diff --git a/prowler/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer.py b/prowler/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer.py new file mode 100644 index 0000000000..752c8586a5 --- /dev/null +++ b/prowler/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer.py @@ -0,0 +1,63 @@ +from prowler.lib.logger import logger +from prowler.providers.aws.services.rds.rds_client import rds_client + + +def fixer(resource_id: str, region: str) -> bool: + """ + Modify the attributes of an RDS DB snapshot or DB cluster snapshot to remove public access. + Specifically, this fixer removes the 'all' value from the 'restore' attribute to + prevent the snapshot from being publicly accessible for both DB snapshots and DB cluster snapshots. + + Requires the rds:ModifyDBSnapshotAttribute or rds:ModifyDBClusterSnapshotAttribute permissions. + { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": "rds:ModifyDBSnapshotAttribute", + "Resource": "*" + }, + { + "Effect": "Allow", + "Action": "rds:ModifyDBClusterSnapshotAttribute", + "Resource": "*" + } + ] + } + + Args: + resource_id (str): The DB snapshot or DB cluster snapshot identifier. + region (str): AWS region where the snapshot exists. + + Returns: + bool: True if the operation is successful (public access is removed), False otherwise. + """ + try: + regional_client = rds_client.regional_clients[region] + + # Check if the resource is a DB Cluster Snapshot or a DB Instance Snapshot + try: + regional_client.describe_db_cluster_snapshots( + DBClusterSnapshotIdentifier=resource_id + ) + # If the describe call is successful, it's a DB cluster snapshot + regional_client.modify_db_cluster_snapshot_attribute( + DBClusterSnapshotIdentifier=resource_id, + AttributeName="restore", + ValuesToRemove=["all"], + ) + except regional_client.exceptions.DBClusterSnapshotNotFoundFault: + # If the DB cluster snapshot doesn't exist, it's an instance snapshot + regional_client.modify_db_snapshot_attribute( + DBSnapshotIdentifier=resource_id, + AttributeName="restore", + ValuesToRemove=["all"], + ) + + except Exception as error: + logger.error( + f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return False + else: + return True diff --git a/tests/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer_test.py b/tests/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer_test.py new file mode 100644 index 0000000000..6a915bfb1c --- /dev/null +++ b/tests/providers/aws/services/rds/rds_snapshots_public_access/rds_snapshots_public_access_fixer_test.py @@ -0,0 +1,187 @@ +from unittest import mock + +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider + + +class Test_rds_snapshots_public_access: + @mock_aws + def test_rds_private_snapshot(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_public_access.rds_snapshots_public_access_fixer.rds_client", + new=RDS(aws_provider), + ): + # Test Fixer + from prowler.providers.aws.services.rds.rds_snapshots_public_access.rds_snapshots_public_access_fixer import ( + fixer, + ) + + assert fixer("snapshot-1", AWS_REGION_US_EAST_1) + + @mock_aws + def test_rds_public_snapshot(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_public_access.rds_snapshots_public_access_fixer.rds_client", + new=RDS(aws_provider), + ) as service_client: + + service_client.db_snapshots[0].public = True + + # Test Fixer + from prowler.providers.aws.services.rds.rds_snapshots_public_access.rds_snapshots_public_access_fixer import ( + fixer, + ) + + assert fixer("snapshot-1", AWS_REGION_US_EAST_1) + + @mock_aws + def test_rds_cluster_private_snapshot(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_public_access.rds_snapshots_public_access_fixer.rds_client", + new=RDS(aws_provider), + ): + # Test Fixer + from prowler.providers.aws.services.rds.rds_snapshots_public_access.rds_snapshots_public_access_fixer import ( + fixer, + ) + + assert fixer("snapshot-1", AWS_REGION_US_EAST_1) + + @mock_aws + def test_rds_cluster_public_snapshot(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_public_access.rds_snapshots_public_access_fixer.rds_client", + new=RDS(aws_provider), + ) as service_client: + + service_client.db_cluster_snapshots[0].public = True + + # Test Fixer + from prowler.providers.aws.services.rds.rds_snapshots_public_access.rds_snapshots_public_access_fixer import ( + fixer, + ) + + assert fixer("snapshot-1", AWS_REGION_US_EAST_1) + + @mock_aws + def test_rds_cluster_public_snapshot_error(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_public_access.rds_snapshots_public_access_fixer.rds_client", + new=RDS(aws_provider), + ) as service_client: + + service_client.db_cluster_snapshots[0].public = True + + # Test Fixer + from prowler.providers.aws.services.rds.rds_snapshots_public_access.rds_snapshots_public_access_fixer import ( + fixer, + ) + + assert not fixer("snapshot-2", AWS_REGION_US_EAST_1)