diff options
Diffstat (limited to 'idatui')
| -rw-r--r-- | idatui/app.py | 485 | ||||
| -rw-r--r-- | idatui/pane.py | 22 | ||||
| -rw-r--r-- | idatui/rpc.py | 132 | ||||
| -rw-r--r-- | idatui/trace.py | 115 |
4 files changed, 712 insertions, 42 deletions
diff --git a/idatui/app.py b/idatui/app.py index 71cd9c4..1edc65e 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -98,6 +98,11 @@ _S_CURSOR = Style(bgcolor="#2a313c") _S_TRAIL_NOW = Style(bgcolor="#3f3410") _S_TRAIL_PAST = Style(bgcolor="#2b1c17") # warm: behind you _S_TRAIL_FUTURE = Style(bgcolor="#152230") # cool: ahead of you +#: Hex with a trace loaded: bytes the trace SAW at this timestamp vs bytes we're +#: still showing from the file. The distinction matters more than the values — +#: one is evidence, the other is an assumption. +_S_HEX_LIVE = Style(color="#9ece6a") +_S_HEX_STALE = Style(color="#5e6875") _S_DIM = Style(color="#7c8b9e", italic=True) _S_MATCH = Style(bgcolor="#7a5c00") # all search matches _S_MATCH_CUR = Style(bgcolor="#d0a215", color="#12161c") # the current match @@ -1561,6 +1566,10 @@ class HexView(ScrollView, can_focus=True): super().__init__(id="hex") self.model = None self.total = 0 + #: Trace to read memory from, and the timestamp to read it at. When set, + #: the dump shows what memory HELD then rather than what the file holds. + self.trace = None + self.trace_idx = 0 self._internal_top: int | None = None # scroll target we set ourselves # -- public API -------------------------------------------------------- # @@ -1747,6 +1756,15 @@ class HexView(ScrollView, can_focus=True): return Strip([Segment("".ljust(width), _S_HEX)]) va, data = model.row(r) cur_row, cur_col = self.cursor // 16, self.cursor % 16 + # With a trace loaded the row shows what memory HELD at the current + # timestamp, not what the file contains. Only the bytes the trace + # actually saw are overlaid: the rest stay the database's, dimmed, so + # you can always tell evidence from the file's idea of the world. + tmem = tknown = None + if self.trace is not None and data is not None: + tmem, tknown = self.trace.memory(va, len(data), self.trace_idx) + if not any(tknown): + tmem = tknown = None fo = model.file_offset(va) fo_str = f"{fo:08X}" if fo is not None else "--------" segs: list[Segment] = [ @@ -1761,15 +1779,29 @@ class HexView(ScrollView, can_focus=True): if i == 8: segs.append(Segment(" ", _S_HEX)) if i < n: - st = _S_CELL if (r == cur_row and i == cur_col) else _S_HEX - segs.append(Segment(f"{data[i]:02X} ", st)) + live = tknown is not None and tknown[i] + val = tmem[i] if live else data[i] + if r == cur_row and i == cur_col: + st = _S_CELL + elif tknown is None: + st = _S_HEX + else: + st = _S_HEX_LIVE if live else _S_HEX_STALE + segs.append(Segment(f"{val:02X} ", st)) else: segs.append(Segment(" ", _S_HEX)) segs.append(Segment(" |", _S_DIM)) for i in range(16): if i < n: - ch = chr(data[i]) if 32 <= data[i] < 127 else "." - st = _S_CELL if (r == cur_row and i == cur_col) else _S_ASCII + live = tknown is not None and tknown[i] + val = tmem[i] if live else data[i] + ch = chr(val) if 32 <= val < 127 else "." + if r == cur_row and i == cur_col: + st = _S_CELL + elif tknown is None: + st = _S_ASCII + else: + st = _S_HEX_LIVE if live else _S_HEX_STALE else: ch, st = " ", _S_ASCII segs.append(Segment(ch, st)) @@ -2309,6 +2341,87 @@ class HelpScreen(ModalScreen): self.dismiss(None) +class RegWriteScreen(ModalScreen): + """Registers, and the instruction that set each one. + + "Which instruction set this register to its current value?" is the question + a trace exists to answer, and seeking backwards to it is a single keypress + here rather than a manual walk. Forward is offered too, but backward is what + people actually want — you notice a bad value after it has been used. + """ + + BINDINGS = [ + Binding("escape", "close", "Close"), + Binding("down,ctrl+n", "cursor_down", show=False), + Binding("up,ctrl+p", "cursor_up", show=False), + Binding("enter", "choose", show=False, priority=True), + Binding("f", "choose_forward", show=False), + ] + + def __init__(self, rows, idx: int) -> None: + super().__init__() + self._rows = rows # (name, value, last_write, next_write) + self._idx = idx + + def compose(self) -> ComposeResult: + with Vertical(id="pal-box"): + yield Static(f" registers at t={self._idx:,} \u2014 Enter seeks to the " + f"write, f seeks forward", id="pal-title", markup=False) + yield OptionList(id="pal-list") + + def on_mount(self) -> None: + ol = self.query_one(OptionList) + opts = [] + for name, val, last, nxt in self._rows: + label = Text() + label.append(f" {name:>4} ", _S_MNEM) + label.append(f"{val:#018x} " if val > 0xFFFFFFFF else f"{val:#010x} ", + _S_INSN) + if last is None: + label.append("never written in this trace", _S_DIM) + elif last == self._idx: + label.append("set by THIS instruction", _S_DATA) + else: + label.append(f"set at t={last:,}", _S_LABEL) + label.append(f" ({self._idx - last:,} steps back)", _S_DIM) + if nxt is not None: + label.append(f" next t={nxt:,}", _S_DIM) + opts.append(Option(label)) + ol.add_options(opts) + ol.highlighted = 0 + ol.focus() + + def 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 _pick(self, forward: bool) -> None: + i = self.query_one(OptionList).highlighted + if i is None or not (0 <= i < len(self._rows)): + self.dismiss(None) + return + _name, _val, last, nxt = self._rows[i] + self.dismiss(nxt if forward else last) + + def action_choose(self) -> None: + self._pick(False) + + def action_choose_forward(self) -> None: + self._pick(True) + + def on_option_list_option_selected(self, event) -> None: # type: ignore[no-untyped-def] + self._pick(False) + + def action_close(self) -> None: + self.dismiss(None) + + class TraceDock(Vertical): """Registers and a timeline for the loaded execution trace, docked right. @@ -2325,6 +2438,7 @@ class TraceDock(Vertical): def compose(self) -> ComposeResult: yield Static("", id="trace-head", markup=False) yield Static("", id="trace-regs", markup=False) + yield Static("", id="trace-stack", markup=False) yield TraceTimeline(id="trace-timeline") def show(self, trace, idx: int) -> None: @@ -2367,11 +2481,49 @@ class TraceDock(Vertical): body.append(f"{v:#018x}\n" if v > 0xFFFFFFFF else f"{v:#010x}\n", _S_DATA if hot else (_S_LABEL if name == pc else _S_INSN)) self.query_one("#trace-regs", Static).update(body) + self._render_stack(t) tl = self.query_one(TraceTimeline) tl.idx = self.idx tl.refresh() + STACK_WORDS = 8 + + def _render_stack(self, t) -> None: # type: ignore[no-untyped-def] + """The stack as of this instant, read out of the trace. + + This is where a trace's memory actually is: on two real traces, NONE of + the accesses fell inside the image — every one was stack or heap. A + memory view that could only address the image would have nothing to show. + + Bytes the trace never saw are printed as '??' rather than zeros. A trace + knows what it observed and nothing else, and quietly rendering unseen + memory as zero would invent facts. + """ + sp_name = next((r for r in ("rsp", "esp", "sp") if r in t.reg_at), "") + sp = t.register(sp_name, self.idx) if sp_name else None + out = Text() + if sp is None: + self.query_one("#trace-stack", Static).update(out) + return + width = 8 if (t.info and "64" in (t.info.arch or "")) else 4 + out.append(f" stack ({sp_name})\n", _S_DIM) + for k in range(self.STACK_WORDS): + a = sp + k * width + data, known = t.memory_raw(a, width, self.idx) + out.append(" \u25b8" if k == 0 else " ", _S_MNEM) + out.append(f"{a:012x} ", _S_ADDR) + if all(known): + v = int.from_bytes(data, "little") + out.append(f"{v:0{width * 2}x}\n", _S_DATA if k == 0 else _S_INSN) + elif any(known): + out.append("".join(f"{b:02x}" if known[i] else "??" + for i, b in enumerate(data)) + "\n", _S_INSN) + else: + out.append("?" * (width * 2) + "\n", _S_SEP) + self.query_one("#trace-stack", Static).update(out) + + class TraceTimeline(Static): """The trace as a vertical bar: where you are, and where you've been. @@ -3134,7 +3286,7 @@ class IdaTui(App): #xref-list { height: auto; max-height: 100%; } /* every #pal-box palette centres, not just the symbol one */ SymbolPalette, StringsPalette, ProjectPalette, - LoadOptionsScreen { align: center middle; } + LoadOptionsScreen, RegWriteScreen { 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; } @@ -3147,6 +3299,7 @@ class IdaTui(App): #trace-dock { dock: right; width: 34; background: $surface; border-left: solid $panel; } #trace-head { height: 2; padding: 0 1; background: $panel; } #trace-regs { height: auto; padding: 1 0 0 0; } + #trace-stack { height: auto; padding: 1 0 0 0; } #trace-timeline { height: 1fr; padding: 1 0 0 0; } #load-note { height: 2; padding: 1 1 0 1; color: $text-muted; } /* Cap the processor list so the ADDRESS FIELD is always on screen: with the @@ -3195,6 +3348,12 @@ class IdaTui(App): Binding("ctrl+l", "load_options", "Reload as…", show=False), # Trace stepping. ] / [ move one instruction, } / { step over a # call by following the stack pointer. + # Seeking, as opposed to stepping: jump to the next/previous time THIS + # thing was touched, where "this thing" is whatever the focused view + # addresses — an instruction in the code views, a byte in hex. + Binding("greater_than_sign", "seek_next_hit", "Next hit", show=False), + Binding("less_than_sign", "seek_prev_hit", "Prev hit", show=False), + Binding("W", "seek_reg_write", "Reg writes", show=False), Binding("right_square_bracket", "step_fwd", "Step", show=False), Binding("left_square_bracket", "step_back", "Step back", show=False), Binding("right_curly_bracket", "step_over_fwd", "Step over", show=False), @@ -3242,10 +3401,15 @@ class IdaTui(App): self._open_path = open_path self._ttl = ttl self._load_args = load_args or "" # IDA switches for a headerless blob + self._title = (os.path.basename(open_path) if open_path else "") self._trace_path = trace_path or "" # Tenet execution trace to explore self._trace = None # the loaded Trace, once analysed self._trail_map = [] # decomp_map for _trail_map_ea self._trail_map_ea = None + self._pending_trace_line = None # step waiting on a re-decompile + self._trail_line_of: dict[int, int] = {} # ea -> pseudocode line + self._trail_span = None # ea span of that function + self._trail_eas: list[int] = [] # sorted keys of _trail_line_of self._t = 0 # current timestamp in that trace self._do_keepalive = keepalive self._rpc_path = rpc_path @@ -3576,8 +3740,12 @@ class IdaTui(App): self._flash_until = _time.monotonic() + 8.0 elif self._flash and _time.monotonic() < self._flash_until: text = self._flash - if self._binary: # project mode: always say which binary you're in - text = f"[{self._binary}] {text}" + # Always say WHICH file this is. In project mode that's the binary's + # label; otherwise the filename we opened. Cheap on purpose — _module() + # asks the worker, and this runs on every status write. + tag = self._binary or self._title + if tag: + text = f"[{tag}] {text}" # An image with no functions at all is nearly always a blob described # wrongly, and that stays true as you scroll around — so it belongs in # the status bar, not in a one-off message the next write clobbers. @@ -3723,6 +3891,7 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged + self._title = os.path.basename(self._open_path) return client if not self._open_path: self.app.call_from_thread( @@ -3745,7 +3914,6 @@ class IdaTui(App): self._func_index = idx self.app.call_from_thread(lambda: self.query_one("#func-table", DataTable).clear()) last = 0 - module = self._module() while not idx.complete: idx.load_next_page() rows = idx.window(last, len(idx) - last) @@ -3753,14 +3921,14 @@ class IdaTui(App): if rows: self.app.call_from_thread(self._append_rows, rows) self.app.call_from_thread( - self._status, f"{module} — {last} functions…" + self._status, f"{last} functions…" ) # If a filter is active (typed during load), re-apply it over the full set. if self._filter_term: self.app.call_from_thread(self._apply_filter, self._filter_term) else: self.app.call_from_thread( - self._status, f"{module} — {len(idx)} functions (Ctrl+N: find symbol)") + self._status, f"{len(idx)} functions (Ctrl+N: find symbol)") # Land somewhere useful instead of an empty pane: main() if present, # otherwise pop the fuzzy symbol picker. self.app.call_from_thread(self._auto_land) @@ -3866,7 +4034,19 @@ class IdaTui(App): if fn is not None: self._open_function(fn.addr, fn.name) elif len(self._func_index): - self.action_symbols() + # No main(): land on the first function rather than pushing the + # symbol palette. A modal as the *startup* state leaves a human + # staring at a picker over an empty pane, and silently swallows + # every keystroke an RPC driver injects while `ping` still says + # ready:true. Landing somewhere real is better for both; Ctrl+N is + # one keypress away. + first = self._func_index.get(0) + if first is not None: + self._open_function(first.addr, first.name) + self._status(f"no entry function — opened {first.name} " + "(Ctrl+N: find symbol)") + else: + self.action_symbols() else: self._land_without_functions() @@ -3977,7 +4157,7 @@ class IdaTui(App): if term: self._status(f"filter '{term}': {len(matched)}/{total}") else: - self._status(f"{self._module()} — {total} functions") + self._status(f"{total} functions") def _apply_pending_filter(self) -> None: self._filter_timer = None @@ -4179,6 +4359,7 @@ class IdaTui(App): self._binary = label self._pool.set_active(label) self._open_path = self._project.by_label(label).staged + self._title = os.path.basename(self._open_path) self._active = st.active if st else "listing" self._split = st.split if st else False self._filter_term = st.filter_term if st else "" @@ -5299,12 +5480,38 @@ class IdaTui(App): return # The label shows in the listing's head rows -> invalidate + reopen. self.program.bump_items() + # Naming the address *of a function start* is a function rename by any + # other name. Without this the cached index kept the old name, so + # `functions`/`names`/resolve/the palette all reported the rename had + # not happened -- and a driver that trusts those readbacks redoes work + # it already did. + fn = None + try: + fn = self.program.function_of(addr) + except Exception: # noqa: BLE001 + fn = None + is_func_start = fn is not None and fn.addr == addr lm = self.program.listing(addr) - label = self.program.region_label(addr) + label = name if is_func_start else self.program.region_label(addr) idx = max(lm.ensure_ea(addr), 0) if lm is not None else 0 - self.app.call_from_thread(self._open_at_named, label, addr, idx, name) + self.app.call_from_thread(self._open_at_named, label, addr, idx, name, + is_func_start) - def _open_at_named(self, label: str, addr: int, idx: int, name: str) -> None: + def _open_at_named(self, label: str, addr: int, idx: int, name: str, + is_func_start: bool = False) -> None: + if is_func_start: + self.program.bump_names() + if self._func_index is not None: + self._func_index.update_name(addr, name) + for e in self._nav: + if e.ea == addr: + e.name = name + try: + table = self.query_one("#func-table", DataTable) + name_col = list(table.columns.keys())[1] + table.update_cell(str(addr), name_col, name) + except Exception: # noqa: BLE001 -- row filtered out / not streamed + pass self._open_at(addr, label, idx, False, -1, 0, True) self._dirty = True self._status(f"named {addr:#x} → {name} (Ctrl+S to save)") @@ -5485,11 +5692,99 @@ class IdaTui(App): t = self._trace if t is None or not t.length: return + # A seek invalidates any navigation still in flight. They run in workers + # and finish out of order: the trace's opening seek lands on the entry + # point, takes a while, and used to arrive AFTER later seeks — dragging + # the cursor back to _start while the trace was elsewhere, permanently. + # + # Bumped HERE and not in _goto_ea. Doing it for every navigation is the + # more general rule ("the last thing you asked for wins") but it also + # lets an ordinary follow be dropped by whatever navigates next, and the + # only evidence I have is about seeks. Narrow fix for the measured bug. + self._nav_seq += 1 self._t = max(0, min(int(idx), t.length - 1)) self.query_one(TraceDock).show(t, self._t) self._paint_trail() - if follow: - self._goto_ea(t.ip(self._t), push=False) + if not follow: + return + pc = t.ip(self._t) + if self._split and self._seek_split(pc): + return + # Stay in whichever view you're reading. Without prefer_decomp a step + # from the pseudocode navigates to an address, which opens the listing — + # so stepping through C threw you out of C on the first keypress. + self._goto_ea(pc, push=False, + prefer_decomp=(self._active == "decomp")) + + def _seek_split(self, pc: int) -> bool: + """Put BOTH panes on ``pc``. True if handled. + + Normal navigation moves one pane and gives the companion a band, never a + cursor — that rule exists so the two can't chase each other. A trace step + isn't navigation though: time is a single global position, and both views + are showing the same instant, so both cursors belong on it. + + The scroll anchoring is unchanged: after placing the cursors, the usual + _sync_split still bands the companion and aligns it to the driver's + screen row, so the eye tracks straight across. + """ + lst = self.query_one(ListingView) + dec = self.query_one(DecompView) + if lst.model is None: + return False + row = lst.model.ensure_ea(pc) + if row is None or row < 0: + return False # not in this listing (other segment): full nav + lst.cursor = row + lst._scroll_cursor_into_view() + + # Has execution actually left the decompiled function? Ask the map the + # trail painting keeps, which is keyed to what the decompiler currently + # HOLDS. _split_range comes from the guarded async path and lags, so a + # stale one made every step look like a function change: the decompiler + # bounced main -> PLT stub -> main, each bounce costing a synchronous + # 769-line map fetch on the UI thread. + span = self._trail_span + inside = (pc in self._trail_line_of + or (span is not None and span[0] <= pc <= span[1])) + if not inside: + self._pending_trace_line = pc + self._resync_decomp_async(pc) + return True + self._place_decomp_at(pc) + self._sync_split(self._active) + return True + + def _place_decomp_at(self, pc: int) -> None: + """Move the pseudocode cursor to the line covering ``pc``. + + Uses the map the trail painting already keeps (keyed to the decompiler's + CURRENTLY loaded function), not the split view's _split_ea2line. That one + is refreshed by a guarded async path — it drops a result if _cur moved + while it was in flight — and a burst of steps moves _cur constantly, so + during stepping it is frequently a map of the function you just left. + """ + dec = self.query_one(DecompView) + line = None + if self._trail_map_ea == dec.loaded_ea and self._trail_line_of: + # EXACT match only. The decompiler doesn't attribute every + # instruction to a line (about half of main's aren't), and the + # tempting fallback — the nearest mapped instruction at or before + # the pc — is unsound: C lines are not monotonic in address, so + # 0x24a8 early in main resolved to line 708, "sub_2040();", near the + # end. A cursor that jumps to an unrelated statement is worse than + # one that waits; the trail still marks where we are. + line = self._trail_line_of.get(pc) + if line is None: + line = self._split_ea2line.get(pc) + if line is None: + line = dec.line_for_ea(pc) + if line is not None: + dec.goto(line, dec.cursor_x) + + @work(thread=True, exclusive=True, group="split-resync") + def _resync_decomp_async(self, ea: int) -> None: + self._resync_decomp(ea) def _paint_trail(self) -> None: """Push the execution trail into the code views. @@ -5500,6 +5795,13 @@ class IdaTui(App): t = self._trace if t is None: return + try: + hx = self.query_one(HexView) + hx.trace, hx.trace_idx = t, self._t + if hx.display: + hx.refresh() + except Exception: # noqa: BLE001 -- not mounted yet + pass trail = t.trail(self._t) try: lst = self.query_one(ListingView) @@ -5530,13 +5832,16 @@ class IdaTui(App): dec.trail = {} return if self._trail_map_ea != ea: - # Cached per function: decomp_map is an RPC and stepping is - # interactive, so paying it per keystroke would be felt. + # One index, built once per decompiled function and shared with the + # split view (_apply_split_map fills the same fields). decomp_map is + # an RPC and stepping is interactive, so paying it per keystroke — + # or twice, once for each of two parallel maps — would be felt. try: - self._trail_map = self.program.decomp_map(ea) + self._apply_split_map(ea, self.program.decomp_map(ea)) except Exception: # noqa: BLE001 - self._trail_map = [] - self._trail_map_ea = ea + self._trail_map, self._trail_map_ea = [], ea + self._trail_line_of, self._trail_eas = {}, [] + self._trail_span = None rank = {"future": 0, "past": 1, "now": 2} lines: dict[int, str] = {} for i, eas in enumerate(self._trail_map or []): @@ -5549,6 +5854,12 @@ class IdaTui(App): lines[i] = best dec.trail = lines dec.refresh() + pend, self._pending_trace_line = self._pending_trace_line, None + if pend is not None and self._split: + # The function was still decompiling when the step happened; land + # now that its line map exists. + self._place_decomp_at(pend) + self._sync_split(self._active) def _step(self, delta: int) -> None: if self._trace is None: @@ -5556,6 +5867,97 @@ class IdaTui(App): return self._seek(self._t + delta) + def _seek_hit(self, direction: int) -> None: + """Seek to the next/previous time the focused view's subject was touched. + + Two different questions with one pair of keys, because the answer to + "which thing?" is already on screen: in a code view it's the instruction + under the cursor ("when else did this run?"), in hex it's the byte under + the cursor ("who else touched this?"). + """ + t = self._trace + if t is None: + self._status("no trace loaded (--trace FILE)") + return + if self._active == "hex": + hx = self._try_view(HexView) + va = hx.cursor_va() if hx is not None else None + if va is None: + return + stamps = t.memory_accesses(va, 1) + what = f"access to {va:#x}" + else: + view = self._active_code_view() + if isinstance(view, DecompView): + # A C line is not one address, so ask about the whole statement: + # "when else did this line run?" is the question, and it's the + # union of its instructions' executions. Falling back to the + # line's single /*ea*/ marker would answer a narrower question + # and often no question at all, since most lines have no marker. + line = view.cursor + eas = [] + if (self._trail_map_ea == view.loaded_ea + and 0 <= line < len(self._trail_map or [])): + eas = list(self._trail_map[line]) + if not eas: + one = view._line_ea(line) + eas = [one] if one is not None else [] + if not eas: + self._status("this line has no instructions to seek on", + priority=True) + return + stamps = sorted({x for e in eas for x in t.executions(e)}) + what = f"execution of C line {line + 1}" + else: + ea = view._cursor_ea() if view is not None else None + if ea is None: + self._status("no address on this line", priority=True) + return + stamps = list(t.executions(ea)) + what = f"execution of {ea:#x}" + if not stamps: + self._status(f"no {what} in this trace", priority=True) + return + import bisect as _b + if direction > 0: + i = _b.bisect_right(stamps, self._t) + else: + i = _b.bisect_left(stamps, self._t) - 1 + if not (0 <= i < len(stamps)): + edge = "last" if direction > 0 else "first" + self._status(f"already at the {edge} {what} " + f"({len(stamps)} in the trace)", priority=True) + return + self._seek(stamps[i]) + self._status(f"{what}: {i + 1} of {len(stamps)} @ t={stamps[i]:,}", + priority=True) + + def action_seek_next_hit(self) -> None: + self._seek_hit(1) + + def action_seek_prev_hit(self) -> None: + self._seek_hit(-1) + + def action_seek_reg_write(self) -> None: + """W: which instruction set each register to its current value.""" + t = self._trace + if t is None: + self._status("no trace loaded (--trace FILE)") + return + rows = [] + for name in t.registers: + v = t.register(name, self._t) + if v is None: + continue + rows.append((name, v, t.last_write(name, self._t), + t.next_write(name, self._t))) + if rows: + self.push_screen(RegWriteScreen(rows, self._t), self._on_reg_write_chosen) + + def _on_reg_write_chosen(self, idx) -> None: # type: ignore[no-untyped-def] + if idx is not None: + self._seek(int(idx)) + def action_step_fwd(self) -> None: self._step(1) @@ -5675,7 +6077,8 @@ class IdaTui(App): idx = max(lm.ensure_ea(ea), 0) if lm is not None else 0 name = fn.name if fn is not None else self.program.region_label(ea) self.app.call_from_thread( - self._open_at, ea, name, idx, push, -1, 0, fn is None, focus_name) + self._open_at_if_current, seq, ea, name, idx, push, fn is None, + focus_name) def _open_decomp_entry(self, fn_addr: int, fn_name: str, dec_idx: int, dec_cursor_x: int, push: bool, @@ -5848,6 +6251,18 @@ class IdaTui(App): cur = row_of(fallback_ea) return (cur, row_of(a.top_ea)) + def _open_at_if_current(self, seq: int, ea: int, name: str, cursor: int, + push: bool, is_region: bool, + focus_name: str | None) -> None: + """Apply a navigation result only if it's still the one being awaited. + + The decompiler path has had this since 756589a; the listing path hadn't, + so a slow navigation could still land on top of a newer one. + """ + if seq != self._nav_seq: + return + self._open_at(ea, name, cursor, push, -1, 0, is_region, focus_name) + def _open_at(self, ea: int, name: str, cursor: int, push: bool, dec_cursor: int = -1, dec_cursor_x: int = 0, is_region: bool = False, focus_name: str | None = None, @@ -6461,8 +6876,18 @@ class IdaTui(App): self.app.call_from_thread(self._apply_split_map, ea, m) def _apply_split_map(self, ea: int, m: list) -> None: - if not self._split or self._cur is None or self._cur.ea != ea: - return # left split / navigated away + """Index the per-line instruction map for the decompiled function. + + Keyed to what the DECOMPILER holds, not to _cur, and not conditional on + split being on. The old guard dropped the result whenever _cur had moved + while the fetch was in flight — during trace stepping that is almost + always — leaving the split view working from the map of the function you + just left. _cur follows the cursor; this map describes the pseudocode on + screen, and those are different things. + """ + dec = self._try_view(DecompView) + if dec is not None and dec.loaded_ea is not None and ea != dec.loaded_ea: + return # a stale fetch for a function we no longer show self._split_eamap = m self._split_ea2line = {} alleas = [] @@ -6473,7 +6898,15 @@ class IdaTui(App): # ea span of the decompiled function: when the listing cursor leaves it, # _sync_split re-points the decomp to the function under the cursor. self._split_range = (min(alleas), max(alleas)) if alleas else None - self._sync_split(self._active) # re-link with the region map + # ONE index, shared with the trace path: it used to keep a parallel copy + # of exactly this, fetched separately and keyed differently, which is how + # the two ended up describing different functions. + self._trail_map, self._trail_map_ea = m, ea + self._trail_line_of = dict(self._split_ea2line) + self._trail_eas = sorted(self._trail_line_of) + self._trail_span = self._split_range + if self._split: + self._sync_split(self._active) # re-link with the region map def on_decomp_view_cursor_moved(self, msg: DecompView.CursorMoved) -> None: dv = self._try_view(DecompView) diff --git a/idatui/pane.py b/idatui/pane.py index 1a46a4f..37a5f0c 100644 --- a/idatui/pane.py +++ b/idatui/pane.py @@ -248,18 +248,29 @@ def stop(args) -> int: if not rows: print("error: no matching pane (need --sock or --pane)", file=sys.stderr) return 2 + killed: list[str] = [] for r in rows: sock, pane = r.get("sock"), r.get("pane") + quit_ok = False if sock and os.path.exists(sock): try: # ask it to quit gracefully first with RpcClient(sock) as c: c.call("quit") - time.sleep(0.4) + quit_ok = True except (OSError, RpcError, ConnectionError): pass + # Wait for the pane to actually go away. Quitting runs App.on_unmount, + # which writes every dirty database; a 90 MB .i64 takes tens of seconds. + # Killing the pane on a fixed short sleep truncated that save and + # silently destroyed the session's work, so block on the real signal. + if pane and quit_ok: + deadline = time.monotonic() + float(args.timeout) + while time.monotonic() < deadline and _pane_alive(pane): + time.sleep(0.25) if pane and _pane_alive(pane): subprocess.run(["tmux", "kill-pane", "-t", pane], capture_output=True) + killed.append(pane) if sock: try: os.unlink(sock) @@ -271,6 +282,12 @@ def stop(args) -> int: out = {"stopped": [r.get("sock") or r.get("pane") for r in rows]} if reaped: out["reaped_workers"] = reaped + if killed: + # Only ever reached on timeout: say so, because it means a save may have + # been cut short rather than "clean teardown". + out["force_killed"] = killed + out["warning"] = (f"pane(s) did not exit within {args.timeout}s and were " + "killed; unsaved database changes may be lost") print(json.dumps(out)) return 0 @@ -335,6 +352,9 @@ def main(argv: list[str]) -> int: st = sub.add_parser("stop", help="graceful quit + kill the pane") st.add_argument("--sock") st.add_argument("--pane") + st.add_argument("--timeout", type=float, default=600.0, + help="seconds to wait for the pane to exit (it saves dirty " + "databases on the way out) before force-killing it") st.set_defaults(fn=stop) ls = sub.add_parser("list", help="list tracked panes") diff --git a/idatui/rpc.py b/idatui/rpc.py index 7741f48..717f4df 100644 --- a/idatui/rpc.py +++ b/idatui/rpc.py @@ -59,7 +59,7 @@ METHODS = { "text": "{text,delay_ms?,settle?} type a literal string into the focused input", "goto/open": "{target,delay_ms?} g-prompt to a name or 0xADDR", "rename": "{name,word?,delay_ms?} rename the token under (or 'word') the cursor", - "comment": "{text,delay_ms?} comment the current line", + "comment": "{text} comment the current line (use \\n for newlines)", "retype": "{proto,word?,delay_ms?} set the prototype/type under the cursor", "follow": "{word?} follow the reference under (or 'word') the cursor", "cursor_on": "{word,line?,occurrence?=1} place the cursor on a token", @@ -104,6 +104,33 @@ def _active_widget(app): _MODALS = ("XrefsScreen", "SymbolPalette", "StructEditor", "ConfirmScreen") +#: ``drive raw`` (and any k=v CLI) hands every param through as a *string*. +#: Handlers that did ``int(...)`` coped; the ones that compared directly blew up +#: with e.g. "'<' not supported between instances of 'int' and 'str'". Coerce the +#: known-numeric names once, centrally, instead of at every call site. +_INT_PARAMS = ("lines", "limit", "max", "n", "index", "line", "col", + "occurrence", "delay_ms", "direction", "addr", "count") +_FLOAT_PARAMS = ("timeout",) + + +def _coerce_params(params: dict[str, Any]) -> dict[str, Any]: + out = dict(params) + for k in _INT_PARAMS: + v = out.get(k) + if isinstance(v, str) and v.strip(): + try: + out[k] = int(v, 0) + except ValueError: + pass + for k in _FLOAT_PARAMS: + v = out.get(k) + if isinstance(v, str) and v.strip(): + try: + out[k] = float(v) + except ValueError: + pass + return out + def _modal_snapshot(app) -> dict[str, Any] | None: """Describe the top modal screen, if any, enough to drive it.""" @@ -148,6 +175,14 @@ def _cursor_info(app, w) -> dict[str, Any]: "scroll_y": round(w.scroll_offset.y)} +def _where(app) -> str: + """Short 'name @ 0xea' for error messages that need to say where we ended up.""" + cur = getattr(app, "_cur", None) + if cur is None: + return "nowhere" + return f"{getattr(cur, 'name', '?')} @ {getattr(cur, 'ea', 0):#x}" + + def _readiness(app) -> dict[str, Any]: """Whether the app is drivable yet, and how far function-loading has got. (Cheap: no network — never call client.health() here.)""" @@ -452,9 +487,15 @@ class RpcServer: return {"id": rid, "error": {"message": f"{type(e).__name__}: {msg}"}} # -- composed helpers (semantic verbs) -------------------------------- # - async def _press(self, keys, pred=None, timeout=20.0): + async def _press(self, keys, pred=None, timeout=20.0, what=""): await self.app._press_keys([str(k) for k in keys]) - await settle(self.app, pred, timeout=timeout) + ok = await settle(self.app, pred, timeout=timeout) + if pred is not None and not ok: + # Never report success for an action that did not happen: the caller + # would go on to edit whatever the *previous* location was. + raise TimeoutError( + f"{what or 'action'} did not complete within {timeout}s " + f"(still at {_where(self.app)}); retry with a larger timeout=") return snapshot(self.app) async def _fill_prompt(self, open_key, input_id, value, delay_ms, clear): @@ -466,7 +507,14 @@ class RpcServer: await settle(app, lambda: app.query_one(f"#{input_id}", Input).display, timeout=10) inp = app.query_one(f"#{input_id}", Input) if not inp.display: - raise RuntimeError(f"{input_id!r} prompt did not open (word under cursor?)") + # Say *why*. The old message always blamed the word under the cursor, + # which sent readers hunting for a cursor problem when the real cause + # was usually a modal eating the opening keystroke. + modal = type(app.screen).__name__ + why = (f"modal {modal!r} has focus and ate the {open_key!r} keystroke" + if modal in _MODALS or modal != "Screen" + else "no renameable token under the cursor") + raise RuntimeError(f"{input_id!r} prompt did not open: {why}") if clear: inp.value = "" await app._press_keys(_text_to_keys(value, delay_ms)) @@ -485,8 +533,32 @@ class RpcServer: want = fn.addr if fn else ea return lambda: app._cur is not None and app._cur.ea == want + #: Verbs that drive the *main* app by injecting keystrokes. If a modal is on + #: top it eats those keys, so they must refuse rather than silently no-op. + _NEEDS_NO_MODAL = { + "goto", "open", "rename", "comment", "retype", "follow", "back", + "toggle_view", "hex", "save", "search", "move", "cursor", "cursor_on", + } + #: Modals the driver is expected to interact with (they have their own verbs). + _DRIVABLE_MODALS = {"XrefsScreen", "SymbolPalette", "StructEditor", + "ProjectPalette", "QuitScreen"} + + def _modal_kind(self) -> str | None: + scr = self.app.screen + name = type(scr).__name__ + return name if name in _MODALS or name in self._DRIVABLE_MODALS else None + async def _dispatch(self, method: str, params: dict[str, Any]) -> Any: app = self.app + params = _coerce_params(params) + if method in self._NEEDS_NO_MODAL: + modal = self._modal_kind() + if modal is not None: + raise RuntimeError( + f"modal {modal!r} is on top and will swallow this verb's " + f"keystrokes; dismiss it first (close) or use its own verb " + f"(select/symbols/xrefs). Note: a binary with no entry " + f"function can land in the symbol palette on startup.") if method in (None, "ping"): module = None try: @@ -498,9 +570,25 @@ class RpcServer: if method == "methods": return METHODS if method == "quit": + # Route through the same teardown a human gets, so a dirty database + # is written instead of dropped. `app.exit()` alone skips the dirty + # check entirely, and the caller (pane stop) then kills the pane -- + # which used to destroy a whole session's annotations. + dirty = list(app._dirty_labels()) + save = params.get("save", True) + if isinstance(save, str): + save = save.lower() not in ("0", "false", "no", "") + + def _go(): + if dirty and save: + app._on_quit_choice("save") # saves, then exits + else: + app._on_quit_choice("discard") + # answer first, then tear down (so this response still gets written) - asyncio.get_running_loop().call_later(0.2, app.exit) - return {"ok": True, "quitting": True} + asyncio.get_running_loop().call_later(0.2, _go) + return {"ok": True, "quitting": True, "saving": bool(dirty and save), + "dirty": dirty} if method in _PROGRAM_METHODS and app.program is None: raise ValueError("not ready: still connecting / loading functions") @@ -642,7 +730,15 @@ class RpcServer: target = str(params.get("target", "")) pred = self._goto_target_pred(target) await self._fill_prompt("g", "goto", target, delay, clear=False) - await settle(app, pred, timeout=timeout) + ok = await settle(app, pred, timeout=timeout) + if pred is not None and not ok: + # A goto that silently "succeeds" without moving is worse than an + # error: on a big database the listing build can outrun the + # default timeout, and every subsequent rename/comment then lands + # on the function the caller *used* to be looking at. + raise TimeoutError( + f"goto {target!r} did not land within {timeout}s " + f"(still at {_where(app)}); retry with a larger timeout=") return snapshot(app) if method == "rename": @@ -650,7 +746,13 @@ class RpcServer: await settle(app, timeout=timeout) return snapshot(app) if method == "comment": - await self._fill_prompt("semicolon", "comment", str(params["text"]), delay, + # Comments can be long; skip the per-char delay so the agent isn't + # blocked for seconds watching the typing animation. Also: the + # prompt is single-line, so literal newlines (0x0a) get swallowed by + # the Input widget. The app's _do_comment converts the two-char + # sequence '\n' into a real newline for IDA, so we escape here. + ctext = str(params["text"]).replace("\n", "\\n") + await self._fill_prompt("semicolon", "comment", ctext, 0, clear=True) await settle(app, timeout=timeout) return snapshot(app) @@ -661,7 +763,8 @@ class RpcServer: if method == "follow": depth = len(app._nav) - return await self._press(["enter"], lambda: len(app._nav) > depth, timeout) + return await self._press(["enter"], lambda: len(app._nav) > depth, + timeout, "follow") if method == "back": return await self._press(["escape"], timeout=timeout) if method == "toggle_view": @@ -685,12 +788,14 @@ class RpcServer: return False return False - return await self._press(["tab"], _toggled, timeout) + return await self._press(["tab"], _toggled, timeout, "toggle_view") if method == "hex": - return await self._press(["backslash"], lambda: app._active == "hex", timeout) + return await self._press(["backslash"], lambda: app._active == "hex", + timeout, "hex") if method == "xrefs": return await self._press( - ["x"], lambda: type(app.screen).__name__ == "XrefsScreen", timeout) + ["x"], lambda: type(app.screen).__name__ == "XrefsScreen", + timeout, "xrefs") if method == "symbols": await app._press_keys(["ctrl+n"]) await settle(app, lambda: type(app.screen).__name__ == "SymbolPalette", timeout=10) @@ -701,7 +806,8 @@ class RpcServer: return snapshot(app) if method == "structs": return await self._press( - ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor", timeout) + ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor", + timeout, "structs") if method == "close": return await self._press(["escape"], timeout=timeout) if method == "save": diff --git a/idatui/trace.py b/idatui/trace.py index 2afc8f6..931f918 100644 --- a/idatui/trace.py +++ b/idatui/trace.py @@ -76,7 +76,14 @@ class TraceInfo: @dataclass class MemOp: - """One memory access made by one instruction.""" + """One memory access made by one instruction. + + ``addr`` is the address the TRACE recorded — not slid onto the database. + Most accesses are stack or heap, which have no counterpart in the database + at all, and applying the image's relocation to a stack pointer produces a + nonsense address (it went negative in testing). Only addresses inside the + image can be translated, and the caller knows when that applies. + """ addr: int data: bytes @@ -116,6 +123,10 @@ class Trace: mem_row: array.array = field(default_factory=lambda: array.array("I")) #: applied so trace addresses line up with the database (see ``rebase``). slide: int = 0 + #: accesses ordered by address, built on first memory query. + _mem_order: list | None = None + _mem_starts: list = field(default_factory=list) + _mem_maxlen: int = 0 # -- construction ------------------------------------------------------ # @classmethod @@ -266,11 +277,111 @@ class Trace: out = [] for k in range(lo, hi): off, ln = self.mem_off[k], self.mem_len[k] - out.append(MemOp(addr=self.mem_addr[k] + self.slide, + out.append(MemOp(addr=self.mem_addr[k], data=bytes(self.mem_blob[off:off + ln]), write=bool(self.mem_write[k]))) return out + # -- memory state ------------------------------------------------------- # + def _mem_index(self) -> None: + """Order the accesses by address, once. + + Queries ask "what was at this window at time t", so the accesses that + matter are the few touching that window — not the tens of thousands in + the trace. Sorting by address makes those a bisect away; sorting by time + (the order they arrive in) would mean scanning everything per repaint. + """ + if self._mem_order is not None: + return + order = sorted(range(len(self.mem_addr)), key=lambda k: self.mem_addr[k]) + self._mem_order = order + self._mem_starts = [self.mem_addr[k] for k in order] + self._mem_maxlen = max(self.mem_len) if len(self.mem_len) else 0 + + def memory_raw(self, addr: int, length: int, + idx: int | None = None) -> tuple[bytes, bytes]: + """Memory at a TRACE address (no slide). + + The stack lives here. Measured on two real traces, 0% of memory accesses + fall inside the image — every one is stack or heap — so a query that + insists on database addresses can't answer the question anyone actually + has about memory in a trace. + """ + return self.memory(addr + self.slide, length, idx) + + def memory(self, addr: int, length: int, + idx: int | None = None) -> tuple[bytes, bytes]: + """``(data, known)`` for ``length`` bytes at ``addr`` as of ``idx``. + + ``known`` is a byte-per-byte mask: a trace only says what it saw, so a + byte nobody read or wrote is genuinely unknown and must not be drawn as + zero. That distinction is the whole value of reading memory from a trace + rather than from the database — the database has the file's bytes, the + trace has what was actually there at that instant. + + Reads count as evidence, not just writes: an instruction reading a byte + reveals what it held then. + """ + if idx is None: + idx = self.length - 1 + length = max(int(length), 0) + out, known = bytearray(length), bytearray(length) + if not length or not len(self.mem_idx): + return bytes(out), bytes(known) + self._mem_index() + raw = addr - self.slide + best = [-1] * length + import bisect as _b + lo = _b.bisect_left(self._mem_starts, raw - self._mem_maxlen) + hi = _b.bisect_right(self._mem_starts, raw + length - 1) + for pos in range(lo, hi): + k = self._mem_order[pos] + t = self.mem_idx[k] + if t > idx: + continue + a, ln = self.mem_addr[k], self.mem_len[k] + s, e = max(a, raw), min(a + ln, raw + length) + if s >= e: + continue + off = self.mem_off[k] + for b in range(s, e): + j = b - raw + # >= not >: several accesses can share a timestamp (an + # instruction that reads and writes), and the later entry on the + # line is the one that stands. + if t >= best[j]: + best[j] = t + out[j] = self.mem_blob[off + (b - a)] + known[j] = 1 + return bytes(out), bytes(known) + + def memory_writes(self, addr: int, length: int) -> list[int]: + """Timestamps that WROTE any byte of ``[addr, addr+length)``.""" + return self._mem_touch(addr, length, writes=True) + + def memory_accesses(self, addr: int, length: int) -> list[int]: + """Timestamps that read or wrote any byte of the range.""" + return self._mem_touch(addr, length, writes=False) + + def _mem_touch(self, addr: int, length: int, writes: bool) -> list[int]: + if not len(self.mem_idx) or length <= 0: + return [] + self._mem_index() + raw = addr - self.slide + import bisect as _b + lo = _b.bisect_left(self._mem_starts, raw - self._mem_maxlen) + hi = _b.bisect_right(self._mem_starts, raw + length - 1) + out = set() + for pos in range(lo, hi): + k = self._mem_order[pos] + a, ln = self.mem_addr[k], self.mem_len[k] + if a + ln <= raw or a >= raw + length: + continue + if writes and not self.mem_write[k]: + continue + out.add(self.mem_idx[k]) + return sorted(out) + # -- execution queries (what painting is built on) ---------------------- # def executions(self, ea: int) -> array.array: """Every timestamp that executed ``ea`` (database address).""" |