fix(gitops): repair trusted-builder receipt replay framing

Reviewed-on: #71
Reviewed-by: owner <owner@cnnetmkt.top>

active-runtime/app/trusted_builder_signer_provisioning.py

@@ -0,0 +1,905 @@
+#!/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())

deploy/libexec/aios-trusted-builder-signer-provisioner

@@ -0,0 +1,9 @@
+#!/usr/bin/python3
+"""Installed entry point for the governed trusted-builder provisioner."""
+from __future__ import annotations
+
+import runpy
+import sys
+
+sys.path.insert(0, "/opt/aios/trusted-candidate-builder/current/app")
+runpy.run_module("trusted_builder_signer_provisioning", run_name="__main__")

deploy/trusted-builder/aios-trusted-builder-package-manifest.json

@@ -0,0 +1,57 @@
+{
+  "activation": {
+    "credential_installation": false,
+    "daemon_reload": true,
+    "enable_units": [
+      "aios-trusted-candidate-builder.socket"
+    ],
+    "service_start": "socket-activation-only",
+    "signer_invocation": false
+  },
+  "files": [
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "active-runtime/app/trusted_builder_signer_provisioning.py"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "active-runtime/app/aios_trusted_candidate_builder.py"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "config/gitops/trusted-candidate-builder-policy.json"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "deploy/systemd/aios-trusted-candidate-builder.service"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "deploy/systemd/aios-trusted-candidate-builder.socket"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "deploy/systemd/aios-gitops-model-api-trusted-builder.conf.example"
+    },
+    {
+      "install_mode": "0755",
+      "source_mode": "0644",
+      "source_path": "deploy/libexec/aios-trusted-builder-signer-provisioner"
+    },
+    {
+      "install_mode": "0644",
+      "source_mode": "0644",
+      "source_path": "deploy/trusted-builder/aios-trusted-builder-package-manifest.json"
+    }
+  ],
+  "package_name": "aios-trusted-candidate-builder",
+  "schema": "aios.trusted-builder-package.v1",
+  "source_root": "/opt/aios/trusted-candidate-builder/current",
+  "version": 1
+}

docs/gitops/trusted-builder-signer-provisioning.md

@@ -0,0 +1,35 @@
+# Trusted-builder signer provisioning
+
+`trusted_builder_signer_provisioning.py` provisions the purpose-bound
+`AIOS_TRUSTED_RELEASE_BUILDER` signer only from an exact
+`aios.trusted-builder-signer-provisioning.v2` payload and detached SSH signature
+for `aios-task-grant-signer` in namespace `aios-owner-approval-v1`.
+
+A fresh payload is bound to the final common Gitea, authority, and governed
+`main` OID; the protected merge tree; the ordered two parents; and the selected
+PR head as the merge's second parent. The provisioner rechecks those facts in
+the authority repository and rejects ancestry or source drift. It does not
+reuse PR69, PR70, or a release-candidate recipe binding.
+
+The payload evidence is an exact allowlist: provisioner, candidate builder,
+policy, service, socket, model-API config template, launcher, and the versioned
+package manifest. Missing, extra, modified, or mode-incompatible evidence is
+rejected. The package manifest is non-secret and declares daemon reload plus
+socket activation, with no credential installation or signer invocation. It
+requires the installed launcher target mode to be `0755`; source checkout mode
+is treated separately from installed package mode. A controlled deterministic
+worktree may hold ordinary source evidence at `0600`, `0640`, `0644`, or `0664`; the
+installed package still enforces its declared target modes.
+
+The provisioner generates the Ed25519 key and builder HMAC only on the host.
+Plaintext exists only in its private temporary directory and is encrypted with
+`systemd-creds --with-key=host`; the service receives encrypted credentials via
+`LoadCredentialEncrypted`. The public keyring and service-principal binding
+contain no private key.
+
+Receipts append at
+`/var/lib/aios-gitops/trusted-builder/provisioning-receipts.jsonl`. An exact
+replay returns `IDEMPOTENT_REPLAY_PASS`; mismatched or unreceipted state fails
+closed. Provisioning is configuration only: activation, restart persistence,
+public-key verification and negative-capability tests are separate gates before
+`TRUSTED_BUILDER_SIGNER_READINESS=PASS`.

tests/test_trusted_builder_signer_provisioning.py

@@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import sys
+
+import pytest
+
+APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
+sys.path.insert(0, str(APP))
+import trusted_builder_signer_provisioning as provisioning  # noqa: E402
+
+
+def _source(tmp_path: Path) -> tuple[Path, list[str]]:
+    root = tmp_path / "source"
+    for relative in provisioning.SOURCE_EVIDENCE_PATHS:
+        path = root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        if relative != provisioning.PACKAGE_MANIFEST_PATH:
+            path.write_text("source:" + relative + "\n", encoding="utf-8")
+            os.chmod(path, 0o644)
+    manifest = root / provisioning.PACKAGE_MANIFEST_PATH
+    manifest.write_bytes(provisioning.pretty_json(provisioning.package_contract_document()))
+    os.chmod(manifest, 0o644)
+    return root, list(provisioning.SOURCE_EVIDENCE_PATHS)
+
+
+def _payload(tmp_path: Path):
+    source, paths = _source(tmp_path)
+    output = tmp_path / "payload.json"
+    value = provisioning.build_payload(
+        source, output, main_oid="c" * 40, tree_oid="d" * 40,
+        pr_number=70, pr_head_oid="e" * 40,
+        merge_parent_oids=["a" * 40, "e" * 40], source_files=paths,
+        nonce="test-provisioning-nonce-001",
+    )
+    return source, output, value
+
+
+def test_fresh_payload_is_bound_to_final_common_main_and_ordered_pr_parent(tmp_path: Path):
+    _source_root, output, payload = _payload(tmp_path)
+    assert output.read_bytes().startswith(b"{")
+    provisioning._validate_payload(payload)
+    assert payload["source"]["gitea_main_oid"] == payload["source"]["governed_main_oid"]
+    assert payload["source"]["merge_parent_oids"][1] == payload["source"]["pr_head_oid"]
+    assert set(payload["source"]["source_evidence_sha256"]) == set(provisioning.SOURCE_EVIDENCE_PATHS)
+    assert set(payload["allowed_capabilities"]) == provisioning.ALLOWED_CAPABILITIES
+    assert set(payload["denied_capabilities"]) == provisioning.DENIED_CAPABILITIES
+
+
+@pytest.mark.parametrize("field", ["allowed_capabilities", "denied_capabilities"])
+def test_payload_rejects_capability_scope_tampering(tmp_path: Path, field: str):
+    _source_root, _output, payload = _payload(tmp_path)
+    payload[field] = ["owner_release_approval"]
+    with pytest.raises(provisioning.ProvisioningError, match="CAPABILITY_SCOPE_INVALID"):
+        provisioning._validate_payload(payload)
+
+
+def test_payload_rejects_main_divergence_and_wrong_second_parent(tmp_path: Path):
+    _source_root, _output, payload = _payload(tmp_path)
+    payload["source"]["governed_main_oid"] = "f" * 40
+    with pytest.raises(provisioning.ProvisioningError, match="SOURCE_MAIN_NOT_CONVERGED"):
+        provisioning._validate_payload(payload)
+    _source_root, _output, payload = _payload(tmp_path / "second")
+    payload["source"]["merge_parent_oids"][1] = "a" * 40
+    with pytest.raises(provisioning.ProvisioningError, match="SOURCE_PR_HEAD_SECOND_PARENT_INVALID"):
+        provisioning._validate_payload(payload)
+
+
+def test_payload_rejects_missing_or_extra_source_evidence(tmp_path: Path):
+    _source_root, _output, payload = _payload(tmp_path)
+    payload["source"]["source_evidence_sha256"].pop(next(iter(payload["source"]["source_evidence_sha256"])))
+    with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_SET_INVALID"):
+        provisioning._validate_payload(payload)
+    _source_root, _output, payload = _payload(tmp_path / "second")
+    payload["source"]["source_evidence_sha256"]["extra.txt"] = "0" * 64
+    with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_SET_INVALID"):
+        provisioning._validate_payload(payload)
+
+
+def test_package_contract_rejects_incompatible_source_mode_and_tampering(tmp_path: Path):
+    source, _paths = _source(tmp_path)
+    service = source / "deploy/systemd/aios-trusted-candidate-builder.service"
+    os.chmod(service, 0o601)
+    with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_MODE_INVALID"):
+        provisioning.verify_package_contract(source)
+    os.chmod(service, 0o644)
+    manifest = source / provisioning.PACKAGE_MANIFEST_PATH
+    value = json.loads(manifest.read_text(encoding="utf-8"))
+    value["activation"]["credential_installation"] = True
+    manifest.write_text(json.dumps(value), encoding="utf-8")
+    with pytest.raises(provisioning.ProvisioningError, match="PACKAGE_CONTRACT_INVALID"):
+        provisioning.verify_package_contract(source)
+
+
+def test_package_contract_declares_socket_activation_no_credentials_and_0755_launcher(tmp_path: Path):
+    source, _paths = _source(tmp_path)
+    provisioning.verify_package_contract(source)
+    value = json.loads((source / provisioning.PACKAGE_MANIFEST_PATH).read_text(encoding="utf-8"))
+    launcher = next(item for item in value["files"] if item["source_path"] == "deploy/libexec/aios-trusted-builder-signer-provisioner")
+    assert launcher["install_mode"] == "0755"
+    assert value["activation"] == {
+        "daemon_reload": True,
+        "enable_units": ["aios-trusted-candidate-builder.socket"],
+        "service_start": "socket-activation-only",
+        "credential_installation": False,
+        "signer_invocation": False,
+    }
+
+
+def test_provisioner_self_test_has_no_mutation(capsys):
+    assert provisioning.main(["--self-test"]) == 0
+    assert capsys.readouterr().out.strip() == "PASS"
+
+
+def _replay_fixture(tmp_path: Path):
+    _source_root, payload_path, payload = _payload(tmp_path)
+    raw = payload_path.read_bytes()
+    signature = tmp_path / "owner.sig"
+    signature.write_bytes(b"owner-signature-test")
+    os.chmod(signature, 0o600)
+
+    config = provisioning.ProvisionerConfig()
+    config.credential_store = tmp_path / "credstore"
+    config.public_keyring = tmp_path / "state" / "public-keyring.json"
+    config.binding_file = tmp_path / "state" / "service-principal-binding.json"
+    config.receipt_log = tmp_path / "state" / "provisioning-receipts.jsonl"
+    config.credential_store.mkdir(mode=0o700)
+    os.chmod(config.credential_store, 0o700)
+    config.public_keyring.parent.mkdir(mode=0o700)
+
+    public = b"p" * 32
+    now = provisioning.utc_now()
+    keyring = provisioning._keyring(payload["signing_identity"], public, now)
+    config.public_keyring.write_bytes(provisioning.pretty_json(keyring))
+    os.chmod(config.public_keyring, 0o600)
+    binding = {
+        "schema": "aios.trusted-builder.service-principal-binding.v1",
+        "purpose": provisioning.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": provisioning.sha256_bytes(public),
+        "source_commit": payload["source"]["gitea_main_oid"],
+        "source_tree": payload["source"]["gitea_main_tree_oid"],
+        "payload_sha256": provisioning.sha256_bytes(raw),
+    }
+    config.binding_file.write_bytes(provisioning.pretty_json(binding))
+    os.chmod(config.binding_file, 0o600)
+    for name in ("signing_private_key", "builder_hmac"):
+        target = config.credential_store / name
+        target.write_bytes(("encrypted-" + name).encode())
+        os.chmod(target, 0o600)
+    signature_bytes = signature.read_bytes()
+    receipt = {
+        "schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
+        "status": "COMPLETED",
+        "created_at": provisioning.iso(now),
+        "operation_key": "payload:" + provisioning.sha256_bytes(raw) + ":signature:" + provisioning.sha256_bytes(signature_bytes),
+        "payload_sha256": provisioning.sha256_bytes(raw),
+        "signature_sha256": provisioning.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": provisioning.PURPOSE,
+        "service_principal": payload["signing_identity"]["service_principal"],
+        "service_principal_bound": True,
+        "public_key_sha256": provisioning.sha256_bytes(public),
+        "private_key_exportable": False,
+        "credential_projection": "systemd-encrypted-host-credential",
+        "negative_capability_tests": {name: True for name in sorted(provisioning.DENIED_CAPABILITIES)},
+        "production_mutation": False,
+        "protected_ref_mutation": False,
+        "tag_mutation": False,
+        "baseline_mutation": False,
+    }
+    return config, payload_path, payload, signature, receipt
+
+
+def test_legacy_multiline_receipt_parse(tmp_path: Path):
+    path = tmp_path / "receipts.jsonl"
+    receipt = {"schema": "legacy", "status": "COMPLETED", "nested": {"value": True}}
+    path.write_bytes(provisioning.pretty_json(receipt))
+    assert provisioning._receipts(path) == [receipt]
+
+
+def test_new_canonical_jsonl_receipt_parse(tmp_path: Path):
+    path = tmp_path / "receipts.jsonl"
+    rows = [{"id": 1}, {"id": 2}]
+    for row in rows:
+        provisioning._append(path, row)
+    assert path.read_bytes() == b"".join(provisioning.canonical_json(row) + b"\n" for row in rows)
+    assert provisioning._receipts(path) == rows
+
+
+@pytest.mark.parametrize(
+    "content",
+    [b'{"status":"COMPLETED"}\nnot-json\n', b'{"status":"COMPLETED"'],
+    ids=["malformed", "truncated"],
+)
+def test_malformed_or_truncated_receipt_fails_closed(tmp_path: Path, content: bytes):
+    path = tmp_path / "receipts.jsonl"
+    path.write_bytes(content)
+    with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_LOG_INVALID"):
+        provisioning._receipts(path)
+
+
+def test_preexisting_valid_state_is_adopted_without_mutation(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    before = {
+        path: path.read_bytes()
+        for path in (
+            config.receipt_log, config.public_keyring, config.binding_file,
+            config.credential_store / "signing_private_key",
+            config.credential_store / "builder_hmac",
+        )
+    }
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    result = provisioning.provision(config, payload_path, signature)
+    assert result["status"] == "IDEMPOTENT_REPLAY_PASS"
+    assert result["mutation_executed"] is False
+    assert result["existing_signer_state_adopted"] is True
+    assert result["private_key_exportable"] is False
+    assert before == {path: path.read_bytes() for path in before}
+
+
+def test_duplicate_replay_is_idempotent(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    first = provisioning.provision(config, payload_path, signature)
+    receipt_bytes = config.receipt_log.read_bytes()
+    second = provisioning.provision(config, payload_path, signature)
+    assert first == second
+    assert config.receipt_log.read_bytes() == receipt_bytes
+
+
+def test_wrong_payload_is_rejected_against_preexisting_state(tmp_path: Path, monkeypatch):
+    config, payload_path, payload, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    payload["nonce"] = "wrong-payload-nonce-0002"
+    payload_path.write_bytes(provisioning.pretty_json(payload))
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="PREEXISTING_STATE_CONFLICT"):
+        provisioning.provision(config, payload_path, signature)
+
+
+def test_wrong_signature_is_rejected(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    monkeypatch.setattr(
+        provisioning,
+        "_run_signature_verify",
+        lambda *_args: (_ for _ in ()).throw(provisioning.ProvisioningError("OWNER_SIGNATURE_VERIFY_FAILED")),
+    )
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="OWNER_SIGNATURE_VERIFY_FAILED"):
+        provisioning.provision(config, payload_path, signature)
+
+
+def test_wrong_purpose_is_rejected(tmp_path: Path):
+    _source_root, _payload_path, payload = _payload(tmp_path)
+    payload["signing_identity"]["purpose"] = "WRONG_PURPOSE"
+    with pytest.raises(provisioning.ProvisioningError, match="SIGNING_IDENTITY_INVALID"):
+        provisioning._validate_payload(payload)
+
+
+def test_replay_rejects_state_binding_tampering(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    receipt["service_principal"] = "wrong.service"
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_CONFLICT"):
+        provisioning.provision(config, payload_path, signature)