diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/control_discovery.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/control_discovery.py | 1055 |
1 files changed, 1055 insertions, 0 deletions
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()) |