From d1b7d3f4d7fcaf681412af277c37b0837f540609 Mon Sep 17 00:00:00 2001 From: shah Date: Sun, 19 Jul 2026 14:43:32 +0000 Subject: [PATCH] Add headless Solana SBF ELF loader --- README.md | 38 +++- loaders/solana_sbf.py | 296 ++++++++++++++++++++++++++++++++ solana/processor.py | 13 +- tests/ida_headless_smoke.py | 53 ++++++ tests/run_headless_ida.py | 135 +++++++++++++++ tests/test_processor_guards.py | 137 +++++++++++++++ tests/test_solana_sbf_loader.py | 265 ++++++++++++++++++++++++++++ 7 files changed, 928 insertions(+), 9 deletions(-) create mode 100644 loaders/solana_sbf.py create mode 100644 tests/ida_headless_smoke.py create mode 100644 tests/run_headless_ida.py create mode 100644 tests/test_processor_guards.py create mode 100644 tests/test_solana_sbf_loader.py diff --git a/README.md b/README.md index d96b687..d2b998e 100644 --- a/README.md +++ b/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 `/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 +`/procs`, then copy `loaders/solana_sbf.py` to +`/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. \ No newline at end of file +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. diff --git a/loaders/solana_sbf.py b/loaders/solana_sbf.py new file mode 100644 index 0000000..77c1f7c --- /dev/null +++ b/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(" 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(" 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 diff --git a/solana/processor.py b/solana/processor.py index f6a4a00..6ffc81e 100644 --- a/solana/processor.py +++ b/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']) diff --git a/tests/ida_headless_smoke.py b/tests/ida_headless_smoke.py new file mode 100644 index 0000000..5be253e --- /dev/null +++ b/tests/ida_headless_smoke.py @@ -0,0 +1,53 @@ +"""Assertions executed inside IDA by run_headless_ida.py.""" + +import json +import os + +import idaapi +import idautils +import ida_loader +import idc + + +def main(): + idaapi.auto_wait() + + extern = idaapi.get_segm_by_name("extern") + extern_names = [] + if extern is not None: + extern_names = [ + name + for address, name in idautils.Names() + if extern.start_ea <= address < extern.end_ea + ] + normalized_extern_names = sorted( + name[6:] if name.startswith("__imp_") else name + for name in extern_names + ) + + result = { + "function_count": sum(1 for _ in idautils.Functions()), + "extern_names": sorted(extern_names), + "extern_is_external": ( + extern is not None and extern.type == idaapi.SEG_XTRN + ), + "normalized_extern_names": normalized_extern_names, + "has_text_segment": idaapi.get_segm_by_name(".text") is not None, + "loader_format": ida_loader.get_file_type_name(), + } + output_path = os.environ["SOLANA_IDA_SMOKE_OUT"] + with open(output_path, "w", encoding="utf-8") as output: + json.dump(result, output, indent=2, sort_keys=True) + + required_imports = {"abort", "sol_log_", "sol_memcpy_"} + passed = ( + result["function_count"] > 0 + and result["has_text_segment"] + and result["extern_is_external"] + and result["loader_format"] == "Solana SBF/eBPF ELF" + and required_imports.issubset(normalized_extern_names) + ) + idc.qexit(0 if passed else 1) + + +main() diff --git a/tests/run_headless_ida.py b/tests/run_headless_ida.py new file mode 100644 index 0000000..7e16afc --- /dev/null +++ b/tests/run_headless_ida.py @@ -0,0 +1,135 @@ +"""Run a reproducible end-to-end import with an installed IDA.""" + +import argparse +import json +import os +from pathlib import Path +import shutil +import struct +import subprocess +import sys +import tempfile + + +ROOT = Path(__file__).resolve().parents[1] +FORMAT = "Solana SBF/eBPF ELF" + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("--idat", required=True, type=Path) + parser.add_argument( + "--base-ida-user", + type=Path, + default=Path(os.environ.get("IDAUSR", Path.home() / ".idapro")), + help=( + "Existing IDA user directory whose local runtime configuration " + "is copied into the isolated test directory." + ), + ) + parser.add_argument( + "--sample", + type=Path, + default=ROOT / "samples" / "hello_world.so", + ) + parser.add_argument( + "--pythonpath", + action="append", + default=[], + help="Additional path visible to IDA's embedded Python; repeat as needed.", + ) + parser.add_argument( + "--machine-id", + type=lambda value: int(value, 0), + choices=(0x00F7, 0x0107), + help=( + "Temporarily set the sample's ELF machine ID before importing it. " + "Accepts decimal or 0x-prefixed values." + ), + ) + parser.add_argument("--timeout", type=int, default=120) + return parser.parse_args() + + +def main(): + args = parse_args() + idat = args.idat.resolve() + sample = args.sample.resolve() + if not idat.is_file(): + raise SystemExit("idat executable not found: %s" % idat) + if not sample.is_file(): + raise SystemExit("sample not found: %s" % sample) + + with tempfile.TemporaryDirectory(prefix="solana-ida-smoke-") as temp_name: + temp = Path(temp_name) + idausr = temp / "idausr" + (idausr / "loaders").mkdir(parents=True) + (idausr / "procs").mkdir() + for config_name in ("ida.reg", "ida-config.json"): + source = args.base_ida_user / config_name + if source.is_file(): + shutil.copy2(source, idausr / config_name) + shutil.copy2(ROOT / "loaders" / "solana_sbf.py", idausr / "loaders") + shutil.copy2(ROOT / "solana-init.py", idausr / "procs") + shutil.copytree(ROOT / "solana", idausr / "procs" / "solana") + + result_path = temp / "result.json" + database_path = temp / "sample.i64" + analysis_sample = sample + if args.machine_id is not None: + sample_data = bytearray(sample.read_bytes()) + if len(sample_data) < 20 or sample_data[:4] != b"\x7fELF": + raise SystemExit("cannot patch machine ID in a non-ELF sample") + struct.pack_into("