hex view: raw image bytes (VA-addressed), synced to the code cursor

New HexView: a line-virtualized 16-bytes/row dump of the whole loaded image,
addressed by virtual address. Format-agnostic — the range/segments come from IDA
(image_range over sections()), not any file header; gaps read back as zeros.

Backslash toggles it, synced to the address under the disasm/pseudocode cursor
(see the naked bytes of some code/data). In hex: hjkl/paging navigate a byte
cursor (status shows the section), Enter jumps the code view to the byte,
Tab/Esc/backslash return to the preferred code view. A third _active state; block
-cached + prefetched like the disasm model.

domain: HexModel + Program.hex_model/read_bytes/image_range (get_bytes regions).
Fixed a self-deadlock (hex_model held _lock while sections() re-locked).
Pilot: sync, actual bytes, byte-step, return. full suite 98/98.

1 files changed, 119 insertions, 0 deletions

diff --git a/idatui/domain.py b/idatui/domain.py
index 279ef49..602d54d 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -35,6 +35,7 @@ from .client import IDAClient, IDAToolError
 # Clamps derived from measured caps (list ~700, disasm ~500). Margin included.
 LIST_PAGE = 500
 DISASM_BLOCK = 256  # instructions per cached/fetched block (<= disasm cap)
+HEX_BLOCK = 4096    # bytes per cached/fetched hex block
 
 _TRUNC_RE = re.compile(r"\[(\d+) chars total\]\s*$")
 
@@ -396,6 +397,86 @@ class DisasmModel:
 
 
 # --------------------------------------------------------------------------- #
+# Hex model: block-cached byte view over the loaded image (VA-addressed)
+# --------------------------------------------------------------------------- #
+class HexModel:
+    """Windowed, block-cached raw bytes of the loaded image, addressed by virtual
+    address. Format-agnostic: the range/segments come from IDA, not from any
+    file header. Gaps between segments read back as zeros."""
+
+    BLOCK = HEX_BLOCK
+
+    def __init__(self, program: "Program", start: int, end: int):
+        self._prog = program
+        self.start = start
+        self.end = end
+        self.size = max(end - start, 0)
+        self._blocks: dict[int, bytes] = {}
+        self._lock = threading.Lock()
+        self._inflight: set[int] = set()
+
+    def total_rows(self) -> int:
+        return (self.size + 15) // 16
+
+    def _fetch_block(self, b: int) -> bytes:
+        addr = self.start + b * self.BLOCK
+        n = min(self.BLOCK, self.end - addr)
+        data = self._prog.read_bytes(addr, n) if n > 0 else b""
+        with self._lock:
+            self._blocks[b] = data
+            self._inflight.discard(b)
+        return data
+
+    def _prefetch(self, b: int) -> None:
+        if b < 0 or b * self.BLOCK >= self.size:
+            return
+        with self._lock:
+            if b in self._blocks or b in self._inflight:
+                return
+            self._inflight.add(b)
+        self._prog.submit(self._fetch_block, b)
+
+    def row(self, r: int, prefetch: bool = True) -> tuple[int, bytes | None]:
+        """Return (va, bytes<=16) for row ``r``, or (va, None) if not yet cached.
+        Non-blocking; used by the virtualized view's render path."""
+        off = r * 16
+        va = self.start + off
+        b = off // self.BLOCK
+        with self._lock:
+            block = self._blocks.get(b)
+        if block is None:
+            return (va, None)
+        bo = off - b * self.BLOCK
+        return (va, block[bo:bo + 16])
+
+    def ensure(self, r0: int, count: int) -> None:
+        """Blocking: fetch the blocks covering rows [r0, r0+count) if missing."""
+        if count <= 0:
+            return
+        b0 = (r0 * 16) // self.BLOCK
+        b1 = ((r0 + count) * 16) // self.BLOCK
+        for b in range(b0, b1 + 1):
+            with self._lock:
+                have = b in self._blocks
+            if not have:
+                self._fetch_block(b)
+
+    def is_cached(self, r0: int, count: int) -> bool:
+        if count <= 0:
+            return True
+        b0 = (r0 * 16) // self.BLOCK
+        b1 = ((r0 + count - 1) * 16) // self.BLOCK
+        with self._lock:
+            return all(b in self._blocks for b in range(b0, b1 + 1))
+
+    def ensure_async(self, r0: int, count: int) -> None:
+        b0 = (r0 * 16) // self.BLOCK
+        b1 = ((r0 + max(count, 1) - 1) * 16) // self.BLOCK
+        for b in range(b0 - 1, b1 + 2):
+            self._prefetch(b)
+
+
+# --------------------------------------------------------------------------- #
 # Program: top-level handle, model registry, prefetch pool
 # --------------------------------------------------------------------------- #
 class Program:
@@ -411,6 +492,7 @@ class Program:
         self._decomp: dict[int, tuple[Decompilation, int]] = {}
         self._name_gen = 0  # bumped on rename; invalidates stale name caches
         self._sections: list[tuple[int, int, str]] | None = None
+        self._hexmodel: "HexModel | None" = None
         self._lock = threading.Lock()
 
     # -- prefetch plumbing ------------------------------------------------- #
@@ -454,6 +536,43 @@ class Program:
             self._sections = secs
         return secs
 
+    def image_range(self) -> tuple[int, int] | None:
+        """[start, end) spanning all loaded segments (the hex view's document)."""
+        secs = self.sections()
+        if not secs:
+            return None
+        return (min(s[0] for s in secs), max(s[1] for s in secs))
+
+    def hex_model(self) -> "HexModel | None":
+        """Cached block-backed byte view over the whole loaded image."""
+        if self._hexmodel is not None:
+            return self._hexmodel
+        rng = self.image_range()  # calls sections() -> don't hold _lock (it re-locks)
+        with self._lock:
+            if self._hexmodel is None and rng is not None:
+                self._hexmodel = HexModel(self, rng[0], rng[1])
+            return self._hexmodel
+
+    def read_bytes(self, ea: int, n: int) -> bytes:
+        """Raw bytes [ea, ea+n) from IDA (gaps read as zero)."""
+        if n <= 0:
+            return b""
+        try:
+            r = self.client.call("get_bytes", regions=[{"addr": hex(ea), "size": int(n)}])
+        except IDAToolError:
+            return b"\x00" * n
+        res = r.get("result", []) if isinstance(r, dict) else []
+        data = res[0].get("data", "") if res and isinstance(res[0], dict) else ""
+        out = bytearray()
+        for tok in data.split():
+            try:
+                out.append(int(tok, 16) & 0xFF)
+            except ValueError:
+                out.append(0)
+        if len(out) < n:  # pad short reads (unmapped tail)
+            out.extend(b"\x00" * (n - len(out)))
+        return bytes(out[:n])
+
     def section_of(self, ea: int) -> str | None:
         """Name of the segment/section containing ``ea`` (e.g. '.got', '.text',
         '.data.rel.ro', 'LOAD'), or None if unmapped."""