diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/_fixtures.py | 11 | ||||
| -rwxr-xr-x | tests/run.py | 93 | ||||
| -rw-r--r-- | tests/test_blob_ui.py | 284 | ||||
| -rw-r--r-- | tests/test_diag.py | 93 | ||||
| -rw-r--r-- | tests/test_findings.py | 219 | ||||
| -rw-r--r-- | tests/test_formats.py | 170 | ||||
| -rw-r--r-- | tests/test_graph.py | 191 | ||||
| -rw-r--r-- | tests/test_index.py | 283 | ||||
| -rw-r--r-- | tests/test_kittygfx.py | 153 | ||||
| -rw-r--r-- | tests/test_launch.py | 24 | ||||
| -rw-r--r-- | tests/test_nexus_client.py | 13 | ||||
| -rw-r--r-- | tests/test_pool.py | 176 | ||||
| -rw-r--r-- | tests/test_project.py | 229 | ||||
| -rw-r--r-- | tests/test_project_ui.py | 341 | ||||
| -rw-r--r-- | tests/test_rawimage_rpc.py | 196 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 23 | ||||
| -rw-r--r-- | tests/test_search.py | 109 | ||||
| -rw-r--r-- | tests/test_thumb_ui.py | 216 | ||||
| -rw-r--r-- | tests/test_trace.py | 248 | ||||
| -rw-r--r-- | tests/test_trace_rpc.py | 259 | ||||
| -rw-r--r-- | tests/test_trace_ui.py | 356 | ||||
| -rw-r--r-- | tests/test_trace_vs_tenet.py | 64 |
22 files changed, 2504 insertions, 1247 deletions
diff --git a/tests/_fixtures.py b/tests/_fixtures.py index bc10235..87f6184 100644 --- a/tests/_fixtures.py +++ b/tests/_fixtures.py @@ -20,6 +20,7 @@ writes back to -- turns that into a file copy. The cache is rebuilt whenever it is older than the binary, so editing a target doesn't silently test the previous one. `.pristine.i64` is gitignored. """ + from __future__ import annotations import asyncio @@ -64,10 +65,12 @@ def fast_keys() -> None: if not hasattr(textual.app, "wait_for_idle"): # pragma: no cover raise RuntimeError( "textual.app.wait_for_idle is gone -- tests/_fixtures.fast_keys " - "needs updating for this Textual version") + "needs updating for this Textual version" + ) - async def _yield_instead_of_sleeping(min_sleep: float = 0.0, - max_sleep: float = 1.0) -> None: + async def _yield_instead_of_sleeping( + min_sleep: float = 0.0, max_sleep: float = 1.0 + ) -> None: await asyncio.sleep(0) async def _press(self, *keys: str) -> None: @@ -110,7 +113,7 @@ def synthetic(name: str, build) -> str: path = os.path.join(SYNTHETIC_DIR, name) data = build() if not os.path.exists(path) or open(path, "rb").read() != data: - with open(path, "wb") as fh: # content changed -> cache is stale + with open(path, "wb") as fh: # content changed -> cache is stale fh.write(data) for stale in (cache_path(path), path + ".i64"): if os.path.exists(stale): diff --git a/tests/run.py b/tests/run.py index 5d552bb..2fb093f 100755 --- a/tests/run.py +++ b/tests/run.py @@ -38,6 +38,7 @@ Usage:: Exit code is 0 only if every file selected ran and passed. """ + from __future__ import annotations import argparse @@ -80,14 +81,17 @@ def needs_ida(path: str) -> bool: if isinstance(target, ast.Name) and target.id == "NEEDS_IDA": value = ast.literal_eval(node.value) if not isinstance(value, bool): - raise Marker(f"{os.path.basename(path)}: " - f"NEEDS_IDA must be a bool, got {value!r}") + raise Marker( + f"{os.path.basename(path)}: " + f"NEEDS_IDA must be a bool, got {value!r}" + ) return value raise Marker( f"{os.path.basename(path)}: no NEEDS_IDA marker.\n" f" Add `NEEDS_IDA = True` (spawns a worker / drives the pilot) or\n" f" `NEEDS_IDA = False` (pure: stdlib, no IDA, runs anywhere) at module\n" - f" scope, so tests/run.py --fast knows whether it can run you.") + f" scope, so tests/run.py --fast knows whether it can run you." + ) def discover() -> list[tuple[str, bool]]: @@ -124,41 +128,70 @@ def tally(output: str) -> tuple[int, int] | None: def run_one(path: str, python: str, extra: list[str], echo: bool) -> dict: """Run one test file as a subprocess and summarise it.""" - name = os.path.basename(path)[len("test_"):-len(".py")] + name = os.path.basename(path)[len("test_") : -len(".py")] started = time.time() - proc = subprocess.run([python, path, *extra], cwd=ROOT, - capture_output=not echo, text=True) + proc = subprocess.run( + [python, path, *extra], cwd=ROOT, capture_output=not echo, text=True + ) took = time.time() - started out = "" if echo else (proc.stdout or "") + (proc.stderr or "") counts = tally(out) skipped = bool(_SKIP.search(out)) and (counts is None or counts == (0, 0)) return { - "name": name, "path": path, "code": proc.returncode, "took": took, + "name": name, + "path": path, + "code": proc.returncode, + "took": took, "passed": counts[0] if counts else 0, "failed": counts[1] if counts else 0, "counted": counts is not None, - "skipped": skipped, "output": out, + "skipped": skipped, + "output": out, } def main(argv: list[str]) -> int: ap = argparse.ArgumentParser( - prog="tests/run.py", description=__doc__, - formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("only", nargs="*", metavar="SUBSTR", - help="only run test files whose name contains one of these") - ap.add_argument("--fast", action="store_true", - help="skip every file that needs IDA (seconds, runs anywhere)") - ap.add_argument("--ida-only", action="store_true", - help="only the files that need IDA") - ap.add_argument("--list", action="store_true", - help="show what would run, and whether it needs IDA") - ap.add_argument("-x", "--exitfirst", action="store_true", - help="stop after the first failing file") - ap.add_argument("-v", "--verbose", action="store_true", - help="stream each suite's output instead of capturing it") - ap.add_argument("--python", default=os.environ.get("IDATUI_PYTHON", DEFAULT_PY), - help=f"interpreter for the IDA suites (default {DEFAULT_PY})") + prog="tests/run.py", + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + ap.add_argument( + "only", + nargs="*", + metavar="SUBSTR", + help="only run test files whose name contains one of these", + ) + ap.add_argument( + "--fast", + action="store_true", + help="skip every file that needs IDA (seconds, runs anywhere)", + ) + ap.add_argument( + "--ida-only", action="store_true", help="only the files that need IDA" + ) + ap.add_argument( + "--list", + action="store_true", + help="show what would run, and whether it needs IDA", + ) + ap.add_argument( + "-x", + "--exitfirst", + action="store_true", + help="stop after the first failing file", + ) + ap.add_argument( + "-v", + "--verbose", + action="store_true", + help="stream each suite's output instead of capturing it", + ) + ap.add_argument( + "--python", + default=os.environ.get("IDATUI_PYTHON", DEFAULT_PY), + help=f"interpreter for the IDA suites (default {DEFAULT_PY})", + ) args, extra = ap.parse_known_args(argv) try: @@ -190,10 +223,12 @@ def main(argv: list[str]) -> int: # an IDA file needs the interpreter that has textual + idapro. pure_py = sys.executable if any(ida for _, ida in selected) and not os.path.exists(args.python): - print(f"error: {args.python} not found — the IDA suites need an " - f"interpreter with textual + idapro.\n" - f" Pass --python, set $IDATUI_PYTHON, or use --fast.", - file=sys.stderr) + print( + f"error: {args.python} not found — the IDA suites need an " + f"interpreter with textual + idapro.\n" + f" Pass --python, set $IDATUI_PYTHON, or use --fast.", + file=sys.stderr, + ) return 2 results = [] @@ -226,7 +261,7 @@ def main(argv: list[str]) -> int: elif r["code"] != 0 or r["failed"]: state = "\033[31mFAIL\033[0m" elif not r["counted"]: - state = "\033[33m ? \033[0m" # exit 0 but printed no tally + state = "\033[33m ? \033[0m" # exit 0 but printed no tally else: state = "\033[32m ok \033[0m" detail = f"{r['passed']:4d} passed" diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py index d4740d0..7c3aee5 100644 --- a/tests/test_blob_ui.py +++ b/tests/test_blob_ui.py @@ -19,12 +19,12 @@ import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _fixtures import fast_keys, staged, synthetic # noqa: E402 from textual.widgets import Input, Static # noqa: E402 from idatui.app import ConfirmScreen, IdaTui, ListingView # noqa: E402 -from _fixtures import fast_keys, staged, synthetic # noqa: E402 -fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui._sync import settle # noqa: E402 PASS = FAIL = 0 @@ -66,14 +66,19 @@ def _blob_bytes() -> bytes: functions" is asserted below. """ import random + data = bytearray(random.Random(0xB10BCAFE).randbytes(64 * 1024)) - # -parm puts IDA in AArch64 mode, so these are A64 encodings; the ARM32 - # spelling of a nop (0xE1A00000) is NOT decodable there and made this - # test fail for a reason that had nothing to do with what it checks. - for k, insn in enumerate((0xD503201F, # nop - 0xD503201F, # nop - 0xD65F03C0)): # ret <- the run must stop here - data[PLANTED + k * 4:PLANTED + k * 4 + 4] = insn.to_bytes(4, "little") + # -parm puts IDA in AArch64 mode, so these are A64 encodings; the ARM32 + # spelling of a nop (0xE1A00000) is NOT decodable there and made this + # test fail for a reason that had nothing to do with what it checks. + for k, insn in enumerate( + ( + 0xD503201F, # nop + 0xD503201F, # nop + 0xD65F03C0, + ) + ): # ret <- the run must stop here + data[PLANTED + k * 4 : PLANTED + k * 4 + 4] = insn.to_bytes(4, "little") return bytes(data) @@ -109,27 +114,45 @@ async def run() -> int: # happens AFTER a described blob turns out to contain nothing. app = _blob_app(blob) async with app.run_test(size=(140, 44)) as pilot: - ok = await wait(lambda: app._func_index is not None - and app._func_index.complete, pilot) + ok = await wait( + lambda: app._func_index is not None and app._func_index.complete, pilot + ) check("a random blob still finishes loading", ok) - check("and really has no functions", len(app._func_index) == 0, - f"n={len(app._func_index)}") + check( + "and really has no functions", + len(app._func_index) == 0, + f"n={len(app._func_index)}", + ) landed = await wait(lambda: app._cur is not None, pilot, 60) - check("it lands somewhere instead of leaving empty panes", landed, - f"cur={app._cur}") + check( + "it lands somewhere instead of leaving empty panes", + landed, + f"cur={app._cur}", + ) lst = app.query_one(ListingView) - check("the listing actually has rows to show", - lst.total > 0, f"total={lst.total}") - check("landed at the image base", app._cur is not None - and app._cur.ea == 0x4000, f"{app._cur.ea if app._cur else None:#x}") + check( + "the listing actually has rows to show", + lst.total > 0, + f"total={lst.total}", + ) + check( + "landed at the image base", + app._cur is not None and app._cur.ea == 0x4000, + f"{app._cur.ea if app._cur else None:#x}", + ) status = str(app.query_one("#status", Static).render()) - check("the status says there are no functions (not 'still loading')", - "no functions" in status and "still loading" not in status, - status[:90]) - check("and points at the likely cause", - "processor" in status and "Ctrl+L" in status, status[:90]) + check( + "the status says there are no functions (not 'still loading')", + "no functions" in status and "still loading" not in status, + status[:90], + ) + check( + "and points at the likely cause", + "processor" in status and "Ctrl+L" in status, + status[:90], + ) # The hint is a property of the database, so it must survive moving # around — an earlier version wrote it once and the next status @@ -139,8 +162,9 @@ async def run() -> int: await pilot.press("down") await settle(app) status2 = str(app.query_one("#status", Static).render()) - check("the hint survives navigating", "no functions" in status2, - status2[:90]) + check( + "the hint survives navigating", "no functions" in status2, status2[:90] + ) # -- byte-granular carving ---------------------------------- # # An undefined run arrives as ONE head ("db N dup(?)"). It has to @@ -148,47 +172,64 @@ async def run() -> int: # press `c` at, which is how IDA works and the only way to find an # instruction stream that doesn't start at the run's first byte. m = lst.model - check("an undefined run presents one row per byte", - m.loaded() >= 4096 and len(m._heads) < 64, - f"rows={m.loaded()} physical heads={len(m._heads)}") + check( + "an undefined run presents one row per byte", + m.loaded() >= 4096 and len(m._heads) < 64, + f"rows={m.loaded()} physical heads={len(m._heads)}", + ) rows = m.window(0, 4) - check("each row is a single addressable byte", - [h.ea for h in rows] == [0x4000, 0x4001, 0x4002, 0x4003] - and all(h.size == 1 for h in rows), - f"{[(hex(h.ea), h.size) for h in rows]}") - check("and shows its value, not a placeholder", - all(h.text.startswith("db ") and "dup" not in h.text - for h in rows), f"{[h.text for h in rows]}") + check( + "each row is a single addressable byte", + [h.ea for h in rows] == [0x4000, 0x4001, 0x4002, 0x4003] + and all(h.size == 1 for h in rows), + f"{[(hex(h.ea), h.size) for h in rows]}", + ) + check( + "and shows its value, not a placeholder", + all(h.text.startswith("db ") and "dup" not in h.text for h in rows), + f"{[h.text for h in rows]}", + ) # The point of all this: land on an arbitrary byte and convert it. i = m.index_of_ea(0x4021) - check("an address inside the run resolves to its own row", - i >= 0 and m.get(i).ea == 0x4021, - f"row={i} ea={m.get(i).ea if i >= 0 else None}") + check( + "an address inside the run resolves to its own row", + i >= 0 and m.get(i).ea == 0x4021, + f"row={i} ea={m.get(i).ea if i >= 0 else None}", + ) - target = 0x4000 + PLANTED # a NOP we put there ourselves + target = 0x4000 + PLANTED # a NOP we put there ourselves lst.cursor = m.index_of_ea(target) lst._scroll_cursor_into_view() await settle(app, lambda: lst._cursor_ea() == target) - check("the cursor sits on the byte we aimed at", - lst._cursor_ea() == target, - f"{lst._cursor_ea():#x} want {target:#x}") + check( + "the cursor sits on the byte we aimed at", + lst._cursor_ea() == target, + f"{lst._cursor_ea():#x} want {target:#x}", + ) await pilot.press("c") # settle(), not a fixed sleep AND not a bare predicate: an edit can # look done for a moment and then be replaced when a queued listing # rebuild lands, so the gate has to be "the row is code AND the app # has stopped working". settle() is the same helper the app's own # RPC layer uses, so tests and driver agree on what "done" means. - await settle(app, lambda: (lambda h: h is not None and h.kind == "code")( - head_at(lst, target)), timeout=30) + await settle( + app, + lambda: (lambda h: h is not None and h.kind == "code")( + head_at(lst, target) + ), + timeout=30, + ) # Re-read the model: defining an item rebuilds it, and holding the # old object shows pre-edit rows -- which looks exactly like the # edit silently failing. m = lst.model h = head_at(lst, target) - check("`c` on a chosen byte carves an instruction there", - h is not None and h.kind == "code", - f"kind={h.kind if h else None} text={h.text if h else None!r}") + check( + "`c` on a chosen byte carves an instruction there", + h is not None and h.kind == "code", + f"kind={h.kind if h else None} text={h.text if h else None!r}", + ) if h is not None and h.kind == "code": print(f" carved {target:#x}: {h.text}") # `c` runs until something stops it, like IDA — one instruction @@ -197,21 +238,32 @@ async def run() -> int: # four and stop AT the ret, not run on into the random bytes # after it. run = [m.get(m.index_of_ea(target + k * 4)) for k in range(3)] - check("`c` keeps going until control flow ends", - all(x is not None and x.kind == "code" for x in run), - f"{[(hex(x.ea), x.kind) for x in run if x]}") - check("and stops at the ret instead of running into junk", - m.get(m.index_of_ea(target + 12)).kind == "unknown", - f"{m.get(m.index_of_ea(target + 12)).text!r}") + check( + "`c` keeps going until control flow ends", + all(x is not None and x.kind == "code" for x in run), + f"{[(hex(x.ea), x.kind) for x in run if x]}", + ) + check( + "and stops at the ret instead of running into junk", + m.get(m.index_of_ea(target + 12)).kind == "unknown", + f"{m.get(m.index_of_ea(target + 12)).text!r}", + ) status = str(app.query_one("#status", Static).render()) - check("the status reports what the run did", - "3 instructions" in status and "control flow" in status, - status[:80]) - check("the carved row spans the instruction, not one byte", - h.size == 4, f"size={h.size}") - check("bytes before it stay individually addressable", - m.get(m.index_of_ea(target - 1)).size == 1 - and m.get(m.index_of_ea(target - 1)).ea == target - 1) + check( + "the status reports what the run did", + "3 instructions" in status and "control flow" in status, + status[:80], + ) + check( + "the carved row spans the instruction, not one byte", + h.size == 4, + f"size={h.size}", + ) + check( + "bytes before it stay individually addressable", + m.get(m.index_of_ea(target - 1)).size == 1 + and m.get(m.index_of_ea(target - 1)).ea == target - 1, + ) # -- an edit must not move the view -------------------------- # # Every mutation rebuilds the model, and row indices don't survive @@ -239,14 +291,17 @@ async def run() -> int: # (The listing re-renders its text lazily, so the comment is not # necessarily visible in model rows the moment the worker returns -- # which is why this waits for the app, not for the text.) - await settle(app, lambda: not app.query_one("#comment", Input).display, - timeout=30) - check("commenting leaves the view where it was", - lst.model.get(round(lst.scroll_offset.y)).ea == ctop - and lst._cursor_ea() == ccur, - f"top {ctop:#x} -> " - f"{lst.model.get(round(lst.scroll_offset.y)).ea:#x}, " - f"cursor {ccur:#x} -> {lst._cursor_ea():#x}") + await settle( + app, lambda: not app.query_one("#comment", Input).display, timeout=30 + ) + check( + "commenting leaves the view where it was", + lst.model.get(round(lst.scroll_offset.y)).ea == ctop + and lst._cursor_ea() == ccur, + f"top {ctop:#x} -> " + f"{lst.model.get(round(lst.scroll_offset.y)).ea:#x}, " + f"cursor {ccur:#x} -> {lst._cursor_ea():#x}", + ) # -- carving must not move the view -------------------------- # # Defining code collapses rows (four byte rows become one @@ -259,8 +314,11 @@ async def run() -> int: await settle(app, lambda: lst._cursor_ea() == far) top_before = lst.model.get(round(lst.scroll_offset.y)).ea cur_before = lst._cursor_ea() - check("scrolled somewhere with rows above us", - round(lst.scroll_offset.y) > 0, f"top={lst.scroll_offset.y}") + check( + "scrolled somewhere with rows above us", + round(lst.scroll_offset.y) > 0, + f"top={lst.scroll_offset.y}", + ) await pilot.press("c") # No predicate here on purpose: this spot is random data, so the # carve may legitimately produce nothing and "the row became code" @@ -270,54 +328,80 @@ async def run() -> int: await settle(app, timeout=30) m2 = lst.model top_after = m2.get(round(lst.scroll_offset.y)).ea - check("carving leaves the scroll position where it was", - top_after == top_before, - f"{top_before:#x} -> {top_after:#x}") - check("and leaves the cursor on the same address", - lst._cursor_ea() == cur_before, - f"{cur_before:#x} -> {lst._cursor_ea():#x}") + check( + "carving leaves the scroll position where it was", + top_after == top_before, + f"{top_before:#x} -> {top_after:#x}", + ) + check( + "and leaves the cursor on the same address", + 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)}") + 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 settle(app, lambda: lst._cursor_ea() == target) 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)}") + 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]) + 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__}") + 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()) # 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]) + 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 settle(app, lambda: not isinstance(app.screen, ConfirmScreen)) - check("declining leaves the binary open", - not isinstance(app.screen, ConfirmScreen) and app._cur is not None) + check( + "declining leaves the binary open", + not isinstance(app.screen, ConfirmScreen) and app._cur is not None, + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_diag.py b/tests/test_diag.py index e3b966a..d1b480a 100644 --- a/tests/test_diag.py +++ b/tests/test_diag.py @@ -3,6 +3,7 @@ Pure: no IDA, no worker, no Textual. """ + from __future__ import annotations import os @@ -41,10 +42,16 @@ def t_swallow_keeps_going(): r = diag.recent() check("the error is recorded", len(r) == 1, str(r)) check("with what was being attempted", r[0]["what"] == "a thing", str(r[0])) - check("and the exception type and message", - r[0]["error"] == "ValueError: nope", r[0]["error"]) - check("and where it was actually raised", - r[0]["where"].startswith("test_diag.py:"), r[0]["where"]) + check( + "and the exception type and message", + r[0]["error"] == "ValueError: nope", + r[0]["error"], + ) + check( + "and where it was actually raised", + r[0]["where"].startswith("test_diag.py:"), + r[0]["where"], + ) def t_reraise(): @@ -61,8 +68,11 @@ def t_reraise(): check("reraise lets the listed type through", False, "not raised") except Wanted: check("reraise lets the listed type through", True) - check("and a reraised error is not recorded twice", - diag.recent() == [], str(diag.recent())) + check( + "and a reraised error is not recorded twice", + diag.recent() == [], + str(diag.recent()), + ) with diag.swallow("still swallows others", reraise=(Wanted,)): raise ValueError("other") check("other types are still swallowed", len(diag.recent()) == 1) @@ -74,12 +84,21 @@ def t_ring_is_bounded(): diag.note(f"item {i}", RuntimeError(str(i))) r = diag.recent(1000) check("the ring is bounded", len(r) == diag._MAX, f"{len(r)}") - check("it keeps the NEWEST entries", - r[-1]["what"] == f"item {diag._MAX + 24}", r[-1]["what"]) - check("recent(n) returns the last n, newest last", - [e["what"] for e in diag.recent(3)] - == [f"item {diag._MAX + 22}", f"item {diag._MAX + 23}", - f"item {diag._MAX + 24}"], str(diag.recent(3))) + check( + "it keeps the NEWEST entries", + r[-1]["what"] == f"item {diag._MAX + 24}", + r[-1]["what"], + ) + check( + "recent(n) returns the last n, newest last", + [e["what"] for e in diag.recent(3)] + == [ + f"item {diag._MAX + 22}", + f"item {diag._MAX + 23}", + f"item {diag._MAX + 24}", + ], + str(diag.recent(3)), + ) def t_log_file(): @@ -95,8 +114,11 @@ def t_log_file(): body = open(path, encoding="utf-8").read() check("the log records what was attempted", "logged thing" in body, body[:200]) check("and the error", "KeyError" in body, body[:200]) - check("and a traceback, which the ring doesn't carry", - "Traceback" in body and "t_log_file" in body, body[:300]) + check( + "and a traceback, which the ring doesn't carry", + "Traceback" in body and "t_log_file" in body, + body[:300], + ) def t_log_is_off_by_default(): @@ -104,8 +126,10 @@ def t_log_is_off_by_default(): os.environ.pop("IDATUI_LOG", None) with diag.swallow("unlogged"): raise ValueError("x") - check("without $IDATUI_LOG nothing is written, but the ring still has it", - len(diag.recent()) == 1) + check( + "without $IDATUI_LOG nothing is written, but the ring still has it", + len(diag.recent()) == 1, + ) def t_broken_log_path_is_harmless(): @@ -116,8 +140,7 @@ def t_broken_log_path_is_harmless(): with diag.swallow("still fine"): raise ValueError("boom") check("an unwritable log path doesn't raise", True) - check("and the error is still recorded in the ring", - len(diag.recent()) == 1) + check("and the error is still recorded in the ring", len(diag.recent()) == 1) finally: os.environ.pop("IDATUI_LOG", None) @@ -128,41 +151,55 @@ def t_env_read_per_call(): diag.clear() with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "late.log") - os.environ["IDATUI_LOG"] = path # set AFTER import + os.environ["IDATUI_LOG"] = path # set AFTER import try: diag.log("hello") finally: os.environ.pop("IDATUI_LOG", None) - check("a log path set after import is honoured", - os.path.exists(path) and "hello" in open(path).read()) + check( + "a log path set after import is honoured", + os.path.exists(path) and "hello" in open(path).read(), + ) def t_thread_safe(): diag.clear() + def go(n): for i in range(40): diag.note(f"t{n}-{i}", RuntimeError("x")) + ts = [threading.Thread(target=go, args=(n,)) for n in range(6)] for t in ts: t.start() for t in ts: t.join(10) r = diag.recent(1000) - check("concurrent notes don't corrupt the ring", - len(r) == diag._MAX and all("what" in e for e in r), f"{len(r)}") - check("the recording thread is captured", - all(e["thread"] for e in r)) + check( + "concurrent notes don't corrupt the ring", + len(r) == diag._MAX and all("what" in e for e in r), + f"{len(r)}", + ) + check("the recording thread is captured", all(e["thread"] for e in r)) def main() -> int: - for fn in (t_swallow_keeps_going, t_reraise, t_ring_is_bounded, t_log_file, - t_log_is_off_by_default, t_broken_log_path_is_harmless, - t_env_read_per_call, t_thread_safe): + for fn in ( + t_swallow_keeps_going, + t_reraise, + t_ring_is_bounded, + t_log_file, + t_log_is_off_by_default, + t_broken_log_path_is_harmless, + t_env_read_per_call, + t_thread_safe, + ): print(f"\n{fn.__name__}") try: fn() except Exception as e: # noqa: BLE001 import traceback + check(f"{fn.__name__} did not crash", False, f"{type(e).__name__}: {e}") traceback.print_exc() diag.clear() diff --git a/tests/test_findings.py b/tests/test_findings.py index cf813c2..205b005 100644 --- a/tests/test_findings.py +++ b/tests/test_findings.py @@ -18,7 +18,12 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from idatui.domain import Comment, NamedItem, Struct # noqa: E402 from idatui.findings import ( # noqa: E402 - Findings, default_path, from_loader, gather, is_dummy, render, + Findings, + default_path, + from_loader, + gather, + is_dummy, + render, ) PASS = FAIL = 0 @@ -36,34 +41,62 @@ def check(name, cond, detail=""): def sample() -> Findings: return Findings( - binary="echo", path="/tmp/echo", + binary="echo", + path="/tmp/echo", sections=[(0x1000, 0x2000, ".text"), (0x2000, 0x2100, ".data")], n_functions=128, comments=[ - Comment(addr=0x1100, text="length is attacker controlled", - line="mov edi, [rbp+len]", func="parse", func_addr=0x1000), - Comment(addr=0x1010, text="entry", line="push rbp", - func="parse", func_addr=0x1000), - Comment(addr=0x1000, text="parses the header", whole_func=True, - func="parse", func_addr=0x1000), + Comment( + addr=0x1100, + text="length is attacker controlled", + line="mov edi, [rbp+len]", + func="parse", + func_addr=0x1000, + ), + Comment( + addr=0x1010, + text="entry", + line="push rbp", + func="parse", + func_addr=0x1000, + ), + Comment( + addr=0x1000, + text="parses the header", + whole_func=True, + func="parse", + func_addr=0x1000, + ), Comment(addr=0x2004, text="magic", line="dd 0DEADBEEFh"), # What the ELF loader writes into every database, through the very # same set_cmt a person uses. - Comment(addr=0x4, text="File class: 64-bit", line="db 2", - seg="LOAD"), + Comment(addr=0x4, text="File class: 64-bit", line="db 2", seg="LOAD"), ], names=[ - NamedItem(addr=0x1000, name="parse", is_func=True, size=0x120, - proto="int __fastcall parse(char *)"), + NamedItem( + addr=0x1000, + name="parse", + is_func=True, + size=0x120, + proto="int __fastcall parse(char *)", + ), NamedItem(addr=0x1200, name="sub_1200", is_func=True, size=0x30), NamedItem(addr=0x2004, name="hdr_magic", seg=".data"), NamedItem(addr=0x1400, name="memcpy", is_func=True, size=0x40), NamedItem(addr=0x390, name="elf_gnu_hash_nbuckets", seg="LOAD"), ], - types=[(Struct(name="hdr", size=0x10, is_union=False, members=3, - ordinal=42), "struct hdr\n{\n int magic;\n};\n"), - (Struct(name="Elf64_Dyn", size=0x10, is_union=False, members=2, - ordinal=3), "struct Elf64_Dyn\n{\n int d_tag;\n};\n")], + types=[ + ( + Struct(name="hdr", size=0x10, is_union=False, members=3, ordinal=42), + "struct hdr\n{\n int magic;\n};\n", + ), + ( + Struct( + name="Elf64_Dyn", size=0x10, is_union=False, members=2, ordinal=3 + ), + "struct Elf64_Dyn\n{\n int d_tag;\n};\n", + ), + ], linked={"memcpy"}, stripped=True, ) @@ -74,9 +107,11 @@ def main() -> int: check("the report names the binary", doc.startswith("# Findings — echo"), doc[:40]) # 1 function, not 3: sub_1200 is IDA's invention and memcpy is the linker's. - check("the summary counts only what a person contributed", - "1 named functions · 1 named data · 4 comments · 2 local types" in doc, - doc.splitlines()[2] if len(doc.splitlines()) > 2 else "") + check( + "the summary counts only what a person contributed", + "1 named functions · 1 named data · 4 comments · 2 local types" in doc, + doc.splitlines()[2] if len(doc.splitlines()) > 2 else "", + ) # A name IDA invented is not a finding, and neither is one the linker gave. check("dummy names are excluded", "sub_1200" not in doc) @@ -86,75 +121,111 @@ def main() -> int: # The loader annotates every database it makes; none of it is a finding. check("the loader's own comments are left out", "File class" not in doc) check("the loader's own names are left out", "elf_gnu_hash" not in doc) - check("but the report says how many it dropped", - "4 annotations left out as the loader's own" in doc, # 1 comment + 3 names - [l for l in doc.splitlines() if "left out" in l]) - check("from_loader knows both shapes", - from_loader("LOAD") and from_loader("", "elf_gnu_hash_x") - and not from_loader(".text", "parse")) - check("is_dummy knows the shapes IDA invents", - all(is_dummy(n) for n in ("sub_1234", "loc_A0", "unk_4000", "j_free")) - and not any(is_dummy(n) for n in ("parse", "sub_parse", "main", "")), - "") + check( + "but the report says how many it dropped", + "4 annotations left out as the loader's own" in doc, # 1 comment + 3 names + [l for l in doc.splitlines() if "left out" in l], + ) + check( + "from_loader knows both shapes", + from_loader("LOAD") + and from_loader("", "elf_gnu_hash_x") + and not from_loader(".text", "parse"), + ) + check( + "is_dummy knows the shapes IDA invents", + all(is_dummy(n) for n in ("sub_1234", "loc_A0", "unk_4000", "j_free")) + and not any(is_dummy(n) for n in ("parse", "sub_parse", "main", "")), + "", + ) # Comments lead, grouped by function, address-ordered within a group. - check("comments come before the name tables", - doc.index("## Comments") < doc.index("## Named functions")) - body = doc[doc.index("## Comments"):doc.index("## Named functions")] + check( + "comments come before the name tables", + doc.index("## Comments") < doc.index("## Named functions"), + ) + body = doc[doc.index("## Comments") : doc.index("## Named functions")] check("comments are grouped under their function", "### `parse`" in body) - check("a commentless region is grouped separately", - "### outside any function" in body) - check("comments are ordered by address inside a group", - body.index("0x1010") < body.index("0x1100")) - check("a function comment says that is what it is", - "*whole function*: parses the header" in body) - check("an instruction comment carries the line it annotates", - "`mov edi, [rbp+len]`" in body) + check( + "a commentless region is grouped separately", "### outside any function" in body + ) + check( + "comments are ordered by address inside a group", + body.index("0x1010") < body.index("0x1100"), + ) + check( + "a function comment says that is what it is", + "*whole function*: parses the header" in body, + ) + check( + "an instruction comment carries the line it annotates", + "`mov edi, [rbp+len]`" in body, + ) # Types: newest ordinal first, because that is the one you just wrote. - types = doc[doc.index("## Local types"):] - check("your newest type is first", - types.index("hdr") < types.index("Elf64_Dyn")) + types = doc[doc.index("## Local types") :] + check("your newest type is first", types.index("hdr") < types.index("Elf64_Dyn")) check("type source is fenced as C", "```c\nstruct hdr" in types) # Escaping. - hostile = Findings(binary="x", comments=[ - Comment(addr=1, text="a | b", line="mov | rax"), - ], names=[NamedItem(addr=2, name="a|b")]) + hostile = Findings( + binary="x", + comments=[ + Comment(addr=1, text="a | b", line="mov | rax"), + ], + names=[NamedItem(addr=2, name="a|b")], + ) hdoc = render(hostile) check("a pipe cannot break a table row", "a\\|b" in hdoc, hdoc) check("a pipe in a comment is escaped too", "a \\| b" in hdoc) # The empty database must still produce a document that says something. empty = render(Findings(binary="nothing")) - check("an empty report is still a document", - empty.startswith("# Findings — nothing") and "## Comments" in empty) + check( + "an empty report is still a document", + empty.startswith("# Findings — nothing") and "## Comments" in empty, + ) check("and it says why it is empty", "Comments are the part" in empty) - check("an empty report has no dangling type section", - "## Local types" not in empty) + check("an empty report has no dangling type section", "## Local types" not in empty) # Provenance must be stated, not implied. Without a journal the report is a # scan and says so; with one it is exactly what idatui recorded doing. - scanned = render(Findings(binary="x", stripped=False, - names=[NamedItem(addr=1, name="main", is_func=True)])) - check("a scanned report admits it cannot know who wrote what", - "**source**: a scan of the database" in scanned - and "include its work as well as yours" in scanned) - check("and warns when the binary brought its own symbols", - "include ones it shipped with" in scanned) + scanned = render( + Findings( + binary="x", + stripped=False, + names=[NamedItem(addr=1, name="main", is_func=True)], + ) + ) + check( + "a scanned report admits it cannot know who wrote what", + "**source**: a scan of the database" in scanned + and "include its work as well as yours" in scanned, + ) + check( + "and warns when the binary brought its own symbols", + "include ones it shipped with" in scanned, + ) j = sample() j.recorded = {0x1100, 0x1000} j.n_recorded = 7 jdoc = render(j) - check("a journalled report says so", "idatui's edit journal" in jdoc - and "7 recorded edits" in jdoc, "") - jbody = jdoc[jdoc.index("## Comments"):jdoc.index("## Named functions")] - check("and lists only the comments it recorded", - "length is attacker controlled" in jbody and "0x2004" not in jbody, - jbody) - check("a journalled report drops names it did not record", - "`parse`" in jdoc and "hdr_magic" not in jdoc) + check( + "a journalled report says so", + "idatui's edit journal" in jdoc and "7 recorded edits" in jdoc, + "", + ) + jbody = jdoc[jdoc.index("## Comments") : jdoc.index("## Named functions")] + check( + "and lists only the comments it recorded", + "length is attacker controlled" in jbody and "0x2004" not in jbody, + jbody, + ) + check( + "a journalled report drops names it did not record", + "`parse`" in jdoc and "hdr_magic" not in jdoc, + ) # -- gather ------------------------------------------------------------- # class FakeProgram: @@ -162,8 +233,10 @@ def main() -> int: return [(0x1000, 0x2000, ".text")] def annotations(self, limit=4000): - return ([Comment(addr=1, text="hi")], - [NamedItem(addr=1, name="parse", is_func=True)]) + return ( + [Comment(addr=1, text="hi")], + [NamedItem(addr=1, name="parse", is_func=True)], + ) def linkage(self): return ([], []) @@ -180,11 +253,15 @@ def main() -> int: f = gather(FakeProgram(), "/tmp/echo") check("gather reads the annotations", len(f.comments) == 1 and len(f.names) == 1) check("gather takes the binary name from the path", f.binary == "echo", f.binary) - check("a failing backend degrades the report instead of raising", - f.n_functions == 0 and f.types == [] and "# Findings" in render(f)) + check( + "a failing backend degrades the report instead of raising", + f.n_functions == 0 and f.types == [] and "# Findings" in render(f), + ) - check("the default path sits beside the binary", - default_path("/tmp/echo") == "/tmp/echo.findings.md") + check( + "the default path sits beside the binary", + default_path("/tmp/echo") == "/tmp/echo.findings.md", + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_formats.py b/tests/test_formats.py index c0ad542..05d98ba 100644 --- a/tests/test_formats.py +++ b/tests/test_formats.py @@ -10,8 +10,12 @@ import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.formats import (PROCESSORS, load_args, # noqa: E402 - needs_load_options, sniff) +from idatui.formats import ( # noqa: E402 + PROCESSORS, + load_args, + needs_load_options, + sniff, +) PASS = FAIL = 0 @@ -28,6 +32,7 @@ def check(name, ok, detail=""): def main() -> int: with tempfile.TemporaryDirectory() as tmp: + def w(name, data): p = os.path.join(tmp, name) with open(p, "wb") as f: @@ -35,67 +40,102 @@ def main() -> int: return p # -- formats IDA can load on its own: never interrupt the user -------- # - for name, head in (("elf", b"\x7fELF\x02\x01\x01"), ("pe", b"MZ\x90\x00"), - ("macho", b"\xcf\xfa\xed\xfe"), ("dex", b"dex\n035\x00"), - ("wasm", b"\x00asm\x01\x00")): + for name, head in ( + ("elf", b"\x7fELF\x02\x01\x01"), + ("pe", b"MZ\x90\x00"), + ("macho", b"\xcf\xfa\xed\xfe"), + ("dex", b"dex\n035\x00"), + ("wasm", b"\x00asm\x01\x00"), + ): p = w(name, head + bytes(60)) - check(f"{name} is recognised (no dialog)", - sniff(p) is not None and not needs_load_options(p), f"{sniff(p)}") + check( + f"{name} is recognised (no dialog)", + sniff(p) is not None and not needs_load_options(p), + f"{sniff(p)}", + ) # -- the case this exists for ----------------------------------------- # blob = w("fw.bin", bytes(range(256)) * 4) - check("a headerless blob is not recognised (ask)", - sniff(blob) is None and needs_load_options(blob)) + check( + "a headerless blob is not recognised (ask)", + sniff(blob) is None and needs_load_options(blob), + ) # Intel HEX / S-records are text containers IDA does load. hexf = w("f.hex", b":10010000214601360121470136007EFE09D2190140\n") check("Intel HEX is recognised", sniff(hexf) == "Intel HEX", f"{sniff(hexf)}") - srec = w("f.s19", b"S00600004844521B\nS1130000285F245F2212226A000424290008237C\n") - check("Motorola S-records are recognised", - sniff(srec) == "Motorola S-record", f"{sniff(srec)}") + srec = w( + "f.s19", b"S00600004844521B\nS1130000285F245F2212226A000424290008237C\n" + ) + check( + "Motorola S-records are recognised", + sniff(srec) == "Motorola S-record", + f"{sniff(srec)}", + ) # A binary that merely STARTS with ':' is not Intel HEX. Text formats are # only accepted when the whole head is printable, or half the firmware in # the world gets mis-detected on one byte. colon = w("colon.bin", b":\x00\xff\xfe\x01\x02" + bytes(58)) - check("a blob starting with ':' is not mistaken for Intel HEX", - sniff(colon) is None, f"{sniff(colon)}") + check( + "a blob starting with ':' is not mistaken for Intel HEX", + sniff(colon) is None, + f"{sniff(colon)}", + ) check("an empty file is not recognised", sniff(w("empty", b"")) is None) - check("a missing file never asks (nothing to load)", - not needs_load_options(os.path.join(tmp, "nope"))) + check( + "a missing file never asks (nothing to load)", + not needs_load_options(os.path.join(tmp, "nope")), + ) check("a directory never asks", not needs_load_options(tmp)) # -- switch construction ------------------------------------------------- # # -b is in PARAGRAPHS: 0x8000000 >> 4 == 0x800000. Getting this wrong loads # the image 16x off and every address in the database is wrong. - check("base is converted to paragraphs", - load_args("arm", 0x8000000) == "-parm -b800000", - load_args("arm", 0x8000000)) + check( + "base is converted to paragraphs", + load_args("arm", 0x8000000) == "-parm -b800000", + load_args("arm", 0x8000000), + ) check("processor alone", load_args("mipsb") == "-pmipsb", load_args("mipsb")) check("base alone", load_args("", 0x10000) == "-b1000", load_args("", 0x10000)) - check("base 0 emits no switch (it's the default)", - load_args("arm", 0) == "-parm", load_args("arm", 0)) + check( + "base 0 emits no switch (it's the default)", + load_args("arm", 0) == "-parm", + load_args("arm", 0), + ) check("nothing in, nothing out", load_args() == "") - check("extra switches pass through", - load_args("arm", 0, "-T binary") == "-parm -T binary") + check( + "extra switches pass through", + load_args("arm", 0, "-T binary") == "-parm -T binary", + ) - check("the processor list leads with the common targets", - [n for n, _ in PROCESSORS[:2]] == ["arm", "arm:ARMv7-A"], - f"{[n for n, _ in PROCESSORS[:3]]}") + check( + "the processor list leads with the common targets", + [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')]}") + 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 @@ -103,16 +143,40 @@ def main() -> int: # output of tools/verify_procs.py; adding a processor without re-running it # fails here on purpose. VERIFIED = { - "arm", "armb", "metapc", "mipsl", "mipsb", "ppc", "ppcl", "sh4", "68k", - "riscv", "tricore", "xtensa", "avr", "z80", "tms320c6", "m32r", "arc", - "h8300", "sparcb", "sparcl", "s390", + "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", + "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", - offered <= VERIFIED, f"unverified: {sorted(offered - VERIFIED)}") + check( + "every offered processor name is IDA-verified", + offered <= VERIFIED, + f"unverified: {sorted(offered - VERIFIED)}", + ) # These are module FILENAMES or common aliases, not -p names. IDA refuses # them; they were in the list until a real run said otherwise. @@ -123,16 +187,28 @@ def main() -> int: def finds(q): ql = q.lower() return [n for n, d in PROCESSORS if ql in n.lower() or ql in d.lower()] - check("typing 'arm64' still finds ARM", "arm" in finds("arm64"), f"{finds('arm64')}") + + check( + "typing 'arm64' still finds ARM", "arm" in finds("arm64"), f"{finds('arm64')}" + ) check("typing 'aarch64' still finds ARM", "arm" in finds("aarch64")) check("typing 'm68k' still finds 68k", "68k" in finds("m68k"), f"{finds('m68k')}") - check("typing 'mips' finds both endiannesses", - set(finds("mips")) == {"mipsl", "mipsb"}, f"{finds('mips')}") - check("every processor entry has a human label", - all(n and d for n, d in PROCESSORS)) - check("endianness is spelled out where it matters", - all(any(w in d.lower() for w in ("endian",)) - for n, d in PROCESSORS if n in ("arm", "armb", "mipsb", "mipsl"))) + check( + "typing 'mips' finds both endiannesses", + set(finds("mips")) == {"mipsl", "mipsb"}, + f"{finds('mips')}", + ) + check( + "every processor entry has a human label", all(n and d for n, d in PROCESSORS) + ) + check( + "endianness is spelled out where it matters", + all( + any(w in d.lower() for w in ("endian",)) + for n, d in PROCESSORS + if n in ("arm", "armb", "mipsb", "mipsl") + ), + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_graph.py b/tests/test_graph.py index a75ff68..e3b9143 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -9,6 +9,7 @@ suite runs anywhere, but when present it is the interesting half: real functions are where the degenerate shapes (switch fan-out, irreducible loops, 400-block monsters) actually live. """ + from __future__ import annotations #: the layout engine is pure: no IDA, no Textual. @@ -55,17 +56,29 @@ def mk(edges: dict[int, list[tuple[int, str]]], n: int | None = None) -> list[G. ids = set(edges) | {d for v in edges.values() for d, _ in v} if n: ids |= set(range(n)) - return [G.Block(id=i, start=0x1000 + i * 0x10, end=0x1000 + i * 0x10 + 8, - succs=list(edges.get(i, []))) for i in sorted(ids)] + return [ + G.Block( + id=i, + start=0x1000 + i * 0x10, + end=0x1000 + i * 0x10 + 8, + succs=list(edges.get(i, [])), + ) + for i in sorted(ids) + ] # ------------------------------------------------------------ invariants + def no_box_overlap(lay: G.Layout) -> bool: for i, a in enumerate(lay.nodes): - for b in lay.nodes[i + 1:]: - if (a.x <= b.right and b.x <= a.right - and a.y <= b.y + b.h - 1 and b.y <= a.y + a.h - 1): + for b in lay.nodes[i + 1 :]: + if ( + a.x <= b.right + and b.x <= a.right + and a.y <= b.y + b.h - 1 + and b.y <= a.y + a.h - 1 + ): return False return True @@ -97,13 +110,16 @@ def all_edges_drawn(lay: G.Layout) -> bool: def invariants(lay: G.Layout, name: str) -> None: check(no_box_overlap(lay), f"{name}: boxes must not overlap") check(no_edge_through_box(lay) == 0, f"{name}: no edge may cross a box") - check(all(n.x >= 0 and n.y >= 0 for n in lay.nodes), - f"{name}: no negative coordinates") + check( + all(n.x >= 0 and n.y >= 0 for n in lay.nodes), + f"{name}: no negative coordinates", + ) check(lay.width > 0 and lay.height > 0, f"{name}: canvas has extent") # ---------------------------------------------------------------- cases + def t_linear() -> None: lay = layout(mk({0: [(1, "uncond")], 1: [(2, "uncond")]}), sizer) invariants(lay, "linear") @@ -113,8 +129,10 @@ def t_linear() -> None: def t_diamond() -> None: - lay = layout(mk({0: [(1, "jump"), (2, "fall")], - 1: [(3, "uncond")], 2: [(3, "uncond")]}), sizer) + lay = layout( + mk({0: [(1, "jump"), (2, "fall")], 1: [(3, "uncond")], 2: [(3, "uncond")]}), + sizer, + ) invariants(lay, "diamond") check(lay.by_id[3].rank == 2, "diamond: join sits below both arms") check(lay.by_id[1].rank == lay.by_id[2].rank, "diamond: arms share a rank") @@ -125,8 +143,10 @@ def t_diamond() -> None: def t_selfloop() -> None: """A self-loop must not stall the ranking — the bug that collapsed a whole function into three layers and made the graph 280 columns wide.""" - lay = layout(mk({0: [(1, "uncond")], 1: [(1, "jump"), (2, "fall")], - 2: [(3, "uncond")]}), sizer) + lay = layout( + mk({0: [(1, "uncond")], 1: [(1, "jump"), (2, "fall")], 2: [(3, "uncond")]}), + sizer, + ) invariants(lay, "selfloop") ranks = [lay.by_id[i].rank for i in (0, 1, 2, 3)] check(ranks == [0, 1, 2, 3], f"selfloop: ranking still stacks ({ranks})") @@ -134,23 +154,36 @@ def t_selfloop() -> None: def t_loop() -> None: - lay = layout(mk({0: [(1, "uncond")], 1: [(2, "jump"), (3, "fall")], - 2: [(1, "uncond")]}), sizer) + lay = layout( + mk({0: [(1, "uncond")], 1: [(2, "jump"), (3, "fall")], 2: [(1, "uncond")]}), + sizer, + ) invariants(lay, "loop") check(any(e.back for e in lay.edges), "loop: a back edge is detected") check(lay.by_id[1].rank < lay.by_id[2].rank, "loop: header above the body") back = [e for e in lay.edges if e.back][0] - check((2, G.E_BACK) in [(a, s) for a, s in lay.pred[1]] - or (1, G.E_BACK) in [(a, s) for a, s in lay.succ[2]], - "loop: the back edge reads 2 -> 1 despite being reversed for layout") + check( + (2, G.E_BACK) in [(a, s) for a, s in lay.pred[1]] + or (1, G.E_BACK) in [(a, s) for a, s in lay.succ[2]], + "loop: the back edge reads 2 -> 1 despite being reversed for layout", + ) def t_switch() -> None: - lay = layout(mk({0: [(i, "switch") for i in range(1, 9)], - **{i: [(9, "uncond")] for i in range(1, 9)}}), sizer) + lay = layout( + mk( + { + 0: [(i, "switch") for i in range(1, 9)], + **{i: [(9, "uncond")] for i in range(1, 9)}, + } + ), + sizer, + ) invariants(lay, "switch") - check(len({lay.by_id[i].rank for i in range(1, 9)}) == 1, - "switch: all cases share a rank") + check( + len({lay.by_id[i].rank for i in range(1, 9)}) == 1, + "switch: all cases share a rank", + ) check(lay.by_id[9].rank == 2, "switch: the join is below the cases") @@ -172,14 +205,26 @@ def t_unreachable_entry() -> None: from, so the engine must never be handed one. """ # entry 0 is a sink; 2 and 3 jump INTO it; 1 and 6 self-loop. - lay = layout(mk({0: [], 1: [(5, "switch"), (1, "fall"), (4, "switch")], - 2: [(5, "jump"), (0, "uncond")], 3: [(0, "switch")], - 4: [], 5: [(4, "jump")], - 6: [(2, "jump"), (6, "switch"), (3, "jump")]}), entry=0) + lay = layout( + mk( + { + 0: [], + 1: [(5, "switch"), (1, "fall"), (4, "switch")], + 2: [(5, "jump"), (0, "uncond")], + 3: [(0, "switch")], + 4: [], + 5: [(4, "jump")], + 6: [(2, "jump"), (6, "switch"), (3, "jump")], + } + ), + entry=0, + ) invariants(lay, "unreachable_entry") check(len(lay.nodes) == 7, "unreachable_entry: every block is placed") - check(lay.stats.get("engine_error") is None, - f"unreachable_entry: no fallback ({lay.stats.get('engine_error')})") + check( + lay.stats.get("engine_error") is None, + f"unreachable_entry: no fallback ({lay.stats.get('engine_error')})", + ) # An entry that reaches nothing at all, with everything hanging off nodes # it cannot see, is the degenerate version of the same thing. @@ -198,8 +243,10 @@ def t_long_edge() -> None: # long edge. Triskel reaches the same end -- an edge that crosses no box, # checked by invariants() above -- without them, so this is engine-specific. if ENGINE == "native": - check(lay.stats["dummies"] >= 4, - f"long_edge: the skip edge is padded ({lay.stats['dummies']} dummies)") + check( + lay.stats["dummies"] >= 4, + f"long_edge: the skip edge is padded ({lay.stats['dummies']} dummies)", + ) check(all_edges_drawn(lay), "long_edge: the long edge is drawn") @@ -212,14 +259,18 @@ def t_empty() -> None: def t_row_query() -> None: """cells_at_row must be windowed: asking for a slice returns only that slice, which is what keeps a 13M-cell graph renderable.""" - lay = layout(mk({0: [(1, "jump"), (2, "fall")], - 1: [(3, "uncond")], 2: [(3, "uncond")]}), sizer) + lay = layout( + mk({0: [(1, "jump"), (2, "fall")], 1: [(3, "uncond")], 2: [(3, "uncond")]}), + sizer, + ) for row in range(lay.height): full = lay.painting.cells_at_row(row, 0, lay.width) part = lay.painting.cells_at_row(row, 5, 12) check(all(5 <= c < 12 for c in part), f"row {row}: window respected") - check(all(full.get(c) == v for c, v in part.items()), - f"row {row}: window agrees with the full row") + check( + all(full.get(c) == v for c, v in part.items()), + f"row {row}: window agrees with the full row", + ) def t_hit_test() -> None: @@ -233,6 +284,7 @@ def t_hit_test() -> None: # ---------------------------------------------------------------- corpus + def t_corpus(path: str) -> None: recs = json.load(open(path)) print(f"\ncorpus: {len(recs)} functions from {path}") @@ -243,9 +295,15 @@ def t_corpus(path: str) -> None: worst_any_name = "" t0 = time.perf_counter() for rec in recs: - blocks = [G.Block(id=b["id"], start=b["start"], end=b["end"], - succs=[(d, k) for d, k in b["succs"]]) - for b in rec["blocks"]] + blocks = [ + G.Block( + id=b["id"], + start=b["start"], + end=b["end"], + succs=[(d, k) for d, k in b["succs"]], + ) + for b in rec["blocks"] + ] lay = layout(blocks, sizer) # A SILENT fallback is the failure mode that matters here: the engine # under test quietly stops being the engine under test, and every @@ -258,46 +316,61 @@ def t_corpus(path: str) -> None: # not that it never happens. if ENGINE != "auto" and lay.stats["engine"] != ENGINE: fellback.append((rec["name"], lay.stats.get("engine_error"))) - check(bool(lay.stats.get("engine_error")), - f"corpus {rec['name']}: a fallback must record its reason") + check( + bool(lay.stats.get("engine_error")), + f"corpus {rec['name']}: a fallback must record its reason", + ) # Time the engine only on the functions it would actually be ASKED for. # `auto` hands anything over AUTO_TRISKEL_MAX_BLOCKS to native, and the # view refuses to draw past 400 blocks at all, so a forced triskel run # on a 495-block monster times a call the app cannot make. - reachable = (ENGINE != "triskel" - or len(blocks) <= G.AUTO_TRISKEL_MAX_BLOCKS) + reachable = ENGINE != "triskel" or len(blocks) <= G.AUTO_TRISKEL_MAX_BLOCKS if reachable and lay.stats["ms"] > worst_ms: worst_ms, worst_name = lay.stats["ms"], rec["name"] if lay.stats["ms"] > worst_any_ms: worst_any_ms, worst_any_name = lay.stats["ms"], rec["name"] check(no_box_overlap(lay), f"corpus {rec['name']}: boxes must not overlap") - check(len(lay.nodes) == len(blocks), - f"corpus {rec['name']}: every block is placed") + check( + len(lay.nodes) == len(blocks), + f"corpus {rec['name']}: every block is placed", + ) # The full cell sweep is O(canvas); only affordable on the small ones, # but that is where a routing bug would show up anyway. if lay.width * lay.height < 400_000: - check(no_edge_through_box(lay) == 0, - f"corpus {rec['name']}: no edge may cross a box") + check( + no_edge_through_box(lay) == 0, + f"corpus {rec['name']}: no edge may cross a box", + ) total = (time.perf_counter() - t0) * 1000 - print(f" laid out {len(recs)} functions in {total:.0f} ms " - f"(worst {worst_ms:.0f} ms: {worst_name})") + print( + f" laid out {len(recs)} functions in {total:.0f} ms " + f"(worst {worst_ms:.0f} ms: {worst_name})" + ) if fellback: print(f" {len(fellback)} fell back to native:") for name, why in fellback: print(f" {name}: {why}") - check(len(fellback) <= max(2, len(recs) // 20), - f"corpus: {ENGINE} fell back on {len(fellback)}/{len(recs)} functions") - check(worst_ms < 2000, f"corpus: worst REACHABLE layout under 2s " - f"({worst_ms:.0f} ms: {worst_name})") + check( + len(fellback) <= max(2, len(recs) // 20), + f"corpus: {ENGINE} fell back on {len(fellback)}/{len(recs)} functions", + ) + check( + worst_ms < 2000, + f"corpus: worst REACHABLE layout under 2s ({worst_ms:.0f} ms: {worst_name})", + ) # Nothing may blow up quadratically even when forced past its own limits. - check(worst_any_ms < 5000, f"corpus: worst layout at any size under 5s " - f"({worst_any_ms:.0f} ms: {worst_any_name})") + check( + worst_any_ms < 5000, + f"corpus: worst layout at any size under 5s " + f"({worst_any_ms:.0f} ms: {worst_any_name})", + ) def main() -> int: global ENGINE print("idatui.graph layout tests") from idatui import graph_triskel + engines = ["native"] if graph_triskel.available(): engines.append("triskel") @@ -306,9 +379,19 @@ def main() -> int: for engine in engines: ENGINE = engine print(f"\nengine: {engine}") - for fn in (t_linear, t_diamond, t_selfloop, t_loop, t_switch, - t_unreachable, t_unreachable_entry, t_long_edge, t_empty, - t_row_query, t_hit_test): + for fn in ( + t_linear, + t_diamond, + t_selfloop, + t_loop, + t_switch, + t_unreachable, + t_unreachable_entry, + t_long_edge, + t_empty, + t_row_query, + t_hit_test, + ): print(f" {fn.__name__}") fn() for path in sys.argv[1:]: diff --git a/tests/test_index.py b/tests/test_index.py index f6761b1..5850821 100644 --- a/tests/test_index.py +++ b/tests/test_index.py @@ -15,8 +15,13 @@ import tempfile import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.index import (KIND_EXPORT, KIND_FUNC, KIND_IMPORT, # noqa: E402 - KIND_STRING, ProjectIndex) +from idatui.index import ( # noqa: E402 + KIND_EXPORT, + KIND_FUNC, + KIND_IMPORT, + KIND_STRING, + ProjectIndex, +) PASS = FAIL = 0 @@ -41,70 +46,106 @@ def main() -> int: check("a fresh index is empty", idx.total() == 0 and idx.counts() == {}) check("an unindexed binary is stale", idx.is_stale("libfoo", src)) - n = idx.reindex("libfoo", [ - (KIND_FUNC, 0x1000, "SSL_CTX_new"), - (KIND_FUNC, 0x1100, "SSL_read"), - (KIND_FUNC, 0x1200, "sub_1200"), - (KIND_STRING, 0x8000, "error opening socket"), - (KIND_STRING, 0x8100, "/etc/ssl/certs"), - ], source=src) + n = idx.reindex( + "libfoo", + [ + (KIND_FUNC, 0x1000, "SSL_CTX_new"), + (KIND_FUNC, 0x1100, "SSL_read"), + (KIND_FUNC, 0x1200, "sub_1200"), + (KIND_STRING, 0x8000, "error opening socket"), + (KIND_STRING, 0x8100, "/etc/ssl/certs"), + ], + source=src, + ) check("reindex reports what it stored", n == 5, f"n={n}") check("the entries are there", idx.total() == 5, f"{idx.total()}") check("an indexed binary is fresh", not idx.is_stale("libfoo", src)) # -- substring search (the thing a prefix index can't do) ----------- # hits = idx.search("SSL", kind=KIND_FUNC) - check("finds symbols by substring", {h.text for h in hits} == - {"SSL_CTX_new", "SSL_read"}, f"{[h.text for h in hits]}") - check("matching ignores case across kinds (SSL also hits /etc/ssl)", - {h.text for h in idx.search("SSL")} == - {"SSL_CTX_new", "SSL_read", "/etc/ssl/certs"}, - f"{[h.text for h in idx.search('SSL')]}") + check( + "finds symbols by substring", + {h.text for h in hits} == {"SSL_CTX_new", "SSL_read"}, + f"{[h.text for h in hits]}", + ) + check( + "matching ignores case across kinds (SSL also hits /etc/ssl)", + {h.text for h in idx.search("SSL")} + == {"SSL_CTX_new", "SSL_read", "/etc/ssl/certs"}, + f"{[h.text for h in idx.search('SSL')]}", + ) hits = idx.search("socket") - check("finds strings by substring mid-text", - len(hits) == 1 and hits[0].kind == KIND_STRING - and hits[0].addr == 0x8000, f"{hits}") - check("hits carry the owning binary", - all(h.binary == "libfoo" for h in idx.search("SSL"))) - check("search is case-insensitive", - {h.text for h in idx.search("ssl_read")} == {"SSL_read"}, - f"{[h.text for h in idx.search('ssl_read')]}") - check("kind filter narrows to strings", - [h.text for h in idx.search("ss", kind=KIND_STRING)] == ["/etc/ssl/certs"], - f"{[h.text for h in idx.search('ss', kind=KIND_STRING)]}") + check( + "finds strings by substring mid-text", + len(hits) == 1 and hits[0].kind == KIND_STRING and hits[0].addr == 0x8000, + f"{hits}", + ) + check( + "hits carry the owning binary", + all(h.binary == "libfoo" for h in idx.search("SSL")), + ) + check( + "search is case-insensitive", + {h.text for h in idx.search("ssl_read")} == {"SSL_read"}, + f"{[h.text for h in idx.search('ssl_read')]}", + ) + check( + "kind filter narrows to strings", + [h.text for h in idx.search("ss", kind=KIND_STRING)] == ["/etc/ssl/certs"], + f"{[h.text for h in idx.search('ss', kind=KIND_STRING)]}", + ) # -- the <3 char fallback (trigram silently matches nothing) -------- # - check("2-char query still works (LIKE fallback)", - {h.text for h in idx.search("ss")} == {"SSL_CTX_new", "SSL_read", - "/etc/ssl/certs"}, - f"{[h.text for h in idx.search('ss')]}") - check("1-char query still works", - len(idx.search("/")) == 1, f"{idx.search('/')}") + check( + "2-char query still works (LIKE fallback)", + {h.text for h in idx.search("ss")} + == {"SSL_CTX_new", "SSL_read", "/etc/ssl/certs"}, + f"{[h.text for h in idx.search('ss')]}", + ) + check( + "1-char query still works", len(idx.search("/")) == 1, f"{idx.search('/')}" + ) check("an empty query matches nothing", idx.search(" ") == []) - check("a query with FTS operators is treated literally", - idx.search('SSL OR "') == [] or True) # must not raise + check( + "a query with FTS operators is treated literally", + idx.search('SSL OR "') == [] or True, + ) # must not raise # -- multi-binary: the whole point ---------------------------------- # - idx.reindex("httpd", [ - (KIND_FUNC, 0x2000, "handle_ssl_request"), - (KIND_STRING, 0x9000, "socket bind failed"), - ]) + idx.reindex( + "httpd", + [ + (KIND_FUNC, 0x2000, "handle_ssl_request"), + (KIND_STRING, 0x9000, "socket bind failed"), + ], + ) hits = idx.search("ssl") - check("search spans binaries", - {h.binary for h in hits} == {"libfoo", "httpd"}, - f"{[(h.binary, h.text) for h in hits]}") - check("counts are per binary", - idx.counts() == {"libfoo": 5, "httpd": 2}, f"{idx.counts()}") + check( + "search spans binaries", + {h.binary for h in hits} == {"libfoo", "httpd"}, + f"{[(h.binary, h.text) for h in hits]}", + ) + check( + "counts are per binary", + idx.counts() == {"libfoo": 5, "httpd": 2}, + f"{idx.counts()}", + ) # -- incremental: reindexing one binary leaves the others alone ----- # idx.reindex("libfoo", [(KIND_FUNC, 0x1000, "SSL_CTX_new_v2")], source=src) - check("reindex replaces only that binary's entries", - idx.counts() == {"libfoo": 1, "httpd": 2}, f"{idx.counts()}") - check("the stale entries are gone", - [h.text for h in idx.search("SSL_read")] == [], - f"{idx.search('SSL_read')}") - check("the other binary survived untouched", - len(idx.search("socket bind")) == 1) + check( + "reindex replaces only that binary's entries", + idx.counts() == {"libfoo": 1, "httpd": 2}, + f"{idx.counts()}", + ) + check( + "the stale entries are gone", + [h.text for h in idx.search("SSL_read")] == [], + f"{idx.search('SSL_read')}", + ) + check( + "the other binary survived untouched", len(idx.search("socket bind")) == 1 + ) # -- staleness follows the source ----------------------------------- # time.sleep(0.01) @@ -112,90 +153,128 @@ def main() -> int: f.write(b"\x7fELF binary rebuilt, different size") os.utime(src, (1, 1)) check("a changed source goes stale", idx.is_stale("libfoo", src)) - check("a missing source does NOT wipe the index", - not idx.is_stale("libfoo", os.path.join(tmp, "gone"))) + check( + "a missing source does NOT wipe the index", + not idx.is_stale("libfoo", os.path.join(tmp, "gone")), + ) # -- forget ----------------------------------------------------------- # idx.forget("httpd") - check("forget drops a binary entirely", - idx.counts() == {"libfoo": 1} and idx.search("socket bind") == [], - f"{idx.counts()}") + check( + "forget drops a binary entirely", + idx.counts() == {"libfoo": 1} and idx.search("socket bind") == [], + f"{idx.counts()}", + ) # -- cross-binary linkage join (phase 3) ------------------------------ # - idx.reindex("app", [ - (KIND_FUNC, 0x1000, "main"), - (KIND_IMPORT, 0x2000, "strcmp"), - (KIND_IMPORT, 0x2008, "read"), - (KIND_IMPORT, 0x2010, "SSL_new"), - ]) - idx.reindex("libc", [ - (KIND_EXPORT, 0x8000, "strcmp"), - (KIND_EXPORT, 0x8100, "read"), - (KIND_EXPORT, 0x8200, "pread"), - (KIND_EXPORT, 0x8300, "read_line"), - (KIND_FUNC, 0x8000, "strcmp"), - ]) + idx.reindex( + "app", + [ + (KIND_FUNC, 0x1000, "main"), + (KIND_IMPORT, 0x2000, "strcmp"), + (KIND_IMPORT, 0x2008, "read"), + (KIND_IMPORT, 0x2010, "SSL_new"), + ], + ) + idx.reindex( + "libc", + [ + (KIND_EXPORT, 0x8000, "strcmp"), + (KIND_EXPORT, 0x8100, "read"), + (KIND_EXPORT, 0x8200, "pread"), + (KIND_EXPORT, 0x8300, "read_line"), + (KIND_FUNC, 0x8000, "strcmp"), + ], + ) idx.reindex("libssl", [(KIND_EXPORT, 0x9000, "SSL_new")]) prov = idx.providers("strcmp", exclude="app") - check("an import resolves to the binary that exports it", - [(h.binary, h.addr) for h in prov] == [("libc", 0x8000)], - f"{[(h.binary, hex(h.addr)) for h in prov]}") + check( + "an import resolves to the binary that exports it", + [(h.binary, h.addr) for h in prov] == [("libc", 0x8000)], + f"{[(h.binary, hex(h.addr)) for h in prov]}", + ) # The whole point of exact(): substring search would drag in pread, # read_line and thread_start, and 'read' is also below the trigram floor # for some engines — an import must bind to its exact name or nothing. prov = idx.providers("read", exclude="app") - check("the join is exact, not substring", - [(h.binary, h.addr) for h in prov] == [("libc", 0x8100)], - f"{[(h.binary, h.text) for h in prov]}") + check( + "the join is exact, not substring", + [(h.binary, h.addr) for h in prov] == [("libc", 0x8100)], + f"{[(h.binary, h.text) for h in prov]}", + ) - check("a short name still resolves (below the trigram floor)", - [h.binary for h in idx.providers("SSL_new", exclude="app")] == ["libssl"], - f"{idx.providers('SSL_new')}") + check( + "a short name still resolves (below the trigram floor)", + [h.binary for h in idx.providers("SSL_new", exclude="app")] == ["libssl"], + f"{idx.providers('SSL_new')}", + ) - check("an unprovided import resolves to nothing", - idx.providers("dlopen", exclude="app") == []) + check( + "an unprovided import resolves to nothing", + idx.providers("dlopen", exclude="app") == [], + ) - check("exclude keeps a binary from resolving to itself", - idx.providers("strcmp", exclude="libc") == [], - f"{idx.providers('strcmp', exclude='libc')}") + check( + "exclude keeps a binary from resolving to itself", + idx.providers("strcmp", exclude="libc") == [], + f"{idx.providers('strcmp', exclude='libc')}", + ) imp = idx.importers("strcmp") - check("the reverse join finds who imports an export", - [(h.binary, h.addr) for h in imp] == [("app", 0x2000)], - f"{[(h.binary, hex(h.addr)) for h in imp]}") + check( + "the reverse join finds who imports an export", + [(h.binary, h.addr) for h in imp] == [("app", 0x2000)], + f"{[(h.binary, hex(h.addr)) for h in imp]}", + ) - check("kind keeps functions out of the linkage join", - [h.binary for h in idx.providers("strcmp")] == ["libc"], - "a KIND_FUNC row named strcmp must not answer as an export") + check( + "kind keeps functions out of the linkage join", + [h.binary for h in idx.providers("strcmp")] == ["libc"], + "a KIND_FUNC row named strcmp must not answer as an export", + ) idx.forget("libc") - check("forgetting a provider unresolves its imports", - idx.providers("strcmp", exclude="app") == []) + check( + "forgetting a provider unresolves its imports", + idx.providers("strcmp", exclude="app") == [], + ) # -- ELF symbol versioning -------------------------------------------- # # The importer sees strrchr@@GLIBC_2.2.5 while the provider may export a # different spelling; raw names would resolve almost nothing. link_name # cuts at the first '@' so both sides meet on the bare symbol. from idatui.domain import link_name - check("link_name strips an ELF version suffix", - link_name("strrchr@@GLIBC_2.2.5") == "strrchr", - link_name("strrchr@@GLIBC_2.2.5")) - check("link_name leaves an unversioned name alone", - link_name("strrchr") == "strrchr") - check("link_name handles a single-@ version", - link_name("SSL_new@OPENSSL_3.0.0") == "SSL_new") - check("link_name doesn't eat a leading @", - link_name("@weird") == "@weird", link_name("@weird")) + + check( + "link_name strips an ELF version suffix", + link_name("strrchr@@GLIBC_2.2.5") == "strrchr", + link_name("strrchr@@GLIBC_2.2.5"), + ) + check( + "link_name leaves an unversioned name alone", + link_name("strrchr") == "strrchr", + ) + check( + "link_name handles a single-@ version", + link_name("SSL_new@OPENSSL_3.0.0") == "SSL_new", + ) + check( + "link_name doesn't eat a leading @", + link_name("@weird") == "@weird", + link_name("@weird"), + ) # -- persistence ------------------------------------------------------ # path = idx.path idx.close() idx2 = ProjectIndex(path) - check("the index persists across sessions", - [h.text for h in idx2.search("SSL_CTX")] == ["SSL_CTX_new_v2"], - f"{idx2.search('SSL_CTX')}") + check( + "the index persists across sessions", + [h.text for h in idx2.search("SSL_CTX")] == ["SSL_CTX_new_v2"], + f"{idx2.search('SSL_CTX')}", + ) idx2.close() print(f"\n{PASS} passed, {FAIL} failed") diff --git a/tests/test_kittygfx.py b/tests/test_kittygfx.py index 38d361c..270b4e7 100644 --- a/tests/test_kittygfx.py +++ b/tests/test_kittygfx.py @@ -55,8 +55,11 @@ class Tty: def cmds(self, action): """Every graphics command with the given ``a=`` action.""" - return [c for c in re.findall(r"\x1b_G([^;\x1b]*)", self.blob) - if f"a={action}" in c.split(",")] + return [ + c + for c in re.findall(r"\x1b_G([^;\x1b]*)", self.blob) + if f"a={action}" in c.split(",") + ] def keys(cmd): @@ -76,8 +79,7 @@ def t_no_termios_falls_back(): try: check("missing termios disables graphics", kittygfx._query_tty(0) is False) except Exception as exc: # the original Windows startup crash - check("missing termios does not escape", False, - f"{type(exc).__name__}: {exc}") + check("missing termios does not escape", False, f"{type(exc).__name__}: {exc}") finally: builtins.__import__ = original_import @@ -103,8 +105,7 @@ def t_probe_failure_is_never_fatal(): check("probe exception disables graphics", kittygfx.supported() is False) check("failed result is cached", kittygfx.supported() is False) except Exception as exc: - check("probe exception does not escape", False, - f"{type(exc).__name__}: {exc}") + check("probe exception does not escape", False, f"{type(exc).__name__}: {exc}") finally: kittygfx._query_tty = original_query kittygfx._supported = original_supported @@ -118,7 +119,7 @@ def main() -> int: t_no_termios_falls_back() t_probe_failure_is_never_fatal() - kittygfx._uploaded[kittygfx.LOGO_ID] = (768, 801) # pretend it's uploaded + kittygfx._uploaded[kittygfx.LOGO_ID] = (768, 801) # pretend it's uploaded # -- the bug: anonymous placements STACK ------------------------------- # # A placement is identified by (image id, placement id). With no p key @@ -128,80 +129,126 @@ def main() -> int: for _ in range(50): kittygfx.place(4, 10, 60, 26) placements = tty.cmds("p") - check("place() emits one command per call", len(placements) == 50, - f"{len(placements)}") - check("every placement carries a placement id (replaces, not stacks)", - all("p" in keys(c) for c in placements), - f"{placements[0] if placements else '(none)'}") - check("the placement id is the same every time (one image on screen)", - len({keys(c)["p"] for c in placements}) == 1, - f"{sorted({keys(c).get('p') for c in placements})}") - check("...and it is non-zero (p=0 means anonymous)", - keys(placements[0])["p"] not in ("0", ""), f"{placements[0]}") + check( + "place() emits one command per call", + len(placements) == 50, + f"{len(placements)}", + ) + check( + "every placement carries a placement id (replaces, not stacks)", + all("p" in keys(c) for c in placements), + f"{placements[0] if placements else '(none)'}", + ) + check( + "the placement id is the same every time (one image on screen)", + len({keys(c)["p"] for c in placements}) == 1, + f"{sorted({keys(c).get('p') for c in placements})}", + ) + check( + "...and it is non-zero (p=0 means anonymous)", + keys(placements[0])["p"] not in ("0", ""), + f"{placements[0]}", + ) # -- the rest of the escape still says what it used to ------------------ # with Tty() as tty: ok = kittygfx.place(4, 10, 60, 26) k = keys(tty.cmds("p")[0]) check("place() reports success", ok) - check("image id, source pixels and cell box are unchanged", - (k["i"], k["s"], k["v"], k["c"], k["r"]) - == (str(kittygfx.LOGO_ID), "768", "801", "60", "26"), f"{k}") - check("the terminal is told not to move the cursor (C=1)", - k.get("C") == "1", f"{k}") - check("the cursor is saved and restored around the placement", - tty.blob.startswith("\x1b[s") and tty.blob.endswith("\x1b[u"), - repr(tty.blob[:8] + "..." + tty.blob[-8:])) - check("the placement is positioned 1-based (row 4 -> line 5)", - "\x1b[5;11H" in tty.blob, repr(tty.blob[:24])) + check( + "image id, source pixels and cell box are unchanged", + (k["i"], k["s"], k["v"], k["c"], k["r"]) + == (str(kittygfx.LOGO_ID), "768", "801", "60", "26"), + f"{k}", + ) + check( + "the terminal is told not to move the cursor (C=1)", + k.get("C") == "1", + f"{k}", + ) + check( + "the cursor is saved and restored around the placement", + tty.blob.startswith("\x1b[s") and tty.blob.endswith("\x1b[u"), + repr(tty.blob[:8] + "..." + tty.blob[-8:]), + ) + check( + "the placement is positioned 1-based (row 4 -> line 5)", + "\x1b[5;11H" in tty.blob, + repr(tty.blob[:24]), + ) # -- deleting still removes EVERY placement of the image ---------------- # # d=i is by image id, so it takes the placement with us regardless of p. with Tty() as tty: kittygfx.clear() k = keys(tty.cmds("d")[0]) - check("clear() deletes by image id (d=i), keeping the upload", - k.get("d") == "i" and k.get("i") == str(kittygfx.LOGO_ID), f"{k}") - check("clear() does not free the image data (lowercase d)", - kittygfx.is_uploaded(), "upload was dropped") + check( + "clear() deletes by image id (d=i), keeping the upload", + k.get("d") == "i" and k.get("i") == str(kittygfx.LOGO_ID), + f"{k}", + ) + check( + "clear() does not free the image data (lowercase d)", + kittygfx.is_uploaded(), + "upload was dropped", + ) with Tty() as tty: kittygfx.delete() k = keys(tty.cmds("d")[0]) check("delete() frees the image data too (d=I)", k.get("d") == "I", f"{k}") - check("...and forgets the upload, so the next place() refuses", - not kittygfx.is_uploaded() and kittygfx.place(0, 0, 10, 10) is False) + check( + "...and forgets the upload, so the next place() refuses", + not kittygfx.is_uploaded() and kittygfx.place(0, 0, 10, 10) is False, + ) # -- refusals ----------------------------------------------------------- # kittygfx._uploaded[kittygfx.LOGO_ID] = (768, 801) with Tty() as tty: - check("a zero-sized box is refused, not sent", - kittygfx.place(0, 0, 0, 10) is False - and kittygfx.place(0, 0, 10, 0) is False and not tty.sent, - f"{tty.sent}") + check( + "a zero-sized box is refused, not sent", + kittygfx.place(0, 0, 0, 10) is False + and kittygfx.place(0, 0, 10, 0) is False + and not tty.sent, + f"{tty.sent}", + ) kittygfx._uploaded.pop(kittygfx.LOGO_ID, None) # -- fit(): aspect ratio against non-square cells ----------------------- # - check("fit() keeps the aspect ratio for 9x22 cells", - kittygfx.fit((768, 801), 60, 99, cell=(9, 22)) == (60, 26), - f"{kittygfx.fit((768, 801), 60, 99, cell=(9, 22))}") - check("fit() shrinks to the row budget instead of overflowing", - kittygfx.fit((768, 801), 60, 10, cell=(9, 22))[1] == 10, - f"{kittygfx.fit((768, 801), 60, 10, cell=(9, 22))}") - check("fit() never returns a zero dimension", - all(v >= 1 for v in kittygfx.fit((768, 801), 1, 1, cell=(9, 22)))) - check("fit() survives a degenerate image size", - kittygfx.fit((0, 0), 60, 26) == (60, 26)) + check( + "fit() keeps the aspect ratio for 9x22 cells", + kittygfx.fit((768, 801), 60, 99, cell=(9, 22)) == (60, 26), + f"{kittygfx.fit((768, 801), 60, 99, cell=(9, 22))}", + ) + check( + "fit() shrinks to the row budget instead of overflowing", + kittygfx.fit((768, 801), 60, 10, cell=(9, 22))[1] == 10, + f"{kittygfx.fit((768, 801), 60, 10, cell=(9, 22))}", + ) + check( + "fit() never returns a zero dimension", + all(v >= 1 for v in kittygfx.fit((768, 801), 1, 1, cell=(9, 22))), + ) + check( + "fit() survives a degenerate image size", + kittygfx.fit((0, 0), 60, 26) == (60, 26), + ) # -- png_size() reads the header, not the pixels ------------------------ # - logo = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), - "logo.png") + logo = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logo.png" + ) if os.path.exists(logo): - check("png_size() reads logo.png's IHDR", - kittygfx.png_size(logo) == (768, 801), f"{kittygfx.png_size(logo)}") + check( + "png_size() reads logo.png's IHDR", + kittygfx.png_size(logo) == (768, 801), + f"{kittygfx.png_size(logo)}", + ) check("png_size() returns None for a non-PNG", kittygfx.png_size(__file__) is None) - check("png_size() returns None for a missing file", - kittygfx.png_size("/nonexistent/nope.png") is None) + check( + "png_size() returns None for a missing file", + kittygfx.png_size("/nonexistent/nope.png") is None, + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_launch.py b/tests/test_launch.py index 35a0b8d..5c05d08 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -10,6 +10,7 @@ is the cheapest guard against someone reintroducing the "helpful" cleanup. Pure: no IDA, no IDA Nexus library, no Textual. """ + from __future__ import annotations import os @@ -53,9 +54,11 @@ def t_no_lock_sweeping(): check("_sweep_locks is gone", not hasattr(launch, "_sweep_locks")) check("the scratch-suffix list is gone", not hasattr(launch, "_LOCK_SUFFIXES")) src = open(launch.__file__, encoding="utf-8").read() - check("the launcher does not remove files at all", - "os.remove" not in src and "shutil.rmtree" not in src, - "launch.py deletes something again") + check( + "the launcher does not remove files at all", + "os.remove" not in src and "shutil.rmtree" not in src, + "launch.py deletes something again", + ) def t_load_args(): @@ -66,10 +69,16 @@ def t_load_args(): # -b is in PARAGRAPHS, not bytes: 0x8000 >> 4 == 0x800. check("a base is converted to paragraphs", "-b800" in a, a) b = _load_args({"base": "0x1000"}) - check("a base given as a hex STRING is accepted (project files write those)", - "-b100" in b, b) - check("no base means no -b switch", "-b" not in _load_args({"processor": "arm"}), - _load_args({"processor": "arm"})) + check( + "a base given as a hex STRING is accepted (project files write those)", + "-b100" in b, + b, + ) + check( + "no base means no -b switch", + "-b" not in _load_args({"processor": "arm"}), + _load_args({"processor": "arm"}), + ) c = _load_args({"ida_args": "-p1"}) check("extra ida_args are passed through", "-p1" in c, c) @@ -81,6 +90,7 @@ def main() -> int: fn() except Exception as e: # noqa: BLE001 import traceback + check(f"{fn.__name__} did not crash", False, f"{type(e).__name__}: {e}") traceback.print_exc() print(f"\n{PASS} passed, {FAIL} failed") diff --git a/tests/test_nexus_client.py b/tests/test_nexus_client.py index ae2bfa7..834a8d6 100644 --- a/tests/test_nexus_client.py +++ b/tests/test_nexus_client.py @@ -5,15 +5,15 @@ from __future__ import annotations import os import queue import sys +import tempfile import threading import time -import tempfile from dataclasses import dataclass sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.errors import IDAConnectionError, IDAToolError # noqa: E402 -from idatui import remote_ops # noqa: E402 import idatui.nexus_client as module # noqa: E402 +from idatui import remote_ops # noqa: E402 +from idatui.errors import IDAConnectionError, IDAToolError # noqa: E402 from idatui.nexus_client import NexusClient, _parse_load_args # noqa: E402 #: Pure: fakes the DatabaseHandle, never touches IDA or the IDA Nexus library. @@ -215,10 +215,7 @@ class FakeRemoteModule: operation_label=label, ) result = response["result"] - if ( - isinstance(result, dict) - and result.get("__remote_ida_status__") == "ok" - ): + if isinstance(result, dict) and result.get("__remote_ida_status__") == "ok": return result.get("__remote_ida_value__") raise module.RemoteError(name, f"remote call failed: {result!r}", 500) @@ -232,6 +229,7 @@ def _open_kwargs_are_real(sent: dict): """ try: import inspect + from ida_nexus import DatabaseHandle as Real except ImportError: return True, "ida_nexus not installed - signature not checked" @@ -249,6 +247,7 @@ def _option_fields_are_real(options): """ try: import dataclasses + from ida_nexus import DatabaseOpenOptions as Real except ImportError: return True, "ida_nexus not installed - fields not checked" diff --git a/tests/test_pool.py b/tests/test_pool.py index 37058bf..203ce95 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -33,8 +33,7 @@ def check(name, cond, detail=""): class FakeClient: """Stands in for a NexusClient lease and records saves/closes.""" - def __init__(self, ref, mem=100, backend="idalib", - discardable=True): + def __init__(self, ref, mem=100, backend="idalib", discardable=True): self.ref = ref self.mem = mem self.backend = backend @@ -82,14 +81,18 @@ def main() -> int: made[ref.label] = c return c - pool = DatabasePool(proj, budget_mb=350, spawn=spawn, - mem_fn=lambda c: c.mem) + pool = DatabasePool(proj, budget_mb=350, spawn=spawn, mem_fn=lambda c: c.mem) # -- lazy spawn + reuse -------------------------------------------- # a = pool.get("bin0") - check("get() spawns a database lease on first use", a is made["bin0"] and a.connected) - check("get() stages the binary first", - os.path.isfile(proj.by_label("bin0").staged)) + check( + "get() spawns a database lease on first use", + a is made["bin0"] and a.connected, + ) + check( + "get() stages the binary first", + os.path.isfile(proj.by_label("bin0").staged), + ) check("get() reuses the resident lease", pool.get("bin0") is a) check("resident() reports it", pool.resident() == ["bin0"], pool.resident()) @@ -97,28 +100,40 @@ def main() -> int: pool.get("bin1") pool.get("bin2") pool.get("bin0") # touch: bin0 becomes most-recent - check("LRU order tracks use", pool.resident() == ["bin1", "bin2", "bin0"], - pool.resident()) + check( + "LRU order tracks use", + pool.resident() == ["bin1", "bin2", "bin0"], + pool.resident(), + ) # -- budget eviction -------------------------------------------------- # check("pool reports its memory", pool.memory_mb() == 300, pool.memory_mb()) pool.get("bin3") # 400MB > 350MB budget -> evict LRU (bin1) - check("exceeding the budget evicts the least-recently-used", - pool.evicted == ["bin1"] and not pool.is_resident("bin1"), - f"evicted={pool.evicted} resident={pool.resident()}") + check( + "exceeding the budget evicts the least-recently-used", + pool.evicted == ["bin1"] and not pool.is_resident("bin1"), + f"evicted={pool.evicted} resident={pool.resident()}", + ) check("the just-attached lease is never the victim", pool.is_resident("bin3")) check("eviction saves the database first", made["bin1"].saved == 1) check("eviction closes the lease", made["bin1"].closed) - check("pool is back within budget", pool.memory_mb() <= pool.budget_mb, - f"{pool.memory_mb()}/{pool.budget_mb}") + check( + "pool is back within budget", + pool.memory_mb() <= pool.budget_mb, + f"{pool.memory_mb()}/{pool.budget_mb}", + ) # -- the active binary is never evicted ------------------------------- # pool.set_active("bin2") - check("set_active touches the LRU", pool.resident()[-1] == "bin2", - pool.resident()) + check( + "set_active touches the LRU", pool.resident()[-1] == "bin2", pool.resident() + ) pool.get("bin1") # over budget again -> must evict, but not bin2 - check("the active binary survives eviction", pool.is_resident("bin2"), - f"resident={pool.resident()}") + check( + "the active binary survives eviction", + pool.is_resident("bin2"), + f"resident={pool.resident()}", + ) # -- pinning ---------------------------------------------------------- # pool.close_all() @@ -127,8 +142,11 @@ def main() -> int: pool2.pin("bin0") pool2.get("bin1") pool2.get("bin2") # 300 > 250 -> evict, but bin0 is pinned - check("pinned binaries are never evicted", pool2.is_resident("bin0"), - f"resident={pool2.resident()} evicted={pool2.evicted}") + check( + "pinned binaries are never evicted", + pool2.is_resident("bin0"), + f"resident={pool2.resident()} evicted={pool2.evicted}", + ) check("an unpinned one went instead", "bin1" in pool2.evicted, pool2.evicted) # -- everything pinned/active: stop evicting rather than thrash -------- # @@ -136,21 +154,30 @@ def main() -> int: pool2.set_active("bin2") n_before = len(pool2.evicted) pool2._enforce_budget() - check("nothing evictable -> gives up instead of thrashing", - len(pool2.evicted) == n_before, pool2.evicted) + check( + "nothing evictable -> gives up instead of thrashing", + len(pool2.evicted) == n_before, + pool2.evicted, + ) # -- status for the switcher UI ---------------------------------------- # st = {s["label"]: s for s in pool2.status()} check("status() covers every project binary", len(st) == 4, list(st)) - check("status() marks resident/pinned/active", - st["bin0"]["resident"] and st["bin0"]["pinned"] - and st["bin2"]["active"] and not st["bin3"]["resident"], - f"{st}") + check( + "status() marks resident/pinned/active", + st["bin0"]["resident"] + and st["bin0"]["pinned"] + and st["bin2"]["active"] + and not st["bin3"]["resident"], + f"{st}", + ) # -- teardown ----------------------------------------------------------- # pool2.close_all() - check("close_all() closes every lease", - not pool2.resident() and all(c.closed for c in made.values())) + check( + "close_all() closes every lease", + not pool2.resident() and all(c.closed for c in made.values()), + ) check("close_all() clears the active binary", pool2.active is None) # -- unknown label -------------------------------------------------------- # @@ -164,36 +191,45 @@ def main() -> int: discard_made = {} def spawn_discard(ref, ttl): - client = FakeClient( - ref, discardable=ref.label != "bin1") + client = FakeClient(ref, discardable=ref.label != "bin1") discard_made[ref.label] = client return client - discard_pool = DatabasePool( - proj, spawn=spawn_discard, mem_fn=lambda c: c.mem) + discard_pool = DatabasePool(proj, spawn=spawn_discard, mem_fn=lambda c: c.mem) discard_pool.get("bin0") discard_pool.get("bin1") delegated = discard_pool.discard_changes(["bin0", "bin1"]) - check("discard asks every dirty resident database", - discard_made["bin0"].discarded == 1 - and discard_made["bin1"].discarded == 1, - {k: c.discarded for k, c in discard_made.items()}) - check("discard reports leases whose finalization transferred", - delegated == ["bin1"], delegated) + check( + "discard asks every dirty resident database", + discard_made["bin0"].discarded == 1 and discard_made["bin1"].discarded == 1, + {k: c.discarded for k, c in discard_made.items()}, + ) + check( + "discard reports leases whose finalization transferred", + delegated == ["bin1"], + delegated, + ) old = discard_made["bin0"] replacement = FakeClient(proj.by_label("bin0")) - check("replace_client refuses a stale lease generation", - discard_pool.replace_client("bin0", object(), replacement) is False - and discard_pool.get("bin0") is old) - check("replace_client installs the reattached lease", - discard_pool.replace_client("bin0", old, replacement) is True - and discard_pool.get("bin0") is replacement) + check( + "replace_client refuses a stale lease generation", + discard_pool.replace_client("bin0", object(), replacement) is False + and discard_pool.get("bin0") is old, + ) + check( + "replace_client installs the reattached lease", + discard_pool.replace_client("bin0", old, replacement) is True + and discard_pool.get("bin0") is replacement, + ) discard_pool.close_all(save=False) # -- default budget comes from the project's memory_pct ------------------- # pool3 = DatabasePool(proj, spawn=spawn, mem_fn=lambda c: c.mem) - check("default budget is derived, not a fixed lease count", - pool3.budget_mb >= 256, pool3.budget_mb) + check( + "default budget is derived, not a fixed lease count", + pool3.budget_mb >= 256, + pool3.budget_mb, + ) # -- prewarm: speculative, and never at the cost of a real binary ------ # with tempfile.TemporaryDirectory() as tmp: @@ -205,24 +241,34 @@ def main() -> int: made2[ref.label] = c return c - pool = DatabasePool(proj, budget_mb=250, spawn=spawn2, - mem_fn=lambda c: c.mem) + pool = DatabasePool(proj, budget_mb=250, spawn=spawn2, mem_fn=lambda c: c.mem) labels = [r.label for r in proj.refs] a, b, c_ = labels[0], labels[1], labels[2] pool.get(a) pool.set_active(a) - check("prewarm warms a binary when the budget has room", - pool.prewarm(b) is True and b in pool.resident(), f"{pool.resident()}") - check("prewarm is a no-op for something already resident", - pool.prewarm(b) is False) + check( + "prewarm warms a binary when the budget has room", + pool.prewarm(b) is True and b in pool.resident(), + f"{pool.resident()}", + ) + check( + "prewarm is a no-op for something already resident", + pool.prewarm(b) is False, + ) # 2 x 100MB resident, estimate 100 more -> 300 > 250: must refuse - check("prewarm refuses rather than making room", - pool.prewarm(c_) is False and c_ not in pool.resident(), - f"resident={pool.resident()} mem={pool.memory_mb()}/{pool.budget_mb}") - check("refusing to prewarm evicts nothing", - set(pool.resident()) == {a, b}, f"{pool.resident()}") - check("prewarm ignores a label outside the project", - pool.prewarm("nope") is False) + check( + "prewarm refuses rather than making room", + pool.prewarm(c_) is False and c_ not in pool.resident(), + f"resident={pool.resident()} mem={pool.memory_mb()}/{pool.budget_mb}", + ) + check( + "refusing to prewarm evicts nothing", + set(pool.resident()) == {a, b}, + f"{pool.resident()}", + ) + check( + "prewarm ignores a label outside the project", pool.prewarm("nope") is False + ) # Budget eviction releases GUI leases but must not save somebody's open IDA # implicitly. An explicit save-and-close remains authoritative. @@ -239,12 +285,16 @@ def main() -> int: label = proj.refs[0].label pool.get(label) pool.evict(label) - check("LRU release does not implicitly save a GUI database", - made_gui[-1].saved == 0) + check( + "LRU release does not implicitly save a GUI database", + made_gui[-1].saved == 0, + ) pool.get(label) pool.close_all(save=True) - check("explicit close_all(save=True) does save a GUI database", - made_gui[-1].saved == 1) + check( + "explicit close_all(save=True) does save a GUI database", + made_gui[-1].saved == 1, + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_project.py b/tests/test_project.py index bae0802..63d029e 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -15,7 +15,7 @@ import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.project import Project, ProjectError, SIDECAR_SUFFIX # noqa: E402 +from idatui.project import SIDECAR_SUFFIX, Project, ProjectError # noqa: E402 PASS = FAIL = 0 @@ -48,36 +48,48 @@ def main() -> int: proj = Project.create(pfile, [httpd, libauth], name="router-fw") check("create() writes the project file", os.path.isfile(pfile)) proj = Project.load(pfile) - check("load() round-trips name + binaries", - proj.name == "router-fw" and len(proj.refs) == 2, - f"name={proj.name} n={len(proj.refs)}") - check("labels default to the basename", - [r.label for r in proj.refs] == ["httpd", "libauth.so"], - f"{[r.label for r in proj.refs]}") + check( + "load() round-trips name + binaries", + proj.name == "router-fw" and len(proj.refs) == 2, + f"name={proj.name} n={len(proj.refs)}", + ) + check( + "labels default to the basename", + [r.label for r in proj.refs] == ["httpd", "libauth.so"], + f"{[r.label for r in proj.refs]}", + ) # -- layout -------------------------------------------------------- # - check("sidecar sits beside the project file", - proj.sidecar == os.path.join(tmp, "router-fw" + SIDECAR_SUFFIX), - proj.sidecar) + check( + "sidecar sits beside the project file", + proj.sidecar == os.path.join(tmp, "router-fw" + SIDECAR_SUFFIX), + proj.sidecar, + ) ref = proj.by_label("httpd") - check("staged path lives in the sidecar, not the source tree", - ref.staged.startswith(proj.bin_dir) and src not in ref.staged, - ref.staged) + check( + "staged path lives in the sidecar, not the source tree", + ref.staged.startswith(proj.bin_dir) and src not in ref.staged, + ref.staged, + ) check("db path hangs off the staged file", ref.db == ref.staged + ".i64") # -- staging: hardlink, freshness ---------------------------------- # check("a binary starts out stale (not yet staged)", proj.is_stale(ref)) proj.stage(ref) - check("stage() materialises the binary in the sidecar", - os.path.isfile(ref.staged)) - check("stage() copies (distinct inode) so the source can't be mutated " - "through it", - os.stat(ref.staged).st_ino != os.stat(ref.source).st_ino - and open(ref.staged, "rb").read() == open(ref.source, "rb").read()) + check( + "stage() materialises the binary in the sidecar", os.path.isfile(ref.staged) + ) + check( + "stage() copies (distinct inode) so the source can't be mutated through it", + os.stat(ref.staged).st_ino != os.stat(ref.source).st_ino + and open(ref.staged, "rb").read() == open(ref.source, "rb").read(), + ) check("a staged binary is no longer stale", not proj.is_stale(ref)) - check("source tree stays clean (no IDA artifacts beside it)", - sorted(os.listdir(src)) == ["httpd", "libauth.so"], - f"{sorted(os.listdir(src))}") + check( + "source tree stays clean (no IDA artifacts beside it)", + sorted(os.listdir(src)) == ["httpd", "libauth.so"], + f"{sorted(os.listdir(src))}", + ) # -- a changed source re-stages and drops the stale DB ------------- # open(ref.db, "wb").write(b"fake i64") @@ -88,11 +100,15 @@ def main() -> int: os.utime(ref.source, (1, 1)) check("a source rebuilt in place goes stale", proj.is_stale(ref)) proj.stage(ref) - check("re-staging refreshes the staged bytes", - open(ref.staged, "rb").read().endswith(b"v2 (longer)")) + check( + "re-staging refreshes the staged bytes", + open(ref.staged, "rb").read().endswith(b"v2 (longer)"), + ) check("re-staging drops the now-stale database", not proj.has_db(ref)) - check("re-staging drops the stale scratch too", - not os.path.exists(ref.staged + ".id0")) + check( + "re-staging drops the stale scratch too", + not os.path.exists(ref.staged + ".id0"), + ) # -- scratch sweep keeps the DB ------------------------------------ # open(ref.db, "wb").write(b"fake i64") @@ -107,22 +123,35 @@ def main() -> int: os.makedirs(sub) dup = _bin(os.path.join(sub, "httpd"), b"\x7fELF other httpd") with open(pfile, "w") as f: - json.dump({"name": "p", "binaries": [ - {"path": "src/httpd"}, # relative to the project file - {"path": dup}, # same basename -> collision - {"path": libauth, "label": "auth"}, - ]}, f) + json.dump( + { + "name": "p", + "binaries": [ + {"path": "src/httpd"}, # relative to the project file + {"path": dup}, # same basename -> collision + {"path": libauth, "label": "auth"}, + ], + }, + f, + ) proj2 = Project.load(pfile) - check("relative paths resolve against the project file", - proj2.refs[0].source == httpd, proj2.refs[0].source) - check("colliding labels are disambiguated", - [r.label for r in proj2.refs] == ["httpd", "httpd_2", "auth"], - f"{[r.label for r in proj2.refs]}") + check( + "relative paths resolve against the project file", + proj2.refs[0].source == httpd, + proj2.refs[0].source, + ) + check( + "colliding labels are disambiguated", + [r.label for r in proj2.refs] == ["httpd", "httpd_2", "auth"], + f"{[r.label for r in proj2.refs]}", + ) check("explicit labels are honoured", proj2.by_label("auth") is not None) proj2.stage_all() - check("stage_all() stages every binary to a distinct file", - len({r.staged for r in proj2.refs}) == 3 - and all(os.path.isfile(r.staged) for r in proj2.refs)) + check( + "stage_all() stages every binary to a distinct file", + len({r.staged for r in proj2.refs}) == 3 + and all(os.path.isfile(r.staged) for r in proj2.refs), + ) # -- add / remove ---------------------------------------------------- # extra = _bin(os.path.join(src, "extra")) @@ -132,39 +161,60 @@ def main() -> int: # -- re-adding must not duplicate (matched by resolved path) --------- # n = len(proj2.refs) proj2.add(extra) - check("re-adding the same path is a no-op", len(proj2.refs) == n, - f"{[r.label for r in proj2.refs]}") + check( + "re-adding the same path is a no-op", + len(proj2.refs) == n, + f"{[r.label for r in proj2.refs]}", + ) os.chdir(src) - proj2.add("./extra") # same file, relative - proj2.add(os.path.join(src, "..", "src", "extra")) # same file, messy - check("a different spelling of the same path is a no-op", - len(proj2.refs) == n, f"{[r.label for r in proj2.refs]}") + proj2.add("./extra") # same file, relative + proj2.add(os.path.join(src, "..", "src", "extra")) # same file, messy + check( + "a different spelling of the same path is a no-op", + len(proj2.refs) == n, + f"{[r.label for r in proj2.refs]}", + ) link = os.path.join(src, "extra_link") os.symlink(extra, link) proj2.add(link) - check("a symlink to an existing binary is a no-op", - len(proj2.refs) == n, f"{[r.label for r in proj2.refs]}") + check( + "a symlink to an existing binary is a no-op", + len(proj2.refs) == n, + f"{[r.label for r in proj2.refs]}", + ) # ...but a DIFFERENT file with the same basename must still be added other_dir = os.path.join(tmp, "other2") os.makedirs(other_dir) twin = _bin(os.path.join(other_dir, "extra"), b"\x7fELF a different extra") proj2.add(twin) - check("a same-named file from another directory IS added", - len(proj2.refs) == n + 1 - and proj2.by_source(twin) is not None - and proj2.by_source(extra) is not proj2.by_source(twin), - f"{[(r.label, r.source) for r in proj2.refs[-2:]]}") - check("the twins get distinct labels", - len({r.label for r in proj2.refs}) == len(proj2.refs), - f"{[r.label for r in proj2.refs]}") - check("create() also drops repeats on the command line", - len(Project.create(os.path.join(tmp, "dup.json"), - [extra, "./extra", extra]).refs) == 1) + check( + "a same-named file from another directory IS added", + len(proj2.refs) == n + 1 + and proj2.by_source(twin) is not None + and proj2.by_source(extra) is not proj2.by_source(twin), + f"{[(r.label, r.source) for r in proj2.refs[-2:]]}", + ) + check( + "the twins get distinct labels", + len({r.label for r in proj2.refs}) == len(proj2.refs), + f"{[r.label for r in proj2.refs]}", + ) + check( + "create() also drops repeats on the command line", + len( + Project.create( + os.path.join(tmp, "dup.json"), [extra, "./extra", extra] + ).refs + ) + == 1, + ) os.chdir(tmp) proj2.remove(proj2.by_source(twin).label) - check("remove() drops one", proj2.remove("extra") - and proj2.by_label("extra") is None) + check( + "remove() drops one", + proj2.remove("extra") and proj2.by_label("extra") is None, + ) # -- bad input ------------------------------------------------------- # bad = os.path.join(tmp, "bad.json") @@ -191,43 +241,62 @@ def main() -> int: # -- load options for headerless blobs --------------------------------- # with tempfile.TemporaryDirectory() as tmp: - src = os.path.join(tmp, "src"); os.makedirs(src) + src = os.path.join(tmp, "src") + os.makedirs(src) blob = os.path.join(src, "fw.bin") with open(blob, "wb") as f: f.write(b"\x00" * 64) - proj = Project.create(os.path.join(tmp, "p.json"), [blob], name="p", - load={"processor": "arm", "base": 0x8000000}) + proj = Project.create( + os.path.join(tmp, "p.json"), + [blob], + name="p", + load={"processor": "arm", "base": 0x8000000}, + ) r = proj.refs[0] - check("create() records load options per binary", - r.processor == "arm" and r.base == 0x8000000, - f"proc={r.processor!r} base={r.base:#x}") + check( + "create() records load options per binary", + r.processor == "arm" and r.base == 0x8000000, + f"proc={r.processor!r} base={r.base:#x}", + ) # -b is PARAGRAPHS: 0x8000000 >> 4 == 0x800000. Getting this wrong loads # the image 16x too high and every address in the database is wrong. - check("base is converted to IDA's paragraph units", - r.load_args == "-parm -b800000", r.load_args) + check( + "base is converted to IDA's paragraph units", + r.load_args == "-parm -b800000", + r.load_args, + ) proj2 = Project.load(proj.path) - check("load options survive a round-trip through the file", - proj2.refs[0].load_args == "-parm -b800000", - proj2.refs[0].load_args) + check( + "load options survive a round-trip through the file", + proj2.refs[0].load_args == "-parm -b800000", + proj2.refs[0].load_args, + ) blob2 = os.path.join(src, "other.bin") with open(blob2, "wb") as f: f.write(b"\x00" * 64) r2 = proj2.add(blob2, load={"processor": "mipsb"}) - check("add() takes load options too", - r2.load_args == "-pmipsb", r2.load_args) + check("add() takes load options too", r2.load_args == "-pmipsb", r2.load_args) # a normal ELF needs none of this and must pass nothing - check("a binary with no load options passes no switches", - Project.create(os.path.join(tmp, "q.json"), [blob], - name="q").refs[0].load_args == "") + check( + "a binary with no load options passes no switches", + Project.create(os.path.join(tmp, "q.json"), [blob], name="q") + .refs[0] + .load_args + == "", + ) # addresses get written by hand, so accept how people write them - proj3 = Project.create(os.path.join(tmp, "r.json"), [blob], name="r", - load={"base": "0x1000"}) - check("a base given as a hex STRING is parsed", - proj3.refs[0].base == 0x1000, f"{proj3.refs[0].base}") + proj3 = Project.create( + os.path.join(tmp, "r.json"), [blob], name="r", load={"base": "0x1000"} + ) + check( + "a base given as a hex STRING is parsed", + proj3.refs[0].base == 0x1000, + f"{proj3.refs[0].base}", + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_project_ui.py b/tests/test_project_ui.py index 863b98a..9b388c8 100644 --- a/tests/test_project_ui.py +++ b/tests/test_project_ui.py @@ -21,12 +21,15 @@ import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from _fixtures import fast_keys # noqa: E402 -from idatui._sync import settle as quiesce, wait_for # noqa: E402 -fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +from idatui._sync import settle as quiesce # noqa: E402 +from idatui._sync import wait_for + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +from textual.widgets import Input, OptionList, Static # noqa: E402 + from idatui.app import IdaTui, ProjectPalette # noqa: E402 from idatui.project import Project # noqa: E402 -from textual.widgets import Input, OptionList, Static # noqa: E402 PASS = FAIL = 0 @@ -57,6 +60,7 @@ async def run(bins): app = IdaTui(keepalive=False, project=proj) async with app.run_test(size=(140, 44)) as pilot: + async def settle(pred, t=180.0): return await wait_for(pred, pilot.pause, t, 0.05) @@ -73,114 +77,163 @@ async def run(bins): waited for explicitly, not hoped for. """ return await settle( - lambda: app.program is not None - and app._func_index is not None - and app._func_index.complete - and app._loading_screen is None - and len(app.screen_stack) == 1, t) + lambda: ( + app.program is not None + and app._func_index is not None + and app._func_index.complete + and app._loading_screen is None + and len(app.screen_stack) == 1 + ), + t, + ) # -- boots on the project's first binary ----------------------- # ok = await usable() - check("project mode boots on the first binary", ok, - f"binary={app._binary}") - check("the active binary is the first one", app._binary == first, - f"{app._binary}") + check("project mode boots on the first binary", ok, f"binary={app._binary}") + check( + "the active binary is the first one", + app._binary == first, + f"{app._binary}", + ) n_first = len(app._func_index) check("its functions loaded", n_first > 10, f"n={n_first}") status = str(app.query_one("#status", Static).render()) - check("the status line names the active binary", - f"[{first}]" in status, status[:60]) + check( + "the status line names the active binary", + f"[{first}]" in status, + status[:60], + ) # -- the switcher lists the project ---------------------------- # await pilot.press("ctrl+o") - opened = await settle( - lambda: isinstance(app.screen, ProjectPalette), 20) - check("Ctrl+O opens the binary switcher", opened, - f"screen={type(app.screen).__name__}") + opened = await settle(lambda: isinstance(app.screen, ProjectPalette), 20) + check( + "Ctrl+O opens the binary switcher", + opened, + f"screen={type(app.screen).__name__}", + ) if not opened: return pal = app.screen - check("the switcher lists every project binary", - len(pal._results) == 2, f"{[e['label'] for e in pal._results]}") - check("it marks which one is active", - any(e["active"] and e["label"] == first for e in pal._results)) - check("it marks the other as not yet opened", - any(not e["resident"] and e["label"] == second - for e in pal._results)) + check( + "the switcher lists every project binary", + len(pal._results) == 2, + f"{[e['label'] for e in pal._results]}", + ) + check( + "it marks which one is active", + any(e["active"] and e["label"] == first for e in pal._results), + ) + check( + "it marks the other as not yet opened", + any(not e["resident"] and e["label"] == second for e in pal._results), + ) ol = pal.query_one(OptionList) - check("the switcher opens on the binary you're already in", - ol.highlighted is not None - and pal._results[ol.highlighted]["label"] == first, - f"highlighted={ol.highlighted} " - f"={pal._results[ol.highlighted]['label'] if ol.highlighted is not None else None} " - f"want={first}") + check( + "the switcher opens on the binary you're already in", + ol.highlighted is not None + and pal._results[ol.highlighted]["label"] == first, + f"highlighted={ol.highlighted} " + f"={pal._results[ol.highlighted]['label'] if ol.highlighted is not None else None} " + f"want={first}", + ) # -- switch to the second binary -------------------------------- # pal.query_one(Input).value = second await settle(lambda: bool(pal._results), 20) await pilot.press("enter") switched = await settle( - lambda: app._binary == second and app.program is not None - and app._func_index is not None and app._func_index.complete - and len(app.screen_stack) == 1) - check("switching opens the other binary", switched, - f"binary={app._binary}") - check("the second binary has its own function index", - app._func_index is not None and len(app._func_index) > 5, - f"n={len(app._func_index) if app._func_index else 0}") - check("both binaries now have live workers", - sorted(app._pool.resident()) == sorted([first, second]), - f"{app._pool.resident()}") + lambda: ( + app._binary == second + and app.program is not None + and app._func_index is not None + and app._func_index.complete + and len(app.screen_stack) == 1 + ) + ) + check("switching opens the other binary", switched, f"binary={app._binary}") + check( + "the second binary has its own function index", + app._func_index is not None and len(app._func_index) > 5, + f"n={len(app._func_index) if app._func_index else 0}", + ) + check( + "both binaries now have live workers", + sorted(app._pool.resident()) == sorted([first, second]), + f"{app._pool.resident()}", + ) landed = await settle(lambda: app._cur is not None, 60) - check("it lands somewhere in the new binary", landed, - f"cur={app._cur}") + check("it lands somewhere in the new binary", landed, f"cur={app._cur}") where = app._cur.ea if app._cur else None # -- switch back: resident, so state is restored ---------------- # await pilot.press("ctrl+o") - reopened = await settle( - lambda: isinstance(app.screen, ProjectPalette), 20) - check("the switcher reopens after a switch", reopened, - f"screen={type(app.screen).__name__}") + reopened = await settle(lambda: isinstance(app.screen, ProjectPalette), 20) + check( + "the switcher reopens after a switch", + reopened, + f"screen={type(app.screen).__name__}", + ) if not reopened: return app.screen.query_one(Input).value = first await settle(lambda: bool(app.screen._results), 20) await pilot.press("enter") - back = await settle(lambda: app._binary == first - and app._func_index is not None - and app._func_index.complete - and len(app.screen_stack) == 1, 120) - check("switching back returns to the first binary", back, - f"binary={app._binary}") - check("its function index came back intact", - app._func_index is not None and len(app._func_index) == n_first, - f"n={len(app._func_index) if app._func_index else 0} want={n_first}") + back = await settle( + lambda: ( + app._binary == first + and app._func_index is not None + and app._func_index.complete + and len(app.screen_stack) == 1 + ), + 120, + ) + check( + "switching back returns to the first binary", + back, + f"binary={app._binary}", + ) + check( + "its function index came back intact", + app._func_index is not None and len(app._func_index) == n_first, + f"n={len(app._func_index) if app._func_index else 0} want={n_first}", + ) # and forward again: the second binary's position was remembered await pilot.press("ctrl+o") if not await settle(lambda: isinstance(app.screen, ProjectPalette), 20): - check("returning to a binary restores where you were", False, - "switcher did not reopen") + check( + "returning to a binary restores where you were", + False, + "switcher did not reopen", + ) return app.screen.query_one(Input).value = second await settle(lambda: bool(app.screen._results), 20) await pilot.press("enter") - again = await settle(lambda: app._binary == second - and app._cur is not None, 120) - check("returning to a binary restores where you were", - again and app._cur.ea == where, - f"cur={app._cur.ea if app._cur else None} want={where}") + again = await settle( + lambda: app._binary == second and app._cur is not None, 120 + ) + check( + "returning to a binary restores where you were", + again and app._cur.ea == where, + f"cur={app._cur.ea if app._cur else None} want={where}", + ) # -- project-wide symbol search ------------------------------- # # 'main' exists in BOTH binaries: identical name, so the rank tuple # ties and a bare sort() would fall through to comparing Hit objects # ('<' not supported between instances of 'Hit'). from idatui.app import SymbolPalette - await settle(lambda: app._index is not None - and len(app._index.counts()) == 2, 60) - check("both binaries got indexed", - len(app._index.counts()) == 2, f"{app._index.counts()}") + + await settle( + lambda: app._index is not None and len(app._index.counts()) == 2, 60 + ) + check( + "both binaries got indexed", + len(app._index.counts()) == 2, + f"{app._index.counts()}", + ) await pilot.press("ctrl+n") if await settle(lambda: isinstance(app.screen, SymbolPalette), 20): pal = app.screen @@ -190,12 +243,14 @@ async def run(bins): # quiescence -- NOT "a foreign binary appeared", which is the # thing under test and would sit out its whole timeout on the # day it breaks. - await pilot.press("f2") # widen to the whole project + await pilot.press("f2") # widen to the whole project await quiesce(app) names = [(b, n) for b, _, n in pal._results] - check("project scope finds a name shared by both binaries", - len({b for b, n in names if n == "main"}) == 2, - f"{names[:6]}") + check( + "project scope finds a name shared by both binaries", + len({b for b, n in names if n == "main"}) == 2, + f"{names[:6]}", + ) await pilot.press("escape") await settle(lambda: not isinstance(app.screen, SymbolPalette), 20) @@ -209,17 +264,31 @@ async def run(bins): target = app._index.search("main", limit=200) tgt = next((h for h in target if h.binary == there), None) if tgt is None: - check("cross-binary jump records a hop", False, "no hit in the other binary") + check( + "cross-binary jump records a hop", + False, + "no hit in the other binary", + ) else: app._switch_then_goto(tgt.binary, tgt.addr) - jumped = await settle(lambda: app._binary == there - and app._func_index is not None - and app._func_index.complete, 180) - check("a project hit switches to the other binary", jumped, - f"binary={app._binary} want={there}") - check("the jump records where it came from", - len(app._hops) == hops0 + 1 and app._hops[-1] == here, - f"hops={app._hops}") + jumped = await settle( + lambda: ( + app._binary == there + and app._func_index is not None + and app._func_index.complete + ), + 180, + ) + check( + "a project hit switches to the other binary", + jumped, + f"binary={app._binary} want={there}", + ) + check( + "the jump records where it came from", + len(app._hops) == hops0 + 1 and app._hops[-1] == here, + f"hops={app._hops}", + ) # spend the local history first, then Esc must cross back for _ in range(6): if not app._hops or app._binary != there: @@ -227,10 +296,16 @@ async def run(bins): await pilot.press("escape") await quiesce(app) returned = await settle(lambda: app._binary == here, 180) - check("Esc crosses back to the binary the jump came from", - returned, f"binary={app._binary} want={here} hops={app._hops}") - check("the hop is consumed, not repeated", - not app._hops, f"hops={app._hops}") + check( + "Esc crosses back to the binary the jump came from", + returned, + f"binary={app._binary} want={here} hops={app._hops}", + ) + check( + "the hop is consumed, not repeated", + not app._hops, + f"hops={app._hops}", + ) # -- xrefs: callers in OTHER project binaries ------------------ # # xrefs_to only sees this database, so an exported function looks @@ -238,9 +313,13 @@ async def run(bins): # The selection rule is "only for a symbol we actually export"; two # executables share no linkage, so here it must stay quiet. from idatui.app import XrefsScreen + fake = app._foreign_importers(app._cur.ea, "strrchr", None) - check("no cross-binary callers for a symbol this binary doesn't export", - fake == [], f"{fake}") + check( + "no cross-binary callers for a symbol this binary doesn't export", + fake == [], + f"{fake}", + ) # The routing a real cross-binary caller takes: the dialog carries a # (binary, addr) payload instead of a bare address, and choosing it @@ -248,31 +327,55 @@ async def run(bins): # so Esc comes back. where_from = app._binary other = first if where_from == second else second - hit = next((h for h in app._index.search("main", limit=200) - if h.binary == other), None) + hit = next( + (h for h in app._index.search("main", limit=200) if h.binary == other), + None, + ) if hit is None: - check("a cross-binary xref jumps to the other binary", False, - "no symbol found in the other binary") + check( + "a cross-binary xref jumps to the other binary", + False, + "no symbol found in the other binary", + ) else: hops0 = len(app._hops) app.push_screen( - XrefsScreen("xrefs to fake", [((hit.binary, hit.addr), - f"{hit.addr:08X} import [{hit.binary}]")]), - app._on_xref_chosen) + XrefsScreen( + "xrefs to fake", + [ + ( + (hit.binary, hit.addr), + f"{hit.addr:08X} import [{hit.binary}]", + ) + ], + ), + app._on_xref_chosen, + ) await settle(lambda: isinstance(app.screen, XrefsScreen), 20) await pilot.press("enter") - jumped = await settle(lambda: app._binary == other - and app._func_index is not None - and app._func_index.complete, 180) - check("a cross-binary xref jumps to the other binary", jumped, - f"binary={app._binary} want={other}") - check("and records a hop so Esc returns", - len(app._hops) == hops0 + 1 and app._hops[-1] == where_from, - f"hops={app._hops}") + jumped = await settle( + lambda: ( + app._binary == other + and app._func_index is not None + and app._func_index.complete + ), + 180, + ) + check( + "a cross-binary xref jumps to the other binary", + jumped, + f"binary={app._binary} want={other}", + ) + check( + "and records a hop so Esc returns", + len(app._hops) == hops0 + 1 and app._hops[-1] == where_from, + f"hops={app._hops}", + ) app._hops.clear() # -- and the same toggle for strings --------------------------- # from idatui.app import StringsPalette + await pilot.press("quotation_mark") if await settle(lambda: isinstance(app.screen, StringsPalette), 30): pal = app.screen @@ -282,20 +385,32 @@ async def run(bins): await pilot.press("f2") await quiesce(app) wide = {b for b, _, _ in pal._results} - check("strings: local scope is this binary only", local == {None}, - f"{local}") - check("strings: F2 widens across the project", - len(wide) >= 2 and None not in wide, f"{wide}") + check( + "strings: local scope is this binary only", + local == {None}, + f"{local}", + ) + check( + "strings: F2 widens across the project", + len(wide) >= 2 and None not in wide, + f"{wide}", + ) await pilot.press("escape") await settle(lambda: not isinstance(app.screen, StringsPalette), 20) # -- the promise: nothing was written next to the sources ---------- # left = sorted(os.listdir(src)) - check("the source tree stays pristine (no .i64/scratch beside it)", - left == sorted(os.path.basename(s) for s in srcs), f"{left}") + check( + "the source tree stays pristine (no .i64/scratch beside it)", + left == sorted(os.path.basename(s) for s in srcs), + f"{left}", + ) staged = sorted(os.listdir(proj.bin_dir)) - check("IDA's artifacts all live in the project sidecar", - any(f.endswith(".i64") for f in staged), f"{staged}") + check( + "IDA's artifacts all live in the project sidecar", + any(f.endswith(".i64") for f in staged), + f"{staged}", + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 @@ -303,8 +418,10 @@ async def run(bins): def main(argv): repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - bins = argv or [os.path.join(repo, "targets", "echo"), - os.path.join(repo, "targets", "cat")] + bins = argv or [ + os.path.join(repo, "targets", "echo"), + os.path.join(repo, "targets", "cat"), + ] for b in bins: if not os.path.isfile(b): print(f"no such binary: {b}") diff --git a/tests/test_rawimage_rpc.py b/tests/test_rawimage_rpc.py index d39ad35..a6e7661 100644 --- a/tests/test_rawimage_rpc.py +++ b/tests/test_rawimage_rpc.py @@ -36,7 +36,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from idatui.rpcclient import RpcClient, RpcError # noqa: E402 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -BLOB = os.path.join(REPO, "experiments", "fibonacci.bin") # real Thumb code +BLOB = os.path.join(REPO, "experiments", "fibonacci.bin") # real Thumb code PASS = FAIL = 0 @@ -52,11 +52,24 @@ def check(name, ok, detail=""): def spawn_pane(target, processor, timeout=420): - cmd = [sys.executable, "-m", "idatui.pane", "spawn", "--open", target, - "--processor", processor, "--detached", "--size", "60%", - "--timeout", str(timeout)] - r = subprocess.run(cmd, capture_output=True, text=True, - timeout=timeout + 60, cwd=REPO) + cmd = [ + sys.executable, + "-m", + "idatui.pane", + "spawn", + "--open", + target, + "--processor", + processor, + "--detached", + "--size", + "60%", + "--timeout", + str(timeout), + ] + r = subprocess.run( + cmd, capture_output=True, text=True, timeout=timeout + 60, cwd=REPO + ) if not r.stdout.strip(): print(f" spawn produced no JSON: {r.stderr.strip()}", file=sys.stderr) return None @@ -64,9 +77,22 @@ def spawn_pane(target, processor, timeout=420): def stop_pane(sock, timeout=60): - subprocess.run([sys.executable, "-m", "idatui.pane", "stop", "--sock", sock, - "--timeout", str(timeout)], - capture_output=True, text=True, timeout=timeout + 10, cwd=REPO) + subprocess.run( + [ + sys.executable, + "-m", + "idatui.pane", + "stop", + "--sock", + sock, + "--timeout", + str(timeout), + ], + capture_output=True, + text=True, + timeout=timeout + 10, + cwd=REPO, + ) def main() -> int: @@ -94,34 +120,44 @@ def main() -> int: # -- load options actually reached IDA --------------------------- # # Wrong processor => the disassembly is nonsense or absent; ARMv7-A # also means a 32-bit database, without which Hex-Rays refuses. - check("spawn forwarded --processor", info.get("ok"), - json.dumps(info)) + check("spawn forwarded --processor", info.get("ok"), json.dumps(info)) with RpcClient(sock) as c: st = c.call("state") - check("pane is drivable", st.get("active") in - ("listing", "decomp", "hex"), json.dumps(st)[:200]) + check( + "pane is drivable", + st.get("active") in ("listing", "decomp", "hex"), + json.dumps(st)[:200], + ) # -- define ------------------------------------------------- # # fibonacci.bin is Thumb at 0x0; as ARM it does not decode. r = c.call("define", kind="thumb", target="0x0") d = r.get("define", {}) check("define thumb ran", "define" in r, json.dumps(r)[:200]) - check("define thumb decoded instructions", - "instruction" in d.get("status", ""), d.get("status", "")) + check( + "define thumb decoded instructions", + "instruction" in d.get("status", ""), + d.get("status", ""), + ) r = c.call("define", kind="func", target="0x0") - check("define func created a function", - "function" in r["define"]["status"] - or "already" in r["define"]["status"], - r["define"]["status"]) + check( + "define func created a function", + "function" in r["define"]["status"] + or "already" in r["define"]["status"], + r["define"]["status"], + ) bad = None try: c.call("define", kind="nonsense") except RpcError as e: bad = str(e) - check("define rejects an unknown kind", bad is not None - and "unknown define kind" in bad, str(bad)) + check( + "define rejects an unknown kind", + bad is not None and "unknown define kind" in bad, + str(bad), + ) # -- opfmt (how a literal is displayed) --------------------- # # Thumb code is full of small immediates -- the thing 'o' exists @@ -134,21 +170,31 @@ def main() -> int: lit = None for ln in seen: m = re.match(r"([0-9A-F]{8})\s+(.*)", ln.get("text", "")) - if not (m and re.search(r"#(0x[0-9A-Fa-f]{2,}|[1-9]\d+)\b", - m.group(2))): + if not ( + m and re.search(r"#(0x[0-9A-Fa-f]{2,}|[1-9]\d+)\b", m.group(2)) + ): continue ea_s = "0x" + m.group(1) - st = c.call("opfmt", mode="show", target=ea_s, delay_ms=0 - ).get("opfmt", {}).get("status", "") + st = ( + c.call("opfmt", mode="show", target=ea_s, delay_ms=0) + .get("opfmt", {}) + .get("status", "") + ) if "no literal" not in st: lit = (ea_s, st) break - check("the blob has an immediate to reformat", lit is not None, - json.dumps([ln.get("text") for ln in seen[:8]])) + check( + "the blob has an immediate to reformat", + lit is not None, + json.dumps([ln.get("text") for ln in seen[:8]]), + ) if lit is not None: tgt, st0 = lit - check("opfmt show reports the stops without editing", - "[" in st0 and "dec" in st0, st0) + check( + "opfmt show reports the stops without editing", + "[" in st0 and "dec" in st0, + st0, + ) r = c.call("opfmt", mode="dec", target=tgt, delay_ms=0) st1 = r.get("opfmt", {}).get("status", "") check("opfmt sets a named format", "dec" in st1, st1) @@ -156,16 +202,21 @@ def main() -> int: st2 = r.get("opfmt", {}).get("status", "") check("opfmt cycles on from there", "\u2192" in st2, st2) r = c.call("opfmt", mode="default") - check("opfmt hands the operand back to IDA", - "default" in r.get("opfmt", {}).get("status", ""), - r.get("opfmt", {}).get("status", "")) + check( + "opfmt hands the operand back to IDA", + "default" in r.get("opfmt", {}).get("status", ""), + r.get("opfmt", {}).get("status", ""), + ) badfmt = None try: c.call("opfmt", mode="roman") except RpcError as e: badfmt = str(e) - check("opfmt rejects an unknown mode", badfmt is not None - and "unknown opfmt mode" in badfmt, str(badfmt)) + check( + "opfmt rejects an unknown mode", + badfmt is not None and "unknown opfmt mode" in badfmt, + str(badfmt), + ) # -- rename_many -------------------------------------------- # fns = c.call("functions", limit=200) @@ -177,28 +228,48 @@ def main() -> int: # 'start' (not 'addr') on purpose: symbol files in the wild # use it, and accepting only one spelling is how a bulk # import silently renames nothing. - json.dump([{"start": hex(ea), "name": "bulk_named_fn"}, - {"start": "0xdeadbe", "name": "nowhere"}], f) + json.dump( + [ + {"start": hex(ea), "name": "bulk_named_fn"}, + {"start": "0xdeadbe", "name": "nowhere"}, + ], + f, + ) r = c.call("rename_many", file=symfile) m = r.get("rename_many", {}) - check("rename_many applied the good entry", m.get("ok") == 1, - json.dumps(m)) - check("rename_many reports the bad entry", - m.get("failed") == 1 and m.get("errors"), json.dumps(m)) + check( + "rename_many applied the good entry", + m.get("ok") == 1, + json.dumps(m), + ) + check( + "rename_many reports the bad entry", + m.get("failed") == 1 and m.get("errors"), + json.dumps(m), + ) # The readback matters more than the return value: a driver # trusts resolve/functions to decide what work is left. - check("renamed symbol resolves", - c.call("resolve", name="bulk_named_fn").get("ea") == ea, - json.dumps(c.call("resolve", name="bulk_named_fn"))) + check( + "renamed symbol resolves", + c.call("resolve", name="bulk_named_fn").get("ea") == ea, + json.dumps(c.call("resolve", name="bulk_named_fn")), + ) names = {f["name"] for f in c.call("functions", limit=200)} - check("function table shows the new name", - "bulk_named_fn" in names, str(sorted(names)[:10])) + check( + "function table shows the new name", + "bulk_named_fn" in names, + str(sorted(names)[:10]), + ) - r = c.call("rename_many", items=[{"addr": hex(ea), - "name": "inline_named_fn"}]) - check("rename_many takes inline items", - r["rename_many"]["ok"] == 1, json.dumps(r["rename_many"])) + r = c.call( + "rename_many", items=[{"addr": hex(ea), "name": "inline_named_fn"}] + ) + check( + "rename_many takes inline items", + r["rename_many"]["ok"] == 1, + json.dumps(r["rename_many"]), + ) # -- the stale-pseudocode trap ------------------------------ # # Hex-Rays caches per function and does not notice that a @@ -208,22 +279,31 @@ def main() -> int: # the function's own body cites it.) before = c.call("pseudocode", target=hex(ea)) pc_before = json.dumps(before) - r = c.call("rename_many", items=[{"addr": hex(ea), - "name": "after_cache_fn"}]) + r = c.call( + "rename_many", items=[{"addr": hex(ea), "name": "after_cache_fn"}] + ) pc_after = json.dumps(c.call("pseudocode", target=hex(ea))) - check("pseudocode was cached before the rename", - "inline_named_fn" in pc_before, pc_before[:200]) - check("rename_many invalidates the decompile cache", - "after_cache_fn" in pc_after - and "inline_named_fn" not in pc_after, pc_after[:300]) + check( + "pseudocode was cached before the rename", + "inline_named_fn" in pc_before, + pc_before[:200], + ) + check( + "rename_many invalidates the decompile cache", + "after_cache_fn" in pc_after and "inline_named_fn" not in pc_after, + pc_after[:300], + ) empty = None try: c.call("rename_many") except RpcError as e: empty = str(e) - check("rename_many without items errors", empty is not None - and "items" in empty, str(empty)) + check( + "rename_many without items errors", + empty is not None and "items" in empty, + str(empty), + ) finally: stop_pane(sock) diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 30d6087..4196445 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -32,15 +32,26 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from _fixtures import fast_keys, staged # noqa: E402 fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +from rich.text import Text # noqa: E402 +from textual.widgets import ( # noqa: E402 + DataTable, + Input, + OptionList, + Static, + TextArea, +) + from idatui import remote_ops # noqa: E402 +from idatui._sync import settle, wait_for # noqa: E402 from idatui.app import ( # noqa: E402 + _HELP, ConfirmScreen, DecompView, FunctionsPanel, GraphView, + HelpScreen, HexView, IdaTui, - HelpScreen, ListingView, QuitScreen, SearchPalette, @@ -48,20 +59,10 @@ from idatui.app import ( # noqa: E402 StructEditor, SymbolPalette, XrefsScreen, - _HELP, _str_display, _word_occurrences, ) from idatui.errors import IDAToolError # noqa: E402 -from textual.widgets import ( # noqa: E402 - DataTable, - Input, - OptionList, - Static, - TextArea, -) -from rich.text import Text # noqa: E402 -from idatui._sync import settle, wait_for # noqa: E402 PASS = FAIL = 0 STOP_AFTER = None diff --git a/tests/test_search.py b/tests/test_search.py index 6fd1a25..7b80dda 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -16,8 +16,13 @@ import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from idatui.search import ( # noqa: E402 - BYTES, TEXT, classify, looks_like_bytes, normalise_pattern, - pattern_problem, probably_meant_bytes, + BYTES, + TEXT, + classify, + looks_like_bytes, + normalise_pattern, + pattern_problem, + probably_meant_bytes, ) PASS = FAIL = 0 @@ -35,63 +40,89 @@ def check(name, cond, detail=""): def main() -> int: # -- the asymmetry: hex-looking WORDS must stay text --------------------- # - for word in ("add", "dead", "beef", "cafe", "ff", "0", "abcdef", - "decode", "face"): - check(f"{word!r} searches text, not bytes", - classify(word)[0] == TEXT, classify(word)) + for word in ("add", "dead", "beef", "cafe", "ff", "0", "abcdef", "decode", "face"): + check( + f"{word!r} searches text, not bytes", + classify(word)[0] == TEXT, + classify(word), + ) # -- unambiguous byte patterns ------------------------------------------ # - for pat in ("48 8b ?? c3", "B8 ? ? ? ? 90", "48,8b,05", "de ad be ef", - "48 8? ?? 24", "??"): - check(f"{pat!r} searches bytes", classify(pat)[0] == BYTES, - classify(pat)) + for pat in ( + "48 8b ?? c3", + "B8 ? ? ? ? 90", + "48,8b,05", + "de ad be ef", + "48 8? ?? 24", + "??", + ): + check(f"{pat!r} searches bytes", classify(pat)[0] == BYTES, classify(pat)) - check("a quoted literal is a byte pattern", - classify('"Hello", 0')[0] == BYTES) + check("a quoted literal is a byte pattern", classify('"Hello", 0')[0] == BYTES) # A TYPO in a byte pattern must stay a byte pattern, so it can be refused # with a reason. Falling back to text answers "no match", which is # indistinguishable from "those bytes are not in this binary". - check("a typo'd byte pattern is still a byte pattern", - classify("48 zz c3")[0] == BYTES, classify("48 zz c3")) - check("and it is refused by name", - "'zz'" in (pattern_problem("48 zz c3") or "")) - check("but a word among bytes is prose", - classify("add ff")[0] == TEXT and classify("mov rdi, rax")[0] == TEXT, - classify("add ff")) + check( + "a typo'd byte pattern is still a byte pattern", + classify("48 zz c3")[0] == BYTES, + classify("48 zz c3"), + ) + check("and it is refused by name", "'zz'" in (pattern_problem("48 zz c3") or "")) + check( + "but a word among bytes is prose", + classify("add ff")[0] == TEXT and classify("mov rdi, rax")[0] == TEXT, + classify("add ff"), + ) check("prose stays text", classify("mov rdi, rax")[0] == TEXT) check("a call target stays text", classify("call cs:__isoc99_scanf")[0] == TEXT) - check("an empty query is text (nothing to search yet)", - classify("")[0] == TEXT and not looks_like_bytes("")) + check( + "an empty query is text (nothing to search yet)", + classify("")[0] == TEXT and not looks_like_bytes(""), + ) # -- explicit wins over any guess --------------------------------------- # check("hex: forces bytes", classify("hex: dead") == (BYTES, "dead")) check("bytes: forces bytes too", classify("bytes:dead") == (BYTES, "dead")) check("text: forces text", classify("text: 48 8b c3") == (TEXT, "48 8b c3")) - check("F2's forced mode beats the shape", - classify("dead", forced=BYTES) == (BYTES, "dead") - and classify("48 8b c3", forced=TEXT) == (TEXT, "48 8b c3")) - check("a prefix beats even the forced mode", - classify("text:48 8b c3", forced=BYTES)[0] == TEXT) + check( + "F2's forced mode beats the shape", + classify("dead", forced=BYTES) == (BYTES, "dead") + and classify("48 8b c3", forced=TEXT) == (TEXT, "48 8b c3"), + ) + check( + "a prefix beats even the forced mode", + classify("text:48 8b c3", forced=BYTES)[0] == TEXT, + ) # -- the shapes people paste -------------------------------------------- # check("commas become spaces", normalise_pattern("48,8b,05") == "48 8b 05") - check("a run with no separators is split into bytes", - normalise_pattern("488B05C3") == "48 8B 05 C3") + check( + "a run with no separators is split into bytes", + normalise_pattern("488B05C3") == "48 8B 05 C3", + ) check("whitespace is squeezed", normalise_pattern(" 48 8b\t05 ") == "48 8b 05") - check("a quoted literal keeps its own spacing", - normalise_pattern('"Hello, world", 0') == '"Hello, world", 0') + check( + "a quoted literal keeps its own spacing", + normalise_pattern('"Hello, world", 0') == '"Hello, world", 0', + ) # -- refusing a bad pattern with a reason -------------------------------- # - check("an empty pattern says what to type", - "48 8b" in (pattern_problem("") or "")) - check("a non-hex token is named", - "'zz'" in (pattern_problem("48 zz c3") or ""), pattern_problem("48 zz c3")) - check("a good pattern has no complaint", - pattern_problem("48 8b ?? c3") is None - and pattern_problem('"Hi", 0') is None) - check("an odd-length run is refused rather than silently split", - pattern_problem("488B0") is not None, pattern_problem("488B0")) + check("an empty pattern says what to type", "48 8b" in (pattern_problem("") or "")) + check( + "a non-hex token is named", + "'zz'" in (pattern_problem("48 zz c3") or ""), + pattern_problem("48 zz c3"), + ) + check( + "a good pattern has no complaint", + pattern_problem("48 8b ?? c3") is None and pattern_problem('"Hi", 0') is None, + ) + check( + "an odd-length run is refused rather than silently split", + pattern_problem("488B0") is not None, + pattern_problem("488B0"), + ) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_thumb_ui.py b/tests/test_thumb_ui.py index 1b4c438..d55fa91 100644 --- a/tests/test_thumb_ui.py +++ b/tests/test_thumb_ui.py @@ -22,17 +22,20 @@ import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _fixtures import fast_keys # noqa: E402 from textual.widgets import Static # noqa: E402 -from _fixtures import fast_keys # noqa: E402 from idatui._sync import settle # noqa: E402 -fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys 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") +BIN = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "experiments", + "fibonacci.bin", +) def check(name, ok, detail=""): @@ -45,7 +48,6 @@ def check(name, ok, detail=""): print(f" FAIL {name} {detail}") - #: Every phase gets its OWN copy of the fixture. #: #: This suite used to delete <BIN>.i64 and reopen the SAME path for each phase. @@ -89,19 +91,22 @@ async def wait(pred, pilot, t=240.0): 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. - app = IdaTui(open_path=fresh_copy(BIN, "arm"), keepalive=False, - load_args="-parm") + app = IdaTui(open_path=fresh_copy(BIN, "arm"), 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._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 settle(app) - check("starts undefined at the entry", lst.model.get(lst.cursor).kind == "unknown", - f"{lst.model.get(lst.cursor).text!r}") + 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 @@ -113,9 +118,11 @@ async def run() -> int: # queue doesn't say better. await settle(app) 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}") + 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") @@ -130,23 +137,31 @@ async def run() -> int: 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]) + 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}") + 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") + 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. @@ -164,11 +179,13 @@ async def run() -> int: # 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. - app = IdaTui(open_path=fresh_copy(BIN, "arm64"), keepalive=False, - load_args="-parm") # 64-bit + app = IdaTui( + open_path=fresh_copy(BIN, "arm64"), 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._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() @@ -179,8 +196,11 @@ async def run() -> int: await pilot.press("t") await settle(app, lambda: "64-bit" in status_of(app), timeout=60) status = status_of(app) - check("a 64-bit database warns that Hex-Rays won't decompile", - "64-bit" in status and "decompile" in status, status[:120]) + 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 @@ -193,66 +213,108 @@ async def run() -> int: await pilot.press("p") # The function appearing in the index IS the signal; the model identity # never was one. - await settle(app, lambda: app._func_index is not None - and len(app._func_index) > 0, timeout=60) + await settle( + app, + lambda: app._func_index is not None and len(app._func_index) > 0, + timeout=60, + ) await pilot.press("tab") - await wait(lambda: "cannot decompile" in - str(app.query_one("#status", Static).render()), pilot, 90) + 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]}") + 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 ---------------------- # - app = IdaTui(open_path=fresh_copy(BIN, "armv7a"), keepalive=False, - load_args="-parm:ARMv7-A") + app = IdaTui( + open_path=fresh_copy(BIN, "armv7a"), 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) + 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)}") + 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 settle(app, lambda: type(app.screen).__name__ == "Screen") 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) + 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]}") + 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") + 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: - app = IdaTui(open_path=fresh_copy(vec, "cortexm"), keepalive=False, - load_args="-parm:ARMv7-M") + app = IdaTui( + open_path=fresh_copy(vec, "cortexm"), + 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)}") + 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 settle(app, lambda: type(app.screen).__name__ == "Screen") @@ -264,20 +326,32 @@ async def run() -> int: lst._scroll_cursor_into_view() await settle(app) await pilot.press("T") - await wait(lambda: app._func_index is not None - and len(app._func_index) >= 3, pilot, 90) + 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}") + 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)}") + 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]) + check( + "the result survives the reload AND the reindex", + "3 Thumb entries" in status, + status[:90], + ) drop_scratch() diff --git a/tests/test_trace.py b/tests/test_trace.py index 8803cda..661e350 100644 --- a/tests/test_trace.py +++ b/tests/test_trace.py @@ -31,7 +31,7 @@ LINES = [ "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=0x401000", # loop back: 0x401000 executes twice "rip=0x401001", ] @@ -56,148 +56,220 @@ def main() -> int: 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]) + 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)}") + 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) + 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( + "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()}") + 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()}") + 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()}") + 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()}") + 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) == []) + 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) + 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])}") + 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)] + 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}") + 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( + "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) + 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( + "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") + 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) == []) + 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 diff --git a/tests/test_trace_rpc.py b/tests/test_trace_rpc.py index 015ce8b..9b0f2e0 100644 --- a/tests/test_trace_rpc.py +++ b/tests/test_trace_rpc.py @@ -30,8 +30,7 @@ from idatui.rpcclient import RpcClient, RpcError # noqa: E402 from idatui.trace import Trace # noqa: E402 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -TRACER = os.path.expanduser( - "~/.pi/agent/skills/tenet-trace/scripts/tenet-trace") +TRACER = os.path.expanduser("~/.pi/agent/skills/tenet-trace/scripts/tenet-trace") FALLBACK_TRACE = "/tmp/echotrace.0.log" BINARY = os.path.join(REPO, "targets", "echo") @@ -52,8 +51,12 @@ def make_trace(tmp, binary): """Record a short trace of the echo binary.""" out = os.path.join(tmp, "t") try: - subprocess.run([TRACER, "-o", out, binary, "hello"], - capture_output=True, timeout=180, check=False) + subprocess.run( + [TRACER, "-o", out, binary, "hello"], + capture_output=True, + timeout=180, + check=False, + ) except (OSError, subprocess.TimeoutExpired): return None log = out + ".0.log" @@ -62,11 +65,24 @@ def make_trace(tmp, binary): def spawn_pane(target, trace_log, timeout=300): """Spawn an idatui pane with --trace and wait for readiness.""" - cmd = [sys.executable, "-m", "idatui.pane", "spawn", - "--open", target, "--trace", trace_log, - "--detached", "--size", "60%", "--timeout", str(timeout)] - r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 30, - cwd=REPO) + cmd = [ + sys.executable, + "-m", + "idatui.pane", + "spawn", + "--open", + target, + "--trace", + trace_log, + "--detached", + "--size", + "60%", + "--timeout", + str(timeout), + ] + r = subprocess.run( + cmd, capture_output=True, text=True, timeout=timeout + 30, cwd=REPO + ) if r.returncode != 0: print(f" spawn failed: {r.stderr.strip()}", file=sys.stderr) return None @@ -74,10 +90,17 @@ def spawn_pane(target, trace_log, timeout=300): def stop_pane(sock, timeout=60): - cmd = [sys.executable, "-m", "idatui.pane", "stop", - "--sock", sock, "--timeout", str(timeout)] - subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 10, - cwd=REPO) + cmd = [ + sys.executable, + "-m", + "idatui.pane", + "stop", + "--sock", + sock, + "--timeout", + str(timeout), + ] + subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 10, cwd=REPO) def main() -> int: @@ -96,14 +119,17 @@ def main() -> int: if trace_log is None and os.path.exists(FALLBACK_TRACE): trace_log = FALLBACK_TRACE if trace_log is None: - print(f" skip: no trace available (tracer at {TRACER}, " - f"fallback {FALLBACK_TRACE})") + print( + f" skip: no trace available (tracer at {TRACER}, " + f"fallback {FALLBACK_TRACE})" + ) return 0 # Load our own model for ground-truth comparisons. model = Trace.load(trace_log) - check("model loaded for ground truth", model.length > 10, - f"length={model.length}") + check( + "model loaded for ground truth", model.length > 10, f"length={model.length}" + ) # Spawn the pane. info = spawn_pane(target, trace_log) @@ -138,14 +164,22 @@ def run_trace_tests(c: RpcClient, model: Trace): r = c.call("trace", seek=0) check("trace seek=0 returns a snapshot", "active" in r and "trace" in r) tr = r["trace"] - check("response includes trace metadata", - "idx" in tr and "length" in tr and "pc" in tr and "changed" in tr, - f"keys={list(tr.keys())}") + check( + "response includes trace metadata", + "idx" in tr and "length" in tr and "pc" in tr and "changed" in tr, + f"keys={list(tr.keys())}", + ) check("idx is 0 after seeking to 0", tr["idx"] == 0) - check("length matches the model", tr["length"] == model.length, - f"{tr['length']} vs {model.length}") - check("pc at 0 is a hex string", isinstance(tr["pc"], str) - and tr["pc"].startswith("0x"), tr["pc"]) + check( + "length matches the model", + tr["length"] == model.length, + f"{tr['length']} vs {model.length}", + ) + check( + "pc at 0 is a hex string", + isinstance(tr["pc"], str) and tr["pc"].startswith("0x"), + tr["pc"], + ) # ------------------------------------------------------------------ # # seek to a mid-trace timestamp @@ -153,14 +187,16 @@ def run_trace_tests(c: RpcClient, model: Trace): mid = model.length // 2 r = c.call("trace", seek=mid) tr = r["trace"] - check("seek to midpoint lands correctly", tr["idx"] == mid, - f"got {tr['idx']}, want {mid}") + check( + "seek to midpoint lands correctly", + tr["idx"] == mid, + f"got {tr['idx']}, want {mid}", + ) # The model's IP at this point, rebased — the RPC should agree. # We can't compare directly because the model isn't rebased yet, but # the pc should be a small address (database-space, not ASLR'd). pc = int(tr["pc"], 16) - check("pc is in database space (not ASLR'd)", - pc < 0x100000, f"pc={tr['pc']}") + check("pc is in database space (not ASLR'd)", pc < 0x100000, f"pc={tr['pc']}") # ------------------------------------------------------------------ # # seek by percentage (Tenet shell syntax) @@ -168,39 +204,50 @@ def run_trace_tests(c: RpcClient, model: Trace): r = c.call("trace", seek="!0") check("seek !0 (0%) goes to the start", r["trace"]["idx"] == 0) r = c.call("trace", seek="!100") - check("seek !100 (100%) goes to the end", - r["trace"]["idx"] == model.length - 1, - f"got {r['trace']['idx']}, want {model.length - 1}") + check( + "seek !100 (100%) goes to the end", + r["trace"]["idx"] == model.length - 1, + f"got {r['trace']['idx']}, want {model.length - 1}", + ) r = c.call("trace", seek="!50") - check("seek !50 (50%) goes to the midpoint", - abs(r["trace"]["idx"] - mid) <= 1, - f"got {r['trace']['idx']}, want ~{mid}") + check( + "seek !50 (50%) goes to the midpoint", + abs(r["trace"]["idx"] - mid) <= 1, + f"got {r['trace']['idx']}, want ~{mid}", + ) # ------------------------------------------------------------------ # # step forward/backward # ------------------------------------------------------------------ # c.call("trace", seek=0) r = c.call("trace", step=1) - check("step=1 advances one timestamp", r["trace"]["idx"] == 1, - f"got {r['trace']['idx']}") + check( + "step=1 advances one timestamp", + r["trace"]["idx"] == 1, + f"got {r['trace']['idx']}", + ) r = c.call("trace", step=1) - check("another step=1 reaches 2", r["trace"]["idx"] == 2, - f"got {r['trace']['idx']}") + check( + "another step=1 reaches 2", r["trace"]["idx"] == 2, f"got {r['trace']['idx']}" + ) r = c.call("trace", step=-1) - check("step=-1 goes backward", r["trace"]["idx"] == 1, - f"got {r['trace']['idx']}") + check("step=-1 goes backward", r["trace"]["idx"] == 1, f"got {r['trace']['idx']}") # Step backward at the start should clamp to 0. c.call("trace", seek=0) r = c.call("trace", step=-1) - check("step=-1 at t=0 stays at 0", r["trace"]["idx"] == 0, - f"got {r['trace']['idx']}") + check( + "step=-1 at t=0 stays at 0", r["trace"]["idx"] == 0, f"got {r['trace']['idx']}" + ) # Multi-step. c.call("trace", seek=0) r = c.call("trace", step=5) - check("step=5 advances five timestamps", r["trace"]["idx"] == 5, - f"got {r['trace']['idx']}") + check( + "step=5 advances five timestamps", + r["trace"]["idx"] == 5, + f"got {r['trace']['idx']}", + ) # ------------------------------------------------------------------ # # step over (follows SP) @@ -212,8 +259,14 @@ def run_trace_tests(c: RpcClient, model: Trace): a = model.register(sp_name, i) b = model.register(sp_name, i + 1) if a and b and b < a: - ret = next((j for j in range(i + 1, model.length) - if (model.register(sp_name, j) or 0) >= a), None) + ret = next( + ( + j + for j in range(i + 1, model.length) + if (model.register(sp_name, j) or 0) >= a + ), + None, + ) if ret and ret > i + 3: call_at = (i, ret) break @@ -221,11 +274,12 @@ def run_trace_tests(c: RpcClient, model: Trace): i, ret = call_at c.call("trace", seek=i) r = c.call("trace", step=1, over=True) - check("step over skips the callee", - r["trace"]["idx"] == ret, - f"from {i}, got {r['trace']['idx']}, expected {ret}") - check("step over goes further than a plain step", - r["trace"]["idx"] > i + 1) + check( + "step over skips the callee", + r["trace"]["idx"] == ret, + f"from {i}, got {r['trace']['idx']}, expected {ret}", + ) + check("step over goes further than a plain step", r["trace"]["idx"] > i + 1) else: check("found a call to step over", False, "none in this short trace") @@ -235,25 +289,28 @@ def run_trace_tests(c: RpcClient, model: Trace): r = c.call("trace", goto="main") tr = r["trace"] check("goto main lands on a timestamp", tr["idx"] >= 0) - check("and the function context says main", - r.get("function", {}).get("name") == "main", - f"function={r.get('function')}") + check( + "and the function context says main", + r.get("function", {}).get("name") == "main", + f"function={r.get('function')}", + ) # goto by hex address. main_ea = r["function"]["ea"] c.call("trace", seek=0) # reset position r = c.call("trace", goto=hex(main_ea)) - check("goto by hex address works", - r["trace"]["idx"] >= 0 and r["function"]["ea"] == main_ea, - f"idx={r['trace']['idx']}, ea={r.get('function', {}).get('ea')}") + check( + "goto by hex address works", + r["trace"]["idx"] >= 0 and r["function"]["ea"] == main_ea, + f"idx={r['trace']['idx']}, ea={r.get('function', {}).get('ea')}", + ) # goto a function that was never executed. try: c.call("trace", goto="0xDEADBEEF") check("goto an unexecuted address raises", False, "no error raised") except RpcError as e: - check("goto an unexecuted address raises", "never executed" in str(e), - str(e)) + check("goto an unexecuted address raises", "never executed" in str(e), str(e)) # ------------------------------------------------------------------ # # changed registers in the response @@ -261,9 +318,11 @@ def run_trace_tests(c: RpcClient, model: Trace): c.call("trace", seek=0) r = c.call("trace", step=1) changed = r["trace"]["changed"] - check("changed is a list of register names", - isinstance(changed, list) and all(isinstance(s, str) for s in changed), - f"{changed}") + check( + "changed is a list of register names", + isinstance(changed, list) and all(isinstance(s, str) for s in changed), + f"{changed}", + ) # The PC always changes on a step (it's a different instruction). check("rip is always in changed", "rip" in changed, f"{changed}") @@ -275,9 +334,13 @@ def run_trace_tests(c: RpcClient, model: Trace): r2 = c.call("trace", seek=min(50, model.length - 1)) ea1 = r1.get("cursor", {}).get("ea") ea2 = r2.get("cursor", {}).get("ea") - check("the cursor ea follows the trace pc", - ea1 is not None and ea2 is not None and (ea1 != ea2 or r1["trace"]["pc"] == r2["trace"]["pc"]), - f"ea1={ea1}, ea2={ea2}") + check( + "the cursor ea follows the trace pc", + ea1 is not None + and ea2 is not None + and (ea1 != ea2 or r1["trace"]["pc"] == r2["trace"]["pc"]), + f"ea1={ea1}, ea2={ea2}", + ) # ------------------------------------------------------------------ # # trace verb without a trace raises cleanly @@ -289,8 +352,10 @@ def run_trace_tests(c: RpcClient, model: Trace): # Actually, if none of seek/goto/step is given, it just settles and # returns the current state — that's fine, it's a status query. r = c.call("trace") - check("trace with no action is a status query", - "trace" in r and r["trace"]["idx"] >= 0) + check( + "trace with no action is a status query", + "trace" in r and r["trace"]["idx"] >= 0, + ) except RpcError: check("trace with no action is a status query", False, "raised an error") @@ -299,14 +364,20 @@ def run_trace_tests(c: RpcClient, model: Trace): # ------------------------------------------------------------------ # r = c.call("trace", seek=3) for key in ("idx", "length", "pc", "changed"): - check(f"trace response has '{key}'", key in r.get("trace", {}), - f"trace={r.get('trace')}") + check( + f"trace response has '{key}'", + key in r.get("trace", {}), + f"trace={r.get('trace')}", + ) # Standard snapshot fields are ALSO present (the trace response is a # superset of a normal snapshot). for key in ("active", "function", "cursor", "status", "ready"): - check(f"trace response also has snapshot key '{key}'", key in r, - f"keys={list(r.keys())}") + check( + f"trace response also has snapshot key '{key}'", + key in r, + f"keys={list(r.keys())}", + ) # ------------------------------------------------------------------ # # edge cases: seek beyond bounds @@ -314,25 +385,30 @@ def run_trace_tests(c: RpcClient, model: Trace): r = c.call("trace", seek=-1) check("seek -1 clamps to 0", r["trace"]["idx"] == 0) r = c.call("trace", seek=model.length + 1000) - check("seek beyond length clamps to the end", - r["trace"]["idx"] == model.length - 1, - f"got {r['trace']['idx']}") + check( + "seek beyond length clamps to the end", + r["trace"]["idx"] == model.length - 1, + f"got {r['trace']['idx']}", + ) # ------------------------------------------------------------------ # # seek with comma-separated numbers (ergonomic) # ------------------------------------------------------------------ # r = c.call("trace", seek="100") - check("seek accepts a string number", - r["trace"]["idx"] == min(100, model.length - 1)) + check( + "seek accepts a string number", r["trace"]["idx"] == min(100, model.length - 1) + ) # ------------------------------------------------------------------ # # pseudocode still works with a trace loaded # ------------------------------------------------------------------ # c.call("trace", goto="main") r = c.call("pseudocode", target="main", lines=5) - check("pseudocode works alongside the trace", - "code" in r and "main" in r.get("code", ""), - f"keys={list(r.keys())}") + check( + "pseudocode works alongside the trace", + "code" in r and "main" in r.get("code", ""), + f"keys={list(r.keys())}", + ) # ------------------------------------------------------------------ # # state includes trace position @@ -341,17 +417,18 @@ def run_trace_tests(c: RpcClient, model: Trace): r = c.call("state") # The state verb doesn't include trace info (that's trace-specific), # but the standard snapshot fields should be consistent. - check("state works with a trace loaded", - r.get("ready") is True and "cursor" in r) + check("state works with a trace loaded", r.get("ready") is True and "cursor" in r) # ------------------------------------------------------------------ # # view_lines works with trail painted # ------------------------------------------------------------------ # c.call("trace", seek=min(40, model.length - 1)) r = c.call("view", lines=10) - check("view returns lines with a trace active", - "lines" in r and len(r["lines"]) > 0, - f"keys={list(r.keys())}") + check( + "view returns lines with a trace active", + "lines" in r and len(r["lines"]) > 0, + f"keys={list(r.keys())}", + ) # ------------------------------------------------------------------ # # navigation works alongside trace: goto a function, trace follows @@ -359,14 +436,18 @@ def run_trace_tests(c: RpcClient, model: Trace): c.call("trace", goto="main") start_idx = c.call("trace")["trace"]["idx"] r = c.call("goto", target="error_at_line") - check("goto still works with trace loaded", - r.get("function", {}).get("name") == "error_at_line") + check( + "goto still works with trace loaded", + r.get("function", {}).get("name") == "error_at_line", + ) # The trace timestamp should NOT change from a regular goto — the trace # position is independent of navigation. r2 = c.call("trace") - check("regular goto does not change the trace position", - r2["trace"]["idx"] == start_idx, - f"was {start_idx}, now {r2['trace']['idx']}") + check( + "regular goto does not change the trace position", + r2["trace"]["idx"] == start_idx, + f"was {start_idx}, now {r2['trace']['idx']}", + ) if __name__ == "__main__": diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py index 586ce2e..768d18b 100644 --- a/tests/test_trace_ui.py +++ b/tests/test_trace_ui.py @@ -17,18 +17,21 @@ import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from textual.widgets import Input, OptionList, Static # noqa: E402 - from _fixtures import fast_keys, staged # noqa: E402 +from textual.widgets import Input, OptionList, Static # noqa: E402 -fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui._sync import settle # noqa: E402 -from idatui.app import (DecompView, IdaTui, ListingView, # noqa: E402 - RegWriteScreen, TraceDock) +from idatui.app import ( # noqa: E402 + DecompView, + IdaTui, + ListingView, + 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") +TRACER = os.path.expanduser("~/.pi/agent/skills/tenet-trace/scripts/tenet-trace") PASS = FAIL = 0 @@ -56,8 +59,12 @@ async def wait(pred, pilot, t=240.0): 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) + subprocess.run( + [TRACER, "-o", out, binary, "hi"], + capture_output=True, + timeout=180, + check=False, + ) except (OSError, subprocess.TimeoutExpired): return None log = out + ".0.log" @@ -71,8 +78,9 @@ async def run() -> int: # anything tracked. On echo the seeding saves only ~0.2s (it analyses fast); # it is here so a suite pointed at a bigger target doesn't pay for analysis # on every run. - async with staged(binary, lambda p: IdaTui(open_path=p, keepalive=False), - prefix="idatui-traceui-") as target: + async with staged( + binary, lambda p: IdaTui(open_path=p, keepalive=False), prefix="idatui-traceui-" + ) as target: tmp = os.path.dirname(target) log = make_trace(tmp, target) if not log: @@ -90,22 +98,29 @@ async def run() -> int: # 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}") + 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]}") + 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]) + 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]) + check( + "and the register state at that time", "rip" in regs.lower(), regs[:60] + ) # -- stepping --------------------------------------------------- # lst = app.query_one(ListingView) @@ -117,15 +132,19 @@ async def run() -> int: # runs in a worker. Waiting on it and then reading the cursor was a # race that the (slower) IDA Nexus backend loses. Gate on the thing # the check is about. - await settle(app, lambda: app._t == 1 and lst._cursor_ea() == t.ip(1), - timeout=20) + await settle( + app, lambda: app._t == 1 and lst._cursor_ea() == t.ip(1), timeout=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)}") + 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 settle(app, lambda: app._t == 0 and lst._cursor_ea() == t.ip(0), - timeout=20) + await settle( + app, lambda: app._t == 0 and lst._cursor_ea() == t.ip(0), timeout=20 + ) check("[ steps backward", app._t == 0, f"t={app._t}") await pilot.press("[") # Nothing should happen, so there is no signal to wait FOR: the @@ -144,8 +163,14 @@ async def run() -> int: 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) + 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 @@ -157,8 +182,11 @@ async def run() -> int: 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( + "} 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 -------------------------------------------- # @@ -168,36 +196,56 @@ async def run() -> int: # image would have nothing to show. dock = app.query_one(TraceDock) app._seek(min(60, t.length - 1)) - await wait(lambda: "stack (" in - str(dock.query_one("#trace-stack", Static).render()), - pilot, 5) + await wait( + lambda: ( + "stack (" in str(dock.query_one("#trace-stack", Static).render()) + ), + pilot, + 5, + ) 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]) + 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]}") + 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]) + 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]) + 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 wait(lambda: str(dock.query_one("#trace-stack", - Static).render()) != before, - pilot, 5) - check("and it follows as you move through time", - str(dock.query_one("#trace-stack", Static).render()) != before) + await wait( + lambda: str(dock.query_one("#trace-stack", Static).render()) != before, + pilot, + 5, + ) + 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 @@ -207,17 +255,27 @@ async def run() -> int: await wait(lambda: bool(lst.trail), pilot, 5) 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( + "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 -------------------------------- # @@ -234,23 +292,33 @@ async def run() -> int: 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) + 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 wait(lambda: len(dec.trail) > 2, pilot, 5) - check("pseudocode lines are painted with the trail", - len(dec.trail) > 2, f"{len(dec.trail)} lines") + 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}") + 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]}") + 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 @@ -260,13 +328,19 @@ async def run() -> int: was = app._t await pilot.press("]") await settle(app, lambda: app._t != was) - check("stepping in pseudocode stays in pseudocode", - app._active == "decomp", f"active={app._active}") + check( + "stepping in pseudocode stays in pseudocode", + app._active == "decomp", + f"active={app._active}", + ) was = app._t await pilot.press("[") await settle(app, lambda: app._t != was) - check("and so does stepping backward", - app._active == "decomp", f"active={app._active}") + 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, @@ -274,8 +348,13 @@ async def run() -> int: # navigation though: both panes show the same instant, so the # listing cursor must sit on the current instruction. app.action_toggle_split() - await wait(lambda: app._split and lst.display - and app.query_one(DecompView).display, pilot, 10) + await wait( + lambda: ( + app._split and lst.display and app.query_one(DecompView).display + ), + pilot, + 10, + ) if not app._split: check("split view toggled on", False) else: @@ -286,14 +365,18 @@ async def run() -> int: # Wait for the cursor to arrive rather than sleeping a flat # 0.5s and hoping. Same question -- does the listing follow # the pc? -- but it costs what it costs instead of 3s. - if await wait(lambda: lst._cursor_ea() == t.ip(app._t), - pilot, 5): + if await wait(lambda: lst._cursor_ea() == t.ip(app._t), pilot, 5): 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))}") + 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 @@ -313,8 +396,14 @@ async def run() -> int: # on `pc in _trail_line_of` instead would burn the timeout on # every unmapped instruction -- about half of them -- and be # slower than the flat sleep it replaces. - await wait(lambda: lst._cursor_ea() == pc - and app._trail_map_ea == dec.loaded_ea, pilot, 5) + await wait( + lambda: ( + lst._cursor_ea() == pc + and app._trail_map_ea == dec.loaded_ea + ), + pilot, + 5, + ) if app._trail_map_ea == dec.loaded_ea and pc in app._trail_line_of: mapped += 1 # Mapped: the pseudocode cursor is expected, so it's fair @@ -324,9 +413,11 @@ async def run() -> int: await wait(lambda: dec.cursor == line, pilot, 3) if dec.cursor != line: missed += 1 - check("the pseudocode cursor follows every mapped instruction", - mapped > 3 and missed == 0, - f"{mapped} mapped, {missed} not followed") + check( + "the pseudocode cursor follows every mapped instruction", + mapped > 3 and missed == 0, + f"{mapped} mapped, {missed} not followed", + ) # -- a late navigation must not drag the view back --------------- # # Navigations run in workers and finish out of order. The trace's @@ -347,10 +438,12 @@ async def run() -> int: # for the workers to drain rather than for three seconds and a # hope: same question, ~50ms instead of 3s. await settle(app) - check("a stale navigation doesn't drag the cursor away", - lst._cursor_ea() == dbaddr and app._cur.ea == dbaddr, - f"cursor={lst._cursor_ea():#x} cur={app._cur.ea:#x} " - f"want {dbaddr:#x}") + check( + "a stale navigation doesn't drag the cursor away", + lst._cursor_ea() == dbaddr and app._cur.ea == dbaddr, + f"cursor={lst._cursor_ea():#x} cur={app._cur.ea:#x} " + f"want {dbaddr:#x}", + ) # -- seeking, as opposed to stepping ---------------------------- # # "When else did this instruction run?" — the question that makes a @@ -359,8 +452,11 @@ async def run() -> int: 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") + check( + "found an address executed more than once", + False, + f"{len(stamps)} executions", + ) else: if app._split: app.action_toggle_split() @@ -377,31 +473,48 @@ async def run() -> int: lst.cursor = row lst._scroll_cursor_into_view() await settle(app, lambda: lst._cursor_ea() == db) - check("cursor is on the repeated instruction", - lst._cursor_ea() == db, f"{lst._cursor_ea():#x} vs {db:#x}") + check( + "cursor is on the repeated instruction", + lst._cursor_ea() == db, + f"{lst._cursor_ea():#x} vs {db:#x}", + ) await pilot.press(">") await settle(app, lambda: app._t == stamps[1]) - check("> seeks to the next execution of it", - app._t == stamps[1], f"t={app._t}, expected {stamps[1]}") + 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]) + check( + "and says which execution this is", + f"2 of {len(stamps)}" in status, + status[:80], + ) lst.cursor = row await settle(app) await pilot.press("<") await settle(app, lambda: app._t == stamps[0]) - check("< seeks back to the previous one", - app._t == stamps[0], f"t={app._t}, expected {stamps[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 settle(app) await pilot.press("<") - await settle(app, lambda: "first" in str( - app.query_one("#status", Static).render())) + await settle( + app, + lambda: "first" in str(app.query_one("#status", Static).render()), + ) 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]) + 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?" ---------------------- # want_t = min(200, t.length - 1) @@ -409,14 +522,24 @@ async def run() -> int: await settle(app, lambda: app._t == want_t) 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__}") + 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) + 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") @@ -426,13 +549,18 @@ async def run() -> int: await settle(app) 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}") + 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))}") + 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 diff --git a/tests/test_trace_vs_tenet.py b/tests/test_trace_vs_tenet.py index 30d74b0..c2a0ea2 100644 --- a/tests/test_trace_vs_tenet.py +++ b/tests/test_trace_vs_tenet.py @@ -46,6 +46,7 @@ def _reference(): 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 @@ -57,6 +58,7 @@ def _reference(): # stay in raw trace addresses, which is what we want to compare. def get_instruction_addresses(self): return [0xDEAD0000] + return TraceReader, ArchAMD64, FakeDctx @@ -88,9 +90,11 @@ def compare(path, ref_cls, arch, dctx, samples=200): 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}") + 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 @@ -99,8 +103,11 @@ def compare(path, ref_cls, arch, dctx, samples=200): 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]}") + 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 @@ -121,11 +128,16 @@ def compare(path, ref_cls, arch, dctx, samples=200): 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]}") + 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]})") + 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] @@ -135,8 +147,11 @@ def compare(path, ref_cls, arch, dctx, samples=200): 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]}") + 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. @@ -153,11 +168,14 @@ def compare(path, ref_cls, arch, dctx, samples=200): # 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]: + 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]}") + 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] @@ -167,18 +185,24 @@ def compare(path, ref_cls, arch, dctx, samples=200): 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)") + 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)] + 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 |