"""Fail-closed authorization boundary for model and external-data egress. The gateway intentionally receives a digest and classifications, never a raw prompt. Callers must assemble and sanitize the payload before this boundary; providers may only be called after a successful decision has been recorded. Remoteness is decided by the policy allowlists, never by a caller supplied flag. A provider that appears in neither allowlist is denied: there is no default-open path through this boundary. """ from __future__ import annotations import hashlib from dataclasses import dataclass, field from enum import Enum from ipaddress import ip_address from urllib.parse import urlsplit from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.extension_points import ( EgressPermissionGate, LocalClassificationGate, RemoteClassificationGate, ) from cyberbrain26.security_plane.permission import SecurityError class DataClassification(str, Enum): """Highest sensitivity carried by a payload or retrieved context.""" PUBLIC = "public" PERSONAL = "personal" SENSITIVE = "sensitive" RESTRICTED = "restricted" @dataclass(frozen=True) class EgressRequest: """Minimal, non-content-bearing description of one outbound request. ``destination`` is the real endpoint the caller intends to reach. It is never trusted as-is: only a sanitised host form reaches the audit record. """ agent_id: str provider_id: str payload_sha256: str classifications: frozenset[DataClassification] = field(default_factory=frozenset) destination: str = "" @dataclass(frozen=True) class EgressPolicy: """Explicit allowlists for remote destinations and permitted data classes. Both allowlists default to empty. A caller therefore has to name every provider on purpose before any egress - local or remote - is permitted. """ allowed_remote_providers: frozenset[str] = field(default_factory=frozenset) allowed_remote_classifications: frozenset[DataClassification] = field( default_factory=lambda: frozenset({DataClassification.PUBLIC}) ) local_provider_ids: frozenset[str] = field(default_factory=frozenset) allowed_local_classifications: frozenset[DataClassification] = field( default_factory=lambda: frozenset( {DataClassification.PUBLIC, DataClassification.PERSONAL, DataClassification.SENSITIVE} ) ) class EgressGate: """Authorize egress only when destination, data class and audit are known. ``audit_log`` is required: a gate that cannot record a decision must not be constructible, otherwise every caller would have to rediscover that at the moment it tries to send data. """ def __init__( self, policy: EgressPolicy | None = None, *, audit_log: AuditLog, base_decision: bool = False, ) -> None: self.policy = policy or EgressPolicy() self.audit_log = audit_log # Extension point gatey - defaultne odrazeji nastaveni politiky, # lze je později nahradit bez změny jádra EgressGate. self.remote_classification_gate = RemoteClassificationGate( allowed=self.policy.allowed_remote_classifications ) self.local_classification_gate = LocalClassificationGate( allowed=self.policy.allowed_local_classifications ) self.egress_permission_gate = EgressPermissionGate(base_decision=base_decision) def authorize(self, request: EgressRequest) -> None: """Raise :class:`SecurityError` unless this specific egress is allowed.""" self._validate_request(request) self._require_audit() # Extension point: overall egress permission (default deny, fail-closed). if not self.egress_permission_gate.evaluate(request): self._deny(request, "egress permission denied") if request.provider_id in self.policy.allowed_remote_providers: if not self.remote_classification_gate.evaluate(request): self._deny(request, "remote classifications denied") self._authorize_remote(request) elif request.provider_id in self.policy.local_provider_ids: if not self.local_classification_gate.evaluate(request): self._deny(request, "local classifications denied") self._authorize_local(request) else: self._deny(request, "provider is in neither the remote nor the local allowlist") self._record(request, result="allowed") def _authorize_remote(self, request: EgressRequest) -> None: """Remote egress additionally has to pass the data-classification gate.""" disallowed = request.classifications - self.policy.allowed_remote_classifications if disallowed: names = ", ".join(sorted(item.value for item in disallowed)) self._deny(request, f"payload classifications are not allowed remotely: {names}") def _authorize_local(self, request: EgressRequest) -> None: """Local egress skips no control at all. Being on the allowlist only proves the owner named the provider; it proves nothing about where the bytes actually go. A provider name is a string a caller controls, so treating "local" as a licence to skip the classification gate would reintroduce the very bypass that ``destination_is_remote`` used to be. Local traffic therefore runs through its own classification policy, and only ``RESTRICTED`` is excluded by default. """ if request.destination and not self._is_loopback_destination(request.destination): self._deny(request, "provider is declared local but the destination is not a loopback address") disallowed = request.classifications - self.policy.allowed_local_classifications if disallowed: names = ", ".join(sorted(item.value for item in disallowed)) self._deny(request, f"payload classifications are not allowed for local egress: {names}") @classmethod def _is_loopback_destination(cls, destination: str) -> bool: """True only for a destination whose host is a loopback literal. Host names such as ``localhost`` are accepted, but nothing that could resolve off-box; resolution is deliberately not attempted here, so an attacker-controlled DNS name cannot decide this branch. """ candidate = destination.strip() if "//" not in candidate: candidate = f"//{candidate}" try: parts = urlsplit(candidate) host = (parts.hostname or "").lower() except ValueError: return False if host in {"localhost", "::1"} or host.startswith("127."): return True try: return ip_address(host).is_loopback except ValueError: return False def _require_audit(self) -> None: """Fail closed when no audit log is present, even though it is required.""" if self.audit_log is None: raise SecurityError("egress denied because audit logging is unavailable") def _deny(self, request: EgressRequest, reason: str) -> None: """Record the refusal when possible, but always end in a refusal. A failing audit write must not turn a denial into an unrelated exception, and it must never let the request through: the ``SecurityError`` is raised whether or not the record succeeded. """ try: self._record(request, result="denied", error=reason) except Exception as error: # noqa: BLE001 - the denial below is the control reason = f"{reason} (audit write failed: {type(error).__name__})" raise SecurityError(f"egress denied: {reason}") @staticmethod def _sanitize_destination(destination: str) -> str: """Reduce a destination to a bare ``host[:port]`` safe for an audit row. Credentials live in the userinfo section of a URL. ``urlsplit().netloc`` keeps that userinfo, so only ``hostname`` and ``port`` may be used. The path, query and fragment are dropped outright: they routinely carry API keys and personal identifiers, and an audit log must never become the place where those accumulate. Percent-encoded separators cannot survive this reduction because the host is re-read from the parsed authority, never copied out of the raw string. """ candidate = destination.strip() if not candidate: return "" # "host:port" without a scheme has no netloc until "//" is supplied. if "//" not in candidate: candidate = f"//{candidate}" try: parts = urlsplit(candidate) host = parts.hostname port = parts.port except ValueError: # A malformed port raises; fall back to the opaque provider name # rather than copying an unparsed string into the audit row. return "" if not host: return "" host = host.lower() return f"{host}:{port}" if port else host def _record(self, request: EgressRequest, *, result: str, error: str = "") -> None: is_remote = request.provider_id in self.policy.allowed_remote_providers raw_destination = request.destination or request.provider_id self.audit_log.record( AuditEventType.DATA_ACCESS, actor=request.agent_id, agent_id=request.agent_id, action="model_egress", resource=self._sanitize_destination(request.provider_id) or request.provider_id, result=result, error=error, network_action="remote" if is_remote else "local", metadata={ "payload_sha256": request.payload_sha256, "classifications": sorted(item.value for item in request.classifications), "destination_host": self._sanitize_destination(raw_destination), "destination_sha256": hashlib.sha256(raw_destination.encode()).hexdigest(), }, ) @staticmethod def _validate_request(request: EgressRequest) -> None: if not request.agent_id.strip() or not request.provider_id.strip(): raise ValueError("egress agent_id and provider_id must not be empty") if len(request.payload_sha256) != 64 or any( character not in "0123456789abcdef" for character in request.payload_sha256.lower() ): raise ValueError("payload_sha256 must be a SHA-256 hexadecimal digest") __all__ = ["DataClassification", "EgressGate", "EgressPolicy", "EgressRequest"]