aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--.gitignore1
-rw-r--r--README.md34
-rw-r--r--TODO103
-rw-r--r--docs/PROJECTS.md52
-rw-r--r--docs/TEXTUAL_NOTES.md2
-rw-r--r--docs/wip-asm-spans.patch247
-rw-r--r--experiments/cortexm.binbin0 -> 1024 bytes
-rwxr-xr-xexperiments/fibonacci.binbin0 -> 7016 bytes
-rw-r--r--experiments/fibonacci.c43
-rwxr-xr-xexperiments/fibonacci.elfbin0 -> 109028 bytes
-rw-r--r--experiments/fibonacci.obin0 -> 3040 bytes
-rw-r--r--idatui/app.py1235
-rw-r--r--idatui/domain.py68
-rw-r--r--idatui/formats.py15
-rw-r--r--idatui/launch.py6
-rw-r--r--idatui/pane.py4
-rw-r--r--idatui/rpc.py33
-rw-r--r--idatui/trace.py495
-rw-r--r--server/patch_server.py364
-rw-r--r--tests/test_blob_ui.py33
-rw-r--r--tests/test_formats.py18
-rw-r--r--tests/test_scenarios.py135
-rw-r--r--tests/test_thumb_ui.py260
-rw-r--r--tests/test_trace.py203
-rw-r--r--tests/test_trace_ui.py374
-rw-r--r--tests/test_trace_vs_tenet.py189
-rw-r--r--tools/verify_procs.py13
27 files changed, 3240 insertions, 687 deletions
diff --git a/.gitignore b/.gitignore
index b1d85aa..6ff0889 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,3 +14,4 @@ build/
# large analysis targets (copied in, not source)
targets/
bin/
+*.pristine.i64
diff --git a/README.md b/README.md
index 8fc59fd..7d7d8d7 100644
--- a/README.md
+++ b/README.md
@@ -98,6 +98,10 @@ nothing:
./ida-tui fw.bin --processor arm --base 0x8000000
```
+ARM images that use Thumb need one more thing: press `t` on the listing to switch
+ARM/Thumb decoding at the cursor (it sets IDA's `T` register, and the segment to
+32-bit, since Thumb doesn't exist in AArch64).
+
`--base` is a real address (IDA's own `-b` is in paragraphs; the conversion is
done for you). In a project the options are recorded per binary, which is what a
multi-image firmware wants. They apply to the first open only — after that the
@@ -108,6 +112,36 @@ multi-image firmware wants. They apply to the first open only — after that the
> reopen. Delete those stale files (never the `.i64`) and retry — `ida-tui` does
> this automatically.
+## Execution traces
+
+Load a [Tenet](https://github.com/gaasedelen/tenet) trace alongside the binary
+and explore it in time:
+
+```sh
+./ida-tui /path/to/binary --trace trace.0.log
+```
+
+A docked pane on the right shows the registers at the current timestamp (the
+ones the current instruction wrote are highlighted) and a timeline. `]` and `[`
+step one instruction forward and back; `}` and `{` step over a call by following
+the stack pointer. The code view follows.
+
+Both code views are painted with the execution trail: where you just came from,
+where you're about to go, and the instruction you're standing on. The pseudocode
+view is painted too — a trace records instructions, but `decomp_map` says which
+instructions each C line covers, so the same trail lands on the decompilation.
+
+The dock also shows the **stack as of that instant**, read out of the trace.
+Bytes the trace never observed print as `??` rather than zeros — a trace knows
+what it saw and nothing else. The hex view (`\`) gets the same treatment: bytes
+the trace saw at this timestamp are shown in green over the file's own contents.
+
+Trace addresses are rebased onto the database automatically — a traced process
+is relocated, so nothing lines up until that's solved.
+
+Traces are recorded separately; see `~/dev/tenet/tenet-original/tracers/` for the
+QEMU tracer.
+
## RPC / driving the TUI
Give the TUI `--rpc <sock>` to expose a unix-socket control channel, then drive
diff --git a/TODO b/TODO
index 78568ae..7a79ba8 100644
--- a/TODO
+++ b/TODO
@@ -72,61 +72,72 @@ done-ish:
-## The scenario suite mutates a PERSISTENT database
+## Test hygiene (fixed, worth remembering)
-tests/test_scenarios.py runs against targets/echo.i64 and every edit it makes is
-saved there. A scenario that undefines an instruction leaves that instruction
-undefined for every later run — decomp_follow_self started failing "for no
-reason" and stayed failing until the .i64 was deleted and re-analysed.
+tests/test_scenarios.py used to run against targets/echo.i64 and IDA saved every
+edit it made, so each run inherited the previous run's damage. It cost real time
+twice: decomp_follow_self "started failing" with no code change (an earlier
+scenario had undefined an instruction), and an edit-position check looked flaky
+one run in three, which nearly got written up as an async race.
-That also poisoned an investigation: an `edit_keeps_view` check looked flaky
-(cursor jumping 0x20c6 -> 0x2094 about one run in three) and was almost
-certainly the database drifting between runs, not a race. Any conclusion drawn
-from repeated runs of a mutating scenario is suspect.
+Now: the suite copies the binary into a temp dir and seeds it from a golden
+database (<target>.pristine.i64, built once, never written back). Every run
+starts from identical bytes and the tracked target is never touched.
-Worth fixing properly: either give the suite a scratch copy of the binary per
-run, or have mutating scenarios undo themselves. Until then, `rm targets/*.i64`
-before trusting a failure that appeared without a code change.
+The general rule this came from: a suite whose result depends on its own history
+can't be trusted to accuse the code. tests/test_blob_ui.py builds a throwaway
+binary; test_project_ui.py stages copies; test_thumb_ui.py deletes the .i64
+before each phase because the T flag and segment bitness are SAVED in it.
-Coverage for "an edit must not move the view" lives in tests/test_blob_ui.py,
-which builds its own throwaway binary and can mutate freely.
+## decomp_map was NOT the problem (corrected)
-## Assembly syntax highlighting (WIP, patch in docs/wip-asm-spans.patch)
+I recorded here that decomp_map returned four entries for cat's main and blamed
+the tool. It doesn't: called directly it returns 769 lines, 475 with addresses,
+for exactly that function. The four-line map belonged to a PLT stub the
+decompiler had momentarily switched to, sampled mid-bounce.
-The listing shows the mnemonic bright and everything after it in one body
-colour. IDA already classifies every token, for every processor, so there is
-nothing to lex — generate_disasm_line() emits \x01<tag>text\x02<tag> and the tag
-says what the text IS. A pygments asm lexer would be a worse guess and would
-need a dialect per architecture.
+The real fault was the resync decision in _seek_split using _split_range, which
+is maintained by a guarded async path and lags. A stale range made every step
+look like a function change, so the decompiler thrashed
+(main -> stub -> main), each bounce paying a synchronous 769-line map fetch on
+the UI thread. Fixed by deciding from the map the trail painting already holds,
+which is keyed to what the decompiler currently HAS loaded.
-The patch: _idatui_spans() in patch_server parses those tags into
-[[kind, text], ...] (insn/reg/num/str/name/seg/cmt/punct), heads rows carry
-"spans", Head.spans holds them, and ListingView renders via _span_segments()
-with a fallback to today's mnemonic/rest split. Verified working on echo:
+Still true and worth knowing: the decompiler attributes only about half of a
+function's instructions to a line, so the pseudocode cursor moves on those and
+waits on the rest. The tempting fallback — nearest mapped address at or before
+the pc — is UNSOUND: C lines are not monotonic in address, and it resolved an
+instruction early in main to a line near the end of the function.
- lea rcx, function; "usage" -> insn text reg punct text name cmt
- coverage over 400 rows: insn 344, reg 402, punct 397, name 156, num 47, cmt 44
+FIXED: the split view and the trace path used to keep two parallel maps of the
+same thing, fetched separately and keyed differently — the split one on _cur (the
+cursor's function), the trace one on the decompiler's loaded function. That is
+how they ended up describing different functions. There is now one index, keyed
+to what the decompiler HOLDS, and both read it.
-Palette rule used: NEUTRALS for the machine (mnemonic brightest, registers at
-body weight since they are most of the text), HUES only where they carry meaning
-(numbers, strings, symbols), structure recedes so commas and brackets stop
-competing with operands.
+## A stray navigation to the entry point during trace seeking
-Two things learned that the patch encodes:
-* The constants are SCOLOR_DATNAME / SCOLOR_CODNAME — there is no SCOLOR_DNAME.
- Guessing fails SILENTLY: an unmapped tag renders as body text, so symbols just
- aren't blue and nothing says why.
-* Spans must be whitespace-collapsed exactly as `text` is, walking characters
- rather than per-span, because a run of spaces straddles spans. The row only
- gets spans when the two agree, so a mismatch degrades instead of corrupting.
+Seen while building M3. After a trace seek, a SECOND _open_at arrives from a
+worker callback and re-points the view at the entry function:
-NOT MERGED because it breaks one check: listing_view "undefining a data head
-yields an unknown run" now reports kind=data. Reproducible, and bisected to this
-change (10/0 with it stashed, 9/1 with it applied) on a fresh .i64. Cause not yet
-found — the span code doesn't touch `kind`, which is computed from the flags
-before spans are attached, so the suspicion is that adding "spans" to the row
-dict perturbs something in the undefine path's re-read (caching keyed on the row
-shape?). Find that before merging.
+ ('_goto_ea', ['0x3160']) <- the seek's own navigation
+ ('_open_at', ['0x3160', 'main']) <- correct
+ ('_open_at', ['0x34d0', 'start']) <- stray, and it wins
-Also still plain: DisasmView (the function-scoped view) renders Line objects from
-the `disasm` tool, which doesn't emit spans. Same treatment needed there.
+The cursor and _cur then sit on 'start' while the trace's pc is elsewhere, and it
+does not settle — measured stable for 3+ seconds. Consequence: a cursor-based
+action taken right after a seek (`>` seeks on the address under the cursor) acts
+on the wrong address.
+
+Not _auto_land: 'start' is not in ENTRY_NAMES, and its guard was set. The
+traceback only shows the Textual callback frame, so the origin is a
+call_from_thread from some worker — most likely a navigation queued earlier
+arriving late, which is the same shape as the stale-result bug fixed in 756589a
+for _open_decomp_entry (that one got a sequence guard; the listing path did not).
+
+To reproduce: seek to an address executed twice, wait for the cursor to reach it,
+then seek again and watch _cur.
+
+Workaround in place: nothing. The M3 tests place the cursor explicitly rather
+than relying on where a seek left it, which is what a user does anyway (you point
+at a line and ask about it) — but the drift is real and worth fixing.
diff --git a/docs/PROJECTS.md b/docs/PROJECTS.md
index 10274ab..fe6c2a8 100644
--- a/docs/PROJECTS.md
+++ b/docs/PROJECTS.md
@@ -297,6 +297,58 @@ conversion happens in `BinaryRef.load_args`, and a base that isn't 16-byte
aligned is rejected rather than silently landing somewhere else. `ida_args`
passes anything else through untouched.
+### ARM/Thumb
+
+Thumb is not a property of the bytes — it's a mode the CPU is in — so a raw image
+gives IDA nothing to detect. At a Thumb entry point it decodes 16-bit
+instructions as 32-bit ARM and produces confident nonsense: `push {r3,lr}`
+(`08 b5`) reads as `SVCLT 0xBF00`, and `c` either refuses or carves garbage.
+
+`t` on the listing switches the mode at the cursor and disassembles in it:
+
+ Thumb @ 0x0 (segment set to 32-bit; Thumb needs ARM32) — 10 instructions
+
+It sets IDA's `T` segment register, and forces the segment to 32-bit when
+turning Thumb on. That second part isn't optional: Thumb doesn't exist in
+AArch64, and a headerless blob loaded with `-parm` comes up 64-bit, so setting
+`T` alone changes nothing and looks broken. Asking for Thumb *is* asking for
+ARM32.
+
+`t` again toggles back — the mode is a guess, and guesses get revised. Both
+states are saved in the `.i64`.
+
+**Pick a 32-bit processor at load, or none of this decompiles.** Bare `arm`
+gives a 64-BIT database (AArch64), and that is decided at load time and cannot be
+changed afterwards — setting the bitness post-hoc makes the decompiler INTERR.
+In a 64-bit database:
+
+* Thumb doesn't exist, so `t` sets a segment flag that means nothing; and
+* Hex-Rays refuses a 32-bit function outright — *"only 64-bit functions can be
+ decompiled in the current database"* — so the disassembly looks right and F5
+ silently produces nothing.
+
+So the list offers `arm:ARMv7-A` (most firmware), `arm:ARMv7-M` / `arm:ARMv6-M`
+(Cortex-M, Thumb only) and `arm:ARMv5TE` alongside 64-bit `arm`. `t` warns when
+it notices the database is 64-bit and points at `Ctrl+L`.
+
+Worth knowing: a correctly-chosen 32-bit ARM database also lets IDA's own
+auto-analysis find Thumb functions — `experiments/fibonacci.bin` yields 54
+functions on load with `arm:ARMv7-A` and none with `arm`.
+
+### Thumb entry points from a vector table
+
+`Shift+T` scans forward from the cursor for **odd pointers** — an ARM function
+pointer carries the mode in bit 0, so a Cortex-M vector table is a list of Thumb
+entry points. IDA won't follow them on a headerless image, because nothing tells
+it those words are pointers at all:
+
+ 3 Thumb entries found, 3 disassembled
+
+A word only counts when it is odd, lands inside a loaded segment, and its target
+is executable and not already data. The even words in a vector table (the initial
+stack pointer) fail the first test, which is the point — marking a data word as
+code corrupts the listing, so a false positive costs more than a miss.
+
### When analysis finds nothing
Zero functions is what a raw image described wrongly looks like — right file,
diff --git a/docs/TEXTUAL_NOTES.md b/docs/TEXTUAL_NOTES.md
index 059b47c..c8b265c 100644
--- a/docs/TEXTUAL_NOTES.md
+++ b/docs/TEXTUAL_NOTES.md
@@ -15,7 +15,7 @@ Hard-won Textual behaviour and the patterns this app relies on. Pairs with
isn't recomputed until layout. Apply the scroll now AND again via
`call_after_refresh` with `refresh(layout=True)`; don't zero `virtual_size` on
load (snaps scroll to 0 → visible flash). See `ColumnCursor._apply_scroll`,
- `DisasmView._on_primed`, `DecompView.show`.
+ `ListingView._on_primed`, `DecompView.show`.
- **Scrolling on already-laid-out content (no reload) leaves a stale frame.** A
plain `scroll_to` moves `scroll_offset` but Textual only repaints on a
*rounded* scroll change, and with no virtual_size/content change there's no
diff --git a/docs/wip-asm-spans.patch b/docs/wip-asm-spans.patch
deleted file mode 100644
index 6b43da6..0000000
--- a/docs/wip-asm-spans.patch
+++ /dev/null
@@ -1,247 +0,0 @@
-diff --git a/idatui/app.py b/idatui/app.py
-index dd7c222..6506016 100644
---- a/idatui/app.py
-+++ b/idatui/app.py
-@@ -53,6 +53,23 @@ from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct
- _S_ADDR = Style(color="#6b7684")
- _S_LABEL = Style(color="#7aa2f7", bold=True)
- _S_INSN = Style(color="#c3cad3")
-+#: IDA's token kinds -> the measured palette. The rule that keeps a dense
-+#: disassembly readable: NEUTRALS for the machine (mnemonic brightest because you
-+#: scan down that column, registers at body weight because they're most of the
-+#: text), HUES only where they mean something (numbers, strings, symbols),
-+#: structure recedes so brackets and commas stop competing with operands.
-+_S_SPAN = {
-+ "insn": Style(color="#e8ecf2"), # 15.3:1 mnemonic / directive
-+ "reg": Style(color="#c3cad3"), # 11.0:1 registers = body weight
-+ "num": Style(color="#d8a657"), # 8.2:1 immediates, offsets
-+ "str": Style(color="#9ece6a"), # 9.9:1 string literals
-+ "name": Style(color="#7aa2f7"), # 7.2:1 symbols / xref targets
-+ "seg": Style(color="#93aee0"), # 8.1:1 segment names
-+ "cmt": Style(color="#7c8b9e", italic=True), # 5.2:1
-+ "punct": Style(color="#626c7a"), # 3.4:1 brackets, commas, +/-
-+ "err": Style(color="#c9762f"), # IDA's own error marker
-+ "text": Style(color="#c3cad3"), # 11.0:1 anything unclassified
-+}
- _S_MNEM = Style(color="#e8ecf2")
- _S_OPBYTES = Style(color="#5e6875") # raw opcode bytes column
- _S_DATA = Style(color="#d8a657")
-@@ -1077,6 +1094,24 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
- return " ".join(f"{b:02X}" for b in raw[:_OP_LIMIT]) + "\u2026"
- return " ".join(f"{b:02X}" for b in raw)
-
-+ @staticmethod
-+ def _span_segments(h: Head, fallback: Style):
-+ """Segments for a row's disassembly text.
-+
-+ Uses IDA's own token classification when the worker supplied it; falls
-+ back to the old mnemonic/rest split so an older worker (or a row whose
-+ spans didn't match the text) still renders.
-+ """
-+ if h.spans:
-+ return [Segment(t, _S_SPAN.get(k, fallback)) for k, t in h.spans]
-+ if h.kind == "code":
-+ mnem, _, rest = h.text.partition(" ")
-+ segs = [Segment(mnem, _S_MNEM)]
-+ if rest:
-+ segs.append(Segment(" " + rest, fallback))
-+ return segs
-+ return [Segment(h.text, fallback)]
-+
- def _op_field(self, h: Head) -> str:
- """The padded opcode-bytes column (empty when hidden). Shared format so
- cursor/search offsets line up."""
-@@ -1301,17 +1336,11 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
- segs.append(Segment(_LST_INDENT, _S_MEMBER))
- if h.name:
- segs.append(Segment(f"{h.name} ", _S_LABEL))
-- if h.kind == "code":
-- mnem, _, rest = h.text.partition(" ")
-- segs.append(Segment(mnem, _S_MNEM))
-- if rest:
-- segs.append(Segment(" " + rest, _S_INSN))
-- elif h.kind == "data":
-- segs.append(Segment(h.text, _S_DATA))
-- elif h.kind == "member":
-+ if h.kind == "member":
- segs.append(Segment(h.text, _S_MEMBER))
- else:
-- segs.append(Segment(h.text, _S_UNK))
-+ base = {"code": _S_INSN, "data": _S_DATA}.get(h.kind, _S_UNK)
-+ segs.extend(self._span_segments(h, base))
- strip = Strip(segs)
- linked = idx in self._link_rows
- if linked:
-diff --git a/idatui/domain.py b/idatui/domain.py
-index e4fbec9..606cf42 100644
---- a/idatui/domain.py
-+++ b/idatui/domain.py
-@@ -100,6 +100,10 @@ class Head:
- text: str
- name: str | None = None
- raw: bytes | None = None # opcode/item bytes (filled in for code by the model)
-+ #: [(kind, text)] from IDA's own colour tags — mnem/reg/num/name/str/punct/…
-+ #: None when the worker didn't provide them (older worker, or the spans
-+ #: disagreed with the plain text, in which case the text wins).
-+ spans: tuple[tuple[str, str], ...] | None = None
-
- @property
- def label(self) -> str | None: # Line-compatible alias
-@@ -107,12 +111,15 @@ class Head:
-
- @classmethod
- def from_raw(cls, d: dict) -> "Head":
-+ sp = d.get("spans")
- return cls(
- ea=_as_int(d["ea"]),
- kind=d.get("kind", "unknown"),
- size=int(d.get("size", 0) or 0),
- text=d.get("text", ""),
- name=d.get("name"),
-+ spans=(tuple((str(k), str(t)) for k, t in sp)
-+ if isinstance(sp, list) and sp else None),
- )
-
-
-diff --git a/server/patch_server.py b/server/patch_server.py
-index 4c806a7..6601a0f 100644
---- a/server/patch_server.py
-+++ b/server/patch_server.py
-@@ -305,12 +305,137 @@ def _idatui_head_row(ea):
- "size": int(ida_bytes.get_item_size(ea)),
- "text": text,
- }
-+ if line:
-+ # Keep IDA's own token classification for syntax highlighting. Built from
-+ # the SAME line as `text`, then whitespace-collapsed identically so the
-+ # two never disagree about what the row says.
-+ spans = _idatui_spans(line)
-+ joined = "".join(t for _k, t in spans)
-+ if " ".join(joined.split()) == text:
-+ row["spans"] = spans
- nm = ida_name.get_ea_name(ea)
- if nm:
- row["name"] = nm
- return row
-
-
-+#: IDA colour tag -> the semantic kind the TUI styles. IDA already classifies
-+#: every token in a disassembly line, for every processor it supports, so there
-+#: is nothing to lex: generate_disasm_line emits \x01<tag>text\x02<tag> and the
-+#: tag says what the text IS. A pygments assembly lexer would be a worse guess at
-+#: this and would need one dialect per architecture.
-+_IDATUI_SPAN_KINDS = {
-+ "insn": ("SCOLOR_INSN", "SCOLOR_KEYWORD", "SCOLOR_ASMDIR", "SCOLOR_MACRO"),
-+ "reg": ("SCOLOR_REG",),
-+ "num": ("SCOLOR_NUMBER", "SCOLOR_CHAR", "SCOLOR_BINPREF"),
-+ "str": ("SCOLOR_STRING",),
-+ # NB the real constant names: DATNAME/CODNAME, not "DNAME". Guessing here
-+ # fails silently — an unmapped tag renders as plain body text, so symbols
-+ # just quietly aren't blue and nothing tells you why.
-+ "name": ("SCOLOR_DATNAME", "SCOLOR_CODNAME", "SCOLOR_LOCNAME",
-+ "SCOLOR_IMPNAME", "SCOLOR_DEMNAME", "SCOLOR_LIBNAME",
-+ "SCOLOR_CNAME", "SCOLOR_DNAME",
-+ "SCOLOR_CREF", "SCOLOR_DREF", "SCOLOR_CREFTAIL", "SCOLOR_DREFTAIL"),
-+ "seg": ("SCOLOR_SEGNAME",),
-+ "cmt": ("SCOLOR_AUTOCMT", "SCOLOR_REGCMT", "SCOLOR_RPTCMT", "SCOLOR_VOIDOP"),
-+ "punct": ("SCOLOR_SYMBOL", "SCOLOR_ALTOP", "SCOLOR_HIDNAME"),
-+ "err": ("SCOLOR_ERROR",),
-+}
-+
-+
-+def _idatui_tag_map():
-+ """{tag character: kind}, built once from whatever this IDA actually has."""
-+ import ida_lines
-+ out = {}
-+ for kind, names in _IDATUI_SPAN_KINDS.items():
-+ for n in names:
-+ v = getattr(ida_lines, n, None)
-+ if isinstance(v, str) and v:
-+ out[v[0]] = kind
-+ elif isinstance(v, int):
-+ out[chr(v)] = kind
-+ return out
-+
-+
-+_IDATUI_TAGS = None
-+
-+
-+def _idatui_spans(line):
-+ """A tagged disasm line as [[kind, text], ...], colour tags resolved.
-+
-+ Unknown tags become 'text' rather than being dropped: a processor module can
-+ emit a colour we don't classify, and losing the characters would corrupt the
-+ line."""
-+ global _IDATUI_TAGS
-+ import ida_lines
-+ if _IDATUI_TAGS is None:
-+ _IDATUI_TAGS = _idatui_tag_map()
-+ on, off, esc = "\x01", "\x02", "\x03"
-+ addr_tag = chr(getattr(ida_lines, "COLOR_ADDR", 0x28))
-+ addr_len = int(getattr(ida_lines, "COLOR_ADDR_SIZE", 16))
-+ spans, stack, buf = [], [], []
-+ i, n = 0, len(line)
-+
-+ def flush():
-+ if buf:
-+ spans.append([stack[-1] if stack else "text", "".join(buf)])
-+ del buf[:]
-+
-+ while i < n:
-+ ch = line[i]
-+ if ch == on and i + 1 < n:
-+ tag = line[i + 1]
-+ if tag == addr_tag:
-+ # An embedded target address, not display text: 16 hex digits
-+ # that must not reach the screen.
-+ i += 2 + addr_len
-+ continue
-+ flush()
-+ stack.append(_IDATUI_TAGS.get(tag, "text"))
-+ i += 2
-+ continue
-+ if ch == off and i + 1 < n:
-+ flush()
-+ if stack:
-+ stack.pop()
-+ i += 2
-+ continue
-+ if ch == esc and i + 1 < n: # escaped literal
-+ buf.append(line[i + 1])
-+ i += 2
-+ continue
-+ buf.append(ch)
-+ i += 1
-+ flush()
-+ # Collapse IDA's column padding EXACTLY as the plain text does. A run of
-+ # spaces can straddle two spans, so this walks characters rather than
-+ # collapsing each span on its own — otherwise the spans and `text` disagree
-+ # about the line and the row silently loses its highlighting.
-+ out, prev_space = [], False
-+ for kind, txt in spans:
-+ acc = []
-+ for ch in txt:
-+ if ch.isspace():
-+ if prev_space:
-+ continue
-+ acc.append(" ")
-+ prev_space = True
-+ else:
-+ acc.append(ch)
-+ prev_space = False
-+ if acc:
-+ out.append([kind, "".join(acc)])
-+ while out and out[0][1] == " ":
-+ out.pop(0)
-+ while out and out[-1][1] == " ":
-+ out.pop()
-+ if out and out[0][1].startswith(" "):
-+ out[0][1] = out[0][1].lstrip()
-+ if out and out[-1][1].endswith(" "):
-+ out[-1][1] = out[-1][1].rstrip()
-+ return [[k, t] for k, t in out if t]
-+
-+
- def _idatui_unknown_row(ea, size):
- """One collapsed row for a run of ``size`` undefined bytes starting at
- ``ea``. A single byte is rendered normally (shows its value); a longer run
diff --git a/experiments/cortexm.bin b/experiments/cortexm.bin
new file mode 100644
index 0000000..173bb10
--- /dev/null
+++ b/experiments/cortexm.bin
Binary files differ
diff --git a/experiments/fibonacci.bin b/experiments/fibonacci.bin
new file mode 100755
index 0000000..6f7b470
--- /dev/null
+++ b/experiments/fibonacci.bin
Binary files differ
diff --git a/experiments/fibonacci.c b/experiments/fibonacci.c
new file mode 100644
index 0000000..26fca0c
--- /dev/null
+++ b/experiments/fibonacci.c
@@ -0,0 +1,43 @@
+#include <stdint.h>
+
+/* Iterative Fibonacci. */
+uint32_t fib_iter(uint32_t n)
+{
+ uint32_t a = 0, b = 1;
+ for (uint32_t i = 0; i < n; i++) {
+ uint32_t t = a + b;
+ a = b;
+ b = t;
+ }
+ return a;
+}
+
+/* Naive recursive Fibonacci. */
+uint32_t fib_rec(uint32_t n)
+{
+ if (n < 2)
+ return n;
+ return fib_rec(n - 1) + fib_rec(n - 2);
+}
+
+/* Memoized Fibonacci (static table). */
+uint32_t fib_memo(uint32_t n)
+{
+ static uint32_t cache[48];
+ if (n < 2)
+ return n;
+ if (n >= sizeof(cache) / sizeof(cache[0]))
+ return fib_iter(n);
+ if (cache[n])
+ return cache[n];
+ return cache[n] = fib_memo(n - 1) + fib_memo(n - 2);
+}
+
+volatile uint32_t sink;
+
+int main(void)
+{
+ for (uint32_t n = 0; n < 20; n++)
+ sink = fib_iter(n) + fib_rec(n) + fib_memo(n);
+ return 0;
+}
diff --git a/experiments/fibonacci.elf b/experiments/fibonacci.elf
new file mode 100755
index 0000000..05ac21d
--- /dev/null
+++ b/experiments/fibonacci.elf
Binary files differ
diff --git a/experiments/fibonacci.o b/experiments/fibonacci.o
new file mode 100644
index 0000000..c474d49
--- /dev/null
+++ b/experiments/fibonacci.o
Binary files differ
diff --git a/idatui/app.py b/idatui/app.py
index dd7c222..55f21ef 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -53,6 +53,23 @@ from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct
_S_ADDR = Style(color="#6b7684")
_S_LABEL = Style(color="#7aa2f7", bold=True)
_S_INSN = Style(color="#c3cad3")
+#: IDA's token kinds -> the measured palette. The rule that keeps a dense
+#: disassembly readable: NEUTRALS for the machine (mnemonic brightest because you
+#: scan down that column, registers at body weight because they're most of the
+#: text), HUES only where they mean something (numbers, strings, symbols),
+#: structure recedes so brackets and commas stop competing with operands.
+_S_SPAN = {
+ "insn": Style(color="#e8ecf2"), # 15.3:1 mnemonic / directive
+ "reg": Style(color="#c3cad3"), # 11.0:1 registers = body weight
+ "num": Style(color="#d8a657"), # 8.2:1 immediates, offsets
+ "str": Style(color="#9ece6a"), # 9.9:1 string literals
+ "name": Style(color="#7aa2f7"), # 7.2:1 symbols / xref targets
+ "seg": Style(color="#93aee0"), # 8.1:1 segment names
+ "cmt": Style(color="#7c8b9e", italic=True), # 5.2:1
+ "punct": Style(color="#626c7a"), # 3.4:1 brackets, commas, +/-
+ "err": Style(color="#c9762f"), # IDA's own error marker
+ "text": Style(color="#c3cad3"), # 11.0:1 anything unclassified
+}
_S_MNEM = Style(color="#e8ecf2")
_S_OPBYTES = Style(color="#5e6875") # raw opcode bytes column
_S_DATA = Style(color="#d8a657")
@@ -74,6 +91,18 @@ _ASM_KEYWORDS = frozenset({
"gs", "ss", "align", "public", "assume", "end",
})
_S_CURSOR = Style(bgcolor="#2a313c")
+#: Execution trails. Deliberately faint: they sit UNDER the code palette and
+#: must not compete with it — the trail says "you came through here", the text
+#: still has to be readable as code. Now is the loudest because there is exactly
+#: one of it.
+_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
@@ -132,6 +161,8 @@ class ViewAnchor:
top_ea: int | None = None # first visible address
cursor_x: int = 0
flash: str | None = None
+ #: The edit changed which functions exist, so the index must be rebuilt.
+ refresh_functions: bool = False
@dataclass
@@ -693,316 +724,12 @@ class SearchMixin:
# --------------------------------------------------------------------------- #
# Virtualized disassembly view
# --------------------------------------------------------------------------- #
-class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True):
- """A line-virtualized disassembly listing for a single function."""
-
- BINDINGS = [
- Binding("j,down", "cursor_down", "Down", show=False),
- Binding("k,up", "cursor_up", "Up", show=False),
- Binding("ctrl+d", "half_page(1)", "½↓", show=False),
- Binding("ctrl+u", "half_page(-1)", "½↑", show=False),
- Binding("pagedown", "page(1)", "PgDn", show=False),
- Binding("pageup", "page(-1)", "PgUp", show=False),
- Binding("home", "goto_top", "Top", show=False),
- Binding("G,end", "goto_bottom", "Bottom", show=False),
- Binding("o", "toggle_opcodes", "Opcodes", show=False),
- Binding("c", "define_code", "Code", show=False),
- Binding("p", "define_func", "Func", show=False),
- Binding("u", "undefine", "Undef", show=False),
- Binding("tab,shift+tab", "app.toggle_view", "Pseudocode", priority=True),
- Binding("f5", "app.toggle_view", "Decompile", priority=True),
- Binding("L", "app.continuous_here", "Listing"),
- *SearchMixin.SEARCH_BINDINGS,
- *NavMixin.NAV_BINDINGS,
- *ColumnCursor.COL_BINDINGS,
- ]
-
- cursor = reactive(0, repaint=False)
- cursor_x = reactive(0, repaint=False)
-
- class CursorMoved(Message):
- """Posted when the disasm cursor moves; carries the instruction ea."""
-
- def __init__(self, index: int, ea: int | None) -> None:
- super().__init__()
- self.index = index
- self.ea = ea
-
- def __init__(self) -> None:
- super().__init__()
- self.model: DisasmModel | None = None
- self.total = 0
- self._name = ""
- self._pending_scroll_y: int | None = None
- self._term = ""
- self._matches: list[int] = []
- self._ranges: dict[int, list[tuple[int, int]]] = {}
- self._search_texts: list[str] | None = None
- self._show_ops = True
- self._op_w = 0 # char width of the hex-bytes field (excl. trailing gap)
-
- def _op_field(self, line) -> str: # type: ignore[no-untyped-def]
- """The padded opcode-bytes column (empty when hidden). Kept identical
- between the rendered strip and the plain text so cursor/search offsets
- line up."""
- if not self._show_ops or self._op_w <= 0:
- return ""
- raw = line.raw or b""
- return " ".join(f"{b:02X}" for b in raw).ljust(self._op_w) + " "
-
- def _line_plain(self, idx: int) -> str | None:
- if self.model is None:
- return None
- line = self.model.cached_line(idx)
- if line is None:
- return None
- s = f"{line.ea:08X} " + self._op_field(line)
- if line.label:
- s += f"{line.label}: "
- return s + line.text
-
- # -- public API -------------------------------------------------------- #
- def load(self, model: DisasmModel, name: str, cursor: int = 0,
- cursor_x: int = 0, scroll_y: int | None = None) -> None:
- self.model = model
- self._name = name
- self.total = 0
- self.cursor = cursor
- self.cursor_x = cursor_x
- self._pending_scroll_y = scroll_y
- # NB: don't zero virtual_size here — that snaps the scroll to 0 and
- # causes a visible jump before _on_primed restores the target scroll.
- self._matches = []
- self._ranges = {}
- self._search_texts = None
- self._prime()
-
- # -- search hooks ------------------------------------------------------ #
- def _fmt(self, line) -> str: # type: ignore[no-untyped-def]
- s = f"{line.ea:08X} " + self._op_field(line)
- if line.label:
- s += f"{line.label}: "
- return s + line.text
-
- def _search_line_count(self) -> int:
- return self.total
-
- def _search_line_text(self, i: int) -> str | None:
- t = self._search_texts
- return t[i] if t is not None and 0 <= i < len(t) else None
-
- def _search_ensure(self, done) -> None:
- if self._search_texts is not None:
- done()
- return
- self._app_status(f"/{self._term}/ indexing {self.total} lines…")
- self._index_for_search(done)
-
- @work(thread=True, exclusive=True, group="search-index")
- def _index_for_search(self, done) -> None:
- model = self.model
- if model is None:
- self.app.call_from_thread(done)
- return
- texts: list[str] = []
- off, total = 0, self.total
- # 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, 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")
- def _prime(self) -> None:
- model = self.model
- if model is None:
- return
- total = model.total()
- height = max(self.size.height, 1)
- model.lines(0, min(total, height + DisasmModel.BLOCK), prefetch=True)
- if self.cursor:
- model.lines(max(self.cursor - 2, 0), height, prefetch=True)
- self.app.call_from_thread(self._on_primed, total)
-
- def _on_primed(self, total: int) -> None:
- self.total = total
- self.virtual_size = Size(0, total)
- self._update_op_w() # provisional width from the primed window
- self._scan_op_width() # settle it against the whole function
- self._clamp_x() # cursor line is now cached; keep the column in range
- if self._pending_scroll_y is not None and self._pending_scroll_y >= 0:
- self._apply_scroll(min(self._pending_scroll_y, max(total - 1, 0)))
- else:
- self._scroll_cursor_into_view()
- self._pending_scroll_y = None
- self.refresh()
-
- # -- rendering --------------------------------------------------------- #
- def render_line(self, y: int) -> Strip:
- model = self.model
- width = self.size.width
- if model is None or self.total == 0:
- return Strip([Segment("".ljust(width), _S_DIM)])
- top = round(self.scroll_offset.y)
- if y == 0: # once per refresh: warm the visible window + a little ahead
- self._ensure_window(top)
- idx = top + y
- if idx >= self.total:
- return Strip([Segment("".ljust(width), _S_INSN)])
- line = model.cached_line(idx)
- is_cursor = idx == self.cursor
- if line is None:
- strip = Strip([Segment(f" {idx:>8} …", _S_DIM)])
- else:
- segs: list[Segment] = [Segment(f"{line.ea:08X} ", _S_ADDR)]
- op = self._op_field(line)
- if op:
- segs.append(Segment(op, _S_OPBYTES))
- if line.label:
- segs.append(Segment(f"{line.label}: ", _S_LABEL))
- mnem, _, rest = line.text.partition(" ")
- segs.append(Segment(mnem, _S_MNEM))
- if rest:
- segs.append(Segment(" " + rest, _S_INSN))
- strip = Strip(segs)
- if idx in self._ranges:
- strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx))
- if is_cursor:
- strip = _cursor_decorate(strip, self._line_plain(idx) or "", self.cursor_x)
- return strip.adjust_cell_length(width, _S_INSN)
-
- def _update_op_w(self) -> bool:
- """Recompute the opcode column width from the model's widest instruction.
- Returns True if it changed."""
- w = 0
- if self.model is not None and self._show_ops:
- mx = self.model.max_raw_len()
- w = max(mx * 3 - 1, 0) if mx > 0 else 0
- if w != self._op_w:
- self._op_w = w
- return True
- return False
-
- @work(thread=True, exclusive=True, group="disasm-opwidth")
- def _scan_op_width(self) -> None:
- model = self.model
- if model is None:
- return
- model.scan_bytes() # fetch all blocks -> stable widest instruction
- self.app.call_from_thread(self._settle_op_w)
-
- def _settle_op_w(self) -> None:
- if self._update_op_w():
- self._search_texts = None # layout changed -> stale offsets
- self.refresh()
-
- def action_toggle_opcodes(self) -> None:
- self._show_ops = not self._show_ops
- self._update_op_w()
- self._search_texts = None # column layout changed -> reindex on next search
- self._ranges = {}
- self._clamp_x()
- self.refresh()
- self._app_status("opcodes " + ("on" if self._show_ops else "off"))
-
- def _ensure_window(self, top: int) -> None:
- if self.model is None:
- return
- height = max(self.size.height, 1)
- start = max(top - DisasmModel.BLOCK, 0)
- count = height + 2 * DisasmModel.BLOCK
- if not self.model.is_cached(top, height):
- self._fetch_window(start, count)
- else:
- self.model.ensure_async(start, count) # warm neighbors
-
- @work(thread=True, exclusive=False, group="disasm-fetch")
- def _fetch_window(self, start: int, count: int) -> None:
- model = self.model
- if model is None:
- return
- model.lines(start, count, prefetch=True)
- self.app.call_from_thread(self.refresh)
-
- # -- navigation -------------------------------------------------------- #
- def _visible_height(self) -> int:
- return max(self.size.height, 1)
-
- def _scroll_cursor_into_view(self) -> None:
- height = self._visible_height()
- top = round(self.scroll_offset.y)
- if self.cursor < top:
- self.scroll_to(y=self.cursor, animate=False)
- elif self.cursor >= top + height:
- self.scroll_to(y=max(self.cursor - height + 1, 0), animate=False)
-
- def _move(self, delta: int) -> None:
- if self.total == 0:
- return
- old = self.cursor
- before = round(self.scroll_offset.y)
- self.cursor = max(0, min(self.total - 1, self.cursor + delta))
- self._clamp_x()
- self._scroll_cursor_into_view()
- if round(self.scroll_offset.y) != before:
- self.refresh() # scrolled: the whole viewport shifted
- else:
- _refresh_lines(self, old, self.cursor) # only the two changed rows
- self._refresh_hl()
- self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea()))
-
- def _after_cursor_move(self) -> None:
- self.post_message(DisasmView.CursorMoved(self.cursor, self._cursor_ea()))
-
- def _cursor_ea(self) -> int | None:
- if self.model is None:
- return None
- line = self.model.cached_line(self.cursor)
- return line.ea if line else None
-
- # -- item structure edits (IDA c/p/u) --------------------------------- #
- def action_define_code(self) -> None:
- self.post_message(EditItemRequested(self, "code"))
-
- def action_define_func(self) -> None:
- self.post_message(EditItemRequested(self, "func"))
-
- def action_undefine(self) -> None:
- self.post_message(EditItemRequested(self, "undef"))
-
- def action_cursor_down(self) -> None:
- self._move(1)
-
- def action_cursor_up(self) -> None:
- self._move(-1)
-
- def action_goto_top(self) -> None:
- self._move(-self.total)
-
- def action_goto_bottom(self) -> None:
- self._move(self.total)
-
-
# --------------------------------------------------------------------------- #
# Virtualized flat listing view (code + data + undefined, per segment)
# --------------------------------------------------------------------------- #
class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True):
"""A line-virtualized *flat* listing over one segment: code, data and
- undefined heads interleaved (IDA's disassembly view), unlike ``DisasmView``
+ undefined heads interleaved (IDA's disassembly view), unlike ``DisasmModel``
which is bounded to one function. Backed by ``ListingModel`` (the ``heads``
server tool). Used for non-function regions and raw segment browsing.
"""
@@ -1025,6 +752,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
Binding("a", "make_string", "Str", show=False),
Binding("p", "define_func", "Func", show=False),
Binding("u", "undefine", "Undef", show=False),
+ Binding("t", "toggle_thumb", "ARM/Thumb", show=False),
+ Binding("T", "thumb_scan", "Scan vectors", show=False),
*SearchMixin.SEARCH_BINDINGS,
*NavMixin.NAV_BINDINGS,
*ColumnCursor.COL_BINDINGS,
@@ -1060,6 +789,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
self._search_loading = False
self._search_pending: list = [] # done-callbacks awaiting the load
self._link_rows: set[int] = set() # split-view: linked instruction rows
+ #: {address: 'now'|'past'|'future'} painted under the code (trace mode).
+ self.trail: dict[int, str] = {}
# -- text helpers ------------------------------------------------------ #
def _head(self, idx: int) -> Head | None:
@@ -1077,6 +808,24 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
return " ".join(f"{b:02X}" for b in raw[:_OP_LIMIT]) + "\u2026"
return " ".join(f"{b:02X}" for b in raw)
+ @staticmethod
+ def _span_segments(h: Head, fallback: Style):
+ """Segments for a row's disassembly text.
+
+ Uses IDA's own token classification when the worker supplied it; falls
+ back to the old mnemonic/rest split so an older worker (or a row whose
+ spans didn't match the text) still renders.
+ """
+ if h.spans:
+ return [Segment(t, _S_SPAN.get(k, fallback)) for k, t in h.spans]
+ if h.kind == "code":
+ mnem, _, rest = h.text.partition(" ")
+ segs = [Segment(mnem, _S_MNEM)]
+ if rest:
+ segs.append(Segment(" " + rest, fallback))
+ return segs
+ return [Segment(h.text, fallback)]
+
def _op_field(self, h: Head) -> str:
"""The padded opcode-bytes column (empty when hidden). Shared format so
cursor/search offsets line up."""
@@ -1301,21 +1050,21 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
segs.append(Segment(_LST_INDENT, _S_MEMBER))
if h.name:
segs.append(Segment(f"{h.name} ", _S_LABEL))
- if h.kind == "code":
- mnem, _, rest = h.text.partition(" ")
- segs.append(Segment(mnem, _S_MNEM))
- if rest:
- segs.append(Segment(" " + rest, _S_INSN))
- elif h.kind == "data":
- segs.append(Segment(h.text, _S_DATA))
- elif h.kind == "member":
+ if h.kind == "member":
segs.append(Segment(h.text, _S_MEMBER))
else:
- segs.append(Segment(h.text, _S_UNK))
+ base = {"code": _S_INSN, "data": _S_DATA}.get(h.kind, _S_UNK)
+ segs.extend(self._span_segments(h, base))
strip = Strip(segs)
linked = idx in self._link_rows
if linked:
strip = strip.apply_style(_S_LINK) # split-view companion band
+ if self.trail and h is not None:
+ kind = self.trail.get(h.ea)
+ if kind is not None:
+ strip = strip.apply_style(
+ _S_TRAIL_NOW if kind == "now" else
+ _S_TRAIL_PAST if kind == "past" else _S_TRAIL_FUTURE)
plain = self._line_plain(idx) if (self._hl_word or idx == self.cursor) else None
if idx in self._ranges:
strip = _overlay_ranges(strip, self._ranges[idx], self._match_style(idx))
@@ -1401,6 +1150,12 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
def action_undefine(self) -> None:
self.post_message(EditItemRequested(self, "undef"))
+ def action_toggle_thumb(self) -> None:
+ self.post_message(EditItemRequested(self, "thumb"))
+
+ def action_thumb_scan(self) -> None:
+ self.post_message(EditItemRequested(self, "thumbscan"))
+
def action_make_data(self) -> None:
self.post_message(MakeDataRequested(self))
@@ -1518,6 +1273,9 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True
self._gutter = 0 # line-number gutter width (cells)
self._line_eas: list[int | None] = [] # per-line address (marker stripped)
self._link_line: int | None = None # split-view: linked pseudocode line
+ #: {line index: 'now'|'past'|'future'} — the execution trail, mapped from
+ #: instructions onto pseudocode via decomp_map.
+ self.trail: dict[int, str] = {}
self._term = ""
self._matches: list[int] = []
self._ranges: dict[int, list[tuple[int, int]]] = {}
@@ -1651,6 +1409,11 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True
base = self._strips[idx]
if linked:
base = base.apply_style(_S_LINK) # split-view companion band
+ kind = self.trail.get(idx) if self.trail else None
+ if kind is not None:
+ base = base.apply_style(
+ _S_TRAIL_NOW if kind == "now" else
+ _S_TRAIL_PAST if kind == "past" else _S_TRAIL_FUTURE)
if idx in self._ranges:
base = _overlay_ranges(base, self._ranges[idx], self._match_style(idx))
if self._hl_word:
@@ -1713,7 +1476,7 @@ class DecompView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True
best, best_ea = i, e
return best
- # -- navigation (mirrors DisasmView) ---------------------------------- #
+ # -- navigation -------------------------------------------------------- #
def _visible_height(self) -> int:
return max(self.size.height, 1)
@@ -1803,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 -------------------------------------------------------- #
@@ -1989,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] = [
@@ -2003,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))
@@ -2551,6 +2341,221 @@ 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.
+
+ Persistent rather than a modal: a trace turns every other view into "state
+ at time T", so the time and the registers are context you read WHILE looking
+ at code, not something you open and dismiss.
+ """
+
+ def __init__(self) -> None:
+ super().__init__(id="trace-dock")
+ self.trace = None
+ self.idx = 0
+
+ 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:
+ self.trace = trace
+ self.idx = idx
+ tl = self.query_one(TraceTimeline)
+ tl.trace, tl.idx = trace, idx
+ self.refresh_state()
+
+ def refresh_state(self) -> None:
+ t = self.trace
+ if t is None:
+ return
+ n = max(t.length, 1)
+ pct = (self.idx + 1) * 100.0 / n
+ head = Text()
+ head.append(f" {self.idx:,}", _S_MNEM)
+ head.append(f" / {t.length - 1:,} ", _S_DIM)
+ head.append(f"{pct:5.1f}%\n", _S_ADDR)
+ # Register values are machine state and stay as the trace recorded them,
+ # but everything else on screen is in database addresses. Showing both
+ # here explains the relationship once, where it's read, instead of
+ # leaving "pc 0x2aed" next to "rip 0x7ffff6faaaed" to be puzzled over.
+ head.append(f" pc {t.ip(self.idx):#x}", _S_LABEL)
+ if t.slide:
+ head.append(f" (trace {t.raw_ip(self.idx):#x})", _S_DIM)
+ self.query_one("#trace-head", Static).update(head)
+
+ # Registers, with the ones THIS instruction wrote called out: that
+ # difference is the entire reason a delta trace is readable.
+ changed = t.changed(self.idx)
+ body = Text()
+ pc = t.pc_name
+ for name in t.registers:
+ v = t.register(name, self.idx)
+ if v is None:
+ continue
+ hot = name in changed
+ body.append(f" {name:>4} ", _S_MNEM if hot else _S_DIM)
+ 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.
+
+ Tenet's timeline is a Qt widget you scroll and drag to zoom. A terminal
+ column can't do that, but it can do the part that matters — show the shape
+ of the trace and your position in it — with one row per N timestamps.
+ """
+
+ def __init__(self, **kw) -> None:
+ super().__init__("", **kw)
+ self.trace = None
+ self.idx = 0
+
+ def render(self) -> Text:
+ t = self.trace
+ out = Text()
+ h = max(self.size.height - 1, 1)
+ if t is None or not t.length:
+ return out
+ out.append(" timeline\n", _S_DIM)
+ h = max(h - 1, 1)
+ per = max(t.length / h, 1.0)
+ here = int(self.idx / per)
+ for row in range(h):
+ if row == here:
+ out.append(" \u25b6", _S_MNEM)
+ out.append(f" {int(row * per):>10,}\n", _S_ADDR)
+ else:
+ out.append(" \u2502\n", _S_SEP if row % 5 else _S_ADDR)
+ return out
+
+
class LoadOptionsScreen(ModalScreen):
"""Ask how to load a file no loader recognised.
@@ -3227,7 +3232,6 @@ class IdaTui(App):
#left { width: 30%; min-width: 42; max-width: 44; border-right: solid $panel; }
#func-table { height: 1fr; }
#func-filter { dock: top; }
- DisasmView { width: 1fr; padding: 0 1; }
DecompView { width: 1fr; }
ListingView { width: 1fr; padding: 0 1; }
#panes.split ListingView { border-right: tall $panel-lighten-2; }
@@ -3282,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; }
@@ -3292,6 +3296,11 @@ 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; }
+ #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
palette default (24) the box outgrew the terminal and the field you need
@@ -3337,6 +3346,18 @@ class IdaTui(App):
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),
+ # 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),
+ Binding("left_curly_bracket", "step_over_back", "Step over back", show=False),
Binding("f1", "help", "Keys", show=False),
Binding("g", "goto", "Goto"),
Binding("slash", "filter", "Filter", show=False),
@@ -3348,7 +3369,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, load_args: str = "") -> None:
+ project=None, load_args: str = "", trace_path: str = "") -> None:
super().__init__()
# Project mode is additive: with no project this is the plain
# single-binary app, unchanged.
@@ -3362,6 +3383,7 @@ class IdaTui(App):
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._flash_until = 0.0 # ...until this monotonic time
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.
@@ -3379,6 +3401,16 @@ 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
self._rpc = None
@@ -3425,14 +3457,19 @@ class IdaTui(App):
fp = FunctionsPanel(id="left")
fp.display = False # overlay-first: reveal the docked pane with Ctrl+B
yield fp
- # The unified continuous listing is the one code view; DisasmView is
- # deprecated (kept only for DisasmModel, still used by the domain).
+ # The unified continuous listing is the one code view. (DisasmModel
+ # is still used by the domain to index a function's instructions.)
lst = ListingView()
yield lst
yield DecompView()
hx = HexView()
hx.display = False
yield hx
+ # Docked right and only shown once a trace is loaded, so a normal
+ # session looks exactly as it did.
+ td = TraceDock()
+ td.display = False
+ yield td
si = Input(id="search")
si.display = False
si.can_focus = False
@@ -3525,8 +3562,9 @@ class IdaTui(App):
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")
+ f"discards the database for this binary \u2014 {n} "
+ f"function{'s' if n != 1 else ''}, plus any names and comments "
+ f"you've added")
self.push_screen(ConfirmScreen("Reload with different options?", note),
self._on_reload_confirmed)
@@ -3685,12 +3723,37 @@ class IdaTui(App):
await self._rpc.stop()
# -- status helper ----------------------------------------------------- #
- def _status(self, text: str) -> None:
- if self._binary: # project mode: always say which binary you're in
- text = f"[{self._binary}] {text}"
+ def _status(self, text: str, priority: bool = False) -> None:
+ """Write the status bar. ``priority`` marks the RESULT of something the
+ user did.
+
+ An action's result is written, and then the reload it triggered writes
+ its own idle status on top — cursor moved, filter re-applied, functions
+ re-counted. I patched that at five separate call sites before admitting
+ it's one problem: routine chatter must not outrank an answer. A priority
+ message holds the bar briefly and is cleared by the next keypress, i.e.
+ when the user has read it and moved on.
+ """
+ import time as _time
+ if priority:
+ self._flash = text
+ self._flash_until = _time.monotonic() + 8.0
+ elif self._flash and _time.monotonic() < self._flash_until:
+ text = self._flash
+ # 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.
+ # It stops being true the moment a function exists, though: latching it
+ # meant the warning survived defining one with `p` and kept telling you
+ # the load was wrong when it no longer was.
+ if self._no_functions and self._func_index is not None and len(self._func_index):
+ self._no_functions = False
if self._no_functions:
text += " \u2014 no functions: wrong processor/base? Ctrl+L to reload"
try:
@@ -3828,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(
@@ -3850,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)
@@ -3858,18 +3921,20 @@ 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)
self._index_binary() # project mode: keep the cross-binary index fresh
+ if self._trace_path and self._trace is None:
+ self._load_trace() # needs the index above: rebasing reads it
@work(thread=True, exclusive=True, group="prewarm")
def _prewarm_provider(self) -> None:
@@ -4080,7 +4145,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
@@ -4282,6 +4347,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 ""
@@ -4624,17 +4690,12 @@ class IdaTui(App):
def on_follow_requested(self, msg: FollowRequested) -> None:
view = msg.view
word = view.word_under_cursor()
- if isinstance(view, DisasmView):
- ea = view._cursor_ea()
- # The instruction's ordinary fall-through edge (to the next line) is
- # an indistinguishable 'code' xref; pass it so follow can skip it and
- # land on a call/jump's real target instead of the next instruction.
- nxt = view.model.cached_line(view.cursor + 1) if view.model else None
- if ea is not None:
- self._follow_disasm(ea, word, nxt.ea if nxt else None)
- elif isinstance(view, ListingView):
+ if isinstance(view, ListingView):
ea = view._cursor_ea()
if ea is not None:
+ # _next_ea() is the following item: follow uses it to skip the
+ # ordinary fall-through edge, which is an indistinguishable
+ # 'code' xref, and land on a call/jump's real target instead.
self._follow_disasm(ea, word, view._next_ea())
elif isinstance(view, DecompView) and view._texts:
self._follow_decomp(view._texts[view.cursor], word,
@@ -4645,17 +4706,11 @@ class IdaTui(App):
return
view = msg.view
word = view.word_under_cursor()
- if isinstance(view, DisasmView):
+ if isinstance(view, ListingView):
ea = view._cursor_ea()
if ea is not None:
# span = this instruction .. the next, to pre-select the dialog
# entry for the site we invoked xrefs from.
- nxt = view.model.cached_line(view.cursor + 1) if view.model else None
- self._push_busy("finding xrefs\u2026")
- self._xrefs_disasm(ea, word, ea, nxt.ea if nxt else None)
- elif isinstance(view, ListingView):
- ea = view._cursor_ea()
- if ea is not None:
self._push_busy("finding xrefs\u2026")
self._xrefs_disasm(ea, word, ea, view._next_ea())
elif isinstance(view, DecompView) and view._texts:
@@ -5047,7 +5102,7 @@ class IdaTui(App):
# -- comments ---------------------------------------------------------- #
def _line_ea_for(self, view) -> int | None: # type: ignore[no-untyped-def]
"""Address of the line under the cursor in either code view."""
- if isinstance(view, (DisasmView, ListingView)):
+ if isinstance(view, ListingView):
return view._cursor_ea()
if isinstance(view, DecompView):
return view._line_ea(view.cursor)
@@ -5426,8 +5481,7 @@ class IdaTui(App):
# -- item structure edits (IDA c/p/u) --------------------------------- #
def on_edit_item_requested(self, msg: EditItemRequested) -> None:
view = msg.view
- ea = (view._cursor_ea()
- if isinstance(view, (DisasmView, ListingView)) else None)
+ ea = view._cursor_ea() if isinstance(view, ListingView) else None
if ea is None:
self._status("no address on this line to (re)define")
return
@@ -5438,7 +5492,8 @@ class IdaTui(App):
anchor: ViewAnchor | None = None) -> None: # worker context
assert self.program is not None
verb = {"code": "defined code", "func": "created function",
- "undef": "undefined", "string": "made string"}[kind]
+ "undef": "undefined", "string": "made string",
+ "thumb": "switched decoding", "thumbscan": "scanned"}[kind]
try:
if kind == "code":
# Keep going until something stops it: one instruction is rarely
@@ -5463,12 +5518,53 @@ class IdaTui(App):
"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 == "thumbscan":
+ # A vector table is a list of Thumb entry points that IDA won't
+ # follow on a headerless image, because nothing tells it those
+ # words are pointers. Scan from the cursor.
+ anchor.refresh_functions = True
+ r = self.program.thumb_scan(ea, ea + 0x400)
+ n, applied = int(r.get("n", 0)), int(r.get("applied", 0))
+ if not n:
+ verb = (f"no Thumb entry pointers in {ea:#x}\u2013{ea+0x400:#x}"
+ " (odd words pointing into the image)")
+ else:
+ verb = (f"{n} Thumb entr{'y' if n == 1 else 'ies'} found, "
+ f"{applied} disassembled")
+ elif kind == "thumb":
+ # Switch the mode, then disassemble in it: flipping T and
+ # leaving the bytes undefined shows nothing, and the reason you
+ # flipped it was to read the code.
+ r = self.program.set_thumb(ea)
+ run = self.program.define_code_run(ea)
+ n = int(run.get("count", 0))
+ mode = "Thumb" if r.get("thumb") else "ARM"
+ verb = f"{mode} @ {ea:#x}"
+ if r.get("forced_32bit"):
+ verb += " (segment set to 32-bit; Thumb needs ARM32)"
+ if r.get("db_64bit"):
+ # Disassembly will look right and F5 will never work.
+ verb += (" \u26a0 this database is 64-bit, so Hex-Rays "
+ "won't decompile it \u2014 Ctrl+L and pick "
+ "arm:ARMv7-A")
+ verb += (f" \u2014 {n} instruction{'s' if n != 1 else ''}"
+ if n else " \u2014 still doesn't decode")
+ # falls through to the shared reload: same cache bump, same
+ # anchor restore, same flash. That is the whole point of having
+ # one path.
elif kind == "func":
- self.program.define_func(ea)
+ anchor.refresh_functions = True
+ r = self.program.define_func(ea)
+ if r.get("start") and r.get("end"):
+ verb = (f"created function {r['start']}\u2013{r['end']}"
+ + (" (end worked out from the code)"
+ if r.get("how") == "explicit-end" else ""))
elif kind == "string":
s = self.program.make_string(ea)
verb = f"made string ({s[:24]!r})" if s else verb
else:
+ # Undefining can destroy a function as easily as `p` creates one.
+ anchor.refresh_functions = True
self.program.undefine(ea)
except Exception as e: # noqa: BLE001 -- surface soft/hard tool errors
self.app.call_from_thread(self._status, f"{kind}: {e}")
@@ -5504,9 +5600,375 @@ class IdaTui(App):
written and lost.
"""
self._dirty = True
- self._flash = anchor.flash
if anchor.flash:
- self._status(anchor.flash)
+ self._status(anchor.flash, priority=True)
+ if anchor.refresh_functions:
+ # Creating (or destroying) a function changes the index that the
+ # names pane, Ctrl+N and the "no functions" hint all read. Without
+ # this, `p` gave you a function the rest of the app couldn't see.
+ self._reindex_functions()
+
+ def _load_trace(self) -> None:
+ """Parse the trace and line it up with the database.
+
+ Runs after the function index exists: rebasing needs the database's
+ addresses, and without it nothing in the trace matches anything on
+ screen (our echo trace runs at 0x7ffff6faa000; the database has that
+ code at 0x2000).
+ """
+ from .trace import Trace
+ path = self._trace_path
+ try:
+ def note(n):
+ self.app.call_from_thread(
+ self._status, f"trace: {n:,} instructions\u2026")
+ trace = Trace.load(path, progress=note)
+ except OSError as e:
+ self.app.call_from_thread(self._status, f"trace: {e}")
+ return
+ if not trace.length:
+ self.app.call_from_thread(
+ self._status, f"trace: {os.path.basename(path)} is empty")
+ return
+ idx = self._func_index
+ addrs = [f.addr for f in idx.all_loaded()] if idx is not None else []
+ slide = trace.rebase(addrs)
+ trace.apply_slide(slide)
+ hit = sum(1 for f in (idx.all_loaded() if idx else []) if trace.executions(f.addr))
+ self.app.call_from_thread(self._trace_ready, trace, slide, hit)
+
+ def _trace_ready(self, trace, slide: int, hit: int) -> None:
+ self._trace = trace
+ self._t = 0
+ dock = self.query_one(TraceDock)
+ dock.display = True
+ dock.show(trace, 0)
+ where = (f"rebased {slide:+#x}" if slide else "no rebase needed")
+ self._status(f"trace: {trace.length:,} instructions, {hit} functions "
+ f"touched ({where})", priority=True)
+ self._seek(0, follow=True)
+
+ # -- trace navigation --------------------------------------------------- #
+ def _seek(self, idx: int, follow: bool = True) -> None:
+ """Move to timestamp ``idx``; ``follow`` takes the code view with it."""
+ t = self._trace
+ if t is None or not t.length:
+ return
+ self._t = max(0, min(int(idx), t.length - 1))
+ self.query_one(TraceDock).show(t, self._t)
+ self._paint_trail()
+ 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.
+
+ Recomputed per seek rather than per repaint: it's ~200 lookups, and a
+ repaint happens far more often than a step.
+ """
+ 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)
+ lst.trail = trail
+ lst.refresh()
+ except Exception: # noqa: BLE001 -- view not mounted yet
+ pass
+ self._paint_trail_decomp(trail)
+
+ def _paint_trail_decomp(self, trail: dict) -> None:
+ """Map the instruction trail onto pseudocode lines.
+
+ This is the thing Tenet can't do: it paints disassembly, because that's
+ where a trace's addresses live. We already have decomp_map (built for
+ the split view) saying which instructions each pseudocode line covers,
+ so the same trail lands on C.
+
+ A line covers many instructions, so it takes the strongest kind present:
+ 'now' wins over 'past' wins over 'future' — if the instruction you are
+ standing on is part of this line, this line is where you are.
+ """
+ try:
+ dec = self.query_one(DecompView)
+ except Exception: # noqa: BLE001
+ return
+ ea = dec.loaded_ea
+ if not dec.display or ea is None or self.program is None:
+ dec.trail = {}
+ return
+ if self._trail_map_ea != ea:
+ # 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._apply_split_map(ea, self.program.decomp_map(ea))
+ except Exception: # noqa: BLE001
+ 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 []):
+ best = None
+ for a in eas:
+ k = trail.get(a)
+ if k is not None and (best is None or rank[k] > rank[best]):
+ best = k
+ if best is not None:
+ 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:
+ self._status("no trace loaded (--trace FILE)")
+ 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)
+
+ def action_step_back(self) -> None:
+ self._step(-1)
+
+ def _step_over(self, direction: int) -> None:
+ """Step over a call by following the stack pointer.
+
+ A call pushes, so the callee runs with SP BELOW where we started;
+ stepping until SP comes back up lands after the call returns. Cheaper
+ and more robust than recognising call instructions per architecture,
+ which is what the mode makes it: if this instruction doesn't call
+ anything, SP is already >= the start and it degenerates to one step.
+ """
+ t = self._trace
+ if t is None:
+ self._status("no trace loaded (--trace FILE)")
+ return
+ sp_name = "rsp" if "rsp" in t.reg_at else ("esp" if "esp" in t.reg_at else "sp")
+ sp0 = t.register(sp_name, self._t)
+ i = self._t + direction
+ limit = 200000 # a runaway search must not hang the UI
+ while 0 <= i < t.length and limit > 0:
+ sp = t.register(sp_name, i)
+ if sp0 is None or sp is None or sp >= sp0:
+ break
+ i += direction
+ limit -= 1
+ self._seek(max(0, min(i, t.length - 1)))
+
+ def action_step_over_fwd(self) -> None:
+ self._step_over(1)
+
+ def action_step_over_back(self) -> None:
+ self._step_over(-1)
+
+ @work(thread=True, exclusive=True, group="load-funcs")
+ def _reindex_functions(self) -> None:
+ """Rebuild the function index in place after an edit changed it.
+
+ Deliberately not _load_functions(): that one is the BOOT path — it
+ clears the table, streams progress and then auto-lands, which would
+ yank the view away from the function you just made.
+ """
+ if self.program is None:
+ return
+ idx = self.program.functions()
+ idx.load_all()
+ self._func_index = idx
+ self.app.call_from_thread(self._after_reindex)
+
+ def _after_reindex(self) -> None:
+ idx = self._func_index
+ if idx is None:
+ return
+ if len(idx):
+ self._no_functions = False
+ self._apply_filter(self._filter_term) # repopulate the names pane
@work(thread=True, exclusive=True, group="save")
def _save(self) -> None:
@@ -5784,6 +6246,8 @@ class IdaTui(App):
view.focus()
def on_key(self, event) -> None: # type: ignore[no-untyped-def]
+ # Any keypress means the result of the last edit has been read.
+ self._flash = None
if event.key != "escape":
return
if self.query_one("#search", Input).display:
@@ -6153,10 +6617,6 @@ class IdaTui(App):
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}]")
@@ -6164,12 +6624,22 @@ class IdaTui(App):
def _load_decomp(self, ea: int, name: str) -> None:
assert self.program is not None
dec = self.program.decompile(ea)
- self.app.call_from_thread(self._apply_decomp, ea, name, dec)
+ why = ""
+ if dec.failed:
+ # Ask Hex-Rays why, in the same worker: the plain tool reports
+ # "Decompilation failed at 0x0" and drops the only useful part.
+ # "Decompile failed" with no reason is indistinguishable from a bug
+ # in this app, and for the common cause (a 32-bit function in a
+ # 64-bit database) the user cannot even guess the fix.
+ why = self.program.decomp_error(ea)
+ self.app.call_from_thread(self._apply_decomp, ea, name, dec, why)
- def _apply_decomp(self, ea: int, name: str, dec) -> None: # type: ignore[no-untyped-def]
+ def _apply_decomp(self, ea: int, name: str, dec, # type: ignore[no-untyped-def]
+ why: str = "") -> None:
view = self.query_one(DecompView)
view.loading = False
if dec.failed:
+ detail = f" \u2014 {why}" if why else ""
# No pseudocode for this function: fall back to the code view rather
# than an error panel. If we came from the continuous listing (F5),
# return there; otherwise show the disassembly.
@@ -6180,13 +6650,17 @@ class IdaTui(App):
self._decomp_return = None
self._cur = ret
self._active = "listing"
+ # Hand the reason over as a flash BEFORE reopening: going back
+ # to the listing reloads it, and the reload writes its own
+ # status afterwards — which is precisely how "F5 does nothing"
+ # looked like nothing at all.
+ msg = f"{name}: cannot decompile{detail}"
+ self._status(msg, priority=True)
self._open_entry(ret, push=False)
- self._status(f"{name} — decompile failed; back to the listing")
return
self._active = "disasm"
+ self._status(f"{name}: cannot decompile{detail}", priority=True)
self._show_active()
- self._status(
- f"{name} — no pseudocode (decompile failed); showing disassembly")
return
if self._active == "decomp":
view.focus() # loading cover had blurred it; restore focus
@@ -6341,8 +6815,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 = []
@@ -6353,7 +6837,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)
@@ -6393,13 +6885,6 @@ 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 e4fbec9..97b042d 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -100,6 +100,10 @@ class Head:
text: str
name: str | None = None
raw: bytes | None = None # opcode/item bytes (filled in for code by the model)
+ #: [(kind, text)] from IDA's own colour tags — mnem/reg/num/name/str/punct/…
+ #: None when the worker didn't provide them (older worker, or the spans
+ #: disagreed with the plain text, in which case the text wins).
+ spans: tuple[tuple[str, str], ...] | None = None
@property
def label(self) -> str | None: # Line-compatible alias
@@ -107,12 +111,15 @@ class Head:
@classmethod
def from_raw(cls, d: dict) -> "Head":
+ sp = d.get("spans")
return cls(
ea=_as_int(d["ea"]),
kind=d.get("kind", "unknown"),
size=int(d.get("size", 0) or 0),
text=d.get("text", ""),
name=d.get("name"),
+ spans=(tuple((str(k), str(t)) for k, t in sp)
+ if isinstance(sp, list) and sp else None),
)
@@ -1366,6 +1373,43 @@ class Program:
if res.get("error"):
raise IDAToolError("define_code", f"@ {ea:#x}: {res['error']}")
+ def decomp_error(self, ea: int) -> str:
+ """Hex-Rays' own reason for refusing ``ea``, or "" if it won't say."""
+ try:
+ r = self.client.call("decomp_error", addr=hex(ea))
+ except IDAToolError:
+ return ""
+ if not isinstance(r, dict):
+ return ""
+ reason = str(r.get("reason") or "")
+ if reason and r.get("bitness") == 64 and "64-bit" in reason:
+ # Say the FIX, not the diagnosis. Hex-Rays' own sentence ("only
+ # 64-bit functions can be decompiled in the current database") is
+ # accurate and useless: it describes the database, not what to do,
+ # and it's long enough that a status bar cuts off the end — which is
+ # exactly where an appended hint would live. This is unfixable in
+ # place (bitness is decided at load), so the whole message is the
+ # instruction.
+ return "this database is 64-bit \u2014 Ctrl+L, pick arm:ARMv7-A"
+ return reason
+
+ def thumb_scan(self, start: int, end: int, apply: bool = True) -> dict:
+ """Find Thumb entry points from odd pointers in ``[start, end)``."""
+ r = self.client.call("thumb_scan", start=hex(start), end=hex(end),
+ apply=bool(apply))
+ if not isinstance(r, dict) or r.get("error"):
+ raise IDAToolError("thumb_scan",
+ f"@ {start:#x}: {(r or {}).get('error', 'failed')}")
+ return r
+
+ def set_thumb(self, ea: int, mode: str = "toggle") -> dict:
+ """Switch ARM/Thumb decoding at ``ea``. Returns the resulting state."""
+ r = self.client.call("set_thumb", addr=hex(ea), mode=mode)
+ if not isinstance(r, dict) or r.get("error"):
+ raise IDAToolError("set_thumb",
+ f"@ {ea:#x}: {(r or {}).get('error', 'failed')}")
+ return r
+
def define_code_run(self, ea: int, limit: int = 20000) -> dict:
"""Disassemble consecutively from ``ea`` until something stops it.
@@ -1382,12 +1426,24 @@ class Program:
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("define_func", f"@ {ea:#x}: {res['error']}")
+ def define_func(self, ea: int) -> dict:
+ """Create a function starting at ``ea`` (IDA's 'p').
+
+ Prefers the injected tool, which works out the end when IDA can't;
+ falls back to the plain one for an older worker.
+ """
+ try:
+ r = self.client.call("define_func_run", addr=hex(ea))
+ except IDAToolError:
+ res = self._first_result(
+ self.client.call("define_func", items=[{"addr": hex(ea)}]))
+ if res.get("error"):
+ raise IDAToolError("define_func", f"@ {ea:#x}: {res['error']}")
+ return {"ok": True, "how": "legacy"}
+ if not isinstance(r, dict) or not r.get("ok"):
+ raise IDAToolError("define_func",
+ f"@ {ea:#x}: {(r or {}).get('error', 'failed')}")
+ return r
def undefine(self, ea: int, size: int | None = None) -> None:
"""Undefine the item at ``ea`` back to raw bytes (IDA's 'u')."""
diff --git a/idatui/formats.py b/idatui/formats.py
index b2f784d..f09bb99 100644
--- a/idatui/formats.py
+++ b/idatui/formats.py
@@ -92,9 +92,18 @@ def needs_load_options(path: str) -> bool:
#: reach for first — arm64, aarch64, mips, m68k — are all invalid. Re-run the
#: script before adding to this list.
PROCESSORS: tuple[tuple[str, str], ...] = (
- # 'arm' covers AArch64 too; arm64/aarch64 are NOT valid -p names, so they
- # live in the label where the filter can still find them.
- ("arm", "ARM / AArch64 / arm64 — little-endian"),
+ # Bare 'arm' gives a 64-BIT database in IDA 9 (AArch64). That matters far
+ # more than it looks: Hex-Rays refuses a 32-bit function in a 64-bit database
+ # ("only 64-bit functions can be decompiled in the current database"), and
+ # Thumb doesn't exist in AArch64 at all — so a 32-bit ARM image loaded as
+ # plain 'arm' disassembles wrongly and can never be decompiled. The database
+ # bitness is fixed at load; it cannot be corrected afterwards (setting it
+ # post-hoc makes the decompiler INTERR). Pick the right one here.
+ ("arm", "ARM64 / AArch64 / arm64 — 64-bit, little-endian"),
+ ("arm:ARMv7-A", "ARM 32-bit (ARMv7-A) — Thumb capable, most firmware"),
+ ("arm:ARMv7-M", "ARM 32-bit (ARMv7-M) — Cortex-M, Thumb only"),
+ ("arm:ARMv6-M", "ARM 32-bit (ARMv6-M) — Cortex-M0/M0+"),
+ ("arm:ARMv5TE", "ARM 32-bit (ARMv5TE) — older SoCs"),
("armb", "ARM — big-endian"),
("metapc", "x86 / x86-64"),
("mipsl", "MIPS — little-endian"),
diff --git a/idatui/launch.py b/idatui/launch.py
index f51e5af..64e1546 100644
--- a/idatui/launch.py
+++ b/idatui/launch.py
@@ -64,6 +64,8 @@ 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)")
+ p.add_argument("--trace", metavar="FILE",
+ help="Tenet execution trace to explore alongside the binary")
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 "
@@ -157,7 +159,9 @@ def main(argv: list[str] | None = None) -> int:
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,
- load_args=_load_args(load)).run()
+ load_args=_load_args(load),
+ trace_path=(os.path.abspath(os.path.expanduser(args.trace))
+ if args.trace else "")).run()
return 0
diff --git a/idatui/pane.py b/idatui/pane.py
index 53afef3..1a46a4f 100644
--- a/idatui/pane.py
+++ b/idatui/pane.py
@@ -165,6 +165,8 @@ def spawn(args) -> int:
inner += ["--rpc", sock]
else:
inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock]
+ if getattr(args, "trace", None):
+ inner += ["--trace", os.path.abspath(os.path.expanduser(args.trace))]
cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner)
split = ["split-window", "-v" if args.vertical else "-h",
@@ -316,6 +318,8 @@ def main(argv: list[str]) -> int:
sp = sub.add_parser("spawn", help="open a TUI pane and wait until ready")
sp.add_argument("--open", metavar="PATH",
help="binary to open (its dir must be writable)")
+ sp.add_argument("--trace", metavar="FILE",
+ help="Tenet execution trace to load alongside the binary")
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)")
diff --git a/idatui/rpc.py b/idatui/rpc.py
index 0921b89..7741f48 100644
--- a/idatui/rpc.py
+++ b/idatui/rpc.py
@@ -72,6 +72,7 @@ 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)",
+ "trace": "{seek|goto|step,over?} navigate the execution trace (seek '!50' = percent)",
"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)",
@@ -531,6 +532,38 @@ class RpcServer:
return functions(app, params.get("filter"), int(params.get("limit", 50)))
# -- projects ------------------------------------------------------ #
+ if method == "trace":
+ if app._trace is None:
+ raise ValueError("no trace loaded (launch with --trace FILE)")
+ t = app._trace
+ if "seek" in params:
+ v = params["seek"]
+ # "!50" seeks a percentage, like Tenet's timestamp shell.
+ if isinstance(v, str) and v.startswith("!"):
+ idx = int(float(v[1:]) * (t.length - 1) / 100.0)
+ else:
+ idx = int(str(v).replace(",", ""), 0) if isinstance(v, str) else int(v)
+ app._seek(idx)
+ elif "goto" in params: # first execution of an address/name
+ tgt = params["goto"]
+ ea = (int(str(tgt), 0) if str(tgt).lower().startswith("0x")
+ else app.program.resolve(str(tgt)))
+ first = t.first_execution(ea)
+ if first is None:
+ raise ValueError(f"{tgt} never executed in this trace")
+ app._seek(first)
+ elif "step" in params:
+ n = int(params.get("step") or 1)
+ over = bool(params.get("over"))
+ for _ in range(abs(n)):
+ (app._step_over if over else app._step)(1 if n > 0 else -1)
+ await settle(app, timeout=float(params.get("timeout", 20.0)))
+ snap = snapshot(app)
+ snap["trace"] = {"idx": app._t, "length": t.length,
+ "pc": hex(t.ip(app._t)),
+ "changed": sorted(t.changed(app._t))}
+ return snap
+
if method == "binaries":
if app._project is None:
raise ValueError("not a project session (launch with --project)")
diff --git a/idatui/trace.py b/idatui/trace.py
new file mode 100644
index 0000000..931f918
--- /dev/null
+++ b/idatui/trace.py
@@ -0,0 +1,495 @@
+"""Reading Tenet execution traces.
+
+A Tenet trace is a line-per-instruction delta log::
+
+ rax=0x3c,rbx=0x0,...,rip=0x7ffff6faaae0 # full state on the first line
+ rip=0x7ffff6faaae4 # then only what changed
+ r9=0x7ffff6762e60,rip=0x7ffff6faaae9,mw=0x7ffff6f9b7c8:08c177f6ff7f0000
+
+Registers that changed, the PC on every line, and every memory access *with its
+bytes*. That is enough to reconstruct any register or any memory address at any
+point in time, forwards or backwards, which is the whole trick.
+
+This is our own reader, not a port. The reference implementation
+(``~/dev/tenet/tenet-original/plugins/tenet/trace/``) packs the trace into
+segments with compressed address/mask tables, which earns its keep for its Qt
+timeline; we need different queries and would rather own the ~400 lines than
+inherit 3700. It is differential-tested against that implementation
+(``tests/test_trace_vs_tenet.py``) so "our own" doesn't quietly mean "different".
+
+The index is built around the query the UI actually asks, which the reference
+answers one address at a time: **which timestamps executed this set of
+addresses**. A listing row is one address, but a pseudocode line covers many, so
+``by_ip`` maps address -> timestamps and set queries are unions of those.
+"""
+
+from __future__ import annotations
+
+import array
+import bisect
+import os
+import re
+from dataclasses import dataclass, field
+
+#: Tenet packs its register delta into a uint32, so a trace arch may name at
+#: most 32 registers. Ours is discovered from the trace instead of declared,
+#: but the cap is worth knowing when a trace looks short of registers.
+MAX_REGISTERS = 32
+
+_MEM_RE = re.compile(r"^m(r|w|rw)$")
+
+
+@dataclass
+class TraceInfo:
+ """The sidecar ``<prefix>.info`` written by our QEMU tracer.
+
+ Optional: a trace from another tracer has none, and everything here can be
+ recovered or guessed from the log itself.
+ """
+
+ arch: str = ""
+ mode: str = ""
+ binary: str = ""
+ start_code: int = 0
+ end_code: int = 0
+ entry_code: int = 0
+ traced: str = ""
+
+ @classmethod
+ def load(cls, path: str) -> "TraceInfo | None":
+ try:
+ with open(path) as f:
+ raw = dict(
+ ln.strip().split("=", 1) for ln in f if "=" in ln)
+ except OSError:
+ return None
+ def num(k):
+ try:
+ return int(raw.get(k, "0"), 0)
+ except ValueError:
+ return 0
+ return cls(arch=raw.get("arch", ""), mode=raw.get("mode", ""),
+ binary=raw.get("binary", ""), start_code=num("start_code"),
+ end_code=num("end_code"), entry_code=num("entry_code"),
+ traced=raw.get("traced", ""))
+
+
+@dataclass
+class MemOp:
+ """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
+ write: bool
+
+ @property
+ def end(self) -> int:
+ return self.addr + len(self.data)
+
+
+@dataclass
+class Trace:
+ """An indexed Tenet trace.
+
+ Timestamps are indices into the executed-instruction sequence: 0 is the
+ first instruction, ``length - 1`` the last.
+ """
+
+ path: str = ""
+ info: TraceInfo | None = None
+ #: PC per timestamp.
+ ips: array.array = field(default_factory=lambda: array.array("Q"))
+ #: name -> (timestamps of change, value at each change). A register's value
+ #: at time t is the last change at or before t; the first line carries a
+ #: full state dump, so every register has an entry at 0.
+ reg_at: dict[str, tuple[array.array, array.array]] = field(default_factory=dict)
+ #: address -> timestamps that executed it (ascending, by construction).
+ by_ip: dict[int, array.array] = field(default_factory=dict)
+ #: memory accesses, parallel arrays indexed by access number.
+ mem_idx: array.array = field(default_factory=lambda: array.array("I"))
+ mem_addr: array.array = field(default_factory=lambda: array.array("Q"))
+ mem_write: bytearray = field(default_factory=bytearray)
+ mem_off: array.array = field(default_factory=lambda: array.array("Q"))
+ mem_len: array.array = field(default_factory=lambda: array.array("H"))
+ mem_blob: bytearray = field(default_factory=bytearray)
+ #: first access number of each timestamp, plus a final sentinel.
+ 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
+ def load(cls, path: str, progress=None, limit: int = 0) -> "Trace":
+ """Parse a text trace. ``progress(lines)`` is called every 50k lines."""
+ t = cls(path=os.path.abspath(path))
+ base = path[:-6] if path.endswith(".0.log") else os.path.splitext(path)[0]
+ t.info = TraceInfo.load(base + ".info")
+ regs: dict[str, tuple[array.array, array.array]] = {}
+ ips, by_ip = t.ips, t.by_ip
+ n = 0
+ with open(path) as f:
+ for line in f:
+ line = line.strip()
+ if not line:
+ continue
+ ip = None
+ t.mem_row.append(len(t.mem_idx))
+ for part in line.split(","):
+ key, _, val = part.partition("=")
+ if not val:
+ continue
+ key = key.strip().lower()
+ if _MEM_RE.match(key):
+ addr_s, _, data_s = val.partition(":")
+ try:
+ addr = int(addr_s, 16)
+ data = bytes.fromhex(data_s)
+ except ValueError:
+ continue
+ # 'mrw' is one access that both reads and writes; record
+ # the write, which is what the memory state follows.
+ t.mem_idx.append(n)
+ t.mem_addr.append(addr)
+ t.mem_write.append(0 if key == "mr" else 1)
+ t.mem_off.append(len(t.mem_blob))
+ t.mem_len.append(len(data))
+ t.mem_blob += data
+ continue
+ try:
+ v = int(val, 16)
+ except ValueError:
+ continue
+ slot = regs.get(key)
+ if slot is None:
+ slot = regs[key] = (array.array("I"), array.array("Q"))
+ slot[0].append(n)
+ slot[1].append(v)
+ ip = v if key in ("rip", "eip", "pc") else ip
+ if ip is None:
+ # No PC on this line: the format says always emit it, and
+ # without it the line cannot be placed. Carry the previous
+ # one rather than dropping the instruction.
+ ip = ips[-1] if ips else 0
+ ips.append(ip)
+ where = by_ip.get(ip)
+ if where is None:
+ where = by_ip[ip] = array.array("I")
+ where.append(n)
+ n += 1
+ if progress is not None and n % 50000 == 0:
+ progress(n)
+ if limit and n >= limit:
+ break
+ t.mem_row.append(len(t.mem_idx))
+ t.reg_at = regs
+ return t
+
+ # -- basics ------------------------------------------------------------ #
+ def __len__(self) -> int:
+ return len(self.ips)
+
+ @property
+ def length(self) -> int:
+ return len(self.ips)
+
+ @property
+ def registers(self) -> list[str]:
+ """Register names in the trace, PC last (the order tracers emit)."""
+ return sorted(self.reg_at, key=lambda r: (r in ("rip", "eip", "pc"), r))
+
+ @property
+ def pc_name(self) -> str:
+ for n in ("rip", "eip", "pc"):
+ if n in self.reg_at:
+ return n
+ return ""
+
+ def ip(self, idx: int) -> int:
+ """PC at ``idx``, in DATABASE addresses (slide applied)."""
+ return self.ips[idx] + self.slide
+
+ def raw_ip(self, idx: int) -> int:
+ return self.ips[idx]
+
+ # -- register state ----------------------------------------------------- #
+ def register(self, name: str, idx: int) -> int | None:
+ """Value of ``name`` at ``idx``, or None if the trace never set it."""
+ slot = self.reg_at.get(name.lower())
+ if slot is None:
+ return None
+ idxs, vals = slot
+ i = bisect.bisect_right(idxs, idx) - 1
+ return vals[i] if i >= 0 else None
+
+ def register_state(self, idx: int) -> dict[str, int]:
+ return {n: v for n in self.reg_at
+ if (v := self.register(n, idx)) is not None}
+
+ def changed(self, idx: int) -> set[str]:
+ """Registers written BY the instruction at ``idx`` (what the line said).
+
+ Used to highlight what an instruction actually did, which is the reason
+ a delta trace is readable at all.
+ """
+ out = set()
+ for n, (idxs, _vals) in self.reg_at.items():
+ i = bisect.bisect_left(idxs, idx)
+ if i < len(idxs) and idxs[i] == idx:
+ out.add(n)
+ return out
+
+ def last_write(self, name: str, idx: int) -> int | None:
+ """Timestamp of the write that produced ``name``'s value at ``idx``.
+
+ "Which instruction set this register?" — the question that motivates a
+ trace explorer in the first place.
+ """
+ slot = self.reg_at.get(name.lower())
+ if slot is None:
+ return None
+ i = bisect.bisect_right(slot[0], idx) - 1
+ return slot[0][i] if i >= 0 else None
+
+ def next_write(self, name: str, idx: int) -> int | None:
+ slot = self.reg_at.get(name.lower())
+ if slot is None:
+ return None
+ i = bisect.bisect_right(slot[0], idx)
+ return slot[0][i] if i < len(slot[0]) else None
+
+ # -- memory ------------------------------------------------------------- #
+ def memory_ops(self, idx: int) -> list[MemOp]:
+ """Accesses made by the instruction at ``idx``."""
+ if not (0 <= idx < len(self.mem_row) - 1):
+ return []
+ lo, hi = self.mem_row[idx], self.mem_row[idx + 1]
+ 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],
+ 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)."""
+ return self.by_ip.get(ea - self.slide, array.array("I"))
+
+ def executions_between(self, ea: int, lo: int, hi: int) -> list[int]:
+ ts = self.executions(ea)
+ a = bisect.bisect_left(ts, lo)
+ b = bisect.bisect_right(ts, hi)
+ return list(ts[a:b])
+
+ def hits(self, eas) -> dict[int, int]:
+ """{address: execution count} for a set of addresses.
+
+ The set form is the point: painting a listing row needs one address, but
+ a pseudocode line covers many, and asking per-address would mean a
+ lookup per instruction per repaint.
+ """
+ out = {}
+ for ea in eas:
+ ts = self.by_ip.get(ea - self.slide)
+ if ts:
+ out[ea] = len(ts)
+ return out
+
+ def prev_ips(self, idx: int, n: int) -> list[int]:
+ """Addresses executed in the ``n`` steps before ``idx`` (nearest first).
+
+ A trail, not all of history: showing every address the trace ever
+ touched says almost nothing on a loop-heavy program, whereas the last
+ few dozen steps say how you GOT here.
+ """
+ lo = max(idx - n, 0)
+ return [self.ips[i] + self.slide for i in range(idx - 1, lo - 1, -1)]
+
+ def next_ips(self, idx: int, n: int) -> list[int]:
+ """Addresses executed in the ``n`` steps after ``idx`` (nearest first)."""
+ hi = min(idx + n + 1, self.length)
+ return [self.ips[i] + self.slide for i in range(idx + 1, hi)]
+
+ def trail(self, idx: int, n: int = 96) -> dict[int, str]:
+ """{address: 'now' | 'past' | 'future'} around ``idx``.
+
+ Where an address appears on both sides — a loop body, which is most of
+ them — the nearer side wins, because that's the one that explains the
+ step you are about to take or just took.
+ """
+ out: dict[int, str] = {}
+ for k, ea in enumerate(self.next_ips(idx, n)):
+ out.setdefault(ea, "future")
+ for k, ea in enumerate(self.prev_ips(idx, n)):
+ prev = out.get(ea)
+ if prev is None:
+ out[ea] = "past"
+ elif prev == "future":
+ # Same distance rule as above, resolved by which loop found it
+ # first would be arbitrary; compare real distances instead.
+ fwd = next((i for i, a in enumerate(self.next_ips(idx, n)) if a == ea), n)
+ if k < fwd:
+ out[ea] = "past"
+ if 0 <= idx < self.length:
+ out[self.ips[idx] + self.slide] = "now"
+ return out
+
+ def first_execution(self, ea: int) -> int | None:
+ ts = self.executions(ea)
+ return ts[0] if ts else None
+
+ def next_execution(self, ea: int, idx: int) -> int | None:
+ ts = self.executions(ea)
+ i = bisect.bisect_right(ts, idx)
+ return ts[i] if i < len(ts) else None
+
+ def prev_execution(self, ea: int, idx: int) -> int | None:
+ ts = self.executions(ea)
+ i = bisect.bisect_left(ts, idx) - 1
+ return ts[i] if i >= 0 else None
+
+ # -- lining the trace up with the database ------------------------------ #
+ def rebase(self, db_addresses) -> int:
+ """Find the slide between trace addresses and database addresses.
+
+ A traced process is relocated (ASLR, or a PIE base the database doesn't
+ share): our echo trace runs at 0x7ffff6faa000 while the database has the
+ same code at 0x2490. Nothing lines up until this is solved, so it is not
+ optional.
+
+ Page offsets survive relocation — only whole pages move — so the low 12
+ bits of an instruction address are invariant. Bucket the database's
+ addresses by those bits, and for each trace address the candidate slides
+ are the differences to database addresses in its bucket. The slide that
+ the most instructions agree on wins.
+ """
+ buckets: dict[int, list[int]] = {}
+ for a in db_addresses:
+ buckets.setdefault(a & 0xFFF, []).append(a)
+ if not buckets:
+ return 0
+ votes: dict[int, int] = {}
+ # A sample is enough and keeps this O(1)-ish on a 10M trace; unique
+ # addresses, because a hot loop shouldn't outvote the rest of the code.
+ for ea in list(self.by_ip)[:4096]:
+ for cand in buckets.get(ea & 0xFFF, ()):
+ votes[cand - ea] = votes.get(cand - ea, 0) + 1
+ if not votes:
+ return 0
+ best, n = max(votes.items(), key=lambda kv: kv[1])
+ return best if n > 1 else 0
+
+ def apply_slide(self, slide: int) -> None:
+ self.slide = int(slide)
diff --git a/server/patch_server.py b/server/patch_server.py
index 4c806a7..160b15b 100644
--- a/server/patch_server.py
+++ b/server/patch_server.py
@@ -305,12 +305,137 @@ def _idatui_head_row(ea):
"size": int(ida_bytes.get_item_size(ea)),
"text": text,
}
+ if line:
+ # Keep IDA's own token classification for syntax highlighting. Built from
+ # the SAME line as `text`, then whitespace-collapsed identically so the
+ # two never disagree about what the row says.
+ spans = _idatui_spans(line)
+ joined = "".join(t for _k, t in spans)
+ if " ".join(joined.split()) == text:
+ row["spans"] = spans
nm = ida_name.get_ea_name(ea)
if nm:
row["name"] = nm
return row
+#: IDA colour tag -> the semantic kind the TUI styles. IDA already classifies
+#: every token in a disassembly line, for every processor it supports, so there
+#: is nothing to lex: generate_disasm_line emits \x01<tag>text\x02<tag> and the
+#: tag says what the text IS. A pygments assembly lexer would be a worse guess at
+#: this and would need one dialect per architecture.
+_IDATUI_SPAN_KINDS = {
+ "insn": ("SCOLOR_INSN", "SCOLOR_KEYWORD", "SCOLOR_ASMDIR", "SCOLOR_MACRO"),
+ "reg": ("SCOLOR_REG",),
+ "num": ("SCOLOR_NUMBER", "SCOLOR_CHAR", "SCOLOR_BINPREF"),
+ "str": ("SCOLOR_STRING",),
+ # NB the real constant names: DATNAME/CODNAME, not "DNAME". Guessing here
+ # fails silently — an unmapped tag renders as plain body text, so symbols
+ # just quietly aren't blue and nothing tells you why.
+ "name": ("SCOLOR_DATNAME", "SCOLOR_CODNAME", "SCOLOR_LOCNAME",
+ "SCOLOR_IMPNAME", "SCOLOR_DEMNAME", "SCOLOR_LIBNAME",
+ "SCOLOR_CNAME", "SCOLOR_DNAME",
+ "SCOLOR_CREF", "SCOLOR_DREF", "SCOLOR_CREFTAIL", "SCOLOR_DREFTAIL"),
+ "seg": ("SCOLOR_SEGNAME",),
+ "cmt": ("SCOLOR_AUTOCMT", "SCOLOR_REGCMT", "SCOLOR_RPTCMT", "SCOLOR_VOIDOP"),
+ "punct": ("SCOLOR_SYMBOL", "SCOLOR_ALTOP", "SCOLOR_HIDNAME"),
+ "err": ("SCOLOR_ERROR",),
+}
+
+
+def _idatui_tag_map():
+ """{tag character: kind}, built once from whatever this IDA actually has."""
+ import ida_lines
+ out = {}
+ for kind, names in _IDATUI_SPAN_KINDS.items():
+ for n in names:
+ v = getattr(ida_lines, n, None)
+ if isinstance(v, str) and v:
+ out[v[0]] = kind
+ elif isinstance(v, int):
+ out[chr(v)] = kind
+ return out
+
+
+_IDATUI_TAGS = None
+
+
+def _idatui_spans(line):
+ """A tagged disasm line as [[kind, text], ...], colour tags resolved.
+
+ Unknown tags become 'text' rather than being dropped: a processor module can
+ emit a colour we don't classify, and losing the characters would corrupt the
+ line."""
+ global _IDATUI_TAGS
+ import ida_lines
+ if _IDATUI_TAGS is None:
+ _IDATUI_TAGS = _idatui_tag_map()
+ on, off, esc = "\x01", "\x02", "\x03"
+ addr_tag = chr(getattr(ida_lines, "COLOR_ADDR", 0x28))
+ addr_len = int(getattr(ida_lines, "COLOR_ADDR_SIZE", 16))
+ spans, stack, buf = [], [], []
+ i, n = 0, len(line)
+
+ def flush():
+ if buf:
+ spans.append([stack[-1] if stack else "text", "".join(buf)])
+ del buf[:]
+
+ while i < n:
+ ch = line[i]
+ if ch == on and i + 1 < n:
+ tag = line[i + 1]
+ if tag == addr_tag:
+ # An embedded target address, not display text: 16 hex digits
+ # that must not reach the screen.
+ i += 2 + addr_len
+ continue
+ flush()
+ stack.append(_IDATUI_TAGS.get(tag, "text"))
+ i += 2
+ continue
+ if ch == off and i + 1 < n:
+ flush()
+ if stack:
+ stack.pop()
+ i += 2
+ continue
+ if ch == esc and i + 1 < n: # escaped literal
+ buf.append(line[i + 1])
+ i += 2
+ continue
+ buf.append(ch)
+ i += 1
+ flush()
+ # Collapse IDA's column padding EXACTLY as the plain text does. A run of
+ # spaces can straddle two spans, so this walks characters rather than
+ # collapsing each span on its own — otherwise the spans and `text` disagree
+ # about the line and the row silently loses its highlighting.
+ out, prev_space = [], False
+ for kind, txt in spans:
+ acc = []
+ for ch in txt:
+ if ch.isspace():
+ if prev_space:
+ continue
+ acc.append(" ")
+ prev_space = True
+ else:
+ acc.append(ch)
+ prev_space = False
+ if acc:
+ out.append([kind, "".join(acc)])
+ while out and out[0][1] == " ":
+ out.pop(0)
+ while out and out[-1][1] == " ":
+ out.pop()
+ if out and out[0][1].startswith(" "):
+ out[0][1] = out[0][1].lstrip()
+ if out and out[-1][1].endswith(" "):
+ out[-1][1] = out[-1][1].rstrip()
+ return [[k, t] for k, t in out if t]
+
+
def _idatui_unknown_row(ea, size):
"""One collapsed row for a run of ``size`` undefined bytes starting at
``ea``. A single byte is rendered normally (shows its value); a longer run
@@ -842,6 +967,245 @@ def define_code_run(
return {"start": hex(start), "end": hex(ea), "count": count,
"stopped": stopped}
+
+@tool
+@idasync
+def set_thumb(
+ addr: Annotated[str, "Address to change the ARM decoding mode at"],
+ mode: Annotated[str, "'toggle', 'on' (Thumb) or 'off' (ARM)"] = "toggle",
+ end: Annotated[str, "Optional exclusive end address (default: this item)"] = "",
+) -> dict:
+ """Switch ARM/Thumb decoding at ``addr`` (IDA's T segment register).
+
+ Thumb is not a property of the bytes, it's a mode the CPU is in, so a raw
+ image gives IDA no way to know: at a Thumb entry point it decodes 16-bit
+ instructions as 32-bit ARM and produces confident nonsense
+ (``push {r3,lr}`` reads as ``SVCLT 0xBF00``).
+
+ Also forces the segment to 32-bit when turning Thumb ON. Thumb does not
+ exist in AArch64, and a headerless blob loaded with -parm defaults to
+ 64-bit — so setting T alone changes nothing and looks broken. Asking for
+ Thumb IS asking for ARM32."""
+ import ida_bytes
+ import ida_idp
+ import ida_segment
+ import ida_segregs
+
+ try:
+ ea = parse_address(addr)
+ except Exception as e:
+ return {"addr": str(addr), "error": str(e)}
+ treg = ida_idp.str2reg("T")
+ if treg is None or treg < 0:
+ return {"addr": hex(ea), "error": "no T register (not an ARM database)"}
+ seg = ida_segment.getseg(ea)
+ if not seg:
+ return {"addr": hex(ea), "error": "no segment"}
+
+ import ida_ida
+ db64 = ida_ida.inf_get_app_bitness() == 64
+ cur = ida_segregs.get_sreg(ea, treg)
+ cur = 0 if cur in (None, 0xFFFFFFFF, -1) else int(cur)
+ want = {"on": 1, "off": 0}.get(str(mode).lower(), 0 if cur else 1)
+
+ changed_bits = False
+ if want and seg.bitness != 1:
+ ida_segment.set_segm_addressing(seg, 1)
+ changed_bits = True
+
+ try:
+ stop = parse_address(end) if end else 0
+ except Exception:
+ stop = 0
+ size = max(int(stop) - ea, 0) or max(ida_bytes.get_item_size(ea), 2)
+ # The bytes are currently decoded in the OLD mode; leaving that item defined
+ # pins the wrong instruction length and the new mode has nothing to apply to.
+ ida_bytes.del_items(ea, 0, size)
+ ok = bool(ida_segregs.split_sreg_range(ea, treg, want, ida_segregs.SR_user))
+ now = ida_segregs.get_sreg(ea, treg)
+ return {"addr": hex(ea), "thumb": bool(now), "was": bool(cur), "ok": ok,
+ "bitness": ida_segment.getseg(ea).bitness,
+ "forced_32bit": changed_bits,
+ # The DATABASE's bitness is fixed at load and can't be corrected
+ # here (setting it post-hoc makes the decompiler INTERR). In a
+ # 64-bit database a 32-bit function disassembles but Hex-Rays
+ # refuses it outright, so say so instead of leaving the user to
+ # discover that F5 does nothing.
+ "db_64bit": bool(db64 and want)}
+
+def _idatui_add_func(ea):
+ """add_func at ``ea``, falling back to an explicit end.
+
+ ida_funcs.add_func(ea) asks IDA to find the end and on carved or
+ freshly-marked code it often can't, failing with no reason given."""
+ import ida_bytes
+ import ida_funcs
+ import ida_segment
+ import idaapi
+
+ if idaapi.get_func(ea) is not None:
+ return True
+ if ida_funcs.add_func(ea):
+ return True
+ seg = ida_segment.getseg(ea)
+ hi = seg.end_ea if seg else ea
+ end = ea
+ while end < hi and ida_bytes.is_code(ida_bytes.get_flags(end)):
+ nxt = ida_bytes.get_item_end(end)
+ if nxt <= end:
+ break
+ end = nxt
+ return bool(end > ea and ida_funcs.add_func(ea, end))
+
+
+@tool
+@idasync
+def define_func_run(
+ addr: Annotated[str, "Entry point of the function to create"],
+) -> dict:
+ """Create a function at ``addr``, working out its end if IDA can't.
+
+ ida_funcs.add_func(ea) asks IDA to find the end itself, and on hand-carved
+ code it often can't — a run that ends in a tail call, or whose last
+ instruction isn't recognised as a return, simply fails with no reason given.
+ You then have a disassembled routine that refuses to become a function, and
+ F5 has nothing to work with.
+
+ So: try IDA's way, and if that fails, use the end of the contiguous
+ instruction run starting at ``addr``."""
+ import ida_bytes
+ import ida_funcs
+ import ida_segment
+ import idaapi
+
+ try:
+ ea = parse_address(addr)
+ except Exception as e:
+ return {"addr": str(addr), "error": str(e), "ok": False}
+ fn = idaapi.get_func(ea)
+ if fn is not None and fn.start_ea == ea:
+ return {"addr": hex(ea), "ok": True, "start": hex(fn.start_ea),
+ "end": hex(fn.end_ea), "how": "existed"}
+ auto = ida_funcs.add_func(ea)
+ if not auto and not _idatui_add_func(ea):
+ return {"addr": hex(ea), "ok": False,
+ "error": f"IDA refused a function at {ea:#x}"}
+ f = idaapi.get_func(ea)
+ if f is None:
+ return {"addr": hex(ea), "ok": False, "error": "function did not stick"}
+ return {"addr": hex(ea), "ok": True, "start": hex(f.start_ea),
+ "end": hex(f.end_ea), "how": "auto" if auto else "explicit-end"}
+
+@tool
+@idasync
+def decomp_error(
+ addr: Annotated[str, "Address of the function that failed to decompile"],
+) -> dict:
+ """Why Hex-Rays refused this function, in its own words.
+
+ The plain decompile tool reports "Decompilation failed at 0x0" and drops the
+ reason, which is the only useful part. Hex-Rays fills in a hexrays_failure_t
+ saying things like "only 64-bit functions can be decompiled in the current
+ database" — that one is unfixable in place (the database's bitness is set at
+ load), so a user who can't see it has no way to know they must reload."""
+ import ida_funcs
+ import ida_hexrays
+ import ida_ida
+
+ try:
+ ea = parse_address(addr)
+ except Exception as e:
+ return {"addr": str(addr), "error": str(e)}
+ out = {"addr": hex(ea), "bitness": ida_ida.inf_get_app_bitness()}
+ fn = ida_funcs.get_func(ea)
+ if fn is None:
+ out["reason"] = "no function here"
+ return out
+ try:
+ if not ida_hexrays.init_hexrays_plugin():
+ out["reason"] = "the decompiler is not available for this processor"
+ return out
+ hf = ida_hexrays.hexrays_failure_t()
+ cf = ida_hexrays.decompile_func(fn, hf)
+ if cf is not None:
+ out["reason"] = "" # it decompiles now
+ return out
+ out["reason"] = hf.desc() or f"error {hf.code}"
+ out["code"] = int(hf.code)
+ out["errea"] = hex(hf.errea)
+ except Exception as e: # noqa: BLE001
+ out["reason"] = f"{type(e).__name__}: {e}"
+ return out
+
+@tool
+@idasync
+def thumb_scan(
+ start: Annotated[str, "Start of the range to scan for entry pointers"] = "",
+ end: Annotated[str, "Exclusive end of the range (default: 1KB from start)"] = "",
+ apply: Annotated[bool, "Mark the targets as Thumb and disassemble them"] = True,
+ limit: Annotated[int, "Max entries to act on"] = 512,
+) -> dict:
+ """Find Thumb entry points from ODD pointers, e.g. a Cortex-M vector table.
+
+ An ARM function pointer carries the mode in bit 0: odd means Thumb. A vector
+ table is therefore a list of Thumb entry points that IDA won't follow on a
+ headerless image, because nothing tells it those words are pointers at all.
+
+ Being wrong here is expensive — marking a data word as code corrupts the
+ listing — so a word only counts when it is odd, lands inside a loaded
+ segment, and its target is EXECUTABLE and not already defined as data. The
+ even words in a vector table (the initial stack pointer) fail the first test,
+ which is the point."""
+ import ida_bytes
+ import ida_funcs
+ import ida_idp
+ import ida_segment
+ import ida_segregs
+ import ida_ua
+
+ seg0 = ida_segment.getseg(parse_address(start)) if start else None
+ if seg0 is None:
+ seg0 = ida_segment.getnseg(0)
+ if seg0 is None:
+ return {"error": "no segments", "found": [], "applied": 0}
+ try:
+ lo = parse_address(start) if start else seg0.start_ea
+ hi = parse_address(end) if end else min(lo + 0x400, seg0.end_ea)
+ except Exception as e:
+ return {"error": str(e), "found": [], "applied": 0}
+
+ treg = ida_idp.str2reg("T")
+ found, applied = [], 0
+ ea = lo
+ while ea + 4 <= hi and len(found) < limit:
+ w = ida_bytes.get_dword(ea)
+ ea += 4
+ if not (w & 1):
+ continue # even: not a Thumb pointer
+ tgt = w & ~1
+ seg = ida_segment.getseg(tgt)
+ if seg is None or not (seg.perm & ida_segment.SEGPERM_EXEC or seg.perm == 0):
+ continue # points outside the image, or at data
+ f = ida_bytes.get_flags(tgt)
+ if ida_bytes.is_data(f):
+ continue # already something else; don't fight it
+ rec = {"at": hex(ea - 4), "value": hex(w), "target": hex(tgt),
+ "was_code": bool(ida_bytes.is_code(f))}
+ found.append(rec)
+ if not apply:
+ continue
+ if treg is not None and treg >= 0:
+ ida_segregs.split_sreg_range(tgt, treg, 1, ida_segregs.SR_user)
+ if not ida_bytes.is_code(ida_bytes.get_flags(tgt)):
+ ida_bytes.del_items(tgt, 0, 2)
+ if ida_ua.create_insn(tgt) <= 0:
+ rec["decoded"] = False
+ continue
+ rec["decoded"] = True
+ rec["function"] = _idatui_add_func(tgt)
+ applied += 1
+ return {"start": hex(lo), "end": hex(hi), "found": found,
+ "applied": applied, "n": len(found)}
'''
SNIPPET = f"{BEGIN}\n{BODY.strip()}\n{END}\n"
diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py
index f930401..b30bba5 100644
--- a/tests/test_blob_ui.py
+++ b/tests/test_blob_ui.py
@@ -214,14 +214,43 @@ async def run() -> int:
lst._cursor_ea() == cur_before,
f"{cur_before:#x} -> {lst._cursor_ea():#x}")
+ # -- `p` after carving: the rest of the app must notice ------ #
+ # The "no functions" hint was latched at load and only cleared on a
+ # reload, so it kept telling you the processor/base were wrong long
+ # after you'd defined a function. The function index was never
+ # rebuilt either, which meant Ctrl+N couldn't find what `p` made.
+ check("no functions yet, and the hint says so",
+ len(app._func_index) == 0
+ and "no functions" in str(app.query_one("#status", Static).render()),
+ f"n={len(app._func_index)}")
+ lst.cursor = lst.model.index_of_ea(target)
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.3)
+ mp = lst.model
+ await pilot.press("p")
+ await wait(lambda: lst.model is not mp and lst.model is not None,
+ pilot, 40)
+ await wait(lambda: app._func_index is not None
+ and len(app._func_index) > 0, pilot, 60)
+ check("`p` creates a function the index can see",
+ len(app._func_index) == 1, f"n={len(app._func_index)}")
+ status = str(app.query_one("#status", Static).render())
+ check("and the stale 'no functions' hint is gone",
+ "no functions" not in status, status[:90])
+ check("the status names the function it made",
+ "created function" in status, status[:90])
+
await pilot.press("ctrl+l")
opened = await wait(lambda: isinstance(app.screen, ConfirmScreen), pilot, 20)
check("Ctrl+L offers to reload with different options", opened,
f"screen={type(app.screen).__name__}")
if opened:
note = str(app.screen.query_one("#confirm-note", Static).render())
- check("and says nothing is lost when there's nothing to lose",
- "nothing is lost" in note, note[:70])
+ # We just made a function, so it must NOT claim nothing is lost —
+ # reloading throws the database away and that is now a real cost.
+ check("the confirmation counts what would be lost",
+ "1 function," in note and "nothing is lost" not in note,
+ note[:80])
await pilot.press("escape")
await pilot.pause(0.3)
check("declining leaves the binary open",
diff --git a/tests/test_formats.py b/tests/test_formats.py
index ab72e8d..1045703 100644
--- a/tests/test_formats.py
+++ b/tests/test_formats.py
@@ -77,9 +77,22 @@ def main() -> int:
load_args("arm", 0, "-T binary") == "-parm -T binary")
check("the processor list leads with the common targets",
- [n for n, _ in PROCESSORS[:3]] == ["arm", "armb", "metapc"],
+ [n for n, _ in PROCESSORS[:2]] == ["arm", "arm:ARMv7-A"],
f"{[n for n, _ in PROCESSORS[:3]]}")
+ # 32-bit ARM has to be offered SEPARATELY from bare 'arm', which gives a
+ # 64-bit database. That isn't cosmetic: Hex-Rays refuses a 32-bit function
+ # in a 64-bit database, and Thumb doesn't exist in AArch64 at all, so a
+ # firmware image loaded as plain 'arm' can never be decompiled — and the
+ # database's bitness cannot be corrected after load.
+ check("a 32-bit ARM variant is offered",
+ any(n.startswith("arm:ARMv") for n, _ in PROCESSORS),
+ f"{[n for n, _ in PROCESSORS if n.startswith('arm')]}")
+ check("and the labels say which is 32- vs 64-bit",
+ all(("32-bit" in d or "64-bit" in d)
+ for n, d in PROCESSORS if n == "arm" or n.startswith("arm:")),
+ f"{[(n, d) for n, d in PROCESSORS if n.startswith('arm')]}")
+
# Every offered name must have been checked against a real IDA, because a
# wrong one is REJECTED (rc=4) with nothing useful said — handing the user a
# dead end from inside the dialog meant to rescue them. This list is the
@@ -89,6 +102,9 @@ def main() -> int:
"arm", "armb", "metapc", "mipsl", "mipsb", "ppc", "ppcl", "sh4", "68k",
"riscv", "tricore", "xtensa", "avr", "z80", "tms320c6", "m32r", "arc",
"h8300", "sparcb", "sparcl", "s390",
+ # variants: tools/verify_procs.py checks these report the base module
+ # AND change the database bitness, which is the reason they exist
+ "arm:ARMv7-A", "arm:ARMv7-M", "arm:ARMv6-M", "arm:ARMv5TE",
}
offered = {n for n, _ in PROCESSORS}
check("every offered processor name is IDA-verified",
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py
index 6ccbf0d..9d77680 100644
--- a/tests/test_scenarios.py
+++ b/tests/test_scenarios.py
@@ -20,12 +20,14 @@ import asyncio
import fnmatch
import os
import re
+import shutil
import sys
+import tempfile
import traceback
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from idatui.app import ( # noqa: E402
- ConfirmScreen, DecompView, DisasmView, FunctionsPanel, HexView, IdaTui,
+ ConfirmScreen, DecompView, FunctionsPanel, HexView, IdaTui,
HelpScreen, ListingView, QuitScreen, StringsPalette, StructEditor,
SymbolPalette, XrefsScreen, _str_display, _word_occurrences,
)
@@ -395,6 +397,84 @@ async def s_load_options(c: Ctx):
await c.wait(lambda: not isinstance(app.screen, LoadOptionsScreen), 10)
+@scenario("asm_highlight")
+async def s_asm_highlight(c: Ctx):
+ """Assembly is highlighted from IDA's OWN token tags.
+
+ generate_disasm_line() emits \x01<tag>text\x02<tag>, and the tag says what
+ the text is — for every processor IDA supports. Nothing is lexed here, so
+ what this pins is that the tags survive the trip: parsed server-side, agreed
+ with the plain text, carried on the Head, and mapped to a style.
+ """
+ from idatui.app import _S_SPAN
+ app = c.app
+ await c.open_biggest("listing")
+ lst = c.lst
+ ok = await c.wait(lambda: lst.model is not None and lst.total > 20, 25)
+ if not ok:
+ c.check("a listing to highlight", False, f"total={lst.total}")
+ return
+ lst.model.ensure(400)
+ rows = [lst.model.get(i) for i in range(min(lst.model.loaded(), 400))]
+ rows = [h for h in rows if h is not None]
+ code = [h for h in rows if h.kind == "code"]
+ c.check("code rows carry IDA's token spans",
+ code and sum(1 for h in code if h.spans) > len(code) * 0.9,
+ f"{sum(1 for h in code if h.spans)}/{len(code)} have spans")
+
+ kinds = {k for h in rows for k, _ in (h.spans or ())}
+ # If a tag isn't mapped it renders as body text and nothing says why, so the
+ # ones that carry real meaning are worth asserting explicitly.
+ for want in ("insn", "reg", "punct"):
+ c.check(f"the palette sees {want} tokens", want in kinds, f"{sorted(kinds)}")
+ c.check("every span kind has a style",
+ all(k in _S_SPAN for k in kinds), f"unstyled: {sorted(kinds - set(_S_SPAN))}")
+
+ # Spans must describe the SAME text the row shows, or the row renders
+ # different characters than search/width calculations think it has.
+ bad = [h for h in rows if h.spans
+ and "".join(t for _k, t in h.spans) != h.text]
+ c.check("spans reconstruct the row text exactly", not bad,
+ f"{[(hex(h.ea), h.text) for h in bad[:2]]}")
+
+ # And the mnemonic must be the loudest thing on the line (the column you
+ # scan), not just any styled token.
+ mn = next((h for h in code if h.spans and h.spans[0][0] == "insn"), None)
+ c.check("the mnemonic is the first span",
+ mn is not None, f"{code[0].spans if code else None}")
+
+
+@scenario("status_names_the_file")
+async def s_status_names_the_file(c: Ctx):
+ """The status bar always says which file you're looking at.
+
+ Obvious once several panes and a project switcher exist: with two idatui
+ windows open, or after switching binaries, "0x2490" alone doesn't say what
+ it belongs to.
+ """
+ from textual.widgets import Static
+ app = c.app
+ await c.open_biggest("listing")
+ await c.pause(0.3)
+ name = os.path.basename(app._open_path)
+ status = str(app.query_one("#status", Static).render())
+ c.check("the status bar names the open file",
+ status.startswith(f"[{name}]"), f"{status[:60]!r} (want [{name}])")
+
+ # It must survive the messages that WRITE the status, not just the idle one.
+ c.lst.focus()
+ await c.press("down")
+ await c.pause(0.3)
+ status = str(app.query_one("#status", Static).render())
+ c.check("and keeps naming it as you move",
+ status.startswith(f"[{name}]"), status[:60])
+
+ # The function-count message used to include the module name itself, which
+ # would now read "[echo] echo — 128 functions".
+ c.check("without saying the name twice",
+ status.count(name) == 1, status[:70])
+
+
@scenario("command_palette")
async def s_command_palette(c: Ctx):
app = c.app
@@ -1860,7 +1940,7 @@ async def s_region_define(c: Ctx):
c.check("listing spans the whole segment (more heads than one function)",
seg is not None and c.lst.total > 1, f"total={c.lst.total} seg={seg}")
- # 'p' on the entry head (re)creates the function and UPGRADES to DisasmView
+ # 'p' on the entry head (re)creates the function
c.lst.focus()
c.lst.cursor, c.lst.cursor_x = c.lst.model.index_of_ea(addr), 0
await c.pause(0.05)
@@ -1949,9 +2029,16 @@ async def s_listing_view(c: Ctx):
try:
c.prog.undefine(dea, size=4)
c.prog.bump_items()
- # reopen the listing so the model reflects the new undefined run
+ # Reopen the listing so the model reflects the new undefined run, and
+ # wait for the model to be REPLACED. Waiting on index_of_ea(dea) >= 0 is
+ # no test at all: dea is in the OLD model too (it's the head we just
+ # undefined), so the predicate passes instantly and we then assert
+ # against pre-edit rows. It only ever passed because the swap happened to
+ # win the race, and it started failing the moment page loads got bigger.
+ stale = c.lst.model
await c.goto_ui(hex(dea))
await c.wait(lambda: app._active == "listing" and c.lst.total > 0
+ and c.lst.model is not stale
and c.lst.model.index_of_ea(dea) >= 0, 25)
ui = c.lst.model.index_of_ea(dea)
c.check("undefining a data head yields an unknown run in the listing",
@@ -2216,7 +2303,49 @@ async def s_func_banners(c: Ctx):
# --------------------------------------------------------------------------- #
# Runner
# --------------------------------------------------------------------------- #
+async def _build_pristine(binary, cache):
+ """Analyse ``binary`` once and keep the resulting database as a golden copy.
+
+ Costs one full analysis, then every later run starts from it instead of
+ re-analysing.
+ """
+ print(f" (building pristine database for {os.path.basename(binary)}\u2026)")
+ app = IdaTui(open_path=binary, keepalive=False)
+ async with app.run_test(size=(140, 44)) as pilot:
+ for _ in range(6000):
+ await pilot.pause(0.05)
+ if app._func_index is not None and app._func_index.complete:
+ break
+ app.program.client.call("idb_save", timeout=600.0)
+ db = binary + ".i64"
+ if os.path.exists(db):
+ shutil.copy2(db, cache)
+
+
async def run(binary, only=None):
+ # The suite EDITS the database — it defines code, undefines items, renames
+ # and comments — and IDA saves those edits. Run that against the tracked
+ # target and every run inherits the last one's damage: decomp_follow_self
+ # started failing with no code change because an earlier scenario had
+ # undefined an instruction, and a position check looked flaky one run in
+ # three for the same reason. A suite whose result depends on its own history
+ # can't be trusted to accuse the code.
+ #
+ # So: work on a scratch copy, seeded from a golden database that nothing
+ # ever writes back to.
+ with tempfile.TemporaryDirectory(prefix="idatui-pilot-") as scratch:
+ target = os.path.join(scratch, os.path.basename(binary))
+ shutil.copy2(binary, target)
+ cache = binary + ".pristine.i64"
+ if not (os.path.exists(cache)
+ and os.path.getmtime(cache) >= os.path.getmtime(binary)):
+ await _build_pristine(target, cache)
+ if os.path.exists(cache):
+ shutil.copy2(cache, target + ".i64")
+ await _run_on(target, only)
+
+
+async def _run_on(binary, only=None):
# Own idalib worker: opens the binary in-process over a unix socket.
app = IdaTui(open_path=binary, keepalive=False)
async with app.run_test(size=(140, 44)) as pilot:
diff --git a/tests/test_thumb_ui.py b/tests/test_thumb_ui.py
new file mode 100644
index 0000000..b0ca00b
--- /dev/null
+++ b/tests/test_thumb_ui.py
@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+"""ARM/Thumb decoding switch, against real Thumb code.
+
+Thumb isn't a property of the bytes — it's a mode the CPU is in — so a raw image
+gives IDA no way to know. At a Thumb entry point it decodes 16-bit instructions
+as 32-bit ARM and produces confident nonsense, and `c` either fails or carves
+garbage. `t` switches the mode.
+
+Uses experiments/fibonacci.bin (real Thumb), so the encodings are not a guess.
+Needs IDA. ~40s.
+"""
+import asyncio
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from textual.widgets import Static # noqa: E402
+
+from idatui.app import DecompView, IdaTui, ListingView # noqa: E402
+
+PASS = FAIL = 0
+BIN = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ "experiments", "fibonacci.bin")
+
+
+def check(name, ok, detail=""):
+ global PASS, FAIL
+ if ok:
+ PASS += 1
+ print(f" ok {name}")
+ else:
+ FAIL += 1
+ print(f" FAIL {name} {detail}")
+
+
+async def wait(pred, pilot, t=240.0):
+ for _ in range(int(t / 0.05)):
+ await pilot.pause(0.05)
+ try:
+ if pred():
+ return True
+ except Exception: # noqa: BLE001
+ pass
+ return False
+
+
+async def run() -> int:
+ # A fresh database every time: the T flag and the segment's addressing mode
+ # are SAVED in the .i64, so a previous run would answer the question for us.
+ for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"):
+ try:
+ os.remove(BIN + ext)
+ except OSError:
+ pass
+
+ app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm")
+ async with app.run_test(size=(140, 44)) as pilot:
+ await wait(lambda: app._func_index is not None
+ and app._func_index.complete, pilot)
+ await wait(lambda: app._cur is not None, pilot, 60)
+ lst = app.query_one(ListingView)
+ lst.focus()
+ lst.cursor = lst.model.index_of_ea(0)
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.3)
+ check("starts undefined at the entry", lst.model.get(lst.cursor).kind == "unknown",
+ f"{lst.model.get(lst.cursor).text!r}")
+
+ # `c` in the wrong mode: this is the failure being fixed. It must NOT
+ # quietly carve garbage — either it refuses, or whatever it makes is not
+ # the Thumb prologue.
+ m0 = lst.model
+ await pilot.press("c")
+ await pilot.pause(2.5)
+ h = lst.model.get(lst.model.index_of_ea(0))
+ check("`c` alone does not produce the Thumb prologue",
+ h is None or h.kind != "code" or "PUSH" not in h.text.upper(),
+ f"{h.text if h else None!r}")
+
+ m1 = lst.model
+ await pilot.press("t")
+ await wait(lambda: lst.model is not m1 and lst.model is not None, pilot, 60)
+ await pilot.pause(0.5)
+
+ status = str(app.query_one("#status", Static).render())
+ check("the status says it switched to Thumb", "Thumb" in status, status[:90])
+ # Thumb doesn't exist in AArch64, and -parm on a headerless blob gives a
+ # 64-bit segment, so setting T alone would change nothing and look broken.
+ check("and says it forced the segment to 32-bit",
+ "32-bit" in status, status[:90])
+
+ m = lst.model
+ rows = [m.get(m.index_of_ea(ea)) for ea in (0x0, 0x2, 0x4)]
+ check("the entry decodes as Thumb",
+ rows[0] is not None and rows[0].kind == "code"
+ and "PUSH" in rows[0].text.upper(),
+ f"{rows[0].text if rows[0] else None!r}")
+ # 16-bit instructions: the addresses are 2 apart, which is the whole
+ # point — in ARM mode these would be one 4-byte instruction.
+ check("instructions are 16-bit wide",
+ all(r is not None and r.kind == "code" and r.size == 2 for r in rows),
+ f"{[(hex(r.ea), r.size, r.text) for r in rows if r]}")
+ check("and it kept disassembling past the first one",
+ sum(1 for i in range(20) if (m.get(i) or h).kind == "code") > 5,
+ "expected a run of instructions, not one")
+
+ # Toggling back must be possible — the mode is a guess and guesses get
+ # revised.
+ m2 = lst.model
+ lst.cursor = lst.model.index_of_ea(0)
+ await pilot.press("t")
+ await wait(lambda: lst.model is not m2 and lst.model is not None, pilot, 60)
+ await pilot.pause(0.5)
+ status = str(app.query_one("#status", Static).render())
+ check("`t` toggles back to ARM", "ARM @" in status, status[:80])
+
+ # -- and the reason a carved function wouldn't decompile ---------------- #
+ # Bare 'arm' gives a 64-BIT database. Thumb doesn't exist there, and
+ # Hex-Rays refuses a 32-bit function outright ("only 64-bit functions can be
+ # decompiled in the current database") — so `t` produces correct-looking
+ # disassembly that F5 can never turn into pseudocode. The database's bitness
+ # is fixed at load and cannot be corrected afterwards, so the only honest
+ # thing is to say so.
+ for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"):
+ try:
+ os.remove(BIN + ext)
+ except OSError:
+ pass
+ app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm") # 64-bit
+ async with app.run_test(size=(140, 44)) as pilot:
+ await wait(lambda: app._func_index is not None
+ and app._func_index.complete, pilot)
+ await wait(lambda: app._cur is not None, pilot, 60)
+ lst = app.query_one(ListingView)
+ lst.focus()
+ lst.cursor = lst.model.index_of_ea(0)
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.3)
+ m = lst.model
+ await pilot.press("t")
+ await wait(lambda: lst.model is not m and lst.model is not None, pilot, 60)
+ await pilot.pause(0.5)
+ status = str(app.query_one("#status", Static).render())
+ check("a 64-bit database warns that Hex-Rays won't decompile",
+ "64-bit" in status and "decompile" in status, status[:120])
+ check("and names the fix", "ARMv7-A" in status, status[:120])
+
+ # And if you ignore that and carry on, the failure has to say WHY. The
+ # plain decompile tool reports "Decompilation failed at 0x0" and drops
+ # the reason, which is the only part that tells you what to do — with it
+ # missing, F5 doing nothing is indistinguishable from a bug in the TUI.
+ lst.cursor = lst.model.index_of_ea(0)
+ await pilot.pause(0.2)
+ mp = lst.model
+ await pilot.press("p")
+ await wait(lambda: lst.model is not mp and lst.model is not None, pilot, 60)
+ await wait(lambda: app._func_index is not None
+ and len(app._func_index) > 0, pilot, 60)
+ await pilot.press("tab")
+ await wait(lambda: "cannot decompile" in
+ str(app.query_one("#status", Static).render()), pilot, 90)
+ status = str(app.query_one("#status", Static).render())
+ # The message must say what to DO. Hex-Rays' own sentence ("only 64-bit
+ # functions can be decompiled in the current database") describes the
+ # database, not the fix, and is long enough that a status bar cuts off
+ # the end — which is where an appended hint would have lived.
+ check("a failed decompile names the fix, not just the diagnosis",
+ "Ctrl+L" in status and "ARMv7-A" in status, status[:130])
+ check("and the reason survives the view reloading under it",
+ "cannot decompile" in status, status[:130])
+ check("the message fits a narrow status bar",
+ len(status) < 110, f"{len(status)} chars: {status[:130]}")
+
+ # -- the whole point: a 32-bit database decompiles ---------------------- #
+ for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"):
+ try:
+ os.remove(BIN + ext)
+ except OSError:
+ pass
+ app = IdaTui(open_path=BIN, keepalive=False, load_args="-parm:ARMv7-A")
+ async with app.run_test(size=(140, 44)) as pilot:
+ await wait(lambda: app._func_index is not None
+ and app._func_index.complete, pilot)
+ # A 32-bit ARM database also lets auto-analysis do its job on Thumb code,
+ # which is why this one lands in the symbol picker rather than nowhere.
+ check("a 32-bit ARM database finds functions by itself",
+ len(app._func_index) > 5, f"n={len(app._func_index)}")
+ await pilot.press("escape")
+ await pilot.pause(0.5)
+ f = app._func_index.all_loaded()[0]
+ app._goto_ea(f.addr, push=True)
+ await wait(lambda: app._cur is not None
+ and app.query_one(ListingView).model is not None, pilot, 60)
+ app.query_one(ListingView).focus()
+ await pilot.press("tab")
+ dec = app.query_one(DecompView)
+ got = await wait(lambda: dec.display and dec._texts, pilot, 90)
+ check("Tab decompiles a Thumb function", got and len(dec._texts) > 3,
+ f"lines={len(dec._texts or [])}")
+ check("and it reads like C",
+ any("(" in t and ")" in t for t in (dec._texts or [])[:3]),
+ f"{(dec._texts or [])[:3]}")
+
+ # -- Thumb entry points from a vector table ----------------------------- #
+ # An ARM function pointer carries the mode in bit 0: odd means Thumb. A
+ # Cortex-M vector table is therefore a list of Thumb entry points, and IDA
+ # won't follow them on a headerless image because nothing says those words
+ # are pointers at all.
+ vec = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ "experiments", "cortexm.bin")
+ if not os.path.isfile(vec):
+ check("the cortexm fixture exists", False, vec)
+ else:
+ for ext in (".i64", ".id0", ".id1", ".id2", ".nam", ".til"):
+ try:
+ os.remove(vec + ext)
+ except OSError:
+ pass
+ app = IdaTui(open_path=vec, keepalive=False, load_args="-parm:ARMv7-M")
+ async with app.run_test(size=(140, 44)) as pilot:
+ await wait(lambda: app._func_index is not None
+ and app._func_index.complete, pilot)
+ check("a bare vector table gives IDA nothing to go on",
+ len(app._func_index) == 0, f"n={len(app._func_index)}")
+ if type(app.screen).__name__ != "Screen":
+ await pilot.press("escape")
+ await pilot.pause(0.5)
+ app._goto_ea(0, push=True)
+ lst = app.query_one(ListingView)
+ await wait(lambda: lst.model is not None, pilot, 60)
+ lst.focus()
+ lst.cursor = lst.model.index_of_ea(0)
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.3)
+ await pilot.press("T")
+ await wait(lambda: app._func_index is not None
+ and len(app._func_index) >= 3, pilot, 90)
+ names = sorted(f.name for f in app._func_index.all_loaded())
+ check("scanning the table finds the Thumb handlers",
+ names == ["sub_200", "sub_240", "sub_280"], f"{names}")
+ # The table also holds an even word (the initial stack pointer), an
+ # even in-range word and an odd word pointing outside the image. All
+ # three must be ignored — marking a data word as code corrupts the
+ # listing, so the cost of a false positive is high.
+ check("and ignores the words that aren't Thumb pointers",
+ len(app._func_index) == 3, f"n={len(app._func_index)}")
+ status = str(app.query_one("#status", Static).render())
+ check("the result survives the reload AND the reindex",
+ "3 Thumb entries" in status, status[:90])
+
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ if not os.path.isfile(BIN):
+ print(f"no such binary: {BIN}")
+ raise SystemExit(2)
+ raise SystemExit(asyncio.run(run()))
diff --git a/tests/test_trace.py b/tests/test_trace.py
new file mode 100644
index 0000000..58c9d81
--- /dev/null
+++ b/tests/test_trace.py
@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+"""Trace model: parsing, queries, and lining a trace up with the database.
+
+Pure stdlib — no IDA, no trace files needed. The differential test
+(test_trace_vs_tenet.py) checks our answers against Tenet's on real traces;
+this one covers what the reference can't arbitrate: the set-shaped queries
+painting needs, and rebasing.
+"""
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from idatui.trace import Trace # noqa: E402
+
+PASS = FAIL = 0
+
+GPR = ["rax", "rbx", "rcx", "rdx", "rbp", "rsp", "rsi", "rdi"]
+FULL = ",".join(f"{r}=0x{0x1000 if r == 'rsp' else 0:x}" for r in GPR)
+
+# push rbp / mov rbp,rsp / mov [rbp-4],0x2a / mov eax,[rbp-4] / pop rbp / jmp back
+LINES = [
+ FULL + ",rip=0x401000",
+ "rsp=0xff8,rip=0x401001,mw=0xff8:0000000000000000",
+ "rbp=0xff8,rip=0x401004",
+ "rip=0x40100b,mw=0xff4:2a000000",
+ "rax=0x2a,rip=0x40100e,mr=0xff4:2a000000",
+ "rbp=0x0,rsp=0x1000,rip=0x401010,mr=0xff8:0000000000000000",
+ "rip=0x401000", # loop back: 0x401000 executes twice
+ "rip=0x401001",
+]
+
+
+def check(name, ok, detail=""):
+ global PASS, FAIL
+ if ok:
+ PASS += 1
+ print(f" ok {name}")
+ else:
+ FAIL += 1
+ print(f" FAIL {name} {detail}")
+
+
+def main() -> int:
+ with tempfile.TemporaryDirectory() as tmp:
+ path = os.path.join(tmp, "t.0.log")
+ with open(path, "w") as f:
+ f.write("\n".join(LINES) + "\n")
+ with open(os.path.join(tmp, "t.info"), "w") as f:
+ f.write("arch=x86_64\nmode=user\nstart_code=0x401000\n")
+ t = Trace.load(path)
+
+ check("every line is one timestamp", t.length == len(LINES), f"{t.length}")
+ check("the sidecar .info is picked up",
+ t.info is not None and t.info.arch == "x86_64" and
+ t.info.start_code == 0x401000)
+ check("PC is tracked per timestamp",
+ [t.ip(i) for i in range(6)] ==
+ [0x401000, 0x401001, 0x401004, 0x40100b, 0x40100e, 0x401010])
+
+ # -- register reconstruction --------------------------------------- #
+ check("a register keeps its value until it changes",
+ t.register("rsp", 0) == 0x1000 and t.register("rsp", 1) == 0xff8
+ and t.register("rsp", 4) == 0xff8 and t.register("rsp", 5) == 0x1000)
+ check("the full first line seeds every register",
+ t.register("rbx", 3) == 0)
+ check("an unknown register is None, not 0",
+ t.register("r15", 0) is None)
+ check("changed() is what the INSTRUCTION did, not the state",
+ t.changed(4) == {"rax", "rip"}, f"{t.changed(4)}")
+
+ # "which instruction set this register?" — the question a trace exists
+ # to answer.
+ check("last_write finds the instruction that set a value",
+ t.last_write("rax", 5) == 4 and t.last_write("rbp", 4) == 2,
+ f"{t.last_write('rax', 5)}, {t.last_write('rbp', 4)}")
+ check("next_write looks forward",
+ t.next_write("rbp", 2) == 5 and t.next_write("rbp", 5) is None)
+
+ # -- memory --------------------------------------------------------- #
+ ops = t.memory_ops(3)
+ check("a write is captured with its bytes",
+ len(ops) == 1 and ops[0].write and ops[0].addr == 0xff4
+ and ops[0].data == bytes.fromhex("2a000000"), f"{ops}")
+ check("a read is captured and marked as a read",
+ [o.write for o in t.memory_ops(4)] == [False])
+ check("an instruction with no memory has none", t.memory_ops(2) == [])
+
+ # -- memory state at a timestamp ------------------------------------ #
+ # The trace wrote 2a000000 at 0xff4 (t=3) and read it back (t=4); it
+ # pushed/popped 8 zero bytes at 0xff8 (t=1 write, t=5 read).
+ d, k = t.memory(0xff4, 4, 3)
+ check("memory reflects a write as of that timestamp",
+ d == bytes.fromhex("2a000000") and k == b"\x01" * 4, f"{d.hex()} {k.hex()}")
+ d, k = t.memory(0xff4, 4, 2)
+ check("and does NOT reflect it before the write happened",
+ k == b"\x00" * 4, f"{d.hex()} known={k.hex()}")
+
+ # A trace only knows what it saw. A byte nobody touched is unknown, and
+ # must not be reported as zero — that distinction is the entire reason
+ # to read memory from a trace instead of from the database.
+ d, k = t.memory(0x5000, 4, t.length - 1)
+ check("untouched memory is unknown, not zero",
+ k == b"\x00" * 4 and d == b"\x00" * 4, f"known={k.hex()}")
+ # 0xff2..0xff3 was never touched; 0xff4..0xff7 came from the write at
+ # t=3 and 0xff8..0xff9 from the push at t=1 — a window can be knowable
+ # from several accesses at different times, which is what makes this
+ # worth a mask rather than a flag.
+ d, k = t.memory(0xff2, 8, 4)
+ check("a partially-covered window marks which bytes are known",
+ k == bytes([0, 0, 1, 1, 1, 1, 1, 1]), f"known={k.hex()}")
+
+ # Reads are evidence too: an instruction reading a byte reveals what it
+ # held at that moment.
+ d, k = t.memory(0xff8, 8, 5)
+ check("a read reveals memory contents",
+ k == b"\x01" * 8, f"known={k.hex()}")
+
+ check("memory_writes lists only the writers",
+ t.memory_writes(0xff4, 4) == [3], f"{t.memory_writes(0xff4, 4)}")
+ check("memory_accesses includes the readers",
+ t.memory_accesses(0xff4, 4) == [3, 4], f"{t.memory_accesses(0xff4, 4)}")
+ check("a never-touched range has no accesses",
+ t.memory_accesses(0x5000, 16) == [])
+
+ # -- execution queries: what painting is built on -------------------- #
+ check("executions lists every timestamp for an address",
+ list(t.executions(0x401000)) == [0, 6], f"{list(t.executions(0x401000))}")
+ check("a never-executed address has none", not len(t.executions(0xdead)))
+ check("executions_between windows the result",
+ t.executions_between(0x401000, 1, 7) == [6])
+ check("next/prev execution step between hits",
+ t.next_execution(0x401000, 0) == 6
+ and t.prev_execution(0x401000, 6) == 0
+ and t.next_execution(0x401000, 6) is None)
+
+ # A pseudocode line covers MANY addresses, so the set form is the one
+ # decompiler painting will call — per-address lookups would mean one
+ # dict hit per instruction per repaint.
+ check("hits() counts a whole set of addresses at once",
+ t.hits([0x401000, 0x401001, 0x401004, 0xdead]) ==
+ {0x401000: 2, 0x401001: 2, 0x401004: 1},
+ f"{t.hits([0x401000, 0x401001, 0x401004, 0xdead])}")
+
+ # -- rebasing -------------------------------------------------------- #
+ # A traced process is relocated; nothing lines up until the slide is
+ # found. Page offsets survive relocation, which is what makes it
+ # findable.
+ db = [0x1000 + (a - 0x401000) for a in
+ (0x401000, 0x401001, 0x401004, 0x40100b, 0x40100e, 0x401010)]
+ slide = t.rebase(db)
+ check("the slide between trace and database is found",
+ slide == 0x1000 - 0x401000, f"{slide:#x}")
+ t.apply_slide(slide)
+ check("addresses come back in database terms",
+ t.ip(0) == 0x1000 and list(t.executions(0x1000)) == [0, 6],
+ f"{t.ip(0):#x}")
+ # Memory addresses are NOT slid: the slide relocates the image, and
+ # these are overwhelmingly stack/heap addresses with no database
+ # counterpart — sliding a stack pointer by the image delta produced a
+ # negative address in testing.
+ check("memory op addresses stay in trace space",
+ t.memory_ops(3)[0].addr == 0xff4, f"{t.memory_ops(3)[0].addr:#x}")
+ check("raw_ip still gives the traced address",
+ t.raw_ip(0) == 0x401000)
+
+ t2 = Trace.load(path)
+ # Page offsets that appear nowhere in the trace. (0xdead0000/0xdead0004
+ # would NOT do: they sit at the same offsets as two traced addresses and
+ # so legitimately agree on a slide — a reminder that this matches on
+ # offsets, not on addresses looking plausible.)
+ check("no match means no slide, not a wrong one",
+ t2.rebase([0xdead0555, 0xbeef0777]) == 0,
+ f"{t2.rebase([0xdead0555, 0xbeef0777]):#x}")
+ check("one lone agreeing address is not enough to claim a slide",
+ t2.rebase([0x1000]) == 0, f"{t2.rebase([0x1000]):#x}")
+ check("an empty database is harmless", t2.rebase([]) == 0)
+
+ # -- robustness ------------------------------------------------------ #
+ p2 = os.path.join(tmp, "odd.0.log")
+ with open(p2, "w") as f:
+ f.write("\n".join([
+ FULL + ",rip=0x401000",
+ "", # blank line
+ "rax=0xnothex,rip=0x401001", # unparseable value
+ "rbx=0x1", # no PC at all
+ "rip=0x401002,mw=0x10:zz", # unparseable memory
+ ]) + "\n")
+ t3 = Trace.load(p2)
+ check("a malformed trace loads instead of raising", t3.length == 4,
+ f"{t3.length}")
+ check("a line with no PC inherits the previous one",
+ t3.ip(2) == 0x401001, f"{t3.ip(2):#x}")
+ check("an unparseable memory entry is dropped, not fatal",
+ t3.memory_ops(3) == [])
+
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py
new file mode 100644
index 0000000..b08ca02
--- /dev/null
+++ b/tests/test_trace_ui.py
@@ -0,0 +1,374 @@
+#!/usr/bin/env python3
+"""The trace dock: registers, timeline, and stepping through time.
+
+Needs IDA and a trace. Generates its own trace with the QEMU tracer if the
+tracer is built; skips with a message otherwise, since neither the emulator nor
+the trace is part of this repo.
+"""
+import asyncio
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from textual.widgets import Input, OptionList, Static # noqa: E402
+
+from idatui.app import (DecompView, IdaTui, ListingView, # noqa: E402
+ RegWriteScreen, TraceDock)
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRACER = os.path.expanduser(
+ "~/.pi/agent/skills/tenet-trace/scripts/tenet-trace")
+PASS = FAIL = 0
+
+
+def check(name, ok, detail=""):
+ global PASS, FAIL
+ if ok:
+ PASS += 1
+ print(f" ok {name}")
+ else:
+ FAIL += 1
+ print(f" FAIL {name} {detail}")
+
+
+async def wait(pred, pilot, t=240.0):
+ for _ in range(int(t / 0.05)):
+ await pilot.pause(0.05)
+ try:
+ if pred():
+ return True
+ except Exception: # noqa: BLE001
+ pass
+ return False
+
+
+def make_trace(tmp, binary):
+ out = os.path.join(tmp, "t")
+ try:
+ subprocess.run([TRACER, "-o", out, binary, "hi"],
+ capture_output=True, timeout=180, check=False)
+ except (OSError, subprocess.TimeoutExpired):
+ return None
+ log = out + ".0.log"
+ return log if os.path.exists(log) else None
+
+
+async def run() -> int:
+ binary = os.path.join(REPO, "targets", "echo")
+ with tempfile.TemporaryDirectory() as tmp:
+ # Scratch copy: opening a binary writes a database beside it, and the
+ # suite must not edit anything tracked.
+ target = os.path.join(tmp, "echo")
+ shutil.copy2(binary, target)
+ log = make_trace(tmp, target)
+ if not log:
+ print(f" skip: could not record a trace (tracer at {TRACER})")
+ return 0
+
+ app = IdaTui(open_path=target, keepalive=False, trace_path=log)
+ async with app.run_test(size=(160, 44)) as pilot:
+ ok = await wait(lambda: app._trace is not None, pilot, 300)
+ check("the trace loads alongside the binary", ok)
+ if not ok:
+ return 1
+ t = app._trace
+ check("it has instructions", t.length > 10, f"{t.length}")
+
+ # Rebasing: the tracer runs the binary relocated, so without a slide
+ # nothing in the trace matches anything on screen.
+ check("trace addresses were rebased onto the database",
+ t.slide != 0 and t.ip(0) < 0x1000000,
+ f"slide={t.slide:#x} ip0={t.ip(0):#x}")
+ idx = app._func_index
+ touched = [f.name for f in idx.all_loaded() if t.executions(f.addr)]
+ check("and now line up with real functions",
+ len(touched) > 1 and "main" in touched, f"{touched[:6]}")
+
+ dock = app.query_one(TraceDock)
+ check("the dock is docked and visible", dock.display)
+ head = str(dock.query_one("#trace-head", Static).render())
+ check("it shows where we are in time", "0" in head and "%" in head,
+ head[:60])
+ regs = str(dock.query_one("#trace-regs", Static).render())
+ check("and the register state at that time", "rip" in regs.lower(),
+ regs[:60])
+
+ # -- stepping --------------------------------------------------- #
+ lst = app.query_one(ListingView)
+ lst.focus()
+ await pilot.pause(0.4)
+ await pilot.press("]")
+ await wait(lambda: app._t == 1, pilot, 20)
+ check("] steps forward one instruction", app._t == 1, f"t={app._t}")
+ check("the code view follows the trace",
+ lst._cursor_ea() == t.ip(1),
+ f"{lst._cursor_ea()} vs {t.ip(1)}")
+ await pilot.press("[")
+ await wait(lambda: app._t == 0, pilot, 20)
+ check("[ steps backward", app._t == 0, f"t={app._t}")
+ await pilot.press("[")
+ await pilot.pause(0.4)
+ check("and stops at the start of the trace", app._t == 0)
+
+ # -- step over -------------------------------------------------- #
+ # A call pushes, so the callee runs with SP below where we started;
+ # stepping until SP comes back up lands after the return. Find a
+ # real call in this trace rather than assuming one is at a fixed
+ # place.
+ sp = "rsp" if "rsp" in t.reg_at else "esp"
+ call_at = None
+ for i in range(1, min(t.length - 1, 400)):
+ a, b = t.register(sp, i), t.register(sp, i + 1)
+ if a and b and b < a:
+ ret = next((j for j in range(i + 1, t.length)
+ if (t.register(sp, j) or 0) >= a), None)
+ if ret and ret > i + 3:
+ call_at = (i, ret)
+ break
+ if call_at is None:
+ check("found a call to step over", False, "none in this trace")
+ else:
+ i, ret = call_at
+ app._seek(i)
+ await wait(lambda: app._t == i, pilot, 20)
+ await pilot.press("}")
+ await wait(lambda: app._t != i, pilot, 30)
+ check("} steps OVER a call instead of into it",
+ app._t == ret, f"{i} -> {app._t}, expected {ret}")
+ check("which is further than a plain step", app._t > i + 1)
+
+ # -- memory at time T -------------------------------------------- #
+ # This is where a trace's memory actually lives: measured 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.
+ dock = app.query_one(TraceDock)
+ app._seek(min(60, t.length - 1))
+ await pilot.pause(0.8)
+ stack = str(dock.query_one("#trace-stack", Static).render())
+ check("the dock shows the stack at this timestamp",
+ "stack (" in stack and len(stack.splitlines()) > 4, stack[:60])
+ sp_name = next(r for r in ("rsp", "esp", "sp") if r in t.reg_at)
+ sp = t.register(sp_name, app._t)
+ check("anchored at the stack pointer",
+ f"{sp:012x}" in stack, f"sp={sp:#x} / {stack[:80]}")
+
+ # A trace knows what it observed and nothing else. Unseen bytes are
+ # printed as '?', never as zeros — rendering them as zero would
+ # invent facts about memory nobody looked at.
+ data, known = t.memory_raw(sp, 8, app._t)
+ if not all(known):
+ check("memory the trace never saw is marked unknown",
+ "?" in stack, stack[:80])
+ else:
+ check("known stack words are shown as values",
+ any(c in "0123456789abcdef" for c in stack), stack[:60])
+
+ # Stepping must move the memory view with time.
+ before = stack
+ app._seek(min(80, t.length - 1))
+ await pilot.pause(0.8)
+ check("and it follows as you move through time",
+ str(dock.query_one("#trace-stack", Static).render()) != before)
+
+ # -- trails ------------------------------------------------------ #
+ # Not "every address the trace ever touched": on a loop-heavy
+ # program that's almost everything and says nothing. The last/next
+ # few dozen steps say how you got here and where you're going.
+ app._seek(min(40, t.length - 1))
+ await pilot.pause(0.6)
+ trail = lst.trail
+ kinds = {k for k in trail.values()}
+ check("the listing is painted with an execution trail",
+ {"now", "past", "future"} <= kinds, f"{sorted(kinds)}")
+ check("'now' is the instruction we're standing on",
+ trail.get(t.ip(app._t)) == "now", f"{trail.get(t.ip(app._t))}")
+ check("the step behind is past, the step ahead is future",
+ trail.get(t.ip(app._t - 1)) == "past"
+ and trail.get(t.ip(app._t + 1)) == "future",
+ f"{trail.get(t.ip(app._t - 1))}, {trail.get(t.ip(app._t + 1))}")
+ painted = [y for y in range(min(lst.size.height, 30))
+ if any(seg.style and seg.style.bgcolor
+ for seg in lst.render_line(y))]
+ check("and it actually reaches the screen", painted, "no tinted rows")
+
+ # -- the same trail on PSEUDOCODE -------------------------------- #
+ # The point of doing this in our app rather than using Tenet: a
+ # trace's addresses are instructions, but decomp_map (built for the
+ # split view) says which instructions each pseudocode line covers,
+ # so the trail lands on C.
+ main_ea = app.program.resolve("main")
+ first = t.first_execution(main_ea)
+ if first is None:
+ check("main was executed in the trace", False)
+ else:
+ app._seek(first + 12)
+ await wait(lambda: app._t == first + 12, pilot, 20)
+ lst.focus()
+ await pilot.press("tab")
+ got = await wait(lambda: app.query_one(DecompView).display
+ and app.query_one(DecompView)._texts, pilot, 120)
+ dec = app.query_one(DecompView)
+ check("pseudocode is available for the traced function", got)
+ app._seek(first + 12)
+ await pilot.pause(1.0)
+ check("pseudocode lines are painted with the trail",
+ len(dec.trail) > 2, f"{len(dec.trail)} lines")
+ now = [i for i, k in dec.trail.items() if k == "now"]
+ check("exactly one pseudocode line is 'now'",
+ len(now) == 1, f"{now}")
+ # The 'now' line must be the one covering the current
+ # instruction, not merely some executed line.
+ covered = app._trail_map[now[0]] if now and app._trail_map else []
+ check("and it's the line covering the current instruction",
+ t.ip(app._t) in covered,
+ f"pc={t.ip(app._t):#x} line covers {[hex(a) for a in covered][:4]}")
+
+ # Stepping must not throw you out of the view you're reading.
+ # A step navigates to an address, and navigating to an address
+ # opens the LISTING unless the decompiler is preferred — so
+ # stepping through C used to drop you into disassembly on the
+ # first keypress. Found by watching a demo, not by a test.
+ await pilot.press("]")
+ await pilot.pause(1.2)
+ check("stepping in pseudocode stays in pseudocode",
+ app._active == "decomp", f"active={app._active}")
+ await pilot.press("[")
+ await pilot.pause(1.2)
+ check("and so does stepping backward",
+ app._active == "decomp", f"active={app._active}")
+
+ # -- split view: a step is a GLOBAL move ------------------------ #
+ # Normal navigation moves one pane and gives the companion a band,
+ # never a cursor, so the two can't chase each other. Time isn't
+ # navigation though: both panes show the same instant, so the
+ # listing cursor must sit on the current instruction.
+ app.action_toggle_split()
+ await pilot.pause(2.0)
+ if not app._split:
+ check("split view toggled on", False)
+ else:
+ base = t.first_execution(main_ea) or 0
+ tracked = 0
+ for k in range(2, 8):
+ app._seek(base + k)
+ await pilot.pause(0.5)
+ if lst._cursor_ea() == t.ip(app._t):
+ tracked += 1
+ check("stepping in split moves the listing cursor to the pc",
+ tracked == 6, f"{tracked}/6 steps tracked")
+ check("and the trail follows in both panes",
+ lst.trail.get(t.ip(app._t)) == "now",
+ f"{lst.trail.get(t.ip(app._t))}")
+
+ # The pseudocode cursor follows too — but only for instructions
+ # the decompiler actually attributes to a line. About half
+ # aren't, and the tempting fallback (nearest mapped address at
+ # or before the pc) is unsound because C lines are not monotonic
+ # in address: an early instruction resolved to a line near the
+ # END of the function. Better to wait than to jump somewhere
+ # unrelated.
+ dec = app.query_one(DecompView)
+ mapped = missed = 0
+ for k in range(2, 30):
+ app._seek(base + k)
+ await pilot.pause(0.3)
+ pc = t.ip(app._t)
+ if app._trail_map_ea == dec.loaded_ea and pc in app._trail_line_of:
+ mapped += 1
+ if dec.cursor != app._trail_line_of[pc]:
+ missed += 1
+ check("the pseudocode cursor follows every mapped instruction",
+ mapped > 3 and missed == 0,
+ f"{mapped} mapped, {missed} not followed")
+
+ # -- seeking, as opposed to stepping ---------------------------- #
+ # "When else did this instruction run?" — the question that makes a
+ # trace more than a very long single-step log.
+ hot = max(t.by_ip, key=lambda a: len(t.by_ip[a]))
+ stamps = list(t.by_ip[hot])
+ db = hot + t.slide
+ if len(stamps) < 2 or lst.model is None:
+ check("found an address executed more than once", False,
+ f"{len(stamps)} executions")
+ else:
+ if app._split:
+ app.action_toggle_split()
+ await pilot.pause(1.0)
+ if app._active != "listing":
+ # focus() does NOT make a view active outside split mode;
+ # Tab is what switches which one is showing.
+ await pilot.press("tab")
+ await wait(lambda: app._active == "listing", pilot, 60)
+ app._seek(stamps[0])
+ await pilot.pause(1.2)
+ lst.focus()
+ row = lst.model.index_of_ea(db)
+ lst.cursor = row
+ lst._scroll_cursor_into_view()
+ await pilot.pause(0.4)
+ check("cursor is on the repeated instruction",
+ lst._cursor_ea() == db, f"{lst._cursor_ea():#x} vs {db:#x}")
+ await pilot.press(">")
+ await pilot.pause(1.0)
+ check("> seeks to the next execution of it",
+ app._t == stamps[1], f"t={app._t}, expected {stamps[1]}")
+ status = str(app.query_one("#status", Static).render())
+ check("and says which execution this is",
+ f"2 of {len(stamps)}" in status, status[:80])
+ lst.cursor = row
+ await pilot.pause(0.2)
+ await pilot.press("<")
+ await pilot.pause(1.0)
+ check("< seeks back to the previous one",
+ app._t == stamps[0], f"t={app._t}, expected {stamps[0]}")
+ # An edge must SAY it's an edge rather than silently doing
+ # nothing, which is indistinguishable from a broken key.
+ lst.cursor = row
+ await pilot.pause(0.2)
+ await pilot.press("<")
+ await pilot.pause(1.0)
+ status = str(app.query_one("#status", Static).render())
+ check("and the first execution says so instead of moving",
+ app._t == stamps[0] and "first" in status, status[:80])
+
+ # -- "which instruction set this register?" ---------------------- #
+ app._seek(min(200, t.length - 1))
+ await pilot.pause(1.0)
+ lst.focus()
+ await pilot.press("W")
+ opened = await wait(lambda: isinstance(app.screen, RegWriteScreen),
+ pilot, 20)
+ check("W lists the registers and where each was set", opened,
+ f"screen={type(app.screen).__name__}")
+ if opened:
+ sc = app.screen
+ pick = next((k for k, (n, v, l, x) in enumerate(sc._rows)
+ if l is not None and l != app._t), None)
+ if pick is None:
+ check("a register was set by an earlier instruction", False)
+ await pilot.press("escape")
+ else:
+ name, _v, last, _n = sc._rows[pick]
+ sc.query_one(OptionList).highlighted = pick
+ await pilot.pause(0.3)
+ await pilot.press("enter")
+ await wait(lambda: app._t == last, pilot, 30)
+ check("choosing one seeks to the write that set it",
+ app._t == last, f"t={app._t}, expected {last}")
+ # The real check: that instruction must actually have
+ # written the register we asked about.
+ check("and that instruction really wrote it",
+ name in t.changed(app._t),
+ f"{name} not in {sorted(t.changed(app._t))}")
+
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(asyncio.run(run()))
diff --git a/tests/test_trace_vs_tenet.py b/tests/test_trace_vs_tenet.py
new file mode 100644
index 0000000..aa6002a
--- /dev/null
+++ b/tests/test_trace_vs_tenet.py
@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+"""Differential test: our trace reader vs Tenet's own.
+
+We wrote our own reader rather than vendoring 3700 lines of packing and mask
+compression we don't need. That is only defensible if "our own" doesn't quietly
+mean "different", so every answer we give is checked against the reference
+implementation on real traces.
+
+Needs ~/dev/tenet/tenet-original (reference) and a trace to compare on; both are
+skipped-with-a-message rather than failed if absent, since neither is part of
+this repo.
+
+ python3 tests/test_trace_vs_tenet.py [trace.0.log ...]
+"""
+import os
+import random
+import sys
+import types
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from idatui.trace import Trace # noqa: E402
+
+TENET = os.path.expanduser("~/dev/tenet/tenet-original/plugins")
+PASS = FAIL = 0
+
+
+def check(name, ok, detail=""):
+ global PASS, FAIL
+ if ok:
+ PASS += 1
+ print(f" ok {name}")
+ else:
+ FAIL += 1
+ print(f" FAIL {name} {detail}")
+
+
+def _reference():
+ """Tenet's TraceReader, with its disassembler dependency stubbed out."""
+ sys.path.insert(0, TENET)
+ log = types.ModuleType("tenet.util.log")
+ log.pmsg = lambda *a, **k: None
+ import tenet # noqa: F401
+ import tenet.util # noqa: F401
+ sys.modules["tenet.util.log"] = log
+ from tenet.trace.arch import ArchAMD64
+ from tenet.trace.reader import TraceReader
+
+ class FakeDctx:
+ # The reference only touches the disassembler to guess an ASLR slide.
+ # An EMPTY list crashes its analysis (indexes [0] unguarded), so hand it
+ # one address that matches nothing: no slide is found and both readers
+ # stay in raw trace addresses, which is what we want to compare.
+ def get_instruction_addresses(self):
+ return [0xDEAD0000]
+ return TraceReader, ArchAMD64, FakeDctx
+
+
+def _truth(path, reg, idx):
+ """The register's value at ``idx`` read straight out of the text.
+
+ Slow (re-reads the file) and only used to settle a disagreement, which is
+ exactly when being slow and obviously-correct is what you want.
+ """
+ reg = reg.lower()
+ last = None
+ with open(path) as f:
+ for i, line in enumerate(f):
+ if i > idx:
+ break
+ for part in line.strip().split(","):
+ k, _, v = part.partition("=")
+ if k.strip().lower() == reg:
+ try:
+ last = int(v, 16)
+ except ValueError:
+ pass
+ return last
+
+
+def compare(path, ref_cls, arch, dctx, samples=200):
+ TraceReader, ArchAMD64, FakeDctx = ref_cls, arch, dctx
+ ours = Trace.load(path)
+ theirs = TraceReader(path, ArchAMD64(), FakeDctx())
+ name = os.path.basename(path)
+
+ check(f"{name}: same length",
+ ours.length == theirs.trace.length,
+ f"{ours.length} vs {theirs.trace.length}")
+ n = min(ours.length, theirs.trace.length)
+ if not n:
+ return
+
+ rnd = random.Random(1234)
+ idxs = sorted({0, n - 1, n // 2} | {rnd.randrange(n) for _ in range(samples)})
+
+ bad = [i for i in idxs if ours.raw_ip(i) != theirs.get_ip(i)]
+ check(f"{name}: same PC at every sampled timestamp", not bad,
+ f"first mismatch at {bad[:1]}")
+
+ # Register reconstruction is the part that is easy to get subtly wrong: a
+ # delta belongs to the line that CAUSED it, and an off-by-one here silently
+ # shifts every value by one instruction.
+ #
+ # Where the two disagree, ARBITRATE with the text rather than assume the
+ # reference is right. It isn't always: a register written on the last line
+ # of a 65535-line segment is missing from the next segment's base state, so
+ # the reference returns a stale value until that register is written again
+ # (measured: wrong for all 179 timestamps of one such window). Blanket
+ # "must agree" would have made us copy that bug to stay green.
+ regs = [r.upper() for r in ours.registers if r not in ("rip",)][:8]
+ ours_wrong, ref_wrong = [], []
+ for i in idxs:
+ for r in regs:
+ mine, ref = ours.register(r, i), theirs.get_register(r, i)
+ if mine == ref:
+ continue
+ true = _truth(path, r, i)
+ (ours_wrong if mine != true else ref_wrong).append((i, r, mine, ref, true))
+ check(f"{name}: register state matches the trace text everywhere",
+ not ours_wrong, f"{ours_wrong[:3]}")
+ if ref_wrong:
+ print(f" (reference disagrees at {len(ref_wrong)} sampled points; "
+ f"the text backs us, e.g. idx {ref_wrong[0][0]} {ref_wrong[0][1]})")
+
+ # Execution queries: what painting is built on.
+ hot = sorted(ours.by_ip, key=lambda a: -len(ours.by_ip[a]))[:5]
+ ex_bad = []
+ for ea in hot:
+ mine = list(ours.executions(ea))
+ ref = list(theirs.get_executions(ea))
+ if mine != ref:
+ ex_bad.append((hex(ea), len(mine), len(ref)))
+ check(f"{name}: same execution timestamps for the hottest addresses",
+ not ex_bad, f"{ex_bad[:3]}")
+
+ # Memory STATE at a timestamp — reconstructed from the deltas, which is the
+ # hard part and the whole point of reading memory from a trace.
+ mem_bad, mem_checked = [], 0
+ for i in idxs[::4]:
+ for op in ours.memory_ops(i)[:2]:
+ n = min(len(op.data), 8)
+ mine, known = ours.memory(op.addr, n, i)
+ ref = theirs.get_memory(op.addr, n, i)
+ if ref is None:
+ continue
+ refb = bytes(ref.data)
+ # Compare only the bytes we claim to know; the reference reports its
+ # own coverage separately and a byte neither has seen is not a
+ # disagreement.
+ for j in range(n):
+ if known[j] and refb[j:j + 1] and mine[j] != refb[j]:
+ mem_bad.append((i, hex(op.addr + j), mine[j], refb[j]))
+ mem_checked += 1
+ check(f"{name}: memory state at a timestamp matches the reference",
+ not mem_bad, f"{mem_bad[:3]}")
+
+ # Memory: the bytes an instruction touched, and which way.
+ with_mem = [i for i in idxs if ours.memory_ops(i)][:40]
+ mem_bad = []
+ for i in with_mem:
+ for op in ours.memory_ops(i):
+ ref = theirs.get_memory(op.addr, len(op.data), i + 1) if op.write else None
+ if ref is not None and bytes(ref.data) != op.data:
+ mem_bad.append((i, hex(op.addr), op.data.hex(), bytes(ref.data).hex()))
+ check(f"{name}: written bytes match the reference's memory state",
+ not mem_bad, f"{mem_bad[:2]}")
+ print(f" ({ours.length:,} instructions, {len(idxs)} sampled, "
+ f"{len(with_mem)} with memory)")
+
+
+def main(argv):
+ if not os.path.isdir(TENET):
+ print(f" skip: reference not found at {TENET}")
+ return 0
+ traces = argv or [p for p in ("/tmp/echotrace.0.log", "/tmp/big.0.log")
+ if os.path.exists(p)]
+ if not traces:
+ print(" skip: no traces to compare (pass one, or run tenet-trace first)")
+ return 0
+ ref = _reference()
+ for path in traces:
+ compare(path, *ref)
+ print(f"\n{PASS} passed, {FAIL} failed")
+ return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main(sys.argv[1:]))
diff --git a/tools/verify_procs.py b/tools/verify_procs.py
index 8012931..323bf1f 100644
--- a/tools/verify_procs.py
+++ b/tools/verify_procs.py
@@ -48,9 +48,18 @@ def main() -> int:
if rc == 0:
ida_auto.auto_wait()
got = ida_ida.inf_get_procname()
+ # "arm:ARMv7-A" selects a VARIANT: inf_get_procname() reports the base
+ # module ("ARM"), so compare that and check the bitness separately —
+ # the variant is only worth offering if it actually changes something.
+ base = name.split(":", 1)[0]
+ bits = ""
+ if rc == 0:
+ import ida_ida
+ bits = f" bitness={ida_ida.inf_get_app_bitness()}"
+ ok = rc == 0 and got.lower() == base.lower()
+ print(f" {'ok ' if ok else 'BAD '} {name:<14} rc={rc} -> {got!r}{bits} {desc}")
+ if rc == 0:
idapro.close_database(save=False)
- ok = rc == 0 and got.lower() == name.lower()
- print(f" {'ok ' if ok else 'BAD '} {name:<12} rc={rc} -> {got!r} {desc}")
if not ok:
bad.append((name, rc, got))
generated by cgit v1.3.1-sl0p (git 2.54.0) at 2026-07-29 23:38:33 +0000 — sl0p fork