From 9583334d0d413d8f4962b5d8fa1eccdbc196dd33 Mon Sep 17 00:00:00 2001 From: blasty Date: Fri, 10 Jul 2026 15:29:21 +0200 Subject: rpc: more verbs — structured reads, modal select, in-view search, save MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introspection (offloaded to a thread so a big fetch can't freeze the render): pseudocode (full body), disassembly (bounded), xrefs_to/from (structured), resolve (name->ea). These let an agent reason about a function and the call graph without scraping screen(). Semantic: select (choose the highlighted/nth item in the open xrefs or symbol-palette list and activate it — the natural follow-up to xrefs/ symbols), search (/ or ? incremental in the active code view), save (Ctrl+S). rpc_smoke exercises all of them end-to-end (18 green). --- docs/RPC.md | 12 ++++++ idatui/rpc.py | 106 +++++++++++++++++++++++++++++++++++++++++++++++++++++ tests/rpc_smoke.py | 56 ++++++++++++++++++++++++++++ 3 files changed, 174 insertions(+) diff --git a/docs/RPC.md b/docs/RPC.md index dcf63b6..0a60cc9 100644 --- a/docs/RPC.md +++ b/docs/RPC.md @@ -45,12 +45,21 @@ acts on stale state. | `text` | `text: str`, `delay_ms?=0`, `settle?`, `timeout?` | type a literal string into the focused input; `delay_ms` interleaves per-char waits for a typed-out look. | ### Introspection (read-only) +Structured reads for an agent to reason without scraping the screen. The heavy +ones (`pseudocode`/`disassembly`/`xrefs_*`) run off the UI loop so the render never +stalls. `target` is a name, `0xADDR`, or omitted (= current function). + | method | params | returns | |--------|--------|---------| | `state` | — | active/pref view, current function `{ea,name}`, cursor `{line,col,word,ea}` (or `{va,byte}` in hex), status text, filter, nav depth, dirty, open `modal`. | | `view` | `lines?` | visible lines of the active code pane, cursor-marked (disasm/decomp; use `screen` for hex). | | `screen` | — | `{width,height,text}` — a full plain-text render of exactly what's on screen. | | `functions` | `filter?`, `limit?=50` | `[{ea,name,size}]`. | +| `pseudocode` | `target?` | `{ea,name,code,failed,error,truncated}` — the **whole** decompiled body. | +| `disassembly` | `target?`, `max?=2000` | `{ea,name,total,lines:[{ea,text}]}`. | +| `xrefs_to` | `target`, `limit?=200` | `[{frm,to,type,fn_addr,fn_name}]` — who references it. | +| `xrefs_from` | `target`, `limit?=200` | `[{frm,to,type,fn_addr,fn_name}]` — what it references. | +| `resolve` | `name` | `{ea}` (or `{ea:null}`). | ### Semantic verbs High-level ops type through the **real prompts** with a per-char delay @@ -70,6 +79,9 @@ predicate so the returned state is final. | `xrefs` | — | `x`: open the xref picker. | | `symbols` | `query?` | Ctrl+N palette, optionally pre-typed. | | `structs` | — | Ctrl+T struct editor. | +| `search` | `term`, `direction?=1` | `/` (or `?`) incremental search in the active code view. | +| `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). | ### Movement (fast — bare keypresses, pump-only settle) diff --git a/idatui/rpc.py b/idatui/rpc.py index e324586..b4b73c6 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -171,6 +171,67 @@ def functions(app, flt: str | None = None, limit: int = 50) -> list[dict[str, An return out +def _target_ea(app, target) -> int | None: + """Resolve a target (None=current function, a name, or 0xADDR/int) to an ea.""" + if target is None: + return app._cur.ea if app._cur else None + if isinstance(target, int): + return target + return app.program.resolve(str(target)) + + +def _func_at(app, ea: int | None): + return app.program.function_of(ea) if ea is not None else None + + +def pseudocode(app, target=None) -> dict[str, Any]: + """Full decompiled body of a function (the whole thing, not the viewport).""" + ea = _target_ea(app, target) + fn = _func_at(app, ea) + dea = fn.addr if fn else ea + if dea is None: + return {"ea": None, "error": "no target"} + d = app.program.decompile(dea) + return {"ea": dea, "name": (fn.name if fn else None), "failed": d.failed, + "error": d.error, "truncated": d.truncated, "code": d.code} + + +def disassembly(app, target=None, max_lines: int = 2000) -> dict[str, Any]: + """Full (bounded) disassembly of a function as text.""" + ea = _target_ea(app, target) + fn = _func_at(app, ea) + dea = fn.addr if fn else ea + if dea is None: + return {"ea": None, "error": "no target"} + m = app.program.disasm(dea, fn.name if fn else None) + total = m.total() + lines = m.lines(0, min(total, max(1, max_lines)), prefetch=False) + return {"ea": dea, "name": (fn.name if fn else None), "total": total, + "lines": [{"ea": ln.ea, "text": ln.text} for ln in lines]} + + +def _xref_dicts(xs, limit: int) -> list[dict[str, Any]]: + return [{"frm": x.frm, "to": x.to, "type": x.type, + "fn_addr": x.fn_addr, "fn_name": x.fn_name} for x in xs[:limit]] + + +def xrefs_to(app, target, limit: int = 200) -> list[dict[str, Any]]: + ea = _target_ea(app, target) + return _xref_dicts(app.program.xrefs_to(ea), limit) if ea is not None else [] + + +def xrefs_from(app, target, limit: int = 200) -> list[dict[str, Any]]: + ea = _target_ea(app, target) + return _xref_dicts(app.program.xrefs_from(ea), limit) if ea is not None else [] + + +def resolve(app, name) -> dict[str, Any]: + try: + return {"ea": app.program.resolve(str(name))} + except Exception: # noqa: BLE001 + return {"ea": None} + + # --------------------------------------------------------------------------- # # Keystroke injection # --------------------------------------------------------------------------- # @@ -314,6 +375,25 @@ class RpcServer: if method == "functions": return functions(app, params.get("filter"), int(params.get("limit", 50))) + # -- structured introspection (heavy: run off the UI loop) -------- # + loop = asyncio.get_running_loop() + if method == "pseudocode": + return await loop.run_in_executor(None, pseudocode, app, params.get("target")) + if method == "disassembly": + mx = int(params.get("max", 2000)) + return await loop.run_in_executor( + None, disassembly, app, params.get("target"), mx) + if method == "xrefs_to": + lim = int(params.get("limit", 200)) + return await loop.run_in_executor( + None, xrefs_to, app, params.get("target"), lim) + if method == "xrefs_from": + lim = int(params.get("limit", 200)) + return await loop.run_in_executor( + None, xrefs_from, app, params.get("target"), lim) + if method == "resolve": + return await loop.run_in_executor(None, resolve, app, params.get("name")) + # -- semantic verbs (typed-with-delay high-level ops) ------------- # delay = int(params.get("delay_ms", TYPE_DELAY_MS)) timeout = float(params.get("timeout", 20.0)) @@ -365,6 +445,32 @@ class RpcServer: ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor", timeout) if method == "close": return await self._press(["escape"], timeout=timeout) + if method == "save": + return await self._press(["ctrl+s"], timeout=timeout) + + if method == "search": + term = str(params.get("term", "")) + open_key = "slash" if int(params.get("direction", 1)) >= 0 else "question_mark" + await self._fill_prompt(open_key, "search", term, delay, clear=True) + await settle(app, timeout=timeout) + return snapshot(app) + + if method == "select": + from textual.widgets import OptionList + scr = app.screen + if type(scr).__name__ not in _MODALS: + raise ValueError("select: no modal list is open") + try: + ol = scr.query_one(OptionList) + except Exception: + raise ValueError("select: the open modal has no list to select from") + idx = params.get("index") + if idx is not None and ol.option_count: + ol.highlighted = max(0, min(ol.option_count - 1, int(idx))) + await drain(app) + await app._press_keys(["enter"]) + await settle(app, timeout=timeout) + return snapshot(app) if method == "move": key = _MOVE_KEYS.get(str(params.get("dir"))) diff --git a/tests/rpc_smoke.py b/tests/rpc_smoke.py index d456552..01418b8 100644 --- a/tests/rpc_smoke.py +++ b/tests/rpc_smoke.py @@ -144,6 +144,62 @@ async def run(db): else: check("found the function name to rename", False, repr(line0_text[:60])) + # --- structured introspection ------------------------------------- # + resp = await c.call("resolve", name=target["name"]) + check("resolve maps a name to its ea", + resp.get("result", {}).get("ea") == target["ea"], str(resp.get("result"))) + + resp = await c.call("pseudocode", target=target["name"]) + pc = resp.get("result", {}) + check("pseudocode returns the full body", + isinstance(pc.get("code"), str) and len(pc["code"]) > 0 and not pc["failed"], + f"failed={pc.get('failed')} len={len(pc.get('code') or '')}") + + resp = await c.call("disassembly", target=target["name"], max=50) + da = resp.get("result", {}) + check("disassembly returns lines with addresses", + isinstance(da.get("lines"), list) and len(da["lines"]) > 0 + and all("ea" in ln and "text" in ln for ln in da["lines"]), + f"total={da.get('total')} n={len(da.get('lines', []))}") + + # a function that is actually referenced, so xrefs_to is non-empty + callee = None + for f in funcs: + xr = (await c.call("xrefs_to", target=f["name"], limit=5)).get("result", []) + if xr: + callee = (f, xr) + break + check("xrefs_to returns structured references", + callee is not None and all("frm" in x for x in callee[1]), + "no referenced function found" if callee is None else "") + + # --- modal select: open xrefs on the callee, pick the first site --- # + if callee is not None: + f, xr = callee + await c.call("goto", target=f["name"], delay_ms=0) + # place the cursor on the function name so xrefs targets it + v = (await c.call("view", lines=1))["result"] + lt = v["lines"][0]["text"] if v.get("lines") else "" + col = lt.find(f["name"]) + if col >= 0: + await c.call("cursor", line=0, col=col + 1) + resp = await c.call("xrefs") + m = resp.get("result", {}).get("modal") or {} + check("xrefs opens the picker with items", + m.get("kind") == "XrefsScreen" and len(m.get("items", [])) > 0, + str(m)[:80]) + resp = await c.call("select", index=0) + check("select follows a picked xref (modal closes, we navigate)", + (resp.get("result", {}).get("modal") is None), str(resp.get("result", {}).get("modal"))) + + # --- in-view search ----------------------------------------------- # + await c.call("goto", target=target["name"], delay_ms=0) + resp = await c.call("search", term="return", delay_ms=0) + st = resp.get("result", {}) + check("search runs without error and returns state", + st.get("active") in ("decomp", "disasm"), str(st.get("active"))) + await c.call("keys", keys=["escape"]) + w.close() print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 -- cgit v1.3.1-sl0p