feat(RDS): Additional RDS checks (#4233)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
sansns-aws
2024-06-20 13:41:08 -04:00
committed by GitHub
co-authored by Sergio Garcia Sergio
parent 95a9f1c458
commit 9ca64e7bdb
19 changed files with 1440 additions and 0 deletions
@@ -0,0 +1,181 @@
from unittest import mock
import botocore
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,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeDBEngineVersions":
return {
"DBEngineVersions": [
{
"Engine": "mysql",
"EngineVersion": "8.0.32",
"DBEngineDescription": "description",
"DBEngineVersionDescription": "description",
},
]
}
return make_api_call(self, operation_name, kwarg)
# Currently have to mock the tests as moto does not return the value for backtrack. Issue: https://github.com/getmoto/moto/issues/7734
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
class Test_rds_cluster_backtrack_enabled:
@mock_aws
def test_no_rds_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_backtrack_enabled.rds_cluster_backtrack_enabled.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_cluster_backtrack_enabled.rds_cluster_backtrack_enabled import (
rds_cluster_backtrack_enabled,
)
check = rds_cluster_backtrack_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_cluster_aurora_mysql_backtrack_disabled(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_cluster_backtrack_enabled.rds_cluster_backtrack_enabled.rds_client",
new=RDS(aws_provider),
):
from prowler.providers.aws.services.rds.rds_cluster_backtrack_enabled.rds_cluster_backtrack_enabled import (
rds_cluster_backtrack_enabled,
)
check = rds_cluster_backtrack_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 does not have backtrack enabled."
)
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_aurora_mysql_backtrack_enabled(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",
},
],
)
db_cluster = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
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_backtrack_enabled.rds_cluster_backtrack_enabled.rds_client",
new=RDS(aws_provider),
) as service_client:
from prowler.providers.aws.services.rds.rds_cluster_backtrack_enabled.rds_cluster_backtrack_enabled import (
rds_cluster_backtrack_enabled,
)
service_client.db_clusters[db_cluster].backtrack = 1
check = rds_cluster_backtrack_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 has backtrack enabled."
)
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 == []
@@ -76,6 +76,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -145,6 +147,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -213,6 +217,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -281,6 +287,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -349,6 +357,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -417,6 +427,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -485,6 +497,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -553,6 +567,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -621,6 +637,8 @@ class Test_rds_instance_certificate_expiration:
deletion_protection=True,
auto_minor_version_upgrade=False,
multi_az=True,
username="test",
iam_auth=False,
region=AWS_REGION,
ca_cert="rds-ca-rsa2048-g1",
cert=[
@@ -0,0 +1,264 @@
from unittest import mock
import botocore
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,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeDBEngineVersions":
return {
"DBEngineVersions": [
{
"Engine": "mysql",
"EngineVersion": "8.0.32",
"DBEngineDescription": "description",
"DBEngineVersionDescription": "description",
},
]
}
return make_api_call(self, operation_name, kwarg)
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
class Test_rds_instance_default_admin:
@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_default_admin.rds_instance_default_admin.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_default_admin.rds_instance_default_admin import (
rds_instance_default_admin,
)
check = rds_instance_default_admin()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_instance_with_default_username(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_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DBName="aurora-postgres",
MasterUsername="postgres",
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
)
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_default_admin.rds_instance_default_admin.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_default_admin.rds_instance_default_admin import (
rds_instance_default_admin,
)
check = rds_instance_default_admin()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is using the default master username."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_instance_without_default_username(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_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DBName="aurora-postgres",
MasterUsername="postgres2",
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
)
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_default_admin.rds_instance_default_admin.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_default_admin.rds_instance_default_admin import (
rds_instance_default_admin,
)
check = rds_instance_default_admin()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is not using the default master username."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_clustered_with_default_username(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="admin",
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_instance_default_admin.rds_instance_default_admin.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_default_admin.rds_instance_default_admin import (
rds_instance_default_admin,
)
check = rds_instance_default_admin()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 is using the default master username."
)
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_clustered_without_default_username(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=[],
)
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_default_admin.rds_instance_default_admin.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_default_admin.rds_instance_default_admin import (
rds_instance_default_admin,
)
check = rds_instance_default_admin()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 is not using the default master username."
)
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 == []
@@ -0,0 +1,408 @@
from unittest import mock
import botocore
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,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeDBEngineVersions":
return {
"DBEngineVersions": [
{
"Engine": "mysql",
"EngineVersion": "8.0.32",
"DBEngineDescription": "description",
"DBEngineVersionDescription": "description",
},
]
}
return make_api_call(self, operation_name, kwarg)
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
class Test_rds_instance_iam_authentication_enabled:
@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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_aurora_instance_without_iam_auth(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_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DBName="aurora-postgres",
EnableIAMDatabaseAuthentication=False,
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Instance db-master-1 does not have IAM authentication enabled."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_postgres_instance_with_iam_auth(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.postgres9.3",
Description="test parameter group",
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
EnableIAMDatabaseAuthentication=True,
DBParameterGroupName="test",
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 has IAM authentication enabled."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_mysql_instance_with_iam_auth(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql",
Description="test parameter group",
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="mysql",
DBName="staging-mysql",
DBInstanceClass="db.m1.small",
EnableIAMDatabaseAuthentication=True,
DBParameterGroupName="test",
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 has IAM authentication enabled."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_mariadb_instance_with_iam_auth(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mariadb",
Description="test parameter group",
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="mariadb",
DBName="staging-mariadb",
DBInstanceClass="db.m1.small",
EnableIAMDatabaseAuthentication=True,
DBParameterGroupName="test",
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 has IAM authentication enabled."
)
assert result[0].resource_id == "db-master-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}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_sqlserver_instance(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.sqlserver18",
Description="test parameter group",
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="sqlserver-ee",
DBName="staging-sqlserver",
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_rds_aurora_postgres_clustered_without_iam_auth(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=[],
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 does not have IAM authentication enabled."
)
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_without_iam_auth(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=[],
)
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_iam_authentication_enabled.rds_instance_iam_authentication_enabled.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_iam_authentication_enabled.rds_instance_iam_authentication_enabled import (
rds_instance_iam_authentication_enabled,
)
check = rds_instance_iam_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 does not have IAM authentication enabled."
)
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 == []
@@ -163,6 +163,9 @@ class Test_rds_instance_multi_az:
deletion_protection=False,
parameter_group="",
multi_az=True,
username="test",
iam_auth=False,
backtrack=0,
region=AWS_REGION_US_EAST_1,
tags=[],
)
@@ -183,6 +186,8 @@ class Test_rds_instance_multi_az:
deletion_protection=False,
parameter_group=[],
multi_az=False,
username="test",
iam_auth=False,
cluster_id="test-cluster",
cluster_arn=cluster_arn,
region=AWS_REGION_US_EAST_1,
@@ -240,6 +245,9 @@ class Test_rds_instance_multi_az:
deletion_protection=False,
parameter_group="",
multi_az=False,
username="test",
iam_auth=False,
backtrack=0,
region=AWS_REGION_US_EAST_1,
tags=[],
)
@@ -260,6 +268,8 @@ class Test_rds_instance_multi_az:
deletion_protection=False,
parameter_group=[],
multi_az=False,
username="test",
iam_auth=False,
cluster_id="test-cluster",
cluster_arn=cluster_arn,
region=AWS_REGION_US_EAST_1,
@@ -0,0 +1,251 @@
from unittest import mock
import botocore
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,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeDBEngineVersions":
return {
"DBEngineVersions": [
{
"Engine": "mysql",
"EngineVersion": "8.0.32",
"DBEngineDescription": "description",
"DBEngineVersionDescription": "description",
},
]
}
return make_api_call(self, operation_name, kwarg)
class Test_rds_snapshots_encrypted:
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_rds_no_snapshots(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_snapshots_encrypted.rds_snapshots_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_snapshots_encrypted.rds_snapshots_encrypted import (
rds_snapshots_encrypted,
)
check = rds_snapshots_encrypted()
result = check.execute()
assert len(result) == 0
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_rds_snapshot_not_encrypted(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_instance(
DBInstanceIdentifier="db-primary-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
)
conn.create_db_snapshot(
DBInstanceIdentifier="db-primary-1", DBSnapshotIdentifier="snapshot-1"
)
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_snapshots_encrypted.rds_snapshots_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_snapshots_encrypted.rds_snapshots_encrypted import (
rds_snapshots_encrypted,
)
check = rds_snapshots_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Instance Snapshot snapshot-1 is not encrypted."
)
assert result[0].resource_id == "snapshot-1"
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_rds_snapshot_encrypted(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_instance(
DBInstanceIdentifier="db-primary-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
)
conn.create_db_snapshot(
DBInstanceIdentifier="db-primary-1", DBSnapshotIdentifier="snapshot-1"
)
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_snapshots_encrypted.rds_snapshots_encrypted.rds_client",
new=RDS(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.rds.rds_snapshots_encrypted.rds_snapshots_encrypted import (
rds_snapshots_encrypted,
)
service_client.db_snapshots[0].encrypted = True
check = rds_snapshots_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance Snapshot snapshot-1 is encrypted."
)
assert result[0].resource_id == "snapshot-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}:snapshot:snapshot-1"
)
assert result[0].resource_tags == []
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_rds_cluster_snapshot_encrypted(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_cluster(
DBClusterIdentifier="db-primary-1",
AllocatedStorage=10,
Engine="postgres",
DBClusterInstanceClass="db.m1.small",
MasterUsername="root",
MasterUserPassword="hunter2000",
)
conn.create_db_cluster_snapshot(
DBClusterIdentifier="db-primary-1", DBClusterSnapshotIdentifier="snapshot-1"
)
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_snapshots_encrypted.rds_snapshots_encrypted.rds_client",
new=RDS(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.rds.rds_snapshots_encrypted.rds_snapshots_encrypted import (
rds_snapshots_encrypted,
)
service_client.db_cluster_snapshots[0].encrypted = True
check = rds_snapshots_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster Snapshot snapshot-1 is encrypted."
)
assert result[0].resource_id == "snapshot-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-snapshot:snapshot-1"
)
assert result[0].resource_tags == []
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_rds_cluster_snapshot_not_encrypted(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_cluster(
DBClusterIdentifier="db-primary-1",
AllocatedStorage=10,
Engine="postgres",
DBClusterInstanceClass="db.m1.small",
MasterUsername="root",
MasterUserPassword="hunter2000",
)
conn.create_db_cluster_snapshot(
DBClusterIdentifier="db-primary-1", DBClusterSnapshotIdentifier="snapshot-1"
)
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_snapshots_encrypted.rds_snapshots_encrypted.rds_client",
new=RDS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_snapshots_encrypted.rds_snapshots_encrypted import (
rds_snapshots_encrypted,
)
check = rds_snapshots_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster Snapshot snapshot-1 is not encrypted."
)
assert result[0].resource_id == "snapshot-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-snapshot:snapshot-1"
)
assert result[0].resource_tags == []