From e5df95c2591843e5b81cf4414670d97758f1e5c5 Mon Sep 17 00:00:00 2001 From: Jonathan Nguyen <87379568+tamg-aws@users.noreply.github.com> Date: Tue, 1 Sep 2026 06:40:45 -0500 Subject: [PATCH] feat(eks): assess Kubernetes network policy enforcement in the Amazon VPC CNI add-on (#12661) --- ...ks-cluster-vpc-cni-network-policy.added.md | 1 + .../aws/aws_ai_security_framework_aws.json | 1 + .../__init__.py | 0 ..._cni_network_policy_enforced.metadata.json | 42 +++ ...cluster_vpc_cni_network_policy_enforced.py | 129 ++++++++ .../providers/aws/services/eks/eks_service.py | 64 ++++ ...er_vpc_cni_network_policy_enforced_test.py | 307 ++++++++++++++++++ .../aws/services/eks/eks_service_test.py | 156 +++++++++ 8 files changed, 700 insertions(+) create mode 100644 prowler/changelog.d/eks-cluster-vpc-cni-network-policy.added.md create mode 100644 prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/__init__.py create mode 100644 prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.metadata.json create mode 100644 prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.py create mode 100644 tests/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced_test.py diff --git a/prowler/changelog.d/eks-cluster-vpc-cni-network-policy.added.md b/prowler/changelog.d/eks-cluster-vpc-cni-network-policy.added.md new file mode 100644 index 0000000000..7002eae1d9 --- /dev/null +++ b/prowler/changelog.d/eks-cluster-vpc-cni-network-policy.added.md @@ -0,0 +1 @@ +`eks_cluster_vpc_cni_network_policy_enforced` check for AWS provider, flagging EKS clusters whose Amazon VPC CNI managed add-on does not enable Kubernetes network policy enforcement, and reporting MANUAL where the EKS API cannot show the setting diff --git a/prowler/compliance/aws/aws_ai_security_framework_aws.json b/prowler/compliance/aws/aws_ai_security_framework_aws.json index 205d5902d1..666530ef53 100644 --- a/prowler/compliance/aws/aws_ai_security_framework_aws.json +++ b/prowler/compliance/aws/aws_ai_security_framework_aws.json @@ -1130,6 +1130,7 @@ "eks_cluster_not_publicly_accessible", "eks_cluster_private_nodes_enabled", "eks_cluster_network_policy_enabled", + "eks_cluster_vpc_cni_network_policy_enforced", "eks_cluster_uses_a_supported_version", "eks_control_plane_logging_all_types_enabled", "eks_cluster_kms_cmk_encryption_in_secrets_enabled", diff --git a/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/__init__.py b/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.metadata.json b/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.metadata.json new file mode 100644 index 0000000000..f9b4bb0b30 --- /dev/null +++ b/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.metadata.json @@ -0,0 +1,42 @@ +{ + "Provider": "aws", + "CheckID": "eks_cluster_vpc_cni_network_policy_enforced", + "CheckTitle": "EKS cluster enforces Kubernetes network policies through the Amazon VPC CNI add-on", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices/Network Reachability", + "TTPs/Lateral Movement" + ], + "ServiceName": "eks", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "AwsEksCluster", + "ResourceGroup": "container", + "Description": "**Amazon EKS clusters** are evaluated for whether the **Amazon VPC CNI** managed add-on sets `enableNetworkPolicy` to `true`, which is what makes the CNI enforce Kubernetes `NetworkPolicy` resources. The policy objects themselves live in the cluster and are not exposed by the EKS API, so only this enforcement precondition is verified.", + "Risk": "Without CNI **network policy enforcement** every `NetworkPolicy` an operator authors is inert, so pods reach every other pod and service in the cluster. That unrestricted east-west path lets one compromised container move **laterally** to sidecars, tool executors and internal APIs, widening the blast radius and easing **data exfiltration**.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/eks/latest/userguide/cni-network-policy.html", + "https://docs.aws.amazon.com/eks/latest/APIReference/API_UpdateAddon.html", + "https://docs.aws.amazon.com/eks/latest/userguide/updating-an-add-on.html" + ], + "Remediation": { + "Code": { + "CLI": "aws eks update-addon --cluster-name --addon-name vpc-cni --resolve-conflicts PRESERVE --configuration-values '{\"enableNetworkPolicy\":\"true\"}'", + "NativeIaC": "```yaml\n# CloudFormation: enable network policy enforcement in the VPC CNI add-on\nResources:\n :\n Type: AWS::EKS::Addon\n Properties:\n ClusterName: \n AddonName: vpc-cni\n ResolveConflicts: PRESERVE\n ConfigurationValues: '{\"enableNetworkPolicy\":\"true\"}' # critical: makes the CNI enforce NetworkPolicy resources\n```", + "Other": "1. Open the AWS Console and go to EKS > Clusters\n2. Select and open the Add-ons tab\n3. Select the Amazon VPC CNI add-on and click Edit\n4. Expand Optional configuration settings and set enableNetworkPolicy to \"true\" in the configuration values\n5. Click Save changes", + "Terraform": "```hcl\n# Enable network policy enforcement in the VPC CNI managed add-on\nresource \"aws_eks_addon\" \"\" {\n cluster_name = \"\"\n addon_name = \"vpc-cni\"\n resolve_conflicts_on_update = \"PRESERVE\"\n\n configuration_values = jsonencode({\n enableNetworkPolicy = \"true\" # critical: makes the CNI enforce NetworkPolicy resources\n })\n}\n```" + }, + "Recommendation": { + "Text": "Set `enableNetworkPolicy` to `true` on the Amazon VPC CNI add-on, then author `NetworkPolicy` resources that allow each workload only its declared dependencies, ideally with `strict` enforcement mode so traffic is denied until the policy is in place. Layer security groups for Pods to reach VPC resources such as databases.", + "Url": "https://hub.prowler.com/check/eks_cluster_vpc_cni_network_policy_enforced" + } + }, + "Categories": [ + "trust-boundaries", + "cluster-security" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "The EKS API exposes only the add-on setting, not the Kubernetes NetworkPolicy resources, so a PASS is the enforcement precondition rather than proof that pod-to-pod traffic is restricted. A FAIL is a statement about the managed add-on, not about the cluster: two documented architectures enforce network policies with this setting false and neither is visible to the EKS API. A third-party policy engine -- the EKS Best Practices Guide recommends Calico and Cilium in its 'ThirdParty Network Policy Engines' section (eks/latest/best-practices/network-security.html) -- would correctly leave it false, since two enforcers are not run together. And a self-managed VPC CNI can be enabled by Helm or by the amazon-vpc-cni ConfigMap key enable-network-policy-controller with the aws-node DaemonSet, two of the three paths AWS documents (eks/latest/userguide/cni-network-policy-configure.html); only the managed add-on path is readable here. A cluster with no vpc-cni managed add-on at all reports MANUAL for the same reason. Detecting Calico or Cilium is deliberately not attempted, because any signal would be a guess presented as a measurement." +} diff --git a/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.py b/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.py new file mode 100644 index 0000000000..cd634c9278 --- /dev/null +++ b/prowler/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced.py @@ -0,0 +1,129 @@ +import json +from typing import Optional + +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.eks.eks_client import eks_client + +VPC_CNI_ADDON_NAME = "vpc-cni" +NETWORK_POLICY_KEY = "enableNetworkPolicy" + + +def parse_configuration_values(configuration_values: Optional[str]) -> Optional[dict]: + """Decode an EKS add-on `configurationValues` blob. + + Args: + configuration_values: The raw JSON string returned by DescribeAddon, which is + absent when no configuration has been supplied for the add-on. + + Returns: + The decoded mapping, an empty mapping when nothing was supplied, or None when + the blob is not a readable JSON object. + """ + if not configuration_values: + return {} + try: + configuration = json.loads(configuration_values) + except ValueError: + return None + return configuration if isinstance(configuration, dict) else None + + +def boolean_configuration_value(value: object) -> Optional[bool]: + """Read a VPC CNI boolean setting. + + The add-on configuration schema types these as a string carrying `"format": "boolean"`, + so the API returns `"true"` rather than `true`; a JSON boolean is accepted as well + because the schema is add-on-version specific and this blob is otherwise untyped. + + Args: + value: The value found in the add-on configuration, if any. + + Returns: + The boolean it denotes, or None when it is absent or not a recognized boolean. + """ + if isinstance(value, bool): + return value + if isinstance(value, str) and value.strip().lower() in ("true", "false"): + return value.strip().lower() == "true" + return None + + +class eks_cluster_vpc_cni_network_policy_enforced(Check): + """Ensure the Amazon VPC CNI add-on enforces Kubernetes network policies. + + Kubernetes NetworkPolicy resources live in the cluster and are not exposed by the + EKS API. What the API does expose is whether the Amazon VPC CNI managed add-on has + network policy enforcement switched on, which is the precondition for any + NetworkPolicy to take effect. + - PASS: The Amazon VPC CNI add-on sets enableNetworkPolicy to true. + - FAIL: The Amazon VPC CNI add-on sets enableNetworkPolicy to false. + - MANUAL: The setting cannot be read, or the cluster does not use the Amazon VPC CNI + managed add-on. + + TWO WAYS A CLUSTER CAN ENFORCE NETWORK POLICIES WITHOUT THIS SETTING BEING TRUE, so a + FAIL is a statement about the managed add-on and not about the cluster. Neither is + fixable by reading more of the AWS API: both live in in-cluster state that + DescribeAddon cannot see. + + 1. A third-party policy engine. The EKS Best Practices Guide recommends Calico and + Cilium for requirements the VPC CNI does not cover, such as Layer 7 and DNS + hostname rules, in its "ThirdParty Network Policy Engines" section + (https://docs.aws.amazon.com/eks/latest/best-practices/network-security.html). + A cluster enforcing through one of those would correctly leave this setting false, + because two enforcers are not run together. + 2. A self-managed VPC CNI. AWS documents three ways to enable the feature and only the + first is visible here + (https://docs.aws.amazon.com/eks/latest/userguide/cni-network-policy-configure.html): + `aws eks update-addon --addon-name vpc-cni` with configurationValues; `helm upgrade + ... aws-vpc-cni`; or the `amazon-vpc-cni` ConfigMap key + `enable-network-policy-controller: "true"` together with policy enforcement in the + aws-node container of the VPC CNI DaemonSet. The second and third leave the EKS + control plane with nothing to report. + + Detecting either is deliberately NOT attempted. Calico and Cilium are invisible to the + AWS API, so any signal would be a guess presented as a measurement. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + for cluster in eks_client.clusters: + report = Check_Report_AWS(metadata=self.metadata(), resource=cluster) + report.status = "MANUAL" + addon = cluster.addons.get(VPC_CNI_ADDON_NAME) + + if addon is None and cluster.addons_discovery_failed: + report.status_extended = f"EKS cluster {cluster.name} add-ons could not be listed, so Kubernetes network policy enforcement in the Amazon VPC CNI add-on cannot be determined." + elif addon is None: + report.status_extended = f"EKS cluster {cluster.name} does not use the Amazon VPC CNI managed add-on, so Kubernetes network policy enforcement cannot be determined from the EKS API. Review the self-managed CNI configuration in the cluster." + elif addon.configuration_discovery_failed: + report.status_extended = f"EKS cluster {cluster.name} Amazon VPC CNI add-on configuration could not be read, so Kubernetes network policy enforcement cannot be determined." + else: + configuration = parse_configuration_values(addon.configuration_values) + if configuration is None: + report.status_extended = f"EKS cluster {cluster.name} Amazon VPC CNI add-on configuration values are not a readable JSON object, so Kubernetes network policy enforcement cannot be determined." + else: + network_policy_enabled = boolean_configuration_value( + configuration.get(NETWORK_POLICY_KEY) + ) + if network_policy_enabled is None: + report.status_extended = f"EKS cluster {cluster.name} Amazon VPC CNI add-on does not set {NETWORK_POLICY_KEY} to true or false, so Kubernetes network policy enforcement cannot be determined." + elif network_policy_enabled: + report.status = "PASS" + report.status_extended = f"EKS cluster {cluster.name} enforces Kubernetes network policies through the Amazon VPC CNI add-on. This does not confirm that NetworkPolicy resources restricting pod-to-pod traffic exist in the cluster." + else: + # States the measurement, not the inference from it. The previous wording -- + # "cluster does not enforce Kubernetes network policies" -- claimed a cluster + # property from an add-on setting, and is false for a cluster enforcing + # through Calico or Cilium, or through a self-managed VPC CNI. Both are + # documented architectures rather than edge cases; see the class docstring. + report.status = "FAIL" + report.status_extended = f"EKS cluster {cluster.name} Amazon VPC CNI managed add-on does not enforce Kubernetes network policies, since it sets {NETWORK_POLICY_KEY} to false. Enforcement by a third-party policy engine or a self-managed VPC CNI is not visible to the EKS API and is not evaluated." + + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/eks/eks_service.py b/prowler/providers/aws/services/eks/eks_service.py index b0ccace49b..ef06feb823 100644 --- a/prowler/providers/aws/services/eks/eks_service.py +++ b/prowler/providers/aws/services/eks/eks_service.py @@ -6,14 +6,19 @@ from prowler.lib.logger import logger from prowler.lib.scan_filters.scan_filters import is_resource_filtered from prowler.providers.aws.lib.service.service import AWSService +# DescribeAddon has no batch form, so only add-ons a check reads are described. +COLLECTED_ADDONS = ("vpc-cni",) + class EKS(AWSService): def __init__(self, provider): + """Collect the audited account's EKS clusters, their configuration and their add-ons.""" # Call AWSService's __init__ super().__init__(__class__.__name__, provider) self.clusters = [] self.__threading_call__(self._list_clusters) self._describe_cluster(self.regional_clients) + self.__threading_call__(self._describe_cluster_addons, self.clusters) def _list_clusters(self, regional_client): logger.info("EKS listing clusters...") @@ -95,12 +100,69 @@ class EKS(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + def _describe_cluster_addons(self, cluster): + """Attach the add-ons named in COLLECTED_ADDONS, with their configuration, to a cluster. + + ListAddons names every add-on installed on the cluster and DescribeAddon then supplies + the ARN and the raw `configurationValues` blob for the ones checks read. A failed listing + sets `addons_discovery_failed` on the cluster and a failed describe sets + `configuration_discovery_failed` on the add-on, so a check can tell an add-on that is + absent from one whose state could not be read instead of reporting both as absent. + """ + logger.info("EKS describing cluster add-ons...") + try: + regional_client = self.regional_clients[cluster.region] + list_addons_paginator = regional_client.get_paginator("list_addons") + for page in list_addons_paginator.paginate(clusterName=cluster.name): + for addon_name in page["addons"]: + if addon_name in COLLECTED_ADDONS: + cluster.addons[addon_name] = EKSAddon(name=addon_name) + except Exception as error: + cluster.addons_discovery_failed = True + logger.error( + f"{cluster.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return + + for addon in cluster.addons.values(): + try: + describe_addon = regional_client.describe_addon( + clusterName=cluster.name, addonName=addon.name + ) + addon.arn = describe_addon["addon"].get("addonArn") + addon.configuration_values = describe_addon["addon"].get( + "configurationValues" + ) + except Exception as error: + addon.configuration_discovery_failed = True + logger.error( + f"{cluster.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + class EKSClusterLoggingEntity(BaseModel): types: list[str] = None enabled: bool = None +class EKSAddon(BaseModel): + """An EKS managed add-on, with the configuration values collected for it. + + Attributes: + name: The add-on name as returned by ListAddons. + arn: The add-on ARN, absent when DescribeAddon could not be read. + configuration_values: The raw JSON blob supplied for the add-on, absent when none + was supplied or when DescribeAddon could not be read. + configuration_discovery_failed: True when DescribeAddon failed, so a check can + tell a setting that is unset from one that could not be read. + """ + + name: str + arn: Optional[str] = None + configuration_values: Optional[str] = None + configuration_discovery_failed: bool = False + + class EKSCluster(BaseModel): name: str arn: str @@ -113,4 +175,6 @@ class EKSCluster(BaseModel): public_access_cidrs: list[str] = [] encryptionConfig: bool = None deletion_protection: bool = None + addons: dict[str, EKSAddon] = {} + addons_discovery_failed: bool = False tags: Optional[list] = [] diff --git a/tests/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced_test.py b/tests/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced_test.py new file mode 100644 index 0000000000..56b9990afe --- /dev/null +++ b/tests/providers/aws/services/eks/eks_cluster_vpc_cni_network_policy_enforced/eks_cluster_vpc_cni_network_policy_enforced_test.py @@ -0,0 +1,307 @@ +from unittest import mock + +import pytest + +from prowler.providers.aws.services.eks.eks_service import EKSAddon, EKSCluster +from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 + +cluster_name = "cluster_test" +cluster_arn = ( + f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster/{cluster_name}" +) +addon_arn = f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:addon/{cluster_name}/vpc-cni/1a2b3c4d" + + +def build_cluster(name=cluster_name, arn=cluster_arn, **kwargs): + """Build an EKSCluster whose add-on state the caller supplies through kwargs. + + The defaults leave `addons` empty and both discovery flags false, which is the shape of a + cluster that carries no managed add-ons rather than one whose add-ons could not be read. + """ + return EKSCluster(name=name, arn=arn, region=AWS_REGION_EU_WEST_1, **kwargs) + + +def vpc_cni_addon(configuration_values=None, configuration_discovery_failed=False): + """Build the `addons` mapping for a cluster carrying the vpc-cni managed add-on. + + `configuration_values` is passed through as the raw JSON string DescribeAddon returns, so a + test can supply the exact blob the API would, including an absent or malformed one. + """ + return { + "vpc-cni": EKSAddon( + name="vpc-cni", + arn=addon_arn, + configuration_values=configuration_values, + configuration_discovery_failed=configuration_discovery_failed, + ) + } + + +def run_check(clusters): + """Execute the check against the given clusters and return its reports. + + The clusters are model objects, so the reports exercise the check's own branching over + already-collected state and no EKS API call takes place. + """ + eks_client = mock.MagicMock + eks_client.clusters = clusters + with mock.patch( + "prowler.providers.aws.services.eks.eks_service.EKS", + eks_client, + ): + from prowler.providers.aws.services.eks.eks_cluster_vpc_cni_network_policy_enforced.eks_cluster_vpc_cni_network_policy_enforced import ( + eks_cluster_vpc_cni_network_policy_enforced, + ) + + return eks_cluster_vpc_cni_network_policy_enforced().execute() + + +class Test_eks_cluster_vpc_cni_network_policy_enforced: + def test_no_clusters(self): + """An account with no EKS clusters must produce no reports at all.""" + assert len(run_check([])) == 0 + + def test_addons_discovery_failed(self): + """A cluster whose ListAddons call failed must be MANUAL, not FAIL. + + The add-on mapping is empty in both this case and the no-managed-add-on case, so the + cluster-level flag is what separates "unknown" from "not installed". + """ + result = run_check([build_cluster(addons_discovery_failed=True)]) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} add-ons could not be listed, so Kubernetes " + "network policy enforcement in the Amazon VPC CNI add-on cannot be determined." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + def test_no_vpc_cni_addon(self): + """A cluster with managed add-ons but no vpc-cni must be MANUAL and name the CNI. + + The EKS API exposes nothing about a CNI it does not manage, so the verdict cannot be + FAIL: the cluster may well enforce network policies through a self-managed CNI. + """ + result = run_check( + [build_cluster(addons={"coredns": EKSAddon(name="coredns")})] + ) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} does not use the Amazon VPC CNI managed " + "add-on, so Kubernetes network policy enforcement cannot be determined " + "from the EKS API. Review the self-managed CNI configuration in the cluster." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + def test_addons_listed_but_empty(self): + """A cluster whose add-ons listed successfully as empty must get the not-installed wording. + + Same MANUAL verdict as a failed listing but a different explanation, so the report does + not tell an operator to fix permissions when the add-on is simply not there. + """ + result = run_check([build_cluster(addons={})]) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} does not use the Amazon VPC CNI managed " + "add-on, so Kubernetes network policy enforcement cannot be determined " + "from the EKS API. Review the self-managed CNI configuration in the cluster." + ) + + def test_addon_configuration_unreadable(self): + """A vpc-cni add-on whose DescribeAddon call failed must be MANUAL. + + The add-on is known to be installed, but its configuration was never read, so network + policy enforcement is undetermined rather than off. + """ + result = run_check( + [build_cluster(addons=vpc_cni_addon(configuration_discovery_failed=True))] + ) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} Amazon VPC CNI add-on configuration could " + "not be read, so Kubernetes network policy enforcement cannot be determined." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + def test_addon_configuration_unreadable_wins_over_stale_values(self): + """A failed describe must stay MANUAL even when the model still carries a true setting. + + Guards the branch order: reading `configuration_values` before checking the failure flag + would report PASS from a value the failed call did not return. + """ + result = run_check( + [ + build_cluster( + addons=vpc_cni_addon( + configuration_values='{"enableNetworkPolicy":"true"}', + configuration_discovery_failed=True, + ) + ) + ] + ) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} Amazon VPC CNI add-on configuration could " + "not be read, so Kubernetes network policy enforcement cannot be determined." + ) + + @pytest.mark.parametrize( + "configuration_values", + ['{"enableNetworkPolicy": "true"', '["enableNetworkPolicy"]', "true", "42"], + ) + def test_addon_configuration_not_a_json_object(self, configuration_values): + """Configuration values that do not decode to a JSON object must be MANUAL. + + Truncated JSON, an array, a bare boolean and a bare number each reach a different line of + the decoder, and none may raise out of the check or be read as an empty configuration. + """ + result = run_check([build_cluster(addons=vpc_cni_addon(configuration_values))]) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} Amazon VPC CNI add-on configuration values " + "are not a readable JSON object, so Kubernetes network policy enforcement " + "cannot be determined." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + @pytest.mark.parametrize( + "configuration_values", + [ + None, + "", + "{}", + '{"enableWindowsIpam": "false"}', + '{"enableNetworkPolicy": "yes"}', + '{"enableNetworkPolicy": 1}', + '{"enableNetworkPolicy": null}', + ], + ) + def test_network_policy_setting_not_a_boolean(self, configuration_values): + """A configuration carrying no recognizable boolean for the setting must be MANUAL, never FAIL. + + Absent, empty, a different key, `"yes"`, `1` and `null` all mean the setting was not + stated. Reporting FAIL on any of them would assert that enforcement is off on the strength + of a value the API never returned. + """ + result = run_check([build_cluster(addons=vpc_cni_addon(configuration_values))]) + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} Amazon VPC CNI add-on does not set " + "enableNetworkPolicy to true or false, so Kubernetes network policy " + "enforcement cannot be determined." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + @pytest.mark.parametrize( + "configuration_values", + [ + '{"enableNetworkPolicy": "true"}', + '{"enableNetworkPolicy": "True"}', + '{"enableNetworkPolicy": true}', + '{"enableNetworkPolicy": "true", "enableWindowsIpam": "false"}', + ], + ) + def test_network_policy_enforced(self, configuration_values): + """A cluster whose vpc-cni add-on enables the setting must PASS, string or boolean. + + The add-on configuration schema types this setting as a string carrying + `"format": "boolean"`, so the API returns `"true"` rather than `true`; `"True"` and a JSON + boolean must land on the same verdict, and an unrelated sibling key must not disturb it. + """ + result = run_check([build_cluster(addons=vpc_cni_addon(configuration_values))]) + + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} enforces Kubernetes network policies through " + "the Amazon VPC CNI add-on. This does not confirm that NetworkPolicy " + "resources restricting pod-to-pod traffic exist in the cluster." + ) + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + @pytest.mark.parametrize( + "configuration_values", + [ + '{"enableNetworkPolicy": "false"}', + '{"enableNetworkPolicy": "False"}', + '{"enableNetworkPolicy": false}', + ], + ) + def test_network_policy_not_enforced(self, configuration_values): + """A cluster whose vpc-cni add-on sets the setting to false must FAIL, string or boolean. + + `"false"`, `"False"` and a JSON `false` are the three forms the setting can arrive in, and + a decoder that only understood one of them would report MANUAL on a real misconfiguration. + """ + result = run_check([build_cluster(addons=vpc_cni_addon(configuration_values))]) + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + f"EKS cluster {cluster_name} Amazon VPC CNI managed add-on does not enforce " + "Kubernetes network policies, since it sets enableNetworkPolicy to false. " + "Enforcement by a third-party policy engine or a self-managed VPC CNI is not " + "visible to the EKS API and is not evaluated." + ) + # The finding must not claim a property of the CLUSTER from an add-on setting: a cluster + # enforcing through Calico or Cilium, or through a self-managed VPC CNI, has this setting + # false and does enforce. Both are documented architectures, so the old wording was false + # of them rather than merely imprecise. + assert "cluster does not enforce" not in result[0].status_extended + assert result[0].resource_id == cluster_name + assert result[0].resource_arn == cluster_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + + def test_multiple_clusters(self): + """Six clusters must yield six reports, in input order, each judged on its own add-on. + + Two enforcing, three not and one with no managed add-on, so a check that carried state + between iterations or reported once per account would not produce this split. + """ + clusters = [] + for index in range(6): + name = f"{cluster_name}_{index}" + arn = f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster/{name}" + if index in (0, 1): + addons = vpc_cni_addon('{"enableNetworkPolicy": "true"}') + elif index in (2, 3, 4): + addons = vpc_cni_addon('{"enableNetworkPolicy": "false"}') + else: + addons = {} + clusters.append(build_cluster(name=name, arn=arn, addons=addons)) + + result = run_check(clusters) + + assert len(result) == 6 + statuses = [report.status for report in result] + assert statuses.count("PASS") == 2 + assert statuses.count("FAIL") == 3 + assert statuses.count("MANUAL") == 1 + assert [report.resource_id for report in result] == [ + f"{cluster_name}_{index}" for index in range(6) + ] diff --git a/tests/providers/aws/services/eks/eks_service_test.py b/tests/providers/aws/services/eks/eks_service_test.py index 86de0e246a..1feb75cc45 100644 --- a/tests/providers/aws/services/eks/eks_service_test.py +++ b/tests/providers/aws/services/eks/eks_service_test.py @@ -1,5 +1,6 @@ from unittest.mock import patch +import botocore from boto3 import client from moto import mock_aws @@ -7,6 +8,7 @@ from prowler.providers.aws.services.eks.eks_service import EKS from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1, + mocked_api_response, set_mocked_aws_provider, ) @@ -14,6 +16,77 @@ cluster_name = "test" cidr_block_vpc = "10.0.0.0/16" cidr_block_subnet_1 = "10.0.0.0/22" cidr_block_subnet_2 = "10.0.4.0/22" +cluster_arn = ( + f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster/{cluster_name}" +) +vpc_cni_addon_arn = f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:addon/{cluster_name}/vpc-cni/1a2b3c4d" +vpc_cni_configuration_values = '{"enableNetworkPolicy":"true"}' + +make_api_call = botocore.client.BaseClient._make_api_call +described_addons = [] + + +def _addon_response(addon_name, arn, configuration_values=None): + """Build a DescribeAddon response, omitting `configurationValues` when none is given. + + The key is absent from the API response for an add-on left at its defaults, so passing None + reproduces that rather than sending an empty string. + """ + addon = {"addonName": addon_name, "addonArn": arn, "clusterName": cluster_name} + if configuration_values is not None: + addon["configurationValues"] = configuration_values + return mocked_api_response("eks", "DescribeAddon", {"addon": addon}) + + +def mock_make_api_call_addons(self, operation_name, kwargs): + """Serve the add-on inventory, with vpc-cni on the SECOND ListAddons page.""" + if operation_name == "ListClusters": + return mocked_api_response("eks", "ListClusters", {"clusters": [cluster_name]}) + if operation_name == "DescribeCluster": + return mocked_api_response( + "eks", + "DescribeCluster", + {"cluster": {"name": cluster_name, "arn": cluster_arn, "version": "1.34"}}, + ) + if operation_name == "ListAddons": + if kwargs.get("nextToken") is None: + return mocked_api_response( + "eks", + "ListAddons", + {"addons": ["coredns"], "nextToken": "second-page"}, + ) + return mocked_api_response("eks", "ListAddons", {"addons": ["vpc-cni"]}) + if operation_name == "DescribeAddon": + described_addons.append(kwargs["addonName"]) + if kwargs["addonName"] == "vpc-cni": + return _addon_response( + "vpc-cni", vpc_cni_addon_arn, vpc_cni_configuration_values + ) + return _addon_response( + "coredns", + f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:addon/{cluster_name}/coredns/5e6f7a8b", + ) + return make_api_call(self, operation_name, kwargs) + + +def mock_make_api_call_list_addons_denied(self, operation_name, kwargs): + """Deny ListAddons and serve every other call, as a role without eks:ListAddons would.""" + if operation_name == "ListAddons": + raise botocore.exceptions.ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return mock_make_api_call_addons(self, operation_name, kwargs) + + +def mock_make_api_call_describe_vpc_cni_denied(self, operation_name, kwargs): + """Deny DescribeAddon for vpc-cni only, so listing succeeds and the per-add-on read fails.""" + if operation_name == "DescribeAddon" and kwargs["addonName"] == "vpc-cni": + raise botocore.exceptions.ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return mock_make_api_call_addons(self, operation_name, kwargs) def mock_generate_regional_clients(provider, service): @@ -139,3 +212,86 @@ class Test_EKS_Service: assert eks.clusters[0].public_access_cidrs == ["0.0.0.0/0"] assert eks.clusters[0].encryptionConfig assert eks.clusters[0].version == "1.10" + + +@patch( + "prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients", + new=mock_generate_regional_clients, +) +class Test_EKS_Service_Addons: + # Test EKS describe cluster add-ons + @mock_aws + @patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_addons, + ) + def test__describe_cluster_addons(self): + """vpc-cni is collected with its ARN and configuration from the SECOND ListAddons page. + + Also asserts coredns costs no DescribeAddon call: it is listed on the cluster but not in + COLLECTED_ADDONS, and DescribeAddon has no batch form, so describing it would be one extra + API call per cluster for data no check reads. + """ + described_addons.clear() + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + eks = EKS(aws_provider) + + assert len(eks.clusters) == 1 + cluster = eks.clusters[0] + assert not cluster.addons_discovery_failed + # vpc-cni is only on the second ListAddons page, so this fails without pagination + assert sorted(cluster.addons) == ["vpc-cni"] + assert cluster.addons["vpc-cni"].name == "vpc-cni" + assert cluster.addons["vpc-cni"].arn == vpc_cni_addon_arn + assert ( + cluster.addons["vpc-cni"].configuration_values + == vpc_cni_configuration_values + ) + assert not cluster.addons["vpc-cni"].configuration_discovery_failed + # coredns is listed but not in COLLECTED_ADDONS, so it costs no DescribeAddon call + assert described_addons == ["vpc-cni"] + + # Test EKS cluster add-ons cannot be listed + @mock_aws + @patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_list_addons_denied, + ) + def test__describe_cluster_addons_list_denied(self): + """A denied ListAddons sets addons_discovery_failed and issues no DescribeAddon call. + + The cluster itself must survive collection: a missing add-on permission may not cost the + scan every other EKS finding for that cluster. + """ + described_addons.clear() + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + eks = EKS(aws_provider) + + assert len(eks.clusters) == 1 + assert eks.clusters[0].addons_discovery_failed + assert eks.clusters[0].addons == {} + assert described_addons == [] + + # Test EKS cluster add-on configuration cannot be described + @mock_aws + @patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_describe_vpc_cni_denied, + ) + def test__describe_cluster_addons_describe_denied(self): + """A denied DescribeAddon flags the add-on, not the cluster, and leaves its fields None. + + The add-on is known to exist because ListAddons succeeded, so the failure belongs on + `configuration_discovery_failed` while `addons_discovery_failed` stays false. + """ + described_addons.clear() + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + eks = EKS(aws_provider) + + assert len(eks.clusters) == 1 + cluster = eks.clusters[0] + assert not cluster.addons_discovery_failed + assert sorted(cluster.addons) == ["vpc-cni"] + assert cluster.addons["vpc-cni"].configuration_discovery_failed + assert cluster.addons["vpc-cni"].configuration_values is None + assert cluster.addons["vpc-cni"].arn is None