debug: green-eye wave 3 - RM harness, local simulator, device map, MCP plugin, value-validation demos, CI pipeline, toolchain docs; add host RM tools rmtool/rmtop

1 files changed, 586 insertions, 0 deletions

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