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

1"""network-posture: the shipped NetworkPolicies decide traffic on the live cluster.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ..checks.cluster import cluster_kubectl 

8from ..checks.network_posture import NetworkPostureProbe 

9from ..checks.opencost import _cost_monitoring_configured 

10from ..checks.platform_workloads import network_policy_enforcement_enabled 

11from ..models import RunContext 

12 

13 

14def action_network_posture(ctx: RunContext) -> dict[str, Any]: 

15 """Require every Region's cluster to enforce the documented zero-trust posture. 

16 

17 For every deployed Region, open the tunnelled kubectl session, start one 

18 digest-pinned listener in ``gco-system``, one in ``gco-jobs``, and a 

19 stand-in model pod in ``gco-inference``, then dial them (and the live 

20 inference-monitor and cost-monitor, and an AWS-hosted HTTPS endpoint) from 

21 throwaway client pods whose exit code is the verdict: same-namespace job 

22 traffic, the inference-monitor's TLS metrics port 9443, and HTTPS egress 

23 must be reachable; cross-namespace ingress into either namespace, the 

24 cost monitor's 8443 from anything but the manifest processor, the model 

25 port 8443 from anything but the inference proxy, and non-443 egress from 

26 ``gco-jobs`` must be blocked; and the loopback-bound plaintext ports 

27 (inference-monitor 9090, cost-monitor 8080) must not answer on the pod 

28 network. ``GET /api/v1/cost/status`` must answer 200, proving the manifest 

29 processor reaches the cost monitor on 8443. Every probe and listener is a 

30 run-labelled Job deleted before the action returns. 

31 

32 When cdk.json sets ``eks_cluster.network_policy_enforcement: false`` the 

33 deny verdicts are recorded as skipped — that mode promises none — and the 

34 reachability and loopback probes still have to pass; with cost monitoring 

35 off the cost-monitor probes are skipped with that reason. 

36 """ 

37 regions: dict[str, Any] = {} 

38 for region in ctx.deployment_regions: 

39 with cluster_kubectl(ctx, region) as kubectl: 

40 regions[region] = NetworkPostureProbe(ctx, region, kubectl).run() 

41 return { 

42 "network_policy_enforcement": network_policy_enforcement_enabled(ctx), 

43 "cost_monitoring": _cost_monitoring_configured(ctx), 

44 "regions": regions, 

45 }