Coverage for scripts / live_release_validation / checks / network_posture.py: 100.00%

1"""Network posture probes: prove the shipped NetworkPolicies decide traffic live. 

2 

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: 

8 

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 ``https://…:9443/metrics``: 

15 **reachable** (``allow-metrics-to-inference-monitor`` admits exactly the 

16 metrics TLS sidecar's port, which is what lets Prometheus scrape it); 

17* ``default`` -> the same pod's plaintext ``:9090``: **blocked** (the monitor 

18 binds it to loopback, so only its TLS sidecar can reach it); 

19* ``default`` -> the live cost-monitor's ``https://…:8443``: **blocked** 

20 (``allow-manifest-processor-to-cost-monitor-ingress`` admits only the 

21 manifest processor), and its plaintext ``:8080``: **blocked** (loopback); 

22* ``default`` and ``gco-jobs`` -> a stand-in model pod in ``gco-inference`` on 

23 ``:8443``: **blocked** (``allow-inference-proxy-ingress`` admits only the 

24 inference proxy; the stand-in carries the ``gco.io/type: inference`` label 

25 those policies select); 

26* ``gco-jobs`` -> ``https://checkip.amazonaws.com/``: **reachable** 

27 (``allow-dns`` plus ``allow-https-egress``); 

28* ``gco-jobs`` -> ``http://checkip.amazonaws.com/`` on port 80: **blocked** 

29 (no rule admits it — the same host over 443 answered, so this is the policy 

30 deciding, not the network); and 

31* ``gco-jobs`` -> ``http://169.254.170.23/v1/credentials``: **reachable** 

32 (``allow-pod-identity-agent`` — the node's EKS Pod Identity Agent, which is 

33 where a ``gco-service-account`` pod gets its AWS credentials; the probe 

34 carries no token, so the agent answers 4xx, and any HTTP answer counts). 

35 Port 80 to the internet is blocked by the previous probe, so this one shows 

36 the rule admitting exactly the link-local endpoint. 

37 

38The one allowed caller of the cost monitor cannot be impersonated by a probe 

39pod without joining its Service, so that leg rides the service path instead: 

40``GET /api/v1/cost/status`` through the Region's API must answer 200, which 

41the manifest processor only does after reaching the cost monitor on 8443 over 

42verified TLS. The model pods' one allowed caller, the inference proxy, is 

43proved the same way by the ``inference`` action's served requests (their 

44Services expose only 8443). 

45 

46Every probe is a digest-pinned BusyBox Job labelled with this run's token 

47(``manifests/netpol-probe-job.yaml``); the listeners are the same image 

48behind ``httpd`` (``manifests/netpol-target-job.yaml`` in ``gco-system`` and 

49``gco-jobs``, ``manifests/netpol-model-target-job.yaml`` in ``gco-inference``). 

50All of them are deleted before the action returns, and each carries 

51``activeDeadlineSeconds`` plus a TTL so a harness that dies mid-probe still 

52leaves nothing behind. 

53 

54A probe's verdict is its steady state, not its first dial. The VPC CNI 

55attaches a new pod's policies in parallel with the pod's start and admits 

56everything until they are in place (standard mode; Auto Mode's NodeClass 

57``networkPolicy: DefaultDeny`` is the strict alternative, which requires a 

58policy for every pod on the node), so a client that dials once at start can 

59read that window instead of the policy — the fourth live run of this branch 

60saw exactly that on the port-80 egress probe. The probe script therefore 

61samples until one answer has held for 30 seconds and at least 45 seconds have 

62passed, exits 43 if the answer is still changing after three minutes, and 

63prints every change of answer; the harness keeps those lines as ``samples``, 

64the closing line as ``settled``, and flags ``attach_window_observed`` when the 

65first answer differed from the steady state. 

66 

67A disruption is not a verdict. Probe and listener pods ask Karpenter to leave 

68their node alone (``karpenter.sh/do-not-disrupt``) and prefer on-demand 

69capacity, but an interruption — a Spot reclaim, scheduled maintenance — can 

70still evict one: a live run lost the HTTPS egress probe 46 seconds into its 

71sampling and then waited out the whole pod timeout on a pod that no longer 

72existed. A probe whose Job failed with no terminated pod left, or whose pod 

73carries ``DisruptionTarget`` without a verdict exit code, is read at once and 

74re-run once from a fresh Job; a second disruption is reported as one. A 

75listener pod that did not last the matrix (any ``httpd`` target, or the 

76inference-monitor or cost-monitor pod a probe dialed) voids every verdict 

77dialed against it, so the action names it instead of reporting mismatches. 

78 

79When cdk.json turns the controller off (``eks_cluster.network_policy_enforcement: 

80false``) the deny verdicts are not promised and those probes are recorded as 

81skipped; the reachability probes and the two loopback probes (a bind address, 

82not a policy) still have to pass. With cost monitoring off there is no cost 

83monitor to dial and its probes are skipped with that reason. 

84""" 

85 

86from __future__ import annotations 

87 

88import copy 

89import json 

90import time 

91from dataclasses import asdict, dataclass 

92from typing import Any 

93 

94from ..context import _job_transport_region 

95from ..models import RunContext 

96from .cluster import KubectlRunner, kubectl_json 

97from .jobs import _load_manifest, _run_token 

98from .opencost import _cost_monitoring_configured 

99from .platform_workloads import network_policy_enforcement_enabled 

100 

101_TARGET_MANIFEST = "netpol-target-job.yaml" 

102_MODEL_TARGET_MANIFEST = "netpol-model-target-job.yaml" 

103_PROBE_MANIFEST = "netpol-probe-job.yaml" 

104_TARGET_PORT = 8080 

105#: The inference-monitor's metrics TLS sidecar; its app binds 9090 to loopback. 

106_INFERENCE_MONITOR_METRICS_PORT = 9443 

107_INFERENCE_MONITOR_LOOPBACK_PORT = 9090 

108#: The cost-monitor's TLS sidecar; its app binds 8080 to loopback. 

109_COST_MONITOR_PORT = 8443 

110_COST_MONITOR_LOOPBACK_PORT = 8080 

111#: The model pods' only Service port (their TLS sidecar). 

112_MODEL_PORT = 8443 

113_MODEL_NAMESPACE = "gco-inference" 

114_COST_STATUS_PATH = "/api/v1/cost/status" 

115_EGRESS_HOST = "checkip.amazonaws.com" 

116#: The EKS Pod Identity Agent's link-local address on every node; the 

117#: cluster points AWS_CONTAINER_CREDENTIALS_FULL_URI at this path. 

118_POD_IDENTITY_AGENT_URL = "http://169.254.170.23/v1/credentials" 

119#: A namespace the manifests leave unpoliced, so a client there proves the 

120#: target namespace's ingress rules and nothing else. 

121_UNPOLICED_NAMESPACE = "default" 

122_REACHABLE_EXIT_CODE = 0 

123_BLOCKED_EXIT_CODE = 42 

124#: The answer kept changing for the probe's whole budget; never a verdict. 

125_UNSETTLED_EXIT_CODE = 43 

126#: Exit codes the probe script chose; anything else means it never finished. 

127_SCRIPT_EXIT_CODES = frozenset({_REACHABLE_EXIT_CODE, _BLOCKED_EXIT_CODE, _UNSETTLED_EXIT_CODE}) 

128#: A probe a disruption voided is re-run from a fresh Job, once. 

129_PROBE_ATTEMPTS = 2 

130#: Auto Mode may have to launch a node for the first pod; a probe then still 

131#: has the image pull and its own three-minute sampling budget ahead of it. 

132_POD_TIMEOUT_SECONDS = 600 

133#: One line per change of answer plus the closing lines; an answer that 

134#: flapped on every sample for the whole budget prints more, and the tail 

135#: keeps the end of that story. 

136_LOG_TAIL_LINES = 40 

137_LOG_LIMIT = 4_000 

138_SAMPLE_PREFIX = "NETPOL_SAMPLE " 

139_SETTLED_PREFIXES = ("NETPOL_SETTLED ", "NETPOL_UNSETTLED ") 

140_COST_MONITORING_DISABLED = "cost monitoring is disabled in cdk.json; there is no cost monitor" 

141 

142 

143class NetworkPostureValidationError(RuntimeError): 

144 """The live cluster does not enforce the documented network posture.""" 

145 

146 

147@dataclass(frozen=True) 

148class ProbeSpec: 

149 """One connection attempt and the verdict the shipped policies promise.""" 

150 

151 name: str 

152 client_namespace: str 

153 url: str 

154 expected: str 

155 rule: str 

156 #: A deny verdict exists only while the policy controller is on. 

157 enforcement_only: bool 

158 #: Dials the cost monitor, which exists only with cost monitoring on. 

159 requires_cost_monitor: bool = False 

160 

161 

162def _probe_specs( 

163 system_target_ip: str, 

164 jobs_target_ip: str, 

165 inference_monitor_ip: str, 

166 *, 

167 model_target_ip: str, 

168 cost_monitor_ip: str | None, 

169) -> tuple[ProbeSpec, ...]: 

170 cost_host = cost_monitor_ip or "cost-monitor-not-deployed" 

171 return ( 

172 ProbeSpec( 

173 "same-namespace", 

174 "gco-jobs", 

175 f"http://{jobs_target_ip}:{_TARGET_PORT}/", 

176 "reachable", 

177 "gco-jobs/allow-same-namespace", 

178 enforcement_only=False, 

179 ), 

180 ProbeSpec( 

181 "cross-jobs", 

182 _UNPOLICED_NAMESPACE, 

183 f"http://{jobs_target_ip}:{_TARGET_PORT}/", 

184 "blocked", 

185 "gco-jobs/default-deny-ingress", 

186 enforcement_only=True, 

187 ), 

188 ProbeSpec( 

189 "cross-system", 

190 "gco-jobs", 

191 f"http://{system_target_ip}:{_TARGET_PORT}/", 

192 "blocked", 

193 "gco-system/default-deny-ingress", 

194 enforcement_only=True, 

195 ), 

196 ProbeSpec( 

197 "metrics-open", 

198 _UNPOLICED_NAMESPACE, 

199 f"https://{inference_monitor_ip}:{_INFERENCE_MONITOR_METRICS_PORT}/metrics", 

200 "reachable", 

201 "gco-system/allow-metrics-to-inference-monitor (metrics-tls-proxy, TCP 9443)", 

202 enforcement_only=False, 

203 ), 

204 ProbeSpec( 

205 "metrics-plaintext", 

206 _UNPOLICED_NAMESPACE, 

207 f"http://{inference_monitor_ip}:{_INFERENCE_MONITOR_LOOPBACK_PORT}/metrics", 

208 "blocked", 

209 "inference-monitor binds :9090 to 127.0.0.1 (METRICS_HOST)", 

210 enforcement_only=False, 

211 ), 

212 ProbeSpec( 

213 "cost-from-default", 

214 _UNPOLICED_NAMESPACE, 

215 f"https://{cost_host}:{_COST_MONITOR_PORT}/", 

216 "blocked", 

217 "gco-system/allow-manifest-processor-to-cost-monitor-ingress " 

218 "(manifest-processor only, TCP 8443)", 

219 enforcement_only=True, 

220 requires_cost_monitor=True, 

221 ), 

222 ProbeSpec( 

223 "cost-plaintext", 

224 _UNPOLICED_NAMESPACE, 

225 f"http://{cost_host}:{_COST_MONITOR_LOOPBACK_PORT}/", 

226 "blocked", 

227 "cost-monitor binds :8080 to 127.0.0.1 (HOST)", 

228 enforcement_only=False, 

229 requires_cost_monitor=True, 

230 ), 

231 ProbeSpec( 

232 "model-from-default", 

233 _UNPOLICED_NAMESPACE, 

234 f"http://{model_target_ip}:{_MODEL_PORT}/", 

235 "blocked", 

236 "gco-inference/allow-inference-proxy-ingress (inference-proxy only, TCP 8443)", 

237 enforcement_only=True, 

238 ), 

239 ProbeSpec( 

240 "model-from-jobs", 

241 "gco-jobs", 

242 f"http://{model_target_ip}:{_MODEL_PORT}/", 

243 "blocked", 

244 "gco-inference/allow-inference-proxy-ingress (inference-proxy only, TCP 8443)", 

245 enforcement_only=True, 

246 ), 

247 ProbeSpec( 

248 "https-egress", 

249 "gco-jobs", 

250 f"https://{_EGRESS_HOST}/", 

251 "reachable", 

252 "gco-jobs/allow-dns + gco-jobs/allow-https-egress", 

253 enforcement_only=False, 

254 ), 

255 ProbeSpec( 

256 "http-egress", 

257 "gco-jobs", 

258 f"http://{_EGRESS_HOST}/", 

259 "blocked", 

260 "gco-jobs egress (no rule admits port 80)", 

261 enforcement_only=True, 

262 ), 

263 ProbeSpec( 

264 "pod-identity-agent", 

265 "gco-jobs", 

266 _POD_IDENTITY_AGENT_URL, 

267 "reachable", 

268 "gco-jobs/allow-pod-identity-agent", 

269 enforcement_only=False, 

270 ), 

271 ) 

272 

273 

274#: The manifest processor's leg to the cost monitor, over the service path. 

275_COST_VIA_MANIFEST_PROCESSOR = ProbeSpec( 

276 "cost-via-manifest-processor", 

277 "gco-system", 

278 f"GET {_COST_STATUS_PATH} -> https://cost-monitor.gco-system.svc.cluster.local:8443", 

279 "reachable", 

280 "gco-system/allow-manifest-processor-to-cost-monitor-egress + -ingress (TCP 8443, " 

281 "verified TLS)", 

282 enforcement_only=False, 

283 requires_cost_monitor=True, 

284) 

285 

286 

287def _dict(value: Any) -> dict[str, Any]: 

288 return value if isinstance(value, dict) else {} 

289 

290 

291def _list(value: Any) -> list[Any]: 

292 return value if isinstance(value, list) else [] 

293 

294 

295def _substitute(value: Any, replacements: dict[str, str]) -> Any: 

296 """Replace manifest placeholders (``__PROBE_URL__``) wherever they appear.""" 

297 if isinstance(value, str): 

298 for token, replacement in replacements.items(): 

299 value = value.replace(token, replacement) 

300 return value 

301 if isinstance(value, list): 

302 return [_substitute(item, replacements) for item in value] 

303 if isinstance(value, dict): 

304 return {key: _substitute(item, replacements) for key, item in value.items()} 

305 return value 

306 

307 

308def _mismatch(item: dict[str, Any]) -> str: 

309 """One report line for a probe whose observed answer broke its promise.""" 

310 if item.get("via") == "api": 

311 detail = f"[HTTP {item['status_code']}]" 

312 else: 

313 detail = f"[phase={item['phase']} exit={item['exit_code']}]" 

314 line = ( 

315 f"{item['name']} ({item['client_namespace']} -> {item['url']}) expected " 

316 f"{item['expected']}, observed {item['observed']} {detail}" 

317 ) 

318 disruptions = item.get("disruptions") or [] 

319 if disruptions: 

320 line += f" after {len(disruptions)} disruption(s), last: {disruptions[-1]['reason']}" 

321 return line 

322 

323 

324class NetworkPostureProbe: 

325 """Run the probe matrix on one Region's cluster and clean up after it.""" 

326 

327 def __init__(self, ctx: RunContext, region: str, kubectl: KubectlRunner) -> None: 

328 self.ctx = ctx 

329 self.region = region 

330 self.kubectl = kubectl 

331 self.token = _run_token(ctx.settings.run_id) 

332 self.timeout = float(ctx.settings.command_timeout_seconds) 

333 with ctx.state_lock: 

334 self.record: dict[str, Any] = ctx.checkpoint.state.setdefault( 

335 "network_posture", {} 

336 ).setdefault(region, {}) 

337 self.record.setdefault("jobs", []) 

338 

339 # -- checkpointing ----------------------------------------------------- 

340 

341 def _persist(self) -> None: 

342 self.ctx.persist() 

343 

344 def _fail(self, message: str) -> NetworkPostureValidationError: 

345 self.record["failure"] = message 

346 self._persist() 

347 return NetworkPostureValidationError(f"network posture in {self.region}: {message}") 

348 

349 # -- Job lifecycle ----------------------------------------------------- 

350 

351 def _job_manifest( 

352 self, 

353 filename: str, 

354 *, 

355 name: str, 

356 namespace: str, 

357 replacements: dict[str, str] | None = None, 

358 ) -> dict[str, Any]: 

359 """Load one checked-in Job (run token applied) and give it this probe's identity.""" 

360 manifests, _name, _namespace = _load_manifest(self.ctx, filename) 

361 job = copy.deepcopy(next(item for item in manifests if item.get("kind") == "Job")) 

362 if replacements: 

363 job = _substitute(job, replacements) 

364 job["metadata"]["name"] = name 

365 job["metadata"]["namespace"] = namespace 

366 return job 

367 

368 def _run(self, *arguments: str, **kwargs: Any) -> tuple[int, str, str]: 

369 code, stdout, stderr = self.kubectl(*arguments, timeout=self.timeout, **kwargs) 

370 return code, stdout, stderr 

371 

372 def _create_job(self, job: dict[str, Any]) -> None: 

373 namespace = job["metadata"]["namespace"] 

374 name = job["metadata"]["name"] 

375 # Record before creating: a Job that exists without a record could not 

376 # be cleaned up, so the record is what authorizes the delete. A re-run 

377 # of a disrupted probe reuses its Job's record. 

378 entry = {"namespace": namespace, "name": name, "deleted": False} 

379 if entry not in self.record["jobs"]: 

380 self.record["jobs"].append(entry) 

381 self._persist() 

382 # Start from a clean slate so a retry of this run never reads a stale 

383 # verdict; foreground cascading waits for the old pod to be gone. 

384 code, _stdout, stderr = self._run( 

385 "delete", 

386 "job", 

387 name, 

388 "--namespace", 

389 namespace, 

390 "--ignore-not-found", 

391 "--cascade=foreground", 

392 "--wait=true", 

393 "--timeout=120s", 

394 ) 

395 if code != 0: 

396 self.record["last_kubectl_error"] = { 

397 "argv": ["delete", "job", name], 

398 "returncode": code, 

399 "stderr": stderr[-_LOG_LIMIT:], 

400 } 

401 raise self._fail(f"could not clear a previous {namespace}/{name}") 

402 code, _stdout, stderr = self._run( 

403 "apply", "--namespace", namespace, "--filename", "-", input=json.dumps(job) 

404 ) 

405 if code != 0: 

406 self.record["last_kubectl_error"] = { 

407 "argv": ["apply", namespace, name], 

408 "returncode": code, 

409 "stderr": stderr[-_LOG_LIMIT:], 

410 } 

411 raise self._fail(f"could not create {namespace}/{name}") 

412 

413 def _live_pods(self, namespace: str, selector: str) -> list[dict[str, Any]]: 

414 """The selector's pods that are not being deleted.""" 

415 payload = kubectl_json( 

416 self.kubectl, 

417 self.record, 

418 "get", 

419 "pods", 

420 "--namespace", 

421 namespace, 

422 "--selector", 

423 selector, 

424 timeout=self.timeout, 

425 ) 

426 return [ 

427 _dict(item) 

428 for item in _list(_dict(payload).get("items")) 

429 if not _dict(_dict(item).get("metadata")).get("deletionTimestamp") 

430 ] 

431 

432 def _job_pod(self, namespace: str, name: str) -> dict[str, Any] | None: 

433 pods = self._live_pods(namespace, f"job-name={name}") 

434 return pods[0] if pods else None 

435 

436 def _job_failure(self, namespace: str, name: str) -> dict[str, str] | None: 

437 """The Job's ``Failed`` condition as ``{reason, message}``, or ``None``.""" 

438 payload = kubectl_json( 

439 self.kubectl, 

440 self.record, 

441 "get", 

442 "job", 

443 name, 

444 "--namespace", 

445 namespace, 

446 timeout=self.timeout, 

447 ) 

448 for entry in _list(_dict(_dict(payload).get("status")).get("conditions")): 

449 condition = _dict(entry) 

450 if condition.get("type") == "Failed" and condition.get("status") == "True": 

451 return { 

452 "reason": str(condition.get("reason") or "Failed"), 

453 "message": str(condition.get("message") or "")[:_LOG_LIMIT], 

454 } 

455 return None 

456 

457 @staticmethod 

458 def _disruption_condition(pod: dict[str, Any]) -> dict[str, str] | None: 

459 """The pod's ``DisruptionTarget`` condition as ``{reason, message}``, or ``None``.""" 

460 for entry in _list(_dict(pod.get("status")).get("conditions")): 

461 condition = _dict(entry) 

462 if condition.get("type") == "DisruptionTarget" and condition.get("status") == "True": 

463 return { 

464 "reason": str(condition.get("reason") or "DisruptionTarget"), 

465 "message": str(condition.get("message") or "")[:_LOG_LIMIT], 

466 } 

467 return None 

468 

469 def _wait(self, deadline: float, what: str) -> None: 

470 if time.monotonic() >= deadline: 

471 raise self._fail(f"{what} did not happen within {_POD_TIMEOUT_SECONDS}s") 

472 time.sleep(self.ctx.settings.poll_interval_seconds) 

473 

474 def _wait_for_listener(self, namespace: str, name: str) -> tuple[str, str]: 

475 """Return ``(pod name, pod IP)`` once the listener pod is Running and Ready.""" 

476 deadline = time.monotonic() + _POD_TIMEOUT_SECONDS 

477 while True: 

478 pod = self._job_pod(namespace, name) 

479 status = _dict(pod.get("status")) if pod else {} 

480 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))] 

481 if status.get("phase") == "Failed": 

482 raise self._fail(f"listener {namespace}/{name} failed before serving") 

483 if ( 

484 pod is not None 

485 and status.get("phase") == "Running" 

486 and containers 

487 and all(entry.get("ready") is True for entry in containers) 

488 and status.get("podIP") 

489 ): 

490 return str(_dict(pod.get("metadata")).get("name")), str(status["podIP"]) 

491 self._wait(deadline, f"listener {namespace}/{name} readiness") 

492 

493 def _wait_for_verdict(self, namespace: str, name: str) -> dict[str, Any]: 

494 """Return the probe pod's terminal phase, exit code, log tail, and disruption. 

495 

496 ``disruption`` is ``None`` for a verdict. It holds the reason when a 

497 disruption took the verdict away: the Job failed with no terminated pod 

498 left to read (the pod was evicted and deleted), or the pod stopped with 

499 an exit code the script never chose while carrying ``DisruptionTarget``. 

500 Both return at once instead of waiting out the deadline. 

501 """ 

502 deadline = time.monotonic() + _POD_TIMEOUT_SECONDS 

503 while True: 

504 pod = self._job_pod(namespace, name) 

505 status = _dict(pod.get("status")) if pod else {} 

506 phase = status.get("phase") 

507 if pod is not None and phase in ("Succeeded", "Failed"): 

508 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))] 

509 terminated = ( 

510 _dict(_dict(containers[0].get("state")).get("terminated")) if containers else {} 

511 ) 

512 exit_code = terminated.get("exitCode") 

513 pod_name = str(_dict(pod.get("metadata")).get("name")) 

514 _code, stdout, stderr = self._run( 

515 "logs", 

516 pod_name, 

517 "--namespace", 

518 namespace, 

519 f"--tail={_LOG_TAIL_LINES}", 

520 ) 

521 disruption = ( 

522 None if exit_code in _SCRIPT_EXIT_CODES else self._disruption_condition(pod) 

523 ) 

524 return { 

525 "pod": pod_name, 

526 "phase": phase, 

527 "exit_code": exit_code, 

528 "output": (stdout or stderr)[-_LOG_LIMIT:], 

529 "disruption": disruption, 

530 } 

531 if pod is None and (failure := self._job_failure(namespace, name)) is not None: 

532 return { 

533 "pod": None, 

534 "phase": "Failed", 

535 "exit_code": None, 

536 "output": "", 

537 "disruption": failure, 

538 } 

539 self._wait(deadline, f"probe {namespace}/{name} completion") 

540 

541 def _probe_verdict( 

542 self, spec: ProbeSpec, job: dict[str, Any] 

543 ) -> tuple[dict[str, Any], list[dict[str, Any]]]: 

544 """Wait for one probe, re-running it from a fresh Job if a disruption voided it.""" 

545 namespace = job["metadata"]["namespace"] 

546 name = job["metadata"]["name"] 

547 disruptions: list[dict[str, Any]] = [] 

548 for attempt in range(1, _PROBE_ATTEMPTS + 1): 

549 verdict = self._wait_for_verdict(namespace, name) 

550 disruption = verdict.pop("disruption") 

551 if disruption is None: 

552 break 

553 disruptions.append({"attempt": attempt, "pod": verdict["pod"], **disruption}) 

554 self.record.setdefault("disruptions", []).append( 

555 {"probe": spec.name, "attempt": attempt, **disruption} 

556 ) 

557 self._persist() 

558 if attempt < _PROBE_ATTEMPTS: 

559 self._create_job(job) 

560 return verdict, disruptions 

561 

562 def _disrupted_listeners( 

563 self, 

564 targets: dict[str, dict[str, str]], 

565 platform_pods: list[tuple[str, str]], 

566 ) -> list[str]: 

567 """Listeners whose pod did not last the matrix; verdicts dialed at them are void. 

568 

569 The inference-monitor and cost-monitor pods the probes dialed are among 

570 them: a replaced pod answers on a different IP, so a probe that dialed 

571 the old one would read ``blocked`` for a reason no policy chose. 

572 """ 

573 dialed = [ 

574 (namespace, f"job-name={target['job']}", target["pod"], f"{namespace}/{target['job']}") 

575 for namespace, target in targets.items() 

576 ] 

577 dialed.extend( 

578 ("gco-system", f"app={app}", pod_name, f"gco-system/{pod_name}") 

579 for app, pod_name in platform_pods 

580 ) 

581 broken: list[str] = [] 

582 for namespace, selector, pod_name, label in dialed: 

583 pod = next( 

584 ( 

585 item 

586 for item in self._live_pods(namespace, selector) 

587 if _dict(item.get("metadata")).get("name") == pod_name 

588 ), 

589 None, 

590 ) 

591 if ( 

592 pod is None 

593 or _dict(pod.get("status")).get("phase") != "Running" 

594 or self._disruption_condition(pod) is not None 

595 ): 

596 broken.append(label) 

597 return broken 

598 

599 def _delete_jobs(self) -> list[dict[str, Any]]: 

600 problems: list[dict[str, Any]] = [] 

601 for entry in self.record["jobs"]: 

602 if entry.get("deleted"): 

603 continue 

604 code, _stdout, stderr = self._run( 

605 "delete", 

606 "job", 

607 entry["name"], 

608 "--namespace", 

609 entry["namespace"], 

610 "--ignore-not-found", 

611 "--wait=false", 

612 ) 

613 if code == 0: 

614 entry["deleted"] = True 

615 else: 

616 problems.append({**entry, "returncode": code, "stderr": stderr[-_LOG_LIMIT:]}) 

617 self._persist() 

618 return problems 

619 

620 # -- the matrix --------------------------------------------------------- 

621 

622 def _ready_platform_pod(self, app: str) -> tuple[str, str]: 

623 """Return ``(pod name, pod IP)`` of one Running, Ready ``gco-system`` pod of ``app``.""" 

624 for item in self._live_pods("gco-system", f"app={app}"): 

625 metadata = _dict(item.get("metadata")) 

626 status = _dict(item.get("status")) 

627 containers = [_dict(entry) for entry in _list(status.get("containerStatuses"))] 

628 if ( 

629 status.get("phase") == "Running" 

630 and containers 

631 and all(entry.get("ready") is True for entry in containers) 

632 and status.get("podIP") 

633 ): 

634 return str(metadata.get("name")), str(status["podIP"]) 

635 raise self._fail(f"no ready {app} pod to probe") 

636 

637 def _cost_via_manifest_processor(self) -> dict[str, Any]: 

638 """Prove the manifest processor reaches the cost monitor through the API.""" 

639 spec = asdict(_COST_VIA_MANIFEST_PROCESSOR) 

640 try: 

641 transport = _job_transport_region(self.ctx, self.region) 

642 except RuntimeError as exc: 

643 return {**spec, "status": "skipped", "reason": str(exc)} 

644 response = self.ctx.aws_client.make_authenticated_request( 

645 method="GET", 

646 path=_COST_STATUS_PATH, 

647 target_region=transport, 

648 ) 

649 observed = "reachable" if response.status_code == 200 else "error" 

650 return { 

651 **spec, 

652 "via": "api", 

653 "status_code": response.status_code, 

654 "observed": observed, 

655 "status": "matched" if observed == spec["expected"] else "mismatch", 

656 } 

657 

658 @staticmethod 

659 def _observed(verdict: dict[str, Any]) -> str: 

660 if verdict["phase"] == "Succeeded" and verdict["exit_code"] == _REACHABLE_EXIT_CODE: 

661 return "reachable" 

662 if verdict["phase"] == "Failed" and verdict["exit_code"] == _BLOCKED_EXIT_CODE: 

663 return "blocked" 

664 if verdict["phase"] == "Failed" and verdict["exit_code"] == _UNSETTLED_EXIT_CODE: 

665 return "unsettled" 

666 return "error" 

667 

668 @staticmethod 

669 def _trace(output: str) -> dict[str, Any]: 

670 """Read the probe script's sampling trace out of its log tail. 

671 

672 ``samples`` holds every change of answer (``t=0s reachable``, 

673 ``t=3s blocked``), ``settled`` the closing line, and 

674 ``attach_window_observed`` is true when the first answer differed from 

675 the steady state — on the VPC CNI, the standard-mode window between the 

676 pod's start and its policies being attached. 

677 """ 

678 samples: list[str] = [] 

679 settled: str | None = None 

680 for line in output.splitlines(): 

681 if line.startswith(_SAMPLE_PREFIX): 

682 samples.append(line.removeprefix(_SAMPLE_PREFIX)) 

683 elif line.startswith(_SETTLED_PREFIXES): 

684 settled = line 

685 return { 

686 "samples": samples, 

687 "settled": settled, 

688 "attach_window_observed": len(samples) > 1, 

689 } 

690 

691 def _start_listeners(self) -> dict[str, dict[str, str]]: 

692 """Start every ``httpd`` target and return ``{namespace: {job, pod, ip}}``.""" 

693 listeners = ( 

694 ("gco-system", _TARGET_MANIFEST, f"gco-live-netpol-target-{self.token}"), 

695 ("gco-jobs", _TARGET_MANIFEST, f"gco-live-netpol-target-{self.token}"), 

696 (_MODEL_NAMESPACE, _MODEL_TARGET_MANIFEST, f"gco-live-netpol-model-{self.token}"), 

697 ) 

698 targets: dict[str, dict[str, str]] = {} 

699 for namespace, manifest, name in listeners: 

700 self._create_job(self._job_manifest(manifest, name=name, namespace=namespace)) 

701 pod_name, pod_ip = self._wait_for_listener(namespace, name) 

702 targets[namespace] = {"job": name, "pod": pod_name, "ip": pod_ip} 

703 return targets 

704 

705 def run(self) -> dict[str, Any]: 

706 """Start the listeners, run every probe, delete everything, judge the matrix.""" 

707 enforcement = network_policy_enforcement_enabled(self.ctx) 

708 cost_monitoring = _cost_monitoring_configured(self.ctx) 

709 self.record["enforcement_configured"] = enforcement 

710 results: list[dict[str, Any]] = [] 

711 evidence: dict[str, Any] = { 

712 "enforcement_configured": enforcement, 

713 "cost_monitoring_configured": cost_monitoring, 

714 "probes": results, 

715 } 

716 try: 

717 targets = self._start_listeners() 

718 evidence["targets"] = targets 

719 self.record["targets"] = targets 

720 monitor_pod, monitor_ip = self._ready_platform_pod("inference-monitor") 

721 evidence["inference_monitor"] = {"pod": monitor_pod, "ip": monitor_ip} 

722 platform_pods = [("inference-monitor", monitor_pod)] 

723 cost_monitor_ip: str | None = None 

724 if cost_monitoring: 

725 cost_pod, cost_monitor_ip = self._ready_platform_pod("cost-monitor") 

726 evidence["cost_monitor"] = {"pod": cost_pod, "ip": cost_monitor_ip} 

727 platform_pods.append(("cost-monitor", cost_pod)) 

728 self._persist() 

729 

730 specs = _probe_specs( 

731 targets["gco-system"]["ip"], 

732 targets["gco-jobs"]["ip"], 

733 monitor_ip, 

734 model_target_ip=targets[_MODEL_NAMESPACE]["ip"], 

735 cost_monitor_ip=cost_monitor_ip, 

736 ) 

737 launched: list[tuple[ProbeSpec, dict[str, Any]]] = [] 

738 for spec in specs: 

739 if spec.requires_cost_monitor and not cost_monitoring: 

740 results.append( 

741 {**asdict(spec), "status": "skipped", "reason": _COST_MONITORING_DISABLED} 

742 ) 

743 continue 

744 if spec.enforcement_only and not enforcement: 

745 results.append( 

746 { 

747 **asdict(spec), 

748 "status": "skipped", 

749 "reason": ( 

750 "eks_cluster.network_policy_enforcement is false in cdk.json; " 

751 "no deny verdict is promised" 

752 ), 

753 } 

754 ) 

755 continue 

756 job = self._job_manifest( 

757 _PROBE_MANIFEST, 

758 name=f"gco-live-netpol-{spec.name}-{self.token}", 

759 namespace=spec.client_namespace, 

760 replacements={"__PROBE_URL__": spec.url}, 

761 ) 

762 self._create_job(job) 

763 launched.append((spec, job)) 

764 for spec, job in launched: 

765 verdict, disruptions = self._probe_verdict(spec, job) 

766 observed = self._observed(verdict) 

767 results.append( 

768 { 

769 **asdict(spec), 

770 **verdict, 

771 **self._trace(verdict["output"]), 

772 "job": job["metadata"]["name"], 

773 "disruptions": disruptions, 

774 "observed": observed, 

775 "status": "matched" if observed == spec.expected else "mismatch", 

776 } 

777 ) 

778 self.record["probes"] = results 

779 self._persist() 

780 if cost_monitoring: 

781 results.append(self._cost_via_manifest_processor()) 

782 else: 

783 results.append( 

784 { 

785 **asdict(_COST_VIA_MANIFEST_PROCESSOR),