diff options
| -rw-r--r-- | docs/RPC.md | 19 | ||||
| -rw-r--r-- | idatui/drive.py | 36 | ||||
| -rw-r--r-- | idatui/pane.py | 16 | ||||
| -rw-r--r-- | idatui/rpc.py | 129 | ||||
| -rw-r--r-- | tests/test_rawimage_rpc.py | 169 |
5 files changed, 368 insertions, 1 deletions
diff --git a/docs/RPC.md b/docs/RPC.md index d1290f5..53ec36b 100644 --- a/docs/RPC.md +++ b/docs/RPC.md @@ -94,6 +94,16 @@ predicate so the returned state is final. | `select` | `index?` | in an open modal list (xrefs/symbols) choose the highlighted (or nth) item and activate it. | | `save` | — | Ctrl+S: persist the `.i64`. | | `close` | — | Escape (dismiss a modal). | +| `define` | `kind`, `target?`, `delay_ms?` | goto `target` (if given) then press the listing key for `kind` ∈ `code`(c) / `func`(p) / `undef`(u) / `thumb`(t) / `thumbscan`(T) / `data`(d) / `string`(a). Leaves hex/decomp for the listing first (those bindings are listing-only). The IDA-side outcome is in `status` (e.g. *defined 228 instructions (0x4370–0x45e8) — control flow ends here*) and in `define.status`. | +| `rename_many` | `items:[{addr,name}]` **or** `file:<json>`, `allow_overwrite?=true` | bulk-apply a symbol map in ONE worker call, then refresh the caches + function table. Accepts `addr`/`start`/`ea`/`address` and `name`/`label`, or a plain `{addr: name}` object. Returns `rename_many:{requested,skipped,ok,failed,errors[]}`. | + +**Raw images: `define` + `rename_many` are the workflow.** A firmware blob loads +with no functions and no names. Point `define thumb` / `define func` at the entry +points you know (IDA's auto-analysis then cascades through the call graph), and +apply the whole symbol file with `rename_many`. Do **not** loop `rename` over a +symbol file: each one costs a navigation (listing page + decompile) plus two +prompt round-trips, i.e. tens of minutes for a few hundred symbols, where +`rename_many` is one call and a few seconds. ### Movement (fast — bare keypresses, pump-only settle) | method | params | effect | @@ -111,6 +121,15 @@ to read the pseudocode → `cursor line=.. col=..` onto a token → `rename name ## Notes / gotchas +- **Load options belong to the first open.** `pane spawn --processor/--base/ + --ida-args` (forwarded to `idatui.launch`) only take effect while there is no + `.i64` yet — IDA bakes them into the database. To change them, delete the + `.i64` (or use Ctrl+L in the TUI) and spawn again. +- **`--processor arm` is AArch64**, and Hex-Rays will not decompile a 32-bit + function in a 64-bit database. For 32-bit ARM firmware use + `--processor arm:ARMv7-A` (see `idatui/formats.py: PROCESSORS`, every name + there verified against a real IDA). + - Settle is the shared `_sync.settle`: drain the message pump, wait for threaded workers, then (for ops with a known outcome) poll a predicate. A verb whose predicate can't be derived (e.g. `rename` on an arbitrary token) falls back to diff --git a/idatui/drive.py b/idatui/drive.py index b6fa641..0e82b0d 100644 --- a/idatui/drive.py +++ b/idatui/drive.py @@ -233,6 +233,39 @@ def cmd_retype(c, args): return _fmt_where(st) +def cmd_define(c, args): + """define <kind> [target ...] — the raw-image workflow (thumb/code/func). + + Several targets are common on a firmware image (a list of entry points from + a symbol file), so take them all and report per-target. + """ + if not args: + raise SystemExit("usage: define <code|func|undef|thumb|thumbscan|data|" + "string> [target ...]") + kind, targets = args[0], (args[1:] or [None]) + out = [] + for t in targets: + try: + st = c.call("define", kind=kind, **({"target": t} if t else {})) + out.append(f" {t or '.'}: {st.get('status', '')}") + except RpcError as e: + out.append(f" {t or '.'}: FAILED: {e}") + return "\n".join(out) + + +def cmd_syms(c, args): + """syms <file.json> — bulk-apply a symbol file ([{addr|start|ea, name}]).""" + if len(args) != 1: + raise SystemExit("usage: syms <symbols.json>") + r = c.call("rename_many", file=os.path.abspath(os.path.expanduser(args[0]))) + m = r.get("rename_many", {}) + out = [f" {m.get('ok', 0)}/{m.get('requested', 0)} renamed" + f" (skipped {m.get('skipped', 0)}, failed {m.get('failed', 0)})"] + for e in m.get("errors", []): + out.append(f" {e.get('addr')}: {e.get('error')}") + return "\n".join(out) + + def cmd_save(c, args): c.call("save") return " saved" @@ -256,7 +289,8 @@ COMMANDS = { "where": cmd_where, "go": cmd_go, "pc": cmd_pc, "dis": cmd_dis, "callees": cmd_callees, "callers": cmd_callers, "names": cmd_names, "rename": cmd_rename, "mv": cmd_mv, "note": cmd_note, "retype": cmd_retype, - "save": cmd_save, "screen": cmd_screen, "raw": cmd_raw, + "save": cmd_save, "screen": cmd_screen, "raw": cmd_raw, "define": cmd_define, + "syms": cmd_syms, "binaries": cmd_binaries, "switch": cmd_switch, } diff --git a/idatui/pane.py b/idatui/pane.py index 37a5f0c..8a2e4a4 100644 --- a/idatui/pane.py +++ b/idatui/pane.py @@ -165,6 +165,14 @@ def spawn(args) -> int: inner += ["--rpc", sock] else: inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock] + # Loading a headerless blob: without these IDA reads a raw firmware image as + # x86 at 0 and analyses to nothing, and the pane comes up ready-but-empty. + # They are launch's options; spawn just forwards them (a project records + # them per binary, so they're only needed on the first open). + for opt in ("processor", "base", "ida_args"): + val = getattr(args, opt, None) + if val: + inner += ["--" + opt.replace("_", "-"), str(val)] if getattr(args, "trace", None): inner += ["--trace", os.path.abspath(os.path.expanduser(args.trace))] cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner) @@ -340,6 +348,14 @@ def main(argv: list[str]) -> int: sp.add_argument("--project", metavar="FILE", help="project file to open instead of a single binary; " "any --open paths are added to it (created if absent)") + sp.add_argument("--processor", metavar="NAME", + help="IDA processor for a headerless blob: arm, armb, " + "mipsb, metapc, … (passed to idatui.launch)") + sp.add_argument("--base", metavar="ADDR", + help="load address for a headerless blob, e.g. 0x8000000 " + "(16-byte aligned)") + sp.add_argument("--ida-args", metavar="STR", dest="ida_args", + help="extra IDA command-line switches, passed through") sp.add_argument("--sock", help="RPC socket path (default: auto in $XDG_RUNTIME_DIR)") sp.add_argument("--python", default=DEFAULT_PY, help=f"python for the TUI ({DEFAULT_PY})") sp.add_argument("--vertical", action="store_true", help="split vertically (stacked)") diff --git a/idatui/rpc.py b/idatui/rpc.py index 717f4df..c356f12 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -38,6 +38,7 @@ _PROGRAM_METHODS = { "goto", "open", "rename", "comment", "retype", "follow", "xrefs", "symbols", "structs", "search", "select", "save", "hex", "toggle_view", "pseudocode", "disassembly", "xrefs_to", "xrefs_from", "resolve", + "define", "rename_many", } # Self-documenting method table (returned by the 'methods' verb). @@ -78,6 +79,23 @@ METHODS = { "close": "dismiss a modal (Escape)", "move": "{dir,n?=1} fast movement (down/up/.../pagedown)", "cursor": "{line?,col?} set the code-pane cursor directly", + "define": "{kind:code|func|undef|thumb|thumbscan|data|string,target?} " + "(re)define bytes at target — the raw-image workflow", + "rename_many": "{items:[{addr,name}] | file:JSON} bulk-apply a symbol file " + "in ONE call (no typing, no navigation)", +} + +# `define` kinds -> the ListingView key that runs them. Driving the real key +# keeps the pane honest (a viewer sees the same thing a human would do) and +# reuses the app's own edit worker, which reports what actually happened. +_DEFINE_KEYS = { + "code": "c", # make code (runs until flow/undecodable) + "func": "p", # make function + "undef": "u", + "thumb": "t", # flip ARM/Thumb at the cursor, then disassemble + "thumbscan": "T", # find Thumb entry pointers in a vector table + "data": "d", + "string": "a", } # Movement keys — driven fast (no typed delay) so the pane still visibly moves. @@ -520,6 +538,80 @@ class RpcServer: await app._press_keys(_text_to_keys(value, delay_ms)) await app._press_keys(["enter"]) + async def _rename_many(self, params: dict[str, Any], timeout: float) -> dict: + """Apply a whole symbol file in one worker call. + + The per-symbol path (goto + typed rename prompt) is the right thing for + one name a human is watching, and hopeless for the case a firmware image + always brings: hundreds of names from a loader map, an emulator's + symbols.json, or another tool's export. Each of those renames costs a + navigation (which pulls a listing page and a decompile) plus two prompt + round-trips, so 400 symbols is tens of minutes of driving and the pane + just flickers. IDA's own rename tool already takes a *list*; this hands + it the whole list, then refreshes the caches and the function table once. + """ + app = self.app + items = params.get("items") + src = params.get("file") + if isinstance(items, str): # `drive raw` hands params through as text + items = json.loads(items) + if items is None: + if not src: + raise ValueError("rename_many needs items=[{addr,name}] or file=<json>") + with open(os.path.expanduser(str(src))) as f: + items = json.load(f) + if isinstance(items, dict): # {"0x4370": "name"} is a natural shape too + items = [{"addr": k, "name": v} for k, v in items.items()] + if not isinstance(items, list) or not items: + raise ValueError("rename_many: items must be a non-empty list") + + ops, skipped = [], 0 + for it in items: + if not isinstance(it, dict): + skipped += 1 + continue + # Accept the field names symbol files actually use. + addr = next((it[k] for k in ("addr", "start", "ea", "address") + if it.get(k) is not None), None) + name = it.get("name") or it.get("label") + if addr is None or not name: + skipped += 1 + continue + ea = int(str(addr), 0) if isinstance(addr, str) else int(addr) + ops.append({"addr": hex(ea), "name": str(name)}) + if not ops: + raise ValueError("rename_many: no usable {addr,name} entries") + + overwrite = params.get("allow_overwrite", True) + if isinstance(overwrite, str): + overwrite = overwrite.lower() not in ("0", "false", "no", "") + batch = {"func": ops, "allow_overwrite": bool(overwrite)} + # The worker is blocking and single-threaded; off the event loop it goes, + # or the TUI freezes for the length of the batch. + res = await asyncio.to_thread(app.program.client.call, "rename", batch=batch) + summary = res.get("summary", {}) if isinstance(res, dict) else {} + failed = [r for r in (res.get("func") or []) if isinstance(r, dict) + and r.get("error")] if isinstance(res, dict) else [] + + # Names live in the IDB, but every cache in front of it is now stale. + app.program.bump_names() + app.program.invalidate_functions() + app._func_index = None + app._load_functions() # re-streams the function table + await settle(app, timeout=timeout) + app._dirty = True + app._status(f"renamed {summary.get('ok', 0)} symbols" + + (f", {len(failed)} failed" if failed else "") + + " (Ctrl+S to save)") + snap = snapshot(app) + snap["rename_many"] = { + "requested": len(ops), "skipped": skipped, + "ok": summary.get("ok", 0), "failed": summary.get("failed", 0), + "errors": [{"addr": r.get("addr"), "error": r.get("error")} + for r in failed[:10]], + } + return snap + def _goto_target_pred(self, target): """A predicate that holds once a goto to ``target`` has landed.""" app = self.app @@ -538,6 +630,7 @@ class RpcServer: _NEEDS_NO_MODAL = { "goto", "open", "rename", "comment", "retype", "follow", "back", "toggle_view", "hex", "save", "search", "move", "cursor", "cursor_on", + "define", } #: Modals the driver is expected to interact with (they have their own verbs). _DRIVABLE_MODALS = {"XrefsScreen", "SymbolPalette", "StructEditor", @@ -741,6 +834,42 @@ class RpcServer: f"(still at {_where(app)}); retry with a larger timeout=") return snapshot(app) + if method == "define": + kind = str(params.get("kind", "code")).lower() + if kind not in _DEFINE_KEYS: + raise ValueError( + f"unknown define kind {kind!r}; one of " + f"{', '.join(sorted(_DEFINE_KEYS))}") + target = params.get("target") + if target not in (None, ""): + # Land on the address first. A raw image is mostly *undefined*, + # so the target usually has no name and no function — the goto + # predicate can't be address-based, only "we moved". + await self._fill_prompt("g", "goto", str(target), delay, + clear=False) + await settle(app, timeout=timeout) + if app._active == "hex": + # backslash leaves hex for the code view (which may be decomp). + await self._press(["backslash"], + lambda: app._active != "hex", timeout, + "leave the hex view") + if app._active == "decomp": + # These bindings live on the listing; in the decompiler the key + # would be swallowed or do something else entirely. + await self._press(["tab"], lambda: app._active == "listing", + timeout, "switch to the listing") + if app._active != "listing": + raise RuntimeError( + f"define needs the listing view, but the active pane is " + f"{app._active!r}") + snap = await self._press([_DEFINE_KEYS[kind]], timeout=timeout, + what=f"define {kind}") + snap["define"] = {"kind": kind, "status": snap.get("status", "")} + return snap + + if method == "rename_many": + return await self._rename_many(params, timeout) + if method == "rename": await self._fill_prompt("n", "rename", str(params["name"]), delay, clear=True) await settle(app, timeout=timeout) diff --git a/tests/test_rawimage_rpc.py b/tests/test_rawimage_rpc.py new file mode 100644 index 0000000..ab7eb96 --- /dev/null +++ b/tests/test_rawimage_rpc.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""The raw-image workflow over the RPC socket: spawn with load options, define, +bulk-apply a symbol file. + +A headerless firmware image is the case where driving IDA from an agent used to +fall apart: + + * ``pane spawn`` could not pass ``--processor``/``--base``, so the pane came up + ready-but-empty (x86 at 0, zero functions) and the only way through was to + hand-write a project file; + * ``c``/``p``/``t`` (make code / make function / ARM-Thumb) existed as key + bindings but had no verb, so a driver had to guess raw keys and hope no + modal was on top; + * every name had to go through the typed rename prompt — a navigation plus two + prompt round-trips each, which is tens of minutes for a 400-symbol map. + +This test spawns a real pane on a real Thumb blob and checks all three. + +Requires: tmux, IDA (idalib). ~2min. + + ~/ida-venv/bin/python tests/test_rawimage_rpc.py +""" +import json +import os +import subprocess +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from idatui.rpcclient import RpcClient, RpcError # noqa: E402 + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +BLOB = os.path.join(REPO, "experiments", "fibonacci.bin") # real Thumb code + +PASS = FAIL = 0 + + +def check(name, ok, detail=""): + global PASS, FAIL + if ok: + PASS += 1 + print(f" ok [{name}]") + else: + FAIL += 1 + print(f" FAIL [{name}] {detail}") + + +def spawn_pane(target, processor, timeout=420): + cmd = [sys.executable, "-m", "idatui.pane", "spawn", "--open", target, + "--processor", processor, "--detached", "--size", "60%", + "--timeout", str(timeout)] + r = subprocess.run(cmd, capture_output=True, text=True, + timeout=timeout + 60, cwd=REPO) + if not r.stdout.strip(): + print(f" spawn produced no JSON: {r.stderr.strip()}", file=sys.stderr) + return None + return json.loads(r.stdout) + + +def stop_pane(sock, timeout=60): + subprocess.run([sys.executable, "-m", "idatui.pane", "stop", "--sock", sock, + "--timeout", str(timeout)], + capture_output=True, text=True, timeout=timeout + 10, cwd=REPO) + + +def main() -> int: + if not os.environ.get("TMUX"): + print("SKIP: not inside tmux") + return 0 + if not os.path.exists(BLOB): + print(f"SKIP: no blob at {BLOB}") + return 0 + + # Work on a copy: the .i64 lands next to the binary and the load options + # only apply to a FIRST open, so a leftover database would silently decide + # what this test measures. + with tempfile.TemporaryDirectory() as tmp: + blob = os.path.join(tmp, "fib.bin") + with open(BLOB, "rb") as src, open(blob, "wb") as dst: + dst.write(src.read()) + + info = spawn_pane(blob, "arm:ARMv7-A") + if not info: + print("SKIP: could not spawn a pane") + return 0 + sock = info["sock"] + try: + # -- load options actually reached IDA --------------------------- # + # Wrong processor => the disassembly is nonsense or absent; ARMv7-A + # also means a 32-bit database, without which Hex-Rays refuses. + check("spawn forwarded --processor", info.get("ok"), + json.dumps(info)) + with RpcClient(sock) as c: + st = c.call("state") + check("pane is drivable", st.get("active") in + ("listing", "decomp", "hex"), json.dumps(st)[:200]) + + # -- define ------------------------------------------------- # + # fibonacci.bin is Thumb at 0x0; as ARM it does not decode. + r = c.call("define", kind="thumb", target="0x0") + d = r.get("define", {}) + check("define thumb ran", "define" in r, json.dumps(r)[:200]) + check("define thumb decoded instructions", + "instruction" in d.get("status", ""), d.get("status", "")) + + r = c.call("define", kind="func", target="0x0") + check("define func created a function", + "function" in r["define"]["status"] + or "already" in r["define"]["status"], + r["define"]["status"]) + + bad = None + try: + c.call("define", kind="nonsense") + except RpcError as e: + bad = str(e) + check("define rejects an unknown kind", bad is not None + and "unknown define kind" in bad, str(bad)) + + # -- rename_many -------------------------------------------- # + fns = c.call("functions", limit=200) + ea = min(f["ea"] for f in fns) if fns else None + check("a function exists to rename", ea is not None) + + symfile = os.path.join(tmp, "syms.json") + with open(symfile, "w") as f: + # 'start' (not 'addr') on purpose: symbol files in the wild + # use it, and accepting only one spelling is how a bulk + # import silently renames nothing. + json.dump([{"start": hex(ea), "name": "bulk_named_fn"}, + {"start": "0xdeadbe", "name": "nowhere"}], f) + r = c.call("rename_many", file=symfile) + m = r.get("rename_many", {}) + check("rename_many applied the good entry", m.get("ok") == 1, + json.dumps(m)) + check("rename_many reports the bad entry", + m.get("failed") == 1 and m.get("errors"), json.dumps(m)) + + # The readback matters more than the return value: a driver + # trusts resolve/functions to decide what work is left. + check("renamed symbol resolves", + c.call("resolve", name="bulk_named_fn").get("ea") == ea, + json.dumps(c.call("resolve", name="bulk_named_fn"))) + names = {f["name"] for f in c.call("functions", limit=200)} + check("function table shows the new name", + "bulk_named_fn" in names, str(sorted(names)[:10])) + + r = c.call("rename_many", items=[{"addr": hex(ea), + "name": "inline_named_fn"}]) + check("rename_many takes inline items", + r["rename_many"]["ok"] == 1, json.dumps(r["rename_many"])) + + empty = None + try: + c.call("rename_many") + except RpcError as e: + empty = str(e) + check("rename_many without items errors", empty is not None + and "items" in empty, str(empty)) + finally: + stop_pane(sock) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) |