feat(redshift): add new check redshift_cluster_in_transit_encryption_enabled (#5271)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>

prowler/providers/aws/services/redshift/redshift_cluster_in_transit_encryption_enabled/redshift_cluster_in_transit_encryption_enabled.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "redshift_cluster_in_transit_encryption_enabled",
+  "CheckTitle": "Check if connections to Amazon Redshift clusters are encrypted in transit.",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "redshift",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:redshift:region:account-id:cluster/cluster-name",
+  "Severity": "medium",
+  "ResourceType": "AwsRedshiftCluster",
+  "Description": "This control checks whether connections to Amazon Redshift clusters are required to use encryption in transit. The control fails if the Redshift cluster parameter 'require_SSL' isn't set to True.",
+  "Risk": "Without encryption in transit, connections to the Redshift cluster are vulnerable to eavesdropping or person-in-the-middle attacks, exposing sensitive data to unauthorized access.",
+  "RelatedUrl": "https://docs.aws.amazon.com/redshift/latest/mgmt/security-encryption-in-transit.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws redshift modify-cluster-parameter-group --parameter-group-name <group-name> --parameters ParameterName=require_ssl,ParameterValue=true,ApplyType=static",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/redshift-controls.html#redshift-2",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that connections to Amazon Redshift clusters use encryption in transit by setting the 'require_ssl' parameter to True.",
+      "Url": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Redshift/redshift-parameter-groups-require-ssl.html"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/redshift/redshift_cluster_in_transit_encryption_enabled/redshift_cluster_in_transit_encryption_enabled.py

@@ -0,0 +1,26 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.redshift.redshift_client import redshift_client
+
+
+class redshift_cluster_in_transit_encryption_enabled(Check):
+    def execute(self):
+        findings = []
+        for cluster in redshift_client.clusters:
+            report = Check_Report_AWS(self.metadata())
+            report.region = cluster.region
+            report.resource_id = cluster.id
+            report.resource_arn = cluster.arn
+            report.resource_tags = cluster.tags
+            report.status = "FAIL"
+            report.status_extended = (
+                f"Redshift Cluster {cluster.id} is not encrypted in transit."
+            )
+            if cluster.require_ssl:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Redshift Cluster {cluster.id} is encrypted in transit."
+                )
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/redshift/redshift_service.py

@@ -13,8 +13,9 @@ class Redshift(AWSService):
         super().__init__(__class__.__name__, provider)
         self.clusters = []
         self.__threading_call__(self._describe_clusters)
-        self._describe_logging_status(self.regional_clients)
-        self._describe_cluster_snapshots(self.regional_clients)
+        self.__threading_call__(self._describe_logging_status, self.clusters)
+        self.__threading_call__(self._describe_cluster_snapshots, self.clusters)
+        self.__threading_call__(self._describe_cluster_parameters, self.clusters)
 
     def _describe_clusters(self, regional_client):
         logger.info("Redshift - describing clusters...")
@@ -41,6 +42,9 @@ class Redshift(AWSService):
                             tags=cluster.get("Tags"),
                             master_username=cluster.get("MasterUsername", ""),
                             database_name=cluster.get("DBName", ""),
+                            parameter_group_name=cluster.get(
+                                "ClusterParameterGroups", [{}]
+                            )[0].get("ParameterGroupName", ""),
                         )
                         self.clusters.append(cluster_to_append)
         except Exception as error:
@@ -48,37 +52,52 @@ class Redshift(AWSService):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
-    def _describe_logging_status(self, regional_clients):
+    def _describe_logging_status(self, cluster):
         logger.info("Redshift - describing logging status...")
         try:
-            for cluster in self.clusters:
-                regional_client = regional_clients[cluster.region]
-                cluster_attributes = regional_client.describe_logging_status(
-                    ClusterIdentifier=cluster.id
-                )
-                if (
-                    "LoggingEnabled" in cluster_attributes
-                    and cluster_attributes["LoggingEnabled"]
-                ):
-                    cluster.logging_enabled = True
-                if "BucketName" in cluster_attributes:
-                    cluster.bucket = cluster_attributes["BucketName"]
+            regional_client = self.regional_clients[cluster.region]
+            cluster_attributes = regional_client.describe_logging_status(
+                ClusterIdentifier=cluster.id
+            )
+            if (
+                "LoggingEnabled" in cluster_attributes
+                and cluster_attributes["LoggingEnabled"]
+            ):
+                cluster.logging_enabled = True
+            if "BucketName" in cluster_attributes:
+                cluster.bucket = cluster_attributes["BucketName"]
 
         except Exception as error:
             logger.error(
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
-    def _describe_cluster_snapshots(self, regional_clients):
+    def _describe_cluster_snapshots(self, cluster):
         logger.info("Redshift - describing logging status...")
         try:
-            for cluster in self.clusters:
-                regional_client = regional_clients[cluster.region]
-                cluster_snapshots = regional_client.describe_cluster_snapshots(
-                    ClusterIdentifier=cluster.id
-                )
-                if "Snapshots" in cluster_snapshots and cluster_snapshots["Snapshots"]:
-                    cluster.cluster_snapshots = True
+            regional_client = self.regional_clients[cluster.region]
+            cluster_snapshots = regional_client.describe_cluster_snapshots(
+                ClusterIdentifier=cluster.id
+            )
+            if "Snapshots" in cluster_snapshots and cluster_snapshots["Snapshots"]:
+                cluster.cluster_snapshots = True
+
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_cluster_parameters(self, cluster):
+        logger.info("Redshift - describing cluster parameter groups...")
+        try:
+            regional_client = self.regional_clients[cluster.region]
+            cluster_parameter_groups = regional_client.describe_cluster_parameters(
+                ClusterParameterGroupName=cluster.parameter_group_name
+            )
+            for parameter_group in cluster_parameter_groups["Parameters"]:
+                if parameter_group["ParameterName"].lower() == "require_ssl":
+                    if parameter_group["ParameterValue"].lower() == "true":
+                        cluster.require_ssl = True
 
         except Exception as error:
             logger.error(
@@ -100,3 +119,5 @@ class Cluster(BaseModel):
     bucket: str = None
     cluster_snapshots: bool = False
     tags: Optional[list] = []
+    parameter_group_name: str = None
+    require_ssl: bool = False

tests/providers/aws/services/redshift/redshift_cluster_in_transit_encryption_enabled/redshift_cluster_in_transit_encryption_enabled_test.py

@@ -0,0 +1,160 @@
+from unittest import mock
+from uuid import uuid4
+
+import botocore
+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,
+)
+
+CLUSTER_ID = str(uuid4())
+CLUSTER_ARN = (
+    f"arn:aws:redshift:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster:{CLUSTER_ID}"
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeClusterParameters":
+        return {
+            "Parameters": [
+                {
+                    "ParameterName": "require_ssl",
+                    "ParameterValue": "true",
+                    "Description": "Require SSL for connections",
+                    "Source": "user",
+                    "DataType": "boolean",
+                    "AllowedValues": "true, false",
+                    "IsModifiable": True,
+                    "MinimumEngineVersion": "1.0",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+class Test_redshift_cluster_in_transit_encryption_enabled:
+    def test_no_clusters(self):
+        from prowler.providers.aws.services.redshift.redshift_service import Redshift
+
+        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.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
+                new=Redshift(aws_provider),
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
+                    redshift_cluster_in_transit_encryption_enabled,
+                )
+
+                check = redshift_cluster_in_transit_encryption_enabled()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_cluster_not_encrypted_in_transit(self):
+        redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
+        redshift_client.create_cluster(
+            DBName="test",
+            ClusterIdentifier=CLUSTER_ID,
+            ClusterType="single-node",
+            NodeType="ds2.xlarge",
+            MasterUsername="awsuser",
+            MasterUserPassword="password",
+            PubliclyAccessible=True,
+            Tags=[
+                {"Key": "test", "Value": "test"},
+            ],
+            Port=9439,
+            Encrypted=False,
+        )
+        from prowler.providers.aws.services.redshift.redshift_service import Redshift
+
+        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.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
+                new=Redshift(aws_provider),
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
+                    redshift_cluster_in_transit_encryption_enabled,
+                )
+
+                check = redshift_cluster_in_transit_encryption_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert result[0].status_extended == (
+                    f"Redshift Cluster {CLUSTER_ID} is not encrypted in transit."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN
+                assert result[0].region == AWS_REGION_EU_WEST_1
+                assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]
+
+    @mock_aws
+    def test_cluster_encrypted_in_transit(self):
+        with mock.patch(
+            "botocore.client.BaseClient._make_api_call", new=mock_make_api_call
+        ):
+            redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
+            redshift_client.create_cluster(
+                DBName="test",
+                ClusterIdentifier=CLUSTER_ID,
+                ClusterType="single-node",
+                NodeType="ds2.xlarge",
+                MasterUsername="user",
+                MasterUserPassword="password",
+                PubliclyAccessible=True,
+                Tags=[
+                    {"Key": "test", "Value": "test"},
+                ],
+                Port=9439,
+                Encrypted=True,
+            )
+            from prowler.providers.aws.services.redshift.redshift_service import (
+                Redshift,
+            )
+
+            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.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
+                    new=Redshift(aws_provider),
+                ):
+                    from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
+                        redshift_cluster_in_transit_encryption_enabled,
+                    )
+
+                    check = redshift_cluster_in_transit_encryption_enabled()
+                    result = check.execute()
+
+                    assert len(result) == 1
+                    assert result[0].status == "PASS"
+                    assert result[0].status_extended == (
+                        f"Redshift Cluster {CLUSTER_ID} is encrypted in transit."
+                    )
+                    assert result[0].resource_id == CLUSTER_ID
+                    assert result[0].resource_arn == CLUSTER_ARN
+                    assert result[0].region == AWS_REGION_EU_WEST_1
+                    assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]

tests/providers/aws/services/redshift/redshift_service_test.py

@@ -43,6 +43,22 @@ def mock_make_api_call(self, operation_name, kwarg):
                 },
             ]
         }
+    if operation_name == "DescribeClusterParameters":
+        return {
+            "Parameters": [
+                {
+                    "ParameterName": "require_ssl",
+                    "ParameterValue": "true",
+                    "Description": "Require SSL for connections",
+                    "Source": "user",
+                    "DataType": "boolean",
+                    "AllowedValues": "true, false",
+                    "IsModifiable": True,
+                    "MinimumEngineVersion": "1.0",
+                },
+            ]
+        }
+
     return make_api_call(self, operation_name, kwarg)
 
 
@@ -94,6 +110,7 @@ class Test_Redshift_Service:
             Tags=[
                 {"Key": "test", "Value": "test"},
             ],
+            ClusterParameterGroupName="default.redshift-1.0",
         )
         aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
         redshift = Redshift(aws_provider)
@@ -113,6 +130,7 @@ class Test_Redshift_Service:
         assert redshift.clusters[0].tags == [
             {"Key": "test", "Value": "test"},
         ]
+        assert redshift.clusters[0].parameter_group_name == "default.redshift-1.0"
         assert redshift.clusters[0].encrypted
         assert redshift.clusters[0].master_username == "user"
         assert redshift.clusters[0].database_name == "test"
@@ -177,3 +195,39 @@ class Test_Redshift_Service:
         assert redshift.clusters[0].logging_enabled
         assert redshift.clusters[0].bucket == test_bucket_name
         assert redshift.clusters[0].cluster_snapshots
+
+    @mock_aws
+    def test_describe_cluster_parameter_groups(self):
+        redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
+        response = redshift_client.create_cluster(
+            DBName="test",
+            ClusterIdentifier=cluster_id,
+            ClusterType="single-node",
+            NodeType="ds2.xlarge",
+            MasterUsername="user",
+            MasterUserPassword="password",
+            PubliclyAccessible=True,
+            Tags=[
+                {"Key": "test", "Value": "test"},
+            ],
+        )
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        redshift = Redshift(aws_provider)
+
+        assert len(redshift.clusters) == 1
+        assert redshift.clusters[0].id == cluster_id
+        assert redshift.clusters[0].region == AWS_REGION_EU_WEST_1
+        assert redshift.clusters[0].public_access
+        assert (
+            redshift.clusters[0].endpoint_address
+            == response["Cluster"]["Endpoint"]["Address"]
+        )
+        assert (
+            redshift.clusters[0].allow_version_upgrade
+            == response["Cluster"]["AllowVersionUpgrade"]
+        )
+        assert redshift.clusters[0].tags == [
+            {"Key": "test", "Value": "test"},
+        ]
+        assert redshift.clusters[0].parameter_group_name == "default.redshift-1.0"
+        assert redshift.clusters[0].require_ssl is True