mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
feat(rds): add new fixer rds_snapshots_public_access_fixer (#5773)
This commit is contained in:
Daniel Barranquero
committed by
GitHub
GitHub
parent
c1b050b8b9
commit
b0bb348480
+63
@@ -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
|
||||
+187
@@ -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)
|
||||
Reference in New Issue
Block a user