Coverage for scripts / live_release_validation / checks / network_posture.py: 100.00%
208 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Network posture probes: prove the shipped NetworkPolicies decide traffic live.
3``03-network-policies.yaml`` documents a zero-trust posture, and on EKS Auto
4Mode that posture is only real once ``06-network-policy-controller.yaml`` has
5switched the policy controller on. The ``network-posture`` action settles the
6question on the deployed cluster by dialing real connections from throwaway
7pods and reading the verdict from each probe's exit code:
9* ``gco-jobs`` -> ``gco-jobs``: **reachable** (``allow-same-namespace``, so a
10 multi-pod job can talk to itself);
11* ``default`` -> ``gco-jobs``: **blocked** (``default-deny-ingress``);
12* ``gco-jobs`` -> ``gco-system``: **blocked** (``default-deny-ingress`` — the
13 job namespace cannot reach the control plane's services);
14* ``default`` -> the live inference-monitor's ``:9090/metrics``: **reachable**
15 (``allow-metrics-to-inference-monitor`` admits exactly the metrics port,
16 which is what lets Prometheus scrape it);
17* ``gco-jobs`` -> ``https://checkip.amazonaws.com/``: **reachable**
18 (``allow-dns`` plus ``allow-https-egress``);
19* ``gco-jobs`` -> ``http://checkip.amazonaws.com/`` on port 80: **blocked**
20 (no rule admits it — the same host over 443 answered, so this is the policy
21 deciding, not the network); and
22* ``gco-jobs`` -> ``http://169.254.170.23/v1/credentials``: **reachable**
23 (``allow-pod-identity-agent`` — the node's EKS Pod Identity Agent, which is
24 where a ``gco-service-account`` pod gets its AWS credentials; the probe
25 carries no token, so the agent answers 4xx, and any HTTP answer counts).
26 Port 80 to the internet is blocked by the previous probe, so this one shows
27 the rule admitting exactly the link-local endpoint.
29Every probe is a digest-pinned BusyBox Job labelled with this run's token
30(``manifests/netpol-probe-job.yaml``); the two listeners are the same image
31behind ``httpd`` (``manifests/netpol-target-job.yaml``). All of them are
32deleted before the action returns, and each carries ``activeDeadlineSeconds``
33plus a TTL so a harness that dies mid-probe still leaves nothing behind.
35A probe's verdict is its steady state, not its first dial. The VPC CNI
36attaches a new pod's policies in parallel with the pod's start and admits
37everything until they are in place (standard mode; Auto Mode's NodeClass
38``networkPolicy: DefaultDeny`` is the strict alternative, which requires a
39policy for every pod on the node), so a client that dials once at start can
40read that window instead of the policy — the fourth live run of this branch
41saw exactly that on the port-80 egress probe. The probe script therefore
42samples until one answer has held for 30 seconds and at least 45 seconds have
43passed, exits 43 if the answer is still changing after three minutes, and
44prints every change of answer; the harness keeps those lines as ``samples``,
45the closing line as ``settled``, and flags ``attach_window_observed`` when the
46first answer differed from the steady state.
48When cdk.json turns the controller off (``eks_cluster.network_policy_enforcement:
49false``) the deny verdicts are not promised and those probes are recorded as
50skipped; the reachability probes still have to pass.
51"""
53from __future__ import annotations
55import copy
56import json
57import time
58from dataclasses import asdict, dataclass
59from typing import Any
61from ..models import RunContext
62from .cluster import KubectlRunner, kubectl_json
63from .jobs import _load_manifest, _run_token
64from .platform_workloads import network_policy_enforcement_enabled
66_TARGET_MANIFEST = "netpol-target-job.yaml"
67_PROBE_MANIFEST = "netpol-probe-job.yaml"
68_TARGET_PORT = 8080
69_INFERENCE_MONITOR_METRICS_PORT = 9090
70_EGRESS_HOST = "checkip.amazonaws.com"
71#: The EKS Pod Identity Agent's link-local address on every node; the
72#: cluster points AWS_CONTAINER_CREDENTIALS_FULL_URI at this path.
73_POD_IDENTITY_AGENT_URL = "http://169.254.170.23/v1/credentials"
74#: A namespace the manifests leave unpoliced, so a client there proves the
75#: target namespace's ingress rules and nothing else.
76_UNPOLICED_NAMESPACE = "default"
77_REACHABLE_EXIT_CODE = 0
78_BLOCKED_EXIT_CODE = 42
79#: The answer kept changing for the probe's whole budget; never a verdict.
80_UNSETTLED_EXIT_CODE = 43
81#: Auto Mode may have to launch a node for the first pod; a probe then still
82#: has the image pull and its own three-minute sampling budget ahead of it.
83_POD_TIMEOUT_SECONDS = 600
84#: One line per change of answer plus the closing lines; an answer that
85#: flapped on every sample for the whole budget prints more, and the tail
86#: keeps the end of that story.
87_LOG_TAIL_LINES = 40
88_LOG_LIMIT = 4_000
89_SAMPLE_PREFIX = "NETPOL_SAMPLE "
90_SETTLED_PREFIXES = ("NETPOL_SETTLED ", "NETPOL_UNSETTLED ")
93class NetworkPostureValidationError(RuntimeError):
94 """The live cluster does not enforce the documented network posture."""
97@dataclass(frozen=True)
98class ProbeSpec:
99 """One connection attempt and the verdict the shipped policies promise."""
101 name: str
102 client_namespace: str
103 url: str
104 expected: str
105 rule: str
106 #: A deny verdict exists only while the policy controller is on.
107 enforcement_only: bool
110def _probe_specs(
111 system_target_ip: str,
112 jobs_target_ip: str,
113 inference_monitor_ip: str,
114) -> tuple[ProbeSpec, ...]:
115 return (
116 ProbeSpec(
117 "same-namespace",
118 "gco-jobs",
119 f"http://{jobs_target_ip}:{_TARGET_PORT}/",
120 "reachable",
121 "gco-jobs/allow-same-namespace",
122 enforcement_only=False,
123 ),
124 ProbeSpec(
125 "cross-jobs",
126 _UNPOLICED_NAMESPACE,
127 f"http://{jobs_target_ip}:{_TARGET_PORT}/",
128 "blocked",
129 "gco-jobs/default-deny-ingress",
130 enforcement_only=True,
131 ),
132 ProbeSpec(
133 "cross-system",
134 "gco-jobs",
135 f"http://{system_target_ip}:{_TARGET_PORT}/",
136 "blocked",
137 "gco-system/default-deny-ingress",
138 enforcement_only=True,
139 ),
140 ProbeSpec(
141 "metrics-open",
142 _UNPOLICED_NAMESPACE,
143 f"http://{inference_monitor_ip}:{_INFERENCE_MONITOR_METRICS_PORT}/metrics",
144 "reachable",
145 "gco-system/allow-metrics-to-inference-monitor",
146 enforcement_only=False,
147 ),
148 ProbeSpec(
149 "https-egress",
150 "gco-jobs",
151 f"https://{_EGRESS_HOST}/",
152 "reachable",
153 "gco-jobs/allow-dns + gco-jobs/allow-https-egress",
154 enforcement_only=False,
155 ),
156 ProbeSpec(
157 "http-egress",
158 "gco-jobs",
159 f"http://{_EGRESS_HOST}/",
160 "blocked",
161 "gco-jobs egress (no rule admits port 80)",
162 enforcement_only=True,
163 ),
164 ProbeSpec(
165 "pod-identity-agent",
166 "gco-jobs",
167 _POD_IDENTITY_AGENT_URL,
168 "reachable",
169 "gco-jobs/allow-pod-identity-agent",
170 enforcement_only=False,
171 ),
172 )
175def _dict(value: Any) -> dict[str, Any]:
176 return value if isinstance(value, dict) else {}
179def _list(value: Any) -> list[Any]:
180 return value if isinstance(value, list) else []
183def _substitute(value: Any, replacements: dict[str, str]) -> Any:
184 """Replace manifest placeholders (``__PROBE_URL__``) wherever they appear."""
185 if isinstance(value, str):
186 for token, replacement in replacements.items():
187 value = value.replace(token, replacement)
188 return value
189 if isinstance(value, list):
190 return [_substitute(item, replacements) for item in value]
191 if isinstance(value, dict):
192 return {key: _substitute(item, replacements) for key, item in value.items()}
193 return value
196class NetworkPostureProbe:
197 """Run the probe matrix on one Region's cluster and clean up after it."""
199 def __init__(self, ctx: RunContext, region: str, kubectl: KubectlRunner) -> None:
200 self.ctx = ctx
201 self.region = region
202 self.kubectl = kubectl
203 self.token = _run_token(ctx.settings.run_id)
204 self.timeout = float(ctx.settings.command_timeout_seconds)
205 with ctx.state_lock:
206 self.record: dict[str, Any] = ctx.checkpoint.state.setdefault(
207 "network_posture", {}
208 ).setdefault(region, {})
209 self.record.setdefault("jobs", [])
211 # -- checkpointing -----------------------------------------------------
213 def _persist(self) -> None:
214 self.ctx.persist()
216 def _fail(self, message: str) -> NetworkPostureValidationError:
217 self.record["failure"] = message
218 self._persist()
219 return NetworkPostureValidationError(f"network posture in {self.region}: {message}")
221 # -- Job lifecycle -----------------------------------------------------
223 def _job_manifest(
224 self,
225 filename: str,
226 *,
227 name: str,
228 namespace: str,
229 replacements: dict[str, str] | None = None,
230 ) -> dict[str, Any]:
231 """Load one checked-in Job (run token applied) and give it this probe's identity."""
232 manifests, _name, _namespace = _load_manifest(self.ctx, filename)
233 job = copy.deepcopy(next(item for item in manifests if item.get("kind") == "Job"))
234 if replacements:
235 job = _substitute(job, replacements)
236 job["metadata"]["name"] = name
237 job["metadata"]["namespace"] = namespace
238 return job
240 def _run(self, *arguments: str, **kwargs: Any) -> tuple[int, str, str]:
241 code, stdout, stderr = self.kubectl(*arguments, timeout=self.timeout, **kwargs)
242 return code, stdout, stderr
244 def _create_job(self, job: dict[str, Any]) -> None:
245 namespace = job["metadata"]["namespace"]
246 name = job["metadata"]["name"]
247 # Record before creating: a Job that exists without a record could not
248 # be cleaned up, so the record is what authorizes the delete.
249 self.record["jobs"].append({"namespace": namespace, "name": name, "deleted": False})
250 self._persist()
251 # Start from a clean slate so a retry of this run never reads a stale
252 # verdict; foreground cascading waits for the old pod to be gone.
253 code, _stdout, stderr = self._run(
254 "delete",
255 "job",
256 name,
257 "--namespace",
258 namespace,
259 "--ignore-not-found",
260 "--cascade=foreground",
261 "--wait=true",
262 "--timeout=120s",
263 )
264 if code != 0:
265 self.record["last_kubectl_error"] = {
266 "argv": ["delete", "job", name],
267 "returncode": code,
268 "stderr": stderr[-_LOG_LIMIT:],
269 }
270 raise self._fail(f"could not clear a previous {namespace}/{name}")
271 code, _stdout, stderr = self._run(
272 "apply", "--namespace", namespace, "--filename", "-", input=json.dumps(job)
273 )
274 if code != 0:
275 self.record["last_kubectl_error"] = {
276 "argv": ["apply", namespace, name],
277 "returncode": code,
278 "stderr": stderr[-_LOG_LIMIT:],
279 }
280 raise self._fail(f"could not create {namespace}/{name}")
282 def _job_pod(self, namespace: str, name: str) -> dict[str, Any] | None:
283 payload = kubectl_json(
284 self.kubectl,
285 self.record,
286 "get",
287 "pods",
288 "--namespace",
289 namespace,
290 "--selector",
291 f"job-name={name}",
292 timeout=self.timeout,
293 )
294 for item in _list(_dict(payload).get("items")):
295 if not _dict(_dict(item).get("metadata")).get("deletionTimestamp"):
296 return _dict(item)
297 return None
299 def _wait(self, deadline: float, what: str) -> None:
300 if time.monotonic() >= deadline:
301 raise self._fail(f"{what} did not happen within {_POD_TIMEOUT_SECONDS}s")
302 time.sleep(self.ctx.settings.poll_interval_seconds)
304 def _wait_for_listener(self, namespace: str, name: str) -> tuple[str, str]:
305 """Return ``(pod name, pod IP)`` once the listener pod is Running and Ready."""
306 deadline = time.monotonic() + _POD_TIMEOUT_SECONDS
307 while True:
308 pod = self._job_pod(namespace, name)
309 status = _dict(pod.get("status")) if pod else {}
310 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))]
311 if status.get("phase") == "Failed":
312 raise self._fail(f"listener {namespace}/{name} failed before serving")
313 if (
314 pod is not None
315 and status.get("phase") == "Running"
316 and containers
317 and all(entry.get("ready") is True for entry in containers)
318 and status.get("podIP")
319 ):
320 return str(_dict(pod.get("metadata")).get("name")), str(status["podIP"])
321 self._wait(deadline, f"listener {namespace}/{name} readiness")
323 def _wait_for_verdict(self, namespace: str, name: str) -> dict[str, Any]:
324 """Return the probe pod's terminal phase, exit code, and log tail."""
325 deadline = time.monotonic() + _POD_TIMEOUT_SECONDS
326 while True:
327 pod = self._job_pod(namespace, name)
328 status = _dict(pod.get("status")) if pod else {}
329 phase = status.get("phase")
330 if phase in ("Succeeded", "Failed"):
331 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))]
332 terminated = (
333 _dict(_dict(containers[0].get("state")).get("terminated")) if containers else {}
334 )
335 exit_code = terminated.get("exitCode")
336 pod_name = str(_dict(_dict(pod).get("metadata")).get("name"))
337 _code, stdout, stderr = self._run(
338 "logs",
339 pod_name,
340 "--namespace",
341 namespace,
342 f"--tail={_LOG_TAIL_LINES}",
343 )
344 return {
345 "pod": pod_name,
346 "phase": phase,
347 "exit_code": exit_code,
348 "output": (stdout or stderr)[-_LOG_LIMIT:],
349 }
350 self._wait(deadline, f"probe {namespace}/{name} completion")
352 def _delete_jobs(self) -> list[dict[str, Any]]:
353 problems: list[dict[str, Any]] = []
354 for entry in self.record["jobs"]:
355 if entry.get("deleted"):
356 continue
357 code, _stdout, stderr = self._run(
358 "delete",
359 "job",
360 entry["name"],
361 "--namespace",
362 entry["namespace"],
363 "--ignore-not-found",
364 "--wait=false",
365 )
366 if code == 0:
367 entry["deleted"] = True
368 else:
369 problems.append({**entry, "returncode": code, "stderr": stderr[-_LOG_LIMIT:]})
370 self._persist()
371 return problems
373 # -- the matrix ---------------------------------------------------------
375 def _inference_monitor_ip(self) -> tuple[str, str]:
376 payload = kubectl_json(
377 self.kubectl,
378 self.record,
379 "get",
380 "pods",
381 "--namespace",
382 "gco-system",
383 "--selector",
384 "app=inference-monitor",
385 timeout=self.timeout,
386 )
387 for item in _list(_dict(payload).get("items")):
388 metadata = _dict(_dict(item).get("metadata"))
389 status = _dict(_dict(item).get("status"))
390 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))]
391 if (
392 not metadata.get("deletionTimestamp")
393 and status.get("phase") == "Running"
394 and containers
395 and all(entry.get("ready") is True for entry in containers)
396 and status.get("podIP")
397 ):
398 return str(metadata.get("name")), str(status["podIP"])
399 raise self._fail("no ready inference-monitor pod to probe")
401 @staticmethod
402 def _observed(verdict: dict[str, Any]) -> str:
403 if verdict["phase"] == "Succeeded" and verdict["exit_code"] == _REACHABLE_EXIT_CODE:
404 return "reachable"
405 if verdict["phase"] == "Failed" and verdict["exit_code"] == _BLOCKED_EXIT_CODE:
406 return "blocked"
407 if verdict["phase"] == "Failed" and verdict["exit_code"] == _UNSETTLED_EXIT_CODE:
408 return "unsettled"
409 return "error"
411 @staticmethod
412 def _trace(output: str) -> dict[str, Any]:
413 """Read the probe script's sampling trace out of its log tail.
415 ``samples`` holds every change of answer (``t=0s reachable``,
416 ``t=3s blocked``), ``settled`` the closing line, and
417 ``attach_window_observed`` is true when the first answer differed from
418 the steady state — on the VPC CNI, the standard-mode window between the
419 pod's start and its policies being attached.
420 """
421 samples: list[str] = []
422 settled: str | None = None
423 for line in output.splitlines():
424 if line.startswith(_SAMPLE_PREFIX):
425 samples.append(line.removeprefix(_SAMPLE_PREFIX))
426 elif line.startswith(_SETTLED_PREFIXES):
427 settled = line
428 return {
429 "samples": samples,
430 "settled": settled,
431 "attach_window_observed": len(samples) > 1,
432 }
434 def run(self) -> dict[str, Any]:
435 """Start the listeners, run every probe, delete everything, judge the matrix."""
436 enforcement = network_policy_enforcement_enabled(self.ctx)
437 self.record["enforcement_configured"] = enforcement
438 results: list[dict[str, Any]] = []
439 evidence: dict[str, Any] = {"enforcement_configured": enforcement, "probes": results}
440 try:
441 targets: dict[str, dict[str, str]] = {}
442 for namespace in ("gco-system", "gco-jobs"):
443 name = f"gco-live-netpol-target-{self.token}"
444 self._create_job(
445 self._job_manifest(_TARGET_MANIFEST, name=name, namespace=namespace)
446 )
447 pod_name, pod_ip = self._wait_for_listener(namespace, name)
448 targets[namespace] = {"job": name, "pod": pod_name, "ip": pod_ip}
449 evidence["targets"] = targets
450 monitor_pod, monitor_ip = self._inference_monitor_ip()
451 evidence["inference_monitor"] = {"pod": monitor_pod, "ip": monitor_ip}
452 self.record["targets"] = targets
453 self._persist()
455 specs = _probe_specs(targets["gco-system"]["ip"], targets["gco-jobs"]["ip"], monitor_ip)
456 launched: list[tuple[ProbeSpec, str]] = []
457 for spec in specs:
458 if spec.enforcement_only and not enforcement:
459 results.append(
460 {
461 **asdict(spec),
462 "status": "skipped",
463 "reason": (
464 "eks_cluster.network_policy_enforcement is false in cdk.json; "
465 "no deny verdict is promised"
466 ),
467 }
468 )
469 continue
470 name = f"gco-live-netpol-{spec.name}-{self.token}"
471 self._create_job(
472 self._job_manifest(
473 _PROBE_MANIFEST,
474 name=name,
475 namespace=spec.client_namespace,
476 replacements={"__PROBE_URL__": spec.url},
477 )
478 )
479 launched.append((spec, name))
480 for spec, name in launched:
481 verdict = self._wait_for_verdict(spec.client_namespace, name)
482 observed = self._observed(verdict)
483 results.append(
484 {
485 **asdict(spec),
486 **verdict,
487 **self._trace(verdict["output"]),
488 "job": name,
489 "observed": observed,
490 "status": "matched" if observed == spec.expected else "mismatch",
491 }
492 )
493 self.record["probes"] = results
494 self._persist()
495 finally:
496 try:
497 evidence["cleanup_problems"] = self._delete_jobs()
498 except Exception as exc: # a cleanup error must never mask the verdict
499 evidence["cleanup_problems"] = [{"error": f"{type(exc).__name__}: {exc}"}]
501 mismatches = [
502 f"{item['name']} ({item['client_namespace']} -> {item['url']}) expected "
503 f"{item['expected']}, observed {item['observed']} "
504 f"[phase={item['phase']} exit={item['exit_code']}]"
505 for item in results
506 if item["status"] == "mismatch"
507 ]
508 if mismatches:
509 raise self._fail("; ".join(mismatches))
510 if evidence["cleanup_problems"]:
511 raise self._fail(
512 f"{len(evidence['cleanup_problems'])} probe Job(s) could not be deleted"
513 )
514 self.record["evidence"] = evidence
515 self._persist()
516 return evidence