feat(redshift): add new check redshift_cluster_encrypted_at_rest (#5262)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 13:01:56 +00:00 · 2024-10-02 23:06:19 +02:00
parent 2ffe7f3ef7
commit 2177704b4b
11 changed files with 250 additions and 23 deletions
@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "redshift_cluster_encrypted_at_rest",
+  "CheckTitle": "Check if Redshift clusters are encrypted at rest.",
+  "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 Amazon Redshift clusters are encrypted at rest. The control fails if a Redshift cluster isn't encrypted at rest.",
+  "Risk": "Without encryption at rest, sensitive data stored in Redshift clusters is vulnerable to unauthorized access, which could lead to data breaches and regulatory non-compliance.",
+  "RelatedUrl": "https://docs.aws.amazon.com/redshift/latest/mgmt/working-with-db-encryption.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws redshift modify-cluster --cluster-identifier <cluster-id> --encrypted --kms-key-id <key-id>",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/redshift-controls.html#redshift-10",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable encryption at rest for your Redshift clusters using KMS to protect sensitive data from unauthorized access.",
+      "Url": "https://docs.aws.amazon.com/redshift/latest/mgmt/changing-cluster-encryption.html"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -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_encrypted_at_rest(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 at rest."
+            )
+            if cluster.encrypted:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Redshift Cluster {cluster.id} is encrypted at rest."
+                )
+
+            findings.append(report)
+
+        return findings
@@ -13,7 +13,7 @@ class redshift_cluster_non_default_username(Check):
            report.resource_tags = cluster.tags
            report.status = "PASS"
            report.status_extended = f"Redshift Cluster {cluster.id} does not have the default Admin username."
-            if cluster.masterusername == "awsuser":
+            if cluster.master_username == "awsuser":
                report.status = "FAIL"
                report.status_extended = (
                    f"Redshift Cluster {cluster.id} has the default Admin username."
@@ -7,7 +7,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
 from prowler.providers.aws.lib.service.service import AWSService


-################################ Redshift
 class Redshift(AWSService):
    def __init__(self, provider):
        # Call AWSService's __init__
@@ -30,24 +29,18 @@ class Redshift(AWSService):
                        cluster_to_append = Cluster(
                            arn=arn,
                            id=cluster["ClusterIdentifier"],
+                            endpoint_address=cluster.get("Endpoint", {}).get(
+                                "Address", ""
+                            ),
+                            public_access=cluster.get("PubliclyAccessible", False),
+                            allow_version_upgrade=cluster.get(
+                                "AllowVersionUpgrade", False
+                            ),
+                            encrypted=cluster.get("Encrypted", False),
                            region=regional_client.region,
                            tags=cluster.get("Tags"),
-                            masterusername=cluster.get("MasterUsername", ""),
+                            master_username=cluster.get("MasterUsername", ""),
                        )
-                        if (
-                            "PubliclyAccessible" in cluster
-                            and cluster["PubliclyAccessible"]
-                        ):
-                            cluster_to_append.public_access = True
-                        if "Endpoint" in cluster and "Address" in cluster["Endpoint"]:
-                            cluster_to_append.endpoint_address = cluster["Endpoint"][
-                                "Address"
-                            ]
-                        if (
-                            "AllowVersionUpgrade" in cluster
-                            and cluster["AllowVersionUpgrade"]
-                        ):
-                            cluster_to_append.allow_version_upgrade = True
                        self.clusters.append(cluster_to_append)
        except Exception as error:
            logger.error(
@@ -96,11 +89,12 @@ class Cluster(BaseModel):
    id: str
    arn: str
    region: str
-    public_access: bool = None
-    masterusername: str = None
+    public_access: bool = False
+    encrypted: bool = False
+    master_username: str = None
    endpoint_address: str = None
-    allow_version_upgrade: bool = None
-    logging_enabled: bool = None
+    allow_version_upgrade: bool = False
+    logging_enabled: bool = False
    bucket: str = None
-    cluster_snapshots: bool = None
+    cluster_snapshots: bool = False
    tags: Optional[list] = []
@@ -17,6 +17,9 @@ class Test_redshift_cluster_audit_logging:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_audit_logging.redshift_cluster_audit_logging import (
                redshift_cluster_audit_logging,
@@ -40,6 +43,9 @@ class Test_redshift_cluster_audit_logging:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_audit_logging.redshift_cluster_audit_logging import (
                redshift_cluster_audit_logging,
@@ -70,6 +76,9 @@ class Test_redshift_cluster_audit_logging:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_audit_logging.redshift_cluster_audit_logging import (
                redshift_cluster_audit_logging,
@@ -17,6 +17,9 @@ class Test_redshift_cluster_automated_snapshot:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automated_snapshot.redshift_cluster_automated_snapshot import (
                redshift_cluster_automated_snapshot,
@@ -40,6 +43,9 @@ class Test_redshift_cluster_automated_snapshot:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automated_snapshot.redshift_cluster_automated_snapshot import (
                redshift_cluster_automated_snapshot,
@@ -69,6 +75,9 @@ class Test_redshift_cluster_automated_snapshot:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automated_snapshot.redshift_cluster_automated_snapshot import (
                redshift_cluster_automated_snapshot,
@@ -17,6 +17,9 @@ class Test_redshift_cluster_automatic_upgrades:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automatic_upgrades.redshift_cluster_automatic_upgrades import (
                redshift_cluster_automatic_upgrades,
@@ -40,6 +43,9 @@ class Test_redshift_cluster_automatic_upgrades:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automatic_upgrades.redshift_cluster_automatic_upgrades import (
                redshift_cluster_automatic_upgrades,
@@ -69,6 +75,9 @@ class Test_redshift_cluster_automatic_upgrades:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_automatic_upgrades.redshift_cluster_automatic_upgrades import (
                redshift_cluster_automatic_upgrades,
@@ -0,0 +1,132 @@
+from unittest import mock
+from uuid import uuid4
+
+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}"
+)
+
+
+class Test_redshift_cluster_encrypted:
+    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_encrypted_at_rest.redshift_cluster_encrypted_at_rest.redshift_client",
+                new=Redshift(aws_provider),
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_encrypted_at_rest.redshift_cluster_encrypted_at_rest import (
+                    redshift_cluster_encrypted_at_rest,
+                )
+
+                check = redshift_cluster_encrypted_at_rest()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_cluster_not_encrypted(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="user",
+            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_encrypted_at_rest.redshift_cluster_encrypted_at_rest.redshift_client",
+                new=Redshift(aws_provider),
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_encrypted_at_rest.redshift_cluster_encrypted_at_rest import (
+                    redshift_cluster_encrypted_at_rest,
+                )
+
+                check = redshift_cluster_encrypted_at_rest()
+                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 at rest."
+                )
+                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(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="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_encrypted_at_rest.redshift_cluster_encrypted_at_rest.redshift_client",
+                new=Redshift(aws_provider),
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_encrypted_at_rest.redshift_cluster_encrypted_at_rest import (
+                    redshift_cluster_encrypted_at_rest,
+                )
+
+                check = redshift_cluster_encrypted_at_rest()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert result[0].status_extended == (
+                    f"Redshift Cluster {CLUSTER_ID} is encrypted at rest."
+                )
+                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"}]
@@ -17,6 +17,9 @@ class Test_redshift_cluster_public_access:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
                redshift_cluster_public_access,
@@ -41,6 +44,9 @@ class Test_redshift_cluster_public_access:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
                redshift_cluster_public_access,
@@ -71,6 +77,9 @@ class Test_redshift_cluster_public_access:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
                redshift_cluster_public_access,
@@ -100,6 +109,9 @@ class Test_redshift_cluster_public_access:
        with mock.patch(
            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
            redshift_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
+            redshift_client,
        ):
            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
                redshift_cluster_public_access,
@@ -90,6 +90,7 @@ class Test_Redshift_Service:
            MasterUsername="user",
            MasterUserPassword="password",
            PubliclyAccessible=True,
+            Encrypted=True,
            Tags=[
                {"Key": "test", "Value": "test"},
            ],
@@ -112,7 +113,8 @@ class Test_Redshift_Service:
        assert redshift.clusters[0].tags == [
            {"Key": "test", "Value": "test"},
        ]
-        assert redshift.clusters[0].masterusername == "user"
+        assert redshift.clusters[0].encrypted
+        assert redshift.clusters[0].master_username == "user"

    @mock_aws
    def test_describe_logging_status(self):