feat(aws): Add new check to ensure RDS DB clusters are encrypted at rest (#4931)

This commit is contained in:
Daniel Barranquero
2024-09-10 19:40:08 +02:00
committed by GitHub
parent 2c6944176f
commit 6883467d2f
4 changed files with 187 additions and 0 deletions
@@ -0,0 +1,30 @@
{
"Provider": "aws",
"CheckID": "rds_cluster_storage_encrypted",
"CheckTitle": "Check if RDS clusters storage is encrypted.",
"CheckType": [],
"ServiceName": "rds",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:rds:region:account-id:db-cluster",
"Severity": "medium",
"ResourceType": "AwsRdsDbCluster",
"Description": "Check if RDS clusters storage is encrypted.",
"Risk": "If not enabled sensitive information at rest is not protected.",
"RelatedUrl": "https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Overview.Encryption.html",
"Remediation": {
"Code": {
"CLI": "aws rds create-db-cluster --db-cluster-identifier <db_cluster_id> --db-cluster-class <cluster_class> --engine <engine> --storage-encrypted true",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/rds-controls.html#rds-27",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable Encryption. Use a CMK where possible. It will provide additional management and privacy benefits.",
"Url": "https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Overview.Encryption.html#Overview.Encryption.Enabling"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,25 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.rds.rds_client import rds_client
class rds_cluster_storage_encrypted(Check):
def execute(self):
findings = []
for db_cluster_arn, db_cluster in rds_client.db_clusters.items():
report = Check_Report_AWS(self.metadata())
report.region = db_cluster.region
report.resource_id = db_cluster.id
report.resource_arn = db_cluster_arn
report.resource_tags = db_cluster.tags
if db_cluster.encrypted:
report.status = "PASS"
report.status_extended = f"RDS cluster {db_cluster.id} is encrypted."
else:
report.status = "FAIL"
report.status_extended = (
f"RDS cluster {db_cluster.id} is not encrypted."
)
findings.append(report)
return findings
@@ -0,0 +1,132 @@
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_cluster_storage_encrypted:
@mock_aws
def test_rds_no_clusters(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_cluster_storage_encrypted.rds_cluster_storage_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_cluster_storage_encrypted.rds_cluster_storage_encrypted import (
rds_cluster_storage_encrypted,
)
check = rds_cluster_storage_encrypted()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_cluster_no_encryption(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-mysql",
DatabaseName="staging-mysql",
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
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_cluster_storage_encrypted.rds_cluster_storage_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_cluster_storage_encrypted.rds_cluster_storage_encrypted import (
rds_cluster_storage_encrypted,
)
check = rds_cluster_storage_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS cluster db-cluster-1 is not encrypted."
)
assert result[0].resource_id == "db-cluster-1"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_cluster_with_encryption(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-mysql",
DatabaseName="staging-mysql",
StorageEncrypted=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
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_cluster_storage_encrypted.rds_cluster_storage_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_cluster_storage_encrypted.rds_cluster_storage_encrypted import (
rds_cluster_storage_encrypted,
)
check = rds_cluster_storage_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS cluster db-cluster-1 is encrypted."
)
assert result[0].resource_id == "db-cluster-1"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
)
assert result[0].resource_tags == []