From b90ae8d52623e9a59d6f5ee68ff2f96537bf965d Mon Sep 17 00:00:00 2001 From: blasty Date: Fri, 10 Jul 2026 15:45:51 +0200 Subject: rpc: cursor_on + word= edits, single-driver gate, colored screen MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cursor_on {word,line?,occurrence?} places the cursor on a token, verified with the app's own tokenizer (so 'main' won't match inside 'domain'); rename/retype/follow gain an optional word= that runs it first — the ergonomic way to edit a named symbol without a manual view+cursor dance. Single-driver gate: the socket now serves one client at a time; a second concurrent connection is refused with 'busy' (no multi-driver support yet, by request). Sequential connections are unaffected. screen gains format=text|html|svg (html/svg are colored — for an out-of-band web viewer). rpc_smoke: 24 green. --- docs/RPC.md | 13 +++++--- idatui/rpc.py | 91 +++++++++++++++++++++++++++++++++++++++++++++++++----- tests/rpc_smoke.py | 31 ++++++++++++++++--- 3 files changed, 118 insertions(+), 17 deletions(-) diff --git a/docs/RPC.md b/docs/RPC.md index 0a60cc9..b8ee87f 100644 --- a/docs/RPC.md +++ b/docs/RPC.md @@ -30,6 +30,10 @@ connection (the UI is single-user). A mutating call returns **after the UI has settled**, and its result is a fresh `state()` snapshot — so the driver never acts on stale state. +**Single-driver:** only one client connection is served at a time; a second +concurrent connection is refused with `error: busy` (no multi-driver support). +Sequential connections (e.g. one `rpcclient` invocation per call) are fine. + ``` -> {"id": 1, "method": "goto", "params": {"target": "main"}} <- {"id": 1, "result": { ...state... }} @@ -53,7 +57,7 @@ stalls. `target` is a name, `0xADDR`, or omitted (= current function). |--------|--------|---------| | `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. | +| `screen` | `format?=text\|html\|svg` | `{width,height,format,text}` — a full render of exactly what's on screen (`html`/`svg` are colored, for an out-of-band viewer). | | `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}]}`. | @@ -69,10 +73,10 @@ predicate so the returned state is final. | method | params | effect | |--------|--------|--------| | `goto` / `open` | `target`, `delay_ms?`, `timeout?` | `g` prompt → name or `0xADDR` → Enter. | -| `rename` | `name`, `delay_ms?` | `n` on the token under the cursor → replace → Enter. | +| `rename` | `name`, `word?`, `delay_ms?` | `n` on the token under the cursor → replace → Enter. `word` first places the cursor on that token (see `cursor_on`). | | `comment` | `text`, `delay_ms?` | `;` on the current line. | -| `retype` | `proto`, `delay_ms?` | `y` on the token/function under the cursor. | -| `follow` | — | Enter: follow the reference under the cursor (waits for the jump). | +| `retype` | `proto`, `word?`, `delay_ms?` | `y` on the token/function under the cursor (`word` places it first). | +| `follow` | `word?` | Enter: follow the reference under the cursor (`word` places it first); waits for the jump. | | `back` | — | Escape: pop the nav stack. | | `toggle_view` | — | Tab: disasm ⇄ pseudocode. | | `hex` | — | `\`: hex view. | @@ -89,6 +93,7 @@ predicate so the returned state is final. |--------|--------|--------| | `move` | `dir`, `n?=1`, `settle?` | `dir` ∈ down/up/left/right/word/wordback/bol/eol/top/bottom/halfdown/halfup/pagedown/pageup. | | `cursor` | `line?`, `col?` | set the cursor directly on the active code pane (disasm/decomp). | +| `cursor_on` | `word`, `line?`, `occurrence?=1` | place the cursor on the *n*-th token equal to `word` (verified with the app's tokenizer). Decomp searches the whole body; disasm only cached/visible lines. Returns `{found, ...state}`. | ## Driving pattern for an agent diff --git a/idatui/rpc.py b/idatui/rpc.py index afe5be3..73e10d2 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -47,7 +47,7 @@ METHODS = { "quit": "gracefully exit the TUI", "state": "-> full state snapshot", "view": "{lines?} -> visible code-pane lines (disasm/decomp)", - "screen": "-> {width,height,text} full plain-text render", + "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}]}", @@ -57,10 +57,11 @@ METHODS = { "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,delay_ms?} rename the token under the cursor", + "rename": "{name,word?,delay_ms?} rename the token under (or 'word') the cursor", "comment": "{text,delay_ms?} comment the current line", - "retype": "{proto,delay_ms?} set the prototype/type under the cursor", - "follow": "follow the reference under the cursor", + "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", @@ -198,8 +199,9 @@ def view_lines(app, lines: int | None = None) -> dict[str, Any]: "cursor": _cursor_info(app, w), "lines": out} -def screen_text(app) -> dict[str, Any]: - """Plain-text render of the whole screen — exactly what the viewer sees.""" +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, @@ -207,7 +209,49 @@ def screen_text(app) -> dict[str, Any]: render = app.screen._compositor.render_update( full=True, screen_stack=app._background_screens, simplify=False) console.print(render) - return {"width": width, "height": height, "text": console.export_text(styles=False)} + 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 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.""" + w = _active_widget(app) + if isinstance(w, HexView): + raise ValueError("cursor_on: not supported in the hex view") + 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) + rows = [line] if line is not None else range(len(texts)) + 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): + if hasattr(w, "_after_cursor_move"): + w._after_cursor_move() + w.refresh() + 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]]: @@ -308,6 +352,7 @@ class RpcServer: 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): @@ -336,6 +381,19 @@ class RpcServer: 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() @@ -347,6 +405,7 @@ class RpcServer: except (ConnectionResetError, BrokenPipeError): pass finally: + self._busy = False try: writer.close() except Exception: # noqa: BLE001 @@ -440,7 +499,7 @@ class RpcServer: if method == "view": return view_lines(app, params.get("lines")) if method == "screen": - return screen_text(app) + return screen_text(app, str(params.get("format", "text"))) if method == "functions": return functions(app, params.get("filter"), int(params.get("limit", 50))) @@ -467,6 +526,22 @@ class RpcServer: 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) diff --git a/tests/rpc_smoke.py b/tests/rpc_smoke.py index ba579a9..5d24978 100644 --- a/tests/rpc_smoke.py +++ b/tests/rpc_smoke.py @@ -134,17 +134,22 @@ async def run(db): isinstance(line0, int) and isinstance(line1, int) and line1 > line0, f"{line0} -> {line1}") - # rename round-trip through the prompt-fill path (place cursor on the - # function's own name via view(), rename, verify, revert via the API) + # cursor_on: place the cursor on the function's own name by token + resp = await c.call("cursor_on", word=target["name"]) + cur = resp.get("result", {}) + check("cursor_on lands on the named token", + cur.get("found") is True and cur.get("cursor", {}).get("word") == target["name"], + str(cur.get("cursor"))) + + # rename using the ergonomic word= (cursor_on + prompt-fill in one call) v = (await c.call("view", lines=1))["result"] line0_text = v["lines"][0]["text"] if v.get("lines") else "" col = line0_text.find(target["name"]) if col >= 0: - await c.call("cursor", line=0, col=col + 1) newname = f"rpc_{os.getpid()}" - await c.call("rename", name=newname, delay_ms=0) + await c.call("rename", name=newname, word=target["name"], delay_ms=0) got = app._func_index.by_addr(target["ea"]) - check("semantic rename updates the function name", + check("rename word= updates the function name", got is not None and got.name == newname, got.name if got else None) app.program.client.call( @@ -208,6 +213,22 @@ async def run(db): st.get("active") in ("decomp", "disasm"), str(st.get("active"))) await c.call("keys", keys=["escape"]) + # colored screen export (for an out-of-band web viewer) + resp = await c.call("screen", format="html") + html = resp.get("result", {}) + check("screen format=html returns an html document", + html.get("format") == "html" and "<" in (html.get("text") or ""), + str(html.get("format"))) + + # single-driver gate: a 2nd connection (while c is open) is refused + r2, w2 = await asyncio.open_unix_connection(sock) + c2 = Conn(r2, w2) + resp2 = await c2.call("ping") + check("second concurrent client is refused (single-driver)", + resp2.get("error") and "busy" in resp2["error"].get("message", ""), + str(resp2)) + w2.close() + try: w.close() except Exception: # noqa: BLE001 -- cgit v1.3.1-sl0p