diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_blob_ui.py | 33 | ||||
| -rw-r--r-- | tests/test_formats.py | 18 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 135 | ||||
| -rw-r--r-- | tests/test_thumb_ui.py | 260 | ||||
| -rw-r--r-- | tests/test_trace.py | 203 | ||||
| -rw-r--r-- | tests/test_trace_ui.py | 374 | ||||
| -rw-r--r-- | tests/test_trace_vs_tenet.py | 189 |
7 files changed, 1206 insertions, 6 deletions
diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py index f930401..b30bba5 100644 --- a/tests/test_blob_ui.py +++ b/tests/test_blob_ui.py @@ -214,14 +214,43 @@ async def run() -> int: lst._cursor_ea() == cur_before, f"{cur_before:#x} -> {lst._cursor_ea():#x}") + # -- `p` after carving: the rest of the app must notice ------ # + # The "no functions" hint was latched at load and only cleared on a + # reload, so it kept telling you the processor/base were wrong long + # after you'd defined a function. The function index was never + # rebuilt either, which meant Ctrl+N couldn't find what `p` made. + check("no functions yet, and the hint says so", + len(app._func_index) == 0 + and "no functions" in str(app.query_one("#status", Static).render()), + f"n={len(app._func_index)}") + lst.cursor = lst.model.index_of_ea(target) + lst._scroll_cursor_into_view() + await pilot.pause(0.3) + mp = lst.model + await pilot.press("p") + await wait(lambda: lst.model is not mp and lst.model is not None, + pilot, 40) + await wait(lambda: app._func_index is not None + and len(app._func_index) > 0, pilot, 60) + check("`p` creates a function the index can see", + len(app._func_index) == 1, f"n={len(app._func_index)}") + status = str(app.query_one("#status", Static).render()) + check("and the stale 'no functions' hint is gone", + "no functions" not in status, status[:90]) + check("the status names the function it made", + "created function" in status, status[:90]) + await pilot.press("ctrl+l") opened = await wait(lambda: isinstance(app.screen, ConfirmScreen), pilot, 20) check("Ctrl+L offers to reload with different options", opened, f"screen={type(app.screen).__name__}") if opened: note = str(app.screen.query_one("#confirm-note", Static).render()) - check("and says nothing is lost when there's nothing to lose", - "nothing is lost" in note, note[:70]) + # We just made a function, so it must NOT claim nothing is lost — + # reloading throws the database away and that is now a real cost. + check("the confirmation counts what would be lost", + "1 function," in note and "nothing is lost" not in note, + note[:80]) await pilot.press("escape") await pilot.pause(0.3) check("declining leaves the binary open", diff --git a/tests/test_formats.py b/tests/test_formats.py index ab72e8d..1045703 100644 --- a/tests/test_formats.py +++ b/tests/test_formats.py @@ -77,9 +77,22 @@ def main() -> int: load_args("arm", 0, "-T binary") == "-parm -T binary") check("the processor list leads with the common targets", - [n for n, _ in PROCESSORS[:3]] == ["arm", "armb", "metapc"], + [n for n, _ in PROCESSORS[:2]] == ["arm", "arm:ARMv7-A"], f"{[n for n, _ in PROCESSORS[:3]]}") + # 32-bit ARM has to be offered SEPARATELY from bare 'arm', which gives a + # 64-bit database. That isn't cosmetic: Hex-Rays refuses a 32-bit function + # in a 64-bit database, and Thumb doesn't exist in AArch64 at all, so a + # firmware image loaded as plain 'arm' can never be decompiled — and the + # database's bitness cannot be corrected after load. + check("a 32-bit ARM variant is offered", + any(n.startswith("arm:ARMv") for n, _ in PROCESSORS), + f"{[n for n, _ in PROCESSORS if n.startswith('arm')]}") + check("and the labels say which is 32- vs 64-bit", + all(("32-bit" in d or "64-bit" in d) + for n, d in PROCESSORS if n == "arm" or n.startswith("arm:")), + f"{[(n, d) for n, d in PROCESSORS if n.startswith('arm')]}") + # Every offered name must have been checked against a real IDA, because a # wrong one is REJECTED (rc=4) with nothing useful said — handing the user a # dead end from inside the dialog meant to rescue them. This list is the @@ -89,6 +102,9 @@ def main() -> int: "arm", "armb", "metapc", "mipsl", "mipsb", "ppc", "ppcl", "sh4", "68k", "riscv", "tricore", "xtensa", "avr", "z80", "tms320c6", "m32r", "arc", "h8300", "sparcb", "sparcl", "s390", + # variants: tools/verify_procs.py checks these report the base module + # AND change the database bitness, which is the reason they exist + "arm:ARMv7-A", "arm:ARMv7-M", "arm:ARMv6-M", "arm:ARMv5TE", } offered = {n for n, _ in PROCESSORS} check("every offered processor name is IDA-verified", diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 6ccbf0d..9d77680 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -20,12 +20,14 @@ 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__)))) from idatui.app import ( # noqa: E402 - ConfirmScreen, DecompView, DisasmView, FunctionsPanel, HexView, IdaTui, + ConfirmScreen, DecompView, FunctionsPanel, HexView, IdaTui, HelpScreen, ListingView, QuitScreen, StringsPalette, StructEditor, SymbolPalette, XrefsScreen, _str_display, _word_occurrences, ) @@ -395,6 +397,84 @@ async def s_load_options(c: Ctx): await c.wait(lambda: not isinstance(app.screen, LoadOptionsScreen), 10) +@scenario("asm_highlight") +async def s_asm_highlight(c: Ctx): + """Assembly is highlighted from IDA's OWN token tags. + + generate_disasm_line() emits \x01<tag>text\x02<tag>, and the tag says what + the text is — for every processor IDA supports. Nothing is lexed here, so + what this pins is that the tags survive the trip: parsed server-side, agreed + with the plain text, carried on the Head, and mapped to a style. + """ + from idatui.app import _S_SPAN + app = c.app + await c.open_biggest("listing") + lst = c.lst + ok = await c.wait(lambda: lst.model is not None and lst.total > 20, 25) + if not ok: + c.check("a listing to highlight", False, f"total={lst.total}") + return + lst.model.ensure(400) + rows = [lst.model.get(i) for i in range(min(lst.model.loaded(), 400))] + rows = [h for h in rows if h is not None] + code = [h for h in rows if h.kind == "code"] + c.check("code rows carry IDA's token spans", + code and sum(1 for h in code if h.spans) > len(code) * 0.9, + f"{sum(1 for h in code if h.spans)}/{len(code)} have spans") + + kinds = {k for h in rows for k, _ in (h.spans or ())} + # If a tag isn't mapped it renders as body text and nothing says why, so the + # ones that carry real meaning are worth asserting explicitly. + for want in ("insn", "reg", "punct"): + c.check(f"the palette sees {want} tokens", want in kinds, f"{sorted(kinds)}") + c.check("every span kind has a style", + all(k in _S_SPAN for k in kinds), f"unstyled: {sorted(kinds - set(_S_SPAN))}") + + # Spans must describe the SAME text the row shows, or the row renders + # different characters than search/width calculations think it has. + bad = [h for h in rows if h.spans + and "".join(t for _k, t in h.spans) != h.text] + c.check("spans reconstruct the row text exactly", not bad, + f"{[(hex(h.ea), h.text) for h in bad[:2]]}") + + # And the mnemonic must be the loudest thing on the line (the column you + # scan), not just any styled token. + mn = next((h for h in code if h.spans and h.spans[0][0] == "insn"), None) + c.check("the mnemonic is the first span", + 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 @@ -1860,7 +1940,7 @@ async def s_region_define(c: Ctx): c.check("listing spans the whole segment (more heads than one function)", seg is not None and c.lst.total > 1, f"total={c.lst.total} seg={seg}") - # 'p' on the entry head (re)creates the function and UPGRADES to DisasmView + # 'p' on the entry head (re)creates the function c.lst.focus() c.lst.cursor, c.lst.cursor_x = c.lst.model.index_of_ea(addr), 0 await c.pause(0.05) @@ -1949,9 +2029,16 @@ async def s_listing_view(c: Ctx): try: c.prog.undefine(dea, size=4) c.prog.bump_items() - # reopen the listing so the model reflects the new undefined run + # Reopen the listing so the model reflects the new undefined run, and + # wait for the model to be REPLACED. Waiting on index_of_ea(dea) >= 0 is + # no test at all: dea is in the OLD model too (it's the head we just + # undefined), so the predicate passes instantly and we then assert + # against pre-edit rows. It only ever passed because the swap happened to + # win the race, and it started failing the moment page loads got bigger. + stale = c.lst.model await c.goto_ui(hex(dea)) await c.wait(lambda: app._active == "listing" and c.lst.total > 0 + and c.lst.model is not stale and c.lst.model.index_of_ea(dea) >= 0, 25) ui = c.lst.model.index_of_ea(dea) c.check("undefining a data head yields an unknown run in the listing", @@ -2216,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_thumb_ui.py b/tests/test_thumb_ui.py new file mode 100644 index 0000000..b0ca00b --- /dev/null +++ b/tests/test_thumb_ui.py @@ -0,0 +1,260 @@ +#!/usr/bin/env python3 +"""ARM/Thumb decoding switch, against real Thumb code. + +Thumb isn't a property of the bytes — it's a mode the CPU is in — so a raw image +gives IDA no way to know. At a Thumb entry point it decodes 16-bit instructions +as 32-bit ARM and produces confident nonsense, and `c` either fails or carves +garbage. `t` switches the mode. + +Uses experiments/fibonacci.bin (real Thumb), so the encodings are not a guess. +Needs IDA. ~40s. +""" +import asyncio +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from textual.widgets import Static # noqa: E402 + +from idatui.app import DecompView, IdaTui, ListingView # noqa: E402 + +PASS = FAIL = 0 +BIN = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "experiments", "fibonacci.bin") + + +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 + + +async def run() -> int: + # A fresh database every time: the T flag and the segment's addressing mode + # are SAVED in the .i64, so a previous run would answer the question for us. + for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + try: + os.remove(BIN + ext) + except OSError: + pass + + app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm") + async with app.run_test(size=(140, 44)) as pilot: + await wait(lambda: app._func_index is not None + and app._func_index.complete, pilot) + await wait(lambda: app._cur is not None, pilot, 60) + lst = app.query_one(ListingView) + lst.focus() + lst.cursor = lst.model.index_of_ea(0) + lst._scroll_cursor_into_view() + await pilot.pause(0.3) + check("starts undefined at the entry", lst.model.get(lst.cursor).kind == "unknown", + f"{lst.model.get(lst.cursor).text!r}") + + # `c` in the wrong mode: this is the failure being fixed. It must NOT + # quietly carve garbage — either it refuses, or whatever it makes is not + # the Thumb prologue. + m0 = lst.model + await pilot.press("c") + await pilot.pause(2.5) + h = lst.model.get(lst.model.index_of_ea(0)) + check("`c` alone does not produce the Thumb prologue", + h is None or h.kind != "code" or "PUSH" not in h.text.upper(), + f"{h.text if h else None!r}") + + m1 = lst.model + await pilot.press("t") + await wait(lambda: lst.model is not m1 and lst.model is not None, pilot, 60) + await pilot.pause(0.5) + + status = str(app.query_one("#status", Static).render()) + check("the status says it switched to Thumb", "Thumb" in status, status[:90]) + # Thumb doesn't exist in AArch64, and -parm on a headerless blob gives a + # 64-bit segment, so setting T alone would change nothing and look broken. + check("and says it forced the segment to 32-bit", + "32-bit" in status, status[:90]) + + m = lst.model + rows = [m.get(m.index_of_ea(ea)) for ea in (0x0, 0x2, 0x4)] + check("the entry decodes as Thumb", + rows[0] is not None and rows[0].kind == "code" + and "PUSH" in rows[0].text.upper(), + f"{rows[0].text if rows[0] else None!r}") + # 16-bit instructions: the addresses are 2 apart, which is the whole + # point — in ARM mode these would be one 4-byte instruction. + check("instructions are 16-bit wide", + all(r is not None and r.kind == "code" and r.size == 2 for r in rows), + f"{[(hex(r.ea), r.size, r.text) for r in rows if r]}") + check("and it kept disassembling past the first one", + sum(1 for i in range(20) if (m.get(i) or h).kind == "code") > 5, + "expected a run of instructions, not one") + + # Toggling back must be possible — the mode is a guess and guesses get + # revised. + m2 = lst.model + lst.cursor = lst.model.index_of_ea(0) + await pilot.press("t") + await wait(lambda: lst.model is not m2 and lst.model is not None, pilot, 60) + await pilot.pause(0.5) + status = str(app.query_one("#status", Static).render()) + check("`t` toggles back to ARM", "ARM @" in status, status[:80]) + + # -- and the reason a carved function wouldn't decompile ---------------- # + # Bare 'arm' gives a 64-BIT database. Thumb doesn't exist there, and + # Hex-Rays refuses a 32-bit function outright ("only 64-bit functions can be + # decompiled in the current database") — so `t` produces correct-looking + # disassembly that F5 can never turn into pseudocode. The database's bitness + # is fixed at load and cannot be corrected afterwards, so the only honest + # thing is to say so. + for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + try: + os.remove(BIN + ext) + except OSError: + pass + app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm") # 64-bit + async with app.run_test(size=(140, 44)) as pilot: + await wait(lambda: app._func_index is not None + and app._func_index.complete, pilot) + await wait(lambda: app._cur is not None, pilot, 60) + lst = app.query_one(ListingView) + lst.focus() + lst.cursor = lst.model.index_of_ea(0) + lst._scroll_cursor_into_view() + await pilot.pause(0.3) + m = lst.model + await pilot.press("t") + await wait(lambda: lst.model is not m and lst.model is not None, pilot, 60) + await pilot.pause(0.5) + status = str(app.query_one("#status", Static).render()) + check("a 64-bit database warns that Hex-Rays won't decompile", + "64-bit" in status and "decompile" in status, status[:120]) + check("and names the fix", "ARMv7-A" in status, status[:120]) + + # And if you ignore that and carry on, the failure has to say WHY. The + # plain decompile tool reports "Decompilation failed at 0x0" and drops + # the reason, which is the only part that tells you what to do — with it + # missing, F5 doing nothing is indistinguishable from a bug in the TUI. + lst.cursor = lst.model.index_of_ea(0) + await pilot.pause(0.2) + mp = lst.model + await pilot.press("p") + await wait(lambda: lst.model is not mp and lst.model is not None, pilot, 60) + await wait(lambda: app._func_index is not None + and len(app._func_index) > 0, pilot, 60) + await pilot.press("tab") + await wait(lambda: "cannot decompile" in + str(app.query_one("#status", Static).render()), pilot, 90) + status = str(app.query_one("#status", Static).render()) + # The message must say what to DO. Hex-Rays' own sentence ("only 64-bit + # functions can be decompiled in the current database") describes the + # database, not the fix, and is long enough that a status bar cuts off + # the end — which is where an appended hint would have lived. + check("a failed decompile names the fix, not just the diagnosis", + "Ctrl+L" in status and "ARMv7-A" in status, status[:130]) + check("and the reason survives the view reloading under it", + "cannot decompile" in status, status[:130]) + check("the message fits a narrow status bar", + len(status) < 110, f"{len(status)} chars: {status[:130]}") + + # -- the whole point: a 32-bit database decompiles ---------------------- # + for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + try: + os.remove(BIN + ext) + except OSError: + pass + app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm:ARMv7-A") + async with app.run_test(size=(140, 44)) as pilot: + await wait(lambda: app._func_index is not None + and app._func_index.complete, pilot) + # A 32-bit ARM database also lets auto-analysis do its job on Thumb code, + # which is why this one lands in the symbol picker rather than nowhere. + check("a 32-bit ARM database finds functions by itself", + len(app._func_index) > 5, f"n={len(app._func_index)}") + await pilot.press("escape") + await pilot.pause(0.5) + f = app._func_index.all_loaded()[0] + app._goto_ea(f.addr, push=True) + await wait(lambda: app._cur is not None + and app.query_one(ListingView).model is not None, pilot, 60) + app.query_one(ListingView).focus() + await pilot.press("tab") + dec = app.query_one(DecompView) + got = await wait(lambda: dec.display and dec._texts, pilot, 90) + check("Tab decompiles a Thumb function", got and len(dec._texts) > 3, + f"lines={len(dec._texts or [])}") + check("and it reads like C", + any("(" in t and ")" in t for t in (dec._texts or [])[:3]), + f"{(dec._texts or [])[:3]}") + + # -- Thumb entry points from a vector table ----------------------------- # + # An ARM function pointer carries the mode in bit 0: odd means Thumb. A + # Cortex-M vector table is therefore a list of Thumb entry points, and IDA + # won't follow them on a headerless image because nothing says those words + # are pointers at all. + vec = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "experiments", "cortexm.bin") + if not os.path.isfile(vec): + check("the cortexm fixture exists", False, vec) + else: + for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + try: + os.remove(vec + ext) + except OSError: + pass + app = IdaTui(open_path=vec, keepalive=False, load_args="-parm:ARMv7-M") + async with app.run_test(size=(140, 44)) as pilot: + await wait(lambda: app._func_index is not None + and app._func_index.complete, pilot) + check("a bare vector table gives IDA nothing to go on", + len(app._func_index) == 0, f"n={len(app._func_index)}") + if type(app.screen).__name__ != "Screen": + await pilot.press("escape") + await pilot.pause(0.5) + app._goto_ea(0, push=True) + lst = app.query_one(ListingView) + await wait(lambda: lst.model is not None, pilot, 60) + lst.focus() + lst.cursor = lst.model.index_of_ea(0) + lst._scroll_cursor_into_view() + await pilot.pause(0.3) + await pilot.press("T") + await wait(lambda: app._func_index is not None + and len(app._func_index) >= 3, pilot, 90) + names = sorted(f.name for f in app._func_index.all_loaded()) + check("scanning the table finds the Thumb handlers", + names == ["sub_200", "sub_240", "sub_280"], f"{names}") + # The table also holds an even word (the initial stack pointer), an + # even in-range word and an odd word pointing outside the image. All + # three must be ignored — marking a data word as code corrupts the + # listing, so the cost of a false positive is high. + check("and ignores the words that aren't Thumb pointers", + len(app._func_index) == 3, f"n={len(app._func_index)}") + status = str(app.query_one("#status", Static).render()) + check("the result survives the reload AND the reindex", + "3 Thumb entries" in status, status[:90]) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + if not os.path.isfile(BIN): + print(f"no such binary: {BIN}") + raise SystemExit(2) + raise SystemExit(asyncio.run(run())) 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_ui.py b/tests/test_trace_ui.py new file mode 100644 index 0000000..b08ca02 --- /dev/null +++ b/tests/test_trace_ui.py @@ -0,0 +1,374 @@ +#!/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") + + # -- 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:])) |