diff options
Diffstat (limited to 'tests/test_trace.py')
| -rw-r--r-- | tests/test_trace.py | 203 |
1 files changed, 203 insertions, 0 deletions
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()) |