#!/usr/bin/env python3 """Collect live installed payloads and issue an independently signed receipt.""" from __future__ import annotations import argparse 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 import collect_installed_composition # noqa: E402 from govoplan_assessment.atomic_io import ( # noqa: E402 AtomicJsonWriteError, atomic_write_json, ) from govoplan_assessment.capability_fit import ( # noqa: E402 canonical_hash, review_capability_fit, ) from govoplan_assessment.evidence import validate_payload # noqa: E402 from govoplan_assessment.installer_receipt import ( # noqa: E402 issue_installer_receipt, ) 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("--receipt-id", required=True) parser.add_argument("--signing-key", required=True, metavar="KEY_ID=/PATH/PRIVATE.pem") parser.add_argument( "--python-artifact", action="append", default=[], required=True, metavar="PACKAGE=/PATH/TO/CONSUMED.whl", help="Exact wheel consumed by this installer; repeat for every package.", ) parser.add_argument("--evidence-output", type=Path, required=True) parser.add_argument("--receipt-output", type=Path, required=True) parser.add_argument("--installed-evidence-schema", type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json") parser.add_argument("--receipt-schema", type=Path, default=META_ROOT / "docs" / "installer-receipt.schema.json") args = parser.parse_args(argv) if args.evidence_output.resolve() == args.receipt_output.resolve(): parser.error("evidence and receipt 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") keyring = read_object(args.keyring, label="published keyring") trusted_keyring = read_object(args.trusted_keyring, label="trusted keyring") evidence_schema = read_object(args.installed_evidence_schema, label="installed evidence schema") receipt_schema = read_object(args.receipt_schema, label="installer receipt schema") release = catalog.get("release") if not isinstance(release, dict) or release.get("keyring_sha256") != canonical_hash(keyring): parser.error("catalog does not pin the supplied published keyring") evidence = collect_installed_composition(assessment=assessment) if validate_payload(payload=evidence, schema=evidence_schema): parser.error("live installed observation does not satisfy its bounded schema") report = review_capability_fit( assessment=assessment, schema=assessment_schema, catalog=catalog, published_keyring=keyring, trusted_keyring=trusted_keyring, workspace_root=None, installed_evidence=evidence, installed_evidence_schema=evidence_schema, installed_evidence_mode="direct_local", ) required_scopes = ( "catalog_signature_and_keyring", "catalog_signature_and_trusted_keyring", "published_keyring_hash", "release_metadata", "installed_artifacts", "installed_record_integrity", ) if any(report["proof_scope"].get(scope, {}).get("valid") is not True for scope in required_scopes): parser.error("trusted catalog and live installed-composition checks must pass before receipt issuance") key_id, private_key = read_signing_key(args.signing_key) consumed_artifacts = parse_package_paths(tuple(args.python_artifact)) receipt = issue_installer_receipt( assessment=assessment, catalog=catalog, installed_evidence=evidence, receipt_id=args.receipt_id, key_id=key_id, private_key=private_key, consumed_artifacts=consumed_artifacts, same_process_observation=True, ) if validate_payload(payload=receipt, schema=receipt_schema): parser.error("issued installer receipt does not satisfy its bounded schema") write_object(args.evidence_output, evidence, label="installed evidence") write_object(args.receipt_output, receipt, label="installer receipt") print(json.dumps({"receipt_id": args.receipt_id, "evidence_output": str(args.evidence_output), "receipt_output": str(args.receipt_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") encoded = path.read_bytes() payload = json.loads(encoded.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_key(value: str) -> tuple[str, Ed25519PrivateKey]: key_id, separator, path_text = value.partition("=") if not separator or not key_id or not path_text: raise SystemExit("--signing-key must 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 installer signing key") from exc if not isinstance(key, Ed25519PrivateKey): raise SystemExit("Installer signing key must be Ed25519") return key_id, key def parse_package_paths(values: tuple[str, ...]) -> dict[str, Path]: result: dict[str, Path] = {} for value in values: package, separator, path_text = value.partition("=") package = package.strip() path_text = path_text.strip() if not separator or not package or not path_text or package in result: raise SystemExit( "--python-artifact must uniquely use PACKAGE=/path/to/consumed.whl" ) result[package] = Path(path_text).expanduser() return result 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())