"""In-process RPC server: puppeteer the live TUI over a unix socket. The TUI renders normally in its terminal (what a viewer/livestream sees); this listener runs on the *same* asyncio loop, so handlers touch the UI directly with no thread marshalling. Every injected key produces the identical on-screen effect a human keyboard would — that's the whole point (drive it "as if a user"). Protocol: newline-delimited JSON, one request/response per line. -> {"id": 1, "method": "keys", "params": {"keys": ["g","m","a","i","n","enter"]}} <- {"id": 1, "result": {...}} (or {"id":1,"error":{"message":..}}) Anyone who can r/w the socket can drive the app (no auth by design; the socket is mode 0600, local only). Method tiers: raw keys, text — inject keystrokes (max fidelity) introspect state, view, screen, functions (semantic verbs — open/goto/rename/... — layer on top in a later pass.) """ from __future__ import annotations import asyncio import io import json import os from typing import Any from rich.console import Console from ._sync import drain, settle from . import diag from .app import DecompView, GraphView, HexView, ListingView, ViewMode PROTO_VERSION = 1 TYPE_DELAY_MS = 35 # default per-char delay for high-level typed ops (aesthetic) # Verbs that dereference app.program — refused with a clear error before load. _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", "opfmt", "graph", } # Self-documenting method table (returned by the 'methods' verb). METHODS = { "ping": "-> {ok,proto,module,ready,functions,complete}", "methods": "-> this table", "quit": "gracefully exit the TUI", "state": "-> full state snapshot", "view": "{lines?} -> visible code-pane lines (disasm/decomp)", "screen": "{format?=text|html|svg} -> {width,height,text} full render", "functions": "{filter?,limit?=50} -> [{ea,name,size}]", "pseudocode": "{target?} -> full decompiled body", "disassembly": "{target?,max?=2000} -> {total,lines:[{ea,text}]}", "xrefs_to": "{target,limit?=200} -> [{frm,to,type,fn_addr,fn_name}]", "xrefs_from": "{target,limit?=200} -> callees/refs; function-scoped for a " "function (decomp refs), address-scoped for a 0xADDR", "resolve": "{name} -> {ea}", "keys": "{keys:[str],settle?,timeout?} raw key injection (supports 'wait:')", "text": "{text,delay_ms?,settle?} type a literal string into the focused input", "goto/open": "{target,delay_ms?} g-prompt to a name or 0xADDR", "rename": "{name,word?,delay_ms?} rename the token under (or 'word') the cursor", "comment": "{text} comment the current line (use \\n for newlines)", "retype": "{proto,word?,delay_ms?} set the prototype/type under the cursor", "follow": "{word?} follow the reference under (or 'word') the cursor", "cursor_on": "{word,line?,occurrence?=1} place the cursor on a token", "back": "pop the nav stack", "toggle_view": "disasm <-> pseudocode", "hex": "hex view", "graph": "{action?=show|open|close|toggle|zoom|block|entry|succ|pred," "target?,blocks?} the control-flow graph: 'show' reports its " "structure (blocks, edges, cursor) without touching it; the others " "drive it. 'block' takes target=", "xrefs": "open the xref picker", "symbols": "{query?} open the symbol palette", "structs": "open the struct editor", "search": "{term,direction?=1} incremental search in the code view", "select": "{index?} choose the highlighted/nth item in the open modal", "save": "persist the .i64 (Ctrl+S)", "trace": "{seek|goto|step,over?} navigate the execution trace (seek '!50' = percent)", "binaries": "-> project binaries {label,active,resident,indexed} (project mode)", "switch": "{binary,addr?} make another project binary active (addr also jumps)", "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)", "opfmt": "{mode?=cycle|back|show|hex|dec|oct|bin|char|offset|stack|" "default,target?,word?,line?,col?} how the literal under the cursor is " "DISPLAYED (IDA's 'o'); works on the listing and on pseudocode " "numbers. 'show' reports the format and the stops without editing", } #: `opfmt` modes that have a real key on the code views. Driving the key keeps #: the pane honest (a viewer sees the same thing a human would do); the named #: formats have no key, so those go through the view's action directly. _OPFMT_KEYS = {"cycle": "o", "back": "O"} _OPFMT_MODES = ("cycle", "back", "show", "hex", "dec", "oct", "bin", "char", "offset", "stack", "default") # `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. _MOVE_KEYS = { "down": "j", "up": "k", "left": "h", "right": "l", "word": "w", "wordback": "b", "bol": "0", "eol": "dollar_sign", "top": "home", "bottom": "G", "halfdown": "ctrl+d", "halfup": "ctrl+u", "pagedown": "pagedown", "pageup": "pageup", } # --------------------------------------------------------------------------- # # State introspection # --------------------------------------------------------------------------- # def _active_widget(app): """The currently *shown* code widget (mirrors app._active).""" if app.is_hex: return app.query_one(HexView) if app.is_graph: return app.query_one(GraphView) if app.is_listing: return app.query_one(ListingView) return app.query_one(DecompView) def graph_info(app, blocks: bool = True) -> dict[str, Any]: """Structured view of the control-flow graph: what a driver actually wants, rather than the box-drawing characters it is rendered as.""" gv = app.query_one(GraphView) if gv.fc is None or gv.lay is None: return {"open": app.is_graph, "loaded": False, "note": "press space (or graph {action:'open'}) on a function"} lay, fc = gv.lay, gv.fc out: dict[str, Any] = { "open": app.is_graph, "loaded": True, "func": {"name": fc.name, "ea": fc.func_ea, "entry": fc.entry}, "zoom": gv.ZOOMS[gv._zoom], "canvas": {"w": lay.width, "h": lay.height}, "stats": dict(lay.stats), "cursor": {"block": gv.cursor_node, "row": gv.cursor_row, "ea": gv._cursor_ea(), "word": gv.word_under_cursor()}, } if blocks: rows = [] for n in lay.nodes: b = gv._blocks.get(n.id) rows.append({ "id": n.id, "start": b.start if b else None, "end": b.end if b else None, "insns": len(b.rows) if b else 0, "rank": n.rank, "box": {"x": n.x, "y": n.y, "w": n.w, "h": n.h}, "succs": [{"id": i, "kind": k} for i, k in lay.succ.get(n.id, [])], "preds": [{"id": i, "kind": k} for i, k in lay.pred.get(n.id, [])], "selfloop": bool(b and any(d == n.id for d, _ in b.succs)), }) out["blocks"] = rows return out _MODALS = ("XrefsScreen", "SymbolPalette", "StructEditor", "ConfirmScreen") #: ``drive raw`` (and any k=v CLI) hands every param through as a *string*. #: Handlers that did ``int(...)`` coped; the ones that compared directly blew up #: with e.g. "'<' not supported between instances of 'int' and 'str'". Coerce the #: known-numeric names once, centrally, instead of at every call site. _INT_PARAMS = ("lines", "limit", "max", "n", "index", "line", "col", "occurrence", "delay_ms", "direction", "addr", "count") _FLOAT_PARAMS = ("timeout",) def _coerce_params(params: dict[str, Any]) -> dict[str, Any]: out = dict(params) for k in _INT_PARAMS: v = out.get(k) if isinstance(v, str) and v.strip(): try: out[k] = int(v, 0) except ValueError: pass for k in _FLOAT_PARAMS: v = out.get(k) if isinstance(v, str) and v.strip(): try: out[k] = float(v) except ValueError: pass return out def _modal_snapshot(app) -> dict[str, Any] | None: """Describe the top modal screen, if any, enough to drive it.""" scr = app.screen name = type(scr).__name__ if name not in _MODALS: return None # the base app Screen is not a modal info: dict[str, Any] = {"kind": name} items = getattr(scr, "_items", None) if isinstance(items, list): try: from textual.widgets import OptionList hl = scr.query_one(OptionList).highlighted except Exception: # noqa: BLE001 hl = None info["highlighted"] = hl info["items"] = [ {"ea": (it[0] if isinstance(it[0], int) else None), "label": str(it[1]) if len(it) > 1 else str(it)} for it in items[:64] ] return info def _cursor_info(app, w) -> dict[str, Any]: if isinstance(w, HexView): return {"kind": "hex", "va": (w.cursor_va() if w.model else None), "byte": w.cursor} if isinstance(w, GraphView): # The graph cursor is (block, row), not a line index -- reporting it as # one would make a driver's `cursor line=` land somewhere arbitrary. return {"kind": "graph", "ea": w._cursor_ea(), "block": w.cursor_node, "row": w.cursor_row, "col": w.cursor_x, "word": w.word_under_cursor(), "text": w._line_plain()} # disasm / decomp share the ColumnCursor surface word = None try: word = w.word_under_cursor() except Exception: # noqa: BLE001 pass ea = None try: ea = app._line_ea_for(w) except Exception: # noqa: BLE001 pass return {"kind": app._active, "line": w.cursor, "col": w.cursor_x, "word": word, "ea": ea, "total": getattr(w, "total", None), "scroll_y": round(w.scroll_offset.y)} def _where(app) -> str: """Short 'name @ 0xea' for error messages that need to say where we ended up.""" cur = getattr(app, "_cur", None) if cur is None: return "nowhere" return f"{getattr(cur, 'name', '?')} @ {getattr(cur, 'ea', 0):#x}" def _readiness(app) -> dict[str, Any]: """Whether the app is drivable yet, and how far function-loading has got. (Cheap: no network — never call client.health() here.)""" idx = app._func_index n = len(idx.all_loaded()) if idx is not None else 0 return { "ready": app.program is not None and idx is not None and n > 0, "functions": n, "complete": bool(idx is not None and idx.complete), } def snapshot(app) -> dict[str, Any]: """A structured view of everything an agent needs to decide the next move.""" cur = app._cur w = _active_widget(app) st = "" try: st = str(app.query_one("#status").render()) except Exception: # noqa: BLE001 pass return { "active": app._active, "pref": app._code_mode(), # kept for wire compat; a constant now "function": ({"ea": cur.ea, "name": cur.name} if cur else None), "cursor": _cursor_info(app, w), "status": st, "filter": app._filter_term, "binary": app._binary, # None outside project mode "nav_depth": len(app._nav), "hops": list(getattr(app, "_hops", [])), "dirty": bool(app._dirty), "modal": _modal_snapshot(app), **_readiness(app), } def view_lines(app, lines: int | None = None) -> dict[str, Any]: """The visible text of the active code pane, cursor-marked. (disasm/decomp; for hex use screen()).""" w = _active_widget(app) if isinstance(w, HexView): return {"active": "hex", "note": "use screen() for the hex grid", "cursor": _cursor_info(app, w)} if isinstance(w, GraphView): return {"active": "graph", "note": "use graph() for structure, " "screen() for the drawing", "cursor": _cursor_info(app, w), "graph": graph_info(app, blocks=False)} top = round(w.scroll_offset.y) height = w.size.height or 40 n = min(lines or height, max(w.total - top, 0)) out = [] for r in range(n): idx = top + r plain = w._line_plain(idx) out.append({"i": idx, "cur": idx == w.cursor, "text": plain if plain is not None else ""}) return {"active": app._active, "top": top, "total": w.total, "cursor": _cursor_info(app, w), "lines": out} def screen_text(app, fmt: str = "text") -> dict[str, Any]: """Render the whole screen exactly as shown. ``fmt``: 'text' (plain, default), 'html' or 'svg' (colored — handy for an out-of-band web viewer).""" width, height = app.size console = Console(width=width, height=height or 40, file=io.StringIO(), force_terminal=True, color_system="truecolor", record=True, legacy_windows=False, safe_box=False) render = app.screen._compositor.render_update( full=True, screen_stack=app._background_screens, simplify=False) console.print(render) out: dict[str, Any] = {"width": width, "height": height, "format": fmt} if fmt == "html": out["text"] = console.export_html(inline_styles=True) elif fmt == "svg": out["text"] = console.export_svg(title=app.title) else: out["text"] = console.export_text(styles=False) return out def place_cursor(w, line=None, col=None) -> None: """Move a code view's cursor and BRING IT INTO VIEW. Setting the cursor without scrolling leaves the pane showing somewhere else entirely, and the next verb then edits a line the operator cannot see — the status describes one thing, the screen shows another. Every programmatic cursor move goes through here for that reason. """ if line is not None: w.cursor = max(0, min(getattr(w, "total", 1) - 1, int(line))) if col is not None: w.cursor_x = max(0, int(col)) if hasattr(w, "_after_cursor_move"): w._after_cursor_move() if hasattr(w, "_scroll_cursor_into_view"): w._scroll_cursor_into_view() if hasattr(w, "_hscroll"): w._hscroll() w.refresh() def cursor_on(app, word: str, line: int | None = None, occurrence: int = 1) -> bool: """Place the cursor on the ``occurrence``-th token equal to ``word`` in the active code pane (optionally restricted to ``line``). Verified with the app's own tokenizer so 'main' won't match inside 'domain'. Disasm scan is limited to already-cached lines (what's on/near screen); decomp searches the whole body. Returns whether it found and moved. Search starts at the VIEWPORT, not at row 0. A continuous listing is the whole segment, so counting from the top finds an occurrence in some unrelated function thousands of rows away -- and the cursor then lands there, off screen, where the next verb edits something the operator cannot see. Wrapping to the rows above keeps every match reachable; landing scrolls, so wherever it goes is visible. """ w = _active_widget(app) if isinstance(w, HexView): raise ValueError("cursor_on: not supported in the hex view") if isinstance(w, GraphView): raise ValueError("cursor_on: not supported in the graph view — use " "graph {action:'block'} or goto") if isinstance(w, DecompView): texts = list(w._texts) else: texts = [(w._line_plain(i) or "") for i in range(getattr(w, "total", 0))] orig = (w.cursor, w.cursor_x) if line is not None: rows = [line] else: # From the top of the viewport, then wrap round to what's above it. top = round(w.scroll_offset.y) rows = list(range(top, len(texts))) + list(range(0, top)) hits = 0 for i in rows: if not (0 <= i < len(texts)): continue t = texts[i] col = t.find(word) while col != -1: w.cursor, w.cursor_x = i, col if w.word_under_cursor() == word: hits += 1 if hits >= max(1, occurrence): place_cursor(w) # scrolls: an off-screen cursor edits blind return True col = t.find(word, col + 1) w.cursor, w.cursor_x = orig # not found: leave the cursor untouched return False def functions(app, flt: str | None = None, limit: int = 50) -> list[dict[str, Any]]: idx = app._func_index if idx is None: return [] flt = (flt or "").lower() out = [] for f in idx.all_loaded(): if flt and flt not in f.name.lower(): continue out.append({"ea": f.addr, "name": f.name, "size": f.size}) if len(out) >= limit: break 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, "kind": x.kind, "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]]: """References *out of* the target. For a function (a bare name / its entry ea) this is whole-body — the callees, string and data refs from the decompiler (the server's xrefs_from is address-scoped, so it only sees the entry instruction). For an explicit mid-function 0xADDR it stays address-scoped.""" ea = _target_ea(app, target) if ea is None: return [] fn = app.program.function_of(ea) if fn is not None and fn.addr == ea: # function target -> decomp outgoing refs out = [] for r in app.program.decompile(ea).refs[:limit]: tf = app.program.function_of(r.addr) is_func = bool(tf and tf.addr == r.addr) out.append({"to": r.addr, "name": r.name or (tf.name if tf else None), "string": r.string, "is_func": is_func, "type": "code" if is_func else "data"}) return out return _xref_dicts(app.program.xrefs_from(ea), limit) def resolve(app, name) -> dict[str, Any]: try: return {"ea": app.program.resolve(str(name))} except Exception: # noqa: BLE001 return {"ea": None} # --------------------------------------------------------------------------- # # Keystroke injection # --------------------------------------------------------------------------- # def _text_to_keys(text: str, delay_ms: int = 0) -> list[str]: """Each printable char becomes a key; interleave wait: for the typed-out aesthetic (App._press_keys understands 'wait:NN').""" keys: list[str] = [] for i, ch in enumerate(text): if delay_ms and i: keys.append(f"wait:{delay_ms}") keys.append(ch) return keys # --------------------------------------------------------------------------- # # Server # --------------------------------------------------------------------------- # class RpcServer: def __init__(self, app, path: str): self.app = app self.path = path self._server: asyncio.AbstractServer | None = None self._busy = False # single-driver gate: one client connection at a time async def start(self) -> None: if os.path.exists(self.path): try: os.unlink(self.path) except OSError: pass self._server = await asyncio.start_unix_server(self._on_client, path=self.path) try: os.chmod(self.path, 0o600) except OSError: pass async def stop(self) -> None: if self._server is not None: self._server.close() try: await self._server.wait_closed() except Exception: # noqa: BLE001 pass try: if os.path.exists(self.path): os.unlink(self.path) except OSError: pass async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: if self._busy: # No multi-driver support yet: refuse a second concurrent client # rather than let two drivers interleave mutations. try: writer.write(json.dumps( {"id": None, "error": {"message": "busy: another client is " "connected (single-driver only)"}}).encode() + b"\n") await writer.drain() writer.close() except Exception: # noqa: BLE001 pass return self._busy = True try: while not reader.at_eof(): line = await reader.readline() if not line: break resp = await self._handle_line(line) writer.write(json.dumps(resp).encode() + b"\n") await writer.drain() except (ConnectionResetError, BrokenPipeError): pass finally: self._busy = False try: writer.close() except Exception: # noqa: BLE001 pass async def _handle_line(self, line: bytes) -> dict[str, Any]: rid = None try: req = json.loads(line.decode()) rid = req.get("id") method = req.get("method") params = req.get("params") or {} result = await self._dispatch(method, params) return {"id": rid, "result": result} except Exception as e: # noqa: BLE001 — report, never kill the connection # ``str(KeyError("msg"))`` returns ``repr("msg")`` (adds quotes), which # mangles our friendly resolve messages; unwrap the single arg instead. if isinstance(e, KeyError) and len(e.args) == 1 and isinstance(e.args[0], str): msg = e.args[0] else: msg = str(e) return {"id": rid, "error": {"message": f"{type(e).__name__}: {msg}"}} # -- composed helpers (semantic verbs) -------------------------------- # async def _press(self, keys, pred=None, timeout=20.0, what=""): await self.app._press_keys([str(k) for k in keys]) ok = await settle(self.app, pred, timeout=timeout) if pred is not None and not ok: # Never report success for an action that did not happen: the caller # would go on to edit whatever the *previous* location was. raise TimeoutError( f"{what or 'action'} did not complete within {timeout}s " f"(still at {_where(self.app)}); retry with a larger timeout=") return snapshot(self.app) async def _graph(self, params, timeout): """Drive / read the control-flow graph. Everything goes through the real keys and the real view state, so a driver sees exactly what a person would -- and 'show' is a pure read, which is what you want between edits. """ app = self.app action = str(params.get("action") or "show").lower() gv = app.query_one(GraphView) want_blocks = params.get("blocks", True) not in (False, "false", "0", 0) if action == "show": return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)} if action in ("open", "toggle", "close"): if action == "open" and app.is_graph: return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)} if action == "close" and not app.is_graph: return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)} want = "graph" if action in ("open", "toggle") and \ not app.is_graph else None res = await self._press( ["space"], (lambda: app.is_graph) if want else (lambda: not app.is_graph), timeout, f"graph {action}") return {**res, "graph": graph_info(app, blocks=want_blocks)} if not app.is_graph: raise ValueError(f"graph {action}: the graph is not open " f"(graph {{action:'open'}} first)") if action == "zoom": before = gv._zoom await self._press(["z"], lambda: gv._zoom != before, timeout, "graph zoom") elif action == "entry": await self._press(["0"], None, timeout, "graph entry") elif action in ("succ", "pred"): before = gv.cursor_node await self._press(["J" if action == "succ" else "K"], lambda: gv.cursor_node != before, timeout, f"graph {action}") elif action == "block": target = params.get("target") if target is None: raise ValueError("graph block: need target=") nid = None s = str(target) if s.startswith("0x") or s.startswith("0X"): ea = int(s, 16) b = gv.fc.block_at(ea) if gv.fc else None if b is None: raise ValueError(f"graph block: {s} is not in this graph") nid = b.id else: nid = int(s) if gv.lay is None or nid not in gv.lay.by_id: raise ValueError(f"graph block: no block {nid}") gv.cursor_node = nid gv.cursor_row = 0 gv.cursor_x = 0 gv._clamp_cursor() gv._center_cursor() gv.refresh() await settle(app, None, timeout=2) else: raise ValueError(f"graph: unknown action {action!r}") return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)} async def _fill_prompt(self, open_key, input_id, value, delay_ms, clear): """Open a prompt (a keystroke), optionally clear its prefill, type the value with the typed-out delay, submit. Returns after the prompt closes.""" from textual.widgets import Input app = self.app await app._press_keys([open_key]) await settle(app, lambda: app.query_one(f"#{input_id}", Input).display, timeout=10) inp = app.query_one(f"#{input_id}", Input) if not inp.display: # Say *why*. The old message always blamed the word under the cursor, # which sent readers hunting for a cursor problem when the real cause # was usually a modal eating the opening keystroke. modal = type(app.screen).__name__ why = (f"modal {modal!r} has focus and ate the {open_key!r} keystroke" if modal in _MODALS or modal != "Screen" else "no renameable token under the cursor") raise RuntimeError(f"{input_id!r} prompt did not open: {why}") if clear: inp.value = "" 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=") 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 -- # including Hex-Rays', which is per-function and does NOT notice that a # *callee* was renamed. That cache is persisted in the .i64, so without # this a batch import leaves pseudocode calling sub_98C0 forever while # the listing (and every readback) says memset. try: await asyncio.to_thread(app.program.client.call, "force_recompile") except Exception: # noqa: BLE001 -- older worker without the tool pass 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 try: ea = app.program.resolve(target) except Exception: # noqa: BLE001 — unknown name; caller falls back to generic return None if app.is_hex: return lambda: app.query_one(HexView).cursor_va() == ea fn = app.program.function_of(ea) want = fn.addr if fn else ea return lambda: app._cur is not None and app._cur.ea == want #: Verbs that drive the *main* app by injecting keystrokes. If a modal is on #: top it eats those keys, so they must refuse rather than silently no-op. _NEEDS_NO_MODAL = { "goto", "open", "rename", "comment", "retype", "follow", "back", "toggle_view", "hex", "save", "search", "move", "cursor", "cursor_on", "define", "opfmt", } #: Modals the driver is expected to interact with (they have their own verbs). _DRIVABLE_MODALS = {"XrefsScreen", "SymbolPalette", "StructEditor", "ProjectPalette", "QuitScreen"} def _modal_kind(self) -> str | None: scr = self.app.screen name = type(scr).__name__ return name if name in _MODALS or name in self._DRIVABLE_MODALS else None async def _dispatch(self, method: str, params: dict[str, Any]) -> Any: app = self.app params = _coerce_params(params) if method in self._NEEDS_NO_MODAL: modal = self._modal_kind() if modal is not None: raise RuntimeError( f"modal {modal!r} is on top and will swallow this verb's " f"keystrokes; dismiss it first (close) or use its own verb " f"(select/symbols/xrefs). Note: a binary with no entry " f"function can land in the symbol palette on startup.") if method in (None, "ping"): module = None try: module = app._module() if app.client else None except Exception: # noqa: BLE001 pass return {"ok": True, "proto": PROTO_VERSION, "module": module, **_readiness(app)} if method == "methods": return METHODS if method == "quit": # Route through the same teardown a human gets, so a dirty database # is written instead of dropped. `app.exit()` alone skips the dirty # check entirely, and the caller (pane stop) then kills the pane -- # which used to destroy a whole session's annotations. dirty = list(app._dirty_labels()) save = params.get("save", True) if isinstance(save, str): save = save.lower() not in ("0", "false", "no", "") def _go(): if dirty and save: app._on_quit_choice("save") # saves, then exits else: app._on_quit_choice("discard") # answer first, then tear down (so this response still gets written) asyncio.get_running_loop().call_later(0.2, _go) return {"ok": True, "quitting": True, "saving": bool(dirty and save), "dirty": dirty} if method in _PROGRAM_METHODS and app.program is None: raise ValueError("not ready: still connecting / loading functions") if method == "keys": keys = params.get("keys") or [] if not isinstance(keys, list): raise ValueError("keys must be a list") await app._press_keys([str(k) for k in keys]) if params.get("settle", True): await settle(app, timeout=float(params.get("timeout", 20.0))) return snapshot(app) if method == "text": text = str(params.get("text", "")) keys = _text_to_keys(text, int(params.get("delay_ms", 0))) await app._press_keys(keys) if params.get("settle", True): await settle(app, timeout=float(params.get("timeout", 20.0))) return snapshot(app) if method == "state": return snapshot(app) if method == "view": return view_lines(app, params.get("lines")) if method == "screen": return screen_text(app, str(params.get("format", "text"))) if method == "functions": return functions(app, params.get("filter"), int(params.get("limit", 50))) # -- projects ------------------------------------------------------ # if method == "diag": # What has been swallowed lately. A driver that got "success" and a # pane showing nothing has no other way to ask; inside a # full-screen TUI there is nowhere for a traceback to go. if params.get("clear"): diag.clear() return {"cleared": True} return {"recent": diag.recent(int(params.get("n", 10))), "log": os.environ.get("IDATUI_LOG") or None} if method == "trace": tc = app.trace_ctl if tc.trace is None: raise ValueError("no trace loaded (launch with --trace FILE)") t = tc.trace if "seek" in params: v = params["seek"] # "!50" seeks a percentage, like Tenet's timestamp shell. if isinstance(v, str) and v.startswith("!"): idx = int(float(v[1:]) * (t.length - 1) / 100.0) else: idx = int(str(v).replace(",", ""), 0) if isinstance(v, str) else int(v) tc.seek(idx) elif "goto" in params: # first execution of an address/name tgt = params["goto"] ea = (int(str(tgt), 0) if str(tgt).lower().startswith("0x") else app.program.resolve(str(tgt))) first = t.first_execution(ea) if first is None: raise ValueError(f"{tgt} never executed in this trace") tc.seek(first) elif "step" in params: n = int(params.get("step") or 1) over = bool(params.get("over")) for _ in range(abs(n)): (tc.step_over if over else tc.step)(1 if n > 0 else -1) await settle(app, timeout=float(params.get("timeout", 20.0))) snap = snapshot(app) snap["trace"] = {"idx": tc.t, "length": t.length, "pc": hex(t.ip(tc.t)), "changed": sorted(t.changed(tc.t))} return snap if method == "binaries": if app._project is None: raise ValueError("not a project session (launch with --project)") counts = app._index.counts() if app._index is not None else {} resident = set(app._pool.resident()) if app._pool is not None else set() return {"active": app._binary, "hops": list(app._hops), "binaries": [{"label": r.label, "source": r.source, "active": r.label == app._binary, "resident": r.label in resident, "indexed": int(counts.get(r.label, 0))} for r in app._project.refs]} if method == "switch": if app._project is None: raise ValueError("not a project session (launch with --project)") label = str(params.get("binary") or params.get("label") or "") if app._project.by_label(label) is None: have = ", ".join(r.label for r in app._project.refs) raise ValueError(f"no such binary {label!r} (have: {have})") addr = params.get("addr") if label == app._binary and addr is None: return snapshot(app) if addr is None: app._switch_binary(label) else: # Same path a project search hit takes, so it records a hop and # Esc comes back here. app._switch_then_goto(label, int(str(addr), 0) if isinstance(addr, str) else int(addr)) await settle(app, lambda: app._binary == label and app._func_index is not None and app._func_index.complete, timeout=float(params.get("timeout", 300.0))) return snapshot(app) # -- 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)) # optional ergonomic: place the cursor on a token before an edit/follow if method in ("rename", "retype", "follow") and params.get("word"): if not cursor_on(app, str(params["word"]), params.get("line"), int(params.get("occurrence", 1))): raise ValueError(f"cursor_on: token {params['word']!r} not found " "in the current view") await drain(app) if method == "cursor_on": found = cursor_on(app, str(params["word"]), params.get("line"), int(params.get("occurrence", 1))) await drain(app) snap = snapshot(app) snap["found"] = found return snap if method in ("goto", "open"): target = str(params.get("target", "")) pred = self._goto_target_pred(target) await self._fill_prompt("g", "goto", target, delay, clear=False) ok = await settle(app, pred, timeout=timeout) if pred is not None and not ok: # A goto that silently "succeeds" without moving is worse than an # error: on a big database the listing build can outrun the # default timeout, and every subsequent rename/comment then lands # on the function the caller *used* to be looking at. raise TimeoutError( f"goto {target!r} did not land within {timeout}s " 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.is_hex: # backslash leaves hex for the code view (which may be decomp). await self._press(["backslash"], lambda: not app.is_hex, timeout, "leave the hex view") if app.is_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.is_listing, timeout, "switch to the listing") if not app.is_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 == "opfmt": mode = str(params.get("mode", "cycle")).lower() if mode not in _OPFMT_MODES: raise ValueError(f"unknown opfmt mode {mode!r}; one of " f"{', '.join(_OPFMT_MODES)}") target = params.get("target") if target not in (None, ""): await self._fill_prompt("g", "goto", str(target), delay, clear=False) await settle(app, timeout=timeout) if app.is_hex: await self._press(["backslash"], lambda: not app.is_hex, timeout, "leave the hex view") view = _active_widget(app) if isinstance(view, HexView): raise RuntimeError("opfmt needs a code view, not the hex view") if params.get("word"): # Land the column on the literal first: WHICH operand gets # reformatted is decided by where the cursor is. if not cursor_on(app, str(params["word"]), params.get("line"), int(params.get("occurrence", 1) or 1)): raise RuntimeError( f"{params['word']!r} is not on screen in this view, so " f"there is no literal to reformat") await drain(app) elif params.get("line") is not None or params.get("col") is not None: place_cursor(view, params.get("line"), params.get("col")) await drain(app) before = _where(app) if mode in _OPFMT_KEYS: snap = await self._press([_OPFMT_KEYS[mode]], timeout=timeout, what=f"opfmt {mode}") else: view.focus() view.action_op_format(mode) await settle(app, timeout=timeout) snap = snapshot(app) snap["opfmt"] = {"mode": mode, "at": before, "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) return snapshot(app) if method == "comment": # Comments can be long; skip the per-char delay so the agent isn't # blocked for seconds watching the typing animation. Also: the # prompt is single-line, so literal newlines (0x0a) get swallowed by # the Input widget. The app's _do_comment converts the two-char # sequence '\n' into a real newline for IDA, so we escape here. ctext = str(params["text"]).replace("\n", "\\n") await self._fill_prompt("semicolon", "comment", ctext, 0, clear=True) await settle(app, timeout=timeout) return snapshot(app) if method == "retype": await self._fill_prompt("y", "retype", str(params["proto"]), delay, clear=True) await settle(app, timeout=timeout) return snapshot(app) if method == "follow": depth = len(app._nav) return await self._press(["enter"], lambda: len(app._nav) > depth, timeout, "follow") if method == "back": return await self._press(["escape"], timeout=timeout) if method == "toggle_view": before = app._active def _toggled(): # Normal case: the shown view flipped. if app._active != before: return True # Fallback case: a tab toward pseudocode on a function Hex-Rays # can't decompile lands back on the LISTING (see # App._apply_decomp), so `_active` never changes and the naive # `_active != before` predicate would block for the full # timeout. Treat "requested decomp but it's known-failed" as # settled (the decompile is cached, so this is cheap). cur = app._cur if before == ViewMode.LISTING and cur is not None: try: return app.program.decompile(cur.ea).failed except Exception: # noqa: BLE001 return False return False return await self._press(["tab"], _toggled, timeout, "toggle_view") if method == "hex": return await self._press(["backslash"], lambda: app.is_hex, timeout, "hex") if method == "graph": return await self._graph(params, timeout) if method == "xrefs": return await self._press( ["x"], lambda: type(app.screen).__name__ == "XrefsScreen", timeout, "xrefs") if method == "symbols": await app._press_keys(["ctrl+n"]) await settle(app, lambda: type(app.screen).__name__ == "SymbolPalette", timeout=10) q = params.get("query") if q: await app._press_keys(_text_to_keys(str(q), delay)) await settle(app, timeout=timeout) return snapshot(app) if method == "structs": return await self._press( ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor", timeout, "structs") 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"))) if key is None: raise ValueError(f"unknown move dir: {params.get('dir')!r} " f"(one of {sorted(_MOVE_KEYS)})") n = max(1, int(params.get("n", 1))) await app._press_keys([key] * n) if params.get("settle", True): await drain(app) # light: pump only, keep movement snappy return snapshot(app) if method == "cursor": w = _active_widget(app) if isinstance(w, HexView): raise ValueError("cursor: not supported in the hex view (use goto)") place_cursor(w, params.get("line"), params.get("col")) await drain(app) return snapshot(app) raise ValueError(f"unknown method: {method!r}")