From b512f6c421ad178c0a326e233e04285278c48720 Mon Sep 17 00:00:00 2001 From: Daniel Barranquero <74871504+danibarranqueroo@users.noreply.github.com> Date: Thu, 11 Sep 2025 15:55:16 +0200 Subject: [PATCH] fix(firehose): false positive in `firehose_stream_encrypted_at_rest` (#8599) Co-authored-by: Sergio Garcia --- prowler/CHANGELOG.md | 1 + .../firehose_stream_encrypted_at_rest.py | 24 ++- ...fka_cluster_encryption_at_rest_uses_cmk.py | 7 +- ...fka_cluster_enhanced_monitoring_enabled.py | 7 +- ...a_cluster_in_transit_encryption_enabled.py | 7 +- .../kafka_cluster_is_public.py | 7 +- ...uster_mutual_tls_authentication_enabled.py | 7 +- ...ka_cluster_unrestricted_access_disabled.py | 7 +- .../kafka_cluster_uses_latest_version.py | 7 +- .../aws/services/kafka/kafka_service.py | 146 ++++++++++++----- ...luster_encryption_at_rest_uses_cmk_test.py | 61 +++++++ ...luster_enhanced_monitoring_enabled_test.py | 74 +++++++-- ...ster_in_transit_encryption_enabled_test.py | 80 ++++++++-- .../kafka_cluster_is_public_test.py | 74 +++++++-- ..._mutual_tls_authentication_enabled_test.py | 74 +++++++-- ...uster_unrestricted_access_disabled_test.py | 74 +++++++-- .../kafka_cluster_uses_latest_version_test.py | 79 +++++++-- .../aws/services/kafka/kafka_service_test.py | 151 +++++++++++------- 18 files changed, 724 insertions(+), 163 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 9df610cdd9..5d873f663b 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -18,6 +18,7 @@ All notable changes to the **Prowler SDK** are documented in this file. ### Fixed - Replaced old check id with new ones for compliance files [(#8682)](https://github.com/prowler-cloud/prowler/pull/8682) +- `firehose_stream_encrypted_at_rest` check false positives and new api call in kafka service [(#8599)](https://github.com/prowler-cloud/prowler/pull/8599) - Replace defender rules policies key to use old name [(#8702)](https://github.com/prowler-cloud/prowler/pull/8702) ## [v5.12.0] (Prowler v5.12.0) diff --git a/prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py b/prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py index 69d1711f73..21ca07a920 100644 --- a/prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py +++ b/prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py @@ -3,6 +3,7 @@ from typing import List from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.firehose.firehose_client import firehose_client from prowler.providers.aws.services.firehose.firehose_service import EncryptionStatus +from prowler.providers.aws.services.kafka.kafka_client import kafka_client from prowler.providers.aws.services.kinesis.kinesis_client import kinesis_client from prowler.providers.aws.services.kinesis.kinesis_service import EncryptionType @@ -37,7 +38,28 @@ class firehose_stream_encrypted_at_rest(Check): report.status = "PASS" report.status_extended = f"Firehose Stream {stream.name} does not have at rest encryption enabled but the source stream {source_stream.name} has at rest encryption enabled." - # Check if the stream has encryption enabled directly + # MSK source - check if the MSK cluster has encryption at rest with CMK + elif stream.delivery_stream_type == "MSKAsSource": + msk_cluster_arn = stream.source.msk.msk_cluster_arn + if msk_cluster_arn: + msk_cluster = None + for cluster in kafka_client.clusters.values(): + if cluster.arn == msk_cluster_arn: + msk_cluster = cluster + break + + if msk_cluster: + # All MSK clusters (both provisioned and serverless) always have encryption at rest enabled by AWS + # AWS MSK always encrypts data at rest - either with AWS managed keys or CMK + report.status = "PASS" + if msk_cluster.kafka_version == "SERVERLESS": + report.status_extended = f"Firehose Stream {stream.name} uses MSK serverless source which always has encryption at rest enabled by default." + else: + report.status_extended = f"Firehose Stream {stream.name} uses MSK provisioned source which always has encryption at rest enabled by AWS (either with AWS managed keys or CMK)." + else: + report.status_extended = f"Firehose Stream {stream.name} uses MSK source which always has encryption at rest enabled by AWS." + + # Check if the stream has encryption enabled directly (DirectPut or DatabaseAsSource cases) elif stream.kms_encryption == EncryptionStatus.ENABLED: report.status = "PASS" report.status_extended = f"Firehose Stream {stream.name} does have at rest encryption enabled." diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk.py b/prowler/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk.py index 2dce9437ad..b110804028 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk.py @@ -12,7 +12,12 @@ class kafka_cluster_encryption_at_rest_uses_cmk(Check): report.status = "FAIL" report.status_extended = f"Kafka cluster '{cluster.name}' does not have encryption at rest enabled with a CMK." - if any( + # Serverless clusters always have encryption at rest enabled by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has encryption at rest enabled by default." + # For provisioned clusters, check if they use a customer managed KMS key + elif any( ( cluster.data_volume_kms_key_id == key.arn and getattr(key, "manager", "") == "CUSTOMER" diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled.py b/prowler/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled.py index 2bf0d38391..03fb098eef 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled.py @@ -13,7 +13,12 @@ class kafka_cluster_enhanced_monitoring_enabled(Check): f"Kafka cluster '{cluster.name}' has enhanced monitoring enabled." ) - if cluster.enhanced_monitoring == "DEFAULT": + # Serverless clusters always have enhanced monitoring enabled by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has enhanced monitoring enabled by default." + # For provisioned clusters, check the enhanced monitoring configuration + elif cluster.enhanced_monitoring == "DEFAULT": report.status = "FAIL" report.status_extended = f"Kafka cluster '{cluster.name}' does not have enhanced monitoring enabled." diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled.py b/prowler/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled.py index 7b37fe5a17..5e9bd77af0 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled.py @@ -11,7 +11,12 @@ class kafka_cluster_in_transit_encryption_enabled(Check): report.status = "FAIL" report.status_extended = f"Kafka cluster '{cluster.name}' does not have encryption in transit enabled." - if ( + # Serverless clusters always have encryption in transit enabled by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has encryption in transit enabled by default." + # For provisioned clusters, check the encryption configuration + elif ( cluster.encryption_in_transit.client_broker == "TLS" and cluster.encryption_in_transit.in_cluster ): diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public.py b/prowler/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public.py index 426ff04490..4dabf61fce 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public.py @@ -13,7 +13,12 @@ class kafka_cluster_is_public(Check): f"Kafka cluster {cluster.name} is publicly accessible." ) - if not cluster.public_access: + # Serverless clusters are always private by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster {cluster.name} is serverless and always private by default." + # For provisioned clusters, check the public access configuration + elif not cluster.public_access: report.status = "PASS" report.status_extended = ( f"Kafka cluster {cluster.name} is not publicly accessible." diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled.py b/prowler/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled.py index 0e07e34107..98dcf15b44 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled.py @@ -11,7 +11,12 @@ class kafka_cluster_mutual_tls_authentication_enabled(Check): report.status = "FAIL" report.status_extended = f"Kafka cluster '{cluster.name}' does not have mutual TLS authentication enabled." - if cluster.tls_authentication: + # Serverless clusters always have TLS authentication enabled by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has TLS authentication enabled by default." + # For provisioned clusters, check the TLS configuration + elif cluster.tls_authentication: report.status = "PASS" report.status_extended = f"Kafka cluster '{cluster.name}' has mutual TLS authentication enabled." diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled.py b/prowler/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled.py index f96abbdb06..219affd8a5 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled.py @@ -13,7 +13,12 @@ class kafka_cluster_unrestricted_access_disabled(Check): f"Kafka cluster '{cluster.name}' has unrestricted access enabled." ) - if not cluster.unauthentication_access: + # Serverless clusters always require authentication by default + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always requires authentication by default." + # For provisioned clusters, check the unauthenticated access configuration + elif not cluster.unauthentication_access: report.status = "PASS" report.status_extended = f"Kafka cluster '{cluster.name}' does not have unrestricted access enabled." diff --git a/prowler/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version.py b/prowler/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version.py index 32c1793384..e19686c9a6 100644 --- a/prowler/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version.py +++ b/prowler/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version.py @@ -13,7 +13,12 @@ class kafka_cluster_uses_latest_version(Check): f"Kafka cluster '{cluster.name}' is using the latest version." ) - if cluster.kafka_version != kafka_client.kafka_versions[-1].version: + # Serverless clusters don't have specific Kafka versions - AWS manages them automatically + if cluster.kafka_version == "SERVERLESS": + report.status = "PASS" + report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and AWS automatically manages the Kafka version." + # For provisioned clusters, check if they're using the latest version + elif cluster.kafka_version != kafka_client.kafka_versions[-1].version: report.status = "FAIL" report.status_extended = ( f"Kafka cluster '{cluster.name}' is not using the latest version." diff --git a/prowler/providers/aws/services/kafka/kafka_service.py b/prowler/providers/aws/services/kafka/kafka_service.py index 4197e543c4..3b52063206 100644 --- a/prowler/providers/aws/services/kafka/kafka_service.py +++ b/prowler/providers/aws/services/kafka/kafka_service.py @@ -15,61 +15,133 @@ class Kafka(AWSService): self.__threading_call__(self._list_kafka_versions) def _list_clusters(self, regional_client): + logger.info(f"Kafka - Listing clusters in region {regional_client.region}...") try: - cluster_paginator = regional_client.get_paginator("list_clusters") + # Use list_clusters_v2 to support both provisioned and serverless clusters + cluster_paginator = regional_client.get_paginator("list_clusters_v2") + logger.info( + f"Kafka - Paginator created for region {regional_client.region}" + ) for page in cluster_paginator.paginate(): + logger.info( + f"Kafka - Processing page with {len(page.get('ClusterInfoList', []))} clusters in region {regional_client.region}" + ) for cluster in page["ClusterInfoList"]: + logger.info( + f"Kafka - Found cluster: {cluster.get('ClusterName', 'Unknown')} in region {regional_client.region}" + ) arn = cluster.get( "ClusterArn", f"{self.account_arn_template}/{cluster.get('ClusterName', '')}", ) + cluster_type = cluster.get("ClusterType", "UNKNOWN") if not self.audit_resources or is_resource_filtered( arn, self.audit_resources ): - self.clusters[cluster.get("ClusterArn", "")] = Cluster( - id=arn.split(":")[-1].split("/")[-1], - name=cluster.get("ClusterName", ""), - arn=arn, - region=regional_client.region, - tags=list(cluster.get("Tags", {})), - state=cluster.get("State", ""), - kafka_version=cluster.get( - "CurrentBrokerSoftwareInfo", {} - ).get("KafkaVersion", ""), - data_volume_kms_key_id=cluster.get("EncryptionInfo", {}) - .get("EncryptionAtRest", {}) - .get("DataVolumeKMSKeyId", ""), - encryption_in_transit=EncryptionInTransit( - client_broker=cluster.get("EncryptionInfo", {}) - .get("EncryptionInTransit", {}) - .get("ClientBroker", "PLAINTEXT"), - in_cluster=cluster.get("EncryptionInfo", {}) - .get("EncryptionInTransit", {}) - .get("InCluster", False), - ), - tls_authentication=cluster.get("ClientAuthentication", {}) - .get("Tls", {}) - .get("Enabled", False), - public_access=cluster.get("BrokerNodeGroupInfo", {}) - .get("ConnectivityInfo", {}) - .get("PublicAccess", {}) - .get("Type", "SERVICE_PROVIDED_EIPS") - != "DISABLED", - unauthentication_access=cluster.get( - "ClientAuthentication", {} + # Handle provisioned clusters + if cluster_type == "PROVISIONED" and "Provisioned" in cluster: + provisioned = cluster["Provisioned"] + self.clusters[cluster.get("ClusterArn", "")] = Cluster( + id=arn.split(":")[-1].split("/")[-1], + name=cluster.get("ClusterName", ""), + arn=arn, + region=regional_client.region, + tags=( + list(cluster.get("Tags", {}).values()) + if cluster.get("Tags") + else [] + ), + state=cluster.get("State", ""), + kafka_version=provisioned.get( + "CurrentBrokerSoftwareInfo", {} + ).get("KafkaVersion", ""), + data_volume_kms_key_id=provisioned.get( + "EncryptionInfo", {} + ) + .get("EncryptionAtRest", {}) + .get("DataVolumeKMSKeyId", ""), + encryption_in_transit=EncryptionInTransit( + client_broker=provisioned.get("EncryptionInfo", {}) + .get("EncryptionInTransit", {}) + .get("ClientBroker", "PLAINTEXT"), + in_cluster=provisioned.get("EncryptionInfo", {}) + .get("EncryptionInTransit", {}) + .get("InCluster", False), + ), + tls_authentication=provisioned.get( + "ClientAuthentication", {} + ) + .get("Tls", {}) + .get("Enabled", False), + public_access=provisioned.get("BrokerNodeGroupInfo", {}) + .get("ConnectivityInfo", {}) + .get("PublicAccess", {}) + .get("Type", "SERVICE_PROVIDED_EIPS") + != "DISABLED", + unauthentication_access=provisioned.get( + "ClientAuthentication", {} + ) + .get("Unauthenticated", {}) + .get("Enabled", False), + enhanced_monitoring=provisioned.get( + "EnhancedMonitoring", "DEFAULT" + ), ) - .get("Unauthenticated", {}) - .get("Enabled", False), - enhanced_monitoring=cluster.get( - "EnhancedMonitoring", "DEFAULT" - ), + logger.info( + f"Kafka - Added provisioned cluster {cluster.get('ClusterName', 'Unknown')} to clusters dict" + ) + + # Handle serverless clusters + elif cluster_type == "SERVERLESS" and "Serverless" in cluster: + # For serverless clusters, encryption is always enabled by default + # We'll create a Cluster object with default encryption values + self.clusters[cluster.get("ClusterArn", "")] = Cluster( + id=arn.split(":")[-1].split("/")[-1], + name=cluster.get("ClusterName", ""), + arn=arn, + region=regional_client.region, + tags=( + list(cluster.get("Tags", {}).values()) + if cluster.get("Tags") + else [] + ), + state=cluster.get("State", ""), + kafka_version="SERVERLESS", # Serverless doesn't have specific Kafka version + data_volume_kms_key_id="AWS_MANAGED", # Serverless uses AWS managed keys + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", # Serverless always has TLS enabled + in_cluster=True, # Serverless always has in-cluster encryption + ), + tls_authentication=True, # Serverless always has TLS authentication + public_access=False, # Serverless clusters are always private + unauthentication_access=False, # Serverless requires authentication + enhanced_monitoring="DEFAULT", + ) + logger.info( + f"Kafka - Added serverless cluster {cluster.get('ClusterName', 'Unknown')} to clusters dict" + ) + + else: + logger.warning( + f"Kafka - Unknown cluster type {cluster_type} for cluster {cluster.get('ClusterName', 'Unknown')}" + ) + else: + logger.info( + f"Kafka - Cluster {cluster.get('ClusterName', 'Unknown')} filtered out by audit_resources" ) + + logger.info( + f"Kafka - Total clusters found in region {regional_client.region}: {len(self.clusters)}" + ) except Exception as error: logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + logger.error( + f"Kafka - Error details in region {regional_client.region}: {str(error)}" + ) def _list_kafka_versions(self, regional_client): try: diff --git a/tests/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk_test.py b/tests/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk_test.py index 236da74917..3673c32717 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk_test.py @@ -162,3 +162,64 @@ class Test_kafka_cluster_encryption_at_rest_uses_cmk: ) assert result[0].resource_tags == [] assert result[0].region == AWS_REGION_US_EAST_1 + + def test_kafka_cluster_serverless_encryption_at_rest(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + kms_client = MagicMock + kms_client.keys = [] + + with ( + patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + ), + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_cluster_encryption_at_rest_uses_cmk.kafka_cluster_encryption_at_rest_uses_cmk.kms_client", + new=kms_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_encryption_at_rest_uses_cmk.kafka_cluster_encryption_at_rest_uses_cmk import ( + kafka_cluster_encryption_at_rest_uses_cmk, + ) + + check = kafka_cluster_encryption_at_rest_uses_cmk() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and always has encryption at rest enabled by default." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 diff --git a/tests/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled_test.py b/tests/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled_test.py index 784d5ebff4..967c6740c0 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled_test.py @@ -4,7 +4,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( Cluster, EncryptionInTransit, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_enhanced_monitoring_enabled: @@ -14,11 +14,11 @@ class Test_kafka_cluster_enhanced_monitoring_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -56,11 +56,11 @@ class Test_kafka_cluster_enhanced_monitoring_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -110,11 +110,11 @@ class Test_kafka_cluster_enhanced_monitoring_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -138,3 +138,57 @@ class Test_kafka_cluster_enhanced_monitoring_enabled: ) assert result[0].resource_tags == [] assert result[0].region == AWS_REGION_US_EAST_1 + + def test_kafka_cluster_serverless_enhanced_monitoring(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_enhanced_monitoring_enabled.kafka_cluster_enhanced_monitoring_enabled import ( + kafka_cluster_enhanced_monitoring_enabled, + ) + + check = kafka_cluster_enhanced_monitoring_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and always has enhanced monitoring enabled by default." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 diff --git a/tests/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled_test.py b/tests/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled_test.py index 78368530ea..a4b8165ec6 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled_test.py @@ -4,7 +4,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( Cluster, EncryptionInTransit, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_in_transit_encryption_enabled: @@ -14,11 +14,11 @@ class Test_kafka_cluster_in_transit_encryption_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -56,11 +56,11 @@ class Test_kafka_cluster_in_transit_encryption_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -110,11 +110,11 @@ class Test_kafka_cluster_in_transit_encryption_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -164,11 +164,11 @@ class Test_kafka_cluster_in_transit_encryption_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -191,3 +191,57 @@ class Test_kafka_cluster_in_transit_encryption_enabled: == "arn:aws:kafka:us-east-1:123456789012:cluster/demo-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" ) assert result[0].region == AWS_REGION_US_EAST_1 + + def test_kafka_cluster_serverless_in_transit_encryption(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_in_transit_encryption_enabled.kafka_cluster_in_transit_encryption_enabled import ( + kafka_cluster_in_transit_encryption_enabled, + ) + + check = kafka_cluster_in_transit_encryption_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and always has encryption in transit enabled by default." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public_test.py b/tests/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public_test.py index 392582b0ee..12459150df 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public_test.py @@ -4,7 +4,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( Cluster, EncryptionInTransit, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_is_public: @@ -14,11 +14,11 @@ class Test_kafka_cluster_is_public: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -56,11 +56,11 @@ class Test_kafka_cluster_is_public: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -110,11 +110,11 @@ class Test_kafka_cluster_is_public: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -138,3 +138,57 @@ class Test_kafka_cluster_is_public: assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" assert result[0].region == AWS_REGION_US_EAST_1 assert result[0].resource_tags == [] + + def test_kafka_cluster_serverless_public(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_is_public.kafka_cluster_is_public import ( + kafka_cluster_is_public, + ) + + check = kafka_cluster_is_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster serverless-cluster-1 is serverless and always private by default." + ) + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled_test.py b/tests/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled_test.py index c5d7f42205..651184f063 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled_test.py @@ -4,7 +4,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( Cluster, EncryptionInTransit, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_mutual_tls_authentication_enabled: @@ -14,11 +14,11 @@ class Test_kafka_cluster_mutual_tls_authentication_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -56,11 +56,11 @@ class Test_kafka_cluster_mutual_tls_authentication_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -110,11 +110,11 @@ class Test_kafka_cluster_mutual_tls_authentication_enabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -138,3 +138,57 @@ class Test_kafka_cluster_mutual_tls_authentication_enabled: ) assert result[0].region == AWS_REGION_US_EAST_1 assert result[0].resource_tags == [] + + def test_kafka_cluster_serverless_mutual_tls_authentication(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_mutual_tls_authentication_enabled.kafka_cluster_mutual_tls_authentication_enabled import ( + kafka_cluster_mutual_tls_authentication_enabled, + ) + + check = kafka_cluster_mutual_tls_authentication_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and always has TLS authentication enabled by default." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled_test.py b/tests/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled_test.py index 22a2c0eb71..17d01e5635 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled_test.py @@ -4,7 +4,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( Cluster, EncryptionInTransit, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_unrestricted_access_disabled: @@ -14,11 +14,11 @@ class Test_kafka_cluster_unrestricted_access_disabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -56,11 +56,11 @@ class Test_kafka_cluster_unrestricted_access_disabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -110,11 +110,11 @@ class Test_kafka_cluster_unrestricted_access_disabled: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -138,3 +138,57 @@ class Test_kafka_cluster_unrestricted_access_disabled: ) assert result[0].region == AWS_REGION_US_EAST_1 assert result[0].resource_tags == [] + + def test_kafka_cluster_serverless_unrestricted_access_disabled(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_unrestricted_access_disabled.kafka_cluster_unrestricted_access_disabled import ( + kafka_cluster_unrestricted_access_disabled, + ) + + check = kafka_cluster_unrestricted_access_disabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and always requires authentication by default." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version_test.py b/tests/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version_test.py index ac44f868e0..cfae707ad9 100644 --- a/tests/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version_test.py +++ b/tests/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version_test.py @@ -5,7 +5,7 @@ from prowler.providers.aws.services.kafka.kafka_service import ( EncryptionInTransit, KafkaVersion, ) -from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider +from tests.providers.aws.utils import AWS_REGION_US_EAST_1 class Test_kafka_cluster_latest_version: @@ -15,11 +15,11 @@ class Test_kafka_cluster_latest_version: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -62,11 +62,11 @@ class Test_kafka_cluster_latest_version: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -121,11 +121,11 @@ class Test_kafka_cluster_latest_version: with ( patch( - "prowler.providers.common.provider.Provider.get_global_provider", - return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, ), patch( - "prowler.providers.aws.services.kafka.kafka_service.Kafka", + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", new=kafka_client, ), ): @@ -149,3 +149,62 @@ class Test_kafka_cluster_latest_version: ) assert result[0].resource_tags == [] assert result[0].region == AWS_REGION_US_EAST_1 + + def test_kafka_cluster_serverless_uses_latest_version(self): + kafka_client = MagicMock + kafka_client.clusters = { + "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6": Cluster( + id="6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + name="serverless-cluster-1", + arn="arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + region=AWS_REGION_US_EAST_1, + tags=[], + state="ACTIVE", + kafka_version="SERVERLESS", + data_volume_kms_key_id="AWS_MANAGED", + encryption_in_transit=EncryptionInTransit( + client_broker="TLS", + in_cluster=True, + ), + tls_authentication=True, + public_access=False, + unauthentication_access=False, + enhanced_monitoring="DEFAULT", + ) + } + + kafka_client.kafka_versions = [ + KafkaVersion(version="1.0.0", status="DEPRECATED"), + KafkaVersion(version="2.8.0", status="ACTIVE"), + ] + + with ( + patch( + "prowler.providers.aws.services.kafka.kafka_service.Kafka", + new=kafka_client, + ), + patch( + "prowler.providers.aws.services.kafka.kafka_client.kafka_client", + new=kafka_client, + ), + ): + from prowler.providers.aws.services.kafka.kafka_cluster_uses_latest_version.kafka_cluster_uses_latest_version import ( + kafka_cluster_uses_latest_version, + ) + + check = kafka_cluster_uses_latest_version() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Kafka cluster 'serverless-cluster-1' is serverless and AWS automatically manages the Kafka version." + ) + assert result[0].resource_id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert ( + result[0].resource_arn + == "arn:aws:kafka:us-east-1:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + ) + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 diff --git a/tests/providers/aws/services/kafka/kafka_service_test.py b/tests/providers/aws/services/kafka/kafka_service_test.py index a7767a0201..e041c4e125 100644 --- a/tests/providers/aws/services/kafka/kafka_service_test.py +++ b/tests/providers/aws/services/kafka/kafka_service_test.py @@ -13,47 +13,67 @@ make_api_call = botocore.client.BaseClient._make_api_call def mock_make_api_call(self, operation_name, kwarg): - if operation_name == "ListClusters": + if operation_name == "ListClustersV2": return { "ClusterInfoList": [ { - "BrokerNodeGroupInfo": { - "BrokerAZDistribution": "DEFAULT", - "ClientSubnets": ["subnet-cbfff283", "subnet-6746046b"], - "InstanceType": "kafka.m5.large", - "SecurityGroups": ["sg-f839b688"], - "StorageInfo": {"EbsStorageInfo": {"VolumeSize": 100}}, - }, + "ClusterType": "PROVISIONED", "ClusterArn": f"arn:aws:kafka:{AWS_REGION_US_EAST_1}:123456789012:cluster/demo-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5", "ClusterName": "demo-cluster-1", - "CreationTime": "2020-07-09T02:31:36.223000+00:00", - "CurrentBrokerSoftwareInfo": {"KafkaVersion": "2.2.1"}, - "CurrentVersion": "K3AEGXETSR30VB", - "EncryptionInfo": { - "EncryptionAtRest": { - "DataVolumeKMSKeyId": f"arn:aws:kms:{AWS_REGION_US_EAST_1}:123456789012:key/a7ca56d5-0768-4b64-a670-339a9fbef81c" - }, - "EncryptionInTransit": { - "ClientBroker": "TLS_PLAINTEXT", - "InCluster": True, - }, - }, - "ClientAuthentication": { - "Tls": {"CertificateAuthorityArnList": [], "Enabled": True}, - "Unauthenticated": {"Enabled": False}, - }, - "EnhancedMonitoring": "DEFAULT", - "OpenMonitoring": { - "Prometheus": { - "JmxExporter": {"EnabledInBroker": False}, - "NodeExporter": {"EnabledInBroker": False}, - } - }, - "NumberOfBrokerNodes": 2, "State": "ACTIVE", "Tags": {}, - "ZookeeperConnectString": f"z-2.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181,z-1.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181,z-3.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181", - } + "Provisioned": { + "BrokerNodeGroupInfo": { + "BrokerAZDistribution": "DEFAULT", + "ClientSubnets": ["subnet-cbfff283", "subnet-6746046b"], + "InstanceType": "kafka.m5.large", + "SecurityGroups": ["sg-f839b688"], + "StorageInfo": {"EbsStorageInfo": {"VolumeSize": 100}}, + "ConnectivityInfo": { + "PublicAccess": {"Type": "SERVICE_PROVIDED_EIPS"} + }, + }, + "CurrentBrokerSoftwareInfo": {"KafkaVersion": "2.2.1"}, + "CurrentVersion": "K3AEGXETSR30VB", + "EncryptionInfo": { + "EncryptionAtRest": { + "DataVolumeKMSKeyId": f"arn:aws:kms:{AWS_REGION_US_EAST_1}:123456789012:key/a7ca56d5-0768-4b64-a670-339a9fbef81c" + }, + "EncryptionInTransit": { + "ClientBroker": "TLS_PLAINTEXT", + "InCluster": True, + }, + }, + "ClientAuthentication": { + "Tls": {"CertificateAuthorityArnList": [], "Enabled": True}, + "Unauthenticated": {"Enabled": False}, + }, + "EnhancedMonitoring": "DEFAULT", + "OpenMonitoring": { + "Prometheus": { + "JmxExporter": {"EnabledInBroker": False}, + "NodeExporter": {"EnabledInBroker": False}, + } + }, + "NumberOfBrokerNodes": 2, + "ZookeeperConnectString": f"z-2.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181,z-1.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181,z-3.demo-cluster-1.xuy0sb.c5.kafka.{AWS_REGION_US_EAST_1}.amazonaws.com:2181", + }, + }, + { + "ClusterType": "SERVERLESS", + "ClusterArn": f"arn:aws:kafka:{AWS_REGION_US_EAST_1}:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6", + "ClusterName": "serverless-cluster-1", + "State": "ACTIVE", + "Tags": {}, + "Serverless": { + "VpcConfigs": [ + { + "SubnetIds": ["subnet-cbfff283", "subnet-6746046b"], + "SecurityGroups": ["sg-f839b688"], + } + ], + }, + }, ] } elif operation_name == "ListKafkaVersions": @@ -86,32 +106,53 @@ class TestKafkaService: assert kafka.__class__.__name__ == "Kafka" assert kafka.session.__class__.__name__ == "Session" assert kafka.audited_account == AWS_ACCOUNT_NUMBER - # Clusters assertions - assert len(kafka.clusters) == 1 - cluster_arn = f"arn:aws:kafka:{AWS_REGION_US_EAST_1}:123456789012:cluster/demo-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" - assert cluster_arn in kafka.clusters + + # Clusters assertions - should now include both provisioned and serverless + assert len(kafka.clusters) == 2 + + # Check provisioned cluster + provisioned_cluster_arn = f"arn:aws:kafka:{AWS_REGION_US_EAST_1}:123456789012:cluster/demo-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" + assert provisioned_cluster_arn in kafka.clusters + provisioned_cluster = kafka.clusters[provisioned_cluster_arn] + assert provisioned_cluster.id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" + assert provisioned_cluster.arn == provisioned_cluster_arn + assert provisioned_cluster.name == "demo-cluster-1" + assert provisioned_cluster.region == AWS_REGION_US_EAST_1 + assert provisioned_cluster.tags == [] + assert provisioned_cluster.state == "ACTIVE" + assert provisioned_cluster.kafka_version == "2.2.1" assert ( - kafka.clusters[cluster_arn].id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-5" - ) - assert kafka.clusters[cluster_arn].arn == cluster_arn - assert kafka.clusters[cluster_arn].name == "demo-cluster-1" - assert kafka.clusters[cluster_arn].region == AWS_REGION_US_EAST_1 - assert kafka.clusters[cluster_arn].tags == [] - assert kafka.clusters[cluster_arn].state == "ACTIVE" - assert kafka.clusters[cluster_arn].kafka_version == "2.2.1" - assert ( - kafka.clusters[cluster_arn].data_volume_kms_key_id + provisioned_cluster.data_volume_kms_key_id == f"arn:aws:kms:{AWS_REGION_US_EAST_1}:123456789012:key/a7ca56d5-0768-4b64-a670-339a9fbef81c" ) assert ( - kafka.clusters[cluster_arn].encryption_in_transit.client_broker - == "TLS_PLAINTEXT" + provisioned_cluster.encryption_in_transit.client_broker == "TLS_PLAINTEXT" ) - assert kafka.clusters[cluster_arn].encryption_in_transit.in_cluster - assert kafka.clusters[cluster_arn].enhanced_monitoring == "DEFAULT" - assert kafka.clusters[cluster_arn].tls_authentication - assert kafka.clusters[cluster_arn].public_access - assert not kafka.clusters[cluster_arn].unauthentication_access + assert provisioned_cluster.encryption_in_transit.in_cluster + assert provisioned_cluster.enhanced_monitoring == "DEFAULT" + assert provisioned_cluster.tls_authentication + assert provisioned_cluster.public_access + assert not provisioned_cluster.unauthentication_access + + # Check serverless cluster + serverless_cluster_arn = f"arn:aws:kafka:{AWS_REGION_US_EAST_1}:123456789012:cluster/serverless-cluster-1/6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert serverless_cluster_arn in kafka.clusters + serverless_cluster = kafka.clusters[serverless_cluster_arn] + assert serverless_cluster.id == "6357e0b2-0e6a-4b86-a0b4-70df934c2e31-6" + assert serverless_cluster.arn == serverless_cluster_arn + assert serverless_cluster.name == "serverless-cluster-1" + assert serverless_cluster.region == AWS_REGION_US_EAST_1 + assert serverless_cluster.tags == [] + assert serverless_cluster.state == "ACTIVE" + assert serverless_cluster.kafka_version == "SERVERLESS" + assert serverless_cluster.data_volume_kms_key_id == "AWS_MANAGED" + assert serverless_cluster.encryption_in_transit.client_broker == "TLS" + assert serverless_cluster.encryption_in_transit.in_cluster + assert serverless_cluster.enhanced_monitoring == "DEFAULT" + assert serverless_cluster.tls_authentication + assert not serverless_cluster.public_access + assert not serverless_cluster.unauthentication_access + # Kafka versions assertions assert len(kafka.kafka_versions) == 2 assert kafka.kafka_versions[0].version == "1.0.0"