Enforce private-by-default rail exposure
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
This commit is contained in:
parent
c6b045051f
commit
004f1c4dc1
7 changed files with 368 additions and 7 deletions
110
tools/exposure.py
Normal file
110
tools/exposure.py
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
"""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}
|
||||
Loading…
Add table
Add a link
Reference in a new issue