#!/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 "" 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())