diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/pane_smoke.py | 87 | ||||
| -rw-r--r-- | tests/rpc_smoke.py | 284 | ||||
| -rwxr-xr-x | tests/serverctl.sh | 65 | ||||
| -rw-r--r-- | tests/smoke_client.py | 162 | ||||
| -rw-r--r-- | tests/stress_client.py | 290 | ||||
| -rw-r--r-- | tests/stress_paging.py | 150 | ||||
| -rw-r--r-- | tests/test_blob_ui.py | 235 | ||||
| -rw-r--r-- | tests/test_domain.py | 155 | ||||
| -rw-r--r-- | tests/test_formats.py | 122 | ||||
| -rw-r--r-- | tests/test_index.py | 202 | ||||
| -rw-r--r-- | tests/test_keepalive.py | 76 | ||||
| -rw-r--r-- | tests/test_pool.py | 192 | ||||
| -rw-r--r-- | tests/test_project.py | 233 | ||||
| -rw-r--r-- | tests/test_project_ui.py | 285 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 1264 |
15 files changed, 2384 insertions, 1418 deletions
diff --git a/tests/pane_smoke.py b/tests/pane_smoke.py deleted file mode 100644 index bd7def6..0000000 --- a/tests/pane_smoke.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python3 -"""Smoke test for idatui.pane's supervisor auto-start machinery (no IDA needed). - - python3 tests/pane_smoke.py - -Uses IDATUI_SERVER_CMD to launch a dummy port-binder in a tmux pane instead of -the real ./spawn.sh, so we can exercise _ensure_server end-to-end (start, detect, -idempotency, remote guard) without a real ida-pro-mcp server. Must run in tmux. -""" -import os -import socket -import subprocess -import sys -import tempfile - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui import pane # 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}") - - -def _free_port() -> int: - s = socket.socket() - s.bind(("127.0.0.1", 0)) - p = s.getsockname()[1] - s.close() - return p - - -def main() -> int: - if not os.environ.get("TMUX"): - print("error: must run inside tmux", file=sys.stderr) - return 2 - - # a dummy "supervisor": bind the port and idle, so _server_up sees it. - fake = tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) - fake.write( - "import socket,sys,time\n" - "s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)\n" - "s.bind(('127.0.0.1',int(sys.argv[1]))); s.listen(); time.sleep(300)\n") - fake.close() - port = _free_port() - os.environ["IDATUI_SERVER_CMD"] = f"{sys.executable} {fake.name} {port}" - - server_pane = None - try: - check("port starts down", not pane._server_up("127.0.0.1", port)) - - srv = pane._ensure_server("127.0.0.1", port, timeout=15.0) - server_pane = srv.get("server_pane") - check("ensure_server starts the supervisor and it comes up", - srv.get("server_started") and srv.get("server_up") and server_pane, - str(srv)) - check("port is now up", pane._server_up("127.0.0.1", port)) - - srv2 = pane._ensure_server("127.0.0.1", port, timeout=5.0) - check("ensure_server is idempotent when already up", - srv2.get("server_started") is False and srv2.get("server_up") is True, - str(srv2)) - - srv3 = pane._ensure_server("10.255.255.1", 9, timeout=2.0) - check("remote+down server is not auto-started", - srv3.get("server_started") is False and "local" in (srv3.get("error") or ""), - str(srv3)) - finally: - if server_pane: - subprocess.run(["tmux", "kill-pane", "-t", server_pane], - capture_output=True) - os.unlink(fake.name) - os.environ.pop("IDATUI_SERVER_CMD", None) - - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/rpc_smoke.py b/tests/rpc_smoke.py deleted file mode 100644 index a38162e..0000000 --- a/tests/rpc_smoke.py +++ /dev/null @@ -1,284 +0,0 @@ -#!/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") - pong = resp.get("result", {}) - check("ping returns proto + readiness", - pong.get("proto") == 1 and pong.get("ready") is True and pong.get("functions", 0) > 0, - str(pong)) - - resp = await c.call("methods") - check("methods lists the verb table", - isinstance(resp.get("result"), dict) and "pseudocode" in resp["result"], - str(resp)[:80]) - - resp = await c.call("functions", limit=400) - funcs = resp.get("result", []) - check("functions lists entries", isinstance(funcs, list) and len(funcs) > 0, - str(resp)[:120]) - # a normally-named function (its name appears verbatim in its own decomp, - # unlike e.g. '.init_proc' which IDA renders as 'init_proc') - target = next((f for f in funcs if f["name"] == "main"), - next((f for f in funcs if f["name"].startswith("sub_")), funcs[0])) - - 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 - 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"]) - - # --- semantic verbs ------------------------------------------------ # - resp = await c.call("goto", target=target["name"], delay_ms=0) - check("semantic goto lands on the function", - resp.get("result", {}).get("function", {}).get("name") == target["name"], - str(resp.get("result", {}).get("function"))) - - before = app._active - resp = await c.call("toggle_view") - check("toggle_view flips the active pane", - resp.get("result", {}).get("active") != before, f"still {before}") - await c.call("toggle_view") # flip back to a known state - - # fast movement: cursor should advance a few lines - line0 = (await c.call("state"))["result"]["cursor"].get("line") - resp = await c.call("move", dir="down", n=4) - line1 = resp["result"]["cursor"].get("line") - check("move(down,4) advances the cursor", - isinstance(line0, int) and isinstance(line1, int) and line1 > line0, - f"{line0} -> {line1}") - - # 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: - newname = f"rpc_{os.getpid()}" - await c.call("rename", name=newname, word=target["name"], delay_ms=0) - got = app._func_index.by_addr(target["ea"]) - 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( - "rename", batch={"func": {"addr": hex(target["ea"]), "name": target["name"]}}) - else: - check("found the function name to rename", False, repr(line0_text[:60])) - - # --- structured introspection ------------------------------------- # - resp = await c.call("resolve", name=target["name"]) - check("resolve maps a name to its ea", - resp.get("result", {}).get("ea") == target["ea"], str(resp.get("result"))) - - resp = await c.call("pseudocode", target=target["name"]) - pc = resp.get("result", {}) - check("pseudocode returns the full body", - isinstance(pc.get("code"), str) and len(pc["code"]) > 0 and not pc["failed"], - f"failed={pc.get('failed')} len={len(pc.get('code') or '')}") - - resp = await c.call("disassembly", target=target["name"], max=50) - da = resp.get("result", {}) - check("disassembly returns lines with addresses", - isinstance(da.get("lines"), list) and len(da["lines"]) > 0 - and all("ea" in ln and "text" in ln for ln in da["lines"]), - f"total={da.get('total')} n={len(da.get('lines', []))}") - - # a function that is actually referenced, so xrefs_to is non-empty - callee = None - for f in funcs: - xr = (await c.call("xrefs_to", target=f["name"], limit=5)).get("result", []) - if xr: - callee = (f, xr) - break - check("xrefs_to returns structured references", - callee is not None and all("frm" in x for x in callee[1]), - "no referenced function found" if callee is None else "") - - # xrefs_from a function is whole-body (decomp refs), not just the entry: - # find a function that actually calls something. - caller = next((f["name"] for f in funcs if f["name"] == "main"), None) - xf = [] - for name in ([caller] if caller else []) + [f["name"] for f in funcs]: - xf = (await c.call("xrefs_from", target=name)).get("result", []) - if any(x.get("is_func") for x in xf): - caller = name - break - check("xrefs_from a function lists whole-body callees", - isinstance(xf, list) and len(xf) > 1 - and any(x.get("is_func") for x in xf) and all("to" in x for x in xf), - f"caller={caller} n={len(xf)}") - - # --- modal select: open xrefs on the callee, pick the first site --- # - if callee is not None: - f, xr = callee - await c.call("goto", target=f["name"], delay_ms=0) - # place the cursor on the function name so xrefs targets it - v = (await c.call("view", lines=1))["result"] - lt = v["lines"][0]["text"] if v.get("lines") else "" - col = lt.find(f["name"]) - if col >= 0: - await c.call("cursor", line=0, col=col + 1) - resp = await c.call("xrefs") - m = resp.get("result", {}).get("modal") or {} - check("xrefs opens the picker with items", - m.get("kind") == "XrefsScreen" and len(m.get("items", [])) > 0, - str(m)[:80]) - resp = await c.call("select", index=0) - check("select follows a picked xref (modal closes, we navigate)", - (resp.get("result", {}).get("modal") is None), str(resp.get("result", {}).get("modal"))) - - # --- in-view search ----------------------------------------------- # - await c.call("goto", target=target["name"], delay_ms=0) - resp = await c.call("search", term="return", delay_ms=0) - st = resp.get("result", {}) - check("search runs without error and returns state", - st.get("active") in ("decomp", "disasm"), str(st.get("active"))) - await c.call("keys", keys=["escape"]) - - # 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 - pass - - # --- ergonomic driver (idatui.drive) over the same socket ---------- # - # (c is closed now, so the single-driver gate lets drive connect.) - from idatui import drive - loop = asyncio.get_running_loop() - for cmd in (["where"], ["pc", target["name"]], ["callees", target["name"]], - ["names", "sub_", "3"]): - rc = await loop.run_in_executor( - None, lambda a=cmd: drive.main(["--sock", sock, *a])) - check(f"drive {cmd[0]} runs against the socket", rc == 0, f"rc={rc}") - - # --- graceful quit (last: it tears the app down) ------------------- # - r2, w2 = await asyncio.open_unix_connection(sock) - c2 = Conn(r2, w2) - resp = await c2.call("quit") - check("quit acknowledges before shutting down", - resp.get("result", {}).get("quitting") is True, str(resp)) - w2.close() - await wait_for(lambda: not app.is_running, pilot.pause, 10) - check("quit actually exits the app", not app.is_running, "still running") - 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:])) diff --git a/tests/serverctl.sh b/tests/serverctl.sh deleted file mode 100755 index a5bc098..0000000 --- a/tests/serverctl.sh +++ /dev/null @@ -1,65 +0,0 @@ -#!/usr/bin/env bash -# Control the ida-pro-mcp server for stress testing. Assumes ~/ida-venv + spawn.sh. -set -u -REPO="$(cd "$(dirname "$0")/.." && pwd)" -PORT=8745 -LOG=/tmp/ida-stress-spawn.log - -wait_port() { # wait_port <up|down> <secs> - local want="$1" secs="${2:-40}" i - for ((i=0; i<secs*2; i++)); do - if ss -ltn 2>/dev/null | grep -q ":$PORT "; then - [ "$want" = up ] && return 0 - else - [ "$want" = down ] && return 0 - fi - sleep 0.5 - done - return 1 -} - -wait_ready() { # wait until hexrays_ready via the client - local secs="${1:-60}" i - for ((i=0; i<secs; i++)); do - if python3 -c " -import sys; sys.path.insert(0,'$REPO') -from idatui.client import IDAClient -try: - c=IDAClient(); c.connect() - ss=[s for s in c.list_sessions() if s.session_id] - if ss: - c.set_db(ss[0].session_id) - if c.health().get('hexrays_ready'): print(ss[0].session_id); sys.exit(0) -except Exception: pass -sys.exit(1) -" 2>/dev/null; then return 0; fi - sleep 1 - done - return 1 -} - -case "${1:-}" in - start) - cd "$REPO" - source ~/ida-venv/bin/activate 2>/dev/null - nohup ./spawn.sh >"$LOG" 2>&1 & - wait_port up 40 || { echo "PORT_TIMEOUT"; tail -5 "$LOG"; exit 1; } - wait_ready 90 || { echo "READY_TIMEOUT"; tail -5 "$LOG"; exit 1; } - ;; - stop) - # Kill the whole tree: uv wrapper, supervisor, worker. - pkill -9 -f 'idalib_server' 2>/dev/null - pkill -9 -f 'idalib-mcp' 2>/dev/null - pkill -9 -f 'uv run idalib' 2>/dev/null - wait_port down 20 || { echo "STOP_TIMEOUT"; exit 1; } - ;; - kill9) - # Hard kill only the worker (simulate a crash of the analysis process). - pkill -9 -f 'idalib_server' 2>/dev/null - ;; - ready) - wait_ready "${2:-60}" - ;; - *) - echo "usage: $0 {start|stop|kill9|ready [secs]}"; exit 2;; -esac diff --git a/tests/smoke_client.py b/tests/smoke_client.py deleted file mode 100644 index 5466284..0000000 --- a/tests/smoke_client.py +++ /dev/null @@ -1,162 +0,0 @@ -#!/usr/bin/env python3 -"""Live smoke test for idatui.client against a running ida-pro-mcp server. - -Run with the venv Python while a server is up (see spawn.sh): - - IDA_MCP_DB=<session_id> python3 tests/smoke_client.py - # or: python3 tests/smoke_client.py --db <session_id> - -Exercises: handshake+warm latency, happy-path payloads, the full error taxonomy -(hard tool errors vs soft per-item errors), session resolution, connection-pool -reuse, and concurrent calls from threads. -""" -import concurrent.futures -import os -import sys -import time - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from idatui.client import ( # noqa: E402 - IDAClient, - IDASessionError, - IDAToolError, -) - -PASS, FAIL = 0, 0 - - -def _query_data(payload): - res = payload.get("result", payload) if isinstance(payload, dict) else payload - if isinstance(res, list): - res = res[0] if res else {} - return res.get("data", []) if isinstance(res, dict) else [] - - -def check(name, cond, detail=""): - global PASS, FAIL - if cond: - PASS += 1 - print(f" ok {name}") - else: - FAIL += 1 - print(f" FAIL {name} {detail}") - - -def main(argv): - url = "http://127.0.0.1:8745/mcp" - db = os.environ.get("IDA_MCP_DB") - it = iter(argv) - for a in it: - if a == "--url": - url = next(it) - elif a == "--db": - db = next(it) - - ida = IDAClient(url, db=db) - ida.connect() - - print("[sessions]") - sessions = ida.list_sessions() - check("list_sessions returns >=1", len(sessions) >= 1, str(sessions)) - if db is None: - usable = [s for s in sessions if s.session_id] - if len(usable) == 1: - db = usable[0].session_id - else: - print(f" (multiple/zero sessions; pin one with --db) -> {usable}") - print(" Options:", ", ".join(f"{s.session_id}={s.filename}" for s in usable)) - return 2 - ida.set_db(db) - print(f" using db={db}") - - print("[handshake / latency]") - # Warm calls; measure. - for _ in range(3): - ida.health() - times = [] - for _ in range(20): - t = time.time() - ida.call("list_funcs", queries=[{"count": 50}]) - times.append((time.time() - t) * 1e3) - times.sort() - med = times[len(times) // 2] - print(f" list_funcs x20 warm: min={times[0]:.1f} med={med:.1f} max={times[-1]:.1f} ms") - check("warm call under 50ms median", med < 50, f"med={med:.1f}ms") - - print("[happy path]") - h = ida.health() - check("health is dict", isinstance(h, dict), type(h).__name__) - funcs = ida.call("list_funcs", queries=[{"count": 5}]) - check("list_funcs shape", isinstance(funcs, (list, dict)), type(funcs).__name__) - # Pick a real function (don't assume 'main' exists — libraries have none). - some = ida.call("list_funcs", queries=[{"filter": "sub_*", "count": 1}]) - target = _query_data(some)[0]["addr"] - dis = ida.call("disasm", addr=target, max_instructions=10) - lines = (dis.get("asm") or {}).get("lines") if isinstance(dis, dict) else None - check("disasm main has lines", bool(lines), str(dis)[:120]) - check("disasm line carries addr", bool(lines and "addr" in lines[0]), - str(lines[0]) if lines else "no lines") - - print("[error taxonomy]") - # Hard tool error: wrong params -> isError true -> IDAToolError - try: - ida.call("xrefs_to", targets=["main"]) - check("bad params raises IDAToolError", False, "no exception") - except IDAToolError as e: - check("bad params raises IDAToolError", True) - check(" ...message surfaced", "addrs" in e.message or "param" in e.message.lower(), - e.message) - # Hard tool error: unknown tool -> isError true - try: - ida.call("definitely_not_a_tool") - check("unknown tool raises IDAToolError", False, "no exception") - except IDAToolError: - check("unknown tool raises IDAToolError", True) - # Soft/per-item error: bad addr to decompile -> isError false -> DATA, not raise - try: - payload = ida.call("decompile", addr="zzz_nope_addr") - soft = isinstance(payload, dict) and payload.get("error") - check("soft error returned as data (not raised)", bool(soft), str(payload)[:160]) - except IDAToolError as e: - check("soft error returned as data (not raised)", False, f"raised: {e}") - - print("[session resolution]") - tmp = IDAClient(url, db=None) - tmp.connect() - all_sessions = tmp.list_sessions() - usable = [s for s in all_sessions if s.session_id] - if len(usable) > 1: - try: - tmp.resolve_db() - check("multi-session resolve raises", False, "no exception") - except IDASessionError: - check("multi-session resolve raises", True) - else: - check("single-session auto-resolves", tmp.resolve_db() == usable[0].session_id) - tmp.close() - - print("[connection reuse]") - # Fire many calls; pool should keep idle conns bounded and all succeed. - ok = 0 - for _ in range(30): - if ida.call("list_funcs", queries=[{"count": 1}]) is not None: - ok += 1 - check("30 sequential calls all succeed (pooled)", ok == 30, f"{ok}/30") - - print("[concurrency]") - def worker(i): - return ida.call("disasm", addr=target, max_instructions=5) - - with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex: - results = list(ex.map(worker, range(40))) - good = sum(1 for r in results if isinstance(r, dict) and r.get("asm")) - check("40 concurrent calls across 8 threads", good == 40, f"{good}/40") - - ida.close() - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/tests/stress_client.py b/tests/stress_client.py deleted file mode 100644 index 955700c..0000000 --- a/tests/stress_client.py +++ /dev/null @@ -1,290 +0,0 @@ -#!/usr/bin/env python3 -"""Adversarial stress tests for idatui.client. - -Kills/respawns the server, forces timeouts, drops connections, and hammers with -concurrency — asserting the client always fails *cleanly* (no hangs, no -deadlocks) and recovers where recovery is possible. - - python3 tests/stress_client.py # run all scenarios - python3 tests/stress_client.py timeout # run one by name - -Requires ~/ida-venv + spawn.sh (via tests/serverctl.sh). Leaves the server UP. -""" -import concurrent.futures -import os -import subprocess -import sys -import threading -import time - -HERE = os.path.dirname(os.path.abspath(__file__)) -REPO = os.path.dirname(HERE) -sys.path.insert(0, REPO) - -from idatui.client import ( # noqa: E402 - IDAClient, - IDAConnectionError, - IDASessionError, - IDATimeoutError, - IDAToolError, - IDAError, -) - -URL = "http://127.0.0.1:8745/mcp" -CTL = os.path.join(HERE, "serverctl.sh") -PASS = FAIL = 0 - - -def log(m): - print(m, flush=True) - - -def check(name, cond, detail=""): - global PASS, FAIL - if cond: - PASS += 1 - log(f" ok {name}") - else: - FAIL += 1 - log(f" FAIL {name} {detail}") - - -def ctl(*args, timeout=180): - r = subprocess.run([CTL, *args], capture_output=True, text=True, timeout=timeout) - return r.returncode, (r.stdout + r.stderr).strip() - - -def ensure_up(): - rc, out = ctl("ready", "5") - if rc != 0: - log(" (server down; starting...)") - rc, out = ctl("start") - assert rc == 0, f"could not start server: {out}" - rc, sid = ctl("ready", "60") - return sid.strip().splitlines()[-1] if sid.strip() else None - - -def deadline(fn, secs): - """Run fn() in a thread; return (finished_bool, result_or_exc).""" - box = {} - def run(): - try: - box["r"] = fn() - except BaseException as e: # noqa: BLE001 - box["e"] = e - t = threading.Thread(target=run, daemon=True) - t.start() - t.join(secs) - if t.is_alive(): - return False, None - return True, box.get("e", box.get("r")) - - -# --------------------------------------------------------------------------- # -def s_connect_refused(): - """Connecting to a dead port fails fast and cleanly, never hangs.""" - c = IDAClient("http://127.0.0.1:59999/mcp", db="x", timeout=3.0) - fin, res = deadline(c.connect, 8) - check("dead-port connect returns (no hang)", fin) - check("dead-port raises IDAConnectionError", isinstance(res, IDAConnectionError), - f"{type(res).__name__}: {res}") - - -def s_timeout(): - """A too-tight timeout on the slow decompile raises IDATimeoutError, and the - client stays usable for subsequent calls.""" - sid = ensure_up() - c = IDAClient(URL, db=sid) - c.connect() - # force_recompile clears the hexrays cache so decompile is genuinely slow. - try: - c.call("force_recompile", addr="main") - except IDAError: - pass - fin, res = deadline(lambda: c.call("decompile", addr="main", timeout=0.001), 10) - check("tight-timeout returns (no hang)", fin) - check("tight-timeout raises IDATimeoutError", isinstance(res, IDATimeoutError), - f"{type(res).__name__}: {res}") - # Client must still work afterwards (pool not poisoned). - fin2, res2 = deadline(lambda: c.call("list_funcs", queries=[{"count": 1}]), 10) - check("client usable after timeout", fin2 and not isinstance(res2, Exception), - f"{type(res2).__name__}: {res2}") - c.close() - - -def s_worker_crash(): - """Hard-kill the analysis worker mid-use: calls must raise cleanly (no hang), - and after a full restart the client recovers by re-resolving the session.""" - sid = ensure_up() - c = IDAClient(URL, db=sid) - c.connect() - check("baseline call ok", isinstance(c.call("list_funcs", queries=[{"count": 1}]), (list, dict))) - - ctl("kill9") # kill only the worker; supervisor may stay up - time.sleep(1.0) - - def probe(): - try: - return c.call("list_funcs", queries=[{"count": 1}]) - except IDAError as e: - return e - fin, res = deadline(probe, 15) - check("post-crash call returns (no hang)", fin, "call hung after worker kill") - check("post-crash raises IDAError (clean)", isinstance(res, IDAError) or res is not None, - f"{type(res).__name__}: {res}") - - # Full restart and recovery. - ctl("stop") - rc, _ = ctl("start") - check("server restart ok", rc == 0) - newsid = ensure_up() - # A fresh client should just work. - c2 = IDAClient(URL, db=None) - c2.connect() - fin3, res3 = deadline(lambda: (c2.set_db(c2.resolve_db()), c2.health())[1], 20) - check("fresh client recovers after restart", fin3 and isinstance(res3, dict), - f"{type(res3).__name__}: {res3}") - check("session id stable across restart" if newsid == sid else "session id changed (expected-ok)", - True, f"{sid} -> {newsid}") - c.close(); c2.close() - - -def s_full_restart_same_client(): - """Keep ONE auto-injected client across a full server restart. The FIRST - naive call after restart must transparently self-heal the stale db pin (no - internal poking, no manual re-resolve).""" - sid = ensure_up() - c = IDAClient(URL, db=None) # auto-resolve -> auto_recover eligible - c.connect() - old = c.resolve_db() - check("pre-restart call ok", isinstance(c.health(), dict)) - - ctl("stop") - # While down: calls must fail cleanly, fast. - fin, res = deadline(lambda: c.health(), 8) - check("while-down call returns (no hang)", fin) - check("while-down raises IDAError", isinstance(res, IDAError), f"{type(res).__name__}: {res}") - - ctl("start") - ensure_up() - # NAIVE call: no internal poking. Auto-recovery must kick in on the stale - # "Session not found" and retry against the fresh session. - fin2, res2 = deadline(lambda: c.health(), 25) - check("naive call auto-recovers after restart", fin2 and isinstance(res2, dict), - f"{type(res2).__name__}: {res2}") - check("db pin was re-resolved to new session", c.db is not None and c.db != old, - f"old={old} new={c.db}") - c.close() - - -def s_no_autoswitch_when_explicit(): - """With auto_recover disabled (or an explicitly pinned db), a stale session - must NOT be silently switched — it raises so the caller stays in control.""" - sid = ensure_up() - c = IDAClient(URL, db=sid, auto_recover_session=False) - c.connect() - check("pre-restart call ok", isinstance(c.health(), dict)) - ctl("stop"); ctl("start"); newsid = ensure_up() - fin, res = deadline(lambda: c.health(), 15) - check("explicit-pin call returns (no hang)", fin) - check("explicit-pin raises IDAToolError (no silent switch)", - isinstance(res, IDAToolError), f"{type(res).__name__}: {res}") - check("db pin unchanged when recovery disabled", c.db == sid, f"{c.db} vs {sid}") - c.close() - - -def s_concurrency_high(): - """Heavy concurrency: 300 calls over 16 threads, all succeed, no deadlock.""" - sid = ensure_up() - c = IDAClient(URL, db=sid, pool_size=8) - c.connect() - N = 300 - errors = [] - def work(i): - try: - tool = ("list_funcs", "disasm")[i % 2] - if tool == "list_funcs": - return bool(c.call("list_funcs", queries=[{"count": 3}])) - return bool(c.call("disasm", addr="main", max_instructions=5)) - except IDAError as e: - errors.append(e); return False - def run_all(): - with concurrent.futures.ThreadPoolExecutor(max_workers=16) as ex: - return sum(ex.map(work, range(N))) - fin, ok = deadline(run_all, 60) - check("300/16 concurrent finished (no deadlock)", fin, "pool deadlocked") - check("300/16 all succeeded", ok == N, f"{ok}/{N}, errs={errors[:3]}") - c.close() - - -def s_concurrency_under_churn(): - """Hammer concurrently while the worker is hard-killed mid-flight. No hang, - no deadlock; errors are clean; after respawn calls succeed again.""" - sid = ensure_up() - c = IDAClient(URL, db=sid, pool_size=8) - c.connect() - stop = threading.Event() - stats = {"ok": 0, "err": 0, "weird": 0} - lock = threading.Lock() - def spinner(): - while not stop.is_set(): - try: - c.call("list_funcs", queries=[{"count": 1}]) - with lock: stats["ok"] += 1 - except IDAError: - with lock: stats["err"] += 1 - except Exception: # noqa: BLE001 -- any non-IDAError is a bug - with lock: stats["weird"] += 1 - time.sleep(0.01) - threads = [threading.Thread(target=spinner, daemon=True) for _ in range(12)] - for t in threads: t.start() - time.sleep(1.0) - ctl("kill9") # crash the worker under load - time.sleep(2.0) - stop.set() - for t in threads: t.join(10) - alive = [t for t in threads if t.is_alive()] - check("no spinner thread hung on worker kill", not alive, f"{len(alive)} stuck") - check("no non-IDAError leaked during churn", stats["weird"] == 0, str(stats)) - log(f" churn stats: {stats}") - # Recover. - ctl("stop"); ctl("start"); ensure_up() - c.close() - - -SCENARIOS = { - "connect_refused": s_connect_refused, - "timeout": s_timeout, - "worker_crash": s_worker_crash, - "full_restart_same_client": s_full_restart_same_client, - "no_autoswitch_when_explicit": s_no_autoswitch_when_explicit, - "concurrency_high": s_concurrency_high, - "concurrency_under_churn": s_concurrency_under_churn, -} - - -def main(argv): - names = argv or list(SCENARIOS) - for name in names: - fn = SCENARIOS.get(name) - if not fn: - log(f"unknown scenario: {name} (have: {', '.join(SCENARIOS)})") - return 2 - log(f"\n[{name}]") - t0 = time.time() - try: - fn() - except Exception as e: # noqa: BLE001 - global FAIL - FAIL += 1 - log(f" FAIL {name} raised {type(e).__name__}: {e}") - log(f" ({time.time() - t0:.1f}s)") - log("\n== ensuring server is UP for subsequent work ==") - sid = ensure_up() - log(f" server ready, db={sid}") - log(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/tests/stress_paging.py b/tests/stress_paging.py deleted file mode 100644 index b821354..0000000 --- a/tests/stress_paging.py +++ /dev/null @@ -1,150 +0,0 @@ -#!/usr/bin/env python3 -"""Stress the 'many lines of text' problem: paging huge listings/functions. - -Validates the windowing strategy the TUI will use so we never hand a widget more -than a viewport-sized slice. Prints concise stats only. - - python3 tests/stress_paging.py --db <session_id> -""" -import os -import statistics -import sys -import time - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.client import IDAClient # noqa: E402 - -URL = "http://127.0.0.1:8745/mcp" -# Verified server cap: list_* honors count up to ~700, then silently collapses -# to a 10-item default. Clamp with margin. -SAFE_PAGE = 500 - - -def ms(fn): - t = time.time() - r = fn() - return (time.time() - t) * 1e3, r - - -def q0(payload): - """Unwrap a *_query/list_* payload: {'result':[{'data':[...],'next_offset':N}]}.""" - res = payload.get("result", payload) - if isinstance(res, list): - res = res[0] if res else {} - return res.get("data", []), res.get("next_offset") - - -def page_all_funcs(c, page=SAFE_PAGE): - """Enumerate the entire function list correctly: advance by len(data), NOT by - next_offset (which is just offset+count and skips over the per-call cap).""" - funcs, offset, pages, t0 = [], 0, 0, time.time() - while True: - payload = c.call("list_funcs", queries=[{"offset": offset, "count": page}]) - data, _ = q0(payload) - funcs.extend(data) - pages += 1 - if len(data) < page: # short page => reached the end - break - offset += len(data) - return funcs, pages, (time.time() - t0) * 1e3 - - -def disasm_meta(c, addr): - # total_instructions is a TOP-LEVEL field, not under 'asm'. - payload = c.call("disasm", addr=addr, max_instructions=1, include_total=True) - return payload.get("total_instructions", payload.get("instruction_count")) - - -def main(argv): - db = None - it = iter(argv) - for a in it: - if a == "--db": - db = next(it) - c = IDAClient(URL, db=db) - c.connect() - if db is None: - c.set_db(c.resolve_db()) - print(f"db={c.db} ({c.health().get('module')})") - - # ---- 1. Full function list via cursor pagination ------------------- # - print("\n[1] page entire function list (cursor)") - funcs, pages, total_ms = page_all_funcs(c, page=SAFE_PAGE) - n = len(funcs) - print(f" {n} functions in {pages} pages, {total_ms:.0f}ms " - f"({total_ms / max(n,1):.3f}ms/func)") - - def sz(f): - s = f["size"] - return int(s, 16) if isinstance(s, str) else s - biggest = sorted(funcs, key=sz, reverse=True)[:5] - print(" biggest by bytes:") - for f in biggest: - print(f" {f['addr']:>12} {sz(f):#8x} {f['name']}") - - # ---- 2. Instruction counts of the biggest -------------------------- # - print("\n[2] instruction totals of biggest funcs") - fattest = None - fattest_n = 0 - for f in biggest: - dt, total = ms(lambda f=f: disasm_meta(c, f["addr"])) - print(f" {f['name']:<28} {str(total):>8} insns (meta {dt:.1f}ms)") - if isinstance(total, int) and total > fattest_n: - fattest, fattest_n = f, total - - # ---- 3. Windowed paging INTO the fattest function ------------------ # - print(f"\n[3] window-scroll fattest func {fattest['name']} ({fattest_n} insns)") - WIN = 60 # a viewport - offsets = list(range(0, max(fattest_n - WIN, 1), max((fattest_n // 12), 1))) - times = [] - for off in offsets: - dt, payload = ms(lambda off=off: c.call( - "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN)) - lines = payload.get("asm", {}).get("lines", []) - times.append(dt) - assert len(lines) <= WIN, f"got {len(lines)} > window {WIN}" - print(f" {len(offsets)} windowed reads @win={WIN}: " - f"min={min(times):.1f} med={statistics.median(times):.1f} " - f"max={max(times):.1f} ms (payload capped to <= {WIN} lines)") - - # ---- 3b. O(offset) latency curve (the key hazard) ------------------ # - print("\n[3b] disasm offset-latency curve (offset paging is O(offset))") - for off in [0, 1000, 5000, 20000, min(50000, fattest_n - WIN)]: - if off < 0: - continue - dt, _ = ms(lambda off=off: c.call( - "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN)) - print(f" offset={off:>6}: {dt:6.1f}ms") - - # ---- 4. Simulated 'hold page-down' throughput ---------------------- # - print("\n[4] rapid sequential scroll (hold page-down)") - reads, t0, off = 0, time.time(), 0 - while time.time() - t0 < 2.0: - c.call("disasm", addr=fattest["addr"], offset=off, max_instructions=WIN) - off = (off + WIN) % max(fattest_n - WIN, 1) - reads += 1 - dur = time.time() - t0 - print(f" {reads} windowed reads in {dur:.1f}s = {reads / dur:.0f} reads/s " - f"(~{reads / dur * WIN:.0f} lines/s)") - - # ---- 5. Decompile: truncation AND hard-failure on monsters --------- # - print("\n[5] decompile on fattest func (may fail on huge funcs)") - dt, payload = ms(lambda: c.call("decompile", addr=fattest["addr"])) - code = payload.get("code") if isinstance(payload, dict) else None - if not code: - err = payload.get("error") if isinstance(payload, dict) else payload - print(f" decompile {dt:.0f}ms -> FAILED (soft error, not raised): {err}") - else: - marker = "chars total]" in code[-40:] - print(f" decompile {dt:.0f}ms, returned {len(code)} chars, " - f"server-truncated={marker}") - if marker: - print(f" tail: ...{code[-60:]!r}") - - c.close() - print("\nOK") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py new file mode 100644 index 0000000..f930401 --- /dev/null +++ b/tests/test_blob_ui.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +"""A blob that analyses to NOTHING must still be usable. + +Zero functions is the normal outcome for a raw image described wrongly — and it +used to leave two empty panes and a status that said "functions still loading…", +which was a lie: loading had finished, there was simply nothing to land on. + +Needs IDA (spawns a real worker). ~40s. +""" +import asyncio +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from textual.widgets import Input, Static # noqa: E402 + +from idatui.app import ConfirmScreen, IdaTui, ListingView # noqa: E402 + +PASS = FAIL = 0 + + +def check(name, ok, detail=""): + global PASS, FAIL + if ok: + PASS += 1 + print(f" ok {name}") + else: + FAIL += 1 + print(f" FAIL {name} {detail}") + + +async def wait(pred, pilot, t=240.0): + for _ in range(int(t / 0.05)): + await pilot.pause(0.05) + try: + if pred(): + return True + except Exception: # noqa: BLE001 + pass + return False + + +async def run() -> int: + with tempfile.TemporaryDirectory() as tmp: + # Random bytes so IDA finds no functions... but with REAL AArch64 + # instructions planted at a known offset. Whether arbitrary random bytes + # happen to decode is chance, and a test that depends on chance tells you + # nothing on the run where it fails. + data = bytearray(os.urandom(64 * 1024)) + # -parm puts IDA in AArch64 mode, so these are A64 encodings; the ARM32 + # spelling of a nop (0xE1A00000) is NOT decodable there and made this + # test fail for a reason that had nothing to do with what it checks. + planted = 0x40 # file offset -> ea 0x4040 + for k, insn in enumerate((0xD503201F, # nop + 0xD503201F, # nop + 0xD65F03C0)): # ret <- the run must stop here + data[planted + k * 4:planted + k * 4 + 4] = insn.to_bytes(4, "little") + blob = os.path.join(tmp, "rnd.bin") + with open(blob, "wb") as f: + f.write(bytes(data)) + + # Skip the dialog by answering up front; this test is about what + # happens AFTER a described blob turns out to contain nothing. + app = IdaTui(open_path=blob, keepalive=False, load_args="-parm -b400") + async with app.run_test(size=(140, 44)) as pilot: + ok = await wait(lambda: app._func_index is not None + and app._func_index.complete, pilot) + check("a random blob still finishes loading", ok) + check("and really has no functions", len(app._func_index) == 0, + f"n={len(app._func_index)}") + + landed = await wait(lambda: app._cur is not None, pilot, 60) + check("it lands somewhere instead of leaving empty panes", landed, + f"cur={app._cur}") + lst = app.query_one(ListingView) + check("the listing actually has rows to show", + lst.total > 0, f"total={lst.total}") + check("landed at the image base", app._cur is not None + and app._cur.ea == 0x4000, f"{app._cur.ea if app._cur else None:#x}") + + status = str(app.query_one("#status", Static).render()) + check("the status says there are no functions (not 'still loading')", + "no functions" in status and "still loading" not in status, + status[:90]) + check("and points at the likely cause", + "processor" in status and "Ctrl+L" in status, status[:90]) + + # The hint is a property of the database, so it must survive moving + # around — an earlier version wrote it once and the next status + # write erased it. + lst.focus() + await pilot.press("down") + await pilot.press("down") + await pilot.pause(0.3) + status2 = str(app.query_one("#status", Static).render()) + check("the hint survives navigating", "no functions" in status2, + status2[:90]) + + # -- byte-granular carving ---------------------------------- # + # An undefined run arrives as ONE head ("db N dup(?)"). It has to + # PRESENT as one row per byte or there is no cursor position to + # press `c` at, which is how IDA works and the only way to find an + # instruction stream that doesn't start at the run's first byte. + m = lst.model + check("an undefined run presents one row per byte", + m.loaded() >= 4096 and len(m._heads) < 64, + f"rows={m.loaded()} physical heads={len(m._heads)}") + rows = m.window(0, 4) + check("each row is a single addressable byte", + [h.ea for h in rows] == [0x4000, 0x4001, 0x4002, 0x4003] + and all(h.size == 1 for h in rows), + f"{[(hex(h.ea), h.size) for h in rows]}") + check("and shows its value, not a placeholder", + all(h.text.startswith("db ") and "dup" not in h.text + for h in rows), f"{[h.text for h in rows]}") + + # The point of all this: land on an arbitrary byte and convert it. + i = m.index_of_ea(0x4021) + check("an address inside the run resolves to its own row", + i >= 0 and m.get(i).ea == 0x4021, + f"row={i} ea={m.get(i).ea if i >= 0 else None}") + + target = 0x4000 + planted # a NOP we put there ourselves + lst.cursor = m.index_of_ea(target) + lst._scroll_cursor_into_view() + await pilot.pause(0.1) + check("the cursor sits on the byte we aimed at", + lst._cursor_ea() == target, + f"{lst._cursor_ea():#x} want {target:#x}") + await pilot.press("c") + await pilot.pause(2.0) + # Defining an item REBUILDS the listing model, so re-read it from the + # view: holding the old object shows the pre-edit rows and looks + # exactly like the edit silently failing. + m = lst.model + h = m.get(m.index_of_ea(target)) + check("`c` on a chosen byte carves an instruction there", + h is not None and h.kind == "code", + f"kind={h.kind if h else None} text={h.text if h else None!r}") + if h is not None and h.kind == "code": + print(f" carved {target:#x}: {h.text}") + # `c` runs until something stops it, like IDA — one instruction + # at a time means pressing it once per opcode for the length of + # a routine. We planted nop/nop/nop/ret, so it must take all + # four and stop AT the ret, not run on into the random bytes + # after it. + run = [m.get(m.index_of_ea(target + k * 4)) for k in range(3)] + check("`c` keeps going until control flow ends", + all(x is not None and x.kind == "code" for x in run), + f"{[(hex(x.ea), x.kind) for x in run if x]}") + check("and stops at the ret instead of running into junk", + m.get(m.index_of_ea(target + 12)).kind == "unknown", + f"{m.get(m.index_of_ea(target + 12)).text!r}") + status = str(app.query_one("#status", Static).render()) + check("the status reports what the run did", + "3 instructions" in status and "control flow" in status, + status[:80]) + check("the carved row spans the instruction, not one byte", + h.size == 4, f"size={h.size}") + check("bytes before it stay individually addressable", + m.get(m.index_of_ea(target - 1)).size == 1 + and m.get(m.index_of_ea(target - 1)).ea == target - 1) + + # -- an edit must not move the view -------------------------- # + # Every mutation rebuilds the model, and row indices don't survive + # that: carving collapses four byte rows into one instruction row. + # All the edit paths go through ViewAnchor, which remembers + # ADDRESSES. This runs in its own app, so unlike the shared scenario + # suite it can mutate the database freely. + lst.cursor = m.index_of_ea(0x4200) + lst._scroll_cursor_into_view() + await pilot.pause(0.4) + ctop = lst.model.get(round(lst.scroll_offset.y)).ea + ccur = lst._cursor_ea() + old = lst.model + await pilot.press("semicolon") + await wait(lambda: app.query_one("#comment", Input).display, pilot, 20) + for ch in "note": + await pilot.press(ch) + await pilot.press("enter") + await wait(lambda: lst.model is not old and lst.model is not None, + pilot, 30) + await pilot.pause(0.4) + check("commenting leaves the view where it was", + lst.model.get(round(lst.scroll_offset.y)).ea == ctop + and lst._cursor_ea() == ccur, + f"top {ctop:#x} -> " + f"{lst.model.get(round(lst.scroll_offset.y)).ea:#x}, " + f"cursor {ccur:#x} -> {lst._cursor_ea():#x}") + + # -- carving must not move the view -------------------------- # + # Defining code collapses rows (four byte rows become one + # instruction), so anything that remembers a row INDEX puts you + # somewhere else afterwards. Scroll down far enough that there is a + # viewport to lose, then check the top ADDRESS is unchanged. + far = 0x4000 + 0x600 + lst.cursor = lst.model.index_of_ea(far) + lst._scroll_cursor_into_view() + await pilot.pause(0.4) + top_before = lst.model.get(round(lst.scroll_offset.y)).ea + cur_before = lst._cursor_ea() + check("scrolled somewhere with rows above us", + round(lst.scroll_offset.y) > 0, f"top={lst.scroll_offset.y}") + await pilot.press("c") + await pilot.pause(2.5) + m2 = lst.model + top_after = m2.get(round(lst.scroll_offset.y)).ea + check("carving leaves the scroll position where it was", + top_after == top_before, + f"{top_before:#x} -> {top_after:#x}") + check("and leaves the cursor on the same address", + lst._cursor_ea() == cur_before, + f"{cur_before:#x} -> {lst._cursor_ea():#x}") + + await pilot.press("ctrl+l") + opened = await wait(lambda: isinstance(app.screen, ConfirmScreen), pilot, 20) + check("Ctrl+L offers to reload with different options", opened, + f"screen={type(app.screen).__name__}") + if opened: + note = str(app.screen.query_one("#confirm-note", Static).render()) + check("and says nothing is lost when there's nothing to lose", + "nothing is lost" in note, note[:70]) + await pilot.press("escape") + await pilot.pause(0.3) + check("declining leaves the binary open", + not isinstance(app.screen, ConfirmScreen) and app._cur is not None) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(run())) diff --git a/tests/test_domain.py b/tests/test_domain.py deleted file mode 100644 index 6875b66..0000000 --- a/tests/test_domain.py +++ /dev/null @@ -1,155 +0,0 @@ -#!/usr/bin/env python3 -"""Validate the domain/paging layer against a real, large binary. - -Run with a session open on a big binary (e.g. libcrypto.so.3, 10k funcs): - - python3 tests/test_domain.py --db <session_id> - -Checks: correct full pagination (advance by len, not next_offset), viewport -slicing across block boundaries, window caching (revisit is instant), prefetch -warming, cached instruction totals, decompile success + hard-failure handling, -and address resolution. -""" -import os -import sys -import time - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.client import IDAClient # noqa: E402 -from idatui.domain import DISASM_BLOCK, LIST_PAGE, Program # noqa: E402 - -URL = "http://127.0.0.1:8745/mcp" -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}") - - -def ms(fn): - t = time.time() - r = fn() - return (time.time() - t) * 1e3, r - - -def main(argv): - db = None - it = iter(argv) - for a in it: - if a == "--db": - db = next(it) - c = IDAClient(URL, db=db) - c.connect() - if db is None: - c.set_db(c.resolve_db()) - prog = Program(c) - module = c.health().get("module") - total_funcs = c.call("survey_binary").get("statistics", {}).get("total_functions") - print(f"db={c.db} module={module} survey_total_functions={total_funcs}") - - # ---- function index: full enumeration correctness ------------------ # - print("\n[function index]") - idx = prog.functions() - dt, _ = ms(lambda: idx.load_all()) - n = len(idx) - check("enumerated all functions (matches survey)", n == total_funcs, - f"got {n} vs survey {total_funcs}") - print(f" loaded {n} funcs in {dt:.0f}ms ({dt / max(n,1):.3f}ms/func), " - f"page={LIST_PAGE}") - # no duplicates, monotonic-ish uniqueness by addr - addrs = [idx.get(i).addr for i in range(min(n, 3000))] - check("no duplicate addrs in first 3000", len(addrs) == len(set(addrs))) - # viewport slice (adapt to binary size) - wstart = min(1000, max(n - 50, 0)) - wlen = min(50, n - wstart) - w = idx.window(wstart, wlen) - check(f"window({wstart},{wlen}) returns {wlen}", len(w) == wlen, str(len(w))) - - # ---- filtered index uses server-side glob -------------------------- # - print("\n[filtered index]") - sub = prog.functions(filter="sub_*") - sub.ensure(10) - check("filter sub_* yields sub_ names", all(f.name.startswith("sub_") for f in sub.window(0, 10)), - str([f.name for f in sub.window(0, 5)])) - - # ---- pick the fattest function for disasm stress ------------------- # - print("\n[disasm windowing]") - fattest = max((idx.get(i) for i in range(n)), key=lambda f: f.size) - dm = prog.disasm(fattest.addr, fattest.name) - dt, total = ms(dm.total) - check("total() returns positive count", total > 0, str(total)) - dt2, total2 = ms(dm.total) - check("total() cached (2nd call ~instant)", dt2 < dt / 2 + 1, f"{dt:.0f}ms -> {dt2:.1f}ms") - print(f" fattest {fattest.name}: {total} insns, total() {dt:.0f}ms then {dt2:.1f}ms") - - # viewport across a block boundary - start = DISASM_BLOCK - 5 - win = dm.lines(start, 60, prefetch=False) - check("viewport spans block boundary, right length", - len(win) == min(60, max(total - start, 0)), f"got {len(win)}") - # addresses strictly increasing and contiguous slice - eas = [ln.ea for ln in win] - check("viewport addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])), - str(eas[:4])) - - # deep window: first slow (O(offset)), revisit instant (cached) - if total > DISASM_BLOCK * 4: - deep = (total // DISASM_BLOCK - 1) * DISASM_BLOCK - dt_cold, a = ms(lambda: dm.lines(deep, 60, prefetch=False)) - dt_warm, b = ms(lambda: dm.lines(deep, 60, prefetch=False)) - check("deep window revisit is cached/instant", dt_warm < dt_cold / 2 + 1, - f"cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms") - check("cached window identical", [l.ea for l in a] == [l.ea for l in b]) - print(f" deep@{deep}: cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms") - - # prefetch warms the next block - print("\n[prefetch]") - dm2 = prog.disasm(fattest.addr + 0) # same model (cached by ea) - fresh = prog.disasm(idx.get(0).addr, idx.get(0).name) - fresh.lines(0, 60, prefetch=True) # should prefetch block 1 - time.sleep(0.3) - check("prefetch warmed a neighbor block", fresh.cached_blocks() >= 2, - f"cached_blocks={fresh.cached_blocks()}") - - # ---- decompile: success + hard-failure ----------------------------- # - print("\n[decompile]") - # a small function likely decompiles - small = min((idx.get(i) for i in range(n)), key=lambda f: f.size if f.size > 4 else 1 << 30) - d_small = prog.decompile(small.addr) - check("small func decompiles or fails cleanly", isinstance(d_small.failed, bool)) - dt_c, _ = ms(lambda: prog.decompile(small.addr)) - check("decompile cached (2nd ~instant)", dt_c < 5, f"{dt_c:.1f}ms") - # the monster should hard-fail as a soft error (not raise) - d_big = prog.decompile(fattest.addr) - check("monster decompile handled (failed flag, no raise)", - d_big.failed or d_big.code is not None, - f"failed={d_big.failed} err={d_big.error}") - print(f" small {small.name}: failed={d_small.failed} " - f"trunc={d_small.truncated} chars={d_small.total_chars}") - print(f" monster {fattest.name}: failed={d_big.failed} err={d_big.error}") - - # ---- resolve ------------------------------------------------------- # - print("\n[resolve]") - check("resolve hex", prog.resolve(hex(fattest.addr)) == fattest.addr) - check("resolve int passthrough", prog.resolve(fattest.addr) == fattest.addr) - named = next((idx.get(i) for i in range(n) if not idx.get(i).name.startswith("sub_")), None) - if named: - try: - r = prog.resolve(named.name) - check("resolve symbol name", r == named.addr, f"{hex(r)} vs {hex(named.addr)} ({named.name})") - except KeyError as e: - check("resolve symbol name", False, str(e)) - - prog.close() - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/tests/test_formats.py b/tests/test_formats.py new file mode 100644 index 0000000..ab72e8d --- /dev/null +++ b/tests/test_formats.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""Format sniffing + load-switch construction (pure stdlib, no IDA).""" +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from idatui.formats import (PROCESSORS, load_args, # noqa: E402 + needs_load_options, sniff) + +PASS = FAIL = 0 + + +def check(name, ok, detail=""): + global PASS, FAIL + if ok: + PASS += 1 + print(f" ok {name}") + else: + FAIL += 1 + print(f" FAIL {name} {detail}") + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp: + def w(name, data): + p = os.path.join(tmp, name) + with open(p, "wb") as f: + f.write(data) + return p + + # -- formats IDA can load on its own: never interrupt the user -------- # + for name, head in (("elf", b"\x7fELF\x02\x01\x01"), ("pe", b"MZ\x90\x00"), + ("macho", b"\xcf\xfa\xed\xfe"), ("dex", b"dex\n035\x00"), + ("wasm", b"\x00asm\x01\x00")): + p = w(name, head + bytes(60)) + check(f"{name} is recognised (no dialog)", + sniff(p) is not None and not needs_load_options(p), f"{sniff(p)}") + + # -- the case this exists for ----------------------------------------- # + blob = w("fw.bin", bytes(range(256)) * 4) + check("a headerless blob is not recognised (ask)", + sniff(blob) is None and needs_load_options(blob)) + + # Intel HEX / S-records are text containers IDA does load. + hexf = w("f.hex", b":10010000214601360121470136007EFE09D2190140\n") + check("Intel HEX is recognised", sniff(hexf) == "Intel HEX", f"{sniff(hexf)}") + srec = w("f.s19", b"S00600004844521B\nS1130000285F245F2212226A000424290008237C\n") + check("Motorola S-records are recognised", + sniff(srec) == "Motorola S-record", f"{sniff(srec)}") + + # A binary that merely STARTS with ':' is not Intel HEX. Text formats are + # only accepted when the whole head is printable, or half the firmware in + # the world gets mis-detected on one byte. + colon = w("colon.bin", b":\x00\xff\xfe\x01\x02" + bytes(58)) + check("a blob starting with ':' is not mistaken for Intel HEX", + sniff(colon) is None, f"{sniff(colon)}") + + check("an empty file is not recognised", sniff(w("empty", b"")) is None) + check("a missing file never asks (nothing to load)", + not needs_load_options(os.path.join(tmp, "nope"))) + check("a directory never asks", not needs_load_options(tmp)) + + # -- switch construction ------------------------------------------------- # + # -b is in PARAGRAPHS: 0x8000000 >> 4 == 0x800000. Getting this wrong loads + # the image 16x off and every address in the database is wrong. + check("base is converted to paragraphs", + load_args("arm", 0x8000000) == "-parm -b800000", + load_args("arm", 0x8000000)) + check("processor alone", load_args("mipsb") == "-pmipsb", load_args("mipsb")) + check("base alone", load_args("", 0x10000) == "-b1000", load_args("", 0x10000)) + check("base 0 emits no switch (it's the default)", + load_args("arm", 0) == "-parm", load_args("arm", 0)) + check("nothing in, nothing out", load_args() == "") + check("extra switches pass through", + load_args("arm", 0, "-T binary") == "-parm -T binary") + + check("the processor list leads with the common targets", + [n for n, _ in PROCESSORS[:3]] == ["arm", "armb", "metapc"], + f"{[n for n, _ in PROCESSORS[:3]]}") + + # Every offered name must have been checked against a real IDA, because a + # wrong one is REJECTED (rc=4) with nothing useful said — handing the user a + # dead end from inside the dialog meant to rescue them. This list is the + # output of tools/verify_procs.py; adding a processor without re-running it + # fails here on purpose. + VERIFIED = { + "arm", "armb", "metapc", "mipsl", "mipsb", "ppc", "ppcl", "sh4", "68k", + "riscv", "tricore", "xtensa", "avr", "z80", "tms320c6", "m32r", "arc", + "h8300", "sparcb", "sparcl", "s390", + } + offered = {n for n, _ in PROCESSORS} + check("every offered processor name is IDA-verified", + offered <= VERIFIED, f"unverified: {sorted(offered - VERIFIED)}") + + # These are module FILENAMES or common aliases, not -p names. IDA refuses + # them; they were in the list until a real run said otherwise. + for wrong in ("h8", "sparc", "arm64", "aarch64", "mips", "m68k"): + check(f"{wrong!r} is not offered (IDA rejects it)", wrong not in offered) + + # ...but someone WILL type them, so the labels have to carry the alias. + def finds(q): + ql = q.lower() + return [n for n, d in PROCESSORS if ql in n.lower() or ql in d.lower()] + check("typing 'arm64' still finds ARM", "arm" in finds("arm64"), f"{finds('arm64')}") + check("typing 'aarch64' still finds ARM", "arm" in finds("aarch64")) + check("typing 'm68k' still finds 68k", "68k" in finds("m68k"), f"{finds('m68k')}") + check("typing 'mips' finds both endiannesses", + set(finds("mips")) == {"mipsl", "mipsb"}, f"{finds('mips')}") + check("every processor entry has a human label", + all(n and d for n, d in PROCESSORS)) + check("endianness is spelled out where it matters", + all(any(w in d.lower() for w in ("endian",)) + for n, d in PROCESSORS if n in ("arm", "armb", "mipsb", "mipsl"))) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_index.py b/tests/test_index.py new file mode 100644 index 0000000..e14f715 --- /dev/null +++ b/tests/test_index.py @@ -0,0 +1,202 @@ +#!/usr/bin/env python3 +"""Unit tests for idatui.index (the project-wide symbol/string index). + +Pure stdlib: no IDA, no textual, no worker. + + python tests/test_index.py +""" +import os +import sys +import tempfile +import time + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from idatui.index import (KIND_EXPORT, KIND_FUNC, KIND_IMPORT, # noqa: E402 + KIND_STRING, ProjectIndex) + +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}") + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp: + src = os.path.join(tmp, "libfoo.so") + with open(src, "wb") as f: + f.write(b"\x7fELF binary") + idx = ProjectIndex(os.path.join(tmp, "idx", "project.db")) + + check("a fresh index is empty", idx.total() == 0 and idx.counts() == {}) + check("an unindexed binary is stale", idx.is_stale("libfoo", src)) + + n = idx.reindex("libfoo", [ + (KIND_FUNC, 0x1000, "SSL_CTX_new"), + (KIND_FUNC, 0x1100, "SSL_read"), + (KIND_FUNC, 0x1200, "sub_1200"), + (KIND_STRING, 0x8000, "error opening socket"), + (KIND_STRING, 0x8100, "/etc/ssl/certs"), + ], source=src) + check("reindex reports what it stored", n == 5, f"n={n}") + check("the entries are there", idx.total() == 5, f"{idx.total()}") + check("an indexed binary is fresh", not idx.is_stale("libfoo", src)) + + # -- substring search (the thing a prefix index can't do) ----------- # + hits = idx.search("SSL", kind=KIND_FUNC) + check("finds symbols by substring", {h.text for h in hits} == + {"SSL_CTX_new", "SSL_read"}, f"{[h.text for h in hits]}") + check("matching ignores case across kinds (SSL also hits /etc/ssl)", + {h.text for h in idx.search("SSL")} == + {"SSL_CTX_new", "SSL_read", "/etc/ssl/certs"}, + f"{[h.text for h in idx.search('SSL')]}") + hits = idx.search("socket") + check("finds strings by substring mid-text", + len(hits) == 1 and hits[0].kind == KIND_STRING + and hits[0].addr == 0x8000, f"{hits}") + check("hits carry the owning binary", + all(h.binary == "libfoo" for h in idx.search("SSL"))) + check("search is case-insensitive", + {h.text for h in idx.search("ssl_read")} == {"SSL_read"}, + f"{[h.text for h in idx.search('ssl_read')]}") + check("kind filter narrows to strings", + [h.text for h in idx.search("ss", kind=KIND_STRING)] == ["/etc/ssl/certs"], + f"{[h.text for h in idx.search('ss', kind=KIND_STRING)]}") + + # -- the <3 char fallback (trigram silently matches nothing) -------- # + check("2-char query still works (LIKE fallback)", + {h.text for h in idx.search("ss")} == {"SSL_CTX_new", "SSL_read", + "/etc/ssl/certs"}, + f"{[h.text for h in idx.search('ss')]}") + check("1-char query still works", + len(idx.search("/")) == 1, f"{idx.search('/')}") + check("an empty query matches nothing", idx.search(" ") == []) + check("a query with FTS operators is treated literally", + idx.search('SSL OR "') == [] or True) # must not raise + + # -- multi-binary: the whole point ---------------------------------- # + idx.reindex("httpd", [ + (KIND_FUNC, 0x2000, "handle_ssl_request"), + (KIND_STRING, 0x9000, "socket bind failed"), + ]) + hits = idx.search("ssl") + check("search spans binaries", + {h.binary for h in hits} == {"libfoo", "httpd"}, + f"{[(h.binary, h.text) for h in hits]}") + check("counts are per binary", + idx.counts() == {"libfoo": 5, "httpd": 2}, f"{idx.counts()}") + + # -- incremental: reindexing one binary leaves the others alone ----- # + idx.reindex("libfoo", [(KIND_FUNC, 0x1000, "SSL_CTX_new_v2")], source=src) + check("reindex replaces only that binary's entries", + idx.counts() == {"libfoo": 1, "httpd": 2}, f"{idx.counts()}") + check("the stale entries are gone", + [h.text for h in idx.search("SSL_read")] == [], + f"{idx.search('SSL_read')}") + check("the other binary survived untouched", + len(idx.search("socket bind")) == 1) + + # -- staleness follows the source ----------------------------------- # + time.sleep(0.01) + with open(src, "wb") as f: + f.write(b"\x7fELF binary rebuilt, different size") + os.utime(src, (1, 1)) + check("a changed source goes stale", idx.is_stale("libfoo", src)) + check("a missing source does NOT wipe the index", + not idx.is_stale("libfoo", os.path.join(tmp, "gone"))) + + # -- forget ----------------------------------------------------------- # + idx.forget("httpd") + check("forget drops a binary entirely", + idx.counts() == {"libfoo": 1} and idx.search("socket bind") == [], + f"{idx.counts()}") + + # -- cross-binary linkage join (phase 3) ------------------------------ # + idx.reindex("app", [ + (KIND_FUNC, 0x1000, "main"), + (KIND_IMPORT, 0x2000, "strcmp"), + (KIND_IMPORT, 0x2008, "read"), + (KIND_IMPORT, 0x2010, "SSL_new"), + ]) + idx.reindex("libc", [ + (KIND_EXPORT, 0x8000, "strcmp"), + (KIND_EXPORT, 0x8100, "read"), + (KIND_EXPORT, 0x8200, "pread"), + (KIND_EXPORT, 0x8300, "read_line"), + (KIND_FUNC, 0x8000, "strcmp"), + ]) + idx.reindex("libssl", [(KIND_EXPORT, 0x9000, "SSL_new")]) + + prov = idx.providers("strcmp", exclude="app") + check("an import resolves to the binary that exports it", + [(h.binary, h.addr) for h in prov] == [("libc", 0x8000)], + f"{[(h.binary, hex(h.addr)) for h in prov]}") + + # The whole point of exact(): substring search would drag in pread, + # read_line and thread_start, and 'read' is also below the trigram floor + # for some engines — an import must bind to its exact name or nothing. + prov = idx.providers("read", exclude="app") + check("the join is exact, not substring", + [(h.binary, h.addr) for h in prov] == [("libc", 0x8100)], + f"{[(h.binary, h.text) for h in prov]}") + + check("a short name still resolves (below the trigram floor)", + [h.binary for h in idx.providers("SSL_new", exclude="app")] == ["libssl"], + f"{idx.providers('SSL_new')}") + + check("an unprovided import resolves to nothing", + idx.providers("dlopen", exclude="app") == []) + + check("exclude keeps a binary from resolving to itself", + idx.providers("strcmp", exclude="libc") == [], + f"{idx.providers('strcmp', exclude='libc')}") + + imp = idx.importers("strcmp") + check("the reverse join finds who imports an export", + [(h.binary, h.addr) for h in imp] == [("app", 0x2000)], + f"{[(h.binary, hex(h.addr)) for h in imp]}") + + check("kind keeps functions out of the linkage join", + [h.binary for h in idx.providers("strcmp")] == ["libc"], + "a KIND_FUNC row named strcmp must not answer as an export") + + idx.forget("libc") + check("forgetting a provider unresolves its imports", + idx.providers("strcmp", exclude="app") == []) + + # -- ELF symbol versioning -------------------------------------------- # + # The importer sees strrchr@@GLIBC_2.2.5 while the provider may export a + # different spelling; raw names would resolve almost nothing. link_name + # cuts at the first '@' so both sides meet on the bare symbol. + from idatui.domain import link_name + check("link_name strips an ELF version suffix", + link_name("strrchr@@GLIBC_2.2.5") == "strrchr", + link_name("strrchr@@GLIBC_2.2.5")) + check("link_name leaves an unversioned name alone", + link_name("strrchr") == "strrchr") + check("link_name handles a single-@ version", + link_name("SSL_new@OPENSSL_3.0.0") == "SSL_new") + check("link_name doesn't eat a leading @", + link_name("@weird") == "@weird", link_name("@weird")) + + # -- persistence ------------------------------------------------------ # + path = idx.path + idx.close() + idx2 = ProjectIndex(path) + check("the index persists across sessions", + [h.text for h in idx2.search("SSL_CTX")] == ["SSL_CTX_new_v2"], + f"{idx2.search('SSL_CTX')}") + idx2.close() + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_keepalive.py b/tests/test_keepalive.py deleted file mode 100644 index 2910e1a..0000000 --- a/tests/test_keepalive.py +++ /dev/null @@ -1,76 +0,0 @@ -#!/usr/bin/env python3 -"""Prove that an interactive session can 'just chill' without the worker idling -out. Uses a deliberately short worker TTL to keep the test fast. - - python3 tests/test_keepalive.py - -Needs ~/ida-venv + a running server (tests/serverctl.sh) and a writable target. -""" -import os -import sys -import time - -HERE = os.path.dirname(os.path.abspath(__file__)) -REPO = os.path.dirname(HERE) -sys.path.insert(0, REPO) -from idatui.client import IDAClient, IDAToolError # 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}") - - -def alive(c, sid): - try: - c.set_db(sid) - c.health() - return True - except IDAToolError: - return False - - -def main(): - target = os.path.join(REPO, "targets", "ls_ttl") - if not os.path.exists(target): - src = os.path.join(REPO, "bin", "ls") - os.makedirs(os.path.dirname(target), exist_ok=True) - import shutil - shutil.copy(src, target) - - c = IDAClient(timeout=300) - c.connect() - SHORT = 12 # worker self-exits after ~12s idle unless kept alive - - print("[heartbeat keeps a short-TTL worker alive]") - sid = c.call("idb_open", input_path=target, idle_ttl_sec=SHORT)["session"]["session_id"] - c.set_db(sid) - ka = c.keepalive(interval=4.0).start() - time.sleep(SHORT * 2 + 2) # idle well past the TTL, but heartbeat is beating - ok = alive(c, sid) - check("alive past 2x TTL with heartbeat", ok, f"beats={ka.beats}") - check("heartbeat actually beat", ka.beats >= 4, f"beats={ka.beats}") - check("heartbeat had no failures", ka.failures == 0, f"failures={ka.failures}") - ka.stop() - - print("\n[bump_idle_ttl makes it effectively immortal]") - sid = c.call("idb_open", input_path=target, idle_ttl_sec=SHORT)["session"]["session_id"] - c.set_db(sid) - c.bump_idle_ttl() # ~1e9 seconds - time.sleep(SHORT * 2 + 6) # zero requests during this window - check("alive past 2x TTL after bump, zero requests", alive(c, sid)) - - c.close() - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/test_pool.py b/tests/test_pool.py new file mode 100644 index 0000000..5c6e2c4 --- /dev/null +++ b/tests/test_pool.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +"""Unit tests for idatui.pool (worker residency: LRU + memory budget). + +Pure stdlib with a fake client injected, so the eviction policy is testable +without spawning real idalib workers. + + python tests/test_pool.py +""" +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from idatui.pool import WorkerPool # noqa: E402 +from idatui.project import Project # 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 FakeClient: + """Stands in for a WorkerClient: records saves/closes, reports fixed memory.""" + + def __init__(self, ref, mem=100): + self.ref = ref + self.mem = mem + self.saved = 0 + self.closed = False + self.connected = False + + def connect(self, progress=None, **kw): + self.connected = True + return self + + def call(self, tool, **kw): + if tool == "idb_save": + self.saved += 1 + return {} + + def close(self, grace=None): + self.closed = True + + +def _mkproject(tmp, n=4, size=64): + src = os.path.join(tmp, "src") + os.makedirs(src, exist_ok=True) + paths = [] + for i in range(n): + p = os.path.join(src, f"bin{i}") + with open(p, "wb") as f: + f.write(b"\x7fELF" + bytes(size)) + paths.append(p) + return Project.create(os.path.join(tmp, "p.json"), paths, name="p") + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp: + proj = _mkproject(tmp) + made = {} + + def spawn(ref, ttl): + c = FakeClient(ref, mem=100) + made[ref.label] = c + return c + + pool = WorkerPool(proj, budget_mb=350, spawn=spawn, + mem_fn=lambda c: c.mem) + + # -- lazy spawn + reuse -------------------------------------------- # + a = pool.get("bin0") + check("get() spawns a worker on first use", a is made["bin0"] and a.connected) + check("get() stages the binary first", + os.path.isfile(proj.by_label("bin0").staged)) + check("get() reuses the resident worker", pool.get("bin0") is a) + check("resident() reports it", pool.resident() == ["bin0"], pool.resident()) + + # -- LRU ordering ---------------------------------------------------- # + pool.get("bin1") + pool.get("bin2") + pool.get("bin0") # touch: bin0 becomes most-recent + check("LRU order tracks use", pool.resident() == ["bin1", "bin2", "bin0"], + pool.resident()) + + # -- budget eviction -------------------------------------------------- # + check("pool reports its memory", pool.memory_mb() == 300, pool.memory_mb()) + pool.get("bin3") # 400MB > 350MB budget -> evict LRU (bin1) + check("exceeding the budget evicts the least-recently-used", + pool.evicted == ["bin1"] and not pool.is_resident("bin1"), + f"evicted={pool.evicted} resident={pool.resident()}") + check("the just-spawned worker is never the victim", pool.is_resident("bin3")) + check("eviction saves the database first", made["bin1"].saved == 1) + check("eviction closes the worker", made["bin1"].closed) + check("pool is back within budget", pool.memory_mb() <= pool.budget_mb, + f"{pool.memory_mb()}/{pool.budget_mb}") + + # -- the active binary is never evicted ------------------------------- # + pool.set_active("bin2") + check("set_active touches the LRU", pool.resident()[-1] == "bin2", + pool.resident()) + pool.get("bin1") # over budget again -> must evict, but not bin2 + check("the active binary survives eviction", pool.is_resident("bin2"), + f"resident={pool.resident()}") + + # -- pinning ---------------------------------------------------------- # + pool.close_all() + pool2 = WorkerPool(proj, budget_mb=250, spawn=spawn, mem_fn=lambda c: c.mem) + pool2.get("bin0") + pool2.pin("bin0") + pool2.get("bin1") + pool2.get("bin2") # 300 > 250 -> evict, but bin0 is pinned + check("pinned binaries are never evicted", pool2.is_resident("bin0"), + f"resident={pool2.resident()} evicted={pool2.evicted}") + check("an unpinned one went instead", "bin1" in pool2.evicted, pool2.evicted) + + # -- everything pinned/active: stop evicting rather than thrash -------- # + pool2.pin("bin2") + pool2.set_active("bin2") + n_before = len(pool2.evicted) + pool2._enforce_budget() + check("nothing evictable -> gives up instead of thrashing", + len(pool2.evicted) == n_before, pool2.evicted) + + # -- status for the switcher UI ---------------------------------------- # + st = {s["label"]: s for s in pool2.status()} + check("status() covers every project binary", len(st) == 4, list(st)) + check("status() marks resident/pinned/active", + st["bin0"]["resident"] and st["bin0"]["pinned"] + and st["bin2"]["active"] and not st["bin3"]["resident"], + f"{st}") + + # -- teardown ----------------------------------------------------------- # + pool2.close_all() + check("close_all() closes every worker", + not pool2.resident() and all(c.closed for c in made.values())) + check("close_all() clears the active binary", pool2.active is None) + + # -- unknown label -------------------------------------------------------- # + try: + pool2.get("nope") + check("an unknown label raises KeyError", False, "no raise") + except KeyError: + check("an unknown label raises KeyError", True) + + # -- default budget comes from the project's memory_pct ------------------- # + pool3 = WorkerPool(proj, spawn=spawn, mem_fn=lambda c: c.mem) + check("default budget is derived, not a fixed worker count", + pool3.budget_mb >= 256, pool3.budget_mb) + + # -- prewarm: speculative, and never at the cost of a real binary ------ # + with tempfile.TemporaryDirectory() as tmp: + proj = _mkproject(tmp, n=3) + made2 = {} + + def spawn2(ref, ttl): + c = FakeClient(ref, mem=100) + made2[ref.label] = c + return c + + pool = WorkerPool(proj, budget_mb=250, spawn=spawn2, + mem_fn=lambda c: c.mem) + labels = [r.label for r in proj.refs] + a, b, c_ = labels[0], labels[1], labels[2] + pool.get(a) + pool.set_active(a) + check("prewarm warms a binary when the budget has room", + pool.prewarm(b) is True and b in pool.resident(), f"{pool.resident()}") + check("prewarm is a no-op for something already resident", + pool.prewarm(b) is False) + # 2 x 100MB resident, estimate 100 more -> 300 > 250: must refuse + check("prewarm refuses rather than making room", + pool.prewarm(c_) is False and c_ not in pool.resident(), + f"resident={pool.resident()} mem={pool.memory_mb()}/{pool.budget_mb}") + check("refusing to prewarm evicts nothing", + set(pool.resident()) == {a, b}, f"{pool.resident()}") + check("prewarm ignores a label outside the project", + pool.prewarm("nope") is False) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..91fd250 --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +"""Unit tests for idatui.project (the multi-binary project model + staging). + +Pure stdlib: no IDA, no textual, no worker — runs anywhere in under a second. + + python tests/test_project.py +""" +import json +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from idatui.project import Project, ProjectError, SIDECAR_SUFFIX # 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}") + + +def _bin(path, content=b"\x7fELF fake binary"): + with open(path, "wb") as f: + f.write(content) + return path + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp: + src = os.path.join(tmp, "src") + os.makedirs(src) + httpd = _bin(os.path.join(src, "httpd")) + libauth = _bin(os.path.join(src, "libauth.so"), b"\x7fELF lib") + pfile = os.path.join(tmp, "router-fw.json") + + # -- create / load round-trip ------------------------------------- # + proj = Project.create(pfile, [httpd, libauth], name="router-fw") + check("create() writes the project file", os.path.isfile(pfile)) + proj = Project.load(pfile) + check("load() round-trips name + binaries", + proj.name == "router-fw" and len(proj.refs) == 2, + f"name={proj.name} n={len(proj.refs)}") + check("labels default to the basename", + [r.label for r in proj.refs] == ["httpd", "libauth.so"], + f"{[r.label for r in proj.refs]}") + + # -- layout -------------------------------------------------------- # + check("sidecar sits beside the project file", + proj.sidecar == os.path.join(tmp, "router-fw" + SIDECAR_SUFFIX), + proj.sidecar) + ref = proj.by_label("httpd") + check("staged path lives in the sidecar, not the source tree", + ref.staged.startswith(proj.bin_dir) and src not in ref.staged, + ref.staged) + check("db path hangs off the staged file", ref.db == ref.staged + ".i64") + + # -- staging: hardlink, freshness ---------------------------------- # + check("a binary starts out stale (not yet staged)", proj.is_stale(ref)) + proj.stage(ref) + check("stage() materialises the binary in the sidecar", + os.path.isfile(ref.staged)) + check("stage() copies (distinct inode) so the source can't be mutated " + "through it", + os.stat(ref.staged).st_ino != os.stat(ref.source).st_ino + and open(ref.staged, "rb").read() == open(ref.source, "rb").read()) + check("a staged binary is no longer stale", not proj.is_stale(ref)) + check("source tree stays clean (no IDA artifacts beside it)", + sorted(os.listdir(src)) == ["httpd", "libauth.so"], + f"{sorted(os.listdir(src))}") + + # -- a changed source re-stages and drops the stale DB ------------- # + open(ref.db, "wb").write(b"fake i64") + open(ref.staged + ".id0", "wb").write(b"scratch") + check("has_db() sees the analysed database", proj.has_db(ref)) + # rewritten IN PLACE (same inode) — the case a hardlink would miss + _bin(ref.source, b"\x7fELF fake binary v2 (longer)") + os.utime(ref.source, (1, 1)) + check("a source rebuilt in place goes stale", proj.is_stale(ref)) + proj.stage(ref) + check("re-staging refreshes the staged bytes", + open(ref.staged, "rb").read().endswith(b"v2 (longer)")) + check("re-staging drops the now-stale database", not proj.has_db(ref)) + check("re-staging drops the stale scratch too", + not os.path.exists(ref.staged + ".id0")) + + # -- scratch sweep keeps the DB ------------------------------------ # + open(ref.db, "wb").write(b"fake i64") + for suf in (".id0", ".id1", ".nam"): + open(ref.staged + suf, "wb").write(b"x") + n = proj.sweep_scratch(ref) + check("sweep_scratch() removes the unpacked working files", n == 3, f"n={n}") + check("sweep_scratch() never touches the .i64", proj.has_db(ref)) + + # -- relative paths + label collisions ----------------------------- # + sub = os.path.join(tmp, "other") + os.makedirs(sub) + dup = _bin(os.path.join(sub, "httpd"), b"\x7fELF other httpd") + with open(pfile, "w") as f: + json.dump({"name": "p", "binaries": [ + {"path": "src/httpd"}, # relative to the project file + {"path": dup}, # same basename -> collision + {"path": libauth, "label": "auth"}, + ]}, f) + proj2 = Project.load(pfile) + check("relative paths resolve against the project file", + proj2.refs[0].source == httpd, proj2.refs[0].source) + check("colliding labels are disambiguated", + [r.label for r in proj2.refs] == ["httpd", "httpd_2", "auth"], + f"{[r.label for r in proj2.refs]}") + check("explicit labels are honoured", proj2.by_label("auth") is not None) + proj2.stage_all() + check("stage_all() stages every binary to a distinct file", + len({r.staged for r in proj2.refs}) == 3 + and all(os.path.isfile(r.staged) for r in proj2.refs)) + + # -- add / remove ---------------------------------------------------- # + extra = _bin(os.path.join(src, "extra")) + proj2.add(extra, label="extra") + check("add() appends a binary", proj2.by_label("extra") is not None) + + # -- re-adding must not duplicate (matched by resolved path) --------- # + n = len(proj2.refs) + proj2.add(extra) + check("re-adding the same path is a no-op", len(proj2.refs) == n, + f"{[r.label for r in proj2.refs]}") + os.chdir(src) + proj2.add("./extra") # same file, relative + proj2.add(os.path.join(src, "..", "src", "extra")) # same file, messy + check("a different spelling of the same path is a no-op", + len(proj2.refs) == n, f"{[r.label for r in proj2.refs]}") + link = os.path.join(src, "extra_link") + os.symlink(extra, link) + proj2.add(link) + check("a symlink to an existing binary is a no-op", + len(proj2.refs) == n, f"{[r.label for r in proj2.refs]}") + + # ...but a DIFFERENT file with the same basename must still be added + other_dir = os.path.join(tmp, "other2") + os.makedirs(other_dir) + twin = _bin(os.path.join(other_dir, "extra"), b"\x7fELF a different extra") + proj2.add(twin) + check("a same-named file from another directory IS added", + len(proj2.refs) == n + 1 + and proj2.by_source(twin) is not None + and proj2.by_source(extra) is not proj2.by_source(twin), + f"{[(r.label, r.source) for r in proj2.refs[-2:]]}") + check("the twins get distinct labels", + len({r.label for r in proj2.refs}) == len(proj2.refs), + f"{[r.label for r in proj2.refs]}") + check("create() also drops repeats on the command line", + len(Project.create(os.path.join(tmp, "dup.json"), + [extra, "./extra", extra]).refs) == 1) + os.chdir(tmp) + proj2.remove(proj2.by_source(twin).label) + check("remove() drops one", proj2.remove("extra") + and proj2.by_label("extra") is None) + + # -- bad input ------------------------------------------------------- # + bad = os.path.join(tmp, "bad.json") + with open(bad, "w") as f: + f.write("{not json") + try: + Project.load(bad) + check("malformed JSON raises ProjectError", False, "no raise") + except ProjectError: + check("malformed JSON raises ProjectError", True) + with open(bad, "w") as f: + json.dump({"binaries": []}, f) + try: + Project.load(bad) + check("an empty project raises ProjectError", False, "no raise") + except ProjectError: + check("an empty project raises ProjectError", True) + missing = proj2.add(os.path.join(tmp, "nope")) + try: + proj2.stage(missing) + check("staging a missing binary raises ProjectError", False, "no raise") + except ProjectError: + check("staging a missing binary raises ProjectError", True) + + # -- load options for headerless blobs --------------------------------- # + with tempfile.TemporaryDirectory() as tmp: + src = os.path.join(tmp, "src"); os.makedirs(src) + blob = os.path.join(src, "fw.bin") + with open(blob, "wb") as f: + f.write(b"\x00" * 64) + proj = Project.create(os.path.join(tmp, "p.json"), [blob], name="p", + load={"processor": "arm", "base": 0x8000000}) + r = proj.refs[0] + check("create() records load options per binary", + r.processor == "arm" and r.base == 0x8000000, + f"proc={r.processor!r} base={r.base:#x}") + # -b is PARAGRAPHS: 0x8000000 >> 4 == 0x800000. Getting this wrong loads + # the image 16x too high and every address in the database is wrong. + check("base is converted to IDA's paragraph units", + r.load_args == "-parm -b800000", r.load_args) + + proj2 = Project.load(proj.path) + check("load options survive a round-trip through the file", + proj2.refs[0].load_args == "-parm -b800000", + proj2.refs[0].load_args) + + blob2 = os.path.join(src, "other.bin") + with open(blob2, "wb") as f: + f.write(b"\x00" * 64) + r2 = proj2.add(blob2, load={"processor": "mipsb"}) + check("add() takes load options too", + r2.load_args == "-pmipsb", r2.load_args) + + # a normal ELF needs none of this and must pass nothing + check("a binary with no load options passes no switches", + Project.create(os.path.join(tmp, "q.json"), [blob], + name="q").refs[0].load_args == "") + + # addresses get written by hand, so accept how people write them + proj3 = Project.create(os.path.join(tmp, "r.json"), [blob], name="r", + load={"base": "0x1000"}) + check("a base given as a hex STRING is parsed", + proj3.refs[0].base == 0x1000, f"{proj3.refs[0].base}") + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_project_ui.py b/tests/test_project_ui.py new file mode 100644 index 0000000..54c06bd --- /dev/null +++ b/tests/test_project_ui.py @@ -0,0 +1,285 @@ +#!/usr/bin/env python3 +"""End-to-end pilot for project mode: two binaries, switching between them. + +Needs idalib (it spawns real workers, one per binary) and textual: + + ~/ida-venv/bin/python tests/test_project_ui.py [bin1 bin2] + +Defaults to targets/echo + targets/cat. The binaries are copied into a temp +source dir first, so the "source tree stays pristine" promise is checkable. +""" +import asyncio +import os +import shutil +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from idatui._sync import wait_for # noqa: E402 +from idatui.app import IdaTui, ProjectPalette # noqa: E402 +from idatui.project import Project # noqa: E402 +from textual.widgets import Input, OptionList, Static # 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}") + + +async def run(bins): + repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + with tempfile.TemporaryDirectory() as tmp: + src = os.path.join(tmp, "src") + os.makedirs(src) + srcs = [] + for b in bins: + dst = os.path.join(src, os.path.basename(b)) + shutil.copy2(b, dst) + srcs.append(dst) + proj = Project.create(os.path.join(tmp, "proj.json"), srcs, name="proj") + proj.stage_all() + first, second = (r.label for r in proj.refs) + + app = IdaTui(keepalive=False, project=proj) + async with app.run_test(size=(140, 44)) as pilot: + async def settle(pred, t=180.0): + return await wait_for(pred, pilot.pause, t, 0.05) + + # -- boots on the project's first binary ----------------------- # + ok = await settle(lambda: app.program is not None + and app._func_index is not None + and app._func_index.complete) + check("project mode boots on the first binary", ok, + f"binary={app._binary}") + check("the active binary is the first one", app._binary == first, + f"{app._binary}") + n_first = len(app._func_index) + check("its functions loaded", n_first > 10, f"n={n_first}") + status = str(app.query_one("#status", Static).render()) + check("the status line names the active binary", + f"[{first}]" in status, status[:60]) + + # -- the switcher lists the project ---------------------------- # + await pilot.press("ctrl+o") + opened = await settle( + lambda: isinstance(app.screen, ProjectPalette), 20) + check("Ctrl+O opens the binary switcher", opened, + f"screen={type(app.screen).__name__}") + if not opened: + return + pal = app.screen + check("the switcher lists every project binary", + len(pal._results) == 2, f"{[e['label'] for e in pal._results]}") + check("it marks which one is active", + any(e["active"] and e["label"] == first for e in pal._results)) + check("it marks the other as not yet opened", + any(not e["resident"] and e["label"] == second + for e in pal._results)) + ol = pal.query_one(OptionList) + check("the switcher opens on the binary you're already in", + ol.highlighted is not None + and pal._results[ol.highlighted]["label"] == first, + f"highlighted={ol.highlighted} " + f"={pal._results[ol.highlighted]['label'] if ol.highlighted is not None else None} " + f"want={first}") + + # -- switch to the second binary -------------------------------- # + pal.query_one(Input).value = second + await pilot.pause(0.2) + await pilot.press("enter") + switched = await settle( + lambda: app._binary == second and app.program is not None + and app._func_index is not None and app._func_index.complete) + check("switching opens the other binary", switched, + f"binary={app._binary}") + check("the second binary has its own function index", + app._func_index is not None and len(app._func_index) > 5, + f"n={len(app._func_index) if app._func_index else 0}") + check("both binaries now have live workers", + sorted(app._pool.resident()) == sorted([first, second]), + f"{app._pool.resident()}") + landed = await settle(lambda: app._cur is not None, 60) + check("it lands somewhere in the new binary", landed, + f"cur={app._cur}") + where = app._cur.ea if app._cur else None + + # -- switch back: resident, so state is restored ---------------- # + await pilot.press("ctrl+o") + reopened = await settle( + lambda: isinstance(app.screen, ProjectPalette), 20) + check("the switcher reopens after a switch", reopened, + f"screen={type(app.screen).__name__}") + if not reopened: + return + app.screen.query_one(Input).value = first + await pilot.pause(0.2) + await pilot.press("enter") + back = await settle(lambda: app._binary == first + and app._func_index is not None + and app._func_index.complete, 120) + check("switching back returns to the first binary", back, + f"binary={app._binary}") + check("its function index came back intact", + app._func_index is not None and len(app._func_index) == n_first, + f"n={len(app._func_index) if app._func_index else 0} want={n_first}") + + # and forward again: the second binary's position was remembered + await pilot.press("ctrl+o") + if not await settle(lambda: isinstance(app.screen, ProjectPalette), 20): + check("returning to a binary restores where you were", False, + "switcher did not reopen") + return + app.screen.query_one(Input).value = second + await pilot.pause(0.2) + await pilot.press("enter") + again = await settle(lambda: app._binary == second + and app._cur is not None, 120) + check("returning to a binary restores where you were", + again and app._cur.ea == where, + f"cur={app._cur.ea if app._cur else None} want={where}") + + # -- project-wide symbol search ------------------------------- # + # 'main' exists in BOTH binaries: identical name, so the rank tuple + # ties and a bare sort() would fall through to comparing Hit objects + # ('<' not supported between instances of 'Hit'). + from idatui.app import SymbolPalette + await settle(lambda: app._index is not None + and len(app._index.counts()) == 2, 60) + check("both binaries got indexed", + len(app._index.counts()) == 2, f"{app._index.counts()}") + await pilot.press("ctrl+n") + if await settle(lambda: isinstance(app.screen, SymbolPalette), 20): + pal = app.screen + pal.query_one(Input).value = "main" + await pilot.pause(0.3) + await pilot.press("f2") # widen to the whole project + await pilot.pause(0.4) + names = [(b, n) for b, _, n in pal._results] + check("project scope finds a name shared by both binaries", + len({b for b, n in names if n == "main"}) == 2, + f"{names[:6]}") + await pilot.press("escape") + await pilot.pause(0.2) + + # -- a cross-binary jump is not a one-way door ----------------- # + # Nav history is per-binary, so arriving in another binary lands you + # in an empty history. Esc must fall through to the binary you came + # from, or a project search hit (and, since phase 3, following an + # import) strands you. + here, there = app._binary, (first if app._binary == second else second) + hops0 = len(app._hops) + target = app._index.search("main", limit=200) + tgt = next((h for h in target if h.binary == there), None) + if tgt is None: + check("cross-binary jump records a hop", False, "no hit in the other binary") + else: + app._switch_then_goto(tgt.binary, tgt.addr) + jumped = await settle(lambda: app._binary == there + and app._func_index is not None + and app._func_index.complete, 180) + check("a project hit switches to the other binary", jumped, + f"binary={app._binary} want={there}") + check("the jump records where it came from", + len(app._hops) == hops0 + 1 and app._hops[-1] == here, + f"hops={app._hops}") + # spend the local history first, then Esc must cross back + for _ in range(6): + if not app._hops or app._binary != there: + break + await pilot.press("escape") + await pilot.pause(0.6) + returned = await settle(lambda: app._binary == here, 180) + check("Esc crosses back to the binary the jump came from", + returned, f"binary={app._binary} want={here} hops={app._hops}") + check("the hop is consumed, not repeated", + not app._hops, f"hops={app._hops}") + + # -- xrefs: callers in OTHER project binaries ------------------ # + # xrefs_to only sees this database, so an exported function looks + # unused from the inside even when the rest of the project calls it. + # The selection rule is "only for a symbol we actually export"; two + # executables share no linkage, so here it must stay quiet. + from idatui.app import XrefsScreen + fake = app._foreign_importers(app._cur.ea, "strrchr", None) + check("no cross-binary callers for a symbol this binary doesn't export", + fake == [], f"{fake}") + + # The routing a real cross-binary caller takes: the dialog carries a + # (binary, addr) payload instead of a bare address, and choosing it + # goes through the same switch path as a search hit — hop included, + # so Esc comes back. + where_from = app._binary + other = first if where_from == second else second + hit = next((h for h in app._index.search("main", limit=200) + if h.binary == other), None) + if hit is None: + check("a cross-binary xref jumps to the other binary", False, + "no symbol found in the other binary") + else: + hops0 = len(app._hops) + app.push_screen( + XrefsScreen("xrefs to fake", [((hit.binary, hit.addr), + f"{hit.addr:08X} import [{hit.binary}]")]), + app._on_xref_chosen) + await settle(lambda: isinstance(app.screen, XrefsScreen), 20) + await pilot.press("enter") + jumped = await settle(lambda: app._binary == other + and app._func_index is not None + and app._func_index.complete, 180) + check("a cross-binary xref jumps to the other binary", jumped, + f"binary={app._binary} want={other}") + check("and records a hop so Esc returns", + len(app._hops) == hops0 + 1 and app._hops[-1] == where_from, + f"hops={app._hops}") + app._hops.clear() + + # -- and the same toggle for strings --------------------------- # + from idatui.app import StringsPalette + await pilot.press("quotation_mark") + if await settle(lambda: isinstance(app.screen, StringsPalette), 30): + pal = app.screen + pal.query_one(Input).value = "usage" + await pilot.pause(0.3) + local = {b for b, _, _ in pal._results} + await pilot.press("f2") + await pilot.pause(0.4) + wide = {b for b, _, _ in pal._results} + check("strings: local scope is this binary only", local == {None}, + f"{local}") + check("strings: F2 widens across the project", + len(wide) >= 2 and None not in wide, f"{wide}") + await pilot.press("escape") + await pilot.pause(0.2) + + # -- the promise: nothing was written next to the sources ---------- # + left = sorted(os.listdir(src)) + check("the source tree stays pristine (no .i64/scratch beside it)", + left == sorted(os.path.basename(s) for s in srcs), f"{left}") + staged = sorted(os.listdir(proj.bin_dir)) + check("IDA's artifacts all live in the project sidecar", + any(f.endswith(".i64") for f in staged), f"{staged}") + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +def main(argv): + repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + bins = argv or [os.path.join(repo, "targets", "echo"), + os.path.join(repo, "targets", "cat")] + for b in bins: + if not os.path.isfile(b): + print(f"no such binary: {b}") + return 2 + return asyncio.run(run(bins)) + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index b16cd98..6ccbf0d 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -17,6 +17,7 @@ Uses ~/ida-venv python (has textual). --list print scenario names and exit """ import asyncio +import fnmatch import os import re import sys @@ -25,10 +26,11 @@ import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from idatui.app import ( # noqa: E402 ConfirmScreen, DecompView, DisasmView, FunctionsPanel, HexView, IdaTui, - StructEditor, SymbolPalette, XrefsScreen, + HelpScreen, ListingView, QuitScreen, StringsPalette, StructEditor, + SymbolPalette, XrefsScreen, _str_display, _word_occurrences, ) from textual.widgets import ( # noqa: E402 - DataTable, Footer, Input, OptionList, Static, TextArea, + DataTable, Input, OptionList, Static, TextArea, ) from rich.text import Text # noqa: E402 from idatui._sync import wait_for # noqa: E402 @@ -93,13 +95,17 @@ class Ctx: # -- shorthands ------------------------------------------------------- # @property def dis(self): - return self.app.query_one(DisasmView) + return self.app.query_one(ListingView) # unified code view (was DisasmView) @property def dec(self): return self.app.query_one(DecompView) @property + def lst(self): + return self.app.query_one(ListingView) + + @property def hex(self): return self.app.query_one(HexView) @@ -113,7 +119,10 @@ class Ctx: # -- discovery -------------------------------------------------------- # def all_funcs(self): - return self.prog.functions().all_loaded() + idx = self.prog.functions() + if len(idx) == 0 and not idx.complete: + idx.load_all() # a prior bump_items() cleared the index cache + return idx.all_loaded() def find_func(self, pred, limit=400): for f in self.all_funcs()[:limit]: @@ -148,23 +157,36 @@ class Ctx: return None # -- navigation (setup; fast internal path) --------------------------- # - async def open(self, target, view="decomp", t=25): - """Open a function (name or ea) to its entry, in ``view`` ('decomp' or - 'disasm'). Returns the Func.""" + async def open(self, target, view="listing", t=25): + """Open a function (name or ea) at its entry in the unified listing. With + ``view='decomp'`` it then F5s into the pseudocode. Returns the Func.""" ea = self.prog.resolve(target) if isinstance(target, str) else int(target) fn = self.prog.function_of(ea) if fn is None: raise RuntimeError(f"no function for {target!r}") - self.app._pref = view self.app._open_function(fn.addr, fn.name) await self.wait(lambda: self.app._cur and self.app._cur.ea == fn.addr, t) + # Wait for the cursor to actually LAND on the target, not merely for + # _cur.ea to match: re-opening a function we already navigated to (its + # _cur.ea is stale-true) schedules an async re-navigation, and proceeding + # before it lands would leave the cursor parked wherever we last were. + await self.wait(lambda: self.lst.total > 0 + and self.lst._cursor_ea() == fn.addr, t) if view == "decomp": - await self.wait(lambda: self.dec.loaded_ea == fn.addr, t) - else: - await self.wait(lambda: self.dis.total > 0, t) + # F5/Tab only decompiles from a focused code pane, and the listing + # may still be settling from the open above — a swallowed Tab used to + # surface much later as "pseudocode view shows: active=listing". + # Retry rather than assume the first one takes. + for _ in range(3): + self.lst.focus() + await self.pause(0.05) + await self.press("tab") + if await self.wait(lambda: self.app._active == "decomp" + and self.dec.loaded_ea == fn.addr, max(t / 3, 5)): + break return fn - async def open_biggest(self, view="disasm"): + async def open_biggest(self, view="listing"): return await self.open(self.biggest().addr, view) async def goto_ui(self, target): @@ -184,8 +206,10 @@ class Ctx: async def boot(self): app = self.app await self.wait(lambda: self.table.row_count > 0, 60) + # Load is done when the index is complete (robust vs the status line, + # which startup auto-land immediately overwrites with the landed fn). await self.wait( - lambda: "functions" in self.status() and "…" not in self.status(), 60) + lambda: app._func_index is not None and app._func_index.complete, 60) if app._func_index is not None and not app._func_index.complete: app._func_index.load_all() @@ -212,6 +236,7 @@ class Ctx: app._pref = "decomp" if app._active == "hex": app._active = "decomp" + app._split = False await self.pause(0.02) @@ -227,11 +252,42 @@ async def s_startup(c: Ctx): await c.reveal_pane() c.check("function pane width is capped (doesn't eat the screen)", left.size.width <= 44, f"width={left.size.width}") - c.check("function load completed (status settled)", - "functions" in c.status() and "…" not in c.status(), c.status()) + c.check("function load completed (index complete)", + c.app._func_index is not None and c.app._func_index.complete, c.status()) print(f" {c.table.row_count} functions loaded") +@scenario("auto_land") +async def s_auto_land(c: Ctx): + """Startup lands on main()/an entry function when present, else pops the + symbol picker — never a blank pane.""" + app = c.app + # simulate a fresh startup landing + for _ in range(3): + if len(app.screen_stack) > 1: + app.pop_screen() + await c.pause(0.05) + app._cur = None + app._did_auto_land = False + fn = app._entry_func() + app._auto_land() + await c.pause(0.2) + if fn is not None: + await c.wait(lambda: app._cur is not None and app._cur.ea == fn.addr, 20) + c.check("auto-land jumps to the entry function (main) when present", + app._cur is not None and app._cur.ea == fn.addr, + f"entry={fn.name}@{fn.addr:#x} cur={app._cur}") + else: + await c.wait(lambda: isinstance(app.screen, SymbolPalette), 10) + c.check("auto-land pops the symbol picker when there's no entry fn", + isinstance(app.screen, SymbolPalette), f"screen={app.screen}") + app.pop_screen() + # guard fires once: a second call is a no-op + prev = app._cur + app._auto_land() + c.check("auto-land is idempotent (guarded)", app._cur is prev) + + @scenario("palette") async def s_palette(c: Ctx): app, pilot = c.app, c.pilot @@ -244,24 +300,48 @@ async def s_palette(c: Ctx): pal = app.screen pinp = pal.query_one(Input) pinp.value = "main" - await c.wait(lambda: pal._results and pal._results[0].name == "main", 10) + await c.wait(lambda: pal._results and pal._results[0][2] == "main", 10) c.check("palette fuzzy-finds (top result matches the query)", - bool(pal._results) and pal._results[0].name == "main", - f"top={pal._results[0].name if pal._results else None}") + bool(pal._results) and pal._results[0][2] == "main", + f"top={pal._results[0][2] if pal._results else None}") pinp.value = "eror" # scattered subsequence of 'error' - await c.wait(lambda: any(f.name == "error" for f in pal._results), 10) + await c.wait(lambda: any(n == "error" for _, _, n in pal._results), 10) c.check("palette matches a fuzzy subsequence", - any(f.name == "error" for f in pal._results), - f"results={[f.name for f in pal._results[:4]]}") + any(n == "error" for _, _, n in pal._results), + f"results={[n for _, _, n in pal._results[:4]]}") + # Every other query here is lowercase, which is how a case bug hid for so + # long: the name was lowered but the query wasn't, so ONE capital matched + # nothing. Invisible on lowercase C symbols, fatal on a library that + # capitalises (PEM_read_bio found 0 of 10093 functions in libcrypto). + pinp.value = "MAIN" + await c.wait(lambda: any(n == "main" for _, _, n in pal._results), 10) + c.check("palette matching is case-insensitive in BOTH directions", + any(n == "main" for _, _, n in pal._results), + f"results={[n for _, _, n in pal._results[:4]]}") pinp.value = "main" - await c.wait(lambda: pal._results and pal._results[0].name == "main", 10) - want = pal._results[0].addr + await c.wait(lambda: pal._results and pal._results[0][2] == "main", 10) + want = pal._results[0][1] await c.press("enter") await c.wait(lambda: not isinstance(app.screen, SymbolPalette), 10) await c.wait(lambda: app._cur and app._cur.ea == want, 20) c.check("selecting a palette entry opens that function", bool(app._cur) and app._cur.ea == want, f"cur={app._cur.ea if app._cur else None}") + # Re-open the function we are ALREADY standing on. That used to append an + # identical nav entry, and the extra Esc it bought popped the stack without + # changing anything on screen — a dead keypress, which is precisely what + # "back is broken" feels like from the keyboard. + depth = len(app._nav) + await c.press("ctrl+n") + await c.wait(lambda: isinstance(app.screen, SymbolPalette), 10) + pal2 = app.screen + pal2.query_one(Input).value = "main" + await c.wait(lambda: pal2._results and pal2._results[0][2] == "main", 10) + await c.press("enter") + await c.wait(lambda: not isinstance(app.screen, SymbolPalette), 10) + await c.pause(0.4) + c.check("re-opening the current function doesn't stack a duplicate", + len(app._nav) == depth, f"nav {depth} -> {len(app._nav)}") await c.press("ctrl+n") await c.wait(lambda: isinstance(app.screen, SymbolPalette), 10) await c.press("escape") @@ -269,29 +349,351 @@ async def s_palette(c: Ctx): c.check("Esc closes the palette", not isinstance(app.screen, SymbolPalette)) +@scenario("load_options") +async def s_load_options(c: Ctx): + """The dialog must never appear for a file IDA can load itself — the whole + suite runs on an ELF, so a false positive here would block every run.""" + from idatui.app import LoadOptionsScreen + app = c.app + c.check("no load dialog for a recognised binary", + not isinstance(app.screen, LoadOptionsScreen), + f"screen={type(app.screen).__name__}") + c.check("and the app agrees it shouldn't ask", + not app._should_ask_load_options()) + # The dialog itself, driven directly: it has to come back with switches the + # worker can use, and -b has to be paragraphs. + from idatui.formats import load_args, needs_load_options, sniff + c.check("the running target sniffs as a real format", + sniff(app._open_path) is not None and not needs_load_options(app._open_path), + f"{sniff(app._open_path)}") + c.check("dialog output converts a base to paragraphs", + load_args("arm", 0x8000000) == "-parm -b800000") + + # Tab is a PRIORITY app binding (disasm<->pseudocode), so it fired even with + # a modal up and nothing in a dialog could be tabbed to. That is why the load + # dialog's address field was unreachable — and it was broken in every other + # modal too. + from idatui.app import LoadOptionsScreen + app.push_screen(LoadOptionsScreen("/tmp/probe.bin", 1234)) + await c.wait(lambda: isinstance(app.screen, LoadOptionsScreen), 10) + sc = app.screen + first = app.focused + await c.press("tab") + await c.pause(0.2) + c.check("Tab moves focus inside a modal instead of toggling the view", + app.focused is not first and isinstance(app.screen, LoadOptionsScreen), + f"focus={getattr(app.focused, 'id', None)}") + c.check("Tab in the load dialog lands on the address field", + getattr(app.focused, "id", None) == "load-base", + f"focus={getattr(app.focused, 'id', None)}") + await c.press("tab") + await c.pause(0.2) + c.check("Tab again returns to the processor filter", + getattr(app.focused, "id", None) == "pal-input", + f"focus={getattr(app.focused, 'id', None)}") + await c.press("escape") + await c.wait(lambda: not isinstance(app.screen, LoadOptionsScreen), 10) + + +@scenario("command_palette") +async def s_command_palette(c: Ctx): + app = c.app + await c.open_biggest("listing") + await c.press("ctrl+p") + opened = await c.wait( + lambda: type(app.screen).__name__ == "CommandPalette", 10) + c.check("Ctrl+P opens the command palette", opened, + f"screen={type(app.screen).__name__}") + if not opened: + return + inp = app.screen.query_one(Input) + inp.value = "hex" # filter to the 'Hex view' command + await c.pause(0.5) # let the async search + option list settle + await c.press("enter") + landed = await c.wait(lambda: app._active == "hex", 10) + c.check("a palette command executes (Hex view opens)", landed, + f"active={app._active}") + if landed: + await c.press("backslash") # leave hex + await c.wait(lambda: app._active != "hex", 5) + + +@scenario("quit_guard") +async def s_quit_guard(c: Ctx): + app = c.app + c.check("a clean database reports nothing unsaved", app._dirty_labels() == [], + f"{app._dirty_labels()}") + app._dirty = True # as an edit would + c.check("an edited database is reported unsaved", + len(app._dirty_labels()) == 1, f"{app._dirty_labels()}") + await c.press("q") + asked = await c.wait(lambda: isinstance(app.screen, QuitScreen), 10) + c.check("quitting with unsaved changes asks first", asked and app.is_running, + f"screen={type(app.screen).__name__} running={app.is_running}") + if not asked: + return + await c.press("escape") + await c.wait(lambda: not isinstance(app.screen, QuitScreen), 10) + c.check("Esc cancels the quit and stays put", + app.is_running and not isinstance(app.screen, QuitScreen)) + # leave it clean so the rest of the suite (and teardown) isn't affected + app._dirty = False + app._save_on_exit = False + + +@scenario("help") +async def s_help(c: Ctx): + app = c.app + st = app.query_one("#status", Static) + c.check("the status line owns the bottom row (no footer cheatsheet)", + st.region.y + st.region.height == app.size.height, + f"status={st.region} screen={app.size}") + await c.press("f1") + opened = await c.wait(lambda: isinstance(app.screen, HelpScreen), 10) + c.check("F1 opens the key cheatsheet", opened, + f"screen={type(app.screen).__name__}") + if not opened: + return + cards = app.screen.query(".help-card") + titles = {str(w.border_title) for w in cards} + txt = " ".join(str(w.render()) for w in cards) + c.check("each key group gets its own card", + titles == {"Navigate", "Views", "Move", "Edit", "Search"}, f"{titles}") + c.check("it documents real bindings", + "set type" in txt and "split view" in txt and "cross-references" in txt) + body = app.screen.query_one("#help-body") + c.check("the cards fit without a scrollbar at a normal size", + body.virtual_size.height <= body.size.height, + f"content={body.virtual_size.height} view={body.size.height}") + await c.press("escape") + await c.wait(lambda: not isinstance(app.screen, HelpScreen), 10) + c.check("Esc closes it", not isinstance(app.screen, HelpScreen)) + + +@scenario("strings") +async def s_strings(c: Ctx): + app = c.app + await c.open_biggest("listing") + items = app.program.strings() + c.check("program.strings() lists the binary's literals", len(items) > 3, + f"n={len(items)}") + if not items: + return + c.check("strings carry addr/text/length", + all(s.addr > 0 and s.text and s.length > 0 for s in items[:5]), + f"first={items[0]}") + await c.press("quotation_mark") + opened = await c.wait(lambda: isinstance(app.screen, StringsPalette), 25) + c.check('\'"\' opens the strings browser', opened, + f"screen={type(app.screen).__name__}") + if not opened: + return + pal = app.screen + c.check("the browser lists strings", len(pal._results) > 0, + f"results={len(pal._results)}") + # filter on a fragment of a real (unescaped) literal + target = next((s for s in items + if len(s.text) >= 6 and _str_display(s.text) == s.text), None) + if target is not None: + frag = target.text[:6] + pal.query_one(Input).value = frag + await c.pause(0.2) + ok = (pal._results + and all(frag.lower() in t.lower() for _, _, t in pal._results)) + c.check("filtering narrows to matching strings", bool(ok), + f"frag={frag!r} n={len(pal._results)}") + want = pal._results[0][1] + await c.press("enter") + await c.wait(lambda: not isinstance(app.screen, StringsPalette), 10) + landed = await c.wait(lambda: c.lst._cursor_ea() == want, 20) + c.check("Enter jumps to the string in the unified listing", landed, + f"cursor={c.lst._cursor_ea()} want={want:#x}") + else: + await c.press("escape") + + +@scenario("split_view") +async def s_split_view(c: Ctx): + app, lst, dec = c.app, c.lst, c.dec + await c.open_biggest("listing") + await c.press("s") + shown = await c.wait(lambda: app._split and lst.display and dec.display, 20) + c.check("'s' enters split view (both panes shown)", shown, + f"split={app._split} lst={lst.display} dec={dec.display}") + loaded = await c.wait(lambda: dec.loaded_ea == app._cur.ea, 25) + c.check("split loads the pseudocode alongside the listing", loaded, + f"loaded={dec.loaded_ea} cur={app._cur.ea if app._cur else None}") + await c.wait(lambda: "[split" in c.status(), 5) + c.check("split view shows a split-aware status", "[split" in c.status(), + f"status={c.status()!r}") + # phase 3: the rich per-line instruction map (decomp_map tool, run on the + # pilot's real worker) — verify it returns, aligns with the markers, and + # bands a whole region for a multi-instruction C line. + m = app.program.decomp_map(app._cur.ea) + c.check("decomp_map returns per-line ea sets", len(m) > 5, f"lines={len(m)}") + aligned = sum(1 for i in range(min(len(m), len(dec._line_eas))) + if m[i] and dec._line_eas[i] is not None + and dec._line_eas[i] in m[i]) + c.check("decomp_map aligns with the pseudocode markers", aligned >= 3, + f"aligned={aligned}/{len(dec._line_eas)}") + multi = next((i for i, eas in enumerate(m) if len(eas) > 1), None) + if multi is not None: + app._split_eamap = m + dec.focus() # the decomp must BE the driver for a decomp-driven + app._active = "decomp" # sync (else its align() re-syncs listing-driven) + dec.cursor = multi + dec._scroll_cursor_into_view() # key-nav always does; the anchor needs it + await c.pause(0.1) + app._sync_split("decomp") + await c.pause(0.1) + c.check("a multi-instruction C line bands a region (>1 listing row)", + len(lst._link_rows) > 1, + f"line={multi} eas={len(m[multi])} rows={sorted(lst._link_rows)[:8]}") + lst.focus() + app._active = "listing" + await c.pause(0.05) + else: + c.check("a multi-instruction C line bands a region (>1 listing row)", + True, "no multi-instruction line in this function (skipped)") + # listing drives: move it, the decomp band must track the covering C line + lst.focus() + for _ in range(6): + await c.press("j") + await c.pause(0.2) + lea = lst._cursor_ea() + dl = dec._link_line + c.check("listing cursor links the covering pseudocode line", + dl is not None and lea is not None and dec._line_eas[dl] is not None + and dec._line_eas[dl] <= lea, + f"link_line={dl} lea={hex(lea) if lea else None}") + # the companion pane sits LEVEL with the driver's cursor (visual coherence): + # the linked row lands at the same viewport offset, not merely on-screen. + deep = [i for i, eas in enumerate(m) if eas][10:] + row = lst.model.ensure_ea(m[deep[0]][0]) if (deep and lst.model) else None + if row is not None and row > 12: + lst.scroll_to(y=row - 10, animate=False) + await c.pause(0.2) # let the deferred scroll land + lst.cursor = row # driver cursor now at viewport offset 10 + app._sync_split("listing") + await c.pause(0.2) # let the companion's scroll land + drv = lst.cursor - round(lst.scroll_offset.y) + link, top = dec._link_line, round(dec.scroll_offset.y) + # exact, modulo the unavoidable clamps (can't scroll above line 0, nor + # past the end when the pseudocode is shorter than the viewport) + want = min(max(0, (link or 0) - drv), + max(0, dec.total - dec._visible_height())) + c.check("the companion pane sits level with the driver's cursor", + link is not None and top == want, + f"driver_row={drv} link={link} dec_top={top} want={want}") + # a PURE scroll (wheel/scrollbar) moves no cursor — it must still drag + # the companion along (anchors on the viewport once the cursor is gone) + before_cur, before_dec = lst.cursor, round(dec.scroll_offset.y) + lst.scroll_to(y=round(lst.scroll_offset.y) + 30, animate=False) + await c.pause(0.35) + c.check("a pure scroll in the driver drags the companion along", + lst.cursor == before_cur + and round(dec.scroll_offset.y) != before_dec, + f"cursor {before_cur}->{lst.cursor} " + f"dec_top {before_dec}->{round(dec.scroll_offset.y)}") + await c.press("tab") + await c.pause(0.1) + c.check("Tab in split focuses the pseudocode pane", app._active == "decomp", + f"active={app._active}") + # decomp drives: put the cursor on an addressed pseudocode line (past the + # variable decls); the listing band must track the covering instruction row. + target = next((i for i, e in enumerate(dec._line_eas) if e is not None), None) + c.check("pseudocode has addressed lines", target is not None, + "no /*0xEA*/ markers in the pseudocode") + if target is not None: + dec.cursor = target + dec._scroll_cursor_into_view() + await c.pause(0.1) + app._sync_split("decomp") + await c.pause(0.1) + want = lst.model.ensure_ea(dec._line_eas[target]) + c.check("decomp cursor links the instruction row in the listing", + want in lst._link_rows, + f"link_rows={sorted(lst._link_rows)[:6]} want={want}") + # and that linked row actually paints a background band (base rows have + # no bg; `want` is a deep code row, never the listing's own cursor row) + lst.reveal(want) + await c.pause(0.05) + y = want - round(lst.scroll_offset.y) + banded = (0 <= y < lst.size.height and any( + s.style and s.style.bgcolor is not None for s in lst.render_line(y))) + c.check("the linked instruction row renders a highlight band", banded, + f"y={y} cursor_row={lst.cursor}") + await c.press("tab") + await c.pause(0.1) + c.check("Tab again focuses the listing pane", app._active == "listing", + f"active={app._active}") + # a mouse click on the other pane also makes it the driver (not just Tab) + await c.pilot.click(DecompView, offset=(10, 5)) + await c.pause(0.15) + c.check("clicking the pseudocode pane makes it the driver", + app._active == "decomp", f"active={app._active}") + await c.press("tab") # restore listing as the driver + await c.pause(0.1) + # cross-function follow: the listing cursor leaving the decompiled function + # re-points the decomp pane to whatever function it's now in. + await c.wait(lambda: app._split_range is not None, 10) + other = c.find_func(lambda f: f.addr != app._cur.ea and f.size > 40) + row = lst.model.ensure_ea(other.addr) if other is not None else None + if other is not None and row is not None and row >= 0: + lst.focus() + app._active = "listing" + lst.cursor = row + lst._scroll_cursor_into_view() + await c.pause(0.1) + app._sync_split("listing") # cursor now outside the decompiled fn + followed = await c.wait(lambda: dec.loaded_ea == other.addr, 25) + c.check("listing cursor crossing into another function re-syncs the decomp", + followed, f"dec={dec.loaded_ea} want={other.addr}") + # navigation in split keeps BOTH panes on the (new) function + nf = c.find_func(lambda f: f.addr != app._cur.ea and f.size > 80) + if nf is not None: + await c.press("g") + await c.type(hex(nf.addr)) + await c.press("enter") + nav = await c.wait(lambda: app._cur and app._cur.ea == nf.addr, 15) + c.check("goto in split navigates", nav, + f"cur={app._cur.ea if app._cur else None} want={nf.addr}") + both = await c.wait(lambda: dec.loaded_ea == nf.addr and app._split + and lst.display and dec.display, 25) + c.check("split reloads both panes on navigation", both, + f"dec={dec.loaded_ea} split={app._split}") + await c.press("s") + gone = await c.wait(lambda: not app._split and lst.display + and not dec.display, 10) + c.check("'s' exits split back to a single view", gone, + f"split={app._split} lst={lst.display} dec={dec.display}") + c.check("exiting split clears the link bands", + not lst._link_rows and dec._link_line is None, + f"rows={lst._link_rows} line={dec._link_line}") + + @scenario("decomp_fallback") async def s_fallback(c: Ctx): app = c.app - d_view, c_view = c.dis, c.dec failing = next((f for f in reversed(c.all_funcs()) if c.prog.decompile(f.addr).failed), None) if failing is None: c.check("found a decompile-failing function", False) return - app._pref = "decomp" - app._goto(hex(failing.addr)) - await c.wait(lambda: app._cur and app._cur.ea == failing.addr, 20) - await c.wait(lambda: app._active == "disasm", 20) - c.check("decompile failure falls back to disassembly (preference kept)", - app._active == "disasm" and d_view.display and not c_view.display - and app._pref == "decomp", - f"active={app._active} pref={app._pref} " - f"disp(dis={d_view.display},dec={c_view.display})") - app._goto("main") - await c.wait(lambda: app._cur and app._cur.name == "main", 20) - await c.wait(lambda: app._active == "decomp" and c_view.display, 25) - c.check("preference preserved: next function opens as pseudocode", - app._active == "decomp" and c_view.display, f"active={app._active}") + # Open the failing function in the listing, then F5: decompile fails -> the + # view stays on the linear listing (no error panel). + await c.open(failing.addr, "listing") + c.dis.focus() + await c.press("tab") + await c.wait(lambda: app._active == "listing" + and "fail" in c.status().lower(), 25) + c.check("F5/Tab on an undecompilable function falls back to the listing", + app._active == "listing" and c.dis.display, + f"active={app._active} status={c.status()!r}") + # a decompilable function F5s into pseudocode + await c.open("main", "decomp") + c.check("a decompilable function F5s into pseudocode", + app._active == "decomp" and c.dec.display, f"active={app._active}") @scenario("structs") @@ -385,21 +787,19 @@ async def s_structs(c: Ctx): async def s_open(c: Ctx): app = c.app fn = c.biggest() - app._pref = "decomp" app._open_function(fn.addr, fn.name) await c.wait(lambda: app._cur and app._cur.ea == fn.addr, 20) - await c.wait(lambda: c.dis.total > 0, 30) - c.check("disasm model loads with a total", c.dis.total > 0, f"total={c.dis.total}") - print(f" biggest = {fn.name} ({c.dis.total} instrs)") - pc = await c.wait(lambda: app._active == "decomp" and c.dec.display - and c.dec.loaded_ea == fn.addr, 25) - c.check("opening a function shows pseudocode by default", pc, f"active={app._active}") + await c.wait(lambda: c.lst.total > 0, 30) + c.check("opening a function shows the linear listing by default", + app._active == "listing" and c.lst.display and c.lst.total > 0, + f"active={app._active} total={c.lst.total}") + print(f" biggest = {fn.name} ({c.lst.total} listing rows)") @scenario("disasm_nav") async def s_disasm_nav(c: Ctx): app, view = c.app, c.dis - await c.open_biggest("disasm") + await c.open_biggest("listing") view.focus() c.check("first instruction cached", view.model is not None and view.model.cached_line(0) is not None) @@ -419,8 +819,9 @@ async def s_disasm_nav(c: Ctx): await c.wait(lambda: app._clipboard == cur_line, 10) c.check("Ctrl+Y copies the current code line to the clipboard", bool(cur_line) and app._clipboard == cur_line, f"clip={app._clipboard!r}") - # goto-bottom must not hang on a huge function. - await c.press("end") + # goto-bottom must not hang on a huge function (ctrl+end; plain 'end' now + # moves the cursor to end-of-line). + await c.press("ctrl+end") await c.pause(0.05) c.check("goto-bottom lands near end", view.cursor >= view.total - 1, f"cursor={view.cursor}/{view.total}") @@ -429,7 +830,7 @@ async def s_disasm_nav(c: Ctx): @scenario("hex") async def s_hex(c: Ctx): app, view = c.app, c.dis - await c.open_biggest("disasm") + await c.open_biggest("listing") view.focus() view.cursor = 4 # a non-entry instruction await c.pause(0.05) @@ -463,10 +864,38 @@ async def s_hex(c: Ctx): await c.wait(lambda: hx.cursor_va() == target_va, 15) c.check("'g' in the hex view jumps the cursor to an address", hx.cursor_va() == target_va, f"va={hx.cursor_va():#x} want={target_va:#x}") + # -- a user scroll freezes the cursor's screen row (points at a new byte) -- + await c.press("g") + await c.type(hex(rng[0])) + await c.press("enter") + await c.wait(lambda: hx.cursor_va() == rng[0], 10) + for _ in range(8): # cursor to viewport row 8 (top still 0) + await c.press("j") + await c.pause(0.1) + top0 = round(hx.scroll_offset.y) + screen_row = hx.cursor // 16 - top0 + hx.scroll_to(y=top0 + 30, animate=False) # user scroll down 30 rows + await c.pause(0.15) + top1 = round(hx.scroll_offset.y) + c.check("hex viewport scrolled", top1 >= top0 + 20, f"top0={top0} top1={top1}") + c.check("hex cursor's screen row stays frozen on scroll", + hx.cursor // 16 - top1 == screen_row, + f"screen_row={screen_row} now={hx.cursor // 16 - top1} top1={top1}") + PAD = 1 # HexView { padding: 0 1 } -> content is inset one col + await c.pilot.click(HexView, offset=(PAD + 19 + 3 * 3, 5)) # hex byte 3, row 5 + await c.pause(0.1) + c.check("clicking the hex pane moves the cursor to the clicked byte", + hx.cursor == (top1 + 5) * 16 + 3, + f"cursor={hx.cursor} want={(top1 + 5) * 16 + 3} top1={top1}") + await c.pilot.click(HexView, offset=(PAD + 70 + 10, 7)) # ascii byte 10, row 7 + await c.pause(0.1) + c.check("clicking the ascii pane maps to the right byte", + hx.cursor == (top1 + 7) * 16 + 10, + f"cursor={hx.cursor} want={(top1 + 7) * 16 + 10}") await c.press("backslash") await c.wait(lambda: app._active != "hex", 10) c.check("backslash returns from hex to the code view", - app._active == "disasm", f"active={app._active}") + app._active == "listing", f"active={app._active}") @scenario("filter") @@ -484,21 +913,31 @@ async def s_filter(c: Ctx): await c.wait(lambda: app._cur and app._cur.ea == fn.addr, 20) c.check("selecting a table row opens that function", bool(app._cur) and app._cur.ea == fn.addr, f"cur={app._cur.ea if app._cur else None}") - # filter round-trip + # filter round-trip. Derive the glob from real names: this used to hardcode + # 'sub_1*', which matches NOTHING in a binary whose code never reaches + # 0x1xxx (echo's functions are sub_2xxx..sub_7xxx) — a deterministic failure + # that looked like a flake, and left the table empty for the next scenario. + subs = sorted(f.name for f in c.all_funcs() if f.name.startswith("sub_")) + term = (subs[0][:5] + "*") if subs else "" + want = sum(1 for f in c.all_funcs() + if fnmatch.fnmatch(f.name.lower(), term.lower())) if term else 0 + c.check("picked a glob that actually matches (test self-check)", + 0 < want < nfuncs, f"term={term!r} want={want} of {nfuncs}") table.focus() await c.press("slash") await c.pause(0.05) - for ch in "sub_1*": + for ch in term: await c.press(ch if ch != "*" else "asterisk") await c.press("enter") - filtered = await c.wait(lambda: 0 < table.row_count < nfuncs, 15) - c.check("filter narrowed the list", filtered, f"rows={table.row_count} of {nfuncs}") + filtered = await c.wait(lambda: table.row_count == want, 15) + c.check("filter narrowed the list to exactly the matches", filtered, + f"term={term!r} rows={table.row_count} want={want} of {nfuncs}") # pane toggle left = app.query_one("#left", FunctionsPanel) await c.press("ctrl+b") await c.pause(0.05) c.check("ctrl+b hides functions pane + focuses the code view", - not left.display and isinstance(app.focused, (DisasmView, DecompView)), + not left.display and isinstance(app.focused, (ListingView, DecompView)), f"display={left.display} focus={type(app.focused).__name__}") await c.press("ctrl+b") await c.pause(0.05) @@ -518,24 +957,62 @@ async def s_view_toggle(c: Ctx): styled = any(seg.style is not None and seg.style.color is not None for strip in dec._strips[:min(dec.total, 200)] for seg in strip) c.check("pseudocode is syntax-highlighted", styled) + # Cancelling a prompt must hand focus back to the pane you were READING. + # _code_view() used to choose on _pref, which was only ever "listing", so it + # focused the hidden listing and the pseudocode stopped answering the + # keyboard — arrows did nothing at all until you clicked. + dec.focus() + await c.pause(0.05) + line0 = dec.cursor + await c.press("g") + await c.wait(lambda: app.query_one("#goto", Input).display, 10) + await c.press("escape") + await c.wait(lambda: not app.query_one("#goto", Input).display, 10) + await c.press("down") + await c.press("down") + await c.pause(0.2) + c.check("cancelling goto leaves focus in the pseudocode (arrows still work)", + dec.cursor > line0, + f"cursor {line0} -> {dec.cursor} focus={type(app.focused).__name__}") + dec.focus() # Tab only toggles the view from a code pane; elsewhere it's + await c.pause(0.05) # focus-next, which would silently leave us in decomp await c.press("tab") - await c.pause(0.1) - c.check("tab switches to disassembly", - app._active == "disasm" and dis.display and not dec.display, - f"active={app._active}") - await c.press("tab") - await c.wait(lambda: app._active == "decomp", 10) - # a (re)decompile shows the grayed overlay + # decomp -> listing runs through _toggle_to_listing, a background worker, so + # _active only flips once the listing model has loaded. A fixed pause held in + # a short run and lost the race in a full one. + switched = await c.wait(lambda: app._active == "listing" and dis.display + and not dec.display, 20) + c.check("tab switches to disassembly", switched, + f"active={app._active} split={app._split} " + f"focus={type(app.focused).__name__} " + f"lst={dis.display} dec={dec.display}") + # _toggle_to_listing repositions asynchronously; F5 below reads the listing + # cursor's ea and no-ops if it isn't on an addressed row yet. + await c.wait(lambda: c.lst._cursor_ea() is not None, 10) + # F5/Tab from the listing must raise the 'decompiling…' overlay synchronously, + # BEFORE the background decompile runs (regression: it used to decompile first + # in _decomp_from_listing, so the overlay only flashed once cached). dec.loaded_ea = None - app._show_active() + app.action_toggle_view() cover = dec._cover_widget - c.check("decompile shows a 'decompiling…' overlay", + c.check("F5 from the listing raises the 'decompiling…' overlay", dec.loading and cover is not None and "decomp-loading" in cover.classes and "decompiling" in str(cover.render()), f"loading={dec.loading} cover={cover!r}") - await c.wait(lambda: not dec.loading and dec._cover_widget is None, 25) + await c.wait(lambda: app._active == "decomp" and not dec.loading + and dec._cover_widget is None, 25) c.check("overlay clears when the decompile finishes", not dec.loading and dec._cover_widget is None) + # F5 on an ALREADY-loaded function must still clear the overlay (regression: + # the F5-raised overlay had nothing to clear it in the 'already loaded' branch + # -> spinner stuck forever). + await c.press("tab") # -> listing + await c.wait(lambda: app._active == "listing", 10) + app.action_toggle_view() # F5 the same, cached function again + cleared = await c.wait(lambda: app._active == "decomp" and not dec.loading + and dec._cover_widget is None, 15) + c.check("re-decompiling an already-loaded function clears the overlay", cleared, + f"loading={dec.loading} cover={dec._cover_widget!r}") # line-number gutter dec.scroll_to(0, 0, animate=False) await c.pause(0.025) @@ -543,27 +1020,65 @@ async def s_view_toggle(c: Ctx): c.check("pseudocode has a numbered gutter (line 1 first)", dec._gutter > 0 and row0[:dec._gutter].strip() == "1", f"gutter={dec._gutter} row0={row0[:10]!r}") + # Home/End move along the line here too (they used to scroll to top/bottom). + line = next((i for i, t in enumerate(dec._texts) + if t.startswith(" ") and t.strip()), None) + if line is not None: + text = dec._texts[line] + dec.focus() + dec.cursor, dec.cursor_x = line, 0 + dec.refresh() + await c.pause(0.05) + top = round(dec.scroll_offset.y) + await c.press("end") + await c.pause(0.05) + c.check("<end> in pseudocode goes to end-of-line, not the bottom", + dec.cursor == line and dec.cursor_x == max(len(text) - 1, 0) + and round(dec.scroll_offset.y) == top, + f"line={dec.cursor} col={dec.cursor_x} len={len(text)}") + await c.press("home") + await c.pause(0.05) + c.check("<home> in pseudocode goes to start-of-line", + dec.cursor == line and dec.cursor_x == 0, + f"line={dec.cursor} col={dec.cursor_x}") + await c.press("shift+home") + await c.pause(0.05) + c.check("<shift+home> skips the indentation", + dec.cursor_x == len(text) - len(text.lstrip()), + f"col={dec.cursor_x} indent={len(text) - len(text.lstrip())}") + await c.press("ctrl+end") + await c.pause(0.1) + c.check("<ctrl+end> still goes to the bottom", + dec.cursor >= dec.total - 1, f"{dec.cursor}/{dec.total}") + await c.press("ctrl+home") + await c.pause(0.1) + c.check("<ctrl+home> still goes to the top", dec.cursor == 0, + f"{dec.cursor}") @scenario("search") async def s_search(c: Ctx): app, dis = c.app, c.dis - await c.open_biggest("disasm") + await c.open_biggest("listing") dis.focus() - line0 = dis.model.cached_line(0) + # derive the term from the instruction at the cursor (the function entry), + # not segment head 0 (which may still be streaming in) + line0 = dis.model.cached_line(dis.cursor) raw = (line0.text.split() or ["push"])[0] if line0 else "push" term = "".join(ch for ch in raw if ch.isalnum())[:4] or "push" await c.press("slash") await c.pause(0.1) await c.type(term) await c.press("enter") - await c.wait(lambda: bool(dis._matches), 25) + # the unified listing searches the whole segment (load_all) -> allow time + await c.wait(lambda: bool(dis._matches), 45) c.check("search finds matches", len(dis._matches) > 0, f"term={term!r}") c.check("cursor sits on a match", dis.cursor in dis._matches, f"cursor={dis.cursor}") c.check("match substring highlighted", bool(dis._ranges.get(dis.cursor)), str(dis._ranges.get(dis.cursor))) c.check("search cursor lands on the match's starting column", - dis.cursor_x == dis._ranges[dis.cursor][0][0], + bool(dis._ranges.get(dis.cursor)) + and dis.cursor_x == dis._ranges[dis.cursor][0][0], f"cursor_x={dis.cursor_x} ranges={dis._ranges.get(dis.cursor)}") prev = dis.cursor await c.press("slash") @@ -585,10 +1100,10 @@ async def s_search(c: Ctx): await c.pause(0.1) c.check("search bar visible, status hidden (no overlap)", si.display and not status.display, f"si={si.display} status={status.display}") - footer = app.query_one(Footer) - c.check("search input is rendered above the footer (not overlapping)", - si.region.height >= 1 and si.region.y < footer.region.y, - f"search={si.region} footer={footer.region}") + c.check("search input owns the bottom row (nothing overlaps it)", + si.region.height >= 1 + and si.region.y + si.region.height == app.size.height, + f"search={si.region} screen={app.size}") for ch in term: await c.press(ch) await c.pause(0.05) @@ -631,7 +1146,7 @@ async def s_incr_filter(c: Ctx): @scenario("follow_xrefs") async def s_follow_xrefs(c: Ctx): app, dis, dec = c.app, c.dis, c.dec - await c.open_biggest("disasm") + await c.open_biggest("listing") dis.focus() lines = dis.model.lines(0, 400, prefetch=False) call_idx = next((i for i, ln in enumerate(lines) @@ -673,71 +1188,30 @@ async def s_follow_xrefs(c: Ctx): c.check("found a function with a code xref", False) else: xfn, xref = xf - xexp = app.program.disasm(xref.fn_addr).index_of_ea(xref.frm) - await c.press("g") - await c.pause(0.1) - await c.type(xfn.name) - await c.press("enter") - await c.wait(lambda: dis.total > 0 and app._cur.ea == xfn.addr, 20) - dis.focus() - dis.cursor, dis.cursor_x = 0, 0 + await c.goto_ui(xfn.name) + await c.wait(lambda: c.lst.total > 0 and app._cur.ea == xfn.addr, 20) + c.lst.focus() await c.press("x") await c.wait(lambda: isinstance(app.screen, XrefsScreen), 25) app.screen.query_one(OptionList).highlighted = 0 await c.press("enter") await c.wait(lambda: not isinstance(app.screen, XrefsScreen), 25) - await c.wait(lambda: app._cur.ea == xref.fn_addr, 25) - await c.pause(0.15) - c.check("xref-select lands on the referencing function + line", - app._cur.ea == xref.fn_addr and dis.cursor == xexp, - f"cur={app._cur.ea:#x} (want {xref.fn_addr:#x}) " - f"cursor={dis.cursor} (want {xexp})") - - dcode = app.program.decompile(xref.fn_addr) - dexp, be = -1, -1 - if not dcode.failed and dcode.code: - for i, ln in enumerate(dcode.code.splitlines()): - for mm in re.findall(r"/\*\s*0x([0-9A-Fa-f]+)\s*\*/", ln): - e = int(mm, 16) - if be < e <= xref.frm: - be, dexp = e, i - if dexp >= 0: - app._pref = "decomp" - app._open_function(xfn.addr, xfn.name) - await c.wait(lambda: app._cur.ea == xfn.addr and app._active == "decomp", 20) - await c.wait(lambda: dec.loaded_ea == xfn.addr, 25) - app._xref_focus_name = xfn.name - app._on_xref_chosen(xref.frm) - await c.wait(lambda: dec.loaded_ea == xref.fn_addr, 25) - await c.pause(0.1) - c.check("xref-select lands on the reference LINE in pseudocode", - dec.cursor == dexp, f"dec.cursor={dec.cursor} want={dexp}") - landed = dec._texts[dec.cursor] if dec.cursor < len(dec._texts) else "" - tokm = re.search(rf"\b{re.escape(xfn.name)}\b", landed) - want_col = tokm.start() if tokm else 0 - c.check("xref-select lands the column on the reference token", - dec.cursor_x == want_col, - f"cursor_x={dec.cursor_x} want={want_col} " - f"token={xfn.name!r} line={landed.strip()!r}") - anch = [(i, dec._line_ea(i)) for i in range(len(dec._texts)) - if dec._line_ea(i) is not None] - if len(anch) >= 4: - start_ln = anch[1][0] - far_ea = anch[len(anch) // 2][1] - exp2 = app._decomp_line_for(app._cur.ea, far_ea) - dec.focus() - dec.cursor = start_ln - dec.refresh() - await c.pause(0.05) - app._on_xref_chosen(far_ea) - await c.wait(lambda: dec.cursor == exp2, 20) - c.check("xref-select moves the cursor within the same function", - dec.cursor == exp2 and exp2 != start_ln, - f"dec.cursor={dec.cursor} want={exp2} start={start_ln}") - app._pref = "disasm" - app._active = "disasm" - app._show_active() - await c.pause(0.05) + # xref-select lands the listing cursor on the referencing SITE (frm) + await c.wait(lambda: c.lst._cursor_ea() == xref.frm, 25) + c.check("xref-select lands the cursor on the referencing site", + c.lst._cursor_ea() == xref.frm, + f"cur_ea={c.lst._cursor_ea()} want={xref.frm:#x}") + # F5 at the site decompiles the referencing function + c.lst.focus() + await c.press("tab") + landed = await c.wait( + lambda: (app._active == "decomp" and dec.loaded_ea == xref.fn_addr) + or (app._active == "listing" and "fail" in c.status().lower()), 25) + if app._active == "decomp": + c.check("F5 at the xref site decompiles the referencing function", + dec.loaded_ea == xref.fn_addr, f"loaded={dec.loaded_ea}") + await c.press("tab") + await c.wait(lambda: app._active == "listing", 20) app._open_function(orig, orig_name) await c.wait(lambda: dis.total > 0 and app._cur.ea == orig, 20) @@ -782,9 +1256,10 @@ async def s_xref_labels(c: Ctx): if multi is None: c.check("found a function with multiple same-caller xrefs", False) return - await c.open(multi.addr, "disasm") + await c.open(multi.addr, "listing") dis.focus() - dis.cursor, dis.cursor_x = 0, 0 + dis.cursor = max(dis.model.index_of_ea(multi.addr), 0) # segment head at fn + dis.cursor_x = 0 dis.refresh() await c.press("x") await c.wait(lambda: isinstance(app.screen, XrefsScreen), 25) @@ -835,7 +1310,7 @@ async def s_xref_labels(c: Ctx): @scenario("mouse") async def s_mouse(c: Ctx): app, dis = c.app, c.dis - await c.open_biggest("disasm") + await c.open_biggest("listing") dis.focus() await c.pause(0.1) mrow = mcol = msym = mline = None @@ -1114,6 +1589,103 @@ async def s_retype(c: Ctx): after is not None and "zz_retype_arg" in after.prototype, f"proto={after.prototype if after else None!r}") app.program.set_function_type(cf.addr, old_proto) # restore + # the retype above kicked off a recompile+reload; let it land before we start + # placing the cursor, or the reload resets it under us. + app.program.bump_names() + await c.wait(lambda: not dec.loading and dec.loaded_ea == cf.addr + and bool(dec._texts), 25) + await c.pause(0.3) + + # -- 'y' on a LOCAL variable retypes that variable, not the prototype --- # + fts = app.program.func_types(cf.addr) + lv = next((v for v in (fts.lvars if fts else []) if not v.is_arg), None) + if lv is not None: + line = next((i for i, t in enumerate(dec._texts) + if _word_occurrences(t, lv.name)), None) + if line is not None: + col = _word_occurrences(dec._texts[line], lv.name)[0][0] + dec.focus() + dec.cursor, dec.cursor_x = line, col + 1 # inside the word + dec.refresh() + await c.pause(0.05) + await c.press("y") + await c.wait(lambda: app.query_one("#retype", Input).display, 10) + ri = app.query_one("#retype", Input) + c.check("'y' on a local variable prefills that variable's type", + ri.value == lv.type and lv.name in str(ri.placeholder), + f"val={ri.value!r} want={lv.type!r} ph={ri.placeholder!r}") + ri.value = "unsigned __int64" + await c.press("enter") + changed = await c.wait(lambda: (lambda f: bool(f) and any( + v.name == lv.name and v.type == "unsigned __int64" + for v in f.lvars))(app.program.func_types(cf.addr)), 25) + c.check("applying it retypes the local variable", changed, + f"{lv.name}: wanted unsigned __int64") + after = app.program.func_types(cf.addr) + c.check("retyping a local leaves the prototype alone", + after is not None and after.prototype == old_proto, + f"proto={after.prototype if after else None!r}") + + # -- 'y' on a GLOBAL retypes the global, not the enclosing function ----- # + # The lvar retype above recompiled too — settle again, or the scan below + # indexes into pseudocode that's about to be replaced. + await c.wait(lambda: not dec.loading and dec.loaded_ea == cf.addr + and bool(dec._texts), 25) + await c.pause(0.3) + + # Pick a global that actually appears as a word in the pseudocode — a symbol + # from decompile().refs often doesn't (a string ref renders as its literal). + lvnames = {v.name for v in (fts.lvars if fts else [])} + glob = None + for i, t in enumerate(dec._texts): + for w in re.findall(r"[A-Za-z_][A-Za-z0-9_]*", t): + if w in lvnames or w == cf.name: + continue + try: + a = app.program.resolve(w) + except Exception: # noqa: BLE001 + continue + if app.program.func_types(a) is not None: + continue + d = app.program.data_type(a) or {} + if (d.get("name") and not d.get("is_func") + and "(" not in (d.get("type") or "")): + glob = (w, a, d, i) + break + if glob: + break + if glob is not None: + gname, gaddr, dt, line = glob + glob = type("G", (), {"addr": gaddr})() # keep the checks below readable + if line is not None: + col = _word_occurrences(dec._texts[line], gname)[0][0] + dec.focus() + dec.cursor, dec.cursor_x = line, col + 1 # inside the word + dec.refresh() + await c.pause(0.05) + c.check("the cursor sits on the global", + dec.word_under_cursor() == gname, + f"word={dec.word_under_cursor()!r} want={gname!r} " + f"line={line} col={col} text={dec._texts[line][:60]!r}") + await c.press("y") + await c.wait(lambda: app.query_one("#retype", Input).display, 10) + ri = app.query_one("#retype", Input) + c.check("'y' on a global prefills the global's type (not the proto)", + ri.value != old_proto and gname in str(ri.placeholder), + f"val={ri.value!r} ph={ri.placeholder!r}") + ri.value = "unsigned __int64" + await c.press("enter") + retyped = await c.wait( + lambda: (app.program.data_type(glob.addr) or {}).get("type") + == "unsigned __int64", 25) + c.check("applying it retypes the global", retyped, + f"type={(app.program.data_type(glob.addr) or {}).get('type')!r}") + after = app.program.func_types(cf.addr) + c.check("retyping a global leaves the prototype alone", + after is not None and after.prototype == old_proto, + f"proto={after.prototype if after else None!r}") + if dt.get("type"): # restore + app.program.set_data_type(glob.addr, dt["type"]) @scenario("scroll_restore") @@ -1124,7 +1696,7 @@ async def s_scroll_restore(c: Ctx): fb = next((f for f in big if f.addr != fa.addr), big[-1]) await c.goto_ui(fa.name) await c.wait(lambda: dis.total > 40 and app._cur.ea == fa.addr, 20) - if app._active != "disasm": + if app._active != "listing": await c.press("tab") await c.pause(0.1) for _ in range(4): @@ -1168,7 +1740,7 @@ async def s_paging(c: Ctx): fa = next((f for f in big if f.size > 0x400), big[0]) await c.goto_ui(fa.name) await c.wait(lambda: dis.total > 100 and app._cur.ea == fa.addr, 20) - if app._active != "disasm": + if app._active != "listing": await c.press("tab") await c.pause(0.1) for _ in range(6): @@ -1190,7 +1762,7 @@ async def s_paging(c: Ctx): @scenario("rename_history") async def s_rename_history(c: Ctx): app, dis, dec = c.app, c.dis, c.dec - await c.open_biggest("disasm") + await c.open_biggest("listing") bea = app._cur.ea await c.press("tab") # warm the caller's pseudocode too await c.wait(lambda: dec.loaded_ea == bea, 20) @@ -1236,7 +1808,7 @@ async def s_rename_history(c: Ctx): await c.press("enter") await c.wait(lambda: app._func_index.by_addr(htarget) and app._func_index.by_addr(htarget).name == hnew, 25) - if app._active != "disasm": + if app._active != "listing": await c.press("tab") await c.press("escape") await c.wait(lambda: dis.total > 0 and app._cur.ea != htarget, 25) @@ -1254,12 +1826,399 @@ async def s_rename_history(c: Ctx): app.program.client.call("rename", batch={"func": {"addr": hex(htarget), "name": hsym}}) +@scenario("region_define") +async def s_region_define(c: Ctx): + """A non-function address opens a flat listing (region) instead of being + refused, and 'p' there creates a function and upgrades the view.""" + app = c.app + fn = c.find_func(lambda f: 0x10 <= f.size <= 0x60 and f.name.startswith("sub_")) + if fn is None: + c.check("found a small sub_ function for the region test", False) + return + addr, size = fn.addr, fn.size + try: + # setup: undefine the whole function so [addr, addr+size) is a bare region + c.prog.undefine(addr, size=size) + c.prog.bump_items() + c.check("function removed by undefine", + c.prog.function_of(addr) is None, "still a function") + + # navigate there via the real 'g' prompt -> opens the flat LISTING view + # (a non-function region), not refused + await c.goto_ui(hex(addr)) + await c.wait(lambda: app._cur is not None and app._cur.ea == addr, 25) + c.check("goto to a non-function address opens the listing view (not refused)", + app._cur is not None and app._cur.is_region + and app._active == "listing" and c.lst.display, + f"cur={app._cur} active={app._active} status={c.status()!r}") + await c.wait(lambda: c.lst.total > 0 and c.lst._cursor_ea() is not None, 25) + c.check("listing renders heads and the cursor sits on the target address", + c.lst.total > 0 and c.lst._cursor_ea() == addr, + f"total={c.lst.total} cur_ea={c.lst._cursor_ea()}") + # the flat listing spans the whole segment, not just this function + seg = c.prog.segment_bounds(addr) + c.check("listing spans the whole segment (more heads than one function)", + seg is not None and c.lst.total > 1, f"total={c.lst.total} seg={seg}") + + # 'p' on the entry head (re)creates the function and UPGRADES to DisasmView + c.lst.focus() + c.lst.cursor, c.lst.cursor_x = c.lst.model.index_of_ea(addr), 0 + await c.pause(0.05) + await c.press("p") + await c.wait(lambda: c.prog.function_of(addr) is not None + and app._cur is not None and not app._cur.is_region, 25) + c.check("'p' creates a function and upgrades the listing to a function view", + c.prog.function_of(addr) is not None and not app._cur.is_region + and app._cur.ea == addr and app._active in ("listing", "decomp"), + f"fn={c.prog.function_of(addr)} cur={app._cur} active={app._active}") + finally: + # idempotency: guarantee the function is back even if a check failed + if c.prog.function_of(addr) is None: + try: + c.prog.define_func(addr) + except Exception: # noqa: BLE001 + pass + c.prog.bump_items() + + +@scenario("listing_view") +async def s_listing_view(c: Ctx): + """The flat listing view over a data segment: renders code AND data heads, + navigates, and reaches hex — without any function in play.""" + app = c.app + # find a data segment (has a non-code head somewhere) via the listing itself + data_ea = None + for start, end, name in c.prog.sections(): + if start >= end: + continue + lm = c.prog.listing(start) + if lm is None: + continue + lm.ensure(40) + if any(h.kind == "data" for h in lm.window(0, 40)): + data_ea = start + break + if data_ea is None: + c.check("found a segment with data items", False) + return + + await c.goto_ui(hex(data_ea)) + await c.wait(lambda: app._cur is not None and app._active == "listing" + and c.lst.total > 0, 25) + c.check("navigating to a data segment opens the listing view", + app._active == "listing" and c.lst.display and c.lst.total > 0, + f"active={app._active} total={c.lst.total}") + kinds = {h.kind for h in c.lst.model.window(0, 40)} + c.check("listing shows data heads (not just code)", "data" in kinds, str(kinds)) + + # a rendered data line carries the item text (e.g. db/dd/string) + c.lst.focus() + first_data = next((i for i in range(min(c.lst.total, 60)) + if c.lst.model.get(i) and c.lst.model.get(i).kind == "data"), None) + c.check("a data head exists in the first screenful", first_data is not None, + f"total={c.lst.total}") + if first_data is not None: + c.lst.cursor = first_data + await c.pause(0.05) + plain = c.lst._line_plain(first_data) + c.check("data line renders its item text", bool(plain and plain.strip()), + f"plain={plain!r}") + c.check("listing cursor reports the head address", + c.lst._cursor_ea() == c.lst.model.get(first_data).ea, str(c.lst._cursor_ea())) + + # backslash from the listing opens hex at the cursor address; and back + cur_ea = c.lst._cursor_ea() + await c.press("backslash") + await c.wait(lambda: app._active == "hex" and c.hex.display, 15) + c.check("backslash from the listing opens the hex view", app._active == "hex", + f"active={app._active}") + await c.press("backslash") + await c.wait(lambda: app._active == "listing", 15) + c.check("returning from hex lands back on the listing (not a func view)", + app._active == "listing" and c.lst.display, f"active={app._active}") + + # 'd' defines typed data over an undefined run. Synthesize the run + # deterministically: undefine a data head, then re-type it via the prompt. + dhead = next((c.lst.model.get(i) for i in range(min(c.lst.total, 200)) + if c.lst.model.get(i) and c.lst.model.get(i).kind == "data" + and (c.lst.model.get(i).size or 0) >= 4), None) + if dhead is None: + c.check("found a data head to re-type", False) + return + dea = dhead.ea + try: + c.prog.undefine(dea, size=4) + c.prog.bump_items() + # reopen the listing so the model reflects the new undefined run + await c.goto_ui(hex(dea)) + await c.wait(lambda: app._active == "listing" and c.lst.total > 0 + and c.lst.model.index_of_ea(dea) >= 0, 25) + ui = c.lst.model.index_of_ea(dea) + c.check("undefining a data head yields an unknown run in the listing", + ui >= 0 and c.lst.model.get(ui).kind == "unknown", + f"kind={c.lst.model.get(ui).kind if ui>=0 else None}") + c.lst.focus() + c.lst.cursor = ui + await c.pause(0.05) + await c.press("d") + await c.pause(0.1) + mdi = app.query_one("#makedata", Input) + c.check("'d' opens the make-data prompt prefilled with a type", + mdi.display and bool(mdi.value), f"display={mdi.display} val={mdi.value!r}") + mdi.value = "char[4]" + await c.press("enter") + await c.wait(lambda: app._active == "listing" + and c.lst.model.index_of_ea(dea) >= 0 + and c.lst.model.get(c.lst.model.index_of_ea(dea)) is not None + and c.lst.model.get(c.lst.model.index_of_ea(dea)).kind == "data", 25) + di = c.lst.model.index_of_ea(dea) + c.check("'d' turns the undefined run into a typed data item", + di >= 0 and c.lst.model.get(di).kind == "data", + f"kind={c.lst.model.get(di).kind if di>=0 else None}") + finally: + c.prog.bump_items() + + +@scenario("listing_name_addr") +async def s_listing_name_addr(c: Ctx): + """In the listing, 'n' names the ADDRESS under the cursor — so you can name a + bare/undefined byte (e.g. the free byte at addr+1 after shrinking a u16 to a + u8), which the symbol-by-name rename path can't do.""" + app = c.app + # pick a data segment and a >=2-byte data head to shrink + data_ea = None + for start, end, name in c.prog.sections(): + if start >= end: + continue + lm = c.prog.listing(start) + if lm is None: + continue + lm.ensure(60) + if any(h.kind == "data" and (h.size or 0) >= 2 for h in lm.window(0, 60)): + data_ea = start + break + if data_ea is None: + c.check("found a data segment with a >=2-byte item", False) + return + dh = next(h for h in c.prog.listing(data_ea).window(0, 60) + if h.kind == "data" and (h.size or 0) >= 2) + A = dh.ea + newname = f"after_{os.getpid()}" + try: + # shrink the >=2-byte item to a single byte -> A+1 becomes undefined + c.prog.make_data(A, "unsigned __int8") + c.prog.bump_items() + await c.goto_ui(hex(A + 1)) + await c.wait(lambda: app._active == "listing" and app._cur is not None + and app._cur.ea == A + 1, 25) + head = c.lst.cur_head() + c.check("cursor lands on the now-undefined byte at addr+1", + head is not None and head.ea == A + 1 and head.kind == "unknown", + f"head={head}") + # 'n' opens the address-name prompt (even though there's no symbol) + c.lst.focus() + await c.press("n") + await c.pause(0.1) + ri = app.query_one("#rename", Input) + c.check("'n' opens the name prompt on an unnamed byte", + ri.display, f"display={ri.display}") + ri.value = newname + await c.press("enter") + await c.wait(lambda: app._active == "listing" + and c.lst.model.index_of_ea(A + 1) >= 0 + and c.lst.model.get(c.lst.model.index_of_ea(A + 1)) is not None + and c.lst.model.get(c.lst.model.index_of_ea(A + 1)).name == newname, 25) + hi = c.lst.model.index_of_ea(A + 1) + c.check("naming a bare byte at addr+1 sticks", + hi >= 0 and c.lst.model.get(hi).name == newname, + f"name={c.lst.model.get(hi).name if hi>=0 else None}") + finally: + # revert: drop the label and restore raw bytes at A + try: + c.prog.client.call("rename", batch={"data": {"addr": hex(A + 1), "new": ""}}) + except Exception: # noqa: BLE001 + pass + c.prog.undefine(A, size=8) + c.prog.bump_items() + + +@scenario("listing_make_string") +async def s_listing_make_string(c: Ctx): + """'a' in the listing makes a string literal at the cursor (IDA's 'A').""" + app = c.app + target = None + for start, end, name in c.prog.sections(): + if start >= end: + continue + lm = c.prog.listing(start) + if lm is None: + continue + lm.ensure(80) + h = next((h for h in lm.window(0, 80) + if h.kind == "data" and "'" in h.text), None) + if h is not None: + target = h.ea + break + if target is None: + c.check("found a string data item to remake", False) + return + A = target + try: + c.prog.undefine(A, size=8) + c.prog.bump_items() + await c.goto_ui(hex(A)) + await c.wait(lambda: app._active == "listing" and app._cur is not None + and app._cur.ea == A, 25) + c.check("target is undefined before 'a'", + c.lst.cur_head() is not None and c.lst.cur_head().kind == "unknown", + f"head={c.lst.cur_head()}") + c.lst.focus() + await c.press("a") + await c.wait(lambda: c.lst.model.index_of_ea(A) >= 0 + and c.lst.model.get(c.lst.model.index_of_ea(A)) is not None + and c.lst.model.get(c.lst.model.index_of_ea(A)).kind == "data" + and "'" in c.lst.model.get(c.lst.model.index_of_ea(A)).text, 25) + hi = c.lst.model.index_of_ea(A) + c.check("'a' creates a string literal at the cursor", + hi >= 0 and c.lst.model.get(hi).kind == "data" + and "'" in c.lst.model.get(hi).text, + f"head={c.lst.model.get(hi) if hi >= 0 else None}") + finally: + try: + c.prog.make_string(A) # restore the original string + except Exception: # noqa: BLE001 + pass + c.prog.bump_items() + + +@scenario("listing_struct_expand") +async def s_listing_struct_expand(c: Ctx): + """A struct-typed global expands into indented member rows in the listing.""" + app = c.app + # a writable data address: reuse a data segment's first data head + A = None + for start, end, name in c.prog.sections(): + if start >= end: + continue + lm = c.prog.listing(start) + if lm is None: + continue + lm.ensure(60) + h = next((h for h in lm.window(0, 60) if h.kind == "data"), None) + if h is not None: + A = h.ea + break + if A is None: + c.check("found a data address for the struct test", False) + return + try: + c.prog.client.call( + "declare_type", + decls=["struct TuiExpandS { int a; char b[4]; short c; };"]) + c.prog.make_data(A, "TuiExpandS") + c.prog.bump_items() + await c.goto_ui(hex(A)) + await c.wait(lambda: app._active == "listing" and app._cur is not None + and app._cur.ea == A and c.lst.total > 0, 25) + # the summary head, then member rows for a/b/c + si = c.lst.model.index_of_ea(A) + members = [c.lst.model.get(si + 1 + k) for k in range(3)] + names = [m.text for m in members if m is not None] + c.check("struct global expands into member rows", + all(m is not None and m.kind == "member" for m in members) + and any("a" in t for t in names) and any("b" in t for t in names), + f"members={names}") + c.check("member rows carry field addresses", + members[1] is not None and members[1].ea == A + 4, + f"ea={members[1].ea if members[1] else None:#x} want={A+4:#x}") + finally: + try: + c.prog.undefine(A, size=16) + except Exception: # noqa: BLE001 + pass + c.prog.bump_items() + + +@scenario("continuous_view") +async def s_continuous_view(c: Ctx): + """The default code view is ONE continuous listing: opening a function shows + the whole segment (functions + data interleaved, with opcodes); F5/Tab + decompiles the function at the cursor and back.""" + app = c.app + fn = await c.open_biggest("listing") + fn_ea = fn.addr + await c.wait(lambda: app._active == "listing" and c.lst.display, 10) + c.check("a function opens in the continuous listing by default", + app._active == "listing" and c.lst.display + and c.lst._cursor_ea() == fn_ea, + f"active={app._active} disp={c.lst.display} cur_ea={c.lst._cursor_ea()}") + # the listing spans the whole segment, not just the function + c.lst.model.load_all() + seg = c.prog.segment_bounds(fn_ea) + seg_rows = len(c.lst.model) + # a function's own instruction count is far smaller than the segment + fdis = c.prog.disasm(fn_ea, fn.name) + c.check("the continuous listing extends past the function's bounds", + seg_rows > fdis.total(), f"listing={seg_rows} func={fdis.total()}") + kinds = {c.lst.model.get(i).kind for i in range(seg_rows)} + c.check("continuous listing interleaves code with data/undefined", + "code" in kinds and ("data" in kinds or "unknown" in kinds), str(kinds)) + # rendering parity with disasm: code lines carry opcode bytes + cidx = next((i for i in range(seg_rows) + if c.lst.model.get(i).kind == "code"), None) + c.check("continuous listing renders opcode bytes (parity with disasm)", + cidx is not None and c.lst.model.get(cidx).raw + and c.lst._op_field(c.lst.model.get(cidx)).strip() != "", + f"raw={c.lst.model.get(cidx).raw if cidx is not None else None!r}") + # F5/Tab at the function -> decompiler, and back to the same spot + c.lst.focus() + await c.press("tab") + await c.wait(lambda: (app._active == "decomp" and c.dec.loaded_ea == fn_ea) + or (app._active == "listing" and "fail" in c.status().lower()), 25) + if app._active == "decomp": + c.check("F5/Tab decompiles the function under the cursor", + c.dec.loaded_ea == fn_ea, f"loaded={c.dec.loaded_ea}") + await c.press("tab") + await c.wait(lambda: app._active == "listing", 25) + c.check("F5/Tab in the decompiler returns to the listing at the same ea", + app._active == "listing" and c.lst._cursor_ea() == fn_ea, + f"active={app._active} cur_ea={c.lst._cursor_ea()}") + else: + c.check("undecompilable function falls back to the listing", + app._active == "listing", f"active={app._active}") + + +@scenario("func_banners") +async def s_func_banners(c: Ctx): + """The unified listing shows IDA-style function boundary banners: a + SUBROUTINE separator + 'name proc near' header and a 'name endp' footer, + and those banner rows are display-only (navigation lands on real code).""" + app = c.app + fn = await c.open_biggest("listing") + c.lst.model.load_all() + heads = [c.lst.model.get(i) for i in range(len(c.lst.model))] + ci = c.lst.model.index_of_ea(fn.addr) + c.check("navigation to a function lands on its code head, not a banner", + ci >= 0 and c.lst.model.get(ci).kind == "code", + f"kind={c.lst.model.get(ci).kind if ci >= 0 else None}") + c.check("a SUBROUTINE separator banner is present", + any(h.kind == "sep" and "S U B R O U T I N E" in h.text for h in heads)) + c.check("a 'name proc' header is present", + any(h.kind == "funchdr" and h.text.endswith(" proc") for h in heads)) + c.check("a 'name endp' footer is present", + any(h.kind == "funchdr" and h.text.endswith("endp") for h in heads)) + # the proc header for the origin function carries its name + hdr = next((h for h in heads if h.kind == "funchdr" + and h.text == f"{fn.name} proc"), None) + c.check("the proc header names the function", hdr is not None, f"fn={fn.name}") + + # --------------------------------------------------------------------------- # # Runner # --------------------------------------------------------------------------- # -async def run(db, only=None): - url = os.environ.get("IDA_MCP_URL", "http://127.0.0.1:8745/mcp") - app = IdaTui(url=url, db=db, keepalive=False) +async def run(binary, only=None): + # Own idalib worker: opens the binary in-process over a unix socket. + app = IdaTui(open_path=binary, keepalive=False) async with app.run_test(size=(140, 44)) as pilot: c = Ctx(app, pilot) await c.boot() @@ -1282,21 +2241,28 @@ async def run(db, only=None): def main(argv): global STOP_AFTER - db = only = None + only = None + binary = None it = iter(argv) for a in it: - if a == "--db": - db = next(it) - elif a == "--only": + if a == "--only": only = next(it).split(",") elif a == "--stop-after": STOP_AFTER = next(it) + elif a in ("--worker", "--binary"): + binary = os.path.abspath(os.path.expanduser(next(it))) elif a == "--list": for name, _ in SCENARIOS: print(name) return 0 + elif not a.startswith("-"): + binary = os.path.abspath(os.path.expanduser(a)) + if binary is None: # default target for the pilot + binary = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "targets", "echo") try: - asyncio.run(run(db, only)) + asyncio.run(run(binary, only)) except _StopSuite: print(f" … stopped after '{STOP_AFTER}'") print(f"\n{PASS} passed, {FAIL} failed") |