DataOps-platform/app/edge_gateway/runtime.py

"""Runnable pull-only edge process with mTLS server-CRL fail-stop semantics."""

from __future__ import annotations

import hashlib
import json
import os
import signal
import stat
import threading
from datetime import UTC, datetime
from pathlib import Path

import requests
from requests.adapters import HTTPAdapter

from app.core.edge_gateway.policy import EdgePolicyError
from app.edge_gateway.agent import EdgeAgent, EdgeRunnerAdapter
from app.edge_gateway.bootstrap import (
    EdgeBootstrapConfig,
    build_server_crl_ssl_context,
    validate_server_crl,
)
from app.edge_gateway.transport import (
    EdgeAuthenticationStopped,
    EdgeTransport,
    EdgeTransportError,
)

MAX_CONFIG_BYTES = 262_144


def _now() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def _read_regular(path: Path, *, maximum: int, private: bool = False) -> bytes:
    if not path.is_absolute() or path.is_symlink():
        raise ValueError("runtime material path is invalid")
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags)
    try:
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_nlink != 1
            or metadata.st_size > maximum
            or (private and stat.S_IMODE(metadata.st_mode) & 0o077)
        ):
            raise ValueError("runtime material metadata is unsafe")
        chunks = bytearray()
        while len(chunks) <= maximum:
            chunk = os.read(descriptor, min(65_536, maximum + 1 - len(chunks)))
            if not chunk:
                break
            chunks.extend(chunk)
        if len(chunks) > maximum or os.fstat(descriptor) != metadata:
            raise ValueError("runtime material changed while reading")
        return bytes(chunks)
    finally:
        os.close(descriptor)


def _read_json(path: Path) -> dict[str, object]:
    try:
        value = json.loads(_read_regular(path, maximum=MAX_CONFIG_BYTES).decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise ValueError("runtime JSON is invalid") from exc
    if not isinstance(value, dict):
        raise ValueError("runtime JSON must be an object")
    return value


def _atomic_health(path: Path, value: dict[str, object]) -> None:
    encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n"
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    temporary = path.parent / f".{path.name}.{os.getpid()}.tmp"
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(temporary, flags, 0o600)
    try:
        os.write(descriptor, encoded)
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
    os.replace(temporary, path)
    directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(directory)
    finally:
        os.close(directory)


class ServerCrlHttpAdapter(HTTPAdapter):
    def __init__(self, ca_bundle_path: str, server_crl_path: str):
        self._ssl_context = build_server_crl_ssl_context(
            ca_bundle_path, server_crl_path
        )
        super().__init__()

    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
        pool_kwargs["ssl_context"] = self._ssl_context
        return super().init_poolmanager(connections, maxsize, block, **pool_kwargs)


class ServerCrlSession(requests.Session):
    """Reload the TLS pool whenever the operator atomically replaces the CRL."""

    def __init__(self, ca_bundle_path: str, server_crl_path: str):
        super().__init__()
        self.trust_env = False
        self._ca_bundle_path = ca_bundle_path
        self._server_crl_path = server_crl_path
        self._server_crl_digest = ""
        self.refresh_server_crl()

    @property
    def server_crl_digest(self) -> str:
        return self._server_crl_digest

    def refresh_server_crl(self) -> bool:
        validate_server_crl(self._ca_bundle_path, self._server_crl_path)
        digest = hashlib.sha256(
            _read_regular(Path(self._server_crl_path), maximum=1_048_576)
        ).hexdigest()
        if digest == self._server_crl_digest:
            return False
        replacement = ServerCrlHttpAdapter(
            self._ca_bundle_path, self._server_crl_path
        )
        previous = self.adapters.get("https://")
        self.mount("https://", replacement)
        self._server_crl_digest = digest
        if previous is not None:
            previous.close()
        return True

    def request(self, *args, **kwargs):
        self.refresh_server_crl()
        return super().request(*args, **kwargs)


def _runtime_self_check(node, request, cancel_requested):
    if (
        node != "quality.check"
        or request.get("operation") != "quality"
        or request.get("purpose") != "quality-evaluation"
        or request.get("classification") != "evidence"
    ):
        raise EdgePolicyError("enterprise source runner handler is not provisioned")
    if cancel_requested():
        raise EdgePolicyError("runtime self-check was cancelled")
    return {
        "classification": "evidence",
        "payload": {
            "check_count": 1,
            "passed_count": 1,
            "scope": "edge_runtime",
            "status": "passed",
        },
    }


def build_runtime_runner() -> EdgeRunnerAdapter:
    """Provide one closed engineering self-check; source handlers remain external."""

    return EdgeRunnerAdapter(_runtime_self_check)


def build_runtime_from_environment() -> tuple[EdgeAgent, Path, float]:
    config_path = Path(os.environ.get("EDGE_CONFIG_PATH", ""))
    credential_path = Path(os.environ.get("EDGE_CREDENTIAL_FILE", ""))
    task_keys_path = Path(os.environ.get("EDGE_TASK_SIGNING_KEYS_PATH", ""))
    release_keys_path = Path(os.environ.get("EDGE_RELEASE_SIGNING_KEYS_PATH", ""))
    server_crl_path = Path(os.environ.get("EDGE_SERVER_CRL_PATH", ""))
    health_path = Path(os.environ.get("EDGE_HEALTH_PATH", "/run/edge/tmp/health.json"))
    if any(not path.is_absolute() for path in (
        config_path, credential_path, task_keys_path, release_keys_path,
        server_crl_path, health_path,
    )):
        raise ValueError("edge runtime paths must be absolute")
    raw = _read_json(config_path)
    credential = _read_regular(
        credential_path, maximum=160, private=True
    ).decode("utf-8").strip()
    task_keys = _read_json(task_keys_path)
    release_keys = _read_json(release_keys_path)
    raw.update({
        "credential": credential,
        "trusted_task_keys": task_keys,
        "trusted_release_keys": release_keys,
        "server_crl_path": str(server_crl_path),
    })
    config = EdgeBootstrapConfig.from_mapping(raw)
    if config.server_crl_path is None:
        raise ValueError("server CRL is required by the runtime")
    session = ServerCrlSession(config.ca_bundle_path, config.server_crl_path)
    transport = EdgeTransport(
        base_url=config.control_url,
        gateway_id=config.gateway_id,
        environment=config.environment,
        network_zone=config.network_zone,
        generation=config.generation,
        credential=config.credential,
        certificate_sha256=config.certificate_sha256,
        allowed_control_hosts=config.allowed_control_hosts,
        allowed_proxy_hosts=config.allowed_proxy_hosts,
        allowed_control_origins=config.allowed_control_origins or None,
        allowed_proxy_origins=config.allowed_proxy_origins or None,
        proxy_url=config.proxy_url,
        client_certificate_path=config.client_certificate_path,
        client_private_key_path=config.client_private_key_path,
        ca_bundle_path=config.ca_bundle_path,
        client=session,
        server_crl_path=config.server_crl_path,
    )
    runner = build_runtime_runner()
    interval = float(os.environ.get("EDGE_POLL_INTERVAL_SECONDS", "5"))
    if not 1 <= interval <= 60:
        raise ValueError("EDGE_POLL_INTERVAL_SECONDS must be between 1 and 60")
    return EdgeAgent(config, transport, runner), health_path, interval


def run() -> int:
    stop = threading.Event()
    for event in (signal.SIGINT, signal.SIGTERM):
        signal.signal(event, lambda _signum, _frame: stop.set())
    try:
        agent, health_path, interval = build_runtime_from_environment()
    except Exception:
        return 2
    stopped = False
    while not stop.is_set():
        try:
            if agent.config.server_crl_path is None:  # pragma: no cover - invariant
                raise EdgeAuthenticationStopped("server CRL is missing")
            crl_next_update = validate_server_crl(
                agent.config.ca_bundle_path, agent.config.server_crl_path
            )
            agent.heartbeat_once()
            cycle = agent.run_once()
            _atomic_health(health_path, {
                "checked_at": _now(),
                "crl_next_update": crl_next_update.isoformat().replace("+00:00", "Z"),
                "gateway_id_digest": agent.gateway_id_digest,
                "last_cycle": cycle,
                "queue": agent.safe_diagnostic(),
                "status": "healthy",
            })
        except EdgeAuthenticationStopped:
            stopped = True
            _atomic_health(health_path, {
                "checked_at": _now(), "gateway_id_digest": agent.gateway_id_digest,
                "status": "stopped",
            })
        except (EdgeTransportError, OSError, ValueError):
            _atomic_health(health_path, {
                "checked_at": _now(), "gateway_id_digest": agent.gateway_id_digest,
                "status": "degraded",
            })
        if stopped:
            stop.wait(interval)
        else:
            stop.wait(interval)
    _atomic_health(health_path, {
        "checked_at": _now(), "gateway_id_digest": agent.gateway_id_digest,
        "status": "stopped",
    })
    return 0


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