Coverage for scripts / live_release_validation / checks / policy.py: 100.00%
93 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Deployed job-validation policy readback checks.
3``GET /api/v1/policy`` exists so a caller can ask "will this cluster admit the
4job I am about to pay to run" before submitting. It answers in three layers: the
5front-door per-manifest caps the manifest processor applies itself, the
6per-container ``LimitRange``, and the namespace aggregate ``ResourceQuota``.
8The two cluster-read layers are deliberately fail-soft -- a Kubernetes read
9failure degrades that namespace to ``{"status": "unavailable", "reason": ...}``
10rather than failing the whole response. That is the right behavior and it is also
11why this check has to exist: **the degraded response is an HTTP 200**, so
12``response.ok`` proves nothing and every transport-level check in this harness
13passes while the endpoint reports nothing useful.
15That is not hypothetical. The 2026-08-26 run was green across all ten actions
16while ``cluster_enforcement."gco-jobs"`` was ``{"status": "unavailable",
17"reason": "403 Forbidden"}`` -- the manifest-processor Role had no grant on
18``resourcequotas``/``limitranges``. A caller reading only the caps would be told
19a manifest is admissible that pod creation then rejects, possibly after they
20provisioned a region on the strength of that answer.
22So the assertions here are all on the response *body*.
23"""
25from __future__ import annotations
27import re
28from typing import Any
30from ..checks.jobs import _response_json
31from ..context import _job_transport_region
32from ..models import RunContext
34#: An AWS ECR registry hostname, e.g. 123456789012.dkr.ecr.us-east-2.amazonaws.com
35_ECR_HOSTNAME = re.compile(r"^(\d{12})\.dkr\.ecr\.([a-z0-9-]+)\.amazonaws\.com(?:\.cn)?$")
37#: Quota keys the deployed ResourceQuota is expected to carry. Sourced from
38#: 04-resource-quotas.yaml, which is substituted from cdk.json at deploy time.
39_EXPECTED_QUOTA_HINTS = ("cpu", "memory")
42def _get_policy(ctx: RunContext, region: str) -> dict[str, Any]:
43 """Fetch one Region's /api/v1/policy through its authorized transport."""
44 response = ctx.aws_client.make_authenticated_request(
45 method="GET",
46 path="/api/v1/policy",
47 target_region=_job_transport_region(ctx, region),
48 )
49 if not response.ok:
50 raise RuntimeError(
51 f"Policy readback for {region} failed: {response.status_code} {response.text}"
52 )
53 return _response_json(response, f"Policy readback for {region}")
56def _validate_identity(ctx: RunContext, region: str, payload: dict[str, Any]) -> None:
57 """Require the response to come from the Region we addressed.
59 A transport that silently answered from the wrong Region would make every
60 other assertion here describe a cluster the caller did not ask about.
61 """
62 observed_region = str(payload.get("region") or "")
63 if observed_region != region:
64 raise RuntimeError(
65 f"Policy readback transport returned Region {observed_region!r}; expected {region!r}"
66 )
67 expected_cluster = f"{ctx.config.project_name}-{region}"
68 observed_cluster = str(payload.get("cluster_id") or "")
69 if observed_cluster != expected_cluster:
70 raise RuntimeError(
71 f"Policy readback for {region} reported cluster_id {observed_cluster!r}; "
72 f"expected {expected_cluster!r}"
73 )
74 source = str(payload.get("source") or "")
75 if source != "deployed-cluster-runtime":
76 raise RuntimeError(
77 f"Policy readback for {region} reported source {source!r}; the endpoint must "
78 "name the deployed runtime as its origin so a caller cannot mistake it for "
79 "a config-file read"
80 )
83def _validate_front_door(region: str, policy: dict[str, Any]) -> dict[str, Any]:
84 """Require the layer-1 caps and allowlists to be present and non-degenerate."""
85 if not policy:
86 raise RuntimeError(f"Policy readback for {region} carried no policy object")
88 caps = policy.get("manifest_caps")
89 if not isinstance(caps, dict):
90 raise RuntimeError(f"Policy readback for {region} omitted manifest_caps")
91 for key in ("max_cpu_millicores", "max_memory_bytes", "max_gpu_count"):
92 value = caps.get(key)
93 if not isinstance(value, int) or value <= 0:
94 raise RuntimeError(
95 f"Policy readback for {region} reported {key}={value!r}; a non-positive "
96 "cap would reject every job that requests that resource"
97 )
99 namespaces = policy.get("allowed_namespaces")
100 if not isinstance(namespaces, list) or not namespaces:
101 raise RuntimeError(
102 f"Policy readback for {region} reported no allowed_namespaces; nothing "
103 "could ever be submitted"
104 )
105 kinds = policy.get("allowed_kinds")
106 if not isinstance(kinds, list) or not kinds:
107 raise RuntimeError(f"Policy readback for {region} reported no allowed_kinds")
109 return {
110 "max_cpu_millicores": caps["max_cpu_millicores"],
111 "max_memory_bytes": caps["max_memory_bytes"],
112 "max_gpu_count": caps["max_gpu_count"],
113 "allowed_namespaces": sorted(str(item) for item in namespaces),
114 "allowed_kinds": sorted(str(item) for item in kinds),
115 "require_accelerator_toleration": bool(policy.get("require_accelerator_toleration")),
116 "validation_enabled": bool(policy.get("validation_enabled")),
117 }
120def _validate_synth_time_ecr_augmentation(
121 ctx: RunContext, region: str, policy: dict[str, Any]
122) -> list[str]:
123 """Require the project's own ECR hostnames to be in the trusted allowlist.
125 CDK appends them at synth time (``_augment_trusted_registries_with_project
126 _ecr``), which is the concrete reason a locally-computed policy is not
127 authoritative: the deployed allowlist is strictly larger than the configured
128 one. If this augmentation silently stopped happening, every job pulling from
129 the project's own registry would be rejected as an untrusted image source --
130 and no offline check would predict it, because the offline view never had
131 those hostnames.
132 """
133 registries = policy.get("trusted_registries")
134 if not isinstance(registries, list) or not registries:
135 raise RuntimeError(f"Policy readback for {region} reported no trusted_registries")
137 account = str(ctx.settings.expected_account or "")
138 augmentation: list[str] = []
139 for entry in registries:
140 match = _ECR_HOSTNAME.match(str(entry))
141 if match and (not account or match.group(1) == account):
142 augmentation.append(str(entry))
144 if not augmentation:
145 raise RuntimeError(
146 f"Policy readback for {region} shows no project ECR registry in "
147 f"trusted_registries {sorted(map(str, registries))!r}. CDK is expected to "
148 "append the project's own ECR hostnames at synth time; without them every "
149 "job pulling a project-built image is rejected as an untrusted source"
150 )
151 return sorted(augmentation)
154def _validate_cluster_enforcement(
155 region: str, enforcement: Any, allowed_namespaces: list[str]
156) -> dict[str, Any]:
157 """Require layers 2 and 3 to be readable for every allowed namespace.
159 This is the assertion the 2026-08-26 regression needed. The endpoint returns
160 200 with a per-namespace ``status`` field, so the failure is only visible
161 here -- in the body, per namespace.
162 """
163 if not isinstance(enforcement, dict) or not enforcement:
164 raise RuntimeError(
165 f"Policy readback for {region} carried no cluster_enforcement object; "
166 "layers 2 and 3 (LimitRange, ResourceQuota) would be unreportable"
167 )
169 missing = sorted(set(allowed_namespaces) - set(enforcement))
170 if missing:
171 raise RuntimeError(
172 f"Policy readback for {region} reports no cluster_enforcement for allowed "
173 f"namespace(s) {missing}; a caller cannot tell whether a job would clear "
174 "the namespace ceilings"
175 )
177 degraded: list[str] = []
178 summary: dict[str, Any] = {}
179 for namespace in sorted(allowed_namespaces):
180 layer = enforcement.get(namespace)
181 if not isinstance(layer, dict):
182 degraded.append(f"{namespace}: not an object")
183 continue
184 status = str(layer.get("status") or "unknown")
185 if status != "ok":
186 degraded.append(f"{namespace}: {status} — {layer.get('reason', 'no reason given')}")
187 continue
189 quotas = layer.get("resource_quotas")
190 limits = layer.get("limit_ranges")
191 if not isinstance(quotas, dict) or not quotas:
192 raise RuntimeError(
193 f"Policy readback for {region} namespace {namespace} reports status ok "
194 "but no ResourceQuota. 04-resource-quotas.yaml is expected to deploy "
195 "one, so an empty result means the aggregate ceiling is unenforced"
196 )
197 if not isinstance(limits, dict) or not limits:
198 raise RuntimeError(
199 f"Policy readback for {region} namespace {namespace} reports status ok "
200 "but no LimitRange; the per-container ceiling is unenforced"
201 )
202 for name, hard in quotas.items():
203 if not isinstance(hard, dict) or not any(
204 any(hint in str(key) for hint in _EXPECTED_QUOTA_HINTS) for key in hard
205 ):
206 raise RuntimeError(
207 f"Policy readback for {region} ResourceQuota/{name} in {namespace} "
208 f"carries no cpu/memory ceiling: {hard!r}"
209 )
210 summary[namespace] = {
211 "status": status,
212 "resource_quotas": sorted(quotas),
213 "limit_ranges": sorted(limits),
214 }
216 if degraded:
217 raise RuntimeError(
218 f"Policy readback for {region} could not read the live ResourceQuota / "
219 f"LimitRange: {'; '.join(degraded)}. The response degrades to HTTP 200, so "
220 "this is invisible to a transport-level check — a caller is told a manifest "
221 "is admissible that pod creation may still reject. Check the "
222 "gco-manifest-processor-role grant on resourcequotas and limitranges."
223 )
224 return summary
227def _validate_region_policy(ctx: RunContext, region: str) -> dict[str, Any]:
228 """Assert one Region's full three-layer policy readback."""
229 payload = _get_policy(ctx, region)
230 _validate_identity(ctx, region, payload)
232 policy = payload.get("policy")
233 policy = policy if isinstance(policy, dict) else {}
234 front_door = _validate_front_door(region, policy)
235 augmentation = _validate_synth_time_ecr_augmentation(ctx, region, policy)
236 enforcement = _validate_cluster_enforcement(
237 region, payload.get("cluster_enforcement"), front_door["allowed_namespaces"]
238 )
240 return {
241 "region": region,
242 "cluster_id": payload.get("cluster_id"),
243 "source": payload.get("source"),
244 "policy": front_door,
245 "synth_time_ecr_registries": augmentation,
246 "cluster_enforcement": enforcement,
247 }