diff options
Diffstat (limited to 'tests/test_domain.py')
| -rw-r--r-- | tests/test_domain.py | 155 |
1 files changed, 0 insertions, 155 deletions
diff --git a/tests/test_domain.py b/tests/test_domain.py deleted file mode 100644 index 6875b66..0000000 --- a/tests/test_domain.py +++ /dev/null @@ -1,155 +0,0 @@ -#!/usr/bin/env python3 -"""Validate the domain/paging layer against a real, large binary. - -Run with a session open on a big binary (e.g. libcrypto.so.3, 10k funcs): - - python3 tests/test_domain.py --db <session_id> - -Checks: correct full pagination (advance by len, not next_offset), viewport -slicing across block boundaries, window caching (revisit is instant), prefetch -warming, cached instruction totals, decompile success + hard-failure handling, -and address resolution. -""" -import os -import sys -import time - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.client import IDAClient # noqa: E402 -from idatui.domain import DISASM_BLOCK, LIST_PAGE, Program # noqa: E402 - -URL = "http://127.0.0.1:8745/mcp" -PASS = FAIL = 0 - - -def check(name, cond, detail=""): - global PASS, FAIL - if cond: - PASS += 1 - print(f" ok {name}") - else: - FAIL += 1 - print(f" FAIL {name} {detail}") - - -def ms(fn): - t = time.time() - r = fn() - return (time.time() - t) * 1e3, r - - -def main(argv): - db = None - it = iter(argv) - for a in it: - if a == "--db": - db = next(it) - c = IDAClient(URL, db=db) - c.connect() - if db is None: - c.set_db(c.resolve_db()) - prog = Program(c) - module = c.health().get("module") - total_funcs = c.call("survey_binary").get("statistics", {}).get("total_functions") - print(f"db={c.db} module={module} survey_total_functions={total_funcs}") - - # ---- function index: full enumeration correctness ------------------ # - print("\n[function index]") - idx = prog.functions() - dt, _ = ms(lambda: idx.load_all()) - n = len(idx) - check("enumerated all functions (matches survey)", n == total_funcs, - f"got {n} vs survey {total_funcs}") - print(f" loaded {n} funcs in {dt:.0f}ms ({dt / max(n,1):.3f}ms/func), " - f"page={LIST_PAGE}") - # no duplicates, monotonic-ish uniqueness by addr - addrs = [idx.get(i).addr for i in range(min(n, 3000))] - check("no duplicate addrs in first 3000", len(addrs) == len(set(addrs))) - # viewport slice (adapt to binary size) - wstart = min(1000, max(n - 50, 0)) - wlen = min(50, n - wstart) - w = idx.window(wstart, wlen) - check(f"window({wstart},{wlen}) returns {wlen}", len(w) == wlen, str(len(w))) - - # ---- filtered index uses server-side glob -------------------------- # - print("\n[filtered index]") - sub = prog.functions(filter="sub_*") - sub.ensure(10) - check("filter sub_* yields sub_ names", all(f.name.startswith("sub_") for f in sub.window(0, 10)), - str([f.name for f in sub.window(0, 5)])) - - # ---- pick the fattest function for disasm stress ------------------- # - print("\n[disasm windowing]") - fattest = max((idx.get(i) for i in range(n)), key=lambda f: f.size) - dm = prog.disasm(fattest.addr, fattest.name) - dt, total = ms(dm.total) - check("total() returns positive count", total > 0, str(total)) - dt2, total2 = ms(dm.total) - check("total() cached (2nd call ~instant)", dt2 < dt / 2 + 1, f"{dt:.0f}ms -> {dt2:.1f}ms") - print(f" fattest {fattest.name}: {total} insns, total() {dt:.0f}ms then {dt2:.1f}ms") - - # viewport across a block boundary - start = DISASM_BLOCK - 5 - win = dm.lines(start, 60, prefetch=False) - check("viewport spans block boundary, right length", - len(win) == min(60, max(total - start, 0)), f"got {len(win)}") - # addresses strictly increasing and contiguous slice - eas = [ln.ea for ln in win] - check("viewport addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])), - str(eas[:4])) - - # deep window: first slow (O(offset)), revisit instant (cached) - if total > DISASM_BLOCK * 4: - deep = (total // DISASM_BLOCK - 1) * DISASM_BLOCK - dt_cold, a = ms(lambda: dm.lines(deep, 60, prefetch=False)) - dt_warm, b = ms(lambda: dm.lines(deep, 60, prefetch=False)) - check("deep window revisit is cached/instant", dt_warm < dt_cold / 2 + 1, - f"cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms") - check("cached window identical", [l.ea for l in a] == [l.ea for l in b]) - print(f" deep@{deep}: cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms") - - # prefetch warms the next block - print("\n[prefetch]") - dm2 = prog.disasm(fattest.addr + 0) # same model (cached by ea) - fresh = prog.disasm(idx.get(0).addr, idx.get(0).name) - fresh.lines(0, 60, prefetch=True) # should prefetch block 1 - time.sleep(0.3) - check("prefetch warmed a neighbor block", fresh.cached_blocks() >= 2, - f"cached_blocks={fresh.cached_blocks()}") - - # ---- decompile: success + hard-failure ----------------------------- # - print("\n[decompile]") - # a small function likely decompiles - small = min((idx.get(i) for i in range(n)), key=lambda f: f.size if f.size > 4 else 1 << 30) - d_small = prog.decompile(small.addr) - check("small func decompiles or fails cleanly", isinstance(d_small.failed, bool)) - dt_c, _ = ms(lambda: prog.decompile(small.addr)) - check("decompile cached (2nd ~instant)", dt_c < 5, f"{dt_c:.1f}ms") - # the monster should hard-fail as a soft error (not raise) - d_big = prog.decompile(fattest.addr) - check("monster decompile handled (failed flag, no raise)", - d_big.failed or d_big.code is not None, - f"failed={d_big.failed} err={d_big.error}") - print(f" small {small.name}: failed={d_small.failed} " - f"trunc={d_small.truncated} chars={d_small.total_chars}") - print(f" monster {fattest.name}: failed={d_big.failed} err={d_big.error}") - - # ---- resolve ------------------------------------------------------- # - print("\n[resolve]") - check("resolve hex", prog.resolve(hex(fattest.addr)) == fattest.addr) - check("resolve int passthrough", prog.resolve(fattest.addr) == fattest.addr) - named = next((idx.get(i) for i in range(n) if not idx.get(i).name.startswith("sub_")), None) - if named: - try: - r = prog.resolve(named.name) - check("resolve symbol name", r == named.addr, f"{hex(r)} vs {hex(named.addr)} ({named.name})") - except KeyError as e: - check("resolve symbol name", False, str(e)) - - prog.close() - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) |
