feat(DocumentDB): Add new DocumentDB check for cluster snapshot visibility (#4702)

2026-07-23 12:31:54 +00:00 · 2024-08-12 20:05:04 +02:00
parent bcc8d5f1fe
commit cb807e4aed
6 changed files with 335 additions and 6 deletions
@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "documentdb_cluster_public_snapshot",
+  "CheckTitle": "Check if DocumentDB manual cluster snapshot is public.",
+  "CheckType": [],
+  "ServiceName": "documentdb",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "critical",
+  "ResourceType": "AwsRDSDBClusterSnapshot",
+  "Description": "Check if DocumentDB manual cluster snapshot is public.",
+  "Risk": "If you share an unencrypted manual snapshot as public, the snapshot is available to all AWS accounts. Public snapshots may result in unintended data exposure.",
+  "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/docdb-cluster-snapshot-public-prohibited.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws docdb modify-db-snapshot-attribute --db-snapshot-identifier <snapshot_id> --attribute-name restore --values-to-remove all",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/documentdb-controls.html#documentdb-3",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To remove public access from a manual snapshot, follow the Sharing a snapshot tutorial.",
+      "Url": "https://docs.aws.amazon.com/documentdb/latest/developerguide/backup_restore-share_cluster_snapshots.html#backup_restore-share_snapshots"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,29 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.documentdb.documentdb_client import (
+    documentdb_client,
+)
+
+
+class documentdb_cluster_public_snapshot(Check):
+    def execute(self):
+        findings = []
+        for db_snap in documentdb_client.db_cluster_snapshots:
+            report = Check_Report_AWS(self.metadata())
+            report.region = db_snap.region
+            report.resource_id = db_snap.id
+            report.resource_arn = db_snap.arn
+            report.resource_tags = db_snap.tags
+            if db_snap.public:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"DocumentDB Cluster Snapshot {db_snap.id} is public."
+                )
+            else:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"DocumentDB Cluster Snapshot {db_snap.id} is not shared publicly."
+                )
+
+            findings.append(report)
+
+        return findings
@@ -1,5 +1,6 @@
 from typing import Optional

+from botocore.client import ClientError
 from pydantic import BaseModel

 from prowler.lib.logger import logger
@@ -15,11 +16,14 @@ class DocumentDB(AWSService):
        super().__init__(self.service_name, provider)
        self.db_instances = {}
        self.db_clusters = {}
-        self.__threading_call__(self.__describe_db_instances__)
-        self.__threading_call__(self.__describe_db_clusters__)
-        self.__list_tags_for_resource__()
+        self.db_cluster_snapshots = []
+        self.__threading_call__(self._describe_db_instances)
+        self.__threading_call__(self._describe_db_clusters)
+        self.__threading_call__(self._describe_db_cluster_snapshots)
+        self.__threading_call__(self._describe_db_cluster_snapshot_attributes)
+        self._list_tags_for_resource()

-    def __describe_db_instances__(self, regional_client):
+    def _describe_db_instances(self, regional_client):
        logger.info("DocumentDB - Describe Instances...")
        try:
            describe_db_instances_paginator = regional_client.get_paginator(
@@ -58,7 +62,7 @@ class DocumentDB(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

-    def __list_tags_for_resource__(self):
+    def _list_tags_for_resource(self):
        logger.info("DocumentDB - List Tags...")
        try:
            for instance_arn, instance in self.db_instances.items():
@@ -77,7 +81,7 @@ class DocumentDB(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

-    def __describe_db_clusters__(self, regional_client):
+    def _describe_db_clusters(self, regional_client):
        logger.info("DocumentDB - Describe Clusters...")
        try:
            describe_db_clusters_paginator = regional_client.get_paginator(
@@ -123,6 +127,62 @@ class DocumentDB(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

+    def _describe_db_cluster_snapshots(self, regional_client):
+        logger.info("DocumentDB - Describe Cluster Snapshots...")
+        try:
+            describe_db_snapshots_paginator = regional_client.get_paginator(
+                "describe_db_cluster_snapshots"
+            )
+            for page in describe_db_snapshots_paginator.paginate():
+                for snapshot in page["DBClusterSnapshots"]:
+                    arn = f"arn:{self.audited_partition}:rds:{regional_client.region}:{self.audited_account}:cluster-snapshot:{snapshot['DBClusterSnapshotIdentifier']}"
+                    if not self.audit_resources or (
+                        is_resource_filtered(
+                            arn,
+                            self.audit_resources,
+                        )
+                    ):
+                        if snapshot["Engine"] == "docdb":
+                            self.db_cluster_snapshots.append(
+                                ClusterSnapshot(
+                                    id=snapshot["DBClusterSnapshotIdentifier"],
+                                    arn=arn,
+                                    cluster_id=snapshot["DBClusterIdentifier"],
+                                    encrypted=snapshot.get("StorageEncrypted", False),
+                                    region=regional_client.region,
+                                    tags=snapshot.get("TagList", []),
+                                )
+                            )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_db_cluster_snapshot_attributes(self, regional_client):
+        logger.info("DocumentDB - Describe Cluster Snapshot Attributes...")
+        try:
+            for snapshot in self.db_cluster_snapshots:
+                if snapshot.region == regional_client.region:
+                    response = regional_client.describe_db_cluster_snapshot_attributes(
+                        DBClusterSnapshotIdentifier=snapshot.id
+                    )["DBClusterSnapshotAttributesResult"]
+                    for att in response["DBClusterSnapshotAttributes"]:
+                        if "all" in att["AttributeValues"]:
+                            snapshot.public = True
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "DBClusterSnapshotNotFoundFault":
+                logger.warning(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            else:
+                logger.error(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+

 class Instance(BaseModel):
    id: str
@@ -151,3 +211,14 @@ class DBCluster(BaseModel):
    parameter_group: str
    region: str
    tags: Optional[list] = []
+
+
+class ClusterSnapshot(BaseModel):
+    id: str
+    cluster_id: str
+    # arn:{partition}:rds:{region}:{account}:cluster-snapshot:{resource_id}
+    arn: str
+    public: bool = False
+    encrypted: bool
+    region: str
+    tags: Optional[list] = []
@@ -0,0 +1,140 @@
+from unittest import mock
+
+from prowler.providers.aws.services.documentdb.documentdb_service import (
+    ClusterSnapshot,
+    DBCluster,
+)
+
+AWS_ACCOUNT_NUMBER = "123456789012"
+AWS_REGION_US_EAST_1 = "us-east-1"
+
+DOC_DB_CLUSTER_NAME = "test-cluster"
+DOC_DB_CLUSTER_ARN = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:{DOC_DB_CLUSTER_NAME}"
+DOC_DB_ENGINE_VERSION = "5.0.0"
+
+
+class Test_documentdb_cluster_public_snapshot:
+    def test_documentdb_no_snapshot(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {}
+        documentdb_client.db_cluster_snapshots = []
+
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_public_snapshot.documentdb_cluster_public_snapshot import (
+                documentdb_cluster_public_snapshot,
+            )
+
+            check = documentdb_cluster_public_snapshot()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_documentdb_cluster_private_snapshot(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {
+            DOC_DB_CLUSTER_ARN: DBCluster(
+                id=DOC_DB_CLUSTER_NAME,
+                arn=DOC_DB_CLUSTER_ARN,
+                engine="docdb",
+                status="available",
+                backup_retention_period=0,
+                encrypted=False,
+                cloudwatch_logs=[],
+                multi_az=True,
+                parameter_group="default.docdb3.6",
+                deletion_protection=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        }
+        documentdb_client.db_cluster_snapshots = [
+            ClusterSnapshot(
+                id="snapshot-1",
+                arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1",
+                cluster_id=DOC_DB_CLUSTER_NAME,
+                encrypted=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        ]
+
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_public_snapshot.documentdb_cluster_public_snapshot import (
+                documentdb_cluster_public_snapshot,
+            )
+
+            check = documentdb_cluster_public_snapshot()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "DocumentDB Cluster Snapshot snapshot-1 is not shared publicly."
+            )
+            assert result[0].resource_id == "snapshot-1"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1"
+            )
+            assert result[0].resource_tags == []
+
+    def test_documentdb_cluster_public_snapshot(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {
+            DOC_DB_CLUSTER_ARN: DBCluster(
+                id=DOC_DB_CLUSTER_NAME,
+                arn=DOC_DB_CLUSTER_ARN,
+                engine="docdb",
+                status="available",
+                backup_retention_period=9,
+                encrypted=True,
+                cloudwatch_logs=[],
+                multi_az=True,
+                parameter_group="default.docdb3.6",
+                deletion_protection=True,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        }
+        documentdb_client.db_cluster_snapshots = [
+            ClusterSnapshot(
+                id="snapshot-1",
+                arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1",
+                cluster_id=DOC_DB_CLUSTER_NAME,
+                encrypted=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        ]
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_public_snapshot.documentdb_cluster_public_snapshot import (
+                documentdb_cluster_public_snapshot,
+            )
+
+            documentdb_client.db_cluster_snapshots[0].public = True
+            check = documentdb_cluster_public_snapshot()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "DocumentDB Cluster Snapshot snapshot-1 is public."
+            )
+            assert result[0].resource_id == "snapshot-1"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1"
+            )
+            assert result[0].resource_tags == []
@@ -2,6 +2,7 @@ import botocore
 from mock import patch

 from prowler.providers.aws.services.documentdb.documentdb_service import (
+    ClusterSnapshot,
    DBCluster,
    DocumentDB,
    Instance,
@@ -77,6 +78,29 @@ def mock_make_api_call(self, operation_name, kwargs):
                },
            ]
        }
+    if operation_name == "DescribeDBClusterSnapshots":
+        return {
+            "DBClusterSnapshots": [
+                {
+                    "DBClusterSnapshotIdentifier": "test-cluster-snapshot",
+                    "DBClusterIdentifier": DOC_DB_CLUSTER_ID,
+                    "Engine": "docdb",
+                    "Status": "available",
+                    "StorageEncrypted": True,
+                    "DBClusterSnapshotArn": f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:test-cluster-snapshot",
+                    "TagList": [{"Key": "snapshot", "Value": "test"}],
+                },
+            ]
+        }
+    if operation_name == "DescribeDBClusterSnapshotAttributes":
+        return {
+            "DBClusterSnapshotAttributesResult": {
+                "DBClusterSnapshotIdentifier": "test-cluster-snapshot",
+                "DBClusterSnapshotAttributes": [
+                    {"AttributeName": "restore", "AttributeValues": ["all"]}
+                ],
+            }
+        }
    return make_api_call(self, operation_name, kwargs)


@@ -158,3 +182,38 @@ class Test_DocumentDB_Service:
                tags=[],
            )
        }
+
+    # Test DocumentDB Describe DB Cluster Snapshots
+    def test_describe_db_cluster_snapshots(self):
+        aws_provider = set_mocked_aws_provider()
+        docdb = DocumentDB(aws_provider)
+        assert docdb.db_cluster_snapshots == [
+            ClusterSnapshot(
+                id="test-cluster-snapshot",
+                arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:test-cluster-snapshot",
+                cluster_id=DOC_DB_CLUSTER_ID,
+                public=True,
+                encrypted=True,
+                region=AWS_REGION_US_EAST_1,
+                tags=[{"Key": "snapshot", "Value": "test"}],
+            )
+        ]
+
+    # Test DocumentDB Describe DB Snapshot Attributes
+    def test_describe_db_cluster_snapshot_attributes(self):
+        aws_provider = set_mocked_aws_provider()
+        docdb = DocumentDB(aws_provider)
+        docdb.db_cluster_snapshots = [
+            ClusterSnapshot(
+                id="test-cluster-snapshot",
+                arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:test-cluster-snapshot",
+                cluster_id=DOC_DB_CLUSTER_ID,
+                encrypted=True,
+                region=AWS_REGION_US_EAST_1,
+                tags=[{"Key": "snapshot", "Value": "test"}],
+            )
+        ]
+        docdb._describe_db_cluster_snapshot_attributes(
+            docdb.regional_clients[AWS_REGION_US_EAST_1]
+        )
+        assert docdb.db_cluster_snapshots[0].public is True