aboutsummaryrefslogtreecommitdiffstats
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/test_scenarios.py75
-rw-r--r--tests/test_trace.py203
-rw-r--r--tests/test_trace_rpc.py369
-rw-r--r--tests/test_trace_ui.py395
-rw-r--r--tests/test_trace_vs_tenet.py189
5 files changed, 1231 insertions, 0 deletions
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py
index 53dad91..9d77680 100644
--- a/tests/test_scenarios.py
+++ b/tests/test_scenarios.py
@@ -20,7 +20,9 @@ import asyncio
import fnmatch
import os
import re
+import shutil
import sys
+import tempfile
import traceback
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -442,6 +444,37 @@ async def s_asm_highlight(c: Ctx):
mn is not None, f"{code[0].spans if code else None}")
+@scenario("status_names_the_file")
+async def s_status_names_the_file(c: Ctx):
+ """The status bar always says which file you're looking at.
+
+ Obvious once several panes and a project switcher exist: with two idatui
+ windows open, or after switching binaries, "0x2490" alone doesn't say what
+ it belongs to.
+ """
+ from textual.widgets import Static
+ app = c.app
+ await c.open_biggest("listing")
+ await c.pause(0.3)
+ name = os.path.basename(app._open_path)
+ status = str(app.query_one("#status", Static).render())
+ c.check("the status bar names the open file",
+ status.startswith(f"[{name}]"), f"{status[:60]!r} (want [{name}])")
+
+ # It must survive the messages that WRITE the status, not just the idle one.
+ c.lst.focus()
+ await c.press("down")
+ await c.pause(0.3)
+ status = str(app.query_one("#status", Static).render())
+ c.check("and keeps naming it as you move",
+ status.startswith(f"[{name}]"), status[:60])
+
+ # The function-count message used to include the module name itself, which
+ # would now read "[echo] echo — 128 functions".
+ c.check("without saying the name twice",
+ status.count(name) == 1, status[:70])
+
+
@scenario("command_palette")
async def s_command_palette(c: Ctx):
app = c.app
@@ -2270,7 +2303,49 @@ async def s_func_banners(c: Ctx):
# --------------------------------------------------------------------------- #
# Runner
# --------------------------------------------------------------------------- #
+async def _build_pristine(binary, cache):
+ """Analyse ``binary`` once and keep the resulting database as a golden copy.
+
+ Costs one full analysis, then every later run starts from it instead of
+ re-analysing.
+ """
+ print(f" (building pristine database for {os.path.basename(binary)}\u2026)")
+ app = IdaTui(open_path=binary, keepalive=False)
+ async with app.run_test(size=(140, 44)) as pilot:
+ for _ in range(6000):
+ await pilot.pause(0.05)
+ if app._func_index is not None and app._func_index.complete:
+ break
+ app.program.client.call("idb_save", timeout=600.0)
+ db = binary + ".i64"
+ if os.path.exists(db):
+ shutil.copy2(db, cache)
+
+
async def run(binary, only=None):
+ # The suite EDITS the database — it defines code, undefines items, renames
+ # and comments — and IDA saves those edits. Run that against the tracked
+ # target and every run inherits the last one's damage: decomp_follow_self
+ # started failing with no code change because an earlier scenario had
+ # undefined an instruction, and a position check looked flaky one run in
+ # three for the same reason. A suite whose result depends on its own history
+ # can't be trusted to accuse the code.
+ #
+ # So: work on a scratch copy, seeded from a golden database that nothing
+ # ever writes back to.
+ with tempfile.TemporaryDirectory(prefix="idatui-pilot-") as scratch:
+ target = os.path.join(scratch, os.path.basename(binary))
+ shutil.copy2(binary, target)
+ cache = binary + ".pristine.i64"
+ if not (os.path.exists(cache)
+ and os.path.getmtime(cache) >= os.path.getmtime(binary)):
+ await _build_pristine(target, cache)
+ if os.path.exists(cache):
+ shutil.copy2(cache, target + ".i64")
+ await _run_on(target, only)
+
+
+async def _run_on(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:
diff --git a/tests/test_trace.py b/tests/test_trace.py
new file mode 100644
index 0000000..58c9d81
--- /dev/null
+++ b/tests/test_trace.py
@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+"""Trace model: parsing, queries, and lining a trace up with the database.
+
+Pure stdlib — no IDA, no trace files needed. The differential test
+(test_trace_vs_tenet.py) checks our answers against Tenet's on real traces;
+this one covers what the reference can't arbitrate: the set-shaped queries
+painting needs, and rebasing.
+"""
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from idatui.trace import Trace # noqa: E402
+
+PASS = FAIL = 0
+
+GPR = ["rax", "rbx", "rcx", "rdx", "rbp", "rsp", "rsi", "rdi"]
+FULL = ",".join(f"{r}=0x{0x1000 if r == 'rsp' else 0:x}" for r in GPR)
+
+# push rbp / mov rbp,rsp / mov [rbp-4],0x2a / mov eax,[rbp-4] / pop rbp / jmp back
+LINES = [
+ FULL + ",rip=0x401000",
+ "rsp=0xff8,rip=0x401001,mw=0xff8:0000000000000000",
+ "rbp=0xff8,rip=0x401004",
+ "rip=0x40100b,mw=0xff4:2a000000",
+ "rax=0x2a,rip=0x40100e,mr=0xff4:2a000000",
+ "rbp=0x0,rsp=0x1000,rip=0x401010,mr=0xff8:0000000000000000",
+ "rip=0x401000", # loop back: 0x401000 executes twice
+ "rip=0x401001",
+]
+
+
+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:
+ path = os.path.join(tmp, "t.0.log")
+ with open(path, "w") as f:
+ f.write("\n".join(LINES) + "\n")
+ with open(os.path.join(tmp, "t.info"), "w") as f:
+ f.write("arch=x86_64\nmode=user\nstart_code=0x401000\n")
+ t = Trace.load(path)
+
+ check("every line is one timestamp", t.length == len(LINES), f"{t.length}")
+ check("the sidecar .info is picked up",
+ t.info is not None and t.info.arch == "x86_64" and
+ t.info.start_code == 0x401000)
+ check("PC is tracked per timestamp",
+ [t.ip(i) for i in range(6)] ==
+ [0x401000, 0x401001, 0x401004, 0x40100b, 0x40100e, 0x401010])
+
+ # -- register reconstruction --------------------------------------- #
+ check("a register keeps its value until it changes",
+ t.register("rsp", 0) == 0x1000 and t.register("rsp", 1) == 0xff8
+ and t.register("rsp", 4) == 0xff8 and t.register("rsp", 5) == 0x1000)
+ check("the full first line seeds every register",
+ t.register("rbx", 3) == 0)
+ check("an unknown register is None, not 0",
+ t.register("r15", 0) is None)
+ check("changed() is what the INSTRUCTION did, not the state",
+ t.changed(4) == {"rax", "rip"}, f"{t.changed(4)}")
+
+ # "which instruction set this register?" — the question a trace exists
+ # to answer.
+ check("last_write finds the instruction that set a value",
+ t.last_write("rax", 5) == 4 and t.last_write("rbp", 4) == 2,
+ f"{t.last_write('rax', 5)}, {t.last_write('rbp', 4)}")
+ check("next_write looks forward",
+ t.next_write("rbp", 2) == 5 and t.next_write("rbp", 5) is None)
+
+ # -- memory --------------------------------------------------------- #
+ ops = t.memory_ops(3)
+ check("a write is captured with its bytes",
+ len(ops) == 1 and ops[0].write and ops[0].addr == 0xff4
+ and ops[0].data == bytes.fromhex("2a000000"), f"{ops}")
+ check("a read is captured and marked as a read",
+ [o.write for o in t.memory_ops(4)] == [False])
+ check("an instruction with no memory has none", t.memory_ops(2) == [])
+
+ # -- memory state at a timestamp ------------------------------------ #
+ # The trace wrote 2a000000 at 0xff4 (t=3) and read it back (t=4); it
+ # pushed/popped 8 zero bytes at 0xff8 (t=1 write, t=5 read).
+ d, k = t.memory(0xff4, 4, 3)
+ check("memory reflects a write as of that timestamp",
+ d == bytes.fromhex("2a000000") and k == b"\x01" * 4, f"{d.hex()} {k.hex()}")
+ d, k = t.memory(0xff4, 4, 2)
+ check("and does NOT reflect it before the write happened",
+ k == b"\x00" * 4, f"{d.hex()} known={k.hex()}")
+
+ # A trace only knows what it saw. A byte nobody touched is unknown, and
+ # must not be reported as zero — that distinction is the entire reason
+ # to read memory from a trace instead of from the database.
+ d, k = t.memory(0x5000, 4, t.length - 1)
+ check("untouched memory is unknown, not zero",
+ k == b"\x00" * 4 and d == b"\x00" * 4, f"known={k.hex()}")
+ # 0xff2..0xff3 was never touched; 0xff4..0xff7 came from the write at
+ # t=3 and 0xff8..0xff9 from the push at t=1 — a window can be knowable
+ # from several accesses at different times, which is what makes this
+ # worth a mask rather than a flag.
+ d, k = t.memory(0xff2, 8, 4)
+ check("a partially-covered window marks which bytes are known",
+ k == bytes([0, 0, 1, 1, 1, 1, 1, 1]), f"known={k.hex()}")
+
+ # Reads are evidence too: an instruction reading a byte reveals what it
+ # held at that moment.
+ d, k = t.memory(0xff8, 8, 5)
+ check("a read reveals memory contents",
+ k == b"\x01" * 8, f"known={k.hex()}")
+
+ check("memory_writes lists only the writers",
+ t.memory_writes(0xff4, 4) == [3], f"{t.memory_writes(0xff4, 4)}")
+ check("memory_accesses includes the readers",
+ t.memory_accesses(0xff4, 4) == [3, 4], f"{t.memory_accesses(0xff4, 4)}")
+ check("a never-touched range has no accesses",
+ t.memory_accesses(0x5000, 16) == [])
+
+ # -- execution queries: what painting is built on -------------------- #
+ check("executions lists every timestamp for an address",
+ list(t.executions(0x401000)) == [0, 6], f"{list(t.executions(0x401000))}")
+ check("a never-executed address has none", not len(t.executions(0xdead)))
+ check("executions_between windows the result",
+ t.executions_between(0x401000, 1, 7) == [6])
+ check("next/prev execution step between hits",
+ t.next_execution(0x401000, 0) == 6
+ and t.prev_execution(0x401000, 6) == 0
+ and t.next_execution(0x401000, 6) is None)
+
+ # A pseudocode line covers MANY addresses, so the set form is the one
+ # decompiler painting will call — per-address lookups would mean one
+ # dict hit per instruction per repaint.
+ check("hits() counts a whole set of addresses at once",
+ t.hits([0x401000, 0x401001, 0x401004, 0xdead]) ==
+ {0x401000: 2, 0x401001: 2, 0x401004: 1},
+ f"{t.hits([0x401000, 0x401001, 0x401004, 0xdead])}")
+
+ # -- rebasing -------------------------------------------------------- #
+ # A traced process is relocated; nothing lines up until the slide is
+ # found. Page offsets survive relocation, which is what makes it
+ # findable.
+ db = [0x1000 + (a - 0x401000) for a in
+ (0x401000, 0x401001, 0x401004, 0x40100b, 0x40100e, 0x401010)]
+ slide = t.rebase(db)
+ check("the slide between trace and database is found",
+ slide == 0x1000 - 0x401000, f"{slide:#x}")
+ t.apply_slide(slide)
+ check("addresses come back in database terms",
+ t.ip(0) == 0x1000 and list(t.executions(0x1000)) == [0, 6],
+ f"{t.ip(0):#x}")
+ # Memory addresses are NOT slid: the slide relocates the image, and
+ # these are overwhelmingly stack/heap addresses with no database
+ # counterpart — sliding a stack pointer by the image delta produced a
+ # negative address in testing.
+ check("memory op addresses stay in trace space",
+ t.memory_ops(3)[0].addr == 0xff4, f"{t.memory_ops(3)[0].addr:#x}")
+ check("raw_ip still gives the traced address",
+ t.raw_ip(0) == 0x401000)
+
+ t2 = Trace.load(path)
+ # Page offsets that appear nowhere in the trace. (0xdead0000/0xdead0004
+ # would NOT do: they sit at the same offsets as two traced addresses and
+ # so legitimately agree on a slide — a reminder that this matches on
+ # offsets, not on addresses looking plausible.)
+ check("no match means no slide, not a wrong one",
+ t2.rebase([0xdead0555, 0xbeef0777]) == 0,
+ f"{t2.rebase([0xdead0555, 0xbeef0777]):#x}")
+ check("one lone agreeing address is not enough to claim a slide",
+ t2.rebase([0x1000]) == 0, f"{t2.rebase([0x1000]):#x}")
+ check("an empty database is harmless", t2.rebase([]) == 0)
+
+ # -- robustness ------------------------------------------------------ #
+ p2 = os.path.join(tmp, "odd.0.log")
+ with open(p2, "w") as f:
+ f.write("\n".join([
+ FULL + ",rip=0x401000",
+ "", # blank line
+ "rax=0xnothex,rip=0x401001", # unparseable value
+ "rbx=0x1", # no PC at all
+ "rip=0x401002,mw=0x10:zz", # unparseable memory
+ ]) + "\n")
+ t3 = Trace.load(p2)
+ check("a malformed trace loads instead of raising", t3.length == 4,
+ f"{t3.length}")
+ check("a line with no PC inherits the previous one",
+ t3.ip(2) == 0x401001, f"{t3.ip(2):#x}")
+ check("an unparseable memory entry is dropped, not fatal",
+ t3.memory_ops(3) == [])
+
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_trace_rpc.py b/tests/test_trace_rpc.py
new file mode 100644
index 0000000..1007385
--- /dev/null
+++ b/tests/test_trace_rpc.py
@@ -0,0 +1,369 @@
+#!/usr/bin/env python3
+"""Trace integration over the RPC socket — the path an agent actually takes.
+
+The UI tests (test_trace_ui.py) drive the trace via in-process keystrokes and
+the Textual pilot. This one spawns a real tmux pane with ``--trace`` and drives
+every trace verb through the RPC socket, validating the JSON responses an agent
+would see. It exercises the serialization, the settle/timeout machinery, and
+the response shape that a driver depends on.
+
+Requires: tmux, IDA (idalib), and a trace. Records a fresh trace with the QEMU
+tracer if built; falls back to /tmp/echotrace.0.log if present; skips with a
+message otherwise.
+
+ ~/ida-venv/bin/python tests/test_trace_rpc.py
+"""
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from idatui.rpcclient import RpcClient, RpcError # noqa: E402
+from idatui.trace import Trace # noqa: E402
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRACER = os.path.expanduser(
+ "~/.pi/agent/skills/tenet-trace/scripts/tenet-trace")
+FALLBACK_TRACE = "/tmp/echotrace.0.log"
+BINARY = os.path.join(REPO, "targets", "echo")
+
+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 make_trace(tmp, binary):
+ """Record a short trace of the echo binary."""
+ out = os.path.join(tmp, "t")
+ try:
+ subprocess.run([TRACER, "-o", out, binary, "hello"],
+ capture_output=True, timeout=180, check=False)
+ except (OSError, subprocess.TimeoutExpired):
+ return None
+ log = out + ".0.log"
+ return log if os.path.exists(log) else None
+
+
+def spawn_pane(target, trace_log, timeout=300):
+ """Spawn an idatui pane with --trace and wait for readiness."""
+ cmd = [sys.executable, "-m", "idatui.pane", "spawn",
+ "--open", target, "--trace", trace_log,
+ "--detached", "--size", "60%", "--timeout", str(timeout)]
+ r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 30,
+ cwd=REPO)
+ if r.returncode != 0:
+ print(f" spawn failed: {r.stderr.strip()}", file=sys.stderr)
+ return None
+ return json.loads(r.stdout)
+
+
+def stop_pane(sock, timeout=60):
+ cmd = [sys.executable, "-m", "idatui.pane", "stop",
+ "--sock", sock, "--timeout", str(timeout)]
+ subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 10,
+ cwd=REPO)
+
+
+def main() -> int:
+ global PASS, FAIL
+ if not os.environ.get("TMUX"):
+ print(" skip: not inside tmux (test spawns a pane)")
+ return 0
+
+ with tempfile.TemporaryDirectory() as tmp:
+ # Scratch copy so the suite never writes into the tracked tree.
+ target = os.path.join(tmp, "echo")
+ shutil.copy2(BINARY, target)
+
+ # Get a trace.
+ trace_log = make_trace(tmp, target)
+ if trace_log is None and os.path.exists(FALLBACK_TRACE):
+ trace_log = FALLBACK_TRACE
+ if trace_log is None:
+ print(f" skip: no trace available (tracer at {TRACER}, "
+ f"fallback {FALLBACK_TRACE})")
+ return 0
+
+ # Load our own model for ground-truth comparisons.
+ model = Trace.load(trace_log)
+ check("model loaded for ground truth", model.length > 10,
+ f"length={model.length}")
+
+ # Spawn the pane.
+ info = spawn_pane(target, trace_log)
+ if info is None or not info.get("ready"):
+ print(f" FAIL: pane did not become ready: {info}")
+ return 1
+ sock = info["sock"]
+ check("pane spawned and ready", True)
+
+ try:
+ with RpcClient(sock) as c:
+ run_trace_tests(c, model)
+ finally:
+ stop_pane(sock)
+
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+def run_trace_tests(c: RpcClient, model: Trace):
+ """All RPC-level trace tests, given a connected client and the model."""
+
+ # ------------------------------------------------------------------ #
+ # Basic: ping should show the trace is loaded
+ # ------------------------------------------------------------------ #
+ p = c.call("ping")
+ check("ping reports ready", p.get("ready") is True)
+
+ # ------------------------------------------------------------------ #
+ # trace verb: seek to timestamp 0
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", seek=0)
+ check("trace seek=0 returns a snapshot", "active" in r and "trace" in r)
+ tr = r["trace"]
+ check("response includes trace metadata",
+ "idx" in tr and "length" in tr and "pc" in tr and "changed" in tr,
+ f"keys={list(tr.keys())}")
+ check("idx is 0 after seeking to 0", tr["idx"] == 0)
+ check("length matches the model", tr["length"] == model.length,
+ f"{tr['length']} vs {model.length}")
+ check("pc at 0 is a hex string", isinstance(tr["pc"], str)
+ and tr["pc"].startswith("0x"), tr["pc"])
+
+ # ------------------------------------------------------------------ #
+ # seek to a mid-trace timestamp
+ # ------------------------------------------------------------------ #
+ mid = model.length // 2
+ r = c.call("trace", seek=mid)
+ tr = r["trace"]
+ check("seek to midpoint lands correctly", tr["idx"] == mid,
+ f"got {tr['idx']}, want {mid}")
+ # The model's IP at this point, rebased — the RPC should agree.
+ # We can't compare directly because the model isn't rebased yet, but
+ # the pc should be a small address (database-space, not ASLR'd).
+ pc = int(tr["pc"], 16)
+ check("pc is in database space (not ASLR'd)",
+ pc < 0x100000, f"pc={tr['pc']}")
+
+ # ------------------------------------------------------------------ #
+ # seek by percentage (Tenet shell syntax)
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", seek="!0")
+ check("seek !0 (0%) goes to the start", r["trace"]["idx"] == 0)
+ r = c.call("trace", seek="!100")
+ check("seek !100 (100%) goes to the end",
+ r["trace"]["idx"] == model.length - 1,
+ f"got {r['trace']['idx']}, want {model.length - 1}")
+ r = c.call("trace", seek="!50")
+ check("seek !50 (50%) goes to the midpoint",
+ abs(r["trace"]["idx"] - mid) <= 1,
+ f"got {r['trace']['idx']}, want ~{mid}")
+
+ # ------------------------------------------------------------------ #
+ # step forward/backward
+ # ------------------------------------------------------------------ #
+ c.call("trace", seek=0)
+ r = c.call("trace", step=1)
+ check("step=1 advances one timestamp", r["trace"]["idx"] == 1,
+ f"got {r['trace']['idx']}")
+ r = c.call("trace", step=1)
+ check("another step=1 reaches 2", r["trace"]["idx"] == 2,
+ f"got {r['trace']['idx']}")
+ r = c.call("trace", step=-1)
+ check("step=-1 goes backward", r["trace"]["idx"] == 1,
+ f"got {r['trace']['idx']}")
+
+ # Step backward at the start should clamp to 0.
+ c.call("trace", seek=0)
+ r = c.call("trace", step=-1)
+ check("step=-1 at t=0 stays at 0", r["trace"]["idx"] == 0,
+ f"got {r['trace']['idx']}")
+
+ # Multi-step.
+ c.call("trace", seek=0)
+ r = c.call("trace", step=5)
+ check("step=5 advances five timestamps", r["trace"]["idx"] == 5,
+ f"got {r['trace']['idx']}")
+
+ # ------------------------------------------------------------------ #
+ # step over (follows SP)
+ # ------------------------------------------------------------------ #
+ # Find a call in the model: SP drops and later recovers.
+ sp_name = "rsp" if "rsp" in model.reg_at else "esp"
+ call_at = None
+ for i in range(1, min(model.length - 1, 400)):
+ a = model.register(sp_name, i)
+ b = model.register(sp_name, i + 1)
+ if a and b and b < a:
+ ret = next((j for j in range(i + 1, model.length)
+ if (model.register(sp_name, j) or 0) >= a), None)
+ if ret and ret > i + 3:
+ call_at = (i, ret)
+ break
+ if call_at is not None:
+ i, ret = call_at
+ c.call("trace", seek=i)
+ r = c.call("trace", step=1, over=True)
+ check("step over skips the callee",
+ r["trace"]["idx"] == ret,
+ f"from {i}, got {r['trace']['idx']}, expected {ret}")
+ check("step over goes further than a plain step",
+ r["trace"]["idx"] > i + 1)
+ else:
+ check("found a call to step over", False, "none in this short trace")
+
+ # ------------------------------------------------------------------ #
+ # goto: first execution of a function by name
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", goto="main")
+ tr = r["trace"]
+ check("goto main lands on a timestamp", tr["idx"] >= 0)
+ check("and the function context says main",
+ r.get("function", {}).get("name") == "main",
+ f"function={r.get('function')}")
+
+ # goto by hex address.
+ main_ea = r["function"]["ea"]
+ c.call("trace", seek=0) # reset position
+ r = c.call("trace", goto=hex(main_ea))
+ check("goto by hex address works",
+ r["trace"]["idx"] >= 0 and r["function"]["ea"] == main_ea,
+ f"idx={r['trace']['idx']}, ea={r.get('function', {}).get('ea')}")
+
+ # goto a function that was never executed.
+ try:
+ c.call("trace", goto="0xDEADBEEF")
+ check("goto an unexecuted address raises", False, "no error raised")
+ except RpcError as e:
+ check("goto an unexecuted address raises", "never executed" in str(e),
+ str(e))
+
+ # ------------------------------------------------------------------ #
+ # changed registers in the response
+ # ------------------------------------------------------------------ #
+ c.call("trace", seek=0)
+ r = c.call("trace", step=1)
+ changed = r["trace"]["changed"]
+ check("changed is a list of register names",
+ isinstance(changed, list) and all(isinstance(s, str) for s in changed),
+ f"{changed}")
+ # The PC always changes on a step (it's a different instruction).
+ check("rip is always in changed", "rip" in changed, f"{changed}")
+
+ # ------------------------------------------------------------------ #
+ # the cursor tracks the trace: seek moves the code view
+ # ------------------------------------------------------------------ #
+ # Seek to two different timestamps and verify the cursor ea changes.
+ r1 = c.call("trace", seek=10)
+ r2 = c.call("trace", seek=min(50, model.length - 1))
+ ea1 = r1.get("cursor", {}).get("ea")
+ ea2 = r2.get("cursor", {}).get("ea")
+ check("the cursor ea follows the trace pc",
+ ea1 is not None and ea2 is not None and (ea1 != ea2 or r1["trace"]["pc"] == r2["trace"]["pc"]),
+ f"ea1={ea1}, ea2={ea2}")
+
+ # ------------------------------------------------------------------ #
+ # trace verb without a trace raises cleanly
+ # ------------------------------------------------------------------ #
+ # (Can't test this on THIS pane since it has a trace. But we can test
+ # the error path by verifying the error message shape for bad params.)
+ try:
+ c.call("trace") # no seek/goto/step — should still return a snapshot
+ # Actually, if none of seek/goto/step is given, it just settles and
+ # returns the current state — that's fine, it's a status query.
+ r = c.call("trace")
+ check("trace with no action is a status query",
+ "trace" in r and r["trace"]["idx"] >= 0)
+ except RpcError:
+ check("trace with no action is a status query", False, "raised an error")
+
+ # ------------------------------------------------------------------ #
+ # snapshot shape: the trace key is present on every trace response
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", seek=3)
+ for key in ("idx", "length", "pc", "changed"):
+ check(f"trace response has '{key}'", key in r.get("trace", {}),
+ f"trace={r.get('trace')}")
+
+ # Standard snapshot fields are ALSO present (the trace response is a
+ # superset of a normal snapshot).
+ for key in ("active", "function", "cursor", "status", "ready"):
+ check(f"trace response also has snapshot key '{key}'", key in r,
+ f"keys={list(r.keys())}")
+
+ # ------------------------------------------------------------------ #
+ # edge cases: seek beyond bounds
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", seek=-1)
+ check("seek -1 clamps to 0", r["trace"]["idx"] == 0)
+ r = c.call("trace", seek=model.length + 1000)
+ check("seek beyond length clamps to the end",
+ r["trace"]["idx"] == model.length - 1,
+ f"got {r['trace']['idx']}")
+
+ # ------------------------------------------------------------------ #
+ # seek with comma-separated numbers (ergonomic)
+ # ------------------------------------------------------------------ #
+ r = c.call("trace", seek="100")
+ check("seek accepts a string number",
+ r["trace"]["idx"] == min(100, model.length - 1))
+
+ # ------------------------------------------------------------------ #
+ # pseudocode still works with a trace loaded
+ # ------------------------------------------------------------------ #
+ c.call("trace", goto="main")
+ r = c.call("pseudocode", target="main", lines=5)
+ check("pseudocode works alongside the trace",
+ "code" in r and "main" in r.get("code", ""),
+ f"keys={list(r.keys())}")
+
+ # ------------------------------------------------------------------ #
+ # state includes trace position
+ # ------------------------------------------------------------------ #
+ c.call("trace", seek=7)
+ r = c.call("state")
+ # The state verb doesn't include trace info (that's trace-specific),
+ # but the standard snapshot fields should be consistent.
+ check("state works with a trace loaded",
+ r.get("ready") is True and "cursor" in r)
+
+ # ------------------------------------------------------------------ #
+ # view_lines works with trail painted
+ # ------------------------------------------------------------------ #
+ c.call("trace", seek=min(40, model.length - 1))
+ r = c.call("view", lines=10)
+ check("view returns lines with a trace active",
+ "lines" in r and len(r["lines"]) > 0,
+ f"keys={list(r.keys())}")
+
+ # ------------------------------------------------------------------ #
+ # navigation works alongside trace: goto a function, trace follows
+ # ------------------------------------------------------------------ #
+ c.call("trace", goto="main")
+ start_idx = c.call("trace")["trace"]["idx"]
+ r = c.call("goto", target="error_at_line")
+ check("goto still works with trace loaded",
+ r.get("function", {}).get("name") == "error_at_line")
+ # The trace timestamp should NOT change from a regular goto — the trace
+ # position is independent of navigation.
+ r2 = c.call("trace")
+ check("regular goto does not change the trace position",
+ r2["trace"]["idx"] == start_idx,
+ f"was {start_idx}, now {r2['trace']['idx']}")
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py
new file mode 100644
index 0000000..32977ad
--- /dev/null
+++ b/tests/test_trace_ui.py
@@ -0,0 +1,395 @@
+#!/usr/bin/env python3
+"""The trace dock: registers, timeline, and stepping through time.
+
+Needs IDA and a trace. Generates its own trace with the QEMU tracer if the
+tracer is built; skips with a message otherwise, since neither the emulator nor
+the trace is part of this repo.
+"""
+import asyncio
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from textual.widgets import Input, OptionList, Static # noqa: E402
+
+from idatui.app import (DecompView, IdaTui, ListingView, # noqa: E402
+ RegWriteScreen, TraceDock)
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRACER = os.path.expanduser(
+ "~/.pi/agent/skills/tenet-trace/scripts/tenet-trace")
+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
+
+
+def make_trace(tmp, binary):
+ out = os.path.join(tmp, "t")
+ try:
+ subprocess.run([TRACER, "-o", out, binary, "hi"],
+ capture_output=True, timeout=180, check=False)
+ except (OSError, subprocess.TimeoutExpired):
+ return None
+ log = out + ".0.log"
+ return log if os.path.exists(log) else None
+
+
+async def run() -> int:
+ binary = os.path.join(REPO, "targets", "echo")
+ with tempfile.TemporaryDirectory() as tmp:
+ # Scratch copy: opening a binary writes a database beside it, and the
+ # suite must not edit anything tracked.
+ target = os.path.join(tmp, "echo")
+ shutil.copy2(binary, target)
+ log = make_trace(tmp, target)
+ if not log:
+ print(f" skip: could not record a trace (tracer at {TRACER})")
+ return 0
+
+ app = IdaTui(open_path=target, keepalive=False, trace_path=log)
+ async with app.run_test(size=(160, 44)) as pilot:
+ ok = await wait(lambda: app._trace is not None, pilot, 300)
+ check("the trace loads alongside the binary", ok)
+ if not ok:
+ return 1
+ t = app._trace
+ check("it has instructions", t.length > 10, f"{t.length}")
+
+ # Rebasing: the tracer runs the binary relocated, so without a slide
+ # nothing in the trace matches anything on screen.
+ check("trace addresses were rebased onto the database",
+ t.slide != 0 and t.ip(0) < 0x1000000,
+ f"slide={t.slide:#x} ip0={t.ip(0):#x}")
+ idx = app._func_index
+ touched = [f.name for f in idx.all_loaded() if t.executions(f.addr)]
+ check("and now line up with real functions",
+ len(touched) > 1 and "main" in touched, f"{touched[:6]}")
+
+ dock = app.query_one(TraceDock)
+ check("the dock is docked and visible", dock.display)
+ head = str(dock.query_one("#trace-head", Static).render())
+ check("it shows where we are in time", "0" in head and "%" in head,
+ head[:60])
+ regs = str(dock.query_one("#trace-regs", Static).render())
+ check("and the register state at that time", "rip" in regs.lower(),
+ regs[:60])
+
+ # -- stepping --------------------------------------------------- #
+ lst = app.query_one(ListingView)
+ lst.focus()
+ await pilot.pause(0.4)
+ await pilot.press("]")
+ await wait(lambda: app._t == 1, pilot, 20)
+ check("] steps forward one instruction", app._t == 1, f"t={app._t}")
+ check("the code view follows the trace",
+ lst._cursor_ea() == t.ip(1),
+ f"{lst._cursor_ea()} vs {t.ip(1)}")
+ await pilot.press("[")
+ await wait(lambda: app._t == 0, pilot, 20)
+ check("[ steps backward", app._t == 0, f"t={app._t}")
+ await pilot.press("[")
+ await pilot.pause(0.4)
+ check("and stops at the start of the trace", app._t == 0)
+
+ # -- step over -------------------------------------------------- #
+ # A call pushes, so the callee runs with SP below where we started;
+ # stepping until SP comes back up lands after the return. Find a
+ # real call in this trace rather than assuming one is at a fixed
+ # place.
+ sp = "rsp" if "rsp" in t.reg_at else "esp"
+ call_at = None
+ for i in range(1, min(t.length - 1, 400)):
+ a, b = t.register(sp, i), t.register(sp, i + 1)
+ if a and b and b < a:
+ ret = next((j for j in range(i + 1, t.length)
+ if (t.register(sp, j) or 0) >= a), None)
+ if ret and ret > i + 3:
+ call_at = (i, ret)
+ break
+ if call_at is None:
+ check("found a call to step over", False, "none in this trace")
+ else:
+ i, ret = call_at
+ app._seek(i)
+ await wait(lambda: app._t == i, pilot, 20)
+ await pilot.press("}")
+ await wait(lambda: app._t != i, pilot, 30)
+ check("} steps OVER a call instead of into it",
+ app._t == ret, f"{i} -> {app._t}, expected {ret}")
+ check("which is further than a plain step", app._t > i + 1)
+
+ # -- memory at time T -------------------------------------------- #
+ # This is where a trace's memory actually lives: measured on two
+ # real traces, NONE of the accesses fell inside the image — every
+ # one was stack or heap. A memory view that could only address the
+ # image would have nothing to show.
+ dock = app.query_one(TraceDock)
+ app._seek(min(60, t.length - 1))
+ await pilot.pause(0.8)
+ stack = str(dock.query_one("#trace-stack", Static).render())
+ check("the dock shows the stack at this timestamp",
+ "stack (" in stack and len(stack.splitlines()) > 4, stack[:60])
+ sp_name = next(r for r in ("rsp", "esp", "sp") if r in t.reg_at)
+ sp = t.register(sp_name, app._t)
+ check("anchored at the stack pointer",
+ f"{sp:012x}" in stack, f"sp={sp:#x} / {stack[:80]}")
+
+ # A trace knows what it observed and nothing else. Unseen bytes are
+ # printed as '?', never as zeros — rendering them as zero would
+ # invent facts about memory nobody looked at.
+ data, known = t.memory_raw(sp, 8, app._t)
+ if not all(known):
+ check("memory the trace never saw is marked unknown",
+ "?" in stack, stack[:80])
+ else:
+ check("known stack words are shown as values",
+ any(c in "0123456789abcdef" for c in stack), stack[:60])
+
+ # Stepping must move the memory view with time.
+ before = stack
+ app._seek(min(80, t.length - 1))
+ await pilot.pause(0.8)
+ check("and it follows as you move through time",
+ str(dock.query_one("#trace-stack", Static).render()) != before)
+
+ # -- trails ------------------------------------------------------ #
+ # Not "every address the trace ever touched": on a loop-heavy
+ # program that's almost everything and says nothing. The last/next
+ # few dozen steps say how you got here and where you're going.
+ app._seek(min(40, t.length - 1))
+ await pilot.pause(0.6)
+ trail = lst.trail
+ kinds = {k for k in trail.values()}
+ check("the listing is painted with an execution trail",
+ {"now", "past", "future"} <= kinds, f"{sorted(kinds)}")
+ check("'now' is the instruction we're standing on",
+ trail.get(t.ip(app._t)) == "now", f"{trail.get(t.ip(app._t))}")
+ check("the step behind is past, the step ahead is future",
+ trail.get(t.ip(app._t - 1)) == "past"
+ and trail.get(t.ip(app._t + 1)) == "future",
+ f"{trail.get(t.ip(app._t - 1))}, {trail.get(t.ip(app._t + 1))}")
+ painted = [y for y in range(min(lst.size.height, 30))
+ if any(seg.style and seg.style.bgcolor
+ for seg in lst.render_line(y))]
+ check("and it actually reaches the screen", painted, "no tinted rows")
+
+ # -- the same trail on PSEUDOCODE -------------------------------- #
+ # The point of doing this in our app rather than using Tenet: a
+ # trace's addresses are instructions, but decomp_map (built for the
+ # split view) says which instructions each pseudocode line covers,
+ # so the trail lands on C.
+ main_ea = app.program.resolve("main")
+ first = t.first_execution(main_ea)
+ if first is None:
+ check("main was executed in the trace", False)
+ else:
+ app._seek(first + 12)
+ await wait(lambda: app._t == first + 12, pilot, 20)
+ lst.focus()
+ await pilot.press("tab")
+ got = await wait(lambda: app.query_one(DecompView).display
+ and app.query_one(DecompView)._texts, pilot, 120)
+ dec = app.query_one(DecompView)
+ check("pseudocode is available for the traced function", got)
+ app._seek(first + 12)
+ await pilot.pause(1.0)
+ check("pseudocode lines are painted with the trail",
+ len(dec.trail) > 2, f"{len(dec.trail)} lines")
+ now = [i for i, k in dec.trail.items() if k == "now"]
+ check("exactly one pseudocode line is 'now'",
+ len(now) == 1, f"{now}")
+ # The 'now' line must be the one covering the current
+ # instruction, not merely some executed line.
+ covered = app._trail_map[now[0]] if now and app._trail_map else []
+ check("and it's the line covering the current instruction",
+ t.ip(app._t) in covered,
+ f"pc={t.ip(app._t):#x} line covers {[hex(a) for a in covered][:4]}")
+
+ # Stepping must not throw you out of the view you're reading.
+ # A step navigates to an address, and navigating to an address
+ # opens the LISTING unless the decompiler is preferred — so
+ # stepping through C used to drop you into disassembly on the
+ # first keypress. Found by watching a demo, not by a test.
+ await pilot.press("]")
+ await pilot.pause(1.2)
+ check("stepping in pseudocode stays in pseudocode",
+ app._active == "decomp", f"active={app._active}")
+ await pilot.press("[")
+ await pilot.pause(1.2)
+ check("and so does stepping backward",
+ app._active == "decomp", f"active={app._active}")
+
+ # -- split view: a step is a GLOBAL move ------------------------ #
+ # Normal navigation moves one pane and gives the companion a band,
+ # never a cursor, so the two can't chase each other. Time isn't
+ # navigation though: both panes show the same instant, so the
+ # listing cursor must sit on the current instruction.
+ app.action_toggle_split()
+ await pilot.pause(2.0)
+ if not app._split:
+ check("split view toggled on", False)
+ else:
+ base = t.first_execution(main_ea) or 0
+ tracked = 0
+ for k in range(2, 8):
+ app._seek(base + k)
+ await pilot.pause(0.5)
+ if lst._cursor_ea() == t.ip(app._t):
+ tracked += 1
+ check("stepping in split moves the listing cursor to the pc",
+ tracked == 6, f"{tracked}/6 steps tracked")
+ check("and the trail follows in both panes",
+ lst.trail.get(t.ip(app._t)) == "now",
+ f"{lst.trail.get(t.ip(app._t))}")
+
+ # The pseudocode cursor follows too — but only for instructions
+ # the decompiler actually attributes to a line. About half
+ # aren't, and the tempting fallback (nearest mapped address at
+ # or before the pc) is unsound because C lines are not monotonic
+ # in address: an early instruction resolved to a line near the
+ # END of the function. Better to wait than to jump somewhere
+ # unrelated.
+ dec = app.query_one(DecompView)
+ mapped = missed = 0
+ for k in range(2, 30):
+ app._seek(base + k)
+ await pilot.pause(0.3)
+ pc = t.ip(app._t)
+ if app._trail_map_ea == dec.loaded_ea and pc in app._trail_line_of:
+ mapped += 1
+ if dec.cursor != app._trail_line_of[pc]:
+ missed += 1
+ check("the pseudocode cursor follows every mapped instruction",
+ mapped > 3 and missed == 0,
+ f"{mapped} mapped, {missed} not followed")
+
+ # -- a late navigation must not drag the view back --------------- #
+ # Navigations run in workers and finish out of order. The trace's
+ # OPENING seek goes to the entry point (t=0 is _start), takes a
+ # while, and used to arrive after later seeks — leaving the cursor
+ # on _start while the trace was elsewhere, and it never settled.
+ # Anything cursor-based done right after a seek then acted on the
+ # wrong address.
+ two = [a for a in t.by_ip if len(t.by_ip[a]) > 1]
+ if two:
+ a = max(two, key=lambda x: len(t.by_ip[x]))
+ s0, s1 = list(t.by_ip[a])[:2]
+ dbaddr = a + t.slide
+ app._seek(s0)
+ await wait(lambda: lst._cursor_ea() == dbaddr, pilot, 60)
+ app._seek(s1)
+ await pilot.pause(3.0) # long enough for a stale one to land
+ check("a stale navigation doesn't drag the cursor away",
+ lst._cursor_ea() == dbaddr and app._cur.ea == dbaddr,
+ f"cursor={lst._cursor_ea():#x} cur={app._cur.ea:#x} "
+ f"want {dbaddr:#x}")
+
+ # -- seeking, as opposed to stepping ---------------------------- #
+ # "When else did this instruction run?" — the question that makes a
+ # trace more than a very long single-step log.
+ hot = max(t.by_ip, key=lambda a: len(t.by_ip[a]))
+ stamps = list(t.by_ip[hot])
+ db = hot + t.slide
+ if len(stamps) < 2 or lst.model is None:
+ check("found an address executed more than once", False,
+ f"{len(stamps)} executions")
+ else:
+ if app._split:
+ app.action_toggle_split()
+ await pilot.pause(1.0)
+ if app._active != "listing":
+ # focus() does NOT make a view active outside split mode;
+ # Tab is what switches which one is showing.
+ await pilot.press("tab")
+ await wait(lambda: app._active == "listing", pilot, 60)
+ app._seek(stamps[0])
+ await pilot.pause(1.2)
+ lst.focus()
+ row = lst.model.index_of_ea(db)
+ lst.cursor = row
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.4)
+ check("cursor is on the repeated instruction",
+ lst._cursor_ea() == db, f"{lst._cursor_ea():#x} vs {db:#x}")
+ await pilot.press(">")
+ await pilot.pause(1.0)
+ check("> seeks to the next execution of it",
+ app._t == stamps[1], f"t={app._t}, expected {stamps[1]}")
+ status = str(app.query_one("#status", Static).render())
+ check("and says which execution this is",
+ f"2 of {len(stamps)}" in status, status[:80])
+ lst.cursor = row
+ await pilot.pause(0.2)
+ await pilot.press("<")
+ await pilot.pause(1.0)
+ check("< seeks back to the previous one",
+ app._t == stamps[0], f"t={app._t}, expected {stamps[0]}")
+ # An edge must SAY it's an edge rather than silently doing
+ # nothing, which is indistinguishable from a broken key.
+ lst.cursor = row
+ await pilot.pause(0.2)
+ await pilot.press("<")
+ await pilot.pause(1.0)
+ status = str(app.query_one("#status", Static).render())
+ check("and the first execution says so instead of moving",
+ app._t == stamps[0] and "first" in status, status[:80])
+
+ # -- "which instruction set this register?" ---------------------- #
+ app._seek(min(200, t.length - 1))
+ await pilot.pause(1.0)
+ lst.focus()
+ await pilot.press("W")
+ opened = await wait(lambda: isinstance(app.screen, RegWriteScreen),
+ pilot, 20)
+ check("W lists the registers and where each was set", opened,
+ f"screen={type(app.screen).__name__}")
+ if opened:
+ sc = app.screen
+ pick = next((k for k, (n, v, l, x) in enumerate(sc._rows)
+ if l is not None and l != app._t), None)
+ if pick is None:
+ check("a register was set by an earlier instruction", False)
+ await pilot.press("escape")
+ else:
+ name, _v, last, _n = sc._rows[pick]
+ sc.query_one(OptionList).highlighted = pick
+ await pilot.pause(0.3)
+ await pilot.press("enter")
+ await wait(lambda: app._t == last, pilot, 30)
+ check("choosing one seeks to the write that set it",
+ app._t == last, f"t={app._t}, expected {last}")
+ # The real check: that instruction must actually have
+ # written the register we asked about.
+ check("and that instruction really wrote it",
+ name in t.changed(app._t),
+ f"{name} not in {sorted(t.changed(app._t))}")
+
+ 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_trace_vs_tenet.py b/tests/test_trace_vs_tenet.py
new file mode 100644
index 0000000..aa6002a
--- /dev/null
+++ b/tests/test_trace_vs_tenet.py
@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+"""Differential test: our trace reader vs Tenet's own.
+
+We wrote our own reader rather than vendoring 3700 lines of packing and mask
+compression we don't need. That is only defensible if "our own" doesn't quietly
+mean "different", so every answer we give is checked against the reference
+implementation on real traces.
+
+Needs ~/dev/tenet/tenet-original (reference) and a trace to compare on; both are
+skipped-with-a-message rather than failed if absent, since neither is part of
+this repo.
+
+ python3 tests/test_trace_vs_tenet.py [trace.0.log ...]
+"""
+import os
+import random
+import sys
+import types
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from idatui.trace import Trace # noqa: E402
+
+TENET = os.path.expanduser("~/dev/tenet/tenet-original/plugins")
+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 _reference():
+ """Tenet's TraceReader, with its disassembler dependency stubbed out."""
+ sys.path.insert(0, TENET)
+ log = types.ModuleType("tenet.util.log")
+ log.pmsg = lambda *a, **k: None
+ import tenet # noqa: F401
+ import tenet.util # noqa: F401
+ sys.modules["tenet.util.log"] = log
+ from tenet.trace.arch import ArchAMD64
+ from tenet.trace.reader import TraceReader
+
+ class FakeDctx:
+ # The reference only touches the disassembler to guess an ASLR slide.
+ # An EMPTY list crashes its analysis (indexes [0] unguarded), so hand it
+ # one address that matches nothing: no slide is found and both readers
+ # stay in raw trace addresses, which is what we want to compare.
+ def get_instruction_addresses(self):
+ return [0xDEAD0000]
+ return TraceReader, ArchAMD64, FakeDctx
+
+
+def _truth(path, reg, idx):
+ """The register's value at ``idx`` read straight out of the text.
+
+ Slow (re-reads the file) and only used to settle a disagreement, which is
+ exactly when being slow and obviously-correct is what you want.
+ """
+ reg = reg.lower()
+ last = None
+ with open(path) as f:
+ for i, line in enumerate(f):
+ if i > idx:
+ break
+ for part in line.strip().split(","):
+ k, _, v = part.partition("=")
+ if k.strip().lower() == reg:
+ try:
+ last = int(v, 16)
+ except ValueError:
+ pass
+ return last
+
+
+def compare(path, ref_cls, arch, dctx, samples=200):
+ TraceReader, ArchAMD64, FakeDctx = ref_cls, arch, dctx
+ ours = Trace.load(path)
+ theirs = TraceReader(path, ArchAMD64(), FakeDctx())
+ name = os.path.basename(path)
+
+ check(f"{name}: same length",
+ ours.length == theirs.trace.length,
+ f"{ours.length} vs {theirs.trace.length}")
+ n = min(ours.length, theirs.trace.length)
+ if not n:
+ return
+
+ rnd = random.Random(1234)
+ idxs = sorted({0, n - 1, n // 2} | {rnd.randrange(n) for _ in range(samples)})
+
+ bad = [i for i in idxs if ours.raw_ip(i) != theirs.get_ip(i)]
+ check(f"{name}: same PC at every sampled timestamp", not bad,
+ f"first mismatch at {bad[:1]}")
+
+ # Register reconstruction is the part that is easy to get subtly wrong: a
+ # delta belongs to the line that CAUSED it, and an off-by-one here silently
+ # shifts every value by one instruction.
+ #
+ # Where the two disagree, ARBITRATE with the text rather than assume the
+ # reference is right. It isn't always: a register written on the last line
+ # of a 65535-line segment is missing from the next segment's base state, so
+ # the reference returns a stale value until that register is written again
+ # (measured: wrong for all 179 timestamps of one such window). Blanket
+ # "must agree" would have made us copy that bug to stay green.
+ regs = [r.upper() for r in ours.registers if r not in ("rip",)][:8]
+ ours_wrong, ref_wrong = [], []
+ for i in idxs:
+ for r in regs:
+ mine, ref = ours.register(r, i), theirs.get_register(r, i)
+ if mine == ref:
+ continue
+ true = _truth(path, r, i)
+ (ours_wrong if mine != true else ref_wrong).append((i, r, mine, ref, true))
+ check(f"{name}: register state matches the trace text everywhere",
+ not ours_wrong, f"{ours_wrong[:3]}")
+ if ref_wrong:
+ print(f" (reference disagrees at {len(ref_wrong)} sampled points; "
+ f"the text backs us, e.g. idx {ref_wrong[0][0]} {ref_wrong[0][1]})")
+
+ # Execution queries: what painting is built on.
+ hot = sorted(ours.by_ip, key=lambda a: -len(ours.by_ip[a]))[:5]
+ ex_bad = []
+ for ea in hot:
+ mine = list(ours.executions(ea))
+ ref = list(theirs.get_executions(ea))
+ if mine != ref:
+ ex_bad.append((hex(ea), len(mine), len(ref)))
+ check(f"{name}: same execution timestamps for the hottest addresses",
+ not ex_bad, f"{ex_bad[:3]}")
+
+ # Memory STATE at a timestamp — reconstructed from the deltas, which is the
+ # hard part and the whole point of reading memory from a trace.
+ mem_bad, mem_checked = [], 0
+ for i in idxs[::4]:
+ for op in ours.memory_ops(i)[:2]:
+ n = min(len(op.data), 8)
+ mine, known = ours.memory(op.addr, n, i)
+ ref = theirs.get_memory(op.addr, n, i)
+ if ref is None:
+ continue
+ refb = bytes(ref.data)
+ # Compare only the bytes we claim to know; the reference reports its
+ # own coverage separately and a byte neither has seen is not a
+ # disagreement.
+ for j in range(n):
+ if known[j] and refb[j:j + 1] and mine[j] != refb[j]:
+ mem_bad.append((i, hex(op.addr + j), mine[j], refb[j]))
+ mem_checked += 1
+ check(f"{name}: memory state at a timestamp matches the reference",
+ not mem_bad, f"{mem_bad[:3]}")
+
+ # Memory: the bytes an instruction touched, and which way.
+ with_mem = [i for i in idxs if ours.memory_ops(i)][:40]
+ mem_bad = []
+ for i in with_mem:
+ for op in ours.memory_ops(i):
+ ref = theirs.get_memory(op.addr, len(op.data), i + 1) if op.write else None
+ if ref is not None and bytes(ref.data) != op.data:
+ mem_bad.append((i, hex(op.addr), op.data.hex(), bytes(ref.data).hex()))
+ check(f"{name}: written bytes match the reference's memory state",
+ not mem_bad, f"{mem_bad[:2]}")
+ print(f" ({ours.length:,} instructions, {len(idxs)} sampled, "
+ f"{len(with_mem)} with memory)")
+
+
+def main(argv):
+ if not os.path.isdir(TENET):
+ print(f" skip: reference not found at {TENET}")
+ return 0
+ traces = argv or [p for p in ("/tmp/echotrace.0.log", "/tmp/big.0.log")
+ if os.path.exists(p)]
+ if not traces:
+ print(" skip: no traces to compare (pass one, or run tenet-trace first)")
+ return 0
+ ref = _reference()
+ for path in traces:
+ compare(path, *ref)
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main(sys.argv[1:]))
generated by cgit v1.3.1-sl0p (git 2.54.0) at 2026-08-02 11:06:56 +0000 — sl0p fork