feat(firehose): add new check firehose_stream_encrypted_at_rest (#5635)

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

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

@@ -1,3 +1,4 @@
+from enum import Enum
 from typing import Dict, List, Optional
 
 from botocore.client import ClientError
@@ -17,6 +18,9 @@ class Firehose(AWSService):
         self.__threading_call__(
             self._list_tags_for_delivery_stream, self.delivery_streams.values()
         )
+        self.__threading_call__(
+            self._describe_delivery_stream, self.delivery_streams.values()
+        )
 
     def _list_delivery_streams(self, regional_client):
         logger.info("Firehose - Listing delivery streams...")
@@ -39,6 +43,7 @@ class Firehose(AWSService):
             )
 
     def _list_tags_for_delivery_stream(self, stream):
+        logger.info(f"Firehose - Listing tags for stream {stream.name}...")
         try:
             stream.tags = (
                 self.regional_clients[stream.region]
@@ -50,9 +55,42 @@ class Firehose(AWSService):
                 f"{stream.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _describe_delivery_stream(self, stream):
+        logger.info(f"Firehose - Describing stream {stream.name}...")
+        try:
+            describe_stream = self.regional_clients[
+                stream.region
+            ].describe_delivery_stream(DeliveryStreamName=stream.name)
+            encryption_config = describe_stream.get(
+                "DeliveryStreamDescription", {}
+            ).get("DeliveryStreamEncryptionConfiguration", {})
+            stream.kms_encryption = EncryptionStatus(
+                encryption_config.get("Status", "DISABLED")
+            )
+            stream.kms_key_arn = encryption_config.get("KeyARN", "")
+        except ClientError as error:
+            logger.error(
+                f"{stream.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class EncryptionStatus(Enum):
+    """Possible values for the status of the encryption of a Firehose stream"""
+
+    ENABLED = "ENABLED"
+    DISABLED = "DISABLED"
+    ENABLING = "ENABLING"
+    DISABLING = "DISABLING"
+    ENABLING_FAILED = "ENABLING_FAILED"
+    DISABLING_FAILED = "DISABLING_FAILED"
+
 
 class DeliveryStream(BaseModel):
+    """Model for a Firehose Delivery Stream"""
+
     arn: str
     name: str
     region: str
+    kms_key_arn: Optional[str] = Field(default_factory=str)
+    kms_encryption: Optional[str] = Field(default_factory=str)
     tags: Optional[List[Dict[str, str]]] = Field(default_factory=list)

prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.metadata.json

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "firehose_stream_encrypted_at_rest",
+  "CheckTitle": "DataFirehose delivery streams should be encrypted at rest.",
+  "CheckType": [
+    "Software and Configuration Checks/Industry and Regulatory Standards/NIST 800-53 Controls"
+  ],
+  "ServiceName": "firehose",
+  "SubServiceName": "Ensure DataFirehose delivery streams are encrypted at rest.",
+  "ResourceIdTemplate": "arn:partition:firehose:region:account-id:deliverystream/delivery-stream-id",
+  "Severity": "medium",
+  "ResourceType": "AwsKinesisFirehoseDeliveryStream",
+  "Description": "",
+  "Risk": "Without encryption at rest, data in Amazon Kinesis Data Firehose delivery streams is vulnerable to unauthorized access if the storage layer is compromised. This increases the risk of sensitive information exposure, potentially leading to data breaches or non-compliance with security regulations.",
+  "RelatedUrl": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingKMSEncryption.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws firehose update-delivery-stream --delivery-stream-name <delivery-stream-name> --delivery-stream-encryption-configuration-input '{ \"KeyType\": \"CUSTOMER_MANAGED_CMK\", \"KeyARN\": \"<kms-key-arn>\" }'",
+      "NativeIaC": "https://docs.prowler.com/checks/aws/general-policies/ensure-aws-kinesis-firehoses-delivery-stream-is-encrypted/",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/datafirehose-controls.html#datafirehose-1",
+      "Terraform": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Firehose/delivery-stream-encrypted-with-kms-customer-master-keys.html"
+    },
+    "Recommendation": {
+      "Text": "Enable server-side encryption for Kinesis Firehose delivery streams using AWS Key Management Service (KMS). This encrypts data at rest, ensuring that sensitive information remains secure and compliant with regulatory standards.",
+      "Url": "https://docs.aws.amazon.com/firehose/latest/dev/encryption.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,43 @@
+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
+
+
+class firehose_stream_encrypted_at_rest(Check):
+    """Check if Firehose Streams are encrypted at rest.
+
+    This class verifies that all Firehose Streams have at rest encryption enabled by checking if KMS encryption is active and a KMS Key is configured.
+    """
+
+    def execute(self) -> List[Check_Report_AWS]:
+        """Execute the Firehose Stream Encrypted at Rest check.
+
+        Iterates over all Firehose Streams and checks if KMS encryption is enabled and a KMS Key is configured.
+
+        Returns:
+            List[Check_Report_AWS]: A list of reports for each Firehose Stream.
+        """
+        findings = []
+        for stream in firehose_client.delivery_streams.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = stream.region
+            report.resource_id = stream.name
+            report.resource_arn = stream.arn
+            report.resource_tags = stream.tags
+            report.status = "PASS"
+            report.status_extended = (
+                f"Firehose Stream {stream.name} does have at rest encryption enabled."
+            )
+
+            if (
+                stream.kms_encryption != EncryptionStatus.ENABLED
+                or not stream.kms_key_arn
+            ):
+                report.status = "FAIL"
+                report.status_extended = f"Firehose Stream {stream.name} does not have at rest encryption enabled."
+
+            findings.append(report)
+
+        return findings

tests/providers/aws/services/firehose/firehose_service_test.py

@@ -1,8 +1,15 @@
 from boto3 import client
 from moto import mock_aws
 
-from prowler.providers.aws.services.firehose.firehose_service import Firehose
-from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
+from prowler.providers.aws.services.firehose.firehose_service import (
+    EncryptionStatus,
+    Firehose,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_Firehose_Service:
@@ -67,3 +74,81 @@ class Test_Firehose_Service:
         assert firehose.delivery_streams[arn].name == delivery_stream_name
         assert firehose.delivery_streams[arn].region == AWS_REGION_EU_WEST_1
         assert firehose.delivery_streams[arn].tags == [{"Key": "key", "Value": "value"}]
+
+    @mock_aws
+    def test_list_tags_for_delivery_stream(self):
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        delivery_stream_name = arn.split("/")[-1]
+
+        # Firehose Client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+
+        assert len(firehose.delivery_streams) == 1
+        assert firehose.delivery_streams[arn].arn == arn
+        assert firehose.delivery_streams[arn].name == delivery_stream_name
+        assert firehose.delivery_streams[arn].region == AWS_REGION_EU_WEST_1
+        assert firehose.delivery_streams[arn].tags == [{"Key": "key", "Value": "value"}]
+
+    @mock_aws
+    def test_describe_delivery_stream(self):
+        # Generate S3 client
+        s3_client = client("s3", region_name=AWS_REGION_EU_WEST_1)
+        s3_client.create_bucket(
+            Bucket="test-bucket",
+            CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_WEST_1},
+        )
+
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        delivery_stream_name = arn.split("/")[-1]
+
+        firehose_client.start_delivery_stream_encryption(
+            DeliveryStreamName=delivery_stream_name,
+            DeliveryStreamEncryptionConfigurationInput={
+                "KeyARN": f"arn:aws:kms:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:key/test-kms-key-id",
+                "KeyType": "CUSTOMER_MANAGED_CMK",
+            },
+        )
+
+        # Firehose Client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+
+        assert len(firehose.delivery_streams) == 1
+        assert firehose.delivery_streams[arn].arn == arn
+        assert firehose.delivery_streams[arn].name == delivery_stream_name
+        assert firehose.delivery_streams[arn].region == AWS_REGION_EU_WEST_1
+        assert firehose.delivery_streams[arn].tags == [{"Key": "key", "Value": "value"}]
+        assert firehose.delivery_streams[arn].kms_encryption == EncryptionStatus.ENABLED
+        assert (
+            firehose.delivery_streams[arn].kms_key_arn
+            == f"arn:aws:kms:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:key/test-kms-key-id"
+        )

tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py

@@ -0,0 +1,195 @@
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_firehose_stream_encrypted_at_rest:
+    @mock_aws
+    def test_no_streams(self):
+        from prowler.providers.aws.services.firehose.firehose_service import Firehose
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest.firehose_client",
+            new=Firehose(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest import (
+                firehose_stream_encrypted_at_rest,
+            )
+
+            check = firehose_stream_encrypted_at_rest()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_stream_kms_encryption_enabled(self):
+        # Generate S3 client
+        s3_client = client("s3", region_name=AWS_REGION_EU_WEST_1)
+        s3_client.create_bucket(
+            Bucket="test-bucket",
+            CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_WEST_1},
+        )
+
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        stream_name = arn.split("/")[-1]
+
+        firehose_client.start_delivery_stream_encryption(
+            DeliveryStreamName=stream_name,
+            DeliveryStreamEncryptionConfigurationInput={
+                "KeyARN": f"arn:aws:kms:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:key/test-kms-key-id",
+                "KeyType": "CUSTOMER_MANAGED_CMK",
+            },
+        )
+
+        from prowler.providers.aws.services.firehose.firehose_service import Firehose
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest.firehose_client",
+                new=Firehose(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest import (
+                    firehose_stream_encrypted_at_rest,
+                )
+
+                check = firehose_stream_encrypted_at_rest()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Firehose Stream {stream_name} does have at rest encryption enabled."
+                )
+
+    @mock_aws
+    def test_stream_kms_encryption_not_enabled(self):
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        stream_name = arn.split("/")[-1]
+
+        from prowler.providers.aws.services.firehose.firehose_service import Firehose
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest.firehose_client",
+                new=Firehose(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest import (
+                    firehose_stream_encrypted_at_rest,
+                )
+
+                check = firehose_stream_encrypted_at_rest()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"Firehose Stream {stream_name} does not have at rest encryption enabled."
+                )
+
+    @mock_aws
+    def test_stream_kms_encryption_disabled(self):
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        stream_name = arn.split("/")[-1]
+
+        firehose_client.start_delivery_stream_encryption(
+            DeliveryStreamName=stream_name,
+            DeliveryStreamEncryptionConfigurationInput={
+                "KeyARN": f"arn:aws:kms:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:key/test-kms-key-id",
+                "KeyType": "CUSTOMER_MANAGED_CMK",
+            },
+        )
+
+        firehose_client.stop_delivery_stream_encryption(DeliveryStreamName=stream_name)
+
+        from prowler.providers.aws.services.firehose.firehose_service import Firehose
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest.firehose_client",
+                new=Firehose(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.firehose.firehose_stream_encrypted_at_rest.firehose_stream_encrypted_at_rest import (
+                    firehose_stream_encrypted_at_rest,
+                )
+
+                check = firehose_stream_encrypted_at_rest()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"Firehose Stream {stream_name} does not have at rest encryption enabled."
+                )