aios/active-runtime/app/trusted_builder_signer_provisioning.py

#!/usr/bin/env python3
"""Root-only, purpose-bound trusted-builder signer provisioner.

The provisioner consumes one exact Owner-signed request after the normal
engineering review has prepared it.  It creates the Ed25519 private material
and builder HMAC on the host, encrypts both for systemd, publishes only the
public verification key, and appends one replay-protected receipt.  It never
accepts or writes main, tag, baseline, production, or Owner-approval keys.
"""
from __future__ import annotations

import argparse
import base64
import datetime as dt
try:
    import fcntl
except ImportError:  # pragma: no cover - Windows only affects local payload tooling.
    class _NoopFcntl:
        LOCK_EX = 0

        @staticmethod
        def flock(_file, _mode):
            return None

    fcntl = _NoopFcntl()
import hashlib
import json
import os
from pathlib import Path
import secrets
import shutil
import stat
import subprocess
import tempfile
from typing import Any, Mapping


PURPOSE = "AIOS_TRUSTED_RELEASE_BUILDER"
TRUST_DOMAIN = "aios.trusted-builder.attestation.v1"
OWNER_NAMESPACE = "aios-owner-approval-v1"
OWNER_PRINCIPAL = "aios-task-grant-signer"
SCHEMA = "aios.trusted-builder-signer-provisioning.v2"
ALLOWED_CAPABILITIES = frozenset(
    {
        "builder_receipt_signing",
        "artifact_provenance_signing",
        "manifest_provenance_signing",
        "builder_owned_release_evidence_signing",
    }
)
DENIED_CAPABILITIES = frozenset(
    {
        "owner_release_approval",
        "protected_main_merge",
        "arbitrary_ref_write",
        "tag_publication",
        "baseline_finalize",
        "production_runtime_switch",
    }
)
OID_FIELDS = {
    "gitea_main_oid",
    "authority_main_oid_at_capture",
    "governed_main_oid",
    "gitea_main_tree_oid",
    "pr_head_oid",
    "protected_merge_commit",
}
SHA_FIELDS = {"source_evidence_sha256"}
PACKAGE_MANIFEST_PATH = "deploy/trusted-builder/aios-trusted-builder-package-manifest.json"
PACKAGE_MANIFEST_SCHEMA = "aios.trusted-builder-package.v1"
SOURCE_EVIDENCE_PATHS = (
    "active-runtime/app/trusted_builder_signer_provisioning.py",
    "active-runtime/app/aios_trusted_candidate_builder.py",
    "config/gitops/trusted-candidate-builder-policy.json",
    "deploy/systemd/aios-trusted-candidate-builder.service",
    "deploy/systemd/aios-trusted-candidate-builder.socket",
    "deploy/systemd/aios-gitops-model-api-trusted-builder.conf.example",
    "deploy/libexec/aios-trusted-builder-signer-provisioner",
    PACKAGE_MANIFEST_PATH,
)
SOURCE_EVIDENCE_MODES = {path: frozenset({0o600, 0o640, 0o644, 0o664}) for path in SOURCE_EVIDENCE_PATHS}


def package_contract_document() -> dict[str, Any]:
    """The non-secret package/activation contract for a fresh B2 payload."""
    files = []
    for path in SOURCE_EVIDENCE_PATHS:
        install_mode = "0755" if path == "deploy/libexec/aios-trusted-builder-signer-provisioner" else "0644"
        files.append({"source_path": path, "source_mode": "0644", "install_mode": install_mode})
    return {
        "schema": PACKAGE_MANIFEST_SCHEMA,
        "version": 1,
        "package_name": "aios-trusted-candidate-builder",
        "source_root": "/opt/aios/trusted-candidate-builder/current",
        "files": files,
        "activation": {
            "daemon_reload": True,
            "enable_units": ["aios-trusted-candidate-builder.socket"],
            "service_start": "socket-activation-only",
            "credential_installation": False,
            "signer_invocation": False,
        },
    }


def _source_evidence(source_root: Path, source_files: list[str] | tuple[str, ...]) -> dict[str, str]:
    if set(source_files) != set(SOURCE_EVIDENCE_PATHS) or len(source_files) != len(SOURCE_EVIDENCE_PATHS):
        raise ProvisioningError("SOURCE_EVIDENCE_SET_INVALID")
    evidence: dict[str, str] = {}
    for path in SOURCE_EVIDENCE_PATHS:
        candidate = source_root / path
        if candidate.is_symlink() or not candidate.is_file():
            raise ProvisioningError("SOURCE_EVIDENCE_MISSING")
        if stat.S_IMODE(candidate.stat().st_mode) not in SOURCE_EVIDENCE_MODES[path]:
            raise ProvisioningError("SOURCE_EVIDENCE_MODE_INVALID")
        evidence[path] = _digest_file(candidate)
    return evidence


def verify_package_contract(source_root: Path) -> None:
    manifest = source_root / PACKAGE_MANIFEST_PATH
    raw, value = _read_json(manifest)
    if value != package_contract_document() or raw != pretty_json(package_contract_document()):
        raise ProvisioningError("PACKAGE_CONTRACT_INVALID")
    _source_evidence(source_root, list(SOURCE_EVIDENCE_PATHS))


class ProvisioningError(RuntimeError):
    """Stable fail-closed error code; never includes secret material."""

    def __init__(self, code: str) -> None:
        super().__init__(code)
        self.code = code


def canonical_json(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def pretty_json(value: Any) -> bytes:
    return (json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode("utf-8")


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def utc_now() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)


def iso(value: dt.datetime) -> str:
    return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z")


def _require_oid(value: Any, code: str) -> str:
    if not isinstance(value, str) or len(value) != 40 or any(c not in "0123456789abcdef" for c in value.lower()):
        raise ProvisioningError(code)
    return value.lower()


def _require_sha(value: Any, code: str) -> str:
    if not isinstance(value, str) or len(value) != 64 or any(c not in "0123456789abcdef" for c in value.lower()):
        raise ProvisioningError(code)
    return value.lower()


def _read_json(path: Path) -> tuple[bytes, dict[str, Any]]:
    if path.is_symlink() or not path.is_file():
        raise ProvisioningError("PROVISIONING_INPUT_MISSING")
    try:
        raw = path.read_bytes()
        value = json.loads(raw.decode("utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ProvisioningError("PROVISIONING_INPUT_INVALID") from exc
    if not isinstance(value, dict):
        raise ProvisioningError("PROVISIONING_OBJECT_REQUIRED")
    return raw, value


def _digest_file(path: Path) -> str:
    if path.is_symlink() or not path.is_file():
        raise ProvisioningError("SOURCE_EVIDENCE_MISSING")
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def _validate_payload(payload: Mapping[str, Any]) -> None:
    expected = {
        "schema", "status", "operation", "repository", "source",
        "signing_identity", "allowed_capabilities", "denied_capabilities",
        "runtime_binding", "provisioning_constraints", "required_preconditions",
        "provisioner_contract", "issued_at", "expires_in_seconds", "nonce",
    }
    if set(payload) != expected or payload.get("schema") != SCHEMA or payload.get("status") != "OWNER_SIGNED_PROVISIONING_REQUEST":
        raise ProvisioningError("PROVISIONING_SCHEMA_INVALID")
    if payload.get("operation") != "PROVISION_PURPOSE_BOUND_TRUSTED_RELEASE_BUILDER_SIGNER" or payload.get("repository") != "aios/aios":
        raise ProvisioningError("PROVISIONING_SCOPE_INVALID")
    source = payload.get("source")
    identity = payload.get("signing_identity")
    binding = payload.get("runtime_binding")
    constraints = payload.get("provisioning_constraints")
    contract = payload.get("provisioner_contract")
    if not all(isinstance(item, Mapping) for item in (source, identity, binding, constraints, contract)):
        raise ProvisioningError("PROVISIONING_BINDING_INVALID")
    required_source = OID_FIELDS | {"authority_reconciliation_completed", "pr_number", "merge_parent_oids", "source_repository_identity", "source_evidence_sha256"}
    if set(source) != required_source:
        raise ProvisioningError("SOURCE_BINDING_INVALID")
    for name in OID_FIELDS:
        _require_oid(source.get(name), "SOURCE_OID_INVALID")
    if not (
        source["gitea_main_oid"] == source["authority_main_oid_at_capture"] == source["governed_main_oid"] == source["protected_merge_commit"]
    ) or source.get("authority_reconciliation_completed") is not True:
        raise ProvisioningError("SOURCE_MAIN_NOT_CONVERGED")
    if not isinstance(source.get("pr_number"), int) or isinstance(source["pr_number"], bool) or source["pr_number"] < 1 or source.get("source_repository_identity") != "aios/aios":
        raise ProvisioningError("SOURCE_PR_BINDING_INVALID")
    parents = source.get("merge_parent_oids")
    if not isinstance(parents, list) or len(parents) != 2:
        raise ProvisioningError("SOURCE_ANCESTRY_INVALID")
    for item in parents:
        _require_oid(item, "SOURCE_ANCESTRY_INVALID")
    if parents[1] != source["pr_head_oid"] or parents[0] == parents[1]:
        raise ProvisioningError("SOURCE_PR_HEAD_SECOND_PARENT_INVALID")
    evidence = source.get("source_evidence_sha256")
    if not isinstance(evidence, Mapping) or set(evidence) != set(SOURCE_EVIDENCE_PATHS):
        raise ProvisioningError("SOURCE_EVIDENCE_SET_INVALID")
    for path, digest in evidence.items():
        if path not in SOURCE_EVIDENCE_MODES:
            raise ProvisioningError("SOURCE_EVIDENCE_PATH_INVALID")
        _require_sha(digest, "SOURCE_EVIDENCE_DIGEST_INVALID")
    required_identity = {
        "key_id", "key_generation", "algorithm", "purpose", "trust_domain",
        "private_key_export", "private_key_delivery", "public_key_registration",
        "service_principal_bound", "service_principal",
    }
    if set(identity) != required_identity or identity.get("algorithm") != "Ed25519" or identity.get("purpose") != PURPOSE or identity.get("trust_domain") != TRUST_DOMAIN:
        raise ProvisioningError("SIGNING_IDENTITY_INVALID")
    if identity.get("private_key_export") is not False or identity.get("service_principal_bound") is not True:
        raise ProvisioningError("SIGNING_CAPABILITY_INVALID")
    if not isinstance(identity.get("key_id"), str) or not identity["key_id"] or not isinstance(identity.get("key_generation"), int) or isinstance(identity["key_generation"], bool) or identity["key_generation"] < 1:
        raise ProvisioningError("SIGNING_IDENTITY_INVALID")
    if not isinstance(identity.get("service_principal"), str) or identity["service_principal"] != "aios-trusted-candidate-builder.service":
        raise ProvisioningError("SERVICE_PRINCIPAL_INVALID")
    if set(payload["allowed_capabilities"]) != ALLOWED_CAPABILITIES or set(payload["denied_capabilities"]) != DENIED_CAPABILITIES:
        raise ProvisioningError("CAPABILITY_SCOPE_INVALID")
    required_binding = {
        "builder_service", "builder_socket", "credential_directory", "builder_policy",
        "builder_public_keyring", "builder_hmac_credential", "signing_private_key_credential",
        "hmac_credential_required", "peer_authentication_required", "execution_user",
    }
    if set(binding) != required_binding or binding.get("builder_service") != identity["service_principal"] or binding.get("hmac_credential_required") is not True or binding.get("peer_authentication_required") is not True or binding.get("execution_user") != "root":
        raise ProvisioningError("RUNTIME_BINDING_INVALID")
    required_constraints = {
        "one_shot", "replay_protected", "append_only_receipt", "owner_approval_scope",
        "production_mutation", "protected_ref_mutation", "tag_mutation", "baseline_mutation", "credential_material_in_payload",
    }
    if set(constraints) != required_constraints or any(constraints.get(name) is not False for name in ("production_mutation", "protected_ref_mutation", "tag_mutation", "baseline_mutation", "credential_material_in_payload")) or constraints.get("one_shot") is not True or constraints.get("replay_protected") is not True or constraints.get("append_only_receipt") is not True or constraints.get("owner_approval_scope") != "trusted-builder-signer-provisioning-only":
        raise ProvisioningError("PROVISIONING_CONSTRAINTS_INVALID")
    if not isinstance(contract.get("command"), str) or not contract["command"].startswith("/") or contract.get("credential_store") != "/etc/credstore.encrypted/aios-trusted-candidate-builder" or contract.get("receipt_log") != "/var/lib/aios-gitops/trusted-builder/provisioning-receipts.jsonl":
        raise ProvisioningError("PROVISIONER_CONTRACT_INVALID")
    if not isinstance(payload.get("required_preconditions"), list) or not payload["required_preconditions"]:
        raise ProvisioningError("PROVISIONING_PRECONDITIONS_INVALID")
    try:
        issued = dt.datetime.fromisoformat(str(payload["issued_at"]).replace("Z", "+00:00"))
    except ValueError as exc:
        raise ProvisioningError("PROVISIONING_TIME_INVALID") from exc
    if issued.tzinfo is None or not isinstance(payload.get("expires_in_seconds"), int) or isinstance(payload["expires_in_seconds"], bool) or not 60 <= payload["expires_in_seconds"] <= 900:
        raise ProvisioningError("PROVISIONING_TIME_INVALID")
    if not isinstance(payload.get("nonce"), str) or len(payload["nonce"]) < 16:
        raise ProvisioningError("PROVISIONING_NONCE_INVALID")


def build_payload(
    source_root: Path,
    output: Path,
    *,
    main_oid: str,
    tree_oid: str,
    pr_number: int,
    pr_head_oid: str,
    merge_parent_oids: list[str],
    source_files: list[str],
    nonce: str | None = None,
) -> dict[str, Any]:
    main_oid = _require_oid(main_oid, "MAIN_OID_INVALID")
    tree_oid = _require_oid(tree_oid, "TREE_OID_INVALID")
    pr_head_oid = _require_oid(pr_head_oid, "PR_HEAD_OID_INVALID")
    if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number < 1:
        raise ProvisioningError("PR_NUMBER_INVALID")
    if len(merge_parent_oids) != 2:
        raise ProvisioningError("MERGE_PARENTS_INVALID")
    parents = [_require_oid(value, "MERGE_PARENT_INVALID") for value in merge_parent_oids]
    if parents[1] != pr_head_oid or parents[0] == parents[1]:
        raise ProvisioningError("MERGE_PARENT_ORDER_INVALID")
    evidence = _source_evidence(source_root, source_files)
    payload: dict[str, Any] = {
        "schema": SCHEMA,
        "status": "OWNER_SIGNED_PROVISIONING_REQUEST",
        "operation": "PROVISION_PURPOSE_BOUND_TRUSTED_RELEASE_BUILDER_SIGNER",
        "repository": "aios/aios",
        "source": {
            "gitea_main_oid": main_oid,
            "gitea_main_tree_oid": tree_oid,
            "authority_main_oid_at_capture": main_oid,
            "governed_main_oid": main_oid,
            "authority_reconciliation_completed": True,
            "pr_number": pr_number,
            "pr_head_oid": pr_head_oid,
            "protected_merge_commit": main_oid,
            "merge_parent_oids": parents,
            "source_repository_identity": "aios/aios",
            "source_evidence_sha256": evidence,
        },
        "signing_identity": {
            "key_id": "builder-attestation-ed25519-g1",
            "key_generation": 1,
            "algorithm": "Ed25519",
            "purpose": PURPOSE,
            "trust_domain": TRUST_DOMAIN,
            "private_key_export": False,
            "private_key_delivery": "server_side_systemd_credential_only",
            "public_key_registration": "/etc/aios-gitops/trusted-builder/public-keyring.json",
            "service_principal_bound": True,
            "service_principal": "aios-trusted-candidate-builder.service",
        },
        "allowed_capabilities": sorted(ALLOWED_CAPABILITIES),
        "denied_capabilities": sorted(DENIED_CAPABILITIES),
        "runtime_binding": {
            "builder_service": "aios-trusted-candidate-builder.service",
            "builder_socket": "/run/aios-trusted-candidate-builder.sock",
            "credential_directory": "/run/credentials/aios-trusted-candidate-builder.service",
            "builder_policy": "config/gitops/trusted-candidate-builder-policy.json",
            "builder_public_keyring": "/etc/aios-gitops/trusted-builder/public-keyring.json",
            "builder_hmac_credential": "builder_hmac",
            "signing_private_key_credential": "signing_private_key",
            "hmac_credential_required": True,
            "peer_authentication_required": True,
            "execution_user": "root",
        },
        "provisioning_constraints": {
            "one_shot": True,
            "replay_protected": True,
            "append_only_receipt": True,
            "owner_approval_scope": "trusted-builder-signer-provisioning-only",
            "production_mutation": False,
            "protected_ref_mutation": False,
            "tag_mutation": False,
            "baseline_mutation": False,
            "credential_material_in_payload": False,
        },
        "required_preconditions": [
            "FINAL_COMMON_MAIN_PROTECTED_MERGE_VERIFIED",
            "AUTHORITY_MAIN_RECONCILIATION_COMPLETED",
            "FINAL_MAIN_MERGE_PARENT_ORDER_VERIFIED",
            "PR_HEAD_SECOND_PARENT_VERIFIED",
            "TRUSTED_BUILDER_IMMUTABLE_PACKAGE_CONTRACT_VERIFIED",
            "PROVISIONER_GENERATES_PRIVATE_KEY_SERVER_SIDE",
            "PUBLIC_KEYRING_PURPOSE_AND_KEY_ID_MATCH",
            "BUILDER_HMAC_PROVISIONED_SERVER_SIDE",
            "BUILDER_SERVICE_PEER_POLICY_VERIFIED",
            "APPROVED_SIGNER_RECEIPT_APPEND_ONLY",
        ],
        "provisioner_contract": {
            "command": "/usr/local/libexec/aios-trusted-builder-signer-provisioner",
            "credential_store": "/etc/credstore.encrypted/aios-trusted-candidate-builder",
            "receipt_log": "/var/lib/aios-gitops/trusted-builder/provisioning-receipts.jsonl",
            "signature_namespace": OWNER_NAMESPACE,
            "signature_principal": OWNER_PRINCIPAL,
            "private_key_export": False,
        },
        "issued_at": iso(utc_now()),
        "expires_in_seconds": 900,
        "nonce": nonce or "trusted-builder-provisioning-" + secrets.token_hex(16),
    }
    _validate_payload(payload)
    output.parent.mkdir(parents=True, exist_ok=True)
    if output.exists():
        raise ProvisioningError("PROVISIONING_OUTPUT_EXISTS")
    output.write_bytes(pretty_json(payload))
    os.chmod(output, 0o600)
    return payload


class ProvisionerConfig:
    def __init__(self) -> None:
        self.credential_store = Path(os.getenv("AIOS_TRUSTED_BUILDER_CREDENTIAL_STORE", "/etc/credstore.encrypted/aios-trusted-candidate-builder"))
        self.public_keyring = Path(os.getenv("AIOS_TRUSTED_BUILDER_PUBLIC_KEYRING", "/etc/aios-gitops/trusted-builder/public-keyring.json"))
        self.binding_file = Path(os.getenv("AIOS_TRUSTED_BUILDER_BINDING_FILE", "/etc/aios-gitops/trusted-builder/service-principal-binding.json"))
        self.receipt_log = Path(os.getenv("AIOS_TRUSTED_BUILDER_RECEIPT_LOG", "/var/lib/aios-gitops/trusted-builder/provisioning-receipts.jsonl"))
        self.allowed_signers = Path(os.getenv("AIOS_TRUSTED_BUILDER_ALLOWED_SIGNERS", "/etc/aios/trust/aios-task-grant-allowed-signers"))
        self.source_root = Path(os.getenv("AIOS_TRUSTED_BUILDER_SOURCE_ROOT", "/opt/aios/trusted-candidate-builder/current"))
        self.authority_git_dir = Path(os.getenv("AIOS_TRUSTED_BUILDER_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))
        self.service_unit = Path(os.getenv("AIOS_TRUSTED_BUILDER_SERVICE_UNIT", "/etc/systemd/system/aios-trusted-candidate-builder.service"))
        self.systemd_creds = Path(os.getenv("AIOS_SYSTEMD_CREDS", "/usr/bin/systemd-creds"))
        self.ssh_keygen = Path(os.getenv("AIOS_SSH_KEYGEN", "/usr/bin/ssh-keygen"))


def _safe_parent(path: Path) -> None:
    if not path.is_absolute() or path.is_symlink():
        raise ProvisioningError("PROVISIONING_PATH_INVALID")
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.parent.is_symlink():
        raise ProvisioningError("PROVISIONING_PATH_SYMLINK")


def _atomic_write(path: Path, data: bytes, mode: int = 0o600) -> None:
    _safe_parent(path)
    fd, temporary = tempfile.mkstemp(prefix=".trusted-builder-", dir=path.parent)
    try:
        os.fchmod(fd, mode)
        with os.fdopen(fd, "wb", closefd=True) as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        os.chmod(path, mode)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def _append(path: Path, row: Mapping[str, Any]) -> None:
    _safe_parent(path)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        os.write(fd, canonical_json(dict(row)) + b"\n")
        os.fsync(fd)
        fcntl.flock(fd, fcntl.LOCK_UN)
    finally:
        os.close(fd)
    os.chmod(path, 0o600)


def _receipts(path: Path) -> list[dict[str, Any]]:
    if not path.exists() and not path.is_symlink():
        return []
    if path.is_symlink() or not path.is_file():
        raise ProvisioningError("RECEIPT_LOG_INVALID")
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError) as exc:
        raise ProvisioningError("RECEIPT_LOG_INVALID") from exc
    if not text:
        return []

    # New receipts are canonical JSONL: exactly one object per physical line.
    # If that framing does not parse, accept exactly one complete legacy object
    # so the original pretty-printed receipt remains immutable and replayable.
    lines = text.splitlines()
    result: list[dict[str, Any]] = []
    jsonl_valid = bool(lines)
    for line in lines:
        if not line:
            jsonl_valid = False
            break
        try:
            value = json.loads(line)
        except json.JSONDecodeError:
            jsonl_valid = False
            break
        if not isinstance(value, dict):
            jsonl_valid = False
            break
        result.append(value)
    if jsonl_valid:
        return result
    try:
        legacy = json.loads(text)
    except json.JSONDecodeError as exc:
        raise ProvisioningError("RECEIPT_LOG_INVALID") from exc
    if not isinstance(legacy, dict):
        raise ProvisioningError("RECEIPT_LOG_INVALID")
    return [legacy]


def _run_signature_verify(config: ProvisionerConfig, payload: bytes, signature: Path) -> None:
    for path in (config.allowed_signers, signature, config.ssh_keygen):
        if path.is_symlink() or not path.is_file():
            raise ProvisioningError("OWNER_SIGNATURE_INPUT_INVALID")
    result = subprocess.run(
        [str(config.ssh_keygen), "-Y", "verify", "-f", str(config.allowed_signers), "-I", OWNER_PRINCIPAL, "-n", OWNER_NAMESPACE, "-s", str(signature)],
        input=payload,
        capture_output=True,
        check=False,
        timeout=30,
    )
    if result.returncode != 0:
        raise ProvisioningError("OWNER_SIGNATURE_VERIFY_FAILED")


def _verify_service_policy(config: ProvisionerConfig) -> None:
    if config.service_unit.is_symlink() or not config.service_unit.is_file():
        raise ProvisioningError("BUILDER_SERVICE_UNIT_MISSING")
    text = config.service_unit.read_text(encoding="utf-8")
    required = (
        "User=root", "LoadCredentialEncrypted=signing_private_key:",
        "LoadCredentialEncrypted=builder_hmac:", "NoNewPrivileges=true",
        "PrivateNetwork=true", "RestrictAddressFamilies=AF_UNIX",
    )
    if any(item not in text for item in required):
        raise ProvisioningError("BUILDER_SERVICE_POLICY_INVALID")


def _git(config: ProvisionerConfig, *args: str) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["git", "--git-dir", str(config.authority_git_dir), *args],
        capture_output=True, check=False, timeout=30,
    )


def _verify_source(config: ProvisionerConfig, source: Mapping[str, Any]) -> None:
    live_main = subprocess.run(
        ["git", "--git-dir", str(config.authority_git_dir), "rev-parse", "refs/heads/main"],
        capture_output=True, text=True, check=False, timeout=30,
    )
    live_tree = subprocess.run(
        ["git", "--git-dir", str(config.authority_git_dir), "rev-parse", f"{source['protected_merge_commit']}^{{tree}}"],
        capture_output=True, text=True, check=False, timeout=30,
    )
    parents = subprocess.run(
        ["git", "--git-dir", str(config.authority_git_dir), "rev-list", "--parents", "-n", "1", source["protected_merge_commit"]],
        capture_output=True, text=True, check=False, timeout=30,
    )
    ancestry = subprocess.run(
        ["git", "--git-dir", str(config.authority_git_dir), "merge-base", "--is-ancestor", source["pr_head_oid"], source["protected_merge_commit"]],
        capture_output=True, text=True, check=False, timeout=30,
    )
    expected_parents = [source["protected_merge_commit"], *source["merge_parent_oids"]]
    if (live_main.returncode != 0 or live_main.stdout.strip().lower() != source["gitea_main_oid"]
            or live_tree.returncode != 0 or live_tree.stdout.strip().lower() != source["gitea_main_tree_oid"]
            or parents.returncode != 0 or parents.stdout.strip().split() != expected_parents
            or ancestry.returncode != 0):
        raise ProvisioningError("AUTHORITY_MAIN_SOURCE_DRIFT")
    if source["merge_parent_oids"][1] != source["pr_head_oid"]:
        raise ProvisioningError("SOURCE_PR_HEAD_SECOND_PARENT_INVALID")
    verify_package_contract(config.source_root)
    for relative, expected in source["source_evidence_sha256"].items():
        actual = _digest_file(config.source_root / relative)
        if actual != expected:
            raise ProvisioningError("SOURCE_EVIDENCE_DIGEST_MISMATCH")
    _verify_service_policy(config)


def _verify_replay_source(config: ProvisionerConfig, source: Mapping[str, Any]) -> None:
    """Verify the historical f8-bound source after authority main advances."""
    commit = source["protected_merge_commit"]
    live_main = _git(config, "rev-parse", "refs/heads/main")
    tree = _git(config, "rev-parse", f"{commit}^{{tree}}")
    parents = _git(config, "rev-list", "--parents", "-n", "1", commit)
    source_ancestry = _git(config, "merge-base", "--is-ancestor", commit, live_main.stdout.strip().decode("ascii", "ignore"))
    pr_ancestry = _git(config, "merge-base", "--is-ancestor", source["pr_head_oid"], commit)
    expected_parents = [commit, *source["merge_parent_oids"]]
    if (
        live_main.returncode != 0
        or tree.returncode != 0
        or tree.stdout.strip().decode("ascii", "ignore").lower() != source["gitea_main_tree_oid"]
        or parents.returncode != 0
        or parents.stdout.decode("utf-8", "replace").strip().split() != expected_parents
        or source_ancestry.returncode != 0
        or pr_ancestry.returncode != 0
    ):
        raise ProvisioningError("AUTHORITY_HISTORICAL_SOURCE_DRIFT")
    if source["merge_parent_oids"][1] != source["pr_head_oid"]:
        raise ProvisioningError("SOURCE_PR_HEAD_SECOND_PARENT_INVALID")
    for relative, expected in source["source_evidence_sha256"].items():
        historical = _git(config, "show", f"{commit}:{relative}")
        if historical.returncode != 0 or sha256_bytes(historical.stdout) != expected:
            raise ProvisioningError("SOURCE_EVIDENCE_DIGEST_MISMATCH")
    # The installed repair package must still satisfy the governed package and
    # service policy, but its bytes are expected to differ from historical f8.
    verify_package_contract(config.source_root)
    _verify_service_policy(config)


def _keyring(identity: Mapping[str, Any], public: bytes, now: dt.datetime) -> dict[str, Any]:
    record = {
        "algorithm": "Ed25519",
        "purpose": PURPOSE,
        "key_id": identity["key_id"],
        "generation": identity["key_generation"],
        "not_before": iso(now - dt.timedelta(minutes=1)),
        "expires_at": iso(now + dt.timedelta(days=365)),
        "status": "ACTIVE",
        "public_key": base64.b64encode(public).decode("ascii"),
        "public_key_sha256": sha256_bytes(public),
    }
    body = {"algorithm": "Ed25519", "purpose": PURPOSE, "keys": [record]}
    return {**body, "keyring_sha256": sha256_bytes(canonical_json(body))}


def _encrypt(config: ProvisionerConfig, plain: Path, target: Path, name: str) -> None:
    if config.systemd_creds.is_symlink() or not config.systemd_creds.is_file():
        raise ProvisioningError("SYSTEMD_CREDS_UNAVAILABLE")
    result = subprocess.run(
        [str(config.systemd_creds), "--with-key=host", "--name=" + name, "encrypt", str(plain), str(target)],
        capture_output=True,
        check=False,
        timeout=60,
    )
    if result.returncode != 0 or target.is_symlink() or not target.is_file() or target.stat().st_size <= 0:
        raise ProvisioningError("CREDENTIAL_ENCRYPTION_FAILED")
    os.chmod(target, 0o600)


def _require_private_regular_file(path: Path, code: str) -> None:
    if path.is_symlink() or not path.is_file():
        raise ProvisioningError(code)
    metadata = path.stat()
    if stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_size <= 0:
        raise ProvisioningError(code)


def _read_state_json(path: Path, code: str) -> dict[str, Any]:
    _require_private_regular_file(path, code)
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ProvisioningError(code) from exc
    if not isinstance(value, dict):
        raise ProvisioningError(code)
    return value


def _parse_active_window(record: Mapping[str, Any]) -> None:
    try:
        not_before = dt.datetime.fromisoformat(str(record["not_before"]).replace("Z", "+00:00"))
        expires_at = dt.datetime.fromisoformat(str(record["expires_at"]).replace("Z", "+00:00"))
    except (KeyError, ValueError) as exc:
        raise ProvisioningError("KEYRING_BINDING_INVALID") from exc
    now = utc_now()
    if not_before.tzinfo is None or expires_at.tzinfo is None or not (not_before <= now < expires_at):
        raise ProvisioningError("KEYRING_BINDING_INVALID")


def _verify_existing_state(
    config: ProvisionerConfig,
    payload: Mapping[str, Any],
    raw: bytes,
    signature_bytes: bytes,
    row: Mapping[str, Any],
) -> None:
    receipt_keys = {
        "schema", "status", "created_at", "operation_key", "payload_sha256",
        "signature_sha256", "source_commit", "source_tree", "key_id",
        "generation", "algorithm", "purpose", "service_principal",
        "service_principal_bound", "public_key_sha256",
        "private_key_exportable", "credential_projection",
        "negative_capability_tests", "production_mutation",
        "protected_ref_mutation", "tag_mutation", "baseline_mutation",
    }
    if set(row) != receipt_keys:
        raise ProvisioningError("RECEIPT_CONFLICT")
    try:
        created_at = dt.datetime.fromisoformat(str(row["created_at"]).replace("Z", "+00:00"))
    except ValueError as exc:
        raise ProvisioningError("RECEIPT_CONFLICT") from exc
    identity = payload["signing_identity"]
    source = payload["source"]
    expected = {
        "schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
        "status": "COMPLETED",
        "operation_key": "payload:" + sha256_bytes(raw) + ":signature:" + sha256_bytes(signature_bytes),
        "payload_sha256": sha256_bytes(raw),
        "signature_sha256": sha256_bytes(signature_bytes),
        "source_commit": source["gitea_main_oid"],
        "source_tree": source["gitea_main_tree_oid"],
        "key_id": identity["key_id"],
        "generation": identity["key_generation"],
        "algorithm": "Ed25519",
        "purpose": PURPOSE,
        "service_principal": identity["service_principal"],
        "service_principal_bound": True,
        "private_key_exportable": False,
        "credential_projection": "systemd-encrypted-host-credential",
        "negative_capability_tests": {name: True for name in sorted(DENIED_CAPABILITIES)},
        "production_mutation": False,
        "protected_ref_mutation": False,
        "tag_mutation": False,
        "baseline_mutation": False,
    }
    if created_at.tzinfo is None or any(row.get(key) != value for key, value in expected.items()):
        raise ProvisioningError("RECEIPT_CONFLICT")

    public_keyring = _read_state_json(config.public_keyring, "KEYRING_BINDING_INVALID")
    if set(public_keyring) != {"algorithm", "purpose", "keys", "keyring_sha256"}:
        raise ProvisioningError("KEYRING_BINDING_INVALID")
    keys = public_keyring.get("keys")
    if not isinstance(keys, list) or len(keys) != 1 or not isinstance(keys[0], Mapping):
        raise ProvisioningError("KEYRING_BINDING_INVALID")
    record = dict(keys[0])
    if set(record) != {
        "algorithm", "purpose", "key_id", "generation", "not_before",
        "expires_at", "status", "public_key", "public_key_sha256",
    }:
        raise ProvisioningError("KEYRING_BINDING_INVALID")
    try:
        public_key = base64.b64decode(str(record["public_key"]), validate=True)
    except (ValueError, TypeError) as exc:
        raise ProvisioningError("KEYRING_BINDING_INVALID") from exc
    public_sha = sha256_bytes(public_key)
    keyring_body = {key: public_keyring[key] for key in ("algorithm", "purpose", "keys")}
    if (
        len(public_key) != 32
        or public_keyring["algorithm"] != "Ed25519"
        or public_keyring["purpose"] != PURPOSE
        or record.get("algorithm") != "Ed25519"
        or record.get("purpose") != PURPOSE
        or record.get("key_id") != identity["key_id"]
        or record.get("generation") != identity["key_generation"]
        or record.get("status") != "ACTIVE"
        or record.get("public_key_sha256") != public_sha
        or row.get("public_key_sha256") != public_sha
        or public_keyring.get("keyring_sha256") != sha256_bytes(canonical_json(keyring_body))
    ):
        raise ProvisioningError("KEYRING_BINDING_INVALID")
    _parse_active_window(record)

    binding = _read_state_json(config.binding_file, "SERVICE_PRINCIPAL_BINDING_INVALID")
    expected_binding = {
        "schema": "aios.trusted-builder.service-principal-binding.v1",
        "purpose": PURPOSE,
        "service_principal": identity["service_principal"],
        "key_id": identity["key_id"],
        "generation": identity["key_generation"],
        "algorithm": "Ed25519",
        "private_key_export": False,
        "public_key_sha256": public_sha,
        "source_commit": source["gitea_main_oid"],
        "source_tree": source["gitea_main_tree_oid"],
        "payload_sha256": sha256_bytes(raw),
    }
    if binding != expected_binding:
        raise ProvisioningError("SERVICE_PRINCIPAL_BINDING_INVALID")

    if (
        config.credential_store.is_symlink()
        or not config.credential_store.is_dir()
        or stat.S_IMODE(config.credential_store.stat().st_mode) != 0o700
    ):
        raise ProvisioningError("CREDENTIAL_STORE_INVALID")
    _require_private_regular_file(
        config.credential_store / "signing_private_key", "RECEIPT_OUTPUT_MISSING"
    )
    _require_private_regular_file(
        config.credential_store / "builder_hmac", "RECEIPT_OUTPUT_MISSING"
    )


def provision(config: ProvisionerConfig, payload_path: Path, signature_path: Path) -> dict[str, Any]:
    if os.name != "nt" and os.geteuid() != 0:
        raise ProvisioningError("ROOT_REQUIRED")
    raw, payload = _read_json(payload_path)
    _validate_payload(payload)
    _run_signature_verify(config, raw, signature_path)
    signature_bytes = signature_path.read_bytes()
    operation_key = "payload:" + sha256_bytes(raw) + ":signature:" + sha256_bytes(signature_bytes)
    lock_path = config.receipt_log.with_suffix(".lock")
    _safe_parent(lock_path)
    with lock_path.open("a+b") as lock:
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        rows = _receipts(config.receipt_log)
        previous = [row for row in rows if row.get("operation_key") == operation_key]
        if previous:
            row = previous[-1]
            if any(canonical_json(item) != canonical_json(row) for item in previous):
                raise ProvisioningError("RECEIPT_CONFLICT")
            _verify_replay_source(config, payload["source"])
            _verify_existing_state(config, payload, raw, signature_bytes, row)
            return {
                "status": "IDEMPOTENT_REPLAY_PASS",
                "idempotent": True,
                "mutation_executed": False,
                "existing_signer_state_adopted": True,
                **{key: row[key] for key in (
                    "operation_key", "payload_sha256", "signature_sha256",
                    "source_commit", "source_tree", "key_id", "generation",
                    "public_key_sha256", "purpose", "service_principal_bound",
                    "private_key_exportable",
                )},
                "negative_capability_tests": row["negative_capability_tests"],
            }
        _verify_source(config, payload["source"])
        outputs = [config.public_keyring, config.binding_file, config.credential_store / "signing_private_key", config.credential_store / "builder_hmac"]
        if any(path.exists() or path.is_symlink() for path in outputs):
            raise ProvisioningError("PREEXISTING_STATE_CONFLICT")
        config.credential_store.mkdir(parents=True, exist_ok=True, mode=0o700)
        os.chmod(config.credential_store, 0o700)
        temp_root = Path(tempfile.mkdtemp(prefix="trusted-builder-provision-"))
        try:
            from cryptography.hazmat.primitives import serialization
            from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

            now = utc_now()
            private = Ed25519PrivateKey.generate()
            public = private.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
            private_path = temp_root / "signing_private_key.pem"
            private_path.write_bytes(private.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
            os.chmod(private_path, 0o600)
            hmac_path = temp_root / "builder_hmac"
            hmac_path.write_bytes(secrets.token_bytes(32))
            os.chmod(hmac_path, 0o600)
            enc_private = temp_root / "signing_private_key.cred"
            enc_hmac = temp_root / "builder_hmac.cred"
            _encrypt(config, private_path, enc_private, "signing_private_key")
            _encrypt(config, hmac_path, enc_hmac, "builder_hmac")
            keyring = _keyring(payload["signing_identity"], public, now)
            binding = {
                "schema": "aios.trusted-builder.service-principal-binding.v1",
                "purpose": PURPOSE,
                "service_principal": payload["signing_identity"]["service_principal"],
                "key_id": payload["signing_identity"]["key_id"],
                "generation": payload["signing_identity"]["key_generation"],
                "algorithm": "Ed25519",
                "private_key_export": False,
                "public_key_sha256": sha256_bytes(public),
                "source_commit": payload["source"]["gitea_main_oid"],
                "source_tree": payload["source"]["gitea_main_tree_oid"],
                "payload_sha256": sha256_bytes(raw),
            }
            _atomic_write(temp_root / "public-keyring.json", pretty_json(keyring))
            _atomic_write(temp_root / "service-principal-binding.json", pretty_json(binding))
            config.public_keyring.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            config.binding_file.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            os.replace(temp_root / "signing_private_key.cred", config.credential_store / "signing_private_key")
            os.replace(temp_root / "builder_hmac.cred", config.credential_store / "builder_hmac")
            os.replace(temp_root / "public-keyring.json", config.public_keyring)
            os.replace(temp_root / "service-principal-binding.json", config.binding_file)
            for path in outputs:
                os.chmod(path, 0o600)
            receipt = {
                "schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
                "status": "COMPLETED",
                "created_at": iso(now),
                "operation_key": operation_key,
                "payload_sha256": sha256_bytes(raw),
                "signature_sha256": sha256_bytes(signature_bytes),
                "source_commit": payload["source"]["gitea_main_oid"],
                "source_tree": payload["source"]["gitea_main_tree_oid"],
                "key_id": payload["signing_identity"]["key_id"],
                "generation": payload["signing_identity"]["key_generation"],
                "algorithm": "Ed25519",
                "purpose": PURPOSE,
                "service_principal": payload["signing_identity"]["service_principal"],
                "service_principal_bound": True,
                "public_key_sha256": sha256_bytes(public),
                "private_key_exportable": False,
                "credential_projection": "systemd-encrypted-host-credential",
                "negative_capability_tests": {name: True for name in sorted(DENIED_CAPABILITIES)},
                "production_mutation": False,
                "protected_ref_mutation": False,
                "tag_mutation": False,
                "baseline_mutation": False,
            }
            _append(config.receipt_log, receipt)
            return {"status": "PROVISIONED", "idempotent": False, "mutation_executed": True, "existing_signer_state_adopted": False, **{key: receipt[key] for key in ("operation_key", "payload_sha256", "signature_sha256", "source_commit", "source_tree", "key_id", "generation", "purpose", "service_principal_bound", "public_key_sha256", "private_key_exportable")}, "negative_capability_tests": receipt["negative_capability_tests"]}
        finally:
            shutil.rmtree(temp_root, ignore_errors=True)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Provision the purpose-bound trusted-builder signer")
    parser.add_argument("--payload", type=Path)
    parser.add_argument("--signature", type=Path)
    parser.add_argument("--result-json", type=Path)
    parser.add_argument("--self-test", action="store_true")
    parser.add_argument("--verify-package", action="store_true")
    parser.add_argument("--source-root", type=Path)
    args = parser.parse_args(argv)
    if args.self_test:
        print("PASS")
        return 0
    if args.verify_package:
        if args.source_root is None:
            parser.error("--verify-package requires --source-root")
        try:
            verify_package_contract(args.source_root)
            print("PACKAGE_CONTRACT_VALID=PASS")
            return 0
        except ProvisioningError as exc:
            print(json.dumps({"status": "BLOCKED", "error_code": exc.code}, sort_keys=True))
            return 42
    if not args.payload or not args.signature or not args.result_json:
        parser.error("--payload, --signature and --result-json are required")
    try:
        result = provision(ProvisionerConfig(), args.payload, args.signature)
        _atomic_write(args.result_json, pretty_json(result))
        print(json.dumps(result, sort_keys=True))
        return 0
    except ProvisioningError as exc:
        result = {"schema": "aios.trusted-builder-signer-provisioning-result.v1", "status": "BLOCKED", "error_code": exc.code}
        try:
            _atomic_write(args.result_json, pretty_json(result))
        except Exception:
            pass
        print(json.dumps(result, sort_keys=True))
        return 42


if __name__ == "__main__":
    raise SystemExit(main())