trace: read Tenet execution traces (model layer)

First slice of the execution-trace viewer: parse, index and query a Tenet trace.
No UI yet — this is the layer everything else stands on, and its shape decides
whether decompiler painting is cheap later.

A Tenet trace is a line-per-instruction delta log: registers that changed, the
PC every line, and each memory access WITH its bytes. That's enough to
reconstruct any register or address at any point in time, in either direction.

Our own reader, not a port. The reference (tenet-original) packs traces into
segments with compressed address/mask tables, which earns its keep for its Qt
timeline; we need different queries and would rather own ~400 lines than inherit
3700.

Indexed around the query the UI actually asks, which the reference answers one
address at a time: WHICH TIMESTAMPS EXECUTED THIS SET OF ADDRESSES. A listing
row is one address, but a pseudocode line covers many (we already have
decomp_map for that), so by_ip maps address -> timestamps and hits() takes a
set. Painting a pseudocode line will be one call, not one per instruction.

Registers are stored as per-register change points, so a value at time t is a
bisect, and "which instruction set this register?" (last_write) is the same
lookup — that being the question a trace explorer exists to answer.

Rebasing is not optional: our echo trace runs at 0x7ffff6faa000 while the
database has that code at 0x2490. Page offsets survive relocation, so the low 12
bits of an instruction address are invariant; bucket the database's addresses by
those bits and take the slide the most trace addresses agree on. Verified
against a real IDB: slide -0x7ffff6fa8000, and it picks out the 12 functions the
trace actually entered (main, start, ...) from 128.

Performance: 176k instructions parse in 350ms (~500k lines/s), so a 10M-line
trace is ~20s and wants a progress callback, which load() takes.

FOUND A BUG IN THE REFERENCE while building the differential test. A register
written on the LAST line of a 65535-line segment is missing from the next
segment's base state, so Tenet returns a stale value until that register is
written again — measured: wrong for all 179 timestamps of one such window. It
survives in the reference because it only shows when the register isn't
rewritten immediately.

That changed how the test works. Rather than "must agree with the reference",
it ARBITRATES with the raw text when they differ: if the text backs us it's
reported and allowed, if the text backs them it fails. Blanket agreement would
have made us copy their bug to stay green.

tests: test_trace.py (27, pure stdlib) covers reconstruction, the set queries,
rebasing (including that a lone agreeing address is not enough, and that
matching is on page offsets rather than addresses looking plausible) and
malformed input; test_trace_vs_tenet.py (10) diffs against the reference on real
traces.

2 files changed, 330 insertions, 0 deletions

diff --git a/tests/test_trace.py b/tests/test_trace.py
new file mode 100644
index 0000000..2803a11
--- /dev/null
+++ b/tests/test_trace.py
@@ -0,0 +1,162 @@
+#!/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) == [])
+
+        # -- 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}")
+        check("memory addresses are slid too",
+              t.memory_ops(3)[0].addr == 0xff4 + slide)
+        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_vs_tenet.py b/tests/test_trace_vs_tenet.py
new file mode 100644
index 0000000..6c78c95
--- /dev/null
+++ b/tests/test_trace_vs_tenet.py
@@ -0,0 +1,168 @@
+#!/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: 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:]))