DataOps-platform/deployment/scripts/edge_gateway_admin.py

#!/usr/bin/env python3
"""Fail-closed local administration for a DataOps enterprise edge gateway."""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import json
import os
import re
import sqlite3
import stat
import sys
import tempfile
from contextlib import contextmanager, suppress
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path, PurePosixPath

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import (
    ec,
    ed448,
    ed25519,
    padding,
    rsa,
)
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID

MAX_JSON_BYTES = 262_144
MAX_BACKUPS = 3
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
VERSION_RE = re.compile(r"^(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})$")
ORIGIN_RE = re.compile(r"^https://[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?(?::[0-9]{1,5})?$")
MANIFEST_FIELDS = frozenset(
    {
        "release_id", "version", "rollback_version", "artifact_digest",
        "artifact_name", "deadline_at", "signature_algorithm", "key_id",
        "manifest_digest", "signature", "status",
    }
)


class AdminError(RuntimeError):
    def __init__(self, code: str, exit_code: int = 2):
        super().__init__(code)
        self.code = code
        self.exit_code = exit_code


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


def _digest(value: object) -> str:
    return hashlib.sha256(_json_bytes(value)).hexdigest()


def _protocol_bytes(value: object) -> bytes:
    def frame(tag: bytes, content: bytes) -> bytes:
        return tag + str(len(content)).encode("ascii") + b":" + content

    if value is None:
        return b"n"
    if isinstance(value, bool):
        return b"b1" if value else b"b0"
    if isinstance(value, (int, float)):
        number = Decimal(value if isinstance(value, int) else str(value))
        if not number.is_finite():
            raise AdminError("INVALID_INPUT")
        text = "0" if number == 0 else format(number.normalize(), "f")
        if "." in text:
            text = text.rstrip("0").rstrip(".")
        return frame(b"d", text.encode("ascii"))
    if isinstance(value, str):
        return frame(b"s", value.encode("utf-8"))
    if isinstance(value, dict):
        if any(not isinstance(key, str) for key in value):
            raise AdminError("INVALID_INPUT")
        encoded = bytearray(b"m" + str(len(value)).encode("ascii") + b":")
        for key in sorted(value, key=lambda item: item.encode("utf-8")):
            encoded.extend(frame(b"k", key.encode("utf-8")))
            encoded.extend(_protocol_bytes(value[key]))
        return bytes(encoded)
    if isinstance(value, (list, tuple)):
        encoded = bytearray(b"l" + str(len(value)).encode("ascii") + b":")
        for item in value:
            encoded.extend(_protocol_bytes(item))
        return bytes(encoded)
    raise AdminError("INVALID_INPUT")


def _protocol_digest(value: object) -> str:
    return hashlib.sha256(_protocol_bytes(value)).hexdigest()


def _output(value: dict[str, object]) -> None:
    rendered = json.dumps(value, ensure_ascii=False, sort_keys=True)
    if len(rendered.encode("utf-8")) > 32_768:
        raise AdminError("OUTPUT_LIMIT")
    print(rendered)


def _root(raw: str, *, create: bool = False) -> Path:
    candidate = Path(raw)
    if not candidate.is_absolute() or "\x00" in raw:
        raise AdminError("INVALID_INPUT")
    protected = {
        Path("/"), Path.home(), Path("/opt"), Path("/var"), Path("/usr"),
        Path("/etc"), Path("/tmp"), Path("/private"), Path("/private/tmp"),
        Path("/Users"), Path("/home"), Path("/root"), Path("/Volumes"),
    }
    normalized = Path(os.path.abspath(candidate))
    if normalized in protected:
        raise AdminError("PATH_POLICY")
    if candidate.exists() and candidate.is_symlink():
        raise AdminError("PATH_POLICY")
    if create:
        with suppress(FileExistsError):
            candidate.mkdir(mode=0o700, parents=False, exist_ok=False)
    if not candidate.is_dir():
        raise AdminError("PATH_POLICY")
    resolved = candidate.resolve(strict=True)
    if resolved != candidate:
        raise AdminError("PATH_POLICY")
    return resolved


def _runtime_identity(root: Path) -> tuple[int, int]:
    record = _read_json(root / "state" / "runtime-identity.json")
    if set(record) != {"runtime_gid", "runtime_uid", "schema_version"}:
        raise AdminError("OWNER_POLICY")
    uid, gid = record["runtime_uid"], record["runtime_gid"]
    if (
        isinstance(uid, bool)
        or not isinstance(uid, int)
        or isinstance(gid, bool)
        or not isinstance(gid, int)
        or uid < 1
        or gid < 1
    ):
        raise AdminError("OWNER_POLICY")
    metadata = root.stat()
    if metadata.st_uid != uid or metadata.st_gid != gid:
        raise AdminError("OWNER_POLICY")
    return uid, gid


def _validated_root(raw: str) -> Path:
    root = _root(raw)
    _runtime_identity(root)
    return root


def _chown_exact(path: Path, uid: int, gid: int) -> None:
    metadata = path.lstat()
    if stat.S_ISLNK(metadata.st_mode):
        raise AdminError("PATH_POLICY")
    if (metadata.st_uid, metadata.st_gid) == (uid, gid):
        return
    if os.geteuid() != 0:
        raise AdminError("OWNER_POLICY")
    os.chown(path, uid, gid, follow_symlinks=False)


def _inside(root: Path, raw: str | Path, *, must_exist: bool = False) -> Path:
    candidate = Path(raw)
    if not candidate.is_absolute() or "\x00" in str(candidate):
        raise AdminError("INVALID_INPUT")
    if candidate.is_symlink():
        raise AdminError("PATH_POLICY")
    try:
        parent = candidate.parent.resolve(strict=True)
        parent.relative_to(root)
    except (OSError, ValueError) as exc:
        raise AdminError("PATH_POLICY") from exc
    current = root
    for part in parent.relative_to(root).parts:
        current /= part
        if current.is_symlink():
            raise AdminError("PATH_POLICY")
    if must_exist:
        try:
            resolved = candidate.resolve(strict=True)
            resolved.relative_to(root)
        except (OSError, ValueError) as exc:
            raise AdminError("PATH_POLICY") from exc
        metadata = candidate.stat()
        if (
            candidate.is_symlink()
            or not stat.S_ISREG(metadata.st_mode)
            or metadata.st_nlink != 1
        ):
            raise AdminError("PATH_POLICY")
    return candidate


def _mkdir(root: Path, name: str, mode: int = 0o700) -> Path:
    path = root / name
    if path.exists() and (path.is_symlink() or not path.is_dir()):
        raise AdminError("PATH_POLICY")
    created = not path.exists()
    path.mkdir(mode=mode, exist_ok=True)
    os.chmod(path, mode)
    parent = path.parent.stat()
    metadata = path.stat()
    if created and os.geteuid() == 0:
        os.chown(path, parent.st_uid, parent.st_gid, follow_symlinks=False)
    elif (metadata.st_uid, metadata.st_gid) != (parent.st_uid, parent.st_gid):
        raise AdminError("OWNER_POLICY")
    return path


def _inherit_parent_owner(descriptor: int, parent: Path) -> None:
    parent_metadata = parent.stat()
    metadata = os.fstat(descriptor)
    expected = (parent_metadata.st_uid, parent_metadata.st_gid)
    if (metadata.st_uid, metadata.st_gid) == expected:
        return
    if os.geteuid() != 0:
        raise AdminError("OWNER_POLICY")
    os.fchown(descriptor, *expected)


def _atomic_write(path: Path, data: bytes, mode: int) -> None:
    if path.exists() and path.is_symlink():
        raise AdminError("PATH_POLICY")
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        os.fchmod(descriptor, mode)
        _inherit_parent_owner(descriptor, path.parent)
        with os.fdopen(descriptor, "wb", closefd=True) as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        os.chmod(path, mode)
        _fsync_directory(path.parent)
    except Exception:
        with suppress(FileNotFoundError):
            os.unlink(temporary)
        raise


def _fsync_directory(path: Path) -> None:
    directory = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(directory)
    finally:
        os.close(directory)


def _atomic_copy(source: Path, target: Path, mode: int) -> None:
    if target.exists() or target.is_symlink():
        raise AdminError("PATH_POLICY")
    descriptor, temporary = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
    try:
        os.fchmod(descriptor, mode)
        _inherit_parent_owner(descriptor, target.parent)
        with source.open("rb") as reader, os.fdopen(descriptor, "wb") as writer:
            for chunk in iter(lambda: reader.read(1_048_576), b""):
                writer.write(chunk)
            writer.flush()
            os.fsync(writer.fileno())
        os.replace(temporary, target)
        os.chmod(target, mode)
        _fsync_directory(target.parent)
    except Exception:
        with suppress(FileNotFoundError):
            os.unlink(temporary)
        raise


def _read_json(path: Path) -> dict[str, object]:
    if not path.is_file() or path.is_symlink() or path.stat().st_size > MAX_JSON_BYTES:
        raise AdminError("INVALID_INPUT")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise AdminError("INVALID_INPUT") from exc
    if not isinstance(value, dict):
        raise AdminError("INVALID_INPUT")
    return value


def _sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    try:
        with path.open("rb") as stream:
            for chunk in iter(lambda: stream.read(1_048_576), b""):
                digest.update(chunk)
    except OSError as exc:
        raise AdminError("INVALID_INPUT") from exc
    return digest.hexdigest()


def _open_root_file(root: Path, raw: str | Path) -> tuple[int, os.stat_result]:
    path = _inside(root, raw)
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise AdminError("PATH_POLICY") from exc
    try:
        metadata = os.fstat(descriptor)
        uid, gid = _runtime_identity(root)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_nlink != 1
            or (metadata.st_uid, metadata.st_gid) != (uid, gid)
        ):
            raise AdminError("PATH_POLICY")
        return descriptor, metadata
    except Exception:
        os.close(descriptor)
        raise


def _read_json_descriptor(descriptor: int) -> dict[str, object]:
    metadata = os.fstat(descriptor)
    if metadata.st_size > MAX_JSON_BYTES:
        raise AdminError("INVALID_INPUT")
    os.lseek(descriptor, 0, os.SEEK_SET)
    try:
        raw = b""
        while len(raw) <= MAX_JSON_BYTES:
            chunk = os.read(descriptor, min(65_536, MAX_JSON_BYTES + 1 - len(raw)))
            if not chunk:
                break
            raw += chunk
        value = json.loads(raw.decode("utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise AdminError("INVALID_INPUT") from exc
    if not isinstance(value, dict):
        raise AdminError("INVALID_INPUT")
    return value


def _copy_descriptor(
    source_descriptor: int, source_before: os.stat_result, target: Path
) -> str:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    target_descriptor = os.open(target, flags, 0o600)
    digest = hashlib.sha256()
    try:
        os.fchmod(target_descriptor, 0o600)
        _inherit_parent_owner(target_descriptor, target.parent)
        os.lseek(source_descriptor, 0, os.SEEK_SET)
        while True:
            chunk = os.read(source_descriptor, 1_048_576)
            if not chunk:
                break
            digest.update(chunk)
            view = memoryview(chunk)
            while view:
                written = os.write(target_descriptor, view)
                view = view[written:]
        os.fsync(target_descriptor)
    finally:
        os.close(target_descriptor)
    source_after = os.fstat(source_descriptor)
    snapshot_fields = (
        "st_dev", "st_ino", "st_mode", "st_nlink", "st_uid", "st_gid",
        "st_size", "st_mtime_ns", "st_ctime_ns",
    )
    if any(
        getattr(source_before, field) != getattr(source_after, field)
        for field in snapshot_fields
    ):
        raise AdminError("RELEASE_REJECTED", 3)
    copied_digest = digest.hexdigest()
    if _sha256_file(target) != copied_digest:
        raise AdminError("RELEASE_REJECTED", 3)
    return copied_digest


def _layout(root: Path) -> None:
    for name in (
        "queue", "artifacts", "releases", "secrets", "certificates",
        "registration", "state",
    ):
        _mkdir(root, name)


@contextmanager
def _process_lock(root: Path, uid: int | None = None, gid: int | None = None):
    if uid is None or gid is None:
        uid, gid = _runtime_identity(root)
    lock_path = root / "state" / "admin.lock"
    flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(lock_path, flags, 0o600)
    except OSError as exc:
        raise AdminError("LOCK_POLICY", 3) from exc
    try:
        if os.geteuid() == 0:
            os.fchown(descriptor, uid, gid)
        os.fchmod(descriptor, 0o600)
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or stat.S_IMODE(metadata.st_mode) != 0o600
            or metadata.st_nlink != 1
            or (metadata.st_uid, metadata.st_gid) != (uid, gid)
        ):
            raise AdminError("LOCK_POLICY", 3)
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise AdminError("CONCURRENT_OPERATION", 3) from exc
        yield
    finally:
        os.close(descriptor)


def _bounded_backups(directory: Path, prefix: str) -> None:
    candidates = sorted(
        (path for path in directory.glob(f"{prefix}*") if path.is_file() and not path.is_symlink()),
        key=lambda path: path.stat().st_mtime_ns,
        reverse=True,
    )
    for path in candidates[MAX_BACKUPS:]:
        path.unlink()


def command_init(args: argparse.Namespace) -> dict[str, object]:
    if (
        isinstance(args.runtime_uid, bool)
        or not 1 <= args.runtime_uid <= 2_147_483_647
        or isinstance(args.runtime_gid, bool)
        or not 1 <= args.runtime_gid <= 2_147_483_647
    ):
        raise AdminError("INVALID_INPUT")
    root = _root(args.root, create=True)
    identity_path = root / "state" / "runtime-identity.json"
    if identity_path.exists():
        uid, gid = _runtime_identity(root)
        if (uid, gid) != (args.runtime_uid, args.runtime_gid):
            raise AdminError("OWNER_POLICY")
        target = root / "edge-config.template.json"
        if not target.is_file() or target.is_symlink():
            raise AdminError("PATH_POLICY")
        with _process_lock(root):
            return {
                "config_template_path": str(target),
                "root": str(root),
                "runtime_gid": gid,
                "runtime_uid": uid,
                "status": "initialized",
            }
    if any(root.iterdir()):
        raise AdminError("PATH_POLICY")
    os.chmod(root, 0o700)
    current = root.stat()
    if (current.st_uid, current.st_gid) != (args.runtime_uid, args.runtime_gid):
        if not args.allow_chown:
            raise AdminError("OWNER_POLICY")
        _chown_exact(root, args.runtime_uid, args.runtime_gid)
    _layout(root)
    for name in (
        "queue", "artifacts", "releases", "secrets", "certificates",
        "registration", "state",
    ):
        _chown_exact(root / name, args.runtime_uid, args.runtime_gid)
    identity = {
        "runtime_gid": args.runtime_gid,
        "runtime_uid": args.runtime_uid,
        "schema_version": 1,
    }
    with _process_lock(root, args.runtime_uid, args.runtime_gid):
        _atomic_write(identity_path, _json_bytes(identity) + b"\n", 0o600)
        _chown_exact(identity_path, args.runtime_uid, args.runtime_gid)
        template = {
            "allowed_control_hosts": [],
            "allowed_proxy_hosts": [],
            "control_origin": None,
            "environment": None,
            "gateway_id": None,
            "network_zone": None,
            "policy_digest": None,
            "proxy_origin": None,
            "runtime_gid": args.runtime_gid,
            "runtime_uid": args.runtime_uid,
            "schema_version": 1,
        }
        target = root / "edge-config.template.json"
        _atomic_write(target, _json_bytes(template) + b"\n", 0o600)
        _chown_exact(target, args.runtime_uid, args.runtime_gid)
    return {
        "config_template_path": str(target),
        "root": str(root),
        "runtime_gid": args.runtime_gid,
        "runtime_uid": args.runtime_uid,
        "status": "initialized",
    }


def _new_csr(root: Path, common_name: str, prefix: str) -> dict[str, object]:
    if not IDENTIFIER_RE.fullmatch(common_name):
        raise AdminError("INVALID_INPUT")
    key_path = root / "secrets" / f"{prefix}.key"
    csr_path = root / "certificates" / f"{prefix}.csr"
    if key_path.exists() or csr_path.exists():
        raise AdminError("MATERIAL_EXISTS")
    private = rsa.generate_private_key(public_exponent=65537, key_size=3072)
    csr = (
        x509.CertificateSigningRequestBuilder()
        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
        .sign(private, hashes.SHA256())
    )
    key_bytes = private.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    )
    csr_bytes = csr.public_bytes(serialization.Encoding.PEM)
    try:
        _atomic_write(key_path, key_bytes, 0o600)
        _atomic_write(csr_path, csr_bytes, 0o644)
    except Exception:
        if key_path.exists():
            key_path.unlink()
        if csr_path.exists():
            csr_path.unlink()
        raise
    return {
        "csr_path": str(csr_path),
        "private_key_path": str(key_path),
        "public_key_sha256": hashlib.sha256(
            csr.public_key().public_bytes(
                serialization.Encoding.DER,
                serialization.PublicFormat.SubjectPublicKeyInfo,
            )
        ).hexdigest(),
        "status": "csr_created",
    }


def command_csr(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    _layout(root)
    return _new_csr(root, args.common_name, "client")


def _verify_x509_signature(value, issuer_public_key) -> None:
    try:
        if isinstance(issuer_public_key, rsa.RSAPublicKey):
            issuer_public_key.verify(
                value.signature, value.tbs_certificate_bytes
                if isinstance(value, x509.Certificate)
                else value.tbs_certlist_bytes,
                padding.PKCS1v15(), value.signature_hash_algorithm,
            )
        elif isinstance(issuer_public_key, ec.EllipticCurvePublicKey):
            issuer_public_key.verify(
                value.signature, value.tbs_certificate_bytes
                if isinstance(value, x509.Certificate)
                else value.tbs_certlist_bytes,
                ec.ECDSA(value.signature_hash_algorithm),
            )
        elif isinstance(issuer_public_key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
            issuer_public_key.verify(
                value.signature, value.tbs_certificate_bytes
                if isinstance(value, x509.Certificate)
                else value.tbs_certlist_bytes,
            )
        else:
            raise AdminError("MATERIAL_INVALID")
    except (InvalidSignature, ValueError, TypeError) as exc:
        raise AdminError("MATERIAL_INVALID") from exc


def _load_material(
    root: Path, certificate_raw: str, ca_raw: str, crl_raw: str,
    prefix: str = "client",
):
    certificate_path = _inside(root, certificate_raw, must_exist=True)
    ca_path = _inside(root, ca_raw, must_exist=True)
    crl_path = _inside(root, crl_raw, must_exist=True)
    key_path = _inside(root, root / "secrets" / f"{prefix}.key", must_exist=True)
    csr_path = _inside(root, root / "certificates" / f"{prefix}.csr", must_exist=True)
    if stat.S_IMODE(key_path.stat().st_mode) & 0o077:
        raise AdminError("KEY_PERMISSION")
    try:
        certificate = x509.load_pem_x509_certificate(certificate_path.read_bytes())
        authorities = x509.load_pem_x509_certificates(ca_path.read_bytes())
        crl = x509.load_pem_x509_crl(crl_path.read_bytes())
        private = serialization.load_pem_private_key(key_path.read_bytes(), password=None)
        csr = x509.load_pem_x509_csr(csr_path.read_bytes())
    except (ValueError, TypeError, OSError) as exc:
        raise AdminError("MATERIAL_INVALID") from exc
    public = certificate.public_key().public_bytes(
        serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
    )
    if public != private.public_key().public_bytes(
        serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
    ) or public != csr.public_key().public_bytes(
        serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
    ) or not csr.is_signature_valid:
        raise AdminError("MATERIAL_MISMATCH")
    now = datetime.now(UTC)
    if certificate.not_valid_before_utc > now or certificate.not_valid_after_utc <= now:
        raise AdminError("MATERIAL_INVALID")
    try:
        usages = certificate.extensions.get_extension_for_class(
            x509.ExtendedKeyUsage
        ).value
    except x509.ExtensionNotFound as exc:
        raise AdminError("MATERIAL_INVALID") from exc
    if ExtendedKeyUsageOID.CLIENT_AUTH not in usages:
        raise AdminError("MATERIAL_INVALID")
    issuer = next(
        (authority for authority in authorities if authority.subject == certificate.issuer),
        None,
    )
    if issuer is None:
        raise AdminError("MATERIAL_INVALID")
    try:
        basic = issuer.extensions.get_extension_for_class(x509.BasicConstraints).value
    except x509.ExtensionNotFound as exc:
        raise AdminError("MATERIAL_INVALID") from exc
    if not basic.ca or issuer.not_valid_after_utc <= now:
        raise AdminError("MATERIAL_INVALID")
    _verify_x509_signature(certificate, issuer.public_key())
    if crl.issuer != issuer.subject or crl.last_update_utc > now or crl.next_update_utc <= now:
        raise AdminError("MATERIAL_INVALID")
    _verify_x509_signature(crl, issuer.public_key())
    if crl.get_revoked_certificate_by_serial_number(certificate.serial_number) is not None:
        raise AdminError("MATERIAL_INVALID")
    return certificate, csr, ca_path, crl, crl_path


def command_register_manifest(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    certificate, csr, ca_path, crl, crl_path = _load_material(
        root, args.certificate, args.ca_bundle, args.crl
    )
    for value in (args.gateway_id, args.environment, args.network_zone):
        if not IDENTIFIER_RE.fullmatch(value):
            raise AdminError("INVALID_INPUT")
    if args.environment not in {"development", "staging", "production"}:
        raise AdminError("INVALID_INPUT")
    if not SHA256_RE.fullmatch(args.policy_digest):
        raise AdminError("INVALID_INPUT")
    if not ORIGIN_RE.fullmatch(args.control_origin):
        raise AdminError("INVALID_INPUT")
    if args.proxy_origin and not ORIGIN_RE.fullmatch(args.proxy_origin):
        raise AdminError("INVALID_INPUT")
    output = root / "registration" / "registration.json"
    if output.is_symlink():
        raise AdminError("PATH_POLICY")
    certificate_digest = certificate.fingerprint(hashes.SHA256()).hex()
    public_digest = hashlib.sha256(
        certificate.public_key().public_bytes(
            serialization.Encoding.DER,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        )
    ).hexdigest()
    material = {
        "allowed_control_hosts": [args.control_origin.removeprefix("https://").split(":", 1)[0]],
        "allowed_proxy_hosts": [] if not args.proxy_origin else [args.proxy_origin.removeprefix("https://").split(":", 1)[0]],
        "certificate_sha256": certificate_digest,
        "certificate_not_after": certificate.not_valid_after_utc.isoformat().replace("+00:00", "Z"),
        "certificate_not_before": certificate.not_valid_before_utc.isoformat().replace("+00:00", "Z"),
        "control_origin": args.control_origin,
        "csr_sha256": hashlib.sha256(
            csr.public_bytes(serialization.Encoding.DER)
        ).hexdigest(),
        "public_key_sha256": public_digest,
        "environment": args.environment,
        "gateway_id": args.gateway_id,
        "network_zone": args.network_zone,
        "policy_digest": args.policy_digest,
        "proxy_origin": args.proxy_origin,
        "ca_bundle_sha256": _sha256_file(ca_path),
        "ca_chain_verified": True,
        "client_auth_eku_verified": True,
        "crl_sha256": _sha256_file(crl_path),
        "crl_verified": True,
        "crl_this_update": crl.last_update_utc.isoformat().replace("+00:00", "Z"),
        "crl_next_update": crl.next_update_utc.isoformat().replace("+00:00", "Z"),
        "schema_version": 1,
    }
    replayed = False
    if output.exists():
        if _read_json(output) != material:
            raise AdminError("IDEMPOTENCY_CONFLICT", 3)
        replayed = True
    else:
        _atomic_write(output, _json_bytes(material) + b"\n", 0o600)
    return {
        "certificate_sha256": certificate_digest,
        "manifest_digest": _digest(material),
        "manifest_path": str(output),
        "replayed": replayed,
        "status": "registration_material_ready",
    }


def command_rotate(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    if not IDENTIFIER_RE.fullmatch(args.request_id) or not SHA256_RE.fullmatch(
        args.expected_current_certificate_digest
    ):
        raise AdminError("INVALID_INPUT")
    request = {
        "common_name": args.common_name,
        "expected_current_certificate_digest": args.expected_current_certificate_digest,
        "request_id": args.request_id,
    }
    request_digest = _digest(request)
    record_path = root / "state" / f"rotation-{args.request_id}.json"
    if record_path.exists():
        record = _read_json(record_path)
        if record.get("request_digest") != request_digest:
            raise AdminError("IDEMPOTENCY_CONFLICT", 3)
        return record
    pending_records = [
        path
        for path in (root / "state").glob("rotation-*.json")
        if path.is_file()
        and not path.is_symlink()
        and _read_json(path).get("status") == "pending_server_approval"
    ]
    if len(pending_records) >= MAX_BACKUPS:
        raise AdminError("ROTATION_LIMIT", 3)
    prefix = f"pending-{request_digest[:16]}"
    result = _new_csr(root, args.common_name, prefix)
    record = {
        **result,
        "expected_current_certificate_digest": args.expected_current_certificate_digest,
        "request_digest": request_digest,
        "request_id": args.request_id,
        "status": "pending_server_approval",
    }
    _atomic_write(record_path, _json_bytes(record) + b"\n", 0o600)
    return record


def _trusted_keys(bindings: list[str]) -> dict[str, bytes]:
    keys: dict[str, bytes] = {}
    for binding in bindings:
        key_id, separator, key_hex = binding.partition("=")
        if (
            not separator
            or not IDENTIFIER_RE.fullmatch(key_id)
            or not re.fullmatch(r"[0-9a-f]{64}", key_hex)
        ):
            raise AdminError("INVALID_INPUT")
        keys[key_id] = bytes.fromhex(key_hex)
    return keys


STATUS_FIELDS = frozenset(
    {
        "certificate_sha256", "expires_at", "gateway_id", "generation",
        "issued_at", "key_id", "signature_algorithm", "status",
        "envelope_digest", "signature",
    }
)


def command_revoke_status(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    local_marker = root / "state" / "REVOKED_STOP"
    status = "revoked" if local_marker.exists() else "not_revoked_locally"
    evidence_digest = None
    if args.server_status:
        evidence_path = _inside(root, args.server_status, must_exist=True)
        evidence = _read_json(evidence_path)
        if set(evidence) != STATUS_FIELDS:
            raise AdminError("STATUS_REJECTED", 3)
        unsigned = {
            key: evidence[key]
            for key in sorted(STATUS_FIELDS - {"envelope_digest", "signature"})
        }
        try:
            issued = datetime.fromisoformat(str(evidence["issued_at"]).replace("Z", "+00:00"))
            expires = datetime.fromisoformat(str(evidence["expires_at"]).replace("Z", "+00:00"))
            generation = int(evidence["generation"])
        except (TypeError, ValueError) as exc:
            raise AdminError("STATUS_REJECTED", 3) from exc
        now = datetime.now(UTC)
        if (
            not str(evidence["issued_at"]).endswith("Z")
            or not str(evidence["expires_at"]).endswith("Z")
            or issued.tzinfo is None
            or expires.tzinfo is None
            or issued.astimezone(UTC) > now
            or expires.astimezone(UTC) <= now
            or evidence["signature_algorithm"] != "Ed25519"
            or evidence["status"] not in {"active", "revoked"}
            or evidence["gateway_id"] != args.expected_gateway_id
            or generation != args.expected_generation
            or evidence["certificate_sha256"] != args.expected_certificate_digest
            or _protocol_digest(unsigned) != evidence["envelope_digest"]
        ):
            raise AdminError("STATUS_REJECTED", 3)
        public = _trusted_keys(args.trusted_key).get(str(evidence["key_id"]))
        if public is None:
            raise AdminError("STATUS_REJECTED", 3)
        try:
            ed25519.Ed25519PublicKey.from_public_bytes(public).verify(
                bytes.fromhex(str(evidence["signature"])), _protocol_bytes(unsigned)
            )
        except (InvalidSignature, ValueError) as exc:
            raise AdminError("STATUS_REJECTED", 3) from exc
        if evidence["status"] == "revoked":
            _atomic_write(local_marker, _json_bytes(evidence) + b"\n", 0o600)
            status = "revoked"
        evidence_digest = evidence["envelope_digest"]
    return {
        "evidence_digest": evidence_digest,
        "local_stop_required": status == "revoked",
        "status": status,
    }


def _version_tuple(value: str) -> tuple[int, int, int, int]:
    if not VERSION_RE.fullmatch(value):
        raise AdminError("RELEASE_REJECTED", 3)
    values = tuple(int(item) for item in value.split("."))
    if any(item > 2_147_483_647 for item in values):
        raise AdminError("RELEASE_REJECTED", 3)
    return (*values, 0)  # type: ignore[return-value]


def _verify_release_manifest(
    manifest: dict[str, object], artifact_name_actual: str,
    artifact_digest_actual: str, trusted_keys: dict[str, bytes],
    *, current_version: str | None = None,
) -> dict[str, object]:
    if set(manifest) != MANIFEST_FIELDS:
        raise AdminError("RELEASE_REJECTED", 3)
    if not IDENTIFIER_RE.fullmatch(str(manifest["release_id"])):
        raise AdminError("RELEASE_REJECTED", 3)
    artifact_name = manifest["artifact_name"]
    if (
        not isinstance(artifact_name, str)
        or PurePosixPath(artifact_name).name != artifact_name
        or artifact_name != artifact_name_actual
        or artifact_name in {".", ".."}
    ):
        raise AdminError("RELEASE_REJECTED", 3)
    if (
        manifest["signature_algorithm"] != "Ed25519"
        or manifest["status"] not in {
            "offered", "accepted", "installed", "rolled_back"
        }
    ):
        raise AdminError("RELEASE_REJECTED", 3)
    version_tuple = _version_tuple(str(manifest["version"]))
    rollback_tuple = _version_tuple(str(manifest["rollback_version"]))
    if version_tuple <= rollback_tuple:
        raise AdminError("RELEASE_REJECTED", 3)
    if current_version is not None and (
        manifest["rollback_version"] != current_version
        or version_tuple <= _version_tuple(current_version)
    ):
        raise AdminError("RELEASE_REJECTED", 3)
    try:
        deadline = datetime.fromisoformat(str(manifest["deadline_at"]).replace("Z", "+00:00"))
    except ValueError as exc:
        raise AdminError("RELEASE_REJECTED", 3) from exc
    if deadline.tzinfo is None or deadline.astimezone(UTC) <= datetime.now(UTC):
        raise AdminError("RELEASE_REJECTED", 3)
    if not str(manifest["deadline_at"]).endswith("Z"):
        raise AdminError("RELEASE_REJECTED", 3)
    unsigned = {
        key: manifest[key]
        for key in sorted(MANIFEST_FIELDS - {"manifest_digest", "signature"})
    }
    if not SHA256_RE.fullmatch(str(manifest["manifest_digest"])) or _protocol_digest(unsigned) != manifest["manifest_digest"]:
        raise AdminError("RELEASE_REJECTED", 3)
    if not SHA256_RE.fullmatch(str(manifest["artifact_digest"])):
        raise AdminError("RELEASE_REJECTED", 3)
    if artifact_digest_actual != manifest["artifact_digest"]:
        raise AdminError("RELEASE_REJECTED", 3)
    public = trusted_keys.get(str(manifest["key_id"]))
    if public is None:
        raise AdminError("RELEASE_REJECTED", 3)
    try:
        ed25519.Ed25519PublicKey.from_public_bytes(public).verify(
            bytes.fromhex(str(manifest["signature"])), _protocol_bytes(unsigned)
        )
    except (ValueError, InvalidSignature) as exc:
        raise AdminError("RELEASE_REJECTED", 3) from exc
    return {
        "artifact_digest": artifact_digest_actual,
        "content_digest": manifest["manifest_digest"],
        "manifest": manifest,
    }


def _bundle_record(manifest: dict[str, object]) -> dict[str, object]:
    unsigned = {
        "artifact_digest": manifest["artifact_digest"],
        "artifact_name": manifest["artifact_name"],
        "manifest_digest": manifest["manifest_digest"],
        "schema_version": 1,
    }
    return {**unsigned, "bundle_digest": _protocol_digest(unsigned)}


def _verified_bundle(root: Path, manifest_digest: str) -> Path:
    if not SHA256_RE.fullmatch(manifest_digest):
        raise AdminError("RELEASE_REJECTED", 3)
    verified = root / "releases" / "verified"
    if verified.is_symlink() or not verified.is_dir():
        raise AdminError("PATH_POLICY")
    bundle = verified / manifest_digest
    if bundle.is_symlink() or not bundle.is_dir():
        raise AdminError("RELEASE_REJECTED", 3)
    return bundle


def _verify_bundle(
    root: Path, manifest_digest: str, trusted_keys: dict[str, bytes]
) -> tuple[dict[str, object], dict[str, object], Path]:
    bundle = _verified_bundle(root, manifest_digest)
    manifest_path = _inside(root, bundle / "manifest.json", must_exist=True)
    manifest = _read_json(manifest_path)
    if manifest.get("manifest_digest") != manifest_digest:
        raise AdminError("RELEASE_REJECTED", 3)
    artifact_name = manifest.get("artifact_name")
    if not isinstance(artifact_name, str):
        raise AdminError("RELEASE_REJECTED", 3)
    artifact_path = _inside(root, bundle / artifact_name, must_exist=True)
    artifact_descriptor, artifact_metadata = _open_root_file(root, artifact_path)
    try:
        digest = hashlib.sha256()
        for chunk in iter(lambda: os.read(artifact_descriptor, 1_048_576), b""):
            digest.update(chunk)
        if os.fstat(artifact_descriptor) != artifact_metadata:
            raise AdminError("RELEASE_REJECTED", 3)
    finally:
        os.close(artifact_descriptor)
    _verify_release_manifest(manifest, artifact_name, digest.hexdigest(), trusted_keys)
    bundle_path = _inside(root, bundle / "bundle.json", must_exist=True)
    record = _read_json(bundle_path)
    expected = _bundle_record(manifest)
    if record != expected or set(bundle.iterdir()) != {
        manifest_path, artifact_path, bundle_path
    }:
        raise AdminError("RELEASE_REJECTED", 3)
    return manifest, record, artifact_path


def command_release_check(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    manifest_path = _inside(root, args.manifest)
    artifact_path = _inside(root, args.artifact)
    manifest_descriptor = -1
    artifact_descriptor = -1
    temporary: Path | None = None
    try:
        manifest_descriptor, manifest_before = _open_root_file(root, manifest_path)
        artifact_descriptor, artifact_before = _open_root_file(root, artifact_path)
        manifest = _read_json_descriptor(manifest_descriptor)
        if os.fstat(manifest_descriptor) != manifest_before:
            raise AdminError("RELEASE_REJECTED", 3)
        artifact_name = manifest.get("artifact_name")
        if (
            not isinstance(artifact_name, str)
            or not artifact_name
            or artifact_name in {".", ".."}
            or len(artifact_name.encode("utf-8")) > 255
            or PurePosixPath(artifact_name).name != artifact_name
            or Path(artifact_name).name != artifact_name
            or artifact_path.name != artifact_name
        ):
            raise AdminError("RELEASE_REJECTED", 3)
        verified_dir = _mkdir(root / "releases", "verified")
        temporary = Path(tempfile.mkdtemp(prefix=".bundle-", dir=verified_dir))
        os.chmod(temporary, 0o700)
        parent_owner = verified_dir.stat()
        if os.geteuid() == 0:
            os.chown(
                temporary, parent_owner.st_uid, parent_owner.st_gid,
                follow_symlinks=False,
            )
        elif (temporary.stat().st_uid, temporary.stat().st_gid) != (
            parent_owner.st_uid, parent_owner.st_gid
        ):
            raise AdminError("OWNER_POLICY")
        copied_artifact = temporary / artifact_name
        copied_digest = _copy_descriptor(
            artifact_descriptor, artifact_before, copied_artifact
        )
        keys = _trusted_keys(args.trusted_key)
        verified = _verify_release_manifest(
            manifest, artifact_path.name, copied_digest, keys,
            current_version=args.current_version,
        )
        replay_path = root / "state" / "release-replay.json"
        replay = _read_json(replay_path) if replay_path.exists() else {}
        content_digest = str(verified["content_digest"])
        prior = replay.get(str(manifest["release_id"]))
        if prior is not None and prior != content_digest:
            raise AdminError("RELEASE_REJECTED", 3)
        final_bundle = verified_dir / str(manifest["manifest_digest"])
        replayed = False
        if final_bundle.exists() or final_bundle.is_symlink():
            _verify_bundle(root, str(manifest["manifest_digest"]), keys)
            copied_artifact.unlink()
            temporary.rmdir()
            replayed = True
        else:
            _atomic_write(
                temporary / "manifest.json", _json_bytes(manifest) + b"\n", 0o600
            )
            _atomic_write(
                temporary / "bundle.json",
                _json_bytes(_bundle_record(manifest)) + b"\n", 0o600,
            )
            _fsync_directory(temporary)
            os.replace(temporary, final_bundle)
            _fsync_directory(verified_dir)
        replay[str(manifest["release_id"])] = content_digest
        _atomic_write(replay_path, _json_bytes(replay) + b"\n", 0o600)
        return {
            "artifact_digest": manifest["artifact_digest"],
            "manifest_digest": manifest["manifest_digest"],
            "replayed": replayed,
            "release_id": manifest["release_id"],
            "status": "verified",
            "verified_bundle_path": str(final_bundle),
            "version": manifest["version"],
        }
    finally:
        if manifest_descriptor >= 0:
            os.close(manifest_descriptor)
        if artifact_descriptor >= 0:
            os.close(artifact_descriptor)
        if temporary is not None and temporary.exists():
            if temporary.is_symlink() or not temporary.is_dir():
                raise AdminError("PATH_POLICY")
            for item in temporary.iterdir():
                if item.is_symlink() or not item.is_file():
                    raise AdminError("PATH_POLICY")
                item.unlink()
            temporary.rmdir()
            _fsync_directory(temporary.parent)


def command_rollback(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    if not args.yes and (
        not sys.stdin.isatty()
        or input("Type ROLLBACK to continue: ").strip() != "ROLLBACK"
    ):
        raise AdminError("CONFIRMATION_REQUIRED", 4)
    if not SHA256_RE.fullmatch(args.expected_current_digest) or not SHA256_RE.fullmatch(args.target_digest):
        raise AdminError("INVALID_INPUT")
    keys = _trusted_keys(args.trusted_key)
    current_path = _inside(root, root / "releases" / "current.json", must_exist=True)
    current_pointer = _read_json(current_path)
    history_dir = _mkdir(root / "state", "rollback-history")
    request_digest = _protocol_digest(
        {
            "expected_current_digest": args.expected_current_digest,
            "target_digest": args.target_digest,
        }
    )
    history_path = history_dir / f"{request_digest}.json"
    intent_path = root / "state" / "rollback-intent.json"
    current, current_bundle, _ = _verify_bundle(
        root, args.expected_current_digest, keys
    )
    target, target_bundle, _ = _verify_bundle(root, args.target_digest, keys)
    expected_pointer = {
        "artifact_name": current["artifact_name"],
        "bundle_digest": current_bundle["bundle_digest"],
        "manifest_digest": current["manifest_digest"],
        "schema_version": 1,
        "version": current["version"],
    }
    pointer = {
        "artifact_name": target["artifact_name"],
        "bundle_digest": target_bundle["bundle_digest"],
        "manifest_digest": target["manifest_digest"],
        "schema_version": 1,
        "version": target["version"],
    }
    if (
        not isinstance(target.get("version"), str)
        or not isinstance(current.get("version"), str)
        or _version_tuple(target["version"]) >= _version_tuple(current["version"])
        or current.get("rollback_version") != target.get("version")
    ):
        raise AdminError("RELEASE_REJECTED", 3)
    if history_path.exists():
        history = _read_json(history_path)
        if current_pointer != pointer:
            raise AdminError("STATE_CONFLICT", 3)
        if intent_path.exists():
            expected_intent = {
                "current_pointer": expected_pointer,
                "request_digest": request_digest,
                "target_pointer": pointer,
            }
            if _read_json(intent_path) != expected_intent:
                raise AdminError("STATE_CONFLICT", 3)
            intent_path.unlink()
            _fsync_directory(intent_path.parent)
        return {**history, "replayed": True}
    if intent_path.exists():
        intent = _read_json(intent_path)
        if (
            intent.get("request_digest") != request_digest
            or intent.get("current_pointer") != expected_pointer
            or intent.get("target_pointer") != pointer
        ):
            raise AdminError("STATE_CONFLICT", 3)
        if current_pointer == pointer:
            recovered = {
                "current_manifest_digest": args.target_digest,
                "previous_manifest_digest": args.expected_current_digest,
                "replayed": True,
                "status": "rolled_back",
                "version": target.get("version"),
            }
            _atomic_write(history_path, _json_bytes(recovered) + b"\n", 0o600)
            intent_path.unlink()
            _fsync_directory(intent_path.parent)
            return recovered
    if current_pointer != expected_pointer:
        raise AdminError("STATE_CONFLICT", 3)
    backup_dir = _mkdir(root / "releases", "backups")
    backup_path = backup_dir / f"current-{args.expected_current_digest}.json"
    if not backup_path.exists():
        _atomic_write(backup_path, _json_bytes(current_pointer) + b"\n", 0o600)
    _bounded_backups(backup_dir, "current-")
    intent = {
        "current_pointer": expected_pointer,
        "request_digest": request_digest,
        "target_pointer": pointer,
    }
    if intent_path.exists() and _read_json(intent_path) != intent:
        raise AdminError("STATE_CONFLICT", 3)
    if not intent_path.exists():
        _atomic_write(intent_path, _json_bytes(intent) + b"\n", 0o600)
    if _read_json(current_path) != current_pointer:
        raise AdminError("STATE_CONFLICT", 3)
    _atomic_write(current_path, _json_bytes(pointer) + b"\n", 0o600)
    result = {
        "current_manifest_digest": args.target_digest,
        "previous_manifest_digest": args.expected_current_digest,
        "replayed": False,
        "status": "rolled_back",
        "version": target.get("version"),
    }
    _atomic_write(history_path, _json_bytes(result) + b"\n", 0o600)
    intent_path.unlink()
    _fsync_directory(intent_path.parent)
    return result


def command_status(args: argparse.Namespace) -> dict[str, object]:
    root = _validated_root(args.root)
    queue_files = [
        path for path in (root / "queue").glob("*.sqlite3")
        if path.is_file() and not path.is_symlink()
    ]
    if len(queue_files) > 1:
        raise AdminError("STATE_CONFLICT", 3)
    queue_status: dict[str, object] = {
        "artifact_cleanup": {}, "events": {}, "outcomes": {}, "releases": {},
        "schema_version": None, "tasks": {},
    }
    active_leases = 0
    if queue_files:
        queue_path = _inside(root, queue_files[0], must_exist=True)
        connection = sqlite3.connect(f"file:{queue_path}?mode=ro&immutable=1", uri=True)
        try:
            queue_status["schema_version"] = int(
                connection.execute("PRAGMA user_version").fetchone()[0]
            )
            table_specs = {
                "tasks": ("edge_tasks", "status"),
                "events": ("edge_outbound_events", "status"),
                "outcomes": ("edge_task_outcomes", "status"),
                "artifact_cleanup": ("edge_local_artifacts", "cleanup_status"),
                "releases": ("edge_release_state", "status"),
            }
            for label, (table, column) in table_specs.items():
                rows = connection.execute(
                    f"SELECT {column},COUNT(*) FROM {table} GROUP BY {column}"  # noqa: S608
                ).fetchall()
                queue_status[label] = {str(status): int(count) for status, count in rows}
            active_leases = sum(
                int(connection.execute(query).fetchone()[0])
                for query in (
                    "SELECT COUNT(*) FROM edge_tasks WHERE status='leased'",
                    "SELECT COUNT(*) FROM edge_outbound_events WHERE status='sending'",
                    "SELECT COUNT(*) FROM edge_task_outcomes WHERE status='sending'",
                    "SELECT COUNT(*) FROM edge_local_artifacts WHERE cleanup_status='deleting'",
                )
            )
        except sqlite3.DatabaseError as exc:
            raise AdminError("QUEUE_STATUS_FAILED", 3) from exc
        finally:
            connection.close()
    current = None
    current_path = root / "releases" / "current.json"
    if current_path.exists():
        pointer = _read_json(current_path)
        current = {
            "manifest_digest": pointer.get("manifest_digest"),
            "version": pointer.get("version"),
        }
    return {
        "active_leases": active_leases,
        "current_release": current,
        "cursor_state": "reconciled_per_cycle_not_persisted",
        "queue": queue_status,
        "registration_present": (root / "registration" / "registration.json").is_file(),
        "rollback_intent_present": (root / "state" / "rollback-intent.json").is_file(),
        "status": "read_only",
    }


def parser() -> argparse.ArgumentParser:
    main = argparse.ArgumentParser(description=__doc__)
    subcommands = main.add_subparsers(dest="command", required=True)
    init = subcommands.add_parser("init")
    init.add_argument("--root", required=True)
    init.add_argument("--runtime-uid", required=True, type=int)
    init.add_argument("--runtime-gid", required=True, type=int)
    init.add_argument("--allow-chown", action="store_true")
    init.set_defaults(handler=command_init)
    csr = subcommands.add_parser("csr")
    csr.add_argument("--root", required=True)
    csr.add_argument("--common-name", required=True)
    csr.set_defaults(handler=command_csr)
    registration = subcommands.add_parser("register-manifest")
    registration.add_argument("--root", required=True)
    registration.add_argument("--certificate", required=True)
    registration.add_argument("--ca-bundle", required=True)
    registration.add_argument("--crl", required=True)
    registration.add_argument("--gateway-id", required=True)
    registration.add_argument("--environment", required=True)
    registration.add_argument("--network-zone", required=True)
    registration.add_argument("--policy-digest", required=True)
    registration.add_argument("--control-origin", required=True)
    registration.add_argument("--proxy-origin")
    registration.set_defaults(handler=command_register_manifest)
    rotate = subcommands.add_parser("rotate")
    rotate.add_argument("--root", required=True)
    rotate.add_argument("--common-name", required=True)
    rotate.add_argument("--request-id", required=True)
    rotate.add_argument("--expected-current-certificate-digest", required=True)
    rotate.set_defaults(handler=command_rotate)
    revoke = subcommands.add_parser("revoke-status")
    revoke.add_argument("--root", required=True)
    revoke.add_argument("--server-status")
    revoke.add_argument("--trusted-key", action="append", default=[])
    revoke.add_argument("--expected-gateway-id")
    revoke.add_argument("--expected-generation", type=int)
    revoke.add_argument("--expected-certificate-digest")
    revoke.set_defaults(handler=command_revoke_status)
    check = subcommands.add_parser("release-check")
    check.add_argument("--root", required=True)
    check.add_argument("--manifest", required=True)
    check.add_argument("--artifact", required=True)
    check.add_argument("--trusted-key", action="append", required=True)
    check.add_argument("--current-version", required=True)
    check.set_defaults(handler=command_release_check)
    rollback = subcommands.add_parser("rollback")
    rollback.add_argument("--root", required=True)
    rollback.add_argument("--expected-current-digest", required=True)
    rollback.add_argument("--target-digest", required=True)
    rollback.add_argument("--trusted-key", action="append", required=True)
    rollback.add_argument("--yes", action="store_true")
    rollback.set_defaults(handler=command_rollback)
    status_command = subcommands.add_parser("status")
    status_command.add_argument("--root", required=True)
    status_command.set_defaults(handler=command_status)
    return main


def main() -> int:
    try:
        args = parser().parse_args()
        if args.command in {"init", "status"}:
            result = args.handler(args)
        else:
            root = _validated_root(args.root)
            with _process_lock(root):
                result = args.handler(args)
        _output(result)
        return 0
    except AdminError as exc:
        _output({"error": {"code": exc.code}, "status": "error"})
        return exc.exit_code
    except (OSError, ValueError, TypeError, MemoryError, RecursionError):
        _output({"error": {"code": "OPERATION_FAILED"}, "status": "error"})
        return 5


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