diff options
| -rw-r--r-- | README.md | 20 | ||||
| -rw-r--r-- | idatui/app.py | 211 | ||||
| -rw-r--r-- | idatui/launch.py | 6 | ||||
| -rw-r--r-- | tests/test_trace_ui.py | 148 |
4 files changed, 383 insertions, 2 deletions
@@ -112,6 +112,26 @@ multi-image firmware wants. They apply to the first open only — after that the > reopen. Delete those stale files (never the `.i64`) and retry — `ida-tui` does > this automatically. +## Execution traces + +Load a [Tenet](https://github.com/gaasedelen/tenet) trace alongside the binary +and explore it in time: + +```sh +./ida-tui /path/to/binary --trace trace.0.log +``` + +A docked pane on the right shows the registers at the current timestamp (the +ones the current instruction wrote are highlighted) and a timeline. `]` and `[` +step one instruction forward and back; `}` and `{` step over a call by following +the stack pointer. The code view follows. + +Trace addresses are rebased onto the database automatically — a traced process +is relocated, so nothing lines up until that's solved. + +Traces are recorded separately; see `~/dev/tenet/tenet-original/tracers/` for the +QEMU tracer. + ## RPC / driving the TUI Give the TUI `--rpc <sock>` to expose a unix-socket control channel, then drive diff --git a/idatui/app.py b/idatui/app.py index ea68620..59e3f26 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -2286,6 +2286,101 @@ class HelpScreen(ModalScreen): self.dismiss(None) +class TraceDock(Vertical): + """Registers and a timeline for the loaded execution trace, docked right. + + Persistent rather than a modal: a trace turns every other view into "state + at time T", so the time and the registers are context you read WHILE looking + at code, not something you open and dismiss. + """ + + def __init__(self) -> None: + super().__init__(id="trace-dock") + self.trace = None + self.idx = 0 + + def compose(self) -> ComposeResult: + yield Static("", id="trace-head", markup=False) + yield Static("", id="trace-regs", markup=False) + yield TraceTimeline(id="trace-timeline") + + def show(self, trace, idx: int) -> None: + self.trace = trace + self.idx = idx + tl = self.query_one(TraceTimeline) + tl.trace, tl.idx = trace, idx + self.refresh_state() + + def refresh_state(self) -> None: + t = self.trace + if t is None: + return + n = max(t.length, 1) + pct = (self.idx + 1) * 100.0 / n + head = Text() + head.append(f" {self.idx:,}", _S_MNEM) + head.append(f" / {t.length - 1:,} ", _S_DIM) + head.append(f"{pct:5.1f}%\n", _S_ADDR) + # Register values are machine state and stay as the trace recorded them, + # but everything else on screen is in database addresses. Showing both + # here explains the relationship once, where it's read, instead of + # leaving "pc 0x2aed" next to "rip 0x7ffff6faaaed" to be puzzled over. + head.append(f" pc {t.ip(self.idx):#x}", _S_LABEL) + if t.slide: + head.append(f" (trace {t.raw_ip(self.idx):#x})", _S_DIM) + self.query_one("#trace-head", Static).update(head) + + # Registers, with the ones THIS instruction wrote called out: that + # difference is the entire reason a delta trace is readable. + changed = t.changed(self.idx) + body = Text() + pc = t.pc_name + for name in t.registers: + v = t.register(name, self.idx) + if v is None: + continue + hot = name in changed + body.append(f" {name:>4} ", _S_MNEM if hot else _S_DIM) + body.append(f"{v:#018x}\n" if v > 0xFFFFFFFF else f"{v:#010x}\n", + _S_DATA if hot else (_S_LABEL if name == pc else _S_INSN)) + self.query_one("#trace-regs", Static).update(body) + tl = self.query_one(TraceTimeline) + tl.idx = self.idx + tl.refresh() + + +class TraceTimeline(Static): + """The trace as a vertical bar: where you are, and where you've been. + + Tenet's timeline is a Qt widget you scroll and drag to zoom. A terminal + column can't do that, but it can do the part that matters — show the shape + of the trace and your position in it — with one row per N timestamps. + """ + + def __init__(self, **kw) -> None: + super().__init__("", **kw) + self.trace = None + self.idx = 0 + + def render(self) -> Text: + t = self.trace + out = Text() + h = max(self.size.height - 1, 1) + if t is None or not t.length: + return out + out.append(" timeline\n", _S_DIM) + h = max(h - 1, 1) + per = max(t.length / h, 1.0) + here = int(self.idx / per) + for row in range(h): + if row == here: + out.append(" \u25b6", _S_MNEM) + out.append(f" {int(row * per):>10,}\n", _S_ADDR) + else: + out.append(" \u2502\n", _S_SEP if row % 5 else _S_ADDR) + return out + + class LoadOptionsScreen(ModalScreen): """Ask how to load a file no loader recognised. @@ -3026,6 +3121,10 @@ class IdaTui(App): #pal-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } #pal-input { border: none; height: 1; margin: 0 1; background: $panel; color: $text; } #pal-list { height: auto; max-height: 24; } + #trace-dock { dock: right; width: 34; background: $surface; border-left: solid $panel; } + #trace-head { height: 2; padding: 0 1; background: $panel; } + #trace-regs { height: auto; padding: 1 0 0 0; } + #trace-timeline { height: 1fr; padding: 1 0 0 0; } #load-note { height: 2; padding: 1 1 0 1; color: $text-muted; } /* Cap the processor list so the ADDRESS FIELD is always on screen: with the palette default (24) the box outgrew the terminal and the field you need @@ -3071,6 +3170,12 @@ class IdaTui(App): Binding("quotation_mark,shift+f12", "strings", "Strings", show=False), Binding("ctrl+o", "switch_binary", "Binaries", show=False), Binding("ctrl+l", "load_options", "Reload as…", show=False), + # Trace stepping. ] / [ move one instruction, } / { step over a + # call by following the stack pointer. + Binding("right_square_bracket", "step_fwd", "Step", show=False), + Binding("left_square_bracket", "step_back", "Step back", show=False), + Binding("right_curly_bracket", "step_over_fwd", "Step over", show=False), + Binding("left_curly_bracket", "step_over_back", "Step over back", show=False), Binding("f1", "help", "Keys", show=False), Binding("g", "goto", "Goto"), Binding("slash", "filter", "Filter", show=False), @@ -3082,7 +3187,7 @@ class IdaTui(App): def __init__(self, open_path: str | None = None, keepalive: bool = True, rpc_path: str | None = None, ttl: int = 1800, - project=None, load_args: str = "") -> None: + project=None, load_args: str = "", trace_path: str = "") -> None: super().__init__() # Project mode is additive: with no project this is the plain # single-binary app, unchanged. @@ -3114,6 +3219,9 @@ class IdaTui(App): self._open_path = open_path self._ttl = ttl self._load_args = load_args or "" # IDA switches for a headerless blob + self._trace_path = trace_path or "" # Tenet execution trace to explore + self._trace = None # the loaded Trace, once analysed + self._t = 0 # current timestamp in that trace self._do_keepalive = keepalive self._rpc_path = rpc_path self._rpc = None @@ -3168,6 +3276,11 @@ class IdaTui(App): hx = HexView() hx.display = False yield hx + # Docked right and only shown once a trace is loaded, so a normal + # session looks exactly as it did. + td = TraceDock() + td.display = False + yield td si = Input(id="search") si.display = False si.can_focus = False @@ -3627,6 +3740,8 @@ class IdaTui(App): # otherwise pop the fuzzy symbol picker. self.app.call_from_thread(self._auto_land) self._index_binary() # project mode: keep the cross-binary index fresh + if self._trace_path and self._trace is None: + self._load_trace() # needs the index above: rebasing reads it @work(thread=True, exclusive=True, group="prewarm") def _prewarm_provider(self) -> None: @@ -5299,6 +5414,100 @@ class IdaTui(App): # this, `p` gave you a function the rest of the app couldn't see. self._reindex_functions() + def _load_trace(self) -> None: + """Parse the trace and line it up with the database. + + Runs after the function index exists: rebasing needs the database's + addresses, and without it nothing in the trace matches anything on + screen (our echo trace runs at 0x7ffff6faa000; the database has that + code at 0x2000). + """ + from .trace import Trace + path = self._trace_path + try: + def note(n): + self.app.call_from_thread( + self._status, f"trace: {n:,} instructions\u2026") + trace = Trace.load(path, progress=note) + except OSError as e: + self.app.call_from_thread(self._status, f"trace: {e}") + return + if not trace.length: + self.app.call_from_thread( + self._status, f"trace: {os.path.basename(path)} is empty") + return + idx = self._func_index + addrs = [f.addr for f in idx.all_loaded()] if idx is not None else [] + slide = trace.rebase(addrs) + trace.apply_slide(slide) + hit = sum(1 for f in (idx.all_loaded() if idx else []) if trace.executions(f.addr)) + self.app.call_from_thread(self._trace_ready, trace, slide, hit) + + def _trace_ready(self, trace, slide: int, hit: int) -> None: + self._trace = trace + self._t = 0 + dock = self.query_one(TraceDock) + dock.display = True + dock.show(trace, 0) + where = (f"rebased {slide:+#x}" if slide else "no rebase needed") + self._status(f"trace: {trace.length:,} instructions, {hit} functions " + f"touched ({where})", priority=True) + self._seek(0, follow=True) + + # -- trace navigation --------------------------------------------------- # + def _seek(self, idx: int, follow: bool = True) -> None: + """Move to timestamp ``idx``; ``follow`` takes the code view with it.""" + t = self._trace + if t is None or not t.length: + return + self._t = max(0, min(int(idx), t.length - 1)) + self.query_one(TraceDock).show(t, self._t) + if follow: + self._goto_ea(t.ip(self._t), push=False) + + def _step(self, delta: int) -> None: + if self._trace is None: + self._status("no trace loaded (--trace FILE)") + return + self._seek(self._t + delta) + + def action_step_fwd(self) -> None: + self._step(1) + + def action_step_back(self) -> None: + self._step(-1) + + def _step_over(self, direction: int) -> None: + """Step over a call by following the stack pointer. + + A call pushes, so the callee runs with SP BELOW where we started; + stepping until SP comes back up lands after the call returns. Cheaper + and more robust than recognising call instructions per architecture, + which is what the mode makes it: if this instruction doesn't call + anything, SP is already >= the start and it degenerates to one step. + """ + t = self._trace + if t is None: + self._status("no trace loaded (--trace FILE)") + return + sp_name = "rsp" if "rsp" in t.reg_at else ("esp" if "esp" in t.reg_at else "sp") + sp0 = t.register(sp_name, self._t) + i = self._t + direction + limit = 200000 # a runaway search must not hang the UI + while 0 <= i < t.length and limit > 0: + sp = t.register(sp_name, i) + if sp0 is None or sp is None or sp >= sp0: + break + i += direction + limit -= 1 + self._seek(max(0, min(i, t.length - 1))) + + def action_step_over_fwd(self) -> None: + self._step_over(1) + + def action_step_over_back(self) -> None: + self._step_over(-1) + @work(thread=True, exclusive=True, group="load-funcs") def _reindex_functions(self) -> None: """Rebuild the function index in place after an edit changed it. diff --git a/idatui/launch.py b/idatui/launch.py index f51e5af..64e1546 100644 --- a/idatui/launch.py +++ b/idatui/launch.py @@ -64,6 +64,8 @@ def main(argv: list[str] | None = None) -> int: help="do not run the keepalive heartbeat") p.add_argument("--rpc", metavar="PATH", help="listen for RPC on this unix socket (puppeteer the TUI)") + p.add_argument("--trace", metavar="FILE", + help="Tenet execution trace to explore alongside the binary") g = p.add_argument_group( "loading a headerless blob", "An ELF/PE/Mach-O says what it is. A raw firmware dump doesn't, and IDA " @@ -157,7 +159,9 @@ def main(argv: list[str] | None = None) -> int: rpc_path = os.path.abspath(os.path.expanduser(args.rpc)) if args.rpc else None IdaTui(open_path=binary, keepalive=not args.no_keepalive, rpc_path=rpc_path, ttl=args.ttl, project=project, - load_args=_load_args(load)).run() + load_args=_load_args(load), + trace_path=(os.path.abspath(os.path.expanduser(args.trace)) + if args.trace else "")).run() return 0 diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py new file mode 100644 index 0000000..a30362f --- /dev/null +++ b/tests/test_trace_ui.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +"""The trace dock: registers, timeline, and stepping through time. + +Needs IDA and a trace. Generates its own trace with the QEMU tracer if the +tracer is built; skips with a message otherwise, since neither the emulator nor +the trace is part of this repo. +""" +import asyncio +import os +import shutil +import subprocess +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from textual.widgets import Static # noqa: E402 + +from idatui.app import IdaTui, ListingView, TraceDock # 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") +PASS = FAIL = 0 + + +def check(name, ok, detail=""): + global PASS, FAIL + if ok: + PASS += 1 + print(f" ok {name}") + else: + FAIL += 1 + print(f" FAIL {name} {detail}") + + +async def wait(pred, pilot, t=240.0): + for _ in range(int(t / 0.05)): + await pilot.pause(0.05) + try: + if pred(): + return True + except Exception: # noqa: BLE001 + pass + return False + + +def make_trace(tmp, binary): + out = os.path.join(tmp, "t") + try: + subprocess.run([TRACER, "-o", out, binary, "hi"], + capture_output=True, timeout=180, check=False) + except (OSError, subprocess.TimeoutExpired): + return None + log = out + ".0.log" + return log if os.path.exists(log) else None + + +async def run() -> int: + binary = os.path.join(REPO, "targets", "echo") + with tempfile.TemporaryDirectory() as tmp: + # Scratch copy: opening a binary writes a database beside it, and the + # suite must not edit anything tracked. + target = os.path.join(tmp, "echo") + shutil.copy2(binary, target) + log = make_trace(tmp, target) + if not log: + print(f" skip: could not record a trace (tracer at {TRACER})") + return 0 + + app = IdaTui(open_path=target, keepalive=False, trace_path=log) + async with app.run_test(size=(160, 44)) as pilot: + ok = await wait(lambda: app._trace is not None, pilot, 300) + check("the trace loads alongside the binary", ok) + if not ok: + return 1 + t = app._trace + check("it has instructions", t.length > 10, f"{t.length}") + + # Rebasing: the tracer runs the binary relocated, so without a slide + # nothing in the trace matches anything on screen. + check("trace addresses were rebased onto the database", + t.slide != 0 and t.ip(0) < 0x1000000, + f"slide={t.slide:#x} ip0={t.ip(0):#x}") + idx = app._func_index + touched = [f.name for f in idx.all_loaded() if t.executions(f.addr)] + check("and now line up with real functions", + len(touched) > 1 and "main" in touched, f"{touched[:6]}") + + dock = app.query_one(TraceDock) + check("the dock is docked and visible", dock.display) + head = str(dock.query_one("#trace-head", Static).render()) + check("it shows where we are in time", "0" in head and "%" in head, + head[:60]) + regs = str(dock.query_one("#trace-regs", Static).render()) + check("and the register state at that time", "rip" in regs.lower(), + regs[:60]) + + # -- stepping --------------------------------------------------- # + lst = app.query_one(ListingView) + lst.focus() + await pilot.pause(0.4) + await pilot.press("]") + await wait(lambda: app._t == 1, pilot, 20) + check("] steps forward one instruction", app._t == 1, f"t={app._t}") + check("the code view follows the trace", + lst._cursor_ea() == t.ip(1), + f"{lst._cursor_ea()} vs {t.ip(1)}") + await pilot.press("[") + await wait(lambda: app._t == 0, pilot, 20) + check("[ steps backward", app._t == 0, f"t={app._t}") + await pilot.press("[") + await pilot.pause(0.4) + check("and stops at the start of the trace", app._t == 0) + + # -- step over -------------------------------------------------- # + # A call pushes, so the callee runs with SP below where we started; + # stepping until SP comes back up lands after the return. Find a + # real call in this trace rather than assuming one is at a fixed + # place. + sp = "rsp" if "rsp" in t.reg_at else "esp" + call_at = None + for i in range(1, min(t.length - 1, 400)): + a, b = t.register(sp, i), t.register(sp, i + 1) + if a and b and b < a: + ret = next((j for j in range(i + 1, t.length) + if (t.register(sp, j) or 0) >= a), None) + if ret and ret > i + 3: + call_at = (i, ret) + break + if call_at is None: + check("found a call to step over", False, "none in this trace") + else: + i, ret = call_at + app._seek(i) + await wait(lambda: app._t == i, pilot, 20) + await pilot.press("}") + await wait(lambda: app._t != i, pilot, 30) + check("} steps OVER a call instead of into it", + app._t == ret, f"{i} -> {app._t}, expected {ret}") + check("which is further than a plain step", app._t > i + 1) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(run())) |