DataOps-platform/app/edge_gateway/bootstrap.py

"""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