#!/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 _same_file_snapshot(before: os.stat_result, after: os.stat_result) -> bool: """Compare mutation-relevant descriptor metadata without treating atime as a write.""" fields = ( "st_dev", "st_ino", "st_mode", "st_nlink", "st_uid", "st_gid", "st_size", "st_mtime_ns", "st_ctime_ns", ) return all(getattr(before, field) == getattr(after, field) for field in fields) 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 not _same_file_snapshot(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())