feat(dynamodb): add new check dynamodb_accelerator_cluster_in_transit_encryption_enabled (#5173)

Co-authored-by: Sergio <sergio@prowler.com>

prowler/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_encryption_enabled/dynamodb_accelerator_cluster_encryption_enabled.py

@@ -12,11 +12,13 @@ class dynamodb_accelerator_cluster_encryption_enabled(Check):
             report.region = cluster.region
             report.resource_tags = cluster.tags
             report.status = "FAIL"
-            report.status_extended = f"DynamoDB cluster {cluster.name} does not have encryption at rest enabled."
+            report.status_extended = (
+                f"DAX cluster {cluster.name} does not have encryption at rest enabled."
+            )
             if cluster.encryption:
                 report.status = "PASS"
                 report.status_extended = (
-                    f"DynamoDB cluster {cluster.name} has encryption at rest enabled."
+                    f"DAX cluster {cluster.name} has encryption at rest enabled."
                 )
             findings.append(report)
         return findings

prowler/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_in_transit_encryption_enabled/dynamodb_accelerator_cluster_in_transit_encryption_enabled.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "dynamodb_accelerator_cluster_in_transit_encryption_enabled",
+  "CheckTitle": "Check if DynamoDB Accelerator (DAX) clusters are encrypted in transit.",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "dynamodb",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:dynamodb:region:account-id:table/table-name",
+  "Severity": "medium",
+  "ResourceType": "Other",
+  "Description": "This control checks whether an Amazon DynamoDB Accelerator (DAX) cluster is encrypted in transit, with the endpoint encryption type set to TLS. The control fails if the DAX cluster isn't encrypted in transit.",
+  "Risk": "Without encryption in transit, DAX clusters are vulnerable to person-in-the-middle attacks or eavesdropping on network traffic, which could lead to unauthorized access or manipulation of sensitive data.",
+  "RelatedUrl": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DAXEncryptionInTransit.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws dax create-cluster --cluster-name <cluster-name> --node-type <node-type> --replication-factor <replication-factor> --cluster-endpoint-encryption-type TLS",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dynamodb-controls.html#dynamodb-7",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that DynamoDB Accelerator (DAX) clusters are encrypted in transit by enabling TLS during cluster creation.",
+      "Url": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DAXEncryptionInTransit.html"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_in_transit_encryption_enabled/dynamodb_accelerator_cluster_in_transit_encryption_enabled.py

@@ -0,0 +1,22 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.dynamodb.dax_client import dax_client
+
+
+class dynamodb_accelerator_cluster_in_transit_encryption_enabled(Check):
+    def execute(self):
+        findings = []
+        for cluster in dax_client.clusters:
+            report = Check_Report_AWS(self.metadata())
+            report.resource_id = cluster.name
+            report.resource_arn = cluster.arn
+            report.region = cluster.region
+            report.resource_tags = cluster.tags
+            report.status = "FAIL"
+            report.status_extended = f"DAX cluster {cluster.name} does not have encryption in transit enabled."
+            if cluster.tls_encryption:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"DAX cluster {cluster.name} has encryption in transit enabled."
+                )
+            findings.append(report)
+        return findings

prowler/providers/aws/services/dynamodb/dynamodb_service.py

@@ -9,7 +9,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
 from prowler.providers.aws.lib.service.service import AWSService
 
 
-################## DynamoDB
 class DynamoDB(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
@@ -147,7 +146,6 @@ class DynamoDB(AWSService):
             )
 
 
-################## DynamoDB DAX
 class DAX(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
@@ -170,15 +168,20 @@ class DAX(AWSService):
                         )
                     ):
                         encryption = False
+                        tls_encryption = False
                         if "SSEDescription" in cluster:
                             if cluster["SSEDescription"]["Status"] == "ENABLED":
                                 encryption = True
+                        if "ClusterEndpointEncryptionType" in cluster:
+                            if cluster["ClusterEndpointEncryptionType"] == "TLS":
+                                tls_encryption = True
                         self.clusters.append(
                             Cluster(
                                 arn=cluster["ClusterArn"],
                                 name=cluster["ClusterName"],
                                 encryption=encryption,
                                 region=regional_client.region,
+                                tls_encryption=tls_encryption,
                             )
                         )
         except Exception as error:
@@ -224,3 +227,4 @@ class Cluster(BaseModel):
     encryption: bool
     region: str
     tags: Optional[list] = []
+    tls_encryption: bool

tests/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_in_transit_encryption_enabled/dynamodb_accelerator_cluster_in_transit_encryption_enabled_test.py

@@ -0,0 +1,123 @@
+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,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_dynamodb_accelerator_cluster_in_transit_encryption_enabled:
+    @mock_aws
+    def test_dax_no_clusters(self):
+        from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dax_client",
+            new=DAX(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled import (
+                dynamodb_accelerator_cluster_in_transit_encryption_enabled,
+            )
+
+            check = dynamodb_accelerator_cluster_in_transit_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_dax_cluster_no_encryption(self):
+        dax_client = client("dax", region_name=AWS_REGION_US_EAST_1)
+        iam_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/aws-service-role/dax.amazonaws.com/AWSServiceRoleForDAX"
+        cluster = dax_client.create_cluster(
+            ClusterName="daxcluster",
+            NodeType="dax.t3.small",
+            ReplicationFactor=3,
+            IamRoleArn=iam_role_arn,
+        )["Cluster"]
+        from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dax_client",
+            new=DAX(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled import (
+                dynamodb_accelerator_cluster_in_transit_encryption_enabled,
+            )
+
+            check = dynamodb_accelerator_cluster_in_transit_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "DAX cluster daxcluster does not have encryption in transit enabled."
+            )
+            assert result[0].resource_id == cluster["ClusterName"]
+            assert result[0].resource_arn == cluster["ClusterArn"]
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_dax_cluster_with_encryption(self):
+        dax_client = client("dax", region_name=AWS_REGION_US_EAST_1)
+        iam_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/aws-service-role/dax.amazonaws.com/AWSServiceRoleForDAX"
+        cluster = dax_client.create_cluster(
+            ClusterName="daxcluster",
+            NodeType="dax.t3.small",
+            ReplicationFactor=3,
+            IamRoleArn=iam_role_arn,
+            ClusterEndpointEncryptionType="TLS",
+        )["Cluster"]
+        from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dax_client",
+            new=DAX(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_in_transit_encryption_enabled.dynamodb_accelerator_cluster_in_transit_encryption_enabled import (
+                dynamodb_accelerator_cluster_in_transit_encryption_enabled,
+            )
+
+            check = dynamodb_accelerator_cluster_in_transit_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "DAX cluster daxcluster has encryption in transit enabled."
+            )
+            assert result[0].resource_id == cluster["ClusterName"]
+            assert result[0].resource_arn == cluster["ClusterArn"]
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags == []

tests/providers/aws/services/dynamodb/dynamodb_service_test.py

@@ -164,6 +164,7 @@ class Test_DynamoDB_Service:
             Tags=[
                 {"Key": "test", "Value": "test"},
             ],
+            ClusterEndpointEncryptionType="TLS",
         )
         dax_client.create_cluster(
             ClusterName="daxcluster2",
@@ -186,6 +187,7 @@ class Test_DynamoDB_Service:
         assert dax.clusters[0].tags == [
             {"Key": "test", "Value": "test"},
         ]
+        assert dax.clusters[0].tls_encryption
 
         assert dax.clusters[1].name == "daxcluster2"
         assert dax.clusters[1].region == AWS_REGION_US_EAST_1
@@ -193,3 +195,4 @@ class Test_DynamoDB_Service:
         assert dax.clusters[1].tags == [
             {"Key": "test", "Value": "test"},
         ]
+        assert not dax.clusters[1].tls_encryption