diff options
| author | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-02 08:36:35 -0400 |
|---|---|---|
| committer | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-02 08:36:35 -0400 |
| commit | ba58dcdc5ad6ca021c723d279f33e3f14e1184df (patch) | |
| tree | 47e43a2dad9a2d4bb40da46243654bff8b1302de /debug/risc-v-gsp-rm-vm/greeneye/memory.py | |
| parent | fe485080e159c983242d699f4a584659df159818 (diff) | |
| download | open-iov-ba58dcdc5ad6ca021c723d279f33e3f14e1184df.tar.gz open-iov-ba58dcdc5ad6ca021c723d279f33e3f14e1184df.zip | |
debug: add green-eye GSP/uproc RISC-V VM runtime analysis tooling
Adds open-iov/debug/risc-v-gsp-rm-vm: a GDB-free instrumentation stack that boots uproc/GSP firmware in qemu-system-riscv64 and provides Frida-like hooks, breakpoints, watchpoints, memory dumps, structure decoding, snapshot diffs, recorded sessions, declarative probes, a plugin API and a Ghidra symbol/hit bridge.
Verified: 25 tests green (including a live QEMU smoke test); the demo uproc boots with correct hook/leave/watch/schema/snapshot/fault evidence; the optional TCG trace plugin builds and recorded 15M trace lines in 3 s; the TU104 570.144 fwimage probe reports its first exception (load access fault at firmware vaddr 0x4009014 reading 0x2000000000000A00).
Documentation: lean tool docs under debug/risc-v-gsp-rm-vm/docs/00-10, a new docs/05-operations/03-risc-v-gsp-rm-vm.md debugging document, and '# TOOL IN DEVELOPMENT - NOT READY YET!' markers in docs/README.md, the onboarding tooling table and the debugging workflows doc.
Vendors qemu-plugin.h from QEMU v8.2.2 (GPL-2.0-or-later), with its license header intact, so the optional TCG plugin builds without a QEMU source tree.
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/memory.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/memory.py | 190 |
1 files changed, 190 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/memory.py b/debug/risc-v-gsp-rm-vm/greeneye/memory.py new file mode 100644 index 0000000..23d9ca0 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/memory.py @@ -0,0 +1,190 @@ +"""Memory inspection helpers: read/write, hexdump, search, snapshots, diffs.""" +from __future__ import annotations + +from dataclasses import dataclass + +from .errors import GreenEyeError + + +@dataclass +class Diff: + addr: int + old: bytes + new: bytes + + @property + def size(self) -> int: + return len(self.old) + + +class Memory: + """A convenience facade over an RSP client (and optional symbol DB).""" + + def __init__(self, client, symbols=None, recorder=None): + self.client = client + self.symbols = symbols + self.recorder = recorder + + # -- basic access ------------------------------------------------------ + + def read(self, addr: int, size: int) -> bytes: + data = self.client.read_mem(addr, size) + if self.recorder is not None: + self.recorder.event("mem_read", addr=addr, size=size) + return data + + def write(self, addr: int, data: bytes | int | str) -> None: + if isinstance(data, int): + data = bytes([data & 0xFF]) + elif isinstance(data, str): + data = data.encode("utf-8") + self.client.write_mem(addr, data) + if self.recorder is not None: + self.recorder.event("mem_write", addr=addr, size=len(data), data=data.hex()) + + def read_u8(self, addr: int) -> int: + return self.read(addr, 1)[0] + + def read_u16(self, addr: int) -> int: + return int.from_bytes(self.read(addr, 2), "little") + + def read_u32(self, addr: int) -> int: + return int.from_bytes(self.read(addr, 4), "little") + + def read_u64(self, addr: int) -> int: + return int.from_bytes(self.read(addr, 8), "little") + + def write_u32(self, addr: int, value: int) -> None: + self.write(addr, value.to_bytes(4, "little")) + + def write_u64(self, addr: int, value: int) -> None: + self.write(addr, value.to_bytes(8, "little")) + + def read_cstring(self, addr: int, max_len: int = 256) -> str: + return self.read(addr, max_len).split(b"\x00", 1)[0].decode("utf-8", "replace") + + def read_ptr(self, addr: int, width: int = 8) -> int: + return int.from_bytes(self.read(addr, width), "little") + + # -- display ----------------------------------------------------------- + + def hexdump(self, addr: int, size: int, width: int = 16, annotate: bool = True) -> str: + data = self.read(addr, size) + lines: list[str] = [] + if annotate and self.symbols is not None: + sym = self.symbols.resolve(addr) + if sym is not None: + lines.append(f"; {sym.label(addr)} (source={sym.source or '?'})") + for offset in range(0, len(data), width): + chunk = data[offset : offset + width] + hex_part = " ".join(f"{b:02x}" for b in chunk) + ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) + lines.append(f"{addr + offset:08x} {hex_part:<{width * 3 - 1}} |{ascii_part}|") + return "\n".join(lines) + + def hexdump_annotated( + self, + addr: int, + size: int, + struct_schema=None, + width: int = 16, + ) -> str: + """Hexdump plus a decoded structure overlay when a schema is given.""" + lines = [self.hexdump(addr, size, width=width, annotate=True)] + if struct_schema is not None: + from .structs import decode, render + + values = decode(struct_schema, self.read(addr, struct_schema.size), base=addr) + lines.append("") + lines.append(render(struct_schema, values)) + return "\n".join(lines) + + # -- search ------------------------------------------------------------ + + def search( + self, + addr: int, + size: int, + pattern: bytes | int, + *, + mask: bytes | None = None, + max_hits: int = 64, + chunk: int = 0x10000, + ) -> list[int]: + if isinstance(pattern, int): + pattern = bytes([pattern & 0xFF]) + if not pattern: + raise GreenEyeError("empty search pattern") + if mask is not None and len(mask) != len(pattern): + raise GreenEyeError("mask length must match pattern length") + hits: list[int] = [] + offset = 0 + overlap = len(pattern) - 1 + carry = b"" + while offset < size and len(hits) < max_hits: + want = min(chunk, size - offset) + blob = carry + self.read(addr + offset - len(carry), want + len(carry)) if carry else self.read(addr + offset, want) + base = addr + offset - len(carry) + for index in range(0, len(blob) - len(pattern) + 1): + candidate = blob[index : index + len(pattern)] + if mask is None: + if candidate == pattern: + hits.append(base + index) + else: + if all((candidate[i] & mask[i]) == (pattern[i] & mask[i]) for i in range(len(pattern))): + hits.append(base + index) + if len(hits) >= max_hits: + break + carry = blob[-overlap:] if overlap else b"" + offset += want + return hits + + def search_u32(self, addr: int, size: int, value: int, **kw) -> list[int]: + return self.search(addr, size, value.to_bytes(4, "little"), **kw) + + def search_string(self, addr: int, size: int, text: str, **kw) -> list[int]: + return self.search(addr, size, text.encode("utf-8"), **kw) + + # -- capture / diff ---------------------------------------------------- + + def snapshot(self, addr: int, size: int, name: str = "region") -> bytes: + data = self.client.read_mem(addr, size) + if self.recorder is not None: + path = self.recorder.snapshot(name, addr, data) + self.recorder.event("snapshot", label=name, addr=addr, size=size, path=path) + return data + + @staticmethod + def diff(old: bytes, new: bytes, base: int = 0) -> list[Diff]: + length = min(len(old), len(new)) + changed = [i for i in range(length) if old[i] != new[i]] + if not changed: + return [] + # coalesce adjacent bytes into runs + runs: list[Diff] = [] + start = changed[0] + prev = changed[0] + for index in changed[1:]: + if index == prev + 1: + prev = index + continue + runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1])) + start = prev = index + runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1])) + return runs + + @staticmethod + def render_diff(diffs: list[Diff], symbols=None, max_lines: int = 64) -> str: + lines: list[str] = [] + for diff in diffs[:max_lines]: + old = " ".join(f"{b:02x}" for b in diff.old[:8]) + new = " ".join(f"{b:02x}" for b in diff.new[:8]) + label = "" + if symbols is not None: + sym = symbols.resolve(diff.addr) + if sym is not None: + label = f" ; {sym.label(diff.addr)}" + lines.append(f"{diff.addr:08x} {old:<24} -> {new:<24}{label}") + if len(diffs) > max_lines: + lines.append(f"... {len(diffs) - max_lines} more changed runs") + return "\n".join(lines) |
