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