#!/usr/bin/env python3 """Issue and immediately verify a sanitized target-boundary proof.""" from __future__ import annotations import argparse from copy import deepcopy from datetime import UTC, datetime import json from pathlib import Path import sys META_ROOT = Path(__file__).resolve().parents[2] for tools_root in ( META_ROOT / "tools" / "assessments", META_ROOT / "tools" / "release", ): if str(tools_root) not in sys.path: sys.path.insert(0, str(tools_root)) from cryptography.hazmat.primitives import serialization # noqa: E402 from cryptography.hazmat.primitives.asymmetric.ed25519 import ( # noqa: E402 Ed25519PrivateKey, ) from govoplan_assessment.atomic_io import ( # noqa: E402 AtomicJsonWriteError, atomic_write_json, ) from govoplan_assessment.boundary_evidence import ( # noqa: E402 issue_boundary_evidence, ) from govoplan_assessment.capability_fit import review_capability_fit # noqa: E402 from govoplan_assessment.evidence import ( # noqa: E402 public_key_material_from_keyrings, validate_payload, ) MAX_INPUT_BYTES = 16 * 1024 * 1024 MAX_OUTPUT_BYTES = 4 * 1024 * 1024 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--assessment", type=Path, required=True) parser.add_argument( "--assessment-schema", type=Path, default=META_ROOT / "docs" / "capability-fit.schema.json", ) parser.add_argument("--catalog", type=Path, required=True) parser.add_argument("--keyring", type=Path, required=True) parser.add_argument("--trusted-keyring", type=Path, required=True) parser.add_argument("--installed-evidence", type=Path, required=True) parser.add_argument( "--installed-evidence-schema", type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json", ) parser.add_argument("--installer-receipt", type=Path, required=True) parser.add_argument( "--installer-receipt-schema", type=Path, default=META_ROOT / "docs" / "installer-receipt.schema.json", ) parser.add_argument("--installer-authority-keyring", type=Path, required=True) parser.add_argument( "--installer-authority-keyring-schema", type=Path, default=META_ROOT / "docs" / "installer-receipt-authority-keyring.schema.json", ) parser.add_argument( "--claims", type=Path, required=True, help="Private target-run claim manifest; artifact paths are not published.", ) parser.add_argument( "--claims-schema", type=Path, default=META_ROOT / "docs" / "capability-fit-boundary-run.schema.json", ) parser.add_argument("--authority-keyring", type=Path, required=True) parser.add_argument( "--authority-keyring-schema", type=Path, default=META_ROOT / "docs" / "capability-fit-proof-authority-keyring.schema.json", ) parser.add_argument( "--signing-key", action="append", default=[], required=True, metavar="KEY_ID=/PATH/PRIVATE.pem", help="Repeat for independent authorities needed to cover all claim scopes.", ) parser.add_argument("--expected-external-provider-subject") parser.add_argument("--output", type=Path, required=True) parser.add_argument("--review-output", type=Path, required=True) parser.add_argument( "--boundary-evidence-schema", type=Path, default=META_ROOT / "docs" / "capability-fit-boundary-evidence.schema.json", ) args = parser.parse_args(argv) if args.output.resolve() == args.review_output.resolve(): parser.error("proof and review output paths must differ") assessment = read_object(args.assessment, label="assessment") assessment_schema = read_object(args.assessment_schema, label="assessment schema") catalog = read_object(args.catalog, label="catalog") published_keyring = read_object(args.keyring, label="published keyring") trusted_keyring = read_object(args.trusted_keyring, label="trusted keyring") installed_evidence = read_object( args.installed_evidence, label="installed evidence" ) installed_schema = read_object( args.installed_evidence_schema, label="installed evidence schema" ) installer_receipt = read_object(args.installer_receipt, label="installer receipt") installer_receipt_schema = read_object( args.installer_receipt_schema, label="installer receipt schema" ) installer_authorities = read_object( args.installer_authority_keyring, label="installer authority keyring" ) installer_authority_schema = read_object( args.installer_authority_keyring_schema, label="installer authority keyring schema", ) claim_manifest = read_object(args.claims, label="target-run claim manifest") claim_schema = read_object(args.claims_schema, label="target-run claim schema") authority_keyring = read_object( args.authority_keyring, label="proof-authority keyring" ) authority_schema = read_object( args.authority_keyring_schema, label="proof-authority keyring schema" ) boundary_schema = read_object( args.boundary_evidence_schema, label="boundary evidence schema" ) _require_schema(claim_manifest, claim_schema, label="target-run claim manifest") _require_schema( authority_keyring, authority_schema, label="proof-authority keyring" ) base_report = review_capability_fit( assessment=assessment, schema=assessment_schema, catalog=catalog, published_keyring=published_keyring, trusted_keyring=trusted_keyring, workspace_root=None, installed_evidence=installed_evidence, installed_evidence_schema=installed_schema, installer_receipt=installer_receipt, installer_receipt_schema=installer_receipt_schema, installer_authority_keyring=installer_authorities, installer_authority_keyring_schema=installer_authority_schema, installed_evidence_mode="imported_unsigned", ) _require_release_origin(base_report, parser=parser) claims = resolve_claim_artifact_paths( claim_manifest.get("claims"), base_dir=args.claims.resolve().parent ) signing_keys = read_signing_keys(tuple(args.signing_key)) expires_at = parse_datetime(claim_manifest.get("expires_at"), label="expires_at") forbidden_keys = public_key_material_from_keyrings( published_keyring, trusted_keyring, installer_authorities, ) try: proof = issue_boundary_evidence( assessment=assessment, installed_evidence=installed_evidence, proof_id=str(claim_manifest.get("proof_id") or ""), claims=claims, authority_keyring=authority_keyring, signing_keys=signing_keys, expires_at=expires_at, expected_external_provider_subject=( args.expected_external_provider_subject ), forbidden_authority_public_keys=forbidden_keys, release_origin_verified=True, ) except ValueError as exc: parser.error(str(exc)) _require_schema(proof, boundary_schema, label="issued boundary evidence") review = review_capability_fit( assessment=assessment, schema=assessment_schema, catalog=catalog, published_keyring=published_keyring, trusted_keyring=trusted_keyring, workspace_root=None, installed_evidence=installed_evidence, installed_evidence_schema=installed_schema, installer_receipt=installer_receipt, installer_receipt_schema=installer_receipt_schema, installer_authority_keyring=installer_authorities, installer_authority_keyring_schema=installer_authority_schema, boundary_evidence=proof, boundary_evidence_schema=boundary_schema, boundary_authority_keyring=authority_keyring, boundary_authority_keyring_schema=authority_schema, installed_evidence_mode="imported_unsigned", expected_external_provider_subject=args.expected_external_provider_subject, ) supplied_scopes = {str(claim.get("scope")) for claim in proof["claims"]} unchecked = sorted( scope for scope in supplied_scopes if review.get("proof_scope", {}).get(scope, {}).get("checked") is not True ) if unchecked: parser.error( "issued proof did not pass immediate verification for scopes: " + ", ".join(unchecked) ) write_object(args.output, proof, label="boundary evidence") write_object(args.review_output, review, label="capability-fit review") print( json.dumps( { "proof_id": proof["proof_id"], "scopes": sorted(supplied_scopes), "reference_readiness": review.get("proof_scope", {}).get( "reference_readiness" ), "output": str(args.output), "review_output": str(args.review_output), }, sort_keys=True, ) ) return 0 def read_object(path: Path, *, label: str) -> dict[str, object]: try: if path.stat().st_size > MAX_INPUT_BYTES: raise ValueError("input exceeds size limit") payload = json.loads(path.read_bytes().decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: raise SystemExit(f"Could not read {label}") from exc if not isinstance(payload, dict): raise SystemExit(f"{label.capitalize()} must be a JSON object") return payload def read_signing_keys(values: tuple[str, ...]) -> dict[str, Ed25519PrivateKey]: result: dict[str, Ed25519PrivateKey] = {} for value in values: key_id, separator, path_text = value.partition("=") if not separator or not key_id or not path_text or key_id in result: raise SystemExit( "--signing-key must uniquely use KEY_ID=/path/to/private.pem" ) try: key = serialization.load_pem_private_key( Path(path_text).expanduser().read_bytes(), password=None ) except (OSError, ValueError) as exc: raise SystemExit("Could not read boundary signing key") from exc if not isinstance(key, Ed25519PrivateKey): raise SystemExit("Boundary signing keys must be Ed25519") result[key_id] = key return result def resolve_claim_artifact_paths( claims: object, *, base_dir: Path ) -> list[dict[str, object]]: if not isinstance(claims, list): raise SystemExit("Target-run claim manifest has no claim list") resolved = deepcopy(claims) for claim in resolved: if not isinstance(claim, dict): continue artifacts = claim.get("artifacts") if not isinstance(artifacts, list): continue for artifact in artifacts: if not isinstance(artifact, dict) or not isinstance( artifact.get("path"), str ): continue path = Path(artifact["path"]).expanduser() artifact["path"] = path if path.is_absolute() else base_dir / path return resolved def parse_datetime(value: object, *, label: str) -> datetime: if not isinstance(value, str): raise SystemExit(f"{label} must be an RFC 3339 date-time") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise SystemExit(f"{label} must be an RFC 3339 date-time") from exc if parsed.tzinfo is None: raise SystemExit(f"{label} must include a timezone") return parsed.astimezone(UTC) def _require_schema( payload: dict[str, object], schema: dict[str, object], *, label: str ) -> None: errors = validate_payload(payload=payload, schema=schema) if errors: raise SystemExit(f"{label.capitalize()} failed schema validation: {errors[0]}") def _require_release_origin( report: dict[str, object], *, parser: argparse.ArgumentParser ) -> None: proof_scope = report.get("proof_scope") if not isinstance(proof_scope, dict): parser.error("capability-fit review did not return proof scope") required = ( "catalog_signature_and_keyring", "catalog_signature_and_trusted_keyring", "published_keyring_hash", "release_metadata", "installed_artifacts", "installed_record_integrity", "installed_release_origin", ) failed = sorted( scope for scope in required if not isinstance(proof_scope.get(scope), dict) or proof_scope[scope].get("valid") is not True ) observation = proof_scope.get("installed_evidence_observation") if ( not isinstance(observation, dict) or observation.get("receipt_authenticated") is not True ): failed.append("installed_evidence_observation.receipt_authenticated") if failed: parser.error( "trusted release and installer evidence must pass before target proof issuance: " + ", ".join(failed) ) def write_object(path: Path, payload: dict[str, object], *, label: str) -> None: try: atomic_write_json(path, payload, max_bytes=MAX_OUTPUT_BYTES) except AtomicJsonWriteError as exc: raise SystemExit(f"Could not securely write {label}") from exc if __name__ == "__main__": raise SystemExit(main())