diff options
Diffstat (limited to 'idatui/app.py')
| -rw-r--r-- | idatui/app.py | 3976 |
1 files changed, 3002 insertions, 974 deletions
diff --git a/idatui/app.py b/idatui/app.py index ea68620..3f67756 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -10,8 +10,8 @@ Design notes: without ever materializing 52k lines in a widget. * All network/domain work runs in Textual worker threads; the UI never blocks. * An address-history stack backs Enter (follow) / Esc (back), IDA-style. -* On startup we bump the worker idle-TTL and run a keepalive heartbeat so the - session never gets reaped while we chill. +* Database lifecycle is lease-based through ida_codemode: matching GUI sessions + are reused, otherwise a shared managed idalib worker is opened on demand. """ from __future__ import annotations @@ -20,7 +20,10 @@ import asyncio import os import re import subprocess +import sys +import time from dataclasses import dataclass, field +from enum import StrEnum from rich.align import Align from rich.segment import Segment @@ -30,7 +33,7 @@ from textual import work from textual.app import App, ComposeResult from textual.binding import Binding from textual.command import DiscoveryHit, Hit, Provider -from textual.containers import Grid, Horizontal, Vertical, VerticalScroll +from textual.containers import Horizontal, Vertical, VerticalScroll from textual.geometry import Region, Size from textual.message import Message from textual.reactive import reactive @@ -43,11 +46,18 @@ from textual.widgets import ( ) from textual.widgets.option_list import Option -from .highlight import highlight_c +from . import graph +from . import kittygfx +from .edit_ctl import EditController +from .prompt import PromptBar +from .trace_ctl import TraceController +from . import findings, search +from .highlight import CTextArea, highlight_c +from .journal import Journal from .errors import IDAToolError, IDAConnectionError -from .worker_client import WorkerClient -from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct +from .codemode_client import CodeModeClient, registered_database +from .domain import Func, Head, ListingModel, Program, Struct # Styles for the disassembly listing. _S_ADDR = Style(color="#6b7684") @@ -83,6 +93,42 @@ _OP_LIMIT = 8 # opcode bytes shown in the 'limited' column mode _JUMP_CONTEXT = 4 # lines of context kept above a jump target (cursor stays on it) _SPLIT_MIN_WIDTH = 100 # need room for two usable code panes side by side + +class ViewMode(StrEnum): + """Which pane is showing (in split: which one has focus). ``IdaTui._active``. + + A ``StrEnum`` rather than an enum, deliberately: ``_active`` is handed + straight to drivers as ``cursor.kind`` over the RPC socket, and the pilot + suite compares it to plain strings. Members ARE strings, so every existing + comparison and every JSON payload keeps working -- what this buys is one + place that says which modes exist, and an AttributeError instead of silence + when one is misspelled. + + There used to be a fifth value, ``"disasm"``, assigned on exactly one path + (a decompile that failed with nowhere to return to) and meaning the same + widget as ``LISTING``. Four sites handled it and five compared against + ``"listing"`` alone, so it silently took the wrong branch in half the app -- + Tab out of a failed decompile flipped to the listing instead of retrying the + decompiler, and rpc.py carried a workaround for a mode change that never + arrived. It is gone; the failure path lands on LISTING like every other + route into the listing. + + Adding a mode means auditing every ``_active`` comparison. Prefer the + predicates on IdaTui (``is_listing``/``in_code``/...) over bare ``==`` so + the next one has fewer places to reach. + """ + + LISTING = "listing" # the unified continuous listing (code + data) + DECOMP = "decomp" # Hex-Rays pseudocode + HEX = "hex" # the hex viewer + GRAPH = "graph" # the CFG graph view + + #: The two that show a code view over a NavEntry, i.e. where a follow, an + #: xref or a rename makes sense. + @classmethod + def code_modes(cls) -> frozenset["ViewMode"]: + return frozenset({cls.LISTING, cls.DECOMP, cls.GRAPH}) + # Tokens that look like identifiers but aren't renamable symbols (so 'n' on them # in the listing names the address instead of trying to rename the token). _ASM_KEYWORDS = frozenset({ @@ -91,11 +137,27 @@ _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 +#: Hex with a trace loaded: bytes the trace SAW at this timestamp vs bytes we're +#: still showing from the file. The distinction matters more than the values — +#: one is evidence, the other is an assumption. +_S_HEX_LIVE = Style(color="#9ece6a") +_S_HEX_STALE = Style(color="#5e6875") _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 _S_NAME_MATCH = Style(bgcolor="#d0a215", color="#12161c") # filter match in a name _S_WORD = Style(bgcolor="#2a3f5f") # identifier under the cursor +#: The operand/literal under the cursor — what `o` would reformat. Distinct from +#: _S_WORD (which marks every occurrence of an identifier): this marks ONE span, +#: the thing a keypress acts on, so it reads as a selection rather than a match. +_S_OPERAND = Style(bgcolor="#3a3560", underline=True) _S_CELL = Style(reverse=True) # the block cursor cell _S_LINENO = Style(color="#626c7a") # pseudocode line-number gutter _S_LINENO_CUR = Style(color="#c3cad3", bold=True) # gutter on the cursor line @@ -114,15 +176,15 @@ _ADDR_MARK_STRIP_RE = re.compile(r"\s*/\*\s*0x[0-9A-Fa-f]+\s*\*/") @dataclass class BinaryState: """Everything that makes one project binary's session resumable across a - switch. Addresses outlive the worker, so nav history survives eviction; the - Program/index only survive while that worker is still resident.""" + switch. Addresses outlive a database lease, so nav history survives eviction; + the Program/index only survive while that lease remains resident.""" label: str program: object | None = None func_index: object | None = None nav: list = field(default_factory=list) cur: object | None = None - active: str = "listing" + active: ViewMode = ViewMode.LISTING split: bool = False filter_term: str = "" dirty: bool = False @@ -237,6 +299,17 @@ class MakeDataRequested(Message): self.view = view +class OpFormatRequested(Message): + """A code view asks to change how the literal under the cursor is DISPLAYED + (IDA's 'o'): hex, decimal, binary, character, offset. ``mode`` is 'cycle', + 'back', or a format by name.""" + + def __init__(self, view, mode: str = "cycle") -> None: # type: ignore[no-untyped-def] + super().__init__() + self.view = view + self.mode = mode + + class NavMixin: """Follow / xrefs actions shared by the code views (bindings live on each view since Textual only merges BINDINGS from DOMNode subclasses).""" @@ -522,13 +595,22 @@ class ColumnCursor: self._page_scroll(direction * (self._visible_height() // 2)) def _apply_scroll(self, y: int, x: int = 0) -> None: - """Set the scroll offset reliably after a (re)load. Applied now and again - after the next refresh — when the view was just shown its size isn't - computed yet, so an immediate scroll_to clamps to 0; the deferred pass - re-applies it and forces a repaint so the pane never shows a stale frame. + """Set the scroll offset reliably after a (re)load. Applied now and, if + that didn't take, again after the next refresh — when the view was just + shown its size isn't computed yet, so an immediate scroll_to clamps to 0; + the deferred pass re-applies it and forces a repaint so the pane never + shows a stale frame. + + The deferred pass is only scheduled when the scroll actually clamped. + ``refresh(layout=True)`` re-arranges the whole screen, and paying that on + every scroll that already landed cost ~25% of the time it takes to move + through a view. """ y = max(0, y) self.scroll_to(x=x, y=y, animate=False) + off = self.scroll_offset + if round(off.y) == y and round(off.x) == x: + return # max_scroll was current, the offset is already where we want def _fix(yy: int = y, xx: int = x) -> None: self.scroll_to(x=xx, y=yy, animate=False) @@ -537,6 +619,73 @@ class ColumnCursor: self.call_after_refresh(_fix) +class _MatchRanges: + """Which lines matched, plus where in each line — the *where* computed lazily. + + Searching a big segment for one character matches most of it: `c` over bash + hits 177 000 lines at 310 000 places. Building a range list for every one of + them costs more than finding them did, and all but the forty on screen are + thrown away unread. + + So the line set is eager (search needs it to count and to jump) and the + offsets within a line are worked out when that line is painted, or when the + cursor lands on it, and cached from then on. + + Quacks like the ``{line: [(start, end), ...]}`` dict it replaced: ``in``, + ``get``, ``[]``, ``items``, ``len``. Assigning a plain ``{}`` to reset stays + valid, because every reader only uses that same subset. + """ + + __slots__ = ("_lines", "_needle", "_n", "_ci", "_text", "_cache") + + def __init__(self, lines, needle: str, n: int, ci: bool, text) -> None: + self._lines = lines # set[int] + self._needle = needle # already case-folded when ci + self._n = n # len(term); the needle may be folded + self._ci = ci + self._text = text # callable: line index -> str | None + self._cache: dict[int, list[tuple[int, int]]] = {} + + def _find(self, i: int) -> list[tuple[int, int]]: + s = self._text(i) + if not s: + return [] + hay = s.lower() if self._ci else s + needle, n = self._needle, self._n + out = [] + j = hay.find(needle) + while j >= 0: + out.append((j, j + n)) + j = hay.find(needle, j + n) + return out + + def __contains__(self, i) -> bool: + return i in self._lines + + def __len__(self) -> int: + return len(self._lines) + + def __iter__(self): + return iter(self._lines) + + def get(self, i, default=None): + if i not in self._lines: + return default + got = self._cache.get(i) + if got is None: + got = self._cache[i] = self._find(i) + return got + + def __getitem__(self, i): + got = self.get(i) + if got is None: + raise KeyError(i) + return got + + def items(self): + return ((i, self.get(i)) for i in sorted(self._lines)) + + class SearchMixin: """Vim-style in-view search shared by the disasm and pseudocode views. @@ -559,6 +708,80 @@ class SearchMixin: _matches: list[int] _ranges: dict[int, list[tuple[int, int]]] + #: What ``_matches`` was last computed from: (term, case-fold, row count, + #: id(line source)). Lets an as-you-type search narrow the previous hits + #: instead of rescanning the segment; see :meth:`_compute_matches`. Any of + #: those changing under us discards it. + _matched_key: tuple | None = None + + #: The searchable body as ONE string (see :meth:`_search_haystack`), its + #: case-folded twin, the character offset each line starts at, and the key + #: they were built for. Dropped when the search ends. + _hay_key: tuple | None = None + _hay: tuple | None = None + + def _search_source_id(self) -> int: + """Identity of whatever supplies the line text. Changes when the view is + pointed at a different model/body, which invalidates a narrowing.""" + return id(getattr(self, "model", None) or getattr(self, "_texts", None)) + + def _reset_search_cache(self, body: bool = False) -> None: + """Forget the narrowing key, and with ``body=True`` the joined body too. + + Every place that resets ``_matches``/``_ranges`` must call this: a stale + prefix would make the next search narrow from an empty list. + + The body is a different question. It is keyed by (row count, line source) + so it invalidates itself when the view is pointed somewhere else or more + rows stream in — which means ending a search does NOT have to throw it + away, and the next `/` over the same segment is then instant instead of + re-joining a quarter of a million lines. It DOES have to go when the + plain text of a row changes without either of those moving, which is + exactly what toggling the opcode-bytes column does. + """ + self._matched_key = None + if body: + self._hay_key = None + self._hay = None + + def _search_haystack(self, count: int, src: int): + """``(starts, blob, blob_folded)`` for the whole body, or None. + + Every line joined with newlines, so finding a term is one C-level + ``str.find`` walk over the segment instead of a python loop that rebuilds + and case-folds 200 000 lines per keystroke. ``starts[i]`` is where line + ``i`` begins; a term can never straddle a line because it cannot contain + a newline. + + Returns None (and the caller falls back to the per-line loop) if folding + changes the length — a couple of unicode codepoints grow when lowered, + and then every offset after them would be wrong. + """ + key = (count, src) + if self._hay_key == key: + return self._hay + starts: list[int] = [] + parts: list[str] = [] + pos = 0 + chunk = 4096 + for base in range(0, count, chunk): + for s in self._search_line_texts(base, min(chunk, count - base)): + if not s: + s = "" + starts.append(pos) + parts.append(s) + pos += len(s) + 1 + while len(starts) < count: # a short window: keep the indices lined up + starts.append(pos) + parts.append("") + pos += 1 + blob = "\n".join(parts) + folded = blob.lower() + hay = None if len(folded) != len(blob) else (starts, blob, folded) + self._hay_key = key + self._hay = hay + return hay + # --- hooks a subclass implements --- def _search_line_count(self) -> int: raise NotImplementedError @@ -566,6 +789,12 @@ class SearchMixin: def _search_line_text(self, i: int) -> str | None: raise NotImplementedError + def _search_line_texts(self, start: int, count: int) -> list: + """``count`` line texts from ``start``. Overridable so a view whose rows + come from a locked model can fetch a window in one go.""" + text_of = self._search_line_text + return [text_of(i) for i in range(start, start + count)] + def _search_ensure(self, done) -> None: """Ensure all line texts are available, then call ``done()`` on the UI thread. Default: assume ready.""" @@ -590,6 +819,7 @@ class SearchMixin: self._term = "" self._matches = [] self._ranges = {} + self._reset_search_cache() self.cursor = getattr(self, "_search_origin", self.cursor) self.cursor_x = getattr(self, "_search_origin_x", self.cursor_x) self.refresh() @@ -628,6 +858,7 @@ class SearchMixin: self._term = "" self._matches = [] self._ranges = {} + self._reset_search_cache() self.cursor = getattr(self, "_search_origin", self.cursor) self.cursor_x = getattr(self, "_search_origin_x", self.cursor_x) self.scroll_to(y=max(self.cursor - self._visible_height() // 2, 0), animate=False) @@ -645,26 +876,70 @@ class SearchMixin: def _compute_matches(self) -> None: term = self._term - needle = term.lower() if getattr(self, "_ci", True) else term + ci = getattr(self, "_ci", True) + needle = term.lower() if ci else term + count = self._search_line_count() + src = self._search_source_id() + # Typing forward can only ever REMOVE lines: a line holding "mov" holds + # "mo". So when the term just grew (and nothing else moved -- same + # case-folding, same body, same number of rows) rescan only the previous + # hits. Search is driven a keystroke at a time, and a segment of bash is + # 224k rows; this is the difference between rescanning all of them per + # keypress and looking at a few thousand. + # + # The row count is part of the key because the listing streams in behind + # the search: rows that arrived after the last pass have never been + # looked at, and narrowing would silently never find them. + n = len(term) matches: list[int] = [] - ranges: dict[int, list[tuple[int, int]]] = {} - for i in range(self._search_line_count()): - s = self._search_line_text(i) - if not s: - continue - hay = s.lower() if getattr(self, "_ci", True) else s - pos, rs = 0, [] - while True: - j = hay.find(needle, pos) - if j < 0: - break - rs.append((j, j + len(term))) - pos = j + len(term) - if rs: + ranges: object = {} + hay = self._search_haystack(count, src) + if hay is not None: + starts, blob, folded = hay + body = folded if ci else blob + nlines = len(starts) + blen = len(body) + line = 0 + j = body.find(needle) + while j >= 0: + # find() walks forward, so the line only ever advances. + while line + 1 < nlines and starts[line + 1] <= j: + line += 1 + matches.append(line) + # Only the LINE matters here; the offsets inside it are worked + # out on demand. So skip the rest of this line rather than + # finding every further occurrence in it. + nxt = starts[line + 1] if line + 1 < nlines else blen + j = body.find(needle, nxt) + ranges = _MatchRanges(set(matches), needle, n, ci, + self._search_line_text) + else: + # Typing forward can only ever REMOVE lines: a line holding "mov" + # holds "mo". So when the term just grew (and nothing else moved -- + # same case-folding, same body, same number of rows) rescan only the + # previous hits. + # + # The row count is part of the key because the listing streams in + # behind the search: rows that arrived after the last pass have never + # been looked at, and narrowing would silently never find them. + rows: object = range(count) + prev = self._matched_key + if (prev is not None and prev[2] == count and prev[3] == src + and prev[1] == ci and term.startswith(prev[0]) and prev[0]): + rows = self._matches + text_of = self._search_line_text + for i in rows: + s = text_of(i) + if not s: + continue + h = s.lower() if ci else s + if h.find(needle) < 0: + continue matches.append(i) - ranges[i] = rs + ranges = _MatchRanges(set(matches), needle, n, ci, text_of) self._matches = matches self._ranges = ranges + self._matched_key = (term, ci, count, src) def search_repeat(self, direction: int, include_current: bool = False) -> None: if not getattr(self, "_term", ""): @@ -698,6 +973,7 @@ class SearchMixin: self._term = "" self._matches = [] self._ranges = {} + self._reset_search_cache() self.refresh() def _match_style(self, idx: int) -> Style: @@ -734,7 +1010,11 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru Binding("end", "col_end", "eol", show=False), Binding("G,ctrl+end", "goto_bottom", "Bottom", show=False), Binding("ctrl+home", "goto_top", "Top", show=False), - Binding("o", "toggle_opcodes", "Opcodes"), + # `o` is IDA's operand-format key, and that muscle memory is worth more + # than the opcode column's old claim on it (moved to B, for bytes). + Binding("o", "op_format('cycle')", "Format"), + Binding("O", "op_format('back')", "Format \u2190", show=False), + Binding("B", "toggle_opcodes", "Bytes", show=False), Binding("c", "define_code", "Code", show=False), Binding("d", "make_data", "Data", show=False), Binding("a", "make_string", "Str", show=False), @@ -769,6 +1049,9 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru self._name = "" self._pending_scroll_y: int | None = None self._pending_focus: str | None = None + #: Operand index to put the cursor column back on once rows land — a + #: reformat can change an operand's width, moving the ones after it. + self._pending_op: int | None = None self._term = "" self._matches: list[int] = [] self._ranges: dict[int, list[tuple[int, int]]] = {} @@ -777,6 +1060,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: @@ -788,19 +1073,24 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def _op_bytes_text(self, h: Head) -> str: """Hex bytes for ``h``, truncated with an ellipsis in 'limited' mode so a - long x86-64 instruction doesn't blow out the column.""" + long x86-64 instruction doesn't blow out the column. + + ``bytes.hex(" ")`` rather than a per-byte f-string generator: this is + called for every row of every plain line, and search builds the plain + line for the whole segment. + """ raw = h.raw or b"" if self._op_mode == 1 and len(raw) > _OP_LIMIT: - return " ".join(f"{b:02X}" for b in raw[:_OP_LIMIT]) + "\u2026" - return " ".join(f"{b:02X}" for b in raw) + return raw[:_OP_LIMIT].hex(" ").upper() + "\u2026" + return raw.hex(" ").upper() @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. + Uses IDA's own token classification when Code Mode supplies it; falls + back to the mnemonic/rest split when spans are absent or disagree with + the plain text. """ if h.spans: return [Segment(t, _S_SPAN.get(k, fallback)) for k, t in h.spans] @@ -819,8 +1109,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru return "" return self._op_bytes_text(h).ljust(self._op_w) + " " - def _line_plain(self, idx: int) -> str | None: - h = self._head(idx) + def _plain_of(self, h: Head | None) -> str | None: if h is None: return None # Function headers and code labels sit at depth 0 (with the address); @@ -833,6 +1122,22 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru extra = _LST_INDENT if h.kind == "member" else "" return base + self._op_field(h) + extra + self._name_prefix(h) + h.text + def _line_plain(self, idx: int) -> str | None: + return self._plain_of(self._head(idx)) + + def _search_line_texts(self, start: int, count: int) -> list: + """A window of plain lines in one model call. + + Building the search body row by row took the model's lock and bisected + its row table a quarter of a million times; ``window`` does both once + for the whole window. + """ + model = self.model + if model is None: + return [] + plain = self._plain_of + return [plain(h) for h in model.window(start, count)] + def _insn_col(self, idx: int) -> int: """Column where the instruction/content text begins, past the address + opcode-bytes gutter — the shift+home target. Mirrors ``_line_plain``'s @@ -860,15 +1165,19 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def load(self, model: ListingModel, name: str, cursor: int = 0, cursor_x: int = 0, scroll_y: int | None = None, focus: str | None = None) -> None: - self.model = model + previous, self.model = self.model, model self._name = name self.total = 0 self.cursor = cursor self.cursor_x = cursor_x self._pending_scroll_y = scroll_y self._pending_focus = focus # token to land the cursor column on + self._pending_op = None self._matches = [] self._ranges = {} + # Navigating inside the same segment reuses the same model, and the + # searchable body with it; only a different model invalidates it. + self._reset_search_cache(body=model is not previous) self._prime() @work(thread=True, exclusive=True, group="listing-prime") @@ -876,11 +1185,21 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru model = self.model if model is None: return - # Load just enough to render the viewport around the cursor, so the - # listing appears immediately even on a huge segment; the rest streams - # in via _grow. (load_all here would blank the pane for seconds.) + # One call gets the WHOLE row index -- every row's address, kind and + # size, and the page boundaries -- so the scrollbar is right immediately + # and _grow has nothing left to stream. The rows arrive text-less and + # materialise a page at a time as they are read. + # + # It is an optimisation, not a contract: an older or unhappy backend + # returns nothing usable and we stream exactly as before. height = max(self.size.height, 1) - model.ensure(self.cursor + height + 2 * ListingModel.PAGE) + if model.build_from_index(): + # Render the viewport HERE, on this worker thread. Reading a + # skeleton page fetches it, and doing that lazily from render_line + # would put an RPC on the UI loop for the first paint. + model.window(max(self.cursor - height, 0), height * 3) + else: + model.ensure(self.cursor + height + 2 * ListingModel.PAGE) self.app.call_from_thread(self._on_primed, len(model), model.complete) if not model.complete: self._grow() @@ -901,6 +1220,16 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru if occ: self.cursor_x = occ[0][0] self._pending_focus = None + if self._pending_op is not None: + # Stay on the operand that was just reformatted: its text can change + # width, which moves every operand after it out from under the + # cursor (and the next press would then hit a different one). + h = self._head(self.cursor) + for lo, _hi, n in (h.ops if h is not None else None) or (): + if n == self._pending_op: + self.cursor_x = self._insn_col(self.cursor) + lo + break + self._pending_op = None self._clamp_x() 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))) @@ -933,6 +1262,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru self._op_mode = (self._op_mode + 1) % 3 self._update_op_w() self._ranges = {} # column layout changed -> stale match offsets + # ...and which rows match at all: the opcode hex is searchable text. + self._reset_search_cache(body=True) self._clamp_x() self.refresh() self._app_status("opcodes: " + {0: "off", 1: f"limited ({_OP_LIMIT} bytes)", @@ -941,13 +1272,23 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru @work(thread=True, exclusive=True, group="listing-grow") def _grow(self) -> None: """Stream the rest of the segment's heads in the background, growing the - virtual size as they land so the scrollbar/paging catch up.""" + virtual size as they land so the scrollbar/paging catch up. + + SKELETON pages: this loop only exists to find out how many rows the + segment has, and it used to render every one of them to do it -- 227k + rows for a 1.2MB bash, ~9s, essentially all never displayed. A skeleton + page has the same rows at the same addresses and no text, is 2.8x + cheaper and costs one round trip instead of two. The first read of one + materialises it through the same path a rename uses, so only what is + actually shown ever gets rendered. _prime (the viewport) still loads + real pages, so what you are looking at is never a skeleton. + """ model = self.model if model is None: return since = 0 while not model.complete: - if model.load_next_page() == 0: + if model.load_next_page(text=False) == 0: break if self.model is not model: # a new load() replaced us return @@ -1045,6 +1386,12 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru linked = idx in self._link_rows if linked: strip = strip.apply_style(_S_LINK) # split-view companion band + if self.trail and h is not None: + kind = self.trail.get(h.ea) + if kind is not None: + strip = strip.apply_style( + _S_TRAIL_NOW if kind == "now" else + _S_TRAIL_PAST if kind == "past" else _S_TRAIL_FUTURE) plain = self._line_plain(idx) if (self._hl_word or idx == self.cursor) else None if idx in self._ranges: strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx)) @@ -1054,8 +1401,30 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru strip = _overlay_ranges(strip, occ, _S_WORD) if idx == self.cursor: strip = _cursor_decorate(strip, plain or "", self.cursor_x) + # The operand 'o' would reformat, marked before you press it. A line + # can hold several literals and the cursor picks one; showing which + # is the difference between an edit you chose and one you got. + # Drawn LAST on purpose: _cursor_decorate paints the word under the + # cursor, and that word is usually the literal itself — so painting + # this first just loses to it. + span = self._cursor_operand(idx) + if span is not None: + strip = _overlay_over(strip, [span], _S_OPERAND) return strip.adjust_cell_length(width, _S_LINK if linked else _S_INSN) + def _cursor_operand(self, idx: int) -> tuple[int, int] | None: + """Screen columns of the operand under the cursor on row ``idx``, or None. + + The extents come from the worker (IDA's own operand markers) and are + offsets into the head's text, so they shift by the same gutter the + cursor column is measured against.""" + h = self._head(idx) + if h is None or not h.ops: + return None + base = self._insn_col(idx) + got = h.op_at(self.cursor_x - base) + return (base + got[0], base + got[1]) if got else None + # -- split-view link highlight ---------------------------------------- # def set_link(self, rows) -> None: rows = set(rows) if rows else set() @@ -1142,6 +1511,18 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def action_make_string(self) -> None: self.post_message(EditItemRequested(self, "string")) + # -- literal display format (IDA 'o') --------------------------------- # + def action_op_format(self, mode: str = "cycle") -> None: + self.post_message(OpFormatRequested(self, mode)) + + def op_col(self) -> int: + """Cursor column inside the head's OWN text — the same string the worker + renders, so it can say which operand the cursor is standing on. -1 when + the cursor is left of it (in the address/opcode gutter), which means "you + didn't pick one, take the first literal".""" + base = self._insn_col(self.cursor) + return self.cursor_x - base if self.cursor_x >= base else -1 + def cur_head(self) -> Head | None: return self._head(self.cursor) @@ -1225,6 +1606,10 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True Binding("end", "col_end", "eol", show=False), Binding("ctrl+home", "goto_top", "Top", show=False), Binding("G,ctrl+end", "goto_bottom", "Bottom", show=False), + # Hex-Rays keeps number formats of its own, so `o` works here too — on + # the C literal under the cursor, not on the instruction's operand. + Binding("o", "op_format('cycle')", "Format"), + Binding("O", "op_format('back')", "Format \u2190", show=False), *SearchMixin.SEARCH_BINDINGS, *NavMixin.NAV_BINDINGS, *ColumnCursor.COL_BINDINGS, @@ -1253,13 +1638,44 @@ 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]]] = {} + #: {line: [(x0, x1, value, ea, opnum)]} — where the number literals are, + #: so the one under the cursor can be marked (Hex-Rays keeps formats per + #: literal, and a C line often has several). + self._nums: dict[int, list[tuple[int, int, str, int, int]]] = {} + #: (ea, opnum) to put the cursor back on once the nums land — a reformat + #: reflows the line, so the old column points at the wrong literal. + self._keep_lit: tuple[int, int] | None = None def _line_plain(self, idx: int) -> str | None: return self._texts[idx] if 0 <= idx < len(self._texts) else None + def set_nums(self, nums: dict) -> None: + self._nums = nums or {} + keep, self._keep_lit = self._keep_lit, None + if keep is not None: + # Land the cursor back on the literal that was just reformatted: + # `48` becoming `0x30` moves everything after it, so holding the + # column would put the next press on a different literal. + for x0, _x1, _v, ea, opnum in self._nums.get(self.cursor, ()): + if (ea, opnum) == keep: + self.cursor_x = x0 + self._hscroll() + break + self.refresh() + + def _cursor_literal(self, idx: int) -> tuple[int, int] | None: + """Columns of the number literal under the cursor on ``idx``, or None.""" + for x0, x1, _v, _ea, _op in self._nums.get(idx, ()): + if x0 <= self.cursor_x < x1: + return (x0, x1) + return None + def _col_offset(self) -> int: return self._gutter @@ -1293,6 +1709,10 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True self.cursor_x = cursor_x self._matches = [] self._ranges = {} + # A whole new body: drop the joined haystack outright rather than trust + # id(self._texts) to differ, since the list it replaces is freed here and + # its address can be handed straight back. + self._reset_search_cache(body=True) # Gutter wide enough for the largest line number + a trailing space. self._gutter = (len(str(total)) + 1) if total else 0 maxw = max((s.cell_length for s in self._strips), default=0) @@ -1386,6 +1806,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: @@ -1394,6 +1819,9 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True base = _overlay_ranges(base, occ, _S_WORD) if idx == self.cursor: base = _cursor_decorate(base, self._texts[idx], self.cursor_x) + span = self._cursor_literal(idx) # the literal `o` would reformat + if span is not None: # (last: see ListingView) + base = _overlay_over(base, [span], _S_OPERAND) code_w = max(width - gw, 0) code = base.crop(x, x + code_w).adjust_cell_length( code_w, _S_LINK if linked else None) @@ -1430,6 +1858,9 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True if round(old_value) != round(new_value): self.post_message(DecompView.Scrolled()) + def action_op_format(self, mode: str = "cycle") -> None: + self.post_message(OpFormatRequested(self, mode)) + def action_col_code_home(self) -> None: """shift+home: first non-blank column — past the C indentation, the pseudocode analogue of the listing's skip-the-address-gutter.""" @@ -1538,6 +1969,10 @@ class HexView(ScrollView, can_focus=True): super().__init__(id="hex") self.model = None self.total = 0 + #: Trace to read memory from, and the timestamp to read it at. When set, + #: the dump shows what memory HELD then rather than what the file holds. + self.trace = None + self.trace_idx = 0 self._internal_top: int | None = None # scroll target we set ourselves # -- public API -------------------------------------------------------- # @@ -1590,7 +2025,11 @@ class HexView(ScrollView, can_focus=True): self.refresh(layout=True) self.scroll_to(y=y, animate=False) - self.call_after_refresh(_fix) + # Only re-apply when the scroll clamped (the view's size wasn't computed + # yet). See ColumnCursor._apply_scroll: the deferred pass drags a full + # layout with it, which is far too expensive to do on every scroll. + if round(self.scroll_offset.y) != y: + self.call_after_refresh(_fix) def _scroll_to_cursor(self, center: bool = False) -> None: height = self._visible_height() @@ -1724,6 +2163,15 @@ class HexView(ScrollView, can_focus=True): return Strip([Segment("".ljust(width), _S_HEX)]) va, data = model.row(r) cur_row, cur_col = self.cursor // 16, self.cursor % 16 + # With a trace loaded the row shows what memory HELD at the current + # timestamp, not what the file contains. Only the bytes the trace + # actually saw are overlaid: the rest stay the database's, dimmed, so + # you can always tell evidence from the file's idea of the world. + tmem = tknown = None + if self.trace is not None and data is not None: + tmem, tknown = self.trace.memory(va, len(data), self.trace_idx) + if not any(tknown): + tmem = tknown = None fo = model.file_offset(va) fo_str = f"{fo:08X}" if fo is not None else "--------" segs: list[Segment] = [ @@ -1733,28 +2181,940 @@ class HexView(ScrollView, can_focus=True): if data is None: segs.append(Segment("… fetching", _S_DIM)) else: + # Emit RUNS, not one segment per byte. A row is 32 cells whose style + # almost never changes (one cursor cell, or a trace boundary), and a + # segment per cell made every hex frame 1540 segments for the + # compositor to cut and merge again. n = len(data) + run: list[str] = [] + run_st = None + + def flush(st=None, _segs=segs) -> None: + nonlocal run, run_st + if run: + _segs.append(Segment("".join(run), run_st)) + run = [] + run_st = st + + def put(text: str, st) -> None: + nonlocal run_st + if st is not run_st: + flush(st) + run.append(text) + + run_st = _S_HEX for i in range(16): if i == 8: - segs.append(Segment(" ", _S_HEX)) + put(" ", _S_HEX) if i < n: - st = _S_CELL if (r == cur_row and i == cur_col) else _S_HEX - segs.append(Segment(f"{data[i]:02X} ", st)) + live = tknown is not None and tknown[i] + val = tmem[i] if live else data[i] + if r == cur_row and i == cur_col: + st = _S_CELL + elif tknown is None: + st = _S_HEX + else: + st = _S_HEX_LIVE if live else _S_HEX_STALE + put(f"{val:02X} ", st) else: - segs.append(Segment(" ", _S_HEX)) - segs.append(Segment(" |", _S_DIM)) + put(" ", _S_HEX) + put(" |", _S_DIM) for i in range(16): if i < n: - ch = chr(data[i]) if 32 <= data[i] < 127 else "." - st = _S_CELL if (r == cur_row and i == cur_col) else _S_ASCII + live = tknown is not None and tknown[i] + val = tmem[i] if live else data[i] + ch = chr(val) if 32 <= val < 127 else "." + if r == cur_row and i == cur_col: + st = _S_CELL + elif tknown is None: + st = _S_ASCII + else: + st = _S_HEX_LIVE if live else _S_HEX_STALE else: ch, st = " ", _S_ASCII - segs.append(Segment(ch, st)) - segs.append(Segment("|", _S_DIM)) + put(ch, st) + put("|", _S_DIM) + flush() return Strip(segs).adjust_cell_length(width, _S_HEX) # --------------------------------------------------------------------------- # +# Graph view +# --------------------------------------------------------------------------- # +_S_GBORDER = Style(color="#4b5565") # box border, idle +_S_GBORDER_CUR = Style(color="#7aa2f7", bold=True) # box border, cursor block +_S_GLABEL = Style(color="#7aa2f7", bold=True) # loc_XXXX in the border +_S_GLABEL_CUR = Style(color="#c0caf5", bold=True) +_S_GDIM = Style(color="#5e6875") +_S_GENTRY = Style(color="#9ece6a", bold=True) # the entry block's label +#: Edge colours follow IDA's convention: green = branch taken, red = falls +#: through, blue = the block's only successor, purple = loops back. +_S_EDGE = { + graph.E_TRUE: Style(color="#5fbf5f"), + graph.E_FALSE: Style(color="#cf5f5f"), + graph.E_UNCOND: Style(color="#5f87d7"), + graph.E_SWITCH: Style(color="#c9a227"), + graph.E_BACK: Style(color="#a06fd0"), +} +#: Same hues, brightened: the edges touching the block you're on. +_S_EDGE_HOT = { + graph.E_TRUE: Style(color="#8ff08f", bold=True), + graph.E_FALSE: Style(color="#ff8f8f", bold=True), + graph.E_UNCOND: Style(color="#9fc4ff", bold=True), + graph.E_SWITCH: Style(color="#ffd75f", bold=True), + graph.E_BACK: Style(color="#d0a0ff", bold=True), +} +_S_MINI_BG = Style(bgcolor="#161b22", color="#3b4453") +_S_MINI_NODE = Style(bgcolor="#161b22", color="#5f7799") +_S_MINI_CUR = Style(bgcolor="#161b22", color="#9ece6a", bold=True) +_S_MINI_VIEW = Style(bgcolor="#233044", color="#c0caf5") +_S_MINI_EDGE = Style(bgcolor="#161b22", color="#2f3945") + +_GPAD = 1 # columns of padding inside a box +_MINI_W, _MINI_H = 30, 14 + + +#: Memo for ``Style + Style``. Rich rebuilds a whole Style on every ``+``, and +#: the graph merges an overlay (cursor band, trail, highlight) over a run of +#: cells that share only a handful of base styles -- so the same pair is +#: recombined hundreds of times per frame. +_STYLE_SUM: dict[tuple, Style] = {} + + +class _CellRow: + """A row of (char, style) cells that coalesces into a Strip. + + The graph is drawn per screen row from three independent sources -- edge + cells, boxes, then the minimap -- which overwrite each other. Composing into + a flat cell array and coalescing once at the end is both simpler and cheaper + than splicing Strips three times. + """ + + __slots__ = ("ch", "st", "width") + + def __init__(self, width: int, base: Style) -> None: + self.width = max(width, 0) + self.ch = [" "] * self.width + self.st = [base] * self.width + + def put(self, i: int, ch: str, style: Style) -> None: + if 0 <= i < self.width: + self.ch[i] = ch + self.st[i] = style + + def text(self, i: int, s: str, style: Style) -> None: + """Write ``s`` at cell ``i``, clipped to the row. + + Slice assignment rather than a call per character: a graph row is drawn + from box borders, instruction text and the minimap, and doing it a cell + at a time made painting one frame thousands of bound-method calls. + Assigning a str to a list slice expands it to characters in C. + """ + if not s: + return + a = i if i > 0 else 0 + b = i + len(s) + if b > self.width: + b = self.width + if b <= a: + return + self.ch[a:b] = s[a - i:b - i] + self.st[a:b] = [style] * (b - a) + + def restyle(self, a: int, b: int, style: Style) -> None: + """Merge ``style`` over the cells in [a, b) (keeps the characters).""" + if a < 0: + a = 0 + if b > self.width: + b = self.width + st = self.st + combine = _STYLE_SUM + for i in range(a, b): + base = st[i] + key = (base, style) + got = combine.get(key) + if got is None: + got = combine[key] = base + style + st[i] = got + + def strip(self) -> Strip: + segs: list[Segment] = [] + if not self.width: + return Strip([]) + run_start = 0 + cur = self.st[0] + for i in range(1, self.width): + if self.st[i] is not cur and self.st[i] != cur: + segs.append(Segment("".join(self.ch[run_start:i]), cur)) + run_start, cur = i, self.st[i] + segs.append(Segment("".join(self.ch[run_start:]), cur)) + return Strip(segs) + + +class GraphView(NavMixin, ScrollView, can_focus=True): + """IDA-style control-flow graph of one function, in character cells. + + The layout comes from ``idatui.graph`` (pure, offline-testable); this class + is only presentation, navigation and hit-testing. Nothing is pre-painted: a + big function is millions of cells, so each screen row is composed on demand + from the edge index plus whichever boxes cover that row -- the same + discipline as ``ListingView.render_line``. + + Box contents are the SAME ``Head`` rows the listing renders, so IDA's own + colour tags, names and operand text come along for free instead of this + growing a second disassembler renderer. + """ + + BINDINGS = [ + Binding("j,down", "cursor_down", "Down", show=False), + Binding("k,up", "cursor_up", "Up", show=False), + Binding("h,left", "cursor_left", "Left", show=False), + Binding("l,right", "cursor_right", "Right", show=False), + Binding("J", "succ_block", "Next block", show=False), + Binding("K", "pred_block", "Prev block", show=False), + Binding("w", "next_block", "Block →", show=False), + Binding("b", "prev_block", "Block ←", show=False), + Binding("0", "goto_entry", "Entry", show=False), + Binding("z", "zoom", "Zoom"), + Binding("m", "minimap", "Minimap", show=False), + Binding("e", "engine", "Engine", show=False), + Binding("f", "center", "Centre", show=False), + Binding("ctrl+d", "pan(12)", "½↓", show=False), + Binding("ctrl+u", "pan(-12)", "½↑", show=False), + Binding("pagedown", "pan(24)", "PgDn", show=False), + Binding("pageup", "pan(-24)", "PgUp", show=False), + Binding("home", "col_home", "bol", show=False), + Binding("end", "col_end", "eol", show=False), + *NavMixin.NAV_BINDINGS, + ] + + cursor_node = reactive(-1, repaint=False) + cursor_row = reactive(0, repaint=False) + cursor_x = reactive(0, repaint=False) + + ZOOMS = ("full", "compact", "collapsed") + + class CursorMoved(Message): + """Posted when the graph cursor lands on a new address.""" + + def __init__(self, ea: int | None, block: int) -> None: + super().__init__() + self.ea = ea + self.block = block + + def __init__(self) -> None: + super().__init__() + self.fc = None # domain.Flowchart + self.lay: graph.Layout | None = None + self.loaded_ea: int | None = None + self._blocks: dict[int, object] = {} + self._zoom = 0 + self._show_minimap = True + #: layout backend; "auto" prefers triskel where it is installed and the + #: function is small enough for it. Cycled with `e`. + self._engine = "auto" + self._mini_cache: tuple | None = None + self._drag: tuple[int, int, float, float] | None = None + self._drag_map = False # the drag started on the minimap + self._hl_word = "" + self.trail: dict[int, str] | None = None + + # -- content ---------------------------------------------------------- # + def set_graph(self, fc, ea: int | None = None) -> None: + """Install a flowchart and lay it out at the current zoom.""" + self.fc = fc + self._blocks = {b.id: b for b in fc.blocks} if fc else {} + self.loaded_ea = fc.func_ea if fc else None + self._relayout() + if fc: + blk = fc.block_at(ea) if ea is not None else None + self.cursor_node = blk.id if blk else fc.entry + self.cursor_row = 0 + if blk is not None and ea is not None: + rows = self._rows(blk.id) + for i, h in enumerate(rows): + if h is not None and h.ea == ea: + self.cursor_row = i + break + self.cursor_x = 0 + self._center_cursor() + self.refresh(layout=True) + + def _relayout(self) -> None: + self._mini_cache = None + if not self.fc or not self.fc.blocks: + self.lay = None + self.virtual_size = Size(0, 0) + return + blocks = [graph.Block(id=b.id, start=b.start, end=b.end, + succs=list(b.succs)) for b in self.fc.blocks] + self.lay = graph.layout(blocks, self._sizer, entry=self.fc.entry, + engine=self._engine) + self.virtual_size = Size(self.lay.width + 2, self.lay.height + 1) + + def _rows(self, nid: int): + """The Head rows shown inside block ``nid`` at the current zoom.""" + b = self._blocks.get(nid) + if b is None: + return [] + if self._zoom == 2: + return [None] # one synthetic summary row + return b.rows + + def _row_plain(self, nid: int, i: int) -> str: + b = self._blocks.get(nid) + if b is None: + return "" + if self._zoom == 2: + n = len(b.rows) + return f"{n} instruction{'s' if n != 1 else ''}" + rows = b.rows + if not (0 <= i < len(rows)): + return "" + h = rows[i] + if self._zoom == 0: + return f"{h.ea:08X} {self._head_text(h)}" + return self._head_text(h) + + @staticmethod + def _head_text(h) -> str: + return (f"{h.name} {h.text}" if h.name else h.text) + + def _sizer(self, b: graph.Block) -> tuple[int, int]: + nid = b.id + rows = self._rows(nid) + n = max(len(rows), 1) + label = f"loc_{b.start:X}" + widest = max([len(label) + 4] + + [len(self._row_plain(nid, i)) for i in range(n)]) + return (widest + 2 * _GPAD + 2, n + 2) + + # -- geometry --------------------------------------------------------- # + def _cur_node(self) -> graph.Node | None: + if self.lay is None: + return None + return self.lay.by_id.get(self.cursor_node) + + def cur_head(self): + rows = self._rows(self.cursor_node) + if self._zoom == 2 or not rows: + b = self._blocks.get(self.cursor_node) + return b.rows[0] if (b and b.rows) else None + i = max(0, min(self.cursor_row, len(rows) - 1)) + return rows[i] + + def _cursor_ea(self) -> int | None: + h = self.cur_head() + return h.ea if h is not None else None + + def _next_ea(self) -> int | None: + """The address after the cursor's instruction -- what ``follow`` uses to + skip the fall-through edge and land on a real branch target.""" + b = self._blocks.get(self.cursor_node) + if b is None: + return None + rows = b.rows + i = max(0, min(self.cursor_row, len(rows) - 1)) if rows else 0 + if rows and i + 1 < len(rows): + return rows[i + 1].ea + return b.end + + def _line_plain(self, _idx=None) -> str: + return self._row_plain(self.cursor_node, self.cursor_row) + + def word_under_cursor(self) -> str: + plain = self._line_plain() + if not plain: + return "" + a, b = _word_bounds(plain, self.cursor_x) + return plain[a:b] if b > a else "" + + def set_highlight(self, word: str) -> None: + if word != self._hl_word: + self._hl_word = word + self.refresh() + + def set_trail(self, trail: dict[int, str] | None) -> None: + self.trail = trail + self.refresh() + + # -- cursor motion ---------------------------------------------------- # + def _clamp_cursor(self) -> None: + if self.lay is None or not self.lay.nodes: + return + if self.cursor_node not in self.lay.by_id: + self.cursor_node = self.lay.nodes[0].id + rows = self._rows(self.cursor_node) + self.cursor_row = max(0, min(self.cursor_row, max(len(rows) - 1, 0))) + plain = self._line_plain() + self.cursor_x = max(0, min(self.cursor_x, max(len(plain) - 1, 0))) + + def _order(self) -> list[int]: + return [n.id for n in self.lay.nodes] if self.lay else [] + + def _moved(self) -> None: + self._clamp_cursor() + self._scroll_to_cursor() + self.refresh() + self.post_message(self.CursorMoved(self._cursor_ea(), self.cursor_node)) + + def action_cursor_down(self) -> None: + rows = self._rows(self.cursor_node) + if self.cursor_row + 1 < len(rows): + self.cursor_row += 1 + else: + order = self._order() + if self.cursor_node in order: + i = order.index(self.cursor_node) + if i + 1 < len(order): + self.cursor_node = order[i + 1] + self.cursor_row = 0 + self._moved() + + def action_cursor_up(self) -> None: + if self.cursor_row > 0: + self.cursor_row -= 1 + else: + order = self._order() + if self.cursor_node in order: + i = order.index(self.cursor_node) + if i > 0: + self.cursor_node = order[i - 1] + self.cursor_row = max(len(self._rows(self.cursor_node)) - 1, 0) + self._moved() + + def action_cursor_left(self) -> None: + self.cursor_x = max(0, self.cursor_x - 1) + self._moved() + + def action_cursor_right(self) -> None: + self.cursor_x = min(len(self._line_plain()), self.cursor_x + 1) + self._moved() + + def action_col_home(self) -> None: + self.cursor_x = 0 + self._moved() + + def action_col_end(self) -> None: + self.cursor_x = max(len(self._line_plain()) - 1, 0) + self._moved() + + def _hop(self, table: dict) -> None: + tgt = table.get(self.cursor_node) or [] + if not tgt: + self.app._status("no edge that way") + return + self.cursor_node = tgt[0][0] + self.cursor_row = 0 + self._moved() + + def action_succ_block(self) -> None: + if self.lay: + self._hop(self.lay.succ) + + def action_pred_block(self) -> None: + if self.lay: + self._hop(self.lay.pred) + + def _step_order(self, delta: int) -> None: + order = self._order() + if not order: + return + i = order.index(self.cursor_node) if self.cursor_node in order else 0 + self.cursor_node = order[max(0, min(len(order) - 1, i + delta))] + self.cursor_row = 0 + self._moved() + + def action_next_block(self) -> None: + self._step_order(1) + + def action_prev_block(self) -> None: + self._step_order(-1) + + def action_goto_entry(self) -> None: + if self.fc: + self.cursor_node = self.fc.entry + self.cursor_row = 0 + self._moved() + self._center_cursor() + + def action_pan(self, rows: int) -> None: + self.scroll_to(y=max(0, self.scroll_offset.y + rows), animate=False) + self._snap_into_view() + + def _viewport_has_block(self) -> bool: + if self.lay is None: + return False + y0 = int(self.scroll_offset.y) + x0 = int(self.scroll_offset.x) + y1, x1 = y0 + self.size.height, x0 + self.size.width + return any(n.y <= y1 and y0 <= n.bottom and n.x <= x1 and x0 <= n.right + for n in self.lay.nodes) + + def _snap_into_view(self) -> None: + """After a pan, if the viewport holds no block at all, ease to the + nearest one. + + Blocks cover a few percent of a laid-out graph -- 4.6% on an 87-block + function, under 1% on a 424-block one -- the rest being the padding that + keeps edges apart. Panning therefore lands in empty space more often + than not, and an empty screen gives you nothing to navigate back by. + Only fires when nothing is visible, so it never fights a deliberate pan. + """ + if self.lay is None or self._viewport_has_block(): + return + cy = self.scroll_offset.y + self.size.height / 2 + cx = self.scroll_offset.x + self.size.width / 2 + n = self._nearest_node(cy, cx) + if n is not None: + self._center_on(n, defer=False) + self.refresh() + + def action_zoom(self) -> None: + self._zoom = (self._zoom + 1) % len(self.ZOOMS) + self._relayout() + self._clamp_cursor() + self._center_cursor() + self.refresh(layout=True) + self.app._graph_status() # keeps the function name; names the zoom + + def action_minimap(self) -> None: + self._show_minimap = not self._show_minimap + self.refresh() + self.app._status(f"graph: minimap {'on' if self._show_minimap else 'off'}") + + def action_engine(self) -> None: + """Cycle the layout engine and redraw the same function with it. + + The two engines disagree about shape more than about correctness -- + native draws wide and short, triskel narrow and tall with far fewer + crossings -- and which one reads better genuinely depends on the + function. Cheaper to look than to argue. + """ + from . import graph_triskel + choices = ["auto", "native"] + (["triskel"] if graph_triskel.available() + else []) + self._engine = choices[(choices.index(self._engine) + 1) % len(choices)] + self._relayout() + self._clamp_cursor() + self._center_cursor() + self.refresh(layout=True) + got = self.lay.stats["engine"] if self.lay else "?" + # Name the interpreter. The launcher runs $IDATUI_PYTHON (default + # ~/ida-venv), which is NOT the repo .venv the tests use, so "not + # installed" on its own sends people to check the wrong python. + note = ("" if graph_triskel.available() + else f" (no pytriskel in {sys.executable})") + # A fallback with no reason is a bug report nobody can file. + if self.lay and self.lay.stats.get("engine_error"): + note = f" \u2014 {self.lay.stats['engine_error']}" + self.app._status(f"graph: engine {self._engine} \u2192 {got}{note}") + + def action_center(self) -> None: + self._center_cursor() + self.refresh() + + def action_copy_line(self) -> None: + plain = self._line_plain() + if not plain: + return + n = self.app._copy(plain) + self.app._status(f"copied line ({n} chars) to clipboard") + + def goto_ea(self, ea: int) -> bool: + """Put the cursor on ``ea`` if it lives in this graph.""" + if not self.fc: + return False + b = self.fc.block_at(ea) + if b is None: + return False + self.cursor_node = b.id + self.cursor_row = 0 + for i, h in enumerate(self._rows(b.id)): + if h is not None and h.ea == ea: + self.cursor_row = i + break + self.cursor_x = 0 + self._clamp_cursor() + self._center_cursor() + self.refresh() + return True + + # -- scrolling -------------------------------------------------------- # + def _cursor_cell(self) -> tuple[int, int] | None: + n = self._cur_node() + if n is None: + return None + row = n.y + 1 + max(0, min(self.cursor_row, n.h - 3)) + col = n.x + 1 + _GPAD + self.cursor_x + return (row, col) + + def _scroll_to_cursor(self) -> None: + cell = self._cursor_cell() + if cell is None: + return + row, col = cell + w, h = self.size.width, self.size.height + if w <= 0 or h <= 0: + return + y, x = self.scroll_offset.y, self.scroll_offset.x + if row < y: + y = row + elif row >= y + h: + y = row - h + 1 + if col < x + 2: + x = max(0, col - 2) + elif col >= x + w - 2: + x = col - w + 3 + if (y, x) != (self.scroll_offset.y, self.scroll_offset.x): + self.scroll_to(y=max(0, y), x=max(0, x), animate=False) + + def _center_on(self, n: graph.Node, defer: bool = True) -> None: + """Bring block ``n`` into the middle of the viewport. + + ``defer=False`` during a drag: layout is already valid then, and + queueing a callback per mouse-move makes the scrub lag behind. + """ + if n is None or self.size.width <= 0: + return + y = max(0, n.y - max(self.size.height // 2 - n.h // 2, 0)) + x = max(0, int(n.cx) - self.size.width // 2) + self.scroll_to(y=y, x=x, animate=False) + if not defer: + return + # Setting virtual_size then scrolling immediately clamps to 0 (max_scroll + # isn't recomputed until layout), so apply it again after the refresh. + def _again() -> None: + self.scroll_to(y=y, x=x, animate=False) + self.call_after_refresh(_again) + + def _center_cursor(self) -> None: + n = self._cur_node() + if n is not None: + self._center_on(n) + + # -- minimap hit-testing ----------------------------------------------- # + def _minimap_rect(self) -> tuple[int, int, int, int] | None: + """(left, top, w, h) of the minimap in CONTENT coordinates, or None. + + The minimap is pinned to the viewport, not the canvas, so these are + screen-relative and the scroll offset must NOT be added. Must agree with + _draw_minimap_row, which is why both take the inset from here. + """ + if not self._show_minimap or self.lay is None: + return None + w, h = self.size.width, self.size.height + if w < _MINI_W + 10 or h < _MINI_H + 2: + return None + return (w - _MINI_W - 2, 0, _MINI_W, _MINI_H) + + def _nearest_node(self, row: float, col: float) -> graph.Node | None: + """The block nearest a canvas point (distance 0 if the point is inside). + + Cells are about twice as tall as they are wide, so the column distance + is halved -- otherwise "nearest" means nearest in cells, which does not + look nearest on screen. + """ + if self.lay is None: + return None + best, best_d = None, None + for n in self.lay.nodes: + dx = 0.0 if n.x <= col <= n.right else min(abs(col - n.x), + abs(col - n.right)) + dy = 0.0 if n.y <= row <= n.bottom else min(abs(row - n.y), + abs(row - n.bottom)) + d = (dx * 0.5) ** 2 + dy ** 2 + if best_d is None or d < best_d: + best, best_d = n, d + return best + + def _minimap_seek(self, x: int, y: int, defer: bool = True) -> bool: + """Treat (x, y) as a point on the minimap and go to the block there. + + Deliberately snaps to the NEAREST BLOCK rather than scrolling to the raw + coordinate. Most of a laid-out graph is the padding that keeps edges + apart, so a coordinate-accurate jump usually parks the viewport in empty + space -- and the cursor, which only moved when the point landed exactly + on a block, stayed behind. Snapping means every click lands on something + and the keyboard carries on from there. + + Returns False if the point isn't on the minimap, so the caller can fall + through to ordinary canvas hit-testing. + """ + rect = self._minimap_rect() + if rect is None or self.lay is None: + return False + left, top, _w, _h = rect + gw, gh = _MINI_W - 2, _MINI_H - 2 + c, r = x - left - 1, y - top - 1 # inside the border + if not (0 <= c < gw and 0 <= r < gh): + return False + lay = self.lay + sx = max(lay.width / gw, 1e-9) + sy = max(lay.height / gh, 1e-9) + cx, cy = (c + 0.5) * sx, (r + 0.5) * sy # centre of that mini-cell + n = self._nearest_node(cy, cx) + if n is None: + self.scroll_to(x=max(0, int(cx - self.size.width / 2)), + y=max(0, int(cy - self.size.height / 2)), + animate=False) + return True + if n.id == self.cursor_node: + return True # already there; don't churn while dragging + self.cursor_node = n.id + self.cursor_row = 0 + self.cursor_x = 0 + self._clamp_cursor() + self._center_on(n, defer=defer) + self.refresh() + self.post_message(self.CursorMoved(self._cursor_ea(), self.cursor_node)) + return True + + # -- mouse ------------------------------------------------------------- # + def on_mouse_down(self, event) -> None: # type: ignore[no-untyped-def] + off = event.get_content_offset(self) + if off is None: + return + if self._minimap_seek(off.x, off.y): + self._drag = None + self._drag_map = True # keep scrubbing while the button is held + return + self._drag_map = False + self._drag = (off.x, off.y, self.scroll_offset.x, self.scroll_offset.y) + + def on_mouse_up(self, event) -> None: # type: ignore[no-untyped-def] + was_pan = self._drag is not None and not self._drag_map + self._drag = None + self._drag_map = False + if was_pan: + self._snap_into_view() # don't leave them adrift in the padding + + def on_mouse_move(self, event) -> None: # type: ignore[no-untyped-def] + if not event.button: + return + off = event.get_content_offset(self) + if off is None: + return + if self._drag_map: + # drag = scrub block to block through the overview + self._minimap_seek(off.x, off.y, defer=False) + return + if self._drag is None: + return + x0, y0, sx, sy = self._drag + self.scroll_to(x=max(0, sx + (x0 - off.x)), y=max(0, sy + (y0 - off.y)), + animate=False) + + def on_click(self, event) -> None: # type: ignore[no-untyped-def] + if self.lay is None: + return + off = event.get_content_offset(self) + if off is None: + return + # The minimap floats over the canvas, so it has to be tested FIRST -- + # otherwise a click on it is read as canvas coordinates and drops the + # cursor into whatever block happens to lie underneath. + if self._minimap_seek(off.x, off.y): + self.focus() + return + row = off.y + int(self.scroll_offset.y) + col = off.x + int(self.scroll_offset.x) + n = self.lay.node_at(row, col) + if n is None or n.block is None: + return + self.focus() + self.cursor_node = n.id + self.cursor_row = max(0, min(row - n.y - 1, + max(len(self._rows(n.id)) - 1, 0))) + self.cursor_x = max(0, col - n.x - 1 - _GPAD) + self._clamp_cursor() + self.refresh() + self.post_message(self.CursorMoved(self._cursor_ea(), self.cursor_node)) + if getattr(event, "chain", 1) >= 2: + self.post_message(FollowRequested(self)) + + # -- rendering -------------------------------------------------------- # + def _edge_styles(self) -> tuple[dict, set]: + hot = self.lay.incident.get(self.cursor_node, set()) if self.lay else set() + return (_S_EDGE, hot) + + def render_line(self, y: int) -> Strip: + width = self.size.width + if self.lay is None or not self.lay.nodes: + return Strip([Segment("".ljust(width), _S_GDIM)]) + row = int(self.scroll_offset.y) + y + col0 = int(self.scroll_offset.x) + out = _CellRow(width, _S_INSN) + base, hot = self._edge_styles() + + # 1. edge cells (an index query, never a painted canvas) + for col, (ch, kind, eid) in self.lay.painting.cells_at_row( + row, col0, col0 + width).items(): + st = (_S_EDGE_HOT if eid in hot else base).get(kind, _S_GDIM) + out.put(col - col0, ch, st) + + # 2. boxes covering this row (they win over edges: nothing routes inside) + for n in self.lay.nodes_at_row(row): + self._draw_node_row(out, n, row, col0) + + # 3. minimap, last, over everything + if self._show_minimap: + self._draw_minimap_row(out, y, width) + return out.strip().adjust_cell_length(width, _S_INSN) + + def _draw_node_row(self, out: _CellRow, n: graph.Node, row: int, + col0: int) -> None: + cur = n.id == self.cursor_node + bs = _S_GBORDER_CUR if cur else _S_GBORDER + left = n.x - col0 + w = n.w + b = self._blocks.get(n.id) + if row == n.y: + # top border carries the label: ┌─ loc_1234 ─────┐ + label = f"loc_{n.block.start:X}" if n.block else "" + if b is not None and b.rows and b.rows[0].name: + label = b.rows[0].name + out.text(left, graph.BOX["tl"] + graph.BOX["h"] * (w - 2) + + graph.BOX["tr"], bs) + tag = f" {label} " + if len(tag) <= w - 4: + st = _S_GENTRY if (self.fc and n.id == self.fc.entry) else ( + _S_GLABEL_CUR if cur else _S_GLABEL) + out.text(left + 2, tag, st) + if n.block is not None and n.block.selfloop: + out.put(left + w - 2, "↺", _S_EDGE[graph.E_BACK]) + return + if row == n.y + n.h - 1: + out.text(left, graph.BOX["bl"] + graph.BOX["h"] * (w - 2) + + graph.BOX["br"], bs) + return + out.put(left, graph.BOX["v"], bs) + out.put(left + w - 1, graph.BOX["v"], bs) + out.text(left + 1, " " * (w - 2), _S_INSN) + i = row - n.y - 1 + rows = self._rows(n.id) + if not (0 <= i < len(rows)): + return + text_col = left + 1 + _GPAD + h = rows[i] + plain = self._row_plain(n.id, i) + if h is None: # collapsed summary + out.text(text_col, plain, _S_GDIM) + else: + c = text_col + if self._zoom == 0: + out.text(c, f"{h.ea:08X} ", _S_ADDR) + c += 10 + if h.name: + out.text(c, f"{h.name} ", _S_LABEL) + c += len(h.name) + 2 + fallback = {"code": _S_INSN, "data": _S_DATA}.get(h.kind, _S_UNK) + if h.spans: + for kind, t in h.spans: + out.text(c, t, _S_SPAN.get(kind, fallback)) + c += len(t) + else: + out.text(c, h.text, fallback) + inner_a, inner_b = left + 1, left + w - 1 + # execution trail (a loaded Tenet trace), same palette as the listing + if self.trail is not None and h is not None: + k = self.trail.get(h.ea) + if k is not None: + out.restyle(inner_a, inner_b, + _S_TRAIL_NOW if k == "now" else + _S_TRAIL_PAST if k == "past" else _S_TRAIL_FUTURE) + if self._hl_word and plain: + for a, bb in _word_occurrences(plain, self._hl_word): + out.restyle(text_col + a, text_col + bb, _S_WORD) + if cur and i == self.cursor_row: + out.restyle(inner_a, inner_b, _S_CURSOR) + x = max(0, min(self.cursor_x, max(len(plain) - 1, 0))) + wa, wb = _word_bounds(plain, x) + if wb > wa: + out.restyle(text_col + wa, text_col + wb, _S_WORD) + if self.has_focus: + out.restyle(text_col + x, text_col + x + 1, _S_CELL) + + # -- minimap ---------------------------------------------------------- # + def _minimap(self) -> list[list[int]]: + """A coarse occupancy grid of the whole graph: 0 empty, 1 edge, 2 block, + 3 the cursor's block. Cached per (layout, zoom, cursor block).""" + key = (id(self.lay), self._zoom, self.cursor_node) + if self._mini_cache and self._mini_cache[0] == key: + return self._mini_cache[1] + gw, gh = _MINI_W - 2, _MINI_H - 2 + grid = [[0] * gw for _ in range(gh)] + lay = self.lay + if lay is not None and lay.width and lay.height: + sx = max(lay.width / gw, 1e-9) + sy = max(lay.height / gh, 1e-9) + for lo, hi, col, _kind, _eid in lay.painting.vruns: + c = min(int(col / sx), gw - 1) + for r in range(min(int(lo / sy), gh - 1), + min(int(hi / sy), gh - 1) + 1): + if not grid[r][c]: + grid[r][c] = 1 + for n in lay.nodes: + mark = 3 if n.id == self.cursor_node else 2 + r0, r1 = int(n.y / sy), int((n.y + n.h - 1) / sy) + c0, c1 = int(n.x / sx), int(n.right / sx) + for r in range(max(r0, 0), min(r1, gh - 1) + 1): + for c in range(max(c0, 0), min(c1, gw - 1) + 1): + if grid[r][c] < mark: + grid[r][c] = mark + self._mini_cache = (key, grid) + return grid + + def _draw_minimap_row(self, out: _CellRow, y: int, width: int) -> None: + # Inset by two: a ScrollView paints its vertical scrollbar over the last + # column, which otherwise eats the minimap's right border. + if self._minimap_rect() is None or not (0 <= y < _MINI_H): + return + left = self._minimap_rect()[0] # one source of truth with the hit-test + grid = self._minimap() + gw, gh = _MINI_W - 2, _MINI_H - 2 + lay = self.lay + sx = max(lay.width / gw, 1e-9) + sy = max(lay.height / gh, 1e-9) + vy0 = int(self.scroll_offset.y / sy) + vy1 = int((self.scroll_offset.y + self.size.height) / sy) + vx0 = int(self.scroll_offset.x / sx) + vx1 = int((self.scroll_offset.x + width) / sx) + + if y == 0: + out.put(left, graph.BOX["tl"], _S_MINI_BG) + for i in range(1, _MINI_W - 1): + out.put(left + i, graph.BOX["h"], _S_MINI_BG) + out.put(left + _MINI_W - 1, graph.BOX["tr"], _S_MINI_BG) + tag = f" {len(lay.nodes)} blocks " + out.text(left + 2, tag, _S_MINI_BG) + return + if y == _MINI_H - 1: + out.put(left, graph.BOX["bl"], _S_MINI_BG) + for i in range(1, _MINI_W - 1): + out.put(left + i, graph.BOX["h"], _S_MINI_BG) + out.put(left + _MINI_W - 1, graph.BOX["br"], _S_MINI_BG) + return + r = y - 1 + out.put(left, graph.BOX["v"], _S_MINI_BG) + out.put(left + _MINI_W - 1, graph.BOX["v"], _S_MINI_BG) + for c in range(gw): + v = grid[r][c] if r < len(grid) else 0 + inview = vy0 <= r <= vy1 and vx0 <= c <= vx1 + if v == 3: + ch, st = "█", _S_MINI_CUR + elif v == 2: + ch, st = "█", _S_MINI_NODE + elif v == 1: + ch, st = "·", _S_MINI_EDGE + else: + ch, st = " ", _S_MINI_BG + if inview and v != 3: + st = _S_MINI_VIEW if v == 0 else st + Style(bgcolor="#233044") + out.put(left + 1 + c, ch, st) + + +# --------------------------------------------------------------------------- # # Function list panel # --------------------------------------------------------------------------- # class FunctionsPanel(Vertical): @@ -1770,6 +3130,58 @@ class FunctionsPanel(Vertical): # --------------------------------------------------------------------------- # +# Shared list navigation for the overlays +# --------------------------------------------------------------------------- # +class OptionListNav: + """Cursor + paging for an overlay whose filter ``Input`` keeps the focus. + + These screens focus a filter box, not the list, so the ``OptionList``'s own + bindings never fire -- the Input sees every key first. These actions forward + to the list on its behalf. + + Paging delegates to the widget's OWN ``action_page_up``/``action_page_down`` + rather than reimplementing it: those know the live viewport height, skip + disabled options and clamp at both ends. A hand-rolled "move by N" here + would have to guess the height and would drift from the list that DOES have + focus (``XrefsScreen``, or the struct list), which pages natively. + + Textual only merges ``BINDINGS`` from ``DOMNode`` subclasses, so a plain + mixin's are silently dropped -- every screen must splat ``*NAV_BINDINGS`` + (or list its own keys) explicitly. Same trap as ``SearchMixin``. + """ + + #: The common key set. ``StructEditor`` deliberately does NOT use this: it + #: binds ctrl+n to "new type", so it lists bare up/down itself. + NAV_BINDINGS = [ + Binding("down,ctrl+n", "cursor_down", show=False), + Binding("up,ctrl+p", "cursor_up", show=False), + Binding("pagedown", "page_down", show=False), + Binding("pageup", "page_up", show=False), + ] + + def _nav_list(self): + """The list to drive, or None when there is nothing to move through.""" + ol = self.query_one(OptionList) + return ol if ol.option_count else None + + def action_cursor_down(self) -> None: + if (ol := self._nav_list()) is not None: + ol.highlighted = min((ol.highlighted or 0) + 1, ol.option_count - 1) + + def action_cursor_up(self) -> None: + if (ol := self._nav_list()) is not None: + ol.highlighted = max((ol.highlighted or 0) - 1, 0) + + def action_page_down(self) -> None: + if (ol := self._nav_list()) is not None: + ol.action_page_down() + + def action_page_up(self) -> None: + if (ol := self._nav_list()) is not None: + ol.action_page_up() + + +# --------------------------------------------------------------------------- # # Xrefs popup # --------------------------------------------------------------------------- # class XrefsScreen(ModalScreen): @@ -1787,8 +3199,8 @@ class XrefsScreen(ModalScreen): self._preselect = preselect def compose(self) -> ComposeResult: - with Vertical(id="xref-box"): - yield Static(self._label, id="xref-title") + with Vertical(id="xref-box") as box: + box.border_title = Text(self._label.strip()) yield OptionList(*[Option(text) for _, text in self._items], id="xref-list") def on_mount(self) -> None: @@ -1841,13 +3253,12 @@ def _fuzzy(name: str, q: str): return (score, tuple(pos)) -class SymbolPalette(ModalScreen): +class SymbolPalette(OptionListNav, ModalScreen): """A command-palette overlay: type to fuzzy-find a symbol, Enter opens it.""" BINDINGS = [ Binding("escape", "close", "Close"), - Binding("down,ctrl+n", "cursor_down", show=False), - Binding("up,ctrl+p", "cursor_up", show=False), + *OptionListNav.NAV_BINDINGS, # F2, not ctrl+a: the focused Input binds "home,ctrl+a" so it would never # reach us. Function keys are untouched by Input. Binding("f2", "scope", "This binary / whole project", show=False), @@ -1868,8 +3279,8 @@ class SymbolPalette(ModalScreen): self._results: list[tuple] = [] def compose(self) -> ComposeResult: - with Vertical(id="pal-box"): - yield Static(" symbols", id="pal-title", markup=False) + with Vertical(id="pal-box") as box: + box.border_title = Text("symbols") yield Input(placeholder="fuzzy find symbol… ↑↓ select · Enter open · Esc close", id="pal-input") yield OptionList(id="pal-list") @@ -1948,18 +3359,8 @@ class SymbolPalette(ModalScreen): more = "+" if len(self._results) == cap else "" hint = " (F2: this binary)" if self._project_scope else ( " (F2: whole project)" if self._index is not None else "") - self.query_one("#pal-title", Static).update( - f" symbols [{scope}]: {len(self._results)}{more}{hint}") - - 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) + self.query_one("#pal-box").border_title = Text( + f"symbols [{scope}]: {len(self._results)}{more}{hint}") def action_choose(self) -> None: ol = self.query_one(OptionList) @@ -1986,14 +3387,178 @@ def _str_display(text: str, limit: int = 200) -> str: return out[:limit] + ("\u2026" if len(out) > limit else "") -class StringsPalette(ModalScreen): +class SearchPalette(OptionListNav, ModalScreen): + """Ctrl+F: search the whole database, by text or by bytes. + + Unlike every other palette here this does NOT filter as you type: each + search walks the image or the listing in the database process, so it runs + on Enter. That gives Enter two jobs, which is fine as long as it is never + ambiguous: while the query differs from what was last searched, Enter + searches; once the results on screen belong to the query in the box, Enter + opens the highlighted one. The title says which it will do. + + Mode is guessed from the query (idatui/search.py) because asking first is a + tax on every search; F2 overrides the guess and a `hex:`/`text:` prefix + settles it outright. + """ + + BINDINGS = [ + Binding("escape", "close", "Close"), + *OptionListNav.NAV_BINDINGS, + Binding("f2", "mode", "Text / bytes", show=False), + ] + LIMIT = 500 + + def __init__(self, program, initial: str = "") -> None: + super().__init__() + self._program = program + self._initial = initial + self._forced: str | None = None # F2: pin the mode + self._hits: list = [] + self._searched: tuple[str, str] | None = None # (mode, query) on screen + self._busy = False + + def compose(self) -> ComposeResult: + with Vertical(id="pal-box") as box: + box.border_title = Text("search") + yield Input(placeholder="text, or bytes like 48 8b ?? c3 \u00b7 " + "Enter search \u00b7 F2 mode \u00b7 Esc close", + id="pal-input") + yield OptionList(id="pal-list") + + def on_mount(self) -> None: + inp = self.query_one("#pal-input", Input) + inp.value = self._initial + inp.focus() + self._retitle() + if self._initial: + self._run() + + # -- mode / title ------------------------------------------------------- # + def _query(self) -> str: + return self.query_one("#pal-input", Input).value.strip() + + def _mode_query(self) -> tuple[str, str]: + return search.classify(self._query(), self._forced) + + def _retitle(self, note: str = "") -> None: + mode, q = self._mode_query() + pinned = "" if self._forced is None else "*" + state = note + if not state: + if self._busy: + state = "searching\u2026" + elif self._searched == (mode, q) and q: + n = len(self._hits) + state = f"{n} hit{'' if n == 1 else 's'} \u2014 Enter opens" + elif q: + state = "Enter searches" + self.query_one("#pal-box").border_title = Text( + f"search [{mode}{pinned}]" + (f": {state}" if state else "")) + + def action_mode(self) -> None: + mode, _ = self._mode_query() + self._forced = search.TEXT if mode == search.BYTES else search.BYTES + self._searched = None # the results on screen are for the old mode + self._retitle() + + def on_input_changed(self, event: Input.Changed) -> None: + event.stop() # modal inputs bubble to the app's own #search handler + self._retitle() + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + mode, q = self._mode_query() + if self._searched == (mode, q) and self._hits: + self.action_choose() + else: + self._run() + + # -- searching ---------------------------------------------------------- # + def _run(self) -> None: + mode, q = self._mode_query() + if not q: + return + if mode == search.BYTES: + problem = search.pattern_problem(q) + if problem: + # Refuse here rather than round-tripping: IDA's own message for + # a bad pattern is empty about half the time. + self._hits, self._searched = [], None + self.query_one(OptionList).clear_options() + self._retitle(problem) + return + q = search.normalise_pattern(q) + self._busy = True + self._retitle() + self._search(mode, q) + + @work(thread=True, exclusive=True, group="dbsearch") + def _search(self, mode: str, query: str) -> None: + try: + hits, err, truncated = self._program.search(query, mode, + limit=self.LIMIT) + except Exception as e: # noqa: BLE001 -- a search must not kill the app + hits, err, truncated = [], str(e), False + self.app.call_from_thread(self._present, mode, query, hits, err, truncated) + + def _present(self, mode: str, query: str, hits: list, err: str | None, + truncated: bool) -> None: + self._busy = False + self._hits = hits + # Remember what these results ARE, not what the box says now: the user + # may have typed on while the search ran, and then Enter must search + # again rather than open a hit from the previous query. + self._searched = (mode, query) if err is None else None + ol = self.query_one(OptionList) + ol.clear_options() + opts = [] + for h in hits: + label = Text() + label.append(f"{h.addr:08X} ", _S_ADDR) + label.append(f"{(h.func or h.seg or ''):<22.22} ", _S_DIM) + body = Text(h.line or "") + if mode == search.TEXT and query: + low, ql = (h.line or "").lower(), query.lower() + at = low.find(ql) + if at >= 0: + body.stylize(_S_NAME_MATCH, at, at + len(query)) + label.append_text(body) + opts.append(Option(label)) + ol.add_options(opts) + if hits: + ol.highlighted = 0 + if err: + self._retitle(err) + elif not hits: + self._retitle("no match") + else: + n = len(hits) + self._retitle(f"{n}{'+' if truncated else ''} " + f"hit{'' if n == 1 else 's'} \u2014 Enter opens") + + # -- moving / choosing --------------------------------------------------- # + def action_choose(self) -> None: + ol = self.query_one(OptionList) + i = ol.highlighted + if i is not None and 0 <= i < len(self._hits): + self.dismiss(self._hits[i]) + + def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: + if 0 <= event.option_index < len(self._hits): + self.dismiss(self._hits[event.option_index]) + + def action_close(self) -> None: + self.dismiss(None) + + +class StringsPalette(OptionListNav, ModalScreen): """Every string in the binary (IDA's Shift+F12), filterable; Enter jumps to it in the unified listing.""" BINDINGS = [ Binding("escape", "close", "Close"), - Binding("down,ctrl+n", "cursor_down", show=False), - Binding("up,ctrl+p", "cursor_up", show=False), + *OptionListNav.NAV_BINDINGS, Binding("f2", "scope", "This binary / whole project", show=False), ] LIMIT = 500 @@ -2014,8 +3579,8 @@ class StringsPalette(ModalScreen): self._results: list[tuple] = [] def compose(self) -> ComposeResult: - with Vertical(id="pal-box"): - yield Static(" strings", id="pal-title", markup=False) + with Vertical(id="pal-box") as box: + box.border_title = Text("strings") yield Input(placeholder="filter strings\u2026 \u2191\u2193 select \u00b7 " "Enter jump \u00b7 Esc close", id="pal-input") yield OptionList(id="pal-list") @@ -2088,18 +3653,8 @@ class StringsPalette(ModalScreen): more = "+" if len(rows) == cap else "" hint = " (F2: this binary)" if self._project_scope else ( " (F2: whole project)" if self._index is not None else "") - self.query_one("#pal-title", Static).update( - f" strings [{scope}]: {len(self._results)}{more} of {len(self._rows)}{hint}") - - 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) + self.query_one("#pal-box").border_title = Text( + f"strings [{scope}]: {len(self._results)}{more} of {len(self._rows)}{hint}") def action_choose(self) -> None: ol = self.query_one(OptionList) @@ -2132,12 +3687,15 @@ _HELP = ( )), ("Views", ( ("Tab / F5", "disassembly \u21c4 pseudocode"), + ("Space", "control-flow graph \u21c4 text"), ("s", "split view: listing + pseudocode"), ("Tab", "in split: switch the driving pane"), ("\\", "hex view"), - ("o", "cycle the opcode-bytes column"), + ("B", "cycle the opcode-bytes column"), ("Ctrl+B", "show/hide the names pane"), ("Ctrl+T", "structs / types editor"), + ("Ctrl+F", "search the database: text or bytes"), + ("Ctrl+E", "export findings as markdown"), ("Ctrl+P", "command palette"), )), ("Move", ( @@ -2159,6 +3717,7 @@ _HELP = ( ("d", "make data"), ("a", "make string"), ("u", "undefine"), + ("o / O", "literal format: hex/dec/bin/char/offset"), ("Ctrl+S", "save the database"), )), ("Search", ( @@ -2166,19 +3725,37 @@ _HELP = ( ("?", "search backward"), ("N", "previous match"), ("Ctrl+Y", "copy the current line"), - ("F1", "this cheatsheet"), + ("F1 / H", "this cheatsheet"), ("q", "quit"), )), + ("Graph (Space)", ( + ("j / k", "line up/down, crossing blocks"), + ("h / l", "column left / right"), + ("J / K", "follow an edge to a successor / predecessor"), + ("w / b", "next / previous block in layout order"), + ("0", "jump to the entry block"), + ("z", "zoom: full \u2192 compact \u2192 collapsed"), + ("m", "show/hide the minimap"), + ("e", "layout engine: auto \u2192 native \u2192 triskel"), + ("f", "centre on the current block"), + ("Enter", "follow (stays in the graph if it lands here)"), + ("drag / click", "pan / put the cursor in a block"), + ("click minimap", "jump the view there (drag to scrub)"), + )), ) class QuitScreen(ModalScreen): - """Asked before exiting with unsaved database changes. Dismisses with - "save", "discard" or None (stay).""" + """Asked before exiting with unsaved database changes. + + Code Mode clients cannot roll a shared database back. The ``d`` choice means + "do not explicitly save": a GUI keeps the changes dirty, while a managed + idalib worker may persist them when its final lease closes. + """ BINDINGS = [ Binding("s", "save", "Save & quit"), - Binding("d", "discard", "Discard & quit"), + Binding("d", "discard", "Leave & quit"), Binding("escape,c", "cancel", "Cancel"), ] @@ -2189,13 +3766,13 @@ class QuitScreen(ModalScreen): def compose(self) -> ComposeResult: what = (f"{len(self._labels)} databases have unsaved changes" if len(self._labels) > 1 else "unsaved changes") - with Vertical(id="quit-box"): - yield Static(f"\u26a0 {what}", id="quit-title") + with Vertical(id="quit-box") as box: + box.border_title = Text(f"\u26a0 {what}") body = Text() for label in self._labels: body.append(f" \u2022 {label}\n", _S_LABEL) yield Static(body, id="quit-list") - yield Static("s save & quit d discard & quit Esc cancel", + yield Static("s save & quit d leave as-is & quit Esc cancel", id="quit-help") def action_save(self) -> None: @@ -2211,7 +3788,7 @@ class QuitScreen(ModalScreen): class HelpScreen(ModalScreen): """F1: the keyboard cheatsheet, replacing the permanent footer.""" - BINDINGS = [Binding("escape,f1,q,question_mark", "close", "Close")] + BINDINGS = [Binding("escape,f1,H,q,question_mark", "close", "Close")] #: widest cell content, +2 for the card's border, +2 for its padding _CARD_PAD = 4 @@ -2222,8 +3799,8 @@ class HelpScreen(ModalScreen): avail = max(self.app.size.width - 6, 20) cols = self._columns(avail) per = -(-len(_HELP) // cols) # ceil, so the columns stay balanced - with Vertical(id="help-box"): - yield Static(" keys", id="help-title", markup=False) + with Vertical(id="help-box") as box: + box.border_title = Text("keys") # Still inside a scroll container, so a genuinely tiny terminal # degrades to scrolling rather than clipping — but spread across the # width it shouldn't come to that. @@ -2239,7 +3816,7 @@ class HelpScreen(ModalScreen): classes="help-card", markup=False) card.border_title = title yield card - yield Static("Esc / F1 to close", id="help-foot") + yield Static("Esc · F1 · H to close", id="help-foot") @staticmethod def _section_widths() -> list[int]: @@ -2286,7 +3863,211 @@ class HelpScreen(ModalScreen): self.dismiss(None) -class LoadOptionsScreen(ModalScreen): +class RegWriteScreen(OptionListNav, ModalScreen): + """Registers, and the instruction that set each one. + + "Which instruction set this register to its current value?" is the question + a trace exists to answer, and seeking backwards to it is a single keypress + here rather than a manual walk. Forward is offered too, but backward is what + people actually want — you notice a bad value after it has been used. + """ + + BINDINGS = [ + Binding("escape", "close", "Close"), + *OptionListNav.NAV_BINDINGS, + Binding("enter", "choose", show=False, priority=True), + Binding("f", "choose_forward", show=False), + ] + + def __init__(self, rows, idx: int) -> None: + super().__init__() + self._rows = rows # (name, value, last_write, next_write) + self._idx = idx + + def compose(self) -> ComposeResult: + with Vertical(id="pal-box") as box: + box.border_title = Text(f"registers at t={self._idx:,}") + box.border_subtitle = Text("Enter seeks to the write \u00b7 f seeks forward") + yield OptionList(id="pal-list") + + def on_mount(self) -> None: + ol = self.query_one(OptionList) + opts = [] + for name, val, last, nxt in self._rows: + label = Text() + label.append(f" {name:>4} ", _S_MNEM) + label.append(f"{val:#018x} " if val > 0xFFFFFFFF else f"{val:#010x} ", + _S_INSN) + if last is None: + label.append("never written in this trace", _S_DIM) + elif last == self._idx: + label.append("set by THIS instruction", _S_DATA) + else: + label.append(f"set at t={last:,}", _S_LABEL) + label.append(f" ({self._idx - last:,} steps back)", _S_DIM) + if nxt is not None: + label.append(f" next t={nxt:,}", _S_DIM) + opts.append(Option(label)) + ol.add_options(opts) + ol.highlighted = 0 + ol.focus() + + def _pick(self, forward: bool) -> None: + i = self.query_one(OptionList).highlighted + if i is None or not (0 <= i < len(self._rows)): + self.dismiss(None) + return + _name, _val, last, nxt = self._rows[i] + self.dismiss(nxt if forward else last) + + def action_choose(self) -> None: + self._pick(False) + + def action_choose_forward(self) -> None: + self._pick(True) + + def on_option_list_option_selected(self, event) -> None: # type: ignore[no-untyped-def] + self._pick(False) + + def action_close(self) -> None: + 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 Static("", id="trace-stack", 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) + self._render_stack(t) + tl = self.query_one(TraceTimeline) + tl.idx = self.idx + tl.refresh() + + + STACK_WORDS = 8 + + def _render_stack(self, t) -> None: # type: ignore[no-untyped-def] + """The stack as of this instant, read out of the trace. + + This is where a trace's memory actually is: on two real traces, NONE of + the accesses fell inside the image — every one was stack or heap. A + memory view that could only address the image would have nothing to show. + + Bytes the trace never saw are printed as '??' rather than zeros. A trace + knows what it observed and nothing else, and quietly rendering unseen + memory as zero would invent facts. + """ + sp_name = next((r for r in ("rsp", "esp", "sp") if r in t.reg_at), "") + sp = t.register(sp_name, self.idx) if sp_name else None + out = Text() + if sp is None: + self.query_one("#trace-stack", Static).update(out) + return + width = 8 if (t.info and "64" in (t.info.arch or "")) else 4 + out.append(f" stack ({sp_name})\n", _S_DIM) + for k in range(self.STACK_WORDS): + a = sp + k * width + data, known = t.memory_raw(a, width, self.idx) + out.append(" \u25b8" if k == 0 else " ", _S_MNEM) + out.append(f"{a:012x} ", _S_ADDR) + if all(known): + v = int.from_bytes(data, "little") + out.append(f"{v:0{width * 2}x}\n", _S_DATA if k == 0 else _S_INSN) + elif any(known): + out.append("".join(f"{b:02x}" if known[i] else "??" + for i, b in enumerate(data)) + "\n", _S_INSN) + else: + out.append("?" * (width * 2) + "\n", _S_SEP) + self.query_one("#trace-stack", Static).update(out) + + +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(OptionListNav, 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 @@ -2301,8 +4082,7 @@ class LoadOptionsScreen(ModalScreen): BINDINGS = [ Binding("escape", "close", "Close"), - Binding("down,ctrl+n", "cursor_down", show=False), - Binding("up,ctrl+p", "cursor_up", show=False), + *OptionListNav.NAV_BINDINGS, Binding("enter", "choose", show=False, priority=True), ] @@ -2318,9 +4098,8 @@ class LoadOptionsScreen(ModalScreen): 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) + with Vertical(id="pal-box") as box: + box.border_title = Text("unrecognised file \u2014 how should IDA load it?") 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) @@ -2383,18 +4162,8 @@ class LoadOptionsScreen(ModalScreen): 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) + self.query_one("#pal-box").border_title = Text( + f"unrecognised file \u2014 processor? ({len(rows)})") def action_choose(self) -> None: ol = self.query_one(OptionList) @@ -2425,14 +4194,13 @@ class LoadOptionsScreen(ModalScreen): self.dismiss({}) -class ProjectPalette(ModalScreen): +class ProjectPalette(OptionListNav, 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).""" BINDINGS = [ Binding("escape", "close", "Close"), - Binding("down,ctrl+n", "cursor_down", show=False), - Binding("up,ctrl+p", "cursor_up", show=False), + *OptionListNav.NAV_BINDINGS, ] def __init__(self, entries: list[dict]) -> None: @@ -2441,8 +4209,8 @@ class ProjectPalette(ModalScreen): self._results: list[dict] = [] def compose(self) -> ComposeResult: - with Vertical(id="pal-box"): - yield Static(" binaries", id="pal-title", markup=False) + with Vertical(id="pal-box") as box: + box.border_title = Text("binaries") yield Input(placeholder="filter binaries\u2026 \u2191\u2193 select \u00b7 " "Enter switch \u00b7 Esc close", id="pal-input") yield OptionList(id="pal-list") @@ -2489,18 +4257,8 @@ class ProjectPalette(ModalScreen): # you are rather than at whatever sorts first. active = next((i for i, e in enumerate(rows) if e["active"]), 0) ol.highlighted = active - self.query_one("#pal-title", Static).update( - f" binaries: {len(rows)} of {len(self._entries)}") - - 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) + self.query_one("#pal-box").border_title = Text( + f"binaries: {len(rows)} of {len(self._entries)}") def action_choose(self) -> None: i = self.query_one(OptionList).highlighted @@ -2532,7 +4290,8 @@ class ConfirmScreen(ModalScreen): self._note = note def compose(self) -> ComposeResult: - with Vertical(id="confirm-box"): + with Vertical(id="confirm-box") as box: + box.border_title = Text("\u26a0 confirm") yield Static(self._message, id="confirm-msg", markup=False) if self._note: yield Static(self._note, id="confirm-note", markup=False) @@ -2545,8 +4304,43 @@ class ConfirmScreen(ModalScreen): self.dismiss(False) -_LOGO_PATH = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logo.ans") +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_LOGO_PATH = os.path.join(_REPO_ROOT, "logo.ans") +#: The same artwork as a real image, for terminals that can draw one. logo.ans +#: is half-blocks (two pixels per cell); this is a transparent PNG at 768px. +LOGO_PNG = os.path.join(_REPO_ROOT, "logo.png") +_LOGO_BOX = (60, 33) # the most room the splash will give the art +#: Rows the loading box spends on everything that is not the artwork: border 2, +#: padding 2, the art's margin 1, title 1, note 1 + margin 1, help 1 + margin 1. +LOGO_CHROME_ROWS = 10 +#: Below this the image is a postage stamp; show the text splash instead. +LOGO_MIN_ROWS = 8 +_logo_cells: tuple[int, int] | None = None + + +def logo_cells(max_rows: int | None = None) -> tuple[int, int]: + """Cell footprint for the image, derived from the artwork and the terminal's + real cell size rather than hardcoded. + + Cells are nowhere near square (9x22 px here, 1:2.44), so a fixed box picked + for one aspect ratio stretches any other. Recomputing means the art can be + replaced without anyone remembering to edit a constant. + + ``max_rows`` shrinks it to the room actually available. The terminal scales + the image into whatever cell box we place it in, so there is no reason for + the splash to be all-or-nothing -- and it WAS all-or-nothing: a 31-row pane + is one row short of the natural size, so the logo silently disappeared + rather than being drawn a little smaller. + """ + global _logo_cells + if _logo_cells is None: + px = kittygfx.png_size(LOGO_PNG) + _logo_cells = kittygfx.fit(px, *_LOGO_BOX) if px else _LOGO_BOX + if max_rows is None or max_rows >= _logo_cells[1]: + return _logo_cells + px = kittygfx.png_size(LOGO_PNG) + return (kittygfx.fit(px, _LOGO_BOX[0], max(max_rows, 1)) if px + else (_LOGO_BOX[0], max(max_rows, 1))) _logo_cache: object = False # False == not yet loaded (None == absent/unreadable) @@ -2577,17 +4371,45 @@ class LoadingScreen(ModalScreen): super().__init__() self._title = title self._note = note + self._image = False # drawing the real image, not the block art + self._cells: tuple[int, int] | None = None # image size, in cells + self._last_place = 0.0 # throttles re-anchoring after a repaint + + def _room(self) -> int: + """Rows left for artwork once the box's own furniture is paid for.""" + return self.app.size.height - LOGO_CHROME_ROWS + + def _fits(self, rows: int) -> bool: + """Room for art of exactly ``rows`` (the block art cannot be resized).""" + return self._room() >= rows and self.app.size.width >= 64 def compose(self) -> ComposeResult: with Vertical(id="loading-box"): - logo = _load_logo() - # Only show the splash art when the terminal can fit it plus the - # title/note/help + box chrome; otherwise fall back to a text-only - # overlay so nothing important is clipped off a small screen. - if logo is not None: - sz = self.app.size - n = len(logo.split("\n")) - if sz.height >= n + 9 and sz.width >= 64: + # A terminal that can draw a real image gets one: same artwork, + # same cell footprint, ~10x the linear resolution of the block art. + # The image is anchored to screen cells rather than composited by + # Textual (no unicode-placeholder support here), so the widget is + # only reserved blank space -- see _place_logo. + # Scale the image to the room there is, rather than demanding its + # natural size and vanishing when one row is missing. + room = self._room() + cols, rows = logo_cells(room) + self._cells = (cols, rows) + kittygfx.log(f"compose: supported={kittygfx.supported()} " + f"app.size={self.app.size} room={room} " + f"cells={cols}x{rows} natural={logo_cells()}") + if (kittygfx.supported() and self.app.size.width >= 64 + and room >= LOGO_MIN_ROWS): + self._image = True + blank = Static("\n" * (rows - 1), id="loading-image") + blank.styles.height = rows + yield blank + else: + logo = _load_logo() + # Only show the splash art when the terminal can fit it plus + # the title/note/help + box chrome; otherwise fall back to a + # text-only overlay so nothing important is clipped. + if logo is not None and self._fits(len(logo.split("\n"))): # Align.center, not the box's align-horizontal: the 1fr # title/note siblings make the child group span the full # width, so container alignment has nothing left to centre. @@ -2602,6 +4424,61 @@ class LoadingScreen(ModalScreen): self.query_one("#loading-note", Static).update(text) except Exception: # noqa: BLE001 -- not mounted yet / already gone pass + # Textual doesn't know the image is there, so a repaint can drop it. + # Re-anchoring is one short escape with no image data; throttled so a + # chatty progress callback can't turn it into a flicker. + if self._image: + now = time.monotonic() + if now - self._last_place > 0.2: + self._last_place = now + self._place_logo() + + # -- the image, which Textual knows nothing about ---------------------- # + def _place_logo(self) -> None: + """Anchor the image over the blank cells reserved for it. + + Deferred to after a refresh because a widget has no screen region until + it has been laid out, and re-run on resize because the region moves. + """ + if not self._image: + return + try: + region = self.query_one("#loading-image", Static).region + except Exception as e: # noqa: BLE001 -- gone already + kittygfx.log(f"place_logo: no widget ({e})") + return + kittygfx.log(f"place_logo: region={region}") + if not region.width or not region.height: + return + # The reserved region is the truth about how much room there is; the + # image is scaled into exactly it, so a resize needs no relayout. + cols, rows = self._cells or logo_cells() + rows = min(rows, region.height) + col = region.x + max((region.width - cols) // 2, 0) # centre it + kittygfx.place(region.y, col, min(cols, region.width), rows) + + def on_mount(self) -> None: + if not self._image: + return + # Upload HERE, not from the launcher: Textual is on the alternate screen + # by now, and an image uploaded to the primary screen cannot be placed + # from the alternate one -- the placement reports success and draws + # nothing at all. + if not kittygfx.upload(LOGO_PNG): + self._image = False + return + self.call_after_refresh(self._place_logo) + + def on_resize(self) -> None: + if self._image: + kittygfx.clear() + self.call_after_refresh(self._place_logo) + + def on_unmount(self) -> None: + # The image is anchored to the screen, not owned by the compositor, so + # it would sit there over the disassembly forever if we didn't say so. + if self._image: + kittygfx.clear() def action_hide(self) -> None: self.dismiss() @@ -2637,14 +4514,27 @@ class StructEditor(ModalScreen): selected struct; Ctrl+S declares (creates or updates) it; Ctrl+N starts a new one; Delete removes the highlighted struct; Esc returns to the list then closes. + + ``/`` from the list opens a fuzzy filter over the struct names -- the same + key and the same matcher as the code views' search and the symbol palette, + rather than a third way to find something by typing. """ BINDINGS = [ Binding("ctrl+s", "save", "Save", priority=True), Binding("ctrl+n", "new", "New", priority=True), Binding("ctrl+y", "copy", "Copy", priority=True), + Binding("slash", "filter", "Filter", show=False), Binding("delete,d", "delete", "Delete", show=False), Binding("escape", "close", "Close"), + # Only ever reached while the FILTER has focus: a focused OptionList + # consumes these itself, so they move its highlight from the prompt + # (type to narrow, arrow to pick, exactly like the symbol palette). + # NOT OptionListNav.NAV_BINDINGS: ctrl+n is "new type" here. + Binding("up", "cursor_up", "Up", show=False), + Binding("down", "cursor_down", "Down", show=False), + Binding("pageup", "page_up", show=False), + Binding("pagedown", "page_down", show=False), ] NEW_TEMPLATE = "struct NewStruct\n{\n int field;\n};\n" @@ -2654,7 +4544,9 @@ class StructEditor(ModalScreen): def __init__(self, program: Program) -> None: super().__init__() self._program = program - self._structs: list[Struct] = [] + self._all: list[Struct] = [] # every struct the database has + self._structs: list[Struct] = [] # the VISIBLE rows (== _all when unfiltered) + self._filter = "" self._loaded: str | None = None # name currently in the editor self._loaded_src: str | None = None # its source, to detect unsaved edits @@ -2677,16 +4569,23 @@ class StructEditor(ModalScreen): then() def compose(self) -> ComposeResult: - with Vertical(id="se-box"): + with Vertical(id="se-box") as box: + box.border_title = Text("structs / types") + box.border_subtitle = Text("Esc close") with Horizontal(id="se-panes"): with Vertical(id="se-left"): - yield Static(" structs", id="se-title") + yield Static("structs", id="se-title") + yield Input(placeholder="fuzzy filter\u2026 \u2191\u2193 pick \u00b7 " + "Enter edit \u00b7 Esc clear", + id="se-filter") yield OptionList(id="se-list") with Vertical(id="se-right"): - yield Static(" C definition", id="se-hint") - yield TextArea("", id="se-edit") + yield Static("C definition", id="se-hint") + yield CTextArea("", id="se-edit") + # Esc is on the border subtitle; repeating it here cost the row. yield Static( - "Enter edit · Ctrl+S save · Ctrl+Y copy · Ctrl+N new · d/Del delete · Esc close", + "Enter edit · / filter · Ctrl+S save · Ctrl+Y copy · Ctrl+N new · " + "d/Del delete", id="se-status") def on_mount(self) -> None: @@ -2704,26 +4603,153 @@ class StructEditor(ModalScreen): self.app.call_from_thread(self._populate, structs, select) def _populate(self, structs: list[Struct], select: str | None) -> None: - self._structs = structs + self._all = structs + self._show(select) + + def _show(self, select: str | None = None) -> None: + """Rebuild the list from ``_all`` through the current filter. + + ``_structs`` stays the VISIBLE rows, because every action here indexes + it by the list's highlighted row -- load, delete and the confirm dialogs + would all address the wrong struct if the two ever drifted apart. + """ + q = self._filter.strip() + # A struct we were told to select but the filter would hide (a rename, or + # one just created under a stale filter) beats the filter -- otherwise a + # successful save looks like the struct vanished. + if q and select is not None and _fuzzy(select, q) is None: + inp = self.query_one("#se-filter", Input) + inp.value = "" + inp.display = False + self._filter = q = "" + rows: list[tuple[Struct, tuple[int, ...]]] = [] + if q: + scored = [] + for s in self._all: + m = _fuzzy(s.name, q) + if m is not None: + scored.append((m[0], m[1], s)) + scored.sort(key=lambda t: (-t[0], t[2].name)) + rows = [(s, pos) for _, pos, s in scored] + else: + rows = [(s, ()) for s in self._all] + self._structs = [s for s, _ in rows] + ol = self.query_one("#se-list", OptionList) ol.clear_options() - for s in structs: + # Ragged rows read as noise, so the name column is padded to the widest + # name actually present (capped, so one monstrous C++ mangling can't + # push the size/count columns off the pane). + width = min(max((len(s.name) for s, _ in rows), default=0), 22) + opts = [] + for s, pos in rows: kw = "union" if s.is_union else "struct" + name = s.name if len(s.name) <= width else s.name[:width - 1] + "\u2026" label = Text() - label.append(s.name, _S_LABEL) - label.append(f" {s.size:#x} {s.members}f {kw}", _S_DIM) - ol.add_option(Option(label)) - if structs: + nm = Text(f"{name:<{width}}", style=_S_LABEL) + for p in pos: + if p < len(name): + nm.stylize(_S_NAME_MATCH, p, p + 1) + label.append_text(nm) + label.append(f"{s.size:>#7x}{s.members:>4}f {kw}", _S_DIM) + opts.append(Option(label)) + ol.add_options(opts) + if rows: idx = 0 if select is not None: - idx = next((i for i, s in enumerate(structs) if s.name == select), 0) + idx = next((i for i, s in enumerate(self._structs) + if s.name == select), 0) ol.highlighted = idx + cap = "structs" + if q: + cap = f"structs {len(rows)}/{len(self._all)}" + self.query_one("#se-title", Static).update(cap) def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: i = event.option_index if 0 <= i < len(self._structs): self._confirm_discard(lambda n=self._structs[i].name: self._load(n)) + # -- filter ("/" from the list) ----------------------------------------- # + def action_filter(self) -> None: + """'/': fuzzy-filter the struct list, as you type.""" + # A '/' typed into the definition is a division operator, and one typed + # into the filter is the Input's business; only the list opens this. + if self.focused is not self.query_one("#se-list", OptionList): + return + inp = self.query_one("#se-filter", Input) + inp.display = True + inp.value = self._filter + inp.focus() + + def _clear_filter(self, focus_list: bool = True) -> None: + inp = self.query_one("#se-filter", Input) + inp.value = "" + inp.display = False + self._filter = "" + self._show(select=self._loaded) + if focus_list: + self.query_one("#se-list", OptionList).focus() + + def on_input_changed(self, event: Input.Changed) -> None: + # Modal Input messages bubble to the App (which owns #func-filter and + # the search prompt) -- stop them here or they drive the main screen. + if event.input.id != "se-filter": + return + event.stop() + self._filter = event.value + self._show() + + def on_input_submitted(self, event: Input.Submitted) -> None: + if event.input.id != "se-filter": + return + event.stop() + # Enter on a filtered list loads the highlighted struct, like Enter on + # the list itself -- typing a name and pressing Enter should open it. + ol = self.query_one("#se-list", OptionList) + i = ol.highlighted + if i is not None and 0 <= i < len(self._structs): + self._confirm_discard(lambda n=self._structs[i].name: self._load(n)) + else: + ol.focus() + + def _filter_focused(self) -> bool: + return self.focused is self.query_one("#se-filter", Input) + + def action_cursor_up(self) -> None: + self._move_highlight(-1) + + def action_cursor_down(self) -> None: + self._move_highlight(1) + + def action_page_up(self) -> None: + self._page(-1) + + def action_page_down(self) -> None: + self._page(1) + + def _list_from_filter(self): + """The type list, but only while the FILTER holds focus. + + When the list itself has focus it handles these keys natively, and + forwarding as well would move the highlight twice. + """ + if not self._filter_focused(): + return None + ol = self.query_one("#se-list", OptionList) + return ol if ol.option_count else None + + def _move_highlight(self, delta: int) -> None: + ol = self._list_from_filter() + if ol is not None: + ol.highlighted = max(0, min((ol.highlighted or 0) + delta, + ol.option_count - 1)) + + def _page(self, direction: int) -> None: + ol = self._list_from_filter() + if ol is not None: + (ol.action_page_down if direction > 0 else ol.action_page_up)() + @work(thread=True, exclusive=True, group="se-load") def _load(self, name: str) -> None: try: @@ -2782,6 +4808,8 @@ class StructEditor(ModalScreen): self._set_status(f"save failed — {msg}", error=True) return self._loaded = name + if name: + self.app.journal.record("type", None, name) if formatted: # Reformat the editor to IDA's canonical layout (keeps cursor at top). ta = self.query_one("#se-edit", TextArea) @@ -2833,6 +4861,7 @@ class StructEditor(ModalScreen): return if getattr(self.app, "_dirty", None) is not None: self.app._dirty = True + self.app.journal.record("del_type", None, name) self._refresh() self._set_status(f"deleted {name}") @@ -2848,10 +4877,15 @@ class StructEditor(ModalScreen): self._set_status(f"copied {what} ({n} chars) to clipboard") def action_close(self) -> None: - # A stray Esc while editing returns to the list instead of discarding. + # Esc backs out one level at a time: out of the definition, then out of + # the filter, and only then out of the dialog. Closing the editor from + # under a half-typed filter is the kind of thing you only do once. if self.focused is self.query_one("#se-edit", TextArea): self.query_one("#se-list", OptionList).focus() return + if self._filter_focused() or self._filter: + self._clear_filter() + return self._confirm_discard(lambda: self.dismiss(None), "Discard unsaved changes and close?") @@ -2902,7 +4936,12 @@ class IdaCommands(Provider): app.action_strings), ("Switch binary…", "another binary in the project (Ctrl+O)", app.action_switch_binary), - ("Keyboard shortcuts", "the key cheatsheet (F1)", app.action_help), + ("Search database…", "disassembly text or a byte pattern with " + "wildcards (Ctrl+F)", app.action_find), + ("Export findings…", "your comments, names and types as markdown " + "(Ctrl+E)", app.action_export), + ("Keyboard shortcuts", "the key cheatsheet (F1 or H)", + app.action_help), ("Follow symbol under cursor", "jump to the referenced symbol (Enter)", lambda: va("follow")), ("Show xrefs to symbol", "cross-references to the cursor symbol (x)", @@ -2915,6 +4954,14 @@ class IdaCommands(Provider): ("Hex view", "raw bytes at the cursor (\\)", app.action_hex), ("Split view (listing ⇄ pseudocode)", "side-by-side synced views (s)", app.action_toggle_split), + ("Graph view (control flow)", + "the function's basic blocks as a graph (Space)", + app.action_toggle_graph), + ("Graph: cycle zoom", + "full → compact → collapsed (z, in the graph)", + lambda: va("zoom")), + ("Graph: toggle minimap", + "the overview box (m, in the graph)", lambda: va("minimap")), ("Rename symbol…", "rename the symbol under the cursor (n)", lambda: va("rename")), ("Set type / prototype…", "retype the symbol under the cursor (y)", @@ -2929,7 +4976,19 @@ class IdaCommands(Provider): lambda: va("make_string")), ("Undefine", "undefine the item at the cursor (u)", lambda: va("undefine")), - ("Toggle opcode bytes", "cycle the opcode-bytes column (o)", + ("Literal format: next", "cycle the literal under the cursor (o)", + lambda: va("op_format", "cycle")), + ("Literal format: previous", "the other way round (O)", + lambda: va("op_format", "back")), + *((f"Literal format: {label}", f"show the literal as {label} ({fmt})", + (lambda f=fmt: va("op_format", f))) + for fmt, label in (("hex", "hexadecimal"), ("dec", "decimal"), + ("oct", "octal"), ("bin", "binary"), + ("char", "a character"), + ("offset", "an offset (reference)"), + ("stack", "a stack variable"), + ("default", "IDA's own choice"))), + ("Toggle opcode bytes", "cycle the opcode-bytes column (B)", lambda: va("toggle_opcodes")), ("Structs / types editor", "view + edit local types (Ctrl+T)", app.action_structs), @@ -2966,6 +5025,7 @@ class IdaTui(App): ListingView { width: 1fr; padding: 0 1; } #panes.split ListingView { border-right: tall $panel-lighten-2; } HexView { width: 1fr; padding: 0 1; } + GraphView { width: 1fr; padding: 0 1; } #search { height: 1; border: none; padding: 0 1; background: $primary-darken-2; color: $text; @@ -2990,42 +5050,77 @@ class IdaTui(App): height: 1; border: none; padding: 0 1; background: $primary-darken-3; color: $text; } + #export { + height: 1; border: none; padding: 0 1; + background: $success-darken-3; color: $text; + } #status { height: 1; background: $panel; color: $text; padding: 0 1; } .decomp-loading { width: 100%; height: 100%; content-align: center middle; background: $panel-darken-1; } - QuitScreen { align: center middle; } - #quit-box { width: 64; height: auto; border: thick $warning; background: $panel; } - #quit-title { dock: top; height: 1; background: $warning; color: $background; text-style: bold; padding: 0 1; } - #quit-list { height: auto; padding: 1 2 0 2; } - #quit-help { height: 1; color: $text-muted; padding: 0 2; margin-top: 1; } - HelpScreen { align: center middle; } - #help-box { width: auto; max-width: 98%; height: auto; max-height: 90%; - border: thick $accent; background: $panel; } - #help-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } + /* ---- modal chrome --------------------------------------------------- + One shape for every dialog: a THIN round border in a colour that + recedes, and the title set into the top border rather than painted as a + solid bar across the full width. The old `thick` frame plus an inverse + title bar was two heavy rectangles around content that is itself the + only thing worth looking at. Size stays per-box; everything else is + here, once. */ + #quit-box, #help-box, #xref-box, #pal-box, #se-box, #confirm-box, + #loading-box, #busy-box { + background: $panel; + border: round $panel-lighten-3; + border-title-color: $accent; + border-title-style: bold; + border-title-align: left; + border-subtitle-color: $text-muted; + border-subtitle-align: right; + } + /* The two that ask a question you can answer wrongly keep the warning hue. */ + #quit-box, #confirm-box { border: round $warning; + border-title-color: $warning; } + /* Lists inside a dialog are content, not another framed panel: drop the + stock OptionList border/background so the dialog has ONE edge. Focus + still reads from the highlight bar (bright when focused, dim when not). */ + #quit-box OptionList, #help-box OptionList, #xref-box OptionList, + #pal-box OptionList, #se-box OptionList { + border: none; background: transparent; padding: 0 1; + } + #quit-box { width: 64; height: auto; padding: 1 1 0 1; } + #quit-list { height: auto; padding: 0 1; } + #quit-help { height: 1; color: $text-muted; padding: 0 1; margin-top: 1; } + #help-box { width: auto; max-width: 98%; height: auto; max-height: 90%; } #help-body { height: auto; max-height: 100%; width: auto; padding: 1 1; } #help-cols { height: auto; width: auto; } .help-col { height: auto; width: auto; margin-right: 1; } .help-card { height: auto; width: auto; padding: 0 1; border: round $panel-lighten-2; } #help-foot { dock: bottom; height: 1; color: $text-muted; padding: 0 2; } - XrefsScreen { align: center middle; } - #xref-box { width: 84; max-height: 70%; height: auto; border: thick $accent; background: $panel; } - #xref-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } + #xref-box { width: 84; max-height: 70%; height: auto; padding: 0 0; } #xref-list { height: auto; max-height: 100%; } - /* every #pal-box palette centres, not just the symbol one */ - SymbolPalette, StringsPalette, ProjectPalette, - LoadOptionsScreen { align: center middle; } + /* A dialog centres because it is a dialog — not because someone remembered + to add it to a list. That list is how SearchPalette shipped pinned to the + top of the screen, and the comment that used to sit here ("every #pal-box + palette centres, not just the symbol one") was itself the second time. + ModalScreen matches subclasses, so every modal below inherits this and + the next one gets it for free; `modal_centering` in the pilot suite + fails if one ever opts out by accident. Textual's own Ctrl+P palette is a + ModalScreen too and wants its stock top alignment, so it opts out here, + deliberately and visibly. */ + ModalScreen { align: center middle; } + CommandPalette { align: center top; } /* 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; } CommandPalette #--results { width: 100%; } - #pal-box { width: 96; max-width: 92%; height: auto; max-height: 80%; - border: thick $accent; background: $panel; } - #pal-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } + #pal-box { width: 96; max-width: 92%; height: auto; max-height: 80%; } #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-stack { 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 @@ -3033,31 +5128,33 @@ class IdaTui(App): 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; } + #confirm-note { height: auto; color: $text-muted; } + #se-box { width: 90%; height: 84%; } #se-panes { height: 1fr; } - #se-left { width: 38; border-right: solid $accent; } + #se-left { width: 38; border-right: solid $panel-lighten-3; } #se-right { width: 1fr; } - #se-title, #se-hint { height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } + /* Column captions, not title bars: the dialog already has one title. */ + #se-title, #se-hint { height: 1; color: $text-muted; text-style: bold; + padding: 0 1; } + /* Hidden until '/'; same one-row prompt shape as the app's own prompts. */ + #se-filter { display: none; height: 1; border: none; padding: 0 1; + background: $primary-darken-2; color: $text; } #se-list { height: 1fr; } - #se-edit { height: 1fr; border: none; } - #se-status { height: 1; background: $panel-darken-2; color: $text-muted; padding: 0 1; } - ConfirmScreen { align: center middle; } - #confirm-box { width: 60; height: auto; border: thick $warning; - background: $panel; padding: 1 2; } + #se-edit { height: 1fr; border: none; padding: 0 1; } + /* A footer, but a quiet one: a shade of the dialog's own background + instead of the old near-black bar. */ + #se-status { height: 1; background: $panel-darken-1; color: $text-muted; + padding: 0 1; } + #confirm-box { width: 60; height: auto; padding: 1 2; } #confirm-msg { height: auto; } #confirm-help { height: 1; color: $text-muted; margin-top: 1; } - LoadingScreen { align: center middle; } - #loading-box { width: 72; height: auto; border: thick $accent; - background: $panel; padding: 1 2; } + #loading-box { width: 72; height: auto; padding: 1 2; } #loading-logo { width: 100%; height: auto; margin-bottom: 1; } + #loading-image { width: 100%; margin-bottom: 1; } #loading-title { width: 1fr; height: 1; text-style: bold; } #loading-note { height: auto; color: $text-muted; margin-top: 1; } #loading-help { height: auto; color: $text-muted; margin-top: 1; } - BusyScreen { align: center middle; } - #busy-box { width: auto; min-width: 26; height: auto; border: thick $accent; - background: $panel; padding: 1 2; } + #busy-box { width: auto; min-width: 26; height: auto; padding: 1 2; } #busy-msg { height: 1; text-style: bold; } #busy-help { height: 1; color: $text-muted; margin-top: 1; } """ @@ -3066,12 +5163,32 @@ class IdaTui(App): Binding("q", "quit", "Quit"), Binding("ctrl+n", "symbols", "Symbols"), Binding("ctrl+t", "structs", "Structs"), + Binding("ctrl+f", "find", "Find"), + Binding("ctrl+e", "export", "Export", show=False), Binding("backslash", "hex", "Hex"), Binding("s", "toggle_split", "Split", show=False), + # IDA's own key for text/graph. Graph mode is opt-in and self-contained: + # with it off nothing else in the app does any extra work. + Binding("space", "toggle_graph", "Graph", 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), - Binding("f1", "help", "Keys", show=False), + # Trace stepping. ] / [ move one instruction, } / { step over a + # call by following the stack pointer. + # Seeking, as opposed to stepping: jump to the next/previous time THIS + # thing was touched, where "this thing" is whatever the focused view + # addresses — an instruction in the code views, a byte in hex. + Binding("greater_than_sign", "seek_next_hit", "Next hit", show=False), + Binding("less_than_sign", "seek_prev_hit", "Prev hit", show=False), + Binding("W", "seek_reg_write", "Reg writes", show=False), + 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), + # H as well as F1: terminals and multiplexers swallow function keys all + # the time (and the one that does it is upstream of us, so there is + # nothing to fix on this side), which left the cheatsheet unreachable. + Binding("f1,H", "help", "Keys", show=False), Binding("g", "goto", "Goto"), Binding("slash", "filter", "Filter", show=False), Binding("ctrl+b", "toggle_functions", "Names", show=False), @@ -3082,7 +5199,7 @@ class IdaTui(App): def __init__(self, open_path: str | None = None, keepalive: bool = True, rpc_path: str | None = None, ttl: int = 1800, - project=None, load_args: str = "") -> None: + project=None, load_args: str = "", trace_path: str = "") -> None: super().__init__() # Project mode is additive: with no project this is the plain # single-binary app, unchanged. @@ -3099,25 +5216,34 @@ class IdaTui(App): 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 + #: Literal positions for the decompilation being loaded (worker thread + #: -> the view, handed over when the pseudocode is applied). + self._pending_nums: dict = {} # 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 self._index = None # project-wide symbol/string index if project is not None: from .index import ProjectIndex - from .pool import WorkerPool - self._pool = WorkerPool(project, ttl=ttl) + from .pool import DatabasePool + self._pool = DatabasePool(project, ttl=ttl) self._index = ProjectIndex( os.path.join(project.index_dir, "project.db")) self._binary = project.refs[0].label 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._load_args = load_args or "" # first-open options for a headerless blob + self._new_database = False # Ctrl+L asks Code Mode for a fresh IDB + self._title = (os.path.basename(open_path) if open_path else "") + #: Where we are in the execution trace, and everything that moves us. + #: Owns the trace state; the _trace/_t/_trail_* properties below + #: forward to it. + self.trace_ctl = TraceController(self, trace_path or "") self._do_keepalive = keepalive self._rpc_path = rpc_path self._rpc = None - self.client: WorkerClient | None = None + self.client: CodeModeClient | None = None self.program: Program | None = None self._loading_screen: LoadingScreen | None = None self._ka = None @@ -3132,8 +5258,9 @@ class IdaTui(App): # 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._active = ViewMode.LISTING # currently shown view (in split: the focused pane) self._split = False # side-by-side listing + pseudocode + self._graph_sticky = False # stay in graph mode across navigations self._split_eamap: list[list[int]] = [] # split: decomp line -> instr EAs self._split_ea2line: dict[int, int] = {} # split: instr EA -> decomp line self._split_range: tuple[int, int] | None = None # decomp'd fn ea span @@ -3146,11 +5273,15 @@ class IdaTui(App): self._conn_screen: LoadingScreen | None = None self._reconnecting = False # a reconnect attempt is in flight self._search_ctx: tuple[object | None, int] = (None, 1) - self._rename_ctx: tuple[object | None, str] = (None, "") - self._rename_addr: int | None = None # set for address-based (listing) naming - self._comment_ctx: tuple[object | None, int, str] = (None, 0, "") - self._retype_ctx: tuple[object | None, str, int, str] = (None, "", 0, "") - self._makedata_ctx: tuple[object | None, int] = (None, 0) + #: The one-line prompts above the footer. Each holds its own context + #: for exactly as long as it is on screen; see idatui/prompt.py. + self.prompts = PromptBar(self, "search", "rename", "comment", + "retype", "makedata", "goto", "export") + #: Everything that writes to the database (idatui/edit_ctl.py). + self.edits = EditController(self) + #: What those writes were, so the findings export can say which + #: comments and names are YOURS -- the database cannot (journal.py). + self.journal = Journal() self._xref_focus_name: str | None = None self._dirty = False @@ -3168,6 +5299,14 @@ class IdaTui(App): hx = HexView() hx.display = False yield hx + gv = GraphView() + gv.display = False + yield gv + # 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 @@ -3192,6 +5331,10 @@ class IdaTui(App): gi.display = False gi.can_focus = False yield gi + xi = Input(id="export") + xi.display = False + xi.can_focus = False + yield xi # markup=False: the status is plain text full of [listing]/[split]/[label] # markers and symbol names that may contain brackets. With Textual markup # on, a single-word marker parses as a style tag and is silently eaten — @@ -3210,8 +5353,8 @@ class IdaTui(App): 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. + # opened, so ask BEFORE Code Mode creates it — once IDA has made a database + # the answer is baked in and changing it requires a fresh-IDB reopen. if self._project is not None: ref = self._pending_load_ref() if ref is not None: @@ -3242,6 +5385,11 @@ class IdaTui(App): if os.path.exists(self._open_path + ".i64") or os.path.exists( os.path.splitext(self._open_path)[0] + ".i64"): return False + try: + if registered_database(self._open_path): + return False + except Exception: + pass # connect() will surface registry failures with full diagnostics return needs_load_options(self._open_path) def action_load_options(self) -> None: @@ -3255,7 +5403,11 @@ class IdaTui(App): forward. """ if not self._can_reload(): - self._status("nothing to reload") + if self.client is not None and self.client.backend == "gui": + self._status( + "reload unavailable for a GUI-owned database — reopen it in IDA") + else: + 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" @@ -3274,10 +5426,13 @@ class IdaTui(App): 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() + # Release our lease first. Code Mode waits for a managed worker's final + # lease grace, then creates the replacement IDB atomically. A GUI-backed + # database is rejected by _can_reload(): the TUI must never close it. + self._release_database() + self._new_database = True + if label is not None and self._pool is not None: + self._pool.recreate_on_next_open(label) self._reset_for_reload() self._load_args = "" if label is not None and self._project is not None: @@ -3289,7 +5444,9 @@ class IdaTui(App): self._pending_switch = None self._ask_load_options(path, label=label) - def _release_worker(self) -> None: + def _release_database(self) -> None: + if self.program is not None: + self.program.close() if self._pool is not None and self._binary is not None: try: self._pool.evict(self._binary, save=False) @@ -3303,23 +5460,6 @@ class IdaTui(App): 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 @@ -3363,6 +5503,11 @@ class IdaTui(App): 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 + try: + if registered_database(ref.staged, output_database=ref.db): + return None + except Exception: + pass from .formats import needs_load_options return ref if needs_load_options(ref.source) else None @@ -3416,10 +5561,6 @@ class IdaTui(App): asyncio.get_running_loop().create_task(_serve()) - async def on_unmount(self) -> None: - if self._rpc is not None: - await self._rpc.stop() - # -- status helper ----------------------------------------------------- # def _status(self, text: str, priority: bool = False) -> None: """Write the status bar. ``priority`` marks the RESULT of something the @@ -3438,8 +5579,12 @@ class IdaTui(App): self._flash_until = _time.monotonic() + 8.0 elif self._flash and _time.monotonic() < self._flash_until: text = self._flash - if self._binary: # project mode: always say which binary you're in - text = f"[{self._binary}] {text}" + # 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. @@ -3480,9 +5625,10 @@ class IdaTui(App): # -- connection loss / recovery --------------------------------------- # def _handle_exception(self, error: BaseException) -> None: - """Intercept a lost-connection error from any worker so the whole app - doesn't die when the analysis server goes away (it can idle out, be - killed, or the box can sleep). Everything else crashes as usual.""" + """Intercept a lost Code Mode lease so the app can rediscover the DB. + + Everything unrelated to database connectivity crashes as usual. + """ from textual.worker import WorkerFailed orig = error.error if isinstance(error, WorkerFailed) else error if isinstance(orig, IDAConnectionError): @@ -3514,15 +5660,15 @@ class IdaTui(App): @work(thread=True, exclusive=True, group="reconnect") def _reconnect(self) -> None: - # The worker died (segfault -> dropped socket). Respawn it: it re-opens - # and re-analyzes the binary in a fresh process, then we rebuild. + # The registered instance disappeared. Rediscover it; Code Mode may find + # a GUI/replacement worker, then we rebuild caches against the new handle. try: if self._open_path is None: self.app.call_from_thread(self._reconnect_failed, "no binary to reopen") return - client = WorkerClient(self._open_path, ttl=self._ttl, - load_args=self._load_args) + client = CodeModeClient(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 @@ -3530,7 +5676,7 @@ class IdaTui(App): return self.app.call_from_thread(self._after_reconnect, client, Program(client)) - def _after_reconnect(self, client: "WorkerClient", program: "Program") -> None: + def _after_reconnect(self, client: "CodeModeClient", program: "Program") -> None: self.client = client self.program = program self._reconnecting = False @@ -3550,13 +5696,13 @@ class IdaTui(App): @work(thread=True, exclusive=True, group="connect") def _connect(self) -> None: try: - client = self._open_worker_client() + client = self._open_database_client() if client is None: return # the opener already reported + dismissed the overlay module = client.health().get("module", "?") if self._do_keepalive: - # Keep the session warm while we run; don't make it immortal, so - # it's reclaimed after the TUI closes. (No-op for the worker.) + # Compatibility shim: DatabaseHandle's SSE lease already owns + # liveness and heartbeat behavior. self._ka = client.keepalive(interval=120.0).start() program = Program(client) except Exception as e: # noqa: BLE001 @@ -3571,31 +5717,33 @@ class IdaTui(App): return self.client = client self.program = program - self.app.call_from_thread(self._status, f"{module} — loading functions…") + self._new_database = False + self.app.call_from_thread( + self._status, f"{module} [{client.backend}] — loading functions…") self._load_functions() - def _open_worker_client(self): # type: ignore[no-untyped-def] - """Our idalib-worker path: spawn the worker (it opens + analyzes the - binary in its own process) and connect. Returns the client, or None.""" - from .worker_client import WorkerClient - if self._pool is not None: # project mode: the pool owns the workers + def _open_database_client(self): # type: ignore[no-untyped-def] + """Attach through Code Mode, reusing a GUI or managed idalib database.""" + if self._pool is not None: # project mode: the pool owns the leases label = self._binary or self._project.refs[0].label client = self._pool.get(label, progress=lambda m: self.app.call_from_thread(self._status, m)) 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( - self._status, "the worker backend needs a binary path") + self._status, "Code Mode needs a database or executable path") self.app.call_from_thread(self._dismiss_loading) return None 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, - load_args=self._load_args) + self._status, f"discovering Code Mode database for {base}…") + client = CodeModeClient(self._open_path, ttl=self._ttl, + load_args=self._load_args, + new_database=self._new_database) client.connect(progress=lambda m: self.app.call_from_thread( self._status, m)) return client @@ -3607,7 +5755,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) @@ -3615,18 +5762,20 @@ class IdaTui(App): if rows: self.app.call_from_thread(self._append_rows, rows) self.app.call_from_thread( - self._status, f"{module} — {last} functions…" + self._status, f"{last} functions…" ) # If a filter is active (typed during load), re-apply it over the full set. if self._filter_term: self.app.call_from_thread(self._apply_filter, self._filter_term) else: self.app.call_from_thread( - self._status, f"{module} — {len(idx)} functions (Ctrl+N: find symbol)") + self._status, f"{len(idx)} functions (Ctrl+N: find symbol)") # Land somewhere useful instead of an empty pane: main() if present, # otherwise pop the fuzzy symbol picker. self.app.call_from_thread(self._auto_land) self._index_binary() # project mode: keep the cross-binary index fresh + if self.trace_ctl.armed: + self._load_trace() # needs the index above: rebasing reads it @work(thread=True, exclusive=True, group="prewarm") def _prewarm_provider(self) -> None: @@ -3666,7 +5815,7 @@ class IdaTui(App): @work(thread=True, exclusive=True, group="index") def _index_binary(self) -> None: """Fold this binary's symbols + strings into the project index, so it can - be searched later even when its worker is gone.""" + be searched later even when its Code Mode lease is gone.""" if self._index is None or self._project is None or self._binary is None: return ref = self._project.by_label(self._binary) @@ -3684,7 +5833,7 @@ class IdaTui(App): 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 + except Exception: # noqa: BLE001 -- indexing is best-effort pass try: n = self._index.reindex(self._binary, entries, source=ref.source) @@ -3726,7 +5875,19 @@ class IdaTui(App): if fn is not None: self._open_function(fn.addr, fn.name) elif len(self._func_index): - self.action_symbols() + # No main(): land on the first function rather than pushing the + # symbol palette. A modal as the *startup* state leaves a human + # staring at a picker over an empty pane, and silently swallows + # every keystroke an RPC driver injects while `ping` still says + # ready:true. Landing somewhere real is better for both; Ctrl+N is + # one keypress away. + first = self._func_index.get(0) + if first is not None: + self._open_function(first.addr, first.name) + self._status(f"no entry function — opened {first.name} " + "(Ctrl+N: find symbol)") + else: + self.action_symbols() else: self._land_without_functions() @@ -3757,7 +5918,13 @@ class IdaTui(App): cursor=0, push=True, is_region=True) def _can_reload(self) -> bool: - """Whether we're able to re-open this binary with different options.""" + """Whether Code Mode can replace this IDB with different options. + + A GUI database is owned by the user and has no remote close/rollback + route. Managed idalib databases can be released and reopened fresh. + """ + if self.client is not None and self.client.backend == "gui": + return False if self._project is not None and self._binary is not None: return True return bool(self._open_path) @@ -3837,7 +6004,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 @@ -3872,8 +6039,7 @@ class IdaTui(App): left = self.query_one("#left", FunctionsPanel) left.display = not left.display if not left.display: - self.query_one(DecompView if self._active == "decomp" - else ListingView).focus() + self._focus_code_view() else: self.query_one("#func-table", DataTable).focus() @@ -3947,6 +6113,9 @@ class IdaTui(App): def _on_quit_choice(self, choice: str | None) -> None: if choice == "discard": + # Code Mode has no rollback/close-without-save operation. For GUI + # sessions this leaves changes dirty in IDA; a managed worker owns + # its final save policy and may persist them on final lease release. self._save_on_exit = False self.exit() elif choice == "save": @@ -3963,7 +6132,7 @@ class IdaTui(App): if self._pool is not None: self._pool.close_all(save=True) # saves each resident worker elif self.program is not None: - self.program.client.call("idb_save", timeout=600.0) + self.program.client.save_database() except Exception as e: # noqa: BLE001 -- still exit, but say so self.app.call_from_thread(self._status, f"save failed: {e}") self.app.call_from_thread(self._finish_exit) @@ -4003,7 +6172,7 @@ class IdaTui(App): 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). + # the pool hand us a lease (attaching + evicting as the budget dictates). if self._binary is not None: self._states[self._binary] = BinaryState( label=self._binary, program=self.program, @@ -4024,8 +6193,8 @@ class IdaTui(App): self.app.call_from_thread(self._switch_failed, label, str(e)) return st = self._states.get(label) - # The Program (and its caches) only survive while that worker does; a - # binary that was evicted comes back with a fresh one. Either way the nav + # The Program (and its caches) only survive while that lease does; an + # evicted binary reattaches. Either way the nav # history is just addresses, so it always survives. reuse = (st is not None and st.program is not None and getattr(st.program, "client", None) is client) @@ -4039,7 +6208,8 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged - self._active = st.active if st else "listing" + self._title = os.path.basename(self._open_path) + self._active = st.active if st else ViewMode.LISTING self._split = st.split if st else False self._filter_term = st.filter_term if st else "" self._dirty = st.dirty if st else False @@ -4061,7 +6231,7 @@ class IdaTui(App): self._did_auto_land = False self._auto_land() return - # Cold (first visit, or the worker was evicted): rebuild the index, then + # Cold (first visit, or the lease was evicted): rebuild the index, then # land back where we were via _pending_restore. self._cur = None self._func_index = None @@ -4109,6 +6279,35 @@ class IdaTui(App): binary=self._binary), self._on_string_chosen) + def action_find(self) -> None: + """Ctrl+F: search the whole database — disassembly text, or bytes.""" + if self.program is None: + self._status("not connected yet") + return + if self._prompt_active(): + return + # Seed it with the word under the cursor: the search you want is + # usually about the thing you are looking at. + seed = "" + view = self._active_code_view() + if view is not None: + try: + seed = view.word_under_cursor() or "" + except Exception: # noqa: BLE001 -- a seed is a nicety, never a + seed = "" # reason not to open the search + self.push_screen(SearchPalette(self.program, seed), self._on_hit_chosen) + + def _on_hit_chosen(self, hit) -> None: # type: ignore[no-untyped-def] + if hit is None: + return + # Navigate to the ITEM, not the matched byte: a pattern can start in + # the middle of an instruction, and there is nothing to put a cursor on + # there. The status names the exact address so it isn't lost. + self._goto_ea(hit.head, push=True) + if hit.addr != hit.head: + self._status(f"match at {hit.addr:#x} (inside {hit.head:#x})", + priority=True) + def _on_string_chosen(self, choice) -> None: # type: ignore[no-untyped-def] if choice is None: return @@ -4118,28 +6317,6 @@ 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).""" @@ -4157,17 +6334,35 @@ class IdaTui(App): return if self._split: # In split mode Tab/F5 just moves focus between the two panes. - self._active = "decomp" if self._active == "listing" else "listing" - (self.query_one(DecompView) if self._active == "decomp" + self._active = ViewMode.DECOMP if self.is_listing else ViewMode.LISTING + (self.query_one(DecompView) if self.is_decomp else self.query_one(ListingView)).focus() self._sync_split(self._active) # re-link from the new driver self._status_for_cur("split") return - if self._active == "hex": + if self.is_hex: self._active = self._code_mode() self._show_active() return - if self._active == "listing": + if self.is_graph: + # F5/Tab out of the graph lands in the pseudocode at the block the + # cursor was on (Space is the key that returns to the listing). + gv = self.query_one(GraphView) + ea = gv._cursor_ea() + self._graph_sticky = False + if ea is None: + self._active = ViewMode.LISTING + self._show_active() + return + gv.display = False + dec = self.query_one(DecompView) + dec.display = True + dec.loading = True + dec.focus() + self._status("decompiling…") + self._decomp_from_listing(ea) + return + if self.is_listing: # F5/Tab in the continuous listing: decompile the function under the # cursor (IDA-style), if the cursor is inside a defined routine. ea = self.query_one(ListingView)._cursor_ea() @@ -4192,7 +6387,7 @@ class IdaTui(App): self._decomp_return = None if ret is not None: self._cur = ret - self._active = "listing" + self._active = ViewMode.LISTING self._open_entry(ret, push=False) elif self._cur is not None: # No F5 snapshot (we arrived via a decomp navigation): show THIS entry @@ -4203,7 +6398,7 @@ class IdaTui(App): ea = dec._line_ea(dec.cursor) self._toggle_to_listing(ea if ea is not None else self._cur.ea) else: - self._active = "listing" + self._active = ViewMode.LISTING self._show_active() @work(thread=True, group="nav") @@ -4216,14 +6411,14 @@ class IdaTui(App): def _apply_toggle_listing(self, idx: int) -> None: cur = self._cur if cur is None: - self._active = "listing" + self._active = ViewMode.LISTING self._show_active() return cur.view = "listing" cur.cursor = idx cur.cursor_x = 0 cur.scroll_y = -1 # derive a viewport (keeps the target in context) - self._active = "listing" + self._active = ViewMode.LISTING self._open_entry(cur, push=False) @work(thread=True, group="nav") @@ -4242,7 +6437,7 @@ class IdaTui(App): # We optimistically raised the pseudocode overlay; drop back to the # listing since there's nothing to decompile here. self.query_one(DecompView).loading = False - self._active = "listing" + self._active = ViewMode.LISTING self._show_active() self._status(msg) @@ -4258,7 +6453,7 @@ class IdaTui(App): entry = NavEntry(ea=fn_addr, name=fn_name, is_region=False, dec_cursor=max(dec_idx, 0)) self._cur = entry - self._active = "decomp" + self._active = ViewMode.DECOMP self._show_active() def action_toggle_split(self) -> None: @@ -4274,12 +6469,115 @@ class IdaTui(App): return self._split = not self._split if self._active not in ("listing", "decomp"): - self._active = "listing" + self._active = ViewMode.LISTING if self._split: self._enter_split(self._cur.ea, self._cur.name) else: self._show_active() + # -- graph mode -------------------------------------------------------- # + #: Above this, a CFG graph stops being a picture and becomes a hairball -- + #: IDA's own is unreadable at this size too. Refusing beats rendering soup. + GRAPH_MAX_BLOCKS = 400 + + def action_toggle_graph(self) -> None: + """Space: swap the code view for the function's control-flow graph.""" + if self._prompt_active(): + return + if self.is_graph: + self._graph_sticky = False + self._active = self._code_mode() + self._show_active() + return + if self._cur is None or self.program is None: + self._status("open a function first") + return + self._graph_sticky = True + gv = self.query_one(GraphView) + ea = self._graph_target_ea() + if gv.loaded_ea is not None and gv.fc is not None \ + and gv.fc.func_ea == self._cur.ea: + self._active = ViewMode.GRAPH + self._split = False + self._show_active() + if ea is not None: + gv.goto_ea(ea) + self._graph_status() + return + self._status(f"{self._cur.name} — building graph…") + self._load_graph(self._cur.ea, ea) + + def _graph_target_ea(self) -> int | None: + """The address the graph should land on: wherever the code view's cursor + is, so Space doesn't lose your place.""" + try: + if self.is_listing: + return self.query_one(ListingView)._cursor_ea() + if self.is_decomp: + dv = self.query_one(DecompView) + return dv._line_ea(dv.cursor) + except Exception: # noqa: BLE001 + pass + return self._cur.ea if self._cur else None + + @work(thread=True, exclusive=True, group="graph") + def _load_graph(self, func_ea: int, want_ea: int | None) -> None: + assert self.program is not None + err = "" + fc = None + try: + fc = self.program.flowchart(func_ea) + except IDAConnectionError: + raise + except Exception as e: # noqa: BLE001 + err = f"{type(e).__name__}: {e}" + self.app.call_from_thread(self._apply_graph, func_ea, want_ea, fc, err) + + def _apply_graph(self, func_ea: int, want_ea: int | None, fc, # type: ignore[no-untyped-def] + err: str) -> None: + if self._cur is None or self._cur.ea != func_ea: + return # a newer navigation won + if not self._graph_sticky and not self.is_graph: + # The load lost its race: the user has since left graph mode (or a + # rename's reload queued one behind their back). Forcing the view + # here drags them back into a graph they already dismissed. + return + if fc is None: + self._status(err or "no control-flow graph for this function " + "(is it a thunk or an import?)") + return + if len(fc.blocks) > self.GRAPH_MAX_BLOCKS: + self._status( + f"{fc.name}: {len(fc.blocks)} blocks — too many to graph " + f"(limit {self.GRAPH_MAX_BLOCKS}); staying in the listing") + return + gv = self.query_one(GraphView) + gv.set_graph(fc, want_ea) + self._active = ViewMode.GRAPH + self._split = False + self._show_active() + self._graph_status() + + def _graph_status(self) -> None: + gv = self.query_one(GraphView) + if gv.lay is None or gv.fc is None: + return + s = gv.lay.stats + loops = f", {s['back']} loop{'s' if s['back'] != 1 else ''}" if s["back"] else "" + eng = "" if s.get("engine") == "native" else f", {s.get('engine')}" + self._status( + f"{gv.fc.name} @ {gv.fc.func_ea:#x} [graph: {s['blocks']} blocks, " + f"{s['edges']} edges{loops}{eng}] " + f"z=zoom({gv.ZOOMS[gv._zoom]}) m=map J/K=edge space=text") + + def on_graph_view_cursor_moved(self, msg: "GraphView.CursorMoved") -> None: + gv = self.query_one(GraphView) + gv.set_highlight(gv.word_under_cursor()) + if msg.ea is not None and gv.fc is not None: + b = gv.fc.block_at(msg.ea) + extra = f" block {b.start:#x}" if b else "" + self._status(f"{gv.fc.name} @ {msg.ea:#x}{extra} [graph]") + @work(thread=True, group="split") def _enter_split(self, ea: int, name: str) -> None: # Load the listing for the current function (bg) then reveal both panes. @@ -4303,17 +6601,19 @@ class IdaTui(App): cursor; press again (or Tab/Esc) to return to the code view.""" if self._cur is None or self.program is None: return - if self._active == "hex": + if self.is_hex: self._active = self._code_mode() self._show_active() return - if self._active in ("listing", "disasm"): + if self.is_listing: ea = self.query_one(ListingView)._cursor_ea() + elif self.is_graph: + ea = self.query_one(GraphView)._cursor_ea() else: dec = self.query_one(DecompView) ea = dec._line_ea(dec.cursor) self._hex_pending_ea = ea if ea is not None else self._cur.ea - self._active = "hex" + self._active = ViewMode.HEX self._show_active() def action_filter(self) -> None: @@ -4324,9 +6624,53 @@ class IdaTui(App): inp.value = self._filter_term inp.focus() + def action_export(self) -> None: + """Ctrl+E: write what you have worked out to a markdown report. + + Prefilled with a path beside the binary, because the common case is + "just write it" and the rare case is worth one edit. + """ + if self.program is None: + self._status("no database open", priority=True) + return + inp = self.query_one("#export", Input) + inp.placeholder = "export findings to… (markdown) — Enter" + inp.can_focus = True + inp.display = True + inp.value = findings.default_path(self._open_path or "findings") + self.query_one("#status", Static).display = False + inp.focus() + + def _end_export(self) -> None: + inp = self.query_one("#export", Input) + inp.display = False + inp.can_focus = False + self.query_one("#status", Static).display = True + + def export_findings(self, path: str | None = None) -> None: + """Gather + write the report off the UI thread (it walks the database).""" + self._status("exporting findings…", priority=True) + self._export_worker(path) + + @work(thread=True, exclusive=True, group="export") + def _export_worker(self, path: str | None) -> None: + try: + self.journal.load(self.program) + self.journal.flush(self.program) + out, f = findings.export(self.program, self._open_path or "", path, + journal=self.journal) + except Exception as e: # noqa: BLE001 -- a bad path is a message, not a crash + self.call_from_thread(self._status, f"export failed: {e}", True) + return + n_named = len(findings._user_names(f)) + self.call_from_thread( + self._status, + f"exported {len(f.comments)} comments, {n_named} names, " + f"{len(f.types)} types → {out}", True) + def action_goto(self) -> None: inp = self.query_one("#goto", Input) - inp.placeholder = ("hex goto: 0xADDR or name — Enter" if self._active == "hex" + inp.placeholder = ("hex goto: 0xADDR or name — Enter" if self.is_hex else "goto: name or 0xADDR — Enter") inp.can_focus = True inp.display = True @@ -4362,8 +6706,7 @@ class IdaTui(App): elif self.query_one("#left", FunctionsPanel).display: table.focus() else: - self.query_one(DecompView if self._active == "decomp" - else ListingView).focus() + self._focus_code_view() # -- input submit (filter / goto) ------------------------------------- # def on_search_requested(self, msg: SearchRequested) -> None: @@ -4381,6 +6724,19 @@ class IdaTui(App): def on_follow_requested(self, msg: FollowRequested) -> None: view = msg.view word = view.word_under_cursor() + if isinstance(view, GraphView): + ea = view._cursor_ea() + if ea is None: + return + # Inside the graph, a jump to a block of THIS function should move + # the cursor, not navigate away and rebuild the whole picture. + tgt = self._graph_local_target(view, ea, word) + if tgt is not None: + view.goto_ea(tgt) + self.post_message(GraphView.CursorMoved(tgt, view.cursor_node)) + return + self._follow_disasm(ea, word, view._next_ea()) + return if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: @@ -4392,11 +6748,39 @@ class IdaTui(App): self._follow_decomp(view._texts[view.cursor], word, view._line_ea(view.cursor)) + def _graph_local_target(self, view: "GraphView", ea: int, + word: str) -> int | None: + """If the cursor's instruction branches somewhere inside this same + graph, return that address.""" + if view.fc is None or self.program is None: + return None + try: + if word and self._looks_like_symbol(word): + t = self.program.resolve(word) + if t and view.fc.block_at(t) is not None: + return t + except Exception: # noqa: BLE001 + pass + try: + for xr in self.program.xrefs_from(ea): + t = xr.to + if t and t != view._next_ea() and view.fc.block_at(t) is not None: + return t + except Exception: # noqa: BLE001 + pass + return None + def on_xrefs_requested(self, msg: XrefsRequested) -> None: if self._xref_active: # one gather at a time; ignore a second 'x' return view = msg.view word = view.word_under_cursor() + if isinstance(view, GraphView): + ea = view._cursor_ea() + if ea is not None: + self._push_busy("finding xrefs…") + self._xrefs_disasm(ea, word, ea, view._next_ea()) + return if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: @@ -4486,7 +6870,7 @@ class IdaTui(App): 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 + Reads the on-disk index, so a provider resolves even when its lease was evicted — the whole reason the index exists. """ if self._index is None or self._project is None or not name: @@ -4659,7 +7043,7 @@ class IdaTui(App): 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. + whether or not its database lease 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. @@ -4709,595 +7093,141 @@ class IdaTui(App): # 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")) + prefer_decomp=(self.is_decomp)) - # -- rename ------------------------------------------------------------ # - @staticmethod - def _is_pseudocode_label(view, name: str) -> bool: - """True if ``name`` is a Hex-Rays goto label in ``view``. The rename tool - has no label category (only func/global/local/stack), so renaming one - fails with a misleading 'local variable not found'; detect it up front - and explain instead. A label is the default ``LABEL_n`` or any token used - as a ``goto`` target.""" - if not isinstance(view, DecompView): - return False - if re.fullmatch(r"LABEL_\d+", name): - return True - body = "\n".join(getattr(view, "_texts", []) or []) - return re.search(rf"\bgoto\s+{re.escape(name)}\b", body) is not None + # -- database edits ---------------------------------------------------- # + # The bodies live in EditController (idatui/edit_ctl.py). What stays here is + # what Textual insists on owning: on_<Message> handlers, which it dispatches + # by name on the DOMNode, and @work entry points, whose worker machinery + # wants a DOMNode host. Both are one-line delegates. def on_rename_requested(self, msg: RenameRequested) -> None: - # In the flat listing, 'n' names the ADDRESS under the cursor (create a - # label), not a symbol-by-name. This is what lets you name a bare/ - # undefined byte — e.g. the free byte at addr+1 after shrinking a u16 to - # a u8 — which the word-under-cursor path can't do (no symbol to rename). - if isinstance(msg.view, ListingView): - ea = msg.view._cursor_ea() - if ea is None: - self._status("no address on this line to name") - return - head = msg.view.cur_head() - word = msg.view.word_under_cursor() - mnem = head.text.split(" ", 1)[0] if (head and head.text) else "" - # If the cursor is on a symbol token (a call/branch target, a data - # reference, or this head's own label) rename THAT symbol; otherwise - # create/rename a label at the head's address (bare/undefined bytes). - if (word and self._looks_like_symbol(word) and word != mnem - and word.lower() not in _ASM_KEYWORDS): - self._rename_ctx = (msg.view, word) - self._rename_addr = None - placeholder = f"rename '{word}' — Enter=apply Esc=cancel" - prefill = word - else: - cur = head.name if (head is not None and head.name) else "" - self._rename_ctx = (msg.view, cur) - self._rename_addr = ea - placeholder = f"name @ {ea:#x} — Enter=apply Esc=cancel" - prefill = cur - self.query_one("#status", Static).display = False - inp = self.query_one("#rename", Input) - inp.placeholder = placeholder - inp.can_focus = True - inp.display = True - inp.value = prefill - inp.focus() - return - if not msg.name: - self._status("nothing to rename under the cursor") - return - if self._is_pseudocode_label(msg.view, msg.name): - self._status( - f"can't rename pseudocode label '{msg.name}' " - "(Hex-Rays goto labels aren't renamable via the API)") + self.edits.request_rename(msg) + + def on_comment_requested(self, msg: CommentRequested) -> None: + self.edits.request_comment(msg) + + def on_retype_requested(self, msg: RetypeRequested) -> None: + if self._cur is None or self.program is None: return - self._rename_ctx = (msg.view, msg.name) - self._rename_addr = None - self.query_one("#status", Static).display = False - inp = self.query_one("#rename", Input) - inp.placeholder = f"rename '{msg.name}' — Enter=apply Esc=cancel" - inp.can_focus = True - inp.display = True - inp.value = msg.name - inp.focus() + self._prepare_retype(msg.view, msg.name) - def _end_rename(self) -> None: - inp = self.query_one("#rename", Input) - inp.display = False - inp.can_focus = False - self._rename_addr = None - self.query_one("#status", Static).display = True - view, _ = self._rename_ctx - if view is not None: - view.focus() + def on_make_data_requested(self, msg: MakeDataRequested) -> None: + self.edits.request_make_data(msg) - # -- 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, ListingView): - return view._cursor_ea() - if isinstance(view, DecompView): - return view._line_ea(view.cursor) - return None + def on_op_format_requested(self, msg: OpFormatRequested) -> None: + self.edits.request_op_format(msg) - @staticmethod - def _existing_comment(view) -> str: - """Current line comment (for prefill), parsed from the rendered text. In - pseudocode a comment is `// text` before the trailing /*0xEA*/ markers; - C has no `//` operator, so the last `//` is unambiguously the comment.""" - if isinstance(view, DecompView) and 0 <= view.cursor < len(view._texts): - s = re.sub(r"(?:/\*\s*0x[0-9A-Fa-f]+\s*\*/\s*)+$", "", view._texts[view.cursor]) - i = s.rfind("//") - return s[i + 2:].strip() if i >= 0 else "" - return "" + def on_edit_item_requested(self, msg: EditItemRequested) -> None: + self.edits.request_edit_item(msg) - def on_comment_requested(self, msg: CommentRequested) -> None: - ea = self._line_ea_for(msg.view) - # Signature / local-declaration lines carry no address; fall back to the - # function's entry ea so commenting the header annotates the function. - func_level = ea is None - if func_level: - ea = self._cur.ea if self._cur else None - if ea is None: - self._status("no address on this line to comment") - return - existing = "" if func_level else self._existing_comment(msg.view) - self._comment_ctx = (msg.view, ea, existing) - self.query_one("#status", Static).display = False - inp = self.query_one("#comment", Input) - inp.placeholder = ( - f"function comment @ {ea:#x} — Enter=apply (empty=clear) Esc=cancel" - if func_level else - f"comment @ {ea:#x} — Enter=apply (empty=clear) Esc=cancel") - inp.can_focus = True - inp.display = True - inp.value = existing - inp.focus() + @work(thread=True, exclusive=True, group="rename") + def _do_rename(self, view, old: str, new: str) -> None: # type: ignore[no-untyped-def] + self.edits.do_rename(view, old, new) - def _end_comment(self) -> None: - inp = self.query_one("#comment", Input) - inp.display = False - inp.can_focus = False - self.query_one("#status", Static).display = True - view, _, _ = self._comment_ctx - if view is not None: - view.focus() + @work(thread=True, exclusive=True, group="rename") + def _do_name_addr(self, addr: int, name: str) -> None: + self.edits.do_name_addr(addr, name) @work(thread=True, exclusive=True, group="comment") def _do_comment(self, ea: int, text: str) -> None: - assert self.program is not None - # The prompt is single-line, so a literal '\n' (backslash-n) means a real - # newline — Hex-Rays renders each as its own '//' line. Lets long notes - # wrap instead of running off the right edge and clipping. - text = text.replace("\\n", "\n") - try: - res = self.program.set_comment(ea, text) - except IDAToolError as e: - self.app.call_from_thread(self._status, f"comment failed: {e.message}") - return - data = res.get("result") if isinstance(res, dict) else None - if isinstance(data, list) and data and isinstance(data[0], dict) and data[0].get("error"): - self.app.call_from_thread(self._status, f"comment failed: {data[0]['error']}") - return - self.app.call_from_thread(self._after_comment, ea, text) + self.edits.do_comment(ea, text) - def _reload_active_code(self) -> None: - """Refresh whichever code view is showing after an edit (comment/rename/ - retype), in place: re-decompile if in the decompiler, else reload the - listing.""" - cur = self._cur - if cur is None: - return - if self._active == "decomp": - # Snapshot the LIVE pseudocode position before forcing a recompile. - # dec_scroll_y isn't tracked on every move, so without this the reload - # falls into show()'s derive path (a bare scroll_to) and leaves a - # stale frame until the next cursor move; capturing the real scroll - # makes show() take the robust _apply_scroll path and repaint now. - dec = self.query_one(DecompView) - if cur.ea == dec.loaded_ea: - cur.dec_cursor = dec.cursor - cur.dec_cursor_x = dec.cursor_x - cur.dec_scroll_y = round(dec.scroll_offset.y) - cur.dec_scroll_x = round(dec.scroll_offset.x) - dec.loaded_ea = None # force re-decompile - self._show_active() - else: - # 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: - cur.cursor = lst.cursor - cur.cursor_x = lst.cursor_x - cur.scroll_y = round(lst.scroll_offset.y) - self._open_entry(cur, push=False) + @work(thread=True, exclusive=True, group="retype") + def _prepare_retype(self, view, word: str | None) -> None: # type: ignore[no-untyped-def] + self.edits.prepare_retype(view, word) - def _after_comment(self, ea: int, text: str) -> None: - # A comment shows in both views but only after Hex-Rays recompiles, so - # reuse the name-generation invalidation (bumps gen -> decompile is - # force_recompiled lazily; disasm/listing caches are cleared). - self.program.bump_names() - self._reload_active_code() - self._dirty = True - verb = "cleared comment" if not text else "commented" - self._status(f"{verb} @ {ea:#x} (Ctrl+S to save)") + @work(thread=True, exclusive=True, group="retype-apply") + def _do_retype(self, kind: str, subject: int, word: str, new: str) -> None: + self.edits.do_retype(kind, subject, word, new) - # -- retype (set type, IDA 'y') --------------------------------------- # - def on_retype_requested(self, msg: RetypeRequested) -> None: - if self._cur is None or self.program is None: - return - self._prepare_retype(msg.view, msg.name) + @work(thread=True, exclusive=True, group="makedata") + def _do_make_data(self, ea: int, type_decl: str, + anchor: ViewAnchor | None = None) -> None: + self.edits.do_make_data(ea, type_decl, anchor) - @work(thread=True, exclusive=True, group="retype") - def _prepare_retype(self, view, word: str | None) -> None: # type: ignore[no-untyped-def] - """Work out whether the cursor is on a local variable or a function, and - fetch the current type/prototype to prefill the prompt.""" - assert self.program is not None and self._cur is not None - ft = self.program.func_types(self._cur.ea) - kind: str | None = None - subject: int = self._cur.ea - prefill = "" - # 1) a local variable (or arg) of the current function - if word and ft is not None: - lv = next((v for v in ft.lvars if v.name == word), None) - if lv is not None: - kind, prefill = "lvar", lv.type - # 2) a symbol under the cursor: a function (retype its prototype) or a - # global/data item (retype the variable). Without the data case a - # global fell through to (3) and silently retyped the ENCLOSING - # function's prototype instead. - if kind is None and self._looks_like_symbol(word): - try: - tgt = self.program.resolve(word) - except Exception: # noqa: BLE001 - tgt = None - if tgt is not None: - tft = self.program.func_types(tgt) - if tft is not None: - kind, subject, prefill = "func", tgt, tft.prototype - else: - dt = self.program.data_type(tgt) - if dt is not None and not dt.get("is_func"): - kind, subject = "data", tgt - prefill = dt.get("type") or self._guess_data_type( - dt.get("size") or 0) - # 3) fall back to the current function itself - if kind is None and ft is not None: - kind, subject, prefill = "func", self._cur.ea, ft.prototype - if kind is None: - self.app.call_from_thread(self._status, "nothing to retype under the cursor") - return - self.app.call_from_thread(self._open_retype, view, kind, subject, word or "", prefill) + @work(thread=True, exclusive=True, group="opformat") + def _do_op_format(self, mode: str, where: str, ea: int, col: int, + line: int = -1) -> None: + self.edits.do_op_format(mode, where, ea, col, line) - @staticmethod - def _guess_data_type(size: int) -> str: - """A sensible prefill when a global carries no type yet.""" - return {1: "unsigned __int8", 2: "unsigned __int16", - 4: "unsigned __int32", 8: "unsigned __int64"}.get( - size, f"char[{size}]" if size > 0 else "void *") + @work(thread=True, exclusive=True, group="edititem") + def _do_edit_item(self, kind: str, ea: int, + anchor: ViewAnchor | None = None) -> None: + self.edits.do_edit_item(kind, ea, anchor) - def _open_retype(self, view, kind: str, subject: int, word: str, # type: ignore[no-untyped-def] - prefill: str) -> None: - self._retype_ctx = (view, kind, subject, word) - self.query_one("#status", Static).display = False - inp = self.query_one("#retype", Input) - label = "prototype" if kind == "func" else f"type for '{word}'" - inp.placeholder = f"{label} — Enter=apply Esc=cancel" - inp.can_focus = True - inp.display = True - inp.value = prefill - inp.focus() + def _reload_active_code(self) -> None: + self.edits.reload_active_code() - def _end_retype(self) -> None: - inp = self.query_one("#retype", Input) - inp.display = False - inp.can_focus = False - self.query_one("#status", Static).display = True - view = self._retype_ctx[0] - if view is not None: - view.focus() + # -- trace ------------------------------------------------------------- # + # Everything below delegates to TraceController (idatui/trace_ctl.py). The + # keys stay declared here because Textual only merges BINDINGS from DOMNode + # subclasses, and the @work entry points stay here because Textual's worker + # machinery wants a DOMNode host. - @work(thread=True, exclusive=True, group="retype-apply") - def _do_retype(self, kind: str, subject: int, word: str, new: str) -> None: - assert self.program is not None - if kind == "func": - err = self.program.set_function_type(subject, new) - elif kind == "data": # a global / data item referenced in the body - err = self.program.set_data_type(subject, new) - else: # lvar of the current function - err = self.program.set_lvar_type(self._cur.ea, word, new) - if err: - self.app.call_from_thread(self._status, f"retype failed: {err}") - return - self.app.call_from_thread(self._after_retype, kind, word) + # Read-only views onto the controller's state. The pilot suite and the RPC + # layer read the trace position by these names; they are properties rather + # than attributes so there is exactly one owner and no copy to drift. + @property + def _trace(self): # type: ignore[no-untyped-def] + return self.trace_ctl.trace - def _after_retype(self, kind: str, word: str) -> None: - # A type change alters the pseudocode (and disasm operand types), so - # recompile via the name-generation invalidation and reopen in place. - self.program.bump_names() - self._reload_active_code() - self._dirty = True - what = "prototype" if kind == "func" else f"'{word}'" - self._status(f"retyped {what} (Ctrl+S to save)") + @property + def _t(self) -> int: + return self.trace_ctl.t - # -- typed data definition (make_data, IDA 'd') ----------------------- # - @staticmethod - def _default_data_type(head) -> str: # type: ignore[no-untyped-def] - """A sensible prefill C type for defining data over ``head``.""" - sz = getattr(head, "size", 0) or 0 - return {1: "unsigned __int8", 2: "unsigned __int16", - 4: "unsigned __int32", 8: "unsigned __int64"}.get( - sz, f"char[{sz}]" if sz > 0 else "unsigned __int8") + @property + def _trail_map(self) -> list: + return self.trace_ctl.trail_map - def on_make_data_requested(self, msg: MakeDataRequested) -> 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 define data") - return - self._makedata_ctx = (view, ea) - self.query_one("#status", Static).display = False - inp = self.query_one("#makedata", Input) - inp.placeholder = (f"data type @ {ea:#x} (e.g. int, char[16], my_struct)" - " — Enter=apply Esc=cancel") - inp.can_focus = True - inp.display = True - head = view.cur_head() if isinstance(view, ListingView) else None - inp.value = self._default_data_type(head) if head is not None else "int" - inp.focus() + @property + def _trail_map_ea(self): # type: ignore[no-untyped-def] + return self.trace_ctl.trail_map_ea - def _end_makedata(self) -> None: - inp = self.query_one("#makedata", Input) - inp.display = False - inp.can_focus = False - self.query_one("#status", Static).display = True - view = self._makedata_ctx[0] - if view is not None: - view.focus() + @property + def _trail_line_of(self) -> dict: + return self.trace_ctl.trail_line_of - @work(thread=True, exclusive=True, group="makedata") - 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) - except Exception as e: # noqa: BLE001 - 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 - _cur, top = self._anchor_rows(anchor, lm, ea) - self.app.call_from_thread( - 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="load-trace") + def _load_trace(self) -> None: + self.trace_ctl.load() - @work(thread=True, exclusive=True, group="rename") - def _do_rename(self, view, old: str, new: str) -> None: # type: ignore[no-untyped-def] - assert self.program is not None - prog, cur = self.program, self._cur - kind = "data" - addr: int | None = None - batch: dict = {"data": {"old": old, "new": new}} - resolved: int | None = None - try: - resolved = prog.resolve(old) - except Exception: # noqa: BLE001 - resolved = None - if resolved is not None: - fn = prog.function_of(resolved) - if fn is not None and fn.addr == resolved: - kind, addr = "func", resolved - batch = {"func": {"addr": hex(resolved), "name": new}} - else: - kind, batch = "data", {"data": {"old": old, "new": new}} - elif isinstance(view, DecompView) and cur is not None: - dec = prog.decompile(cur.ea) - ref = next((r for r in dec.refs if r.name == old), None) - if ref is not None: - fn = prog.function_of(ref.addr) - if fn is not None and fn.addr == ref.addr: - kind, addr = "func", ref.addr - batch = {"func": {"addr": hex(ref.addr), "name": new}} - else: - kind, batch = "data", {"data": {"old": old, "new": new}} - else: - kind = "local" - batch = {"local": {"func_addr": hex(cur.ea), "old": old, "new": new}} - elif cur is not None: # disasm view - if old.startswith(("var_", "arg_")): - kind = "stack" - batch = {"stack": {"func_addr": hex(cur.ea), "old": old, "new": new}} - try: - res = prog.client.call("rename", batch=batch) - except IDAToolError as e: - self.app.call_from_thread(self._status, f"rename failed: {e.message}") - return - summary = res.get("summary", {}) if isinstance(res, dict) else {} - if not (summary.get("ok", 0) > 0 and summary.get("failed", 0) == 0): - msg = "rename failed" - for catk in ("func", "data", "local", "stack"): - items = res.get(catk) if isinstance(res, dict) else None - if isinstance(items, list) and items and items[0].get("error"): - msg = f"rename failed: {items[0]['error']}" - self.app.call_from_thread(self._status, msg) - return - self.app.call_from_thread(self._after_rename, kind, addr, old, new) + def _seek(self, idx: int, follow: bool = True) -> None: + self.trace_ctl.seek(idx, follow) - def _after_rename(self, kind: str, addr: int | None, old: str, new: str) -> None: - cur = self._cur - # A renamed symbol can appear in many functions, so invalidate globally; - # each function refreshes its names the next time it's viewed. - self.program.bump_names() - self._reload_active_code() - if kind == "func" and addr is not None and self._func_index is not None: - self._func_index.update_name(addr, new) - for e in self._nav: - if e.ea == addr: - e.name = new - # Update the one cell in place (a full rebuild would race the - # initial streaming load and duplicate row keys). - table = self.query_one("#func-table", DataTable) - try: - name_col = list(table.columns.keys())[1] - table.update_cell(str(addr), name_col, new) - except Exception: # noqa: BLE001 -- row filtered out / not yet streamed - pass - self._dirty = True - self._status(f"renamed {old} → {new} (Ctrl+S to save)") + def _step(self, delta: int) -> None: + self.trace_ctl.step(delta) - @work(thread=True, exclusive=True, group="rename") - def _do_name_addr(self, addr: int, name: str) -> None: # worker context - """Set a label at ``addr`` (listing 'n'). Works on a bare/undefined byte - — unlike the symbol-by-name path, this names the address directly.""" - assert self.program is not None - try: - res = self.program.client.call( - "rename", batch={"data": {"addr": hex(addr), "new": name}}) - except IDAToolError as e: - self.app.call_from_thread(self._status, f"name failed: {e.message}") - return - summary = res.get("summary", {}) if isinstance(res, dict) else {} - if not (summary.get("ok", 0) > 0 and summary.get("failed", 0) == 0): - err = "name failed" - items = res.get("data") if isinstance(res, dict) else None - if isinstance(items, list) and items and items[0].get("error"): - err = f"name failed: {items[0]['error']}" - self.app.call_from_thread(self._status, err) - return - # The label shows in the listing's head rows -> invalidate + reopen. - self.program.bump_items() - lm = self.program.listing(addr) - label = self.program.region_label(addr) - idx = max(lm.ensure_ea(addr), 0) if lm is not None else 0 - self.app.call_from_thread(self._open_at_named, label, addr, idx, name) + def _step_over(self, direction: int) -> None: + self.trace_ctl.step_over(direction) - def _open_at_named(self, label: str, addr: int, idx: int, name: str) -> None: - self._open_at(addr, label, idx, False, -1, 0, True) - self._dirty = True - self._status(f"named {addr:#x} → {name} (Ctrl+S to save)") + def _paint_trail(self) -> None: + self.trace_ctl.paint_trail() - # -- item structure edits (IDA c/p/u) --------------------------------- # - def on_edit_item_requested(self, msg: EditItemRequested) -> 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._anchor()) + @work(thread=True, exclusive=True, group="split-resync") + def _resync_decomp_async(self, ea: int) -> None: + self._resync_decomp(ea) - @work(thread=True, exclusive=True, group="edititem") - 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", - "thumb": "switched decoding", "thumbscan": "scanned"}[kind] - try: - if kind == "code": - # 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": - 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}") - return - # Structure changed everywhere: drop all item/function/decomp caches. - 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, - 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, None, top) - self.app.call_from_thread(self._edit_done, anchor) + def action_seek_next_hit(self) -> None: + self.trace_ctl.seek_hit(1) - def _edit_done(self, anchor: ViewAnchor) -> None: - """One place where an edit's aftermath is settled. + def action_seek_prev_hit(self) -> None: + self.trace_ctl.seek_hit(-1) - 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 - 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 action_seek_reg_write(self) -> None: + self.trace_ctl.seek_reg_write() + + def action_step_fwd(self) -> None: + self.trace_ctl.step(1) + + def action_step_back(self) -> None: + self.trace_ctl.step(-1) + + def action_step_over_fwd(self) -> None: + self.trace_ctl.step_over(1) + + def action_step_over_back(self) -> None: + self.trace_ctl.step_over(-1) @work(thread=True, exclusive=True, group="load-funcs") def _reindex_functions(self) -> None: @@ -5326,7 +7256,8 @@ class IdaTui(App): def _save(self) -> None: assert self.program is not None try: - self.program.client.call("idb_save", timeout=300.0) + self.journal.flush(self.program) # ride along into the .i64 + self.program.client.save_database() except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"save failed: {e}") return @@ -5381,7 +7312,8 @@ class IdaTui(App): idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 name = fn.name if fn is not None else self.program.region_label(ea) self.app.call_from_thread( - self._open_at, ea, name, idx, push, -1, 0, fn is None, focus_name) + self._open_at_if_current, seq, ea, name, idx, push, 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, @@ -5399,7 +7331,7 @@ class IdaTui(App): # Snapshot where we jumped from so 'back' returns there. If that was # the pseudocode (F5 makes a transient _cur not yet on the stack), # record its position and push it as a decomp entry. - if self._active == "decomp" and self._cur is not None: + if self.is_decomp and self._cur is not None: dv = self.query_one(DecompView) src = self._cur src.view = "decomp" @@ -5554,6 +7486,18 @@ class IdaTui(App): cur = row_of(fallback_ea) return (cur, row_of(a.top_ea)) + def _open_at_if_current(self, seq: int, ea: int, name: str, cursor: int, + push: bool, is_region: bool, + focus_name: str | None) -> None: + """Apply a navigation result only if it's still the one being awaited. + + The decompiler path has had this since 756589a; the listing path hadn't, + so a slow navigation could still land on top of a newer one. + """ + if seq != self._nav_seq: + return + self._open_at(ea, name, cursor, push, -1, 0, is_region, focus_name) + 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, @@ -5602,37 +7546,22 @@ class IdaTui(App): self._flash = None if event.key != "escape": return - if self.query_one("#search", Input).display: - event.stop() - event.prevent_default() - self._end_search(cancel=True) - return - if self.query_one("#rename", Input).display: + # Esc closes whichever prompt is up. This used to be one copy of these + # four lines per prompt, which is how they drifted apart -- search + # cancelled its highlight, goto restored focus, the edit prompts did + # neither consistently. + prompt = self.prompts.active() + if prompt is not None: event.stop() event.prevent_default() - self._end_rename() - return - if self.query_one("#comment", Input).display: - event.stop() - event.prevent_default() - self._end_comment() - return - if self.query_one("#retype", Input).display: - event.stop() - event.prevent_default() - self._end_retype() - return - if self.query_one("#makedata", Input).display: - event.stop() - event.prevent_default() - self._end_makedata() - return - if self.query_one("#goto", Input).display: - event.stop() - event.prevent_default() - self._end_goto() - (self.query_one(HexView) if self._active == "hex" - else self._code_view()).focus() + if prompt.id == "search": + self._end_search(cancel=True) + elif prompt.id in ("goto", "export"): + self._end_goto() if prompt.id == "goto" else self._end_export() + (self.query_one(HexView) if self.is_hex + else (self._code_view() or self.query_one(ListingView))).focus() + else: + prompt.close() return fi = self.query_one("#func-filter", Input) if fi.display: @@ -5657,42 +7586,32 @@ class IdaTui(App): view.repeat_last(direction) self._end_search() return - if inp.id == "rename": - view, old = self._rename_ctx - addr = self._rename_addr # capture before _end_rename clears it - self._end_rename() - if addr is not None: # listing: name this address (create a label) - if value and value != old: - self._do_name_addr(addr, value) - return - if view is not None and value and value != old: - self._do_rename(view, old, value) - return - if inp.id == "comment": - view, ea, existing = self._comment_ctx - self._end_comment() - if view is not None and value != existing: # empty value clears it - self._do_comment(ea, value) - return - if inp.id == "retype": - view, kind, subject, word = self._retype_ctx - self._end_retype() - if view is not None and value: - self._do_retype(kind, subject, word, value) - return - if inp.id == "makedata": - view, ea = self._makedata_ctx - self._end_makedata() - if view is not None and value: - self._do_make_data(ea, value, self._anchor()) + # The edit prompts all submit the same way: take the context the prompt + # was holding (close() hands it over, so it can't be read twice or go + # stale) and let the controller decide what to do with it. + submit = {"rename": self.edits.submit_rename, + "comment": self.edits.submit_comment, + "retype": self.edits.submit_retype, + "makedata": self.edits.submit_make_data}.get(inp.id or "") + if submit is not None: + ctx = self.prompts[inp.id].close() + if ctx is not None: + submit(ctx, value) return if inp.id == "goto": self._end_goto() - (self.query_one(HexView) if self._active == "hex" - else self._code_view()).focus() + (self.query_one(HexView) if self.is_hex + else (self._code_view() or self.query_one(ListingView))).focus() if value: self._goto(value) return + if inp.id == "export": + self._end_export() + (self.query_one(HexView) if self.is_hex + else (self._code_view() or self.query_one(ListingView))).focus() + if value: + self.export_findings(value) + return # filter mode: already applied incrementally; Enter just confirms + closes. self._apply_filter(value) self.query_one("#func-table", DataTable).focus() @@ -5705,15 +7624,63 @@ class IdaTui(App): """ return self._active_code_view() + def _line_ea_for(self, view) -> int | None: # type: ignore[no-untyped-def] + """Address of the line under the cursor in either code view. + + Shared, not an edit helper: the comment prompt, the goto prompt's + readback and rpc.py's `where` all ask the same question. + """ + if isinstance(view, ListingView): + return view._cursor_ea() + if isinstance(view, DecompView): + return view._line_ea(view.cursor) + return None + + # Read _active through these rather than comparing strings. The bare + # comparisons are what let the old "disasm" value take the wrong branch in + # five places, and they are what the next new mode would have to hunt down. + @property + def is_listing(self) -> bool: + return self._active == ViewMode.LISTING + + @property + def is_decomp(self) -> bool: + return self._active == ViewMode.DECOMP + + @property + def is_hex(self) -> bool: + return self._active == ViewMode.HEX + + @property + def is_graph(self) -> bool: + return self._active == ViewMode.GRAPH + + @property + def in_code(self) -> bool: + """A code view over a NavEntry -- where follow/xrefs/rename mean something.""" + return self._active in ViewMode.code_modes() + def _active_code_view(self): # type: ignore[no-untyped-def] - """The currently-shown code widget (for reading the cursor address).""" - if self._active in ("listing", "disasm"): + """The currently-shown code widget (for reading the cursor address). + + Everything downstream trusts ``_active``, so a new mode that forgets to + appear here doesn't degrade -- it returns None and the first caller that + dereferences it crashes the app (which is exactly how graph mode + announced itself the first time it was driven). + """ + if self.is_listing: return self.query_one(ListingView) - if self._active == "decomp": + if self.is_decomp: return self.query_one(DecompView) + if self.is_graph: + return self.query_one(GraphView) return None - def _palette_action(self, name: str) -> None: + def _focus_code_view(self) -> None: + """Put focus back on whichever code view is showing.""" + (self._active_code_view() or self.query_one(ListingView)).focus() + + def _palette_action(self, name: str, *args) -> None: """Run a cursor-scoped code-view action (rename/xrefs/follow/…) picked from the command palette against the active code view.""" view = self._active_code_view() @@ -5725,7 +7692,7 @@ class IdaTui(App): if fn is None: self._status(f"'{name}' isn't available in this view") return - fn() + fn(*args) def action_continuous_here(self) -> None: """'L': open the continuous segment listing at the cursor — one long @@ -5761,9 +7728,16 @@ class IdaTui(App): except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"goto: {e}") return - if self._active == "hex": + if self.is_hex: self.app.call_from_thread(self._hex_goto, ea) return + # A goto that lands inside the graph you're already looking at should + # move the cursor, not tear the picture down and build the same one. + if self.is_graph: + gv = self.query_one(GraphView) + if gv.fc is not None and gv.fc.block_at(ea) is not None: + self.app.call_from_thread(gv.goto_ea, ea) + return # Navigate to the containing function at the right line (handles both a # function name and a mid-function address). self._do_navigate(ea, push=True) @@ -5806,9 +7780,9 @@ class IdaTui(App): self._do_navigate(ea, push) - def _code_mode(self) -> str: + def _code_mode(self) -> ViewMode: """The code view to return to from hex — always the unified listing.""" - return "listing" + return ViewMode.LISTING def _open_entry(self, entry: NavEntry, push: bool) -> None: if self.program is None: @@ -5819,7 +7793,7 @@ class IdaTui(App): if entry.view == "decomp": # This entry was viewed in the decompiler (a jump from pseudocode, or # a back/forward to one) — restore it there instead of the listing. - self._active = "decomp" + self._active = ViewMode.DECOMP dec = self.query_one(DecompView) if dec.loaded_ea == entry.ea: # already decompiled: reposition without a recompile @@ -5848,13 +7822,19 @@ class IdaTui(App): lst.load( lm, entry.name, cursor=entry.cursor, cursor_x=entry.cursor_x, scroll_y=sy, focus=focus) - self._active = "listing" + self._active = ViewMode.LISTING self._show_active() + # Graph mode is sticky: following a call from the graph should land in + # the callee's graph, not dump you back into the listing. The rebuild is + # async and re-checks _cur, so a fast second jump just drops the stale one. + if self._graph_sticky and entry.ea: + self._load_graph(entry.ea, entry.ea) # Prompt overlays that own the keyboard while visible; a background # navigation must not yank focus out from under them (else typed keys leak # into a code view as destructive verbs — e.g. 'u' = undefine). - _PROMPT_IDS = ("search", "rename", "comment", "retype", "goto", "func-filter") + _PROMPT_IDS = ("search", "rename", "comment", "retype", "goto", "export", + "func-filter") def _prompt_active(self) -> bool: for iid in self._PROMPT_IDS: @@ -5869,12 +7849,13 @@ class IdaTui(App): dec = self.query_one(DecompView) lst = self.query_one(ListingView) hx = self.query_one(HexView) + gv = self.query_one(GraphView) # Don't steal focus from an open prompt (search/rename/…) — a late async # navigation completing here would otherwise pull it into the code view. grab = not self._prompt_active() - if self._split and self._active in ("listing", "decomp"): + if self._split and self._active in (ViewMode.LISTING, ViewMode.DECOMP): # Side-by-side: listing (left) + pseudocode (right), one focused. - hx.display = False + hx.display = gv.display = False lst.display = dec.display = True self.query_one("#panes").set_class(True, "split") busy = self._cur is not None and dec.loaded_ea != self._cur.ea @@ -5884,7 +7865,7 @@ class IdaTui(App): else: dec.loading = False if grab: - (dec if self._active == "decomp" else lst).focus() + (dec if self.is_decomp else lst).focus() 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 @@ -5900,13 +7881,19 @@ class IdaTui(App): self._split_eamap = [] self._split_ea2line = {} self._split_range = None - dec.display = lst.display = hx.display = False - if self._active in ("listing", "disasm"): + dec.display = lst.display = hx.display = gv.display = False + if self.is_graph: + gv.display = True + if grab: + gv.focus() + self._graph_status() + return + if self.is_listing: lst.display = True if grab: lst.focus() self._status_for_cur("listing") - elif self._active == "hex": + elif self.is_hex: hx.display = True if grab: hx.focus() @@ -5947,7 +7934,7 @@ class IdaTui(App): self._status("hex: no loaded segments") return hx.load(model, ea) - if self._active == "hex": + if self.is_hex: hx.focus() def _hex_status(self, va: int) -> None: @@ -5977,6 +7964,12 @@ class IdaTui(App): assert self.program is not None dec = self.program.decompile(ea) why = "" + if not dec.failed: + # Where the number literals are, fetched in the same worker as the + # decompile (it is one call and the answer is cached with it) so the + # view can mark the one under the cursor without a round trip per + # keypress. + self._pending_nums = self.program.pc_nums(ea) 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. @@ -5994,14 +7987,14 @@ class IdaTui(App): 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. + # return to exactly where we left; otherwise just show the listing. if self._cur is None or self._cur.ea != ea: return # navigated away; stale result if self._decomp_return is not None: ret = self._decomp_return self._decomp_return = None self._cur = ret - self._active = "listing" + self._active = ViewMode.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" @@ -6010,11 +8003,14 @@ class IdaTui(App): self._status(msg, priority=True) self._open_entry(ret, push=False) return - self._active = "disasm" + # LISTING, not the old "disasm": it is the same widget, and a value + # only this path produced meant half the app took the wrong branch + # for it (see ViewMode). + self._active = ViewMode.LISTING self._status(f"{name}: cannot decompile{detail}", priority=True) self._show_active() return - if self._active == "decomp": + if self.is_decomp: view.focus() # loading cover had blurred it; restore focus # Restore the saved pseudocode position when returning to this function. cur = self._cur @@ -6025,6 +8021,7 @@ class IdaTui(App): sx = cur.dec_scroll_x if same else 0 note = " (truncated)" if dec.truncated else "" view.show(ea, dec.code or "", cursor=c, cursor_x=cx, scroll_y=sy, scroll_x=sx) + view.set_nums(self._pending_nums) if self._split: self._sync_split(self._active) # crude link now self._load_split_map(ea) # then upgrade to the region map @@ -6033,11 +8030,22 @@ class IdaTui(App): self._status( f"{name} @ {ea:#x} [pseudocode {len(dec.code or '')} chars]{note}") - def _sync_split(self, source: str) -> None: + def _sync_split(self, source: str, resync: bool = True) -> None: """Split view: highlight (+ scroll into view) the companion pane's location for the focused pane's cursor. The companion only gets a band + scroll (its cursor never moves), so there is no echo/ping-pong. Uses the - rich per-line ea map (decomp_map) when loaded, else the single marker.""" + rich per-line ea map (decomp_map) when loaded, else the single marker. + + ``resync=False`` means "the decompiler has just been re-pointed for this + anchor, don't ask again". Without it this recursed forever: _split_range + is the min/max of the decomp_map's addresses, which does NOT cover every + address in the function (Hex-Rays doesn't attribute them all), so an + anchor inside the loaded function but outside that span asked for a + resync, got back the function already loaded, re-entered here, and asked + again -- one worker and one lookup_funcs round trip per iteration, for as + long as the cursor sat there. 21,156 calls for a single address in one + scenario, and in the live app an idle split view pegging the worker. + """ if not self._split or self.program is None: return lst = self.query_one(ListingView) @@ -6070,7 +8078,7 @@ class IdaTui(App): dec.set_link(None) return rng = self._split_range - if rng is not None and not (rng[0] <= ea <= rng[1]): + if resync and rng is not None and not (rng[0] <= ea <= rng[1]): # left the decompiled function — follow to whatever function is # under the anchor (the unified view spans many functions). self._resync_decomp(ea) @@ -6096,11 +8104,11 @@ class IdaTui(App): return top, 0 def on_listing_view_scrolled(self, msg: "ListingView.Scrolled") -> None: - if self._split and self._active == "listing": + if self._split and self.is_listing: self._sync_split("listing") def on_decomp_view_scrolled(self, msg: "DecompView.Scrolled") -> None: - if self._split and self._active == "decomp": + if self._split and self.is_decomp: self._sync_split("decomp") @work(thread=True, group="split-resync", exclusive=True) @@ -6120,7 +8128,10 @@ class IdaTui(App): return self._cur.ea, self._cur.name = fn.addr, fn.name if dec.loaded_ea == fn.addr: # already decompiled (scrolled back): just relink - self._sync_split("listing") + # resync=False: we ARE the resync. Re-entering the resync branch is + # how this looped forever for an address the decomp_map's span + # doesn't cover. + self._sync_split("listing", resync=False) return dec.loading = True self._load_decomp(fn.addr, fn.name) # -> _apply_decomp -> map -> re-sync @@ -6129,7 +8140,7 @@ class IdaTui(App): """A split-aware status line reflecting the focused pane + the link.""" if self._cur is None: return - if self._active == "decomp": + if self.is_decomp: dec = self.query_one(DecompView) ea = dec._line_ea(dec.cursor) n = (len(self._split_eamap[dec.cursor]) @@ -6167,8 +8178,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 = [] @@ -6179,7 +8200,12 @@ 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.trace_ctl.adopt_map(ea, m, self._split_ea2line, 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) @@ -6187,7 +8213,7 @@ class IdaTui(App): self._nav[-1].dec_cursor = msg.index self._nav[-1].dec_cursor_x = dv.cursor_x if self._split: - if self._active == "decomp": + if self.is_decomp: self._sync_split("decomp") self._split_status() return @@ -6213,7 +8239,7 @@ class IdaTui(App): if msg.index >= 0: self._nav[-1].scroll_y = round(lst.scroll_offset.y) if self._split: - if self._active == "listing": + if self.is_listing: self._sync_split("listing") self._split_status() return @@ -6224,7 +8250,9 @@ class IdaTui(App): "(c code · p func · u undefine · Enter follow)") # -- teardown ---------------------------------------------------------- # - def on_unmount(self) -> None: + async def on_unmount(self) -> None: + if self._rpc is not None: + await self._rpc.stop() if self._ka is not None: self._ka.stop() if self.program is not None: @@ -6236,7 +8264,7 @@ class IdaTui(App): elif self.client is not None: if self._save_on_exit is None and self._dirty: try: # unexpected teardown with edits: don't drop them - self.client.call("idb_save", timeout=600.0) + self.client.save_database() except Exception: # noqa: BLE001 pass self.client.close() |