"""Fail-closed bootstrap configuration for an enterprise edge agent.""" from __future__ import annotations import re import ssl import stat from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from types import MappingProxyType from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, ed448, ed25519, padding, rsa from app.core.edge_gateway.policy import EdgeEgressPolicy _SHA256 = re.compile(r"^[0-9a-f]{64}$") _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$") _VERSION = re.compile(r"^(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})$") _PLACEHOLDERS = {"change-me", "changeme", "example", "placeholder", "tbd", "todo"} _REQUIRED = frozenset( { "queue_path", "artifact_root", "gateway_id", "credential", "certificate_sha256", "client_certificate_path", "client_private_key_path", "ca_bundle_path", "generation", "environment", "network_zone", "policy_digest", "control_url", "proxy_url", "allowed_control_hosts", "allowed_proxy_hosts", "trusted_release_keys", "trusted_task_keys", "task_authority_clock_skew_seconds", "version", } ) _OPTIONAL = frozenset({"server_crl_path", "allowed_control_origins", "allowed_proxy_origins"}) def _text(value: object, label: str, maximum: int = 255) -> str: if ( not isinstance(value, str) or not value or value.strip() != value or "\x00" in value or len(value.encode("utf-8")) > maximum or value.casefold() in _PLACEHOLDERS ): raise ValueError(f"{label} is not explicitly configured") return value def _identifier(value: object, label: str) -> str: candidate = _text(value, label) if not _IDENTIFIER.fullmatch(candidate): raise ValueError(f"{label} is invalid") return candidate def _hosts(value: object, label: str, *, required: bool) -> frozenset[str]: if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): raise ValueError(f"{label} must be an explicit sequence") items = [_text(item, label) for item in value] if any(item != item.lower() for item in items): raise ValueError(f"{label} must contain lowercase exact hosts") result = frozenset(items) if required and not result: raise ValueError(f"{label} is required") if len(result) != len(value): raise ValueError(f"{label} contains duplicates") return result @dataclass(frozen=True, slots=True) class EdgeBootstrapConfig: queue_path: str artifact_root: str gateway_id: str credential: str = field(repr=False) certificate_sha256: str client_certificate_path: str client_private_key_path: str = field(repr=False) ca_bundle_path: str generation: int environment: str network_zone: str policy_digest: str control_url: str proxy_url: str | None allowed_control_hosts: frozenset[str] allowed_proxy_hosts: frozenset[str] trusted_release_keys: Mapping[str, str] = field(repr=False) trusted_task_keys: Mapping[str, str] = field(repr=False) version: str task_authority_clock_skew_seconds: int = 0 server_crl_path: str | None = None allowed_control_origins: frozenset[str] = frozenset() allowed_proxy_origins: frozenset[str] = frozenset() @classmethod def from_mapping(cls, value: Mapping[str, object]) -> EdgeBootstrapConfig: if ( not isinstance(value, Mapping) or not set(value) >= _REQUIRED or not set(value) <= _REQUIRED | _OPTIONAL ): raise ValueError("edge bootstrap requires the exact configuration fields") queue = _text(value["queue_path"], "queue_path", 2_048) if queue == ":memory:" or "mode=memory" in queue: raise ValueError("edge queue must use an explicit disk path") queue_path = Path(queue).expanduser() if not queue_path.is_absolute(): raise ValueError("edge queue path must be absolute") artifact_root = Path( _text(value["artifact_root"], "artifact_root", 2_048) ).expanduser() if ( not artifact_root.is_absolute() or not artifact_root.is_dir() or artifact_root.is_symlink() ): raise ValueError("artifact_root must be an existing absolute directory") credential = _text(value["credential"], "credential", 160) if ( not credential.startswith("dopg_") or len(credential) < 16 or any(marker in credential.casefold() for marker in _PLACEHOLDERS) ): raise ValueError("credential is invalid") certificate = _text(value["certificate_sha256"], "certificate_sha256", 64) policy_digest = _text(value["policy_digest"], "policy_digest", 64) if not _SHA256.fullmatch(certificate) or not _SHA256.fullmatch(policy_digest): raise ValueError("edge digest binding is invalid") material: dict[str, Path] = {} for field_name in ( "client_certificate_path", "client_private_key_path", "ca_bundle_path", ): candidate = Path(_text(value[field_name], field_name, 2_048)).expanduser() if not candidate.is_absolute() or not candidate.is_file(): raise ValueError(f"{field_name} must be an existing absolute file") material[field_name] = candidate.resolve() key_mode = stat.S_IMODE(material["client_private_key_path"].stat().st_mode) if key_mode & 0o077: raise ValueError("client private key permissions are not secure") try: certificate_object = x509.load_pem_x509_certificate( material["client_certificate_path"].read_bytes() ) ca_object = x509.load_pem_x509_certificate( material["ca_bundle_path"].read_bytes() ) private_key = serialization.load_pem_private_key( material["client_private_key_path"].read_bytes(), password=None ) except (ValueError, OSError) as exc: raise ValueError("mTLS material is invalid") from exc actual_fingerprint = certificate_object.fingerprint(hashes.SHA256()).hex() if actual_fingerprint != certificate: raise ValueError("client certificate fingerprint does not match configuration") public_encoding = { "encoding": serialization.Encoding.DER, "format": serialization.PublicFormat.SubjectPublicKeyInfo, } if certificate_object.public_key().public_bytes(**public_encoding) != private_key.public_key().public_bytes(**public_encoding): raise ValueError("client private key does not match certificate") now = datetime.now(UTC) if certificate_object.not_valid_before_utc > now or certificate_object.not_valid_after_utc <= now: raise ValueError("client certificate is not currently valid") if ca_object.not_valid_after_utc <= now: raise ValueError("CA bundle certificate is expired") generation = value["generation"] if isinstance(generation, bool) or not isinstance(generation, int) or generation < 1: raise ValueError("generation is invalid") control_hosts = _hosts(value["allowed_control_hosts"], "allowed_control_hosts", required=True) proxy_hosts = _hosts(value["allowed_proxy_hosts"], "allowed_proxy_hosts", required=False) proxy_url = value["proxy_url"] if proxy_url is not None: proxy_url = _text(proxy_url, "proxy_url", 2_048) control_url = _text(value["control_url"], "control_url", 2_048) def origins(field_name: str) -> frozenset[str]: raw = value.get(field_name, []) if not isinstance(raw, Sequence) or isinstance(raw, (str, bytes)): raise ValueError(f"{field_name} must be an explicit sequence") result = frozenset(_text(item, field_name, 2_048) for item in raw) if len(result) != len(raw) or any(not item.startswith("https://") for item in result): raise ValueError(f"{field_name} is invalid") return result control_origins = origins("allowed_control_origins") proxy_origins = origins("allowed_proxy_origins") EdgeEgressPolicy( allowed_control_hosts=set(control_hosts), allowed_proxy_hosts=set(proxy_hosts), allowed_control_origins=set(control_origins) or None, allowed_proxy_origins=set(proxy_origins) or None, ).validate_destination(control_url, proxy_url=proxy_url) def trusted_keys(field_name: str) -> Mapping[str, str]: raw_keys = value[field_name] if not isinstance(raw_keys, Mapping) or not raw_keys: raise ValueError(f"{field_name} is required") keys: dict[str, str] = {} for raw_key_id, raw_public_key in raw_keys.items(): key_id = _identifier(raw_key_id, "signing key id") public_key = _text(raw_public_key, "signing public key", 128) try: decoded = bytes.fromhex(public_key) except ValueError as exc: raise ValueError("signing public key must be lowercase hex") from exc if ( public_key != public_key.lower() or len(decoded) != 32 or decoded == bytes(32) ): raise ValueError("signing public key must be a 32-byte Ed25519 key") keys[key_id] = public_key return MappingProxyType(keys) release_keys = trusted_keys("trusted_release_keys") task_keys = trusted_keys("trusted_task_keys") task_authority_clock_skew_seconds = value[ "task_authority_clock_skew_seconds" ] if ( isinstance(task_authority_clock_skew_seconds, bool) or not isinstance(task_authority_clock_skew_seconds, int) or not 0 <= task_authority_clock_skew_seconds <= 300 ): raise ValueError("task authority clock skew is invalid") version = _text(value["version"], "version", 80) if ( not _VERSION.fullmatch(version) or any(int(part) > 2_147_483_647 for part in version.split(".")) ): raise ValueError("version is invalid") server_crl_path: str | None = None if "server_crl_path" in value: candidate = Path( _text(value["server_crl_path"], "server_crl_path", 2_048) ).expanduser() if not candidate.is_absolute() or not candidate.is_file(): raise ValueError("server_crl_path must be an existing absolute file") server_crl_path = str(candidate.resolve()) validate_server_crl(str(material["ca_bundle_path"]), server_crl_path) return cls( queue_path=str(queue_path.resolve()), artifact_root=str(artifact_root.resolve()), gateway_id=_identifier(value["gateway_id"], "gateway_id"), credential=credential, certificate_sha256=certificate, client_certificate_path=str(material["client_certificate_path"]), client_private_key_path=str(material["client_private_key_path"]), ca_bundle_path=str(material["ca_bundle_path"]), generation=generation, environment=_identifier(value["environment"], "environment"), network_zone=_identifier(value["network_zone"], "network_zone"), policy_digest=policy_digest, control_url=control_url.rstrip("/"), proxy_url=proxy_url, allowed_control_hosts=control_hosts, allowed_proxy_hosts=proxy_hosts, trusted_release_keys=release_keys, trusted_task_keys=task_keys, version=version, task_authority_clock_skew_seconds=task_authority_clock_skew_seconds, server_crl_path=server_crl_path, allowed_control_origins=control_origins, allowed_proxy_origins=proxy_origins, ) def _verify_crl_signature(crl: x509.CertificateRevocationList, issuer) -> None: public_key = issuer.public_key() try: if isinstance(public_key, rsa.RSAPublicKey): public_key.verify( crl.signature, crl.tbs_certlist_bytes, padding.PKCS1v15(), crl.signature_hash_algorithm, ) elif isinstance(public_key, ec.EllipticCurvePublicKey): public_key.verify( crl.signature, crl.tbs_certlist_bytes, ec.ECDSA(crl.signature_hash_algorithm), ) elif isinstance(public_key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)): public_key.verify(crl.signature, crl.tbs_certlist_bytes) else: raise ValueError("server CRL issuer key is unsupported") except Exception as exc: raise ValueError("server CRL signature is invalid") from exc def validate_server_crl(ca_bundle_path: str, server_crl_path: str) -> datetime: try: authorities = x509.load_pem_x509_certificates(Path(ca_bundle_path).read_bytes()) crl = x509.load_pem_x509_crl(Path(server_crl_path).read_bytes()) except (OSError, ValueError) as exc: raise ValueError("server CRL material is invalid") from exc issuer = next((item for item in authorities if item.subject == crl.issuer), None) if issuer is None: raise ValueError("server CRL issuer is not in the CA bundle") _verify_crl_signature(crl, issuer) now = datetime.now(UTC) if ( crl.last_update_utc > now or crl.next_update_utc is None or crl.next_update_utc <= now ): raise ValueError("server CRL is not fresh") return crl.next_update_utc def build_server_crl_ssl_context( ca_bundle_path: str, server_crl_path: str ) -> ssl.SSLContext: validate_server_crl(ca_bundle_path, server_crl_path) context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_bundle_path) context.load_verify_locations(cafile=server_crl_path) context.verify_flags |= ssl.VERIFY_CRL_CHECK_LEAF context.check_hostname = True return context