"""ADR-0008 exposure validation for rendered Kubernetes manifests.""" from __future__ import annotations import re from pathlib import Path from typing import Any import yaml class ExposureError(ValueError): """Rendered public exposure is not backed by the required declarations.""" def _load(path: Path | None, label: str) -> dict[str, Any]: if path is None: raise ExposureError(f"public manifest requires --{label}-declaration") try: payload = yaml.safe_load(path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError) as exc: raise ExposureError(f"cannot read {label} declaration {path}: {exc}") from exc if not isinstance(payload, dict): raise ExposureError(f"{label} declaration must be a YAML object") return payload def _public_hosts(documents: list[dict[str, Any]]) -> list[str]: hosts: set[str] = set() public_kinds = 0 for document in documents: kind = document.get("kind") spec = document.get("spec") or {} if kind == "Ingress": public_kinds += 1 for rule in spec.get("rules") or []: host = rule.get("host") if isinstance(rule, dict) else None if host: hosts.add(str(host)) elif kind == "IngressRoute": public_kinds += 1 for route in spec.get("routes") or []: match = str(route.get("match", "")) if isinstance(route, dict) else "" hosts.update(re.findall(r"Host\(`([^`]+)`\)", match)) if public_kinds and not hosts: raise ExposureError("public routing resource has no exact hostname") return sorted(hosts) def _grant_matches(grant: Any, hostname: str, *, allow_port: bool) -> bool: if not isinstance(grant, dict): return False if grant.get("hostname") == hostname: return True return allow_port and grant.get("port") in {80, 443, "80", "443"} def _require_grant_metadata(grant: Any, label: str) -> None: if not isinstance(grant, dict): raise ExposureError(f"{label} grant must be an object") missing = [ field for field in ("reason", "approved_on", "residual_risk_owner") if not grant.get(field) ] if missing: raise ExposureError(f"{label} grant is missing {', '.join(missing)}") def validate_rendered_exposure( rendered: str, *, rapp_declaration: Path | None = None, reef_declaration: Path | None = None, ) -> dict[str, Any]: """Validate all Ingress/IngressRoute hosts against rapp and reef grants.""" try: documents = [item for item in yaml.safe_load_all(rendered) if isinstance(item, dict)] except yaml.YAMLError as exc: raise ExposureError(f"rendered manifest is not valid YAML: {exc}") from exc hosts = _public_hosts(documents) if not hosts: return {"posture": "private", "public_hosts": [], "validated": True} rapp = _load(rapp_declaration, "rapp") reef = _load(reef_declaration, "reef") rapp_exposure = rapp.get("exposure") or {} reef_exposure = reef.get("exposure") or {} if rapp_exposure.get("posture") != "public": raise ExposureError("rapp exposure.posture must be public") if rapp_exposure.get("binding_admission") != "production-approved": raise ExposureError("rapp exposure.binding_admission must be production-approved") if reef_exposure.get("posture") != "public": raise ExposureError("reef exposure.posture must be public") rapp_grant = rapp_exposure.get("grant") reef_grants = reef_exposure.get("grants") or [] _require_grant_metadata(rapp_grant, "rapp") for hostname in hosts: if not _grant_matches(rapp_grant, hostname, allow_port=False): raise ExposureError(f"rapp grant does not name rendered hostname {hostname}") matching_reef_grants = [ grant for grant in reef_grants if _grant_matches(grant, hostname, allow_port=True) ] if not matching_reef_grants: raise ExposureError(f"reef has no public substrate grant for {hostname}") for grant in matching_reef_grants: _require_grant_metadata(grant, "reef") return {"posture": "public", "public_hosts": hosts, "validated": True}