diff options
Diffstat (limited to 'idatui')
| -rw-r--r-- | idatui/app.py | 896 | ||||
| -rw-r--r-- | idatui/domain.py | 231 | ||||
| -rw-r--r-- | idatui/drive.py | 26 | ||||
| -rw-r--r-- | idatui/formats.py | 134 | ||||
| -rw-r--r-- | idatui/highlight.py | 31 | ||||
| -rw-r--r-- | idatui/index.py | 34 | ||||
| -rw-r--r-- | idatui/launch.py | 57 | ||||
| -rw-r--r-- | idatui/pane.py | 32 | ||||
| -rw-r--r-- | idatui/pool.py | 33 | ||||
| -rw-r--r-- | idatui/project.py | 111 | ||||
| -rw-r--r-- | idatui/rpc.py | 42 | ||||
| -rw-r--r-- | idatui/worker.py | 50 | ||||
| -rw-r--r-- | idatui/worker_client.py | 8 |
13 files changed, 1558 insertions, 127 deletions
diff --git a/idatui/app.py b/idatui/app.py index e96483b..dd7c222 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -105,7 +105,6 @@ class BinaryState: func_index: object | None = None nav: list = field(default_factory=list) cur: object | None = None - pref: str = "listing" active: str = "listing" split: bool = False filter_term: str = "" @@ -113,6 +112,29 @@ 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 + + +@dataclass class NavEntry: ea: int name: str @@ -784,13 +806,24 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True return texts: list[str] = [] off, total = 0, self.total + # A freshly-loaded blob is one row per undefined byte, so "all lines" can + # be millions of `db 4Ah` that nobody searches for. Index a bounded + # prefix rather than hang; say so instead of silently finding nothing. + LIMIT = 400_000 + capped = total > LIMIT + total = min(total, LIMIT) while off < total: - lines = model.lines(off, DisasmModel.BLOCK, prefetch=False) + lines = model.lines(off, min(DisasmModel.BLOCK, total - off), + prefetch=False) if not lines: break texts.extend(self._fmt(ln) for ln in lines) off += len(lines) self._search_texts = texts + if capped: + self.app.call_from_thread( + self._app_status, + f"search covers the first {LIMIT:,} lines of {self.total:,}") self.app.call_from_thread(done) @work(thread=True, exclusive=True, group="disasm-prime") @@ -1574,6 +1607,13 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True else: scroll_x = sx self._apply_scroll(min(max(scroll_y, 0), max(total - 1, 0)), max(scroll_x, 0)) + # `cursor` is reactive(repaint=False) and _apply_scroll only repaints via + # call_after_refresh, so a jump that lands in the SAME viewport (Esc back + # to another spot in the function already on screen) moved the cursor + # with nothing to redraw it — the pane kept showing the old highlight + # until the next keypress. Repaint here; _move does the same via + # _refresh_lines. + self.refresh() self._after_cursor_move() def get_loading_widget(self): # type: ignore[override] @@ -2002,8 +2042,10 @@ class XrefsScreen(ModalScreen): BINDINGS = [Binding("escape", "close", "Close")] - def __init__(self, label: str, items: list[tuple[int, str]], + def __init__(self, label: str, items: list[tuple[object, str]], preselect: int = 0) -> None: + # payload is an int address, or (binary, address) for a caller in another + # project binary; dismiss() hands it back untouched. super().__init__() self._label = label self._items = items @@ -2041,9 +2083,15 @@ def _fuzzy(name: str, q: str): if not name: # defensive: never assume a symbol has a name return None nl = name.lower() + # Case-insensitive means BOTH sides: the name was lowered but the query + # wasn't, so a single capital could never match and any query containing one + # returned nothing at all. Invisible on lowercase C symbols (main, strlen), + # fatal on libraries that capitalise — "PEM_read_bio" found 0 of 10093 + # functions in libcrypto while the backend resolved it fine. + ql = q.lower() pos: list[int] = [] i = 0 - for ch in q: + for ch in ql: j = nl.find(ch, i) if j < 0: return None @@ -2051,9 +2099,9 @@ def _fuzzy(name: str, q: str): i = j + 1 span = pos[-1] - pos[0] score = -(span * 2.0) - pos[0] - len(name) * 0.01 - if q in nl: + if ql in nl: score += 50.0 - if nl.startswith(q): + if nl.startswith(ql): score += 100.0 return (score, tuple(pos)) @@ -2503,6 +2551,145 @@ class HelpScreen(ModalScreen): self.dismiss(None) +class LoadOptionsScreen(ModalScreen): + """Ask how to load a file no loader recognised. + + IDA's own answer to an unidentified file is a dialog; ours is this. Without + it the fallback is x86 at address 0, which doesn't fail — it analyses to + nothing, and you're left wondering why a firmware image has no functions. + + Returns ``{"processor": str, "base": int}``, or ``{}`` to load it the way + IDA would have anyway (that IS the right answer sometimes: our sniff only + recognises formats we're sure about, so it says "unknown" for things IDA + can in fact handle). + """ + + BINDINGS = [ + Binding("escape", "close", "Close"), + Binding("down,ctrl+n", "cursor_down", show=False), + Binding("up,ctrl+p", "cursor_up", show=False), + Binding("enter", "choose", show=False, priority=True), + ] + + def __init__(self, path: str, size: int = 0) -> None: + super().__init__() + self._path = path + # NOT self._size: that is Textual's own backing field for outer_size, and + # assigning an int to it crashes the layout with a bewildering + # "'int' object has no attribute 'region'" from deep inside _set_dirty. + self._nbytes = size + self._results: list[tuple[str, str]] = [] + + def compose(self) -> ComposeResult: + from .formats import PROCESSORS + self._all = list(PROCESSORS) + with Vertical(id="pal-box"): + yield Static(" unrecognised file \u2014 how should IDA load it?", + id="pal-title", markup=False) + yield Static(f" {os.path.basename(self._path)} ({self._nbytes:,} bytes) " + f"\u2014 no loader matched; without a processor IDA " + f"assumes x86 at 0", id="load-note", markup=False) + yield Input(placeholder="filter processors\u2026", id="pal-input") + yield OptionList(id="pal-list") + yield Input(placeholder="load address, e.g. 0x8000000 (blank = 0)", + id="load-base") + yield Static(" Enter accept \u00b7 Tab base address \u00b7 " + "Esc load as IDA would", id="load-help", markup=False) + + def on_mount(self) -> None: + self._apply("") + self.query_one("#pal-input", Input).focus() + + def focus_next(self, selector="*"): # type: ignore[override] + """Tab moves between the two things you TYPE into. + + DOM order would stop at the option list on the way, which is + arrow-driven and has nothing to type — and the address you meant to + enter goes into whichever box happened to have focus. Typing an address + into the processor filter is then taken as a processor name, IDA rejects + it, and the open fails; that is a bad enough outcome to be worth + overriding Tab for. + """ + inp = self.query_one("#pal-input", Input) + base = self.query_one("#load-base", Input) + (inp if self.focused is base else base).focus() + return self.focused + + def focus_previous(self, selector="*"): # type: ignore[override] + return self.focus_next() + + def on_input_changed(self, event: Input.Changed) -> None: + event.stop() + if event.input.id == "pal-input": + self._apply(event.value.strip()) + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self.action_choose() + + def _apply(self, query: str) -> None: + q = query.lower() + rows = [(name, desc) for name, desc in self._all + if not q or q in name.lower() or q in desc.lower()] + # An unlisted processor is still valid: IDA has 73 modules and this + # offers 20, so a typed name that matches nothing is taken literally + # rather than refused. + if not rows and q: + rows = [(query.strip(), "use this processor name as typed")] + self._results = rows + ol = self.query_one(OptionList) + ol.clear_options() + opts = [] + for name, desc in rows: + label = Text() + label.append(f" {name:<12}", _S_LABEL) + label.append(desc, _S_DIM) + opts.append(Option(label)) + ol.add_options(opts) + if rows: + ol.highlighted = 0 + self.query_one("#pal-title", Static).update( + f" unrecognised file \u2014 processor? ({len(rows)})") + + def action_cursor_down(self) -> None: + ol = self.query_one(OptionList) + if ol.option_count: + ol.highlighted = min((ol.highlighted or 0) + 1, ol.option_count - 1) + + def action_cursor_up(self) -> None: + ol = self.query_one(OptionList) + if ol.option_count: + ol.highlighted = max((ol.highlighted or 0) - 1, 0) + + def action_choose(self) -> None: + ol = self.query_one(OptionList) + i = ol.highlighted + if i is None or not (0 <= i < len(self._results)): + self.dismiss({}) + return + raw = self.query_one("#load-base", Input).value.strip() + base = 0 + if raw: + try: + base = int(raw, 0) + except ValueError: + self.query_one("#load-help", Static).update( + f" {raw!r} is not an address \u2014 try 0x8000000") + self.query_one("#load-base", Input).focus() + return + if base % 16: + # IDA's -b is in paragraphs, so an unaligned base can't be + # expressed and would quietly load somewhere else. + self.query_one("#load-help", Static).update( + f" {base:#x} must be 16-byte aligned") + self.query_one("#load-base", Input).focus() + return + self.dismiss({"processor": self._results[i][0], "base": base}) + + def action_close(self) -> None: + self.dismiss({}) + + class ProjectPalette(ModalScreen): """The project's binaries; Enter switches to one. Shows which are resident (a live worker, so switching is instant) vs cold (needs an open).""" @@ -2604,13 +2791,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: @@ -3091,7 +3281,8 @@ class IdaTui(App): #xref-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } #xref-list { height: auto; max-height: 100%; } /* every #pal-box palette centres, not just the symbol one */ - SymbolPalette, StringsPalette, ProjectPalette { align: center middle; } + SymbolPalette, StringsPalette, ProjectPalette, + LoadOptionsScreen { align: center middle; } /* Give the stock Ctrl+P command palette side padding instead of full width; the input + results inherit this width (results is an overlay, so pin it). */ CommandPalette > Vertical { width: 80%; max-width: 120; } @@ -3101,6 +3292,14 @@ class IdaTui(App): #pal-title { dock: top; height: 1; background: $accent; color: $background; text-style: bold; padding: 0 1; } #pal-input { border: none; height: 1; margin: 0 1; background: $panel; color: $text; } #pal-list { height: auto; max-height: 24; } + #load-note { height: 2; padding: 1 1 0 1; color: $text-muted; } + /* Cap the processor list so the ADDRESS FIELD is always on screen: with the + palette default (24) the box outgrew the terminal and the field you need + was clipped off the bottom, which read as "Tab does nothing". */ + LoadOptionsScreen #pal-list { max-height: 12; } + #load-base { border: none; height: 1; margin: 1 1 0 1; background: $panel; color: $text; } + #load-help { height: 1; padding: 0 1; color: $text-muted; } + #confirm-note { height: auto; padding: 0 1; color: $text-muted; } StructEditor { align: center middle; } #se-box { width: 90%; height: 84%; border: thick $accent; background: $panel; } #se-panes { height: 1fr; } @@ -3137,6 +3336,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), @@ -3148,7 +3348,7 @@ class IdaTui(App): def __init__(self, open_path: str | None = None, keepalive: bool = True, rpc_path: str | None = None, ttl: int = 1800, - project=None) -> None: + project=None, load_args: str = "") -> None: super().__init__() # Project mode is additive: with no project this is the plain # single-binary app, unchanged. @@ -3158,6 +3358,12 @@ class IdaTui(App): self._states: dict[str, BinaryState] = {} self._pending_restore = None # entry to reopen after a switch self._goto_after_switch = None # cross-binary search hit to land on + self._hops: list[str] = [] # binaries a navigation crossed FROM + self._load_for_label = None # project binary the dialog is for + self._no_functions = False # analysis produced nothing at all + self._flash: str | None = None # message a pending reload must keep + self._pending_switch = None # switch waiting on that answer + self._nav_seq = 0 # bumped per navigation; drops stale ones # None = teardown wasn't an explicit quit (crash/kill): save defensively. # False = the user chose discard, or we already saved on the way out. self._save_on_exit: bool | None = None @@ -3172,6 +3378,7 @@ class IdaTui(App): open_path = project.refs[0].staged self._open_path = open_path self._ttl = ttl + self._load_args = load_args or "" # IDA switches for a headerless blob self._do_keepalive = keepalive self._rpc_path = rpc_path self._rpc = None @@ -3187,7 +3394,9 @@ class IdaTui(App): self._filter_timer = None self._sort_col = 0 # 0=addr, 1=name, 2=size self._sort_reverse = False - self._pref = "listing" # unified: the linear listing is the code view + # ONE notion of "which pane you're in": _active, kept in step with focus + # (on_descendant_focus does that while split). There used to be a second, + # _pref, but it was only ever assigned "listing" — see _code_mode(). self._active = "listing" # currently shown view (in split: the focused pane) self._split = False # side-by-side listing + pseudocode self._split_eamap: list[list[int]] = [] # split: decomp line -> instr EAs @@ -3263,13 +3472,191 @@ class IdaTui(App): # The names pane is an overlay now (Ctrl+N); focus the default code view # (pseudocode) so app bindings work before anything is opened. self.query_one(ListingView).focus() + if self._rpc_path: + self._start_rpc() + # A file no loader recognises has to be described before it can be + # opened, so ask BEFORE the worker starts — once IDA has made a database + # the answer is baked in and changing it means deleting the .i64. + if self._project is not None: + ref = self._pending_load_ref() + if ref is not None: + self._ask_load_options(ref.source, label=ref.label) + return + elif self._should_ask_load_options(): + self._ask_load_options(self._open_path) + return # Show a loading overlay immediately so a slow open/analysis (big binary) # isn't just dead air behind empty panes; dismissed once we land. self._loading_screen = LoadingScreen(self._loading_title()) self.push_screen(self._loading_screen) self._connect() - if self._rpc_path: - self._start_rpc() + + def _should_ask_load_options(self) -> bool: + """Ask only when nobody has already answered, and only when it matters. + + Skipped when: options came from the command line or the project (the + user already said); a database exists (the answer is recorded in it, and + re-passing switches fails the open); or the file is a format IDA + recognises, which is nearly always. + """ + if not self._open_path: + return False + if self._load_args: + return False + from .formats import needs_load_options + if os.path.exists(self._open_path + ".i64") or os.path.exists( + os.path.splitext(self._open_path)[0] + ".i64"): + return False + return needs_load_options(self._open_path) + + def action_load_options(self) -> None: + """Ctrl+L: re-open this binary with different load options. + + The database IDA already built has the old processor and base baked into + it and takes precedence over any switches, so re-loading means throwing + it away. That destroys names and comments, hence the confirmation — but + for the case this exists for (a blob loaded as the wrong architecture, + which analysed to nothing) there is nothing to lose and no other way + forward. + """ + if not self._can_reload(): + self._status("nothing to reload") + return + n = len(self._func_index) if self._func_index else 0 + note = ("this image has no functions, so nothing is lost" + if n == 0 else + f"discards the database for this binary \u2014 {n} functions, " + f"plus any names and comments you've added") + self.push_screen(ConfirmScreen("Reload with different options?", note), + self._on_reload_confirmed) + + def _on_reload_confirmed(self, yes) -> None: # type: ignore[no-untyped-def] + if not yes: + return + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Drop the worker first: it holds the database open, and the .i64 can't + # be removed (or rebuilt) underneath a live one. + self._release_worker() + self._drop_database() + self._reset_for_reload() + self._load_args = "" + if label is not None and self._project is not None: + self._project.set_load(label, processor="", base=0) + i = self._project._refs.index(self._project.by_label(label)) + self._project._entries[i].pop("processor", None) + self._project._entries[i].pop("base", None) + self._project.save() + self._pending_switch = None + self._ask_load_options(path, label=label) + + def _release_worker(self) -> None: + if self._pool is not None and self._binary is not None: + try: + self._pool.evict(self._binary, save=False) + except Exception: # noqa: BLE001 + pass + elif self.client is not None: + try: + self.client.close() + except Exception: # noqa: BLE001 + pass + self.client = None + self.program = None + + def _drop_database(self) -> None: + """Remove the .i64 (and any unpacked scratch) so the next open re-reads + the raw image with new options.""" + base = self._open_path + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + base = ref.staged + if not base: + return + for suffix in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"): + for cand in (base + suffix, os.path.splitext(base)[0] + suffix): + try: + os.remove(cand) + except OSError: + pass + + def _reset_for_reload(self) -> None: + self._no_functions = False + self._func_index = None + self._cur = None + self._nav = [] + self._did_auto_land = False + self._pending_restore = None + self._split = False + self.query_one(DecompView).loaded_ea = None + + def _retry_load_options(self) -> None: + """Re-ask after IDA refused what we told it.""" + path, label = self._open_path, None + if self._project is not None and self._binary is not None: + ref = self._project.by_label(self._binary) + if ref is not None: + path, label = ref.source, ref.label + # Clear the rejected answer or _pending_load_ref would see the + # binary as already described and never ask again. + self._project.set_load(label, processor="", base=0) + self._project._entries[self._project._refs.index(ref)].pop( + "processor", None) + self._project.save() + self._load_args = "" + if path: + self._status("those load options were rejected \u2014 try again") + self._ask_load_options(path, label=label) + + def _pending_load_ref(self, label: str | None = None): # type: ignore[no-untyped-def] + """The project binary about to be opened, if it needs describing. + + Checked against the SOURCE: staging may not have happened yet, and the + question is about the bytes, not where they were copied to. + """ + if self._project is None: + return None + label = label or self._binary or self._project.refs[0].label + ref = self._project.by_label(label) + if ref is None or ref.load_args: + return None + if os.path.exists(ref.db) or os.path.exists( + os.path.splitext(ref.staged)[0] + ".i64"): + return None # already analysed: the .i64 records how + from .formats import needs_load_options + return ref if needs_load_options(ref.source) else None + + def _ask_load_options(self, path: str, label: str | None = None) -> None: + try: + size = os.path.getsize(path) + except OSError: + size = 0 + self._load_for_label = label + self.push_screen(LoadOptionsScreen(path, size), self._on_load_options) + + def _on_load_options(self, choice) -> None: # type: ignore[no-untyped-def] + from .formats import load_args + choice = choice or {} + label, self._load_for_label = self._load_for_label, None + proc, base = choice.get("processor", ""), int(choice.get("base", 0) or 0) + if proc: + if label is not None and self._project is not None: + # Persist it: the answer belongs to the binary, not to this run. + self._project.set_load(label, proc, base) + else: + self._load_args = load_args(proc, base) + self._status(f"loading as {proc} @ {base:#x}") + if self._pending_switch is not None: + label2, self._pending_switch = self._pending_switch, None + self._switch_binary(label2) + return + self._loading_screen = LoadingScreen(self._loading_title()) + self.push_screen(self._loading_screen) + self._connect() def _loading_title(self) -> str: return os.path.basename(self._open_path) if self._open_path else "database" @@ -3301,6 +3688,11 @@ 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. + if self._no_functions: + text += " \u2014 no functions: wrong processor/base? Ctrl+L to reload" try: self.query_one("#status", Static).update(text) except Exception: # noqa: BLE001 -- status bar transiently unavailable @@ -3372,7 +3764,8 @@ class IdaTui(App): self.app.call_from_thread(self._reconnect_failed, "no binary to reopen") return - client = WorkerClient(self._open_path, ttl=self._ttl) + client = WorkerClient(self._open_path, ttl=self._ttl, + load_args=self._load_args) client.connect(progress=lambda m: self.app.call_from_thread( self._conn_note, m)) except Exception as e: # noqa: BLE001 @@ -3412,6 +3805,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 @@ -3438,7 +3837,8 @@ class IdaTui(App): base = os.path.basename(self._open_path) self.app.call_from_thread( self._status, f"starting worker — initial auto-analysis of {base}…") - client = WorkerClient(self._open_path, ttl=self._ttl) + client = WorkerClient(self._open_path, ttl=self._ttl, + load_args=self._load_args) client.connect(progress=lambda m: self.app.call_from_thread( self._status, m)) return client @@ -3471,6 +3871,41 @@ class IdaTui(App): self.app.call_from_thread(self._auto_land) self._index_binary() # project mode: keep the cross-binary index fresh + @work(thread=True, exclusive=True, group="prewarm") + def _prewarm_provider(self) -> None: + """Warm the binary this one leans on hardest, once we're idle. + + Not "the next in the list" — phase 3 tells us something better. The + binary providing the most of this one's imports is where a follow is + most likely to take you, so paying its startup now is the switch you + would otherwise wait for. Refuses to evict anything (see pool.prewarm), + so at a tight budget this simply does nothing. + """ + if self._pool is None or self._index is None or self._binary is None: + return + try: + imps, _ = self.program.linkage() + except Exception: # noqa: BLE001 + return + if not imps: + return + from collections import Counter + votes: Counter = Counter() + for name in {i.name for i in imps}: + for h in self._index.providers(name, exclude=self._binary): + votes[h.binary] += 1 + resident = set(self._pool.resident()) + cand = next((b for b, _ in votes.most_common() if b not in resident), None) + if cand is None: + return + n = votes[cand] + try: + if self._pool.prewarm(cand): + self.app.call_from_thread( + self._status, f"pre-warmed {cand} (provides {n} imports)") + except Exception: # noqa: BLE001 -- speculative work must never surface + pass + @work(thread=True, exclusive=True, group="index") def _index_binary(self) -> None: """Fold this binary's symbols + strings into the project index, so it can @@ -3480,7 +3915,7 @@ class IdaTui(App): ref = self._project.by_label(self._binary) if ref is None or not self._index.is_stale(self._binary, ref.source): return - from .index import KIND_FUNC, KIND_STRING + from .index import KIND_EXPORT, KIND_FUNC, KIND_IMPORT, KIND_STRING idx = self._func_index entries = [(KIND_FUNC, f.addr, f.name) for f in (idx.all_loaded() if idx else [])] try: @@ -3489,12 +3924,19 @@ class IdaTui(App): except Exception: # noqa: BLE001 -- symbols alone are still worth indexing pass try: + imps, exps = self.program.linkage() + entries += [(KIND_IMPORT, i.addr, i.name) for i in imps] + entries += [(KIND_EXPORT, e.addr, e.name) for e in exps] + except Exception: # noqa: BLE001 -- an old worker has no list_linkage + pass + try: n = self._index.reindex(self._binary, entries, source=ref.source) except Exception as e: # noqa: BLE001 self.app.call_from_thread(self._status, f"indexing failed: {e}") return self.app.call_from_thread( - self._status, f"indexed {self._binary}: {n} symbols + strings") + self._status, f"indexed {self._binary}: {n} symbols, strings + linkage") + self._prewarm_provider() # -- initial landing --------------------------------------------------- # #: function names tried (in order) as the startup landing spot @@ -3526,8 +3968,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 @@ -3674,7 +4150,15 @@ class IdaTui(App): def _switch_then_goto(self, binary: str, addr: int) -> None: """A project-wide hit in another binary: switch to it, then jump there - once its index has loaded.""" + once its index has loaded. + + Records the hop so Esc can come back. Nav history is per-binary, so + without this a cross-binary jump — a project search hit, or following an + import into the library that implements it — is a one-way door: you + arrive in a binary whose history is empty and nothing takes you back. + """ + if self._binary is not None and binary != self._binary: + self._hops.append(self._binary) self._goto_after_switch = addr self._switch_binary(binary) @@ -3754,13 +4238,20 @@ class IdaTui(App): self._switch_binary(label) def _switch_binary(self, label: str) -> None: + # A blob nobody has described yet has to be described before its worker + # opens it — same as at boot, just reached by switching instead. + ref = self._pending_load_ref(label) + if ref is not None and self._load_for_label is None: + self._pending_switch = label + self._ask_load_options(ref.source, label=label) + return # Snapshot what we're leaving so coming back restores the view, then let # the pool hand us a worker (spawning + evicting as the budget dictates). if self._binary is not None: self._states[self._binary] = BinaryState( label=self._binary, program=self.program, func_index=self._func_index, nav=list(self._nav), cur=self._cur, - pref=self._pref, active=self._active, split=self._split, + active=self._active, split=self._split, filter_term=self._filter_term, dirty=self._dirty) self._loading_screen = LoadingScreen(label, note="switching\u2026") self.push_screen(self._loading_screen) @@ -3791,7 +4282,6 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged - self._pref = st.pref if st else "listing" self._active = st.active if st else "listing" self._split = st.split if st else False self._filter_term = st.filter_term if st else "" @@ -3871,9 +4361,41 @@ class IdaTui(App): return self._goto_ea(addr, push=True) # land on the literal in the listing + def on_descendant_focus(self, event) -> None: # type: ignore[no-untyped-def] + """Keep ``_active`` in step with focus while split. + + Tab moves both together, but focus also moves on its own — a click, or a + pane focusing itself after a load — and then ``_active`` still names the + pane you're NOT in. Everything downstream trusts ``_active``: follow + resolves the word under that pane's cursor and pushes history for it, so + Enter in the pseudocode would follow something from the listing and the + next Esc got spent undoing it. + """ + if not self._split: + return + w = self.focused + mode = ("decomp" if isinstance(w, DecompView) + else "listing" if isinstance(w, ListingView) else None) + if mode is None or mode == self._active: + return + self._active = mode + self._sync_split(mode) # re-link the band from the new driver + if not self.query_one(DecompView).loading: + self._status_for_cur("split") # never clobber "decompiling…" + def action_toggle_view(self) -> None: """Tab: switch the code pane between disassembly and pseudocode (or leave the hex view back to the preferred code view).""" + # Tab is a PRIORITY app binding, so it fires even while a modal is up and + # nothing inside a dialog could ever be tabbed to. Hand it back to the + # dialog: this is the only reason the load dialog's address field was + # unreachable, and it was broken the same way in every other modal. + if self.screen is not self.screen_stack[0]: + try: + self.screen.focus_next() + except Exception: # noqa: BLE001 -- screen with nothing focusable + pass + return if self._cur is None: return if self._split: @@ -4067,8 +4589,19 @@ class IdaTui(App): self.clear_filter() # Esc on the list clears an active filter return if len(self._nav) > 1: + self._nav_seq += 1 # invalidate any navigation still in flight self._nav.pop() - self._open_entry(self._nav[-1], push=False) + # Say something immediately: going back can need a decompile, and + # until it lands the panes still show where you were — with no + # feedback Esc reads as a no-op. + dest = self._nav[-1] + self._status(f"\u25c2 back to {dest.name}\u2026") + self._open_entry(dest, push=False) + elif self._hops: + # Local history is spent, but we got here from another binary. + label = self._hops.pop() + self._status(f"\u25c2 back to {label}\u2026") + self._switch_binary(label) # _states restores its nav and position elif self.query_one("#left", FunctionsPanel).display: table.focus() else: @@ -4186,8 +4719,55 @@ class IdaTui(App): if tgt is None or tgt.to is None: self.app.call_from_thread(self._status, "nothing to follow here") return + if self._follow_import(tgt.to): + return self._do_navigate(tgt.to, push=True) + def _import_stub(self, ea: int) -> str | None: + """The import name at ``ea``, if ``ea`` is one of this binary's import + stubs. That's the dead end phase 3 exists to open up: a call to strcmp + reaches the PLT/extern entry and Hex-Rays has nothing to decompile, + because the code lives in a library this binary only references.""" + try: + imps, _ = self.program.linkage() + except Exception: # noqa: BLE001 + return None + for i in imps: + if i.addr == ea: + return i.name + return None + + def _cross_binary_impl(self, name: str) -> tuple[str, int] | None: + """``(binary, addr)`` of a project binary that EXPORTS ``name``. + + Reads the on-disk index, so a provider resolves even when its worker was + evicted — the whole reason the index exists. + """ + if self._index is None or self._project is None or not name: + return None + try: + hits = self._index.providers(name, exclude=self._binary) + except Exception: # noqa: BLE001 + return None + return (hits[0].binary, hits[0].addr) if hits else None + + def _follow_import(self, ea: int) -> bool: + """Follow an import stub into the binary that implements it. True when + it was handled (caller must not also navigate locally).""" + name = self._import_stub(ea) + if not name: + return False + found = self._cross_binary_impl(name) + if found is None: + # Leave the local navigation alone: landing on the stub is still the + # honest answer when nothing in the project provides the symbol. + return False + label, addr = found + self.app.call_from_thread( + self._status, f"{name} \u2192 {label} (import resolved)") + self.app.call_from_thread(self._switch_then_goto, label, addr) + return True + @work(thread=True, group="nav") def _follow_decomp(self, line: str, word: str | None, line_ea: int | None = None) -> None: @@ -4215,6 +4795,8 @@ class IdaTui(App): if addr is None: self.app.call_from_thread(self._status, "nothing to follow on this line") return + if self._follow_import(addr): + return # Following FROM pseudocode: keep the reader in the decompiler when the # target is decompilable, instead of dropping to the linear listing. self._do_navigate(addr, push=True, prefer_decomp=True) @@ -4320,9 +4902,45 @@ class IdaTui(App): kind = x.kind or x.type or "?" items.append((x.frm, f"{x.frm:08X} {kind:<6} {loc}")) preselect = self._xref_preselect(xr, here_ea, here_end) + # Callers in OTHER project binaries. xrefs_to only ever sees this + # database, so an exported function looks unused from the inside even + # when half the project calls it — the import side of the linkage index + # is the only place that knowledge exists. + for label_b, addr_b, text_b in self._foreign_importers(subj, subj_name, fn): + items.append(((label_b, addr_b), text_b)) self.app.call_from_thread(self._present_xrefs, label, items, focus, preselect) - def _present_xrefs(self, label: str, items: list[tuple[int, str]], + def _foreign_importers(self, subj: int, subj_name, fn): # type: ignore[no-untyped-def] + """Project binaries that IMPORT the symbol at ``subj`` — the other half + of the phase-3 join, read from the on-disk index so a caller shows up + whether or not its worker is resident. + + Only for a symbol this binary actually exports: a local name that + happens to collide with another binary's import isn't a caller of ours. + """ + if self._index is None or self._project is None or self._binary is None: + return [] + name = subj_name if self._looks_like_symbol(subj_name) else None + if name is None and fn is not None and fn.addr == subj: + name = fn.name + if not name: + return [] + from .domain import link_name + name = link_name(name) + try: + _, exports = self.program.linkage() + except Exception: # noqa: BLE001 + return [] + if not any(e.name == name for e in exports): + return [] # we don't export it; nobody imports it FROM US + try: + hits = self._index.importers(name, exclude=self._binary) + except Exception: # noqa: BLE001 + return [] + return [(h.binary, h.addr, f"{h.addr:08X} import [{h.binary}] {name}") + for h in hits] + + def _present_xrefs(self, label: str, items: list[tuple[object, str]], focus_name: str | None = None, preselect: int = 0) -> None: if not self._xref_active: return # cancelled (Esc) while we were still gathering @@ -4335,12 +4953,17 @@ class IdaTui(App): self._status(f"{label}: {len(items)}") self.push_screen(XrefsScreen(label, items, preselect), self._on_xref_chosen) - def _on_xref_chosen(self, addr: int | None) -> None: - if addr is not None: - # If xrefs was invoked from the decompiler, land the jump back in the - # decompiler (when the target is decompilable) rather than the listing. - self._goto_ea(addr, push=True, focus_name=self._xref_focus_name, - prefer_decomp=(self._active == "decomp")) + def _on_xref_chosen(self, addr) -> None: # type: ignore[no-untyped-def] + if addr is None: + return + if isinstance(addr, tuple): # a caller in another project binary + binary, ea = addr + self._switch_then_goto(binary, ea) # records a hop, so Esc returns + return + # If xrefs was invoked from the decompiler, land the jump back in the + # decompiler (when the target is decompilable) rather than the listing. + self._goto_ea(addr, push=True, focus_name=self._xref_focus_name, + prefer_decomp=(self._active == "decomp")) # -- rename ------------------------------------------------------------ # @staticmethod @@ -4513,13 +5136,17 @@ class IdaTui(App): dec.loaded_ea = None # force re-decompile self._show_active() else: - # Capture the LIVE listing position straight from the widget (the - # source of truth) rather than trusting nav-entry tracking, which can - # go stale. bump_names() discards the segment model, so the reload - # rebuilds it; feeding the true cursor/scroll keeps the edited line on - # screen even on a huge segment (otherwise it primes/scrolls to a - # stale index and the renamed line lands off-screen — looking like the - # rename never applied). + # Capture the LIVE position from the widget (the source of truth) + # rather than trusting nav-entry tracking, which goes stale. Capture + # it as ADDRESSES via the anchor: bump_names() discards the segment + # model so the reload rebuilds it, and an edit that changes how many + # rows an item takes makes the old indices point somewhere else. + # Index capture is CORRECT here and an anchor is not: a rename or + # comment doesn't change how many rows anything takes, and the model + # this rebuilds is constructed empty — index_of_ea on it returns -1 + # until pages load, so an anchor would resolve to nothing while + # costing an extra model build on the UI thread. Address anchoring is + # for the edit paths that DO change row structure (see _do_edit_item). lst = self.query_one(ListingView) cur.view = "listing" if lst.model is not None: @@ -4672,7 +5299,8 @@ class IdaTui(App): view.focus() @work(thread=True, exclusive=True, group="makedata") - def _do_make_data(self, ea: int, type_decl: str) -> None: # worker context + def _do_make_data(self, ea: int, type_decl: str, + anchor: ViewAnchor | None = None) -> None: # worker context assert self.program is not None try: self.program.make_data(ea, type_decl) @@ -4680,12 +5308,15 @@ class IdaTui(App): self.app.call_from_thread(self._status, f"make data: {e}") return self.program.bump_items() + anchor = anchor or ViewAnchor() + anchor.flash = f"data ({type_decl}) @ {ea:#x} (Ctrl+S to save)" name = self.program.region_label(ea) lm = self.program.listing(ea) idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 - self.app.call_from_thread(self._open_at, ea, name, idx, False, -1, 0, True) + _cur, top = self._anchor_rows(anchor, lm, ea) self.app.call_from_thread( - self._edit_item_done, f"data ({type_decl})", ea) + self._open_at, ea, name, idx, False, -1, 0, True, None, top) + self.app.call_from_thread(self._edit_done, anchor) @work(thread=True, exclusive=True, group="rename") def _do_rename(self, view, old: str, new: str) -> None: # type: ignore[no-untyped-def] @@ -4794,21 +5425,44 @@ 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, (DisasmView, 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] 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 == "func": self.program.define_func(ea) elif kind == "string": @@ -4823,23 +5477,36 @@ 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) @work(thread=True, exclusive=True, group="save") def _save(self) -> None: @@ -4867,6 +5534,10 @@ class IdaTui(App): def _do_navigate(self, ea: int, push: bool, focus_name: str | None = None, prefer_decomp: bool = False) -> None: # worker context assert self.program is not None + # Which navigation this is. Decompiling below can take a while, and if + # you press Esc (or jump again) in the meantime this result is stale — + # applying it anyway silently undoes what you just did. + seq = self._nav_seq fn = self.program.function_of(ea) # Jumping FROM the decompiler: stay in pseudocode when the target is a # decompilable function, landing on the line that matches ``ea``. @@ -4886,7 +5557,8 @@ class IdaTui(App): dec_idx, col = self._decomp_locate(fn.addr, ea, focus_name or fn.name) self.app.call_from_thread( - self._open_decomp_entry, fn.addr, fn.name, dec_idx, col, push) + self._open_decomp_entry, fn.addr, fn.name, dec_idx, col, + push, seq) return # Otherwise: everything opens the one continuous listing at ``ea``. A # function name is used for the status label; a region gets a segment @@ -4898,9 +5570,17 @@ class IdaTui(App): self._open_at, ea, name, idx, push, -1, 0, fn is None, focus_name) def _open_decomp_entry(self, fn_addr: int, fn_name: str, dec_idx: int, - dec_cursor_x: int, push: bool) -> None: + dec_cursor_x: int, push: bool, + seq: int | None = None) -> None: """Open ``fn_addr`` in the decompiler as a real navigation (nav history - aware), landing on pseudocode line ``dec_idx`` column ``dec_cursor_x``.""" + aware), landing on pseudocode line ``dec_idx`` column ``dec_cursor_x``. + + ``seq`` is the navigation this result belongs to; if the user has + navigated since (Esc, another jump), the result is stale and dropped — + otherwise a slow decompile lands afterwards and undoes their Esc. + """ + if seq is not None and seq != self._nav_seq: + return if push: # Snapshot where we jumped from so 'back' returns there. If that was # the pseudocode (F5 makes a transient _cur not yet on the stack), @@ -4913,14 +5593,14 @@ class IdaTui(App): src.dec_cursor_x = dv.cursor_x src.dec_scroll_y = round(dv.scroll_offset.y) if not self._nav or self._nav[-1] is not src: - self._nav.append(src) + self._push_nav(src) # never stack a second copy of a spot else: self._save_current_pos() self._decomp_return = None # a real navigation abandons the F5 return entry = NavEntry(ea=fn_addr, name=fn_name, view="decomp", dec_cursor=dec_idx, dec_cursor_x=dec_cursor_x) if push: - self._nav.append(entry) + self._push_nav(entry) self._open_entry(entry, push=False) def _decomp_col_for(self, fn_addr: int, line_idx: int, name: str) -> int: @@ -4992,9 +5672,78 @@ class IdaTui(App): best_ea, best_idx = e, i return best_idx + @staticmethod + def _same_spot(a: NavEntry, b: NavEntry) -> bool: + """Same function, same view, same line — i.e. going from a to b is not a + move, so it has no business being a step in the history.""" + if a.ea != b.ea or a.view != b.view: + return False + return (a.dec_cursor == b.dec_cursor if a.view == "decomp" + else a.cursor == b.cursor) + + def _push_nav(self, entry: NavEntry) -> None: + """Append to the nav stack unless that would duplicate where we already are. + + Opening the place you're standing on — the app auto-lands on main, then + you pick main out of Ctrl+N — used to append an identical entry. The Esc + it created popped the stack without changing anything on screen: a dead + keypress, which is exactly what "back doesn't work" feels like. Replace + the top instead, so the newer entry's metadata still wins. + """ + top = self._nav[-1] if self._nav else None + if top is not None and self._same_spot(top, entry): + self._nav[-1] = entry + return + self._nav.append(entry) + + # -- view state across a model rebuild --------------------------------- # + def _anchor(self, flash: str | None = None) -> ViewAnchor: + """Capture where we're looking, BEFORE an edit rebuilds the model. + + Must run on the UI thread: it reads live widget state. + """ + a = ViewAnchor(view=self._active, flash=flash) + view = self._active_code_view() + model = getattr(view, "model", None) + if view is None or model is None: + return a + a.cursor_x = getattr(view, "cursor_x", 0) + try: + a.ea = view._cursor_ea() + except Exception: # noqa: BLE001 + a.ea = None + top = round(view.scroll_offset.y) + h = model.cached_line(top) or model.get(top) + a.top_ea = getattr(h, "ea", None) + return a + + @staticmethod + def _anchor_rows(a: ViewAnchor, model, fallback_ea: int | None = None): + """(cursor_row, top_row) for ``a`` in a freshly built ``model``. + + -1 means "no opinion" — the caller's own default wins. + """ + if model is None: + return (-1, -1) + + def row_of(ea): + if ea is None: + return -1 + try: + i = model.index_of_ea(ea) + except Exception: # noqa: BLE001 + return -1 + return i if i >= 0 else -1 + + cur = row_of(a.ea if a.ea is not None else fallback_ea) + if cur < 0 and fallback_ea is not None: + cur = row_of(fallback_ea) + return (cur, row_of(a.top_ea)) + def _open_at(self, ea: int, name: str, cursor: int, push: bool, dec_cursor: int = -1, dec_cursor_x: int = 0, - is_region: bool = False, focus_name: str | None = None) -> None: + is_region: bool = False, focus_name: str | None = None, + scroll_y: int = -1) -> None: if push: self._save_current_pos() self._decomp_return = None # a real navigation abandons the F5 return @@ -5002,11 +5751,13 @@ class IdaTui(App): # the row is loaded, instead of column 0. self._pending_focus_name = focus_name entry = NavEntry(ea=ea, name=name, cursor=cursor, is_region=is_region) + if scroll_y >= 0: + entry.scroll_y = scroll_y if dec_cursor >= 0: entry.dec_cursor = dec_cursor entry.dec_cursor_x = dec_cursor_x if push: - self._nav.append(entry) + self._push_nav(entry) self._open_entry(entry, push=False) def on_input_changed(self, event: Input.Changed) -> None: @@ -5117,7 +5868,7 @@ class IdaTui(App): view, ea = self._makedata_ctx self._end_makedata() if view is not None and value: - self._do_make_data(ea, value) + self._do_make_data(ea, value, self._anchor()) return if inp.id == "goto": self._end_goto() @@ -5131,7 +5882,12 @@ class IdaTui(App): self.query_one("#func-table", DataTable).focus() def _code_view(self): # type: ignore[no-untyped-def] - return self.query_one(DecompView if self._pref == "decomp" else ListingView) + """The code pane the user is in — where focus belongs after a prompt closes. + + This used to pick on _pref, which was always "listing", so closing the + goto prompt while reading pseudocode threw focus into the listing. + """ + return self._active_code_view() def _active_code_view(self): # type: ignore[no-untyped-def] """The currently-shown code widget (for reading the cursor address).""" @@ -5305,14 +6061,21 @@ class IdaTui(App): hx.display = False lst.display = dec.display = True self.query_one("#panes").set_class(True, "split") - if self._cur is not None and dec.loaded_ea != self._cur.ea: + busy = self._cur is not None and dec.loaded_ea != self._cur.ea + if busy: dec.loading = True self._load_decomp(self._cur.ea, self._cur.name) else: dec.loading = False if grab: (dec if self._active == "decomp" else lst).focus() - self._status_for_cur("split") + if busy and self._cur is not None: + # Keep the in-flight message: the idle status used to overwrite + # it, so travelling history in split showed nothing at all while + # a decompile ran and Esc looked like a no-op. + self._status(f"{self._cur.name} \u2014 decompiling\u2026") + else: + self._status_for_cur("split") return self._split = False # a single-view target (hex, etc.) leaves split self.query_one("#panes").set_class(False, "split") @@ -5386,10 +6149,14 @@ class IdaTui(App): self._show_active() def on_hex_view_to_code(self, msg: HexView.ToCode) -> None: - self._active = self._pref + self._active = self._code_mode() self._goto_ea(msg.va, push=True) def _status_for_cur(self, mode: str) -> None: + flash, self._flash = self._flash, None + if flash: + self._status(flash) # the edit that caused this reload wins + return if self._cur is not None: self._status(f"{self._cur.name} @ {self._cur.ea:#x} [{mode}]") @@ -5626,6 +6393,13 @@ class IdaTui(App): return ea = msg.ea if ea is not None: + # A pending flash (the result of an edit that caused this reload) + # outranks the idle hint: the cursor lands here as part of the + # reload, so this handler would otherwise always have the last word. + flash, self._flash = self._flash, None + if flash: + self._status(flash) + return sec = self.program.section_of(ea) if self.program else None self._status(f"{sec or '?'} @ {ea:#x} [listing] " "(c code · p func · u undefine · Enter follow)") diff --git a/idatui/domain.py b/idatui/domain.py index ef37ab9..e4fbec9 100644 --- a/idatui/domain.py +++ b/idatui/domain.py @@ -176,6 +176,35 @@ class StrLit: type: str = "" +def link_name(raw: str) -> str: + """A linkage name reduced to what actually joins across binaries. + + ELF symbol versioning means the importer sees ``strrchr@@GLIBC_2.2.5`` while + the provider may export ``strrchr``, ``strrchr@GLIBC_2.2.5`` or the versioned + spelling — comparing raw names silently resolves almost nothing. Cut at the + first '@' so both sides meet on the bare symbol. + """ + n = (raw or "").strip() + at = n.find("@") + return n[:at] if at > 0 else n + + +@dataclass(frozen=True) +class Linkage: + """One import or export: a name this binary takes from, or offers to, other + modules. ``module`` is set for imports (the library IDA attributes it to), + ``ordinal`` for exports. + + ``name`` is the joinable name; ``raw`` keeps the spelling IDA reported, which + is what the user sees in the listing. + """ + addr: int + name: str + module: str = "" + ordinal: int = 0 + raw: str = "" + + @dataclass class Decompilation: ea: int @@ -554,6 +583,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 @@ -620,14 +659,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 @@ -636,9 +677,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 @@ -650,8 +749,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 @@ -662,12 +762,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): + 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): - return i + # 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: @@ -675,7 +782,7 @@ class ListingModel: if self._load_next_page() == 0: break if progress: - progress(len(self._heads)) + progress(self._rows) @property def complete(self) -> bool: @@ -684,23 +791,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: @@ -712,7 +854,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 @@ -818,6 +960,7 @@ class Program: self._decomp: dict[int, tuple[Decompilation, int]] = {} self._decomp_maps: dict[int, tuple[list[list[int]], int]] = {} # line->ea sets self._strings: list["StrLit"] | None = None # whole-binary string literals + self._linkage: tuple[list["Linkage"], list["Linkage"]] | None = None self._name_gen = 0 # bumped on rename; invalidates stale name caches self._segments_cache: list[tuple[int, int, int, str]] | None = None self._sections: list[tuple[int, int, str]] | None = None @@ -1221,14 +1364,30 @@ 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_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) -> 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']}") + raise IDAToolError("define_func", f"@ {ea:#x}: {res['error']}") def undefine(self, ea: int, size: int | None = None) -> None: """Undefine the item at ``ea`` back to raw bytes (IDA's 'u').""" @@ -1237,7 +1396,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`` @@ -1248,7 +1407,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. @@ -1257,7 +1416,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: @@ -1312,6 +1471,34 @@ class Program: self._strings = out return out + def linkage(self) -> tuple[list[Linkage], list[Linkage]]: + """``(imports, exports)`` for this binary, cached. ``([], [])`` if the + tool is unavailable — an old worker must not break the caller.""" + with self._lock: + hit = self._linkage + if hit is not None: + return hit + try: + payload = self.client.call("list_linkage", kind="both") + except IDAToolError: + return ([], []) + if not isinstance(payload, dict): + return ([], []) + imps = [Linkage(addr=_as_int(r.get("addr", 0)), + name=link_name(r.get("name", "")), + module=r.get("module", "") or "", + raw=r.get("name", "") or "") + for r in payload.get("imports", []) if isinstance(r, dict)] + exps = [Linkage(addr=_as_int(r.get("addr", 0)), + name=link_name(r.get("name", "")), + ordinal=int(r.get("ordinal", 0) or 0), + raw=r.get("name", "") or "") + for r in payload.get("exports", []) if isinstance(r, dict)] + out = ([i for i in imps if i.name], [e for e in exps if e.name]) + with self._lock: + self._linkage = out + return out + def decomp_map(self, ea: int) -> list[list[int]]: """Per-pseudocode-line instruction coverage for the split-view region highlight: a list aligned to the decompiled lines, each the EAs the diff --git a/idatui/drive.py b/idatui/drive.py index 6c96335..b6fa641 100644 --- a/idatui/drive.py +++ b/idatui/drive.py @@ -165,6 +165,31 @@ def cmd_names(c, args): or "(no match)" +def cmd_binaries(c, args): + """Project inventory: which binaries, which is active, which have a live + worker, how much of each is indexed.""" + r = c.call("binaries") + out = [] + for b in r["binaries"]: + mark = "*" if b["active"] else ("~" if b["resident"] else " ") + out.append(f" {mark} {b['label']:<28} indexed={b['indexed']:<7} {b['source']}") + if r.get("hops"): + out.append(f" (Esc returns to: {' <- '.join(r['hops'])})") + return "\n".join(out) + "\n * active ~ worker resident" + + +def cmd_switch(c, args): + if not args: + raise SystemExit("usage: switch <binary> [addr|name]") + p = {"binary": args[0]} + if len(args) > 1: + a = args[1] + p["addr"] = a if a.lower().startswith("0x") else c.call("resolve", name=a)["ea"] + r = c.call("switch", **p) + fn = (r.get("function") or {}).get("name") + return f" now on {r.get('binary') or args[0]}" + (f" @ {fn}" if fn else "") + + def _rename_one(c, old, new): c.call("goto", target=old, delay_ms=0) st = c.call("rename", name=new, word=old, delay_ms=0) @@ -232,6 +257,7 @@ COMMANDS = { "callees": cmd_callees, "callers": cmd_callers, "names": cmd_names, "rename": cmd_rename, "mv": cmd_mv, "note": cmd_note, "retype": cmd_retype, "save": cmd_save, "screen": cmd_screen, "raw": cmd_raw, + "binaries": cmd_binaries, "switch": cmd_switch, } diff --git a/idatui/formats.py b/idatui/formats.py new file mode 100644 index 0000000..b2f784d --- /dev/null +++ b/idatui/formats.py @@ -0,0 +1,134 @@ +"""Recognising what a file is, and what to ask when we can't. + +IDA picks a loader by looking at the file. When one matches, it knows the +processor, the load address and often the entry point, and there is nothing to +ask. When none matches — a raw firmware dump, a flash image, a decompressed blob +— IDA falls back to a plain binary load: **x86 at address 0**. That is a silent +wrong answer, not an error; the database opens, analysis runs, and finds nothing. + +Interactive IDA handles this by asking. This module is the "should we ask, and +what are the sensible answers" half of doing the same. + +The sniff deliberately only recognises formats IDA definitely has a loader for. +Being wrong in the cautious direction costs one dismissible dialog; being wrong +the other way is the silent-nothing case we're trying to kill. +""" + +from __future__ import annotations + +import os + +#: (magic, offset, name) for formats IDA loads without help. +_MAGIC: tuple[tuple[bytes, int, str], ...] = ( + (b"\x7fELF", 0, "ELF"), + (b"MZ", 0, "PE/MZ"), + (b"\xfe\xed\xfa\xce", 0, "Mach-O"), + (b"\xfe\xed\xfa\xcf", 0, "Mach-O 64"), + (b"\xce\xfa\xed\xfe", 0, "Mach-O (LE)"), + (b"\xcf\xfa\xed\xfe", 0, "Mach-O 64 (LE)"), + (b"\xca\xfe\xba\xbe", 0, "Mach-O fat/Java class"), + (b"dex\n", 0, "Dalvik"), + (b"\x00asm", 0, "WebAssembly"), + (b"!<arch>", 0, "ar archive"), + (b"\x4c\x01", 0, "COFF i386"), + (b"\x64\x86", 0, "COFF x64"), + (b"\x00\x00\x03\xf3", 0, "Amiga hunk"), + (b"\x7fCGC", 0, "CGC"), +) + +#: Text-ish container formats: recognised by their first line, not a magic word. +_TEXT_PREFIX: tuple[tuple[bytes, str], ...] = ( + (b":", "Intel HEX"), + (b"S0", "Motorola S-record"), + (b"S1", "Motorola S-record"), + (b"S2", "Motorola S-record"), + (b"S3", "Motorola S-record"), +) + + +def sniff(path: str) -> str | None: + """The container format of ``path``, or None when nothing recognises it. + + None is the interesting answer: it means IDA will guess, and its guess is + x86-at-zero regardless of what the bytes actually are. + """ + try: + with open(path, "rb") as f: + head = f.read(64) + except OSError: + return None + if not head: + return None + for magic, off, name in _MAGIC: + if head[off:off + len(magic)] == magic: + return name + # Only treat a text prefix as a format if the whole head is printable — + # a raw blob starting with 0x3a (':') is far more likely than Intel HEX. + if all(32 <= b < 127 or b in (9, 10, 13) for b in head): + for prefix, name in _TEXT_PREFIX: + if head.startswith(prefix): + return name + return None + + +def needs_load_options(path: str) -> bool: + """True when we should ask how to load ``path`` rather than let IDA guess.""" + return os.path.isfile(path) and sniff(path) is None + + +#: Processors worth offering, as (IDA -p name, human label). +#: +#: IDA ships 73 processor modules, most of which are museum pieces; a list that +#: long is a worse answer than a short one plus free text. These are the targets +#: that actually turn up in firmware work, endianness spelled out because +#: getting it wrong is the most common way to end up with zero functions. +#: +#: EVERY NAME HERE IS VERIFIED against a real IDA by opening a scratch blob with +#: ``-p<name>`` and reading back ``inf_get_procname()`` — see +#: ``tools/verify_procs.py``. That is not ceremony: a wrong name is REJECTED by +#: IDA (rc=4) with no useful message, which is the same silent-failure class this +#: dialog exists to prevent. Two of the first twenty were wrong ("h8" and +#: "sparc" are module filenames, not processor names), and the aliases people +#: 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"), + ("armb", "ARM — big-endian"), + ("metapc", "x86 / x86-64"), + ("mipsl", "MIPS — little-endian"), + ("mipsb", "MIPS — big-endian"), + ("ppc", "PowerPC — big-endian"), + ("ppcl", "PowerPC — little-endian"), + ("sh4", "SuperH SH-4"), + ("68k", "Motorola 68000 / m68k"), + ("riscv", "RISC-V"), + ("tricore", "Infineon TriCore"), + ("xtensa", "Tensilica Xtensa (ESP32 etc.)"), + ("avr", "Atmel AVR"), + ("z80", "Zilog Z80"), + ("tms320c6", "TI TMS320C6x DSP"), + ("m32r", "Renesas M32R"), + ("arc", "Synopsys ARC"), + ("h8300", "Renesas H8/300"), + ("sparcb", "SPARC — big-endian"), + ("sparcl", "SPARC — little-endian"), + ("s390", "IBM S/390"), +) + + +def load_args(processor: str = "", base: int = 0, extra: str = "") -> str: + """``processor``/``base`` as IDA command-line switches. + + ``-b`` is in PARAGRAPHS, not bytes: ``-b1000`` loads at 0x10000. Every caller + goes through here so that conversion is done in exactly one place. + """ + parts = [] + if processor: + parts.append(f"-p{processor}") + if base: + parts.append(f"-b{int(base) >> 4:x}") + if extra: + parts.append(extra) + return " ".join(parts) diff --git a/idatui/highlight.py b/idatui/highlight.py index 9fc692f..5f504bf 100644 --- a/idatui/highlight.py +++ b/idatui/highlight.py @@ -15,19 +15,28 @@ from pygments.lexers import CLexer from pygments.token import Token # Token -> style, checked in priority order (first hierarchical match wins). -# Palette roughly follows a dark IDE theme. +# +# Same measured palette as the listing (see the theme notes): one hue = one +# meaning across BOTH panes, so a string is the same green and a symbol the same +# blue whether you're reading disassembly or pseudocode. Contrast ratios are +# against the app background #12161c; signal sits at 7:1+ and structure recedes +# to 3-4:1 so punctuation stops competing with the code. +# +# Control keywords take the brightest NEUTRAL rather than a hue, mirroring the +# mnemonic column: they're the skeleton you scan for, and a hue there would +# claim a meaning the rest of the palette already assigns. _STYLES: list[tuple[object, Style]] = [ - (Token.Comment, Style(color="grey54", italic=True)), - (Token.Keyword.Type, Style(color="#4ec9b0")), # __int64, char, ... - (Token.Keyword, Style(color="#c586c0", bold=True)), # if/else/return/goto - (Token.Name.Builtin, Style(color="#4ec9b0")), - (Token.Literal.String, Style(color="#ce9178")), # "..." - (Token.Literal.Number, Style(color="#b5cea8")), # 0x10, 42 - (Token.Operator, Style(color="#d4d4d4")), - (Token.Punctuation, Style(color="grey70")), - (Token.Name, Style(color="#dcdcaa")), # identifiers / calls + (Token.Comment, Style(color="#7c8b9e", italic=True)), # 5.2:1 commentary + (Token.Keyword.Type, Style(color="#93aee0")), # 8.1:1 type info + (Token.Keyword, Style(color="#e8ecf2", bold=True)), # 15.3:1 control flow + (Token.Name.Builtin, Style(color="#93aee0")), # 8.1:1 type info + (Token.Literal.String, Style(color="#9ece6a")), # 9.9:1 strings + (Token.Literal.Number, Style(color="#d8a657")), # 8.2:1 data/number + (Token.Operator, Style(color="#c3cad3")), # 11.0:1 body + (Token.Punctuation, Style(color="#626c7a")), # 3.4:1 structure + (Token.Name, Style(color="#7aa2f7")), # 7.2:1 symbol names ] -_DEFAULT = Style(color="#d4d4d4") +_DEFAULT = Style(color="#c3cad3") # 11.0:1 body _lexer = CLexer(stripnl=False, ensurenl=False) diff --git a/idatui/index.py b/idatui/index.py index e7e4a41..751ad1f 100644 --- a/idatui/index.py +++ b/idatui/index.py @@ -32,6 +32,9 @@ MIN_TRIGRAM = 3 KIND_FUNC = "func" KIND_STRING = "string" +#: Cross-binary linkage: what a binary takes from, and offers to, other modules. +KIND_IMPORT = "import" +KIND_EXPORT = "export" @dataclass(frozen=True) @@ -152,6 +155,37 @@ class ProjectIndex: return [] # malformed FTS expression: treat as no matches return [Hit(binary=b, kind=k, addr=int(a), text=t) for b, k, a, t in rows] + def exact(self, name: str, kind: str, exclude: str | None = None) -> list[Hit]: + """Every entry whose text is EXACTLY ``name``, for the linkage join. + + Deliberately not ``search()``: an import must resolve to the export of + that name, not to everything containing it (``read`` would otherwise + match ``pread``, ``read_line``, ``thread_start``). Exact match also + works below the trigram floor, which matters — plenty of real exports + are one or two characters. + """ + n = (name or "").strip() + if not n: + return [] + sql = "SELECT binary, kind, addr, text FROM entries WHERE text = ? AND kind = ?" + args: list = [n, kind] + if exclude: + sql += " AND binary <> ?" + args.append(exclude) + rows = self._db.execute(sql + " ORDER BY binary, addr", args).fetchall() + return [Hit(binary=b, kind=k, addr=int(a), text=t) for b, k, a, t in rows] + + def providers(self, name: str, exclude: str | None = None) -> list[Hit]: + """Binaries in the project that EXPORT ``name`` (skip ``exclude``, the + binary asking). This is the 'follow an import to its implementation' + half of the join.""" + return self.exact(name, KIND_EXPORT, exclude) + + def importers(self, name: str, exclude: str | None = None) -> list[Hit]: + """Binaries in the project that IMPORT ``name`` — 'who in the project + calls this export'.""" + return self.exact(name, KIND_IMPORT, exclude) + # -- introspection ------------------------------------------------------ # def counts(self) -> dict[str, int]: """Indexed entry count per binary.""" diff --git a/idatui/launch.py b/idatui/launch.py index 0ae63a1..f51e5af 100644 --- a/idatui/launch.py +++ b/idatui/launch.py @@ -23,6 +23,13 @@ import sys _LOCK_SUFFIXES = (".id0", ".id1", ".id2", ".nam", ".til") +def _load_args(load: dict) -> str: + """``load`` as IDA switches, for the single-binary path (no project ref).""" + from .formats import load_args + return load_args(load.get("processor", ""), int(load.get("base", 0) or 0), + str(load.get("ida_args", "") or "")) + + def _log(msg: str) -> None: print(f"ida-tui: {msg}", file=sys.stderr) @@ -57,8 +64,37 @@ def main(argv: list[str] | None = None) -> int: help="do not run the keepalive heartbeat") p.add_argument("--rpc", metavar="PATH", help="listen for RPC on this unix socket (puppeteer the TUI)") + g = p.add_argument_group( + "loading a headerless blob", + "An ELF/PE/Mach-O says what it is. A raw firmware dump doesn't, and IDA " + "falls back to x86 at address 0 — which analyses to nothing. These say " + "how to read it, and are recorded per binary in a project.") + g.add_argument("--processor", metavar="NAME", + help="IDA processor: arm, armb (big-endian), mipsb, metapc, …") + g.add_argument("--base", metavar="ADDR", + help="load address, e.g. 0x8000000 (any base; NOT paragraphs)") + g.add_argument("--ida-args", metavar="STR", dest="ida_args", + help="extra IDA command-line switches, passed through as-is") args = p.parse_args(argv) + load: dict = {} + if args.processor: + load["processor"] = args.processor + if args.base: + try: + base = int(args.base, 0) + except ValueError: + _log(f"--base must be a number (got {args.base!r})") + return 2 + if base % 16: + # -b is in paragraphs, so an unaligned base cannot be expressed and + # would silently load somewhere else. + _log(f"--base must be 16-byte aligned (got {base:#x})") + return 2 + load["base"] = base + if args.ida_args: + load["ida_args"] = args.ida_args + project = None binary = None if args.project: @@ -67,12 +103,20 @@ def main(argv: list[str] | None = None) -> int: try: if os.path.isfile(ppath): project = Project.load(ppath) - for b in args.binary: # extend an existing project - project.add(b) - if args.binary: - project.save() + if args.binary: # extend an existing project, skipping repeats + before = len(project.refs) + for b in args.binary: + project.add(b, load=load or None) + added = len(project.refs) - before + dupes = len(args.binary) - added + if added: + project.save() + _log(f"added {added} binary(ies) to {ppath}") + if dupes: + _log(f"{dupes} already in the project (matched by path) " + f"— left alone") elif args.binary: - project = Project.create(ppath, args.binary) + project = Project.create(ppath, args.binary, load=load or None) _log(f"created project {ppath} with {len(project.refs)} binaries") else: _log(f"no such project: {ppath} (pass binaries to create it)") @@ -112,7 +156,8 @@ def main(argv: list[str] | None = None) -> int: return 1 rpc_path = os.path.abspath(os.path.expanduser(args.rpc)) if args.rpc else None IdaTui(open_path=binary, keepalive=not args.no_keepalive, - rpc_path=rpc_path, ttl=args.ttl, project=project).run() + rpc_path=rpc_path, ttl=args.ttl, project=project, + load_args=_load_args(load)).run() return 0 diff --git a/idatui/pane.py b/idatui/pane.py index e34b822..53afef3 100644 --- a/idatui/pane.py +++ b/idatui/pane.py @@ -131,15 +131,20 @@ def spawn(args) -> int: if not os.environ.get("TMUX"): print("error: not inside tmux (spawn creates a tmux pane)", file=sys.stderr) return 2 - if not args.open: - print("error: pass --open <binary>", file=sys.stderr) + if not args.open and not getattr(args, "project", None): + print("error: pass --open <binary> or --project <file>", file=sys.stderr) return 2 sock = args.sock or os.path.join(_sockdir(), f"idatui-{secrets.token_hex(3)}.sock") - target = os.path.abspath(os.path.expanduser(args.open)) - if not os.path.exists(target): + project = (os.path.abspath(os.path.expanduser(args.project)) + if getattr(args, "project", None) else None) + target = os.path.abspath(os.path.expanduser(args.open)) if args.open else None + if target is not None and not os.path.exists(target): print(f"error: no such binary: {target}", file=sys.stderr) return 2 + if project is not None and target is None and not os.path.exists(project): + print(f"error: no such project: {project}", file=sys.stderr) + return 2 # Reap workers leaked by previously-stopped/crashed panes so we don't spawn # into a full IDA_MCP_MAX_WORKERS (which makes the new TUI hang forever, @@ -151,7 +156,15 @@ def spawn(args) -> int: # the command the pane runs: the launcher spawns a private idalib worker for # this binary and becomes the TUI, so kill-pane tears the whole thing down. - inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock] + if project is not None: + # launch takes: --project FILE [binaries...]; extra binaries are added to + # the project (and a missing project file is created from them). + inner = [args.python, "-m", "idatui.launch", "--project", project] + if target is not None: + inner.append(target) + inner += ["--rpc", sock] + else: + inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock] cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner) split = ["split-window", "-v" if args.vertical else "-h", @@ -166,8 +179,8 @@ def spawn(args) -> int: split.append(cmd) pane = _tmux(*split) - row = {"sock": sock, "pane": pane, "target": target, - "kind": "open", "started": time.time()} + row = {"sock": sock, "pane": pane, "target": project or target, + "kind": "project" if project else "open", "started": time.time()} reg = [r for r in _load_registry() if r.get("sock") != sock] reg.append(row) _save_registry(reg) @@ -301,8 +314,11 @@ def main(argv: list[str]) -> int: sub = p.add_subparsers(dest="cmd", required=True) sp = sub.add_parser("spawn", help="open a TUI pane and wait until ready") - sp.add_argument("--open", metavar="PATH", required=True, + sp.add_argument("--open", metavar="PATH", help="binary to open (its dir must be writable)") + sp.add_argument("--project", metavar="FILE", + help="project file to open instead of a single binary; " + "any --open paths are added to it (created if absent)") sp.add_argument("--sock", help="RPC socket path (default: auto in $XDG_RUNTIME_DIR)") sp.add_argument("--python", default=DEFAULT_PY, help=f"python for the TUI ({DEFAULT_PY})") sp.add_argument("--vertical", action="store_true", help="split vertically (stacked)") diff --git a/idatui/pool.py b/idatui/pool.py index df852f8..ae37c25 100644 --- a/idatui/pool.py +++ b/idatui/pool.py @@ -56,7 +56,7 @@ def _pss_mb(pid: int | None) -> int: def _default_spawn(ref: BinaryRef, ttl: int): # pragma: no cover - needs idalib from .worker_client import WorkerClient - return WorkerClient(ref.staged, ttl=ttl) + return WorkerClient(ref.staged, ttl=ttl, load_args=ref.load_args) class WorkerPool: @@ -129,6 +129,37 @@ class WorkerPool: self._enforce_budget(protect=label) return client + def prewarm(self, label: str, progress=None) -> bool: + """Spawn a worker for ``label`` only if it fits the budget AS IT STANDS. + + Pre-warming must never cost residency: evicting a binary the user + actually visited to speculatively load one they haven't is a straight + downgrade, and the eviction would also throw away that binary's caches. + So this refuses rather than making room, and returns False. + + The cost of a worker that doesn't exist yet can only be estimated; the + largest resident one is the best evidence available (they are all the + same program with a different database). With nothing resident we have + no evidence at all, so we allow one — that is the case where the budget + is certainly free. + """ + if label in self._clients: + return False + if self.project.by_label(label) is None: + return False + used = self.memory_mb() + est = max((self._mem(c) for c in self._clients.values()), default=0) + if used + est > self.budget_mb: + return False + self.get(label, progress=progress) + # get() enforces the budget protecting the NEW label; if that had to + # evict, our estimate was wrong and the speculative worker is the one + # that should go — never a binary the user chose. + if self.memory_mb() > self.budget_mb and label != self.active: + self.evict(label) + return False + return True + def _touch(self, label: str) -> None: if label in self._lru: self._lru.remove(label) diff --git a/idatui/project.py b/idatui/project.py index 2d417df..e2fc542 100644 --- a/idatui/project.py +++ b/idatui/project.py @@ -52,12 +52,44 @@ class BinaryRef: label: str # unique within the project; names the staged file source: str # absolute path to the original binary staged: str # absolute path IDA actually opens (inside the sidecar) + #: How to LOAD it. Only meaningful for a headerless blob: an ELF/PE says what + #: it is, a raw firmware image doesn't, and IDA defaults to metapc at 0. + processor: str = "" # IDA processor name: arm, armb, mipsb, metapc, … + base: int = 0 # load address (natural, e.g. 0x8000000) + ida_args: str = "" # escape hatch: extra IDA command-line switches @property def db(self) -> str: """The database IDA creates for the staged file.""" return self.staged + ".i64" + @property + def load_args(self) -> str: + """``processor``/``base`` as IDA command-line switches. + + ``-b`` is in PARAGRAPHS, not bytes — ``-b1000`` loads at 0x10000. That + trap is worth hiding: projects say ``"base": "0x8000000"`` and the one + conversion lives in ``formats.load_args``. + """ + from .formats import load_args + return load_args(self.processor, self.base, self.ida_args) + + +def _as_addr(v) -> int: + """A load address from JSON: int, or a string in any base ("0x8000000"). + + Addresses are written by hand in a project file, so accept how people write + them rather than demanding decimal. + """ + if v is None or v == "": + return 0 + if isinstance(v, int): + return v + try: + return int(str(v), 0) + except ValueError: + return 0 + def _stat_key(path: str) -> tuple[int, int] | None: """(size, mtime) identity used to spot a source that changed under us.""" @@ -109,7 +141,12 @@ class Project: e = {"path": e} if not isinstance(e, dict) or not e.get("path"): raise ProjectError(f"project {path}: bad binary entry {e!r}") - norm.append({k: e[k] for k in ("path", "label") if e.get(k)}) + # Keep every recognised key: a whitelist of path/label silently + # dropped the load options on the first save, so a blob's processor + # and base vanished the moment the project was reopened. + norm.append({k: e[k] for k in + ("path", "label", "processor", "base", "ida_args") + if e.get(k) not in (None, "")}) name = raw.get("name") or os.path.splitext(os.path.basename(path))[0] try: pct = int(raw.get("memory_pct", DEFAULT_MEMORY_PCT)) @@ -119,12 +156,25 @@ class Project: @classmethod def create(cls, path: str, binaries: list[str], name: str | None = None, - memory_pct: int = DEFAULT_MEMORY_PCT) -> "Project": - """Write a new project file listing ``binaries`` (an ad-hoc project).""" + memory_pct: int = DEFAULT_MEMORY_PCT, load: dict | None = None) -> "Project": + """Write a new project file listing ``binaries`` (an ad-hoc project). + + ``load`` carries per-binary load options (processor/base/ida_args) that + apply to every binary given here — a headerless blob needs them, and one + command line normally adds blobs of the same kind. + """ if not binaries: raise ProjectError("a project needs at least one binary") - entries = [{"path": os.path.abspath(os.path.expanduser(b))} - for b in binaries] + entries, seen = [], set() + for b in binaries: # the same file twice on one command line is a typo + p = os.path.abspath(os.path.expanduser(b)) + key = os.path.realpath(p) + if key in seen: + continue + seen.add(key) + e = {"path": p} + e.update({k: v for k, v in (load or {}).items() if v}) + entries.append(e) path = os.path.abspath(os.path.expanduser(path)) proj = cls(path, name or os.path.splitext(os.path.basename(path))[0], entries, memory_pct) @@ -177,21 +227,66 @@ class Project: n += 1 label = f"{label}_{n}" used.add(label) - refs.append(BinaryRef(label=label, source=src, - staged=os.path.join(self.bin_dir, label))) + refs.append(BinaryRef( + label=label, source=src, + staged=os.path.join(self.bin_dir, label), + processor=str(e.get("processor") or ""), + base=_as_addr(e.get("base")), + ida_args=str(e.get("ida_args") or ""))) return tuple(refs) @property 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) - def add(self, binary: str, label: str | None = None) -> BinaryRef: + def by_source(self, binary: str) -> BinaryRef | None: + """The entry for ``binary``, matched by resolved path. + + Identity is the real path, not the file name: a project can legitimately + hold two different ``foo.elf`` from different directories (the labels + disambiguate them), but the same file must not be listed twice — and + ``./a.elf``, ``/abs/a.elf`` and a symlink to it are all the same file. + """ + key = os.path.realpath(os.path.abspath(os.path.expanduser(binary))) + return next((r for r in self._refs + if os.path.realpath(r.source) == key), None) + + def add(self, binary: str, label: str | None = None, + load: dict | None = None) -> BinaryRef: + """Add a binary, or return the existing entry if it's already here.""" + existing = self.by_source(binary) + if existing is not None: + return existing entry = {"path": os.path.abspath(os.path.expanduser(binary))} if label: entry["label"] = label + entry.update({k: v for k, v in (load or {}).items() if v}) self._entries.append(entry) self._refs = self._build_refs() return self._refs[-1] diff --git a/idatui/rpc.py b/idatui/rpc.py index 59eec86..0921b89 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -72,6 +72,8 @@ METHODS = { "search": "{term,direction?=1} incremental search in the code view", "select": "{index?} choose the highlighted/nth item in the open modal", "save": "persist the .i64 (Ctrl+S)", + "binaries": "-> project binaries {label,active,resident,indexed} (project mode)", + "switch": "{binary,addr?} make another project binary active (addr also jumps)", "close": "dismiss a modal (Escape)", "move": "{dir,n?=1} fast movement (down/up/.../pagedown)", "cursor": "{line?,col?} set the code-pane cursor directly", @@ -168,12 +170,14 @@ def snapshot(app) -> dict[str, Any]: pass return { "active": app._active, - "pref": app._pref, + "pref": app._code_mode(), # kept for wire compat; a constant now "function": ({"ea": cur.ea, "name": cur.name} if cur else None), "cursor": _cursor_info(app, w), "status": st, "filter": app._filter_term, + "binary": app._binary, # None outside project mode "nav_depth": len(app._nav), + "hops": list(getattr(app, "_hops", [])), "dirty": bool(app._dirty), "modal": _modal_snapshot(app), **_readiness(app), @@ -526,6 +530,42 @@ class RpcServer: if method == "functions": return functions(app, params.get("filter"), int(params.get("limit", 50))) + # -- projects ------------------------------------------------------ # + if method == "binaries": + if app._project is None: + raise ValueError("not a project session (launch with --project)") + counts = app._index.counts() if app._index is not None else {} + resident = set(app._pool.resident()) if app._pool is not None else set() + return {"active": app._binary, "hops": list(app._hops), + "binaries": [{"label": r.label, "source": r.source, + "active": r.label == app._binary, + "resident": r.label in resident, + "indexed": int(counts.get(r.label, 0))} + for r in app._project.refs]} + + if method == "switch": + if app._project is None: + raise ValueError("not a project session (launch with --project)") + label = str(params.get("binary") or params.get("label") or "") + if app._project.by_label(label) is None: + have = ", ".join(r.label for r in app._project.refs) + raise ValueError(f"no such binary {label!r} (have: {have})") + addr = params.get("addr") + if label == app._binary and addr is None: + return snapshot(app) + if addr is None: + app._switch_binary(label) + else: + # Same path a project search hit takes, so it records a hop and + # Esc comes back here. + app._switch_then_goto(label, int(str(addr), 0) + if isinstance(addr, str) else int(addr)) + await settle(app, lambda: app._binary == label + and app._func_index is not None + and app._func_index.complete, + timeout=float(params.get("timeout", 300.0))) + return snapshot(app) + # -- structured introspection (heavy: run off the UI loop) -------- # loop = asyncio.get_running_loop() if method == "pseudocode": diff --git a/idatui/worker.py b/idatui/worker.py index bfefa4a..a4e3509 100644 --- a/idatui/worker.py +++ b/idatui/worker.py @@ -78,13 +78,48 @@ def _ensure_tools_injected() -> None: sys.stderr.write(f"idatui: tool injection skipped: {e}\n") -def _open_and_register(binpath: str): +def _has_database(binpath: str) -> bool: + """Whether IDA already has a database for ``binpath``. + + IDA names it ``<file>.i64`` (keeping the extension), but a database made + from ``foo.bin`` can also appear as ``foo.i64`` depending on how it was + created — check both, because guessing wrong here means re-passing load + switches to an existing database, which fails the open. + """ + return (os.path.exists(binpath + ".i64") + or os.path.exists(os.path.splitext(binpath)[0] + ".i64")) + + +def _open_and_register(binpath: str, load_args: str = ""): """Open the DB (main thread) then import ida-pro-mcp so every @tool registers - against this live database. Returns (tools_dict, module_name, save_fn).""" + against this live database. Returns (tools_dict, module_name, save_fn). + + ``load_args`` is passed to IDA as command-line switches, which is the only + way to tell it how to read a headerless blob: a raw firmware image has no + format to detect, so without ``-p<processor>`` it loads as metapc at 0 and + finds nothing. Ignored once a database exists — the .i64 already records how + it was loaded, and re-passing conflicting switches is how you corrupt one. + """ _ensure_tools_injected() # before any ida_pro_mcp import import idapro idapro.enable_console_messages(False) - if idapro.open_database(binpath, run_auto_analysis=True): # nonzero == failure + args = load_args or None + if args and _has_database(binpath): + # The .i64 already records how this image was loaded. Passing the + # switches again on reopen makes IDA fail outright (rc != 0) — the load + # options belong to the FIRST open only. + 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 " @@ -109,8 +144,8 @@ def _open_and_register(binpath: str): return MCP_SERVER.tools.methods, module, save -def serve(sockpath: str, binpath: str) -> None: - tools, module, save = _open_and_register(binpath) +def serve(sockpath: str, binpath: str, load_args: str = "") -> None: + tools, module, save = _open_and_register(binpath, load_args) sid = uuid.uuid4().hex[:8] def dispatch(name: str, args: dict): @@ -179,10 +214,11 @@ def serve(sockpath: str, binpath: str) -> None: def main(argv=None) -> None: argv = argv if argv is not None else sys.argv[1:] if len(argv) < 2: - sys.stderr.write("usage: python -m idatui.worker <sock> <binary>\n") + sys.stderr.write( + "usage: python -m idatui.worker <sock> <binary> [ida-load-args]\n") raise SystemExit(2) try: - serve(argv[0], argv[1]) + serve(argv[0], argv[1], argv[2] if len(argv) > 2 else "") except SystemExit: raise except BaseException as e: # noqa: BLE001 -- surface a clean cause + code 1 diff --git a/idatui/worker_client.py b/idatui/worker_client.py index 586ff21..79a6db9 100644 --- a/idatui/worker_client.py +++ b/idatui/worker_client.py @@ -71,8 +71,9 @@ class _NoopKeepAlive: class WorkerClient: def __init__(self, binary_path: str, *, ttl: int = 0, - python: str | None = None) -> None: + python: str | None = None, load_args: str = "") -> None: self._bin = os.path.abspath(os.path.expanduser(binary_path)) + self._load_args = load_args or "" # IDA switches for a headerless blob self._python = python or _find_worker_python() tag = f"{os.getpid()}-{uuid.uuid4().hex[:8]}" self._sock_path = f"/tmp/idatui-worker-{tag}.sock" @@ -93,8 +94,11 @@ class WorkerClient: # run worker.py as a SCRIPT (not -m idatui.worker) so we don't # import the textual-dependent idatui package __init__ under the # IDA python, which usually has no textual. + argv = [self._python, _WORKER_PY, self._sock_path, self._bin] + if self._load_args: + argv.append(self._load_args) self._proc = subprocess.Popen( - [self._python, _WORKER_PY, self._sock_path, self._bin], + argv, stdout=open(self._log_path, "wb"), stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, ) |