diff options
| -rw-r--r-- | idatui/trace.py | 345 | ||||
| -rw-r--r-- | tests/test_trace.py | 162 | ||||
| -rw-r--r-- | tests/test_trace_vs_tenet.py | 168 |
3 files changed, 675 insertions, 0 deletions
diff --git a/idatui/trace.py b/idatui/trace.py new file mode 100644 index 0000000..514a97a --- /dev/null +++ b/idatui/trace.py @@ -0,0 +1,345 @@ +"""Reading Tenet execution traces. + +A Tenet trace is a line-per-instruction delta log:: + + rax=0x3c,rbx=0x0,...,rip=0x7ffff6faaae0 # full state on the first line + rip=0x7ffff6faaae4 # then only what changed + r9=0x7ffff6762e60,rip=0x7ffff6faaae9,mw=0x7ffff6f9b7c8:08c177f6ff7f0000 + +Registers that changed, the PC on every line, and every memory access *with its +bytes*. That is enough to reconstruct any register or any memory address at any +point in time, forwards or backwards, which is the whole trick. + +This is our own reader, not a port. The reference implementation +(``~/dev/tenet/tenet-original/plugins/tenet/trace/``) packs the trace into +segments with compressed address/mask tables, which earns its keep for its Qt +timeline; we need different queries and would rather own the ~400 lines than +inherit 3700. It is differential-tested against that implementation +(``tests/test_trace_vs_tenet.py``) so "our own" doesn't quietly mean "different". + +The index is built 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, so +``by_ip`` maps address -> timestamps and set queries are unions of those. +""" + +from __future__ import annotations + +import array +import bisect +import os +import re +from dataclasses import dataclass, field + +#: Tenet packs its register delta into a uint32, so a trace arch may name at +#: most 32 registers. Ours is discovered from the trace instead of declared, +#: but the cap is worth knowing when a trace looks short of registers. +MAX_REGISTERS = 32 + +_MEM_RE = re.compile(r"^m(r|w|rw)$") + + +@dataclass +class TraceInfo: + """The sidecar ``<prefix>.info`` written by our QEMU tracer. + + Optional: a trace from another tracer has none, and everything here can be + recovered or guessed from the log itself. + """ + + arch: str = "" + mode: str = "" + binary: str = "" + start_code: int = 0 + end_code: int = 0 + entry_code: int = 0 + traced: str = "" + + @classmethod + def load(cls, path: str) -> "TraceInfo | None": + try: + with open(path) as f: + raw = dict( + ln.strip().split("=", 1) for ln in f if "=" in ln) + except OSError: + return None + def num(k): + try: + return int(raw.get(k, "0"), 0) + except ValueError: + return 0 + return cls(arch=raw.get("arch", ""), mode=raw.get("mode", ""), + binary=raw.get("binary", ""), start_code=num("start_code"), + end_code=num("end_code"), entry_code=num("entry_code"), + traced=raw.get("traced", "")) + + +@dataclass +class MemOp: + """One memory access made by one instruction.""" + + addr: int + data: bytes + write: bool + + @property + def end(self) -> int: + return self.addr + len(self.data) + + +@dataclass +class Trace: + """An indexed Tenet trace. + + Timestamps are indices into the executed-instruction sequence: 0 is the + first instruction, ``length - 1`` the last. + """ + + path: str = "" + info: TraceInfo | None = None + #: PC per timestamp. + ips: array.array = field(default_factory=lambda: array.array("Q")) + #: name -> (timestamps of change, value at each change). A register's value + #: at time t is the last change at or before t; the first line carries a + #: full state dump, so every register has an entry at 0. + reg_at: dict[str, tuple[array.array, array.array]] = field(default_factory=dict) + #: address -> timestamps that executed it (ascending, by construction). + by_ip: dict[int, array.array] = field(default_factory=dict) + #: memory accesses, parallel arrays indexed by access number. + mem_idx: array.array = field(default_factory=lambda: array.array("I")) + mem_addr: array.array = field(default_factory=lambda: array.array("Q")) + mem_write: bytearray = field(default_factory=bytearray) + mem_off: array.array = field(default_factory=lambda: array.array("Q")) + mem_len: array.array = field(default_factory=lambda: array.array("H")) + mem_blob: bytearray = field(default_factory=bytearray) + #: first access number of each timestamp, plus a final sentinel. + mem_row: array.array = field(default_factory=lambda: array.array("I")) + #: applied so trace addresses line up with the database (see ``rebase``). + slide: int = 0 + + # -- construction ------------------------------------------------------ # + @classmethod + def load(cls, path: str, progress=None, limit: int = 0) -> "Trace": + """Parse a text trace. ``progress(lines)`` is called every 50k lines.""" + t = cls(path=os.path.abspath(path)) + base = path[:-6] if path.endswith(".0.log") else os.path.splitext(path)[0] + t.info = TraceInfo.load(base + ".info") + regs: dict[str, tuple[array.array, array.array]] = {} + ips, by_ip = t.ips, t.by_ip + n = 0 + with open(path) as f: + for line in f: + line = line.strip() + if not line: + continue + ip = None + t.mem_row.append(len(t.mem_idx)) + for part in line.split(","): + key, _, val = part.partition("=") + if not val: + continue + key = key.strip().lower() + if _MEM_RE.match(key): + addr_s, _, data_s = val.partition(":") + try: + addr = int(addr_s, 16) + data = bytes.fromhex(data_s) + except ValueError: + continue + # 'mrw' is one access that both reads and writes; record + # the write, which is what the memory state follows. + t.mem_idx.append(n) + t.mem_addr.append(addr) + t.mem_write.append(0 if key == "mr" else 1) + t.mem_off.append(len(t.mem_blob)) + t.mem_len.append(len(data)) + t.mem_blob += data + continue + try: + v = int(val, 16) + except ValueError: + continue + slot = regs.get(key) + if slot is None: + slot = regs[key] = (array.array("I"), array.array("Q")) + slot[0].append(n) + slot[1].append(v) + ip = v if key in ("rip", "eip", "pc") else ip + if ip is None: + # No PC on this line: the format says always emit it, and + # without it the line cannot be placed. Carry the previous + # one rather than dropping the instruction. + ip = ips[-1] if ips else 0 + ips.append(ip) + where = by_ip.get(ip) + if where is None: + where = by_ip[ip] = array.array("I") + where.append(n) + n += 1 + if progress is not None and n % 50000 == 0: + progress(n) + if limit and n >= limit: + break + t.mem_row.append(len(t.mem_idx)) + t.reg_at = regs + return t + + # -- basics ------------------------------------------------------------ # + def __len__(self) -> int: + return len(self.ips) + + @property + def length(self) -> int: + return len(self.ips) + + @property + def registers(self) -> list[str]: + """Register names in the trace, PC last (the order tracers emit).""" + return sorted(self.reg_at, key=lambda r: (r in ("rip", "eip", "pc"), r)) + + @property + def pc_name(self) -> str: + for n in ("rip", "eip", "pc"): + if n in self.reg_at: + return n + return "" + + def ip(self, idx: int) -> int: + """PC at ``idx``, in DATABASE addresses (slide applied).""" + return self.ips[idx] + self.slide + + def raw_ip(self, idx: int) -> int: + return self.ips[idx] + + # -- register state ----------------------------------------------------- # + def register(self, name: str, idx: int) -> int | None: + """Value of ``name`` at ``idx``, or None if the trace never set it.""" + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + idxs, vals = slot + i = bisect.bisect_right(idxs, idx) - 1 + return vals[i] if i >= 0 else None + + def register_state(self, idx: int) -> dict[str, int]: + return {n: v for n in self.reg_at + if (v := self.register(n, idx)) is not None} + + def changed(self, idx: int) -> set[str]: + """Registers written BY the instruction at ``idx`` (what the line said). + + Used to highlight what an instruction actually did, which is the reason + a delta trace is readable at all. + """ + out = set() + for n, (idxs, _vals) in self.reg_at.items(): + i = bisect.bisect_left(idxs, idx) + if i < len(idxs) and idxs[i] == idx: + out.add(n) + return out + + def last_write(self, name: str, idx: int) -> int | None: + """Timestamp of the write that produced ``name``'s value at ``idx``. + + "Which instruction set this register?" — the question that motivates a + trace explorer in the first place. + """ + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + i = bisect.bisect_right(slot[0], idx) - 1 + return slot[0][i] if i >= 0 else None + + def next_write(self, name: str, idx: int) -> int | None: + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + i = bisect.bisect_right(slot[0], idx) + return slot[0][i] if i < len(slot[0]) else None + + # -- memory ------------------------------------------------------------- # + def memory_ops(self, idx: int) -> list[MemOp]: + """Accesses made by the instruction at ``idx``.""" + if not (0 <= idx < len(self.mem_row) - 1): + return [] + lo, hi = self.mem_row[idx], self.mem_row[idx + 1] + out = [] + for k in range(lo, hi): + off, ln = self.mem_off[k], self.mem_len[k] + out.append(MemOp(addr=self.mem_addr[k] + self.slide, + data=bytes(self.mem_blob[off:off + ln]), + write=bool(self.mem_write[k]))) + return out + + # -- execution queries (what painting is built on) ---------------------- # + def executions(self, ea: int) -> array.array: + """Every timestamp that executed ``ea`` (database address).""" + return self.by_ip.get(ea - self.slide, array.array("I")) + + def executions_between(self, ea: int, lo: int, hi: int) -> list[int]: + ts = self.executions(ea) + a = bisect.bisect_left(ts, lo) + b = bisect.bisect_right(ts, hi) + return list(ts[a:b]) + + def hits(self, eas) -> dict[int, int]: + """{address: execution count} for a set of addresses. + + The set form is the point: painting a listing row needs one address, but + a pseudocode line covers many, and asking per-address would mean a + lookup per instruction per repaint. + """ + out = {} + for ea in eas: + ts = self.by_ip.get(ea - self.slide) + if ts: + out[ea] = len(ts) + return out + + def first_execution(self, ea: int) -> int | None: + ts = self.executions(ea) + return ts[0] if ts else None + + def next_execution(self, ea: int, idx: int) -> int | None: + ts = self.executions(ea) + i = bisect.bisect_right(ts, idx) + return ts[i] if i < len(ts) else None + + def prev_execution(self, ea: int, idx: int) -> int | None: + ts = self.executions(ea) + i = bisect.bisect_left(ts, idx) - 1 + return ts[i] if i >= 0 else None + + # -- lining the trace up with the database ------------------------------ # + def rebase(self, db_addresses) -> int: + """Find the slide between trace addresses and database addresses. + + A traced process is relocated (ASLR, or a PIE base the database doesn't + share): our echo trace runs at 0x7ffff6faa000 while the database has the + same code at 0x2490. Nothing lines up until this is solved, so it is not + optional. + + Page offsets survive relocation — only whole pages move — so the low 12 + bits of an instruction address are invariant. Bucket the database's + addresses by those bits, and for each trace address the candidate slides + are the differences to database addresses in its bucket. The slide that + the most instructions agree on wins. + """ + buckets: dict[int, list[int]] = {} + for a in db_addresses: + buckets.setdefault(a & 0xFFF, []).append(a) + if not buckets: + return 0 + votes: dict[int, int] = {} + # A sample is enough and keeps this O(1)-ish on a 10M trace; unique + # addresses, because a hot loop shouldn't outvote the rest of the code. + for ea in list(self.by_ip)[:4096]: + for cand in buckets.get(ea & 0xFFF, ()): + votes[cand - ea] = votes.get(cand - ea, 0) + 1 + if not votes: + return 0 + best, n = max(votes.items(), key=lambda kv: kv[1]) + return best if n > 1 else 0 + + def apply_slide(self, slide: int) -> None: + self.slide = int(slide) 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:])) |