diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_trace.py | 162 | ||||
| -rw-r--r-- | tests/test_trace_vs_tenet.py | 168 |
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:])) |