004f1c4dc1
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""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}
|