fix(firehose): false positive in firehose_stream_encrypted_at_rest (#8599)

Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>

Author: danibarranqueroo

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)

prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py

@@ -3,6 +3,7 @@
 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 @@ def execute(self) -> List[Check_Report_AWS]:
                         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."

prowler/providers/aws/services/kafka/kafka_cluster_encryption_at_rest_uses_cmk/kafka_cluster_encryption_at_rest_uses_cmk.py

@@ -12,7 +12,12 @@ def execute(self):
             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"

prowler/providers/aws/services/kafka/kafka_cluster_enhanced_monitoring_enabled/kafka_cluster_enhanced_monitoring_enabled.py

@@ -13,7 +13,12 @@ def execute(self):
                 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."
 

prowler/providers/aws/services/kafka/kafka_cluster_in_transit_encryption_enabled/kafka_cluster_in_transit_encryption_enabled.py

@@ -11,7 +11,12 @@ def execute(self):
             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
             ):

prowler/providers/aws/services/kafka/kafka_cluster_is_public/kafka_cluster_is_public.py

@@ -13,7 +13,12 @@ def execute(self):
                 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."

prowler/providers/aws/services/kafka/kafka_cluster_mutual_tls_authentication_enabled/kafka_cluster_mutual_tls_authentication_enabled.py

@@ -11,7 +11,12 @@ def execute(self):
             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."
 

prowler/providers/aws/services/kafka/kafka_cluster_unrestricted_access_disabled/kafka_cluster_unrestricted_access_disabled.py

@@ -13,7 +13,12 @@ def execute(self):
                 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."
 

prowler/providers/aws/services/kafka/kafka_cluster_uses_latest_version/kafka_cluster_uses_latest_version.py

@@ -13,7 +13,12 @@ def execute(self):
                 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."

prowler/providers/aws/services/kafka/kafka_service.py

@@ -15,61 +15,133 @@ def __init__(self, provider):
         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"
+                                ),
+                            )
+                            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"
                             )
-                            .get("Unauthenticated", {})
-                            .get("Enabled", False),
-                            enhanced_monitoring=cluster.get(
-                                "EnhancedMonitoring", "DEFAULT"
-                            ),
+
+                        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: