Coverage for scripts / live_release_validation / checks / internal_pki.py: 100.00%
115 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Internal PKI checks: every in-cluster TLS hop chains to the one GCO internal CA.
3The in-cluster HTTPS hops (ALB to the API pods, manifest-processor to
4cost-monitor, cost-monitor to OpenCost, Prometheus to the service metrics,
5the Grafana rotator to Grafana, inference-proxy to the model pods, MLflow's
6gco-jobs clients to the tracking server) verify
7their peers against a single cert-manager CA: a self-signed bootstrap
8``ClusterIssuer`` issues the ``cert-manager/gco-internal-ca`` CA Certificate,
9and the ``gco-internal-ca`` ``ClusterIssuer`` signs every leaf from it. A
10client that trusts only that CA fails closed, so a leaf that was never
11issued, or was issued by anything else, is a broken hop rather than a
12cosmetic difference.
14On the live cluster this requires, per Region:
16* the ``gco-internal-ca`` ClusterIssuer ``Ready``;
17* the CA Certificate ``Ready``, ``isCA``, issued by the bootstrap issuer;
18* every GCO leaf Certificate the configuration deploys ``Ready`` with
19 ``spec.issuerRef`` naming the ``gco-internal-ca`` ClusterIssuer (the
20 cost-monitor and OpenCost leaves only with cost monitoring, the Grafana
21 leaf and the monitoring trust bundle only with cluster observability, the
22 MLflow leaf and the trust-manager bundle source only with MLflow); and
23* the retired per-namespace ``gco-api-selfsigned`` Issuer gone.
25A missing object, a wrong issuer, or the retired Issuer fails on the first
26look; a Certificate cert-manager is still issuing is polled to a deadline.
27Reads use the fully qualified ``*.cert-manager.io`` resources so another
28controller's ``Certificate`` kind (ACK's ACM controller, for one) can never
29answer instead.
30"""
32from __future__ import annotations
34import time
35from dataclasses import dataclass
36from typing import Any
38from ..models import RunContext
39from .cluster import KubectlRunner, kubectl_json
40from .opencost import _cost_monitoring_configured
42INTERNAL_CA_ISSUER = "gco-internal-ca"
43BOOTSTRAP_ISSUER = "gco-internal-ca-bootstrap"
44CA_CERTIFICATE_NAMESPACE = "cert-manager"
45CA_CERTIFICATE_NAME = "gco-internal-ca"
46LEGACY_ISSUER_NAMESPACE = "gco-system"
47LEGACY_ISSUER_NAME = "gco-api-selfsigned"
48_CERTIFICATES = "certificates.cert-manager.io"
49_CLUSTER_ISSUERS = "clusterissuers.cert-manager.io"
50_ISSUERS = "issuers.cert-manager.io"
51#: Issuance takes seconds; this covers a renewal that happens to be in flight.
52_READY_TIMEOUT_SECONDS = 300
55@dataclass(frozen=True)
56class LeafCertificate:
57 """One GCO leaf Certificate and the configuration that deploys it."""
59 namespace: str
60 name: str
61 #: ``None`` for always; otherwise the cdk.json feature that gates it.
62 feature: str | None = None
65LEAF_CERTIFICATES: tuple[LeafCertificate, ...] = (
66 LeafCertificate("gco-system", "health-monitor-tls"),
67 LeafCertificate("gco-system", "manifest-processor-tls"),
68 LeafCertificate("gco-system", "inference-proxy-tls"),
69 LeafCertificate("gco-system", "inference-monitor-tls"),
70 LeafCertificate("gco-system", "cost-monitor-tls", "cost_monitoring"),
71 LeafCertificate("gco-inference", "gco-inference-tls"),
72 LeafCertificate("monitoring", "opencost-tls", "cost_monitoring"),
73 LeafCertificate("monitoring", "grafana-tls", "cluster_observability"),
74 LeafCertificate("monitoring", "gco-monitoring-trust", "cluster_observability"),
75 LeafCertificate("monitoring", "mlflow-tls", "mlflow"),
76 LeafCertificate("trust-manager", "gco-internal-ca-source", "mlflow"),
77)
80class InternalPkiValidationError(RuntimeError):
81 """The live cluster's certificates do not chain to the GCO internal CA."""
84def cluster_observability_configured(ctx: RunContext) -> bool:
85 """Return whether cdk.json keeps cluster observability on (default on)."""
86 block = ctx.cdk_context.get("cluster_observability")
87 if isinstance(block, dict) and "enabled" in block:
88 return bool(block["enabled"])
89 return True
92def mlflow_configured(ctx: RunContext) -> bool:
93 """Return whether cdk.json deploys MLflow: its own toggle (default on) and observability."""
94 if not cluster_observability_configured(ctx):
95 return False
96 block = ctx.cdk_context.get("cluster_observability")
97 mlflow = block.get("mlflow") if isinstance(block, dict) else None
98 if isinstance(mlflow, dict) and "enabled" in mlflow:
99 return bool(mlflow["enabled"])
100 return True
103def expected_leaf_certificates(
104 ctx: RunContext,
105) -> tuple[list[LeafCertificate], dict[str, str]]:
106 """Return the leaves this configuration deploys and the skipped ones with reasons."""
107 enabled = {
108 "cost_monitoring": _cost_monitoring_configured(ctx),
109 "cluster_observability": cluster_observability_configured(ctx),
110 "mlflow": mlflow_configured(ctx),
111 }
112 expected: list[LeafCertificate] = []
113 skipped: dict[str, str] = {}
114 for leaf in LEAF_CERTIFICATES:
115 if leaf.feature is None or enabled[leaf.feature]:
116 expected.append(leaf)
117 else:
118 skipped[f"{leaf.namespace}/{leaf.name}"] = f"{leaf.feature} is disabled in cdk.json"
119 return expected, skipped
122def _dict(value: Any) -> dict[str, Any]:
123 return value if isinstance(value, dict) else {}
126def _ready(obj: dict[str, Any]) -> bool:
127 return any(
128 _dict(entry).get("type") == "Ready" and _dict(entry).get("status") == "True"
129 for entry in _dict(obj.get("status")).get("conditions") or []
130 )
133def _issuer_ref(obj: dict[str, Any]) -> dict[str, Any]:
134 ref = _dict(_dict(obj.get("spec")).get("issuerRef"))
135 return {
136 "name": ref.get("name"),
137 # cert-manager reads an omitted kind as a namespaced Issuer.
138 "kind": ref.get("kind") or "Issuer",
139 "group": ref.get("group") or "cert-manager.io",
140 }
143def _cluster_issuer_ref(ref: dict[str, Any], name: str) -> bool:
144 return ref == {"name": name, "kind": "ClusterIssuer", "group": "cert-manager.io"}
147def _snapshot(
148 ctx: RunContext,
149 kubectl: KubectlRunner,
150 record: dict[str, Any],
151 leaves: list[LeafCertificate],
152) -> dict[str, Any]:
153 timeout = float(ctx.settings.command_timeout_seconds)
154 violations: list[str] = []
155 pending: list[str] = []
157 issuer = kubectl_json(
158 kubectl, record, "get", _CLUSTER_ISSUERS, INTERNAL_CA_ISSUER, timeout=timeout
159 )
160 issuer_ready = issuer is not None and _ready(issuer)
161 if issuer is None:
162 violations.append(f"ClusterIssuer {INTERNAL_CA_ISSUER} is missing")
163 elif not issuer_ready:
164 pending.append(f"ClusterIssuer {INTERNAL_CA_ISSUER} is not Ready")
166 ca = kubectl_json(
167 kubectl,
168 record,
169 "get",
170 _CERTIFICATES,
171 CA_CERTIFICATE_NAME,
172 "--namespace",
173 CA_CERTIFICATE_NAMESPACE,
174 timeout=timeout,
175 )
176 ca_label = f"Certificate {CA_CERTIFICATE_NAMESPACE}/{CA_CERTIFICATE_NAME}"
177 ca_evidence: dict[str, Any] = {"exists": ca is not None}
178 if ca is None:
179 violations.append(f"{ca_label} is missing")
180 else:
181 ref = _issuer_ref(ca)
182 ca_evidence.update(
183 {
184 "ready": _ready(ca),
185 "is_ca": _dict(ca.get("spec")).get("isCA") is True,
186 "issuer": ref["name"],
187 "issuer_kind": ref["kind"],
188 }
189 )
190 if not _cluster_issuer_ref(ref, BOOTSTRAP_ISSUER):
191 violations.append(f"{ca_label} is not issued by ClusterIssuer {BOOTSTRAP_ISSUER}")
192 if not ca_evidence["is_ca"]:
193 violations.append(f"{ca_label} is not a CA certificate")
194 if not ca_evidence["ready"]:
195 pending.append(f"{ca_label} is not Ready")
197 certificates: dict[str, Any] = {}
198 for leaf in leaves:
199 label = f"{leaf.namespace}/{leaf.name}"
200 certificate = kubectl_json(
201 kubectl,
202 record,
203 "get",
204 _CERTIFICATES,
205 leaf.name,
206 "--namespace",
207 leaf.namespace,
208 timeout=timeout,
209 )
210 if certificate is None:
211 certificates[label] = {"exists": False}
212 violations.append(f"Certificate {label} is missing")
213 continue
214 ref = _issuer_ref(certificate)
215 ready = _ready(certificate)
216 certificates[label] = {
217 "exists": True,
218 "ready": ready,
219 "issuer": ref["name"],
220 "issuer_kind": ref["kind"],
221 }
222 if not _cluster_issuer_ref(ref, INTERNAL_CA_ISSUER):
223 violations.append(
224 f"Certificate {label} is issued by {ref['kind']} {ref['name']!r}, "
225 f"not ClusterIssuer {INTERNAL_CA_ISSUER}"
226 )
227 elif not ready:
228 pending.append(f"Certificate {label} is not Ready")
230 legacy = kubectl_json(
231 kubectl,
232 record,
233 "get",
234 _ISSUERS,
235 LEGACY_ISSUER_NAME,
236 "--namespace",
237 LEGACY_ISSUER_NAMESPACE,
238 timeout=timeout,
239 )
240 if legacy is not None:
241 violations.append(
242 f"retired Issuer {LEGACY_ISSUER_NAMESPACE}/{LEGACY_ISSUER_NAME} still exists"
243 )
244 return {
245 "cluster_issuer": {
246 "name": INTERNAL_CA_ISSUER,
247 "exists": issuer is not None,
248 "ready": issuer_ready,
249 },
250 "ca_certificate": ca_evidence,
251 "leaf_certificates": certificates,
252 "legacy_issuer_present": legacy is not None,
253 "violations": violations,
254 "pending": pending,
255 }
258def verify_internal_pki(ctx: RunContext, region: str, kubectl: KubectlRunner) -> dict[str, Any]:
259 """Poll one Region until the internal CA and every expected leaf are Ready.
261 Returns the settled snapshot; raises ``InternalPkiValidationError`` on the
262 first snapshot with a violation, or when a Certificate is still not Ready
263 at the deadline. Every observation is checkpointed under
264 ``internal_pki.<region>`` first.
265 """
266 leaves, skipped = expected_leaf_certificates(ctx)
267 with ctx.state_lock:
268 record: dict[str, Any] = ctx.checkpoint.state.setdefault("internal_pki", {}).setdefault(
269 region, {}
270 )
271 deadline = time.monotonic() + _READY_TIMEOUT_SECONDS
272 while True:
273 snapshot = _snapshot(ctx, kubectl, record, leaves)
274 snapshot["skipped_certificates"] = skipped
275 with ctx.state_lock:
276 record["last_snapshot"] = snapshot
277 ctx.persist()
278 if snapshot["violations"]:
279 raise InternalPkiValidationError(
280 f"internal PKI in {region} is broken: " + "; ".join(snapshot["violations"])
281 )
282 if not snapshot["pending"]:
283 return snapshot
284 if time.monotonic() >= deadline:
285 raise InternalPkiValidationError(
286 f"internal PKI in {region} did not become Ready within "
287 f"{_READY_TIMEOUT_SECONDS}s: " + "; ".join(snapshot["pending"])
288 )
289 time.sleep(ctx.settings.poll_interval_seconds)