#!/usr/bin/env python3 """Re-run a capability-fit assessment against a trusted release catalog.""" from __future__ import annotations import argparse import json import os from pathlib import Path import sys META_ROOT = Path(__file__).resolve().parents[2] ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): if str(tools_root) not in sys.path: sys.path.insert(0, str(tools_root)) from govoplan_assessment import ( # noqa: E402 collect_installed_composition, render_review, review_capability_fit, ) from govoplan_assessment.evidence import validate_payload # noqa: E402 from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402 TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE" MAX_ASSESSMENT_INPUT_BYTES = 16 * 1024 * 1024 MAX_EVIDENCE_INPUT_BYTES = 4 * 1024 * 1024 def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate a fit assessment and identify conclusions needing review after release drift." ) parser.add_argument( "--assessment", type=Path, default=META_ROOT / "docs" / "capability-fit-current.json", ) parser.add_argument( "--schema", type=Path, default=META_ROOT / "docs" / "capability-fit.schema.json", ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument( "--public", action="store_true", help="Use the fixed public GovOPlaN stable catalog and keyring URLs.", ) source.add_argument( "--catalog", type=Path, help="Read a catalog from a local file." ) parser.add_argument( "--keyring", type=Path, help="Published or candidate keyring whose hash is pinned by --catalog.", ) parser.add_argument( "--trusted-keyring", type=Path, help=( "Independently pinned local keyring used as the catalog signature trust " f"root; defaults to ${TRUSTED_KEYRING_FILE_ENV}." ), ) parser.add_argument( "--workspace-root", type=Path, default=META_ROOT.parent, help="Workspace containing repository checkouts for local tag provenance.", ) parser.add_argument( "--skip-tag-provenance", action="store_true", help="Compare signed metadata without checking local annotated tags.", ) installed = parser.add_mutually_exclusive_group() installed.add_argument( "--installed-evidence", type=Path, help="Validate an existing bounded installed-composition evidence document.", ) installed.add_argument( "--collect-installed-evidence", type=Path, help=( "Collect the current interpreter's GovOPlaN distributions, write the " "bounded evidence document, and include it in this review. Loading " "module entry points executes installed GovOPlaN manifest factories." ), ) parser.add_argument( "--installed-evidence-schema", type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json", ) parser.add_argument( "--boundary-evidence", type=Path, help=( "Externally issued target/provider/production proof bound to the " "installed-evidence digest." ), ) parser.add_argument( "--boundary-evidence-schema", type=Path, default=META_ROOT / "docs" / "capability-fit-boundary-evidence.schema.json", ) parser.add_argument( "--boundary-authority-keyring", type=Path, help=( "Independently provisioned keyring authorizing proof signers for " "specific target/provider/production scopes." ), ) parser.add_argument( "--boundary-authority-keyring-schema", type=Path, default=META_ROOT / "docs" / "capability-fit-proof-authority-keyring.schema.json", ) parser.add_argument( "--verification-time", help="Optional ISO-8601 time for reproducible boundary-proof verification.", ) parser.add_argument( "--json", action="store_true", help="Print the machine-readable review report." ) parser.add_argument( "--output", type=Path, help="Also write the machine-readable review report to this path.", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if args.catalog is not None and args.keyring is None: raise SystemExit("--keyring is required with --catalog") if args.public and args.keyring is not None: raise SystemExit("--keyring cannot be combined with --public") if (args.boundary_evidence is None) != (args.boundary_authority_keyring is None): raise SystemExit( "--boundary-evidence and --boundary-authority-keyring are required together" ) if args.boundary_evidence is not None and ( args.installed_evidence is None and args.collect_installed_evidence is None ): raise SystemExit( "--boundary-evidence requires --installed-evidence or --collect-installed-evidence" ) if ( args.collect_installed_evidence is not None and args.output is not None and args.collect_installed_evidence.resolve() == args.output.resolve() ): raise SystemExit("Evidence and review output paths must differ") configured_trusted_keyring = os.getenv(TRUSTED_KEYRING_FILE_ENV, "").strip() trusted_keyring_path = args.trusted_keyring or ( Path(configured_trusted_keyring) if configured_trusted_keyring else None ) if trusted_keyring_path is None: raise SystemExit( f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required" ) assessment = read_object( args.assessment, label="assessment", max_bytes=MAX_ASSESSMENT_INPUT_BYTES, ) schema = read_object( args.schema, label="schema", max_bytes=MAX_ASSESSMENT_INPUT_BYTES, ) if args.public: catalog = fetch_public_json( f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/stable.json", label="catalog", ) published_keyring = fetch_public_json( f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json", label="published keyring", ) else: catalog = read_object( args.catalog, label="catalog", max_bytes=MAX_ASSESSMENT_INPUT_BYTES ) published_keyring = read_object( args.keyring, label="published keyring", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) trusted_keyring = read_object( trusted_keyring_path, label="trusted keyring", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) installed_evidence = None installed_evidence_schema = None if ( args.installed_evidence is not None or args.collect_installed_evidence is not None ): installed_evidence_schema = read_object( args.installed_evidence_schema, label="installed evidence schema", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) if args.installed_evidence is not None: installed_evidence = read_object( args.installed_evidence, label="installed evidence", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) else: installed_evidence = collect_installed_composition(assessment=assessment) collection_schema_errors = validate_payload( payload=installed_evidence, schema=installed_evidence_schema, ) if collection_schema_errors: raise SystemExit( "Collected installed evidence did not satisfy its bounded schema" ) write_object(args.collect_installed_evidence, installed_evidence) boundary_evidence = None boundary_evidence_schema = None boundary_authority_keyring = None boundary_authority_keyring_schema = None if args.boundary_evidence is not None: boundary_evidence = read_object( args.boundary_evidence, label="boundary evidence", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) boundary_evidence_schema = read_object( args.boundary_evidence_schema, label="boundary evidence schema", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) boundary_authority_keyring = read_object( args.boundary_authority_keyring, label="boundary authority keyring", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) boundary_authority_keyring_schema = read_object( args.boundary_authority_keyring_schema, label="boundary authority keyring schema", max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) report = review_capability_fit( assessment=assessment, schema=schema, catalog=catalog, published_keyring=published_keyring, trusted_keyring=trusted_keyring, workspace_root=None if args.skip_tag_provenance else args.workspace_root.resolve(), installed_evidence=installed_evidence, installed_evidence_schema=installed_evidence_schema, boundary_evidence=boundary_evidence, boundary_evidence_schema=boundary_evidence_schema, boundary_authority_keyring=boundary_authority_keyring, boundary_authority_keyring_schema=boundary_authority_keyring_schema, verification_time=parse_datetime(args.verification_time), ) encoded = json.dumps(report, indent=2, sort_keys=True) + "\n" if args.output is not None: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(encoded, encoding="utf-8") sys.stdout.write(encoded if args.json else render_review(report)) return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]] def read_object(path: Path | None, *, label: str, max_bytes: int) -> dict[str, object]: if path is None: raise SystemExit(f"Missing {label} path") try: if path.stat().st_size > max_bytes: raise SystemExit( f"{label.capitalize()} exceeds the {max_bytes}-byte input limit" ) with path.open("rb") as handle: encoded = handle.read(max_bytes + 1) if len(encoded) > max_bytes: raise SystemExit( f"{label.capitalize()} exceeds the {max_bytes}-byte input limit" ) payload = json.loads(encoded.decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise SystemExit(f"Could not read {label} {path}: {exc}") from exc if not isinstance(payload, dict): raise SystemExit(f"{label.capitalize()} must be a JSON object: {path}") return payload def write_object(path: Path | None, payload: dict[str, object]) -> None: if path is None: raise SystemExit("Missing installed evidence output path") encoded = json.dumps(payload, indent=2, sort_keys=True) + "\n" if len(encoded.encode("utf-8")) > MAX_EVIDENCE_INPUT_BYTES: raise SystemExit( "Collected installed evidence exceeds the bounded output limit" ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(encoded, encoding="utf-8") def parse_datetime(value: str | None): if value is None: return None from datetime import datetime try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise SystemExit("--verification-time must be an ISO-8601 timestamp") from exc if parsed.tzinfo is None: raise SystemExit("--verification-time must include a timezone") return parsed def fetch_public_json(url: str, *, label: str) -> dict[str, object]: result = fetch_json(url) if result.get("ok") is not True or not isinstance(result.get("payload"), dict): raise SystemExit(f"Could not fetch fixed public {label} endpoint.") return result["payload"] if __name__ == "__main__": raise SystemExit(main())