ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 21:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(rds): Add AWS RDS clusters to transport encryption check (#4028)

Browse Source
This commit is contained in:
madereddy
2024-05-21 09:22:39 -04:00
committed by GitHub
parent 716d38814f
commit 2f57f1f594
4 changed files with 206 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/rds/rds_instance_transport_encrypted/rds_instance_transport_encrypted.py
+19
View File
@@ -54,6 +54,25 @@ class rds_instance_transport_encrypted(Check):
):
report.status = "PASS"
report.status_extended = f"RDS Instance {db_instance.id} connections use SSL encryption."
findings.append(report)
for db_cluster in rds_client.db_clusters:
report = Check_Report_AWS(self.metadata())
report.region = rds_client.db_clusters[db_cluster].region
report.resource_id = rds_client.db_clusters[db_cluster].id
report.resource_arn = db_cluster
report.resource_tags = rds_client.db_clusters[db_cluster].tags
report.status = "FAIL"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections are not encrypted."
# Check RDS Clusters that support TLS encryption
if rds_client.db_clusters[db_cluster].force_ssl == "1":
report.status = "PASS"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections use SSL encryption."
if rds_client.db_clusters[db_cluster].require_secure_transport == "ON":
report.status = "PASS"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections use SSL encryption."
findings.append(report)
return findings
prowler/providers/aws/services/rds/rds_service.py
+24 -1
View File
@@ -17,6 +17,7 @@ class RDS(AWSService):
self.db_clusters = {}
self.db_snapshots = []
self.db_engines = {}
self.db_cluster_parameters = {}
self.db_cluster_snapshots = []
self.__threading_call__(self.__describe_db_instances__)
self.__threading_call__(self.__describe_db_certificate__)
@@ -190,6 +191,9 @@ class RDS(AWSService):
is_resource_filtered(db_cluster_arn, self.audit_resources)
):
if cluster["Engine"] != "docdb":
describe_db_parameters_paginator = (
regional_client.get_paginator("describe_db_parameters")
)
db_cluster = DBCluster(
id=cluster["DBClusterIdentifier"],
arn=db_cluster_arn,
@@ -213,8 +217,25 @@ class RDS(AWSService):
region=regional_client.region,
tags=cluster.get("TagList", []),
)
for page in describe_db_parameters_paginator.paginate(
DBParameterGroupName=cluster["DBClusterParameterGroup"]
):
for parameter in page["Parameters"]:
if parameter["ParameterName"] == "rds.force_ssl":
db_cluster.force_ssl = parameter[
"ParameterValue"
]
if (
parameter["ParameterName"]
== "require_secure_transport"
):
db_cluster.require_secure_transport = parameter[
"ParameterValue"
]
# We must use a unique value as the dict key to have unique keys
self.db_clusters[db_cluster_arn] = db_cluster
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -347,7 +368,9 @@ class DBCluster(BaseModel):
deletion_protection: bool
auto_minor_version_upgrade: bool
multi_az: bool
parameter_group: Optional[str]
parameter_group: str
force_ssl: Optional[bool]
require_secure_transport: Optional[str]
region: str
tags: Optional[list] = []
tests/providers/aws/services/rds/rds_instance_deletion_protection/rds_instance_deletion_protection_test.py
+12
View File
@@ -147,12 +147,18 @@ class Test_rds_instance_deletion_protection:
@mock_aws
def test_rds_instance_without_cluster_deletion_protection(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="postgres",
DatabaseName="staging-postgres",
DeletionProtection=False,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[
@@ -205,12 +211,18 @@ class Test_rds_instance_deletion_protection:
@mock_aws
def test_rds_instance_with_cluster_deletion_protection(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="postgres",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[
tests/providers/aws/services/rds/rds_instance_transport_encrypted/rds_instance_transport_encrypted_test.py
+151 -2
View File
@@ -112,6 +112,17 @@ class Test_rds_instance_transport_encrypted:
DBParameterGroupFamily="default.aurora-postgresql14",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
@@ -119,7 +130,7 @@ class Test_rds_instance_transport_encrypted:
DBName="aurora-postgres",
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
DBClusterIdentifier="cluster-postgres",
DBClusterIdentifier="db-cluster-1",
)
from prowler.providers.aws.services.rds.rds_service import RDS
@@ -141,7 +152,19 @@ class Test_rds_instance_transport_encrypted:
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 0
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections are 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_postgres_rds_instance_no_ssl(self):
@@ -390,3 +413,129 @@ class Test_rds_instance_transport_encrypted:
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_postgres_clustered_instance_ssl(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.aurora-postgresql14",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.modify_db_parameter_group(
DBParameterGroupName="test",
Parameters=[
{
"ParameterName": "rds.force_ssl",
"ParameterValue": "1",
"ApplyMethod": "immediate",
},
],
)
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_transport_encrypted.rds_instance_transport_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted import (
rds_instance_transport_encrypted,
)
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections use SSL encryption."
)
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_aurora_mysql_clustered_instance_ssl(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-mysql",
DatabaseName="staging-mysql",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.modify_db_parameter_group(
DBParameterGroupName="test",
Parameters=[
{
"ParameterName": "require_secure_transport",
"ParameterValue": "ON",
"ApplyMethod": "immediate",
},
],
)
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_transport_encrypted.rds_instance_transport_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted import (
rds_instance_transport_encrypted,
)
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections use SSL encryption."
)
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 == []