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