feat(dms): new check dms_endpoint_ssl_enabled (#4968)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com> Co-authored-by: Rubén De la Torre Vico <rubendltv22@gmail.com>
2026-07-24 21:11:53 +00:00 · 2024-09-18 22:46:56 +01:00
parent 2b41445d57
commit 34679c98d6
5 changed files with 234 additions and 0 deletions
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "dms_endpoint_ssl_enabled",
+  "CheckTitle": "Ensure SSL mode is enabled in DMS endpoint",
+  "CheckType": ["Effects", "Data Exposure"],
+  "ServiceName": "dms",
+  "SubServiceName": "endpoint",
+  "ResourceIdTemplate": "arn:partition:dms:region:account-id:endpoint:resource-id",
+  "Severity": "high",
+  "ResourceType": "AwsDmsEndpoint",
+  "Description": "This check ensures that SSL mode is enabled for all AWS Database Migration Service (DMS) endpoints. Enabling SSL provides encryption in transit for data transferred through these endpoints.",
+  "Risk": "Without SSL enabled, data transferred through DMS endpoints is not encrypted, potentially exposing sensitive information to unauthorized access or interception during transit.",
+  "RelatedUrl": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Security.SSL.html",
+  "Remediation": {
+      "Code": {
+          "CLI": "aws dms modify-endpoint --endpoint-arn <endpoint_arn> --ssl-mode require",
+          "NativeIaC": "",
+          "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dms-controls.html#dms-9",
+          "Terraform": ""
+      },
+      "Recommendation": {
+          "Text": "Enable SSL mode for all DMS endpoints. Use 'require' as the minimum SSL mode, and consider using 'verify-ca' or 'verify-full' for higher security.",
+          "Url": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Security.SSL.html"
+      }
+  },
+  "Categories": [
+      "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,25 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.dms.dms_client import dms_client
+
+
+class dms_endpoint_ssl_enabled(Check):
+    def execute(self):
+        findings = []
+        for endpoint_arn, endpoint in dms_client.endpoints.items():
+            report = Check_Report_AWS(self.metadata())
+            report.resource_id = endpoint.id
+            report.resource_arn = endpoint_arn
+            report.region = dms_client.audited_region
+
+            if endpoint.ssl_mode == "none":
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"DMS Endpoint {endpoint_arn} is not using SSL."
+                )
+            else:
+                report.status = "PASS"
+                report.status_extended = f"DMS Endpoint {endpoint_arn} is using SSL with mode: {endpoint.ssl_mode}."
+
+            findings.append(report)
+
+        return findings
@@ -11,7 +11,9 @@ class DMS(AWSService):
        # Call AWSService's __init__
        super().__init__(__class__.__name__, provider)
        self.instances = []
+        self.endpoints = {}
        self.__threading_call__(self._describe_replication_instances)
+        self.__threading_call__(self._describe_endpoints)

    def _describe_replication_instances(self, regional_client):
        logger.info("DMS - Describing DMS Replication Instances...")
@@ -49,6 +51,32 @@ class DMS(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

+    def _describe_endpoints(self, regional_client):
+        logger.info("DMS - Describing DMS Endpoints...")
+        try:
+            describe_endpoints_paginator = regional_client.get_paginator(
+                "describe_endpoints"
+            )
+            for page in describe_endpoints_paginator.paginate():
+                for endpoint in page["Endpoints"]:
+                    arn = endpoint["EndpointArn"]
+                    if not self.audit_resources or (
+                        is_resource_filtered(arn, self.audit_resources)
+                    ):
+                        self.endpoints[arn] = Endpoint(
+                            id=endpoint["EndpointIdentifier"],
+                            ssl_mode=endpoint.get("SslMode", False),
+                        )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class Endpoint(BaseModel):
+    id: str
+    ssl_mode: str
+

 class RepInstance(BaseModel):
    id: str
@@ -0,0 +1,149 @@
+from unittest import mock
+
+from prowler.providers.aws.services.dms.dms_service import Endpoint
+from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
+
+
+class Test_dms_endpoint_ssl_enabled:
+
+    def test_dms_no_endpoints(self):
+        dms_client = mock.MagicMock
+        dms_client.endpoints = {}
+
+        with mock.patch(
+            "prowler.providers.aws.services.dms.dms_service.DMS",
+            new=dms_client,
+        ):
+            from prowler.providers.aws.services.dms.dms_endpoint_ssl_enabled.dms_endpoint_ssl_enabled import (
+                dms_endpoint_ssl_enabled,
+            )
+
+            check = dms_endpoint_ssl_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_dms_endpoint_ssl_none(self):
+        dms_client = mock.MagicMock
+        dms_client.endpoints = {
+            "test-endpoint-no-ssl": Endpoint(id="test-endpoint-no-ssl", ssl_mode="none")
+        }
+        dms_client.audited_account = AWS_ACCOUNT_NUMBER
+        dms_client.audited_partition = "aws"
+        dms_client.audited_region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.aws.services.dms.dms_service.DMS",
+            new=dms_client,
+        ):
+            from prowler.providers.aws.services.dms.dms_endpoint_ssl_enabled.dms_endpoint_ssl_enabled import (
+                dms_endpoint_ssl_enabled,
+            )
+
+            check = dms_endpoint_ssl_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].resource_id == "test-endpoint-no-ssl"
+            assert result[0].resource_arn == "test-endpoint-no-ssl"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].status_extended
+                == "DMS Endpoint test-endpoint-no-ssl is not using SSL."
+            )
+
+    def test_dms_endpoint_ssl_require(self):
+        dms_client = mock.MagicMock
+        dms_client.endpoints = {
+            "test-endpoint-ssl-require": Endpoint(
+                id="test-endpoint-ssl-require", ssl_mode="require"
+            )
+        }
+        dms_client.audited_account = AWS_ACCOUNT_NUMBER
+        dms_client.audited_partition = "aws"
+        dms_client.audited_region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.aws.services.dms.dms_service.DMS",
+            new=dms_client,
+        ):
+            from prowler.providers.aws.services.dms.dms_endpoint_ssl_enabled.dms_endpoint_ssl_enabled import (
+                dms_endpoint_ssl_enabled,
+            )
+
+            check = dms_endpoint_ssl_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == "test-endpoint-ssl-require"
+            assert result[0].resource_arn == "test-endpoint-ssl-require"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].status_extended
+                == "DMS Endpoint test-endpoint-ssl-require is using SSL with mode: require."
+            )
+
+    def test_dms_endpoint_ssl_verify_ca(self):
+        dms_client = mock.MagicMock
+        dms_client.endpoints = {
+            "test-endpoint-ssl-verify-ca": Endpoint(
+                id="test-endpoint-ssl-verify-ca", ssl_mode="verify-ca"
+            )
+        }
+        dms_client.audited_account = AWS_ACCOUNT_NUMBER
+        dms_client.audited_partition = "aws"
+        dms_client.audited_region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.aws.services.dms.dms_service.DMS",
+            new=dms_client,
+        ):
+            from prowler.providers.aws.services.dms.dms_endpoint_ssl_enabled.dms_endpoint_ssl_enabled import (
+                dms_endpoint_ssl_enabled,
+            )
+
+            check = dms_endpoint_ssl_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == "test-endpoint-ssl-verify-ca"
+            assert result[0].resource_arn == "test-endpoint-ssl-verify-ca"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].status_extended
+                == "DMS Endpoint test-endpoint-ssl-verify-ca is using SSL with mode: verify-ca."
+            )
+
+    def test_dms_endpoint_ssl_verify_full(self):
+        dms_client = mock.MagicMock
+        dms_client.endpoints = {
+            "test-endpoint-ssl-verify-full": Endpoint(
+                id="test-endpoint-ssl-verify-full", ssl_mode="verify-full"
+            )
+        }
+        dms_client.audited_account = AWS_ACCOUNT_NUMBER
+        dms_client.audited_partition = "aws"
+        dms_client.audited_region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.aws.services.dms.dms_service.DMS",
+            new=dms_client,
+        ):
+            from prowler.providers.aws.services.dms.dms_endpoint_ssl_enabled.dms_endpoint_ssl_enabled import (
+                dms_endpoint_ssl_enabled,
+            )
+
+            check = dms_endpoint_ssl_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == "test-endpoint-ssl-verify-full"
+            assert result[0].resource_arn == "test-endpoint-ssl-verify-full"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].status_extended
+                == "DMS Endpoint test-endpoint-ssl-verify-full is using SSL with mode: verify-full."
+            )