d-OAuth-Server/scripts/check_manifests.py

#!/usr/bin/env python3
"""Cluster-manifest policy lint (task 46).

Validates `deploy/k8s` (and `compose.yaml` lightly) against the cluster
contract WITHOUT a live cluster:

- Deployment: 3 replicas, non-root 65532, read-only rootfs, RuntimeDefault
  seccomp, ALL capabilities dropped, requests+limits, startup/liveness on
  `/health/live` + readiness on `/health/ready` via httpGet against the
  plaintext health port (8081), `terminationGracePeriodSeconds >= 45`,
  topology spread, and NO per-replica migration.
- Migration Job: exactly one, runs `migrate`, `restartPolicy: Never`.
- PDB: `minAvailable: 2`.
- Service: never exposes the health port (8081).
- No baked secrets: no PEM blocks in manifests; `REPLACE_ME` placeholders
  only inside `secret.yaml`; no secret-shaped keys in the ConfigMap.
- L7-vs-L4 mTLS constraint: an L7 Ingress that terminates TLS while
  in-process client-cert auth is enabled (`D_OAUTH_TLS__CLIENT_CA_PATH`
  set, or client-cert Ingress annotations) FAILS. mTLS needs L4/TCP
  passthrough (NLB gateway / TLSRoute) or explicit trusted-proxy mode.
  This constraint is DOCUMENTED (docs/DEPLOYMENT.md) and lint-checked
  here; it is explicitly NOT exercised end-to-end in CI.

Usage:
  python3 scripts/check_manifests.py [--manifest-dir deploy/k8s]
                                     [--strict-digest]

`--strict-digest` (used by supply-chain CI after the pin step) additionally
requires every workload image to be digest-pinned (`@sha256:<64 hex>`).
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

import yaml

HEALTH_PORT = 8081
PUBLIC_PORT = 8080
TERMINATION_GRACE_MIN = 45
SECRET_NAME_RE = re.compile(r"(?i)(pepper|password|secret|private[_-]?key|totp)")
PEM_RE = re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----|-----BEGIN CERTIFICATE-----")
DIGEST_RE = re.compile(r"@sha256:[0-9a-f]{64}$")
HIGH_ENTROPY_VALUE_RE = re.compile(r"^[A-Za-z0-9+/=_-]{32,}$")

CLIENT_CERT_ANNOTATIONS = (
  "nginx.ingress.kubernetes.io/auth-tls-verify-client",
  "nginx.ingress.kubernetes.io/auth-tls-secret",
  "haproxy.org/ssl-verify-client",
  "traefik.ingress.kubernetes.io/router.tls.options",
)


class Failures:
  def __init__(self) -> None:
    self.items: list[str] = []

  def add(self, msg: str) -> None:
    self.items.append(msg)

  def ok(self) -> bool:
    return not self.items


def load_manifests(manifest_dir: Path) -> tuple[list[dict], Failures]:
  fails = Failures()
  docs: list[dict] = []
  files = sorted(manifest_dir.glob("*.yaml"))
  if not files:
    fails.add(f"no manifests found in {manifest_dir}")
    return docs, fails
  for path in files:
    if path.name.endswith(".example.yaml"):
      continue
    try:
      loaded = list(yaml.safe_load_all(path.read_text()))
    except yaml.YAMLError as exc:
      fails.add(f"{path.name}: invalid YAML: {exc}")
      continue
    for doc in loaded:
      if doc:
        doc["_source"] = path.name
        docs.append(doc)
  return docs, fails


def by_kind(docs: list[dict], kind: str) -> list[dict]:
  return [d for d in docs if d.get("kind") == kind]


def check_pem_and_placeholders(docs: list[dict], fails: Failures) -> None:
  for doc in docs:
    src = doc.get("_source", "?")
    text = yaml.safe_dump({k: v for k, v in doc.items() if not k.startswith("_")})
    if PEM_RE.search(text):
      fails.add(f"{src}: baked key material (PEM block) in a manifest")
    if src != "secret.yaml" and "REPLACE_ME" in text:
      fails.add(f"{src}: placeholder belongs only in secret.yaml")


def check_configmap(doc: dict, fails: Failures) -> None:
  data = doc.get("data") or {}
  for key in data:
    if SECRET_NAME_RE.search(key):
      fails.add(f"configmap: secret-shaped key {key!r} must live in the Secret")
  if (data.get("D_OAUTH_TLS__CLIENT_CA_PATH") or "").strip():
    pass  # mTLS on: L7 check below becomes load-bearing.


def mtls_enabled(docs: list[dict]) -> bool:
  for doc in by_kind(docs, "ConfigMap"):
    data = doc.get("data") or {}
    if (data.get("D_OAUTH_TLS__CLIENT_CA_PATH") or "").strip():
      return True
  return False


def check_deployment(doc: dict, fails: Failures, strict_digest: bool) -> None:
  name = doc.get("metadata", {}).get("name", "?")
  spec = doc.get("spec", {})
  if spec.get("replicas") != 3:
    fails.add(f"deployment/{name}: replicas must be 3")
  tpl = spec.get("template", {}).get("spec", {})
  if tpl.get("terminationGracePeriodSeconds", 0) < TERMINATION_GRACE_MIN:
    fails.add(f"deployment/{name}: terminationGracePeriodSeconds must be >= 45")
  pod_sc = tpl.get("securityContext", {})
  if pod_sc.get("runAsNonRoot") is not True:
    fails.add(f"deployment/{name}: runAsNonRoot must be true")
  if pod_sc.get("runAsUser") != 65532:
    fails.add(f"deployment/{name}: runAsUser must be 65532")
  if (pod_sc.get("seccompProfile") or {}).get("type") != "RuntimeDefault":
    fails.add(f"deployment/{name}: seccompProfile.type must be RuntimeDefault")
  if not tpl.get("topologySpreadConstraints"):
    fails.add(f"deployment/{name}: topologySpreadConstraints required")
  for container in tpl.get("containers", []):
    cname = container.get("name", "?")
    csc = container.get("securityContext", {})
    if csc.get("readOnlyRootFilesystem") is not True:
      fails.add(f"deployment/{name}/{cname}: readOnlyRootFilesystem must be true")
    if "ALL" not in (csc.get("capabilities") or {}).get("drop", []):
      fails.add(f"deployment/{name}/{cname}: capabilities.drop must include ALL")
    res = container.get("resources", {})
    if not res.get("requests") or not res.get("limits"):
      fails.add(f"deployment/{name}/{cname}: requests and limits required")
    args = " ".join(
      str(a)
      for a in (
        list(container.get("command", [])) + list(container.get("args", []))
      )
    )
    if "migrate" in args:
      fails.add(f"deployment/{name}/{cname}: replicas must never migrate")
    probes = {
      "startupProbe": ("/health/live", True),
      "livenessProbe": ("/health/live", True),
      "readinessProbe": ("/health/ready", True),
    }
    for probe_name, (path, _) in probes.items():
      probe = container.get(probe_name) or {}
      http_get = probe.get("httpGet") or {}
      if http_get.get("path") != path or http_get.get("port") != HEALTH_PORT:
        fails.add(
          f"deployment/{name}/{cname}: {probe_name} must httpGet "
          f"{path} on port {HEALTH_PORT} (plaintext health listener)"
        )
    image = container.get("image", "")
    if strict_digest and not DIGEST_RE.search(image):
      fails.add(f"deployment/{name}/{cname}: image must be digest-pinned")


def check_job(doc: dict, fails: Failures, strict_digest: bool) -> None:
  name = doc.get("metadata", {}).get("name", "?")
  spec = doc.get("spec", {}).get("template", {}).get("spec", {})
  if spec.get("restartPolicy") != "Never":
    fails.add(f"job/{name}: restartPolicy must be Never")
  containers = spec.get("containers", [])
  if not any(
    "migrate" in " ".join(str(a) for a in (list(c.get("command", [])) + list(c.get("args", []))))
    for c in containers
  ):
    fails.add(f"job/{name}: exactly one migrate Job must run `/d-oauth-server migrate`")
  for container in containers:
    if strict_digest and not DIGEST_RE.search(container.get("image", "")):
      fails.add(f"job/{name}: image must be digest-pinned")


def check_pdb(doc: dict, fails: Failures) -> None:
  if doc.get("spec", {}).get("minAvailable") != 2:
    fails.add("poddisruptionbudget: minAvailable must be 2")


def check_service(doc: dict, fails: Failures) -> None:
  ports = doc.get("spec", {}).get("ports", [])
  exposed = {p.get("port") for p in ports} | {p.get("targetPort") for p in ports}
  if HEALTH_PORT in exposed:
    fails.add("service: must not expose the plaintext health port 8081")


def check_l7_mtls(docs: list[dict], fails: Failures) -> None:
  """An L7 ingress that terminates TLS strips the client certificate."""
  ingresses = by_kind(docs, "Ingress")
  if not ingresses:
    return
  client_auth = mtls_enabled(docs)
  for ing in ingresses:
    name = ing.get("metadata", {}).get("name", "?")
    annotations = (ing.get("metadata", {}) or {}).get("annotations", {}) or {}
    terminates = bool((ing.get("spec", {}) or {}).get("tls"))
    wants_client_cert = client_auth or any(a in annotations for a in CLIENT_CERT_ANNOTATIONS)
    if terminates and wants_client_cert:
      fails.add(
        f"ingress/{name}: L7 TLS termination strips the client certificate "
        "while in-process client-cert auth is enabled — use L4/TCP "
        "passthrough (NLB gateway / TLSRoute) or explicit trusted-proxy mode"
      )


def main() -> int:
  parser = argparse.ArgumentParser()
  parser.add_argument("--manifest-dir", default="deploy/k8s")
  parser.add_argument("--strict-digest", action="store_true")
  args = parser.parse_args()

  fails = Failures()
  docs, load_fails = load_manifests(Path(args.manifest_dir))
  fails.items.extend(load_fails.items)

  kinds = {d.get("kind") for d in docs}
  for required in ("Deployment", "Service", "ConfigMap", "Secret", "Job", "PodDisruptionBudget"):
    kind_name = "Job" if required == "Job" else required
    if required not in kinds:
      fails.add(f"missing required kind: {required} ({kind_name})")

  check_pem_and_placeholders(docs, fails)
  for doc in by_kind(docs, "ConfigMap"):
    check_configmap(doc, fails)
  for doc in by_kind(docs, "Deployment"):
    check_deployment(doc, fails, args.strict_digest)
  for doc in by_kind(docs, "Job"):
    check_job(doc, fails, args.strict_digest)
  for doc in by_kind(docs, "PodDisruptionBudget"):
    check_pdb(doc, fails)
  for doc in by_kind(docs, "Service"):
    check_service(doc, fails)
  check_l7_mtls(docs, fails)

  if fails.ok():
    print(f"manifest lint: OK ({len(docs)} documents, strict-digest={args.strict_digest})")
    return 0
  print(f"manifest lint: {len(fails.items)} failure(s):")
  for item in fails.items:
    print(f"  - {item}")
  return 1


if __name__ == "__main__":
  sys.exit(main())