diff options
Diffstat (limited to 'idatui')
| -rw-r--r-- | idatui/app.py | 59 | ||||
| -rw-r--r-- | idatui/domain.py | 78 |
2 files changed, 130 insertions, 7 deletions
diff --git a/idatui/app.py b/idatui/app.py index d7bc390..1db32ff 100644 --- a/idatui/app.py +++ b/idatui/app.py @@ -50,6 +50,7 @@ _S_ADDR = Style(color="grey58") _S_LABEL = Style(color="yellow", bold=True) _S_INSN = Style(color="white") _S_MNEM = Style(color="cyan") +_S_OPBYTES = Style(color="grey50") # raw opcode bytes column _S_CURSOR = Style(bgcolor="grey30") _S_DIM = Style(color="grey42", italic=True) _S_MATCH = Style(bgcolor="#7a5c00") # all search matches @@ -579,6 +580,7 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True Binding("pageup", "page(-1)", "PgUp", show=False), Binding("home", "goto_top", "Top", show=False), Binding("G,end", "goto_bottom", "Bottom", show=False), + Binding("o", "toggle_opcodes", "Opcodes", show=False), Binding("tab,shift+tab", "app.toggle_view", "Pseudocode", priority=True), *SearchMixin.SEARCH_BINDINGS, *NavMixin.NAV_BINDINGS, @@ -606,6 +608,17 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True self._matches: list[int] = [] self._ranges: dict[int, list[tuple[int, int]]] = {} self._search_texts: list[str] | None = None + self._show_ops = True + self._op_w = 0 # char width of the hex-bytes field (excl. trailing gap) + + def _op_field(self, line) -> str: # type: ignore[no-untyped-def] + """The padded opcode-bytes column (empty when hidden). Kept identical + between the rendered strip and the plain text so cursor/search offsets + line up.""" + if not self._show_ops or self._op_w <= 0: + return "" + raw = line.raw or b"" + return " ".join(f"{b:02X}" for b in raw).ljust(self._op_w) + " " def _line_plain(self, idx: int) -> str | None: if self.model is None: @@ -613,7 +626,7 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True line = self.model.cached_line(idx) if line is None: return None - s = f"{line.ea:08X} " + s = f"{line.ea:08X} " + self._op_field(line) if line.label: s += f"{line.label}: " return s + line.text @@ -635,9 +648,8 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True self._prime() # -- search hooks ------------------------------------------------------ # - @staticmethod - def _fmt(line) -> str: # type: ignore[no-untyped-def] - s = f"{line.ea:08X} " + def _fmt(self, line) -> str: # type: ignore[no-untyped-def] + s = f"{line.ea:08X} " + self._op_field(line) if line.label: s += f"{line.label}: " return s + line.text @@ -688,6 +700,8 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True def _on_primed(self, total: int) -> None: self.total = total self.virtual_size = Size(0, total) + self._update_op_w() # provisional width from the primed window + self._scan_op_width() # settle it against the whole function self._clamp_x() # cursor line is now cached; keep the column in range if self._pending_scroll_y is not None and self._pending_scroll_y >= 0: self._apply_scroll(min(self._pending_scroll_y, max(total - 1, 0))) @@ -714,6 +728,9 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True strip = Strip([Segment(f" {idx:>8} …", _S_DIM)]) else: segs: list[Segment] = [Segment(f"{line.ea:08X} ", _S_ADDR)] + op = self._op_field(line) + if op: + segs.append(Segment(op, _S_OPBYTES)) if line.label: segs.append(Segment(f"{line.label}: ", _S_LABEL)) mnem, _, rest = line.text.partition(" ") @@ -727,6 +744,40 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True strip = _cursor_decorate(strip, self._line_plain(idx) or "", self.cursor_x) return strip.adjust_cell_length(width, _S_INSN) + def _update_op_w(self) -> bool: + """Recompute the opcode column width from the model's widest instruction. + Returns True if it changed.""" + w = 0 + if self.model is not None and self._show_ops: + mx = self.model.max_raw_len() + w = max(mx * 3 - 1, 0) if mx > 0 else 0 + if w != self._op_w: + self._op_w = w + return True + return False + + @work(thread=True, exclusive=True, group="disasm-opwidth") + def _scan_op_width(self) -> None: + model = self.model + if model is None: + return + model.scan_bytes() # fetch all blocks -> stable widest instruction + self.app.call_from_thread(self._settle_op_w) + + def _settle_op_w(self) -> None: + if self._update_op_w(): + self._search_texts = None # layout changed -> stale offsets + self.refresh() + + def action_toggle_opcodes(self) -> None: + self._show_ops = not self._show_ops + self._update_op_w() + self._search_texts = None # column layout changed -> reindex on next search + self._ranges = {} + self._clamp_x() + self.refresh() + self._app_status("opcodes " + ("on" if self._show_ops else "off")) + def _ensure_window(self, top: int) -> None: if self.model is None: return diff --git a/idatui/domain.py b/idatui/domain.py index cc95b21..3dd21ea 100644 --- a/idatui/domain.py +++ b/idatui/domain.py @@ -27,7 +27,7 @@ import re import threading import urllib.request from concurrent.futures import ThreadPoolExecutor -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Callable from .client import IDAClient, IDAToolError @@ -68,6 +68,7 @@ class Line: ea: int text: str label: str | None = None + raw: bytes | None = None # opcode bytes; filled in by DisasmModel post-fetch @classmethod def from_raw(cls, d: dict) -> "Line": @@ -270,6 +271,9 @@ class DisasmModel: self._blocks: dict[int, list[Line]] = {} self._total: int | None = None self._ea_list: list[int] | None = None + self._max_raw = 0 # widest opcode length seen (bytes) + self._func_end: int | None = None + self._func_end_done = False self._lock = threading.Lock() self._inflight: set[int] = set() @@ -287,18 +291,85 @@ class DisasmModel: self._total = int(total) return self._total + def _function_end(self) -> int | None: + """End address (exclusive) of this function; used to size the final + instruction's opcode bytes. Cached (one lookup per model).""" + if self._func_end_done: + return self._func_end + end: int | None = None + try: + fn = self._prog.function_of(self.ea) + if fn is not None: + end = fn.addr + fn.size + except Exception: # noqa: BLE001 -- best-effort; falls back to a width guess + end = None + with self._lock: + self._func_end = end + self._func_end_done = True + return end + + def _attach_bytes(self, lines: list[Line], end_ea: int | None) -> list[Line]: + """Read the opcode bytes for ``lines`` in one request and slice them per + instruction using consecutive addresses (variable-length safe).""" + if not lines: + return lines + last_end = end_ea + if last_end is None or last_end <= lines[-1].ea: + last_end = lines[-1].ea + 15 # x86 max insn len; only the final line + start = lines[0].ea + data = self._prog.read_bytes(start, last_end - start) + out: list[Line] = [] + biggest = 0 + for i, ln in enumerate(lines): + nxt = lines[i + 1].ea if i + 1 < len(lines) else last_end + length = max(nxt - ln.ea, 0) + off = ln.ea - start + b = bytes(data[off:off + length]) + biggest = max(biggest, len(b)) + out.append(replace(ln, raw=b)) + with self._lock: + if biggest > self._max_raw: + self._max_raw = biggest + return out + def _fetch_block(self, b: int) -> list[Line]: + # Over-fetch one instruction so the block knows where its last + # instruction ends (variable-length archs give no size field). payload = self._prog.client.call( "disasm", addr=hex(self.ea), offset=b * self.BLOCK, - max_instructions=self.BLOCK, + max_instructions=self.BLOCK + 1, ) raw = payload.get("asm", {}).get("lines", []) if isinstance(payload, dict) else [] - lines = [Line.from_raw(r) for r in raw] + fetched = [Line.from_raw(r) for r in raw] + lines = fetched[:self.BLOCK] + if len(fetched) > self.BLOCK: + end_ea: int | None = fetched[self.BLOCK].ea + else: # this block ends the function + end_ea = self._function_end() + lines = self._attach_bytes(lines, end_ea) with self._lock: self._blocks[b] = lines self._inflight.discard(b) return lines + def max_raw_len(self) -> int: + """Widest opcode length (bytes) across the blocks fetched so far.""" + with self._lock: + return self._max_raw + + def scan_bytes(self) -> int: + """Fetch every block (populating opcode bytes) and return the widest + instruction length across the whole function. Used to size the opcode + column so its padding doesn't jump as the listing streams in.""" + total = self.total() + off = 0 + while off < total: + got = self.lines(off, self.BLOCK, prefetch=False) + if not got: + break + off += len(got) + return self.max_raw_len() + def _get_block(self, b: int) -> list[Line]: with self._lock: hit = self._blocks.get(b) @@ -395,6 +466,7 @@ class DisasmModel: self._blocks.clear() self._total = None self._ea_list = None + self._max_raw = 0 # --------------------------------------------------------------------------- # |