#!/usr/bin/env python3 """AIOS durable autonomous release-campaign authorization. The campaign receipt represents one explicit Owner policy delegation. It is not a reusable Owner credential and it is not an unrestricted Git/production token. Normal release operations are allowed only when all machine gates in the campaign policy succeed. Formal tag conflicts, non-fast-forward main divergence, history rewrite, private-signing-material exposure, repository identity conflict or campaign revocation remain fail-closed/manual-review conditions. """ from __future__ import annotations from dataclasses import dataclass import hashlib import json import os from pathlib import Path import re import stat from typing import Any, Mapping OID_RE = re.compile(r"^[0-9a-f]{40}$") CAMPAIGN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{3,127}$") SCHEMA = "aios.release-campaign-authorization.v1" ALLOWED_ENGINEERING = frozenset({ "create_task_branch", "create_worktree", "append_commit", "push_task_branch", "repair_task_branch", "create_pr", "update_pr", "run_required_checks", "publish_truthful_status", "protected_merge", }) ALLOWED_RECONCILIATION = frozenset({ "fetch_gitea_main", "verify_ancestor", "authority_main_ff_cas", "refresh_runtime_context", }) ALLOWED_RELEASE = frozenset({ "deterministic_build", "retain_artifact", "materialize_approved_artifact", "staging_install", "health_check", "restart_test", "rollback_readiness_test", "create_evidence_commit", "generate_exact_binding", "derive_campaign_acceptance", }) ALLOWED_PRODUCTION = frozenset({ "create_annotated_tag", "publish_tag", "verify_tag", "deploy_exact_artifact", "verify_production_health", "update_runtime_registry", "update_production_baseline", "update_baseline_sidecar", }) DENIED = frozenset({ "force_push", "non_fast_forward_main_write", "direct_main_push", "published_history_rewrite", "formal_tag_rewrite", "formal_tag_delete", "overwrite_conflicting_same_name_tag", "expose_private_signing_material", "commit_plaintext_secret", "repository_identity_change", "weaken_branch_protection", }) class CampaignError(RuntimeError): pass def canonical_json(value: Any) -> bytes: return ( json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n" ).encode("utf-8") def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def secure_json(path: Path) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise CampaignError(f"invalid_file:{path}") st = path.lstat() if st.st_uid != 0: raise CampaignError(f"root_owner_required:{path}") if stat.S_IMODE(st.st_mode) & 0o022: raise CampaignError(f"group_world_writable_rejected:{path}") try: value = json.loads(path.read_text(encoding="utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CampaignError(f"invalid_json:{path}") from exc if not isinstance(value, dict): raise CampaignError(f"json_object_required:{path}") return value def _exact_set(value: Any, expected: frozenset[str], name: str) -> None: if not isinstance(value, list) or frozenset(value) != expected: raise CampaignError(f"{name}_mismatch") def validate_policy(value: Mapping[str, Any]) -> dict[str, Any]: required = { "schema", "campaign_id", "generation", "owner_identity", "owner_role", "repository", "lineage_anchor_oid", "control_plane_branch", "activation_state", "active", "revoked", "valid_until_revoked", "allowed_engineering", "allowed_reconciliation", "allowed_release", "allowed_production", "denied", "normal_owner_interaction_required", } if set(value) != required: raise CampaignError("campaign_fields_invalid") if value["schema"] != SCHEMA: raise CampaignError("campaign_schema_invalid") if not isinstance(value["campaign_id"], str) or not CAMPAIGN_RE.fullmatch(value["campaign_id"]): raise CampaignError("campaign_id_invalid") if value["generation"] != 1: raise CampaignError("campaign_generation_invalid") if value["owner_identity"] != "aiman" or value["owner_role"] != "owner": raise CampaignError("campaign_owner_invalid") if value["repository"] != "aios/aios": raise CampaignError("campaign_repository_invalid") if not isinstance(value["lineage_anchor_oid"], str) or not OID_RE.fullmatch(value["lineage_anchor_oid"]): raise CampaignError("campaign_anchor_invalid") branch = value["control_plane_branch"] if branch != "agent/autonomous-release-campaign-v1-20260807": raise CampaignError("campaign_branch_invalid") if value["activation_state"] not in {"PENDING_CONTROL_PLANE_MERGE", "ACTIVE"}: raise CampaignError("campaign_activation_state_invalid") if value["revoked"] is not False or value["valid_until_revoked"] is not True: raise CampaignError("campaign_lifecycle_invalid") if value["normal_owner_interaction_required"] is not False: raise CampaignError("campaign_owner_interaction_policy_invalid") if value["activation_state"] == "ACTIVE" and value["active"] is not True: raise CampaignError("campaign_active_state_invalid") if value["activation_state"] == "PENDING_CONTROL_PLANE_MERGE" and value["active"] is not False: raise CampaignError("campaign_pending_state_invalid") _exact_set(value["allowed_engineering"], ALLOWED_ENGINEERING, "allowed_engineering") _exact_set(value["allowed_reconciliation"], ALLOWED_RECONCILIATION, "allowed_reconciliation") _exact_set(value["allowed_release"], ALLOWED_RELEASE, "allowed_release") _exact_set(value["allowed_production"], ALLOWED_PRODUCTION, "allowed_production") _exact_set(value["denied"], DENIED, "denied") return dict(value) @dataclass(frozen=True) class CampaignAuthorization: path: Path data: dict[str, Any] digest: str @classmethod def load( cls, path: Path, *, sidecar: Path | None = None, revocation: Path | None = None, require_active: bool = True, ) -> "CampaignAuthorization": data = validate_policy(secure_json(path)) raw = path.read_bytes() digest = sha256_bytes(raw) sidecar = sidecar or path.with_suffix(path.suffix + ".sha256") if not sidecar.is_file() or sidecar.is_symlink(): raise CampaignError("campaign_digest_sidecar_missing") token = sidecar.read_text(encoding="ascii").strip().split()[0] if token != digest: raise CampaignError("campaign_digest_sidecar_mismatch") if revocation and revocation.exists(): revoked = secure_json(revocation) if revoked.get("campaign_id") == data["campaign_id"] and revoked.get("revoked") is True: raise CampaignError("campaign_revoked") if require_active and (data["activation_state"] != "ACTIVE" or data["active"] is not True): raise CampaignError("campaign_not_active") return cls(path=path, data=data, digest=digest) @property def campaign_id(self) -> str: return str(self.data["campaign_id"]) @property def anchor_oid(self) -> str: return str(self.data["lineage_anchor_oid"]) def assert_operation(self, operation: str) -> None: if operation in DENIED: raise CampaignError(f"campaign_operation_denied:{operation}") universe = ( ALLOWED_ENGINEERING | ALLOWED_RECONCILIATION | ALLOWED_RELEASE | ALLOWED_PRODUCTION ) if operation not in universe: raise CampaignError(f"campaign_operation_out_of_scope:{operation}")