diff options
Diffstat (limited to 'idatui/app.py')
| -rw-r--r-- | idatui/app.py | 1804 |
1 files changed, 1403 insertions, 401 deletions
diff --git a/idatui/app.py b/idatui/app.py index e96483b..9d99122 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -53,6 +53,23 @@ from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct _S_ADDR = Style(color="#6b7684") _S_LABEL = Style(color="#7aa2f7", bold=True) _S_INSN = Style(color="#c3cad3") +#: IDA's token kinds -> the measured palette. The rule that keeps a dense +#: disassembly readable: NEUTRALS for the machine (mnemonic brightest because you +#: scan down that column, registers at body weight because they're most of the +#: text), HUES only where they mean something (numbers, strings, symbols), +#: structure recedes so brackets and commas stop competing with operands. +_S_SPAN = { + "insn": Style(color="#e8ecf2"), # 15.3:1 mnemonic / directive + "reg": Style(color="#c3cad3"), # 11.0:1 registers = body weight + "num": Style(color="#d8a657"), # 8.2:1 immediates, offsets + "str": Style(color="#9ece6a"), # 9.9:1 string literals + "name": Style(color="#7aa2f7"), # 7.2:1 symbols / xref targets + "seg": Style(color="#93aee0"), # 8.1:1 segment names + "cmt": Style(color="#7c8b9e", italic=True), # 5.2:1 + "punct": Style(color="#626c7a"), # 3.4:1 brackets, commas, +/- + "err": Style(color="#c9762f"), # IDA's own error marker + "text": Style(color="#c3cad3"), # 11.0:1 anything unclassified +} _S_MNEM = Style(color="#e8ecf2") _S_OPBYTES = Style(color="#5e6875") # raw opcode bytes column _S_DATA = Style(color="#d8a657") @@ -74,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 @@ -105,7 +129,6 @@ class BinaryState: func_index: object | None = None nav: list = field(default_factory=list) cur: object | None = None - pref: str = "listing" active: str = "listing" split: bool = False filter_term: str = "" @@ -113,6 +136,31 @@ class BinaryState: @dataclass +class ViewAnchor: + """Where the user is looking, expressed in ADDRESSES. + + Every path that rebuilds a model must round-trip through this. Row indices + do NOT survive a rebuild: defining code collapses four undefined byte rows + into one instruction row, undefining does the reverse, and a rename can add + or remove banner rows above a function. Anything that remembers an index + puts the user somewhere else afterwards, which reads as "the edit jumped my + screen" or, worse, "the edit didn't apply". + + ``flash`` travels with it because the same rebuild also decides what the + status bar says: the reload writes its own status when it lands, so an edit + that doesn't hand its message over here gets silently overwritten. + """ + + view: str = "listing" + ea: int | None = None # cursor address + 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 class NavEntry: ea: int name: str @@ -671,305 +719,12 @@ class SearchMixin: # --------------------------------------------------------------------------- # # Virtualized disassembly view # --------------------------------------------------------------------------- # -class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True): - """A line-virtualized disassembly listing for a single function.""" - - BINDINGS = [ - Binding("j,down", "cursor_down", "Down", show=False), - Binding("k,up", "cursor_up", "Up", show=False), - Binding("ctrl+d", "half_page(1)", "½↓", show=False), - Binding("ctrl+u", "half_page(-1)", "½↑", show=False), - Binding("pagedown", "page(1)", "PgDn", show=False), - Binding("pageup", "page(-1)", "PgUp", show=False), - Binding("home", "goto_top", "Top", show=False), - Binding("G,end", "goto_bottom", "Bottom", show=False), - Binding("o", "toggle_opcodes", "Opcodes", show=False), - Binding("c", "define_code", "Code", show=False), - Binding("p", "define_func", "Func", show=False), - Binding("u", "undefine", "Undef", show=False), - Binding("tab,shift+tab", "app.toggle_view", "Pseudocode", priority=True), - Binding("f5", "app.toggle_view", "Decompile", priority=True), - Binding("L", "app.continuous_here", "Listing"), - *SearchMixin.SEARCH_BINDINGS, - *NavMixin.NAV_BINDINGS, - *ColumnCursor.COL_BINDINGS, - ] - - cursor = reactive(0, repaint=False) - cursor_x = reactive(0, repaint=False) - - class CursorMoved(Message): - """Posted when the disasm cursor moves; carries the instruction ea.""" - - def __init__(self, index: int, ea: int | None) -> None: - super().__init__() - self.index = index - self.ea = ea - - def __init__(self) -> None: - super().__init__() - self.model: DisasmModel | None = None - self.total = 0 - self._name = "" - self._pending_scroll_y: int | None = None - self._term = "" - self._matches: list[int] = [] - self._ranges: dict[int, list[tuple[int, int]]] = {} - self._search_texts: list[str] | None = None - self._show_ops = True - self._op_w = 0 # char width of the hex-bytes field (excl. trailing gap) - - def _op_field(self, line) -> str: # type: ignore[no-untyped-def] - """The padded opcode-bytes column (empty when hidden). Kept identical - between the rendered strip and the plain text so cursor/search offsets - line up.""" - if not self._show_ops or self._op_w <= 0: - return "" - raw = line.raw or b"" - return " ".join(f"{b:02X}" for b in raw).ljust(self._op_w) + " " - - def _line_plain(self, idx: int) -> str | None: - if self.model is None: - return None - line = self.model.cached_line(idx) - if line is None: - return None - s = f"{line.ea:08X} " + self._op_field(line) - if line.label: - s += f"{line.label}: " - return s + line.text - - # -- public API -------------------------------------------------------- # - def load(self, model: DisasmModel, name: str, cursor: int = 0, - cursor_x: int = 0, scroll_y: int | None = None) -> None: - self.model = model - self._name = name - self.total = 0 - self.cursor = cursor - self.cursor_x = cursor_x - self._pending_scroll_y = scroll_y - # NB: don't zero virtual_size here — that snaps the scroll to 0 and - # causes a visible jump before _on_primed restores the target scroll. - self._matches = [] - self._ranges = {} - self._search_texts = None - self._prime() - - # -- search hooks ------------------------------------------------------ # - def _fmt(self, line) -> str: # type: ignore[no-untyped-def] - s = f"{line.ea:08X} " + self._op_field(line) - if line.label: - s += f"{line.label}: " - return s + line.text - - def _search_line_count(self) -> int: - return self.total - - def _search_line_text(self, i: int) -> str | None: - t = self._search_texts - return t[i] if t is not None and 0 <= i < len(t) else None - - def _search_ensure(self, done) -> None: - if self._search_texts is not None: - done() - return - self._app_status(f"/{self._term}/ indexing {self.total} lines…") - self._index_for_search(done) - - @work(thread=True, exclusive=True, group="search-index") - def _index_for_search(self, done) -> None: - model = self.model - if model is None: - self.app.call_from_thread(done) - return - texts: list[str] = [] - off, total = 0, self.total - while off < total: - lines = model.lines(off, DisasmModel.BLOCK, prefetch=False) - if not lines: - break - texts.extend(self._fmt(ln) for ln in lines) - off += len(lines) - self._search_texts = texts - self.app.call_from_thread(done) - - @work(thread=True, exclusive=True, group="disasm-prime") - def _prime(self) -> None: - model = self.model - if model is None: - return - total = model.total() - height = max(self.size.height, 1) - model.lines(0, min(total, height + DisasmModel.BLOCK), prefetch=True) - if self.cursor: - model.lines(max(self.cursor - 2, 0), height, prefetch=True) - self.app.call_from_thread(self._on_primed, total) - - def _on_primed(self, total: int) -> None: - self.total = total - self.virtual_size = Size(0, total) - self._update_op_w() # provisional width from the primed window - self._scan_op_width() # settle it against the whole function - self._clamp_x() # cursor line is now cached; keep the column in range - if self._pending_scroll_y is not None and self._pending_scroll_y >= 0: - self._apply_scroll(min(self._pending_scroll_y, max(total - 1, 0))) - else: - self._scroll_cursor_into_view() - self._pending_scroll_y = None - self.refresh() - - # -- rendering --------------------------------------------------------- # - def render_line(self, y: int) -> Strip: - model = self.model - width = self.size.width - if model is None or self.total == 0: - return Strip([Segment("".ljust(width), _S_DIM)]) - top = round(self.scroll_offset.y) - if y == 0: # once per refresh: warm the visible window + a little ahead - self._ensure_window(top) - idx = top + y - if idx >= self.total: - return Strip([Segment("".ljust(width), _S_INSN)]) - line = model.cached_line(idx) - is_cursor = idx == self.cursor - if line is None: - strip = Strip([Segment(f" {idx:>8} …", _S_DIM)]) - else: - segs: list[Segment] = [Segment(f"{line.ea:08X} ", _S_ADDR)] - op = self._op_field(line) - if op: - segs.append(Segment(op, _S_OPBYTES)) - if line.label: - segs.append(Segment(f"{line.label}: ", _S_LABEL)) - mnem, _, rest = line.text.partition(" ") - segs.append(Segment(mnem, _S_MNEM)) - if rest: - segs.append(Segment(" " + rest, _S_INSN)) - strip = Strip(segs) - if idx in self._ranges: - strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx)) - if is_cursor: - strip = _cursor_decorate(strip, self._line_plain(idx) or "", self.cursor_x) - return strip.adjust_cell_length(width, _S_INSN) - - def _update_op_w(self) -> bool: - """Recompute the opcode column width from the model's widest instruction. - Returns True if it changed.""" - w = 0 - if self.model is not None and self._show_ops: - mx = self.model.max_raw_len() - w = max(mx * 3 - 1, 0) if mx > 0 else 0 - if w != self._op_w: - self._op_w = w - return True - return False - - @work(thread=True, exclusive=True, group="disasm-opwidth") - def _scan_op_width(self) -> None: - model = self.model - if model is None: - return - model.scan_bytes() # fetch all blocks -> stable widest instruction - self.app.call_from_thread(self._settle_op_w) - - def _settle_op_w(self) -> None: - if self._update_op_w(): - self._search_texts = None # layout changed -> stale offsets - self.refresh() - - def action_toggle_opcodes(self) -> None: - self._show_ops = not self._show_ops - self._update_op_w() - self._search_texts = None # column layout changed -> reindex on next search - self._ranges = {} - self._clamp_x() - self.refresh() - self._app_status("opcodes " + ("on" if self._show_ops else "off")) - - def _ensure_window(self, top: int) -> None: - if self.model is None: - return - height = max(self.size.height, 1) - start = max(top - DisasmModel.BLOCK, 0) - count = height + 2 * DisasmModel.BLOCK - if not self.model.is_cached(top, height): - self._fetch_window(start, count) - else: - self.model.ensure_async(start, count) # warm neighbors - - @work(thread=True, exclusive=False, group="disasm-fetch") - def _fetch_window(self, start: int, count: int) -> None: - model = self.model - if model is None: - return - model.lines(start, count, prefetch=True) - self.app.call_from_thread(self.refresh) - - # -- navigation -------------------------------------------------------- # - def _visible_height(self) -> int: - return max(self.size.height, 1) - - def _scroll_cursor_into_view(self) -> None: - height = self._visible_height() - top = round(self.scroll_offset.y) - if self.cursor < top: - self.scroll_to(y=self.cursor, animate=False) - elif self.cursor >= top + height: - self.scroll_to(y=max(self.cursor - height + 1, 0), animate=False) - - def _move(self, delta: int) -> None: - if self.total == 0: - return - old = self.cursor - before = round(self.scroll_offset.y) - self.cursor = max(0, min(self.total - 1, self.cursor + delta)) - self._clamp_x() - self._scroll_cursor_into_view() - if round(self.scroll_offset.y) != before: - self.refresh() # scrolled: the whole viewport shifted - else: - _refresh_lines(self, old, self.cursor) # only the two changed rows - self._refresh_hl() - self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea())) - - def _after_cursor_move(self) -> None: - self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea())) - - def _cursor_ea(self) -> int | None: - if self.model is None: - return None - line = self.model.cached_line(self.cursor) - return line.ea if line else None - - # -- item structure edits (IDA c/p/u) --------------------------------- # - def action_define_code(self) -> None: - self.post_message(EditItemRequested(self, "code")) - - def action_define_func(self) -> None: - self.post_message(EditItemRequested(self, "func")) - - def action_undefine(self) -> None: - self.post_message(EditItemRequested(self, "undef")) - - def action_cursor_down(self) -> None: - self._move(1) - - def action_cursor_up(self) -> None: - self._move(-1) - - def action_goto_top(self) -> None: - self._move(-self.total) - - def action_goto_bottom(self) -> None: - self._move(self.total) - - # --------------------------------------------------------------------------- # # Virtualized flat listing view (code + data + undefined, per segment) # --------------------------------------------------------------------------- # class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True): """A line-virtualized *flat* listing over one segment: code, data and - undefined heads interleaved (IDA's disassembly view), unlike ``DisasmView`` + undefined heads interleaved (IDA's disassembly view), unlike ``DisasmModel`` which is bounded to one function. Backed by ``ListingModel`` (the ``heads`` server tool). Used for non-function regions and raw segment browsing. """ @@ -992,6 +747,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru Binding("a", "make_string", "Str", show=False), 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, @@ -1027,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: @@ -1044,6 +803,24 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru return " ".join(f"{b:02X}" for b in raw[:_OP_LIMIT]) + "\u2026" return " ".join(f"{b:02X}" for b in raw) + @staticmethod + def _span_segments(h: Head, fallback: Style): + """Segments for a row's disassembly text. + + Uses IDA's own token classification when the worker supplied it; falls + back to the old mnemonic/rest split so an older worker (or a row whose + spans didn't match the text) still renders. + """ + if h.spans: + return [Segment(t, _S_SPAN.get(k, fallback)) for k, t in h.spans] + if h.kind == "code": + mnem, _, rest = h.text.partition(" ") + segs = [Segment(mnem, _S_MNEM)] + if rest: + segs.append(Segment(" " + rest, fallback)) + return segs + return [Segment(h.text, fallback)] + def _op_field(self, h: Head) -> str: """The padded opcode-bytes column (empty when hidden). Shared format so cursor/search offsets line up.""" @@ -1268,21 +1045,21 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru segs.append(Segment(_LST_INDENT, _S_MEMBER)) if h.name: segs.append(Segment(f"{h.name} ", _S_LABEL)) - if h.kind == "code": - mnem, _, rest = h.text.partition(" ") - segs.append(Segment(mnem, _S_MNEM)) - if rest: - segs.append(Segment(" " + rest, _S_INSN)) - elif h.kind == "data": - segs.append(Segment(h.text, _S_DATA)) - elif h.kind == "member": + if h.kind == "member": segs.append(Segment(h.text, _S_MEMBER)) else: - segs.append(Segment(h.text, _S_UNK)) + base = {"code": _S_INSN, "data": _S_DATA}.get(h.kind, _S_UNK) + segs.extend(self._span_segments(h, base)) strip = Strip(segs) 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)) @@ -1368,6 +1145,12 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def action_undefine(self) -> None: self.post_message(EditItemRequested(self, "undef")) + 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)) @@ -1485,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]]] = {} @@ -1574,6 +1360,13 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True else: scroll_x = sx self._apply_scroll(min(max(scroll_y, 0), max(total - 1, 0)), max(scroll_x, 0)) + # `cursor` is reactive(repaint=False) and _apply_scroll only repaints via + # call_after_refresh, so a jump that lands in the SAME viewport (Esc back + # to another spot in the function already on screen) moved the cursor + # with nothing to redraw it — the pane kept showing the old highlight + # until the next keypress. Repaint here; _move does the same via + # _refresh_lines. + self.refresh() self._after_cursor_move() def get_loading_widget(self): # type: ignore[override] @@ -1611,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: @@ -1673,7 +1471,7 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True best, best_ea = i, e return best - # -- navigation (mirrors DisasmView) ---------------------------------- # + # -- navigation -------------------------------------------------------- # def _visible_height(self) -> int: return max(self.size.height, 1) @@ -2002,8 +1800,10 @@ class XrefsScreen(ModalScreen): BINDINGS = [Binding("escape", "close", "Close")] - def __init__(self, label: str, items: list[tuple[int, str]], + def __init__(self, label: str, items: list[tuple[object, str]], preselect: int = 0) -> None: + # payload is an int address, or (binary, address) for a caller in another + # project binary; dismiss() hands it back untouched. super().__init__() self._label = label self._items = items @@ -2041,9 +1841,15 @@ def _fuzzy(name: str, q: str): if not name: # defensive: never assume a symbol has a name return None nl = name.lower() + # Case-insensitive means BOTH sides: the name was lowered but the query + # wasn't, so a single capital could never match and any query containing one + # returned nothing at all. Invisible on lowercase C symbols (main, strlen), + # fatal on libraries that capitalise — "PEM_read_bio" found 0 of 10093 + # functions in libcrypto while the backend resolved it fine. + ql = q.lower() pos: list[int] = [] i = 0 - for ch in q: + for ch in ql: j = nl.find(ch, i) if j < 0: return None @@ -2051,9 +1857,9 @@ def _fuzzy(name: str, q: str): i = j + 1 span = pos[-1] - pos[0] score = -(span * 2.0) - pos[0] - len(name) * 0.01 - if q in nl: + if ql in nl: score += 50.0 - if nl.startswith(q): + if nl.startswith(ql): score += 100.0 return (score, tuple(pos)) @@ -2503,6 +2309,240 @@ 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. + + IDA's own answer to an unidentified file is a dialog; ours is this. Without + it the fallback is x86 at address 0, which doesn't fail — it analyses to + nothing, and you're left wondering why a firmware image has no functions. + + Returns ``{"processor": str, "base": int}``, or ``{}`` to load it the way + IDA would have anyway (that IS the right answer sometimes: our sniff only + recognises formats we're sure about, so it says "unknown" for things IDA + can in fact handle). + """ + + BINDINGS = [ + Binding("escape", "close", "Close"), + Binding("down,ctrl+n", "cursor_down", show=False), + Binding("up,ctrl+p", "cursor_up", show=False), + Binding("enter", "choose", show=False, priority=True), + ] + + def __init__(self, path: str, size: int = 0) -> None: + super().__init__() + self._path = path + # NOT self._size: that is Textual's own backing field for outer_size, and + # assigning an int to it crashes the layout with a bewildering + # "'int' object has no attribute 'region'" from deep inside _set_dirty. + self._nbytes = size + self._results: list[tuple[str, str]] = [] + + def compose(self) -> ComposeResult: + from .formats import PROCESSORS + self._all = list(PROCESSORS) + with Vertical(id="pal-box"): + yield Static(" unrecognised file \u2014 how should IDA load it?", + id="pal-title", markup=False) + yield Static(f" {os.path.basename(self._path)} ({self._nbytes:,} bytes) " + f"\u2014 no loader matched; without a processor IDA " + f"assumes x86 at 0", id="load-note", markup=False) + yield Input(placeholder="filter processors\u2026", id="pal-input") + yield OptionList(id="pal-list") + yield Input(placeholder="load address, e.g. 0x8000000 (blank = 0)", + id="load-base") + yield Static(" Enter accept \u00b7 Tab base address \u00b7 " + "Esc load as IDA would", id="load-help", markup=False) + + def on_mount(self) -> None: + self._apply("") + self.query_one("#pal-input", Input).focus() + + def focus_next(self, selector="*"): # type: ignore[override] + """Tab moves between the two things you TYPE into. + + DOM order would stop at the option list on the way, which is + arrow-driven and has nothing to type — and the address you meant to + enter goes into whichever box happened to have focus. Typing an address + into the processor filter is then taken as a processor name, IDA rejects + it, and the open fails; that is a bad enough outcome to be worth + overriding Tab for. + """ + inp = self.query_one("#pal-input", Input) + base = self.query_one("#load-base", Input) + (inp if self.focused is base else base).focus() + return self.focused + + def focus_previous(self, selector="*"): # type: ignore[override] + return self.focus_next() + + def on_input_changed(self, event: Input.Changed) -> None: + event.stop() + if event.input.id == "pal-input": + self._apply(event.value.strip()) + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self.action_choose() + + def _apply(self, query: str) -> None: + q = query.lower() + rows = [(name, desc) for name, desc in self._all + if not q or q in name.lower() or q in desc.lower()] + # An unlisted processor is still valid: IDA has 73 modules and this + # offers 20, so a typed name that matches nothing is taken literally + # rather than refused. + if not rows and q: + rows = [(query.strip(), "use this processor name as typed")] + self._results = rows + ol = self.query_one(OptionList) + ol.clear_options() + opts = [] + for name, desc in rows: + label = Text() + label.append(f" {name:<12}", _S_LABEL) + label.append(desc, _S_DIM) + opts.append(Option(label)) + ol.add_options(opts) + if rows: + ol.highlighted = 0 + self.query_one("#pal-title", Static).update( + f" unrecognised file \u2014 processor? ({len(rows)})") + + def action_cursor_down(self) -> None: + ol = self.query_one(OptionList) + if ol.option_count: + ol.highlighted = min((ol.highlighted or 0) + 1, ol.option_count - 1) + + def action_cursor_up(self) -> None: + ol = self.query_one(OptionList) + if ol.option_count: + ol.highlighted = max((ol.highlighted or 0) - 1, 0) + + def action_choose(self) -> None: + ol = self.query_one(OptionList) + i = ol.highlighted + if i is None or not (0 <= i < len(self._results)): + self.dismiss({}) + return + raw = self.query_one("#load-base", Input).value.strip() + base = 0 + if raw: + try: + base = int(raw, 0) + except ValueError: + self.query_one("#load-help", Static).update( + f" {raw!r} is not an address \u2014 try 0x8000000") + self.query_one("#load-base", Input).focus() + return + if base % 16: + # IDA's -b is in paragraphs, so an unaligned base can't be + # expressed and would quietly load somewhere else. + self.query_one("#load-help", Static).update( + f" {base:#x} must be 16-byte aligned") + self.query_one("#load-base", Input).focus() + return + self.dismiss({"processor": self._results[i][0], "base": base}) + + def action_close(self) -> None: + self.dismiss({}) + + class ProjectPalette(ModalScreen): """The project's binaries; Enter switches to one. Shows which are resident (a live worker, so switching is instant) vs cold (needs an open).""" @@ -2604,13 +2644,16 @@ class ConfirmScreen(ModalScreen): Binding("escape,n", "cancel", "Cancel"), ] - def __init__(self, message: str) -> None: + def __init__(self, message: str, note: str = "") -> None: super().__init__() self._message = message + self._note = note def compose(self) -> ComposeResult: with Vertical(id="confirm-box"): - yield Static(self._message, id="confirm-msg") + yield Static(self._message, id="confirm-msg", markup=False) + if self._note: + yield Static(self._note, id="confirm-note", markup=False) yield Static("[Enter/y] confirm [Esc/n] cancel", id="confirm-help") def action_confirm(self) -> None: @@ -3037,7 +3080,6 @@ class IdaTui(App): #left { width: 30%; min-width: 42; max-width: 44; border-right: solid $panel; } #func-table { height: 1fr; } #func-filter { dock: top; } - DisasmView { width: 1fr; padding: 0 1; } DecompView { width: 1fr; } ListingView { width: 1fr; padding: 0 1; } #panes.split ListingView { border-right: tall $panel-lighten-2; } @@ -3091,7 +3133,8 @@ class IdaTui(App): #xref-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } #xref-list { height: auto; max-height: 100%; } /* every #pal-box palette centres, not just the symbol one */ - SymbolPalette, StringsPalette, ProjectPalette { align: center middle; } + SymbolPalette, StringsPalette, ProjectPalette, + LoadOptionsScreen { align: center middle; } /* Give the stock Ctrl+P command palette side padding instead of full width; the input + results inherit this width (results is an overlay, so pin it). */ CommandPalette > Vertical { width: 80%; max-width: 120; } @@ -3101,6 +3144,18 @@ 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 + was clipped off the bottom, which read as "Tab does nothing". */ + LoadOptionsScreen #pal-list { max-height: 12; } + #load-base { border: none; height: 1; margin: 1 1 0 1; background: $panel; color: $text; } + #load-help { height: 1; padding: 0 1; color: $text-muted; } + #confirm-note { height: auto; padding: 0 1; color: $text-muted; } StructEditor { align: center middle; } #se-box { width: 90%; height: 84%; border: thick $accent; background: $panel; } #se-panes { height: 1fr; } @@ -3137,6 +3192,13 @@ class IdaTui(App): Binding("s", "toggle_split", "Split", show=False), 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), @@ -3148,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) -> 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. @@ -3158,6 +3220,13 @@ class IdaTui(App): self._states: dict[str, BinaryState] = {} self._pending_restore = None # entry to reopen after a switch self._goto_after_switch = None # cross-binary search hit to land on + self._hops: list[str] = [] # binaries a navigation crossed FROM + 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. # False = the user chose discard, or we already saved on the way out. self._save_on_exit: bool | None = None @@ -3172,6 +3241,17 @@ class IdaTui(App): open_path = project.refs[0].staged 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._pending_trace_line = None # step waiting on a re-decompile + self._trail_line_of: dict[int, int] = {} # ea -> pseudocode line + self._trail_span = None # ea span of that function + self._trail_eas: list[int] = [] # sorted keys of _trail_line_of + self._t = 0 # current timestamp in that trace self._do_keepalive = keepalive self._rpc_path = rpc_path self._rpc = None @@ -3187,7 +3267,9 @@ class IdaTui(App): self._filter_timer = None self._sort_col = 0 # 0=addr, 1=name, 2=size self._sort_reverse = False - self._pref = "listing" # unified: the linear listing is the code view + # ONE notion of "which pane you're in": _active, kept in step with focus + # (on_descendant_focus does that while split). There used to be a second, + # _pref, but it was only ever assigned "listing" — see _code_mode(). self._active = "listing" # currently shown view (in split: the focused pane) self._split = False # side-by-side listing + pseudocode self._split_eamap: list[list[int]] = [] # split: decomp line -> instr EAs @@ -3216,14 +3298,19 @@ class IdaTui(App): fp = FunctionsPanel(id="left") fp.display = False # overlay-first: reveal the docked pane with Ctrl+B yield fp - # The unified continuous listing is the one code view; DisasmView is - # deprecated (kept only for DisasmModel, still used by the domain). + # The unified continuous listing is the one code view. (DisasmModel + # is still used by the domain to index a function's instructions.) lst = ListingView() yield lst yield DecompView() 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 @@ -3263,13 +3350,192 @@ class IdaTui(App): # The names pane is an overlay now (Ctrl+N); focus the default code view # (pseudocode) so app bindings work before anything is opened. self.query_one(ListingView).focus() + if self._rpc_path: + self._start_rpc() + # A file no loader recognises has to be described before it can be + # opened, so ask BEFORE the worker starts — once IDA has made a database + # the answer is baked in and changing it means deleting the .i64. + if self._project is not None: + ref = self._pending_load_ref() + if ref is not None: + self._ask_load_options(ref.source, label=ref.label) + return + elif self._should_ask_load_options(): + self._ask_load_options(self._open_path) + return # Show a loading overlay immediately so a slow open/analysis (big binary) # isn't just dead air behind empty panes; dismissed once we land. self._loading_screen = LoadingScreen(self._loading_title()) self.push_screen(self._loading_screen) self._connect() - if self._rpc_path: - self._start_rpc() + + def _should_ask_load_options(self) -> bool: + """Ask only when nobody has already answered, and only when it matters. + + Skipped when: options came from the command line or the project (the + user already said); a database exists (the answer is recorded in it, and + re-passing switches fails the open); or the file is a format IDA + recognises, which is nearly always. + """ + if not self._open_path: + return False + if self._load_args: + return False + from .formats import needs_load_options + if os.path.exists(self._open_path + ".i64") or os.path.exists( + os.path.splitext(self._open_path)[0] + ".i64"): + return False + return needs_load_options(self._open_path) + + def action_load_options(self) -> None: + """Ctrl+L: re-open this binary with different load options. + + The database IDA already built has the old processor and base baked into + it and takes precedence over any switches, so re-loading means throwing + it away. That destroys names and comments, hence the confirmation — but + for the case this exists for (a blob loaded as the wrong architecture, + which analysed to nothing) there is nothing to lose and no other way + forward. + """ + if not self._can_reload(): + self._status("nothing to reload") + return + 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} " + 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) + + def _on_reload_confirmed(self, yes) -> None: # type: ignore[no-untyped-def] + if not yes: + return + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Drop the worker first: it holds the database open, and the .i64 can't + # be removed (or rebuilt) underneath a live one. + self._release_worker() + self._drop_database() + self._reset_for_reload() + self._load_args = "" + if label is not None and self._project is not None: + self._project.set_load(label, processor="", base=0) + i = self._project._refs.index(self._project.by_label(label)) + self._project._entries[i].pop("processor", None) + self._project._entries[i].pop("base", None) + self._project.save() + self._pending_switch = None + self._ask_load_options(path, label=label) + + def _release_worker(self) -> None: + if self._pool is not None and self._binary is not None: + try: + self._pool.evict(self._binary, save=False) + except Exception: # noqa: BLE001 + pass + elif self.client is not None: + try: + self.client.close() + except Exception: # noqa: BLE001 + pass + self.client = None + self.program = None + + def _drop_database(self) -> None: + """Remove the .i64 (and any unpacked scratch) so the next open re-reads + the raw image with new options.""" + base = self._open_path + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + base = ref.staged + if not base: + return + for suffix in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + for cand in (base + suffix, os.path.splitext(base)[0] + suffix): + try: + os.remove(cand) + except OSError: + pass + + def _reset_for_reload(self) -> None: + self._no_functions = False + self._func_index = None + self._cur = None + self._nav = [] + self._did_auto_land = False + self._pending_restore = None + self._split = False + self.query_one(DecompView).loaded_ea = None + + def _retry_load_options(self) -> None: + """Re-ask after IDA refused what we told it.""" + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Clear the rejected answer or _pending_load_ref would see the + # binary as already described and never ask again. + self._project.set_load(label, processor="", base=0) + self._project._entries[self._project._refs.index(ref)].pop( + "processor", None) + self._project.save() + self._load_args = "" + if path: + self._status("those load options were rejected \u2014 try again") + self._ask_load_options(path, label=label) + + def _pending_load_ref(self, label: str | None = None): # type: ignore[no-untyped-def] + """The project binary about to be opened, if it needs describing. + + Checked against the SOURCE: staging may not have happened yet, and the + question is about the bytes, not where they were copied to. + """ + if self._project is None: + return None + label = label or self._binary or self._project.refs[0].label + ref = self._project.by_label(label) + if ref is None or ref.load_args: + return None + if os.path.exists(ref.db) or os.path.exists( + os.path.splitext(ref.staged)[0] + ".i64"): + return None # already analysed: the .i64 records how + from .formats import needs_load_options + return ref if needs_load_options(ref.source) else None + + def _ask_load_options(self, path: str, label: str | None = None) -> None: + try: + size = os.path.getsize(path) + except OSError: + size = 0 + self._load_for_label = label + self.push_screen(LoadOptionsScreen(path, size), self._on_load_options) + + def _on_load_options(self, choice) -> None: # type: ignore[no-untyped-def] + from .formats import load_args + choice = choice or {} + label, self._load_for_label = self._load_for_label, None + proc, base = choice.get("processor", ""), int(choice.get("base", 0) or 0) + if proc: + if label is not None and self._project is not None: + # Persist it: the answer belongs to the binary, not to this run. + self._project.set_load(label, proc, base) + else: + self._load_args = load_args(proc, base) + self._status(f"loading as {proc} @ {base:#x}") + if self._pending_switch is not None: + label2, self._pending_switch = self._pending_switch, None + self._switch_binary(label2) + return + self._loading_screen = LoadingScreen(self._loading_title()) + self.push_screen(self._loading_screen) + self._connect() def _loading_title(self) -> str: return os.path.basename(self._open_path) if self._open_path else "database" @@ -3298,9 +3564,39 @@ 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: self.query_one("#status", Static).update(text) except Exception: # noqa: BLE001 -- status bar transiently unavailable @@ -3372,7 +3668,8 @@ class IdaTui(App): self.app.call_from_thread(self._reconnect_failed, "no binary to reopen") return - client = WorkerClient(self._open_path, ttl=self._ttl) + client = WorkerClient(self._open_path, ttl=self._ttl, + load_args=self._load_args) client.connect(progress=lambda m: self.app.call_from_thread( self._conn_note, m)) except Exception as e: # noqa: BLE001 @@ -3412,6 +3709,12 @@ class IdaTui(App): except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"connect failed: {e}") self.app.call_from_thread(self._dismiss_loading) + # A load we described ourselves and IDA refused: offer the dialog + # again rather than leaving an empty app with an error in the status + # bar. Getting the processor wrong is an ordinary mistake and should + # cost one more keypress, not a restart. + if "load options" in str(e): + self.app.call_from_thread(self._retry_load_options) return self.client = client self.program = program @@ -3429,6 +3732,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( @@ -3438,7 +3742,8 @@ class IdaTui(App): base = os.path.basename(self._open_path) self.app.call_from_thread( self._status, f"starting worker — initial auto-analysis of {base}…") - client = WorkerClient(self._open_path, ttl=self._ttl) + client = WorkerClient(self._open_path, ttl=self._ttl, + load_args=self._load_args) client.connect(progress=lambda m: self.app.call_from_thread( self._status, m)) return client @@ -3450,7 +3755,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) @@ -3458,18 +3762,55 @@ 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: + """Warm the binary this one leans on hardest, once we're idle. + + Not "the next in the list" — phase 3 tells us something better. The + binary providing the most of this one's imports is where a follow is + most likely to take you, so paying its startup now is the switch you + would otherwise wait for. Refuses to evict anything (see pool.prewarm), + so at a tight budget this simply does nothing. + """ + if self._pool is None or self._index is None or self._binary is None: + return + try: + imps, _ = self.program.linkage() + except Exception: # noqa: BLE001 + return + if not imps: + return + from collections import Counter + votes: Counter = Counter() + for name in {i.name for i in imps}: + for h in self._index.providers(name, exclude=self._binary): + votes[h.binary] += 1 + resident = set(self._pool.resident()) + cand = next((b for b, _ in votes.most_common() if b not in resident), None) + if cand is None: + return + n = votes[cand] + try: + if self._pool.prewarm(cand): + self.app.call_from_thread( + self._status, f"pre-warmed {cand} (provides {n} imports)") + except Exception: # noqa: BLE001 -- speculative work must never surface + pass @work(thread=True, exclusive=True, group="index") def _index_binary(self) -> None: @@ -3480,7 +3821,7 @@ class IdaTui(App): ref = self._project.by_label(self._binary) if ref is None or not self._index.is_stale(self._binary, ref.source): return - from .index import KIND_FUNC, KIND_STRING + from .index import KIND_EXPORT, KIND_FUNC, KIND_IMPORT, KIND_STRING idx = self._func_index entries = [(KIND_FUNC, f.addr, f.name) for f in (idx.all_loaded() if idx else [])] try: @@ -3489,12 +3830,19 @@ class IdaTui(App): except Exception: # noqa: BLE001 -- symbols alone are still worth indexing pass try: + imps, exps = self.program.linkage() + entries += [(KIND_IMPORT, i.addr, i.name) for i in imps] + entries += [(KIND_EXPORT, e.addr, e.name) for e in exps] + except Exception: # noqa: BLE001 -- an old worker has no list_linkage + pass + try: n = self._index.reindex(self._binary, entries, source=ref.source) except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"indexing failed: {e}") return self.app.call_from_thread( - self._status, f"indexed {self._binary}: {n} symbols + strings") + self._status, f"indexed {self._binary}: {n} symbols, strings + linkage") + self._prewarm_provider() # -- initial landing --------------------------------------------------- # #: function names tried (in order) as the startup landing spot @@ -3526,8 +3874,42 @@ class IdaTui(App): fn = self._entry_func() if fn is not None: self._open_function(fn.addr, fn.name) - else: + elif len(self._func_index): self.action_symbols() + else: + self._land_without_functions() + + def _land_without_functions(self) -> None: + """Analysis found nothing. Show the bytes and say so. + + Falling through to the symbol picker here left two empty panes and + "functions still loading…" — which is a lie, loading had finished. There + is always something to look at: the segments exist even when IDA + recognised no code in them, so open the listing at the start of the image. + + Zero functions is also the signal that a blob was described wrongly. It's + exactly what a good image loaded as the wrong processor looks like, so + the status says so rather than leaving you to guess. + """ + start = None + try: + regions = self.program.file_regions() + if regions: + start = regions[0][0] + except Exception: # noqa: BLE001 + pass + self._no_functions = self._can_reload() + if start is None: + self._status("no functions and no segments \u2014 nothing to show") + return + self._open_at(start, self.program.section_of(start) or "image", + cursor=0, push=True, is_region=True) + + def _can_reload(self) -> bool: + """Whether we're able to re-open this binary with different options.""" + if self._project is not None and self._binary is not None: + return True + return bool(self._open_path) def _entry_func(self) -> Func | None: """The best startup landing function (exact-name match against @@ -3604,7 +3986,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 @@ -3674,7 +4056,15 @@ class IdaTui(App): def _switch_then_goto(self, binary: str, addr: int) -> None: """A project-wide hit in another binary: switch to it, then jump there - once its index has loaded.""" + once its index has loaded. + + Records the hop so Esc can come back. Nav history is per-binary, so + without this a cross-binary jump — a project search hit, or following an + import into the library that implements it — is a one-way door: you + arrive in a binary whose history is empty and nothing takes you back. + """ + if self._binary is not None and binary != self._binary: + self._hops.append(self._binary) self._goto_after_switch = addr self._switch_binary(binary) @@ -3754,13 +4144,20 @@ class IdaTui(App): self._switch_binary(label) def _switch_binary(self, label: str) -> None: + # A blob nobody has described yet has to be described before its worker + # opens it — same as at boot, just reached by switching instead. + ref = self._pending_load_ref(label) + if ref is not None and self._load_for_label is None: + self._pending_switch = label + self._ask_load_options(ref.source, label=label) + return # Snapshot what we're leaving so coming back restores the view, then let # the pool hand us a worker (spawning + evicting as the budget dictates). if self._binary is not None: self._states[self._binary] = BinaryState( label=self._binary, program=self.program, func_index=self._func_index, nav=list(self._nav), cur=self._cur, - pref=self._pref, active=self._active, split=self._split, + active=self._active, split=self._split, filter_term=self._filter_term, dirty=self._dirty) self._loading_screen = LoadingScreen(label, note="switching\u2026") self.push_screen(self._loading_screen) @@ -3791,7 +4188,7 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged - self._pref = st.pref if st else "listing" + 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 "" @@ -3871,9 +4268,41 @@ class IdaTui(App): return self._goto_ea(addr, push=True) # land on the literal in the listing + def on_descendant_focus(self, event) -> None: # type: ignore[no-untyped-def] + """Keep ``_active`` in step with focus while split. + + Tab moves both together, but focus also moves on its own — a click, or a + pane focusing itself after a load — and then ``_active`` still names the + pane you're NOT in. Everything downstream trusts ``_active``: follow + resolves the word under that pane's cursor and pushes history for it, so + Enter in the pseudocode would follow something from the listing and the + next Esc got spent undoing it. + """ + if not self._split: + return + w = self.focused + mode = ("decomp" if isinstance(w, DecompView) + else "listing" if isinstance(w, ListingView) else None) + if mode is None or mode == self._active: + return + self._active = mode + self._sync_split(mode) # re-link the band from the new driver + if not self.query_one(DecompView).loading: + self._status_for_cur("split") # never clobber "decompiling…" + def action_toggle_view(self) -> None: """Tab: switch the code pane between disassembly and pseudocode (or leave the hex view back to the preferred code view).""" + # Tab is a PRIORITY app binding, so it fires even while a modal is up and + # nothing inside a dialog could ever be tabbed to. Hand it back to the + # dialog: this is the only reason the load dialog's address field was + # unreachable, and it was broken the same way in every other modal. + if self.screen is not self.screen_stack[0]: + try: + self.screen.focus_next() + except Exception: # noqa: BLE001 -- screen with nothing focusable + pass + return if self._cur is None: return if self._split: @@ -4067,8 +4496,19 @@ class IdaTui(App): self.clear_filter() # Esc on the list clears an active filter return if len(self._nav) > 1: + self._nav_seq += 1 # invalidate any navigation still in flight self._nav.pop() - self._open_entry(self._nav[-1], push=False) + # Say something immediately: going back can need a decompile, and + # until it lands the panes still show where you were — with no + # feedback Esc reads as a no-op. + dest = self._nav[-1] + self._status(f"\u25c2 back to {dest.name}\u2026") + self._open_entry(dest, push=False) + elif self._hops: + # Local history is spent, but we got here from another binary. + label = self._hops.pop() + self._status(f"\u25c2 back to {label}\u2026") + self._switch_binary(label) # _states restores its nav and position elif self.query_one("#left", FunctionsPanel).display: table.focus() else: @@ -4091,17 +4531,12 @@ class IdaTui(App): def on_follow_requested(self, msg: FollowRequested) -> None: view = msg.view word = view.word_under_cursor() - if isinstance(view, DisasmView): - ea = view._cursor_ea() - # The instruction's ordinary fall-through edge (to the next line) is - # an indistinguishable 'code' xref; pass it so follow can skip it and - # land on a call/jump's real target instead of the next instruction. - nxt = view.model.cached_line(view.cursor + 1) if view.model else None - if ea is not None: - self._follow_disasm(ea, word, nxt.ea if nxt else None) - elif isinstance(view, ListingView): + if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: + # _next_ea() is the following item: follow uses it to skip the + # ordinary fall-through edge, which is an indistinguishable + # 'code' xref, and land on a call/jump's real target instead. self._follow_disasm(ea, word, view._next_ea()) elif isinstance(view, DecompView) and view._texts: self._follow_decomp(view._texts[view.cursor], word, @@ -4112,17 +4547,11 @@ class IdaTui(App): return view = msg.view word = view.word_under_cursor() - if isinstance(view, DisasmView): + if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: # span = this instruction .. the next, to pre-select the dialog # entry for the site we invoked xrefs from. - nxt = view.model.cached_line(view.cursor + 1) if view.model else None - self._push_busy("finding xrefs\u2026") - self._xrefs_disasm(ea, word, ea, nxt.ea if nxt else None) - elif isinstance(view, ListingView): - ea = view._cursor_ea() - if ea is not None: self._push_busy("finding xrefs\u2026") self._xrefs_disasm(ea, word, ea, view._next_ea()) elif isinstance(view, DecompView) and view._texts: @@ -4186,8 +4615,55 @@ class IdaTui(App): if tgt is None or tgt.to is None: self.app.call_from_thread(self._status, "nothing to follow here") return + if self._follow_import(tgt.to): + return self._do_navigate(tgt.to, push=True) + def _import_stub(self, ea: int) -> str | None: + """The import name at ``ea``, if ``ea`` is one of this binary's import + stubs. That's the dead end phase 3 exists to open up: a call to strcmp + reaches the PLT/extern entry and Hex-Rays has nothing to decompile, + because the code lives in a library this binary only references.""" + try: + imps, _ = self.program.linkage() + except Exception: # noqa: BLE001 + return None + for i in imps: + if i.addr == ea: + return i.name + return None + + def _cross_binary_impl(self, name: str) -> tuple[str, int] | None: + """``(binary, addr)`` of a project binary that EXPORTS ``name``. + + Reads the on-disk index, so a provider resolves even when its worker was + evicted — the whole reason the index exists. + """ + if self._index is None or self._project is None or not name: + return None + try: + hits = self._index.providers(name, exclude=self._binary) + except Exception: # noqa: BLE001 + return None + return (hits[0].binary, hits[0].addr) if hits else None + + def _follow_import(self, ea: int) -> bool: + """Follow an import stub into the binary that implements it. True when + it was handled (caller must not also navigate locally).""" + name = self._import_stub(ea) + if not name: + return False + found = self._cross_binary_impl(name) + if found is None: + # Leave the local navigation alone: landing on the stub is still the + # honest answer when nothing in the project provides the symbol. + return False + label, addr = found + self.app.call_from_thread( + self._status, f"{name} \u2192 {label} (import resolved)") + self.app.call_from_thread(self._switch_then_goto, label, addr) + return True + @work(thread=True, group="nav") def _follow_decomp(self, line: str, word: str | None, line_ea: int | None = None) -> None: @@ -4215,6 +4691,8 @@ class IdaTui(App): if addr is None: self.app.call_from_thread(self._status, "nothing to follow on this line") return + if self._follow_import(addr): + return # Following FROM pseudocode: keep the reader in the decompiler when the # target is decompilable, instead of dropping to the linear listing. self._do_navigate(addr, push=True, prefer_decomp=True) @@ -4320,9 +4798,45 @@ class IdaTui(App): kind = x.kind or x.type or "?" items.append((x.frm, f"{x.frm:08X} {kind:<6} {loc}")) preselect = self._xref_preselect(xr, here_ea, here_end) + # Callers in OTHER project binaries. xrefs_to only ever sees this + # database, so an exported function looks unused from the inside even + # when half the project calls it — the import side of the linkage index + # is the only place that knowledge exists. + for label_b, addr_b, text_b in self._foreign_importers(subj, subj_name, fn): + items.append(((label_b, addr_b), text_b)) self.app.call_from_thread(self._present_xrefs, label, items, focus, preselect) - def _present_xrefs(self, label: str, items: list[tuple[int, str]], + def _foreign_importers(self, subj: int, subj_name, fn): # type: ignore[no-untyped-def] + """Project binaries that IMPORT the symbol at ``subj`` — the other half + of the phase-3 join, read from the on-disk index so a caller shows up + whether or not its worker is resident. + + Only for a symbol this binary actually exports: a local name that + happens to collide with another binary's import isn't a caller of ours. + """ + if self._index is None or self._project is None or self._binary is None: + return [] + name = subj_name if self._looks_like_symbol(subj_name) else None + if name is None and fn is not None and fn.addr == subj: + name = fn.name + if not name: + return [] + from .domain import link_name + name = link_name(name) + try: + _, exports = self.program.linkage() + except Exception: # noqa: BLE001 + return [] + if not any(e.name == name for e in exports): + return [] # we don't export it; nobody imports it FROM US + try: + hits = self._index.importers(name, exclude=self._binary) + except Exception: # noqa: BLE001 + return [] + return [(h.binary, h.addr, f"{h.addr:08X} import [{h.binary}] {name}") + for h in hits] + + def _present_xrefs(self, label: str, items: list[tuple[object, str]], focus_name: str | None = None, preselect: int = 0) -> None: if not self._xref_active: return # cancelled (Esc) while we were still gathering @@ -4335,12 +4849,17 @@ class IdaTui(App): self._status(f"{label}: {len(items)}") self.push_screen(XrefsScreen(label, items, preselect), self._on_xref_chosen) - def _on_xref_chosen(self, addr: int | None) -> None: - if addr is not None: - # If xrefs was invoked from the decompiler, land the jump back in the - # decompiler (when the target is decompilable) rather than the listing. - self._goto_ea(addr, push=True, focus_name=self._xref_focus_name, - prefer_decomp=(self._active == "decomp")) + def _on_xref_chosen(self, addr) -> None: # type: ignore[no-untyped-def] + if addr is None: + return + if isinstance(addr, tuple): # a caller in another project binary + binary, ea = addr + self._switch_then_goto(binary, ea) # records a hop, so Esc returns + return + # If xrefs was invoked from the decompiler, land the jump back in the + # decompiler (when the target is decompilable) rather than the listing. + self._goto_ea(addr, push=True, focus_name=self._xref_focus_name, + prefer_decomp=(self._active == "decomp")) # -- rename ------------------------------------------------------------ # @staticmethod @@ -4424,7 +4943,7 @@ class IdaTui(App): # -- comments ---------------------------------------------------------- # def _line_ea_for(self, view) -> int | None: # type: ignore[no-untyped-def] """Address of the line under the cursor in either code view.""" - if isinstance(view, (DisasmView, ListingView)): + if isinstance(view, ListingView): return view._cursor_ea() if isinstance(view, DecompView): return view._line_ea(view.cursor) @@ -4513,13 +5032,17 @@ class IdaTui(App): dec.loaded_ea = None # force re-decompile self._show_active() else: - # Capture the LIVE listing position straight from the widget (the - # source of truth) rather than trusting nav-entry tracking, which can - # go stale. bump_names() discards the segment model, so the reload - # rebuilds it; feeding the true cursor/scroll keeps the edited line on - # screen even on a huge segment (otherwise it primes/scrolls to a - # stale index and the renamed line lands off-screen — looking like the - # rename never applied). + # Capture the LIVE position from the widget (the source of truth) + # rather than trusting nav-entry tracking, which goes stale. Capture + # it as ADDRESSES via the anchor: bump_names() discards the segment + # model so the reload rebuilds it, and an edit that changes how many + # rows an item takes makes the old indices point somewhere else. + # Index capture is CORRECT here and an anchor is not: a rename or + # comment doesn't change how many rows anything takes, and the model + # this rebuilds is constructed empty — index_of_ea on it returns -1 + # until pages load, so an anchor would resolve to nothing while + # costing an extra model build on the UI thread. Address anchoring is + # for the edit paths that DO change row structure (see _do_edit_item). lst = self.query_one(ListingView) cur.view = "listing" if lst.model is not None: @@ -4672,7 +5195,8 @@ class IdaTui(App): view.focus() @work(thread=True, exclusive=True, group="makedata") - def _do_make_data(self, ea: int, type_decl: str) -> None: # worker context + def _do_make_data(self, ea: int, type_decl: str, + anchor: ViewAnchor | None = None) -> None: # worker context assert self.program is not None try: self.program.make_data(ea, type_decl) @@ -4680,12 +5204,15 @@ class IdaTui(App): self.app.call_from_thread(self._status, f"make data: {e}") return self.program.bump_items() + anchor = anchor or ViewAnchor() + anchor.flash = f"data ({type_decl}) @ {ea:#x} (Ctrl+S to save)" name = self.program.region_label(ea) lm = self.program.listing(ea) idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 - self.app.call_from_thread(self._open_at, ea, name, idx, False, -1, 0, True) + _cur, top = self._anchor_rows(anchor, lm, ea) self.app.call_from_thread( - self._edit_item_done, f"data ({type_decl})", ea) + self._open_at, ea, name, idx, False, -1, 0, True, None, top) + self.app.call_from_thread(self._edit_done, anchor) @work(thread=True, exclusive=True, group="rename") def _do_rename(self, view, old: str, new: str) -> None: # type: ignore[no-untyped-def] @@ -4794,27 +5321,91 @@ class IdaTui(App): # -- item structure edits (IDA c/p/u) --------------------------------- # def on_edit_item_requested(self, msg: EditItemRequested) -> None: - ea = (msg.view._cursor_ea() - if isinstance(msg.view, (DisasmView, ListingView)) else None) + view = msg.view + ea = view._cursor_ea() if isinstance(view, ListingView) else None if ea is None: self._status("no address on this line to (re)define") return - self._do_edit_item(msg.kind, ea) + self._do_edit_item(msg.kind, ea, self._anchor()) @work(thread=True, exclusive=True, group="edititem") - def _do_edit_item(self, kind: str, ea: int) -> None: # worker context + def _do_edit_item(self, kind: str, ea: int, + anchor: ViewAnchor | None = None) -> None: # worker context assert self.program is not None verb = {"code": "defined code", "func": "created function", - "undef": "undefined", "string": "made string"}[kind] + "undef": "undefined", "string": "made string", + "thumb": "switched decoding", "thumbscan": "scanned"}[kind] try: if kind == "code": - self.program.define_code(ea) + # Keep going until something stops it: one instruction is rarely + # what you want, and on a raw image it means pressing `c` once + # per opcode for the length of a function. + r = self.program.define_code_run(ea) + n, why = int(r.get("count", 0)), r.get("stopped", "") + if n == 0 and why == "defined": + # Already code/data here — a no-op, not a failure. Saying + # "failed to create instruction" for it would be a lie. + self.app.call_from_thread( + self._status, f"already defined @ {ea:#x}") + return + if n == 0: + raise IDAToolError("define_code", + f"@ {ea:#x}: Failed to create instruction") + end = int(str(r.get("end", hex(ea))), 0) + reason = {"undecodable": "hit bytes that don't decode", + "flow": "control flow ends here", + "defined": "ran into existing code/data", + "segment": "end of segment", + "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 + # flipped it was to read the code. + r = self.program.set_thumb(ea) + run = self.program.define_code_run(ea) + n = int(run.get("count", 0)) + mode = "Thumb" if r.get("thumb") else "ARM" + verb = f"{mode} @ {ea:#x}" + if r.get("forced_32bit"): + verb += " (segment set to 32-bit; Thumb needs ARM32)" + if r.get("db_64bit"): + # Disassembly will look right and F5 will never work. + verb += (" \u26a0 this database is 64-bit, so Hex-Rays " + "won't decompile it \u2014 Ctrl+L and pick " + "arm:ARMv7-A") + verb += (f" \u2014 {n} instruction{'s' if n != 1 else ''}" + if n else " \u2014 still doesn't decode") + # falls through to the shared reload: same cache bump, same + # anchor restore, same flash. That is the whole point of having + # one path. elif kind == "func": - self.program.define_func(ea) + 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']}" + + (" (end worked out from the code)" + if r.get("how") == "explicit-end" else "")) elif kind == "string": 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}") @@ -4823,23 +5414,304 @@ class IdaTui(App): self.program.bump_items() # Re-resolve: a define_func upgrades the region to a real function view; # anything else re-reads the (still function-less) listing in place. + anchor = anchor or ViewAnchor() + anchor.flash = f"{verb} @ {ea:#x} (Ctrl+S to save)" fn = self.program.function_of(ea) if fn is not None: model = self.program.disasm(fn.addr, fn.name) idx = 0 if ea == fn.addr else model.index_of_ea(ea) + _cur, top = self._anchor_rows(anchor, model, ea) self.app.call_from_thread( - self._open_at, fn.addr, fn.name, idx, False, -1, 0, False) + self._open_at, fn.addr, fn.name, idx, False, -1, 0, False, + None, top) else: name = self.program.region_label(ea) lm = self.program.listing(ea) idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 + _cur, top = self._anchor_rows(anchor, lm, ea) self.app.call_from_thread( - self._open_at, ea, name, idx, False, -1, 0, True) - self.app.call_from_thread(self._edit_item_done, verb, ea) + self._open_at, ea, name, idx, False, -1, 0, True, None, top) + self.app.call_from_thread(self._edit_done, anchor) - def _edit_item_done(self, verb: str, ea: int) -> None: + def _edit_done(self, anchor: ViewAnchor) -> None: + """One place where an edit's aftermath is settled. + + The reload this edit triggered will write its own status when it lands — + after this — so the message is handed over as a flash rather than + written and lost. + """ self._dirty = True - self._status(f"{verb} @ {ea:#x} (Ctrl+S to save)") + if 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 not follow: + return + pc = t.ip(self._t) + if self._split and self._seek_split(pc): + return + # Stay in whichever view you're reading. Without prefer_decomp a step + # from the pseudocode navigates to an address, which opens the listing — + # so stepping through C threw you out of C on the first keypress. + self._goto_ea(pc, push=False, + prefer_decomp=(self._active == "decomp")) + + def _seek_split(self, pc: int) -> bool: + """Put BOTH panes on ``pc``. True if handled. + + Normal navigation moves one pane and gives the companion a band, never a + cursor — that rule exists so the two can't chase each other. A trace step + isn't navigation though: time is a single global position, and both views + are showing the same instant, so both cursors belong on it. + + The scroll anchoring is unchanged: after placing the cursors, the usual + _sync_split still bands the companion and aligns it to the driver's + screen row, so the eye tracks straight across. + """ + lst = self.query_one(ListingView) + dec = self.query_one(DecompView) + if lst.model is None: + return False + row = lst.model.ensure_ea(pc) + if row is None or row < 0: + return False # not in this listing (other segment): full nav + lst.cursor = row + lst._scroll_cursor_into_view() + + # Has execution actually left the decompiled function? Ask the map the + # trail painting keeps, which is keyed to what the decompiler currently + # HOLDS. _split_range comes from the guarded async path and lags, so a + # stale one made every step look like a function change: the decompiler + # bounced main -> PLT stub -> main, each bounce costing a synchronous + # 769-line map fetch on the UI thread. + span = self._trail_span + inside = (pc in self._trail_line_of + or (span is not None and span[0] <= pc <= span[1])) + if not inside: + self._pending_trace_line = pc + self._resync_decomp_async(pc) + return True + self._place_decomp_at(pc) + self._sync_split(self._active) + return True + + def _place_decomp_at(self, pc: int) -> None: + """Move the pseudocode cursor to the line covering ``pc``. + + Uses the map the trail painting already keeps (keyed to the decompiler's + CURRENTLY loaded function), not the split view's _split_ea2line. That one + is refreshed by a guarded async path — it drops a result if _cur moved + while it was in flight — and a burst of steps moves _cur constantly, so + during stepping it is frequently a map of the function you just left. + """ + dec = self.query_one(DecompView) + line = None + if self._trail_map_ea == dec.loaded_ea and self._trail_line_of: + # EXACT match only. The decompiler doesn't attribute every + # instruction to a line (about half of main's aren't), and the + # tempting fallback — the nearest mapped instruction at or before + # the pc — is unsound: C lines are not monotonic in address, so + # 0x24a8 early in main resolved to line 708, "sub_2040();", near the + # end. A cursor that jumps to an unrelated statement is worse than + # one that waits; the trail still marks where we are. + line = self._trail_line_of.get(pc) + if line is None: + line = self._split_ea2line.get(pc) + if line is None: + line = dec.line_for_ea(pc) + if line is not None: + dec.goto(line, dec.cursor_x) + + @work(thread=True, exclusive=True, group="split-resync") + def _resync_decomp_async(self, ea: int) -> None: + self._resync_decomp(ea) + + 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: + # One index, built once per decompiled function and shared with the + # split view (_apply_split_map fills the same fields). decomp_map is + # an RPC and stepping is interactive, so paying it per keystroke — + # or twice, once for each of two parallel maps — would be felt. + try: + self._apply_split_map(ea, self.program.decomp_map(ea)) + except Exception: # noqa: BLE001 + self._trail_map, self._trail_map_ea = [], ea + self._trail_line_of, self._trail_eas = {}, [] + self._trail_span = None + 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() + pend, self._pending_trace_line = self._pending_trace_line, None + if pend is not None and self._split: + # The function was still decompiling when the step happened; land + # now that its line map exists. + self._place_decomp_at(pend) + self._sync_split(self._active) + + 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: @@ -4867,6 +5739,10 @@ class IdaTui(App): def _do_navigate(self, ea: int, push: bool, focus_name: str | None = None, prefer_decomp: bool = False) -> None: # worker context assert self.program is not None + # Which navigation this is. Decompiling below can take a while, and if + # you press Esc (or jump again) in the meantime this result is stale — + # applying it anyway silently undoes what you just did. + seq = self._nav_seq fn = self.program.function_of(ea) # Jumping FROM the decompiler: stay in pseudocode when the target is a # decompilable function, landing on the line that matches ``ea``. @@ -4886,7 +5762,8 @@ class IdaTui(App): dec_idx, col = self._decomp_locate(fn.addr, ea, focus_name or fn.name) self.app.call_from_thread( - self._open_decomp_entry, fn.addr, fn.name, dec_idx, col, push) + self._open_decomp_entry, fn.addr, fn.name, dec_idx, col, + push, seq) return # Otherwise: everything opens the one continuous listing at ``ea``. A # function name is used for the status label; a region gets a segment @@ -4898,9 +5775,17 @@ class IdaTui(App): self._open_at, ea, name, idx, push, -1, 0, fn is None, focus_name) def _open_decomp_entry(self, fn_addr: int, fn_name: str, dec_idx: int, - dec_cursor_x: int, push: bool) -> None: + dec_cursor_x: int, push: bool, + seq: int | None = None) -> None: """Open ``fn_addr`` in the decompiler as a real navigation (nav history - aware), landing on pseudocode line ``dec_idx`` column ``dec_cursor_x``.""" + aware), landing on pseudocode line ``dec_idx`` column ``dec_cursor_x``. + + ``seq`` is the navigation this result belongs to; if the user has + navigated since (Esc, another jump), the result is stale and dropped — + otherwise a slow decompile lands afterwards and undoes their Esc. + """ + if seq is not None and seq != self._nav_seq: + return if push: # Snapshot where we jumped from so 'back' returns there. If that was # the pseudocode (F5 makes a transient _cur not yet on the stack), @@ -4913,14 +5798,14 @@ class IdaTui(App): src.dec_cursor_x = dv.cursor_x src.dec_scroll_y = round(dv.scroll_offset.y) if not self._nav or self._nav[-1] is not src: - self._nav.append(src) + self._push_nav(src) # never stack a second copy of a spot else: self._save_current_pos() self._decomp_return = None # a real navigation abandons the F5 return entry = NavEntry(ea=fn_addr, name=fn_name, view="decomp", dec_cursor=dec_idx, dec_cursor_x=dec_cursor_x) if push: - self._nav.append(entry) + self._push_nav(entry) self._open_entry(entry, push=False) def _decomp_col_for(self, fn_addr: int, line_idx: int, name: str) -> int: @@ -4992,9 +5877,78 @@ class IdaTui(App): best_ea, best_idx = e, i return best_idx + @staticmethod + def _same_spot(a: NavEntry, b: NavEntry) -> bool: + """Same function, same view, same line — i.e. going from a to b is not a + move, so it has no business being a step in the history.""" + if a.ea != b.ea or a.view != b.view: + return False + return (a.dec_cursor == b.dec_cursor if a.view == "decomp" + else a.cursor == b.cursor) + + def _push_nav(self, entry: NavEntry) -> None: + """Append to the nav stack unless that would duplicate where we already are. + + Opening the place you're standing on — the app auto-lands on main, then + you pick main out of Ctrl+N — used to append an identical entry. The Esc + it created popped the stack without changing anything on screen: a dead + keypress, which is exactly what "back doesn't work" feels like. Replace + the top instead, so the newer entry's metadata still wins. + """ + top = self._nav[-1] if self._nav else None + if top is not None and self._same_spot(top, entry): + self._nav[-1] = entry + return + self._nav.append(entry) + + # -- view state across a model rebuild --------------------------------- # + def _anchor(self, flash: str | None = None) -> ViewAnchor: + """Capture where we're looking, BEFORE an edit rebuilds the model. + + Must run on the UI thread: it reads live widget state. + """ + a = ViewAnchor(view=self._active, flash=flash) + view = self._active_code_view() + model = getattr(view, "model", None) + if view is None or model is None: + return a + a.cursor_x = getattr(view, "cursor_x", 0) + try: + a.ea = view._cursor_ea() + except Exception: # noqa: BLE001 + a.ea = None + top = round(view.scroll_offset.y) + h = model.cached_line(top) or model.get(top) + a.top_ea = getattr(h, "ea", None) + return a + + @staticmethod + def _anchor_rows(a: ViewAnchor, model, fallback_ea: int | None = None): + """(cursor_row, top_row) for ``a`` in a freshly built ``model``. + + -1 means "no opinion" — the caller's own default wins. + """ + if model is None: + return (-1, -1) + + def row_of(ea): + if ea is None: + return -1 + try: + i = model.index_of_ea(ea) + except Exception: # noqa: BLE001 + return -1 + return i if i >= 0 else -1 + + cur = row_of(a.ea if a.ea is not None else fallback_ea) + if cur < 0 and fallback_ea is not None: + cur = row_of(fallback_ea) + return (cur, row_of(a.top_ea)) + def _open_at(self, ea: int, name: str, cursor: int, push: bool, dec_cursor: int = -1, dec_cursor_x: int = 0, - is_region: bool = False, focus_name: str | None = None) -> None: + is_region: bool = False, focus_name: str | None = None, + scroll_y: int = -1) -> None: if push: self._save_current_pos() self._decomp_return = None # a real navigation abandons the F5 return @@ -5002,11 +5956,13 @@ class IdaTui(App): # the row is loaded, instead of column 0. self._pending_focus_name = focus_name entry = NavEntry(ea=ea, name=name, cursor=cursor, is_region=is_region) + if scroll_y >= 0: + entry.scroll_y = scroll_y if dec_cursor >= 0: entry.dec_cursor = dec_cursor entry.dec_cursor_x = dec_cursor_x if push: - self._nav.append(entry) + self._push_nav(entry) self._open_entry(entry, push=False) def on_input_changed(self, event: Input.Changed) -> None: @@ -5033,6 +5989,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: @@ -5117,7 +6075,7 @@ class IdaTui(App): view, ea = self._makedata_ctx self._end_makedata() if view is not None and value: - self._do_make_data(ea, value) + self._do_make_data(ea, value, self._anchor()) return if inp.id == "goto": self._end_goto() @@ -5131,7 +6089,12 @@ class IdaTui(App): self.query_one("#func-table", DataTable).focus() def _code_view(self): # type: ignore[no-untyped-def] - return self.query_one(DecompView if self._pref == "decomp" else ListingView) + """The code pane the user is in — where focus belongs after a prompt closes. + + This used to pick on _pref, which was always "listing", so closing the + goto prompt while reading pseudocode threw focus into the listing. + """ + return self._active_code_view() def _active_code_view(self): # type: ignore[no-untyped-def] """The currently-shown code widget (for reading the cursor address).""" @@ -5305,14 +6268,21 @@ class IdaTui(App): hx.display = False lst.display = dec.display = True self.query_one("#panes").set_class(True, "split") - if self._cur is not None and dec.loaded_ea != self._cur.ea: + busy = self._cur is not None and dec.loaded_ea != self._cur.ea + if busy: dec.loading = True self._load_decomp(self._cur.ea, self._cur.name) else: dec.loading = False if grab: (dec if self._active == "decomp" else lst).focus() - self._status_for_cur("split") + if busy and self._cur is not None: + # Keep the in-flight message: the idle status used to overwrite + # it, so travelling history in split showed nothing at all while + # a decompile ran and Esc looked like a no-op. + self._status(f"{self._cur.name} \u2014 decompiling\u2026") + else: + self._status_for_cur("split") return self._split = False # a single-view target (hex, etc.) leaves split self.query_one("#panes").set_class(False, "split") @@ -5386,7 +6356,7 @@ class IdaTui(App): self._show_active() def on_hex_view_to_code(self, msg: HexView.ToCode) -> None: - self._active = self._pref + self._active = self._code_mode() self._goto_ea(msg.va, push=True) def _status_for_cur(self, mode: str) -> None: @@ -5397,12 +6367,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. @@ -5413,13 +6393,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 @@ -5574,8 +6558,18 @@ class IdaTui(App): self.app.call_from_thread(self._apply_split_map, ea, m) def _apply_split_map(self, ea: int, m: list) -> None: - if not self._split or self._cur is None or self._cur.ea != ea: - return # left split / navigated away + """Index the per-line instruction map for the decompiled function. + + Keyed to what the DECOMPILER holds, not to _cur, and not conditional on + split being on. The old guard dropped the result whenever _cur had moved + while the fetch was in flight — during trace stepping that is almost + always — leaving the split view working from the map of the function you + just left. _cur follows the cursor; this map describes the pseudocode on + screen, and those are different things. + """ + dec = self._try_view(DecompView) + if dec is not None and dec.loaded_ea is not None and ea != dec.loaded_ea: + return # a stale fetch for a function we no longer show self._split_eamap = m self._split_ea2line = {} alleas = [] @@ -5586,7 +6580,15 @@ class IdaTui(App): # ea span of the decompiled function: when the listing cursor leaves it, # _sync_split re-points the decomp to the function under the cursor. self._split_range = (min(alleas), max(alleas)) if alleas else None - self._sync_split(self._active) # re-link with the region map + # ONE index, shared with the trace path: it used to keep a parallel copy + # of exactly this, fetched separately and keyed differently, which is how + # the two ended up describing different functions. + self._trail_map, self._trail_map_ea = m, ea + self._trail_line_of = dict(self._split_ea2line) + self._trail_eas = sorted(self._trail_line_of) + self._trail_span = self._split_range + if self._split: + self._sync_split(self._active) # re-link with the region map def on_decomp_view_cursor_moved(self, msg: DecompView.CursorMoved) -> None: dv = self._try_view(DecompView) |