From d4e9940beb4f51e01c92c8b1ffd44f611343e823 Mon Sep 17 00:00:00 2001
From: Prowler Bot <bot@prowler.com>
Date: Tue, 10 Dec 2024 19:30:41 +0100
Subject: [PATCH] fix(aws): check AWS Owned keys in
 `firehose_stream_encrypted_at_rest` (#6121)

Co-authored-by: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com>
---
 .../firehose_stream_encrypted_at_rest.py      |  5 +-
 .../firehose_stream_encrypted_at_rest_test.py | 74 +++++++++++++++++--
 2 files changed, 69 insertions(+), 10 deletions(-)

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 fe63558366..0430b3d184 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
@@ -31,10 +31,7 @@ class firehose_stream_encrypted_at_rest(Check):
                 f"Firehose Stream {stream.name} does have at rest encryption enabled."
             )
 
-            if (
-                stream.kms_encryption != EncryptionStatus.ENABLED
-                or not stream.kms_key_arn
-            ):
+            if stream.kms_encryption != EncryptionStatus.ENABLED:
                 report.status = "FAIL"
                 report.status_extended = f"Firehose Stream {stream.name} does not have at rest encryption enabled."
 
diff --git a/tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py b/tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py
index d2a1fa40a9..da00aa8900 100644
--- a/tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py
+++ b/tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py
@@ -17,12 +17,15 @@ class Test_firehose_stream_encrypted_at_rest:
 
         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),
+        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 (
@@ -94,6 +97,65 @@ class Test_firehose_stream_encrypted_at_rest:
                     == f"Firehose Stream {stream_name} does have at rest encryption enabled."
                 )
 
+    @mock_aws
+    def test_stream_kms_encryption_enabled_aws_managed_key(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={
+                "KeyType": "AWS_OWNED_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