diff options
Diffstat (limited to 'idatui/rpc.py')
| -rw-r--r-- | idatui/rpc.py | 725 |
1 files changed, 512 insertions, 213 deletions
diff --git a/idatui/rpc.py b/idatui/rpc.py index 1262ebb..cfeb0ca 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -17,6 +17,7 @@ Method tiers: introspect state, view, screen, functions (semantic verbs — open/goto/rename/... — layer on top in a later pass.) """ + from __future__ import annotations import asyncio @@ -27,8 +28,8 @@ from typing import Any from rich.console import Console -from ._sync import drain, settle from . import diag +from ._sync import drain, settle from .app import DecompView, GraphView, HexView, ListingView, ViewMode PROTO_VERSION = 1 @@ -36,10 +37,31 @@ 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", "export", "find", + "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", + "export", + "find", } # Self-documenting method table (returned by the 'methods' verb). @@ -55,7 +77,7 @@ METHODS = { "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", + "function (decomp refs), address-scoped for a 0xADDR", "resolve": "{name} -> {ea}", "keys": "{keys:[str],settle?,timeout?} raw key injection (supports 'wait:<ms>')", "text": "{text,delay_ms?,settle?} type a literal string into the focused input", @@ -69,17 +91,17 @@ METHODS = { "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=<id|0xADDR>", + "target?,blocks?} the control-flow graph: 'show' reports its " + "structure (blocks, edges, cursor) without touching it; the others " + "drive it. 'block' takes target=<id|0xADDR>", "xrefs": "open the xref picker", "symbols": "{query?} open the symbol palette", "structs": "open the struct editor", "export": "{path?,types?=true} write the session's comments/names/types as " - "a markdown report -> {path,comments,names,types}", + "a markdown report -> {path,comments,names,types}", "find": "{query,mode?=auto|text|bytes,limit?=500,regex?,case?} search the " - "WHOLE database: disassembly text, or a byte pattern with " - "wildcards (48 8b ?? c3) -> {mode,hits:[{addr,head,line,func}]}", + "WHOLE database: disassembly text, or a byte pattern with " + "wildcards (48 8b ?? c3) -> {mode,hits:[{addr,head,line,func}]}", "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)", @@ -90,42 +112,62 @@ METHODS = { "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", + "(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)", + "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", + "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") +_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 + "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 + "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", + "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", } @@ -148,8 +190,11 @@ def graph_info(app, blocks: bool = True) -> dict[str, Any]: 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"} + 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, @@ -158,24 +203,30 @@ def graph_info(app, blocks: bool = True) -> dict[str, Any]: "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()}, + "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)), - }) + 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 @@ -186,8 +237,20 @@ _MODALS = ("XrefsScreen", "SymbolPalette", "StructEditor", "ConfirmScreen") #: 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") +_INT_PARAMS = ( + "lines", + "limit", + "max", + "n", + "index", + "line", + "col", + "occurrence", + "delay_ms", + "direction", + "addr", + "count", +) _FLOAT_PARAMS = ("timeout",) @@ -221,13 +284,16 @@ def _modal_snapshot(app) -> dict[str, Any] | 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)} + { + "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 @@ -235,14 +301,23 @@ def _modal_snapshot(app) -> dict[str, Any] | None: 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} + 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()} + 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: @@ -254,9 +329,15 @@ def _cursor_info(app, w) -> dict[str, Any]: 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)} + 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: @@ -290,12 +371,12 @@ def snapshot(app) -> dict[str, Any]: pass return { "active": app._active, - "pref": app._code_mode(), # kept for wire compat; a constant now + "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 + "binary": app._binary, # None outside project mode "nav_depth": len(app._nav), "hops": list(getattr(app, "_hops", [])), "dirty": bool(app._dirty), @@ -309,13 +390,18 @@ def view_lines(app, lines: int | None = None) -> dict[str, Any]: 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)} + 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)} + 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)) @@ -323,21 +409,39 @@ def view_lines(app, lines: int | None = None) -> dict[str, Any]: 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} + 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) + 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) + 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": @@ -388,8 +492,10 @@ def cursor_on(app, word: str, line: int | None = None, occurrence: int = 1) -> b 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") + 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: @@ -412,7 +518,7 @@ def cursor_on(app, word: str, line: int | None = None, occurrence: int = 1) -> b if w.word_under_cursor() == word: hits += 1 if hits >= max(1, occurrence): - place_cursor(w) # scrolls: an off-screen cursor edits blind + 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 @@ -455,8 +561,14 @@ def pseudocode(app, target=None) -> dict[str, Any]: 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} + 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]: @@ -469,13 +581,26 @@ def disassembly(app, target=None, max_lines: int = 2000) -> dict[str, Any]: 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]} + 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]] + 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]]: @@ -497,9 +622,15 @@ def xrefs_from(app, target, limit: int = 200) -> list[dict[str, Any]]: 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"}) + 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) @@ -560,15 +691,25 @@ class RpcServer: except OSError: pass - async def _on_client(self, reader: asyncio.StreamReader, - writer: asyncio.StreamWriter) -> None: + 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") + 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 @@ -604,7 +745,11 @@ class RpcServer: 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): + if ( + isinstance(e, KeyError) + and len(e.args) == 1 + and isinstance(e.args[0], str) + ): msg = e.args[0] else: msg = str(e) @@ -619,7 +764,8 @@ class RpcServer: # 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=") + f"(still at {_where(self.app)}); retry with a larger timeout=" + ) return snapshot(self.app) async def _graph(self, params, timeout): @@ -641,18 +787,21 @@ class RpcServer: 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 + 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}") + (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)") + raise ValueError( + f"graph {action}: the graph is not open (graph {{action:'open'}} first)" + ) if action == "zoom": before = gv._zoom await self._press(["z"], lambda: gv._zoom != before, timeout, "graph zoom") @@ -660,9 +809,12 @@ class RpcServer: 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}") + 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: @@ -694,18 +846,23 @@ class RpcServer: """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) + 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") + 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 = "" @@ -727,14 +884,14 @@ class RpcServer: app = self.app items = params.get("items") src = params.get("file") - if isinstance(items, str): # `drive raw` hands params through as text + 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 + 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") @@ -745,8 +902,14 @@ class RpcServer: 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) + 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 @@ -764,8 +927,15 @@ class RpcServer: # or the TUI freezes for the length of the batch. res = await asyncio.to_thread(app.program.client.invoke, "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 [] + 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 @@ -779,18 +949,23 @@ class RpcServer: app.program.bump_names() app.program.invalidate_functions() app._func_index = None - app._load_functions() # re-streams the function table + 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)") + 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]], + "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 @@ -810,13 +985,31 @@ class RpcServer: #: 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", + "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"} + _DRIVABLE_MODALS = { + "XrefsScreen", + "SymbolPalette", + "StructEditor", + "ProjectPalette", + "QuitScreen", + } def _modal_kind(self) -> str | None: scr = self.app.screen @@ -833,15 +1026,20 @@ class RpcServer: 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.") + 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)} + return { + "ok": True, + "proto": PROTO_VERSION, + "module": module, + **_readiness(app), + } if method == "methods": return METHODS if method == "quit": @@ -856,14 +1054,18 @@ class RpcServer: def _go(): if dirty and save: - app._on_quit_choice("save") # saves, then exits + 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} + 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") @@ -902,8 +1104,10 @@ class RpcServer: 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} + return { + "recent": diag.recent(int(params.get("n", 10))), + "log": os.environ.get("IDATUI_LOG") or None, + } if method == "trace": tc = app.trace_ctl @@ -916,12 +1120,19 @@ class RpcServer: 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) + 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 + 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))) + 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") @@ -933,9 +1144,12 @@ class RpcServer: (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))} + 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": @@ -943,12 +1157,20 @@ class RpcServer: 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]} + 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: @@ -965,30 +1187,41 @@ class RpcServer: 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))) + 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")) + 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) + 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) + 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) + None, xrefs_from, app, params.get("target"), lim + ) if method == "resolve": return await loop.run_in_executor(None, resolve, app, params.get("name")) @@ -998,15 +1231,24 @@ class RpcServer: # 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") + 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))) + found = cursor_on( + app, + str(params["word"]), + params.get("line"), + int(params.get("occurrence", 1)), + ) await drain(app) snap = snapshot(app) snap["found"] = found @@ -1024,7 +1266,8 @@ class RpcServer: # 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=") + f"(still at {_where(app)}); retry with a larger timeout=" + ) return snapshot(app) if method == "define": @@ -1032,80 +1275,95 @@ class RpcServer: if kind not in _DEFINE_KEYS: raise ValueError( f"unknown define kind {kind!r}; one of " - f"{', '.join(sorted(_DEFINE_KEYS))}") + 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 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") + 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") + 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}") + 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)}") + raise ValueError( + f"unknown opfmt mode {mode!r}; one of {', '.join(_OPFMT_MODES)}" + ) target = params.get("target") if target not in (None, ""): - await self._fill_prompt("g", "goto", str(target), delay, - clear=False) + 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") + 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)): + 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") + 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}") + 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", "")} + 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 self._fill_prompt( + "n", "rename", str(params["name"]), delay, clear=True + ) await settle(app, timeout=timeout) return snapshot(app) if method == "comment": @@ -1115,19 +1373,21 @@ class RpcServer: # 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 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 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") + 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": @@ -1158,17 +1418,23 @@ class RpcServer: # call that LEAVES hex could never be satisfied and always timed # out -- a driver could open the hex view but never close it. was_hex = app.is_hex - return await self._press(["backslash"], lambda: app.is_hex != was_hex, - timeout, "hex") + return await self._press( + ["backslash"], lambda: app.is_hex != was_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") + ["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) + 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)) @@ -1176,10 +1442,14 @@ class RpcServer: return snapshot(app) if method == "structs": return await self._press( - ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor", - timeout, "structs") + ["ctrl+t"], + lambda: type(app.screen).__name__ == "StructEditor", + timeout, + "structs", + ) if method == "find": from . import search as _search + q = str(params.get("query", "")) forced = params.get("mode") forced = None if forced in (None, "auto") else str(forced) @@ -1190,33 +1460,57 @@ class RpcServer: raise ValueError(f"find: {problem}") cleaned = _search.normalise_pattern(cleaned) hits, err, truncated = await asyncio.to_thread( - app.program.search, cleaned, mode, + app.program.search, + cleaned, + mode, limit=int(params.get("limit", 500)), - regex=bool(params.get("regex")), case=bool(params.get("case"))) + regex=bool(params.get("regex")), + case=bool(params.get("case")), + ) if err: raise ValueError(f"find: {err}") - return {"mode": mode, "query": cleaned, "truncated": truncated, - "hits": [{"addr": hex(h.addr), "head": hex(h.head), - "line": h.line, "func": h.func, - "seg": h.seg} for h in hits]} + return { + "mode": mode, + "query": cleaned, + "truncated": truncated, + "hits": [ + { + "addr": hex(h.addr), + "head": hex(h.head), + "line": h.line, + "func": h.func, + "seg": h.seg, + } + for h in hits + ], + } if method == "export": # Deliberately NOT driven through the prompt: this is the one verb # whose whole point is the file it leaves behind, and a driver needs # the path back, not a screenshot of a prompt closing. from . import findings + path = params.get("path") app.journal.load(app.program) app.journal.flush(app.program) out, f = await asyncio.to_thread( - findings.export, app.program, app._open_path or "", + findings.export, + app.program, + app._open_path or "", str(path) if path else None, - types=bool(params.get("types", True)), journal=app.journal) + types=bool(params.get("types", True)), + journal=app.journal, + ) app._status(f"exported findings → {out}", priority=True) await drain(app) - return {"path": out, "comments": len(f.comments), - "names": len(findings._user_names(f)), - "types": len(f.types), "functions": f.n_functions, - "bytes": os.path.getsize(out) if os.path.exists(out) else 0} + return { + "path": out, + "comments": len(f.comments), + "names": len(findings._user_names(f)), + "types": len(f.types), + "functions": f.n_functions, + "bytes": os.path.getsize(out) if os.path.exists(out) else 0, + } if method == "close": return await self._press(["escape"], timeout=timeout) if method == "save": @@ -1224,13 +1518,16 @@ class RpcServer: if method == "search": term = str(params.get("term", "")) - open_key = "slash" if int(params.get("direction", 1)) >= 0 else "question_mark" + 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") @@ -1249,8 +1546,10 @@ class RpcServer: 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)})") + 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): |