diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/__init__.py b/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.metadata.json b/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.metadata.json new file mode 100644 index 0000000000..a67bd57531 --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.metadata.json @@ -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 --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": "" +} diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.py b/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.py new file mode 100644 index 0000000000..5b6563b19e --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot.py @@ -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 diff --git a/prowler/providers/aws/services/neptune/neptune_service.py b/prowler/providers/aws/services/neptune/neptune_service.py index 5b4bed0a6e..d6d706961b 100644 --- a/prowler/providers/aws/services/neptune/neptune_service.py +++ b/prowler/providers/aws/services/neptune/neptune_service.py @@ -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] = [] diff --git a/tests/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot_test.py b/tests/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot_test.py new file mode 100644 index 0000000000..caecd222ab --- /dev/null +++ b/tests/providers/aws/services/neptune/neptune_cluster_public_snapshot/neptune_cluster_public_snapshot_test.py @@ -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 == [] diff --git a/tests/providers/aws/services/neptune/neptune_service_test.py b/tests/providers/aws/services/neptune/neptune_service_test.py index 0427867916..c804c6853e 100644 --- a/tests/providers/aws/services/neptune/neptune_service_test.py +++ b/tests/providers/aws/services/neptune/neptune_service_test.py @@ -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