feat(aws): Add new Neptune check for cluster snapshot visibility (#4709)

2026-07-24 13:01:56 +00:00 · 2024-08-13 14:27:35 +02:00
parent 62a1d91869
commit e2d211c188
6 changed files with 345 additions and 7 deletions
@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "neptune_cluster_public_snapshot",
+  "CheckTitle": "Check if NeptuneDB manual cluster snapshot is public.",
+  "CheckType": [],
+  "ServiceName": "neptune",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "critical",
+  "ResourceType": "AwsNeptuneDBClusterSnapshot",
+  "Description": "Check if NeptuneDB 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/neptune/latest/userguide/security-considerations.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws neptune modify-db-cluster-snapshot-attribute --db-cluster-snapshot-identifier <snapshot_id> --attribute-name restore --values-to-remove all",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/neptune-controls.html#neptune-3",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To remove public access from a manual snapshot, follow the AWS documentation on NeptuneDB snapshots.",
+      "Url": "https://docs.aws.amazon.com/neptune/latest/userguide/security-considerations.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.neptune.neptune_client import neptune_client
+
+
+class neptune_cluster_public_snapshot(Check):
+    def execute(self):
+        findings = []
+        for db_snap in neptune_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"NeptuneDB Cluster Snapshot {db_snap.id} is public."
+                )
+            else:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"NeptuneDB 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
@@ -14,11 +15,14 @@ class Neptune(AWSService):
        self.service_name = "neptune"
        super().__init__(self.service_name, provider)
        self.clusters = {}
-        self.__threading_call__(self.__describe_clusters__)
-        self.__threading_call__(self.__describe_db_subnet_groups__)
-        self.__list_tags_for_resource__()
+        self.db_cluster_snapshots = []
+        self.__threading_call__(self._describe_clusters)
+        self.__threading_call__(self._describe_db_subnet_groups)
+        self.__threading_call__(self._describe_db_cluster_snapshots)
+        self.__threading_call__(self._describe_db_cluster_snapshot_attributes)
+        self._list_tags_for_resource()

-    def __describe_clusters__(self, regional_client):
+    def _describe_clusters(self, regional_client):
        logger.info("Neptune - Describing DB Clusters...")
        try:
            for cluster in regional_client.describe_db_clusters(
@@ -54,7 +58,7 @@ class Neptune(AWSService):
                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

-    def __describe_db_subnet_groups__(self, regional_client):
+    def _describe_db_subnet_groups(self, regional_client):
        logger.info("Neptune - Describing DB Subnet Groups...")
        try:
            for cluster in self.clusters.values():
@@ -79,7 +83,7 @@ class Neptune(AWSService):
                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

-    def __list_tags_for_resource__(self):
+    def _list_tags_for_resource(self):
        logger.info("Neptune - Listing Tags...")
        try:
            for cluster in self.clusters.values():
@@ -97,6 +101,62 @@ class Neptune(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

+    def _describe_db_cluster_snapshots(self, regional_client):
+        logger.info("NeptuneDB - 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}:neptune:{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"] == "neptune":
+                            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("NeptuneDB - 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 Cluster(BaseModel):
    arn: str
@@ -112,3 +172,13 @@ class Cluster(BaseModel):
    db_subnet_group_id: str
    subnets: Optional[list]
    tags: Optional[list]
+
+
+class ClusterSnapshot(BaseModel):
+    id: str
+    cluster_id: str
+    arn: str
+    public: bool = False
+    encrypted: bool
+    region: str
+    tags: Optional[list] = []
@@ -0,0 +1,147 @@
+from unittest import mock
+
+from prowler.providers.aws.services.neptune.neptune_service import (
+    Cluster,
+    ClusterSnapshot,
+)
+
+AWS_ACCOUNT_NUMBER = "123456789012"
+AWS_REGION_US_EAST_1 = "us-east-1"
+
+NEPTUNE_CLUSTER_NAME = "test-cluster"
+NEPTUNE_CLUSTER_ARN = f"arn:aws:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:{NEPTUNE_CLUSTER_NAME}"
+
+
+class Test_neptune_cluster_public_snapshot:
+    def test_neptune_no_snapshot(self):
+        neptune_client = mock.MagicMock
+        neptune_client.clusters = {}
+        neptune_client.db_cluster_snapshots = []
+
+        with mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_service.Neptune",
+            new=neptune_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot.neptune_client",
+            new=neptune_client,
+        ):
+            from prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot import (
+                neptune_cluster_public_snapshot,
+            )
+
+            check = neptune_cluster_public_snapshot()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_neptune_cluster_private_snapshot(self):
+        neptune_client = mock.MagicMock
+        neptune_client.clusters = {
+            NEPTUNE_CLUSTER_ARN: Cluster(
+                name=NEPTUNE_CLUSTER_NAME,
+                arn=NEPTUNE_CLUSTER_ARN,
+                id="test-cluster-id",
+                backup_retention_period=7,
+                encrypted=True,
+                kms_key="kms-key-id",
+                multi_az=False,
+                iam_auth=False,
+                deletion_protection=False,
+                db_subnet_group_id="subnet-group",
+                region=AWS_REGION_US_EAST_1,
+            )
+        }
+        neptune_client.db_cluster_snapshots = [
+            ClusterSnapshot(
+                id="snapshot-1",
+                arn=f"arn:aws:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1",
+                cluster_id=NEPTUNE_CLUSTER_NAME,
+                encrypted=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        ]
+
+        with mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_service.Neptune",
+            new=neptune_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot.neptune_client",
+            new=neptune_client,
+        ):
+            from prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot import (
+                neptune_cluster_public_snapshot,
+            )
+
+            check = neptune_cluster_public_snapshot()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "NeptuneDB 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:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1"
+            )
+            assert result[0].resource_tags == []
+
+    def test_neptune_cluster_public_snapshot(self):
+        neptune_client = mock.MagicMock
+        neptune_client.clusters = {
+            NEPTUNE_CLUSTER_ARN: Cluster(
+                name=NEPTUNE_CLUSTER_NAME,
+                arn=NEPTUNE_CLUSTER_ARN,
+                id="test-cluster-id",
+                backup_retention_period=7,
+                encrypted=True,
+                kms_key="kms-key-id",
+                multi_az=False,
+                iam_auth=False,
+                deletion_protection=False,
+                db_subnet_group_id="subnet-group",
+                region=AWS_REGION_US_EAST_1,
+            )
+        }
+        neptune_client.db_cluster_snapshots = [
+            ClusterSnapshot(
+                id="snapshot-1",
+                arn=f"arn:aws:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1",
+                cluster_id=NEPTUNE_CLUSTER_NAME,
+                encrypted=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=[],
+            )
+        ]
+
+        with mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_service.Neptune",
+            new=neptune_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot.neptune_client",
+            new=neptune_client,
+        ):
+            from prowler.providers.aws.services.neptune.neptune_cluster_public_snapshot.neptune_cluster_public_snapshot import (
+                neptune_cluster_public_snapshot,
+            )
+
+            neptune_client.db_cluster_snapshots[0].public = True
+            check = neptune_cluster_public_snapshot()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "NeptuneDB 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:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:snapshot-1"
+            )
+            assert result[0].resource_tags == []
@@ -3,7 +3,11 @@ from boto3 import client
 from mock import patch
 from moto import mock_aws

-from prowler.providers.aws.services.neptune.neptune_service import Cluster, Neptune
+from prowler.providers.aws.services.neptune.neptune_service import (
+    Cluster,
+    ClusterSnapshot,
+    Neptune,
+)
 from tests.providers.aws.utils import (
    AWS_ACCOUNT_NUMBER,
    AWS_REGION_US_EAST_1,
@@ -66,6 +70,30 @@ def mock_make_api_call(self, operation_name, kwargs):
    if operation_name == "ListTagsForResource":
        return {"TagList": NEPTUNE_CLUSTER_TAGS}

+    if operation_name == "DescribeDBClusterSnapshots":
+        return {
+            "DBClusterSnapshots": [
+                {
+                    "DBClusterSnapshotIdentifier": "test-cluster-snapshot",
+                    "DBClusterIdentifier": NEPTUNE_CLUSTER_NAME,
+                    "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)


@@ -156,3 +184,39 @@ class Test_Neptune_Service:
            subnets=[SUBNET_1, SUBNET_2],
            tags=NEPTUNE_CLUSTER_TAGS,
        )
+
+    def test_describe_db_cluster_snapshots(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        neptune = Neptune(aws_provider)
+
+        expected_snapshot = ClusterSnapshot(
+            id="test-cluster-snapshot",
+            arn=f"arn:aws:neptune:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster-snapshot:test-cluster-snapshot",
+            cluster_id=NEPTUNE_CLUSTER_NAME,
+            encrypted=True,
+            region=AWS_REGION_US_EAST_1,
+            tags=[{"Key": "snapshot", "Value": "test"}],
+        )
+
+        neptune.db_cluster_snapshots = [expected_snapshot]
+
+        assert neptune.db_cluster_snapshots[0] == expected_snapshot
+
+    def test_describe_db_cluster_snapshot_attributes(self):
+        aws_provider = set_mocked_aws_provider()
+        neptune = Neptune(aws_provider)
+        neptune.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=NEPTUNE_CLUSTER_NAME,
+                encrypted=True,
+                region=AWS_REGION_US_EAST_1,
+                tags=[{"Key": "snapshot", "Value": "test"}],
+            )
+        ]
+        neptune._describe_db_cluster_snapshot_attributes(
+            neptune.regional_clients[AWS_REGION_US_EAST_1]
+        )
+        assert neptune.db_cluster_snapshots[0].public is True