feat(aws): Add new RDS check to verify that db instances copy tags to snapshots (#4806)

This commit is contained in:
Daniel Barranquero
2024-08-22 16:44:26 +02:00
committed by GitHub
parent dec3e652c5
commit 460acf2860
6 changed files with 216 additions and 0 deletions
@@ -0,0 +1,30 @@
{
"Provider": "aws",
"CheckID": "rds_instance_copy_tags_to_snapshots",
"CheckTitle": "Check if RDS DB instances have copy tags to snapshots enabled",
"CheckType": [],
"ServiceName": "rds",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:rds:region:account-id:db-instance",
"Severity": "low",
"ResourceType": "AwsRdsDbInstance",
"Description": "Check if RDS DB instances have copy tags to snapshots enabled, Aurora instances can not have this feature enabled at this level, they will have it at cluster level",
"Risk": "If RDS instances are not configured to copy tags to snapshots, it could lead to compliance issues as the snapshots will not inherit necessary metadata such as environment, owner, or purpose tags. This could result in inefficient tracking and management of RDS resources and their snapshots.",
"RelatedUrl": "https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_Tagging.html#USER_Tagging.CopyTags",
"Remediation": {
"Code": {
"CLI": "aws rds modify-db-instance --db-instance-identifier <instance-identifier> --copy-tags-to-snapshot",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/rds-controls.html#rds-17",
"Terraform": ""
},
"Recommendation": {
"Text": "Ensure that the `CopyTagsToSnapshot` setting is enabled for all RDS instances to propagate instance tags to their snapshots for improved tracking and compliance.",
"Url": "https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Overview.DBInstance.Modifying.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,28 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.rds.rds_client import rds_client
class rds_instance_copy_tags_to_snapshots(Check):
def execute(self):
findings = []
for db_instance in rds_client.db_instances:
if db_instance.engine not in [
"aurora",
"aurora-mysql",
"aurora-postgresql",
]:
report = Check_Report_AWS(self.metadata())
report.region = db_instance.region
report.resource_id = db_instance.id
report.resource_arn = db_instance.arn
report.resource_tags = db_instance.tags
if db_instance.copy_tags_to_snapshot:
report.status = "PASS"
report.status_extended = f"RDS Instance {db_instance.id} has copy tags to snapshots enabled."
else:
report.status = "FAIL"
report.status_extended = f"RDS Instance {db_instance.id} does not have copy tags to snapshots enabled."
findings.append(report)
return findings
@@ -98,6 +98,9 @@ class RDS(AWSService):
"ReadReplicaSourceDBInstanceIdentifier"
),
ca_cert=instance.get("CACertificateIdentifier"),
copy_tags_to_snapshot=instance.get(
"CopyTagsToSnapshot"
),
)
)
except Exception as error:
@@ -254,6 +257,9 @@ class RDS(AWSService):
),
region=regional_client.region,
tags=cluster.get("TagList", []),
copy_tags_to_snapshot=cluster.get(
"CopyTagsToSnapshot"
),
)
# We must use a unique value as the dict key to have unique keys
self.db_clusters[db_cluster_arn] = db_cluster
@@ -499,6 +505,7 @@ class DBInstance(BaseModel):
replica_source: Optional[str]
ca_cert: Optional[str]
cert: list[Certificate] = []
copy_tags_to_snapshot: Optional[bool]
class DBCluster(BaseModel):
@@ -522,6 +529,7 @@ class DBCluster(BaseModel):
require_secure_transport: str = "OFF"
region: str
tags: Optional[list] = []
copy_tags_to_snapshot: Optional[bool]
class DBSnapshot(BaseModel):
@@ -0,0 +1,146 @@
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
class Test_rds_instance_copy_tags_to_snapshots_to_snapshots:
@mock_aws
def test_rds_no_instances(self):
from prowler.providers.aws.services.rds.rds_service import RDS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots import (
rds_instance_copy_tags_to_snapshots,
)
check = rds_instance_copy_tags_to_snapshots()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_aurora_instance(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_instance(
DBInstanceIdentifier="test-instance",
Engine="aurora-postgresql",
DBInstanceClass="db.t2.micro",
AllocatedStorage=5,
)
from prowler.providers.aws.services.rds.rds_service import RDS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots import (
rds_instance_copy_tags_to_snapshots,
)
check = rds_instance_copy_tags_to_snapshots()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_instance_without_copy_tags(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_instance(
DBInstanceIdentifier="test-instance",
Engine="mysql",
DBInstanceClass="db.t2.micro",
AllocatedStorage=5,
CopyTagsToSnapshot=False,
)
from prowler.providers.aws.services.rds.rds_service import RDS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots import (
rds_instance_copy_tags_to_snapshots,
)
check = rds_instance_copy_tags_to_snapshots()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Instance test-instance does not have copy tags to snapshots enabled."
)
assert result[0].resource_id == "test-instance"
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:test-instance"
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_rds_instance_with_copy_tags(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_instance(
DBInstanceIdentifier="test-instance",
Engine="mysql",
DBInstanceClass="db.t2.micro",
AllocatedStorage=5,
CopyTagsToSnapshot=True,
)
from prowler.providers.aws.services.rds.rds_service import RDS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_instance_copy_tags_to_snapshots.rds_instance_copy_tags_to_snapshots import (
rds_instance_copy_tags_to_snapshots,
)
check = rds_instance_copy_tags_to_snapshots()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance test-instance has copy tags to snapshots enabled."
)
assert result[0].resource_id == "test-instance"
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:test-instance"
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -93,6 +93,7 @@ class Test_RDS_Service:
Tags=[
{"Key": "test", "Value": "test"},
],
CopyTagsToSnapshot=True,
)
# RDS client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -117,6 +118,7 @@ class Test_RDS_Service:
{"Key": "test", "Value": "test"},
]
assert "test" in rds.db_instances[0].parameter_groups
assert rds.db_instances[0].copy_tags_to_snapshot
@mock_aws
def test__describe_db_parameters__(self):
@@ -233,6 +235,7 @@ class Test_RDS_Service:
Tags=[
{"Key": "test", "Value": "test"},
],
CopyTagsToSnapshot=True,
)
# RDS client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -262,6 +265,7 @@ class Test_RDS_Service:
assert rds.db_clusters[db_cluster_arn].parameter_group == "test"
assert rds.db_clusters[db_cluster_arn].force_ssl == "0"
assert rds.db_clusters[db_cluster_arn].require_secure_transport == "OFF"
assert rds.db_clusters[db_cluster_arn].copy_tags_to_snapshot
# Test RDS Describe DB Cluster Snapshots
@mock_aws