aboutsummaryrefslogtreecommitdiffstats
path: root/experiments
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--experiments/bench_ops.py116
-rw-r--r--experiments/bench_pack_trace.py79
-rw-r--r--experiments/call_census.py174
-rw-r--r--experiments/cfg_dump.py41
-rw-r--r--experiments/graph_shot.py39
-rw-r--r--experiments/graph_smoke.py35
-rw-r--r--experiments/graph_spike.py70
-rw-r--r--experiments/inproc_spike.py112
-rw-r--r--experiments/modal_shot.py70
-rw-r--r--experiments/opfmt_tools.py241
-rw-r--r--experiments/profile_client.py76
-rw-r--r--experiments/profile_remote.py73
-rw-r--r--experiments/splash_place_count.py102
-rw-r--r--experiments/worker_smoke.py15
14 files changed, 1049 insertions, 194 deletions
diff --git a/experiments/bench_ops.py b/experiments/bench_ops.py
new file mode 100644
index 0000000..1c0ad69
--- /dev/null
+++ b/experiments/bench_ops.py
@@ -0,0 +1,116 @@
+"""Time a realistic idatui operation mix against whatever ida-nexus is installed.
+
+The companion to `bench_pack_trace.py`: that one isolates a single workaround,
+this one answers "how much faster is the whole client, on real operations".
+
+**It deliberately does not import anything version-specific**, so the SAME file
+can measure an OLD idatui checkout (with its `sys.settrace` strip and packing
+workarounds) and the current one. To compare across versions, copy it somewhere
+outside the repo first -- `git checkout` of an older commit would otherwise
+replace or delete it::
+
+ cp experiments/bench_ops.py /tmp/
+ # C: current client, current library
+ PYTHONPATH=. ~/ida-venv/bin/python /tmp/bench_ops.py
+
+ # B: current client against the OLD library (shows what the workarounds were for)
+ git -C ~/dev/ida-nexus checkout 4195f21
+ PYTHONPATH=. ~/ida-venv/bin/python /tmp/bench_ops.py
+
+ # A: the client as it SHIPPED on the old library, workarounds and all
+ git checkout 8550474 # the commit before the workaround removal
+ PYTHONPATH=. ~/ida-venv/bin/python /tmp/bench_ops.py
+
+ git checkout main && git -C ~/dev/ida-nexus checkout main # ALWAYS restore
+
+ida-nexus is installed **editable** into both venvs, so checking that repo out
+swaps the backend under the TUI with no reinstall -- which is what makes this A/B
+cheap.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import statistics
+import time
+
+from idatui import remote_ops
+from idatui.nexus_client import NexusClient
+
+
+def bench(fn, reps: int) -> tuple[float, float]:
+ """Best-of and median wall time in ms; best-of resists co-tenant noise."""
+ samples = []
+ for _ in range(reps):
+ started = time.perf_counter()
+ fn()
+ samples.append((time.perf_counter() - started) * 1000.0)
+ return min(samples), statistics.median(samples)
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser()
+ ap.add_argument("target", nargs="?", default="targets/bash")
+ ap.add_argument("--reps", type=int, default=20)
+ args = ap.parse_args()
+
+ client = NexusClient(os.path.abspath(args.target))
+ client.connect()
+ handle = client._handle
+
+ # Work on the biggest function we can find, so the payload-heavy operations
+ # are actually payload-heavy.
+ index = client.call(remote_ops.list_funcs, queries=[{"offset": 0, "count": 60}])
+ funcs = (index.get("result") or [{}])[0].get("data") or []
+ if not funcs:
+ print("VERDICT: FAIL - no functions")
+ return 1
+ big = max(funcs, key=lambda f: f.get("size") or 0)
+ ea = big["addr"] if isinstance(big["addr"], str) else hex(big["addr"])
+
+ ops = [
+ # Synthetic: isolates the per-operation floor (execute_sync marshalling).
+ ("empty round trip", lambda: handle.execute_python("result = 1")),
+ # Payload-dominated: what _PACK_EPILOGUE was written for.
+ (
+ "list_funcs 500",
+ lambda: client.call(
+ remote_ops.list_funcs, queries=[{"offset": 0, "count": 500}]
+ ),
+ ),
+ (
+ "heads 200 (listing page)",
+ lambda: client.call(remote_ops.heads, addr=ea, count=200, annotate=True),
+ ),
+ # IDA-work-dominated: Hex-Rays, nothing upstream can move.
+ ("decompile (warm)", lambda: client.call(remote_ops.decompile, addr=ea)),
+ ("flowchart (graph)", lambda: client.call(remote_ops.flowchart, addr=ea)),
+ # Round-trip-dominated: small payload, so only the floor matters.
+ (
+ "xrefs_to",
+ lambda: client.call(remote_ops.xref_query, direction="to", addr=ea),
+ ),
+ ]
+
+ print(
+ f"# target={os.path.basename(args.target)} func={ea} reps={args.reps} "
+ f"backend={client.backend}"
+ )
+ results = {}
+ for name, fn in ops:
+ try:
+ for _ in range(3): # warm caches; the first sample is always an outlier
+ fn()
+ best, med = bench(fn, args.reps)
+ results[name] = med
+ print(f"{name:28} best {best:8.3f}ms median {med:8.3f}ms")
+ except Exception as exc: # one broken op must not lose the other five
+ print(f"{name:28} FAILED: {type(exc).__name__}: {str(exc)[:60]}")
+ client.close()
+ print("RESULT " + ";".join(f"{k}={v:.3f}" for k, v in results.items()))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/experiments/bench_pack_trace.py b/experiments/bench_pack_trace.py
new file mode 100644
index 0000000..2b53afd
--- /dev/null
+++ b/experiments/bench_pack_trace.py
@@ -0,0 +1,79 @@
+"""Measure cold installation versus warm calls for typed remote modules.
+
+Historical note: this file used to benchmark ``_PACK_EPILOGUE``. Application
+scripts are no longer strings and packing is gone; the relevant design cost is
+now the one-time content-addressed module installation versus steady-state calls.
+
+Usage::
+
+ PYTHONPATH=. python experiments/bench_pack_trace.py [FILE]
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import statistics
+import time
+
+from idatui import remote_ops
+from idatui.nexus_client import NexusClient
+
+
+def timed(function, reps: int = 1) -> tuple[object, float]:
+ samples = []
+ result = None
+ for _ in range(reps):
+ started = time.perf_counter()
+ result = function()
+ samples.append((time.perf_counter() - started) * 1000.0)
+ return result, statistics.median(samples)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("target", nargs="?", default="targets/bash")
+ parser.add_argument("--reps", type=int, default=25)
+ parser.add_argument("--rows", type=int, default=200)
+ args = parser.parse_args()
+
+ client = NexusClient(os.path.abspath(args.target)).connect()
+
+ index, operations_cold = timed(
+ lambda: client.call(remote_ops.list_funcs, queries=[{"offset": 0, "count": 40}])
+ )
+ funcs = (index.get("result") or [{}])[0].get("data") or []
+ biggest = max(funcs, key=lambda function: function.get("size") or 0, default=None)
+ if not biggest:
+ print("VERDICT: FAIL - no functions")
+ return 1
+ addr = biggest["addr"]
+ if isinstance(addr, int):
+ addr = hex(addr)
+
+ _, operations_warm = timed(
+ lambda: client.call(
+ remote_ops.list_funcs, queries=[{"offset": 0, "count": 40}]
+ ),
+ args.reps,
+ )
+ page = lambda: client.call( # noqa: E731
+ remote_ops.heads, addr=addr, count=args.rows, annotate=True
+ )
+ payload, tools_cold = timed(page)
+ _, tools_warm = timed(page, args.reps)
+
+ print(f"target {os.path.basename(args.target)} backend={client.backend}")
+ print(f"function {addr}")
+ print(f"listing rows {len(payload.get('heads', []))}")
+ print(
+ f"operations.py cold {operations_cold:8.3f}ms warm {operations_warm:8.3f}ms"
+ )
+ print(f"remote_tools.py cold {tools_cold:8.3f}ms warm {tools_warm:8.3f}ms")
+ print(f"tools install overhead {tools_cold / max(tools_warm, 0.001):.2f}x one time")
+ client.close()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/experiments/call_census.py b/experiments/call_census.py
new file mode 100644
index 0000000..4c0f9d3
--- /dev/null
+++ b/experiments/call_census.py
@@ -0,0 +1,174 @@
+"""Count backend round-trips per user action.
+
+Answers "are we batching, or paying a round-trip per item?" with numbers rather
+than intent. Wraps ``NexusClient.invoke`` on the live app, drives a headless
+Pilot through realistic actions, and reports calls + wall time + which
+operations were used for each.
+
+ PYTHONPATH=. ~/ida-venv/bin/python experiments/call_census.py [BINARY]
+
+Read it as: an action costing 1-8 calls is amortised (the snippet looped inside
+the database); an action whose call count scales with the number of rows or
+symbols on screen is a round-trip-per-item bug worth fixing.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import collections
+import os
+import sys
+import time
+
+sys.path.insert(
+ 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests")
+)
+from _fixtures import fast_keys, staged # noqa: E402
+
+fast_keys()
+
+from idatui.app import IdaTui, ListingView # noqa: E402
+from idatui.nexus_client import NexusClient # noqa: E402
+
+
+class Census:
+ """Patch invoke() once; measure named spans against it."""
+
+ def __init__(self) -> None:
+ self.ops: collections.Counter = collections.Counter()
+ self.n = 0
+ original = NexusClient.invoke
+
+ def counting(client, operation, *a, **kw):
+ self.n += 1
+ self.ops[operation] += 1
+ return original(client, operation, *a, **kw)
+
+ NexusClient.invoke = counting
+ self._original = original
+
+ def restore(self) -> None:
+ NexusClient.invoke = self._original
+
+ def span(self, label: str):
+ return _Span(self, label)
+
+
+class _Span:
+ def __init__(self, census: Census, label: str) -> None:
+ self.c, self.label = census, label
+
+ def __enter__(self):
+ self.n0 = self.c.n
+ self.ops0 = self.c.ops.copy()
+ self.t0 = time.perf_counter()
+ return self
+
+ def __exit__(self, *exc):
+ ms = (time.perf_counter() - self.t0) * 1000
+ used = self.c.ops - self.ops0
+ detail = " ".join(
+ f"{k}x{v}" if v > 1 else k
+ for k, v in sorted(used.items(), key=lambda kv: -kv[1])
+ )
+ print(
+ f" {self.label:<34} {self.c.n - self.n0:>3} calls {ms:7.1f}ms {detail}"
+ )
+ return False
+
+
+async def main() -> int:
+ binary = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "targets/bash")
+ async with staged(
+ binary, lambda p: IdaTui(open_path=p, keepalive=False), prefix="idatui-census-"
+ ) as target:
+ app = IdaTui(open_path=target, keepalive=False)
+ census = Census()
+ try:
+ async with app.run_test(size=(140, 44)) as pilot:
+ for _ in range(200):
+ if getattr(app, "_cur", None) is not None:
+ break
+ await pilot.pause(0.05)
+ print(f"\n# {os.path.basename(binary)} — backend calls per action\n")
+
+ # THE BACKGROUND GROWER MUST FINISH FIRST.
+ #
+ # ListingView._grow streams the WHOLE segment in 500-head pages
+ # on a worker thread, so it lands calls continuously no matter
+ # what the user is doing. Measuring an action while it runs
+ # attributes its traffic to that action -- every span comes out
+ # at a near-identical "~1 call per 10ms of pause", which says
+ # nothing about the action. Drain it, report it as its own line,
+ # then measure against a quiet backend.
+ def listing_done() -> bool:
+ try:
+ m = app.query_one(ListingView).model
+ except Exception:
+ return False
+ return m is not None and m.complete
+
+ with census.span("boot: stream the whole segment"):
+ for _ in range(2000):
+ if listing_done():
+ break
+ await pilot.pause(0.05)
+ await pilot.pause(0.4)
+ drained = listing_done()
+ print(
+ f" {'(grower finished: ' + str(drained) + ')':<34}\n"
+ f" -- everything below is on a QUIET backend --\n"
+ )
+
+ with census.span("scroll one page (pagedown)"):
+ await pilot.press("pagedown")
+ await pilot.pause(0.2)
+
+ with census.span("scroll 20 pages"):
+ for _ in range(20):
+ await pilot.press("pagedown")
+ await pilot.pause(0.5)
+
+ with census.span("switch to pseudocode (tab)"):
+ await pilot.press("tab")
+ await pilot.pause(0.6)
+
+ with census.span("cursor down x30 in pseudocode"):
+ for _ in range(30):
+ await pilot.press("down")
+ await pilot.pause(0.3)
+
+ with census.span("open graph (space)"):
+ await pilot.press("space")
+ await pilot.pause(0.8)
+
+ with census.span("open symbol palette (ctrl+n)"):
+ await pilot.press("ctrl+n")
+ await pilot.pause(0.4)
+
+ with census.span("type 5 chars into the palette"):
+ for ch in "write":
+ await pilot.press(ch)
+ await pilot.pause(0.4)
+ await pilot.press("escape")
+ await pilot.pause(0.2)
+
+ with census.span("hex view (backslash)"):
+ await pilot.press("backslash")
+ await pilot.pause(0.5)
+
+ with census.span("scroll hex 10 pages"):
+ for _ in range(10):
+ await pilot.press("pagedown")
+ await pilot.pause(0.4)
+
+ print(f"\n {'TOTAL':<34} {census.n:>3} calls")
+ top = ", ".join(f"{k}x{v}" for k, v in census.ops.most_common(6))
+ print(f" most-used ops: {top}\n")
+ finally:
+ census.restore()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(asyncio.run(main()))
diff --git a/experiments/cfg_dump.py b/experiments/cfg_dump.py
index 30b6ba9..bef24e7 100644
--- a/experiments/cfg_dump.py
+++ b/experiments/cfg_dump.py
@@ -15,6 +15,7 @@ taken branch) or "switch" (n-way). That is exactly the input a graph view needs;
everything after this point (layering, ordering, routing, rendering) is pure
python and needs no IDA at all.
"""
+
from __future__ import annotations
import argparse
@@ -34,12 +35,12 @@ def dump(path: str, want: list[str], max_blocks: int) -> list[dict]:
if idapro.open_database(local, run_auto_analysis=True) != 0:
raise SystemExit(f"failed to open {local}")
+ import ida_bytes
import ida_funcs
import ida_gdl
import ida_lines
- import ida_bytes
- import idautils
import idaapi
+ import idautils
out = []
try:
@@ -60,7 +61,8 @@ def dump(path: str, want: list[str], max_blocks: int) -> list[dict]:
ea = bb.start_ea
while ea < bb.end_ea and ea != idaapi.BADADDR:
txt = ida_lines.tag_remove(
- ida_lines.generate_disasm_line(ea, 0) or "")
+ ida_lines.generate_disasm_line(ea, 0) or ""
+ )
lines.append(txt.rstrip())
nxt = ida_bytes.next_head(ea, bb.end_ea)
if nxt <= ea:
@@ -78,13 +80,15 @@ def dump(path: str, want: list[str], max_blocks: int) -> list[dict]:
else:
kind = "jump"
succs.append([index[s.start_ea], kind])
- blocks.append({
- "id": index[bb.start_ea],
- "start": bb.start_ea,
- "end": bb.end_ea,
- "lines": lines,
- "succs": succs,
- })
+ blocks.append(
+ {
+ "id": index[bb.start_ea],
+ "start": bb.start_ea,
+ "end": bb.end_ea,
+ "lines": lines,
+ "succs": succs,
+ }
+ )
if max_blocks and len(blocks) > max_blocks:
continue
out.append({"name": name, "ea": fea, "blocks": blocks})
@@ -98,10 +102,19 @@ def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("binary")
ap.add_argument("-o", "--out", default="/tmp/cfg.json")
- ap.add_argument("-f", "--func", action="append", default=[],
- help="only functions whose name contains this (repeatable)")
- ap.add_argument("--max-blocks", type=int, default=0,
- help="skip functions with more blocks than this")
+ ap.add_argument(
+ "-f",
+ "--func",
+ action="append",
+ default=[],
+ help="only functions whose name contains this (repeatable)",
+ )
+ ap.add_argument(
+ "--max-blocks",
+ type=int,
+ default=0,
+ help="skip functions with more blocks than this",
+ )
args = ap.parse_args()
recs = dump(os.path.abspath(args.binary), args.func, args.max_blocks)
diff --git a/experiments/graph_shot.py b/experiments/graph_shot.py
index 5f0d45a..e3eb209 100644
--- a/experiments/graph_shot.py
+++ b/experiments/graph_shot.py
@@ -1,11 +1,12 @@
"""Render the graph view headless at a chosen size and print it.
- ~/ida-venv/bin/python experiments/graph_shot.py [func] [cols] [rows] [zoom]
+ ~/ida-venv/bin/python experiments/graph_shot.py [func] [cols] [rows] [zoom] [engine]
The pane a person runs this in is usually too small to judge the layout, and the
pilot lays out synchronously at whatever size you ask for -- so this is the way
to actually look at the thing.
"""
+
import asyncio
import os
import shutil
@@ -19,6 +20,7 @@ func = sys.argv[1] if len(sys.argv) > 1 else "sub_2297"
cols = int(sys.argv[2]) if len(sys.argv) > 2 else 170
rows = int(sys.argv[3]) if len(sys.argv) > 3 else 55
zoom = int(sys.argv[4]) if len(sys.argv) > 4 else 0
+engine = sys.argv[5] if len(sys.argv) > 5 else None
src, tmp = f"{REPO}/targets/echo", "/tmp/echo_shot"
shutil.copy(src, tmp)
@@ -28,26 +30,33 @@ for e in ".i64 .id0 .id1 .id2 .nam .til".split():
except OSError:
pass
-from idatui.app import IdaTui, GraphView # noqa: E402
-from idatui._sync import wait_for # noqa: E402
-from idatui.rpc import screen_text # noqa: E402
+from idatui._sync import wait_for # noqa: E402
+from idatui.app import GraphView, IdaTui # noqa: E402
+from idatui.rpc import screen_text # noqa: E402
async def main() -> None:
app = IdaTui(tmp, keepalive=False)
async with app.run_test(size=(cols, rows)) as pilot:
- await wait_for(lambda: app.program is not None and app._cur is not None,
- pilot.pause, 120)
+ await wait_for(
+ lambda: app.program is not None and app._cur is not None, pilot.pause, 120
+ )
ea = app.program.resolve(func)
fn = app.program.function_of(ea)
app._open_function(fn.addr, fn.name)
- await wait_for(lambda: app._cur is not None and app._cur.ea == fn.addr,
- pilot.pause, 60)
+ await wait_for(
+ lambda: app._cur is not None and app._cur.ea == fn.addr, pilot.pause, 60
+ )
await pilot.pause(0.2)
await pilot.press("space")
gv = app.query_one(GraphView)
- ok = await wait_for(lambda: app._active == "graph" and gv.lay is not None,
- pilot.pause, 90)
+ if engine:
+ gv._engine = engine
+ gv._relayout()
+ app._graph_status()
+ ok = await wait_for(
+ lambda: app._active == "graph" and gv.lay is not None, pilot.pause, 90
+ )
if not ok:
print("graph never opened:", app.query_one("#status").render())
return
@@ -58,8 +67,14 @@ async def main() -> None:
print(screen_text(app)["text"])
print()
print("status:", app.query_one("#status").render())
- print("stats :", gv.lay.stats, "canvas",
- f"{gv.lay.width}x{gv.lay.height}", "zoom", gv.ZOOMS[gv._zoom])
+ print(
+ "stats :",
+ gv.lay.stats,
+ "canvas",
+ f"{gv.lay.width}x{gv.lay.height}",
+ "zoom",
+ gv.ZOOMS[gv._zoom],
+ )
app._save_on_exit = False
diff --git a/experiments/graph_smoke.py b/experiments/graph_smoke.py
index 1b6e21e..92fe1f5 100644
--- a/experiments/graph_smoke.py
+++ b/experiments/graph_smoke.py
@@ -5,6 +5,7 @@ idatui.graph layout, through a real idalib worker.
Wants: VERDICT: OK
"""
+
import os
import shutil
import sys
@@ -22,16 +23,17 @@ for e in ".i64 .id0 .id1 .id2 .nam .til".split():
except OSError:
pass
-from idatui.worker_client import WorkerClient # noqa: E402
-from idatui.domain import Program # noqa: E402
-from idatui import graph as G # noqa: E402
+from idatui.worker_client import WorkerClient # noqa: E402
+
+from idatui import graph as G # noqa: E402
+from idatui.domain import Program # noqa: E402
want = sys.argv[1] if len(sys.argv) > 1 else "main"
print("spawning worker + opening echo\u2026", flush=True)
t = time.time()
cl = WorkerClient(tmp)
cl.connect(progress=lambda m: None)
-print(f" worker ready in {time.time()-t:.2f}s", flush=True)
+print(f" worker ready in {time.time() - t:.2f}s", flush=True)
prog = Program(cl)
ok = True
@@ -40,7 +42,7 @@ print(f"resolve({want!r}) = {ea:#x}", flush=True)
t = time.time()
fc = prog.flowchart(ea)
-print(f"flowchart() -> {time.time()-t:.2f}s", flush=True)
+print(f"flowchart() -> {time.time() - t:.2f}s", flush=True)
if fc is None:
print("VERDICT: FAIL (no flowchart)")
raise SystemExit(1)
@@ -55,8 +57,10 @@ if not nrows:
empty = [b for b in fc.blocks if not b.rows]
if empty:
ok = False
- print(f" !! {len(empty)} blocks have NO rows, e.g. "
- f"{[hex(b.start) for b in empty[:4]]}")
+ print(
+ f" !! {len(empty)} blocks have NO rows, e.g. "
+ f"{[hex(b.start) for b in empty[:4]]}"
+ )
b0 = fc.blocks[fc.entry]
print(f" entry block {b0.start:#x}-{b0.end:#x}:")
@@ -74,20 +78,20 @@ for b in fc.blocks:
ok = False
print(f" !! block {b.start:#x} has rows outside its range")
-blocks = [G.Block(id=b.id, start=b.start, end=b.end, succs=list(b.succs))
- for b in fc.blocks]
+blocks = [
+ G.Block(id=b.id, start=b.start, end=b.end, succs=list(b.succs)) for b in fc.blocks
+]
def sizer(b):
src_b = fc.blocks[b.id]
- w = max([len(f"loc_{b.start:X}")]
- + [len(h.text) + 12 for h in src_b.rows]) + 4
+ w = max([len(f"loc_{b.start:X}")] + [len(h.text) + 12 for h in src_b.rows]) + 4
return (w, len(src_b.rows) + 3)
t = time.time()
lay = G.layout(blocks, sizer, entry=fc.entry)
-print(f"layout() -> {(time.time()-t)*1000:.0f} ms {lay.stats}", flush=True)
+print(f"layout() -> {(time.time() - t) * 1000:.0f} ms {lay.stats}", flush=True)
print(f" canvas {lay.width}x{lay.height}")
ok &= len(lay.nodes) == len(blocks)
@@ -97,9 +101,10 @@ ok &= covered == {b.id for b in blocks}
if covered != {b.id for b in blocks}:
print(f" !! row index misses {sorted({b.id for b in blocks} - covered)[:5]}")
-hits = sum(1 for r in range(min(lay.height, 400))
- if lay.painting.cells_at_row(r, 0, lay.width))
-print(f" {hits} of the first {min(lay.height,400)} rows carry edge cells")
+hits = sum(
+ 1 for r in range(min(lay.height, 400)) if lay.painting.cells_at_row(r, 0, lay.width)
+)
+print(f" {hits} of the first {min(lay.height, 400)} rows carry edge cells")
ok &= hits > 0
cl.close()
diff --git a/experiments/graph_spike.py b/experiments/graph_spike.py
index 547c829..b3c5dc2 100644
--- a/experiments/graph_spike.py
+++ b/experiments/graph_spike.py
@@ -11,6 +11,7 @@ milliseconds when you're changing layout heuristics.
python3 experiments/graph_spike.py /tmp/cfg-echo.json --func sub_61D0
python3 experiments/graph_spike.py /tmp/cfg-echo.json --stats
"""
+
from __future__ import annotations
import argparse
@@ -23,8 +24,10 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from idatui import graph as G # noqa: E402
COLOR = {
- G.E_UNCOND: "\033[38;5;39m", G.E_TRUE: "\033[38;5;40m",
- G.E_FALSE: "\033[38;5;203m", G.E_SWITCH: "\033[38;5;178m",
+ G.E_UNCOND: "\033[38;5;39m",
+ G.E_TRUE: "\033[38;5;40m",
+ G.E_FALSE: "\033[38;5;203m",
+ G.E_SWITCH: "\033[38;5;178m",
G.E_BACK: "\033[38;5;135m",
}
DIM, RESET = "\033[38;5;244m", "\033[0m"
@@ -37,11 +40,19 @@ def build(rec: dict, max_lines: int):
for b in rec["blocks"]:
lines = list(b["lines"])
if max_lines and len(lines) > max_lines:
- lines = lines[:max_lines - 1] + [f"... {len(b['lines']) - max_lines + 1} more"]
+ lines = lines[: max_lines - 1] + [
+ f"... {len(b['lines']) - max_lines + 1} more"
+ ]
texts[b["id"]] = lines
- blocks = [G.Block(id=b["id"], start=b["start"], end=b["end"],
- succs=[(d, k) for d, k in b["succs"]])
- for b in rec["blocks"]]
+ blocks = [
+ G.Block(
+ id=b["id"],
+ start=b["start"],
+ end=b["end"],
+ succs=[(d, k) for d, k in b["succs"]],
+ )
+ for b in rec["blocks"]
+ ]
def sizer(b: G.Block) -> tuple[int, int]:
lines = texts[b.id]
@@ -57,7 +68,8 @@ def render(lay: G.Layout, texts: dict[int, list[str]], color: bool) -> str:
for row in range(lay.height):
cells: dict[int, tuple[str, str]] = {}
for col, (ch, kind, _eid) in lay.painting.cells_at_row(
- row, 0, lay.width).items():
+ row, 0, lay.width
+ ).items():
cells[col] = (ch, COLOR.get(kind, ""))
for n in lay.nodes_at_row(row):
label = f"loc_{n.block.start:X}" if n.block else ""
@@ -107,41 +119,59 @@ def main() -> int:
ap.add_argument("corpus")
ap.add_argument("--func", help="function name (default: the smallest)")
ap.add_argument("--stats", action="store_true", help="lay out the whole corpus")
- ap.add_argument("--max-lines", type=int, default=8,
- help="collapse blocks longer than this (0 = never)")
+ ap.add_argument(
+ "--max-lines",
+ type=int,
+ default=8,
+ help="collapse blocks longer than this (0 = never)",
+ )
ap.add_argument("--no-color", action="store_true")
+ ap.add_argument(
+ "--engine",
+ choices=G.ENGINES,
+ default=None,
+ help="layout backend (default: $IDATUI_GRAPH_ENGINE or auto)",
+ )
args = ap.parse_args()
recs = json.load(open(args.corpus))
if args.stats:
- print(f"{'blocks':>7} {'nodes':>6} {'dummy':>6} {'layer':>6} "
- f"{'canvas':>12} {'ms':>8} name")
+ print(
+ f"{'blocks':>7} {'nodes':>6} {'dummy':>6} {'layer':>6} "
+ f"{'canvas':>12} {'ms':>8} name"
+ )
tot = 0.0
for r in sorted(recs, key=lambda r: len(r["blocks"])):
blocks, sizer, _ = build(r, args.max_lines)
- lay = G.layout(blocks, sizer)
+ lay = G.layout(blocks, sizer, engine=args.engine)
s = lay.stats
tot += s["ms"]
- print(f"{s['blocks']:>7} {s['nodes']:>6} {s['dummies']:>6} "
- f"{s['layers']:>6} {lay.width:>5}x{lay.height:<6} "
- f"{s['ms']:>8.1f} {r['name']}")
+ print(
+ f"{s['blocks']:>7} {s['nodes']:>6} {s['dummies']:>6} "
+ f"{s['layers']:>6} {lay.width:>5}x{lay.height:<6} "
+ f"{s['ms']:>8.1f} {r['name']}"
+ )
print(f"total {tot:.0f} ms over {len(recs)} functions")
return 0
if args.func:
rec = next((r for r in recs if r["name"] == args.func), None)
if rec is None:
- print("no such function; have: "
- f"{', '.join(r['name'] for r in recs[:20])}", file=sys.stderr)
+ print(
+ f"no such function; have: {', '.join(r['name'] for r in recs[:20])}",
+ file=sys.stderr,
+ )
return 1
else:
rec = min(recs, key=lambda r: len(r["blocks"]))
blocks, sizer, texts = build(rec, args.max_lines)
- lay = G.layout(blocks, sizer)
+ lay = G.layout(blocks, sizer, engine=args.engine)
print(render(lay, texts, color=not args.no_color))
- print(f"\n{rec['name']}: {lay.stats} canvas {lay.width}x{lay.height}",
- file=sys.stderr)
+ print(
+ f"\n{rec['name']}: {lay.stats} canvas {lay.width}x{lay.height}",
+ file=sys.stderr,
+ )
return 0
diff --git a/experiments/inproc_spike.py b/experiments/inproc_spike.py
index 469aa21..1b12204 100644
--- a/experiments/inproc_spike.py
+++ b/experiments/inproc_spike.py
@@ -25,6 +25,7 @@ Key facts this spike encodes (all verified):
another process, so the TUI's event loop never blocks. In-process, a slow
call (decompile ~150ms, analysis seconds) blocks the UI for its duration.
"""
+
from __future__ import annotations
import argparse
@@ -60,13 +61,21 @@ class DirectBackend:
def __init__(self, path: str) -> None:
import idapro
+
idapro.enable_console_messages(False)
t = time.time()
rc = idapro.open_database(path, run_auto_analysis=True)
self.open_secs = time.time() - t
if rc:
raise RuntimeError(f"open_database({path!r}) failed rc={rc}")
- import ida_bytes, ida_funcs, ida_hexrays, ida_name, idaapi, idautils, idc
+ import ida_bytes
+ import ida_funcs
+ import ida_hexrays
+ import ida_name
+ import idaapi
+ import idautils
+ import idc
+
self._idapro = idapro
self.idaapi, self.idautils, self.idc = idaapi, idautils, idc
self.ida_bytes, self.ida_hexrays = ida_bytes, ida_hexrays
@@ -108,6 +117,7 @@ class McpBackend:
if REPO not in sys.path: # idalib's init can reset sys.path out from under us
sys.path.insert(0, REPO)
from idatui.client import IDAClient
+
self.IDAClient = IDAClient
self.c = IDAClient(url, db=db)
self.c.connect()
@@ -150,8 +160,12 @@ class McpBackend:
return r.get("code", "") if isinstance(r, dict) else str(r)
def xrefs_to(self, ea):
- r = self.c.call("xref_query", queries=[{"addr": hex(ea), "direction": "to",
- "include_fn": True, "count": 2000}])
+ r = self.c.call(
+ "xref_query",
+ queries=[
+ {"addr": hex(ea), "direction": "to", "include_fn": True, "count": 2000}
+ ],
+ )
res = r.get("result", []) if isinstance(r, dict) else []
refs = res[0].get("refs", []) if res and isinstance(res[0], dict) else []
return [int(x["frm"], 16) for x in refs if x.get("frm")]
@@ -200,14 +214,21 @@ def _recv(sock):
return _unpack(_recvn(sock, n))
-_WORKER_OPS = ("functions", "resolve", "read_bytes", "disasm_line",
- "decompile", "xrefs_to")
+_WORKER_OPS = (
+ "functions",
+ "resolve",
+ "read_bytes",
+ "disasm_line",
+ "decompile",
+ "xrefs_to",
+)
def _worker_main(sockpath: str, dbpath: str) -> None:
"""Runs in a child process. Opens idalib on ITS main thread (constraint
satisfied), then serves one client serially over a unix socket."""
import socket as sk
+
direct = DirectBackend(dbpath)
ops = {name: getattr(direct, name) for name in _WORKER_OPS}
try:
@@ -237,10 +258,17 @@ class UnixWorkerBackend:
def __init__(self, dbpath: str) -> None:
import socket as sk
+
self.sockpath = f"/tmp/inproc_spike_{os.getpid()}.sock"
self.proc = __import__("subprocess").Popen(
- [sys.executable, os.path.abspath(__file__),
- "--worker", self.sockpath, dbpath])
+ [
+ sys.executable,
+ os.path.abspath(__file__),
+ "--worker",
+ self.sockpath,
+ dbpath,
+ ]
+ )
deadline = time.time() + 120
self.sock = None
while time.time() < deadline:
@@ -263,12 +291,23 @@ class UnixWorkerBackend:
raise RuntimeError(val)
return val
- def functions(self): return self._call("functions")
- def resolve(self, name): return self._call("resolve", name)
- def read_bytes(self, ea, n): return self._call("read_bytes", ea, n)
- def disasm_line(self, ea): return self._call("disasm_line", ea)
- def decompile(self, ea): return self._call("decompile", ea)
- def xrefs_to(self, ea): return self._call("xrefs_to", ea)
+ def functions(self):
+ return self._call("functions")
+
+ def resolve(self, name):
+ return self._call("resolve", name)
+
+ def read_bytes(self, ea, n):
+ return self._call("read_bytes", ea, n)
+
+ def disasm_line(self, ea):
+ return self._call("disasm_line", ea)
+
+ def decompile(self, ea):
+ return self._call("decompile", ea)
+
+ def xrefs_to(self, ea):
+ return self._call("xrefs_to", ea)
def close(self):
try:
@@ -321,6 +360,7 @@ def bench(target: str, n: int) -> None:
try:
import socket
+
socket.create_connection(("127.0.0.1", 8745), 0.3).close()
backends["mcp"] = McpBackend()
print("benching the running mcp server on :8745 too", flush=True)
@@ -338,15 +378,15 @@ def bench(target: str, n: int) -> None:
return sample[_nx[0]]
ops = {
- "resolve(main)": (lambda b: b.resolve("main"), n),
- "read_bytes(16)": (lambda b: b.read_bytes(main, 16), n),
- "read_bytes(4096)": (lambda b: b.read_bytes(main, 4096), n),
- "disasm_line": (lambda b: b.disasm_line(main), n),
- "xrefs_to": (lambda b: b.xrefs_to(_next_sample()), min(n, 200)),
- "decompile(cached)": (lambda b: b.decompile(main), min(n, 40)),
+ "resolve(main)": (lambda b: b.resolve("main"), n),
+ "read_bytes(16)": (lambda b: b.read_bytes(main, 16), n),
+ "read_bytes(4096)": (lambda b: b.read_bytes(main, 4096), n),
+ "disasm_line": (lambda b: b.disasm_line(main), n),
+ "xrefs_to": (lambda b: b.xrefs_to(_next_sample()), min(n, 200)),
+ "decompile(cached)": (lambda b: b.decompile(main), min(n, 40)),
}
names = list(backends)
- hdr = f"{'op':20}" + "".join(f"{nm+' us':>14}" for nm in names)
+ hdr = f"{'op':20}" + "".join(f"{nm + ' us':>14}" for nm in names)
print(hdr)
print("-" * len(hdr))
results: dict[str, dict[str, float]] = {nm: {} for nm in names}
@@ -371,7 +411,7 @@ def bench(target: str, n: int) -> None:
if nm == "mcp":
continue
v = results[nm].get(label, float("nan"))
- parts.append(f"{nm} {m/v:.0f}x" if v == v and v else f"{nm} -")
+ parts.append(f"{nm} {m / v:.0f}x" if v == v and v else f"{nm} -")
print(f" {label:20} {' '.join(parts)}")
for b in backends.values():
@@ -408,10 +448,12 @@ def _build_spike_app(backend: "DirectBackend"):
for ea, nm in backend.functions():
ol.add_option(Option(f"{ea:08x} {nm}", id=str(ea)))
yield ol
- yield Static("select a function, press F5 to decompile inline",
- id="code")
- yield Static("in-process idalib — every call runs on the UI thread",
- id="status")
+ yield Static(
+ "select a function, press F5 to decompile inline", id="code"
+ )
+ yield Static(
+ "in-process idalib — every call runs on the UI thread", id="status"
+ )
yield Footer()
def on_mount(self):
@@ -428,13 +470,13 @@ def _build_spike_app(backend: "DirectBackend"):
if ea is None:
return
t = time.perf_counter()
- code = backend.decompile(ea) # <-- BLOCKS the event loop
+ code = backend.decompile(ea) # <-- BLOCKS the event loop
dt = (time.perf_counter() - t) * 1e3
- self.query_one("#code", Static).update(
- "\n".join(code.splitlines()[:40]))
+ self.query_one("#code", Static).update("\n".join(code.splitlines()[:40]))
self.query_one("#status", Static).update(
f"decompiled {backend.idc.get_func_name(ea)} in {dt:.0f} ms "
- f"(UI was frozen for those {dt:.0f} ms)")
+ f"(UI was frozen for those {dt:.0f} ms)"
+ )
def action_decompile_all(self):
funcs = backend.functions()
@@ -442,14 +484,15 @@ def _build_spike_app(backend: "DirectBackend"):
n = 0
for ea, _ in funcs:
try:
- backend.decompile(ea) # <-- long, uninterruptible freeze
+ backend.decompile(ea) # <-- long, uninterruptible freeze
n += 1
except Exception: # noqa: BLE001
pass
dt = (time.perf_counter() - t) * 1e3
self.query_one("#status", Static).update(
f"decompiled {n} funcs in {dt:.0f} ms — the whole UI was frozen "
- f"the entire time (no spinner, no input)")
+ f"the entire time (no spinner, no input)"
+ )
def action_bytes(self):
ea = self._sel_ea()
@@ -459,7 +502,8 @@ def _build_spike_app(backend: "DirectBackend"):
b = backend.read_bytes(ea, 64)
dt = (time.perf_counter() - t) * 1e6
self.query_one("#status", Static).update(
- f"read 64 bytes in {dt:.1f} us: {b[:16].hex()}…")
+ f"read 64 bytes in {dt:.1f} us: {b[:16].hex()}…"
+ )
return Spike()
@@ -472,7 +516,9 @@ def _open_copy(target: str) -> "DirectBackend":
os.remove(tmp + e)
except OSError:
pass
- print("opening in-process (blocks the terminal until analysis is done)…", flush=True)
+ print(
+ "opening in-process (blocks the terminal until analysis is done)…", flush=True
+ )
return DirectBackend(tmp) # MAIN THREAD open, before the event loop starts
diff --git a/experiments/modal_shot.py b/experiments/modal_shot.py
index 2e6cd7a..82f1e2f 100644
--- a/experiments/modal_shot.py
+++ b/experiments/modal_shot.py
@@ -11,6 +11,7 @@ IDA and no .i64 -- it runs under any python with textual, in about a second.
The app it boots is a bare `App` carrying `IdaTui.CSS` and the app theme, which
is exactly what the modals resolve their styles against.
"""
+
import asyncio
import os
import sys
@@ -18,16 +19,23 @@ import sys
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, REPO)
-from textual.app import App, ComposeResult # noqa: E402
-from textual.widgets import Static # noqa: E402
+from textual.app import App, ComposeResult # noqa: E402
+from textual.widgets import Static # noqa: E402
-from idatui.app import ( # noqa: E402
- IDATUI_THEME, IdaTui, BusyScreen, ConfirmScreen, HelpScreen, LoadingScreen,
- QuitScreen, StructEditor, XrefsScreen,
+from idatui._sync import wait_for # noqa: E402
+from idatui.app import ( # noqa: E402
+ IDATUI_THEME,
+ BusyScreen,
+ ConfirmScreen,
+ HelpScreen,
+ IdaTui,
+ LoadingScreen,
+ QuitScreen,
+ StructEditor,
+ XrefsScreen,
)
-from idatui.domain import Struct # noqa: E402
-from idatui.rpc import screen_text # noqa: E402
-from idatui._sync import wait_for # noqa: E402
+from idatui.domain import Struct # noqa: E402
+from idatui.rpc import screen_text # noqa: E402
class _StubProgram:
@@ -36,14 +44,15 @@ class _StubProgram:
_STRUCTS = [
Struct(name="timespec", size=0x10, members=2, is_union=False, ordinal=1),
Struct(name="stat", size=0x90, members=15, is_union=False, ordinal=2),
- Struct(name="pthread_mutex_t", size=0x28, members=4, is_union=True,
- ordinal=3),
+ Struct(name="pthread_mutex_t", size=0x28, members=4, is_union=True, ordinal=3),
]
- _SRC = ("struct timespec\n"
- "{\n"
- " __time_t tv_sec; /* seconds */\n"
- " __syscall_slong_t tv_nsec;\n"
- "};\n")
+ _SRC = (
+ "struct timespec\n"
+ "{\n"
+ " __time_t tv_sec; /* seconds */\n"
+ " __syscall_slong_t tv_nsec;\n"
+ "};\n"
+ )
def list_structs(self):
return list(self._STRUCTS)
@@ -57,8 +66,7 @@ def _make(name: str):
if name == "structs":
return StructEditor(_StubProgram())
if name == "confirm":
- return ConfirmScreen("Delete struct 'timespec' ?",
- "This cannot be undone.")
+ return ConfirmScreen("Delete struct 'timespec' ?", "This cannot be undone.")
if name == "quit":
return QuitScreen(["echo", "libc.so.6"])
if name == "help":
@@ -68,10 +76,13 @@ def _make(name: str):
if name == "loading":
return LoadingScreen("echo", "opening database\u2026")
if name == "xrefs":
- return XrefsScreen(" xrefs to main", [
- (0x1234, " sub_2297+0x1c call main"),
- (0x5678, " _start+0x21 mov rdi, main"),
- ])
+ return XrefsScreen(
+ " xrefs to main",
+ [
+ (0x1234, " sub_2297+0x1c call main"),
+ (0x5678, " _start+0x21 mov rdi, main"),
+ ],
+ )
raise SystemExit(f"unknown modal {name!r}; --list to see them")
@@ -84,8 +95,9 @@ class _Shot(App):
def compose(self) -> ComposeResult:
# A little content underneath, so the modal's edge is visible against
# something rather than floating on an empty screen.
- yield Static("\n".join(" .... the view behind the dialog ...."
- for _ in range(60)))
+ yield Static(
+ "\n".join(" .... the view behind the dialog ...." for _ in range(60))
+ )
def on_mount(self) -> None:
self.register_theme(IDATUI_THEME)
@@ -108,12 +120,14 @@ async def main() -> None:
await pilot.pause()
if name == "structs":
await wait_for(
- lambda: bool(getattr(app.screen, "_structs", None)),
- pilot.pause, 5)
+ lambda: bool(getattr(app.screen, "_structs", None)), pilot.pause, 5
+ )
app.screen.on_option_list_option_selected(
- type("E", (), {"option_index": 0})())
- await wait_for(lambda: "{" in app.screen.query_one(
- "#se-edit").text, pilot.pause, 5)
+ type("E", (), {"option_index": 0})()
+ )
+ await wait_for(
+ lambda: "{" in app.screen.query_one("#se-edit").text, pilot.pause, 5
+ )
await pilot.pause()
print(f"\n=== {name} " + "=" * (cols - len(name) - 5))
print(screen_text(app)["text"])
diff --git a/experiments/opfmt_tools.py b/experiments/opfmt_tools.py
index 535688c..7654677 100644
--- a/experiments/opfmt_tools.py
+++ b/experiments/opfmt_tools.py
@@ -8,6 +8,7 @@ does to the database, what Hex-Rays will and won't render.
python3 experiments/opfmt_tools.py # 29 checks, ~40s
"""
+
import importlib.util
import os
import shutil
@@ -30,15 +31,21 @@ if os.path.exists(seed):
shutil.copy(seed, tmp + ".i64")
import idapro # noqa: E402
+
idapro.enable_console_messages(False)
assert idapro.open_database(tmp, run_auto_analysis=True) == 0
import ida_auto # noqa: E402
+
ida_auto.auto_wait()
-import ida_typeinf, idaapi, ida_bytes, ida_lines # noqa: E402,F401
+import ida_bytes
+import ida_lines
+import ida_typeinf # noqa: E402,F401
+import idaapi
spec = importlib.util.spec_from_file_location(
- "_patch", os.path.join(REPO, "server", "patch_server.py"))
+ "_patch", os.path.join(REPO, "server", "patch_server.py")
+)
patch = importlib.util.module_from_spec(spec)
spec.loader.exec_module(patch)
@@ -101,14 +108,25 @@ while e < ea + 0x400:
if target:
break
e = ida_bytes.next_head(e, ea + 0x800)
-print("target:", hex(target[0]), "n=", target[1], "value", hex(target[2]),
- "|", line(target[0]))
+print(
+ "target:",
+ hex(target[0]),
+ "n=",
+ target[1],
+ "value",
+ hex(target[2]),
+ "|",
+ line(target[0]),
+)
tea, tn, tv = target
r = op_format(addr=hex(tea), mode="show")
print(" show:", r)
-check("show reports the operand and its choices",
- r.get("n") == tn and "hex" in r.get("choices", []), str(r))
+check(
+ "show reports the operand and its choices",
+ r.get("n") == tn and "hex" in r.get("choices", []),
+ str(r),
+)
check("show doesn't change anything", r.get("applied") is False, str(r))
seen = []
@@ -116,15 +134,25 @@ for i in range(8):
r = op_format(addr=hex(tea), mode="cycle")
seen.append((r.get("format"), r.get("text")))
print(f" cycle -> {r.get('prev')} -> {r.get('format')}: {r.get('text')}")
-check("cycling returns to where it started",
- seen[0][0] == seen[len(r.get("choices", []))][0]
- if len(seen) > len(r.get("choices", [])) else True, str(seen))
-check("decimal renders differently from hex",
- any(s[1] != seen[0][1] for s in seen), str(seen))
+check(
+ "cycling returns to where it started",
+ seen[0][0] == seen[len(r.get("choices", []))][0]
+ if len(seen) > len(r.get("choices", []))
+ else True,
+ str(seen),
+)
+check(
+ "decimal renders differently from hex",
+ any(s[1] != seen[0][1] for s in seen),
+ str(seen),
+)
r = op_format(addr=hex(tea), mode="dec")
-check("explicit dec sticks", r.get("format") == "dec" and str(tv) in r.get("text", ""),
- str(r))
+check(
+ "explicit dec sticks",
+ r.get("format") == "dec" and str(tv) in r.get("text", ""),
+ str(r),
+)
r = op_format(addr=hex(tea), mode="back")
print(" back ->", r.get("format"), r.get("text"))
check("back steps the ring the other way", r.get("format") == "hex", str(r))
@@ -142,29 +170,55 @@ while e < ea + 0x800:
if off_target:
break
e = ida_bytes.next_head(e, ea + 0x800)
-print("target:", off_target and hex(off_target[0]), "|",
- off_target and line(off_target[0]))
+print(
+ "target:",
+ off_target and hex(off_target[0]),
+ "|",
+ off_target and line(off_target[0]),
+)
if off_target:
import ida_name # noqa: E402
+
oe, on = off_target
v, _w = NS["_idatui_op_value"](oe, on)
named = bool(ida_name.get_ea_name(v))
r = op_format(addr=hex(oe), n=on, mode="show")
- print(" ", hex(oe), "n=", on, "|", r.get("text"), r.get("choices"),
- "target named:", named)
- check("an unnamed target is not a cycle stop (it would invent a name)",
- ("offset" in r.get("choices", [])) == named, str(r))
+ print(
+ " ",
+ hex(oe),
+ "n=",
+ on,
+ "|",
+ r.get("text"),
+ r.get("choices"),
+ "target named:",
+ named,
+ )
+ check(
+ "an unnamed target is not a cycle stop (it would invent a name)",
+ ("offset" in r.get("choices", [])) == named,
+ str(r),
+ )
r = op_format(addr=hex(oe), n=on, mode="offset")
print(" offset ->", r.get("text"))
- check("but asking explicitly makes the reference",
- r.get("format") == "offset" and "offset" in r.get("text", ""), str(r))
+ check(
+ "but asking explicitly makes the reference",
+ r.get("format") == "offset" and "offset" in r.get("text", ""),
+ str(r),
+ )
r = op_format(addr=hex(oe), n=on, mode="show")
- check("and from then on the ring includes it",
- "offset" in r.get("choices", []), str(r))
+ check(
+ "and from then on the ring includes it",
+ "offset" in r.get("choices", []),
+ str(r),
+ )
r = op_format(addr=hex(oe), n=on, mode="hex")
print(" hex ->", r.get("text"))
- check("and back to a number", r.get("format") == "hex"
- and "offset" not in r.get("text", ""), str(r))
+ check(
+ "and back to a number",
+ r.get("format") == "hex" and "offset" not in r.get("text", ""),
+ str(r),
+ )
op_format(addr=hex(oe), n=on, mode="default")
else:
print(" (no literal-that-is-an-address in this function)")
@@ -175,22 +229,32 @@ spans = NS["_idatui_op_spans"](tea, txt)
print(" text:", repr(txt), "spans:", spans)
if len(spans) >= 2:
r = op_format(addr=hex(tea), col=spans[0][0], mode="show")
- check("a column inside operand 0 picks operand 0 (or the first literal)",
- r.get("n") in (spans[0][2], NS["_idatui_op_candidates"](tea)[0]), str(r))
+ check(
+ "a column inside operand 0 picks operand 0 (or the first literal)",
+ r.get("n") in (spans[0][2], NS["_idatui_op_candidates"](tea)[0]),
+ str(r),
+ )
r = op_format(addr=hex(tea), col=spans[-1][0], mode="show")
- check("a column inside the last operand picks it", r.get("n") == spans[-1][2],
- str(r))
+ check(
+ "a column inside the last operand picks it", r.get("n") == spans[-1][2], str(r)
+ )
print("\n=== listing: an unmapped value refuses to become an offset ===")
r = op_format(addr=hex(tea), n=tn, mode="offset")
print(" ", r.get("error") or r)
-check("offset on a non-address is refused, not invented",
- bool(r.get("error")) or ida_bytes.is_mapped(tv), str(r))
+check(
+ "offset on a non-address is refused, not invented",
+ bool(r.get("error")) or ida_bytes.is_mapped(tv),
+ str(r),
+)
print("\n=== listing: char is only offered when it renders as one ===")
r = op_format(addr=hex(tea), n=tn, mode="show")
-check("0x%x isn't offered as a char" % tv,
- ("char" in r["choices"]) == NS["_idatui_printable"](tv), str(r))
+check(
+ "0x%x isn't offered as a char" % tv,
+ ("char" in r["choices"]) == NS["_idatui_printable"](tv),
+ str(r),
+)
print("\n=== listing: a stack variable can be cycled AND put back ===")
stk = None
@@ -208,23 +272,32 @@ if stk:
orig = line(se)
r = op_format(addr=hex(se), n=sn, mode="cycle")
print(" cycle ->", r.get("format"), r.get("text"), "|", r.get("warn"))
- check("leaving a stack variable says so, and how to undo it",
- "stack" in (r.get("warn") or ""), str(r))
- ring = [op_format(addr=hex(se), n=sn, mode="cycle")
- for _ in range(len(r["choices"]))]
+ check(
+ "leaving a stack variable says so, and how to undo it",
+ "stack" in (r.get("warn") or ""),
+ str(r),
+ )
+ ring = [
+ op_format(addr=hex(se), n=sn, mode="cycle") for _ in range(len(r["choices"]))
+ ]
print(" ring:", [(x["format"], x["text"]) for x in ring])
- check("the ring is the same at every step (a lap comes home)",
- [x["format"] for x in ring] == r["choices"][1:] + r["choices"][:1],
- f"{[x['format'] for x in ring]} vs {r['choices']}")
+ check(
+ "the ring is the same at every step (a lap comes home)",
+ [x["format"] for x in ring] == r["choices"][1:] + r["choices"][:1],
+ f"{[x['format'] for x in ring]} vs {r['choices']}",
+ )
r = op_format(addr=hex(se), n=sn, mode="stack")
- check("'stack' puts the frame variable back",
- r.get("format") == "stack" and r.get("text") == orig,
- f"{r.get('text')!r} want {orig!r}")
+ check(
+ "'stack' puts the frame variable back",
+ r.get("format") == "stack" and r.get("text") == orig,
+ f"{r.get('text')!r} want {orig!r}",
+ )
else:
print(" (no stack-variable operand found)")
print("\n=== data item ===")
import ida_segment # noqa: E402
+
seg = ida_segment.get_segm_by_name(".data")
if seg:
de = seg.start_ea
@@ -257,8 +330,11 @@ while e < seg.end_ea and ue is None:
if ue is not None:
r = op_format(addr=hex(ue), mode="cycle")
print(" ", hex(ue), "->", r.get("error"))
- check("undefined bytes are refused with the fix, not a silent no-op",
- "define" in (r.get("error") or ""), str(r))
+ check(
+ "undefined bytes are refused with the fix, not a silent no-op",
+ "define" in (r.get("error") or ""),
+ str(r),
+ )
else:
print(" (no undefined bytes)")
@@ -268,6 +344,7 @@ print(" show without a line:", r.get("error"))
check("no line is an error, not a guess", bool(r.get("error")), str(r))
import ida_hexrays # noqa: E402
+
cf = ida_hexrays.decompile(parse_address("main"))
sv = cf.get_pseudocode()
pcline = None
@@ -276,8 +353,13 @@ for i in range(len(sv)):
if nums and nums[0]["value"] > 9:
pcline = (i, nums[0])
break
-print(" line", pcline[0], repr(ida_lines.tag_remove(sv[pcline[0]].line).strip()),
- "num:", pcline[1])
+print(
+ " line",
+ pcline[0],
+ repr(ida_lines.tag_remove(sv[pcline[0]].line).strip()),
+ "num:",
+ pcline[1],
+)
i, num = pcline
r = pc_num_format(addr="main", line=i, mode="show")
check("show finds the literal", r.get("ea") == hex(num["ea"]), str(r))
@@ -293,8 +375,11 @@ print(" bin ->", r.get("error"))
check("binary is refused with a reason", bool(r.get("error")), str(r))
r = pc_num_format(addr="main", line=i, mode="default")
print(" default ->", r.get("text"))
-check("default restores Hex-Rays' own choice",
- r.get("text") == ida_lines.tag_remove(sv[i].line).strip(), str(r))
+check(
+ "default restores Hex-Rays' own choice",
+ r.get("text") == ida_lines.tag_remove(sv[i].line).strip(),
+ str(r),
+)
r = pc_num_format(addr="main", line=i, mode="show")
check("and the format reads back as default", r.get("format") == "default", str(r))
print("\n=== pseudocode: the ring visits every stop ===")
@@ -303,10 +388,16 @@ for _ in range(len(r.get("choices", [])) * 2):
rr = pc_num_format(addr="main", line=i, mode="cycle")
ring.append((rr.get("format"), rr.get("text")))
print(" ", [x[0] for x in ring])
-check("every stop in the ring is reached",
- set(x[0] for x in ring) == set(r["choices"]), f"{ring} vs {r['choices']}")
-check("the ring's renderings are distinct",
- len({x[1] for x in ring}) >= len(r["choices"]) - 1, str(ring))
+check(
+ "every stop in the ring is reached",
+ set(x[0] for x in ring) == set(r["choices"]),
+ f"{ring} vs {r['choices']}",
+)
+check(
+ "the ring's renderings are distinct",
+ len({x[1] for x in ring}) >= len(r["choices"]) - 1,
+ str(ring),
+)
pc_num_format(addr="main", line=i, mode="default")
print("\n=== pseudocode: col picks the literal ===")
@@ -327,9 +418,12 @@ if multi:
col = len(NS["_idatui_compact"](plain[:x]).rstrip()) if x else 0
r = pc_num_format(addr="main", line=k, col=col, mode="show")
print(" col", col, "->", r.get("ea"), r.get("value"))
- check("a column selects the number under it",
- r.get("value") == hex(nums[1]["value"])
- or r.get("value") == hex(nums[0]["value"]), str(r))
+ check(
+ "a column selects the number under it",
+ r.get("value") == hex(nums[1]["value"])
+ or r.get("value") == hex(nums[0]["value"]),
+ str(r),
+ )
else:
print(" (no line with two literals)")
@@ -342,8 +436,11 @@ if row and row.get("ops"):
t = row["text"]
for lo, hi, n in row["ops"]:
print(f" op{n}: {t[lo:hi]!r}")
- check("the extents index the row's own text",
- all(t[lo:hi].strip() for lo, hi, n in row["ops"]), str(row["ops"]))
+ check(
+ "the extents index the row's own text",
+ all(t[lo:hi].strip() for lo, hi, n in row["ops"]),
+ str(row["ops"]),
+ )
# and they agree with what op_format picks for a column inside them
ok = True
for lo, hi, n in row["ops"]:
@@ -361,8 +458,11 @@ if regop:
lo, hi, n = regop
r = op_format(addr=hex(tea), col=(lo + hi) // 2, mode="cycle")
print(" ", repr(row["text"][lo:hi]), "->", r.get("error"))
- check("it names the operand and the one that CAN change",
- bool(r.get("error")) and "operand" in r["error"], str(r))
+ check(
+ "it names the operand and the one that CAN change",
+ bool(r.get("error")) and "operand" in r["error"],
+ str(r),
+ )
else:
print(" (this instruction has no register-only operand)")
@@ -376,15 +476,24 @@ for rec in pn["nums"]:
two = next((v for v in multi2.values() if len(v) >= 2), None)
if two:
import ida_hexrays as _hx
+
cf2 = _hx.decompile(parse_address("main"))
- disp = NS["_idatui_compact"](ida_lines.tag_remove(cf2.get_pseudocode()[two[0]["line"]].line))
+ disp = NS["_idatui_compact"](
+ ida_lines.tag_remove(cf2.get_pseudocode()[two[0]["line"]].line)
+ )
print(" line:", repr(disp.strip()))
for rec in two:
- print(f" x{rec['x0']}..{rec['x1']} = {disp[rec['x0']:rec['x1']]!r} value {rec['value']}")
- check("spans land on the literals in the DISPLAYED text",
- all(disp[r0["x0"]:r0["x1"]].strip() for r0 in two), str(two))
- check("distinct literals get distinct spans",
- two[0]["x0"] != two[1]["x0"], str(two))
+ print(
+ f" x{rec['x0']}..{rec['x1']} = {disp[rec['x0'] : rec['x1']]!r} value {rec['value']}"
+ )
+ check(
+ "spans land on the literals in the DISPLAYED text",
+ all(disp[r0["x0"] : r0["x1"]].strip() for r0 in two),
+ str(two),
+ )
+ check(
+ "distinct literals get distinct spans", two[0]["x0"] != two[1]["x0"], str(two)
+ )
print(f"\n{OK} passed, {FAIL} failed")
idapro.close_database(save=False)
diff --git a/experiments/profile_client.py b/experiments/profile_client.py
new file mode 100644
index 0000000..b03ed46
--- /dev/null
+++ b/experiments/profile_client.py
@@ -0,0 +1,76 @@
+"""Profile the CLIENT half of streaming a segment.
+
+`profile_remote.py` profiles inside the database process. This one profiles the
+other side: unpickling a page, building Heads and maintaining the model's
+indexes. Once the backend got cheap that half became the majority of boot, and
+nothing else here can see it.
+
+ PYTHONPATH=. ~/ida-venv/bin/python experiments/profile_client.py [BINARY] [--pages N]
+
+Time spent in `invoke` is the backend + transport; everything below it in the
+`tottime` list is ours and is what this file is for.
+"""
+
+from __future__ import annotations
+
+import argparse
+import cProfile
+import io
+import os
+import pstats
+import time
+
+from idatui import remote_ops
+from idatui.domain import Program
+from idatui.nexus_client import NexusClient
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser()
+ ap.add_argument("binary", nargs="?", default="targets/bash")
+ ap.add_argument("--pages", type=int, default=60)
+ ap.add_argument("--lines", type=int, default=16)
+ ap.add_argument(
+ "--text", action="store_true", help="load full pages instead of skeletons"
+ )
+ args = ap.parse_args()
+
+ client = NexusClient(os.path.abspath(args.binary))
+ client.connect()
+ program = Program(client)
+ regions = client.call(remote_ops.file_regions)
+ rows = regions.get("regions") or regions.get("result") or []
+ text_seg = next((r for r in rows if ".text" in str(r.get("name", ""))), rows[0])
+ model = program.listing(int(str(text_seg["start"]), 16))
+ assert model is not None
+ model.load_next_page() # prime one page, and install the remote lib
+
+ want_text = bool(args.text)
+ pr = cProfile.Profile()
+ started = time.perf_counter()
+ pr.enable()
+ loaded = 0
+ for _ in range(args.pages):
+ if model.complete:
+ break
+ n = model.load_next_page(text=want_text)
+ if n == 0:
+ break
+ loaded += 1
+ pr.disable()
+ wall = (time.perf_counter() - started) * 1000
+
+ print(
+ f"# {os.path.basename(args.binary)} pages={loaded} "
+ f"text={want_text} {wall:.0f}ms ({wall / max(loaded, 1):.2f}ms/page)"
+ )
+ buf = io.StringIO()
+ pstats.Stats(pr, stream=buf).sort_stats("tottime").print_stats(args.lines)
+ print(buf.getvalue())
+ program.close()
+ client.close()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/experiments/profile_remote.py b/experiments/profile_remote.py
new file mode 100644
index 0000000..fbf17dc
--- /dev/null
+++ b/experiments/profile_remote.py
@@ -0,0 +1,73 @@
+"""Profile persistent ida-tui operations inside the IDA process.
+
+The profiler itself is a typed ``RemoteModule`` function in ``remote_tools.py``;
+this file contains no generated Python source or knowledge of remote module names.
+
+Usage::
+
+ PYTHONPATH=. python experiments/profile_remote.py [BINARY]
+ PYTHONPATH=. python experiments/profile_remote.py --op decompile
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+
+from idatui import remote_ops
+from idatui.nexus_client import NexusClient
+
+CALLS = {
+ "heads": ("heads", {"count": 500, "annotate": True}),
+ "heads_plain": ("heads", {"count": 500, "annotate": False}),
+ "heads_skeleton": (
+ "heads",
+ {"count": 500, "annotate": True, "text": False},
+ ),
+ "decompile": ("decompile", {}),
+}
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("binary", nargs="?", default="targets/bash")
+ parser.add_argument("--op", default="heads", choices=sorted(CALLS))
+ parser.add_argument("--reps", type=int, default=20)
+ parser.add_argument("--addr", default=None, help="default: the .text start")
+ args = parser.parse_args()
+
+ client = NexusClient(os.path.abspath(args.binary)).connect()
+ addr = args.addr
+ if addr is None:
+ regions = client.call(remote_ops.file_regions)
+ rows = regions.get("regions") or regions.get("result") or []
+ text = next(
+ (row for row in rows if ".text" in str(row.get("name", ""))),
+ None,
+ )
+ addr = (text or rows[0])["start"] if rows else "0x0"
+
+ operation, call_args = CALLS[args.op]
+ call_args = {"addr": addr, **call_args}
+ print(
+ f"# {os.path.basename(args.binary)} op={args.op} "
+ f"addr={addr} reps={args.reps}"
+ )
+
+ # Install the persistent tool module and warm its caches before profiling.
+ client.call(remote_ops.heads, addr=addr, count=500, annotate=True)
+ out = client.call(
+ remote_ops.profile_remote,
+ operation=operation,
+ args=call_args,
+ reps=args.reps,
+ )
+ per = out["total"] / out["reps"] * 1000
+ print(f"# {out['total'] * 1000:.0f}ms total, {per:.1f}ms per call\n")
+ print(out["stats"])
+ client.close()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/experiments/splash_place_count.py b/experiments/splash_place_count.py
new file mode 100644
index 0000000..4490b9e
--- /dev/null
+++ b/experiments/splash_place_count.py
@@ -0,0 +1,102 @@
+#!/usr/bin/env python
+"""Count what the splash actually SENDS to the terminal.
+
+Pushes a real LoadingScreen onto a bare Textual app with kitty graphics forced
+on and kittygfx._write captured, then drives it the way the app does (a status
+write per progress tick) and tallies the escapes.
+
+ PYTHONPATH=. ~/ida-venv/bin/python experiments/splash_place_count.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+import sys
+
+os.environ["IDATUI_KITTY"] = "1"
+
+from textual.app import App, ComposeResult # noqa: E402
+from textual.widgets import Static # noqa: E402
+
+from idatui import kittygfx # noqa: E402
+
+SENT: list[str] = []
+
+
+def _fake_write(data: str) -> bool:
+ SENT.append(data)
+ return True
+
+
+kittygfx._write = _fake_write # type: ignore[assignment]
+kittygfx._cell = (9, 22)
+
+from idatui.app import LoadingScreen # noqa: E402
+
+
+def tally() -> dict[str, int]:
+ blob = "".join(SENT)
+ return {
+ "uploads (a=t)": len(re.findall(r"\x1b_G[^;]*a=t", blob)),
+ "placements (a=p)": len(re.findall(r"\x1b_G[^;]*a=p", blob)),
+ "deletes (a=d)": len(re.findall(r"\x1b_G[^;]*a=d", blob)),
+ "bytes": len(blob),
+ }
+
+
+class Host(App):
+ CSS = "#loading-box { width: 70; height: auto; }"
+
+ def compose(self) -> ComposeResult:
+ yield Static("host")
+
+
+async def main() -> None:
+ ticks = int(sys.argv[1]) if len(sys.argv) > 1 else 40
+ app = Host()
+ async with app.run_test(size=(100, 45)) as pilot:
+ screen = LoadingScreen("target")
+ app.push_screen(screen)
+ await pilot.pause()
+ await pilot.pause()
+ print(f"image mode: {screen._image}")
+ after_mount = tally()
+ print("after mount:", after_mount)
+
+ # what the app does: _status() -> loading_screen.update_note(), once
+ # per progress write. Spread over ~4s of wall clock like a real load.
+ for i in range(ticks):
+ screen.update_note(f"analyzing… {i}")
+ await asyncio.sleep(0.1)
+ await pilot.pause()
+ print(f"after {ticks} progress notes:", tally())
+
+ screen.dismiss()
+ await pilot.pause()
+ print("after dismiss:", tally())
+
+ d = tally()
+ blob = "".join(SENT)
+ cmds = re.findall(r"\x1b_G([^;\x1b]*)", blob)
+ ids = {
+ dict(kv.split("=", 1) for kv in c.split(",") if "=" in kv).get("p")
+ for c in cmds
+ if "a=p" in c.split(",")
+ }
+ onscreen = "unbounded (anonymous)" if None in ids else len(ids)
+ print()
+ print(
+ f"=> {d['placements (a=p)']} place escapes sent, {d['deletes (a=d)']} deletes"
+ )
+ print(f" images actually on screen: {onscreen}")
+ print(" A placement is identified by (image id, placement id). An a=p with")
+ print(" no p= key is ANONYMOUS and stacks a fresh copy every time; with a")
+ print(" p= key the terminal replaces the previous one. logo.png is RGBA, so")
+ print(" stacking also composites its soft edges towards solid.")
+ m = re.search(r"\x1b_G(a=p[^;\x1b]*)", blob)
+ print(" placement escape:", m.group(1) if m else "(none)")
+
+
+asyncio.run(main())
diff --git a/experiments/worker_smoke.py b/experiments/worker_smoke.py
index 9b55138..08799a4 100644
--- a/experiments/worker_smoke.py
+++ b/experiments/worker_smoke.py
@@ -1,23 +1,26 @@
-"""Exercise the real domain.Program through an IDA Code Mode lease.
+"""Exercise the real domain.Program through an IDA Nexus lease.
-A matching registered GUI is reused; otherwise Code Mode starts a managed
+A matching registered GUI is reused; otherwise IDA Nexus starts a managed
idalib worker. Usage: ``uv run python experiments/worker_smoke.py FILE``.
"""
+
from __future__ import annotations
import os
import sys
import time
-from idatui.codemode_client import CodeModeClient
from idatui.domain import Program
+from idatui.nexus_client import NexusClient
def main() -> int:
- target = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "experiments/fibonacci.elf")
- print(f"attaching Code Mode to {target}…", flush=True)
+ target = os.path.abspath(
+ sys.argv[1] if len(sys.argv) > 1 else "experiments/fibonacci.elf"
+ )
+ print(f"attaching IDA Nexus to {target}…", flush=True)
started = time.time()
- client = CodeModeClient(target)
+ client = NexusClient(target)
client.connect(progress=lambda message: print(f" {message}", flush=True))
print(
f" ready in {time.time() - started:.2f}s; backend={client.backend}; "
generated by cgit v1.3.1-sl0p (git 2.54.0) at 2026-08-27 18:03:06 +0000 — sl0p fork