diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness')
22 files changed, 7504 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/__init__.py b/debug/risc-v-gsp-rm-vm/harness/__init__.py new file mode 100644 index 0000000..d99d547 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/__init__.py @@ -0,0 +1,11 @@ +"""green-eye automation harness. + +Two stdlib-only orchestration tools: + +* :mod:`harness.portability_matrix` - multi-version/multi-target probe + demo + orchestration with a cross-target delta report (roadmap R6, H4.1); +* :mod:`harness.mutation_harness` - bounded, seeded mutation of the demo + firmware's mailbox input surface (roadmap R5, H4.3). + +Both are development tools; see docs/18-HARNESS-AUTOMATION.md. +""" diff --git a/debug/risc-v-gsp-rm-vm/harness/ci_release.py b/debug/risc-v-gsp-rm-vm/harness/ci_release.py new file mode 100644 index 0000000..dbc1f6f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/ci_release.py @@ -0,0 +1,444 @@ +#!/usr/bin/env python3 +"""Release triage orchestration for green-eye (planned docs/21). + +Given an old and a new firmware target it runs, in order: + +1. ``harness/version_diff.py`` -- image/symbol/corpus/probe/behavior diff; +2. ``harness/portability_matrix.py`` -- probe + scenario + cross-target deltas; +3. ``harness/mutation_harness.py`` -- a small deterministic mutation smoke. + +Every step writes under ``--out`` and the whole directory is hashed into +``bundle-manifest.json`` (the evidence bundle). A failing step is recorded and +the remaining steps still run; the process exits non-zero only when a step +reports an error. + +``--dry-run`` prints the plan and exits 0 without opening targets, and the +RM 610 gate (docs/23) is enforced before any target file is read. Nothing is +ever downloaded. + +Run from the tool root: + + python3 harness/ci_release.py --old old.elf --new new.elf --out dumps/release + python3 harness/ci_release.py --old old.elf --new new.elf --dry-run +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +from dataclasses import dataclass +from pathlib import Path + +try: + from .common import DEMO_ELF, ensure_dir, qemu_available, sha256_file, write_json, write_text + from .mutation_harness import HarnessOptions, run_harness + from .portability_matrix import MatrixOptions, TargetSpec, render_markdown, run_matrix + from .version_diff import ( + RM610_TOKEN, + Rm610GateError, + VersionDiffOptions, + check_rm610_gate, + run_version_diff, + ) +except ImportError: # direct script execution (python3 harness/ci_release.py) + from common import ( # type: ignore[no-redef] + DEMO_ELF, + ensure_dir, + qemu_available, + sha256_file, + write_json, + write_text, + ) + from mutation_harness import HarnessOptions, run_harness # type: ignore[no-redef] + from portability_matrix import MatrixOptions, TargetSpec, render_markdown, run_matrix # type: ignore[no-redef] + from version_diff import ( # type: ignore[no-redef] + RM610_TOKEN, + Rm610GateError, + VersionDiffOptions, + check_rm610_gate, + run_version_diff, + ) + +FORMAT = "greeneye-ci-release/1" +PLAN_FORMAT = "greeneye-ci-release-plan/1" +BUNDLE_FORMAT = "greeneye-release-bundle/1" + + +@dataclass +class ReleaseOptions: + old: str + new: str + out: str | None = None + old_label: str | None = None + new_label: str | None = None + release: str | None = None + approval_token: str | None = None + old_symbols: str | None = None + new_symbols: str | None = None + matrix_symbols: str | None = None + old_ghidra_names: str | None = None + new_ghidra_names: str | None = None + old_ghidra_index: str | None = None + new_ghidra_index: str | None = None + old_ghidra_exports: str | None = None + new_ghidra_exports: str | None = None + old_ghidra_bias: int = 0 + new_ghidra_bias: int = 0 + corpus_diff: tuple[str, str] | None = None + corpus_report: str | None = None + stimulus: str | None = None + probe_instructions: int = 2000 + behavior_duration: float = 0.4 + max_rows: int = 40 + mutation_kernel: str = DEMO_ELF + mutation_iterations: int = 8 + mutation_seed: int = 7 + mutation_timeout: float = 0.25 + mutation_settle: float = 0.1 + allow_foreign_mutation: bool = False + dry_run: bool = False + + def labels(self) -> list[str]: + return [ + value + for value in ( + self.release, + self.old_label, + self.new_label, + os.path.basename(self.old), + os.path.basename(self.new), + os.path.basename(self.mutation_kernel) if self.mutation_kernel else None, + ) + if value + ] + + +def _version_diff_options(options: ReleaseOptions, out_dir: str) -> VersionDiffOptions: + return VersionDiffOptions( + old=options.old, + new=options.new, + out=out_dir, + old_label=options.old_label, + new_label=options.new_label, + release=options.release, + old_symbols=options.old_symbols, + new_symbols=options.new_symbols, + old_ghidra_names=options.old_ghidra_names, + new_ghidra_names=options.new_ghidra_names, + old_ghidra_index=options.old_ghidra_index, + new_ghidra_index=options.new_ghidra_index, + old_ghidra_exports=options.old_ghidra_exports, + new_ghidra_exports=options.new_ghidra_exports, + old_ghidra_bias=options.old_ghidra_bias, + new_ghidra_bias=options.new_ghidra_bias, + corpus_diff=options.corpus_diff, + corpus_report=options.corpus_report, + stimulus=options.stimulus, + probe_instructions=options.probe_instructions, + behavior_duration=options.behavior_duration, + max_rows=options.max_rows, + approval_token=options.approval_token, + ) + + +# --------------------------------------------------------------------------- +# plan (dry run) +# --------------------------------------------------------------------------- + + +def build_plan(options: ReleaseOptions) -> dict: + out = os.path.abspath(options.out) if options.out else "<out>" + vd_out = os.path.join(out, "version-diff") + matrix_out = os.path.join(out, "portability-matrix") + mutation_out = os.path.join(out, "mutation-smoke") + vd_argv = [sys.executable, "harness/version_diff.py", "--old", options.old, "--new", options.new, + "--out", vd_out, "--probe-instructions", str(options.probe_instructions), + "--behavior-duration", str(options.behavior_duration), "--max-rows", str(options.max_rows)] + if options.old_label: + vd_argv += ["--old-label", options.old_label] + if options.new_label: + vd_argv += ["--new-label", options.new_label] + if options.release: + vd_argv += ["--release", options.release] + if options.old_symbols: + vd_argv += ["--old-symbols", options.old_symbols] + if options.new_symbols: + vd_argv += ["--new-symbols", options.new_symbols] + if options.stimulus: + vd_argv += ["--stimulus", options.stimulus] + matrix_argv = [sys.executable, "harness/portability_matrix.py", "--target", options.old, + "--target", options.new, "--out", matrix_out, + "--probe-instructions", str(options.probe_instructions), + "--scenario-duration", str(options.behavior_duration), + "--max-rows", str(options.max_rows)] + mutation_argv = [sys.executable, "harness/mutation_harness.py", "--kernel", options.mutation_kernel, + "--out", mutation_out, "--iterations", str(options.mutation_iterations), + "--seed", str(options.mutation_seed), "--timeout", str(options.mutation_timeout), + "--settle", str(options.mutation_settle)] + if options.allow_foreign_mutation: + mutation_argv.append("--allow-foreign-elf") + return { + "format": PLAN_FORMAT, + "tool": "harness/ci_release.py", + "dry_run": True, + "status": "planned", + "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))}, + "old": {"label": options.old_label or Path(options.old).stem, "path": options.old}, + "new": {"label": options.new_label or Path(options.new).stem, "path": options.new}, + "out": out, + "steps": [ + {"name": "version_diff", "out": vd_out, "argv": vd_argv}, + {"name": "portability_matrix", "out": matrix_out, "argv": matrix_argv}, + {"name": "mutation_smoke", "out": mutation_out, "argv": mutation_argv}, + ], + "manifest": os.path.join(out, "bundle-manifest.json"), + "release_report": os.path.join(out, "release-report.json"), + } + + +# --------------------------------------------------------------------------- +# steps +# --------------------------------------------------------------------------- + + +def step_version_diff(options: ReleaseOptions, out_dir: str) -> dict: + report = run_version_diff(_version_diff_options(options, out_dir)) + sections = {name: section["status"] for name, section in report["sections"].items()} + status = "error" if "error" in sections.values() else "ok" + return { + "status": status, + "out": os.path.basename(out_dir), + "report": "version-diff.json", + "sections": sections, + "summary": report["summary"], + } + + +def step_portability_matrix(options: ReleaseOptions, out_dir: str) -> dict: + matrix = run_matrix( + [TargetSpec(path=os.path.abspath(options.old), name=Path(options.old).stem), + TargetSpec(path=os.path.abspath(options.new), name=Path(options.new).stem)], + MatrixOptions( + out=out_dir, + probe_instructions=options.probe_instructions, + scenario_duration=options.behavior_duration, + max_rows=options.max_rows, + symbols_json=options.matrix_symbols, + ), + ) + ensure_dir(out_dir) + write_json(os.path.join(out_dir, "portability-report.json"), matrix) + write_text( + os.path.join(out_dir, "portability-report.md"), + render_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)), + ) + statuses = [target["status"] for target in matrix["targets"]] + status = "error" if "error" in statuses else "ok" + return { + "status": status, + "report": "portability-report.json", + "targets": {target["name"]: target["status"] for target in matrix["targets"]}, + "comparisons": [ + {"a": comp["a"], "b": comp["b"], "status": comp["status"], "all_zero": comp["all_zero"]} + for comp in matrix["comparisons"] + ], + } + + +def step_mutation_smoke(options: ReleaseOptions, out_dir: str) -> dict: + if not qemu_available(): + return {"status": "skipped", "reason": "qemu-system-riscv64 not found"} + if not os.path.isfile(options.mutation_kernel): + return {"status": "skipped", "reason": f"mutation kernel not found: {options.mutation_kernel}"} + harness = HarnessOptions( + kernel=os.path.abspath(options.mutation_kernel), + out=out_dir, + iterations=options.mutation_iterations, + seed=options.mutation_seed, + timeout=options.mutation_timeout, + settle=options.mutation_settle, + allow_foreign_elf=options.allow_foreign_mutation, + ) + findings = run_harness(harness) + outcomes: dict[str, int] = {} + for finding in findings: + outcomes[finding["outcome"]] = outcomes.get(finding["outcome"], 0) + 1 + return { + "status": "ok", + "report": "findings.jsonl", + "iterations": len(findings), + "outcomes": dict(sorted(outcomes.items())), + } + + +def _run_step(name: str, fn, *args) -> dict: + try: + detail = fn(*args) + except Exception as exc: # a failed step must not hide the others + return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"} + entry = {"name": name} + entry.update(detail) + return entry + + +# --------------------------------------------------------------------------- +# bundle +# --------------------------------------------------------------------------- + + +def build_manifest(out_dir: str) -> dict: + files: list[dict] = [] + for root, dirs, names in os.walk(out_dir): + dirs.sort() + for name in sorted(names): + path = os.path.join(root, name) + rel = os.path.relpath(path, out_dir) + if rel == "bundle-manifest.json": + continue + files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)}) + files.sort(key=lambda row: row["path"]) + return {"format": BUNDLE_FORMAT, "tool": "harness/ci_release.py", "files": files} + + +def run_release(options: ReleaseOptions) -> dict: + check_rm610_gate(options.labels(), approval_token=options.approval_token) + if not options.out: + raise ValueError("--out is required unless --dry-run is used") + out = os.path.abspath(options.out) + ensure_dir(out) + steps = [ + _run_step("version_diff", step_version_diff, options, os.path.join(out, "version-diff")), + _run_step("portability_matrix", step_portability_matrix, options, + os.path.join(out, "portability-matrix")), + _run_step("mutation_smoke", step_mutation_smoke, options, os.path.join(out, "mutation-smoke")), + ] + report = { + "format": FORMAT, + "tool": "harness/ci_release.py", + "dry_run": False, + "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))}, + "old": {"label": options.old_label or Path(options.old).stem, "path": os.path.basename(options.old)}, + "new": {"label": options.new_label or Path(options.new).stem, "path": os.path.basename(options.new)}, + "steps": steps, + "errors": [step["name"] for step in steps if step["status"] == "error"], + "manifest": "bundle-manifest.json", + } + write_json(os.path.join(out, "release-report.json"), report) + manifest = build_manifest(out) + write_json(os.path.join(out, "bundle-manifest.json"), manifest) + return report + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="ci_release", description=__doc__.splitlines()[0]) + parser.add_argument("--old", required=True, help="old target (RISC-V ELF or raw .fwimage)") + parser.add_argument("--new", required=True, help="new target (RISC-V ELF or raw .fwimage)") + parser.add_argument("--out", help="evidence bundle directory (required unless --dry-run)") + parser.add_argument("--dry-run", action="store_true", help="print the plan and exit without touching targets") + parser.add_argument("--old-label", help="report label for the old target") + parser.add_argument("--new-label", help="report label for the new target") + parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23") + parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})") + parser.add_argument("--old-symbols", help="green-eye symbol DB JSON for the old target") + parser.add_argument("--new-symbols", help="green-eye symbol DB JSON for the new target") + parser.add_argument("--matrix-symbols", help="one symbol DB JSON applied to both matrix targets") + parser.add_argument("--old-ghidra-names", help="Ghidra names.tsv for the old target") + parser.add_argument("--new-ghidra-names", help="Ghidra names.tsv for the new target") + parser.add_argument("--old-ghidra-index", help="Ghidra decompiled_index.tsv for the old target") + parser.add_argument("--new-ghidra-index", help="Ghidra decompiled_index.tsv for the new target") + parser.add_argument("--old-ghidra-exports", help="Ghidra exports.tsv / 3-column export for the old target") + parser.add_argument("--new-ghidra-exports", help="Ghidra exports.tsv / 3-column export for the new target") + parser.add_argument("--old-ghidra-bias", type=lambda value: int(value, 0), default=0) + parser.add_argument("--new-ghidra-bias", type=lambda value: int(value, 0), default=0) + parser.add_argument("--corpus-diff", nargs=2, metavar=("OLD_TSV", "NEW_TSV"), + help="diff two decompiled_index.tsv corpus exports") + parser.add_argument("--corpus-report", help="attach a previously produced corpus diff report") + parser.add_argument("--stimulus", help="green-eye script replacing the builtin demo scenario") + parser.add_argument("--probe-instructions", type=int, default=2000) + parser.add_argument("--behavior-duration", type=float, default=0.4) + parser.add_argument("--max-rows", type=int, default=40) + parser.add_argument("--mutation-kernel", default=DEMO_ELF, help="kernel for the mutation smoke (default: demo)") + parser.add_argument("--mutation-iterations", type=int, default=8) + parser.add_argument("--mutation-seed", type=int, default=7) + parser.add_argument("--mutation-timeout", type=float, default=0.25) + parser.add_argument("--mutation-settle", type=float, default=0.1) + parser.add_argument("--allow-foreign-mutation", action="store_true", + help="permit a non-demo mutation kernel (QEMU only)") + return parser + + +def options_from_args(args: argparse.Namespace) -> ReleaseOptions: + return ReleaseOptions( + old=args.old, + new=args.new, + out=args.out, + old_label=args.old_label, + new_label=args.new_label, + release=args.release, + approval_token=args.approval_token, + old_symbols=args.old_symbols, + new_symbols=args.new_symbols, + matrix_symbols=args.matrix_symbols, + old_ghidra_names=args.old_ghidra_names, + new_ghidra_names=args.new_ghidra_names, + old_ghidra_index=args.old_ghidra_index, + new_ghidra_index=args.new_ghidra_index, + old_ghidra_exports=args.old_ghidra_exports, + new_ghidra_exports=args.new_ghidra_exports, + old_ghidra_bias=args.old_ghidra_bias, + new_ghidra_bias=args.new_ghidra_bias, + corpus_diff=tuple(args.corpus_diff) if args.corpus_diff else None, # type: ignore[arg-type] + corpus_report=args.corpus_report, + stimulus=args.stimulus, + probe_instructions=args.probe_instructions, + behavior_duration=args.behavior_duration, + max_rows=args.max_rows, + mutation_kernel=args.mutation_kernel, + mutation_iterations=args.mutation_iterations, + mutation_seed=args.mutation_seed, + mutation_timeout=args.mutation_timeout, + mutation_settle=args.mutation_settle, + allow_foreign_mutation=args.allow_foreign_mutation, + dry_run=args.dry_run, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + options = options_from_args(args) + try: + check_rm610_gate(options.labels(), approval_token=options.approval_token) + except Rm610GateError as exc: + print(f"[release] RM610 gate refused: {exc}", file=sys.stderr) + return 3 + if options.dry_run: + plan = build_plan(options) + print(json.dumps(plan, indent=2, sort_keys=True)) + print(f"[release] dry-run: {len(plan['steps'])} steps planned, no targets touched") + return 0 + if not options.out: + parser.error("--out is required unless --dry-run is used") + try: + report = run_release(options) + except ValueError as exc: + parser.error(str(exc)) + except Exception as exc: + print(f"[release] error: {type(exc).__name__}: {exc}", file=sys.stderr) + return 2 + for step in report["steps"]: + reason = f" reason={step['reason']}" if step.get("reason") else "" + print(f"[release] {step['name']}: status={step['status']}{reason}") + print(f"[release] wrote {os.path.join(os.path.abspath(options.out), 'release-report.json')}") + print(f"[release] wrote {os.path.join(os.path.abspath(options.out), 'bundle-manifest.json')}") + return 1 if report["errors"] else 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/common.py b/debug/risc-v-gsp-rm-vm/harness/common.py new file mode 100644 index 0000000..74dabe0 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/common.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +"""Shared helpers for the green-eye automation harness. + +The harness scripts run from the tool root without installation, so this +module bootstraps the tool root onto ``sys.path`` before importing greeneye. +The QEMU backend is the only execution backend the harness claims: the matrix +must stay comparable to the CLI ``probe``/``demo`` paths and must not depend +on the in-development native simulator. +""" +from __future__ import annotations + +import contextlib +import hashlib +import json +import os +import shutil +import sys +import tempfile +from dataclasses import dataclass, field + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.analyze import parse_qemu_faults # noqa: E402 +from greeneye.elf import FirmwareImage, load_elf, write_flat_kernel # noqa: E402 +from greeneye.errors import GreenEyeError, TargetError # noqa: E402 +from greeneye.rsp import RspClient # noqa: E402 +from greeneye.session import Session # noqa: E402 +from greeneye.symbols import SymbolDB # noqa: E402 +from greeneye.target import QemuTarget, find_qemu # noqa: E402 + +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +DEMO_SCHEMA = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json") + +#: Raw images are relocated to the top of QEMU's virt RAM, matching `greeneye probe`. +LOAD_ADDR = 0x80000000 +#: Refuse images whose flat form would be unreasonably large. +MAX_FLAT_BYTES = 256 * 1024 * 1024 +#: Cap the QEMU memory size the harness will request. +MAX_MEMORY_MB = 2048 +#: Default bound when reading a QEMU log window (trap loops flood `-d int`). +MAX_LOG_BYTES = 1 << 20 + + +def sha256_file(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() + + +def qemu_available() -> bool: + try: + find_qemu() + return True + except TargetError: + return False + + +def qemu_path() -> str: + return find_qemu() + + +def load_target(path: str) -> FirmwareImage: + if not os.path.isfile(path): + raise GreenEyeError(f"target not found: {path}") + image = load_elf(path) + if not image.is_riscv: + raise GreenEyeError(f"{path}: not a RISC-V ELF image") + span = image.max_vaddr - image.min_vaddr + if span > MAX_FLAT_BYTES: + raise GreenEyeError(f"{path}: image span {span:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap") + return image + + +def memory_mb_for(image: FirmwareImage) -> int: + span = image.max_vaddr - image.min_vaddr + return min(MAX_MEMORY_MB, max(256, span // (1024 * 1024) + 64)) + + +def ensure_dir(path: str) -> str: + os.makedirs(path, exist_ok=True) + return path + + +def write_json(path: str, payload: dict) -> str: + ensure_dir(os.path.dirname(path)) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +def write_text(path: str, text: str) -> str: + ensure_dir(os.path.dirname(path)) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def relpath(path: str, root: str) -> str: + return os.path.relpath(path, root) + + +def counter_delta(a: dict, b: dict) -> list[dict]: + """Stable union delta of two count maps (B minus A), sorted by key.""" + rows = [] + for key in sorted(set(a) | set(b)): + a_count = int(a.get(key, 0)) + b_count = int(b.get(key, 0)) + rows.append({"key": key, "a": a_count, "b": b_count, "delta": b_count - a_count}) + return rows + + +def read_log_window(path: str | None, offset: int = 0, max_bytes: int = MAX_LOG_BYTES) -> tuple[str, bool]: + """Read a bounded QEMU log window; returns ``(text, capped)``. + + Trap loops emit a ``-d int`` line per faulting instruction, so an + unbounded read can be hundreds of megabytes. Faults of interest (the + first exception) are at the start of the window. + """ + if not path: + return "", False + try: + with open(path, "rb") as fh: + fh.seek(offset) + data = fh.read(max_bytes) + except (FileNotFoundError, OSError): + return "", False + capped = len(data) == max_bytes + text = data.decode("utf-8", "replace") + if capped: + text = text.rsplit("\n", 1)[0] + return text, capped + + +def read_faults_window(path: str | None, offset: int = 0, max_bytes: int = MAX_LOG_BYTES) -> tuple[list[dict], bool]: + text, capped = read_log_window(path, offset=offset, max_bytes=max_bytes) + return parse_qemu_faults(text), capped + + +@dataclass +class FlatSession: + """A live QEMU session around a relocated flat kernel.""" + + image: FirmwareImage + bias: int + session: Session + target: QemuTarget + workdir: str + flat_kernel: str + owns_workdir: bool = field(default=False) + + +@contextlib.contextmanager +def open_flat_session( + image: FirmwareImage, + target_path: str, + *, + record: str | bool | None, + name: str, + symbols: SymbolDB | None = None, + log_faults: bool = True, + workdir: str | None = None, + meta: dict | None = None, +): + """Start QEMU on the target's flat relocated image and yield a session. + + Relocation (``LOAD_ADDR - image.min_vaddr``) mirrors ``greeneye probe``: + firmware linked for GPU physical addresses has to be placed into virt RAM + before QEMU can execute it. Symbols are loaded at the relocated addresses. + """ + owns_workdir = workdir is None + if workdir is None: + workdir = tempfile.mkdtemp(prefix="greeneye-harness-") + flat_kernel = os.path.join(workdir, "flat.bin") + write_flat_kernel(target_path, flat_kernel, load_addr=LOAD_ADDR) + bias = LOAD_ADDR - image.min_vaddr + target = QemuTarget( + kernel=flat_kernel, + memory_mb=memory_mb_for(image), + dlog="guest_errors,unimp,int" if log_faults else None, + workdir=workdir, + ) + target.start(paused=True) + client = RspClient(target.host, target.port) + client.connect() + client.sync() + db = symbols if symbols is not None else SymbolDB() + session_meta = {"backend": "qemu", "bias": bias, "firmware": target_path, "harness": name} + session_meta.update(meta or {}) + session = Session( + client=client, + target=target, + symbols=db, + record=record, + name=name, + meta=session_meta, + ) + session.load_image(image, bias=bias, load_symbols=symbols is None) + try: + yield FlatSession( + image=image, + bias=bias, + session=session, + target=target, + workdir=workdir, + flat_kernel=flat_kernel, + owns_workdir=owns_workdir, + ) + finally: + try: + session.close() + finally: + if owns_workdir: + shutil.rmtree(workdir, ignore_errors=True) diff --git a/debug/risc-v-gsp-rm-vm/harness/control_discovery.py b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py new file mode 100644 index 0000000..eecf082 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py @@ -0,0 +1,1055 @@ +#!/usr/bin/env python3 +"""Unknown-call discovery over firmware control surfaces (H4, "Sandsifter for RM"). + +The problem this tool attacks: NVIDIA ships new GSP images on a hidden +schedule and new RM messages/controls/services appear without documentation. +Instead of waiting for a new control to be noticed, sweep the input space and +ask the firmware itself which calls exist, which code path each one takes, and +what it changes. + +Each probe uses a fresh QEMU session, settles the firmware through its +self-test, hooks the command dispatcher (and every handler the control block +declares), applies exactly one command with a deterministic argument pattern, +runs a bounded window and classifies the outcome: + + known-handler the documented/id-table handler for this id ran + injected-slot the id routed through the injected-service handler table + default-path the dispatcher's catch-all ran (undiscovered id) + trap the id reached an ``ebreak`` (debugger trap) + fault the id produced a load/store/fetch/illegal exception + no-progress the mailbox was never consumed within the budget + +Findings are written to ``findings.jsonl`` and a markdown report groups ids by +class, listing unknown/undiscovered ids explicitly. + +Surfaces (``--surface``): + +* ``demo`` -- the bundled uproc mailbox command space (works today on QEMU); +* ``rm`` -- an RM control/method entry point for the future simulator or a + live session. It is implemented behind the same :class:`Surface` + abstraction (a probe function + an effect observer), so a real RM surface is + a new class, not a rewrite. When no entry point is exposed the tool prints + a documented message and exits cleanly. + +Safety: the bundled demo (or an explicitly allowed foreign ELF) under QEMU, +fresh session per probe, bounded budgets, host writes limited to ``--out``. +Never hardware, a guest, an existing gdbstub or the native simulator by +default. See docs/19-CONTROL-DISCOVERY.md. + +Run from the tool root: + + python3 harness/control_discovery.py --surface demo --out dumps/discovery +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import random +import shutil +import struct +import sys +import time +from collections import Counter +from dataclasses import dataclass, field + +try: + from .common import ( + DEMO_ELF, + MAX_LOG_BYTES, + ensure_dir, + load_target, + qemu_available, + read_faults_window, + sha256_file, + ) +except ImportError: # direct script execution (python3 harness/control_discovery.py) + from common import ( # type: ignore[no-redef] + DEMO_ELF, + MAX_LOG_BYTES, + ensure_dir, + load_target, + qemu_available, + read_faults_window, + sha256_file, + ) + +from greeneye import launch # noqa: E402 +from greeneye.errors import GreenEyeError, RspTimeout # noqa: E402 + +FORMAT = "greeneye-control-discovery/1" +MASK64 = (1 << 64) - 1 + +#: gsp_ctrl field offsets (firmware/uproc_demo/gsp_ctrl.schema.json). +CTRL_MAGIC = 0 +CTRL_STATE = 8 +CTRL_RPC_COUNT = 16 +CTRL_LAST_RPC = 24 +CTRL_MAILBOX = 32 +CTRL_ARG0 = 40 +CTRL_ARG1 = 48 +CTRL_ARG2 = 56 +CTRL_RESULT = 64 +CTRL_BUF = 72 +CTRL_BUF_SIZE = 256 +CTRL_HANDLER_COUNT = 328 +CTRL_HANDLERS = 336 +CTRL_SIZE = 400 + +#: The demo dispatcher's documented switch cases. +DEMO_CASE_HANDLERS = { + 1: "rpc_ping", + 2: "rpc_echo", + 3: "rpc_fill", + 4: "rpc_sum", + 5: "rpc_state", + 6: "rpc_fault", +} +#: Mailbox commands 16..23 select handler-table slots 0..7 (docs/12 section 4). +INJECT_CMD_BASE = 16 +INJECT_CMD_END = 24 + +SETTLE_MARKER = "SELFTEST DONE" +SETTLE_SLICE = 0.02 +RUN_SLICE = 0.02 +TRAP_GRACE = 0.05 +#: gdb signals that carry a hazard (illegal, trap, bus, segv); SIGINT from the +#: budget interrupt is wall-clock dependent and must never enter a finding. +HAZARD_SIGNALS = (4, 5, 7, 11) + +PATTERNS = ("zero", "sentinel", "boundary") +OUTCOMES = ( + "known-handler", + "injected-slot", + "method-return", + "default-path", + "trap", + "fault", + "no-progress", + "settle-failed", + "error", +) +DEFAULT_RECORD_CLASSES = ("injected-slot", "method-return", "trap", "fault", "default-path", "settle-failed", "error") + + +# --------------------------------------------------------------------------- +# deterministic probe generation +# --------------------------------------------------------------------------- + + +@dataclass +class Probe: + index: int + cmd: int + pattern: str + args: tuple[int, int, int] + + @property + def name(self) -> str: + return f"{self.index:04d}-id{self.cmd:03d}-{self.pattern}" + + +def pattern_args(pattern: str, seed: int, cmd: int) -> tuple[int, int, int]: + """Deterministic argument vector per (pattern, seed, cmd). + + The per-command RNG stream keeps sentinels distinct across ids while a + fixed seed makes the whole sweep replayable. + """ + if pattern == "zero": + return (0, 0, 0) + rng = random.Random(seed * 1_000_003 + cmd * 97 + 0x5EED) + if pattern == "sentinel": + return tuple(rng.getrandbits(64) for _ in range(3)) # type: ignore[return-value] + if pattern == "boundary": + return (MASK64, 1, 0x7FFF_FFFF_FFFF_FFFF) + raise GreenEyeError(f"unknown pattern {pattern!r} (expected one of {', '.join(PATTERNS)})") + + +def generate_probes(max_id: int, patterns: list[str], seed: int, iterations: int = 0) -> list[Probe]: + probes: list[Probe] = [] + for cmd in range(1, max_id + 1): + for pattern in patterns: + probes.append(Probe(index=len(probes), cmd=cmd, pattern=pattern, args=pattern_args(pattern, seed, cmd))) + if iterations > 0: + probes = probes[:iterations] + return probes + + +# --------------------------------------------------------------------------- +# effect observer +# --------------------------------------------------------------------------- + + +@dataclass +class WatchedField: + name: str + offset: int + size: int + fmt: str # u8 | u32 | u64 | bytes + + +@dataclass +class EffectObserver: + """Captures the effect half of a surface: field snapshots + store PCs.""" + + watched: tuple[WatchedField, ...] = () + hits: list[dict] = field(default_factory=list) + + def reset(self) -> None: + self.hits.clear() + + def arm(self, session, base: int) -> None: + """Install write watchpoints for every watched field.""" + for watched in self.watched: + if watched.fmt == "bytes": + continue # byte-range watchpoints are not portable; hash at snapshot time + session.watch(base + watched.offset, size=watched.size, kind="w", callback=self._make_callback(watched)) + + def _make_callback(self, watched: WatchedField): + def callback(hit) -> None: + self.hits.append( + { + "field": watched.name, + "pc": hit.pc, + "symbol": hit.symbol.name if hit.symbol else None, + "addr": hit.addr, + } + ) + + return callback + + def snapshot(self, session, base: int) -> dict: + values: dict = {} + for watched in self.watched: + raw = session.memory.read(base + watched.offset, watched.size) + if watched.fmt == "bytes": + values[watched.name] = { + "sha256": hashlib.sha256(raw).hexdigest()[:16], + "len": len(raw), + "preview": raw[:16].hex(), + } + elif watched.fmt == "u8": + values[watched.name] = raw[0] + elif watched.fmt == "u32": + values[watched.name] = struct.unpack_from("<I", raw, 0)[0] + else: + values[watched.name] = struct.unpack_from("<Q", raw, 0)[0] + return values + + def delta(self, before: dict, after: dict) -> dict: + changed = [ + {"field": name, "before": before.get(name), "after": after.get(name)} + for name in before + if before.get(name) != after.get(name) + ] + return {"changed": changed, "watch": list(self.hits)} + + +# --------------------------------------------------------------------------- +# surfaces +# --------------------------------------------------------------------------- + + +class Surface: + """A sweepable input surface: command space + probe action + effects. + + A future surface is a subclass; the sweep, classification and reporting + machinery is shared. ``probe`` applies one command, ``observer`` returns + the effect model, and ``handler_map``/``command_for_addr`` model the + entry points so ids can be attributed to code. + """ + + name = "abstract" + + def describe(self) -> str: + return self.name + + def available(self) -> tuple[bool, str | None]: + return True, None + + def documented_handlers(self, session) -> dict[int, str]: + return {} + + def handler_map(self, session) -> dict[int, tuple[int, str | None]]: + """Map command id -> (entry address, symbol name) for declared handlers.""" + return {} + + def command_hook(self, session) -> str | None: + return None + + def is_injected(self, cmd: int) -> bool: + return False + + def probe(self, session, cmd: int, args: tuple[int, int, int], *, timeout: float) -> int | None: + """Apply one command. Returns a direct return value when the surface is + a call entry point, else None (effects carry the reply).""" + raise NotImplementedError + + def observer(self) -> EffectObserver: + return EffectObserver() + + def command_for_addr(self, session, addr: int) -> list[dict]: + """Ids whose declared handler resolves to ``addr`` (trigger discovery).""" + out: list[dict] = [] + for cmd, (entry, name) in sorted(self.handler_map(session).items()): + if entry == addr: + out.append({"id": cmd, "handler": name}) + return out + + +class DemoSurface(Surface): + """The uproc demo mailbox surface (gsp_ctrl.handlers + 1..6 switch).""" + + name = "demo" + + def documented_handlers(self, session) -> dict[int, str]: + return { + cmd: name for cmd, name in DEMO_CASE_HANDLERS.items() if session.symbols.get(name) is not None + } + + def handler_map(self, session) -> dict[int, tuple[int, str | None]]: + mapping: dict[int, tuple[int, str | None]] = {} + for cmd, name in self.documented_handlers(session).items(): + mapping[cmd] = (session.symbols.lookup(name), name) + ctrl = session.symbols.get("gsp_ctrl") + if ctrl is not None: + for slot in range(8): + entry = session.memory.read_u64(ctrl.addr + CTRL_HANDLERS + slot * 8) + if entry: + symbol = session.symbols.resolve(entry) + mapping[INJECT_CMD_BASE + slot] = (entry, symbol.name if symbol else None) + return mapping + + def command_hook(self, session) -> str | None: + return "rpc_dispatch" if session.symbols.get("rpc_dispatch") is not None else None + + def is_injected(self, cmd: int) -> bool: + return INJECT_CMD_BASE <= cmd < INJECT_CMD_END + + def probe(self, session, cmd: int, args: tuple[int, int, int], *, timeout: float) -> int | None: + ctrl = session.symbols.lookup("gsp_ctrl") + # Args first: the firmware polls the mailbox last and consumes all four. + session.memory.write_u64(ctrl + CTRL_ARG0, args[0]) + session.memory.write_u64(ctrl + CTRL_ARG1, args[1]) + session.memory.write_u64(ctrl + CTRL_ARG2, args[2]) + session.memory.write_u64(ctrl + CTRL_MAILBOX, cmd) + return None + + def observer(self) -> EffectObserver: + return EffectObserver( + watched=( + WatchedField("state", CTRL_STATE, 4, "u32"), + WatchedField("rpc_count", CTRL_RPC_COUNT, 8, "u64"), + WatchedField("last_rpc", CTRL_LAST_RPC, 8, "u64"), + WatchedField("result", CTRL_RESULT, 8, "u64"), + WatchedField("buf", CTRL_BUF, CTRL_BUF_SIZE, "bytes"), + ) + ) + + def snapshot_extra(self, session, base: int) -> dict: + raw = session.memory.read(base, CTRL_SIZE) + return { + "magic": struct.unpack_from("<I", raw, CTRL_MAGIC)[0], + "handler_count": struct.unpack_from("<Q", raw, CTRL_HANDLER_COUNT)[0], + } + + +class RMSurface(Surface): + """Future RM control/method entry point (roadmap H1.3/H4). + + On a simulator or live session that exposes an RM method dispatcher, the + surface is: call the method entry with ``(control, a0, a1, a2)`` and read + the reply buffer at ``reply_addr``. Today no such entry is exposed, so + ``available()`` reports why and the CLI exits cleanly. + """ + + name = "rm" + + def __init__(self, entry: str | None = None, reply_addr: int | None = None, reply_size: int = 0x100): + self.entry = entry + self.reply_addr = reply_addr + self.reply_size = reply_size + + def describe(self) -> str: + bits = [f"entry={self.entry or '<none>'}"] + bits.append(f"reply={self.reply_addr:#x}/{self.reply_size:#x}" if self.reply_addr else "reply=<none>") + return "rm(" + ", ".join(bits) + ")" + + def available(self) -> tuple[bool, str | None]: + if not self.entry: + return False, ( + "surface 'rm' unavailable: no RM control/method entry point is exposed by the " + "simulator or a live session yet (roadmap H1.3). Pass --rm-entry SYMBOL once the " + "RM method dispatcher exists; the demo surface works today with --surface demo." + ) + return True, None + + def is_injected(self, cmd: int) -> bool: + return False + + def probe(self, session, cmd: int, args: tuple[int, int, int], *, timeout: float) -> int | None: + return session.call(self.entry, (cmd, args[0], args[1], args[2]), timeout=timeout) + + def observer(self) -> EffectObserver: + if not self.reply_addr: + return EffectObserver() + return EffectObserver( + watched=(WatchedField("reply", self.reply_addr, self.reply_size, "bytes"),) + ) + + def snapshot_extra(self, session, base: int) -> dict: + return {} + + +def make_surface(args) -> Surface: + if args.surface == "demo": + return DemoSurface() + if args.surface == "rm": + return RMSurface(entry=args.rm_entry, reply_addr=args.rm_reply, reply_size=args.rm_reply_size) + raise GreenEyeError(f"unknown surface {args.surface!r}") + + +# --------------------------------------------------------------------------- +# one probe +# --------------------------------------------------------------------------- + + +@dataclass +class Finding: + probe: int + cmd: int + pattern: str + args: list[int] + outcome: str = "error" + handler: str | None = None + handler_entry: int | None = None + documented_handler: str | None = None + dispatch_seen: bool = False + dispatch_pc: int | None = None + dispatch_args: list[int] | None = None + call_retval: int | None = None + rpc_count_delta: int = 0 + effects: dict = field(default_factory=dict) + fault: dict | None = None + signals: list[dict] = field(default_factory=list) + settle_ok: bool = True + session: str | None = None + reason: str | None = None + + def to_json(self) -> dict: + return { + "probe": self.probe, + "cmd": self.cmd, + "pattern": self.pattern, + "args": self.args, + "outcome": self.outcome, + "handler": self.handler, + "handler_entry": self.handler_entry, + "documented_handler": self.documented_handler, + "dispatch_seen": self.dispatch_seen, + "dispatch_pc": self.dispatch_pc, + "dispatch_args": self.dispatch_args, + "call_retval": self.call_retval, + "rpc_count_delta": self.rpc_count_delta, + "effects": self.effects, + "fault": self.fault, + "signals": self.signals, + "settle_ok": self.settle_ok, + "session": self.session, + "reason": self.reason, + } + + +def _compact_fault(fault: dict) -> dict: + return { + "cause": fault.get("cause"), + "desc": fault.get("desc"), + "epc": fault.get("epc"), + "tval": fault.get("tval"), + "access": fault.get("access"), + } + + +def classify_outcome( + *, + cmd: int, + is_injected: bool, + handler: str | None, + dispatch_seen: bool, + call_retval: int | None, + faults: list[dict], + signals: list[dict], + settle_ok: bool, +) -> str: + """Priority-ordered classification of one probe result (pure function).""" + if not settle_ok: + return "settle-failed" + causes = [int(fault.get("cause", 0)) for fault in faults] + if any(cause == 3 for cause in causes) or any(sig.get("signal") == 5 for sig in signals): + return "trap" + if causes or any(sig.get("signal") in (4, 7, 11) for sig in signals): + return "fault" + if call_retval is not None: + return "method-return" + if not dispatch_seen: + return "no-progress" + if is_injected and handler: + return "injected-slot" + if handler: + return "known-handler" + return "default-path" + + +class ProbeRunner: + """Runs one probe in a fresh session and builds the finding.""" + + def __init__(self, surface: Surface, options: "DiscoveryOptions", image, kernel_sha: str): + self.surface = surface + self.options = options + self.image = image + self.kernel_sha = kernel_sha + + def run(self, probe: Probe) -> Finding: + finding = Finding(probe=probe.index, cmd=probe.cmd, pattern=probe.pattern, args=list(probe.args)) + staging = os.path.join(self.options.out, ".staging", probe.name) + ensure_dir(staging) + qemu_log = os.path.join(staging, "qemu-faults.log") + # Record into staging unconditionally: the classification decides whether + # the session is evidence worth keeping, and re-running a probe to + # record it later would double the sweep cost. + record_path = staging + session = None + try: + extra_args = None + if self.options.backend == "qemu": + extra_args = ["-d", "guest_errors,unimp,int", "-D", qemu_log] + session = launch( + kernel=self.options.kernel, + record=record_path, + name=f"discovery-{probe.name}", + backend=self.options.backend, + extra_args=extra_args, + ) + finding.settle_ok = self._settle(session, self.options.settle) + if not finding.settle_ok: + finding.reason = f"{SETTLE_MARKER!r} not observed within the settle budget" + return self._finish(session, probe, finding) + + observer = self.surface.observer() + before = observer.snapshot(session, self._base(session)) + before.update(self._extra(self.surface, session)) + observer.reset() + observer.arm(session, self._base(session)) + + dispatches: list[dict] = [] + handlers: list[dict] = [] + signals: list[dict] = [] + self._install_hooks(session, dispatches, handlers, signals) + observer.reset() + + log_offset = os.path.getsize(qemu_log) if os.path.exists(qemu_log) else 0 + call_retval: int | None = None + try: + call_retval = self.surface.probe(session, probe.cmd, probe.args, timeout=self.options.timeout) + except RspTimeout as exc: + finding.reason = f"probe call timed out: {exc}" + self._run_window(session, dispatches, handlers, signals, observer) + + faults: list[dict] = [] + if self.options.backend == "qemu": + faults, _capped = read_faults_window(qemu_log, log_offset, self.options.max_log_bytes) + + after = observer.snapshot(session, self._base(session)) + after.update(self._extra(self.surface, session)) + effects = observer.delta(before, after) + effects["before"] = before + effects["after"] = after + + dispatch = dispatches[0] if dispatches else None + handler = handlers[0] if handlers else None + mapped = self.surface.handler_map(session) + finding.documented_handler = self.surface.documented_handlers(session).get(probe.cmd) + finding.dispatch_seen = dispatch is not None + finding.dispatch_pc = dispatch.get("addr") if dispatch else None + finding.dispatch_args = dispatch.get("args") if dispatch else None + finding.handler = handler.get("symbol") if handler else None + finding.handler_entry = handler.get("addr") if handler else None + if handler is None and probe.cmd in mapped: + entry, name = mapped[probe.cmd] + finding.handler = name + finding.handler_entry = entry + finding.call_retval = call_retval + finding.rpc_count_delta = int(effects.get("after", {}).get("rpc_count", 0)) - int( + effects.get("before", {}).get("rpc_count", 0) + ) + finding.effects = effects + finding.fault = _compact_fault(faults[0]) if faults else None + finding.signals = signals + finding.outcome = classify_outcome( + cmd=probe.cmd, + is_injected=self.surface.is_injected(probe.cmd), + handler=finding.handler, + dispatch_seen=dispatch is not None, + call_retval=call_retval, + faults=faults, + signals=signals, + settle_ok=True, + ) + return self._finish(session, probe, finding) + except Exception as exc: # one bad probe must not end the sweep + finding.outcome = "error" + finding.reason = f"{type(exc).__name__}: {exc}" + return self._finish(session, probe, finding) + + # -- helpers ----------------------------------------------------------- + + def _base(self, session) -> int: + return session.symbols.lookup("gsp_ctrl") if session.symbols.get("gsp_ctrl") else 0 + + def _extra(self, surface: Surface, session) -> dict: + extra = getattr(surface, "snapshot_extra", None) + if extra is None: + return {} + return extra(session, self._base(session)) + + def _install_hooks(self, session, dispatches: list[dict], handlers: list[dict], signals: list[dict]) -> None: + def on_signal(hit) -> None: + stop = hit.stop + if stop is not None and stop.signal in HAZARD_SIGNALS: + signals.append({"signal": stop.signal, "pc": hit.pc}) + + session.on_event("stop", on_signal) + hook_name = self.surface.command_hook(session) + if hook_name: + session.hook( + hook_name, + on_enter=lambda hit: dispatches.append( + { + "addr": hit.addr, + "symbol": hook_name, + "args": [hit.arg(0), hit.arg(1), hit.arg(2), hit.arg(3)], + "caller": hit.regs.get("ra", 0), + } + ), + listener=f"discovery:{hook_name}", + ) + for cmd, (entry, name) in sorted(self.surface.handler_map(session).items()): + session.hook( + entry, + on_enter=lambda hit, name=name: handlers.append( + { + "addr": hit.addr, + "symbol": hit.symbol.name if hit.symbol else name, + "args": [hit.arg(0), hit.arg(1), hit.arg(2)], + "caller": hit.regs.get("ra", 0), + } + ), + listener=f"discovery:handler-{entry:#x}", + ) + + def _settle(self, session, budget: float) -> bool: + deadline = time.monotonic() + budget + while time.monotonic() < deadline: + if SETTLE_MARKER in session.target.read_uart(): + return True + session.run(duration=SETTLE_SLICE, max_stops=1) + return SETTLE_MARKER in session.target.read_uart() + + def _run_window(self, session, dispatches, handlers, signals, observer) -> None: + """Two-phase bounded window: wait for dispatch, then for completion. + + Completion is a ``result`` store (the dispatcher writes it after every + handler) or a trap/grace for ids that never return to the dispatcher. + """ + deadline = time.monotonic() + self.options.timeout + handler_seen_at: float | None = None + while time.monotonic() < deadline: + if not session.target.alive(): + return + remaining = max(0.005, min(RUN_SLICE, deadline - time.monotonic())) + session.run(duration=remaining, max_stops=64) + if handlers and handler_seen_at is None: + handler_seen_at = time.monotonic() + result_written = any(hit.get("field") == "result" for hit in observer.hits) + trapped = any(sig.get("signal") == 5 for sig in signals) or session.pc < 0x100 + if dispatches and (result_written or trapped): + return + if handler_seen_at is not None and time.monotonic() - handler_seen_at >= TRAP_GRACE: + # Slow/hung handler: one grace window after the entry, then stop. + return + + def _keep(self, outcome: str) -> bool: + return self.options.record_any or outcome in self.options.record_classes + + def _finish(self, session, probe: Probe, finding: Finding) -> Finding: + if session is not None: + try: + session.close() + except Exception: + pass + staging = os.path.join(self.options.out, ".staging", probe.name) + if self._keep(finding.outcome) and os.path.isdir(staging): + final = os.path.join(self.options.out, "sessions", probe.name) + ensure_dir(os.path.dirname(final)) + if os.path.isdir(final): + shutil.rmtree(final, ignore_errors=True) + os.replace(staging, final) + finding.session = os.path.join("sessions", probe.name) + else: + shutil.rmtree(staging, ignore_errors=True) + return finding + + +# --------------------------------------------------------------------------- +# report +# --------------------------------------------------------------------------- + + +def aggregate(findings: list[Finding]) -> dict: + per_id: dict[int, dict] = {} + for finding in findings: + row = per_id.setdefault( + finding.cmd, + { + "id": finding.cmd, + "patterns": [], + "outcomes": [], + "handlers": [], + "documented_handler": finding.documented_handler, + "results": [], + "rpc_count_delta": 0, + "sessions": [], + "faults": [], + }, + ) + row["patterns"].append(finding.pattern) + row["outcomes"].append(finding.outcome) + if finding.handler: + row["handlers"].append(finding.handler) + result = (finding.effects or {}).get("after", {}).get("result") + if result is not None and finding.dispatch_seen: + row["results"].append(result) + row["rpc_count_delta"] += finding.rpc_count_delta + if finding.session: + row["sessions"].append(finding.session) + if finding.fault: + row["faults"].append(finding.fault) + for row in per_id.values(): + row["outcomes"] = sorted(set(row["outcomes"])) + row["handlers"] = sorted(set(row["handlers"])) + row["results"] = sorted(set(row["results"])) + return dict(sorted(per_id.items())) + + +def primary_class(row: dict) -> str: + """One label per id for grouping: handle, injected, trap, fault, unknown.""" + outcomes = set(row["outcomes"]) + if "trap" in outcomes: + return "trap/fault" + if "fault" in outcomes: + return "trap/fault" + if "known-handler" in outcomes and row["documented_handler"]: + return "known-handler" + if "injected-slot" in outcomes: + return "injected-slot" + if "method-return" in outcomes: + return "method-return" + if outcomes & {"default-path", "no-progress"}: + return "unknown/undiscovered" + return "unknown/undiscovered" + + +def render_markdown(report: dict, options: "DiscoveryOptions") -> str: + per_id = report["ids"] + outcome_counts = Counter(finding["outcome"] for finding in report["findings"]) + grouped: dict[str, list[int]] = {} + for row in per_id.values(): + grouped.setdefault(primary_class(row), []).append(row["id"]) + lines = [ + "# green-eye control discovery report", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/control_discovery.py`.", + "", + f"* format: `{FORMAT}`", + f"* surface: `{report['surface']}`", + f"* kernel: `{report['kernel']}`", + f"* kernel sha256: `{report['kernel_sha256']}`", + f"* backend: `{report['backend']}`", + f"* ids swept: 1..{report['max_id']} ({report['probes_run']} probes), " + f"seed={report['seed']}, patterns={','.join(report['patterns'])}", + f"* per-probe timeout: {report['timeout']}s; settle budget: {report['settle']}s", + "", + "## Outcome counts", + "", + ] + lines += _table( + ["outcome", "count"], + [[outcome, outcome_counts[outcome]] for outcome in sorted(outcome_counts)], + ) + lines += ["", "## Ids by class", ""] + for klass in ("known-handler", "injected-slot", "method-return", "trap/fault", "unknown/undiscovered"): + ids = grouped.get(klass, []) + lines += [f"* **{klass}**: " + (", ".join(str(i) for i in ids) if ids else "_none_")] + lines += ["", "## Known handler ids", ""] + known = [row for row in per_id.values() if row["documented_handler"]] + lines += _render_rows(known) + lines += ["", "## Injected-slot dispatch ids", ""] + injected = [row for row in per_id.values() if "injected-slot" in row["outcomes"]] + lines += _render_rows(injected) + lines += ["", "## Unknown / undiscovered ids", ""] + unknown = [ + row + for row in per_id.values() + if not row["documented_handler"] and not (set(row["outcomes"]) & {"trap", "fault"}) + ] + if unknown: + lines += _render_rows(unknown) + lines += [ + "", + "These ids reached the dispatcher's catch-all (result ``~id``) and no handler " + "table entry; they are the candidate space for new/unmapped controls.", + ] + else: + lines += ["_none_"] + lines += ["", "## Traps and faults", ""] + hazards = [row for row in per_id.values() if set(row["outcomes"]) & {"trap", "fault"}] + lines += _render_rows(hazards) + lines += ["", "## All findings by id", ""] + lines += _render_rows(list(per_id.values())) + lines += ["", "## Reproduce", "", "```bash", report["reproduce"], "```", ""] + return "\n".join(lines) + "\n" + + +def _render_rows(rows: list[dict]) -> list[str]: + if not rows: + return ["_none_"] + return _table( + ["id", "documented handler", "observed handler", "outcomes", "results", "rpc delta", "sessions"], + [ + [ + row["id"], + row["documented_handler"] or "-", + ", ".join(row["handlers"]) or "-", + ", ".join(row["outcomes"]), + ", ".join(f"{value:#x}" for value in row["results"][:4]) or "-", + row["rpc_count_delta"], + ", ".join(f"`{path}`" for path in row["sessions"]) or "-", + ] + for row in rows + ], + ) + + +def _table(headers: list[str], rows: list[list]) -> list[str]: + lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] + for row in rows: + lines.append("| " + " | ".join(str(cell) for cell in row) + " |") + return lines + + +# --------------------------------------------------------------------------- +# orchestration +# --------------------------------------------------------------------------- + + +@dataclass +class DiscoveryOptions: + surface: str + out: str + kernel: str = DEMO_ELF + max_id: int = 63 + patterns: list[str] = field(default_factory=lambda: list(PATTERNS)) + seed: int = 1 + iterations: int = 0 + timeout: float = 0.25 + settle: float = 2.0 + record_all: bool = False + record_classes: tuple[str, ...] = DEFAULT_RECORD_CLASSES + backend: str = "qemu" + allow_foreign_elf: bool = False + max_log_bytes: int = MAX_LOG_BYTES + rm_entry: str | None = None + rm_reply: int | None = None + rm_reply_size: int = 0x100 + + @property + def record_any(self) -> bool: + return self.record_all + + +def run_discovery(options: DiscoveryOptions) -> dict: + if options.backend == "qemu" and not qemu_available(): + raise GreenEyeError("qemu-system-riscv64 not found; install qemu-system-misc or set GREENEYE_QEMU") + if not os.path.isfile(options.kernel): + raise GreenEyeError(f"kernel not found: {options.kernel}") + if not options.allow_foreign_elf and os.path.abspath(options.kernel) != os.path.abspath(DEMO_ELF): + raise GreenEyeError( + "control discovery is restricted to the bundled demo firmware; " + "pass --allow-foreign-elf to sweep another ELF under QEMU" + ) + image = load_target(options.kernel) + kernel_sha = sha256_file(options.kernel) + surface = make_surface(options) + ensure_dir(options.out) + probes = generate_probes(options.max_id, options.patterns, options.seed, options.iterations) + runner = ProbeRunner(surface, options, image, kernel_sha) + findings: list[Finding] = [] + findings_path = os.path.join(options.out, "findings.jsonl") + with open(findings_path, "w", encoding="utf-8") as fh: + for probe in probes: + finding = runner.run(probe) + findings.append(finding) + fh.write(json.dumps(finding.to_json(), sort_keys=True) + "\n") + fh.flush() + print( + f"[discovery] {finding.probe + 1}/{len(probes)} id={finding.cmd} {finding.pattern} " + f"-> {finding.outcome}" + (f" ({finding.handler})" if finding.handler else ""), + file=sys.stderr, + ) + shutil.rmtree(os.path.join(options.out, ".staging"), ignore_errors=True) + report = { + "format": FORMAT, + "tool": "harness/control_discovery.py", + "surface": surface.describe(), + "kernel": options.kernel, + "kernel_sha256": kernel_sha, + "backend": options.backend, + "max_id": options.max_id, + "patterns": options.patterns, + "seed": options.seed, + "timeout": options.timeout, + "settle": options.settle, + "probes_run": len(findings), + "ids": aggregate(findings), + "findings": [finding.to_json() for finding in findings], + "reproduce": reproduce_command(options), + } + write_json(os.path.join(options.out, "discovery-report.json"), report) + with open(os.path.join(options.out, "control-discovery.md"), "w", encoding="utf-8") as fh: + fh.write(render_markdown(report, options)) + return report + + +def reproduce_command(options: DiscoveryOptions) -> str: + foreign = " --allow-foreign-elf" if options.allow_foreign_elf else "" + record = " --record-all" if options.record_all else "" + rm = "" + if options.surface == "rm": + rm = f" --rm-entry {options.rm_entry}" if options.rm_entry else "" + if options.rm_reply: + rm += f" --rm-reply {options.rm_reply:#x}" + return ( + f"python3 harness/control_discovery.py --surface {options.surface} --kernel {options.kernel} " + f"--out {options.out} --max-id {options.max_id} --patterns {','.join(options.patterns)} " + f"--seed {options.seed} --timeout {options.timeout} --settle {options.settle}" + f"{record}{foreign}{rm}" + ) + + +def write_json(path: str, payload: dict) -> str: + ensure_dir(os.path.dirname(path)) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def parse_patterns(text: str) -> list[str]: + patterns = [part.strip() for part in text.split(",") if part.strip()] + unknown = [pattern for pattern in patterns if pattern not in PATTERNS] + if unknown: + raise argparse.ArgumentTypeError(f"unknown pattern(s) {unknown}; expected {', '.join(PATTERNS)}") + return patterns + + +def parse_record_classes(text: str) -> tuple[str, ...]: + classes = tuple(part.strip() for part in text.split(",") if part.strip()) + unknown = [klass for klass in classes if klass not in OUTCOMES] + if unknown: + raise argparse.ArgumentTypeError(f"unknown outcome class(es) {unknown}; expected {', '.join(OUTCOMES)}") + return classes + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="control_discovery", description=__doc__.splitlines()[0]) + parser.add_argument("--surface", choices=("demo", "rm"), default="demo", help="input surface to sweep") + parser.add_argument("--kernel", default=DEMO_ELF, help="demo/re-hosted ELF (default: bundled demo)") + parser.add_argument("--out", required=True, help="output directory for findings and reports") + parser.add_argument("--max-id", type=int, default=63, help="highest command id to sweep (0..4096)") + parser.add_argument("--patterns", type=parse_patterns, default=list(PATTERNS), help="comma list: zero,sentinel,boundary") + parser.add_argument("--seed", type=int, default=1, help="deterministic pattern seed") + parser.add_argument("--iterations", type=int, default=0, help="cap total probes (0 = all)") + parser.add_argument("--timeout", type=float, default=0.25, help="bounded observation window per probe (seconds)") + parser.add_argument("--settle", type=float, default=2.0, help="budget to observe the self-test before probing") + parser.add_argument("--record-all", action="store_true", help="record every probe (default: interesting classes only)") + parser.add_argument( + "--record-classes", + type=parse_record_classes, + default=DEFAULT_RECORD_CLASSES, + help="outcome classes whose sessions are kept", + ) + parser.add_argument("--backend", choices=("qemu", "native"), default="qemu", help="execution backend (qemu is the supported default)") + parser.add_argument("--allow-foreign-elf", action="store_true", help="permit a non-demo ELF under QEMU") + parser.add_argument("--max-log-bytes", type=int, default=MAX_LOG_BYTES, help="QEMU fault-log window per probe") + parser.add_argument("--rm-entry", help="RM surface: method/control entry symbol (future, H1.3)") + parser.add_argument("--rm-reply", type=lambda value: int(value, 0), help="RM surface: reply buffer address") + parser.add_argument("--rm-reply-size", type=lambda value: int(value, 0), default=0x100, help="RM surface: reply buffer size") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if not 1 <= args.max_id <= 4096: + parser.error("--max-id must be in 1..4096") + if args.iterations < 0: + parser.error("--iterations must be >= 0") + if not 0.01 <= args.timeout <= 30.0: + parser.error("--timeout must be in 0.01..30.0") + options = DiscoveryOptions( + surface=args.surface, + out=os.path.abspath(args.out), + kernel=os.path.abspath(args.kernel), + max_id=args.max_id, + patterns=args.patterns, + seed=args.seed, + iterations=args.iterations, + timeout=args.timeout, + settle=args.settle, + record_all=args.record_all, + record_classes=args.record_classes, + backend=args.backend, + allow_foreign_elf=args.allow_foreign_elf, + max_log_bytes=args.max_log_bytes, + rm_entry=args.rm_entry, + rm_reply=args.rm_reply, + rm_reply_size=args.rm_reply_size, + ) + surface = make_surface(args) + available, reason = surface.available() + if not available: + print(f"[discovery] skipped: {reason}") + return 0 + try: + report = run_discovery(options) + except GreenEyeError as exc: + print(f"control discovery error: {exc}", file=sys.stderr) + return 2 + counts = Counter(finding["outcome"] for finding in report["findings"]) + print(f"[discovery] {report['probes_run']} probe(s) on surface={report['surface']}, out={options.out}") + for outcome in sorted(counts): + print(f"[discovery] {outcome}: {counts[outcome]}") + print(f"[discovery] wrote {os.path.join(options.out, 'findings.jsonl')}") + print(f"[discovery] wrote {os.path.join(options.out, 'control-discovery.md')}") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/mutation_harness.py b/debug/risc-v-gsp-rm-vm/harness/mutation_harness.py new file mode 100644 index 0000000..7b558bd --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/mutation_harness.py @@ -0,0 +1,586 @@ +#!/usr/bin/env python3 +"""Bounded deterministic mutation of the demo firmware input surface (R5, H4.3). + +Each iteration starts a fresh QEMU session, settles the demo firmware to the +end of its self-test, mutates the mailbox command/args (and optionally a RAM +region), runs it for a bounded budget, and classifies the outcome: normal, +trap (ebreak / control transfer to an invalid PC), fault (first new QEMU +exception, parsed from the guest_errors/int log), no-progress, or +unexpected-result against a small demo oracle. Every attempt is recorded as a +green-eye session and a compact finding. + +Strategies: invalid command IDs, boundary values, single bit flips of valid +inputs, and buffer length boundaries for the ECHO/FILL/SUM commands. + +Safety: the harness launches its own QEMU on the bundled (or explicitly +allowed) demo/re-hosted ELF only. It never attaches to hardware, a guest or a +kernel module, and it never uses the native simulator or an existing gdbstub. + +Run from the tool root: + + python3 harness/mutation_harness.py --out dumps/mutation --iterations 64 --seed 7 +""" +from __future__ import annotations + +import argparse +import json +import os +import random +import shutil +import struct +import sys +import tempfile +from collections import Counter +from dataclasses import dataclass + +try: + from .common import ( + DEMO_ELF, + LOAD_ADDR, + MAX_LOG_BYTES, + ensure_dir, + load_target, + memory_mb_for, + open_flat_session, + qemu_available, + read_faults_window, + sha256_file, + ) +except ImportError: # direct script execution (python3 harness/mutation_harness.py) + from common import ( # type: ignore[no-redef] + DEMO_ELF, + LOAD_ADDR, + MAX_LOG_BYTES, + ensure_dir, + load_target, + memory_mb_for, + open_flat_session, + qemu_available, + read_faults_window, + sha256_file, + ) + +from greeneye.errors import GreenEyeError # noqa: E402 + +FORMAT = "greeneye-mutation-findings/1" +MASK64 = (1 << 64) - 1 +VALID_CMDS = (1, 2, 3, 4, 5, 6) +INVALID_CMDS = (0, 7, 8, 15, 24, 25, 31, 32, 63, 0x100, 0xFFFF, 0xFFFFFFFF, MASK64) +BOUNDARY_VALUES = ( + 0, + 1, + 2, + 3, + 4, + 7, + 8, + 16, + 32, + 63, + 64, + 65, + 128, + 0xFF, + 0x100, + 0x101, + 1024, + 0x7FFFFFFF, + 0x80000000, + 0x40000000, + 0xFFFFFFFF, + 0x8000000000000000, + MASK64, +) +BUFFER_LENGTHS = (0, 1, 2, 31, 32, 33, 255, 256, 257, 512, 4096, 0x10000, 0x10000000, MASK64) +STRATEGIES = ("cmd-invalid", "arg-boundary", "bit-flip", "buffer-length") + +CTRL_MAGIC = 0 +CTRL_VERSION = 4 +CTRL_STATE = 8 +CTRL_RPC_COUNT = 16 +CTRL_LAST_RPC = 24 +CTRL_MAILBOX = 32 +CTRL_ARG0 = 40 +CTRL_ARG1 = 48 +CTRL_ARG2 = 56 +CTRL_RESULT = 64 +CTRL_HANDLER_COUNT = 328 +CTRL_BUF_OFFSET = 72 # char buf[256] +CTRL_SIZE = 400 +SETTLE_MARKER = "SELFTEST DONE" +SETTLE_RETRIES = 3 + + +@dataclass +class Mutation: + iteration: int + strategy: str + cmd: int = 0 + arg0: int | str = 0 + arg1: int = 0 + arg2: int = 0 + region: dict | None = None + + def to_json(self) -> dict: + return { + "cmd": self.cmd, + "arg0": self.arg0, + "arg1": self.arg1, + "arg2": self.arg2, + "region": self.region, + } + + +@dataclass +class HarnessOptions: + kernel: str + out: str + iterations: int = 50 + seed: int = 1 + timeout: float = 0.25 + settle: float = 0.2 + phase_a: float = 0.05 + memory_region: tuple[int, int] | None = None + allow_foreign_elf: bool = False + max_log_bytes: int = MAX_LOG_BYTES + kernel_sha256: str = "" + is_demo: bool = False + + +# --------------------------------------------------------------------------- +# mutation generation +# --------------------------------------------------------------------------- + + +def generate_mutations(iterations: int, seed: int, region: tuple[int, int] | None = None) -> list[Mutation]: + """Deterministic strategy rotation; per-iteration RNG seeded by (seed, index).""" + strategies = STRATEGIES + (("mem-region",) if region else ()) + mutations: list[Mutation] = [] + for index in range(iterations): + # Integer seed: deterministic across Python versions/processes. + rng = random.Random(seed * 1_000_003 + index) + strategy = strategies[index % len(strategies)] + mutation = Mutation(iteration=index, strategy=strategy) + if strategy == "cmd-invalid": + mutation.cmd = rng.choice(INVALID_CMDS) + elif strategy == "arg-boundary": + mutation.cmd = rng.choice(VALID_CMDS) + args = [0, 0, 0] + args[rng.randrange(3)] = rng.choice(BOUNDARY_VALUES) + mutation.arg0, mutation.arg1, mutation.arg2 = args + elif strategy == "bit-flip": + values = { + "cmd": rng.choice(VALID_CMDS), + "arg0": rng.choice((0, 1, 0x41, 0xFF)), + "arg1": rng.choice((0, 1, 32)), + "arg2": 0, + } + target = rng.choice(("cmd", "arg0", "arg1", "arg2")) + values[target] = (values[target] ^ (1 << rng.randrange(64))) & MASK64 + mutation.cmd = values["cmd"] + mutation.arg0, mutation.arg1, mutation.arg2 = values["arg0"], values["arg1"], values["arg2"] + elif strategy == "buffer-length": + mutation.cmd = rng.choice((2, 3, 4)) + mutation.arg0 = "ctrl_buf" if mutation.cmd in (2, 4) else rng.choice((0, 0x41)) + mutation.arg1 = rng.choice(BUFFER_LENGTHS) + elif strategy == "mem-region": + addr, size = region # type: ignore[misc] + mutation.region = { + "addr": addr, + "size": size, + "offset": rng.randrange(size), + "xor": 1 << rng.randrange(8), + "old": None, + "new": None, + } + mutations.append(mutation) + return mutations + + +def resolve_arg0(mutation: Mutation, ctrl: int) -> int: + if mutation.arg0 == "ctrl_buf": + return ctrl + CTRL_BUF_OFFSET + return int(mutation.arg0) + + +# --------------------------------------------------------------------------- +# demo oracle (documented subset only; everything else is not judged) +# --------------------------------------------------------------------------- + + +def expected_for(mutation: Mutation, before: dict, ctrl: int) -> dict: + cmd = mutation.cmd + if cmd == 0: + return {"mode": "same"} + if cmd == 1: + return {"mode": "value", "value": (resolve_arg0(mutation, ctrl) ^ 0xBEEF) & MASK64} + if cmd == 3: + return {"mode": "value", "value": min(mutation.arg1, 256)} + if cmd == 5: + return {"mode": "one_of", "values": [0x10, 0x20, 0x30, 0x40]} + if 16 <= cmd <= 21: + return {"mode": None, "reason": "injected handler slot result is service-dependent"} + if cmd in (2, 4, 6): + return {"mode": None, "reason": "memory/trap-dependent handler"} + return {"mode": "value", "value": (~cmd) & MASK64} + + +def expected_ok(expectation: dict, mutation: Mutation, before: dict, after: dict) -> bool | None: + mode = expectation.get("mode") + if mode is None: + return None + if mode == "same": + return after["result"] == before["result"] and after["last_rpc"] == before["last_rpc"] + if mode == "value": + return after["result"] == expectation["value"] and after["last_rpc"] == mutation.cmd + if mode == "one_of": + return after["result"] in expectation["values"] and after["last_rpc"] == mutation.cmd + return None + + +# --------------------------------------------------------------------------- +# iteration execution +# --------------------------------------------------------------------------- + + +def read_ctrl_state(session, ctrl: int) -> dict: + raw = session.memory.read(ctrl, CTRL_SIZE) + u32 = lambda offset: struct.unpack_from("<I", raw, offset)[0] # noqa: E731 + u64 = lambda offset: struct.unpack_from("<Q", raw, offset)[0] # noqa: E731 + return { + "magic": u32(CTRL_MAGIC), + "version": u32(CTRL_VERSION), + "state": u32(CTRL_STATE), + "rpc_count": u64(CTRL_RPC_COUNT), + "last_rpc": u64(CTRL_LAST_RPC), + "result": u64(CTRL_RESULT), + "handler_count": u64(CTRL_HANDLER_COUNT), + } + + +def apply_mutation(session, ctrl: int, mutation: Mutation) -> None: + if mutation.region: + region = mutation.region + data = bytearray(session.memory.read(region["addr"], region["size"])) + region["old"] = data[region["offset"]] + data[region["offset"]] ^= region["xor"] + region["new"] = data[region["offset"]] + session.memory.write(region["addr"], bytes(data)) + session.memory.write_u64(ctrl + CTRL_MAILBOX, mutation.cmd) + session.memory.write_u64(ctrl + CTRL_ARG0, resolve_arg0(mutation, ctrl)) + session.memory.write_u64(ctrl + CTRL_ARG1, mutation.arg1) + session.memory.write_u64(ctrl + CTRL_ARG2, mutation.arg2) + + +def _compact_fault(fault: dict) -> dict: + return { + "cause": fault.get("cause"), + "desc": fault.get("desc"), + "epc": fault.get("epc"), + "tval": fault.get("tval"), + "access": fault.get("access"), + } + + +def classify( + mutation: Mutation, + before: dict, + after: dict, + faults: list[dict], + trap_loop: bool, + exit_seen: bool, + settle_ok: bool, + ctrl: int, +) -> tuple[str, dict]: + delta = after["rpc_count"] - before["rpc_count"] + detail = { + "observed": { + "result": after["result"], + "last_rpc": after["last_rpc"], + "rpc_count_delta": delta, + }, + "expected": None, + "expected_ok": None, + } + if not settle_ok: + detail["reason"] = f"{SETTLE_MARKER!r} not observed within the settle budget" + return "settle-failed", detail + if exit_seen: + return "exit", detail + if faults: + fault = faults[0] + detail["fault"] = _compact_fault(fault) + return ("trap" if fault.get("cause") == 3 else "fault"), detail + if trap_loop: + return "trap", detail + if mutation.cmd != 0 and delta <= 0: + return "no-progress", detail + expectation = expected_for(mutation, before, ctrl) + detail["expected"] = expectation + verdict = expected_ok(expectation, mutation, before, after) + detail["expected_ok"] = verdict + if verdict is False: + return "unexpected-result", detail + return "ok", detail + + +def _settle(session, fs, options: HarnessOptions) -> bool: + for _ in range(SETTLE_RETRIES): + session.run(duration=options.settle, max_stops=1) + if SETTLE_MARKER in fs.target.read_uart(): + return True + return False + + +def run_iteration(mutation: Mutation, options: HarnessOptions, image) -> dict: + session_rel = os.path.join("sessions", f"iter-{mutation.iteration:04d}") + finding = { + "iteration": mutation.iteration, + "strategy": mutation.strategy, + "mutation": mutation.to_json(), + "outcome": "error", + "detail": {}, + "session": session_rel, + } + workdir = tempfile.mkdtemp(prefix="greeneye-mut-") + try: + with open_flat_session( + image, + options.kernel, + record=os.path.join(options.out, session_rel), + name=f"mut-{mutation.iteration:04d}", + symbols=None, + log_faults=True, + workdir=workdir, + ) as fs: + session = fs.session + ctrl_symbol = session.symbols.get("gsp_ctrl") + if ctrl_symbol is None: + raise GreenEyeError("gsp_ctrl symbol not found: the mutation harness targets the demo control block") + ctrl = ctrl_symbol.addr + settle_ok = _settle(session, fs, options) + before = read_ctrl_state(session, ctrl) + faults: list[dict] = [] + trap_loop = False + exit_seen = False + after = before + if settle_ok: + log_offset = os.path.getsize(fs.target.qemu_log) if os.path.exists(fs.target.qemu_log) else 0 + apply_mutation(session, ctrl, mutation) + phase_a = min(options.phase_a, options.timeout) + session.run(duration=phase_a, max_stops=1) + faults, _capped = read_faults_window(fs.target.qemu_log, log_offset, options.max_log_bytes) + trap_loop = session.pc < 0x100 + after = read_ctrl_state(session, ctrl) + # Slow/hung handlers get one confirmation window before we call + # it no-progress; faults are still picked up if they appear. + if not faults and not trap_loop and mutation.cmd != 0 and after["rpc_count"] == before["rpc_count"]: + session.run(duration=max(0.01, options.timeout - phase_a), max_stops=1) + more, _capped = read_faults_window(fs.target.qemu_log, log_offset, options.max_log_bytes) + faults = more or faults + trap_loop = session.pc < 0x100 + after = read_ctrl_state(session, ctrl) + exit_seen = not fs.target.alive() + outcome, detail = classify(mutation, before, after, faults, trap_loop, exit_seen, settle_ok, ctrl) + finding["outcome"] = outcome + finding["detail"] = detail + except Exception as exc: # one bad iteration must not end the run + finding["outcome"] = "error" + finding["detail"] = {"reason": f"{type(exc).__name__}: {exc}"} + finally: + shutil.rmtree(workdir, ignore_errors=True) + return finding + + +# --------------------------------------------------------------------------- +# run + report +# --------------------------------------------------------------------------- + + +def run_harness(options: HarnessOptions) -> list[dict]: + if not qemu_available(): + raise GreenEyeError("qemu-system-riscv64 not found; install qemu-system-misc or set GREENEYE_QEMU") + if not os.path.isfile(options.kernel): + raise GreenEyeError(f"kernel not found: {options.kernel}") + image = load_target(options.kernel) + options.kernel_sha256 = sha256_file(options.kernel) + demo_sha = sha256_file(DEMO_ELF) if os.path.isfile(DEMO_ELF) else None + options.is_demo = options.kernel_sha256 == demo_sha or os.path.abspath(options.kernel) == os.path.abspath(DEMO_ELF) + if not options.is_demo and not options.allow_foreign_elf: + raise GreenEyeError( + "mutation is restricted to the bundled demo firmware; " + "pass --allow-foreign-elf to mutate another re-hosted ELF under QEMU" + ) + if options.memory_region: + addr, size = options.memory_region + limit = LOAD_ADDR + memory_mb_for(image) * 1024 * 1024 + if size <= 0 or addr < LOAD_ADDR or addr + size > limit: + raise GreenEyeError( + f"--memory-region {addr:#x}:{size:#x} is outside QEMU RAM " + f"[{LOAD_ADDR:#x}, {limit:#x})" + ) + ensure_dir(options.out) + findings: list[dict] = [] + mutations = generate_mutations(options.iterations, options.seed, options.memory_region) + findings_path = os.path.join(options.out, "findings.jsonl") + with open(findings_path, "w", encoding="utf-8") as fh: + for mutation in mutations: + finding = run_iteration(mutation, options, image) + findings.append(finding) + fh.write(json.dumps(finding, sort_keys=True) + "\n") + fh.flush() + write_summary(options, findings) + return findings + + +def write_summary(options: HarnessOptions, findings: list[dict]) -> str: + classes = Counter(finding["outcome"] for finding in findings) + strategies = Counter(finding["strategy"] for finding in findings) + lines = [ + "# green-eye mutation findings", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/mutation_harness.py`.", + "", + f"* format: `{FORMAT}`", + f"* kernel: `{options.kernel}`", + f"* kernel sha256: `{options.kernel_sha256}`", + f"* demo restriction: {'demo' if options.is_demo else 'foreign ELF (explicitly allowed)'}", + f"* seed: {options.seed}, iterations: {options.iterations}", + f"* timeout: {options.timeout}s, settle: {options.settle}s", + f"* memory region: {_fmt_region(options.memory_region)}", + "", + "## Outcome classes", + "", + ] + lines += _table( + ["outcome", "count", "first iteration"], + [ + [outcome, classes[outcome], next(f["iteration"] for f in findings if f["outcome"] == outcome)] + for outcome in sorted(classes) + ], + ) + lines += ["", "## Strategy coverage", ""] + lines += _table( + ["strategy", "iterations", "outcomes"], + [ + [ + strategy, + strategies[strategy], + ", ".join( + f"{outcome}={count}" + for outcome, count in sorted(Counter(f["outcome"] for f in findings if f["strategy"] == strategy).items()) + ), + ] + for strategy in sorted(strategies) + ], + ) + notable = [finding for finding in findings if finding["outcome"] != "ok"][:20] + lines += ["", "## Notable findings (first 20 non-ok)", ""] + if not notable: + lines += ["_all iterations completed normally_", ""] + for finding in notable: + lines += [ + f"* iteration {finding['iteration']} `{finding['strategy']}` " + f"cmd={finding['mutation']['cmd']:#x} -> **{finding['outcome']}** " + f"{_short_json(finding['detail'])}", + ] + lines += ["", "## Reproduce", "", "```bash", _reproduce_command(options), "```", ""] + path = os.path.join(options.out, "summary.md") + with open(path, "w", encoding="utf-8") as fh: + fh.write("\n".join(lines)) + return path + + +def _reproduce_command(options: HarnessOptions) -> str: + region = f" --memory-region {_fmt_region(options.memory_region)}" if options.memory_region else "" + foreign = " --allow-foreign-elf" if not options.is_demo else "" + return ( + f"python3 harness/mutation_harness.py --kernel {options.kernel} --out {options.out} " + f"--iterations {options.iterations} --seed {options.seed} --timeout {options.timeout} " + f"--settle {options.settle}{region}{foreign}" + ) + + +def _fmt_region(region: tuple[int, int] | None) -> str: + return "-" if not region else f"{region[0]:#x}:{region[1]:#x}" + + +def _short_json(value, limit: int = 120) -> str: + text = json.dumps(value, sort_keys=True) + return f"`{text}`" if len(text) <= limit else f"`{text[:limit - 1]}...`" + + +def _table(headers: list[str], rows: list[list]) -> list[str]: + lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] + for row in rows: + lines.append("| " + " | ".join(str(cell) for cell in row) + " |") + return lines + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def parse_region(text: str) -> tuple[int, int]: + try: + addr_text, size_text = text.split(":", 1) + addr = int(addr_text, 0) + size = int(size_text, 0) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"expected ADDR:SIZE, got {text!r}") from exc + return addr, size + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="mutation_harness", description=__doc__.splitlines()[0]) + parser.add_argument("--kernel", default=DEMO_ELF, help="demo/re-hosted ELF to mutate (default: bundled demo)") + parser.add_argument("--out", required=True, help="output directory for findings and recorded sessions") + parser.add_argument("--iterations", type=int, default=50, help="mutation count (1..10000)") + parser.add_argument("--seed", type=int, default=1, help="deterministic RNG seed") + parser.add_argument("--timeout", type=float, default=0.25, help="total observation budget per mutation (seconds)") + parser.add_argument("--settle", type=float, default=0.2, help="settle budget per attempt before mutating (seconds)") + parser.add_argument("--phase-a", type=float, default=0.05, help="first observation window before confirmation") + parser.add_argument("--memory-region", type=parse_region, help="optional RAM region to mutate, ADDR:SIZE") + parser.add_argument("--allow-foreign-elf", action="store_true", help="permit a non-demo ELF (QEMU only)") + parser.add_argument("--max-log-bytes", type=int, default=MAX_LOG_BYTES, help="QEMU log window per fault check") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if not 1 <= args.iterations <= 10000: + parser.error("--iterations must be in 1..10000") + if not 0.01 <= args.timeout <= 10.0: + parser.error("--timeout must be in 0.01..10.0") + if not 0.0 <= args.settle <= 10.0: + parser.error("--settle must be in 0..10.0") + options = HarnessOptions( + kernel=os.path.abspath(args.kernel), + out=os.path.abspath(args.out), + iterations=args.iterations, + seed=args.seed, + timeout=args.timeout, + settle=args.settle, + phase_a=args.phase_a, + memory_region=args.memory_region, + allow_foreign_elf=args.allow_foreign_elf, + max_log_bytes=args.max_log_bytes, + ) + try: + findings = run_harness(options) + except GreenEyeError as exc: + print(f"mutation harness error: {exc}", file=sys.stderr) + return 2 + classes = Counter(finding["outcome"] for finding in findings) + print(f"[mutation] {len(findings)} iteration(s), out={options.out}") + for outcome in sorted(classes): + print(f"[mutation] {outcome}: {classes[outcome]}") + print(f"[mutation] wrote {os.path.join(options.out, 'findings.jsonl')}") + print(f"[mutation] wrote {os.path.join(options.out, 'summary.md')}") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py b/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py new file mode 100644 index 0000000..82af458 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py @@ -0,0 +1,779 @@ +#!/usr/bin/env python3 +"""Multi-version portability orchestration (roadmap R6, H4.1). + +Given one or more firmware ELF targets (a demo or re-hosted build, a raw +``.fwimage``, a future version), run the same deterministic scenario against +each on the QEMU backend: + +1. bounded probe: relocate into virt RAM and run/step a fixed budget, + capturing the first exception via ``parse_qemu_faults``; +2. symbol DB build: ELF symbols plus optional Ghidra exports / an existing + green-eye JSON DB, saved per target; +3. scripted demo run with handler hooks and a control-block watchpoint, + recorded as a normal green-eye session; +4. cross-target comparison with ``greeneye.analyze.compare_sessions`` plus + explicit hit/watch/write profile deltas and the docs/06 M1-M6 skeleton + (metrics without a reference capture are marked "not measured"). + +Outputs ``portability-report.json`` and ``portability-report.md`` under +``--out DIR`` with stable ordering. Unsupported targets are skipped with a +reason instead of aborting the matrix. + +Run from the tool root: + + python3 harness/portability_matrix.py --target firmware/uproc_demo/build/uproc_demo.elf --out dumps/matrix +""" +from __future__ import annotations + +import argparse +import hashlib +import os +import struct +from collections import Counter +from dataclasses import dataclass +from pathlib import Path + +try: + from .common import ( + LOAD_ADDR, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + load_target, + open_flat_session, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) +except ImportError: # direct script execution (python3 harness/portability_matrix.py) + from common import ( # type: ignore[no-redef] + LOAD_ADDR, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + load_target, + open_flat_session, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) + +from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 +from greeneye.record import load_events # noqa: E402 + +FORMAT = "greeneye-portability-matrix/1" +RPC_HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault") +CTRL_SIZE = 400 +CTRL_RPC_COUNT = 16 +CTRL_STATE = 8 +CTRL_HANDLER_COUNT = 328 + +METRIC_DEFS = ( + ("M1", "Inventory coverage"), + ("M2", "Reply agreement"), + ("M3", "Ordering agreement"), + ("M4", "State agreement"), + ("M5", "Error-path agreement"), + ("M6", "Memory side-effects"), +) +NO_REFERENCE = "agreement is not measured: needs a reference capture (docs/06 5)" + + +@dataclass +class TargetSpec: + path: str + name: str + + +@dataclass +class MatrixOptions: + out: str + probe_instructions: int = 2000 + probe_duration: float = 0.5 + scenario_duration: float = 0.4 + max_stops: int = 5000 + max_rows: int = 40 + symbols_json: str | None = None + ghidra_names: str | None = None + ghidra_index: str | None = None + ghidra_exports: str | None = None + ghidra_callgraph: str | None = None + ghidra_bias: int = 0 + scenario_script: str | None = None + log_bytes: int = MAX_LOG_BYTES + + +def _skipped_probe(reason: str | None) -> dict: + return { + "status": "skipped", + "reason": reason, + "session": None, + "instructions": None, + "stopped_pc": None, + "stopped_symbol": None, + "fault_count": 0, + "faults_capped": False, + "first_fault": None, + } + + +def _skipped_scenario(reason: str | None) -> dict: + return { + "status": "skipped", + "reason": reason, + "session": None, + "hooks": [], + "uart_banner": None, + "ctrl": None, + "snapshot": None, + "event_counts": {}, + "bp": {}, + "watch": {}, + "writes": {}, + "fault_count": 0, + "faults_capped": False, + "first_fault": None, + } + + +# --------------------------------------------------------------------------- +# per-target steps +# --------------------------------------------------------------------------- + + +def build_symbols(spec: TargetSpec, bias: int, out_dir: str, options: MatrixOptions): + """ELF symbols + optional Ghidra exports / JSON DB, saved per target.""" + db = SymbolDB() + db.load_elf(spec.path, bias=bias, source="elf") + sources = ["elf"] + if options.symbols_json: + if not os.path.isfile(options.symbols_json): + raise FileNotFoundError(f"symbol DB not found: {options.symbols_json}") + db.load_json(options.symbols_json) + sources.append(os.path.basename(options.symbols_json)) + ghidra_inputs = any( + ( + options.ghidra_names, + options.ghidra_index, + options.ghidra_exports, + options.ghidra_callgraph, + ) + ) + if ghidra_inputs: + from greeneye.ghidra import build_symbol_db + + gdb = build_symbol_db( + names_tsv=options.ghidra_names, + index_tsv=options.ghidra_index, + exports_tsv=options.ghidra_exports, + callgraph_tsv=options.ghidra_callgraph, + bias=options.ghidra_bias, + ) + for symbol in gdb: + db.add(symbol) + sources.append("ghidra") + rel = os.path.join("symbols", f"{spec.name}.json") + out_path = os.path.join(out_dir, rel) + ensure_dir(os.path.dirname(out_path)) + db.save_json( + out_path, + meta={"bias": bias, "sources": sources, "firmware": os.path.basename(spec.path)}, + ) + return db, sources, rel + + +def run_probe(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: + if not qemu_ok: + return _skipped_probe("qemu-system-riscv64 not found") + result = _skipped_probe(None) + result["instructions"] = options.probe_instructions + session_rel = os.path.join("sessions", f"{spec.name}-probe") + try: + with open_flat_session( + image, + spec.path, + record=os.path.join(out_dir, session_rel), + name=f"{spec.name}-probe", + symbols=db, + log_faults=True, + ) as fs: + session = fs.session + if options.probe_instructions > 0: + session.step(options.probe_instructions) + else: + session.run(duration=options.probe_duration) + result["stopped_pc"] = session.pc + symbol = session.symbols.resolve(session.pc) + result["stopped_symbol"] = symbol.name if symbol else None + text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + result["first_fault"] = faults[0] if faults else None + result["session"] = session_rel + if text: + fs.session.recorder.meta["qemu_log_blob"] = fs.session.recorder.blob("qemu.log", text.encode("utf-8")) + fs.session.recorder.meta["faults_capped"] = capped + result["status"] = "ok" + except Exception as exc: # a failed step must not abort the matrix + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +def install_scenario(session) -> list[str]: + hooks: list[str] = [] + for name in RPC_HANDLERS: + symbol = session.symbols.get(name) + if symbol is None: + continue + session.hook(symbol.addr, on_enter=lambda hit: None, listener=f"matrix:{name}") + hooks.append(name) + return hooks + + +def read_ctrl(session, symbol) -> dict | None: + if symbol is None: + return None + raw = session.memory.read(symbol.addr, CTRL_SIZE) + u32 = lambda offset: struct.unpack_from("<I", raw, offset)[0] # noqa: E731 + u64 = lambda offset: struct.unpack_from("<Q", raw, offset)[0] # noqa: E731 + return { + "addr": f"{symbol.addr:#x}", + "magic": u32(0), + "version": u32(4), + "state": u32(CTRL_STATE), + "rpc_count": u64(CTRL_RPC_COUNT), + "last_rpc": u64(24), + "result": u64(64), + "handler_count": u64(CTRL_HANDLER_COUNT), + } + + +def profile_from_events(events: list[dict]) -> dict: + counts: Counter = Counter() + bp: Counter = Counter() + watch: Counter = Counter() + writes: Counter = Counter() + for event in events: + kind = event.get("type", "?") + counts[kind] += 1 + if kind == "bp_hit": + bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1 + elif kind == "watch_hit": + watch[f"{event.get('addr', 0):#x}:{event.get('kind') or 'w'}"] += 1 + elif kind == "mem_write": + writes[f"{(event.get('addr', 0) >> 12) << 12:#x}"] += 1 + return { + "event_counts": dict(sorted(counts.items())), + "bp": {key: bp[key] for key in sorted(bp)}, + "watch": {key: watch[key] for key in sorted(watch)}, + "writes": {key: writes[key] for key in sorted(writes)}, + } + + +def run_scenario(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: + if not qemu_ok: + return _skipped_scenario("qemu-system-riscv64 not found") + result = _skipped_scenario(None) + session_rel = os.path.join("sessions", f"{spec.name}-scenario") + try: + with open_flat_session( + image, + spec.path, + record=os.path.join(out_dir, session_rel), + name=f"{spec.name}-scenario", + symbols=db, + log_faults=True, + ) as fs: + session = fs.session + result["hooks"] = install_scenario(session) + ctrl_symbol = session.symbols.get("gsp_ctrl") + ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None + if ctrl_addr is not None: + session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w", listener="matrix:rpc_count") + if options.scenario_script: + import greeneye as ge + + script_path = Path(options.scenario_script) + namespace = { + "ge": ge, + "greeneye": ge, + "session": session, + "s": session, + "memory": session.memory, + "symbols": session.symbols, + "recorder": session.recorder, + } + exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace) + else: + session.run(duration=options.scenario_duration, max_stops=options.max_stops) + if ctrl_addr is not None: + session.memory.write_u64(ctrl_addr + 32, 1) # PING + session.memory.write_u64(ctrl_addr + 40, 0x1234) + session.run(duration=options.scenario_duration, max_stops=options.max_stops) + result["ctrl"] = read_ctrl(session, ctrl_symbol) + if ctrl_addr is not None: + data = session.snapshot(ctrl_addr, CTRL_SIZE, name="ctrl") + result["snapshot"] = {"label": "ctrl", "size": len(data), "sha256": hashlib.sha256(data).hexdigest()} + uart = fs.target.read_uart() + result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None + text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + result["first_fault"] = faults[0] if faults else None + result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel)))) + result["session"] = session_rel + result["status"] = "ok" + except Exception as exc: # a failed step must not abort the matrix + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +# --------------------------------------------------------------------------- +# metrics + comparisons +# --------------------------------------------------------------------------- + + +def metric_rows(probe: dict, scenario: dict) -> list[dict]: + """docs/06 M1-M6 skeleton; observed data is local, agreement stays unmeasured.""" + bp = scenario.get("bp") or {} + ctrl = scenario.get("ctrl") or {} + snapshot = scenario.get("snapshot") or {} + probe_fault = (probe.get("first_fault") or {}).get("desc") + scenario_fault = (scenario.get("first_fault") or {}).get("desc") + rows = [ + { + "id": "M1", + "name": "Inventory coverage", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "rpc_handlers_hit": sorted(k for k in bp if k in RPC_HANDLERS), + "other_hit_sites": sorted(k for k in bp if k not in RPC_HANDLERS), + "declared_handlers": ctrl.get("handler_count"), + }, + }, + { + "id": "M2", + "name": "Reply agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"note": "reply fields are not decoded by the builtin scenario", "result": ctrl.get("result")}, + }, + { + "id": "M3", + "name": "Ordering agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"events": sum((scenario.get("event_counts") or {}).values())}, + }, + { + "id": "M4", + "name": "State agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"final_state": ctrl.get("state"), "rpc_state_hits": bp.get("rpc_state", 0)}, + }, + { + "id": "M5", + "name": "Error-path agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "probe_first_fault": probe_fault, + "scenario_first_fault": scenario_fault, + "scenario_fault_count": scenario.get("fault_count", 0), + }, + }, + { + "id": "M6", + "name": "Memory side-effects", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "written_pages": len(scenario.get("writes") or {}), + "ctrl_snapshot_sha256": snapshot.get("sha256"), + }, + }, + ] + return rows + + +def compare_targets(a: dict, b: dict, out_dir: str, options: MatrixOptions) -> dict: + entry = { + "a": a["name"], + "b": b["name"], + "status": "ok", + "reason": None, + "compare_report": None, + "deltas": {"bp": [], "watch": [], "writes": []}, + "all_zero": True, + "metrics": [], + } + a_scenario = a.get("scenario") or {} + b_scenario = b.get("scenario") or {} + if a_scenario.get("status") != "ok" or b_scenario.get("status") != "ok": + entry["status"] = "skipped" + entry["reason"] = "a scenario session is not available on both targets" + return entry + a_dir = os.path.join(out_dir, a_scenario.get("session") or "") + b_dir = os.path.join(out_dir, b_scenario.get("session") or "") + try: + report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows) + rel = os.path.join("compare", f"{a['name']}--{b['name']}.md") + write_text(os.path.join(out_dir, rel), report) + entry["compare_report"] = rel + except Exception as exc: + entry["reason"] = f"{type(exc).__name__}: {exc}" + deltas = {"bp": [], "watch": [], "writes": []} + for name in ("bp", "watch", "writes"): + deltas[name] = counter_delta(a_scenario.get(name) or {}, b_scenario.get(name) or {}) + entry["deltas"] = deltas + entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows) + a_metrics = {row["id"]: row for row in a.get("metrics") or []} + b_metrics = {row["id"]: row for row in b.get("metrics") or []} + entry["metrics"] = [ + { + "id": metric_id, + "name": name, + "status": "not measured", + "reason": NO_REFERENCE, + "a_observed": (a_metrics.get(metric_id) or {}).get("observed"), + "b_observed": (b_metrics.get(metric_id) or {}).get("observed"), + } + for metric_id, name in METRIC_DEFS + ] + return entry + + +# --------------------------------------------------------------------------- +# report assembly + rendering +# --------------------------------------------------------------------------- + + +def analyze_target(spec: TargetSpec, options: MatrixOptions, qemu_ok: bool) -> dict: + target = { + "name": spec.name, + "path": spec.path, + "sha256": "", + "kind": "elf", + "entry": None, + "status": "skipped", + "skip_reason": None, + "probe": _skipped_probe(None), + "symbols": {"count": 0, "sources": [], "path": None}, + "scenario": _skipped_scenario(None), + "metrics": metric_rows({}, {}), + } + if not os.path.isfile(spec.path): + target["skip_reason"] = "file not found" + return target + target["sha256"] = sha256_file(spec.path) + try: + image = load_target(spec.path) + except Exception as exc: + target["kind"] = "unknown" + target["skip_reason"] = f"{type(exc).__name__}: {exc}" + return target + target["entry"] = f"{image.entry:#x}" + bias = LOAD_ADDR - image.min_vaddr + db = SymbolDB() + try: + db, sources, sym_rel = build_symbols(spec, bias, options.out, options) + target["symbols"] = {"count": len(db), "sources": sources, "path": sym_rel} + except Exception as exc: + target["symbols"] = {"count": 0, "sources": [], "path": None, "reason": f"{type(exc).__name__}: {exc}"} + db.load_elf(spec.path, bias=bias, source="elf") + target["symbols"]["count"] = len(db) + target["symbols"]["sources"] = ["elf"] + target["probe"] = run_probe(spec, image, db, options.out, options, qemu_ok) + target["scenario"] = run_scenario(spec, image, db, options.out, options, qemu_ok) + target["metrics"] = metric_rows(target["probe"], target["scenario"]) + statuses = {target["probe"]["status"], target["scenario"]["status"]} + if statuses == {"ok"}: + target["status"] = "ok" + elif "ok" in statuses: + target["status"] = "partial" + elif statuses == {"skipped"}: + target["status"] = "skipped" + target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") + else: + target["status"] = "error" + target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") + return target + + +def run_matrix(specs: list[TargetSpec], options: MatrixOptions) -> dict: + ensure_dir(options.out) + qemu_ok = qemu_available() + targets = [analyze_target(spec, options, qemu_ok) for spec in specs] + comparisons: list[dict] = [] + for i in range(len(targets)): + for j in range(i + 1, len(targets)): + comparisons.append(compare_targets(targets[i], targets[j], options.out, options)) + report = { + "format": FORMAT, + "tool": "harness/portability_matrix.py", + "backend": "qemu", + "qemu_available": qemu_ok, + "options": { + "probe_instructions": options.probe_instructions, + "probe_duration": options.probe_duration, + "scenario_duration": options.scenario_duration, + "scenario_script": options.scenario_script, + "max_rows": options.max_rows, + "symbols_json": options.symbols_json, + "ghidra_bias": options.ghidra_bias, + }, + "targets": targets, + "comparisons": comparisons, + "skipped": [ + {"name": target["name"], "reason": target["skip_reason"]} + for target in targets + if target["status"] == "skipped" + ], + } + return report + + +def _table(headers: list[str], rows: list[list]) -> list[str]: + lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] + for row in rows: + lines.append("| " + " | ".join(str(cell) for cell in row) + " |") + return lines + + +def _profile_tables(title: str, profile: dict, max_rows: int) -> list[str]: + lines = [f"**{title}**", ""] + if not profile: + return lines + ["_none observed_", ""] + rows = [[f"`{key}`", count] for key, count in sorted(profile.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:max_rows]] + lines += _table(["key", "hits"], rows) + [""] + return lines + + +def render_markdown(report: dict, options: MatrixOptions) -> str: + lines = [ + "# green-eye portability delta report", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/portability_matrix.py`.", + "", + f"* format: `{report['format']}` (green-eye {_greeneye_version()})", + f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})", + f"* targets: {len(report['targets'])}", + "", + "## Targets", + "", + ] + lines += _table( + ["target", "status", "sha256", "probe", "symbols", "scenario events"], + [ + [ + f"`{target['name']}`", + target["status"], + f"`{(target['sha256'] or '')[:16]}`", + f"{target['probe']['status']} ({target['probe']['fault_count']} faults)", + target["symbols"]["count"], + sum((target["scenario"].get("event_counts") or {}).values()), + ] + for target in report["targets"] + ], + ) + lines += [""] + + for target in report["targets"]: + lines += [f"### {target['name']}", ""] + lines += [f"* path: `{target['path']}`", f"* entry: `{target['entry']}`", f"* sha256: `{target['sha256']}`"] + if target["skip_reason"]: + lines += [f"* note: {target['skip_reason']}"] + lines += [""] + + probe = target["probe"] + lines += ["#### Probe", ""] + fault = probe.get("first_fault") + stopped_pc = probe.get("stopped_pc") + probe_rows = [ + ["status", probe["status"]], + ["instructions", probe.get("instructions")], + ["stopped pc", f"`{stopped_pc:#x}`" if stopped_pc is not None else None], + ["stopped symbol", probe.get("stopped_symbol")], + ["fault count", probe.get("fault_count")], + ["faults capped", probe.get("faults_capped")], + ] + if fault: + probe_rows.append( + ["first fault", f"cause {fault.get('cause')} {fault.get('desc')} @ epc `{fault.get('epc', 0):#x}`"] + ) + if probe.get("reason"): + probe_rows.append(["reason", probe["reason"]]) + lines += _table(["field", "value"], probe_rows) + [""] + + lines += ["#### Symbols", ""] + lines += _table( + ["count", "sources", "path"], + [[target["symbols"]["count"], ", ".join(target["symbols"]["sources"]), target["symbols"]["path"]]], + ) + [""] + + scenario = target["scenario"] + lines += ["#### Scenario", ""] + lines += _table( + ["status", "session", "hooks", "uart banner", "faults"], + [[ + scenario["status"], + f"`{scenario.get('session')}`", + ", ".join(scenario.get("hooks") or []), + scenario.get("uart_banner"), + scenario.get("fault_count"), + ]], + ) + [""] + lines += _profile_tables("Breakpoint hits", scenario.get("bp") or {}, options.max_rows) + lines += _profile_tables("Watchpoint hits", scenario.get("watch") or {}, options.max_rows) + lines += _profile_tables("Written pages", scenario.get("writes") or {}, options.max_rows) + + lines += ["#### M1-M6", ""] + lines += [f"> {NO_REFERENCE}", ""] + lines += _table( + ["#", "metric", "status", "observed"], + [ + [row["id"], row["name"], row["status"], _short_json(row["observed"])] + for row in target["metrics"] + ], + ) + [""] + + lines += ["## Cross-target comparisons", ""] + if not report["comparisons"]: + lines += ["_single target: no comparison_", ""] + for comp in report["comparisons"]: + lines += [f"### {comp['a']} vs {comp['b']}", ""] + lines += [f"* status: {comp['status']}", f"* zero deltas: {'yes' if comp['all_zero'] else 'no'}"] + if comp.get("compare_report"): + lines += [f"* compare report: `{comp['compare_report']}`"] + if comp.get("reason"): + lines += [f"* note: {comp['reason']}"] + lines += [""] + for name in ("bp", "watch", "writes"): + rows = [row for row in comp["deltas"][name] if row["delta"] != 0] + rows.sort(key=lambda row: (-abs(row["delta"]), row["key"])) + lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""] + if rows: + lines += _table( + ["key", comp["a"], comp["b"], "delta"], + [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]], + ) + [""] + else: + lines += ["_no deltas_", ""] + lines += ["**M1-M6 (not measured: reference capture required)**", ""] + lines += _table( + ["#", "metric", "A observed", "B observed"], + [ + [row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])] + for row in comp["metrics"] + ], + ) + [""] + + if report["skipped"]: + lines += ["## Skipped targets", ""] + lines += _table(["target", "reason"], [[f"`{item['name']}`", item["reason"]] for item in report["skipped"]]) + lines += [""] + return "\n".join(lines) + + +def _short_json(value, limit: int = 72) -> str: + import json + + if value is None: + return "-" + text = json.dumps(value, sort_keys=True) + return f"`{text}`" if len(text) <= limit else f"`{text[:limit - 1]}...`" + + +def _greeneye_version() -> str: + try: + import greeneye + + return greeneye.__version__ + except Exception: # pragma: no cover + return "unknown" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="portability_matrix", description=__doc__.splitlines()[0]) + parser.add_argument("targets", nargs="*", help="firmware ELF paths") + parser.add_argument("--target", action="append", default=[], help="firmware ELF path (repeatable)") + parser.add_argument("--out", required=True, help="output directory for the report, sessions and symbols") + parser.add_argument("--symbols", dest="symbols_json", help="existing greeneye symbol DB JSON") + parser.add_argument("--ghidra-names", help="Ghidra names.tsv export") + parser.add_argument("--ghidra-index", help="Ghidra decompiled_index.tsv export") + parser.add_argument("--ghidra-exports", help="Ghidra exports.tsv") + parser.add_argument("--ghidra-callgraph", help="Ghidra callgraph.tsv") + parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0) + parser.add_argument("--probe-instructions", type=int, default=2000, help="probe step budget (0 = wall-clock run)") + parser.add_argument("--probe-duration", type=float, default=0.5, help="probe wall budget when --probe-instructions=0") + parser.add_argument("--scenario-duration", type=float, default=0.4, help="budget for each scenario phase") + parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per scenario phase") + parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports") + parser.add_argument("--scenario-script", help="replace the builtin scenario with a green-eye script") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + targets = list(args.targets) + list(args.target) + if not targets: + parser.error("at least one target is required (positional or --target)") + options = MatrixOptions( + out=os.path.abspath(args.out), + probe_instructions=args.probe_instructions, + probe_duration=args.probe_duration, + scenario_duration=args.scenario_duration, + max_stops=args.max_stops, + max_rows=args.max_rows, + symbols_json=args.symbols_json, + ghidra_names=args.ghidra_names, + ghidra_index=args.ghidra_index, + ghidra_exports=args.ghidra_exports, + ghidra_callgraph=args.ghidra_callgraph, + ghidra_bias=args.ghidra_bias, + scenario_script=args.scenario_script, + ) + specs = _name_specs(targets) + report = run_matrix(specs, options) + json_path = write_json(os.path.join(options.out, "portability-report.json"), report) + md_path = write_text(os.path.join(options.out, "portability-report.md"), render_markdown(report, options)) + print(f"[matrix] {len(specs)} target(s), out={options.out}") + for target in report["targets"]: + print(f"[matrix] {target['name']}: status={target['status']} probe={target['probe']['status']} " + f"scenario={target['scenario']['status']}") + print(f"[matrix] wrote {json_path}") + print(f"[matrix] wrote {md_path}") + return 0 + + +def _name_specs(paths: list[str]) -> list[TargetSpec]: + seen: dict[str, int] = {} + specs: list[TargetSpec] = [] + for path in paths: + base = Path(path).stem + seen[base] = seen.get(base, 0) + 1 + name = base if seen[base] == 1 else f"{base}-{seen[base]}" + specs.append(TargetSpec(path=os.path.abspath(path), name=name)) + return specs + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/reach.py b/debug/risc-v-gsp-rm-vm/harness/reach.py new file mode 100644 index 0000000..9df0c4c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/reach.py @@ -0,0 +1,946 @@ +#!/usr/bin/env python3 +"""Reachability answers for a decompiled firmware target (H4.1/H4.5). + +Given a symbol name or address taken from Ghidra or the symbol DB, answer the +practical questions a reimplementation needs: *is this code reachable, how do +I trigger it, what arguments does it need, and what happens when it runs?* + +Methods attempted (``--method auto`` tries them in this order and stops at the +first hit unless ``--all-methods`` is given): + +* ``stimulus`` -- run a supplied green-eye script (``--stimulus script.py``) + or probe spec (``--probe probe.json``) under instrumentation; +* ``trigger`` -- discover the dispatch entry that routes to the target (the + demo's handler table), apply that command, and run bounded; +* ``run`` -- ``Session.run_until(target)`` from launch, which catches the + natural boot/self-test path; +* ``call`` -- ``Session.call(target, args)`` with ``--args`` and + ``--arg-count`` auto-fill (zeros or sentinels). + +On a hit the report records the entry register state, the prior hook events +that led there, the caller, and the effects after resume (field deltas and +watchpoint stores, optional recorded-session trace summary). A repeatable +probe spec is written to ``--emit-probe`` (default ``<out>/probe.json``) so a +later run (or docs/21 CI) can replay the exact route. + +Demonstrated on the demo firmware: ``rpc_ping`` via ``call``, ``rpc_sum`` via +``run`` (self-test), ``rpc_fault`` via ``trigger`` (mailbox 6). See +docs/19-CONTROL-DISCOVERY.md. + +Run from the tool root: + + python3 harness/reach.py --target rpc_fault --method trigger --out dumps/reach-fault +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import sys +import time +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path + +try: + from .common import DEMO_ELF, ensure_dir, qemu_available, read_faults_window, sha256_file + from .control_discovery import DemoSurface, SETTLE_MARKER, SETTLE_SLICE +except ImportError: # direct script execution (python3 harness/reach.py) + from common import DEMO_ELF, ensure_dir, qemu_available, read_faults_window, sha256_file # type: ignore[no-redef] + from control_discovery import DemoSurface, SETTLE_MARKER, SETTLE_SLICE # type: ignore[no-redef] + +from greeneye import launch # noqa: E402 +from greeneye.errors import GreenEyeError, RspTimeout # noqa: E402 +from greeneye.record import load_events # noqa: E402 +from greeneye.symbols import SymbolDB # noqa: E402 + +FORMAT = "greeneye-reach-report/1" +PROBE_FORMAT = "greeneye-reach-probe/1" +SENTINEL_BASE = 0xA5A5_0000_0000_0000 +RUN_SLICE = 0.02 +GRACE = 0.10 +AUTO_ORDER = ("stimulus", "trigger", "run", "call") +TRAP_SIGNALS = (5,) +FAULT_SIGNALS = (4, 7, 11) + + +@dataclass +class ReachOptions: + kernel: str = DEMO_ELF + out: str = "" + target: str = "" + method: str = "auto" + args: list[int] = field(default_factory=list) + arg_count: int = 0 + arg_fill: str = "zero" + timeout: float = 0.5 + settle: float = 2.0 + stimulus: str | None = None + probe: str | None = None + emit_probe: str | None = None + record: str | None = None + backend: str = "qemu" + max_prior: int = 12 + all_methods: bool = False + allow_foreign_elf: bool = False + watch: list[dict] = field(default_factory=list) + + +@dataclass +class Attempt: + method: str + ok: bool = False + reason: str | None = None + entry: dict | None = None + retval: int | None = None + dispatch: list[dict] = field(default_factory=list) + prior: list[dict] = field(default_factory=list) + effects: dict = field(default_factory=dict) + trap: bool = False + fault: bool = False + trigger: dict | None = None + session: str | None = None + hit_count: int = 0 + + def summary(self) -> dict: + return { + "method": self.method, + "ok": self.ok, + "reason": self.reason, + "entry": self.entry, + "retval": self.retval, + "dispatch": self.dispatch[:4], + "prior_events": self.prior, + "effects": self.effects, + "trap": self.trap, + "fault": self.fault, + "trigger": self.trigger, + "session": self.session, + "hit_count": self.hit_count, + } + + +def build_args(options: ReachOptions) -> list[int]: + """Explicit --args plus deterministic --arg-count auto-fill.""" + args = list(options.args) + count = max(options.arg_count, len(args)) + if count > 8: + raise GreenEyeError(f"--arg-count {count}: the RISC-V ABI passes at most 8 integer args") + while len(args) < count: + if options.arg_fill == "sentinel": + args.append(SENTINEL_BASE | len(args)) + else: + args.append(0) + return args + + +class Collector: + """Ordered hook/watch evidence for one reachability attempt.""" + + def __init__( + self, + session, + surface: DemoSurface | None, + target_addr: int, + target_name: str, + options: ReachOptions, + fault_log: str | None = None, + ): + self.session = session + self.surface = surface + self.target_addr = target_addr + self.target_name = target_name + self.options = options + self.fault_log = fault_log + self.fault_offset = 0 + self.faults: list[dict] = [] + self.hits: list[dict] = [] + self.entry: dict | None = None + self.prior: list[dict] = [] + self.dispatch: list[dict] = [] + self.signals: list[dict] = [] + self.watch: list[dict] = [] + self.observer = surface.observer() if surface is not None else None + self.before: dict = {} + self.base = 0 + + # -- installation ------------------------------------------------------ + + def install(self) -> None: + self.session.on_event("stop", self._on_stop) + if self.fault_log and os.path.exists(self.fault_log): + self.fault_offset = os.path.getsize(self.fault_log) + if self.surface is not None and self.session.symbols.get("gsp_ctrl"): + self.base = self.session.symbols.lookup("gsp_ctrl") + if self.observer is not None: + self.observer.reset() + self.observer.arm(self.session, self.base) + for spec in self.options.watch: + self._install_watch(spec) + hooked: set[int] = set() + self._hook_addr(self.target_addr, self.target_name) + hooked.add(self.target_addr) + if self.surface is not None: + command_hook = self.surface.command_hook(self.session) + if command_hook: + addr = self.session.symbols.lookup(command_hook) + self._hook_addr(addr, command_hook, stage="dispatch") + hooked.add(addr) + for cmd, (entry, name) in sorted(self.surface.handler_map(self.session).items()): + if entry in hooked: + continue + self._hook_addr(entry, name or f"handler-{entry:#x}", stage="handler", cmd=cmd) + hooked.add(entry) + + def _install_watch(self, spec: dict) -> None: + target = spec.get("symbol", spec.get("addr")) + if target is None: + return + if isinstance(target, str) and not target.lower().startswith("0x"): + base = self.session.symbols.lookup(target) + else: + base = int(target, 0) if isinstance(target, str) else int(target) + offset = int(spec.get("offset", 0)) + size = int(spec.get("size", 8)) + kind = spec.get("kind", "w") + label = spec.get("name") or f"{target}+{offset:#x}" + + def callback(hit, label=label): + self.watch.append({"name": label, "pc": hit.pc, "symbol": hit.symbol.name if hit.symbol else None}) + + self.session.watch(base + offset, size=size, kind=kind, callback=callback) + + def _hook_addr(self, addr: int, name: str, stage: str = "handler", cmd: int | None = None) -> None: + def callback(hit, stage=stage, cmd=cmd): + item = { + "stage": stage, + "symbol": hit.symbol.name if hit.symbol else name, + "addr": hit.addr, + "args": [hit.arg(i) for i in range(4)], + } + if stage == "dispatch": + self.dispatch.append(item) + if hit.addr == self.target_addr: + self.entry = { + "symbol": hit.symbol.name if hit.symbol else name, + "pc": hit.addr, + "args": [hit.arg(i) for i in range(4)], + "ra": hit.regs.get("ra", 0), + "sp": hit.regs.get("sp", 0), + "cmd": cmd, + # The demo compiler's jump table clobbers a0 at handler entry; + # the dispatcher's own args are the authoritative call model. + "dispatch_args": self.dispatch[-1]["args"] if self.dispatch else None, + } + self.prior = list(self.hits[-self.options.max_prior :]) + self.hits.append(item) + + self.session.hook(addr, on_enter=callback, listener=f"reach:{name}") + + def _on_stop(self, hit) -> None: + stop = hit.stop + if stop is not None and stop.signal is not None: + self.signals.append({"signal": stop.signal, "pc": hit.pc}) + + # -- accounting -------------------------------------------------------- + + def snapshot_before(self) -> None: + if self.observer is not None: + self.before = self.observer.snapshot(self.session, self.base) + + def effects(self, retval: int | None) -> dict: + before = self.before + after = self.observer.snapshot(self.session, self.base) if self.observer is not None else {} + changed = [ + {"field": name, "before": before.get(name), "after": after.get(name)} + for name in before + if before.get(name) != after.get(name) + ] + faults: list[dict] = [] + if self.fault_log: + faults, _capped = read_faults_window(self.fault_log, self.fault_offset) + causes = [int(fault.get("cause", 0)) for fault in faults] + trap = any(cause == 3 for cause in causes) or any( + sig.get("signal") in TRAP_SIGNALS for sig in self.signals + ) or self.session.pc < 0x100 + # A trap often drags a fetch-fault loop behind it (mtvec=0); do not + # double-label the consequence as an independent fault. + fault = (not trap) and ( + any(cause != 3 for cause in causes) + or any(sig.get("signal") in FAULT_SIGNALS for sig in self.signals) + ) + effects = { + "changed": changed, + "watch": list(self.watch) + (list(self.observer.hits) if self.observer is not None else []), + "retval": retval, + "trap": bool(trap), + "fault": bool(fault), + "pc_after": self.session.pc, + } + if faults: + effects["faults"] = [ + {"cause": fault.get("cause"), "desc": fault.get("desc"), "epc": fault.get("epc"), + "access": fault.get("access")} + for fault in faults[:4] + ] + return effects + + +# --------------------------------------------------------------------------- +# session plumbing +# --------------------------------------------------------------------------- + + +class SessionHandle: + def __init__(self, session, record_path: str | None, fault_log: str | None = None): + self.session = session + self.path = record_path + self.fault_log = fault_log + + +def open_session(options: ReachOptions, method: str, record: bool) -> SessionHandle: + record_path = os.path.join(options.record, method) if record and options.record else None + if record_path: + ensure_dir(record_path) + fault_log = None + extra_args = None + if options.backend == "qemu": + fault_log = os.path.join(options.out, ".faults", f"{method}.log") + ensure_dir(os.path.dirname(fault_log)) + extra_args = ["-d", "guest_errors,unimp,int", "-D", fault_log] + session = launch( + kernel=options.kernel, + backend=options.backend, + name=f"reach-{method}", + record=record_path, + extra_args=extra_args, + ) + return SessionHandle(session, record_path, fault_log) + + +def settle(session, budget: float) -> bool: + deadline = time.monotonic() + budget + while time.monotonic() < deadline: + if SETTLE_MARKER in session.target.read_uart(): + return True + session.run(duration=SETTLE_SLICE, max_stops=1) + return SETTLE_MARKER in session.target.read_uart() + + +def run_window(session, collector: Collector, timeout: float) -> None: + """Run bounded until the target entry is hit, then a short grace window.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if collector.entry is not None: + break + if not session.target.alive(): + return + remaining = max(0.005, min(RUN_SLICE, deadline - time.monotonic())) + session.run(duration=remaining, max_stops=64) + if collector.entry is not None: + session.run(duration=min(GRACE, max(0.01, timeout)), max_stops=64) + + +# --------------------------------------------------------------------------- +# methods +# --------------------------------------------------------------------------- + + +def attempt_call(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str, + record: bool) -> Attempt: + attempt = Attempt(method="call") + args = build_args(options) + handle = open_session(options, "call", record) + session = handle.session + attempt.session = handle.path + try: + collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log) + collector.install() + collector.snapshot_before() + retval: int | None = None + try: + retval = session.call(target_addr, args, timeout=options.timeout) + except RspTimeout as exc: + attempt.reason = f"call did not return: {exc}" + except GreenEyeError as exc: + attempt.reason = f"call failed: {exc}" + attempt.entry = collector.entry + attempt.prior = collector.prior + attempt.dispatch = collector.dispatch + attempt.hit_count = len(collector.hits) + attempt.retval = retval + attempt.effects = collector.effects(retval) + attempt.trap = attempt.effects["trap"] + attempt.fault = attempt.effects["fault"] + attempt.ok = collector.entry is not None + if not attempt.ok and attempt.reason is None: + attempt.reason = "target entry hook never fired" + finally: + session.close() + return attempt + + +def attempt_run(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str, + record: bool) -> Attempt: + attempt = Attempt(method="run") + handle = open_session(options, "run", record) + session = handle.session + attempt.session = handle.path + try: + collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log) + collector.install() + collector.snapshot_before() + try: + session.run_until(target_addr, timeout=options.timeout) + except RspTimeout as exc: + attempt.reason = f"run_until not reached: {exc}" + except GreenEyeError as exc: + attempt.reason = f"run_until failed: {exc}" + if collector.entry is not None: + session.run(duration=min(GRACE, options.timeout), max_stops=64) + attempt.entry = collector.entry + attempt.prior = collector.prior + attempt.dispatch = collector.dispatch + attempt.hit_count = len(collector.hits) + attempt.effects = collector.effects(None) + attempt.trap = attempt.effects["trap"] + attempt.fault = attempt.effects["fault"] + attempt.ok = collector.entry is not None + if not attempt.ok and attempt.reason is None: + attempt.reason = "target never executed on the boot path" + finally: + session.close() + return attempt + + +def attempt_trigger(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str, + record: bool) -> Attempt: + attempt = Attempt(method="trigger") + if surface is None: + attempt.reason = "no command surface available (gsp_ctrl/rpc_dispatch missing)" + return attempt + args = build_args(options)[:3] + while len(args) < 3: + args.append(0) + handle = open_session(options, "trigger", record) + session = handle.session + attempt.session = handle.path + try: + if not settle(session, options.settle): + attempt.reason = f"{SETTLE_MARKER!r} not observed; trigger state unknown" + return attempt + candidates = surface.command_for_addr(session, target_addr) + if not candidates: + attempt.reason = "target is not in the declared handler/dispatch table" + return attempt + # Prefer the documented switch case over the injected-slot alias. + candidates.sort(key=lambda item: (item["id"] >= 16, item["id"])) + chosen = candidates[0] + collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log) + collector.install() + collector.snapshot_before() + surface.probe(session, int(chosen["id"]), (args[0], args[1], args[2]), timeout=options.timeout) + run_window(session, collector, options.timeout) + attempt.entry = collector.entry + attempt.prior = collector.prior + attempt.dispatch = collector.dispatch + attempt.hit_count = len(collector.hits) + attempt.effects = collector.effects(None) + attempt.trap = attempt.effects["trap"] + attempt.fault = attempt.effects["fault"] + attempt.ok = collector.entry is not None + attempt.trigger = {"surface": surface.name, "id": chosen["id"], "handler": chosen["handler"], "args": args, + "candidates": candidates} + if not attempt.ok and attempt.reason is None: + attempt.reason = f"mailbox command {chosen['id']} did not reach the target within the budget" + finally: + session.close() + return attempt + + +def attempt_stimulus(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str, + record: bool) -> Attempt: + attempt = Attempt(method="stimulus") + if not options.stimulus: + attempt.reason = "no --stimulus script supplied" + return attempt + handle = open_session(options, "stimulus", record) + session = handle.session + attempt.session = handle.path + try: + collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log) + collector.install() + collector.snapshot_before() + import greeneye as ge + + script_path = Path(options.stimulus) + namespace = { + "ge": ge, + "greeneye": ge, + "session": session, + "s": session, + "memory": session.memory, + "symbols": session.symbols, + "recorder": session.recorder, + } + try: + exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace) + except Exception as exc: # a stimulus failure is a finding, not a crash + attempt.reason = f"stimulus raised {type(exc).__name__}: {exc}" + if collector.entry is None: + run_window(session, collector, options.timeout) + attempt.entry = collector.entry + attempt.prior = collector.prior + attempt.dispatch = collector.dispatch + attempt.hit_count = len(collector.hits) + attempt.effects = collector.effects(None) + attempt.trap = attempt.effects["trap"] + attempt.fault = attempt.effects["fault"] + attempt.ok = collector.entry is not None + if not attempt.ok and attempt.reason is None: + attempt.reason = "stimulus completed without reaching the target" + finally: + session.close() + return attempt + + +METHOD_FUNCS = { + "call": attempt_call, + "run": attempt_run, + "trigger": attempt_trigger, + "stimulus": attempt_stimulus, +} + + +# --------------------------------------------------------------------------- +# orchestration, report, probe spec +# --------------------------------------------------------------------------- + + +def attempt_order(options: ReachOptions, target_addr: int, surface: DemoSurface | None) -> list[str]: + if options.method != "auto": + return [options.method] + order = list(AUTO_ORDER) + if not options.stimulus: + order.remove("stimulus") + return order + + +def run_reach(options: ReachOptions) -> dict: + if options.backend == "qemu" and not qemu_available(): + raise GreenEyeError("qemu-system-riscv64 not found; install qemu-system-misc or set GREENEYE_QEMU") + if not os.path.isfile(options.kernel): + raise GreenEyeError(f"kernel not found: {options.kernel}") + if not options.allow_foreign_elf and os.path.abspath(options.kernel) != os.path.abspath(DEMO_ELF): + raise GreenEyeError( + "reachability is restricted to the bundled demo firmware; " + "pass --allow-foreign-elf to run against another ELF under QEMU" + ) + if not options.target: + raise GreenEyeError("--target is required (symbol name or 0x address)") + ensure_dir(options.out) + # Resolve the target once on a short-lived probe session so the report can + # name the address even if every method misses. + target_addr, target_name, symbols = resolve_target(options) + surface = DemoSurface() if symbols.get("gsp_ctrl") is not None else None + attempts: list[Attempt] = [] + for method in attempt_order(options, target_addr, surface): + attempt = METHOD_FUNCS[method](options, surface, target_addr, target_name, bool(options.record)) + attempts.append(attempt) + if attempt.ok and not options.all_methods: + break + success = next((attempt for attempt in attempts if attempt.ok), None) + report = build_report(options, target_addr, target_name, attempts, success) + shutil.rmtree(os.path.join(options.out, ".faults"), ignore_errors=True) + if report["reachable"]: + probe_path = options.emit_probe or os.path.join(options.out, "probe.json") + ensure_dir(os.path.dirname(os.path.abspath(probe_path))) + write_json(probe_path, suggested_probe(options, target_name, success)) + report["probe_spec"] = probe_path + write_json(os.path.join(options.out, "reach-report.json"), report) + with open(os.path.join(options.out, "reach-report.md"), "w", encoding="utf-8") as fh: + fh.write(render_markdown(report, options)) + return report + + +def resolve_target(options: ReachOptions) -> tuple[int, str, SymbolDB]: + """Resolve the target address with a paused session, then close it.""" + session = launch(kernel=options.kernel, backend=options.backend, name="reach-resolve") + try: + symbols = session.symbols + addr = session.resolve(options.target) + symbol = symbols.resolve(addr) + name = symbol.name if symbol else options.target + return addr, name, symbols + finally: + session.close() + + +def suggested_probe(options: ReachOptions, target_name: str, success: Attempt | None) -> dict: + method = success.method if success is not None else options.method + spec = { + "format": PROBE_FORMAT, + "target": target_name, + "kernel": options.kernel, + "method": method, + "args": list(options.args), + "arg_count": options.arg_count, + "arg_fill": options.arg_fill, + "timeout": options.timeout, + "settle": options.settle, + "stimulus": options.stimulus, + "trigger": success.trigger if success is not None and success.method == "trigger" else None, + "watch": options.watch, + } + return spec + + +def build_report(options: ReachOptions, target_addr: int, target_name: str, attempts: list[Attempt], + success: Attempt | None) -> dict: + reachable = success is not None + entry = success.entry if success is not None else None + prior = success.prior if success is not None else [] + return { + "format": FORMAT, + "tool": "harness/reach.py", + "target": target_name, + "target_addr": target_addr, + "kernel": options.kernel, + "kernel_sha256": sha256_file(options.kernel) if os.path.isfile(options.kernel) else None, + "backend": options.backend, + "reachable": reachable, + "method": success.method if success is not None else None, + "methods_tried": [attempt.summary() for attempt in attempts], + "entry": entry, + "prior_events": prior, + "dispatch": success.dispatch if success is not None else [], + "required_args": required_args(entry, success), + "effects": success.effects if success is not None else {}, + "trigger": success.trigger if success is not None else None, + "trace_window": trace_window(success), + "session": success.session if success is not None else next( + (attempt.session for attempt in attempts if attempt.session), None + ), + "probe_spec": None, + "reproduce": reproduce_command(options), + "reason": None if reachable else (attempts[-1].reason if attempts else "no method attempted"), + } + + +def required_args(entry: dict | None, success: Attempt | None) -> dict | None: + if entry is None: + return None + caller = entry.get("ra", 0) + dispatch_args = entry.get("dispatch_args") + if dispatch_args: + return { + "source": "firmware-dispatch", + "cmd": dispatch_args[0], + "a0": dispatch_args[1] if len(dispatch_args) > 1 else None, + "a1": dispatch_args[2] if len(dispatch_args) > 2 else None, + "a2": dispatch_args[3] if len(dispatch_args) > 3 else None, + "handler_entry_args": entry.get("args"), + "caller": f"{caller:#x}" if caller else None, + } + args = entry.get("args") or [] + return { + "source": "call-args" if success is not None and success.method == "call" else "firmware-dispatch", + "a0": args[0] if len(args) > 0 else None, + "a1": args[1] if len(args) > 1 else None, + "a2": args[2] if len(args) > 2 else None, + "a3": args[3] if len(args) > 3 else None, + "caller": f"{caller:#x}" if caller else None, + } + + +def trace_window(success: Attempt | None) -> dict | None: + if success is None: + return None + counts = Counter(item.get("stage", "hook") for item in success.prior) + window = {"hook_counts": dict(sorted(counts.items())), "prior": success.prior} + if success.session and os.path.isdir(success.session): + try: + events = load_events(success.session) + except Exception: + return window + window["session_event_counts"] = dict(sorted(Counter(event.get("type", "?") for event in events).items())) + return window + + +def render_markdown(report: dict, options: ReachOptions) -> str: + lines = [ + "# green-eye reachability report", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/reach.py`.", + "", + f"* format: `{FORMAT}`", + f"* target: `{report['target']}` @ `{report['target_addr']:#x}`", + f"* kernel: `{report['kernel']}` (sha256 `{(report['kernel_sha256'] or '')[:16]}`)", + f"* backend: `{report['backend']}`", + f"* reachable: **{'yes' if report['reachable'] else 'no'}**" + + (f" via `{report['method']}`" if report["reachable"] else ""), + "", + "## Methods tried", + "", + ] + lines += _table( + ["method", "result", "hits", "reason"], + [ + [ + attempt["method"], + "reachable" if attempt["ok"] else "miss", + attempt["hit_count"], + attempt["reason"] or "-", + ] + for attempt in report["methods_tried"] + ], + ) + if report["reachable"]: + entry = report["entry"] or {} + args = entry.get("args") or [] + req = report["required_args"] or {} + lines += ["", "## Entry state", ""] + lines += _table( + ["field", "value"], + [ + ["symbol", entry.get("symbol")], + ["pc", f"`{entry.get('pc', 0):#x}`"], + ["args a0..a3", ", ".join(f"`{value:#x}`" for value in args[:4]) or "-"], + ["ra", f"`{entry.get('ra', 0):#x}`"], + ["sp", f"`{entry.get('sp', 0):#x}`"], + ], + ) + lines += ["", "## Required args discovered", ""] + lines += _table( + ["source", "cmd", "a0", "a1", "a2", "a3", "caller"], + [[ + req.get("source"), + f"`{req['cmd']:#x}`" if req.get("cmd") is not None else "-", + *[f"`{req[key]:#x}`" if req.get(key) is not None else "-" for key in ("a0", "a1", "a2", "a3")], + f"`{req['caller']}`" if req.get("caller") else "-", + ]], + ) + if report.get("trigger"): + trigger = report["trigger"] + lines += ["", "## Trigger", ""] + lines += _table( + ["surface", "command id", "handler", "args"], + [[trigger["surface"], trigger["id"], trigger["handler"], + ", ".join(f"`{value:#x}`" for value in trigger["args"])]], + ) + lines += ["", "## Prior events (hooks that led here)", ""] + if report["prior_events"]: + lines += _table( + ["#", "stage", "symbol", "addr", "args"], + [ + [index, item.get("stage"), item.get("symbol"), f"`{item.get('addr', 0):#x}`", + ", ".join(f"`{value:#x}`" for value in item.get("args", []))] + for index, item in enumerate(report["prior_events"]) + ], + ) + else: + lines += ["_none: the target was the first hooked event_"] + lines += ["", "## Effects after resume", ""] + effects = report["effects"] + changed = effects.get("changed") or [] + lines += _table( + ["field", "before", "after"], + [[row["field"], _short(row["before"]), _short(row["after"])] for row in changed] or [["-", "-", "-"]], + ) + lines += ["", f"* return value: {_hex(effects.get('retval'))}", + f"* trap (ebreak) observed: {effects.get('trap')}", + f"* fault observed: {effects.get('fault')}", + f"* watchpoint stores: {len(effects.get('watch') or [])}"] + window = report.get("trace_window") or {} + if window.get("session_event_counts"): + lines += ["", "## Trace window (recorded session)", "", + f"`{json.dumps(window['session_event_counts'], sort_keys=True)}`"] + else: + lines += ["", "## Not reached", "", + f"* reason: {report.get('reason') or 'unknown'}", + "* absence of a hit is evidence about this stimulus at this fidelity, not proof of dead code."] + lines += ["", "## Reproduce", "", "```bash", report["reproduce"], "```"] + if report.get("probe_spec"): + lines += ["", f"* repeatable probe spec: `{report['probe_spec']}`"] + lines += [""] + return "\n".join(lines) + + +def _short(value) -> str: + text = json.dumps(value, sort_keys=True) + return f"`{text[:48]}`" if len(text) <= 48 else f"`{text[:47]}...`" + + +def _hex(value) -> str: + return f"`{value:#x}`" if isinstance(value, int) else "-" + + +def _table(headers: list[str], rows: list[list]) -> list[str]: + lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] + for row in rows: + lines.append("| " + " | ".join(str(cell) for cell in row) + " |") + return lines + + +def reproduce_command(options: ReachOptions) -> str: + args = f" --args {','.join(hex(value) for value in options.args)}" if options.args else "" + arg_count = f" --arg-count {options.arg_count}" if options.arg_count else "" + stimulus = f" --stimulus {options.stimulus}" if options.stimulus else "" + methods = " --all-methods" if options.all_methods else "" + foreign = " --allow-foreign-elf" if options.allow_foreign_elf else "" + record = f" --record {options.record}" if options.record else "" + return ( + f"python3 harness/reach.py --target {options.target} --method {options.method} " + f"--kernel {options.kernel} --out {options.out} --timeout {options.timeout} " + f"--settle {options.settle}{args}{arg_count}{stimulus}{methods}{record}{foreign}" + ) + + +def write_json(path: str, payload: dict) -> str: + ensure_dir(os.path.dirname(path)) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +# --------------------------------------------------------------------------- +# probe spec loading +# --------------------------------------------------------------------------- + + +def load_probe_spec(path: str) -> dict: + with open(path, "r", encoding="utf-8") as fh: + spec = json.load(fh) + if not isinstance(spec, dict) or "target" not in spec: + raise GreenEyeError(f"{path}: reachability probe spec must be an object with a 'target'") + return spec + + +def apply_probe_spec(options: ReachOptions, spec: dict) -> None: + if not options.target: + options.target = str(spec.get("target", "")) + if spec.get("kernel") and os.path.abspath(options.kernel) == os.path.abspath(DEMO_ELF): + options.kernel = os.path.abspath(str(spec["kernel"])) + for attr, key in ( + ("method", "method"), + ("stimulus", "stimulus"), + ): + if getattr(options, attr) in (None, "", "auto") and spec.get(key): + setattr(options, attr, spec[key]) + if not options.args and spec.get("args"): + options.args = [int(value) for value in spec["args"]] + if options.arg_count == 0 and spec.get("arg_count"): + options.arg_count = int(spec["arg_count"]) + if options.arg_fill == "zero" and spec.get("arg_fill"): + options.arg_fill = str(spec["arg_fill"]) + if options.timeout == 0.5 and spec.get("timeout"): + options.timeout = float(spec["timeout"]) + if options.settle == 2.0 and spec.get("settle"): + options.settle = float(spec["settle"]) + if not options.watch and spec.get("watch"): + options.watch = list(spec["watch"]) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def parse_args_list(text: str) -> list[int]: + values: list[int] = [] + for part in text.replace(" ", ",").split(","): + if part: + values.append(int(part, 0)) + return values + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="reach", description=__doc__.splitlines()[0]) + parser.add_argument("--target", help="symbol name or 0x address to reach") + parser.add_argument("--kernel", default=DEMO_ELF, help="demo/re-hosted ELF (default: bundled demo)") + parser.add_argument("--out", default=None, help="output directory (default: dumps/reach-<target>)") + parser.add_argument( + "--method", + choices=("auto", "stimulus", "trigger", "run", "call"), + default="auto", + help="methods to attempt; auto tries stimulus/trigger/run/call and stops at the first hit", + ) + parser.add_argument("--all-methods", action="store_true", help="attempt every method, not just until the first hit") + parser.add_argument("--args", type=parse_args_list, default=[], help="comma-separated integer args (a0..a7)") + parser.add_argument("--arg-count", type=int, default=0, help="auto-fill a0..a(n-1) when --args is short") + parser.add_argument("--arg-fill", choices=("zero", "sentinel"), default="zero", help="auto-fill value policy") + parser.add_argument("--timeout", type=float, default=0.5, help="per-attempt budget (seconds)") + parser.add_argument("--settle", type=float, default=2.0, help="self-test settle budget before trigger") + parser.add_argument("--stimulus", help="green-eye script to run under instrumentation") + parser.add_argument("--probe", help="JSON probe spec to replay (see --emit-probe)") + parser.add_argument("--emit-probe", help="where to write the repeatable probe spec (default <out>/probe.json)") + parser.add_argument("--record", nargs="?", const=True, default=None, help="record attempts under DIR/<method>") + parser.add_argument("--backend", choices=("qemu", "native"), default="qemu") + parser.add_argument("--max-prior", type=int, default=12, help="prior hook events kept in the report") + parser.add_argument("--allow-foreign-elf", action="store_true", help="permit a non-demo ELF under QEMU") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.probe: + try: + spec = load_probe_spec(args.probe) + except (OSError, ValueError, GreenEyeError) as exc: + print(f"reach error: {exc}", file=sys.stderr) + return 2 + else: + spec = {} + target_hint = args.target or str(spec.get("target", "target")) + out = args.out or os.path.join("dumps", "reach-" + _safe_name(target_hint)) + if isinstance(args.record, str): + record_base: str | None = os.path.abspath(args.record) + elif args.record: + record_base = os.path.join(os.path.abspath(out), "sessions") + else: + record_base = None + options = ReachOptions( + kernel=os.path.abspath(args.kernel), + out=os.path.abspath(out), + target=args.target or "", + method=args.method, + args=args.args, + arg_count=args.arg_count, + arg_fill=args.arg_fill, + timeout=args.timeout, + settle=args.settle, + stimulus=args.stimulus, + probe=args.probe, + emit_probe=os.path.abspath(args.emit_probe) if args.emit_probe else None, + record=record_base, + backend=args.backend, + max_prior=args.max_prior, + all_methods=args.all_methods, + allow_foreign_elf=args.allow_foreign_elf, + ) + if spec: + apply_probe_spec(options, spec) + if not options.target: + parser.error("--target is required (or provide a probe spec with 'target')") + try: + report = run_reach(options) + except GreenEyeError as exc: + print(f"reach error: {exc}", file=sys.stderr) + return 2 + if report["reachable"]: + print(f"[reach] {report['target']} reachable via {report['method']} ({os.path.join(options.out, 'reach-report.md')})") + else: + print(f"[reach] {report['target']} not reached: {report['reason']}") + return 0 if report["reachable"] else 1 + + +def _safe_name(text: str) -> str: + return "".join(ch if ch.isalnum() or ch in "-_." else "-" for ch in text) + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md b/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md new file mode 100644 index 0000000..e1a450c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md @@ -0,0 +1,83 @@ +# green-eye value-validation demos + +Runnable demonstrations that tie green-eye tool capabilities to **real** +Open-IOV scenarios (cited from `Handoff-vGPU-IO.md` and the project docs) and +measure the time-to-value against the static/Ghidra-only baseline. + +Each demo: + +* names the real scenario (handoff PART/addendum + date) and what it cost; +* prints the question, the green-eye commands used, the answer, the evidence + artifact and the elapsed wall-clock; +* writes a machine-readable `result.json` under `artifacts/<demo>/`; +* labels what is **measured on the analog** vs **projected for the real + case**; device-model-blocked scenarios say so explicitly. + +> green-eye is development-only tooling (docs/00); these demos never touch +> hardware, a guest or a kernel module. The real `.fwimage` cannot boot +> without the H2 device model (docs/06 §3), so every demo is honest about +> being a *shape* reproduction on the bundled `uproc_demo` firmware + native +> RV64 simulator / QEMU. + +## Run + +From the tool root (`open-iov/debug/risc-v-gsp-rm-vm`): + +```bash +make demo # once: build firmware/uproc_demo/build/uproc_demo.elf +make sdk # optional (s06): builds sdk/build/injected_demo.elf + +python3 harness/value_demos/run_all.py # all 10, results JSON +python3 harness/value_demos/run_all.py --only s01,s04,s05 # subset +python3 harness/value_demos/s01_reply_consumer.py # one demo, verbose + +VALUE_DEMO_BACKEND=qemu python3 harness/value_demos/run_all.py # force QEMU +``` + +Defaults: native backend (deterministic, in-process) for s01–s08; QEMU for +s09/s10 (with a native fallback if QEMU is absent). `VALUE_DEMO_BACKEND` +overrides all of them. No hardware is required. + +The aggregate lands in `artifacts/results.json`; each demo writes under +`artifacts/<name>/` (results, sessions, TSVs, reports). The runner enforces a +per-demo timeout and the full suite runs in ~18 s on the reference host +(budget: each demo < ~30 s, suite < ~3 min). + +## Demos + +| # | Script | Real scenario (source, date) | Static/Ghidra-only baseline | Analog answer | +|---|---|---|---|---| +| s01 | `s01_reply_consumer.py` | 0x59 RM-init chase; fn72/fn73 hidden error (PART 91 + ADDENDA 2–7, 2026-10-01) | cycles 2–13, ~5 h, 10+ supervised boots; reply was a generic `0x59` | hook dispatcher + watch `result`: every command consumed and its exact reply decoded (1.14 s) | +| s02 | `s02_counter_publication.py` | 20 GR per-channel counters; only 4/20 moved (PARTs 46–64, 2026-09-24…29) | ~5 days; slot layout re-derived per boot; ~2–3 s pin vs ~2 s deadline | 20/20 slots published, each store PC captured (1.34 s) | +| s03 | `s03_unknown_command.py` | unknown/new control discovery; injected slots 16–23 (docs/19, 2026-10-02) | read the indirect dispatch table per stripped image | 24 IDs swept + classified: 6 known, 8 injected-slot, 10 default (5.96 s) | +| s04 | `s04_run_until_function.py` | RM-client ladder reachability per rung (PART 91 ADDENDUM 9 cycles 18–25, 2026-10-01) | ~1 h 41 m / 8 supervised cycles | natural run / timeout / trigger / direct call in one session (0.24 s) | +| s05 | `s05_snapshot_diff.py` | fn72 identity fix + 1656-B static-config reply (PART 91 ADDENDUM 6/7, 2026-10-01) | malformed field suspected statically; a live boot per fix iteration | every changed byte attributed to a named schema field (0.50 s) | +| s06 | `s06_inject_service.py` | P6 service wiring / guest CE copies (PART 92 ADDENDA 5–11, 2026-10-02) | kernel rebuild + supervised cycle (~5–8 min) per rung | service blob loaded, registered, invoked both routes (0.63 s) | +| s07 | `s07_version_diff.py` | 570.144 vs 570.211 pairing disproof; semantic diff (PART 91 cycles 2–3, 2026-10-01) | one full boot per version + manual log compare | same stimulus on two builds; exactly one reply field moved (1.14 s) | +| s08 | `s08_mutation_triage.py` | error-path corpus / fail-closed rules (R5; docs/19 §3.4, docs/18 §2) | hand-derived asserts and NOCAT records across cycles | 7 seeded mutations classified ok/fault/no-progress (1.82 s) | +| s09 | `s09_grabbing_a_ghidra_address.py` | static/dynamic join; GPFIFO GET fault `0x40000` vs `0x101000` (PART 91 ADDENDUM 9 cycle 24) | hypothesis unproven until the decoded status-queue fault (8 cycles) | Ghidra addr + bias = runtime; hook hit + `greeneye_hits.tsv` (0.72 s) | +| s10 | `s10_full_session_evidence.py` | evidence pack for a milestone (docs/14 §6; cycles 11–13, 2026-10-01) | hashes/sessions/logs/compare gathered by hand per cycle | one run → analyze + compare + hits + manifest + M1–M6 (4.92 s) | + +Times are the measured green-eye path on the reference host (native for +s01–s08, QEMU for s09/s10); see `artifacts/results.json` for a fresh run. + +## Projected vs measured + +* **Measured:** every analog result, elapsed time, session size, trace + slowdown and slot/reply count in `result.json`. +* **Projected (labelled `projected` in each result):** the real-case time + saving. No demo claims a real-GSP result the simulator cannot reach. +* **Blocked (`blocked_on_device_model: true`):** all real-firmware claims + wait on the H2.2 device model; the workflow is demonstrated on the analog. + +## Tests + +```bash +python3 -m unittest tests.test_value_demos -v # 5 tests, ~5 s +``` + +The suite runs the fastest demos end-to-end as subprocesses, asserts their +answers and checks `run_all.py` emits `artifacts/results.json`. It skips +cleanly when QEMU or the demo firmware is absent. See +[`docs/25-VALUE-VALIDATION.md`](../../docs/25-VALUE-VALIDATION.md) for the +methodology, overhead analysis and scenario representativeness argument. diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py new file mode 100644 index 0000000..87cabe7 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py @@ -0,0 +1 @@ +"""green-eye value-validation demos (see docs/25-VALUE-VALIDATION.md).""" diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py new file mode 100644 index 0000000..f70ef1f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""Shared plumbing for the green-eye value-validation demos. + +Every demo maps one *real* Open-IOV scenario (cited from the handoffs) onto a +runnable demonstration on the bundled demo firmware / native RV64 simulator, +prints the question, the green-eye commands, the answer, the evidence artifact +and the measured wall-clock, and emits one machine-readable ``RESULT_JSON`` +line so ``run_all.py`` can aggregate. + +Stdlib only; no hardware; runs from the tool root +(``open-iov/debug/risc-v-gsp-rm-vm``). +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(os.path.dirname(HERE)) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +ARTIFACT_ROOT = os.path.join(HERE, "artifacts") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +SCHEMA_JSON = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json") +SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") + +CTRL_MAGIC = 0x47535031 +CTRL_RPC_COUNT = 0x10 +CTRL_LAST_RPC = 0x18 +CTRL_MAILBOX = 0x20 +CTRL_ARG0 = 0x28 +CTRL_ARG1 = 0x30 +CTRL_ARG2 = 0x38 +CTRL_RESULT = 0x40 +CTRL_BUF = 0x48 +CTRL_SIZE = 0x190 +PING_XOR = 0xBEEF +INJECT_CMD_BASE = 16 +SERVICE_SLOT = 6 + +#: Per-demo wall-clock ceiling the runner enforces (subprocess timeout). +DEMO_TIMEOUT_S = 45.0 +#: Runner ceiling so the whole suite stays well under ~3 min. +SUITE_BUDGET_S = 180.0 + + +class DemoError(RuntimeError): + """A demo precondition failed in a way the caller should report, not mask.""" + + +def qemu_available() -> bool: + return bool(shutil.which("qemu-system-riscv64")) + + +def demo_available() -> bool: + return os.path.isfile(DEMO_ELF) + + +def choose_backend(preferred: str = "native") -> str: + """Environment override wins; QEMU is used only when preferred and present.""" + env = os.environ.get("VALUE_DEMO_BACKEND") + if env: + return env + if preferred == "qemu": + return "qemu" if qemu_available() else "native" + return "native" + + +def sha256_file(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() + + +class Stopwatch: + def __init__(self) -> None: + self.t0 = time.monotonic() + self.elapsed = 0.0 + + def stop(self) -> float: + self.elapsed = time.monotonic() - self.t0 + return self.elapsed + + +def artifact_dir(name: str) -> str: + path = os.path.join(ARTIFACT_ROOT, name) + os.makedirs(path, exist_ok=True) + return path + + +def write_json(path: str, payload) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +def write_text(path: str, text: str) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def load_schema(ge): + with open(SCHEMA_JSON, "r", encoding="utf-8") as fh: + return ge.schema_from_json(json.load(fh)) + + +def decode_ctrl(ge, data: bytes, schema=None) -> dict: + schema = schema or load_schema(ge) + out = schema.decode(data, base=0) + if isinstance(out.get("buf"), (bytes, bytearray)): + out["buf"] = bytes(out["buf"]).split(b"\x00", 1)[0].decode("utf-8", "replace") + return out + + +def short_hex(value) -> str: + if value is None: + return "None" + if isinstance(value, str): + return value + return f"{value:#x}" if isinstance(value, int) else str(value) + + +def render_banner(demo: str, question: str, scenario: str, baseline: str, commands) -> None: + print(f"=== {demo} ===") + print(f"real scenario : {scenario}") + print(f"baseline : {baseline}") + print(f"question : {question}") + print("commands :") + for command in commands: + print(f" - {command}") + + +def finish(result: dict, artifact: str, elapsed: float) -> dict: + """Persist the per-demo result and print the human + machine summary.""" + result["elapsed_s"] = round(elapsed, 4) + result["artifact"] = os.path.relpath(artifact, TOOL_ROOT) if artifact else None + result.setdefault("success", True) + if artifact: + write_json(os.path.join(artifact, "result.json"), result) + print(f"answer : {result.get('answer')}") + print(f"evidence : {result.get('artifact')}") + print(f"analog time : {elapsed:.3f} s ({result.get('backend')} backend)") + if result.get("real_baseline"): + print(f"real baseline : {result['real_baseline']}") + if result.get("projected"): + print(f"projected : {result['projected']}") + if result.get("blocked_on_device_model"): + print("blocked : real case blocked on H2 device model; workflow shown on analog") + if not result.get("success", True): + print(f"status : FAILED - {result.get('failure', 'unknown')}") + print("RESULT_JSON: " + json.dumps(result, sort_keys=True)) + sys.stdout.flush() + return result + + +def timed_demo(fn, artifact: str) -> dict: + watch = Stopwatch() + try: + result = fn() + except Exception as exc: # report, never hide + import traceback + + result = { + "success": False, + "failure": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc()[-2000:], + } + return finish(result, artifact, watch.stop()) + + +def scenario( + *, + ids, + baseline, + projection, + blocked=False, + device=None, +) -> dict: + return { + "citation_ids": ids, + "static_baseline": baseline, + "projection": projection, + "blocked_on_device_model": blocked, + "device_model_needed": device, + } diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py new file mode 100644 index 0000000..808c227 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Run every value-validation demo and emit one machine-readable results JSON. + +Usage (from the tool root, ``open-iov/debug/risc-v-gsp-rm-vm``): + + python3 harness/value_demos/run_all.py + python3 harness/value_demos/run_all.py --only s01,s03,s05,s09 + VALUE_DEMO_BACKEND=qemu python3 harness/value_demos/run_all.py + +Each demo runs in its own subprocess with a hard timeout, so one wedged demo +cannot take the suite down. The aggregate lands in +``harness/value_demos/artifacts/results.json``. +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(os.path.dirname(HERE)) +sys.path.insert(0, HERE) + +from _common import ARTIFACT_ROOT, SUITE_BUDGET_S, write_json # noqa: E402 + +DEMOS = [ + ("s01", "s01_reply_consumer.py"), + ("s02", "s02_counter_publication.py"), + ("s03", "s03_unknown_command.py"), + ("s04", "s04_run_until_function.py"), + ("s05", "s05_snapshot_diff.py"), + ("s06", "s06_inject_service.py"), + ("s07", "s07_version_diff.py"), + ("s08", "s08_mutation_triage.py"), + ("s09", "s09_grabbing_a_ghidra_address.py"), + ("s10", "s10_full_session_evidence.py"), +] + + +def _run_one(key: str, script: str, timeout: float) -> dict: + path = os.path.join(HERE, script) + start = time.monotonic() + try: + proc = subprocess.run( + [sys.executable, path], + cwd=TOOL_ROOT, + capture_output=True, + text=True, + timeout=timeout, + ) + stdout, stderr, returncode = proc.stdout, proc.stderr, proc.returncode + except subprocess.TimeoutExpired as exc: + stdout = (exc.stdout or b"").decode("utf-8", "replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") + stderr = (exc.stderr or b"").decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") + returncode = -9 + elapsed = time.monotonic() - start + result = None + for line in reversed(stdout.splitlines()): + if line.startswith("RESULT_JSON: "): + try: + result = json.loads(line[len("RESULT_JSON: ") :]) + except json.JSONDecodeError: + result = None + break + if result is None: + result = { + "success": False, + "failure": f"no RESULT_JSON (exit {returncode})", + "stderr_tail": stderr[-800:], + } + result["demo_key"] = key + result["runner_elapsed_s"] = round(elapsed, 4) + result["exit_code"] = returncode + return result + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--only", default="", help="comma-separated demo keys (e.g. s01,s03)") + parser.add_argument("--timeout", type=float, default=45.0, help="per-demo timeout seconds") + parser.add_argument("--out", default=os.path.join(ARTIFACT_ROOT, "results.json")) + args = parser.parse_args() + + selected = [item for item in DEMOS if not args.only or item[0] in args.only.split(",")] + if args.only and not selected: + print(f"no demos match --only {args.only!r}", file=sys.stderr) + return 2 + + started = time.monotonic() + results = [] + for key, script in selected: + print(f"--- running {script} ---", flush=True) + result = _run_one(key, script, args.timeout) + results.append(result) + status = "PASS" if result.get("success") else "FAIL" + print(f"--- {key}: {status} ({result['runner_elapsed_s']:.2f}s) ---", flush=True) + if time.monotonic() - started > SUITE_BUDGET_S: + print(f"[run_all] suite budget {SUITE_BUDGET_S:.0f}s exceeded; continuing", flush=True) + + total = time.monotonic() - started + passed = [r for r in results if r.get("success")] + payload = { + "format": "greeneye-value-demos-results/1", + "tool": "harness/value_demos/run_all.py", + "backend_env": os.environ.get("VALUE_DEMO_BACKEND", "native (default)"), + "demos_selected": [k for k, _ in selected], + "demos_passed": [r["demo_key"] for r in passed], + "demos_failed": [r["demo_key"] for r in results if not r.get("success")], + "total_elapsed_s": round(total, 3), + "measured_elapsed_s": round(sum(r["runner_elapsed_s"] for r in results), 3), + "results": results, + } + write_json(args.out, payload) + print(f"\n{len(passed)}/{len(results)} demos passed in {total:.1f}s; results: {args.out}") + print(f"RESULT_JSON: {json.dumps({k: payload[k] for k in ('format', 'demos_passed', 'demos_failed', 'total_elapsed_s')}, sort_keys=True)}") + return 0 if len(passed) == len(results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py new file mode 100644 index 0000000..b86f272 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""s01 - reply consumer: did the server consume the command, and what came back? + +Real scenario (Handoff-vGPU-IO.md, PART 91 + ADDENDA 2-7, 2026-10-01): +the physical backend booted the real GSP-RM and sent the pre-boot +``GSP_SET_SYSTEM_INFO`` (fn72) / ``SET_REGISTRY`` (fn73) RPCs; the only reply +the host could see was a generic ``GSP_INIT_DONE rpc_result=0x59`` +(``NV_ERR_OPERATING_SYSTEM``) wrapped by ``RmStartup``. Cycles 2-13 +(~11:40-16:52, ten supervised hardware cycles) were spent chasing which +collector step consumed which field because the RM log ring was never mapped +(LOGRM id bug) and the failure was un-journaled. Static/Ghidra-only work +(REPORT3-10) produced hypotheses (display, devid, chipset, ACPI) that needed a +live ``0x59``/success reply to accept or reject. + +Analog: hook the demo dispatcher, watch the control block's ``result`` field, +post known/unknown/fault commands and decode the exact reply per command. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s01_reply_consumer") + commands = [ + 'ge.launch(kernel=DEMO_ELF, backend="%s")' % backend, + 's.hook("rpc_dispatch", on_enter=record cmd/args)', + "s.watch(gsp_ctrl+0x40, size=8, kind='w') # result", + "s.memory.write_u64(ctrl+0x20, cmd); s.run(duration=...) # mailbox post", + "s.memory.read_u64(ctrl+0x40) # decode reply", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3", + "PART 91 ADDENDUM 2-7 (2026-10-01): NOCAT, fn72/fn73, LOGRM id, identity", + ], + baseline=( + "cycles 2-13, ~5 h wall-clock and 10+ supervised hardware boots; " + "each candidate fix cost one full boot cycle (~5-6 min) and the " + "only visible reply was a generic 0x59" + ), + projection=( + "real GSP-RM reply consumption is blocked on the H2 device model; " + "on the analog the same observe->decode step is one hook + one " + "watchpoint per command" + ), + blocked=True, + device="H2.2 QEMU device model (RM message loop / real .fwimage)", + ) + C.render_banner( + "s01_reply_consumer", + "When the host posts a command, did the dispatcher consume it and what exact reply came back?", + "0x59 RM-init chase: generic reply hid which collector step consumed which field", + scenario["static_baseline"], + commands, + ) + + cases = [ + {"cmd": 1, "arg0": 0x1234, "label": "PING"}, + {"cmd": 7, "arg0": 0, "label": "unknown-id"}, + {"cmd": 6, "arg0": 0, "label": "FAULT"}, + ] + replies = [] + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s01") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatches = [] + stores = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit: dispatches.append( + {"cmd": hit.arg(0), "arg0": hit.arg(1), "arg1": hit.arg(2), "arg2": hit.arg(3)} + ), + ) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stores.append(hit.pc)) + session.run(duration=0.3) + for case in cases: + before_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + session.memory.write_u64(ctrl + C.CTRL_ARG0, case["arg0"]) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, case["cmd"]) + session.run(duration=0.25) + after_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + expected = { + 1: (0x1234 ^ C.PING_XOR), + 7: (~7) & 0xFFFFFFFFFFFFFFFF, + 6: None, + }[case["cmd"]] + replies.append( + { + "label": case["label"], + "cmd": case["cmd"], + "arg0": case["arg0"], + "consumed": after_count > before_count, + "result": result, + "expected": expected, + "result_ok": (expected is None) or (result == expected), + "rpc_count_delta": after_count - before_count, + } + ) + if case["cmd"] == 6: + # ebreak trap: the dispatcher's post-handler stores never run. + session.step(1) + known = replies[0] + unknown = replies[1] + fault = replies[2] + payload = { + "success": known["result_ok"] and unknown["result_ok"] and not fault["consumed"], + "backend": backend, + "firmware": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT), + "firmware_sha256": C.sha256_file(C.DEMO_ELF), + "dispatches": dispatches, + "reply_store_pcs": stores, + "replies": replies, + "question": "consumed and decoded every posted command", + "answer": ( + f"PING consumed, reply {known['result']:#x} == arg0 ^ 0xbeef " + f"(expected {known['expected']:#x}); unknown id returns ~cmd " + f"({unknown['result']:#x}); FAULT consumed={fault['consumed']} " + f"(trap before the reply store) - the exact first-error visibility " + "gap the 0x59 chase lacked" + ), + "success_criteria": { + "pass": "3/3 dispatches observed; PING/unknown replies match the ABI; FAULT traps before result", + "fail": "a posted command is not consumed or a reply disagrees with the ABI", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "reply.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s01_reply_consumer"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py new file mode 100644 index 0000000..690f60f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""s02 - counter publication: which of N slots advanced, and when? + +Real scenario (Handoff-vGPU-IO.md PART 46-64, 2026-09-24..2026-09-29): +``cuCtxCreate`` waits on 20 GR per-channel completion counters. The host saw +the client's releases but could not translate the counter page; for many boots +only 4 of 20 client-visible slots ever moved (PART 49). The exact slot layout +(page VA ``0x20400f000``, slots at ``+0xec0``, 16-byte stride) was only proven +by a raw live page dump (PART 64, 2026-09-29 ~16:55), and the host raced a +~2-3 s pin path against the client's ~2 s GR-phase deadline. Static reading +could not tell which slots were published: the page's GPA changed every boot +and stale translations pointed at the previous client's page (PART 51). + +Analog: post 20 sequential commands and watch the demo's published result +field; prove every slot was observed, with the publish PC recorded per slot. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +SLOTS = 20 + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s02_counter_publication") + commands = [ + 'ge.launch(kernel=DEMO_ELF, backend="%s")' % backend, + "s.watch(gsp_ctrl+0x40, size=8, kind='w', callback=record pc) # publication", + "for slot in 1..20: write mailbox cmd=5 (STATE); run(duration=0.05)", + "read gsp_ctrl.result after every slot; compare against the expected stage value", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 49/50 (2026-09-29): only 4/20 counters moved; all 20 reached 1 after the queue fix", + "PART 64 (2026-09-29 ~16:55): raw page dump proves 20 slots at 16-byte stride", + "PART 46/48 (2026-09-29): rescue/sweep latency and tracker waits", + ], + baseline=( + "PARTs 46->64: ~5 days of sessions; PART 49 'only 4 of 20 counters " + "ever moved'; each boot re-derived the slot layout because the " + "counter page GPA changed (PART 51/53)" + ), + projection=( + "on real silicon the 20 slots are per-channel memory pages; the " + "analog publishes 20 logical slots through the same watchpoint " + "shape, one store each" + ), + blocked=True, + device="H2.2 device model + real GSP-RM counter pages", + ) + C.render_banner( + "s02_counter_publication", + "Of the N client-visible slots, which ones advanced, and where was each published?", + "20 GR per-channel completion counters: 4/20 published for days until a live page dump", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s02") + slots = [] + try: + ctrl = session.symbols.lookup("gsp_ctrl") + publishes = [] + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: publishes.append(hit.pc)) + session.run(duration=0.25) + before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + for slot in range(SLOTS): + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 5) # STATE + session.run(duration=0.05) + value = session.memory.read_u64(ctrl + C.CTRL_RESULT) + slots.append( + { + "slot": slot + 1, + "result": value, + "result_hex": f"{value:#x}", + "published": value in (0x10, 0x20, 0x30, 0x40), + } + ) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + published = [s for s in slots if s["published"]] + payload = { + "success": len(published) == SLOTS and (after - before) == SLOTS, + "backend": backend, + "slots_requested": SLOTS, + "slots_published": len(published), + "rpc_count_delta": after - before, + "watch_stores": len(publishes), + "distinct_values": sorted({s["result_hex"] for s in published}), + "per_slot": slots, + "publish_pcs": sorted(set(publishes)), + "question": f"did all {SLOTS} slots publish, and can each publication be attributed?", + "answer": ( + f"{len(published)}/{SLOTS} slots published ({after - before} dispatcher " + f"completions, {len(publishes)} stores to gsp_ctrl.result, values " + f"{sorted({s['result_hex'] for s in published})}); every store PC is captured, " + "exactly the evidence class that took a live 512-byte page dump on the real path" + ), + "success_criteria": { + "pass": f"{SLOTS}/{SLOTS} slots observed with rpc_count advancing once per slot", + "fail": "any slot does not publish or the count delta disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "counters.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s02_counter_publication"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py new file mode 100644 index 0000000..162bc47 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""s03 - unknown command discovery: which IDs are handled, injected, or new? + +Real scenario (docs/19-CONTROL-DISCOVERY.md, measured 2026-10-02): a swept +input surface is how new/undocumented controls are found between firmware +versions. On the demo, IDs 16..23 were *discovered* to fall through the +switch into the injected-service handler table (16..21 alias the built-in +handlers, 22..23 are NULL slots returning ``~id``) - wiring a new image can +add silently. Statically this requires reading an indirect dispatch table +per version; behaviorally it is a bounded sweep. + +Analog: 24 command IDs, zero arguments, deterministic, fresh session per ID +(the same safety strategy as the real discovery tool), classified by hooks on +the dispatcher and every handler-table entry. +""" +from __future__ import annotations + +import json +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault") + + +def _classify(cmd: int, handler_hits, dispatch_seen, result, trap) -> str: + if 16 <= cmd <= 23: + return "injected-slot" + if 1 <= cmd <= 6: + return "known-handler" + if trap: + return "trap" + if result is not None and result == (~cmd) & 0xFFFFFFFFFFFFFFFF: + return "default-path" + return "no-progress" + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s03_unknown_command") + commands = [ + 'for cmd in 1..24: ge.launch(backend="%s"); hook(rpc_dispatch + 6 handlers)' % backend, + "write mailbox cmd with zero args; run bounded; classify", + "record: outcome, handler alias, result, trap signal, rpc_count delta", + ] + scenario = C.scenario( + ids=[ + "docs/19-CONTROL-DISCOVERY.md (2026-10-02): 189-probe demo sweep in ~36 s", + "Handoff-vGPU-IO.md PART 91 (2026-10-01): new controls appear between images (0x59 era)", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 4.1: static diff says code changed, not which input activates it", + ], + baseline=( + "static-only: read the indirect dispatcher table in Ghidra for every " + "image; on the real corpus the demo sweep took ~36 s for 189 probes " + "and still found an undocumented injected-slot class" + ), + projection=( + "the RM surface (control IDs by class, method opcodes) is gated on " + "H1.3/H2.2; the sweep/classification shape demonstrated here is " + "exactly what docs/19 section 5 designs for it" + ), + blocked=True, + device="H1.3/H2.2: RM dispatch table exposed to the simulator", + ) + C.render_banner( + "s03_unknown_command", + "Which command IDs are handled, which route to injected slots, and which are new/default?", + "unknown-control discovery between stripped firmware versions", + scenario["static_baseline"], + commands, + ) + + findings = [] + for cmd in range(1, 25): + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name=f"vd-s03-{cmd:02d}") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatch = [] + handler_hits = [] + signals = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit, d=dispatch: d.append( + {"cmd": hit.arg(0), "arg0": hit.arg(1), "arg1": hit.arg(2), "arg2": hit.arg(3)} + ), + ) + for name in HANDLERS: + try: + session.hook(name, on_enter=lambda hit, n=name: handler_hits.append(n)) + except Exception: + pass + session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None) + session.run(duration=0.15) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, cmd) + session.run(duration=0.15) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + trap = 5 in [s for s in signals if s] + outcome = _classify(cmd, handler_hits, bool(dispatch), result, trap) + findings.append( + { + "cmd": cmd, + "outcome": outcome, + "trap": trap, + "signals": [s for s in signals if s], + "handler_alias": handler_hits[0] if handler_hits else None, + "dispatch_seen": bool(dispatch), + "result": result, + "result_hex": f"{result:#x}", + "expected_default": f"{(~cmd) & 0xFFFFFFFFFFFFFFFF:#x}", + } + ) + finally: + session.close() + + by_class: dict[str, list[int]] = {} + for finding in findings: + by_class.setdefault(finding["outcome"], []).append(finding["cmd"]) + injected = by_class.get("injected-slot", []) + aliases = {f["cmd"]: f["handler_alias"] for f in findings if f["cmd"] in injected} + payload = { + "success": len(injected) == 8 and len(by_class.get("known-handler", [])) == 6, + "backend": backend, + "ids_swept": len(findings), + "classes": {key: {"count": len(v), "ids": v} for key, v in sorted(by_class.items())}, + "injected_aliases": aliases, + "findings": findings, + "question": "enumerate handled / injected / default command space in one bounded sweep", + "answer": ( + f"{len(by_class.get('known-handler', []))} known handlers, " + f"{len(injected)} injected-slot IDs ({injected}), aliases {aliases}, " + f"{len(by_class.get('default-path', []))} default-path IDs; " + "the same shape (sweep + dispatcher/handler hooks) is what found the " + "silent 16..23 wiring the static map did not highlight" + ), + "success_criteria": { + "pass": "IDs 1..6 classified known, 16..23 classified injected-slot, remaining default; cmd 6 traps", + "fail": "an ID is misclassified or a fresh-session probe cannot be replayed", + }, + "real_case": scenario, + } + os.makedirs(out, exist_ok=True) + with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh: + for finding in findings: + fh.write(json.dumps(finding, sort_keys=True) + "\n") + lines = ["# control discovery (analog)", "", f"* backend: {backend}", f"* ids swept: {len(findings)}", ""] + for key, value in sorted(by_class.items()): + lines.append(f"* {key}: {value}") + lines.append("") + C.write_text(os.path.join(out, "control-discovery.md"), "\n".join(lines) + "\n") + C.write_json(os.path.join(out, "discovery.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s03_unknown_command"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py new file mode 100644 index 0000000..a74c83f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""s04 - run-until a function: is this code on the live path, and with what args? + +Real scenario (docs/19-CONTROL-DISCOVERY.md 4.4 and +Handoff-vGPU-IO.md PART 91 ADDENDUM 9, 2026-10-01 cycles 18-25): the +``gsp-phys-rm-call`` RM-client ladder asked, per cycle, whether a symbolized +function had executed under the current stimulus (VAS commit, token, BIND, +fetch, semaphore). Cycles 18->25 (~1 h 41 m, eight supervised boots) walked +the ladder one reachability question at a time; ``reach.py`` answers that +question in ~0.5 s per target, and a ``run_until`` timeout means "not executed +under this stimulus at this fidelity", never "unreachable". + +Analog: run to the self-test's ``rpc_sum``, show ``rpc_fault`` is *not* on the +natural path (timeout), then trigger it through the mailbox and reach it with +the dispatch arguments captured; finally call ``rpc_ping`` directly +(Frida NativeFunction) to show the unit-test route. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + from greeneye.errors import RspTimeout + + backend = C.choose_backend("native") + out = C.artifact_dir("s04_run_until_function") + commands = [ + 's.run_until("rpc_sum", timeout=5.0) # natural path', + 's.run_until("rpc_fault", timeout=0.2) # not on natural path', + 'write mailbox cmd=6; s.run_until("rpc_fault") # trigger route', + 's.call("rpc_ping", (0x41,)) # Frida-like direct call', + ] + scenario = C.scenario( + ids=[ + "docs/19-CONTROL-DISCOVERY.md 4.4 (2026-10-02): reach.py measured ~0.5 s per target", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 9 (2026-10-01): RM-client ladder cycles 18-25", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 5: run_until timeouts are not unreachability", + ], + baseline=( + "cycles 18-25 took ~1 h 41 m of supervised hardware cycles (full " + "boot + RPC ladder per hypothesis) to answer the same class of " + "'did this function execute under this stimulus' question" + ), + projection=( + "real RM ladder reachability needs the live GSP-RM (H1.3/H2.2); " + "run_until/hook/call already work against the re-hosted firmware " + "shape and every probe spec emitted here replays" + ), + blocked=True, + device="H1.3 live RM method entry / H2.2 device model", + ) + C.render_banner( + "s04_run_until_function", + "Is a symbolized function on the live path under this stimulus, what args does it take, and can it be called directly?", + "RM-client ladder: reachability of each rung, one supervised boot per question", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s04") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatch_args = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit: dispatch_args.append( + {"cmd": hit.arg(0), "a0": hit.arg(1), "a1": hit.arg(2), "a2": hit.arg(3)} + ), + ) + natural_hit = session.run_until("rpc_sum", timeout=5.0) + natural = { + "target": "rpc_sum", + "method": "run", + "reachable": True, + "pc": f"{natural_hit.addr:#x}", + "dispatch_args": list(dispatch_args), + } + timeout_hit = None + try: + timeout_hit = session.run_until("rpc_fault", timeout=0.2) + timeout_seen = False + except RspTimeout: + timeout_seen = True + fault_before = { + "target": "rpc_fault", + "method": "run_until", + "reachable": timeout_hit is not None, + "timeout": timeout_seen, + "reading": "not executed under this stimulus at this fidelity (docs/14 section 5)", + } + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 6) + trigger_hit = session.run_until("rpc_fault", timeout=5.0) + triggered = { + "target": "rpc_fault", + "method": "trigger (mailbox cmd=6)", + "reachable": True, + "pc": f"{trigger_hit.addr:#x}", + "dispatch_args": list(dispatch_args[-1:]), + } + direct = session.call("rpc_ping", (0x41,)) + payload = { + "success": natural["reachable"] and fault_before["timeout"] and triggered["reachable"] + and direct == (0x41 ^ C.PING_XOR), + "backend": backend, + "natural_path": natural, + "before_trigger": fault_before, + "after_trigger": triggered, + "direct_call": {"target": "rpc_ping", "args": [0x41], "result": direct, "expected": 0x41 ^ C.PING_XOR}, + "question": "reachability and required arguments per symbolized function", + "answer": ( + f"rpc_sum reached on the self-test path (pc {natural['pc']}); rpc_fault " + f"times out (reachable=False) until mailbox cmd=6 triggers it " + f"(pc {triggered['pc']}, args {triggered['dispatch_args']}); direct call " + f"rpc_ping(0x41)={direct:#x} - three reachability answers, one session" + ), + "success_criteria": { + "pass": "natural target reached, un-triggered target times out, trigger route reaches it, direct call matches", + "fail": "a route reports reachable without a hit, or the direct call disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "reach.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s04_run_until_function"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py new file mode 100644 index 0000000..fd89131 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""s05 - snapshot diff: which control-block fields did one command change? + +Real scenario (Handoff-vGPU-IO.md PART 91 ADDENDUM 6, 2026-10-01 cycle 12/13): +the fn72 ``GSP_SET_SYSTEM_INFO`` payload carried *unpacked* identity fields and +the reference sends raw config-space dwords; the server read the device id +back as zero and only a field-level comparison of the live payload/reply +could confirm the fix ("all zeroed fn72 fields cannot matter to this failure" +was settled by decoding, not by reading). Static review (REVIEW-FN72.md, +cycle 13's 1656-byte static-config reply) still needed a live before/after of +the control block to call the fix verified. + +Analog: snapshot the decoded control block before and after one command, +attribute every changed byte run to its named schema field, and show ECHO's +buffer side-effect separately. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def _fields_hit(schema, diffs) -> list[str]: + names = [] + for diff in diffs: + for field in schema.fields: + start = field.offset + end = field.offset + field.nbytes + if start <= diff.addr < end: + if field.name not in names: + names.append(field.name) + return names + + +def run() -> dict: + import greeneye as ge + from greeneye.memory import Memory + + backend = C.choose_backend("native") + out = C.artifact_dir("s05_snapshot_diff") + commands = [ + 's.snapshot(gsp_ctrl, 0x188, "before")', + "write mailbox cmd=2 (ECHO) with a known string; run", + 's.snapshot(gsp_ctrl, 0x188, "after")', + "Memory.diff(before, after); decode both via gsp_ctrl.schema.json", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 ADDENDUM 6 cycle 12 (2026-10-01): unpacked fn72 identity fields", + "PART 91 ADDENDUM 7 (2026-10-01): 1656-byte static-config reply decoded field-by-field", + "docs/06-ACCURACY.md M6: memory side-effect agreement / snapshot hashes", + ], + baseline=( + "the malformed identity value was suspected statically (REVIEW-FN72) " + "but accepting the fix required a live boot per iteration; ten NOCAT " + "cycles ran before the field-level root cause was visible" + ), + projection=( + "the same before/after diff runs against any firmware region once " + "the H2 device model exposes the real RM control pages" + ), + blocked=True, + device="H2.2 RM control/reply pages for the real GSP-RM", + ) + C.render_banner( + "s05_snapshot_diff", + "Which named control-block fields did this one command change?", + "fn72/static-config field-level validation (identity fix, cycle 12/13)", + scenario["static_baseline"], + commands, + ) + + schema = C.load_schema(ge) + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s05") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.run(duration=0.25) + before = session.snapshot(ctrl, C.CTRL_SIZE, "before-echo") + message = b"hello-green-eye" + # Locate the string literal in the loaded image so the stimulus is + # self-describing (no hardcoded address). + hits = session.memory.search(0x80000000, 0x2000, message) + literal = hits[0] if hits else 0 + session.memory.write_u64(ctrl + C.CTRL_ARG0, literal) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 2) + session.run(duration=0.25) + after = session.snapshot(ctrl, C.CTRL_SIZE, "after-echo") + + before_dec = C.decode_ctrl(ge, before, schema) + after_dec = C.decode_ctrl(ge, after, schema) + diffs = Memory.diff(before, after) + changed_fields = _fields_hit(schema, diffs) + changed_names = [name for name in changed_fields if before_dec.get(name) != after_dec.get(name)] + rendered = Memory.render_diff(diffs, symbols=session.symbols) + payload = { + "success": bool(changed_names) + and set(changed_names) >= {"rpc_count", "last_rpc", "result", "buf"} + and after_dec.get("result") == len(message), + "backend": backend, + "changed_field_names": changed_names, + "changed_byte_runs": len(diffs), + "before": {k: v for k, v in before_dec.items() if k in changed_names}, + "after": {k: v for k, v in after_dec.items() if k in changed_names}, + "rendered_diff": rendered, + "question": "attribute every changed byte in one region to a named field", + "answer": ( + f"ECHO changed {len(changed_names)} named fields {changed_names}; " + f"buf={after_dec.get('buf')!r} result={after_dec.get('result')} " + f"(len={len(message)}); {len(diffs)} byte runs exactly covered by the schema - " + "the field-level 'what did the reply actually consume' check" + ), + "success_criteria": { + "pass": "rpc_count/last_rpc/result/buf change, result equals the ECHO length, no unattributed run", + "fail": "a region changes with no named field, or the ECHO reply length disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "snapshot-diff.json"), payload) + C.write_text( + os.path.join(out, "snapshot-diff.md"), + "# snapshot diff (analog)\n\n" + f"* changed fields: {changed_names}\n" + f"* changed byte runs: {len(diffs)}\n\n" + "```\n" + rendered + "\n```\n", + ) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s05_snapshot_diff"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py new file mode 100644 index 0000000..04dd001 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""s06 - inject a service: add an RM engine without rebuilding the firmware. + +Real scenario (Handoff-vGPU-IO.md PART 92 ADDENDUM 5-11, 2026-10-02): the P6 +work wired guest-originated CE copies through the real engine; every change to +the service/wiring path cost a kernel rebuild plus a supervised physical boot +cycle (P6a/b/c, ~5-8 min each, one device-write experiment per cycle) and was +only accepted after guest smoke ``ALL OK 64/64``. docs/12-INJECTION-SDK.md +defines the in-simulator equivalent: load a position-independent service blob +and register it with the firmware's dispatch table (commands 16..23). + +Analog: load ``sdk/build/injected_demo.elf`` into slot 6, invoke it directly +(Frida NativeFunction) and through the firmware mailbox, and record the +``service_injected`` evidence event. No kernel rebuild, no hardware. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + from greeneye.inject import invoke_remote, invoke_via_firmware, load_service + + backend = C.choose_backend("native") + out = C.artifact_dir("s06_inject_service") + commands = [ + 'load_service(s, "sdk/build/injected_demo.elf", slot=6)', + "invoke_remote(s, svc, cmd=1, a0=0x41) # debugger-side call", + "invoke_via_firmware(s, svc, cmd=1, a0=0x41) # mailbox cmd=16", + "read gsp_ctrl.result / handlers[6]", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 92 ADDENDUM 5-11 (2026-10-02): P6 service/wiring rungs", + "PART 92 ADDENDUM 6 (2026-10-02): p6run7 first real engine copy after service wiring", + "docs/12-INJECTION-SDK.md: blob/ABI contract for injecting RM services", + ], + baseline=( + "each P6 rung was a kernel rebuild plus a supervised physical cycle " + "(~5-8 min boots; P6a->P6c spanned ~02:20-08:30) with fail-safe " + "release discipline after every run" + ), + projection=( + "on real GSP-RM the registry is the RM service table (M5), gated on " + "H2.2; the load/register/invoke/observe shape is identical to the " + "demo handler-table slot" + ), + blocked=True, + device="H2.2 device model + M5 RM service registry", + ) + C.render_banner( + "s06_inject_service", + "Can a new service be loaded, registered and invoked without rebuilding the firmware?", + "guest-originated CE service wiring (P6a-P6c): kernel rebuild + supervised boot per rung", + scenario["static_baseline"], + commands, + ) + + if not os.path.isfile(C.SDK_ELF): + payload = { + "success": False, + "skipped": True, + "failure": "sdk/build/injected_demo.elf not built (run: make sdk)", + "backend": backend, + "real_case": scenario, + } + C.write_json(os.path.join(out, "inject.json"), payload) + return payload + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=os.path.join(out, "session"), name="vd-s06") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.run(duration=0.25) + service = load_service(session, C.SDK_ELF, slot=C.SERVICE_SLOT) + remote_ping = invoke_remote(session, service, 1, 0x41) + remote_sum = invoke_remote(session, service, 2, 0x10, 0x20) + via_fw_ping = invoke_via_firmware(session, service, 1, 0x41) + via_fw_sum = invoke_via_firmware(session, service, 2, 0x10, 0x20) + handlers7 = session.memory.read_u64(ctrl + 0x150 + 8 * C.SERVICE_SLOT) + payload = { + "success": remote_ping == (0x41 ^ C.PING_XOR) and remote_sum == 0x30 + and via_fw_ping == remote_ping and via_fw_sum == remote_sum, + "backend": backend, + "service": service.summary(), + "service_sha256": C.sha256_file(C.SDK_ELF), + "handlers_slot6": f"{handlers7:#x}", + "remote": {"ping": remote_ping, "sum": remote_sum}, + "via_firmware": {"ping": via_fw_ping, "sum": via_fw_sum}, + "mailbox_command": C.INJECT_CMD_BASE + C.SERVICE_SLOT, + "session": os.path.relpath(os.path.join(out, "session"), C.TOOL_ROOT), + "question": "load/register/invoke a service blob end-to-end, both routes", + "answer": ( + f"service loaded at {service.load_addr:#x}, slot {service.slot} " + f"(handlers[{service.slot}]={handlers7:#x}); remote ping={remote_ping:#x} " + f"sum={remote_sum:#x}; via mailbox cmd={C.INJECT_CMD_BASE + C.SERVICE_SLOT} " + f"ping={via_fw_ping:#x} sum={via_fw_sum:#x} - both paths agree, one session" + ), + "success_criteria": { + "pass": "remote and firmware-mailbox invocations return the SDK ABI values and agree", + "fail": "a route disagrees, the slot is not registered, or the blob cannot load", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "inject.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s06_inject_service"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py new file mode 100644 index 0000000..e4748d7 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +"""s07 - version diff: same stimulus against two firmware versions. + +Real scenario (Handoff-vGPU-IO.md PART 91, 2026-10-01 cycles 2-3): the staged +570.211 image and a verified 570.144 image both failed ``GSP_INIT_DONE`` with +``rpc_result=0x59``; disproving the image-pairing hypothesis cost one full +supervised boot cycle per version, and docs/14 section 4.1 exists because +"the code changed" and "the behaviour moved" must be separated before the +emulation is touched. Sorting a new image into the decision table +(static changed x hit/write profile changed) is the H4.1 triage. + +Analog: byte-patch the demo's PING reply constant (0xbeef -> 0xbef0) to make a +labelled "new version", run the *same* recorded stimulus on both, and diff the +sessions and the decoded control block. The answer is one live-path field. +""" +from __future__ import annotations + +import os +import shutil + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +#: addiw a1, a1, -0x111 (the 0xbeef PING key) -> -0x110 (0xbef0) +ADDW_OLD = bytes.fromhex("9b85f5ee") +ADDW_NEW = bytes.fromhex("9b8505ef") +STRING_OLD = b"hello-green-eye" +STRING_NEW = b"hello-green-eyf" + + +def _make_variant(src: str, dst: str) -> dict: + with open(src, "rb") as fh: + data = bytearray(fh.read()) + count = data.count(ADDW_OLD) + if count == 1: + offset = data.find(ADDW_OLD) + old_bytes = bytes(data[offset : offset + 4]) + data[offset : offset + 4] = ADDW_NEW + kind = "ping-key-0xbeef->0xbef0" + else: + count = data.count(STRING_OLD) + if count != 1: + raise C.DemoError(f"no unique patch site (addiw x{data.count(ADDW_OLD)}, string x{count})") + offset = data.find(STRING_OLD) + old_bytes = bytes(data[offset : offset + len(STRING_OLD)]) + data[offset : offset + len(STRING_OLD)] = STRING_NEW + kind = "echo-string" + with open(dst, "wb") as fh: + fh.write(data) + return { + "kind": kind, + "offset": offset, + "old_bytes": old_bytes.hex(), + "source_sha256": C.sha256_file(src), + "variant_sha256": C.sha256_file(dst), + } + + +def run() -> dict: + import greeneye as ge + from greeneye.analyze import compare_sessions + + backend = C.choose_backend("native") + out = C.artifact_dir("s07_version_diff") + va = os.path.join(out, "uproc_demo_vA.elf") + vb = os.path.join(out, "uproc_demo_vB.elf") + shutil.copyfile(C.DEMO_ELF, va) + patch = _make_variant(C.DEMO_ELF, vb) + + commands = [ + "byte-patch PING key 0xbeef -> 0xbef0 to create version B", + 'for v in (A,B): s = ge.launch(v, backend="%s", record=...)' % backend, + "identical stimulus: mailbox cmd=1 arg0=0x1234; snapshot+decode gsp_ctrl", + "compare_sessions(A, B, symbols) + decoded field diff", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3: 570.211 vs 570.144 pairing disproven", + "PART 91 ADDENDUM 6 (2026-10-01): LOGRM id + fn72 identity changed behaviour between cycles", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 4.1: version-diff decision table", + ], + baseline=( + "one full supervised boot (~5-6 min) per version plus manual log " + "comparison; the pairing hypothesis took cycles 2 and 3 back-to-back" + ), + projection=( + "green-eye cannot boot the real image without the H2 device model; " + "the same same-stimulus/compare-sessions loop runs on re-hosted " + "firmware and the produced symbol/session deltas are the H4.1 artifact" + ), + blocked=True, + device="H2.2 device model for real .fwimage versions", + ) + C.render_banner( + "s07_version_diff", + "On the same stimulus, which live-path behaviour moved between two firmware versions?", + "570.211 vs 570.144 image pairing; semantic diff before touching the emulation", + scenario["static_baseline"], + commands, + ) + + schema = C.load_schema(ge) + results = {} + for label, path in (("A", va), ("B", vb)): + session = ge.launch(kernel=path, backend=backend, record=os.path.join(out, f"session-{label}"), name=f"vd-s07-{label}") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatches = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit, d=dispatches: d.append({"cmd": hit.arg(0), "arg0": hit.arg(1)}), + ) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w") + session.run(duration=0.25) + session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1) + session.run(duration=0.25) + decoded = C.decode_ctrl(ge, session.snapshot(ctrl, C.CTRL_SIZE, f"ctrl-{label}"), schema) + session.note("version", label=label, patch=patch["kind"], reply=decoded.get("result")) + results[label] = { + "path": os.path.relpath(path, C.TOOL_ROOT), + "sha256": C.sha256_file(path), + "dispatches": dispatches, + "reply": decoded.get("result"), + "rpc_count": decoded.get("rpc_count"), + "last_rpc": decoded.get("last_rpc"), + } + finally: + session.close() + + compare_md = compare_sessions( + os.path.join(out, "session-A"), os.path.join(out, "session-B"), symbols=session.symbols, max_rows=20 + ) + C.write_text(os.path.join(out, "compare-sessions.md"), compare_md) + moved = sorted( + key + for key in ("reply", "rpc_count", "last_rpc") + if results["A"][key] != results["B"][key] + ) + payload = { + "success": results["A"]["reply"] == 0xACDB and results["B"]["reply"] == 0xACC4 and moved == ["reply"], + "backend": backend, + "patch": patch, + "version_A": results["A"], + "version_B": results["B"], + "moved_fields": moved, + "decision_table_row": { + "static_code_changed": True, + "live_profile_changed": bool(moved), + "reading": "new logic on a live path - prioritise (docs/14 4.1)", + }, + "compare_report": os.path.relpath(os.path.join(out, "compare-sessions.md"), C.TOOL_ROOT), + "question": "separate 'the code changed' from 'the behaviour moved' on a fixed stimulus", + "answer": ( + f"version A replies {results['A']['reply']:#x}, version B replies {results['B']['reply']:#x}; " + f"moved fields {moved}; the patch site is {patch['kind']} at file offset {patch['offset']:#x} - " + "the decision-table row is 'code changed + profile changed -> prioritise'" + ), + "success_criteria": { + "pass": "exactly the patched reply field moves on the live path; both sessions replay from the same stimulus", + "fail": "unrelated fields move or a version cannot be replayed", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "version-diff.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s07_version_diff"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py new file mode 100644 index 0000000..8fcaed1 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""s08 - mutation triage: mutate one field, classify the outcome. + +Real scenario (docs/18-HARNESS-AUTOMATION.md R5 and +Handoff-vGPU-IO.md PART 77 / docs/19 3.4, 2026-09-30..2026-10-02): the +error-path corpus (sentinel/boundary pointers on ECHO/SUM, invalid command +IDs) is how the emulation's fail-closed rules are derived. Statically the +demo's 4-of-24 outcomes were invisible; the discovery sweep found +``rpc_echo``/``rpc_sum`` fault on sentinel/boundary pointers while +``rpc_ping``/``rpc_state``/``rpc_fill`` stay on the handler path. + +Analog: a seeded mutation table over command/arguments, one fresh session per +attempt (the documented safety strategy), classified as ok / trap / fault / +no-progress from stop signals, ``rpc_count`` and the result field. +""" +from __future__ import annotations + +import json +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +FAULT_SIGNALS = {4, 7, 11} # SIGILL, SIGBUS, SIGSEGV +TRAP_SIGNAL = 5 # SIGTRAP (ebreak) + + +def _probe(ge, mutation: dict, backend: str) -> dict: + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s08") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + signals = [] + session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None) + session.run(duration=0.15) + literal = 0 + if mutation.get("arg0") == "hello": + hits = session.memory.search(0x80000000, 0x2000, b"hello-green-eye") + literal = hits[0] if hits else ctrl + C.CTRL_BUF + before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + session.memory.write_u64(ctrl + C.CTRL_ARG0, literal if mutation.get("arg0") == "hello" else mutation.get("arg0", 0)) + session.memory.write_u64(ctrl + C.CTRL_ARG1, mutation.get("arg1", 0)) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, mutation["cmd"]) + # max_stops=1: a faulted target keeps re-faulting at the same PC, so + # stop at the first signal instead of spinning the fault loop. + session.run(duration=0.2, max_stops=1) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + if after == before and not any(signals): + session.run(duration=0.3, max_stops=1) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + if any(s == TRAP_SIGNAL for s in signals): + outcome = "trap" + elif any(s in FAULT_SIGNALS for s in signals): + outcome = "fault" + elif after > before: + outcome = "ok" + else: + outcome = "no-progress" + return { + "outcome": outcome, + "rpc_count_delta": after - before, + "result": result, + "result_hex": f"{result:#x}", + "signals": [s for s in signals if s], + "consumed": after > before, + } + finally: + session.close() + + +def run() -> dict: + import greeneye as ge + + out = C.artifact_dir("s08_mutation_triage") + backend = C.choose_backend("native") + mutations = [ + {"strategy": "arg-boundary", "cmd": 2, "arg0": 0, "arg1": 0, "expect": "fault"}, + {"strategy": "valid", "cmd": 2, "arg0": "hello", "arg1": 0, "expect": "ok"}, + {"strategy": "arg-boundary", "cmd": 2, "arg0": 0xFFFFFFFFFFFFFFFF, "arg1": 0, "expect": "fault"}, + {"strategy": "valid", "cmd": 4, "arg0": "hello", "arg1": 5, "expect": "ok"}, + {"strategy": "buffer-length", "cmd": 4, "arg0": "hello", "arg1": 0x8000000000000000, "expect": "no-progress"}, + {"strategy": "buffer-length", "cmd": 3, "arg0": 0x41, "arg1": 4096, "expect": "ok"}, + {"strategy": "cmd-invalid", "cmd": 7, "arg0": 0, "arg1": 0, "expect": "ok"}, + ] + commands = [ + "for each seeded mutation: fresh native session, settle, one mailbox write", + "run bounded; classify from stop signals + rpc_count + result", + "write findings.jsonl (deterministic order) for corpus diffing", + ] + scenario = C.scenario( + ids=[ + "docs/18-HARNESS-AUTOMATION.md 2 (R5): bounded seeded mutation, outcome classes", + "docs/19-CONTROL-DISCOVERY.md 3.4 (2026-10-02): sentinel/boundary pointer faults found by the sweep", + "Handoff-vGPU-IO.md PARTs 3-9 (2026-09-21..23): hand-derived assert/error paths", + ], + baseline=( + "statically the fault surface is invisible; the project derived error " + "paths by hand from asserts and NOCAT records across many physical " + "cycles; a 64-iteration QEMU mutation run is the tracked baseline" + ), + projection=( + "real RM controls land in the same mutation harness once the RM " + "surface exists (docs/19 section 5); one device-write experiment " + "per physical cycle is the current cap" + ), + blocked=True, + device="H1.3 RM control surface for mutation of real controls", + ) + C.render_banner( + "s08_mutation_triage", + "Mutate one input field deterministically and classify the outcome (ok/trap/fault/no-progress).", + "error-path corpus for fail-closed emulation rules (R5/H4.3)", + scenario["static_baseline"], + commands, + ) + + findings = [] + for index, mutation in enumerate(mutations): + observed = _probe(ge, mutation, backend) + finding = {"iteration": index, **mutation, **observed, "matched_expectation": observed["outcome"] == mutation["expect"]} + findings.append(finding) + classes: dict[str, int] = {} + for finding in findings: + classes[finding["outcome"]] = classes.get(finding["outcome"], 0) + 1 + payload = { + "success": all(f["matched_expectation"] for f in findings), + "backend": backend, + "mutations": len(findings), + "outcome_classes": classes, + "findings": findings, + "question": "enumerate and classify one-field mutations without hardware", + "answer": ( + f"{len(findings)} mutations -> {classes}; all expectations matched; " + "faults localised to the dereferencing handlers, the oversized SUM " + "stays no-progress, FILL clamps at the buffer size - a corpus entry " + "per mutation with no device writes" + ), + "success_criteria": { + "pass": "every seeded mutation produces its documented outcome class", + "fail": "an outcome contradicts the case table or a session leaks", + }, + "real_case": scenario, + } + with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh: + for finding in findings: + fh.write(json.dumps(finding, sort_keys=True) + "\n") + C.write_json(os.path.join(out, "mutation.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s08_mutation_triage"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py new file mode 100644 index 0000000..1577772 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""s09 - grab a Ghidra address: did the firmware actually execute it? + +Real scenario (docs/05-GHIDRA-BRIDGE.md, docs/13, docs/14 5; and +Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24, 2026-10-01): a static +hypothesis (the GPFIFO GET encodes entry VA ``0x101000``) was only accepted +after the decoded status-queue fault showed the engine actually fetched +``0x40000`` because the GET field dropped the high VA bits (fix ``6f9cbfa``). +The static/dynamic join is exactly this: Ghidra owns addresses and +decompilation, green-eye answers "did it execute here". + +Analog: export a Ghidra-style ``names.tsv`` from the demo ELF *with a base +offset* (the classic program-address vs runtime-address trap), load it with +the right bias, hook the raw Ghidra address, trigger the path, and write the +resulting ``greeneye_hits.tsv`` annotation back for Ghidra. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +#: Pretend the Ghidra project was based at runtime - 0x1000 (the bias pitfall). +BIAS = 0x1000 + + +def _ghidra_tsv(ge, out_path: str) -> dict: + image = ge.load_elf(C.DEMO_ELF) + rows = [] + for symbol in image.symbols: + if not getattr(symbol, "is_function", False) or not symbol.name: + continue + program = symbol.value - BIAS + rows.append((program, symbol.name, symbol.size)) + rows.sort() + with open(out_path, "w", encoding="utf-8") as fh: + for program, name, size in rows: + fh.write(f"{program:#x}\t{name}\tsize={size}\n") + return {"functions": len(rows), "bias": BIAS} + + +def run() -> dict: + import greeneye as ge + from greeneye.ghidra import export_hits + + backend = C.choose_backend("qemu") + out = C.artifact_dir("s09_grabbing_a_ghidra_address") + names_tsv = os.path.join(out, "ghidra_names.tsv") + facts = _ghidra_tsv(ge, names_tsv) + runtime_rpc_fault = ge.load_elf(C.DEMO_ELF) + addr_rpc_fault = None + for symbol in runtime_rpc_fault.symbols: + if symbol.name == "rpc_fault": + addr_rpc_fault = symbol.value + if addr_rpc_fault is None: + raise C.DemoError("rpc_fault not found in demo ELF") + ghidra_addr = addr_rpc_fault - BIAS + + commands = [ + f'write Ghidra names.tsv from ELF (program = runtime - {BIAS:#x})', + f'SymbolDB.load_names_tsv(names, bias={BIAS:#x})', + f'hook(ghidra_addr {ghidra_addr:#x} + bias) # raw static address', + "trigger mailbox cmd=6; hits recorded; export_hits -> greeneye_hits.tsv", + ] + scenario = C.scenario( + ids=[ + "docs/05-GHIDRA-BRIDGE.md / docs/13: static/dynamic address join", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24 (2026-10-01): GET fault at 0x40000 vs hypothesis 0x101000", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 5: program vs runtime addresses and bias", + ], + baseline=( + "cycle 24's decoded status-queue fault ended a run of eight " + "supervised cycles (18-25, ~1 h 41 m); the static GET-encoding " + "hypothesis was unproven until a live PC/fault address existed" + ), + projection=( + "for the real image the same names.tsv + bias + hook loop runs once " + "H2.2 exposes execution of the .fwimage; the bias arithmetic demoed " + "here is exactly what docs/05 warns about" + ), + blocked=True, + device="H2.2 device model to execute the real .fwimage", + ) + C.render_banner( + "s09_grabbing_a_ghidra_address", + "Ghidra gave an address: does the firmware execute exactly there, and can the hit be annotated back?", + "static/dynamic join for the GPFIFO GET fault (0x40000 vs 0x101000)", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=os.path.join(out, "session"), name="vd-s09") + try: + session.symbols.load_names_tsv(names_tsv, bias=BIAS) + ctrl = session.symbols.lookup("gsp_ctrl") + resolved = session.resolve(ghidra_addr + BIAS) + hits = [] + session.hook(ghidra_addr + BIAS, on_enter=lambda hit: hits.append(hit.addr)) + session.run(duration=0.25) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 6) + session.run(duration=0.25) + hits_tsv = export_hits(os.path.join(out, "session"), os.path.join(out, "greeneye_hits.tsv"), symbols=session.symbols) + payload = { + "success": len(hits) >= 1 and resolved == addr_rpc_fault, + "backend": backend, + "ghidra_functions": facts["functions"], + "ghidra_names_tsv": os.path.relpath(names_tsv, C.TOOL_ROOT), + "bias": BIAS, + "ghidra_addr": f"{ghidra_addr:#x}", + "runtime_addr": f"{resolved:#x}", + "program_addr_used_without_bias": f"{ghidra_addr:#x}", + "hits": hits, + "hits_tsv": os.path.relpath(hits_tsv, C.TOOL_ROOT), + "question": "join a static Ghidra address to a runtime hit, with the bias handled explicitly", + "answer": ( + f"Ghidra addr {ghidra_addr:#x} + bias {BIAS:#x} = runtime {resolved:#x}; " + f"the hook fired {len(hits)}x at that exact address; export_hits wrote " + "the annotated greeneye_hits.tsv - the static hypothesis is now a " + "recorded dynamic fact (and using the program address raw would miss by 0x1000)" + ), + "success_criteria": { + "pass": "the biased address resolves to the real function and the hook records the hit; TSV annotates it", + "fail": "the hit lands at a different address or the TSV is empty", + }, + "real_case": scenario, + } + with open(hits_tsv, "r", encoding="utf-8") as fh: + payload["greeneye_hits_tsv"] = fh.read() + C.write_json(os.path.join(out, "ghidra-join.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s09_grabbing_a_ghidra_address"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py new file mode 100644 index 0000000..f6cd48f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""s10 - full session evidence: one run produces the whole evidence pack. + +Real scenario (docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6; Handoff-vGPU-IO.md +PART 91 ADDENDA 6-7 and PART 92 ADDENDA 6-11, 2026-10-01..10-02): a finding +is only a finding with firmware hashes, baseline/candidate sessions, an +analyze report, a compare report, the M1-M6 table, ``greeneye_hits.tsv`` and +an unexplained-delta list. The physical milestone cycles each gathered this +by hand across separate boots and log windows (cycle evidence dirs +``backups/evidence-20261001-cycle{11,12,13}``, P6 evidence dirs). + +Analog: one recorded session with hooks/watchpoints/notes/snapshots, a second +identical run for determinism, an analyze report, a session compare, an +exported hits TSV and a manifest with hashes - plus measured tool overhead +(startup, session size, trace slowdown). +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def _stimulus(ge, record, backend: str) -> dict: + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=record, name=os.path.basename(record)) + try: + ctrl = session.symbols.lookup("gsp_ctrl") + stats = {"dispatches": [], "sum_hits": 0, "result_stores": 0} + session.hook( + "rpc_dispatch", + on_enter=lambda hit: stats["dispatches"].append(hit.arg(0)), + ) + session.hook("rpc_sum", on_enter=lambda hit: stats.__setitem__("sum_hits", stats["sum_hits"] + 1)) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stats.__setitem__("result_stores", stats["result_stores"] + 1)) + session.run(duration=0.4) + session.snapshot(ctrl, C.CTRL_SIZE, "selftest-done") + session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1) + session.run(duration=0.2) + session.note("stimulus", cmd=1, arg0=0x1234, expected=0x1234 ^ C.PING_XOR) + session.snapshot(ctrl, C.CTRL_SIZE, "after-ping") + stats["ping_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 5) + session.run(duration=0.2) + stats["state_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT) + stats["rpc_count"] = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + return stats + finally: + session.close() + + +def _dir_size(path: str) -> int: + total = 0 + for root, _dirs, files in os.walk(path): + for name in files: + total += os.path.getsize(os.path.join(root, name)) + return total + + +def _trace_overhead(ge, out: str) -> dict: + import time + + def timed(trace): + record = os.path.join(out, "trace-off" if not trace else "trace-on") + start = time.monotonic() + session = ge.launch(kernel=C.DEMO_ELF, backend="native", record=record, trace=trace, name="vd-s10-trace") + run_start = time.monotonic() + try: + session.run(duration=0.25) + finally: + run_s = time.monotonic() - run_start + session.close() + return run_s, time.monotonic() - start + + off_run, off_total = timed(None) + on_run, on_total = timed(True) + return { + "untraced_run_s": round(off_run, 4), + "traced_run_s": round(on_run, 4), + "run_slowdown_x": round(on_run / off_run, 2) if off_run else None, + "untraced_total_s": round(off_total, 4), + "traced_total_s": round(on_total, 4), + "total_slowdown_x": round(on_total / off_total, 2) if off_total else None, + "note": "total includes writing blobs/trace.jsonl (~200k exec lines) at close", + } + + +def run() -> dict: + import time + + import greeneye as ge + from greeneye.analyze import compare_sessions, render_summary + from greeneye.ghidra import export_hits + + backend = C.choose_backend("qemu") + out = C.artifact_dir("s10_full_session_evidence") + commands = [ + 'ge.launch(backend="%s", record="session-A") + hooks/watch/notes/snapshots' % backend, + "repeat the identical stimulus into session-B", + "render_summary(A); compare_sessions(A, B, --symbols); export_hits(A)", + "hash firmware + artifacts into manifest.json with the M1-M6 skeleton", + ] + scenario = C.scenario( + ids=[ + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6: evidence-pack checklist and report template", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 6-7 (2026-10-01): cycle 11-13 evidence dirs", + "PART 92 ADDENDUM 6-11 (2026-10-02): P6 evidence dirs and per-cycle release transcripts", + "docs/06-ACCURACY.md 5: M1-M6 measurement contract", + ], + baseline=( + "every physical milestone gathered hash + session + logs + compare " + "by hand across separate boots and log windows; cycles 11-13 and " + "the P6 runs each have their own evidence directory" + ), + projection=( + "on real firmware the same one-session pack is produced once H2.2 " + "executes the image; the manifest/M1-M6 format is backend-agnostic" + ), + blocked=True, + device="H2.2 device model for real .fwimage session evidence", + ) + C.render_banner( + "s10_full_session_evidence", + "Can one run produce the complete evidence pack (sessions, analyze, compare, hits, hashes, M1-M6)?", + "milestone evidence packs gathered manually across boots", + scenario["static_baseline"], + commands, + ) + + start = time.monotonic() + session_a = _stimulus(ge, os.path.join(out, "session-A"), backend) + session_b = _stimulus(ge, os.path.join(out, "session-B"), backend) + + symbols = ge.SymbolDB() + symbols.load_elf(ge.load_elf(C.DEMO_ELF)) + report = render_summary(os.path.join(out, "session-A"), symbols=symbols, max_rows=30) + C.write_text(os.path.join(out, "analyze.md"), report) + compare = compare_sessions(os.path.join(out, "session-A"), os.path.join(out, "session-B"), symbols=symbols, max_rows=30) + C.write_text(os.path.join(out, "compare.md"), compare) + hits = export_hits(os.path.join(out, "session-A"), os.path.join(out, "greeneye_hits.tsv"), symbols=symbols) + trace = _trace_overhead(ge, out) + size_a = _dir_size(os.path.join(out, "session-A")) + size_b = _dir_size(os.path.join(out, "session-B")) + deltas_match = session_a["dispatches"] == session_b["dispatches"] and session_a["ping_reply"] == session_b["ping_reply"] + manifest = { + "format": "greeneye-value-evidence/1", + "firmware": { + "path": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT), + "sha256": C.sha256_file(C.DEMO_ELF), + }, + "sessions": { + "a": {"path": os.path.relpath(os.path.join(out, "session-A"), C.TOOL_ROOT), "bytes": size_a, "sha256_manifest": C.sha256_file(os.path.join(out, "session-A", "events.jsonl"))}, + "b": {"path": os.path.relpath(os.path.join(out, "session-B"), C.TOOL_ROOT), "bytes": size_b, "sha256_manifest": C.sha256_file(os.path.join(out, "session-B", "events.jsonl"))}, + }, + "stimulus": {"commands": [1, 5], "arg0": 0x1234, "note": "PING then STATE"}, + "observed": {"a": session_a, "b": session_b, "delta_match": deltas_match}, + "metrics": [ + {"id": "M1", "name": "inventory coverage", "status": "observed", "value": "dispatch inventory [1,2,3,4,5,5,5] self-test + [1,5] stimulus"}, + {"id": "M2", "name": "reply agreement", "status": "not measured", "reason": "no physical reference capture (needs H2/production trace)"}, + {"id": "M3", "name": "ordering agreement", "status": "observed", "value": "identical dispatch sequence across A/B"}, + {"id": "M4", "name": "state agreement", "status": "observed", "value": "rpc_state stage cycling on the control block"}, + {"id": "M5", "name": "error-path agreement", "status": "not measured", "reason": "error path demoed in s08; no reference to agree against"}, + {"id": "M6", "name": "memory side-effects", "status": "observed", "value": "2 named region snapshots (selftest-done, after-ping) per session"}, + ], + "artifacts": { + "analyze": "analyze.md", + "compare": "compare.md", + "hits": os.path.relpath(hits, out), + "manifest": "manifest.json", + }, + "overhead": { + "session_a_bytes": size_a, + "session_b_bytes": size_b, + "trace": trace, + "doc16_trace_slowdown": "~2.4x exec trace on the demo workload (docs/16 measured 0.54 MIPS traced vs 1.5 MIPS untraced)", + }, + "unexplained_deltas": [] if deltas_match else ["session A/B observed values differ"], + } + payload = { + "success": deltas_match and session_a["ping_reply"] == (0x1234 ^ C.PING_XOR) and size_a > 0, + "backend": backend, + "elapsed_within_demo_s": round(time.monotonic() - start, 3), + "manifest": manifest, + "question": "one run -> hash + sessions + analyze + compare + hits + M1-M6, plus tool overhead", + "answer": ( + f"one {backend} stimulus pair produced {session_a['dispatches']} dispatches, " + f"PING reply {session_a['ping_reply']:#x}, STATE reply {session_a['state_reply']:#x}, " + f"{session_a['result_stores']} publication stores; session {size_a} B; " + f"A/B deterministic={deltas_match}; analyze+compare+hits+manifest written" + ), + "success_criteria": { + "pass": "reply matches the ABI, both sessions agree, all artifacts exist, deltas list is empty", + "fail": "nondeterminism, missing artifact, or an unexplained delta", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "manifest.json"), manifest) + C.write_json(os.path.join(out, "evidence.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s10_full_session_evidence"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/version_diff.py b/debug/risc-v-gsp-rm-vm/harness/version_diff.py new file mode 100644 index 0000000..a95f244 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/version_diff.py @@ -0,0 +1,1509 @@ +#!/usr/bin/env python3 +"""Version diff: old vs new firmware targets (roadmap H4.1, planned docs/21). + +Compares two firmware images (a built ELF or a raw ``.fwimage``) section by +section; every section degrades to ``status: skipped`` with a reason instead +of aborting the report: + +1. image/shape -- entry, PT_LOAD layout, ELF section sizes, hashes, span; +2. symbol inventory -- ELF ``.symtab`` plus Ghidra names/index/exports TSVs + (including the 3-column ``ExportForGreenEye`` form) and + green-eye JSON DBs; added/removed/moved/resized with + the anonymous-shift matching discipline of + ``userspace/cuda/tools/diff_corpus.py``; +3. decompiled hints -- a corpus diff over two ``decompiled_index.tsv`` files, + an attached corpus report, or an explicit "not + supplied"; +4. probe -- bounded QEMU first-touch probe of both targets with + ``parse_qemu_faults`` and firmware-vaddr translation of + the epc and access addresses; +5. behavior -- the shared demo scenario (or a ``--stimulus`` script) + under QEMU, ``analyze.compare_sessions`` plus + hit/watch/write deltas and the docs/06 M1-M6 skeleton. + +Everything is local and stdlib-only: nothing is downloaded. The RM 610 gate +(docs/23) is enforced before any target file is opened. + +Run from the tool root: + + python3 harness/version_diff.py --old old.elf --new new.elf --out dumps/vdiff +""" +from __future__ import annotations + +import argparse +import contextlib +import os +import re +import shutil +import statistics +import struct +import sys +import tempfile +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path + +try: + from .common import ( + LOAD_ADDR, + MAX_FLAT_BYTES, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + memory_mb_for, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) + from .portability_matrix import ( + CTRL_RPC_COUNT, + METRIC_DEFS, + NO_REFERENCE, + install_scenario, + metric_rows, + profile_from_events, + read_ctrl, + ) +except ImportError: # direct script execution (python3 harness/version_diff.py) + from common import ( # type: ignore[no-redef] + LOAD_ADDR, + MAX_FLAT_BYTES, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + memory_mb_for, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) + from portability_matrix import ( # type: ignore[no-redef] + CTRL_RPC_COUNT, + METRIC_DEFS, + NO_REFERENCE, + install_scenario, + metric_rows, + profile_from_events, + read_ctrl, + ) + +from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 +from greeneye.elf import EM_RISCV, FirmwareImage, Segment, load_elf, write_flat_kernel # noqa: E402 +from greeneye.errors import GreenEyeError # noqa: E402 +from greeneye.record import load_events # noqa: E402 +from greeneye.rsp import RspClient # noqa: E402 +from greeneye.session import Session # noqa: E402 +from greeneye.symbols import Symbol # noqa: E402 +from greeneye.target import QemuTarget # noqa: E402 + +FORMAT = "greeneye-version-diff/1" +RM610_DOC = "docs/23-RM610-READINESS.md" +RM610_TOKEN = "RM610-APPROVED" +RM610_ENV = "GREENEYE_RM610_APPROVED" +RM610_RE = re.compile(r"rm[\s._-]*610", re.IGNORECASE) +PT_LOAD = 1 + + +class Rm610GateError(GreenEyeError): + """A target/release carries an RM 610 label without explicit approval.""" + + +def is_rm610_label(value: str | None) -> bool: + return bool(value) and RM610_RE.search(str(value)) is not None + + +def rm610_approved(token: str | None = None, environ: dict | None = None) -> bool: + """Explicit approval is ``--approval-token RM610-APPROVED`` or the env flag.""" + env = os.environ if environ is None else environ + token_ok = (token or "").strip() == RM610_TOKEN + env_ok = str(env.get(RM610_ENV, "")).strip().lower() in ("1", "true", "yes", "approved") + return token_ok or env_ok + + +def check_rm610_gate( + labels: list[str] | tuple[str, ...], + *, + approval_token: str | None = None, + environ: dict | None = None, +) -> None: + """Refuse RM 610 labels before anything is read, opened or downloaded.""" + blocked = sorted({str(label) for label in labels if is_rm610_label(label)}) + if not blocked: + return + if rm610_approved(approval_token, environ): + return + raise Rm610GateError( + "target/release label(s) " + ", ".join(repr(label) for label in blocked) + " indicate RM 610. " + f"Refused without explicit approval: pass --approval-token {RM610_TOKEN} or set " + f"{RM610_ENV}=1. See {RM610_DOC}. Nothing was downloaded and no target file was opened." + ) + + +# --------------------------------------------------------------------------- +# target loading (ELF or raw) +# --------------------------------------------------------------------------- + + +@dataclass +class LoadedTarget: + path: str + name: str + kind: str # elf | raw | unsupported | missing + image: FirmwareImage | None = None + sha256: str = "" + size: int = 0 + entry: int = 0 + bias: int = 0 + reason: str | None = None + + @property + def ok(self) -> bool: + return self.kind in ("elf", "raw") + + +def _read_bytes(path: str) -> bytes: + with open(path, "rb") as fh: + return fh.read() + + +def load_any_target(path: str) -> LoadedTarget: + """Read a target as RISC-V ELF or as a raw relocated image (no download).""" + name = Path(path).stem + if not os.path.isfile(path): + return LoadedTarget(path=path, name=name, kind="missing", reason="file not found") + size = os.path.getsize(path) + sha = sha256_file(path) + with open(path, "rb") as fh: + magic = fh.read(4) + if magic == b"\x7fELF": + try: + image = load_elf(path) + except Exception as exc: # malformed ELF -> skipped, never fatal + return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, + reason=f"{type(exc).__name__}: {exc}") + if not image.is_riscv: + return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, + reason=f"not a RISC-V ELF (e_machine={image.e_machine})") + span = image.max_vaddr - image.min_vaddr + if span > MAX_FLAT_BYTES: + return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, + reason=f"image span {span:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap") + return LoadedTarget(path=path, name=name, kind="elf", image=image, sha256=sha, size=size, + entry=image.entry, bias=LOAD_ADDR - image.min_vaddr) + if size == 0: + return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=0, + reason="empty raw image") + if size > MAX_FLAT_BYTES: + return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, + reason=f"raw image {size:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap") + data = _read_bytes(path) + segment = Segment(p_type=PT_LOAD, p_flags=0x7, p_offset=0, p_vaddr=0, p_paddr=0, + p_filesz=len(data), p_memsz=len(data), p_align=0x1000) + image = FirmwareImage(path=path, is64=True, e_machine=EM_RISCV, entry=0, segments=[segment], + symbols=[], _blob=data) + return LoadedTarget(path=path, name=name, kind="raw", image=image, sha256=sha, size=size, + entry=0, bias=LOAD_ADDR) + + +# --------------------------------------------------------------------------- +# 1. image / shape +# --------------------------------------------------------------------------- + + +def _cstr(blob: bytes, offset: int) -> str: + if offset >= len(blob): + return "" + end = blob.find(b"\x00", offset) + if end == -1: + end = len(blob) + return blob[offset:end].decode("utf-8", "replace") + + +def elf_sections(data: bytes) -> list[dict]: + """Section name/type/addr/size table (minimal reader; [] when unusable).""" + if len(data) < 64 or data[:4] != b"\x7fELF" or data[4] not in (1, 2): + return [] + is64 = data[4] == 2 + try: + if is64: + sh_off = struct.unpack_from("<Q", data, 40)[0] + shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 58) + sh_fmt = "<IIQQQQIIQQ" + else: + sh_off = struct.unpack_from("<I", data, 32)[0] + shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 46) + sh_fmt = "<IIIIIIIIII" + except struct.error: + return [] + sh_size = struct.calcsize(sh_fmt) + if not sh_off or sh_off + sh_size > len(data): + return [] + shstr = b"" + if 0 <= shstrndx < shnum: + base = sh_off + shstrndx * shentsize + if base + sh_size <= len(data): + sh = struct.unpack_from(sh_fmt, data, base) + shstr = data[sh[4] : sh[4] + sh[5]] + sections: list[dict] = [] + for index in range(shnum): + base = sh_off + index * shentsize + if base + sh_size > len(data): + break + sh = struct.unpack_from(sh_fmt, data, base) + sections.append( + { + "name": _cstr(shstr, sh[0]) or f"<section-{index}>", + "type": sh[1], + "addr": sh[3], + "offset": sh[4], + "size": sh[5], + } + ) + return sections + + +def shape_payload(loaded: LoadedTarget) -> dict: + payload = { + "kind": loaded.kind, + "sha256": loaded.sha256, + "size": loaded.size, + "reason": loaded.reason, + } + if not loaded.ok or loaded.image is None: + return payload + image = loaded.image + payload.update( + { + "entry": loaded.entry, + "bias": loaded.bias, + "min_vaddr": image.min_vaddr, + "max_vaddr": image.max_vaddr, + "span": image.max_vaddr - image.min_vaddr, + "loads": [ + { + "index": index, + "vaddr": seg.p_vaddr, + "paddr": seg.p_paddr, + "filesz": seg.p_filesz, + "memsz": seg.p_memsz, + "flags": seg.p_flags, + "align": seg.p_align, + } + for index, seg in enumerate(image.load_segments) + ], + } + ) + if loaded.kind == "elf": + payload["sections"] = elf_sections(image.read()) + else: + payload["sections"] = [] + payload["note"] = "raw image: one implicit load at vaddr 0; no ELF section table" + return payload + + +def _scalar_delta(a, b, name: str) -> dict: + return {"a": a, "b": b, "delta": b - a if isinstance(a, int) and isinstance(b, int) else None, + "changed": a != b} + + +def _cap(rows: list, max_rows: int) -> list: + return rows[:max_rows] + + +def image_diff(old: LoadedTarget, new: LoadedTarget, max_rows: int) -> dict: + a, b = shape_payload(old), shape_payload(new) + result = {"status": "skipped", "reason": None, "a": a, "b": b, "same": None, "deltas": None} + if not old.ok or not new.ok: + result["reason"] = "; ".join( + f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok + ) + return result + + loads_a, loads_b = a["loads"], b["loads"] + load_added = [payload for index, payload in enumerate(loads_b) if index >= len(loads_a)] + load_removed = [payload for index, payload in enumerate(loads_a) if index >= len(loads_b)] + load_changed = [] + for index, (la, lb) in enumerate(zip(loads_a, loads_b)): + fields = {key: _scalar_delta(la[key], lb[key], key) for key in ("vaddr", "filesz", "memsz", "flags")} + if any(row["changed"] for row in fields.values()): + load_changed.append({"index": index, "fields": fields}) + + sec_a = {row["name"]: row for row in a.get("sections", [])} + sec_b = {row["name"]: row for row in b.get("sections", [])} + sec_added = [{key: sec_b[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_b) - set(sec_a))] + sec_removed = [{key: sec_a[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_a) - set(sec_b))] + sec_moved, sec_resized = [], [] + for name in sorted(set(sec_a) & set(sec_b)): + ra, rb = sec_a[name], sec_b[name] + if ra["addr"] != rb["addr"]: + sec_moved.append({"name": name, "old_addr": ra["addr"], "new_addr": rb["addr"]}) + if ra["size"] != rb["size"]: + sec_resized.append({"name": name, "old_size": ra["size"], "new_size": rb["size"]}) + + deltas = { + "entry": _scalar_delta(a["entry"], b["entry"], "entry"), + "span": _scalar_delta(a["span"], b["span"], "span"), + "file_size": _scalar_delta(a["size"], b["size"], "size"), + "loads": { + "added": _cap(load_added, max_rows), + "removed": _cap(load_removed, max_rows), + "changed": _cap(load_changed, max_rows), + }, + "sections": { + "added": _cap(sec_added, max_rows), + "removed": _cap(sec_removed, max_rows), + "moved": _cap(sec_moved, max_rows), + "resized": _cap(sec_resized, max_rows), + }, + } + same = ( + not deltas["entry"]["changed"] + and not deltas["span"]["changed"] + and not deltas["file_size"]["changed"] + and not any(deltas["loads"][key] for key in ("added", "removed", "changed")) + and not any(deltas["sections"][key] for key in ("added", "removed", "moved", "resized")) + ) + result.update({"status": "ok", "same": same, "deltas": deltas}) + return result + + +# --------------------------------------------------------------------------- +# 2. symbol inventory (diff_corpus.py matching discipline) +# --------------------------------------------------------------------------- + + +def looks_like_three_column_export(path: str) -> bool: + """``addr<TAB>name<TAB>size`` (ExportForGreenEye) vs names/exports TSVs.""" + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + for line in fh: + line = line.strip() + if not line or line.startswith("#"): + continue + cols = line.split("\t") + if len(cols) >= 3 and [c.strip() for c in cols[:3]] == ["addr", "name", "size"]: + return True + if len(cols) == 3: + try: + int(cols[0], 16) + int(cols[2], 0) + return True + except ValueError: + return False + return False + except OSError: + return False + return False + + +def load_three_column_export(db: SymbolDB, path: str, bias: int = 0) -> int: + count = 0 + with open(path, "r", encoding="utf-8", errors="replace") as fh: + for line in fh: + line = line.rstrip("\n") + if not line or line.startswith("#"): + continue + cols = line.split("\t") + if len(cols) != 3 or cols[0].strip() == "addr": + continue + try: + addr = int(cols[0], 16) + size = int(cols[2], 0) + except ValueError: + continue + if not cols[1]: + continue + db.add(Symbol(addr=addr + bias, name=cols[1], size=size, source="ghidra-export")) + count += 1 + return count + + +def load_symbol_sources( + loaded: LoadedTarget, + *, + symbols_json: str | None = None, + ghidra_names: str | None = None, + ghidra_index: str | None = None, + ghidra_exports: str | None = None, + ghidra_bias: int = 0, +) -> tuple[SymbolDB, list[str], list[str]]: + """Merge every supplied symbol source for one side into one SymbolDB.""" + db = SymbolDB() + sources: list[str] = [] + notes: list[str] = [] + if loaded.ok and loaded.image is not None and loaded.image.symbols: + count = db.load_elf(loaded.image, bias=loaded.bias, source="elf") + if count: + sources.append(f"elf:{count}") + if symbols_json: + if os.path.isfile(symbols_json): + count = db.load_json(symbols_json) + sources.append(f"{os.path.basename(symbols_json)}:{count}") + else: + notes.append(f"symbol DB not found: {symbols_json}") + if ghidra_names: + if os.path.isfile(ghidra_names): + count = db.load_names_tsv(ghidra_names, bias=ghidra_bias) + sources.append(f"{os.path.basename(ghidra_names)}:{count}") + else: + notes.append(f"Ghidra names TSV not found: {ghidra_names}") + if ghidra_index: + if os.path.isfile(ghidra_index): + count = db.load_index_tsv(ghidra_index, bias=ghidra_bias) + sources.append(f"{os.path.basename(ghidra_index)}:{count}") + else: + notes.append(f"Ghidra index TSV not found: {ghidra_index}") + if ghidra_exports: + if os.path.isfile(ghidra_exports): + if looks_like_three_column_export(ghidra_exports): + count = load_three_column_export(db, ghidra_exports, bias=ghidra_bias) + sources.append(f"{os.path.basename(ghidra_exports)}:{count} (3-column export)") + else: + count = db.load_index_tsv(ghidra_exports, bias=ghidra_bias) + sources.append(f"{os.path.basename(ghidra_exports)}:{count}") + else: + notes.append(f"Ghidra exports TSV not found: {ghidra_exports}") + return db, sources, notes + + +def inventory_records(db: SymbolDB) -> list[dict]: + records = [ + { + "name": symbol.name, + "addr": symbol.addr, + "size": symbol.size, + "source": symbol.source, + } + for symbol in db + ] + records.sort(key=lambda row: (row["addr"], row["name"], row["size"])) + return records + + +def _is_named(name: str) -> bool: + return bool(name) and not name.startswith("FUN_") + + +def _record_brief(record: dict | None) -> dict | None: + if record is None: + return None + return {"addr": record["addr"], "name": record["name"], "size": record["size"]} + + +def inventory_diff(a_records: list[dict], b_records: list[dict], max_rows: int = 40) -> dict: + """Named-by-name + anonymous-at-shifted-address matching (diff_corpus.py).""" + a_by_addr = {row["addr"]: row for row in a_records} + b_by_addr = {row["addr"]: row for row in b_records} + named_a: dict[str, list[int]] = defaultdict(list) + named_b: dict[str, list[int]] = defaultdict(list) + for row in a_records: + if _is_named(row["name"]): + named_a[row["name"]].append(row["addr"]) + for row in b_records: + if _is_named(row["name"]): + named_b[row["name"]].append(row["addr"]) + + matched_named: list[tuple[str, int, int]] = [] + for name in sorted(set(named_a) & set(named_b)): + for addr_a, addr_b in zip(sorted(named_a[name]), sorted(named_b[name])): + matched_named.append((name, addr_a, addr_b)) + + deltas = [addr_b - addr_a for _, addr_a, addr_b in matched_named] + shift = int(statistics.median(deltas)) if deltas else 0 + dominant = (sum(1 for value in deltas if abs(value - shift) <= 0x40) / len(deltas)) if deltas else 0.0 + + used_a = {addr_a for _, addr_a, _ in matched_named} + used_b = {addr_b for _, _, addr_b in matched_named} + moved = [(name, addr_a, addr_b) for name, addr_a, addr_b in matched_named if addr_a != addr_b] + resized_named = [ + (name, addr_a, addr_b) + for name, addr_a, addr_b in matched_named + if a_by_addr[addr_a]["size"] != b_by_addr[addr_b]["size"] + ] + + shifted: list[tuple[int, int]] = [] + for row in sorted(a_records, key=lambda item: item["addr"]): + addr_a = row["addr"] + if addr_a in used_a or _is_named(row["name"]): + continue + addr_b = addr_a + shift + if addr_b in b_by_addr and addr_b not in used_b: + shifted.append((addr_a, addr_b)) + used_a.add(addr_a) + used_b.add(addr_b) + shifted_resized = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] != b_by_addr[fb]["size"]] + shifted_ok = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] == b_by_addr[fb]["size"]] + + removed = sorted(addr for addr in a_by_addr if addr not in used_a) + added = sorted(addr for addr in b_by_addr if addr not in used_b) + matched_total = len(matched_named) + len(shifted) + named_count_a = sum(len(v) for v in named_a.values()) + named_count_b = sum(len(v) for v in named_b.values()) + + return { + "shift": shift, + "shift_dominant": round(dominant, 4), + "matched_named": len(matched_named), + "matched_shifted": len(shifted_ok), + "matched_shifted_resized": len(shifted_resized), + "moved": _cap( + [{"name": name, "old_addr": fa, "new_addr": fb, "delta": fb - fa} for name, fa, fb in moved], + max_rows, + ), + "resized": _cap( + [ + { + "name": name, + "old_addr": fa, + "new_addr": fb, + "old_size": a_by_addr[fa]["size"], + "new_size": b_by_addr[fb]["size"], + } + for name, fa, fb in resized_named + ], + max_rows, + ), + "resized_anonymous": _cap( + [ + { + "old_addr": fa, + "new_addr": fb, + "old_size": a_by_addr[fa]["size"], + "new_size": b_by_addr[fb]["size"], + } + for fa, fb in shifted_resized + ], + max_rows, + ), + "added": _cap([_record_brief(b_by_addr[addr]) for addr in added], max_rows), + "removed": _cap([_record_brief(a_by_addr[addr]) for addr in removed], max_rows), + "totals": { + "a": len(a_records), + "b": len(b_records), + "named_a": named_count_a, + "named_b": named_count_b, + "anonymous_a": sum(1 for row in a_records if not _is_named(row["name"])), + "anonymous_b": sum(1 for row in b_records if not _is_named(row["name"])), + "added": len(added), + "removed": len(removed), + "moved": len(moved), + "resized_named": len(resized_named), + }, + "rates": { + "match_a": round(matched_total / len(a_records), 4) if a_records else 0.0, + "match_b": round(matched_total / len(b_records), 4) if b_records else 0.0, + "named_transfer": round(len(matched_named) / named_count_a, 4) if named_count_a else 0.0, + }, + } + + +def symbol_section( + old: LoadedTarget, + new: LoadedTarget, + old_db: SymbolDB, + old_sources: list[str], + old_notes: list[str], + new_db: SymbolDB, + new_sources: list[str], + new_notes: list[str], + options: "VersionDiffOptions", + out_dir: str, +) -> dict: + result = { + "status": "skipped", + "reason": None, + "a": {"count": 0, "sources": old_sources, "notes": old_notes, "db": None}, + "b": {"count": 0, "sources": new_sources, "notes": new_notes, "db": None}, + } + if not old.ok or not new.ok: + result["reason"] = "; ".join( + f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok + ) + return result + a_records = inventory_records(old_db) + b_records = inventory_records(new_db) + result["a"]["count"] = len(a_records) + result["b"]["count"] = len(b_records) + if not a_records and not b_records: + result["reason"] = ( + "no symbols on either side: no ELF .symtab and no Ghidra/JSON symbol sources supplied" + ) + return result + result.update(inventory_diff(a_records, b_records, options.max_rows)) + result["status"] = "ok" + for key, db, loaded, sources, notes in ( + ("a", old_db, old, old_sources, old_notes), + ("b", new_db, new, new_sources, new_notes), + ): + if not len(db): + continue + rel = os.path.join("symbols", f"{loaded.name}.json") + path = os.path.join(out_dir, rel) + ensure_dir(os.path.dirname(path)) + db.save_json(path, meta={"firmware": os.path.basename(loaded.path), "bias": loaded.bias, + "sources": sources, "notes": notes}) + result[key]["db"] = rel + return result + + +# --------------------------------------------------------------------------- +# 3. decompiled-change hints +# --------------------------------------------------------------------------- + + +def corpus_records(path: str) -> list[dict]: + db = SymbolDB() + if looks_like_three_column_export(path): + load_three_column_export(db, path, bias=0) + else: + db.load_index_tsv(path, bias=0) + return inventory_records(db) + + +def corpus_diff(path_a: str, path_b: str, max_rows: int = 40) -> dict: + diff = inventory_diff(corpus_records(path_a), corpus_records(path_b), max_rows) + diff["a"] = os.path.basename(path_a) + diff["b"] = os.path.basename(path_b) + return diff + + +def corpus_section(options: "VersionDiffOptions", out_dir: str) -> dict: + if options.corpus_report: + if not os.path.isfile(options.corpus_report): + return {"status": "skipped", "reason": f"corpus report not found: {options.corpus_report}", + "mode": "corpus-report"} + data = _read_bytes(options.corpus_report) + preview = data.decode("utf-8", "replace") + return { + "status": "ok", + "reason": None, + "mode": "corpus-report", + "report": { + "path": os.path.basename(options.corpus_report), + "sha256": sha256_file(options.corpus_report), + "bytes": len(data), + "preview": preview[:4000], + "truncated": len(preview) > 4000, + }, + } + if options.corpus_diff: + path_a, path_b = options.corpus_diff + missing = [path for path in (path_a, path_b) if not os.path.isfile(path)] + if missing: + return {"status": "skipped", "mode": "corpus-diff", + "reason": "corpus TSV not found: " + ", ".join(missing)} + diff = corpus_diff(path_a, path_b, options.max_rows) + rel = os.path.join("corpus", "corpus-diff.json") + write_json(os.path.join(out_dir, rel), diff) + return {"status": "ok", "reason": None, "mode": "corpus-diff", "diff": diff, "report": rel} + return { + "status": "skipped", + "reason": "not supplied: pass --corpus-diff OLD_TSV NEW_TSV or --corpus-report FILE", + "mode": None, + } + + +# --------------------------------------------------------------------------- +# QEMU sessions for ELF and raw targets +# --------------------------------------------------------------------------- + + +@dataclass +class OpenTarget: + loaded: LoadedTarget + session: Session + target: QemuTarget + workdir: str + owns_workdir: bool = False + + +@contextlib.contextmanager +def open_target_session( + loaded: LoadedTarget, + *, + record: str | bool | None, + name: str, + symbols: SymbolDB | None = None, + log_faults: bool = True, + workdir: str | None = None, +): + """Relocate the target to LOAD_ADDR, start QEMU and yield a live session.""" + owns_workdir = workdir is None + if workdir is None: + workdir = tempfile.mkdtemp(prefix="greeneye-vdiff-") + flat_kernel = os.path.join(workdir, "flat.bin") + assert loaded.image is not None + if loaded.kind == "elf": + write_flat_kernel(loaded.path, flat_kernel, load_addr=LOAD_ADDR) + else: + with open(flat_kernel, "wb") as fh: + fh.write(loaded.image.read()) + target = QemuTarget( + kernel=flat_kernel, + memory_mb=memory_mb_for(loaded.image), + dlog="guest_errors,unimp,int" if log_faults else None, + workdir=workdir, + ) + target.start(paused=True) + client = RspClient(target.host, target.port) + client.connect() + client.sync() + db = symbols if symbols is not None else SymbolDB() + session = Session( + client=client, + target=target, + symbols=db, + record=record, + name=name, + meta={ + "backend": "qemu", + "bias": loaded.bias, + "firmware": loaded.path, + "kind": loaded.kind, + "harness": "version_diff", + }, + ) + session.load_image(loaded.image, bias=loaded.bias, load_symbols=symbols is None) + try: + yield OpenTarget(loaded=loaded, session=session, target=target, workdir=workdir, + owns_workdir=owns_workdir) + finally: + try: + session.close() + finally: + if owns_workdir: + shutil.rmtree(workdir, ignore_errors=True) + + +def enrich_fault(fault: dict, bias: int, span: int) -> dict: + """Add firmware-vaddr translations for epc and the access address.""" + out = { + "cause": fault.get("cause"), + "desc": fault.get("desc"), + "epc": fault.get("epc"), + "tval": fault.get("tval"), + "access": fault.get("access"), + } + epc = fault.get("epc") + out["epc_firmware"] = epc - bias if isinstance(epc, int) else None + access = fault.get("access") or {} + access_addr = access.get("access_addr") + if isinstance(access_addr, int): + if LOAD_ADDR <= access_addr < LOAD_ADDR + span: + out["access_firmware"] = access_addr - bias + else: + out["access_firmware"] = None + else: + out["access_firmware"] = None + return out + + +# --------------------------------------------------------------------------- +# 4. probe first-touch diff +# --------------------------------------------------------------------------- + + +def _skipped_probe(reason: str | None) -> dict: + return { + "status": "skipped", + "reason": reason, + "session": None, + "instructions": None, + "stopped_pc": None, + "stopped_symbol": None, + "fault_count": 0, + "faults_capped": False, + "first_fault": None, + } + + +def run_probe( + loaded: LoadedTarget, + db: SymbolDB, + options: "VersionDiffOptions", + out_dir: str, + qemu_ok: bool, +) -> dict: + if not qemu_ok: + return _skipped_probe("qemu-system-riscv64 not found") + if not loaded.ok: + return _skipped_probe(loaded.reason) + result = _skipped_probe(None) + result["instructions"] = options.probe_instructions + session_rel = os.path.join("sessions", f"{loaded.name}-probe") + try: + with open_target_session( + loaded, + record=os.path.join(out_dir, session_rel), + name=f"{loaded.name}-probe", + symbols=db, + log_faults=True, + ) as ot: + session = ot.session + if options.probe_instructions > 0: + session.step(options.probe_instructions) + else: + session.run(duration=options.probe_duration) + result["stopped_pc"] = session.pc + symbol = session.symbols.resolve(session.pc) + result["stopped_symbol"] = symbol.name if symbol else None + text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0 + result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None + result["session"] = session_rel + if text: + session.recorder.meta["qemu_log_blob"] = session.recorder.blob("qemu.log", text.encode("utf-8")) + session.recorder.meta["faults_capped"] = capped + result["status"] = "ok" + except Exception as exc: # a failed probe must not abort the diff + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +def _probe_signature(probe: dict) -> tuple: + first = probe.get("first_fault") or {} + return ( + probe.get("stopped_pc"), + first.get("desc"), + first.get("cause"), + first.get("epc_firmware"), + first.get("access_firmware"), + (first.get("access") or {}).get("access_addr"), + ) + + +def probe_diff( + old: LoadedTarget, + new: LoadedTarget, + old_db: SymbolDB, + new_db: SymbolDB, + options: "VersionDiffOptions", + out_dir: str, + qemu_ok: bool, +) -> dict: + a = run_probe(old, old_db, options, out_dir, qemu_ok) + b = run_probe(new, new_db, options, out_dir, qemu_ok) + statuses = {a["status"], b["status"]} + if statuses == {"ok"}: + status, reason = "ok", None + elif "error" in statuses: + status = "error" + reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe failed" + else: + status = "skipped" + reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe not run" + changed = _probe_signature(a) != _probe_signature(b) + if status == "ok": + summary = "first fault and stopped PC unchanged" if not changed else "first fault or stopped PC changed" + else: + summary = None + return {"status": status, "reason": reason, "a": a, "b": b, "changed": changed, "summary": summary} + + +# --------------------------------------------------------------------------- +# 5. behavior diff +# --------------------------------------------------------------------------- + + +def _skipped_behavior(reason: str | None) -> dict: + return { + "status": "skipped", + "reason": reason, + "session": None, + "hooks": [], + "uart_banner": None, + "ctrl": None, + "event_counts": {}, + "bp": {}, + "watch": {}, + "writes": {}, + "fault_count": 0, + "faults_capped": False, + "first_fault": None, + } + + +def run_behavior( + loaded: LoadedTarget, + db: SymbolDB, + options: "VersionDiffOptions", + out_dir: str, + qemu_ok: bool, +) -> dict: + if not qemu_ok: + return _skipped_behavior("qemu-system-riscv64 not found") + if not loaded.ok: + return _skipped_behavior(loaded.reason) + result = _skipped_behavior(None) + session_rel = os.path.join("sessions", f"{loaded.name}-behavior") + try: + with open_target_session( + loaded, + record=os.path.join(out_dir, session_rel), + name=f"{loaded.name}-behavior", + symbols=db, + log_faults=True, + ) as ot: + session = ot.session + result["hooks"] = install_scenario(session) + ctrl_symbol = session.symbols.get("gsp_ctrl") + ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None + if ctrl_addr is not None: + session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w", + listener="version-diff:rpc_count") + if options.stimulus: + import greeneye as ge + + script_path = Path(options.stimulus) + namespace = { + "ge": ge, + "greeneye": ge, + "session": session, + "s": session, + "memory": session.memory, + "symbols": session.symbols, + "recorder": session.recorder, + } + exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace) + else: + session.run(duration=options.behavior_duration, max_stops=options.max_stops) + if ctrl_addr is not None: + session.memory.write_u64(ctrl_addr + 32, 1) # PING + session.memory.write_u64(ctrl_addr + 40, 0x1234) + session.run(duration=options.behavior_duration, max_stops=options.max_stops) + result["ctrl"] = read_ctrl(session, ctrl_symbol) + uart = ot.target.read_uart() + result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None + text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0 + result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None + result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel)))) + result["session"] = session_rel + result["status"] = "ok" + except Exception as exc: # a failed scenario must not abort the diff + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +def behavior_diff( + old: LoadedTarget, + new: LoadedTarget, + old_db: SymbolDB, + new_db: SymbolDB, + probe_a: dict, + probe_b: dict, + options: "VersionDiffOptions", + out_dir: str, + qemu_ok: bool, +) -> dict: + a = run_behavior(old, old_db, options, out_dir, qemu_ok) + b = run_behavior(new, new_db, options, out_dir, qemu_ok) + entry = { + "status": "skipped", + "reason": None, + "a": a, + "b": b, + "compare_report": None, + "deltas": {"bp": [], "watch": [], "writes": []}, + "all_zero": None, + "metrics": [], + } + a["metrics"] = metric_rows(probe_a, a) + b["metrics"] = metric_rows(probe_b, b) + if a["status"] != "ok" or b["status"] != "ok": + reasons = [row.get("reason") for row in (a, b) if row.get("reason")] + entry["reason"] = "; ".join(reasons) or "a behavior session is not available on both targets" + entry["status"] = "error" if "error" in {a["status"], b["status"]} else "skipped" + return entry + a_dir = os.path.join(out_dir, a["session"]) + b_dir = os.path.join(out_dir, b["session"]) + try: + report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows) + rel = os.path.join("compare", f"{old.name}--{new.name}.md") + write_text(os.path.join(out_dir, rel), report) + entry["compare_report"] = rel + except Exception as exc: + entry["reason"] = f"{type(exc).__name__}: {exc}" + deltas = { + name: counter_delta(a.get(name) or {}, b.get(name) or {}) + for name in ("bp", "watch", "writes") + } + entry["deltas"] = deltas + entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows) + a_metrics = {row["id"]: row for row in a["metrics"]} + b_metrics = {row["id"]: row for row in b["metrics"]} + entry["metrics"] = [ + { + "id": metric_id, + "name": name, + "status": "not measured", + "reason": NO_REFERENCE, + "a_observed": (a_metrics.get(metric_id) or {}).get("observed"), + "b_observed": (b_metrics.get(metric_id) or {}).get("observed"), + } + for metric_id, name in METRIC_DEFS + ] + entry["status"] = "ok" + return entry + + +# --------------------------------------------------------------------------- +# report assembly + rendering +# --------------------------------------------------------------------------- + + +@dataclass +class VersionDiffOptions: + old: str + new: str + out: str + old_label: str | None = None + new_label: str | None = None + release: str | None = None + old_symbols: str | None = None + new_symbols: str | None = None + old_ghidra_names: str | None = None + new_ghidra_names: str | None = None + old_ghidra_index: str | None = None + new_ghidra_index: str | None = None + old_ghidra_exports: str | None = None + new_ghidra_exports: str | None = None + old_ghidra_bias: int = 0 + new_ghidra_bias: int = 0 + corpus_diff: tuple[str, str] | None = None + corpus_report: str | None = None + stimulus: str | None = None + probe_instructions: int = 2000 + probe_duration: float = 0.5 + behavior_duration: float = 0.4 + max_stops: int = 5000 + max_rows: int = 40 + log_bytes: int = MAX_LOG_BYTES + approval_token: str | None = None + + def labels(self) -> list[str]: + return [ + value + for value in ( + self.release, + self.old_label, + self.new_label, + os.path.basename(self.old), + os.path.basename(self.new), + ) + if value + ] + + def label_for(self, side: str) -> str: + explicit = self.old_label if side == "old" else self.new_label + path = self.old if side == "old" else self.new + return explicit or Path(path).stem + + +def _target_brief(label: str, loaded: LoadedTarget, qemu_ok: bool, options: VersionDiffOptions) -> dict: + return { + "label": label, + "path": os.path.basename(loaded.path), + "kind": loaded.kind, + "status": "ok" if loaded.ok else "skipped", + "reason": loaded.reason, + "sha256": loaded.sha256, + "size": loaded.size, + "entry": loaded.entry if loaded.ok else None, + "bias": loaded.bias if loaded.ok else None, + "probe_instructions": options.probe_instructions, + "qemu_available": qemu_ok, + } + + +def run_version_diff(options: VersionDiffOptions) -> dict: + """Run every section, write JSON+markdown under ``--out``, return the report.""" + check_rm610_gate(options.labels(), approval_token=options.approval_token) + ensure_dir(options.out) + old_label, new_label = options.label_for("old"), options.label_for("new") + qemu_ok = qemu_available() + old = load_any_target(options.old) + new = load_any_target(options.new) + + if old.ok: + old_db, old_sources, old_notes = load_symbol_sources( + old, symbols_json=options.old_symbols, ghidra_names=options.old_ghidra_names, + ghidra_index=options.old_ghidra_index, ghidra_exports=options.old_ghidra_exports, + ghidra_bias=options.old_ghidra_bias, + ) + else: + old_db, old_sources, old_notes = SymbolDB(), [], [] + if new.ok: + new_db, new_sources, new_notes = load_symbol_sources( + new, symbols_json=options.new_symbols, ghidra_names=options.new_ghidra_names, + ghidra_index=options.new_ghidra_index, ghidra_exports=options.new_ghidra_exports, + ghidra_bias=options.new_ghidra_bias, + ) + else: + new_db, new_sources, new_notes = SymbolDB(), [], [] + + image = image_diff(old, new, options.max_rows) + symbols = symbol_section(old, new, old_db, old_sources, old_notes, new_db, new_sources, new_notes, + options, options.out) + corpus = corpus_section(options, options.out) + probe = probe_diff(old, new, old_db, new_db, options, options.out, qemu_ok) + behavior = behavior_diff(old, new, old_db, new_db, probe["a"], probe["b"], options, options.out, qemu_ok) + + sections = {"image": image, "symbols": symbols, "corpus": corpus, "probe": probe, "behavior": behavior} + report = { + "format": FORMAT, + "tool": "harness/version_diff.py", + "backend": "qemu", + "qemu_available": qemu_ok, + "old": _target_brief(old_label, old, qemu_ok, options), + "new": _target_brief(new_label, new, qemu_ok, options), + "options": { + "probe_instructions": options.probe_instructions, + "probe_duration": options.probe_duration, + "behavior_duration": options.behavior_duration, + "max_rows": options.max_rows, + "stimulus": os.path.basename(options.stimulus) if options.stimulus else None, + "corpus_report": os.path.basename(options.corpus_report) if options.corpus_report else None, + "corpus_diff": [os.path.basename(p) for p in options.corpus_diff] if options.corpus_diff else None, + }, + "sections": sections, + "summary": { + "sections_ok": sorted(name for name, section in sections.items() if section["status"] == "ok"), + "sections_skipped": { + name: section.get("reason") + for name, section in sections.items() + if section["status"] == "skipped" + }, + "sections_error": { + name: section.get("reason") + for name, section in sections.items() + if section["status"] == "error" + }, + }, + } + write_json(os.path.join(options.out, "version-diff.json"), report) + write_text(os.path.join(options.out, "version-diff.md"), render_markdown(report, options)) + return report + + +# --------------------------------------------------------------------------- +# markdown rendering +# --------------------------------------------------------------------------- + + +def _table(headers: list[str], rows: list[list]) -> list[str]: + lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] + for row in rows: + lines.append("| " + " | ".join(str(cell) for cell in row) + " |") + return lines + + +def _hex(value) -> str: + return f"`{value:#x}`" if isinstance(value, int) else "-" + + +def render_markdown(report: dict, options: VersionDiffOptions) -> str: + old, new = report["old"], report["new"] + lines = [ + "# green-eye version diff", + "", + "> **# TOOL IN DEVELOPMENT - NOT READY YET!** Generated by `harness/version_diff.py`.", + "", + f"* format: `{report['format']}`", + f"* targets: `{old['label']}` -> `{new['label']}`", + f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})", + "", + "## Targets", + "", + ] + lines += _table( + ["side", "label", "kind", "size", "sha256", "entry"], + [ + [side, f"`{item['label']}`", item["kind"], item["size"], f"`{(item['sha256'] or '')[:16]}`", + _hex(item["entry"])] + for side, item in (("old", old), ("new", new)) + ], + ) + lines += [""] + + image = report["sections"]["image"] + lines += ["## 1. Image / shape", "", f"* status: {image['status']}"] + if image.get("reason"): + lines += [f"* reason: {image['reason']}"] + if image["status"] == "ok": + deltas = image["deltas"] + lines += _table( + ["field", "old", "new", "delta"], + [ + ["entry", _hex(deltas["entry"]["a"]), _hex(deltas["entry"]["b"]), _hex(deltas["entry"]["delta"])], + ["span", _hex(deltas["span"]["a"]), _hex(deltas["span"]["b"]), _hex(deltas["span"]["delta"])], + ["file size", deltas["file_size"]["a"], deltas["file_size"]["b"], deltas["file_size"]["delta"]], + ], + ) + lines += ["", f"* identical shape: {'yes' if image['same'] else 'no'}", ""] + lines += _table( + ["load delta", "count"], + [[key, len(image["deltas"]["loads"][key])] for key in ("added", "removed", "changed")] + + [["section " + key, len(image["deltas"]["sections"][key])] + for key in ("added", "removed", "moved", "resized")], + ) + lines += [""] + + symbols = report["sections"]["symbols"] + lines += ["## 2. Symbol inventory", "", f"* status: {symbols['status']}"] + if symbols.get("reason"): + lines += [f"* reason: {symbols['reason']}"] + if symbols["status"] == "ok": + lines += _table( + ["side", "symbols", "named", "anonymous", "sources"], + [ + [side, symbols[side]["count"], symbols["totals"][f"named_{side}"], + symbols["totals"][f"anonymous_{side}"], ", ".join(symbols[side]["sources"]) or "-"] + for side in ("a", "b") + ], + ) + lines += [""] + lines += _table( + ["address shift", "dominant", "matched named", "matched anon", "moved", "resized", "added", "removed"], + [[ + _hex(symbols["shift"]), f"{symbols['shift_dominant'] * 100:.1f}%", symbols["matched_named"], + symbols["matched_shifted"], symbols["totals"]["moved"], symbols["totals"]["resized_named"], + symbols["totals"]["added"], symbols["totals"]["removed"], + ]], + ) + lines += [""] + for key, title in (("moved", "Moved named"), ("resized", "Resized named"), ("added", "Added"), + ("removed", "Removed")): + rows = symbols.get(key) or [] + lines += [f"**{title} (first {options.max_rows})**", ""] + if not rows: + lines += ["_none_", ""] + continue + if key in ("added", "removed"): + lines += _table(["addr", "name", "size"], [[_hex(r["addr"]), f"`{r['name']}`", r["size"]] for r in rows]) + elif key == "moved": + lines += _table(["name", "old addr", "new addr", "delta"], + [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), _hex(r["delta"])] for r in rows]) + else: + lines += _table(["name", "old addr", "new addr", "old size", "new size"], + [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), r["old_size"], r["new_size"]] + for r in rows]) + lines += [""] + else: + lines += [""] + + corpus = report["sections"]["corpus"] + lines += ["## 3. Decompiled-change hints", "", f"* status: {corpus['status']}", + f"* mode: {corpus.get('mode') or '-'}"] + if corpus.get("reason"): + lines += [f"* reason: {corpus['reason']}"] + if corpus["status"] == "ok" and corpus.get("mode") == "corpus-diff": + diff = corpus["diff"] + lines += _table( + ["functions A", "functions B", "shift", "added", "removed", "moved", "resized"], + [[diff["totals"]["a"], diff["totals"]["b"], _hex(diff["shift"]), diff["totals"]["added"], + diff["totals"]["removed"], diff["totals"]["moved"], diff["totals"]["resized_named"]]], + ) + elif corpus["status"] == "ok" and corpus.get("mode") == "corpus-report": + info = corpus["report"] + lines += [f"* report: `{info['path']}` ({info['bytes']} bytes, sha256 `{info['sha256'][:16]}`)"] + lines += [""] + + probe = report["sections"]["probe"] + lines += ["## 4. Probe first-touch", "", f"* status: {probe['status']}"] + if probe.get("reason"): + lines += [f"* reason: {probe['reason']}"] + if probe["status"] == "ok": + lines += [f"* {probe['summary']}", ""] + lines += _table( + ["side", "stopped pc", "stopped symbol", "fault", "epc (fw vaddr)", "access", "access (fw vaddr)"], + [ + [ + side, + _hex(probe[side].get("stopped_pc")), + probe[side].get("stopped_symbol") or "-", + (probe[side].get("first_fault") or {}).get("desc") or "none", + _hex((probe[side].get("first_fault") or {}).get("epc_firmware")), + _hex(((probe[side].get("first_fault") or {}).get("access") or {}).get("access_addr")), + _hex((probe[side].get("first_fault") or {}).get("access_firmware")), + ] + for side in ("a", "b") + ], + ) + lines += [""] + + behavior = report["sections"]["behavior"] + lines += ["## 5. Behavior", "", f"* status: {behavior['status']}"] + if behavior.get("reason"): + lines += [f"* reason: {behavior['reason']}"] + if behavior["status"] == "ok": + lines += _table( + ["side", "session", "hooks", "uart banner", "faults", "events"], + [ + [ + side, + f"`{behavior[side].get('session')}`", + ", ".join(behavior[side].get("hooks") or []), + behavior[side].get("uart_banner") or "-", + behavior[side].get("fault_count"), + sum((behavior[side].get("event_counts") or {}).values()), + ] + for side in ("a", "b") + ], + ) + lines += [f"* zero deltas: {'yes' if behavior['all_zero'] else 'no'}"] + if behavior.get("compare_report"): + lines += [f"* compare report: `{behavior['compare_report']}`"] + lines += [""] + for name in ("bp", "watch", "writes"): + rows = [row for row in behavior["deltas"][name] if row["delta"] != 0] + rows.sort(key=lambda row: (-abs(row["delta"]), str(row["key"]))) + lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""] + if rows: + lines += _table(["key", "old", "new", "delta"], + [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]]) + else: + lines += ["_no deltas_"] + lines += [""] + lines += [f"> {NO_REFERENCE}", ""] + lines += _table( + ["#", "metric", "old observed", "new observed"], + [[row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])] + for row in behavior["metrics"]], + ) + lines += [""] + + summary = report["summary"] + lines += ["## Summary", ""] + lines += [f"* ok: {', '.join(summary['sections_ok']) or '-'}"] + if summary["sections_skipped"]: + for name, reason in sorted(summary["sections_skipped"].items()): + lines += [f"* skipped `{name}`: {reason}"] + if summary["sections_error"]: + for name, reason in sorted(summary["sections_error"].items()): + lines += [f"* error `{name}`: {reason}"] + return "\n".join(lines) + "\n" + + +def _short_json(value, limit: int = 72) -> str: + import json + + if value is None: + return "-" + text = json.dumps(value, sort_keys=True) + return f"`{text}`" if len(text) <= limit else f"`{text[: limit - 1]}...`" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _add_side_symbols(parser: argparse.ArgumentParser, side: str) -> None: + parser.add_argument(f"--{side}-symbols", help="green-eye symbol DB JSON for the %s target" % side) + parser.add_argument(f"--{side}-ghidra-names", help="Ghidra names.tsv for the %s target" % side) + parser.add_argument(f"--{side}-ghidra-index", help="Ghidra decompiled_index.tsv for the %s target" % side) + parser.add_argument( + f"--{side}-ghidra-exports", + help="Ghidra exports.tsv or 3-column ExportForGreenEye TSV for the %s target" % side, + ) + parser.add_argument(f"--{side}-ghidra-bias", type=lambda value: int(value, 0), default=0, + help="bias added to %s Ghidra symbol addresses" % side) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="version_diff", description=__doc__.splitlines()[0]) + parser.add_argument("--old", required=True, help="old target (RISC-V ELF or raw .fwimage)") + parser.add_argument("--new", required=True, help="new target (RISC-V ELF or raw .fwimage)") + parser.add_argument("--out", required=True, help="output directory for the report, sessions and DBs") + parser.add_argument("--old-label", help="report label for the old target (default: file stem)") + parser.add_argument("--new-label", help="report label for the new target (default: file stem)") + parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23") + parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})") + _add_side_symbols(parser, "old") + _add_side_symbols(parser, "new") + parser.add_argument("--corpus-diff", nargs=2, metavar=("OLD_TSV", "NEW_TSV"), + help="diff two decompiled_index.tsv corpus exports") + parser.add_argument("--corpus-report", help="attach a previously produced corpus diff report") + parser.add_argument("--stimulus", help="green-eye script replacing the builtin demo scenario") + parser.add_argument("--probe-instructions", type=int, default=2000, + help="probe step budget (0 = wall-clock --probe-duration)") + parser.add_argument("--probe-duration", type=float, default=0.5, + help="probe wall budget when --probe-instructions=0") + parser.add_argument("--behavior-duration", type=float, default=0.4, + help="budget for each behavior phase") + parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per phase") + parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports") + parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES, help="QEMU log window per fault check") + return parser + + +def options_from_args(args: argparse.Namespace) -> VersionDiffOptions: + corpus = tuple(args.corpus_diff) if args.corpus_diff else None + return VersionDiffOptions( + old=args.old, + new=args.new, + out=os.path.abspath(args.out), + old_label=args.old_label, + new_label=args.new_label, + release=args.release, + old_symbols=args.old_symbols, + new_symbols=args.new_symbols, + old_ghidra_names=args.old_ghidra_names, + new_ghidra_names=args.new_ghidra_names, + old_ghidra_index=args.old_ghidra_index, + new_ghidra_index=args.new_ghidra_index, + old_ghidra_exports=args.old_ghidra_exports, + new_ghidra_exports=args.new_ghidra_exports, + old_ghidra_bias=args.old_ghidra_bias, + new_ghidra_bias=args.new_ghidra_bias, + corpus_diff=corpus, # type: ignore[arg-type] + corpus_report=args.corpus_report, + stimulus=args.stimulus, + probe_instructions=args.probe_instructions, + probe_duration=args.probe_duration, + behavior_duration=args.behavior_duration, + max_stops=args.max_stops, + max_rows=args.max_rows, + log_bytes=args.log_bytes, + approval_token=args.approval_token, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + options = options_from_args(args) + try: + report = run_version_diff(options) + except Rm610GateError as exc: + print(f"[version-diff] RM610 gate refused: {exc}", file=sys.stderr) + return 3 + except GreenEyeError as exc: + print(f"[version-diff] error: {exc}", file=sys.stderr) + return 2 + print(f"[version-diff] {report['old']['label']} -> {report['new']['label']}") + for name in ("image", "symbols", "corpus", "probe", "behavior"): + section = report["sections"][name] + print(f"[version-diff] {name}: status={section['status']}" + + (f" reason={section['reason']}" if section.get("reason") else "")) + print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.json')}") + print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.md')}") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) |