rpc: unix-socket puppeteering server (raw keys + introspection + screen)

Run the TUI with --rpc <sock> and it renders normally while an asyncio
listener on the same loop lets another process drive it. Handlers touch
the UI directly (same loop, no thread hop), so every injected key has the
identical on-screen effect a keyboard would — the point for livestreaming.

Newline-delimited JSON. v1 methods: keys/text (raw injection, via the same
_press_keys the pilot uses; text can interleave wait:<ms> for a typed-out
look), state/view/screen/functions (introspection; screen() is a full
plain-text render of exactly what the viewer sees). No auth by design —
the socket is 0600, local only.

tests/rpc_smoke.py drives it end-to-end over a real socket (7 green).

1 files changed, 121 insertions, 0 deletions

diff --git a/tests/rpc_smoke.py b/tests/rpc_smoke.py
new file mode 100644
index 0000000..92fc801
--- /dev/null
+++ b/tests/rpc_smoke.py
@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""End-to-end smoke test for the RPC server, in-process over a real unix socket.
+
+    ~/ida-venv/bin/python tests/rpc_smoke.py --db <session_id>
+
+Boots the TUI headless with --rpc on a temp socket, connects a JSONL client over
+that socket (same loop), and exercises the raw + introspection primitives.
+"""
+import asyncio
+import json
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from idatui.app import IdaTui  # noqa: E402
+from idatui.client import DEFAULT_URL  # noqa: E402
+from idatui._sync import wait_for  # noqa: E402
+from textual.widgets import DataTable  # noqa: E402
+
+PASS = FAIL = 0
+
+
+def check(name, cond, detail=""):
+    global PASS, FAIL
+    if cond:
+        PASS += 1
+        print(f"  ok   {name}")
+    else:
+        FAIL += 1
+        print(f"  FAIL {name}   {detail}")
+
+
+class Conn:
+    """Tiny JSONL client over a unix socket."""
+    def __init__(self, r, w):
+        self.r, self.w = r, w
+        self._id = 0
+
+    async def call(self, method, **params):
+        self._id += 1
+        self.w.write(json.dumps({"id": self._id, "method": method,
+                                 "params": params}).encode() + b"\n")
+        await self.w.drain()
+        line = await self.r.readline()
+        return json.loads(line.decode())
+
+
+async def run(db):
+    sock = os.path.join(tempfile.gettempdir(), f"idatui-rpc-{os.getpid()}.sock")
+    url = os.environ.get("IDA_MCP_URL", DEFAULT_URL)
+    app = IdaTui(url=url, db=db, keepalive=False, rpc_path=sock)
+    async with app.run_test(size=(140, 44)) as pilot:
+        # boot: functions loaded + socket up
+        await wait_for(lambda: app.query_one("#func-table", DataTable).row_count > 0,
+                       pilot.pause, 60)
+        await wait_for(lambda: os.path.exists(sock), pilot.pause, 20)
+        if app._func_index is not None and not app._func_index.complete:
+            app._func_index.load_all()
+
+        r, w = await asyncio.open_unix_connection(sock)
+        c = Conn(r, w)
+
+        resp = await c.call("ping")
+        check("ping returns proto", resp.get("result", {}).get("proto") == 1, str(resp))
+
+        resp = await c.call("functions", limit=5)
+        funcs = resp.get("result", [])
+        check("functions lists entries", isinstance(funcs, list) and len(funcs) > 0,
+              str(resp)[:120])
+
+        resp = await c.call("state")
+        st = resp.get("result", {})
+        check("state has an active view", st.get("active") in ("decomp", "disasm", "hex"),
+              str(st)[:120])
+
+        # drive it like a user: goto a function by name via raw keys
+        target = next((f for f in funcs if f["name"] == "main"), funcs[0])
+        keys = ["g"] + list(target["name"]) + ["enter"]
+        resp = await c.call("keys", keys=keys)
+        st = resp.get("result", {})
+        check("keys(goto) navigates to the function",
+              st.get("function", {}).get("name") == target["name"],
+              f"got {st.get('function')}")
+
+        resp = await c.call("view", lines=6)
+        v = resp.get("result", {})
+        check("view returns visible lines with a cursor",
+              isinstance(v.get("lines"), list) and any(l.get("cur") for l in v["lines"]),
+              str(v)[:120])
+
+        resp = await c.call("screen")
+        scr = resp.get("result", {})
+        check("screen returns a text grid",
+              isinstance(scr.get("text"), str) and target["name"] in scr["text"],
+              f"len={len(scr.get('text','')) if scr else 0}")
+
+        # raw text primitive into the goto input, then escape out
+        from textual.widgets import Input
+        await c.call("keys", keys=["g"])
+        await c.call("text", text="sub_", settle=False)
+        val = app.query_one("#goto", Input).value
+        check("text primitive types into the focused input", val == "sub_", f"val={val!r}")
+        await c.call("keys", keys=["escape"])
+
+        w.close()
+    print(f"\n{PASS} passed, {FAIL} failed")
+    return 1 if FAIL else 0
+
+
+def main(argv):
+    db = None
+    it = iter(argv)
+    for a in it:
+        if a == "--db":
+            db = next(it)
+    return asyncio.run(run(db))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))