"""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)