diff options
Diffstat (limited to 'idatui')
| -rw-r--r-- | idatui/app.py | 435 | ||||
| -rw-r--r-- | idatui/domain.py | 29 | ||||
| -rw-r--r-- | idatui/launch.py | 6 | ||||
| -rw-r--r-- | idatui/pane.py | 4 | ||||
| -rw-r--r-- | idatui/rpc.py | 33 | ||||
| -rw-r--r-- | idatui/trace.py | 384 |
6 files changed, 861 insertions, 30 deletions
diff --git a/idatui/app.py b/idatui/app.py index ec55911..cc9cfa3 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -91,6 +91,13 @@ _ASM_KEYWORDS = frozenset({ "gs", "ss", "align", "public", "assume", "end", }) _S_CURSOR = Style(bgcolor="#2a313c") +#: Execution trails. Deliberately faint: they sit UNDER the code palette and +#: must not compete with it — the trail says "you came through here", the text +#: still has to be readable as code. Now is the loudest because there is exactly +#: one of it. +_S_TRAIL_NOW = Style(bgcolor="#3f3410") +_S_TRAIL_PAST = Style(bgcolor="#2b1c17") # warm: behind you +_S_TRAIL_FUTURE = Style(bgcolor="#152230") # cool: ahead of you _S_DIM = Style(color="#7c8b9e", italic=True) _S_MATCH = Style(bgcolor="#7a5c00") # all search matches _S_MATCH_CUR = Style(bgcolor="#d0a215", color="#12161c") # the current match @@ -149,6 +156,8 @@ class ViewAnchor: top_ea: int | None = None # first visible address cursor_x: int = 0 flash: str | None = None + #: The edit changed which functions exist, so the index must be rebuilt. + refresh_functions: bool = False @dataclass @@ -739,6 +748,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru Binding("p", "define_func", "Func", show=False), Binding("u", "undefine", "Undef", show=False), Binding("t", "toggle_thumb", "ARM/Thumb", show=False), + Binding("T", "thumb_scan", "Scan vectors", show=False), *SearchMixin.SEARCH_BINDINGS, *NavMixin.NAV_BINDINGS, *ColumnCursor.COL_BINDINGS, @@ -774,6 +784,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru self._search_loading = False self._search_pending: list = [] # done-callbacks awaiting the load self._link_rows: set[int] = set() # split-view: linked instruction rows + #: {address: 'now'|'past'|'future'} painted under the code (trace mode). + self.trail: dict[int, str] = {} # -- text helpers ------------------------------------------------------ # def _head(self, idx: int) -> Head | None: @@ -1042,6 +1054,12 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru linked = idx in self._link_rows if linked: strip = strip.apply_style(_S_LINK) # split-view companion band + if self.trail and h is not None: + kind = self.trail.get(h.ea) + if kind is not None: + strip = strip.apply_style( + _S_TRAIL_NOW if kind == "now" else + _S_TRAIL_PAST if kind == "past" else _S_TRAIL_FUTURE) plain = self._line_plain(idx) if (self._hl_word or idx == self.cursor) else None if idx in self._ranges: strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx)) @@ -1130,6 +1148,9 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def action_toggle_thumb(self) -> None: self.post_message(EditItemRequested(self, "thumb")) + def action_thumb_scan(self) -> None: + self.post_message(EditItemRequested(self, "thumbscan")) + def action_make_data(self) -> None: self.post_message(MakeDataRequested(self)) @@ -1247,6 +1268,9 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True self._gutter = 0 # line-number gutter width (cells) self._line_eas: list[int | None] = [] # per-line address (marker stripped) self._link_line: int | None = None # split-view: linked pseudocode line + #: {line index: 'now'|'past'|'future'} — the execution trail, mapped from + #: instructions onto pseudocode via decomp_map. + self.trail: dict[int, str] = {} self._term = "" self._matches: list[int] = [] self._ranges: dict[int, list[tuple[int, int]]] = {} @@ -1380,6 +1404,11 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True base = self._strips[idx] if linked: base = base.apply_style(_S_LINK) # split-view companion band + kind = self.trail.get(idx) if self.trail else None + if kind is not None: + base = base.apply_style( + _S_TRAIL_NOW if kind == "now" else + _S_TRAIL_PAST if kind == "past" else _S_TRAIL_FUTURE) if idx in self._ranges: base = _overlay_ranges(base, self._ranges[idx], self._match_style(idx)) if self._hl_word: @@ -2280,6 +2309,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. @@ -3020,6 +3144,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 @@ -3065,6 +3193,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), @@ -3076,7 +3210,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. @@ -3090,6 +3224,7 @@ class IdaTui(App): self._load_for_label = None # project binary the dialog is for self._no_functions = False # analysis produced nothing at all self._flash: str | None = None # message a pending reload must keep + self._flash_until = 0.0 # ...until this monotonic time self._pending_switch = None # switch waiting on that answer self._nav_seq = 0 # bumped per navigation; drops stale ones # None = teardown wasn't an explicit quit (crash/kill): save defensively. @@ -3107,6 +3242,12 @@ class IdaTui(App): self._open_path = open_path self._ttl = ttl self._load_args = load_args or "" # IDA switches for a headerless blob + self._title = (os.path.basename(open_path) if open_path else "") + self._trace_path = trace_path or "" # Tenet execution trace to explore + self._trace = None # the loaded Trace, once analysed + self._trail_map = [] # decomp_map for _trail_map_ea + self._trail_map_ea = None + self._t = 0 # current timestamp in that trace self._do_keepalive = keepalive self._rpc_path = rpc_path self._rpc = None @@ -3161,6 +3302,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 @@ -3253,8 +3399,9 @@ class IdaTui(App): n = len(self._func_index) if self._func_index else 0 note = ("this image has no functions, so nothing is lost" if n == 0 else - f"discards the database for this binary \u2014 {n} functions, " - f"plus any names and comments you've added") + f"discards the database for this binary \u2014 {n} " + f"function{'s' if n != 1 else ''}, plus any names and comments " + f"you've added") self.push_screen(ConfirmScreen("Reload with different options?", note), self._on_reload_confirmed) @@ -3413,12 +3560,37 @@ class IdaTui(App): await self._rpc.stop() # -- status helper ----------------------------------------------------- # - def _status(self, text: str) -> None: - if self._binary: # project mode: always say which binary you're in - text = f"[{self._binary}] {text}" + def _status(self, text: str, priority: bool = False) -> None: + """Write the status bar. ``priority`` marks the RESULT of something the + user did. + + An action's result is written, and then the reload it triggered writes + its own idle status on top — cursor moved, filter re-applied, functions + re-counted. I patched that at five separate call sites before admitting + it's one problem: routine chatter must not outrank an answer. A priority + message holds the bar briefly and is cleared by the next keypress, i.e. + when the user has read it and moved on. + """ + import time as _time + if priority: + self._flash = text + self._flash_until = _time.monotonic() + 8.0 + elif self._flash and _time.monotonic() < self._flash_until: + text = self._flash + # Always say WHICH file this is. In project mode that's the binary's + # label; otherwise the filename we opened. Cheap on purpose — _module() + # asks the worker, and this runs on every status write. + tag = self._binary or self._title + if tag: + text = f"[{tag}] {text}" # An image with no functions at all is nearly always a blob described # wrongly, and that stays true as you scroll around — so it belongs in # the status bar, not in a one-off message the next write clobbers. + # It stops being true the moment a function exists, though: latching it + # meant the warning survived defining one with `p` and kept telling you + # the load was wrong when it no longer was. + if self._no_functions and self._func_index is not None and len(self._func_index): + self._no_functions = False if self._no_functions: text += " \u2014 no functions: wrong processor/base? Ctrl+L to reload" try: @@ -3556,6 +3728,7 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged + self._title = os.path.basename(self._open_path) return client if not self._open_path: self.app.call_from_thread( @@ -3578,7 +3751,6 @@ class IdaTui(App): self._func_index = idx self.app.call_from_thread(lambda: self.query_one("#func-table", DataTable).clear()) last = 0 - module = self._module() while not idx.complete: idx.load_next_page() rows = idx.window(last, len(idx) - last) @@ -3586,18 +3758,20 @@ class IdaTui(App): if rows: self.app.call_from_thread(self._append_rows, rows) self.app.call_from_thread( - self._status, f"{module} — {last} functions…" + self._status, f"{last} functions…" ) # If a filter is active (typed during load), re-apply it over the full set. if self._filter_term: self.app.call_from_thread(self._apply_filter, self._filter_term) else: self.app.call_from_thread( - self._status, f"{module} — {len(idx)} functions (Ctrl+N: find symbol)") + self._status, f"{len(idx)} functions (Ctrl+N: find symbol)") # Land somewhere useful instead of an empty pane: main() if present, # 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: @@ -3808,7 +3982,7 @@ class IdaTui(App): if term: self._status(f"filter '{term}': {len(matched)}/{total}") else: - self._status(f"{self._module()} — {total} functions") + self._status(f"{total} functions") def _apply_pending_filter(self) -> None: self._filter_timer = None @@ -4010,6 +4184,7 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged + self._title = os.path.basename(self._open_path) self._active = st.active if st else "listing" self._split = st.split if st else False self._filter_term = st.filter_term if st else "" @@ -5155,7 +5330,7 @@ class IdaTui(App): assert self.program is not None verb = {"code": "defined code", "func": "created function", "undef": "undefined", "string": "made string", - "thumb": "switched decoding"}[kind] + "thumb": "switched decoding", "thumbscan": "scanned"}[kind] try: if kind == "code": # Keep going until something stops it: one instruction is rarely @@ -5180,6 +5355,19 @@ class IdaTui(App): "limit": "instruction limit"}.get(why, why) verb = (f"defined {n} instruction{'s' if n != 1 else ''} " f"({ea:#x}\u2013{end:#x}) \u2014 {reason}") + elif kind == "thumbscan": + # A vector table is a list of Thumb entry points that IDA won't + # follow on a headerless image, because nothing tells it those + # words are pointers. Scan from the cursor. + anchor.refresh_functions = True + r = self.program.thumb_scan(ea, ea + 0x400) + n, applied = int(r.get("n", 0)), int(r.get("applied", 0)) + if not n: + verb = (f"no Thumb entry pointers in {ea:#x}\u2013{ea+0x400:#x}" + " (odd words pointing into the image)") + else: + verb = (f"{n} Thumb entr{'y' if n == 1 else 'ies'} found, " + f"{applied} disassembled") elif kind == "thumb": # Switch the mode, then disassemble in it: flipping T and # leaving the bytes undefined shows nothing, and the reason you @@ -5202,6 +5390,7 @@ class IdaTui(App): # anchor restore, same flash. That is the whole point of having # one path. elif kind == "func": + anchor.refresh_functions = True r = self.program.define_func(ea) if r.get("start") and r.get("end"): verb = (f"created function {r['start']}\u2013{r['end']}" @@ -5211,6 +5400,8 @@ class IdaTui(App): s = self.program.make_string(ea) verb = f"made string ({s[:24]!r})" if s else verb else: + # Undefining can destroy a function as easily as `p` creates one. + anchor.refresh_functions = True self.program.undefine(ea) except Exception as e: # noqa: BLE001 -- surface soft/hard tool errors self.app.call_from_thread(self._status, f"{kind}: {e}") @@ -5246,9 +5437,190 @@ class IdaTui(App): written and lost. """ self._dirty = True - self._flash = anchor.flash if anchor.flash: - self._status(anchor.flash) + self._status(anchor.flash, priority=True) + if anchor.refresh_functions: + # Creating (or destroying) a function changes the index that the + # names pane, Ctrl+N and the "no functions" hint all read. Without + # 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) + self._paint_trail() + if follow: + self._goto_ea(t.ip(self._t), push=False) + + def _paint_trail(self) -> None: + """Push the execution trail into the code views. + + Recomputed per seek rather than per repaint: it's ~200 lookups, and a + repaint happens far more often than a step. + """ + t = self._trace + if t is None: + return + trail = t.trail(self._t) + try: + lst = self.query_one(ListingView) + lst.trail = trail + lst.refresh() + except Exception: # noqa: BLE001 -- view not mounted yet + pass + self._paint_trail_decomp(trail) + + def _paint_trail_decomp(self, trail: dict) -> None: + """Map the instruction trail onto pseudocode lines. + + This is the thing Tenet can't do: it paints disassembly, because that's + where a trace's addresses live. We already have decomp_map (built for + the split view) saying which instructions each pseudocode line covers, + so the same trail lands on C. + + A line covers many instructions, so it takes the strongest kind present: + 'now' wins over 'past' wins over 'future' — if the instruction you are + standing on is part of this line, this line is where you are. + """ + try: + dec = self.query_one(DecompView) + except Exception: # noqa: BLE001 + return + ea = dec.loaded_ea + if not dec.display or ea is None or self.program is None: + dec.trail = {} + return + if self._trail_map_ea != ea: + # Cached per function: decomp_map is an RPC and stepping is + # interactive, so paying it per keystroke would be felt. + try: + self._trail_map = self.program.decomp_map(ea) + except Exception: # noqa: BLE001 + self._trail_map = [] + self._trail_map_ea = ea + rank = {"future": 0, "past": 1, "now": 2} + lines: dict[int, str] = {} + for i, eas in enumerate(self._trail_map or []): + best = None + for a in eas: + k = trail.get(a) + if k is not None and (best is None or rank[k] > rank[best]): + best = k + if best is not None: + lines[i] = best + dec.trail = lines + dec.refresh() + + 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. + + Deliberately not _load_functions(): that one is the BOOT path — it + clears the table, streams progress and then auto-lands, which would + yank the view away from the function you just made. + """ + if self.program is None: + return + idx = self.program.functions() + idx.load_all() + self._func_index = idx + self.app.call_from_thread(self._after_reindex) + + def _after_reindex(self) -> None: + idx = self._func_index + if idx is None: + return + if len(idx): + self._no_functions = False + self._apply_filter(self._filter_term) # repopulate the names pane @work(thread=True, exclusive=True, group="save") def _save(self) -> None: @@ -5526,6 +5898,8 @@ class IdaTui(App): view.focus() def on_key(self, event) -> None: # type: ignore[no-untyped-def] + # Any keypress means the result of the last edit has been read. + self._flash = None if event.key != "escape": return if self.query_one("#search", Input).display: @@ -5895,10 +6269,6 @@ class IdaTui(App): self._goto_ea(msg.va, push=True) def _status_for_cur(self, mode: str) -> None: - flash, self._flash = self._flash, None - if flash: - self._status(flash) # the edit that caused this reload wins - return if self._cur is not None: self._status(f"{self._cur.name} @ {self._cur.ea:#x} [{mode}]") @@ -5906,12 +6276,22 @@ class IdaTui(App): def _load_decomp(self, ea: int, name: str) -> None: assert self.program is not None dec = self.program.decompile(ea) - self.app.call_from_thread(self._apply_decomp, ea, name, dec) + why = "" + if dec.failed: + # Ask Hex-Rays why, in the same worker: the plain tool reports + # "Decompilation failed at 0x0" and drops the only useful part. + # "Decompile failed" with no reason is indistinguishable from a bug + # in this app, and for the common cause (a 32-bit function in a + # 64-bit database) the user cannot even guess the fix. + why = self.program.decomp_error(ea) + self.app.call_from_thread(self._apply_decomp, ea, name, dec, why) - def _apply_decomp(self, ea: int, name: str, dec) -> None: # type: ignore[no-untyped-def] + def _apply_decomp(self, ea: int, name: str, dec, # type: ignore[no-untyped-def] + why: str = "") -> None: view = self.query_one(DecompView) view.loading = False if dec.failed: + detail = f" \u2014 {why}" if why else "" # No pseudocode for this function: fall back to the code view rather # than an error panel. If we came from the continuous listing (F5), # return there; otherwise show the disassembly. @@ -5922,13 +6302,17 @@ class IdaTui(App): self._decomp_return = None self._cur = ret self._active = "listing" + # Hand the reason over as a flash BEFORE reopening: going back + # to the listing reloads it, and the reload writes its own + # status afterwards — which is precisely how "F5 does nothing" + # looked like nothing at all. + msg = f"{name}: cannot decompile{detail}" + self._status(msg, priority=True) self._open_entry(ret, push=False) - self._status(f"{name} — decompile failed; back to the listing") return self._active = "disasm" + self._status(f"{name}: cannot decompile{detail}", priority=True) self._show_active() - self._status( - f"{name} — no pseudocode (decompile failed); showing disassembly") return if self._active == "decomp": view.focus() # loading cover had blurred it; restore focus @@ -6135,13 +6519,6 @@ class IdaTui(App): return ea = msg.ea if ea is not None: - # A pending flash (the result of an edit that caused this reload) - # outranks the idle hint: the cursor lands here as part of the - # reload, so this handler would otherwise always have the last word. - flash, self._flash = self._flash, None - if flash: - self._status(flash) - return sec = self.program.section_of(ea) if self.program else None self._status(f"{sec or '?'} @ {ea:#x} [listing] " "(c code · p func · u undefine · Enter follow)") diff --git a/idatui/domain.py b/idatui/domain.py index 9047d82..97b042d 100644 --- a/idatui/domain.py +++ b/idatui/domain.py @@ -1373,6 +1373,35 @@ class Program: if res.get("error"): raise IDAToolError("define_code", f"@ {ea:#x}: {res['error']}") + def decomp_error(self, ea: int) -> str: + """Hex-Rays' own reason for refusing ``ea``, or "" if it won't say.""" + try: + r = self.client.call("decomp_error", addr=hex(ea)) + except IDAToolError: + return "" + if not isinstance(r, dict): + return "" + reason = str(r.get("reason") or "") + if reason and r.get("bitness") == 64 and "64-bit" in reason: + # Say the FIX, not the diagnosis. Hex-Rays' own sentence ("only + # 64-bit functions can be decompiled in the current database") is + # accurate and useless: it describes the database, not what to do, + # and it's long enough that a status bar cuts off the end — which is + # exactly where an appended hint would live. This is unfixable in + # place (bitness is decided at load), so the whole message is the + # instruction. + return "this database is 64-bit \u2014 Ctrl+L, pick arm:ARMv7-A" + return reason + + def thumb_scan(self, start: int, end: int, apply: bool = True) -> dict: + """Find Thumb entry points from odd pointers in ``[start, end)``.""" + r = self.client.call("thumb_scan", start=hex(start), end=hex(end), + apply=bool(apply)) + if not isinstance(r, dict) or r.get("error"): + raise IDAToolError("thumb_scan", + f"@ {start:#x}: {(r or {}).get('error', 'failed')}") + return r + def set_thumb(self, ea: int, mode: str = "toggle") -> dict: """Switch ARM/Thumb decoding at ``ea``. Returns the resulting state.""" r = self.client.call("set_thumb", addr=hex(ea), mode=mode) 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/idatui/pane.py b/idatui/pane.py index 53afef3..1a46a4f 100644 --- a/idatui/pane.py +++ b/idatui/pane.py @@ -165,6 +165,8 @@ def spawn(args) -> int: inner += ["--rpc", sock] else: inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock] + if getattr(args, "trace", None): + inner += ["--trace", os.path.abspath(os.path.expanduser(args.trace))] cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner) split = ["split-window", "-v" if args.vertical else "-h", @@ -316,6 +318,8 @@ def main(argv: list[str]) -> int: sp = sub.add_parser("spawn", help="open a TUI pane and wait until ready") sp.add_argument("--open", metavar="PATH", help="binary to open (its dir must be writable)") + sp.add_argument("--trace", metavar="FILE", + help="Tenet execution trace to load alongside the binary") sp.add_argument("--project", metavar="FILE", help="project file to open instead of a single binary; " "any --open paths are added to it (created if absent)") diff --git a/idatui/rpc.py b/idatui/rpc.py index 0921b89..7741f48 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -72,6 +72,7 @@ METHODS = { "search": "{term,direction?=1} incremental search in the code view", "select": "{index?} choose the highlighted/nth item in the open modal", "save": "persist the .i64 (Ctrl+S)", + "trace": "{seek|goto|step,over?} navigate the execution trace (seek '!50' = percent)", "binaries": "-> project binaries {label,active,resident,indexed} (project mode)", "switch": "{binary,addr?} make another project binary active (addr also jumps)", "close": "dismiss a modal (Escape)", @@ -531,6 +532,38 @@ class RpcServer: return functions(app, params.get("filter"), int(params.get("limit", 50))) # -- projects ------------------------------------------------------ # + if method == "trace": + if app._trace is None: + raise ValueError("no trace loaded (launch with --trace FILE)") + t = app._trace + if "seek" in params: + v = params["seek"] + # "!50" seeks a percentage, like Tenet's timestamp shell. + if isinstance(v, str) and v.startswith("!"): + idx = int(float(v[1:]) * (t.length - 1) / 100.0) + else: + idx = int(str(v).replace(",", ""), 0) if isinstance(v, str) else int(v) + app._seek(idx) + elif "goto" in params: # first execution of an address/name + tgt = params["goto"] + ea = (int(str(tgt), 0) if str(tgt).lower().startswith("0x") + else app.program.resolve(str(tgt))) + first = t.first_execution(ea) + if first is None: + raise ValueError(f"{tgt} never executed in this trace") + app._seek(first) + elif "step" in params: + n = int(params.get("step") or 1) + over = bool(params.get("over")) + for _ in range(abs(n)): + (app._step_over if over else app._step)(1 if n > 0 else -1) + await settle(app, timeout=float(params.get("timeout", 20.0))) + snap = snapshot(app) + snap["trace"] = {"idx": app._t, "length": t.length, + "pc": hex(t.ip(app._t)), + "changed": sorted(t.changed(app._t))} + return snap + if method == "binaries": if app._project is None: raise ValueError("not a project session (launch with --project)") diff --git a/idatui/trace.py b/idatui/trace.py new file mode 100644 index 0000000..2afc8f6 --- /dev/null +++ b/idatui/trace.py @@ -0,0 +1,384 @@ +"""Reading Tenet execution traces. + +A Tenet trace is a line-per-instruction delta log:: + + rax=0x3c,rbx=0x0,...,rip=0x7ffff6faaae0 # full state on the first line + rip=0x7ffff6faaae4 # then only what changed + r9=0x7ffff6762e60,rip=0x7ffff6faaae9,mw=0x7ffff6f9b7c8:08c177f6ff7f0000 + +Registers that changed, the PC on every line, and every memory access *with its +bytes*. That is enough to reconstruct any register or any memory address at any +point in time, forwards or backwards, which is the whole trick. + +This is our own reader, not a port. The reference implementation +(``~/dev/tenet/tenet-original/plugins/tenet/trace/``) packs the trace into +segments with compressed address/mask tables, which earns its keep for its Qt +timeline; we need different queries and would rather own the ~400 lines than +inherit 3700. It is differential-tested against that implementation +(``tests/test_trace_vs_tenet.py``) so "our own" doesn't quietly mean "different". + +The index is built around the query the UI actually asks, which the reference +answers one address at a time: **which timestamps executed this set of +addresses**. A listing row is one address, but a pseudocode line covers many, so +``by_ip`` maps address -> timestamps and set queries are unions of those. +""" + +from __future__ import annotations + +import array +import bisect +import os +import re +from dataclasses import dataclass, field + +#: Tenet packs its register delta into a uint32, so a trace arch may name at +#: most 32 registers. Ours is discovered from the trace instead of declared, +#: but the cap is worth knowing when a trace looks short of registers. +MAX_REGISTERS = 32 + +_MEM_RE = re.compile(r"^m(r|w|rw)$") + + +@dataclass +class TraceInfo: + """The sidecar ``<prefix>.info`` written by our QEMU tracer. + + Optional: a trace from another tracer has none, and everything here can be + recovered or guessed from the log itself. + """ + + arch: str = "" + mode: str = "" + binary: str = "" + start_code: int = 0 + end_code: int = 0 + entry_code: int = 0 + traced: str = "" + + @classmethod + def load(cls, path: str) -> "TraceInfo | None": + try: + with open(path) as f: + raw = dict( + ln.strip().split("=", 1) for ln in f if "=" in ln) + except OSError: + return None + def num(k): + try: + return int(raw.get(k, "0"), 0) + except ValueError: + return 0 + return cls(arch=raw.get("arch", ""), mode=raw.get("mode", ""), + binary=raw.get("binary", ""), start_code=num("start_code"), + end_code=num("end_code"), entry_code=num("entry_code"), + traced=raw.get("traced", "")) + + +@dataclass +class MemOp: + """One memory access made by one instruction.""" + + addr: int + data: bytes + write: bool + + @property + def end(self) -> int: + return self.addr + len(self.data) + + +@dataclass +class Trace: + """An indexed Tenet trace. + + Timestamps are indices into the executed-instruction sequence: 0 is the + first instruction, ``length - 1`` the last. + """ + + path: str = "" + info: TraceInfo | None = None + #: PC per timestamp. + ips: array.array = field(default_factory=lambda: array.array("Q")) + #: name -> (timestamps of change, value at each change). A register's value + #: at time t is the last change at or before t; the first line carries a + #: full state dump, so every register has an entry at 0. + reg_at: dict[str, tuple[array.array, array.array]] = field(default_factory=dict) + #: address -> timestamps that executed it (ascending, by construction). + by_ip: dict[int, array.array] = field(default_factory=dict) + #: memory accesses, parallel arrays indexed by access number. + mem_idx: array.array = field(default_factory=lambda: array.array("I")) + mem_addr: array.array = field(default_factory=lambda: array.array("Q")) + mem_write: bytearray = field(default_factory=bytearray) + mem_off: array.array = field(default_factory=lambda: array.array("Q")) + mem_len: array.array = field(default_factory=lambda: array.array("H")) + mem_blob: bytearray = field(default_factory=bytearray) + #: first access number of each timestamp, plus a final sentinel. + mem_row: array.array = field(default_factory=lambda: array.array("I")) + #: applied so trace addresses line up with the database (see ``rebase``). + slide: int = 0 + + # -- construction ------------------------------------------------------ # + @classmethod + def load(cls, path: str, progress=None, limit: int = 0) -> "Trace": + """Parse a text trace. ``progress(lines)`` is called every 50k lines.""" + t = cls(path=os.path.abspath(path)) + base = path[:-6] if path.endswith(".0.log") else os.path.splitext(path)[0] + t.info = TraceInfo.load(base + ".info") + regs: dict[str, tuple[array.array, array.array]] = {} + ips, by_ip = t.ips, t.by_ip + n = 0 + with open(path) as f: + for line in f: + line = line.strip() + if not line: + continue + ip = None + t.mem_row.append(len(t.mem_idx)) + for part in line.split(","): + key, _, val = part.partition("=") + if not val: + continue + key = key.strip().lower() + if _MEM_RE.match(key): + addr_s, _, data_s = val.partition(":") + try: + addr = int(addr_s, 16) + data = bytes.fromhex(data_s) + except ValueError: + continue + # 'mrw' is one access that both reads and writes; record + # the write, which is what the memory state follows. + t.mem_idx.append(n) + t.mem_addr.append(addr) + t.mem_write.append(0 if key == "mr" else 1) + t.mem_off.append(len(t.mem_blob)) + t.mem_len.append(len(data)) + t.mem_blob += data + continue + try: + v = int(val, 16) + except ValueError: + continue + slot = regs.get(key) + if slot is None: + slot = regs[key] = (array.array("I"), array.array("Q")) + slot[0].append(n) + slot[1].append(v) + ip = v if key in ("rip", "eip", "pc") else ip + if ip is None: + # No PC on this line: the format says always emit it, and + # without it the line cannot be placed. Carry the previous + # one rather than dropping the instruction. + ip = ips[-1] if ips else 0 + ips.append(ip) + where = by_ip.get(ip) + if where is None: + where = by_ip[ip] = array.array("I") + where.append(n) + n += 1 + if progress is not None and n % 50000 == 0: + progress(n) + if limit and n >= limit: + break + t.mem_row.append(len(t.mem_idx)) + t.reg_at = regs + return t + + # -- basics ------------------------------------------------------------ # + def __len__(self) -> int: + return len(self.ips) + + @property + def length(self) -> int: + return len(self.ips) + + @property + def registers(self) -> list[str]: + """Register names in the trace, PC last (the order tracers emit).""" + return sorted(self.reg_at, key=lambda r: (r in ("rip", "eip", "pc"), r)) + + @property + def pc_name(self) -> str: + for n in ("rip", "eip", "pc"): + if n in self.reg_at: + return n + return "" + + def ip(self, idx: int) -> int: + """PC at ``idx``, in DATABASE addresses (slide applied).""" + return self.ips[idx] + self.slide + + def raw_ip(self, idx: int) -> int: + return self.ips[idx] + + # -- register state ----------------------------------------------------- # + def register(self, name: str, idx: int) -> int | None: + """Value of ``name`` at ``idx``, or None if the trace never set it.""" + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + idxs, vals = slot + i = bisect.bisect_right(idxs, idx) - 1 + return vals[i] if i >= 0 else None + + def register_state(self, idx: int) -> dict[str, int]: + return {n: v for n in self.reg_at + if (v := self.register(n, idx)) is not None} + + def changed(self, idx: int) -> set[str]: + """Registers written BY the instruction at ``idx`` (what the line said). + + Used to highlight what an instruction actually did, which is the reason + a delta trace is readable at all. + """ + out = set() + for n, (idxs, _vals) in self.reg_at.items(): + i = bisect.bisect_left(idxs, idx) + if i < len(idxs) and idxs[i] == idx: + out.add(n) + return out + + def last_write(self, name: str, idx: int) -> int | None: + """Timestamp of the write that produced ``name``'s value at ``idx``. + + "Which instruction set this register?" — the question that motivates a + trace explorer in the first place. + """ + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + i = bisect.bisect_right(slot[0], idx) - 1 + return slot[0][i] if i >= 0 else None + + def next_write(self, name: str, idx: int) -> int | None: + slot = self.reg_at.get(name.lower()) + if slot is None: + return None + i = bisect.bisect_right(slot[0], idx) + return slot[0][i] if i < len(slot[0]) else None + + # -- memory ------------------------------------------------------------- # + def memory_ops(self, idx: int) -> list[MemOp]: + """Accesses made by the instruction at ``idx``.""" + if not (0 <= idx < len(self.mem_row) - 1): + return [] + lo, hi = self.mem_row[idx], self.mem_row[idx + 1] + out = [] + for k in range(lo, hi): + off, ln = self.mem_off[k], self.mem_len[k] + out.append(MemOp(addr=self.mem_addr[k] + self.slide, + data=bytes(self.mem_blob[off:off + ln]), + write=bool(self.mem_write[k]))) + return out + + # -- execution queries (what painting is built on) ---------------------- # + def executions(self, ea: int) -> array.array: + """Every timestamp that executed ``ea`` (database address).""" + return self.by_ip.get(ea - self.slide, array.array("I")) + + def executions_between(self, ea: int, lo: int, hi: int) -> list[int]: + ts = self.executions(ea) + a = bisect.bisect_left(ts, lo) + b = bisect.bisect_right(ts, hi) + return list(ts[a:b]) + + def hits(self, eas) -> dict[int, int]: + """{address: execution count} for a set of addresses. + + The set form is the point: painting a listing row needs one address, but + a pseudocode line covers many, and asking per-address would mean a + lookup per instruction per repaint. + """ + out = {} + for ea in eas: + ts = self.by_ip.get(ea - self.slide) + if ts: + out[ea] = len(ts) + return out + + def prev_ips(self, idx: int, n: int) -> list[int]: + """Addresses executed in the ``n`` steps before ``idx`` (nearest first). + + A trail, not all of history: showing every address the trace ever + touched says almost nothing on a loop-heavy program, whereas the last + few dozen steps say how you GOT here. + """ + lo = max(idx - n, 0) + return [self.ips[i] + self.slide for i in range(idx - 1, lo - 1, -1)] + + def next_ips(self, idx: int, n: int) -> list[int]: + """Addresses executed in the ``n`` steps after ``idx`` (nearest first).""" + hi = min(idx + n + 1, self.length) + return [self.ips[i] + self.slide for i in range(idx + 1, hi)] + + def trail(self, idx: int, n: int = 96) -> dict[int, str]: + """{address: 'now' | 'past' | 'future'} around ``idx``. + + Where an address appears on both sides — a loop body, which is most of + them — the nearer side wins, because that's the one that explains the + step you are about to take or just took. + """ + out: dict[int, str] = {} + for k, ea in enumerate(self.next_ips(idx, n)): + out.setdefault(ea, "future") + for k, ea in enumerate(self.prev_ips(idx, n)): + prev = out.get(ea) + if prev is None: + out[ea] = "past" + elif prev == "future": + # Same distance rule as above, resolved by which loop found it + # first would be arbitrary; compare real distances instead. + fwd = next((i for i, a in enumerate(self.next_ips(idx, n)) if a == ea), n) + if k < fwd: + out[ea] = "past" + if 0 <= idx < self.length: + out[self.ips[idx] + self.slide] = "now" + return out + + def first_execution(self, ea: int) -> int | None: + ts = self.executions(ea) + return ts[0] if ts else None + + def next_execution(self, ea: int, idx: int) -> int | None: + ts = self.executions(ea) + i = bisect.bisect_right(ts, idx) + return ts[i] if i < len(ts) else None + + def prev_execution(self, ea: int, idx: int) -> int | None: + ts = self.executions(ea) + i = bisect.bisect_left(ts, idx) - 1 + return ts[i] if i >= 0 else None + + # -- lining the trace up with the database ------------------------------ # + def rebase(self, db_addresses) -> int: + """Find the slide between trace addresses and database addresses. + + A traced process is relocated (ASLR, or a PIE base the database doesn't + share): our echo trace runs at 0x7ffff6faa000 while the database has the + same code at 0x2490. Nothing lines up until this is solved, so it is not + optional. + + Page offsets survive relocation — only whole pages move — so the low 12 + bits of an instruction address are invariant. Bucket the database's + addresses by those bits, and for each trace address the candidate slides + are the differences to database addresses in its bucket. The slide that + the most instructions agree on wins. + """ + buckets: dict[int, list[int]] = {} + for a in db_addresses: + buckets.setdefault(a & 0xFFF, []).append(a) + if not buckets: + return 0 + votes: dict[int, int] = {} + # A sample is enough and keeps this O(1)-ish on a 10M trace; unique + # addresses, because a hot loop shouldn't outvote the rest of the code. + for ea in list(self.by_ip)[:4096]: + for cand in buckets.get(ea & 0xFFF, ()): + votes[cand - ea] = votes.get(cand - ea, 0) + 1 + if not votes: + return 0 + best, n = max(votes.items(), key=lambda kv: kv[1]) + return best if n > 1 else 0 + + def apply_slide(self, slide: int) -> None: + self.slide = int(slide) |