DataOps-platform/deployment/enterprise/enterprise_delivery.py

#!/usr/bin/env python3
"""P3-WP07 local, fail-closed enterprise delivery evidence controls."""

from __future__ import annotations

import argparse
import contextlib
import fcntl
import hashlib
import json
import os
import re
import stat
import tempfile
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

ROOT = Path(__file__).resolve().parents[2]
HEAD = "20260811_495"
ENVIRONMENTS = ("development", "test", "staging", "production")
TRUST_STORE = ROOT / "deployment/enterprise/trust_store.json"
MAX_APPROVAL_WINDOW = timedelta(hours=24)
CLOCK_SKEW = timedelta(minutes=5)
LOCAL_TEST_PRIVATE_KEY = bytes.fromhex(
    "4f3b2775a4df921336c3882f071d9edce4a6e946b928345f7bf68b55d4eaa3d8"
)


class DeliveryError(ValueError):
    pass


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


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


def sha_file(path: Path) -> str:
    return sha_bytes(safe_bytes(path))


def safe_bytes(path: Path) -> bytes:
    """Read one stable regular file through directory FDs without link traversal."""
    path = path.absolute()
    parts = path.parts
    if len(parts) < 2 or parts[0] != os.sep:
        raise DeliveryError(f"evidence path must be absolute: {path}")
    descriptor = os.open(
        os.sep, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0)
    )
    try:
        for component in parts[1:-1]:
            next_descriptor = os.open(
                component,
                os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0),
                dir_fd=descriptor,
            )
            info = os.fstat(next_descriptor)
            if not stat.S_ISDIR(info.st_mode):
                os.close(next_descriptor)
                raise DeliveryError(f"evidence path contains a non-directory: {path}")
            os.close(descriptor)
            descriptor = next_descriptor
        name = parts[-1]
        before_name = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
        if not stat.S_ISREG(before_name.st_mode) or before_name.st_nlink != 1:
            raise DeliveryError(f"evidence file must be an unlinked regular file: {path}")
        handle = os.open(
            name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=descriptor
        )
        with os.fdopen(handle, "rb") as stream:
            before = os.fstat(stream.fileno())
            if (
                before.st_ino != before_name.st_ino
                or before.st_dev != before_name.st_dev
                or before.st_nlink != 1
            ):
                raise DeliveryError(f"evidence file changed while reading: {path}")
            value = stream.read()
            after = os.fstat(stream.fileno())
    except OSError as exc:
        raise DeliveryError(f"unsafe evidence path (symbolic links are forbidden): {path}") from exc
    finally:
        os.close(descriptor)
    if (
        before.st_ino != after.st_ino
        or before.st_dev != after.st_dev
        or after.st_nlink != 1
    ):
        raise DeliveryError(f"evidence file changed while reading: {path}")
    return value


def read_json(path: Path) -> dict[str, Any]:
    return read_json_bytes(safe_bytes(path), str(path))


def read_json_bytes(value: bytes, label: str) -> dict[str, Any]:
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError as exc:
        raise DeliveryError(f"invalid JSON evidence: {label}") from exc
    if not isinstance(parsed, dict):
        raise DeliveryError(f"JSON object required: {label}")
    return parsed


def require_digest(value: str, field: str) -> str:
    if len(value) != 64 or set(value) - set("0123456789abcdef"):
        raise DeliveryError(f"{field} must be a lowercase sha256 digest")
    return value


def safe_state_root(path: Path) -> Path:
    path = path.absolute()
    if path.is_symlink() or any(
        part.is_symlink() for part in [path, *path.parents] if part.exists()
    ):
        raise DeliveryError("state root and ancestors must not be symbolic links")
    path.mkdir(mode=0o700, parents=True, exist_ok=True)
    info = os.lstat(path)
    if not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o077:
        raise DeliveryError("state root must be a private directory")
    return path


def private_dir_fd(root: Path, *parts: str) -> tuple[Path, int]:
    """Open an owned state directory without following any controlled path part."""
    current = safe_state_root(root)
    descriptor = os.open(
        current, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0)
    )
    try:
        for part in parts:
            if not part or "/" in part or part in {".", ".."}:
                raise DeliveryError("invalid state directory component")
            with contextlib.suppress(FileExistsError):
                os.mkdir(part, mode=0o700, dir_fd=descriptor)
            next_descriptor = os.open(
                part,
                os.O_RDONLY
                | os.O_DIRECTORY
                | getattr(os, "O_NOFOLLOW", 0),
                dir_fd=descriptor,
            )
            info = os.fstat(next_descriptor)
            if not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o077:
                os.close(next_descriptor)
                raise DeliveryError("state subdirectory must be a private directory")
            os.close(descriptor)
            descriptor = next_descriptor
            current = current / part
    except OSError as exc:
        os.close(descriptor)
        raise DeliveryError("unsafe state subdirectory") from exc
    except Exception:
        os.close(descriptor)
        raise
    return current, descriptor


def state_read_json(descriptor: int, name: str) -> dict[str, Any]:
    if "/" in name or name in {"", ".", ".."}:
        raise DeliveryError("invalid state file name")
    try:
        handle = os.open(
            name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=descriptor
        )
        with os.fdopen(handle, "rb") as stream:
            before = os.fstat(stream.fileno())
            if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
                raise DeliveryError("state file must be an unlinked regular file")
            value = stream.read()
            after = os.fstat(stream.fileno())
    except FileNotFoundError:
        raise
    except OSError as exc:
        raise DeliveryError("unsafe state file") from exc
    if before.st_ino != after.st_ino or after.st_nlink != 1:
        raise DeliveryError("state file changed while reading")
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError as exc:
        raise DeliveryError("invalid state JSON") from exc
    if not isinstance(parsed, dict):
        raise DeliveryError("state JSON object required")
    return parsed


def state_write_json(descriptor: int, name: str, value: dict[str, Any]) -> None:
    """Atomic, fsync-backed replacement within a previously verified directory fd."""
    if "/" in name or name in {"", ".", ".."}:
        raise DeliveryError("invalid state file name")
    temporary = f".tmp-{os.getpid()}-{os.urandom(8).hex()}"
    try:
        handle = os.open(
            temporary,
            os.O_WRONLY
            | os.O_CREAT
            | os.O_EXCL
            | getattr(os, "O_NOFOLLOW", 0),
            0o600,
            dir_fd=descriptor,
        )
        with os.fdopen(handle, "wb") as stream:
            stream.write(
                json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True).encode()
                + b"\n"
            )
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, name, src_dir_fd=descriptor, dst_dir_fd=descriptor)
        os.fsync(descriptor)
    except OSError as exc:
        raise DeliveryError("unsafe state write") from exc
    finally:
        with contextlib.suppress(FileNotFoundError):
            os.unlink(temporary, dir_fd=descriptor)


def state_file(root: Path, name: str) -> Path:
    """Only for non-state output paths; controlled state uses descriptor helpers."""
    path = root / name
    if path.exists() and (path.is_symlink() or os.lstat(path).st_nlink != 1):
        raise DeliveryError("state file must not be linked")
    return path


class Lock:
    def __init__(self, root: Path) -> None:
        self.root = root
        self.stream: Any | None = None
        self.descriptor: int | None = None

    def __enter__(self) -> Lock:
        _, self.descriptor = private_dir_fd(self.root)
        descriptor = os.open(
            ".lock",
            os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0),
            0o600,
            dir_fd=self.descriptor,
        )
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
            os.close(descriptor)
            os.close(self.descriptor)
            raise DeliveryError("state lock must be an unlinked regular file")
        self.stream = os.fdopen(descriptor, "r+")
        fcntl.flock(self.stream.fileno(), fcntl.LOCK_EX)
        return self

    def __exit__(self, *_: Any) -> None:
        assert self.stream
        fcntl.flock(self.stream.fileno(), fcntl.LOCK_UN)
        self.stream.close()
        assert self.descriptor is not None
        os.close(self.descriptor)


def write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(prefix=".tmp-", dir=path.parent)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(
                json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True).encode()
                + b"\n"
            )
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, 0o600)
        os.replace(temporary, path)
        directory = os.open(path.parent, os.O_RDONLY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def evidence_time(value: dict[str, Any], label: str) -> None:
    try:
        issued = datetime.fromisoformat(str(value["issued_at"]).replace("Z", "+00:00"))
        expiry = datetime.fromisoformat(str(value["expires_at"]).replace("Z", "+00:00"))
    except (KeyError, ValueError, TypeError) as exc:
        raise DeliveryError(f"{label} evidence timestamps required") from exc
    now = datetime.now(UTC)
    if issued.astimezone(UTC) > now or expiry.astimezone(UTC) <= now or issued >= expiry:
        raise DeliveryError(f"{label} evidence expired or not yet valid")
    if value.get("provider") not in {"fake_local_only", "external"}:
        raise DeliveryError(f"{label} evidence provider required")


def approval_time(value: dict[str, Any]) -> None:
    issued_text, expires_text = value.get("issued_at"), value.get("expires_at")
    if not isinstance(issued_text, str) or not isinstance(expires_text, str):
        raise DeliveryError("approval evidence timestamps required")
    rfc3339_utc = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$"
    if not re.fullmatch(rfc3339_utc, issued_text) or not re.fullmatch(
        rfc3339_utc, expires_text
    ):
        raise DeliveryError("approval evidence timestamps must be UTC RFC3339")
    try:
        issued = datetime.fromisoformat(issued_text.replace("Z", "+00:00"))
        expires = datetime.fromisoformat(expires_text.replace("Z", "+00:00"))
    except ValueError as exc:
        raise DeliveryError("approval evidence timestamps required") from exc
    now = datetime.now(UTC)
    if (
        issued > now + CLOCK_SKEW
        or expires <= now
        or issued >= expires
        or expires - issued > MAX_APPROVAL_WINDOW
    ):
        raise DeliveryError("approval evidence expired, not yet valid, or window exceeds limit")


def validate_sbom(sbom: dict[str, Any], artifact_sha: str) -> None:
    if (
        sbom.get("bomFormat") != "CycloneDX"
        or not isinstance(sbom.get("specVersion"), str)
        or not str(sbom.get("serialNumber", "")).startswith("urn:uuid:")
        or not isinstance(sbom.get("version"), int)
        or not isinstance(sbom.get("components"), list)
    ):
        raise DeliveryError("CycloneDX SBOM schema minimum required")
    properties = sbom.get("metadata", {}).get("properties", [])
    if not isinstance(properties, list) or not any(
        isinstance(item, dict)
        and item.get("name") == "dataops:artifact-sha256"
        and item.get("value") == artifact_sha
        for item in properties
    ):
        raise DeliveryError("CycloneDX SBOM must bind actual artifact")


def approved_signer(provider: str, key_id: str, test_only: bool) -> dict[str, Any]:
    store = read_json(TRUST_STORE)
    configured = store.get("providers", {}).get(provider)
    if not isinstance(configured, dict) or not configured.get("enabled"):
        raise DeliveryError("signing provider disabled or not pre-approved")
    if configured.get("test_only") and not test_only:
        raise DeliveryError("test-only signing provider requires explicit test-only mode")
    signer = configured.get("approved_signers", {}).get(key_id)
    if not isinstance(signer, dict):
        raise DeliveryError("signer key id is not in controlled trust store")
    if signer.get("algorithm") != "Ed25519" or not signer.get("trust_root"):
        raise DeliveryError("signer trust root is incomplete")
    try:
        public_key = bytes.fromhex(str(signer["public_key"]))
    except (KeyError, TypeError, ValueError) as exc:
        raise DeliveryError("controlled signer public key is invalid") from exc
    if len(public_key) != 32 or sha_bytes(public_key) != signer.get("public_key_sha256"):
        raise DeliveryError("controlled signer public key digest is invalid")
    return signer


def approved_backup_encryption(provider: str, key_reference: str, test_only: bool) -> None:
    configured = read_json(TRUST_STORE).get("backup_encryption_providers", {}).get(provider)
    if not isinstance(configured, dict) or not configured.get("enabled"):
        raise DeliveryError("backup encryption provider disabled or not pre-approved")
    if configured.get("test_only") and not test_only:
        raise DeliveryError("test-only backup encryption requires explicit test-only mode")
    if key_reference not in configured.get("approved_key_references", []):
        raise DeliveryError("backup key reference is not pre-approved")


def verify_controlled_signature(
    evidence: dict[str, Any], signed: dict[str, Any], label: str, test_only: bool
) -> None:
    if any(field in evidence for field in ("public_key", "trust_root", "approved_signers")):
        raise DeliveryError(f"{label} must not carry a caller supplied trust root")
    provider, key_id = evidence.get("provider"), evidence.get("key_id")
    if not isinstance(provider, str) or not isinstance(key_id, str):
        raise DeliveryError(f"{label} provider and key id required")
    if evidence.get("algorithm") != "Ed25519":
        raise DeliveryError(f"{label} algorithm must be Ed25519")
    try:
        signature = bytes.fromhex(str(evidence["signature"]))
        if len(signature) != 64:
            raise ValueError("Ed25519 signature length")
        signer = approved_signer(provider, key_id, test_only)
        Ed25519PublicKey.from_public_bytes(
            bytes.fromhex(str(signer["public_key"]))
        ).verify(signature, canonical(signed))
    except (KeyError, TypeError, ValueError) as exc:
        raise DeliveryError(f"invalid {label}") from exc
    except Exception as exc:
        raise DeliveryError(f"{label} verification failed") from exc


def candidate_evidence(args: argparse.Namespace) -> dict[str, Any]:
    # Each caller supplied evidence file is opened, inode-checked and read once.
    raw = {name: safe_bytes(getattr(args, name)) for name in ("artifact", "sbom", "scan", "license", "signature")}
    artifact_sha = sha_bytes(raw["artifact"])
    sbom_sha = sha_bytes(raw["sbom"])
    sbom = read_json_bytes(raw["sbom"], "sbom")
    validate_sbom(sbom, artifact_sha)
    signature = read_json_bytes(raw["signature"], "signature")
    signature_sha = sha_bytes(raw["signature"])
    scan = read_json_bytes(raw["scan"], "scan")
    license_value = read_json_bytes(raw["license"], "license")
    if any(field in signature for field in ("public_key", "trust_root", "approved_signers")):
        raise DeliveryError("signature evidence must not carry a caller supplied trust root")
    evidence_time(scan, "scan")
    evidence_time(license_value, "license")
    provider = signature.get("provider")
    key_id = signature.get("key_id")
    if not isinstance(provider, str) or not isinstance(key_id, str):
        raise DeliveryError("signature provider and key id required")
    if (
        scan.get("artifact_sha256") != artifact_sha
        or scan.get("sbom_sha256") != sbom_sha
        or scan.get("signature_sha256") != signature_sha
        or scan.get("provider") != provider
        or scan.get("key_id") != key_id
        or scan.get("status") != "clear"
    ):
        raise DeliveryError("scan evidence does not clear actual artifact")
    if (
        license_value.get("artifact_sha256") != artifact_sha
        or license_value.get("sbom_sha256") != sbom_sha
        or license_value.get("signature_sha256") != signature_sha
        or license_value.get("provider") != provider
        or license_value.get("key_id") != key_id
        or license_value.get("status") != "clear"
    ):
        raise DeliveryError("license evidence does not clear actual SBOM")
    signed = {"artifact_sha256": artifact_sha, "sbom_sha256": sbom_sha}
    if (
        signature.get("algorithm") != "Ed25519"
        or signature.get("provider") != provider
        or signature.get("artifact_sha256") != artifact_sha
        or signature.get("sbom_sha256") != sbom_sha
    ):
        raise DeliveryError("signature evidence does not bind actual artifact and SBOM")
    try:
        evidence_time(signature, "signature")
        signer = approved_signer(provider, key_id, bool(args.test_only))
        public_key = bytes.fromhex(str(signer["public_key"]))
        Ed25519PublicKey.from_public_bytes(public_key).verify(
            bytes.fromhex(str(signature["signature"])), canonical(signed)
        )
    except (KeyError, ValueError, TypeError) as exc:
        raise DeliveryError("invalid signature evidence") from exc
    except Exception as exc:
        raise DeliveryError("signature evidence verification failed") from exc
    return {
        "artifact_sha256": artifact_sha,
        "sbom_sha256": sbom_sha,
        "scan_sha256": sha_bytes(raw["scan"]),
        "license_sha256": sha_bytes(raw["license"]),
        "signature_sha256": signature_sha,
    }


def command_candidate(args: argparse.Namespace) -> dict[str, Any]:
    root = safe_state_root(args.state_dir)
    if args.environment not in ENVIRONMENTS or not args.approval_ref.strip():
        raise DeliveryError("known environment and approval reference required")
    evidence = candidate_evidence(args)
    value = {
        "environment": args.environment,
        "version": args.version,
        "approval_ref": args.approval_ref,
        "migration_head": HEAD,
        **evidence,
    }
    value["bundle_digest"] = sha_bytes(
        canonical(
            {
                "artifact_sha256": evidence["artifact_sha256"],
                "sbom_sha256": evidence["sbom_sha256"],
                "openapi_sha256": sha_file(ROOT / "docs/architecture/OPENAPI.yaml"),
                "config_schema_sha256": sha_file(
                    ROOT / "deployment/helm/dataops-platform/values.schema.json"
                ),
                "migration_head": HEAD,
            }
        )
    )
    value["candidate_digest"] = sha_bytes(canonical(value))
    with Lock(root):
        _, candidates_fd = private_dir_fd(root, "candidates")
        try:
            filename = f"{value['candidate_digest']}.json"
            try:
                existing = state_read_json(candidates_fd, filename)
            except FileNotFoundError:
                existing = None
            if existing is not None and existing != value:
                raise DeliveryError("candidate collision")
            if existing is None:
                state_write_json(candidates_fd, filename, value)
        finally:
            os.close(candidates_fd)
    return {
        "status": "candidate_created",
        "candidate_digest": value["candidate_digest"],
    }


def load_candidate(root: Path, digest: str) -> dict[str, Any]:
    require_digest(digest, "candidate_digest")
    _, descriptor = private_dir_fd(root, "candidates")
    try:
        return state_read_json(descriptor, f"{digest}.json")
    finally:
        os.close(descriptor)


def current_record(root: Path, environment: str) -> dict[str, Any]:
    _, descriptor = private_dir_fd(root, "current")
    try:
        try:
            return state_read_json(descriptor, f"{environment}.json")
        except FileNotFoundError:
            return {"current_digest": "none", "bundle_digest": "none"}
    finally:
        os.close(descriptor)


def write_current(root: Path, environment: str, value: dict[str, Any]) -> None:
    _, descriptor = private_dir_fd(root, "current")
    try:
        state_write_json(descriptor, f"{environment}.json", value)
    finally:
        os.close(descriptor)


def append_history(root: Path, environment: str, value: dict[str, Any]) -> None:
    _, descriptor = private_dir_fd(root, "history", environment)
    try:
        state_write_json(descriptor, f"{value['current_digest']}.json", value)
    finally:
        os.close(descriptor)


def command_promote(args: argparse.Namespace) -> dict[str, Any]:
    root = safe_state_root(args.state_dir)
    target_index = ENVIRONMENTS.index(args.environment)
    candidate = load_candidate(root, args.candidate_digest)
    if candidate.get("environment") != args.environment:
        raise DeliveryError("candidate environment mismatch")
    with Lock(root):
        current = current_record(root, args.environment)
        if current["current_digest"] == args.candidate_digest:
            return {
                "status": "promoted_replayed",
                "current_digest": args.candidate_digest,
            }
        if target_index:
            if current["current_digest"] != "none":
                raise DeliveryError("current digest CAS conflict")
            previous = current_record(root, ENVIRONMENTS[target_index - 1])
            if previous.get("current_digest") != args.expected_base_digest:
                raise DeliveryError(
                    "promotion must use adjacent prior environment current digest"
                )
            if previous.get("bundle_digest") != candidate.get("bundle_digest"):
                raise DeliveryError(
                    "promotion must retain the exact adjacent artifact bundle"
                )
        elif current["current_digest"] != args.expected_base_digest:
            raise DeliveryError("current digest CAS conflict")
        value = {
            "current_digest": args.candidate_digest,
            "previous_digest": args.expected_base_digest
            if target_index
            else current["current_digest"],
            "approval_ref": candidate["approval_ref"],
            "environment": args.environment,
            "artifact_sha256": candidate["artifact_sha256"],
            "bundle_digest": candidate["bundle_digest"],
            "previous_bundle_digest": previous.get("bundle_digest")
            if target_index
            else current.get("bundle_digest"),
        }
        write_current(root, args.environment, value)
        append_history(root, args.environment, value)
    return {"status": "promoted", "current_digest": args.candidate_digest}


def command_rollback(args: argparse.Namespace) -> dict[str, Any]:
    root = safe_state_root(args.state_dir)
    with Lock(root):
        current = current_record(root, args.environment)
        if not args.approval_ref.strip() or args.target_digest != current.get(
            "previous_digest"
        ):
            raise DeliveryError(
                "rollback requires approval and exact recorded previous digest"
            )
        target = load_candidate(root, args.target_digest)
        if target.get("bundle_digest") != current.get("previous_bundle_digest", target.get("bundle_digest")):
            raise DeliveryError("rollback must retain recorded artifact bundle history")
        value = {
            **current,
            "current_digest": args.target_digest,
            "previous_digest": current["current_digest"],
            "previous_bundle_digest": current.get("bundle_digest"),
            "artifact_sha256": target["artifact_sha256"],
            "bundle_digest": target["bundle_digest"],
            "rollback_approval_ref": args.approval_ref,
        }
        write_current(root, args.environment, value)
        append_history(root, args.environment, value)
    return {"status": "rollback_recorded", "current_digest": args.target_digest}


def command_backup(args: argparse.Namespace) -> dict[str, Any]:
    if args.environment == "production" and args.encryption_provider != "external":
        raise DeliveryError("production backup requires an approved external encryption provider")
    if args.encryption_provider not in {
        "fake",
        "external",
        "disabled",
    } or not args.key_reference.startswith(("kms://", "hsm://", "secretref://")):
        raise DeliveryError("safe provider and key reference required")
    approved_backup_encryption(
        args.encryption_provider, args.key_reference, bool(args.test_only)
    )
    if args.encryption_provider == "external" and (
        args.backup_signature is None or not args.signer_key_id
    ):
        raise DeliveryError("external backup requires a controlled signer key id and signature")
    if args.encryption_provider == "fake" and not args.test_only:
        raise DeliveryError("fake backup encryption requires explicit test-only mode")
    directory = safe_state_root(args.backup_dir)
    release = {
        "path": args.release_artifact.name,
        "sha256": sha_file(args.release_artifact),
    }
    entries = {
        "postgres": {**release, "mode": "logical_or_snapshot"},
        "neo4j": {**release, "mode": "snapshot"},
        "minio": {**release, "mode": "snapshot"},
        "config": {
            "path": args.config_artifact.name,
            "sha256": sha_file(args.config_artifact),
            "mode": "digest_only",
        },
        "key_references": {"reference": args.key_reference, "mode": "reference_only"},
    }
    manifest = {
        "environment": args.environment,
        "components": entries,
        "encryption_provider": args.encryption_provider,
        "key_reference": args.key_reference,
        "migration_head": HEAD,
        "signature_provider": "external"
        if args.encryption_provider == "external"
        else "fake_local_only",
        "signer_key_id": args.signer_key_id
        if args.encryption_provider == "external"
        else "wp07-local-test",
    }
    manifest["manifest_sha256"] = sha_bytes(canonical(manifest))
    signature_payload = {
        "manifest_sha256": manifest["manifest_sha256"],
        "encryption_provider": manifest["encryption_provider"],
        "key_reference": manifest["key_reference"],
    }
    if args.encryption_provider == "external":
        supplied = read_json(args.backup_signature)
        if (
            supplied.get("provider") != "external"
            or supplied.get("key_id") != args.signer_key_id
        ):
            raise DeliveryError("external backup signature signer mismatch")
        verify_controlled_signature(supplied, signature_payload, "backup signature", False)
        manifest["integrity_signature"] = {
            "algorithm": "Ed25519",
            "provider": "external",
            "key_id": args.signer_key_id,
            "signature": supplied["signature"],
        }
    else:
        signer = approved_signer("fake_local_only", "wp07-local-test", True)
        private = Ed25519PrivateKey.from_private_bytes(LOCAL_TEST_PRIVATE_KEY)
        manifest["integrity_signature"] = {
            "algorithm": "Ed25519",
            "provider": "fake_local_only",
            "key_id": "wp07-local-test",
            "signature": private.sign(canonical(signature_payload)).hex(),
        }
        if signer.get("public_key") != private.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw
        ).hex():
            raise DeliveryError("local test signer does not match controlled trust store")
    path = state_file(directory, "backup-manifest.json")
    write_json(path, manifest)
    _, backup_fd = private_dir_fd(directory, "backup")
    try:
        state_write_json(backup_fd, "latest.json", manifest)
    finally:
        os.close(backup_fd)
    return {"status": "planned", "manifest_path": str(path)}


def command_restore(args: argparse.Namespace) -> dict[str, Any]:
    value = read_json(args.backup_manifest)
    check = {
        key: value[key]
        for key in value
        if key not in {"manifest_sha256", "integrity_signature"}
    }
    if value.get("manifest_sha256") != sha_bytes(canonical(check)):
        raise DeliveryError("backup manifest digest mismatch")
    try:
        signature = value["integrity_signature"]
        if not isinstance(signature, dict) or signature.get("algorithm") != "Ed25519":
            raise DeliveryError("backup signature algorithm invalid")
        if (
            signature.get("provider") != value.get("signature_provider")
            or signature.get("key_id") != value.get("signer_key_id")
        ):
            raise DeliveryError("backup signature signer identity mismatch")
        if (
            value.get("encryption_provider") == "external"
            and value.get("signature_provider") != "external"
        ) or (
            value.get("encryption_provider") == "fake"
            and value.get("signature_provider") != "fake_local_only"
        ):
            raise DeliveryError("backup encryption and signer provider mismatch")
        signature_payload = {
            "manifest_sha256": value["manifest_sha256"],
            "encryption_provider": value["encryption_provider"],
            "key_reference": value["key_reference"],
        }
        verify_controlled_signature(
            signature, signature_payload, "backup signature", bool(args.test_only)
        )
    except DeliveryError:
        raise
    except (KeyError, TypeError, ValueError) as exc:
        raise DeliveryError("backup signature invalid") from exc
    except Exception as exc:
        raise DeliveryError("backup signature verification failed") from exc
    if args.target == "production" or args.target != args.confirm_target:
        raise DeliveryError("restore requires exact new non-production target")
    if args.environment == "production" and value.get("encryption_provider") != "external":
        raise DeliveryError("production restore requires an externally encrypted backup")
    approved_backup_encryption(
        str(value.get("encryption_provider")), str(value.get("key_reference")), bool(args.test_only)
    )
    components = value.get("components", {})
    if set(components) != {"postgres", "neo4j", "minio", "config", "key_references"}:
        raise DeliveryError("backup component set incomplete")
    parent = args.backup_manifest.parent
    for name, item in components.items():
        if name == "key_references":
            if not str(item.get("reference", "")).startswith(
                ("kms://", "hsm://", "secretref://")
            ):
                raise DeliveryError("backup key reference invalid")
            continue
        path = parent / str(item["path"])
        if sha_file(path) != item["sha256"]:
            raise DeliveryError("backup component checksum mismatch")
    if value.get("migration_head") != HEAD:
        raise DeliveryError("backup compatibility migration mismatch")
    restore_root = safe_state_root(args.backup_manifest.parent)
    with Lock(restore_root):
        _, restore_fd = private_dir_fd(restore_root, "restore")
        try:
            try:
                state_read_json(restore_fd, f"{args.target}.json")
            except FileNotFoundError:
                pass
            else:
                raise DeliveryError("restore target has already been used")
            state_write_json(
                restore_fd,
                f"{args.target}.json",
                {
                    "target": args.target,
                    "manifest_sha256": value["manifest_sha256"],
                    "status": "restore_to_new_target_planned",
                },
            )
        finally:
            os.close(restore_fd)
    return {"status": "restore_to_new_target_planned", "target": args.target}


def command_compatibility(args: argparse.Namespace) -> dict[str, Any]:
    def release(artifact_path: Path, sbom_path: Path, label: str) -> dict[str, Any]:
        artifact = safe_bytes(artifact_path)
        sbom_bytes = safe_bytes(sbom_path)
        artifact_sha = sha_bytes(artifact)
        sbom = read_json_bytes(sbom_bytes, f"{label} SBOM")
        validate_sbom(sbom, artifact_sha)
        components = sbom.get("components", [])
        identity = [
            {
                "bom-ref": component.get("bom-ref", ""),
                "name": component.get("name", ""),
                "version": component.get("version", ""),
                "purl": component.get("purl", ""),
            }
            for component in components
            if isinstance(component, dict)
        ]
        if len(identity) != len(components):
            raise DeliveryError(f"{label} SBOM component identity is malformed")
        identity.sort(key=canonical)
        return {
            "artifact_sha256": artifact_sha,
            "sbom_sha256": sha_bytes(sbom_bytes),
            "sbom_serial": sbom.get("serialNumber"),
            "component_identity_sha256": sha_bytes(canonical(identity)),
        }

    releases = {
        "base": release(args.base_artifact, args.base_sbom, "base"),
        "candidate": release(args.candidate_artifact, args.candidate_sbom, "candidate"),
        "rollback": release(args.rollback_artifact, args.rollback_sbom, "rollback"),
    }

    def contract(path: Path | None, label: str, bound_release: dict[str, Any]) -> dict[str, Any]:
        if path is None:
            raise DeliveryError("base, candidate and rollback contract manifests are required")
        supplied = read_json(path)
        source: dict[str, Path] = {}
        for field in ("openapi", "config", "chart", "values"):
            value = supplied.get(field)
            if not isinstance(value, str) or not value:
                raise DeliveryError(f"{label} compatibility contract {field} is required")
            source[field] = Path(value)
        if (
            supplied.get("artifact_sha256") != bound_release["artifact_sha256"]
            or supplied.get("sbom_sha256") != bound_release["sbom_sha256"]
            or supplied.get("component_identity_sha256")
            != bound_release["component_identity_sha256"]
        ):
            raise DeliveryError(f"{label} contract does not bind actual artifact/SBOM identity")
        chain = supplied.get("migration_chain")
        head = supplied.get("migration_head")
        if (
            not isinstance(head, str)
            or not isinstance(chain, list)
            or not chain
            or not all(isinstance(item, str) and item for item in chain)
            or len(set(chain)) != len(chain)
            or chain[-1] != head
        ):
            raise DeliveryError(f"{label} migration chain and head are required")
        raw = {field: safe_bytes(item) for field, item in source.items()}
        config = read_json_bytes(raw["config"], f"{label} config schema")
        required = config.get("required", [])
        if not isinstance(required, list) or not all(isinstance(item, str) for item in required):
            raise DeliveryError(f"{label} config required list is invalid")
        value = {
            "openapi_sha256": sha_bytes(raw["openapi"]),
            "config_schema_sha256": sha_bytes(raw["config"]),
            "chart_sha256": sha_bytes(raw["chart"]),
            "values_sha256": sha_bytes(raw["values"]),
            "migration_head": head,
            "migration_chain": chain,
            "openapi_operations": sorted(
                re.findall(r"^  (/[^:\s]+):", raw["openapi"].decode(errors="strict"), re.MULTILINE)
            ),
            "config_required": sorted(required),
        }
        value["contract_sha256"] = sha_bytes(canonical(value))
        return value

    contracts = {
        "base": contract(args.base_contract, "base", releases["base"]),
        "candidate": contract(args.candidate_contract, "candidate", releases["candidate"]),
        "rollback": contract(args.rollback_contract, "rollback", releases["rollback"]),
    }
    base, candidate_contract, rollback_contract = (
        contracts["base"], contracts["candidate"], contracts["rollback"]
    )
    if not set(base["openapi_operations"]).issubset(candidate_contract["openapi_operations"]):
        raise DeliveryError("candidate OpenAPI removes a base operation")
    if not set(candidate_contract["config_required"]).issubset(base["config_required"]):
        raise DeliveryError("candidate config introduces a newly required setting")
    if candidate_contract["migration_chain"][: len(base["migration_chain"])] != base["migration_chain"]:
        raise DeliveryError("candidate migration chain is not an ordered extension of base")
    if (
        base["chart_sha256"] != candidate_contract["chart_sha256"]
        or base["values_sha256"] != candidate_contract["values_sha256"]
        or rollback_contract != base
    ):
        raise DeliveryError("base/candidate/rollback deployment contract is incompatible")
    if releases["rollback"] != releases["base"]:
        raise DeliveryError("rollback must restore the exact tested base artifact and SBOM")
    if releases["candidate"] != releases["base"]:
        if args.compatibility_approval is None:
            raise DeliveryError("changed artifact/SBOM requires a bound compatibility approval")
        approval = read_json(args.compatibility_approval)
        approval_time(approval)
        approval_payload = {
            "base": releases["base"],
            "candidate": releases["candidate"],
            "base_contract_sha256": base["contract_sha256"],
            "candidate_contract_sha256": candidate_contract["contract_sha256"],
            "approval_ref": approval.get("approval_ref"),
            "issued_at": approval.get("issued_at"),
            "expires_at": approval.get("expires_at"),
        }
        if (
            approval.get("base") != releases["base"]
            or approval.get("candidate") != releases["candidate"]
            or approval.get("base_contract_sha256") != base["contract_sha256"]
            or approval.get("candidate_contract_sha256")
            != candidate_contract["contract_sha256"]
            or not str(approval.get("approval_ref", "")).strip()
        ):
            raise DeliveryError("compatibility approval does not bind actual release pair")
        verify_controlled_signature(
            approval, approval_payload, "compatibility approval", False
        )
    value = {
        "base_version": args.base_version,
        "candidate_version": args.candidate_version,
        "migration_head": candidate_contract["migration_head"],
        "contracts": contracts,
        "openapi_sha256": candidate_contract["openapi_sha256"],
        "config_schema_sha256": candidate_contract["config_schema_sha256"],
        "chart_sha256": candidate_contract["chart_sha256"],
        "releases": releases,
        "comparison": {
            "base_to_candidate": "verified",
            "candidate_to_rollback": "verified_exact_base",
            "openapi": "base_operations_retained",
            "config_schema": "no_new_required_setting",
            "chart_values": "base_candidate_equal",
            "migration": "ancestry_and_rollback_checked",
        },
        "rollback": releases["rollback"],
    }
    value["matrix_sha256"] = sha_bytes(canonical(value))
    write_json(args.output, value)
    return {"status": "compatible", "matrix_sha256": value["matrix_sha256"]}


def parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    c = sub.add_parser("candidate")
    c.add_argument("--state-dir", type=Path, required=True)
    c.add_argument("--environment", required=True)
    c.add_argument("--version", required=True)
    c.add_argument("--artifact", type=Path, required=True)
    c.add_argument("--sbom", type=Path, required=True)
    c.add_argument("--scan", type=Path, required=True)
    c.add_argument("--license", type=Path, required=True)
    c.add_argument("--signature", type=Path, required=True)
    c.add_argument("--approval-ref", required=True)
    c.add_argument("--test-only", action="store_true")
    p = sub.add_parser("promote")
    p.add_argument("--state-dir", type=Path, required=True)
    p.add_argument("--environment", required=True)
    p.add_argument("--candidate-digest", required=True)
    p.add_argument("--expected-base-digest", required=True)
    r = sub.add_parser("rollback")
    r.add_argument("--state-dir", type=Path, required=True)
    r.add_argument("--environment", required=True)
    r.add_argument("--target-digest", required=True)
    r.add_argument("--approval-ref", required=True)
    b = sub.add_parser("backup-plan")
    b.add_argument("--backup-dir", type=Path, required=True)
    b.add_argument("--release-artifact", type=Path, required=True)
    b.add_argument("--config-artifact", type=Path, required=True)
    b.add_argument("--key-reference", required=True)
    b.add_argument("--encryption-provider", required=True)
    b.add_argument("--environment", default="test")
    b.add_argument("--signer-key-id")
    b.add_argument("--backup-signature", type=Path)
    b.add_argument("--test-only", action="store_true")
    rs = sub.add_parser("restore-plan")
    rs.add_argument("--backup-manifest", type=Path, required=True)
    rs.add_argument("--target", required=True)
    rs.add_argument("--confirm-target", required=True)
    rs.add_argument("--environment", required=True)
    rs.add_argument("--test-only", action="store_true")
    cm = sub.add_parser("compatibility")
    cm.add_argument("--output", type=Path, required=True)
    cm.add_argument("--base-version", required=True)
    cm.add_argument("--candidate-version", required=True)
    cm.add_argument("--base-artifact", type=Path, required=True)
    cm.add_argument("--base-sbom", type=Path, required=True)
    cm.add_argument("--candidate-artifact", type=Path, required=True)
    cm.add_argument("--candidate-sbom", type=Path, required=True)
    cm.add_argument("--rollback-artifact", type=Path, required=True)
    cm.add_argument("--rollback-sbom", type=Path, required=True)
    cm.add_argument("--compatibility-approval", type=Path)
    cm.add_argument("--base-contract", type=Path)
    cm.add_argument("--candidate-contract", type=Path)
    cm.add_argument("--rollback-contract", type=Path)
    return parser


def main() -> int:
    args = parser().parse_args()
    try:
        result = {
            "candidate": command_candidate,
            "promote": command_promote,
            "rollback": command_rollback,
            "backup-plan": command_backup,
            "restore-plan": command_restore,
            "compatibility": command_compatibility,
        }[args.command](args)
        print(json.dumps(result, sort_keys=True))
        return 0
    except DeliveryError as exc:
        print(json.dumps({"status": "rejected", "error": str(exc)}, sort_keys=True))
        return 2


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