feat(neptune): add new check neptune_cluster_copy_tags_to_snapshots (#5062)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Daniel Barranquero
2024-09-18 15:23:44 +02:00
committed by GitHub
parent c425e8249b
commit f0cd924016
6 changed files with 181 additions and 2 deletions
@@ -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 <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": ""
}
@@ -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
@@ -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]
@@ -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 == []
@@ -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=[],
)