diff options
| -rw-r--r-- | TODO | 39 | ||||
| -rw-r--r-- | idatui/app.py | 59 | ||||
| -rw-r--r-- | idatui/domain.py | 78 | ||||
| -rw-r--r-- | plan/rpc.md | 20 | ||||
| -rw-r--r-- | rehearsed-engineer.md | 6 | ||||
| -rw-r--r-- | systemd/idatui-mcp.service | 34 | ||||
| -rwxr-xr-x | systemd/install.sh | 23 | ||||
| -rw-r--r-- | uv.lock | 4 |
8 files changed, 253 insertions, 10 deletions
@@ -1,5 +1,40 @@ -[ ] RPC endpoint for robot-spectator-ida -[ ] rename -> history stack pop -> old name +## INCOMPLETE LIST OF TODO's +## ---------------------------------------------------------------------------- + +bugs: + +current: +[x] RPC endpoint for robot-spectator-ida + -> progressssss + +ez: +[ ] opcodes (toggle) in disas view + +doable: +[ ] stress test: big fucking binaries (canon rtos blob) + +hard: +[ ] deal with PLT stubs and such (oh god here we go) +[ ] how to deal with non-function-body regions? + -> we want to be able to do data/type definitions in .data etc. + +crazy: +[ ] multi/split view ala ghidra? + -> would be interesting to make arbitrary compositions of disas/decomp/hex + views and keep them all in sync cursor wise etc. + +done-ish: +[x] if `drive pc` hits a decompilation error it somehow causes a long timeout on client side +[x] remove stupid textual chrome/scaffolding +[~] hex editor? (viewer at least) +[x] make unit tests not suck (why unittest? because we're reward hackers!) +[x] (re-)typing +[x] struct editor +[ ] cant rename local label + -> api limitations, might fix (not super important) +[x] rename -> history stack pop -> old name [x] page up/down cursor x/y preserve [x] hlsearch should jump cursor [x] makes names pane sortable (addr/name columns) + + 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 # --------------------------------------------------------------------------- # diff --git a/plan/rpc.md b/plan/rpc.md new file mode 100644 index 0000000..1f6b119 --- /dev/null +++ b/plan/rpc.md @@ -0,0 +1,20 @@ +## RPC functionality + +we need some way to drive an instance of our IDA tui programatically, +with all UI interactions actually being shown as if a regular user was driving +the software. + +the rationale is we'll be doing some machine-assisted reverse engineering +and livestreaming the work/progress on our little terminal streaming platform, +sl0p.foo ! + +come up with a plan on how to architect this, remember the machine (you, the LLM) +that will eventually be driving our TUI is running inside a tmux pane. +I think it'd be most sensible if you assume we run the TUI in some kind of rpc/server +mode in a second pane and talk to it over unixsocket/tcp. + +I think we want some "highlevel" RPC primitives for UI actions, but also a +more raw RPC primitive that lets us send "keystrokes" to the TUI. + +carefully go through our current architecture and the requirements I vaguely +sketched out above and propose an implementation. diff --git a/rehearsed-engineer.md b/rehearsed-engineer.md new file mode 100644 index 0000000..e7fd23a --- /dev/null +++ b/rehearsed-engineer.md @@ -0,0 +1,6 @@ + alright, we're going try to solve a CTF challenge you can find in ~/re300_files + you should be able to analyze it using our ida TUI in a fresh tmux pane. + we're livestreaming this work, so Id prefer you interact with the TUI where + possible to make this (somewhat) entertaining to watch. remember to populate the ida idb + with good comments, names, types and other metadata that can help us understand + and follow your reverse engineering progress. diff --git a/systemd/idatui-mcp.service b/systemd/idatui-mcp.service new file mode 100644 index 0000000..9afcb73 --- /dev/null +++ b/systemd/idatui-mcp.service @@ -0,0 +1,34 @@ +[Unit] +Description=idatui ida-pro-mcp supervisor (idalib-mcp on 127.0.0.1:8745) +Documentation=https://github.com/mrexodia/ida-pro-mcp +# Wait for network in case a worker ever needs to resolve/pull anything. +After=network-online.target +Wants=network-online.target +# If it dies >5 times in 60s, stop flapping and surface a hard failure. +StartLimitIntervalSec=60 +StartLimitBurst=5 + +[Service] +Type=simple +WorkingDirectory=%h/dev/ida-tui-maybe +ExecStart=%h/dev/ida-tui-maybe/spawn.sh + +# Be explicit about PATH so the unit doesn't depend on an interactive login +# shell. %h/.local/bin is where `idalib-mcp` lives (uv tool install); uv itself +# and /usr/bin/python are on the system PATH. +Environment=PATH=%h/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin +# Tunables (override with a drop-in: systemctl --user edit idatui-mcp). +Environment=IDA_MCP_MAX_WORKERS=8 +Environment=IDA_MCP_HOST=127.0.0.1 +Environment=IDA_MCP_PORT=8745 + +# Always bring it back; observe restarts/exits in `journalctl --user -u idatui-mcp`. +Restart=always +RestartSec=2 + +# Give the supervisor time to shut its workers down cleanly. +TimeoutStopSec=30 +KillMode=mixed + +[Install] +WantedBy=default.target diff --git a/systemd/install.sh b/systemd/install.sh new file mode 100755 index 0000000..0a118b7 --- /dev/null +++ b/systemd/install.sh @@ -0,0 +1,23 @@ +#!/bin/sh +# Install & enable the idatui-mcp user service. Re-run after editing the unit. +set -eu + +UNIT=idatui-mcp.service +SRC=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)/$UNIT +DEST_DIR="${XDG_CONFIG_HOME:-$HOME/.config}/systemd/user" + +mkdir -p "$DEST_DIR" +ln -sf "$SRC" "$DEST_DIR/$UNIT" + +# Keep the service alive after logout / across reboots without a login session. +loginctl enable-linger "$(id -un)" || true + +systemctl --user daemon-reload +systemctl --user enable --now "$UNIT" + +echo +echo "Installed. Handy commands:" +echo " systemctl --user status idatui-mcp" +echo " systemctl --user restart idatui-mcp" +echo " journalctl --user -u idatui-mcp -f" +echo " systemctl --user edit idatui-mcp # drop-in overrides (ports, workers)" @@ -21,13 +21,15 @@ dev = [ { name = "pytest" }, ] tui = [ + { name = "pygments" }, { name = "textual" }, ] [package.metadata] requires-dist = [ + { name = "pygments", marker = "extra == 'tui'", specifier = ">=2" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8" }, - { name = "textual", marker = "extra == 'tui'", specifier = ">=0.60" }, + { name = "textual", marker = "extra == 'tui'", specifier = ">=8" }, ] provides-extras = ["tui", "dev"] |