diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/__init__.py b/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.metadata.json b/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.metadata.json new file mode 100644 index 0000000000..28eaa61069 --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.metadata.json @@ -0,0 +1,34 @@ +{ + "Provider": "aws", + "CheckID": "neptune_cluster_copy_tags_to_snapshots", + "CheckTitle": "Check if Neptune DB clusters are configured to copy tags to snapshots.", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "neptune", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:rds:region:account-id:cluster:db-cluster-identifier", + "Severity": "low", + "ResourceType": "AwsRdsDbCluster", + "Description": "This check ensures that Neptune DB clusters are configured to copy all tags to snapshots when the snapshots are created.", + "Risk": "If tags are not copied to snapshots, the snapshots may lack necessary metadata for identification, governance, and access control, leading to potential mismanagement and security risks.", + "RelatedUrl": "https://docs.aws.amazon.com/neptune/latest/userguide/tagging.html#tagging-overview", + "Remediation": { + "Code": { + "CLI": "aws neptune modify-db-cluster --db-cluster-identifier --copy-tags-to-snapshot --apply-immediately", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/neptune-controls.html#neptune-8", + "Terraform": "" + }, + "Recommendation": { + "Text": "Configure your Neptune DB clusters to copy tags to snapshots when the snapshots are created.", + "Url": "https://docs.aws.amazon.com/neptune/latest/userguide/tagging.html#tagging-overview" + } + }, + "Categories": [ + "trustboundaries" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.py b/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.py new file mode 100644 index 0000000000..e9982b663b --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots.py @@ -0,0 +1,22 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.neptune.neptune_client import neptune_client + + +class neptune_cluster_copy_tags_to_snapshots(Check): + def execute(self): + findings = [] + for cluster_arn, cluster in neptune_client.clusters.items(): + 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"Neptune DB Cluster {cluster.id} is not configured to copy tags to snapshots." + if cluster.copy_tags_to_snapshot: + report.status = "PASS" + report.status_extended = f"Neptune DB Cluster {cluster.id} is configured to copy tags to snapshots." + + 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 106fd9caa2..90b8969aa4 100644 --- a/prowler/providers/aws/services/neptune/neptune_service.py +++ b/prowler/providers/aws/services/neptune/neptune_service.py @@ -50,10 +50,10 @@ class Neptune(AWSService): multi_az=cluster["MultiAZ"], iam_auth=cluster.get("IAMDatabaseAuthenticationEnabled", False), deletion_protection=cluster.get("DeletionProtection", False), + copy_tags_to_snapshot=cluster.get("CopyTagsToSnapshot", False), db_subnet_group_id=cluster["DBSubnetGroup"], region=regional_client.region, ) - except Exception as error: logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" @@ -169,6 +169,7 @@ class Cluster(BaseModel): multi_az: bool iam_auth: bool deletion_protection: bool + copy_tags_to_snapshot: Optional[bool] region: str db_subnet_group_id: str subnets: Optional[list] diff --git a/tests/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots_test.py b/tests/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots_test.py new file mode 100644 index 0000000000..98ce7bc49b --- /dev/null +++ b/tests/providers/aws/services/neptune/neptune_cluster_copy_tags_to_snapshots/neptune_cluster_copy_tags_to_snapshots_test.py @@ -0,0 +1,121 @@ +from unittest import mock + +from prowler.providers.aws.services.neptune.neptune_service import Cluster +from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1 + + +class Test_neptune_cluster_copy_tags_to_snapshots: + def test_neptune_no_clusters(self): + neptune_client = mock.MagicMock + neptune_client.clusters = {} + + with mock.patch( + "prowler.providers.aws.services.neptune.neptune_service.Neptune", + new=neptune_client, + ), mock.patch( + "prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots.neptune_client", + new=neptune_client, + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots import ( + neptune_cluster_copy_tags_to_snapshots, + ) + + check = neptune_cluster_copy_tags_to_snapshots() + result = check.execute() + + assert len(result) == 0 + + def test_neptune_cluster_copy_tags_disabled(self): + neptune_client = mock.MagicMock + cluster_arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1" + neptune_client.clusters = { + cluster_arn: Cluster( + arn=cluster_arn, + name="db-cluster-1", + id="db-cluster-1", + region=AWS_REGION_US_EAST_1, + tags=[], + copy_tags_to_snapshot=False, + backup_retention_period=7, + encrypted=True, + kms_key="arn:aws:kms:us-east-1:123456789012:key/abcd1234-a123-456a-a12b-a123b4cd56ef", + multi_az=False, + iam_auth=False, + deletion_protection=False, + db_subnet_group_id="subnet-1234abcd", + ) + } + + with mock.patch( + "prowler.providers.aws.services.neptune.neptune_service.Neptune", + new=neptune_client, + ), mock.patch( + "prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots.neptune_client", + new=neptune_client, + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots import ( + neptune_cluster_copy_tags_to_snapshots, + ) + + check = neptune_cluster_copy_tags_to_snapshots() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "Neptune DB Cluster db-cluster-1 is not configured to copy tags to snapshots." + ) + assert result[0].resource_id == "db-cluster-1" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_arn == cluster_arn + assert result[0].resource_tags == [] + + def test_neptune_cluster_copy_tags_enabled(self): + neptune_client = mock.MagicMock + cluster_arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-2" + neptune_client.clusters = { + cluster_arn: Cluster( + arn=cluster_arn, + name="db-cluster-2", + id="db-cluster-2", + region=AWS_REGION_US_EAST_1, + tags=[], + copy_tags_to_snapshot=True, + backup_retention_period=7, + encrypted=True, + kms_key="arn:aws:kms:us-east-1:123456789012:key/abcd1234-a123-456a-a12b-a123b4cd56ef", + multi_az=False, + iam_auth=False, + deletion_protection=False, + db_subnet_group_id="subnet-1234abcd", + ) + } + + with mock.patch( + "prowler.providers.aws.services.neptune.neptune_service.Neptune", + new=neptune_client, + ), mock.patch( + "prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots.neptune_client", + new=neptune_client, + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_copy_tags_to_snapshots.neptune_cluster_copy_tags_to_snapshots import ( + neptune_cluster_copy_tags_to_snapshots, + ) + + check = neptune_cluster_copy_tags_to_snapshots() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Neptune DB Cluster db-cluster-2 is configured to copy tags to snapshots." + ) + assert result[0].resource_id == "db-cluster-2" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_arn == cluster_arn + 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 802b261569..7f893aa6fd 100644 --- a/tests/providers/aws/services/neptune/neptune_service_test.py +++ b/tests/providers/aws/services/neptune/neptune_service_test.py @@ -149,7 +149,7 @@ class Test_Neptune_Service: cluster = neptune_client.create_db_cluster( AvailabilityZones=[AWS_REGION_US_EAST_1_AZA, AWS_REGION_US_EAST_1_AZB], BackupRetentionPeriod=1, - CopyTagsToSnapshot=True, + CopyTagsToSnapshot=False, Engine=NEPTUNE_ENGINE, DatabaseName=NEPTUNE_CLUSTER_NAME, DBClusterIdentifier=NEPTUNE_CLUSTER_NAME, @@ -183,6 +183,7 @@ class Test_Neptune_Service: db_subnet_group_id=SUBNET_GROUP_NAME, subnets=[SUBNET_1, SUBNET_2], tags=NEPTUNE_CLUSTER_TAGS, + copy_tags_to_snapshot=False, cloudwatch_logs=[], )