Add headless Solana SBF ELF loader

README.md

@@ -4,7 +4,20 @@ This is the processor plugin for IDA Pro that adds the ability to analyze Solana
 
 ## How to use
 
-Install `requirements.txt`. Copy `solana-init.py` script and the `solana` folder to the directory `<ida pro installation>/procs` and select the processor on a Solana program file loading to IDA.
+Install `requirements.txt`. Copy `solana-init.py` and the `solana` folder to
+`<ida pro installation>/procs`, then copy `loaders/solana_sbf.py` to
+`<ida pro installation>/loaders`.
+
+The loader recognizes 64-bit little-endian Solana/eBPF ELF files, selects the
+`EBPF` processor automatically, and creates external symbols for imported
+Solana runtime functions. This also makes unattended imports possible:
+
+```sh
+idat -A -T"Solana SBF/eBPF ELF" program.so
+```
+
+Without the custom loader, IDA's stock ELF loader may ask for confirmation
+before the processor module gets a chance to handle an unknown machine ID.
 
 To dump a program from Solana mainnet use the following command:
 
@@ -24,6 +37,27 @@ Then select Yes:
 
 ![](./img/3.png)
 
+## Tests
+
+The loader tests use the included sample and do not require IDA:
+
+```sh
+python3 -m unittest discover -s tests -v
+```
+
+If IDA is installed, the same sample can be imported in a real headless
+session. Install the Python dependencies where IDA's embedded Python can find
+them, or pass that site-packages directory explicitly:
+
+```sh
+python3 tests/run_headless_ida.py \
+  --idat /path/to/idat \
+  --pythonpath /path/to/site-packages
+```
+
+Use `--machine-id 0x107` (or another supported value) to exercise the same
+headless path against a temporary copy of the sample with that ELF machine ID.
+
 
 ## FLIRT signatures
 
@@ -42,4 +76,4 @@ Proceed to the [solana-ida-signatures-factory](https://github.com/PassKeyRa/sola
 
 ## Thanks
 
-Thanks to Clément Berthaux (clement (dot) berthaux (at) synacktiv (dot) com) and Michael Zandi (the (dot) zandi (at) gmail (dot) com) for developing the EBPF processor plugin, which is the base for this plugin.
+Thanks to Clément Berthaux (clement (dot) berthaux (at) synacktiv (dot) com) and Michael Zandi (the (dot) zandi (at) gmail (dot) com) for developing the EBPF processor plugin, which is the base for this plugin.

loaders/solana_sbf.py

@@ -0,0 +1,296 @@
+"""IDA loader for Solana SBF ELF binaries.
+
+IDA's stock ELF loader may ask the user to confirm unknown Solana/eBPF machine
+IDs before a processor module can handle the file.  That prompt prevents
+unattended imports.  This loader recognizes the supported ELF variants,
+selects the Solana eBPF processor, maps allocated sections, and creates an
+external segment for imported runtime symbols.
+"""
+
+from __future__ import annotations
+
+import struct
+from typing import Dict, List, Optional
+
+import ida_idp
+import ida_loader
+import idaapi
+import idc
+
+
+FORMAT = "Solana SBF/eBPF ELF"
+EM_BPF = 247
+EM_SBPF = 263
+MACHINES = {EM_BPF, EM_SBPF}
+ET_DYN = 3
+ELFOSABI_NONE = 0
+
+SHT_NOBITS = 8
+SHT_DYNSYM = 11
+SHF_WRITE = 0x1
+SHF_ALLOC = 0x2
+SHF_EXECINSTR = 0x4
+SHN_UNDEF = 0
+EXTERN_STRIDE = 8
+
+
+def _read_at(li, offset: int, size: int) -> bytes:
+    li.seek(offset)
+    return li.read(size)
+
+
+def _elf_header(li) -> Optional[Dict[str, int]]:
+    header = _read_at(li, 0, 64)
+    if (
+        len(header) < 64
+        or header[:4] != b"\x7fELF"
+        or header[4] != 2
+        or header[5] != 1
+    ):
+        return None
+
+    values = struct.unpack("<16sHHIQQQIHHHHHH", header)
+    return {
+        "type": values[1],
+        "machine": values[2],
+        "osabi": header[7],
+        "entry": values[4],
+        "shoff": values[6],
+        "shentsize": values[11],
+        "shnum": values[12],
+        "shstrndx": values[13],
+    }
+
+
+def _is_supported_elf(elf_header: Optional[Dict[str, int]]) -> bool:
+    return bool(
+        elf_header
+        and elf_header["type"] == ET_DYN
+        and elf_header["machine"] in MACHINES
+        and elf_header["osabi"] == ELFOSABI_NONE
+    )
+
+
+def _section_headers(li, elf_header: Dict[str, int]) -> List[Dict[str, int]]:
+    section_offset = elf_header["shoff"]
+    section_size = elf_header["shentsize"]
+    section_count = elf_header["shnum"]
+    sections = []
+
+    if section_offset <= 0 or section_size < 64 or section_count <= 0:
+        return sections
+
+    for index in range(section_count):
+        raw = _read_at(li, section_offset + index * section_size, section_size)
+        if len(raw) < 64:
+            break
+        values = struct.unpack("<IIQQQQIIQQ", raw[:64])
+        sections.append(
+            {
+                "idx": index,
+                "name_off": values[0],
+                "type": values[1],
+                "flags": values[2],
+                "addr": values[3],
+                "offset": values[4],
+                "size": values[5],
+                "link": values[6],
+                "addralign": values[8],
+                "entsize": values[9],
+            }
+        )
+
+    string_table_index = elf_header["shstrndx"]
+    if not 0 <= string_table_index < len(sections):
+        return sections
+
+    string_table = sections[string_table_index]
+    names = _read_at(li, string_table["offset"], string_table["size"])
+    for section in sections:
+        name_offset = section["name_off"]
+        name_end = names.find(b"\x00", name_offset)
+        if 0 <= name_offset < len(names) and name_end != -1:
+            section["name"] = (
+                names[name_offset:name_end].decode("utf-8", "replace")
+                or "sec_%d" % section["idx"]
+            )
+        else:
+            section["name"] = "sec_%d" % section["idx"]
+    return sections
+
+
+def _dynamic_imports(li, sections: List[Dict[str, int]]) -> List[str]:
+    imports = []
+    seen = set()
+
+    for section in sections:
+        entry_size = section.get("entsize", 0)
+        if section["type"] != SHT_DYNSYM or entry_size < 24:
+            continue
+
+        string_table_index = section.get("link", -1)
+        if not 0 <= string_table_index < len(sections):
+            continue
+        string_table = sections[string_table_index]
+        names = _read_at(li, string_table["offset"], string_table["size"])
+
+        for index in range(section["size"] // entry_size):
+            raw = _read_at(
+                li,
+                section["offset"] + index * entry_size,
+                entry_size,
+            )
+            if len(raw) < 24:
+                continue
+
+            name_offset, _info, _other, section_index, _value, _size = (
+                struct.unpack("<IBBHQQ", raw[:24])
+            )
+            if section_index != SHN_UNDEF or name_offset == 0:
+                continue
+            if not 0 <= name_offset < len(names):
+                continue
+
+            name_end = names.find(b"\x00", name_offset)
+            if name_end == -1:
+                continue
+            name = names[name_offset:name_end].decode("utf-8", "replace")
+            if name and name not in seen:
+                seen.add(name)
+                imports.append(name)
+
+    return imports
+
+
+def accept_file(li, _filename):
+    elf_header = _elf_header(li)
+    if _is_supported_elf(elf_header):
+        return {
+            "format": FORMAT,
+            "processor": "EBPF",
+            "options": ida_loader.ACCEPT_FIRST,
+        }
+    return 0
+
+
+def _segment_class(section: Dict[str, int]) -> str:
+    flags = section["flags"]
+    if flags & SHF_EXECINSTR:
+        return "CODE"
+    if flags & SHF_WRITE:
+        return "DATA"
+    return "CONST"
+
+
+def _align(value: int, alignment: int) -> int:
+    return (value + alignment - 1) & ~(alignment - 1)
+
+
+def _configure_segment(
+    start: int,
+    name: str,
+    segment_class: str,
+    segment_type: Optional[int] = None,
+) -> None:
+    segment = idaapi.getseg(start)
+    if segment is None:
+        return
+
+    idaapi.set_segm_addressing(segment, 2)
+    if segment_type is not None:
+        segment.type = segment_type
+    idc.set_segm_name(start, name)
+    idc.set_segm_class(start, segment_class)
+
+
+def _create_extern_segment(
+    max_loaded_ea: int,
+    imports: List[str],
+) -> Optional[int]:
+    if not imports:
+        return None
+
+    size = len(imports) * EXTERN_STRIDE
+    base = _align(max_loaded_ea + 0x1000, 0x10000)
+    for attempt in range(32):
+        start = base + attempt * 0x10000
+        end = start + size
+        if idaapi.getseg(start) is not None:
+            continue
+        if not idaapi.add_segm(0, start, end, "extern", "XTRN"):
+            continue
+        _configure_segment(start, "extern", "XTRN", idaapi.SEG_XTRN)
+        if idaapi.get_segm_by_name("extern") is not None:
+            return start
+    return None
+
+
+def load_file(li, _neflags, _format):
+    elf_header = _elf_header(li)
+    if not _is_supported_elf(elf_header):
+        return 0
+
+    idaapi.inf_set_app_bitness(64)
+    idaapi.set_processor_type("EBPF", ida_idp.SETPROC_LOADER)
+
+    sections = _section_headers(li, elf_header)
+    loaded = False
+    max_loaded_ea = 0
+    for section in sections:
+        if (
+            not section["flags"] & SHF_ALLOC
+            or section["addr"] == 0
+            or section["size"] == 0
+        ):
+            continue
+
+        start = int(section["addr"])
+        end = start + int(section["size"])
+        name = section.get("name") or "sec_%d" % section["idx"]
+        segment_class = _segment_class(section)
+        if not idaapi.add_segm(0, start, end, name, segment_class):
+            continue
+        _configure_segment(start, name, segment_class)
+
+        if section["type"] != SHT_NOBITS:
+            li.file2base(
+                int(section["offset"]),
+                start,
+                end,
+                ida_loader.FILEREG_PATCHABLE,
+            )
+        loaded = True
+        max_loaded_ea = max(max_loaded_ea, end)
+
+    if not loaded:
+        return 0
+
+    imports = _dynamic_imports(li, sections)
+    extern_base = _create_extern_segment(max_loaded_ea, imports)
+    if extern_base is not None:
+        for index, name in enumerate(imports):
+            address = extern_base + index * EXTERN_STRIDE
+            idc.set_name(address, name, idc.SN_NOCHECK | idc.SN_PUBLIC)
+            idaapi.add_func(address, address + EXTERN_STRIDE)
+
+    entry = int(elf_header["entry"])
+    if entry:
+        idaapi.add_entry(entry, entry, "_start", 1)
+        idc.create_insn(entry)
+
+    extern_segment = idaapi.get_segm_by_name("extern")
+    extern_status = (
+        "0x%x" % extern_segment.start_ea if extern_segment is not None else "none"
+    )
+    print(
+        "[solana_sbf_loader] loaded %s machine=0x%x entry=0x%x "
+        "imports=%d extern=%s"
+        % (
+            FORMAT,
+            elf_header["machine"],
+            entry,
+            len(imports),
+            extern_status,
+        )
+    )
+    return 1

solana/processor.py

@@ -90,18 +90,13 @@ def __init__(self):
         self.functions = {}
         self.sorted_strings = []
 
-    def ev_loader_elf_machine(self, li, machine_type, p_procname, p_pd, loader, reader): # doesn't work from ida python for some reason
-        if machine_type == 247:
-            p_procname = 'Solana VM'
-        return machine_type
-
     def ev_newfile(self, fname):
         for ea, name in idautils.Names():
             name = decode_name(name)
             self.functions[name] = ea
             idaapi.set_name(ea, name, SN_NOCHECK | SN_FORCE) # demangle function names
             seg = idaapi.getseg(ea)
-            if seg.type == idaapi.SEG_XTRN: # create external functions
+            if seg is not None and seg.type == idaapi.SEG_XTRN: # create external functions
                 idaapi.add_func(ea, ea+8)
 
         self.relocations, self.funcs, self.rodata, self.symtab = process_relocations(fname)
@@ -347,7 +342,11 @@ def _ana_call(self, insn):
         insn[0].dtype = dt_dword
 
         if insn.ea in self.relocations:
-            extern_ea = idaapi.get_segm_by_name("extern").start_ea
+            extern_seg = idaapi.get_segm_by_name("extern")
+            if extern_seg is None:
+                insn[0].addr = idaapi.BADADDR
+                return
+            extern_ea = extern_seg.start_ea
             target_addr = idaapi.get_name_ea(extern_ea, self.relocations[insn.ea]['name'])
             if target_addr == idaapi.BADADDR:
                 target_addr = idaapi.get_name_ea(extern_ea, "__imp_" + self.relocations[insn.ea]['name'])