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.platform_workloads import network_policy_enforcement_enabled 

10from ..models import RunContext 

11 

12 

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

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

15 

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

17 digest-pinned listener in ``gco-system`` and one in ``gco-jobs``, then dial 

18 them (and the live inference-monitor's metrics port, and an AWS-hosted 

19 HTTPS endpoint) from throwaway client pods whose exit code is the verdict: 

20 same-namespace job traffic, the metrics port, and HTTPS egress must be 

21 reachable; cross-namespace ingress into either namespace and non-443 

22 egress from ``gco-jobs`` must be blocked. Every probe and listener is a 

23 run-labelled Job deleted before the action returns. 

24 

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

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

27 reachability probes still have to pass. 

28 """ 

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

30 for region in ctx.deployment_regions: 

31 with cluster_kubectl(ctx, region) as kubectl: 

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

33 return { 

34 "network_policy_enforcement": network_policy_enforcement_enabled(ctx), 

35 "regions": regions, 

36 }