#!/usr/bin/env python3 """Automatically accept an exact release binding under an ACTIVE campaign.""" from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import stat import tempfile from typing import Any, Mapping from release_campaign_authorization import CampaignAuthorization, CampaignError from trusted_release_execution_plan import PlanError, TrustedReleasePlan class AcceptanceError(RuntimeError): pass def secure_json(path: Path) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise AcceptanceError(f"invalid_file:{path}") st = path.lstat() if st.st_uid != 0 or stat.S_IMODE(st.st_mode) & 0o022: raise AcceptanceError(f"unsafe_file:{path}") value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise AcceptanceError(f"json_object_required:{path}") return value def atomic_new(path: Path, value: Mapping[str, Any]) -> None: if path.exists() or path.is_symlink(): raise AcceptanceError("acceptance_receipt_exists") path.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: os.fchmod(fd, 0o600) if os.geteuid() == 0: os.fchown(fd, 0, 0) with os.fdopen(fd, "w", encoding="utf-8", closefd=True) as stream: json.dump(dict(value), stream, sort_keys=True, indent=2) stream.write("\n") stream.flush() os.fsync(stream.fileno()) os.replace(name, path) os.chmod(path, 0o600) finally: try: os.unlink(name) except FileNotFoundError: pass def evaluate( campaign: CampaignAuthorization, binding: Mapping[str, Any], evidence: Mapping[str, Any], artifact: Mapping[str, Any], staging: Mapping[str, Any], main: Mapping[str, Any], tag: Mapping[str, Any], release_plan: Mapping[str, Any] | None = None, ) -> dict[str, Any]: required_binding = { "version", "target_oid", "evidence_closure_oid", "artifact_sha256", "artifact_size", "approval_plan_id", } if set(binding) != required_binding: raise AcceptanceError("binding_fields_invalid") if main.get("repository") != "aios/aios": raise AcceptanceError("repository_identity_mismatch") if main.get("target_in_main_lineage") is not True: raise AcceptanceError("target_outside_main_lineage") if evidence.get("object_type") != "commit": raise AcceptanceError("evidence_not_commit") if evidence.get("target_is_ancestor") is not True: raise AcceptanceError("target_not_evidence_ancestor") if evidence.get("closure_oid") != binding["evidence_closure_oid"]: raise AcceptanceError("evidence_oid_mismatch") if artifact.get("sha256") != binding["artifact_sha256"]: raise AcceptanceError("artifact_digest_mismatch") if artifact.get("size") != binding["artifact_size"]: raise AcceptanceError("artifact_size_mismatch") if artifact.get("retained") is not True: raise AcceptanceError("artifact_not_retained") if staging.get("install") != "PASS": raise AcceptanceError("staging_install_failed") if staging.get("health") != "PASS": raise AcceptanceError("staging_health_failed") if staging.get("restart") != "PASS": raise AcceptanceError("staging_restart_failed") if staging.get("rollback_readiness") != "PASS": raise AcceptanceError("rollback_readiness_failed") if tag.get("conflict") is not False: raise AcceptanceError("tag_conflict") trusted_plan: TrustedReleasePlan | None = None if release_plan is not None: try: trusted_plan = TrustedReleasePlan.from_mapping(release_plan) trusted_plan.assert_binding(binding) except PlanError as exc: raise AcceptanceError(f"trusted_release_plan_invalid:{exc}") from exc if trusted_plan.value["campaign_authorization_sha256"] != campaign.digest: raise AcceptanceError("campaign_authorization_digest_mismatch") if int(trusted_plan.value["campaign_generation"]) != int(campaign.data["generation"]): raise AcceptanceError("campaign_generation_mismatch") if main.get("tree_oid") not in (None, trusted_plan.value["tree_oid"]): raise AcceptanceError("tree_oid_mismatch") if evidence.get("release_plan_sha256") not in (None, trusted_plan.release_plan_sha256): raise AcceptanceError("evidence_release_plan_mismatch") if artifact.get("manifest_sha256") not in (None, trusted_plan.value["manifest_sha256"]): raise AcceptanceError("manifest_digest_mismatch") if artifact.get("package_closure_sha256") not in (None, trusted_plan.value["package_closure_sha256"]): raise AcceptanceError("package_closure_digest_mismatch") campaign.assert_operation("derive_campaign_acceptance") result = { "schema": "aios.campaign-release-acceptance.v1", "accepted": True, "binding_replaced": False, "campaign_id": campaign.campaign_id, "campaign_generation": campaign.data["generation"], **dict(binding), "owner_interaction_required": False, } if trusted_plan is not None: result.update({ "candidate_id": trusted_plan.value["candidate_id"], "release_plan_sha256": trusted_plan.release_plan_sha256, "candidate_binding_sha256": trusted_plan.candidate_binding_sha256, "campaign_authorization_sha256": trusted_plan.value["campaign_authorization_sha256"], "campaign_generation": trusted_plan.value["campaign_generation"], "tree_oid": trusted_plan.value["tree_oid"], "manifest_sha256": trusted_plan.value["manifest_sha256"], "package_closure_sha256": trusted_plan.value["package_closure_sha256"], "deployment_plan_sha256": trusted_plan.value["deployment_plan_sha256"], "policy_sha256": trusted_plan.value["policy_sha256"], "rollback": trusted_plan.value["rollback"], "baseline": trusted_plan.value["baseline"], }) return result def main() -> int: p = argparse.ArgumentParser() p.add_argument("--campaign", type=Path, required=True) p.add_argument("--revocation", type=Path) p.add_argument("--binding", type=Path, required=True) p.add_argument("--evidence", type=Path, required=True) p.add_argument("--artifact", type=Path, required=True) p.add_argument("--staging", type=Path, required=True) p.add_argument("--main-facts", type=Path, required=True) p.add_argument("--tag-facts", type=Path, required=True) p.add_argument("--release-plan", type=Path) p.add_argument("--output", type=Path, required=True) a = p.parse_args() campaign = CampaignAuthorization.load( a.campaign, revocation=a.revocation, require_active=True, ) receipt = evaluate( campaign, secure_json(a.binding), secure_json(a.evidence), secure_json(a.artifact), secure_json(a.staging), secure_json(a.main_facts), secure_json(a.tag_facts), secure_json(a.release_plan) if a.release_plan else None, ) atomic_new(a.output, receipt) print(json.dumps(receipt, sort_keys=True)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (CampaignError, AcceptanceError) as exc: print(json.dumps({"status": "BLOCKED", "reason": str(exc)}, sort_keys=True)) raise SystemExit(2)