ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(aws): check AWS Owned keys in firehose_stream_encrypted_at_rest (#6121)

Browse Source 
Co-authored-by: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com>
This commit is contained in:
Prowler Bot
2024-12-10 19:30:41 +01:00
committed by GitHub
parent 8558034eae
commit d4e9940beb
2 changed files with 69 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest.py
+1 -4
View File
@@ -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."
tests/providers/aws/services/firehose/firehose_stream_encrypted_at_rest/firehose_stream_encrypted_at_rest_test.py
+68 -6
View File
@@ -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