From 002e3d7d13c414ea874abbee8d82803dd3a72fe5 Mon Sep 17 00:00:00 2001 From: Arthur Hanson Rasmusson Date: Fri, 2 Oct 2026 10:58:21 -0400 Subject: debug: green-eye wave 3 - RM harness, local simulator, device map, MCP plugin, value-validation demos, CI pipeline, toolchain docs; add host RM tools rmtool/rmtop --- debug/risc-v-gsp-rm-vm/greeneye/__init__.py | 41 +- debug/risc-v-gsp-rm-vm/greeneye/analyze.py | 157 ++++++- debug/risc-v-gsp-rm-vm/greeneye/cli.py | 126 +++++- debug/risc-v-gsp-rm-vm/greeneye/device_map.py | 555 +++++++++++++++++++++++++ debug/risc-v-gsp-rm-vm/greeneye/ghidra.py | 2 + debug/risc-v-gsp-rm-vm/greeneye/session.py | 2 + debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py | 390 +++++++++++++++++ debug/risc-v-gsp-rm-vm/greeneye/symbols.py | 17 +- 8 files changed, 1268 insertions(+), 22 deletions(-) create mode 100644 debug/risc-v-gsp-rm-vm/greeneye/device_map.py create mode 100644 debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py (limited to 'debug/risc-v-gsp-rm-vm/greeneye') diff --git a/debug/risc-v-gsp-rm-vm/greeneye/__init__.py b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py index 8b1c1e5..717160c 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/__init__.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py @@ -85,13 +85,46 @@ def launch( name: str = "session", load_symbols: bool = True, keep_alive: bool = False, + backend: str = "qemu", + trace=None, ) -> Session: - """Launch QEMU (paused), attach, and return a ready Session. + """Launch a target (paused), attach, and return a ready Session. - The target starts paused at its reset vector; use Session.load_image() to - load an ELF (which also sets PC), or QemuTarget.run_kernel() style helper - launch with kernel= for the fast `-kernel` path. + ``backend="qemu"`` (default) starts ``qemu-system-riscv64`` exactly as + before. ``backend="native"`` uses the in-process deterministic RV64IMAC + simulator (M1) via ``greeneye.sim_backend``; the Session contract is the + same. ``trace`` (native only) enables the R4 exec/access trace: ``True``, + a dict of ``{"exec", "start", "end", "max_events"}``, or a + ``sim.trace.Trace``; it is saved to the session as ``blobs/trace.jsonl``. + The target starts paused at its reset vector; use + Session.load_image() to load an ELF (which also sets PC), or launch with + kernel= for the fast `-kernel` path. """ + if backend in ("native", "sim"): + from .sim_backend import launch_native + + return launch_native( + kernel=kernel, + memory_mb=memory_mb, + record=record, + symbols=symbols, + name=name, + load_symbols=load_symbols, + keep_alive=keep_alive, + trace=trace, + ) + if trace is not None: + from .errors import TargetError + + raise TargetError( + "trace= is native-only; for QEMU build qemu/greeneye-trace.so and " + "pass -plugin (see qemu/README.md)" + ) + if backend != "qemu": + from .errors import TargetError + + raise TargetError(f"unknown backend {backend!r} (use 'qemu' or 'native')") + target = QemuTarget( machine=machine, memory_mb=memory_mb, diff --git a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py index 66429e9..1c7e930 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py @@ -6,12 +6,14 @@ offline comparison between runs (bare-metal vs VFIO, old firmware vs new). """ from __future__ import annotations +import json import os import re from collections import Counter, defaultdict from .record import load_events, load_meta +_TRACE_KINDS = ("exec", "access", "read", "write") _ACCESS_RE = re.compile( r"Invalid (read|write) at addr (0x[0-9a-fA-F]+), size (\d+), region '([^']*)', reason: (\w+)" ) @@ -90,13 +92,56 @@ def describe_fault(fault: dict, bias: int = 0) -> str: return "\n".join(lines) +def _trace_kind(record: dict) -> str | None: + """Native/TCG trace records use ``t``; folded session events use ``type``.""" + return record.get("type") or record.get("t") + + +def _number(value, default: int = 0) -> int: + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value, 0) + except ValueError: + return default + return default + + +def load_trace(session_dir: str, meta: dict | None = None) -> list[dict]: + """Load the trace blob referenced by ``meta['trace_blob']`` (R4).""" + meta = meta if meta is not None else load_meta(session_dir) + rel = meta.get("trace_blob") + if not rel: + return [] + path = rel if os.path.isabs(rel) else os.path.join(session_dir, rel) + records: list[dict] = [] + try: + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + record = json.loads(line) + if _trace_kind(record) != "summary": + records.append(record) + except (OSError, ValueError): + return [] + return records + + def render_summary(session_dir: str, symbols=None, max_rows: int = 40) -> str: meta = load_meta(session_dir) events = load_events(session_dir) - return render_summary_from(meta, events, symbols=symbols, max_rows=max_rows) + return render_summary_from( + meta, events, symbols=symbols, max_rows=max_rows, trace=load_trace(session_dir, meta) + ) -def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: int = 40) -> str: +def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: int = 40, + trace: list[dict] | None = None) -> str: + if trace is None: + trace = [e for e in events if _trace_kind(e) in _TRACE_KINDS] lines: list[str] = [] lines.append(f"# green-eye session summary: {meta.get('name', '?')}") lines.append("") @@ -194,6 +239,79 @@ def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: lines.append(f"| `{page:#x}` | {count} | `{sym.name if sym else ''}` |") lines.append("") + # Execution/access trace coverage (native R4 or folded TCG trace) + if trace: + def access_kind(record): + kind = _trace_kind(record) + if kind == "access": + return "write" if str(record.get("rw", "")).startswith("w") else "read" + return kind + + execs = [r for r in trace if access_kind(r) == "exec"] + reads = [r for r in trace if access_kind(r) == "read"] + writes = [r for r in trace if access_kind(r) == "write"] + lines.append("## Execution trace") + lines.append("") + blob = meta.get("trace_blob") + if blob: + lines.append(f"* trace blob: `{blob}`") + pcs = Counter() + for record in execs: + pcs[_number(record.get("pc"))] += 1 + suffix = f", unique PCs: {len(pcs)}" if execs else " (exec=off)" + lines.append(f"* exec events: {len(execs)}{suffix}") + read_pages = Counter() + for record in reads: + read_pages[(_number(record.get("addr")) >> 12) << 12] += 1 + write_pages = Counter() + for record in writes: + write_pages[(_number(record.get("addr")) >> 12) << 12] += 1 + lines.append( + f"* accesses: {len(reads)} reads ({len(read_pages)} pages), " + f"{len(writes)} writes ({len(write_pages)} pages)" + ) + config = meta.get("trace") or {} + if config: + bits = [f"exec={'on' if config.get('exec') else 'off'}"] + for key in ("start", "end"): + if config.get(key) is not None: + bits.append(f"{key}={config[key]:#x}" if isinstance(config[key], int) else f"{key}={config[key]}") + if config.get("max_events"): + bits.append(f"max_events={config['max_events']}") + if config.get("dropped"): + bits.append(f"dropped={config['dropped']}") + lines.append("* trace config: " + ", ".join(bits)) + lines.append("") + + if execs: + by_sym = defaultdict(lambda: [0, 0]) # unique PCs, exec count + for pc, count in pcs.items(): + sym = symbols.resolve(pc) if symbols else None + entry = by_sym[sym.name if sym else f"{pc:#x}"] + entry[0] += 1 + entry[1] += count + lines.append("### Unique PCs by symbol") + lines.append("") + lines.append("| unique PCs | execs | symbol |") + lines.append("|---:|---:|---|") + for name, (unique, count) in sorted(by_sym.items(), key=lambda kv: -kv[1][1])[:max_rows]: + lines.append(f"| {unique} | {count} | `{name}` |") + lines.append("") + + if read_pages or write_pages: + lines.append("### Accessed pages") + lines.append("") + lines.append("| page | reads | writes | symbol |") + lines.append("|---|---:|---:|---|") + for page in sorted(set(read_pages) | set(write_pages), + key=lambda p: -(read_pages.get(p, 0) + write_pages.get(p, 0)))[:max_rows]: + sym = symbols.resolve(page) if symbols else None + lines.append( + f"| `{page:#x}` | {read_pages.get(page, 0)} | {write_pages.get(page, 0)} " + f"| `{sym.name if sym else ''}` |" + ) + lines.append("") + # Coverage steps = [e for e in events if e.get("type") == "step"] if steps: @@ -277,9 +395,16 @@ def compare_sessions(a_dir: str, b_dir: str, symbols=None, max_rows: int = 40) - """Diff two recorded sessions at the symbolic event level (A/B tooling).""" a_meta, b_meta = load_meta(a_dir), load_meta(b_dir) a_events, b_events = load_events(a_dir), load_events(b_dir) + a_trace, b_trace = load_trace(a_dir, a_meta), load_trace(b_dir, b_meta) lines = [f"# green-eye session comparison", "", f"* A: {a_meta.get('name')} ({a_dir})", f"* B: {b_meta.get('name')} ({b_dir})", ""] - def profile(events): + def trace_kind(record: dict) -> str: + kind = _trace_kind(record) + if kind == "access": + return "write" if str(record.get("rw", "")).startswith("w") else "read" + return kind + + def profile(events, trace): bp = Counter() watch = Counter() writes = Counter() @@ -290,10 +415,13 @@ def compare_sessions(a_dir: str, b_dir: str, symbols=None, max_rows: int = 40) - watch[(event.get("addr"), event.get("kind"))] += 1 elif event.get("type") == "mem_write": writes[(event.get("addr", 0) >> 12) << 12] += 1 + for record in trace: + if trace_kind(record) == "write": + writes[(_number(record.get("addr")) >> 12) << 12] += 1 return bp, watch, writes - a_bp, a_watch, a_writes = profile(a_events) - b_bp, b_watch, b_writes = profile(b_events) + a_bp, a_watch, a_writes = profile(a_events, a_trace) + b_bp, b_watch, b_writes = profile(b_events, b_trace) lines.append("## Breakpoint profile delta (A vs B)") lines.append("") @@ -318,4 +446,23 @@ def compare_sessions(a_dir: str, b_dir: str, symbols=None, max_rows: int = 40) - for page in sorted(set(a_writes) | set(b_writes), key=lambda k: -(a_writes.get(k, 0) + b_writes.get(k, 0)))[:max_rows]: lines.append(f"| `{page:#x}` | {a_writes.get(page, 0)} | {b_writes.get(page, 0)} |") lines.append("") + + def trace_totals(trace): + totals = Counter() + for record in trace: + kind = trace_kind(record) + if kind in ("exec", "read", "write"): + totals[kind] += 1 + return totals + + a_totals, b_totals = trace_totals(a_trace), trace_totals(b_trace) + if a_trace or b_trace: + lines.append("## Trace stream totals (A vs B)") + lines.append("") + lines.append("| stream | A | B | delta |") + lines.append("|---|---:|---:|---:|") + for kind in ("exec", "read", "write"): + a_count, b_count = a_totals.get(kind, 0), b_totals.get(kind, 0) + lines.append(f"| {kind} | {a_count} | {b_count} | {b_count - a_count:+d} |") + lines.append("") return "\n".join(lines) + "\n" diff --git a/debug/risc-v-gsp-rm-vm/greeneye/cli.py b/debug/risc-v-gsp-rm-vm/greeneye/cli.py index 94102a4..e7023ec 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/cli.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/cli.py @@ -1,13 +1,17 @@ """green-eye command line interface. Commands: - info environment/tooling check + info environment/tooling check (incl. native simulator availability) demo boot the bundled uproc demo and run a canned instrumentation pass run run a Python instrumentation script against a target - attach interactive REPL against a QEMU gdbstub + attach interactive REPL over a gdbstub or the native simulator + launch start a target and keep it alive for a later attach probe run an arbitrary firmware (default: the real GSP fwimage) for N instructions and report the first divergence record boot a firmware and record a memory-activity session + (native: --trace collects an exec/access trace blob) + inject load and invoke an injected RV64 service (Frida-like) + ghidra serve a live session to GhidraScripts over the JSON-lines bridge analyze render a recorded session to a symbol-annotated report compare A/B two recorded sessions at the symbolic level symbols build/inspect a green-eye symbol database from Ghidra exports @@ -82,14 +86,20 @@ def _start_session( record, name: str, plugins: list[str] | None = None, + backend: str = "qemu", + trace=None, ): if port: + if backend != "qemu": + raise GreenEyeError("--backend native cannot attach to an existing gdbstub port") + if trace is not None: + raise GreenEyeError("--trace is native-only; use the qemu/greeneye-trace.so plugin for QEMU") db = SymbolDB() if kernel: db.load_elf(kernel) session = ge.attach(port=port, record=record, symbols=db, name=name) else: - session = ge.launch(kernel=kernel, record=record, name=name) + session = ge.launch(kernel=kernel, record=record, name=name, backend=backend, trace=trace) manager = PluginManager(session) manager.load_all(plugins) return session, manager @@ -101,6 +111,63 @@ def _print_session_dir(session) -> None: print(f"[green-eye] session: {directory}") +def _add_backend_arg(parser) -> None: + """Select the execution backend; qemu stays the default reference path.""" + parser.add_argument( + "--backend", + choices=("qemu", "native"), + default="qemu", + help="execution backend: qemu (default) or the native RV64IMAC simulator", + ) + + +def _add_trace_args(parser) -> None: + """Native exec/access tracing (R4), mirroring qemu/greeneye_trace.c.""" + parser.add_argument( + "--trace", + action="store_true", + help="native backend: record an exec/access trace into the session", + ) + parser.add_argument( + "--trace-exec", + choices=("on", "off"), + default="on", + help="trace instruction exec events (default on; 'off' records accesses only)", + ) + parser.add_argument( + "--trace-start", + type=lambda v: int(v, 0), + default=None, + help="only trace PCs >= this address (inclusive)", + ) + parser.add_argument( + "--trace-end", + type=lambda v: int(v, 0), + default=None, + help="only trace PCs <= this address (inclusive)", + ) + parser.add_argument( + "--trace-max", + type=int, + default=None, + help="cap buffered trace events (0 = unlimited)", + ) + + +def _trace_spec(args): + """Build a launch_native trace spec from parsed args (None when disabled).""" + if not getattr(args, "trace", False): + return None + spec = {"exec": args.trace_exec == "on"} + if args.trace_start is not None: + spec["start"] = args.trace_start + if args.trace_end is not None: + spec["end"] = args.trace_end + if args.trace_max is not None: + spec["max_events"] = args.trace_max + return spec + + # --------------------------------------------------------------------------- # commands # --------------------------------------------------------------------------- @@ -121,6 +188,13 @@ def cmd_info(args) -> int: version = subprocess.run([resolved, "--version"], capture_output=True, text=True).stdout.strip() print(f"qemu version: {version}") + try: + from .sim_backend import SIM_IPS # noqa: F401 + + print("native simulator: available (deterministic RV64IMAC, pure-Python reference)") + print("native trace: available (--trace exec/access, --trace-start/--trace-end filters)") + except Exception as exc: # pragma: no cover - only on a broken install + print(f"native simulator: unavailable ({exc})") for label, path in [ ("demo firmware", DEMO_ELF), ("demo schema", DEMO_SCHEMA), @@ -142,7 +216,10 @@ def cmd_demo(args) -> int: raise GreenEyeError(f"demo firmware not built: {elf}") schema = _load_schema(DEMO_SCHEMA) record = True if args.record is None else args.record - session, _manager = _start_session(kernel=str(elf), port=None, record=record, name="demo") + session, _manager = _start_session( + kernel=str(elf), port=None, record=record, name="demo", backend=args.backend, + trace=_trace_spec(args), + ) hits: list[str] = [] try: ctrl = session.symbols.lookup("gsp_ctrl") @@ -204,7 +281,13 @@ def cmd_run(args) -> int: if not kernel and not args.port: kernel = str(_build_demo()) session, manager = _start_session( - kernel=kernel, port=args.port, record=args.record or True, name=script.stem, plugins=args.plugin + kernel=kernel, + port=args.port, + record=args.record or True, + name=script.stem, + plugins=args.plugin, + backend=args.backend, + trace=_trace_spec(args), ) namespace = { "ge": ge, @@ -214,6 +297,8 @@ def cmd_run(args) -> int: "symbols": session.symbols, "recorder": session.recorder, "greeneye": ge, + "__file__": str(script), + "__name__": "__main__", } try: source = script.read_text(encoding="utf-8") @@ -411,7 +496,13 @@ def cmd_attach(args) -> int: elif not args.port: kernel = str(_build_demo()) session, manager = _start_session( - kernel=kernel, port=args.port, record=args.record, name="repl", plugins=args.plugin + kernel=kernel, + port=args.port, + record=args.record, + name="repl", + plugins=args.plugin, + backend=args.backend, + trace=_trace_spec(args), ) try: GreenEyeRepl(session, manager).cmdloop() @@ -443,7 +534,10 @@ def cmd_launch(args) -> int: def cmd_record(args) -> int: _ensure_dumps_env() elf = args.kernel or str(_build_demo()) - session, _manager = _start_session(kernel=str(elf), port=None, record=args.record or True, name=args.name) + session, _manager = _start_session( + kernel=str(elf), port=None, record=args.record or True, name=args.name, + backend=args.backend, trace=_trace_spec(args), + ) try: for spec in args.watch or []: addr_or_sym, size = (spec.split(":", 1) + ["8"])[:2] @@ -550,7 +644,8 @@ def cmd_inject(args) -> int: ) kernel = args.kernel or str(_build_demo()) session, _manager = _start_session( - kernel=str(kernel), port=args.port, record=args.record or True, name="inject", plugins=args.plugin + kernel=str(kernel), port=args.port, record=args.record or True, name="inject", + plugins=args.plugin, backend=args.backend, trace=_trace_spec(args), ) try: service = load_service(session, str(service_path), load_addr=args.load_addr, slot=args.slot) @@ -578,7 +673,8 @@ def cmd_ghidra(args) -> int: elif not args.target_port: kernel = str(_build_demo()) session, _manager = _start_session( - kernel=kernel, port=args.target_port, record=args.record or True, name="ghidra", plugins=args.plugin + kernel=kernel, port=args.target_port, record=args.record or True, name="ghidra", + plugins=args.plugin, backend=args.backend, trace=_trace_spec(args), ) bridge = GhidraBridge(session, port=args.bridge_port) port = bridge.start() @@ -669,6 +765,8 @@ def build_parser() -> argparse.ArgumentParser: demo.add_argument("--no-build", action="store_true", help="do not rebuild the demo") demo.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or default dumps/)") demo.add_argument("--duration", type=float, default=1.0) + _add_backend_arg(demo) + _add_trace_args(demo) demo.set_defaults(fn=cmd_demo) run = sub.add_parser("run", help="run a Python instrumentation script") @@ -677,6 +775,8 @@ def build_parser() -> argparse.ArgumentParser: run.add_argument("--port", type=int) run.add_argument("--record", nargs="?", const=True, default=None) run.add_argument("--plugin", action="append", default=[]) + _add_backend_arg(run) + _add_trace_args(run) run.set_defaults(fn=cmd_run) attach = sub.add_parser("attach", help="interactive REPL over a gdbstub (launches if no --port)") @@ -684,6 +784,8 @@ def build_parser() -> argparse.ArgumentParser: attach.add_argument("--kernel", help="firmware ELF (default: built demo)") attach.add_argument("--record", nargs="?", const=True, default=None) attach.add_argument("--plugin", action="append", default=[]) + _add_backend_arg(attach) + _add_trace_args(attach) attach.set_defaults(fn=cmd_attach) launch = sub.add_parser("launch", help="start a target and keep it alive for attach") @@ -713,6 +815,8 @@ def build_parser() -> argparse.ArgumentParser: inject.add_argument("--no-firmware", action="store_true", help="only make the direct debugger-side call") inject.add_argument("--record", nargs="?", const=True, default=None) inject.add_argument("--plugin", action="append", default=[]) + _add_backend_arg(inject) + _add_trace_args(inject) inject.set_defaults(fn=cmd_inject) record = sub.add_parser("record", help="boot and record a memory-activity session") @@ -722,6 +826,8 @@ def build_parser() -> argparse.ArgumentParser: record.add_argument("--watch", action="append", help="addr|symbol[:size]") record.add_argument("--snapshot", action="append", help="addr|symbol:len") record.add_argument("--record", nargs="?", const=True, default=None) + _add_backend_arg(record) + _add_trace_args(record) record.set_defaults(fn=cmd_record) ghidra = sub.add_parser("ghidra", help="serve a live session to GhidraScripts (JSON-lines bridge)") @@ -730,6 +836,8 @@ def build_parser() -> argparse.ArgumentParser: ghidra.add_argument("--bridge-port", type=int, default=7000) ghidra.add_argument("--record", nargs="?", const=True, default=None) ghidra.add_argument("--plugin", action="append", default=[]) + _add_backend_arg(ghidra) + _add_trace_args(ghidra) ghidra.set_defaults(fn=cmd_ghidra) analyze = sub.add_parser("analyze", help="render a recorded session report") diff --git a/debug/risc-v-gsp-rm-vm/greeneye/device_map.py b/debug/risc-v-gsp-rm-vm/greeneye/device_map.py new file mode 100644 index 0000000..11ac656 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/device_map.py @@ -0,0 +1,555 @@ +"""Versioned first-touch device map (roadmap H2.1 -> H2.2 start). + +A device map is JSON data, not code: it records which MMIO addresses a +firmware image touched first, at which width and direction, with observed +values, hit counts and provenance. ``harvest_*`` builds one from the QEMU +logs a ``greeneye probe`` session records; the native simulator +(:class:`sim.machine.Machine`) consults it so unmodelled accesses return a +mapped behaviour instead of faulting (``unknown_mmio="lenient"``). + +Schema ``greeneye-device-map/1``:: + + { + "schema": "greeneye-device-map/1", + "meta": {"bias": 2147483648, "firmware": "...", "firmware_sha256": "..."}, + "entries": [ + {"address": 2305843009213696512, "width": 4, "kind": "r", + "values": [], "hits": 1, "first_epc": 2214629396, + "first_epc_fw": 67145716, "source": "probe", "notes": "fault_load"}, + ... + ] + } + +Entries are emitted in a deterministic order (address, width, kind) and the +file contains no timestamps or host-specific paths in semantic fields, so two +harvests of the same log are byte-identical. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +from dataclasses import dataclass + +from .errors import GreenEyeError + +SCHEMA = "greeneye-device-map/1" +KINDS = ("r", "w") +SOURCES = ("probe", "manual", "trace") +_RAM_BASE = 0x80000000 +#: QEMU probe first fault for the TU104 570.144 image (docs/06-ACCURACY.md ยง3). +QEMU_REFERENCE = { + "firmware_vaddr": 0x4009014, + "access_addr": 0x2000000000000A00, + "size": 4, + "rw": "read", + "cause": 5, + "desc": "load access fault", +} + +_BEHAVIOUR_ZERO = "zero" +_BEHAVIOUR_VALUE = "value:" + + +def _fw_pc(pc: int, bias: int) -> int: + """Translate a relocated PC back to the firmware's own vaddr. + + A PC of 0 is a trap-vector fetch (not relocated code) and stays 0; PCs + below the relocation bias are not attributable and stay untranslated. + """ + if not bias or pc == 0 or pc < bias: + return pc + return pc - bias + + +@dataclass(frozen=True) +class DeviceEntry: + """One observed (address, width, direction) MMIO touch.""" + + address: int + width: int + kind: str # "r" | "w" + values: tuple[int, ...] = () + hits: int = 1 + first_epc: int = 0 # PC as observed in the run (relocated) + first_epc_fw: int = 0 # PC - bias: the firmware's own virtual address + source: str = "manual" # probe | manual | trace + notes: str = "" + + def key(self) -> tuple[int, int, str]: + return (self.address, self.width, self.kind) + + def to_dict(self) -> dict: + return { + "address": self.address, + "width": self.width, + "kind": self.kind, + "values": list(self.values), + "hits": self.hits, + "first_epc": self.first_epc, + "first_epc_fw": self.first_epc_fw, + "source": self.source, + "notes": self.notes, + } + + @classmethod + def from_dict(cls, data: dict) -> "DeviceEntry": + return cls( + address=int(data["address"]), + width=int(data["width"]), + kind=str(data["kind"]), + values=tuple(int(v) for v in data.get("values", ())), + hits=int(data.get("hits", 1)), + first_epc=int(data.get("first_epc", 0)), + first_epc_fw=int(data.get("first_epc_fw", 0)), + source=str(data.get("source", "manual")), + notes=str(data.get("notes", "")), + ) + + +class DeviceMap: + """A versioned collection of :class:`DeviceEntry` records.""" + + def __init__(self, *, meta: dict | None = None): + self.meta: dict = dict(meta or {}) + self._entries: dict[tuple[int, int, str], DeviceEntry] = {} + + # -- construction ------------------------------------------------------ + + def record( + self, + addr: int, + size: int, + kind: str, + value: int | None = None, + *, + pc: int = 0, + bias: int = 0, + source: str = "manual", + notes: str = "", + ) -> DeviceEntry: + """Add or merge one observation; returns the resulting entry.""" + if kind not in KINDS: + raise GreenEyeError(f"device map: access kind must be r or w, got {kind!r}") + if size <= 0: + raise GreenEyeError(f"device map: width must be positive, got {size}") + key = (addr, size, kind) + entry = self._entries.get(key) + if entry is None: + entry = DeviceEntry( + address=addr, + width=size, + kind=kind, + values=(value,) if value is not None else (), + hits=1, + first_epc=pc, + first_epc_fw=_fw_pc(pc, bias), + source=source, + notes=notes, + ) + else: + values = entry.values + if value is not None and value not in values: + values = values + (value,) + entry = DeviceEntry( + address=entry.address, + width=entry.width, + kind=entry.kind, + values=values, + hits=entry.hits + 1, + first_epc=entry.first_epc or pc, + first_epc_fw=entry.first_epc_fw or _fw_pc(pc, bias), + source=entry.source, + notes=entry.notes or notes, + ) + self._entries[key] = entry + return entry + + def add(self, entry: DeviceEntry) -> DeviceEntry: + """Merge a whole entry (used by :meth:`merge`).""" + key = entry.key() + current = self._entries.get(key) + if current is None: + self._entries[key] = entry + return entry + values = current.values + for value in entry.values: + if value not in values: + values = values + (value,) + merged = DeviceEntry( + address=current.address, + width=current.width, + kind=current.kind, + values=values, + hits=current.hits + entry.hits, + first_epc=current.first_epc or entry.first_epc, + first_epc_fw=current.first_epc_fw or entry.first_epc_fw, + source=current.source, + notes=current.notes or entry.notes, + ) + self._entries[key] = merged + return merged + + def merge(self, other: "DeviceMap") -> "DeviceMap": + """Merge ``other`` into this map (deterministic given call order).""" + for key in sorted(other._entries): + self.add(other._entries[key]) + for key, value in other.meta.items(): + self.meta.setdefault(key, value) + return self + + # -- queries ----------------------------------------------------------- + + def entries(self) -> list[DeviceEntry]: + """All entries, sorted by ``(address, width, kind)``.""" + return [self._entries[key] for key in sorted(self._entries)] + + def __len__(self) -> int: + return len(self._entries) + + def _find(self, addr: int, size: int, kind: str) -> DeviceEntry | None: + exact = self._entries.get((addr, size, kind)) + if exact is not None: + return exact + for key in sorted(self._entries): # width fallback: smallest width first + k_addr, _k_size, k_kind = key + if k_addr == addr and k_kind == kind: + return self._entries[key] + return None + + def lookup(self, addr: int, size: int, write: bool = False) -> str: + """Behaviour spec for an access: ``zero``, ``value:0x...`` or ``unimplemented``. + + Exact (address, width, direction) wins; a different width at the same + address is still evidence for the register. A write to an address + known only for reads is absorbed (``zero``); a read of a write-only + register reads back zero. An address with no entry at all is + ``unimplemented``. + """ + kind = "w" if write else "r" + entry = self._find(addr, size, kind) + if entry is None: + if self._find(addr, size, "w" if not write else "r") is not None: + return _BEHAVIOUR_ZERO # known address, opposite direction + return "unimplemented" + if not write and entry.values: + return f"{_BEHAVIOUR_VALUE}0x{entry.values[-1]:x}" + return _BEHAVIOUR_ZERO + + # -- persistence ------------------------------------------------------- + + def to_dict(self) -> dict: + return { + "schema": SCHEMA, + "meta": dict(self.meta), + "entries": [entry.to_dict() for entry in self.entries()], + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), indent=1, sort_keys=True) + "\n" + + @classmethod + def from_dict(cls, data: dict) -> "DeviceMap": + schema = data.get("schema") + if schema != SCHEMA: + raise GreenEyeError(f"device map: unsupported schema {schema!r} (want {SCHEMA!r})") + dm = cls(meta=data.get("meta") or {}) + for raw in data.get("entries", ()): + dm.add(DeviceEntry.from_dict(raw)) + return dm + + @classmethod + def from_json(cls, text: str) -> "DeviceMap": + try: + data = json.loads(text) + except ValueError as exc: + raise GreenEyeError(f"device map: invalid JSON ({exc})") from exc + if not isinstance(data, dict): + raise GreenEyeError("device map: top level must be an object") + return cls.from_dict(data) + + def save(self, path: str) -> str: + with open(path, "w", encoding="utf-8") as fh: + fh.write(self.to_json()) + return path + + @classmethod + def load(cls, path: str) -> "DeviceMap": + try: + with open(path, "r", encoding="utf-8") as fh: + text = fh.read() + except OSError as exc: + raise GreenEyeError(f"device map: cannot read {path}: {exc}") from exc + return cls.from_json(text) + + +def firmware_sha256(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +# -- harvest --------------------------------------------------------------- + + +def harvest_from_text(log_text: str, bias: int = 0, source: str = "probe") -> DeviceMap: + """Build a map from an arbitrary QEMU ``guest_errors,unimp,int`` log. + + Only records that pair an invalid access with its exception contribute an + entry, using the faulting PC (translated by ``bias``) and the access + address/width/direction. Fetch-loop faults (no access) are skipped. + """ + from .analyze import parse_qemu_faults + + dm = DeviceMap(meta={"bias": bias, "source": source}) + for fault in parse_qemu_faults(log_text): + access = fault.get("access") + if not access: + continue + kind = "r" if access.get("rw") == "read" else "w" + desc = fault.get("desc") or "" + reason = access.get("reason") or "" + dm.record( + int(access["access_addr"]), + int(access["size"]), + kind, + pc=int(fault.get("epc", 0)), + bias=bias, + source=source, + notes=f"{desc} reason={reason}".strip(), + ) + return dm + + +def _fault_to_map(fault: dict, bias: int, source: str = "probe") -> DeviceMap: + dm = DeviceMap(meta={"bias": bias, "source": source}) + access = fault.get("access") + if access: + kind = "r" if access.get("rw") == "read" else "w" + dm.record( + int(access["access_addr"]), + int(access["size"]), + kind, + pc=int(fault.get("epc", 0)), + bias=bias, + source=source, + notes=str(fault.get("desc") or ""), + ) + return dm + + +def harvest_probe_session(session_dir: str, bias: int | None = None) -> DeviceMap: + """Harvest a ``greeneye probe`` session directory into a device map. + + Reads ``session.json`` for the relocation bias and firmware identity, then + the ``qemu_log_blob`` (falling back to ``blobs/qemu.log``, ``qemu.log`` or + the stored ``qemu_first_fault`` record when no blob was kept). ``bias`` + overrides the session's recorded relocation bias when given. + """ + session_file = os.path.join(session_dir, "session.json") + try: + with open(session_file, "r", encoding="utf-8") as fh: + session = json.load(fh) + except FileNotFoundError as exc: + raise GreenEyeError(f"not a greeneye session directory: {session_dir}") from exc + except ValueError as exc: + raise GreenEyeError(f"{session_file}: invalid JSON ({exc})") from exc + + if bias is None: + bias = int(session.get("bias", 0) or 0) + log_text: str | None = None + blob = session.get("qemu_log_blob") + candidates = [blob] if blob else [] + candidates += ["blobs/qemu.log", "qemu.log"] + log_path: str | None = None + for candidate in candidates: + if candidate and os.path.exists(os.path.join(session_dir, candidate)): + log_path = os.path.join(session_dir, candidate) + break + if log_path is not None: + with open(log_path, "r", encoding="utf-8", errors="replace") as fh: + log_text = fh.read() + + first = session.get("qemu_first_fault") + if log_text is not None: + dm = harvest_from_text(log_text, bias=bias) + elif isinstance(first, dict): + dm = _fault_to_map(first, bias) + else: + dm = DeviceMap(meta={"bias": bias}) + + meta = dm.meta + meta["session"] = os.path.basename(os.path.abspath(session_dir)) + if session.get("firmware"): + meta["firmware"] = session["firmware"] + if session.get("entry") is not None: + meta["entry"] = int(session["entry"]) + if session.get("qemu_fault_count") is not None: + meta["qemu_fault_count"] = int(session["qemu_fault_count"]) + fw = session.get("firmware") + if fw and os.path.isfile(fw): + meta["firmware_sha256"] = firmware_sha256(fw) + version_file = os.path.join(os.path.dirname(fw), "gsp_version.txt") + if os.path.isfile(version_file): + with open(version_file, "r", encoding="utf-8", errors="replace") as fh: + meta["firmware_version"] = "".join( + ch for ch in fh.read() if ch.isprintable() + ).strip() + return dm + + +# -- CLI ------------------------------------------------------------------- + + +def _default_image() -> str: + tool_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + return os.path.join( + os.path.dirname(os.path.dirname(tool_root)), + "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin", + ) + + +def _cmd_harvest(args) -> int: + dm = harvest_probe_session(args.session_dir, bias=args.bias) + print(f"[device-map] harvested {len(dm)} entries from {args.session_dir}", file=sys.stderr) + if args.out: + dm.save(args.out) + print(f"[device-map] wrote {args.out}", file=sys.stderr) + else: + sys.stdout.write(dm.to_json()) + return 0 + + +def _cmd_show(args) -> int: + dm = DeviceMap.load(args.map) + if args.json: + sys.stdout.write(dm.to_json()) + return 0 + print(f"map: {args.map}") + for key in sorted(dm.meta): + print(f"meta {key}: {dm.meta[key]}") + print(f"{'address':>20} {'w':>2} {'rw':>3} {'hits':>7} {'first_epc_fw':>14} {'values':>10} source") + for entry in dm.entries(): + values = ",".join(f"{v:#x}" for v in entry.values) or "-" + print( + f"{entry.address:#20x} {entry.width:>2} {entry.kind:>3} {entry.hits:>7} " + f"{entry.first_epc_fw:#14x} {values:>10} {entry.source}" + ) + return 0 + + +def _cmd_merge(args) -> int: + merged = DeviceMap() + for path in args.maps: + merged.merge(DeviceMap.load(path)) + merged.save(args.out) + print(f"[device-map] merged {len(args.maps)} maps -> {args.out} ({len(merged)} entries)") + return 0 + + +def _run_real_image(args) -> tuple: + """Load the fwimage into a lenient native Machine and run a fixed budget.""" + from .elf import load_elf + from sim.machine import Machine + + image = load_elf(args.image) + bias = args.bias + end = image.max_vaddr + bias + need = end - _RAM_BASE + (1 << 20) + memory_mb = max(256, (need + (1 << 20) - 1) // (1 << 20)) + dm = DeviceMap(meta={"source": "learn", "image": os.path.basename(args.image), "bias": bias}) + machine = Machine( + memory_mb=memory_mb, device_map=dm, unknown_mmio="lenient", learn_mmio=True, bias=bias + ) + machine.load_segments(image.flat_load_plan(bias=bias), image.relocated_entry(bias=bias)) + reason, pc, a, b = machine.run(args.instructions) + return machine, image, dm, reason, pc, a, b + + +_REASONS = { + 0: "budget", 1: "breakpoint", 2: "watch", 3: "ebreak", 4: "fault", 5: "step", 6: "ecall", +} + + +def _cmd_demo(args) -> int: + if not os.path.isfile(args.image): + print(f"[device-map] image not found: {args.image}", file=sys.stderr) + return 2 + machine, image, dm, reason, pc, a, b = _run_real_image(args) + digest = firmware_sha256(args.image) + print(f"image: {args.image}") + print(f"sha256: {digest}") + print(f"elf: entry={image.entry:#x} vaddr=[{image.min_vaddr:#x},{image.max_vaddr:#x})") + print(f"bias: {args.bias:#x} relocated entry: {image.relocated_entry(args.bias):#x}") + print(f"budget: {args.instructions} instructions") + print(f"stopped: reason={_REASONS.get(reason, reason)} pc={pc:#x} (fw {pc - args.bias:#x})") + print(f"unknown mmio accesses: {machine.mmio_unknown}") + for line in machine.mmio_log[: args.first]: + print( + f" first unknown: pc={line['pc']:#x} (fw {line['pc'] - args.bias:#x}) " + f"addr={line['addr']:#x} size={line['size']} rw={line['rw']}" + ) + if machine.mmio_log and machine.mmio_log[0]["addr"] == QEMU_REFERENCE["access_addr"]: + delta = machine.insns - 6 # QEMU traps on the sixth instruction + print( + "[device-map] qemu probe first fault: " + f"fw vaddr {QEMU_REFERENCE['firmware_vaddr']:#x} reading " + f"{QEMU_REFERENCE['access_addr']:#x} ({QEMU_REFERENCE['desc']})" + ) + print( + "[device-map] verdict: first-touch address reproduced; lenient execution " + f"continued for {machine.insns} instructions ({delta} past the qemu trap)" + ) + else: + print("[device-map] verdict: no unknown access at the documented first-touch address") + if args.out: + dm.save(args.out) + print(f"[device-map] learned map ({len(dm)} entries) -> {args.out}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python3 -m greeneye.device_map", + description="Harvest, inspect and merge green-eye first-touch device maps.", + ) + sub = parser.add_subparsers(dest="cmd", required=True) + + harvest = sub.add_parser("harvest", help="harvest a greeneye probe session directory") + harvest.add_argument("session_dir") + harvest.add_argument("--out") + harvest.add_argument("--bias", type=lambda s: int(s, 0), default=None) + harvest.set_defaults(func=_cmd_harvest) + + show = sub.add_parser("show", help="print a device map") + show.add_argument("map") + show.add_argument("--json", action="store_true", help="emit the raw JSON") + show.set_defaults(func=_cmd_show) + + merge = sub.add_parser("merge", help="merge maps into one (self, then others)") + merge.add_argument("maps", nargs="+") + merge.add_argument("--out", required=True) + merge.set_defaults(func=_cmd_merge) + + demo = sub.add_parser("demo", help="bounded lenient run of a real fwimage") + demo.add_argument("--image", default=_default_image()) + demo.add_argument("--bias", type=lambda s: int(s, 0), default=_RAM_BASE) + demo.add_argument("--instructions", type=int, default=200_000) + demo.add_argument("--first", type=int, default=8) + demo.add_argument("--out", help="save the learned device map here") + demo.set_defaults(func=_cmd_demo) + + args = parser.parse_args(argv) + try: + return args.func(args) + except GreenEyeError as exc: + print(f"[device-map] error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py index 6986d74..399a355 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py @@ -73,6 +73,8 @@ def corpus_location(symbol, corpus_root: str) -> str | None: def render_ghidra_command(project_dir: str, project_name: str, program: str, hits_tsv: str) -> str: + """Render the ``analyzeHeadless`` command that runs MarkGreenEyeHits.java + against ``hits_tsv`` (the file written by :func:`export_hits`).""" return ( "analyzeHeadless {proj} {name} -process {program} -noanalysis " "-scriptPath /debug/risc-v-gsp-rm-vm/ghidra " diff --git a/debug/risc-v-gsp-rm-vm/greeneye/session.py b/debug/risc-v-gsp-rm-vm/greeneye/session.py index fa54f77..fffc60e 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/session.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/session.py @@ -227,6 +227,8 @@ class Session: raise GreenEyeError(f"cannot resolve target {target!r}") def symbol_at(self, addr: int) -> Symbol | None: + """Return the symbol containing ``addr``, or the nearest one (alias of + ``symbols.resolve``); ``None`` when the database is empty.""" return self.symbols.resolve(addr) # -- breakpoints / hooks ---------------------------------------------- diff --git a/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py b/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py new file mode 100644 index 0000000..d17076f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py @@ -0,0 +1,390 @@ +"""Native green-eye execution backend (M1): deterministic RV64IMAC in Python. + +``ge.launch(kernel=..., backend="native")`` builds a :class:`SimTarget` around +:class:`sim.machine.Machine` and exposes a :class:`SimClient` whose method +surface matches :class:`greeneye.rsp.RspClient`, so ``greeneye.session.Session`` +runs unchanged against it. ``trace=`` enables the M1.5 exec/access trace +(:mod:`sim.trace`), written to the session as ``blobs/trace.jsonl``. +``device_map=`` and ``unknown_mmio="lenient"`` (H2.1/H2.2 start) serve +unmodelled MMIO accesses from a harvested first-touch map instead of faulting; +lenient accesses are logged to ``target.mmio_log`` and counted in +``target.mmio_unknown``. + +Virtual time is instruction count: ``continue_(timeout)`` executes at most +``timeout * SIM_IPS`` instructions (SIM_IPS = 1 MIPS), deterministically. The +wall-clock budgets ``Session.run`` derives from its deadline therefore bound +the run; the stop/watch/hook sequence within a fixed instruction budget is +reproducible on every host. +""" +from __future__ import annotations + +import os +import sys + +from .elf import load_elf +from .errors import GreenEyeError, RspProtocolError, RspTimeout +from .rsp import Register, StopReply +from .session import Session +from .symbols import SymbolDB + +_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _TOOL_ROOT not in sys.path: # `sim/` lives beside the greeneye package + sys.path.insert(0, _TOOL_ROOT) + +from sim.machine import ( # noqa: E402 + R_BUDGET, + R_FAULT, + R_WATCH, + Machine, +) +from sim.trace import make_trace # noqa: E402 +#: Virtual instructions per second used to convert wall timeouts to budgets. +SIM_IPS = 1_000_000 +#: Safety valve for ``continue_(timeout=None)``: never run unbounded. +MAX_UNBOUNDED_BUDGET = 20_000_000 + +#: Canonical register names, as exposed by QEMU's target.xml for riscv:rv64. +ABI_NAMES = ( + ["zero", "ra", "sp", "gp", "tp", "t0", "t1", "t2", "s0", "s1"] + + [f"a{i}" for i in range(8)] + + [f"s{i}" for i in range(2, 12)] + + [f"t{i}" for i in range(3, 7)] +) +#: xN -> ABI-name aliases (x0 == zero, x1 == ra, ...). +_X_TO_ABI = {f"x{i}": ABI_NAMES[i] for i in range(32)} +_X_TO_ABI["fp"] = "s0" +_X_TO_ABI["return"] = "ra" +_ABI_TO_INDEX = {name: i for i, name in enumerate(ABI_NAMES)} + + +class SimClient: + """Drop-in replacement for :class:`RspClient` over a native Machine.""" + + def __init__(self, machine: Machine, *, sim_ips: int = SIM_IPS, trace=None): + self.machine = machine + self.sim_ips = max(1, int(sim_ips)) + self.timeout = 10.0 + self.packet_size = 0x1000 + self.trace = trace # sim.trace.Trace | None + self._recorder = None + self.registers = [ + Register(number=i, name=name, bitsize=64, offset=i * 8) + for i, name in enumerate(ABI_NAMES) + ] + [Register(number=32, name="pc", bitsize=64, offset=32 * 8)] + self._by_name = {r.name: r for r in self.registers} + self._by_name.update(_X_TO_ABI) + self._by_number = {r.number: r for r in self.registers} + self.features = "PacketSize=1000;swbreak+;hwbreak+;greeneye-native+" + self.connected = True + self._bps: set[int] = set() + self._hw_bps: set[int] = set() + self._watches: dict[tuple[int, int, str], None] = {} + + # -- lifecycle --------------------------------------------------------- + + def connect(self, retries: int = 1, retry_delay: float = 0.1) -> None: + self.connected = True + + def sync(self) -> None: + pass # no server to hand-shake with + + def close(self) -> None: + self.connected = False + if self.trace is not None: + self.trace.flush(self._recorder) + + def __enter__(self) -> "SimClient": + return self + + def __exit__(self, *exc) -> None: + self.close() + + # -- registers --------------------------------------------------------- + + def read_registers(self) -> dict[str, int]: + x = self.machine.x + out = {name: x[i] for i, name in enumerate(ABI_NAMES)} + out["fp"] = out["s0"] # QEMU's target.xml names x8 "fp" + out["pc"] = self.machine.pc + return out + + def _index(self, name_or_number: str | int) -> int: + if isinstance(name_or_number, int): + if 0 <= name_or_number < 32: + return name_or_number + if name_or_number == 32: + return 32 + raise RspProtocolError(f"unknown register number {name_or_number}") + key = _X_TO_ABI.get(name_or_number, name_or_number) + if key == "pc": + return 32 + index = _ABI_TO_INDEX.get(key) + if index is None: + raise RspProtocolError(f"unknown register {name_or_number!r}") + return index + + def read_reg(self, name_or_number: str | int) -> int: + index = self._index(name_or_number) + if index == 32: + return self.machine.pc + return self.machine.reg(index) + + def write_reg(self, name_or_number: str | int, value: int) -> None: + index = self._index(name_or_number) + if index == 32: + self.machine.pc = value & ((1 << 64) - 1) + else: + self.machine.set_reg(index, value) + + @property + def pc(self) -> int: + return self.machine.pc + + # -- memory ------------------------------------------------------------ + + def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes: + try: + return self.machine.read_mem(addr, size) + except Exception as exc: + raise RspProtocolError(str(exc)) from exc + + def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None: + try: + self.machine.write_mem(addr, data) + except Exception as exc: + raise RspProtocolError(str(exc)) from exc + + def read_cstring(self, addr: int, max_len: int = 256) -> str: + raw = self.read_mem(addr, max_len) + return raw.split(b"\x00", 1)[0].decode("utf-8", "replace") + + # -- breakpoints / watchpoints ---------------------------------------- + + def set_sw_break(self, addr: int, kind: int = 4) -> None: + self._bps.add(addr) + + def clear_sw_break(self, addr: int, kind: int = 4) -> None: + self._bps.discard(addr) + + def set_hw_break(self, addr: int, kind: int = 4) -> None: + self._hw_bps.add(addr) + + def clear_hw_break(self, addr: int, kind: int = 4) -> None: + self._hw_bps.discard(addr) + + _WATCH_KINDS = {"w": 2, "r": 3, "a": 4} + _WATCH_REVERSE = {"w": "watch", "r": "rwatch", "a": "awatch"} + + def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + if kind not in self._WATCH_KINDS: + raise RspProtocolError(f"unknown watchpoint kind {kind!r}") + self._watches[(addr, size, kind)] = None + + def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + self._watches.pop((addr, size, kind), None) + + def _watcher_list(self): + out = [] + for (addr, size, kind) in self._watches: + mask = (2 if kind in ("r", "a") else 0) | (1 if kind in ("w", "a") else 0) + out.append((addr, addr + size, kind, mask)) + return tuple(out) + + # -- run control ------------------------------------------------------- + + def _budget(self, timeout: float | None) -> int | None: + if timeout is None: + return None + return max(1, int(timeout * self.sim_ips)) + + def _to_stop(self, result) -> StopReply: + reason, pc, a, b = result + if reason == R_WATCH: + kind = self._WATCH_REVERSE.get(b, "watch") + return StopReply( + raw=f"T05watch:{a:x};", + signal=5, + watch_addr=a, + watch_kind=kind, + ) + if reason == R_FAULT: + signal = 4 if a == 2 else 11 + return StopReply(raw=f"T0{signal:x}", signal=signal) + # bp, ebreak, ecall and step all report SIGTRAP, like the gdbstub. + return StopReply(raw="T05", signal=5) + + def continue_(self, timeout: float | None = None) -> StopReply: + budget = self._budget(timeout) + if budget is None: + budget = MAX_UNBOUNDED_BUDGET + result = self.machine.run( + budget, + breakpoints=frozenset(self._bps | self._hw_bps), + watchers=self._watcher_list(), + trace=self.trace, + ) + if result[0] == R_BUDGET: + raise RspTimeout(f"native budget exhausted after {budget} instructions") + return self._to_stop(result) + + def step(self, timeout: float | None = None) -> StopReply: + result = self.machine.step(watchers=self._watcher_list(), trace=self.trace) + return self._to_stop(result) + + def interrupt(self, timeout: float | None = None) -> StopReply: + return StopReply(raw="T02", signal=2) + + # -- monitor ----------------------------------------------------------- + + def monitor(self, cmd: str) -> str: + return "" + + +class SimTarget: + """QemuTarget-shaped handle for a native Machine (no process, no socket).""" + + def __init__( + self, + machine: Machine, + *, + memory_mb: int = 256, + ram_base: int = 0x80000000, + sim_ips: int = SIM_IPS, + keep_alive: bool = False, + trace=None, + ): + self.machine = machine + self.memory_mb = memory_mb + self.ram_base = ram_base + self.keep_alive = keep_alive + self.client = SimClient(machine, sim_ips=sim_ips, trace=trace) + self.kernel: str | None = None + self.workdir: str | None = None + self._alive = True + + # QemuTarget lifecycle surface ---------------------------------------- + + def start(self, paused: bool = True, timeout: float = 0.0) -> "SimTarget": + return self + + def wait_for_port(self, timeout: float = 0.0) -> None: + return None + + def alive(self) -> bool: + return self._alive + + def kill(self, timeout: float = 0.0) -> None: + self._alive = False + self.client.close() + + def read_uart(self) -> str: + return self.machine.uart.text() + + @property + def mmio_log(self) -> list[dict]: + """Lenient-mode log (empty unless ``unknown_mmio="lenient"``).""" + return self.machine.mmio_log + + @property + def mmio_unknown(self) -> int: + """Count of unmodelled MMIO accesses served leniently.""" + return self.machine.mmio_unknown + + def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool: + # Execution is synchronous: by the time the caller looks, the marker + # either exists or the budget was too small. No polling needed. + return marker in self.read_uart() + + def read_qemu_log(self) -> str: + return "" + + @property + def command_line(self) -> list[str]: + return [f"native-sim:{self.kernel or ''}"] + + def shell_command(self) -> str: + return " ".join(self.command_line) + + def __repr__(self) -> str: # pragma: no cover - cosmetic + return f"SimTarget(pc={self.machine.pc:#x}, alive={self._alive})" + + +def launch_native( + *, + kernel: str | None = None, + memory_mb: int = 256, + record=None, + symbols: SymbolDB | None = None, + name: str = "session", + load_symbols: bool = True, + keep_alive: bool = False, + sim_ips: int = SIM_IPS, + trace=None, + device_map=None, + unknown_mmio: str = "fault", + learn_mmio: bool = False, + bias: int = 0, +) -> Session: + """Boot the native simulator and return a ready Session. + + ``trace`` enables the R4 execution/access trace: ``True`` for defaults, a + dict of ``{"exec": bool, "start": int, "end": int, "max_events": int}``, + or a pre-built :class:`sim.trace.Trace`. The trace is written to the + session as ``blobs/trace.jsonl`` when the Session closes; tracing implies + recording, so a session directory is created when none was given. + + ``device_map`` is a :class:`greeneye.device_map.DeviceMap` (or a path to a + saved map) describing modelled MMIO; ``unknown_mmio="lenient"`` serves + unmodelled accesses from it (or zero) instead of faulting, logging each in + ``machine.mmio_log`` and counting ``machine.mmio_unknown``. + ``learn_mmio=True`` writes the observed values back into the map. + ``bias`` relocates PT_LOAD segments and the entry for images linked for + GPU physical addresses (e.g. ``0x80000000`` for the real GSP fwimage). + """ + try: + tracer = make_trace(trace) + except ValueError as exc: + raise GreenEyeError(f"invalid trace spec: {exc}") from exc + if tracer is not None and not record: + record = True # the trace artifact lives in the session + if isinstance(device_map, (str, os.PathLike)): + from .device_map import DeviceMap + + device_map = DeviceMap.load(os.fspath(device_map)) + machine = Machine( + memory_mb=memory_mb, + device_map=device_map, + unknown_mmio=unknown_mmio, + learn_mmio=learn_mmio, + bias=bias, + ) + target = SimTarget( + machine, + memory_mb=memory_mb, + sim_ips=sim_ips, + keep_alive=keep_alive, + trace=tracer, + ) + target.kernel = kernel + symdb = symbols or SymbolDB() + meta = {"backend": "native", "sim_ips": sim_ips, "firmware": kernel} + if unknown_mmio != "fault" or device_map is not None: + meta["unknown_mmio"] = unknown_mmio + session = Session( + client=target.client, + target=target, + symbols=symdb, + record=record, + name=name, + meta=meta, + ) + if tracer is not None: + target.client._recorder = session.recorder + session.recorder.meta["trace"] = {**tracer.config()} + if kernel: + image = load_elf(kernel) + if not image.is_riscv: + raise GreenEyeError(f"{kernel}: not a RISC-V ELF image") + machine.load_segments(image.flat_load_plan(bias=bias), image.relocated_entry(bias=bias)) + session.load_image(image, bias=bias, load_symbols=load_symbols) + return session diff --git a/debug/risc-v-gsp-rm-vm/greeneye/symbols.py b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py index 7b2e9dd..cb11f7f 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/symbols.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py @@ -25,7 +25,7 @@ import json import os from dataclasses import dataclass, field as _field -from .errors import SymbolNotFound +from .errors import GreenEyeError, SymbolNotFound @dataclass @@ -193,9 +193,15 @@ class SymbolDB: return count def load_json(self, path: str) -> int: - with open(path, "r", encoding="utf-8") as fh: - payload = json.load(fh) - symbols = payload.get("symbols", payload if isinstance(payload, list) else []) + """Load a green-eye JSON symbol DB; raise GreenEyeError on bad input.""" + try: + with open(path, "r", encoding="utf-8") as fh: + payload = json.load(fh) + except (OSError, json.JSONDecodeError) as exc: + raise GreenEyeError(f"cannot load symbol DB {path!r}: {exc}") from exc + if not isinstance(payload, (dict, list)): + raise GreenEyeError(f"symbol DB {path!r} must be a JSON object or list") + symbols = payload.get("symbols", []) if isinstance(payload, dict) else payload count = 0 for item in symbols: try: @@ -276,12 +282,15 @@ class SymbolDB: return f"{addr:#x} {sym.label(addr)}" def nearest(self, addr: int, max_delta: int = 0x10000) -> Symbol | None: + """Alias of :meth:`resolve` kept for call sites that read better as + "nearest symbol" (same containing/preceding/max-delta rules).""" return self.resolve(addr, max_delta=max_delta) def names(self) -> list[str]: return sorted(self._by_name) def by_subsystem(self, subsystem: str) -> list[Symbol]: + """Return all symbols tagged with ``subsystem`` (loaders set it when known).""" return [s for s in self._symbols if s.subsystem == subsystem] -- cgit v1.2.3