diff options
Diffstat (limited to 'idatui')
| -rw-r--r-- | idatui/app.py | 902 | ||||
| -rw-r--r-- | idatui/domain.py | 224 | ||||
| -rw-r--r-- | idatui/formats.py | 15 | ||||
| -rw-r--r-- | idatui/project.py | 23 | ||||
| -rw-r--r-- | idatui/worker.py | 9 |
5 files changed, 775 insertions, 398 deletions
diff --git a/idatui/app.py b/idatui/app.py index f861f42..1ea630e 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -53,6 +53,23 @@ from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct _S_ADDR = Style(color="#6b7684") _S_LABEL = Style(color="#7aa2f7", bold=True) _S_INSN = Style(color="#c3cad3") +#: IDA's token kinds -> the measured palette. The rule that keeps a dense +#: disassembly readable: NEUTRALS for the machine (mnemonic brightest because you +#: scan down that column, registers at body weight because they're most of the +#: text), HUES only where they mean something (numbers, strings, symbols), +#: structure recedes so brackets and commas stop competing with operands. +_S_SPAN = { + "insn": Style(color="#e8ecf2"), # 15.3:1 mnemonic / directive + "reg": Style(color="#c3cad3"), # 11.0:1 registers = body weight + "num": Style(color="#d8a657"), # 8.2:1 immediates, offsets + "str": Style(color="#9ece6a"), # 9.9:1 string literals + "name": Style(color="#7aa2f7"), # 7.2:1 symbols / xref targets + "seg": Style(color="#93aee0"), # 8.1:1 segment names + "cmt": Style(color="#7c8b9e", italic=True), # 5.2:1 + "punct": Style(color="#626c7a"), # 3.4:1 brackets, commas, +/- + "err": Style(color="#c9762f"), # IDA's own error marker + "text": Style(color="#c3cad3"), # 11.0:1 anything unclassified +} _S_MNEM = Style(color="#e8ecf2") _S_OPBYTES = Style(color="#5e6875") # raw opcode bytes column _S_DATA = Style(color="#d8a657") @@ -112,6 +129,31 @@ class BinaryState: @dataclass +class ViewAnchor: + """Where the user is looking, expressed in ADDRESSES. + + Every path that rebuilds a model must round-trip through this. Row indices + do NOT survive a rebuild: defining code collapses four undefined byte rows + into one instruction row, undefining does the reverse, and a rename can add + or remove banner rows above a function. Anything that remembers an index + puts the user somewhere else afterwards, which reads as "the edit jumped my + screen" or, worse, "the edit didn't apply". + + ``flash`` travels with it because the same rebuild also decides what the + status bar says: the reload writes its own status when it lands, so an edit + that doesn't hand its message over here gets silently overwritten. + """ + + view: str = "listing" + ea: int | None = None # cursor address + top_ea: int | None = None # first visible address + cursor_x: int = 0 + flash: str | None = None + #: The edit changed which functions exist, so the index must be rebuilt. + refresh_functions: bool = False + + +@dataclass class NavEntry: ea: int name: str @@ -670,305 +712,12 @@ class SearchMixin: # --------------------------------------------------------------------------- # # Virtualized disassembly view # --------------------------------------------------------------------------- # -class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True): - """A line-virtualized disassembly listing for a single function.""" - - BINDINGS = [ - Binding("j,down", "cursor_down", "Down", show=False), - Binding("k,up", "cursor_up", "Up", show=False), - Binding("ctrl+d", "half_page(1)", "½↓", show=False), - Binding("ctrl+u", "half_page(-1)", "½↑", show=False), - Binding("pagedown", "page(1)", "PgDn", show=False), - Binding("pageup", "page(-1)", "PgUp", show=False), - Binding("home", "goto_top", "Top", show=False), - Binding("G,end", "goto_bottom", "Bottom", show=False), - Binding("o", "toggle_opcodes", "Opcodes", show=False), - Binding("c", "define_code", "Code", show=False), - Binding("p", "define_func", "Func", show=False), - Binding("u", "undefine", "Undef", show=False), - Binding("tab,shift+tab", "app.toggle_view", "Pseudocode", priority=True), - Binding("f5", "app.toggle_view", "Decompile", priority=True), - Binding("L", "app.continuous_here", "Listing"), - *SearchMixin.SEARCH_BINDINGS, - *NavMixin.NAV_BINDINGS, - *ColumnCursor.COL_BINDINGS, - ] - - cursor = reactive(0, repaint=False) - cursor_x = reactive(0, repaint=False) - - class CursorMoved(Message): - """Posted when the disasm cursor moves; carries the instruction ea.""" - - def __init__(self, index: int, ea: int | None) -> None: - super().__init__() - self.index = index - self.ea = ea - - def __init__(self) -> None: - super().__init__() - self.model: DisasmModel | None = None - self.total = 0 - self._name = "" - self._pending_scroll_y: int | None = None - self._term = "" - self._matches: list[int] = [] - self._ranges: dict[int, list[tuple[int, int]]] = {} - self._search_texts: list[str] | None = None - self._show_ops = True - self._op_w = 0 # char width of the hex-bytes field (excl. trailing gap) - - def _op_field(self, line) -> str: # type: ignore[no-untyped-def] - """The padded opcode-bytes column (empty when hidden). Kept identical - between the rendered strip and the plain text so cursor/search offsets - line up.""" - if not self._show_ops or self._op_w <= 0: - return "" - raw = line.raw or b"" - return " ".join(f"{b:02X}" for b in raw).ljust(self._op_w) + " " - - def _line_plain(self, idx: int) -> str | None: - if self.model is None: - return None - line = self.model.cached_line(idx) - if line is None: - return None - s = f"{line.ea:08X} " + self._op_field(line) - if line.label: - s += f"{line.label}: " - return s + line.text - - # -- public API -------------------------------------------------------- # - def load(self, model: DisasmModel, name: str, cursor: int = 0, - cursor_x: int = 0, scroll_y: int | None = None) -> None: - self.model = model - self._name = name - self.total = 0 - self.cursor = cursor - self.cursor_x = cursor_x - self._pending_scroll_y = scroll_y - # NB: don't zero virtual_size here — that snaps the scroll to 0 and - # causes a visible jump before _on_primed restores the target scroll. - self._matches = [] - self._ranges = {} - self._search_texts = None - self._prime() - - # -- search hooks ------------------------------------------------------ # - def _fmt(self, line) -> str: # type: ignore[no-untyped-def] - s = f"{line.ea:08X} " + self._op_field(line) - if line.label: - s += f"{line.label}: " - return s + line.text - - def _search_line_count(self) -> int: - return self.total - - def _search_line_text(self, i: int) -> str | None: - t = self._search_texts - return t[i] if t is not None and 0 <= i < len(t) else None - - def _search_ensure(self, done) -> None: - if self._search_texts is not None: - done() - return - self._app_status(f"/{self._term}/ indexing {self.total} lines…") - self._index_for_search(done) - - @work(thread=True, exclusive=True, group="search-index") - def _index_for_search(self, done) -> None: - model = self.model - if model is None: - self.app.call_from_thread(done) - return - texts: list[str] = [] - off, total = 0, self.total - while off < total: - lines = model.lines(off, DisasmModel.BLOCK, prefetch=False) - if not lines: - break - texts.extend(self._fmt(ln) for ln in lines) - off += len(lines) - self._search_texts = texts - self.app.call_from_thread(done) - - @work(thread=True, exclusive=True, group="disasm-prime") - def _prime(self) -> None: - model = self.model - if model is None: - return - total = model.total() - height = max(self.size.height, 1) - model.lines(0, min(total, height + DisasmModel.BLOCK), prefetch=True) - if self.cursor: - model.lines(max(self.cursor - 2, 0), height, prefetch=True) - self.app.call_from_thread(self._on_primed, total) - - def _on_primed(self, total: int) -> None: - self.total = total - self.virtual_size = Size(0, total) - self._update_op_w() # provisional width from the primed window - self._scan_op_width() # settle it against the whole function - self._clamp_x() # cursor line is now cached; keep the column in range - if self._pending_scroll_y is not None and self._pending_scroll_y >= 0: - self._apply_scroll(min(self._pending_scroll_y, max(total - 1, 0))) - else: - self._scroll_cursor_into_view() - self._pending_scroll_y = None - self.refresh() - - # -- rendering --------------------------------------------------------- # - def render_line(self, y: int) -> Strip: - model = self.model - width = self.size.width - if model is None or self.total == 0: - return Strip([Segment("".ljust(width), _S_DIM)]) - top = round(self.scroll_offset.y) - if y == 0: # once per refresh: warm the visible window + a little ahead - self._ensure_window(top) - idx = top + y - if idx >= self.total: - return Strip([Segment("".ljust(width), _S_INSN)]) - line = model.cached_line(idx) - is_cursor = idx == self.cursor - if line is None: - strip = Strip([Segment(f" {idx:>8} …", _S_DIM)]) - else: - segs: list[Segment] = [Segment(f"{line.ea:08X} ", _S_ADDR)] - op = self._op_field(line) - if op: - segs.append(Segment(op, _S_OPBYTES)) - if line.label: - segs.append(Segment(f"{line.label}: ", _S_LABEL)) - mnem, _, rest = line.text.partition(" ") - segs.append(Segment(mnem, _S_MNEM)) - if rest: - segs.append(Segment(" " + rest, _S_INSN)) - strip = Strip(segs) - if idx in self._ranges: - strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx)) - if is_cursor: - strip = _cursor_decorate(strip, self._line_plain(idx) or "", self.cursor_x) - return strip.adjust_cell_length(width, _S_INSN) - - def _update_op_w(self) -> bool: - """Recompute the opcode column width from the model's widest instruction. - Returns True if it changed.""" - w = 0 - if self.model is not None and self._show_ops: - mx = self.model.max_raw_len() - w = max(mx * 3 - 1, 0) if mx > 0 else 0 - if w != self._op_w: - self._op_w = w - return True - return False - - @work(thread=True, exclusive=True, group="disasm-opwidth") - def _scan_op_width(self) -> None: - model = self.model - if model is None: - return - model.scan_bytes() # fetch all blocks -> stable widest instruction - self.app.call_from_thread(self._settle_op_w) - - def _settle_op_w(self) -> None: - if self._update_op_w(): - self._search_texts = None # layout changed -> stale offsets - self.refresh() - - def action_toggle_opcodes(self) -> None: - self._show_ops = not self._show_ops - self._update_op_w() - self._search_texts = None # column layout changed -> reindex on next search - self._ranges = {} - self._clamp_x() - self.refresh() - self._app_status("opcodes " + ("on" if self._show_ops else "off")) - - def _ensure_window(self, top: int) -> None: - if self.model is None: - return - height = max(self.size.height, 1) - start = max(top - DisasmModel.BLOCK, 0) - count = height + 2 * DisasmModel.BLOCK - if not self.model.is_cached(top, height): - self._fetch_window(start, count) - else: - self.model.ensure_async(start, count) # warm neighbors - - @work(thread=True, exclusive=False, group="disasm-fetch") - def _fetch_window(self, start: int, count: int) -> None: - model = self.model - if model is None: - return - model.lines(start, count, prefetch=True) - self.app.call_from_thread(self.refresh) - - # -- navigation -------------------------------------------------------- # - def _visible_height(self) -> int: - return max(self.size.height, 1) - - def _scroll_cursor_into_view(self) -> None: - height = self._visible_height() - top = round(self.scroll_offset.y) - if self.cursor < top: - self.scroll_to(y=self.cursor, animate=False) - elif self.cursor >= top + height: - self.scroll_to(y=max(self.cursor - height + 1, 0), animate=False) - - def _move(self, delta: int) -> None: - if self.total == 0: - return - old = self.cursor - before = round(self.scroll_offset.y) - self.cursor = max(0, min(self.total - 1, self.cursor + delta)) - self._clamp_x() - self._scroll_cursor_into_view() - if round(self.scroll_offset.y) != before: - self.refresh() # scrolled: the whole viewport shifted - else: - _refresh_lines(self, old, self.cursor) # only the two changed rows - self._refresh_hl() - self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea())) - - def _after_cursor_move(self) -> None: - self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea())) - - def _cursor_ea(self) -> int | None: - if self.model is None: - return None - line = self.model.cached_line(self.cursor) - return line.ea if line else None - - # -- item structure edits (IDA c/p/u) --------------------------------- # - def action_define_code(self) -> None: - self.post_message(EditItemRequested(self, "code")) - - def action_define_func(self) -> None: - self.post_message(EditItemRequested(self, "func")) - - def action_undefine(self) -> None: - self.post_message(EditItemRequested(self, "undef")) - - def action_cursor_down(self) -> None: - self._move(1) - - def action_cursor_up(self) -> None: - self._move(-1) - - def action_goto_top(self) -> None: - self._move(-self.total) - - def action_goto_bottom(self) -> None: - self._move(self.total) - - # --------------------------------------------------------------------------- # # Virtualized flat listing view (code + data + undefined, per segment) # --------------------------------------------------------------------------- # class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True): """A line-virtualized *flat* listing over one segment: code, data and - undefined heads interleaved (IDA's disassembly view), unlike ``DisasmView`` + undefined heads interleaved (IDA's disassembly view), unlike ``DisasmModel`` which is bounded to one function. Backed by ``ListingModel`` (the ``heads`` server tool). Used for non-function regions and raw segment browsing. """ @@ -991,6 +740,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru Binding("a", "make_string", "Str", show=False), Binding("p", "define_func", "Func", show=False), Binding("u", "undefine", "Undef", show=False), + Binding("t", "toggle_thumb", "ARM/Thumb", show=False), *SearchMixin.SEARCH_BINDINGS, *NavMixin.NAV_BINDINGS, *ColumnCursor.COL_BINDINGS, @@ -1043,6 +793,24 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru return " ".join(f"{b:02X}" for b in raw[:_OP_LIMIT]) + "\u2026" return " ".join(f"{b:02X}" for b in raw) + @staticmethod + def _span_segments(h: Head, fallback: Style): + """Segments for a row's disassembly text. + + Uses IDA's own token classification when the worker supplied it; falls + back to the old mnemonic/rest split so an older worker (or a row whose + spans didn't match the text) still renders. + """ + if h.spans: + return [Segment(t, _S_SPAN.get(k, fallback)) for k, t in h.spans] + if h.kind == "code": + mnem, _, rest = h.text.partition(" ") + segs = [Segment(mnem, _S_MNEM)] + if rest: + segs.append(Segment(" " + rest, fallback)) + return segs + return [Segment(h.text, fallback)] + def _op_field(self, h: Head) -> str: """The padded opcode-bytes column (empty when hidden). Shared format so cursor/search offsets line up.""" @@ -1267,17 +1035,11 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru segs.append(Segment(_LST_INDENT, _S_MEMBER)) if h.name: segs.append(Segment(f"{h.name} ", _S_LABEL)) - if h.kind == "code": - mnem, _, rest = h.text.partition(" ") - segs.append(Segment(mnem, _S_MNEM)) - if rest: - segs.append(Segment(" " + rest, _S_INSN)) - elif h.kind == "data": - segs.append(Segment(h.text, _S_DATA)) - elif h.kind == "member": + if h.kind == "member": segs.append(Segment(h.text, _S_MEMBER)) else: - segs.append(Segment(h.text, _S_UNK)) + base = {"code": _S_INSN, "data": _S_DATA}.get(h.kind, _S_UNK) + segs.extend(self._span_segments(h, base)) strip = Strip(segs) linked = idx in self._link_rows if linked: @@ -1367,6 +1129,9 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru def action_undefine(self) -> None: self.post_message(EditItemRequested(self, "undef")) + def action_toggle_thumb(self) -> None: + self.post_message(EditItemRequested(self, "thumb")) + def action_make_data(self) -> None: self.post_message(MakeDataRequested(self)) @@ -1679,7 +1444,7 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True best, best_ea = i, e return best - # -- navigation (mirrors DisasmView) ---------------------------------- # + # -- navigation -------------------------------------------------------- # def _visible_height(self) -> int: return max(self.size.height, 1) @@ -2566,6 +2331,24 @@ class LoadOptionsScreen(ModalScreen): self._apply("") self.query_one("#pal-input", Input).focus() + def focus_next(self, selector="*"): # type: ignore[override] + """Tab moves between the two things you TYPE into. + + DOM order would stop at the option list on the way, which is + arrow-driven and has nothing to type — and the address you meant to + enter goes into whichever box happened to have focus. Typing an address + into the processor filter is then taken as a processor name, IDA rejects + it, and the open fails; that is a bad enough outcome to be worth + overriding Tab for. + """ + inp = self.query_one("#pal-input", Input) + base = self.query_one("#load-base", Input) + (inp if self.focused is base else base).focus() + return self.focused + + def focus_previous(self, selector="*"): # type: ignore[override] + return self.focus_next() + def on_input_changed(self, event: Input.Changed) -> None: event.stop() if event.input.id == "pal-input": @@ -2739,13 +2522,16 @@ class ConfirmScreen(ModalScreen): Binding("escape,n", "cancel", "Cancel"), ] - def __init__(self, message: str) -> None: + def __init__(self, message: str, note: str = "") -> None: super().__init__() self._message = message + self._note = note def compose(self) -> ComposeResult: with Vertical(id="confirm-box"): - yield Static(self._message, id="confirm-msg") + yield Static(self._message, id="confirm-msg", markup=False) + if self._note: + yield Static(self._note, id="confirm-note", markup=False) yield Static("[Enter/y] confirm [Esc/n] cancel", id="confirm-help") def action_confirm(self) -> None: @@ -3172,7 +2958,6 @@ class IdaTui(App): #left { width: 30%; min-width: 42; max-width: 44; border-right: solid $panel; } #func-table { height: 1fr; } #func-filter { dock: top; } - DisasmView { width: 1fr; padding: 0 1; } DecompView { width: 1fr; } ListingView { width: 1fr; padding: 0 1; } #panes.split ListingView { border-right: tall $panel-lighten-2; } @@ -3238,8 +3023,13 @@ class IdaTui(App): #pal-input { border: none; height: 1; margin: 0 1; background: $panel; color: $text; } #pal-list { height: auto; max-height: 24; } #load-note { height: 2; padding: 1 1 0 1; color: $text-muted; } + /* Cap the processor list so the ADDRESS FIELD is always on screen: with the + palette default (24) the box outgrew the terminal and the field you need + was clipped off the bottom, which read as "Tab does nothing". */ + LoadOptionsScreen #pal-list { max-height: 12; } #load-base { border: none; height: 1; margin: 1 1 0 1; background: $panel; color: $text; } #load-help { height: 1; padding: 0 1; color: $text-muted; } + #confirm-note { height: auto; padding: 0 1; color: $text-muted; } StructEditor { align: center middle; } #se-box { width: 90%; height: 84%; border: thick $accent; background: $panel; } #se-panes { height: 1fr; } @@ -3276,6 +3066,7 @@ class IdaTui(App): Binding("s", "toggle_split", "Split", show=False), Binding("quotation_mark,shift+f12", "strings", "Strings", show=False), Binding("ctrl+o", "switch_binary", "Binaries", show=False), + Binding("ctrl+l", "load_options", "Reload as…", show=False), Binding("f1", "help", "Keys", show=False), Binding("g", "goto", "Goto"), Binding("slash", "filter", "Filter", show=False), @@ -3298,6 +3089,10 @@ class IdaTui(App): self._pending_restore = None # entry to reopen after a switch self._goto_after_switch = None # cross-binary search hit to land on self._hops: list[str] = [] # binaries a navigation crossed FROM + self._load_for_label = None # project binary the dialog is for + self._no_functions = False # analysis produced nothing at all + self._flash: str | None = None # message a pending reload must keep + self._pending_switch = None # switch waiting on that answer self._nav_seq = 0 # bumped per navigation; drops stale ones # None = teardown wasn't an explicit quit (crash/kill): save defensively. # False = the user chose discard, or we already saved on the way out. @@ -3360,8 +3155,8 @@ class IdaTui(App): fp = FunctionsPanel(id="left") fp.display = False # overlay-first: reveal the docked pane with Ctrl+B yield fp - # The unified continuous listing is the one code view; DisasmView is - # deprecated (kept only for DisasmModel, still used by the domain). + # The unified continuous listing is the one code view. (DisasmModel + # is still used by the domain to index a function's instructions.) lst = ListingView() yield lst yield DecompView() @@ -3412,8 +3207,13 @@ class IdaTui(App): # A file no loader recognises has to be described before it can be # opened, so ask BEFORE the worker starts — once IDA has made a database # the answer is baked in and changing it means deleting the .i64. - if self._should_ask_load_options(): - self._ask_load_options() + if self._project is not None: + ref = self._pending_load_ref() + if ref is not None: + self._ask_load_options(ref.source, label=ref.label) + return + elif self._should_ask_load_options(): + self._ask_load_options(self._open_path) return # Show a loading overlay immediately so a slow open/analysis (big binary) # isn't just dead air behind empty panes; dismissed once we land. @@ -3429,8 +3229,8 @@ class IdaTui(App): re-passing switches fails the open); or the file is a format IDA recognises, which is nearly always. """ - if self._project is not None or not self._open_path: - return False # project mode carries per-binary options already + if not self._open_path: + return False if self._load_args: return False from .formats import needs_load_options @@ -3439,21 +3239,152 @@ class IdaTui(App): return False return needs_load_options(self._open_path) - def _ask_load_options(self) -> None: + def action_load_options(self) -> None: + """Ctrl+L: re-open this binary with different load options. + + The database IDA already built has the old processor and base baked into + it and takes precedence over any switches, so re-loading means throwing + it away. That destroys names and comments, hence the confirmation — but + for the case this exists for (a blob loaded as the wrong architecture, + which analysed to nothing) there is nothing to lose and no other way + forward. + """ + if not self._can_reload(): + self._status("nothing to reload") + return + n = len(self._func_index) if self._func_index else 0 + note = ("this image has no functions, so nothing is lost" + if n == 0 else + f"discards the database for this binary \u2014 {n} " + f"function{'s' if n != 1 else ''}, plus any names and comments " + f"you've added") + self.push_screen(ConfirmScreen("Reload with different options?", note), + self._on_reload_confirmed) + + def _on_reload_confirmed(self, yes) -> None: # type: ignore[no-untyped-def] + if not yes: + return + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Drop the worker first: it holds the database open, and the .i64 can't + # be removed (or rebuilt) underneath a live one. + self._release_worker() + self._drop_database() + self._reset_for_reload() + self._load_args = "" + if label is not None and self._project is not None: + self._project.set_load(label, processor="", base=0) + i = self._project._refs.index(self._project.by_label(label)) + self._project._entries[i].pop("processor", None) + self._project._entries[i].pop("base", None) + self._project.save() + self._pending_switch = None + self._ask_load_options(path, label=label) + + def _release_worker(self) -> None: + if self._pool is not None and self._binary is not None: + try: + self._pool.evict(self._binary, save=False) + except Exception: # noqa: BLE001 + pass + elif self.client is not None: + try: + self.client.close() + except Exception: # noqa: BLE001 + pass + self.client = None + self.program = None + + def _drop_database(self) -> None: + """Remove the .i64 (and any unpacked scratch) so the next open re-reads + the raw image with new options.""" + base = self._open_path + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + base = ref.staged + if not base: + return + for suffix in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + for cand in (base + suffix, os.path.splitext(base)[0] + suffix): + try: + os.remove(cand) + except OSError: + pass + + def _reset_for_reload(self) -> None: + self._no_functions = False + self._func_index = None + self._cur = None + self._nav = [] + self._did_auto_land = False + self._pending_restore = None + self._split = False + self.query_one(DecompView).loaded_ea = None + + def _retry_load_options(self) -> None: + """Re-ask after IDA refused what we told it.""" + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Clear the rejected answer or _pending_load_ref would see the + # binary as already described and never ask again. + self._project.set_load(label, processor="", base=0) + self._project._entries[self._project._refs.index(ref)].pop( + "processor", None) + self._project.save() + self._load_args = "" + if path: + self._status("those load options were rejected \u2014 try again") + self._ask_load_options(path, label=label) + + def _pending_load_ref(self, label: str | None = None): # type: ignore[no-untyped-def] + """The project binary about to be opened, if it needs describing. + + Checked against the SOURCE: staging may not have happened yet, and the + question is about the bytes, not where they were copied to. + """ + if self._project is None: + return None + label = label or self._binary or self._project.refs[0].label + ref = self._project.by_label(label) + if ref is None or ref.load_args: + return None + if os.path.exists(ref.db) or os.path.exists( + os.path.splitext(ref.staged)[0] + ".i64"): + return None # already analysed: the .i64 records how + from .formats import needs_load_options + return ref if needs_load_options(ref.source) else None + + def _ask_load_options(self, path: str, label: str | None = None) -> None: try: - size = os.path.getsize(self._open_path) + size = os.path.getsize(path) except OSError: size = 0 - self.push_screen(LoadOptionsScreen(self._open_path, size), - self._on_load_options) + self._load_for_label = label + self.push_screen(LoadOptionsScreen(path, size), self._on_load_options) def _on_load_options(self, choice) -> None: # type: ignore[no-untyped-def] from .formats import load_args choice = choice or {} - if choice.get("processor"): - self._load_args = load_args(choice["processor"], choice.get("base", 0)) - self._status(f"loading as {choice['processor']} " - f"@ {choice.get('base', 0):#x}") + label, self._load_for_label = self._load_for_label, None + proc, base = choice.get("processor", ""), int(choice.get("base", 0) or 0) + if proc: + if label is not None and self._project is not None: + # Persist it: the answer belongs to the binary, not to this run. + self._project.set_load(label, proc, base) + else: + self._load_args = load_args(proc, base) + self._status(f"loading as {proc} @ {base:#x}") + if self._pending_switch is not None: + label2, self._pending_switch = self._pending_switch, None + self._switch_binary(label2) + return self._loading_screen = LoadingScreen(self._loading_title()) self.push_screen(self._loading_screen) self._connect() @@ -3488,6 +3419,16 @@ class IdaTui(App): def _status(self, text: str) -> None: if self._binary: # project mode: always say which binary you're in text = f"[{self._binary}] {text}" + # An image with no functions at all is nearly always a blob described + # wrongly, and that stays true as you scroll around — so it belongs in + # the status bar, not in a one-off message the next write clobbers. + # It stops being true the moment a function exists, though: latching it + # meant the warning survived defining one with `p` and kept telling you + # the load was wrong when it no longer was. + if self._no_functions and self._func_index is not None and len(self._func_index): + self._no_functions = False + if self._no_functions: + text += " \u2014 no functions: wrong processor/base? Ctrl+L to reload" try: self.query_one("#status", Static).update(text) except Exception: # noqa: BLE001 -- status bar transiently unavailable @@ -3600,6 +3541,12 @@ class IdaTui(App): except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"connect failed: {e}") self.app.call_from_thread(self._dismiss_loading) + # A load we described ourselves and IDA refused: offer the dialog + # again rather than leaving an empty app with an error in the status + # bar. Getting the processor wrong is an ordinary mistake and should + # cost one more keypress, not a restart. + if "load options" in str(e): + self.app.call_from_thread(self._retry_load_options) return self.client = client self.program = program @@ -3757,8 +3704,42 @@ class IdaTui(App): fn = self._entry_func() if fn is not None: self._open_function(fn.addr, fn.name) - else: + elif len(self._func_index): self.action_symbols() + else: + self._land_without_functions() + + def _land_without_functions(self) -> None: + """Analysis found nothing. Show the bytes and say so. + + Falling through to the symbol picker here left two empty panes and + "functions still loading…" — which is a lie, loading had finished. There + is always something to look at: the segments exist even when IDA + recognised no code in them, so open the listing at the start of the image. + + Zero functions is also the signal that a blob was described wrongly. It's + exactly what a good image loaded as the wrong processor looks like, so + the status says so rather than leaving you to guess. + """ + start = None + try: + regions = self.program.file_regions() + if regions: + start = regions[0][0] + except Exception: # noqa: BLE001 + pass + self._no_functions = self._can_reload() + if start is None: + self._status("no functions and no segments \u2014 nothing to show") + return + self._open_at(start, self.program.section_of(start) or "image", + cursor=0, push=True, is_region=True) + + def _can_reload(self) -> bool: + """Whether we're able to re-open this binary with different options.""" + if self._project is not None and self._binary is not None: + return True + return bool(self._open_path) def _entry_func(self) -> Func | None: """The best startup landing function (exact-name match against @@ -3993,6 +3974,13 @@ class IdaTui(App): self._switch_binary(label) def _switch_binary(self, label: str) -> None: + # A blob nobody has described yet has to be described before its worker + # opens it — same as at boot, just reached by switching instead. + ref = self._pending_load_ref(label) + if ref is not None and self._load_for_label is None: + self._pending_switch = label + self._ask_load_options(ref.source, label=label) + return # Snapshot what we're leaving so coming back restores the view, then let # the pool hand us a worker (spawning + evicting as the budget dictates). if self._binary is not None: @@ -4134,6 +4122,16 @@ class IdaTui(App): def action_toggle_view(self) -> None: """Tab: switch the code pane between disassembly and pseudocode (or leave the hex view back to the preferred code view).""" + # Tab is a PRIORITY app binding, so it fires even while a modal is up and + # nothing inside a dialog could ever be tabbed to. Hand it back to the + # dialog: this is the only reason the load dialog's address field was + # unreachable, and it was broken the same way in every other modal. + if self.screen is not self.screen_stack[0]: + try: + self.screen.focus_next() + except Exception: # noqa: BLE001 -- screen with nothing focusable + pass + return if self._cur is None: return if self._split: @@ -4362,17 +4360,12 @@ class IdaTui(App): def on_follow_requested(self, msg: FollowRequested) -> None: view = msg.view word = view.word_under_cursor() - if isinstance(view, DisasmView): - ea = view._cursor_ea() - # The instruction's ordinary fall-through edge (to the next line) is - # an indistinguishable 'code' xref; pass it so follow can skip it and - # land on a call/jump's real target instead of the next instruction. - nxt = view.model.cached_line(view.cursor + 1) if view.model else None - if ea is not None: - self._follow_disasm(ea, word, nxt.ea if nxt else None) - elif isinstance(view, ListingView): + if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: + # _next_ea() is the following item: follow uses it to skip the + # ordinary fall-through edge, which is an indistinguishable + # 'code' xref, and land on a call/jump's real target instead. self._follow_disasm(ea, word, view._next_ea()) elif isinstance(view, DecompView) and view._texts: self._follow_decomp(view._texts[view.cursor], word, @@ -4383,17 +4376,11 @@ class IdaTui(App): return view = msg.view word = view.word_under_cursor() - if isinstance(view, DisasmView): + if isinstance(view, ListingView): ea = view._cursor_ea() if ea is not None: # span = this instruction .. the next, to pre-select the dialog # entry for the site we invoked xrefs from. - nxt = view.model.cached_line(view.cursor + 1) if view.model else None - self._push_busy("finding xrefs\u2026") - self._xrefs_disasm(ea, word, ea, nxt.ea if nxt else None) - elif isinstance(view, ListingView): - ea = view._cursor_ea() - if ea is not None: self._push_busy("finding xrefs\u2026") self._xrefs_disasm(ea, word, ea, view._next_ea()) elif isinstance(view, DecompView) and view._texts: @@ -4785,7 +4772,7 @@ class IdaTui(App): # -- comments ---------------------------------------------------------- # def _line_ea_for(self, view) -> int | None: # type: ignore[no-untyped-def] """Address of the line under the cursor in either code view.""" - if isinstance(view, (DisasmView, ListingView)): + if isinstance(view, ListingView): return view._cursor_ea() if isinstance(view, DecompView): return view._line_ea(view.cursor) @@ -4874,13 +4861,17 @@ class IdaTui(App): dec.loaded_ea = None # force re-decompile self._show_active() else: - # Capture the LIVE listing position straight from the widget (the - # source of truth) rather than trusting nav-entry tracking, which can - # go stale. bump_names() discards the segment model, so the reload - # rebuilds it; feeding the true cursor/scroll keeps the edited line on - # screen even on a huge segment (otherwise it primes/scrolls to a - # stale index and the renamed line lands off-screen — looking like the - # rename never applied). + # Capture the LIVE position from the widget (the source of truth) + # rather than trusting nav-entry tracking, which goes stale. Capture + # it as ADDRESSES via the anchor: bump_names() discards the segment + # model so the reload rebuilds it, and an edit that changes how many + # rows an item takes makes the old indices point somewhere else. + # Index capture is CORRECT here and an anchor is not: a rename or + # comment doesn't change how many rows anything takes, and the model + # this rebuilds is constructed empty — index_of_ea on it returns -1 + # until pages load, so an anchor would resolve to nothing while + # costing an extra model build on the UI thread. Address anchoring is + # for the edit paths that DO change row structure (see _do_edit_item). lst = self.query_one(ListingView) cur.view = "listing" if lst.model is not None: @@ -5033,7 +5024,8 @@ class IdaTui(App): view.focus() @work(thread=True, exclusive=True, group="makedata") - def _do_make_data(self, ea: int, type_decl: str) -> None: # worker context + def _do_make_data(self, ea: int, type_decl: str, + anchor: ViewAnchor | None = None) -> None: # worker context assert self.program is not None try: self.program.make_data(ea, type_decl) @@ -5041,12 +5033,15 @@ class IdaTui(App): self.app.call_from_thread(self._status, f"make data: {e}") return self.program.bump_items() + anchor = anchor or ViewAnchor() + anchor.flash = f"data ({type_decl}) @ {ea:#x} (Ctrl+S to save)" name = self.program.region_label(ea) lm = self.program.listing(ea) idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 - self.app.call_from_thread(self._open_at, ea, name, idx, False, -1, 0, True) + _cur, top = self._anchor_rows(anchor, lm, ea) self.app.call_from_thread( - self._edit_item_done, f"data ({type_decl})", ea) + self._open_at, ea, name, idx, False, -1, 0, True, None, top) + self.app.call_from_thread(self._edit_done, anchor) @work(thread=True, exclusive=True, group="rename") def _do_rename(self, view, old: str, new: str) -> None: # type: ignore[no-untyped-def] @@ -5155,27 +5150,78 @@ class IdaTui(App): # -- item structure edits (IDA c/p/u) --------------------------------- # def on_edit_item_requested(self, msg: EditItemRequested) -> None: - ea = (msg.view._cursor_ea() - if isinstance(msg.view, (DisasmView, ListingView)) else None) + view = msg.view + ea = view._cursor_ea() if isinstance(view, ListingView) else None if ea is None: self._status("no address on this line to (re)define") return - self._do_edit_item(msg.kind, ea) + self._do_edit_item(msg.kind, ea, self._anchor()) @work(thread=True, exclusive=True, group="edititem") - def _do_edit_item(self, kind: str, ea: int) -> None: # worker context + def _do_edit_item(self, kind: str, ea: int, + anchor: ViewAnchor | None = None) -> None: # worker context assert self.program is not None verb = {"code": "defined code", "func": "created function", - "undef": "undefined", "string": "made string"}[kind] + "undef": "undefined", "string": "made string", + "thumb": "switched decoding"}[kind] try: if kind == "code": - self.program.define_code(ea) + # Keep going until something stops it: one instruction is rarely + # what you want, and on a raw image it means pressing `c` once + # per opcode for the length of a function. + r = self.program.define_code_run(ea) + n, why = int(r.get("count", 0)), r.get("stopped", "") + if n == 0 and why == "defined": + # Already code/data here — a no-op, not a failure. Saying + # "failed to create instruction" for it would be a lie. + self.app.call_from_thread( + self._status, f"already defined @ {ea:#x}") + return + if n == 0: + raise IDAToolError("define_code", + f"@ {ea:#x}: Failed to create instruction") + end = int(str(r.get("end", hex(ea))), 0) + reason = {"undecodable": "hit bytes that don't decode", + "flow": "control flow ends here", + "defined": "ran into existing code/data", + "segment": "end of segment", + "limit": "instruction limit"}.get(why, why) + verb = (f"defined {n} instruction{'s' if n != 1 else ''} " + f"({ea:#x}\u2013{end:#x}) \u2014 {reason}") + elif kind == "thumb": + # Switch the mode, then disassemble in it: flipping T and + # leaving the bytes undefined shows nothing, and the reason you + # flipped it was to read the code. + r = self.program.set_thumb(ea) + run = self.program.define_code_run(ea) + n = int(run.get("count", 0)) + mode = "Thumb" if r.get("thumb") else "ARM" + verb = f"{mode} @ {ea:#x}" + if r.get("forced_32bit"): + verb += " (segment set to 32-bit; Thumb needs ARM32)" + if r.get("db_64bit"): + # Disassembly will look right and F5 will never work. + verb += (" \u26a0 this database is 64-bit, so Hex-Rays " + "won't decompile it \u2014 Ctrl+L and pick " + "arm:ARMv7-A") + verb += (f" \u2014 {n} instruction{'s' if n != 1 else ''}" + if n else " \u2014 still doesn't decode") + # falls through to the shared reload: same cache bump, same + # anchor restore, same flash. That is the whole point of having + # one path. elif kind == "func": - self.program.define_func(ea) + anchor.refresh_functions = True + r = self.program.define_func(ea) + if r.get("start") and r.get("end"): + verb = (f"created function {r['start']}\u2013{r['end']}" + + (" (end worked out from the code)" + if r.get("how") == "explicit-end" else "")) elif kind == "string": s = self.program.make_string(ea) verb = f"made string ({s[:24]!r})" if s else verb else: + # Undefining can destroy a function as easily as `p` creates one. + anchor.refresh_functions = True self.program.undefine(ea) except Exception as e: # noqa: BLE001 -- surface soft/hard tool errors self.app.call_from_thread(self._status, f"{kind}: {e}") @@ -5184,23 +5230,64 @@ class IdaTui(App): self.program.bump_items() # Re-resolve: a define_func upgrades the region to a real function view; # anything else re-reads the (still function-less) listing in place. + anchor = anchor or ViewAnchor() + anchor.flash = f"{verb} @ {ea:#x} (Ctrl+S to save)" fn = self.program.function_of(ea) if fn is not None: model = self.program.disasm(fn.addr, fn.name) idx = 0 if ea == fn.addr else model.index_of_ea(ea) + _cur, top = self._anchor_rows(anchor, model, ea) self.app.call_from_thread( - self._open_at, fn.addr, fn.name, idx, False, -1, 0, False) + self._open_at, fn.addr, fn.name, idx, False, -1, 0, False, + None, top) else: name = self.program.region_label(ea) lm = self.program.listing(ea) idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 + _cur, top = self._anchor_rows(anchor, lm, ea) self.app.call_from_thread( - self._open_at, ea, name, idx, False, -1, 0, True) - self.app.call_from_thread(self._edit_item_done, verb, ea) + self._open_at, ea, name, idx, False, -1, 0, True, None, top) + self.app.call_from_thread(self._edit_done, anchor) + + def _edit_done(self, anchor: ViewAnchor) -> None: + """One place where an edit's aftermath is settled. - def _edit_item_done(self, verb: str, ea: int) -> None: + The reload this edit triggered will write its own status when it lands — + after this — so the message is handed over as a flash rather than + written and lost. + """ self._dirty = True - self._status(f"{verb} @ {ea:#x} (Ctrl+S to save)") + self._flash = anchor.flash + if anchor.flash: + self._status(anchor.flash) # and again from the reload, via _flash + 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() + + @work(thread=True, exclusive=True, group="load-funcs") + def _reindex_functions(self) -> None: + """Rebuild the function index in place after an edit changed it. + + Deliberately not _load_functions(): that one is the BOOT path — it + clears the table, streams progress and then auto-lands, which would + yank the view away from the function you just made. + """ + if self.program is None: + return + idx = self.program.functions() + idx.load_all() + self._func_index = idx + self.app.call_from_thread(self._after_reindex) + + def _after_reindex(self) -> None: + idx = self._func_index + if idx is None: + return + if len(idx): + self._no_functions = False + self._apply_filter(self._filter_term) # repopulate the names pane @work(thread=True, exclusive=True, group="save") def _save(self) -> None: @@ -5390,9 +5477,54 @@ class IdaTui(App): return self._nav.append(entry) + # -- view state across a model rebuild --------------------------------- # + def _anchor(self, flash: str | None = None) -> ViewAnchor: + """Capture where we're looking, BEFORE an edit rebuilds the model. + + Must run on the UI thread: it reads live widget state. + """ + a = ViewAnchor(view=self._active, flash=flash) + view = self._active_code_view() + model = getattr(view, "model", None) + if view is None or model is None: + return a + a.cursor_x = getattr(view, "cursor_x", 0) + try: + a.ea = view._cursor_ea() + except Exception: # noqa: BLE001 + a.ea = None + top = round(view.scroll_offset.y) + h = model.cached_line(top) or model.get(top) + a.top_ea = getattr(h, "ea", None) + return a + + @staticmethod + def _anchor_rows(a: ViewAnchor, model, fallback_ea: int | None = None): + """(cursor_row, top_row) for ``a`` in a freshly built ``model``. + + -1 means "no opinion" — the caller's own default wins. + """ + if model is None: + return (-1, -1) + + def row_of(ea): + if ea is None: + return -1 + try: + i = model.index_of_ea(ea) + except Exception: # noqa: BLE001 + return -1 + return i if i >= 0 else -1 + + cur = row_of(a.ea if a.ea is not None else fallback_ea) + if cur < 0 and fallback_ea is not None: + cur = row_of(fallback_ea) + return (cur, row_of(a.top_ea)) + def _open_at(self, ea: int, name: str, cursor: int, push: bool, dec_cursor: int = -1, dec_cursor_x: int = 0, - is_region: bool = False, focus_name: str | None = None) -> None: + is_region: bool = False, focus_name: str | None = None, + scroll_y: int = -1) -> None: if push: self._save_current_pos() self._decomp_return = None # a real navigation abandons the F5 return @@ -5400,6 +5532,8 @@ class IdaTui(App): # the row is loaded, instead of column 0. self._pending_focus_name = focus_name entry = NavEntry(ea=ea, name=name, cursor=cursor, is_region=is_region) + if scroll_y >= 0: + entry.scroll_y = scroll_y if dec_cursor >= 0: entry.dec_cursor = dec_cursor entry.dec_cursor_x = dec_cursor_x @@ -5431,6 +5565,8 @@ class IdaTui(App): view.focus() def on_key(self, event) -> None: # type: ignore[no-untyped-def] + # Any keypress means the result of the last edit has been read. + self._flash = None if event.key != "escape": return if self.query_one("#search", Input).display: @@ -5515,7 +5651,7 @@ class IdaTui(App): view, ea = self._makedata_ctx self._end_makedata() if view is not None and value: - self._do_make_data(ea, value) + self._do_make_data(ea, value, self._anchor()) return if inp.id == "goto": self._end_goto() @@ -5800,6 +5936,13 @@ class IdaTui(App): self._goto_ea(msg.va, push=True) def _status_for_cur(self, mode: str) -> None: + if self._flash: + # Shown, NOT consumed. A reload emits several of these (prime, then + # cursor), so consuming on the first one meant the second erased the + # message the user was meant to read. It clears on the next keypress + # instead — i.e. when they've moved on. + self._status(self._flash) + return if self._cur is not None: self._status(f"{self._cur.name} @ {self._cur.ea:#x} [{mode}]") @@ -5807,12 +5950,22 @@ class IdaTui(App): def _load_decomp(self, ea: int, name: str) -> None: assert self.program is not None dec = self.program.decompile(ea) - self.app.call_from_thread(self._apply_decomp, ea, name, dec) + why = "" + if dec.failed: + # Ask Hex-Rays why, in the same worker: the plain tool reports + # "Decompilation failed at 0x0" and drops the only useful part. + # "Decompile failed" with no reason is indistinguishable from a bug + # in this app, and for the common cause (a 32-bit function in a + # 64-bit database) the user cannot even guess the fix. + why = self.program.decomp_error(ea) + self.app.call_from_thread(self._apply_decomp, ea, name, dec, why) - def _apply_decomp(self, ea: int, name: str, dec) -> None: # type: ignore[no-untyped-def] + def _apply_decomp(self, ea: int, name: str, dec, # type: ignore[no-untyped-def] + why: str = "") -> None: view = self.query_one(DecompView) view.loading = False if dec.failed: + detail = f" \u2014 {why}" if why else "" # No pseudocode for this function: fall back to the code view rather # than an error panel. If we came from the continuous listing (F5), # return there; otherwise show the disassembly. @@ -5823,13 +5976,20 @@ class IdaTui(App): self._decomp_return = None self._cur = ret self._active = "listing" + # Hand the reason over as a flash BEFORE reopening: going back + # to the listing reloads it, and the reload writes its own + # status afterwards — which is precisely how "F5 does nothing" + # looked like nothing at all. + msg = f"{name}: cannot decompile{detail}" + self._flash = msg self._open_entry(ret, push=False) - self._status(f"{name} — decompile failed; back to the listing") + self._status(msg) return self._active = "disasm" + msg = f"{name}: cannot decompile{detail}" + self._flash = msg self._show_active() - self._status( - f"{name} — no pseudocode (decompile failed); showing disassembly") + self._status(msg) return if self._active == "decomp": view.focus() # loading cover had blurred it; restore focus @@ -6036,6 +6196,12 @@ class IdaTui(App): return ea = msg.ea if ea is not None: + # A pending flash (the result of an edit that caused this reload) + # outranks the idle hint: the cursor lands here as part of the + # reload, so this handler would otherwise always have the last word. + if self._flash: + self._status(self._flash) + return sec = self.program.section_of(ea) if self.program else None self._status(f"{sec or '?'} @ {ea:#x} [listing] " "(c code · p func · u undefine · Enter follow)") diff --git a/idatui/domain.py b/idatui/domain.py index 681542e..a92ea49 100644 --- a/idatui/domain.py +++ b/idatui/domain.py @@ -100,6 +100,10 @@ class Head: text: str name: str | None = None raw: bytes | None = None # opcode/item bytes (filled in for code by the model) + #: [(kind, text)] from IDA's own colour tags — mnem/reg/num/name/str/punct/… + #: None when the worker didn't provide them (older worker, or the spans + #: disagreed with the plain text, in which case the text wins). + spans: tuple[tuple[str, str], ...] | None = None @property def label(self) -> str | None: # Line-compatible alias @@ -107,12 +111,15 @@ class Head: @classmethod def from_raw(cls, d: dict) -> "Head": + sp = d.get("spans") return cls( ea=_as_int(d["ea"]), kind=d.get("kind", "unknown"), size=int(d.get("size", 0) or 0), text=d.get("text", ""), name=d.get("name"), + spans=(tuple((str(k), str(t)) for k, t in sp) + if isinstance(sp, list) and sp else None), ) @@ -583,6 +590,16 @@ class ListingModel: self.name = name or f"seg @ {seg_start:#x}" self._heads: list[Head] = [] self._by_ea: dict[int, int] = {} + # Logical rows != physical heads. A run of undefined bytes arrives as ONE + # head ("db 2044 dup(?)") because materialising millions of one-byte rows + # for a .bss would be absurd — but you must still be able to put the + # cursor on any byte in it and press `c`, exactly as in IDA. So a run of + # N bytes PRESENTS as N rows and the text for each is synthesised on + # demand. _row_at[i] is the logical row where physical head i starts. + self._row_at: list[int] = [] + self._head_eas: list[int] = [] # parallel to _heads, for bisect + self._rows = 0 # total logical rows loaded + self._ubytes: dict[int, bytes] = {} # lazily-read bytes for those rows self._next: int | None = seg_start # next address to fetch from self._done = False self._max_raw = 0 # widest opcode length (bytes) seen, for the op column @@ -649,14 +666,16 @@ class ListingModel: continue page = self._attach_opcode_bytes(page) with self._lock: - base = len(self._heads) - for i, h in enumerate(page): + for h in page: # Banner/label rows (function headers, separators, code labels) # are display-only; don't index them so navigation lands on the # real code/data head at that address. if h.kind not in ("sep", "funchdr", "label"): - self._by_ea.setdefault(h.ea, base + i) + self._by_ea.setdefault(h.ea, self._rows) + self._row_at.append(self._rows) + self._head_eas.append(h.ea) self._heads.append(h) + self._rows += self._span(h) nxt = cur.get("next") if nxt is None: self._done = True @@ -665,9 +684,67 @@ class ListingModel: self._next = _as_int(nxt) return len(rows) + @staticmethod + def _span(h: Head) -> int: + """How many logical rows head ``h`` occupies.""" + return h.size if (h.kind == "unknown" and h.size > 1) else 1 + + def _phys(self, row: int) -> tuple[int, int]: + """(physical head index, byte offset into it) for logical ``row``.""" + import bisect + i = bisect.bisect_right(self._row_at, row) - 1 + if i < 0: + return (-1, 0) + return (i, row - self._row_at[i]) + + def _unknown_bytes(self, ea: int, n: int) -> bytes: + """Bytes behind an undefined run, read in blocks and cached. + + Undefined rows are the ones you carve, so their VALUES are the whole + point — "db ?" with no byte tells you nothing about where an instruction + stream might start. + """ + BLK = 1024 + out = bytearray() + a = ea + while len(out) < n: + b0 = (a // BLK) * BLK + blk = self._ubytes.get(b0) + if blk is None: + try: + blk = self._prog.read_bytes(b0, BLK) + except Exception: # noqa: BLE001 + blk = b"" + self._ubytes[b0] = blk + off = a - b0 + take = min(BLK - off, n - len(out)) + chunk = blk[off:off + take] if blk else b"" + if not chunk: + break + out += chunk + a += len(chunk) + return bytes(out) + + def _row_head(self, i: int, off: int) -> Head: + """The Head for one logical row: the physical head, or a synthesised + single-byte row inside an undefined run. + + The run's FIRST row is synthesised too. Leaving "db 2044 dup(?)" there + would say the row covers 2044 bytes when it now covers one, and the + column of byte values would start an address late. + """ + h = self._heads[i] + if self._span(h) == 1: + return h + ea = h.ea + off + b = self._unknown_bytes(ea, 1) + text = f"db {b[0]:02X}h" if b else "db ?" + return Head(ea=ea, kind="unknown", size=1, text=text, + name=h.name if off == 0 else None) + def ensure(self, n: int) -> None: - """Ensure at least ``n`` heads are loaded (or all, if fewer exist).""" - while not self._done and len(self._heads) < n: + """Ensure at least ``n`` logical rows are loaded (or all, if fewer).""" + while not self._done and self._rows < n: if self._load_next_page() == 0: break @@ -679,8 +756,9 @@ class ListingModel: if idx >= 0: return idx with self._lock: - have = len(self._heads) - last_ea = self._heads[-1].ea if self._heads else -1 + have = self._rows + last_ea = (self._heads[-1].ea + max(self._heads[-1].size, 1) - 1 + if self._heads else -1) done = self._done if done or (have and last_ea >= ea): # Loaded past ea without an exact head hit: return the first head @@ -691,12 +769,19 @@ class ListingModel: def _first_at_or_after(self, ea: int) -> int: with self._lock: - heads = self._heads - for i, h in enumerate(heads): + j = self._head_index_at(ea) + if j >= 0: + h = self._heads[j] if h.ea <= ea < h.ea + max(h.size, 1): - return i + off = (ea - h.ea) if self._span(h) > 1 else 0 + return self._row_at[j] + off + for i, h in enumerate(self._heads): + if h.ea <= ea < h.ea + max(h.size, 1): + # Inside an undefined run, land on the exact BYTE. + off = (ea - h.ea) if self._span(h) > 1 else 0 + return self._row_at[i] + off if h.ea > ea: - return i + return self._row_at[i] return -1 def load_all(self, progress: Callable[[int], None] | None = None) -> None: @@ -704,7 +789,7 @@ class ListingModel: if self._load_next_page() == 0: break if progress: - progress(len(self._heads)) + progress(self._rows) @property def complete(self) -> bool: @@ -713,23 +798,58 @@ class ListingModel: def loaded(self) -> int: with self._lock: - return len(self._heads) + return self._rows def __len__(self) -> int: return self.loaded() def get(self, i: int) -> Head | None: with self._lock: - return self._heads[i] if 0 <= i < len(self._heads) else None + if not (0 <= i < self._rows): + return None + j, off = self._phys(i) + if j < 0: + return None + span = self._span(self._heads[j]) + h = self._heads[j] + # Synthesis reads bytes, so do it OUTSIDE the lock: an RPC under the + # model lock deadlocks the page loader that is filling it. + return self._row_head(j, off) if span > 1 else h def window(self, start: int, count: int) -> list[Head]: + """``count`` logical rows from ``start`` (synthesising undefined ones).""" self.ensure(start + count) with self._lock: - return list(self._heads[start:start + count]) + rows = min(self._rows, start + count) + spans = [self._phys(i) for i in range(max(start, 0), max(rows, 0))] + heads = self._heads + plain = [(j, off, heads[j]) for j, off in spans if j >= 0] + return [self._row_head(j, off) if self._span(h) > 1 else h + for j, off, h in plain] def index_of_ea(self, ea: int) -> int: with self._lock: - return self._by_ea.get(ea, -1) + hit = self._by_ea.get(ea) + if hit is not None: + return hit + # An address INSIDE an undefined run is a real row now, not a + # mid-item address: that is what makes `g <addr>` + `c` work + # anywhere in a blob. Heads are address-ordered, so bisect rather + # than scan — a big listing has hundreds of thousands of them and + # this is on the navigation path. + j = self._head_index_at(ea) + if j >= 0: + h = self._heads[j] + if self._span(h) > 1 and h.ea <= ea < h.ea + h.size: + return self._row_at[j] + (ea - h.ea) + return -1 + + def _head_index_at(self, ea: int) -> int: + """Index of the physical head containing ``ea`` (caller holds the lock).""" + import bisect + eas = self._head_eas + i = bisect.bisect_right(eas, ea) - 1 + return i if 0 <= i < len(self._heads) else -1 # -- DisasmModel-compatible accessors (unified model) ------------------ # def cached_line(self, idx: int) -> Head | None: @@ -741,7 +861,7 @@ class ListingModel: def is_cached(self, start: int, count: int) -> bool: with self._lock: - return start + count <= len(self._heads) + return start + count <= self._rows def ensure_async(self, start: int, count: int) -> None: pass # the background grower streams the rest in; nothing to prefetch @@ -1251,14 +1371,64 @@ class Program: res = self._first_result( self.client.call("define_code", items=[{"addr": hex(ea)}])) if res.get("error"): - raise IDAToolError(f"define code @ {ea:#x}: {res['error']}") + raise IDAToolError("define_code", f"@ {ea:#x}: {res['error']}") - def define_func(self, ea: int) -> None: - """Create a function starting at ``ea`` (IDA's 'p').""" - res = self._first_result( - self.client.call("define_func", items=[{"addr": hex(ea)}])) - if res.get("error"): - raise IDAToolError(f"create function @ {ea:#x}: {res['error']}") + def decomp_error(self, ea: int) -> str: + """Hex-Rays' own reason for refusing ``ea``, or "" if it won't say.""" + try: + r = self.client.call("decomp_error", addr=hex(ea)) + except IDAToolError: + return "" + if not isinstance(r, dict): + return "" + reason = str(r.get("reason") or "") + if reason and r.get("bitness") == 64 and "64-bit" in reason: + # Unfixable in place: the database's bitness is decided at load. + reason += " \u2014 Ctrl+L and pick arm:ARMv7-A" + return reason + + def set_thumb(self, ea: int, mode: str = "toggle") -> dict: + """Switch ARM/Thumb decoding at ``ea``. Returns the resulting state.""" + r = self.client.call("set_thumb", addr=hex(ea), mode=mode) + if not isinstance(r, dict) or r.get("error"): + raise IDAToolError("set_thumb", + f"@ {ea:#x}: {(r or {}).get('error', 'failed')}") + return r + + def define_code_run(self, ea: int, limit: int = 20000) -> dict: + """Disassemble consecutively from ``ea`` until something stops it. + + Falls back to a single instruction when the worker predates the tool, so + an old worker degrades to the previous behaviour instead of failing. + """ + try: + r = self.client.call("define_code_run", addr=hex(ea), limit=int(limit)) + except IDAToolError: + self.define_code(ea) + return {"count": 1, "stopped": "single", "end": hex(ea)} + if not isinstance(r, dict) or r.get("error"): + raise IDAToolError("define_code_run", + f"@ {ea:#x}: {(r or {}).get('error', 'failed')}") + return r + + def define_func(self, ea: int) -> dict: + """Create a function starting at ``ea`` (IDA's 'p'). + + Prefers the injected tool, which works out the end when IDA can't; + falls back to the plain one for an older worker. + """ + try: + r = self.client.call("define_func_run", addr=hex(ea)) + except IDAToolError: + res = self._first_result( + self.client.call("define_func", items=[{"addr": hex(ea)}])) + if res.get("error"): + raise IDAToolError("define_func", f"@ {ea:#x}: {res['error']}") + return {"ok": True, "how": "legacy"} + if not isinstance(r, dict) or not r.get("ok"): + raise IDAToolError("define_func", + f"@ {ea:#x}: {(r or {}).get('error', 'failed')}") + return r def undefine(self, ea: int, size: int | None = None) -> None: """Undefine the item at ``ea`` back to raw bytes (IDA's 'u').""" @@ -1267,7 +1437,7 @@ class Program: item["size"] = int(size) res = self._first_result(self.client.call("undefine", items=[item])) if res.get("error"): - raise IDAToolError(f"undefine @ {ea:#x}: {res['error']}") + raise IDAToolError("undefine", f"@ {ea:#x}: {res['error']}") def make_data(self, ea: int, type_decl: str, name: str | None = None) -> None: """Create a typed data item at ``ea`` (IDA's 'd', but typed). ``type_decl`` @@ -1278,7 +1448,7 @@ class Program: res = self._first_result(self.client.call("make_data", items=[item])) if res.get("ok") is False or res.get("error"): raise IDAToolError( - f"make data @ {ea:#x}: {res.get('error') or 'rejected'}") + "make_data", f"@ {ea:#x}: {res.get('error') or 'rejected'}") def make_string(self, ea: int, length: int = 0, kind: str = "c") -> str: """Create a string literal at ``ea`` (IDA's 'A'); auto-length when 0. @@ -1287,7 +1457,7 @@ class Program: res = r if isinstance(r, dict) else {} if not res.get("ok"): raise IDAToolError( - f"make string @ {ea:#x}: {res.get('error') or 'rejected'}") + "make_string", f"@ {ea:#x}: {res.get('error') or 'rejected'}") return res.get("text", "") def region_label(self, ea: int) -> str: diff --git a/idatui/formats.py b/idatui/formats.py index b2f784d..f09bb99 100644 --- a/idatui/formats.py +++ b/idatui/formats.py @@ -92,9 +92,18 @@ def needs_load_options(path: str) -> bool: #: reach for first — arm64, aarch64, mips, m68k — are all invalid. Re-run the #: script before adding to this list. PROCESSORS: tuple[tuple[str, str], ...] = ( - # 'arm' covers AArch64 too; arm64/aarch64 are NOT valid -p names, so they - # live in the label where the filter can still find them. - ("arm", "ARM / AArch64 / arm64 — little-endian"), + # Bare 'arm' gives a 64-BIT database in IDA 9 (AArch64). That matters far + # more than it looks: Hex-Rays refuses a 32-bit function in a 64-bit database + # ("only 64-bit functions can be decompiled in the current database"), and + # Thumb doesn't exist in AArch64 at all — so a 32-bit ARM image loaded as + # plain 'arm' disassembles wrongly and can never be decompiled. The database + # bitness is fixed at load; it cannot be corrected afterwards (setting it + # post-hoc makes the decompiler INTERR). Pick the right one here. + ("arm", "ARM64 / AArch64 / arm64 — 64-bit, little-endian"), + ("arm:ARMv7-A", "ARM 32-bit (ARMv7-A) — Thumb capable, most firmware"), + ("arm:ARMv7-M", "ARM 32-bit (ARMv7-M) — Cortex-M, Thumb only"), + ("arm:ARMv6-M", "ARM 32-bit (ARMv6-M) — Cortex-M0/M0+"), + ("arm:ARMv5TE", "ARM 32-bit (ARMv5TE) — older SoCs"), ("armb", "ARM — big-endian"), ("metapc", "x86 / x86-64"), ("mipsl", "MIPS — little-endian"), diff --git a/idatui/project.py b/idatui/project.py index 8f2674d..e2fc542 100644 --- a/idatui/project.py +++ b/idatui/project.py @@ -239,6 +239,29 @@ class Project: def refs(self) -> tuple[BinaryRef, ...]: return self._refs + def set_load(self, label: str, processor: str = "", base: int = 0, + ida_args: str = "") -> BinaryRef | None: + """Record how ``label`` should be loaded, and persist it. + + Answered once: the dialog that asks writes the answer here, so reopening + the project doesn't ask again — and neither does adding the same blob to + another project, since it travels with the entry. + """ + ref = self.by_label(label) + if ref is None: + return None + i = self._refs.index(ref) + e = self._entries[i] + if processor: + e["processor"] = processor + if base: + e["base"] = int(base) + if ida_args: + e["ida_args"] = ida_args + self._refs = self._build_refs() + self.save() + return self._refs[i] + def by_label(self, label: str) -> BinaryRef | None: return next((r for r in self._refs if r.label == label), None) diff --git a/idatui/worker.py b/idatui/worker.py index 26f0816..a4e3509 100644 --- a/idatui/worker.py +++ b/idatui/worker.py @@ -111,6 +111,15 @@ def _open_and_register(binpath: str, load_args: str = ""): args = None if idapro.open_database(binpath, run_auto_analysis=True, args=args): # nonzero == failure + if args: + # With load switches in play they are the likeliest culprit by far: + # IDA refuses an unknown -p name with no diagnostic of its own, so + # saying "the database is locked" here sends people hunting a + # problem they don't have. + raise RuntimeError( + f"failed to open {binpath} with load options {args!r}: IDA " + f"rejected them \u2014 an unknown processor name is the usual " + f"cause (see tools/verify_procs.py for the valid ones)") raise RuntimeError( f"failed to open {binpath}: the .i64 is likely held by a running " f"ida-mcp worker (try: pkill -f idalib) or wedged from a crash " |