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

fix(aws): change protected_by_backup_plan checks (#5204)

Browse Source 
Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Daniel Barranquero
2024-09-26 16:33:12 +02:00
committed by GitHub
parent 31bff99b3d
commit d0681a9e20
8 changed files with 633 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/backup/backup_service.py
+21 -29
View File
@@ -24,8 +24,8 @@ class Backup(AWSService):
self.__threading_call__(self._list_tags, self.backup_plans)
self.backup_report_plans = []
self.__threading_call__(self._list_backup_report_plans)
self.protected_resources = {}
self.__threading_call__(self._list_protected_resources)
self.protected_resources = []
self.__threading_call__(self._list_backup_selections)
def _list_backup_vaults(self, regional_client):
logger.info("Backup - Listing Backup Vaults...")
@@ -141,28 +141,27 @@ class Backup(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _list_protected_resources(self, regional_client):
logger.info("Backup - Listing Protected Resources...")
def _list_backup_selections(self, regional_client):
logger.info("Backup - Listing Backup Selections...")
try:
list_protected_resources_paginator = regional_client.get_paginator(
"list_protected_resources"
for backup_plan in self.backup_plans:
paginator = regional_client.get_paginator("list_backup_selections")
for page in paginator.paginate(BackupPlanId=backup_plan.id):
for selection in page.get("BackupSelectionsList", []):
selection_id = selection.get("SelectionId")
if selection_id:
backup_selection = regional_client.get_backup_selection(
BackupPlanId=backup_plan.id, SelectionId=selection_id
)["BackupSelection"]
self.protected_resources.extend(
backup_selection.get("Resources", [])
)
except ClientError as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
for page in list_protected_resources_paginator.paginate():
for resource in page.get("Results", []):
arn = resource.get("ResourceArn", "")
if not self.audit_resources or (
is_resource_filtered(
arn,
self.audit_resources,
)
):
self.protected_resources[arn] = ProtectedResource(
arn=arn,
resource_type=resource.get("ResourceType"),
region=regional_client.region,
last_backup_time=resource.get("LastBackupTime"),
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -209,10 +208,3 @@ class BackupReportPlan(BaseModel):
name: str
last_attempted_execution_date: Optional[datetime]
last_successful_execution_date: Optional[datetime]
class ProtectedResource(BaseModel):
arn: str
resource_type: str
region: str
last_backup_time: Optional[datetime]
prowler/providers/aws/services/dynamodb/dynamodb_table_protected_by_backup_plan/dynamodb_table_protected_by_backup_plan.py
+7 -1
View File
@@ -16,11 +16,17 @@ class dynamodb_table_protected_by_backup_plan(Check):
report.status_extended = (
f"DynamoDB table {table.name} is not protected by a backup plan."
)
if table_arn in backup_client.protected_resources:
if (
table_arn in backup_client.protected_resources
or "arn:aws:dynamodb:*:*:table/*" in backup_client.protected_resources
or "*" in backup_client.protected_resources
):
report.status = "PASS"
report.status_extended = (
f"DynamoDB table {table.name} is protected by a backup plan."
)
findings.append(report)
return findings
prowler/providers/aws/services/ec2/ec2_ebs_volume_protected_by_backup_plan/ec2_ebs_volume_protected_by_backup_plan.py
+10 -6
View File
@@ -12,16 +12,20 @@ class ec2_ebs_volume_protected_by_backup_plan(Check):
report.resource_id = volume.id
report.resource_arn = volume.arn
report.resource_tags = volume.tags
if volume.arn in backup_client.protected_resources:
report.status = "FAIL"
report.status_extended = (
f"EBS Volume {volume.id} is not protected by a backup plan."
)
if (
volume.arn in backup_client.protected_resources
or "arn:aws:ec2:*:*:volume/*" in backup_client.protected_resources
or "*" in backup_client.protected_resources
):
report.status = "PASS"
report.status_extended = (
f"EBS Volume {volume.id} is protected by a backup plan."
)
else:
report.status = "FAIL"
report.status_extended = (
f"EBS Volume {volume.id} is not protected by a backup plan."
)
findings.append(report)
return findings
prowler/providers/aws/services/rds/rds_instance_protected_by_backup_plan/rds_instance_protected_by_backup_plan.py
+12 -6
View File
@@ -12,15 +12,21 @@ class rds_instance_protected_by_backup_plan(Check):
report.resource_id = db_instance.id
report.resource_arn = db_instance_arn
report.resource_tags = db_instance.tags
if db_instance_arn in backup_client.protected_resources:
report.status = "FAIL"
report.status_extended = (
f"RDS Instance {db_instance.id} is not protected by a backup plan."
)
if (
db_instance_arn in backup_client.protected_resources
or "arn:aws:dynamodb:*:*:instance:*"
in backup_client.protected_resources
or "*" in backup_client.protected_resources
):
report.status = "PASS"
report.status_extended = (
f"RDS Instance {db_instance.id} is protected by a backup plan."
)
else:
report.status = "FAIL"
report.status_extended = (
f"RDS Instance {db_instance.id} is not protected by a backup plan."
)
findings.append(report)
return findings
tests/providers/aws/services/backup/backup_service_test.py
+28 -13
View File
@@ -58,16 +58,33 @@ def mock_make_api_call(self, operation_name, kwarg):
}
]
}
if operation_name == "ListProtectedResources":
if operation_name == "ListBackupSelections":
return {
"Results": [
"BackupSelectionsList": [
{
"ResourceArn": "arn:aws:rds:eu-west-1:123456789012:db:my-db-instance",
"ResourceType": "RDS",
"LastBackupTime": datetime(2015, 1, 1),
"SelectionId": "selection-id-1",
"SelectionName": "TestSelection",
"BackupPlanId": "ID-TestBackupPlan",
"CreationDate": datetime(2015, 1, 1),
"CreatorRequestId": "request-id-1",
"IamRoleArn": "arn:aws:iam::123456789012:role/service-role/AWSBackupDefaultServiceRole",
}
]
}
if operation_name == "GetBackupSelection":
return {
"BackupSelection": {
"SelectionName": "TestSelection",
"IamRoleArn": "arn:aws:iam::123456789012:role/service-role/AWSBackupDefaultServiceRole",
"Resources": [
"arn:aws:dynamodb:eu-west-1:123456789012:table/MyDynamoDBTable"
],
},
"SelectionId": "selection-id-1",
"BackupPlanId": "ID-TestBackupPlan",
"CreationDate": datetime(2015, 1, 1),
"CreatorRequestId": "request-id-1",
}
return make_api_call(self, operation_name, kwarg)
@@ -178,23 +195,21 @@ class TestBackupService:
2015, 1, 1
)
# Test Backup List Backup Selections
@mock_aws
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
new=mock_generate_regional_clients,
)
def test_list_protected_resources(self):
def test_list_backup_selections(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
backup = Backup(aws_provider)
assert len(backup.protected_resources) == 1
arn = "arn:aws:rds:eu-west-1:123456789012:db:my-db-instance"
protected_resource = backup.protected_resources.get(arn)
assert protected_resource is not None
assert protected_resource.arn == arn
assert protected_resource.resource_type == "RDS"
assert protected_resource.region == AWS_REGION_EU_WEST_1
assert protected_resource.last_backup_time == datetime(2015, 1, 1)
assert (
"arn:aws:dynamodb:eu-west-1:123456789012:table/MyDynamoDBTable"
in backup.protected_resources
)
@mock_aws
def test_list_tags(self):
tests/providers/aws/services/dynamodb/dynamodb_table_protected_by_backup_plan/dynamodb_table_protected_by_backup_plan_test.py
+150 -102
View File
@@ -1,7 +1,5 @@
from unittest import mock
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
@@ -11,39 +9,6 @@ from tests.providers.aws.utils import (
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 == "CreateBackupSelection":
return {
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1",
],
}
elif operation_name == "ListProtectedResources":
return {
"Results": [
{
"ResourceArn": f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1",
"ResourceType": "DynamoDB",
"LastBackupTime": "2023-08-23T00:00:00Z",
}
]
}
elif operation_name == "ListBackupPlans":
return {
"BackupPlans": [
{
"BackupPlanId": "test-backup-plan-id",
"BackupPlanName": "test-backup-plan",
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_dynamodb_table_protected_by_backup_plan:
@mock_aws
@@ -196,86 +161,169 @@ class Test_dynamodb_table_protected_by_backup_plan:
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_dynamodb_table_with_backup_plan(self):
with patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call):
backup = client("backup", region_name=AWS_REGION_US_EAST_1)
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
table = dynamodb_client.create_table(
TableName="test1",
AttributeDefinitions=[
{"AttributeName": "client", "AttributeType": "S"},
{"AttributeName": "app", "AttributeType": "S"},
],
KeySchema=[
{"AttributeName": "client", "KeyType": "HASH"},
{"AttributeName": "app", "KeyType": "RANGE"},
],
DeletionProtectionEnabled=True,
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]
backup.create_backup_plan(
BackupPlan={
"BackupPlanName": "test-backup-plan",
"Rules": [
{
"RuleName": "DailyBackup",
"TargetBackupVaultName": "test-vault",
"ScheduleExpression": "cron(0 12 * * ? *)",
"Lifecycle": {"DeleteAfterDays": 30},
"RecoveryPointTags": {
"Type": "Daily",
},
},
],
}
)
backup.create_backup_selection(
BackupPlanID={
backup.list_backup_plans()["BackupPlans"][0]["BackupPlanId"]
},
BackupPlanSelection={
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"{table['TableArn']}",
],
},
)
dynamodb_client = mock.MagicMock()
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.dynamodb.dynamodb_service import (
DynamoDB,
dynamodb_client.tables = {
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1": Table(
name="test1",
region=AWS_REGION_US_EAST_1,
)
}
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
backup = mock.MagicMock()
backup.protected_resources = [
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1"
]
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.common.provider.Provider.get_global_provider",
return_value=aws_provider,
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=DynamoDB(aws_provider),
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "DynamoDB table test1 is protected by a backup plan."
)
assert result[0].resource_id == table["TableName"]
assert result[0].resource_arn == table["TableArn"]
assert result[0].resource_id == "test1"
assert (
result[0].resource_arn
== f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1"
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
def test_dynamodb_table_with_backup_plan_via_wildcard_all_tables(self):
dynamodb_client = mock.MagicMock()
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = {
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1": Table(
name="test1",
region=AWS_REGION_US_EAST_1,
)
}
backup = mock.MagicMock()
backup.protected_resources = ["arn:aws:dynamodb:*:*:table/*"]
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.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "DynamoDB table test1 is protected by a backup plan."
)
assert result[0].resource_id == "test1"
assert (
result[0].resource_arn
== f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1"
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
def test_dynamodb_table_with_backup_plan_via_wildcard_all_resources(self):
dynamodb_client = mock.MagicMock()
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = {
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1": Table(
name="test1",
region=AWS_REGION_US_EAST_1,
)
}
backup = mock.MagicMock()
backup.protected_resources = ["*"]
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.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
new=dynamodb_client,
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "DynamoDB table test1 is protected by a backup plan."
)
assert result[0].resource_id == "test1"
assert (
result[0].resource_arn
== f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1"
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
tests/providers/aws/services/ec2/ec2_ebs_volume_protected_by_backup_plan/ec2_ebs_volume_protected_by_backup_plan_test.py
+142 -49
View File
@@ -1,6 +1,5 @@
from unittest import mock
import botocore
from boto3 import client
from moto import mock_aws
@@ -10,52 +9,10 @@ from tests.providers.aws.utils import (
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 == "CreateBackupSelection":
return {
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester",
],
}
elif operation_name == "ListProtectedResources":
return {
"Results": [
{
"ResourceArn": f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester",
"ResourceType": "EC2",
"LastBackupTime": "2023-08-23T00:00:00Z",
}
]
}
elif operation_name == "ListBackupPlans":
return {
"BackupPlans": [
{
"BackupPlanId": "test-backup-plan-id",
"BackupPlanName": "test-backup-plan",
}
]
}
elif operation_name == "DescribeVolumes":
return {
"Volumes": [
{
"VolumeId": "volume-tester",
"Encrypted": True,
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_ec2_ebs_volume_protected_by_backup_plan:
@mock_aws
def test_ec2_no_instances(self):
def test_ec2_no_volumes(self):
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.ec2.ec2_service import EC2
@@ -189,10 +146,24 @@ class Test_ec2_ebs_volume_protected_by_backup_plan:
)
assert result[0].resource_tags is None
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_ec2_instance_with_backup_plan(self):
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.ec2.ec2_service import EC2
ec2_client = mock.MagicMock()
from prowler.providers.aws.services.ec2.ec2_service import Volume
ec2_client.volumes = [
Volume(
id="volume-tester",
arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester",
tags=None,
region=AWS_REGION_US_EAST_1,
encrypted=False,
)
]
backup_client = mock.MagicMock()
backup_client.protected_resources = [
f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester"
]
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -202,10 +173,132 @@ class Test_ec2_ebs_volume_protected_by_backup_plan:
):
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.ec2_client",
new=EC2(aws_provider),
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_client.ec2_client",
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
new=backup_client,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup_client,
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan import (
ec2_ebs_volume_protected_by_backup_plan,
)
check = ec2_ebs_volume_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "EBS Volume volume-tester is protected by a backup plan."
)
assert result[0].resource_id == "volume-tester"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester"
)
assert result[0].resource_tags is None
def test_ec2_instance_with_backup_plan_via_volume_wildcard(self):
ec2_client = mock.MagicMock()
from prowler.providers.aws.services.ec2.ec2_service import Volume
ec2_client.volumes = [
Volume(
id="volume-tester",
arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester",
tags=None,
region=AWS_REGION_US_EAST_1,
encrypted=False,
)
]
backup_client = mock.MagicMock()
backup_client.protected_resources = ["arn:aws:ec2:*:*:volume/*"]
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.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.ec2_client",
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_client.ec2_client",
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.backup_client",
new=backup_client,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup_client,
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan import (
ec2_ebs_volume_protected_by_backup_plan,
)
check = ec2_ebs_volume_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "EBS Volume volume-tester is protected by a backup plan."
)
assert result[0].resource_id == "volume-tester"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester"
)
assert result[0].resource_tags is None
def test_ec2_instance_with_backup_plan_via_all_wildcard(self):
ec2_client = mock.MagicMock()
from prowler.providers.aws.services.ec2.ec2_service import Volume
ec2_client.volumes = [
Volume(
id="volume-tester",
arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:volume/volume-tester",
tags=None,
region=AWS_REGION_US_EAST_1,
encrypted=False,
)
]
backup_client = mock.MagicMock()
backup_client.protected_resources = ["*"]
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.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.ec2_client",
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_client.ec2_client",
new=ec2_client,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan.backup_client",
new=backup_client,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup_client,
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_ebs_volume_protected_by_backup_plan.ec2_ebs_volume_protected_by_backup_plan import (
tests/providers/aws/services/rds/rds_instance_protected_by_backup_plan/rds_instance_protected_by_backup_plan_test.py
+263 -133
View File
@@ -1,7 +1,5 @@
from unittest import mock
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
@@ -11,39 +9,6 @@ from tests.providers.aws.utils import (
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 == "CreateBackupSelection":
return {
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
],
}
elif operation_name == "ListProtectedResources":
return {
"Results": [
{
"ResourceArn": f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
"ResourceType": "RDS",
"LastBackupTime": "2023-08-23T00:00:00Z",
}
]
}
elif operation_name == "ListBackupPlans":
return {
"BackupPlans": [
{
"BackupPlanId": "test-backup-plan-id",
"BackupPlanName": "test-backup-plan",
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_rds_instance_protected_by_backup_plan:
@mock_aws
@@ -123,36 +88,43 @@ class Test_rds_instance_protected_by_backup_plan:
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_instance_without_backup_plan(self):
instance = client("rds", region_name=AWS_REGION_US_EAST_1)
backup = client("backup", region_name=AWS_REGION_US_EAST_1)
instance.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
)
backup.create_backup_plan(
BackupPlan={
"BackupPlanName": "test-backup-plan",
"Rules": [
{
"RuleName": "DailyBackup",
"TargetBackupVaultName": "test-vault",
"ScheduleExpression": "cron(0 12 * * ? *)",
"Lifecycle": {"DeleteAfterDays": 30},
"RecoveryPointTags": {
"Type": "Daily",
},
},
],
}
)
instance = mock.MagicMock()
backup = mock.MagicMock()
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.rds.rds_service import RDS
from prowler.providers.aws.services.rds.rds_service import DBInstance
arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
instance.db_instances = {
arn: DBInstance(
id="db-master-1",
arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
endpoint={
"Address": "db-master-1.c9akciq32.rds.amazonaws.com",
"Port": 5432,
},
engine_version="13.3",
status="available",
public=False,
encrypted=True,
deletion_protection=False,
auto_minor_version_upgrade=True,
multi_az=False,
username="admin",
iam_auth=False,
name="db-master-1",
region="us-east-1",
instance_class="db.m1.small",
engine="postgres",
db_name="staging-postgres",
allocated_storage=10,
tags=[],
)
}
backup.protected_resources = [
f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-2"
]
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -162,10 +134,16 @@ class Test_rds_instance_protected_by_backup_plan:
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.rds_client",
new=RDS(aws_provider),
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_client.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan import (
@@ -189,81 +167,233 @@ class Test_rds_instance_protected_by_backup_plan:
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_instance_with_backup_plan(self):
with patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call):
instance = client("rds", region_name=AWS_REGION_US_EAST_1)
backup = client("backup", region_name=AWS_REGION_US_EAST_1)
instance.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
)
backup.create_backup_plan(
BackupPlan={
"BackupPlanName": "test-backup-plan",
"Rules": [
{
"RuleName": "DailyBackup",
"TargetBackupVaultName": "test-vault",
"ScheduleExpression": "cron(0 12 * * ? *)",
"Lifecycle": {"DeleteAfterDays": 30},
"RecoveryPointTags": {
"Type": "Daily",
},
},
],
}
)
backup.create_backup_selection(
BackupPlanID={
backup.list_backup_plans()["BackupPlans"][0]["BackupPlanId"]
},
BackupPlanSelection={
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
],
instance = mock.MagicMock()
from prowler.providers.aws.services.rds.rds_service import DBInstance
arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
instance.db_instances = {
arn: DBInstance(
id="db-master-1",
arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
endpoint={
"Address": "db-master-1.c9akciq32.rds.amazonaws.com",
"Port": 5432,
},
engine_version="13.3",
status="available",
public=False,
encrypted=True,
deletion_protection=False,
auto_minor_version_upgrade=True,
multi_az=False,
username="admin",
iam_auth=False,
name="db-master-1",
region="us-east-1",
instance_class="db.m1.small",
engine="postgres",
db_name="staging-postgres",
allocated_storage=10,
tags=[],
)
}
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.rds.rds_service import RDS
backup = mock.MagicMock()
backup.protected_resources = [arn]
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
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.common.provider.Provider.get_global_provider",
return_value=aws_provider,
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_client.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.rds_client",
new=RDS(aws_provider),
), mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan import (
rds_instance_protected_by_backup_plan,
)
# Test Check
from prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan import (
rds_instance_protected_by_backup_plan,
)
check = rds_instance_protected_by_backup_plan()
result = check.execute()
check = rds_instance_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is protected by a backup plan."
)
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 == []
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is protected by a backup plan."
)
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 == []
def test_rds_instance_with_backup_plan_via_instance_wildcard(self):
instance = mock.MagicMock()
from prowler.providers.aws.services.rds.rds_service import DBInstance
arn = "arn:aws:dynamodb:*:*:instance:*"
instance.db_instances = {
f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1": DBInstance(
id="db-master-1",
arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
endpoint={
"Address": "db-master-1.c9akciq32.rds.amazonaws.com",
"Port": 5432,
},
engine_version="13.3",
status="available",
public=False,
encrypted=True,
deletion_protection=False,
auto_minor_version_upgrade=True,
multi_az=False,
username="admin",
iam_auth=False,
name="db-master-1",
region="us-east-1",
instance_class="db.m1.small",
engine="postgres",
db_name="staging-postgres",
allocated_storage=10,
tags=[],
)
}
backup = mock.MagicMock()
backup.protected_resources = [arn]
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_protected_by_backup_plan.rds_instance_protected_by_backup_plan.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_client.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan import (
rds_instance_protected_by_backup_plan,
)
check = rds_instance_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is protected by a backup plan."
)
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 == []
def test_rds_instance_with_backup_plan_via_all_wildcard(self):
instance = mock.MagicMock()
from prowler.providers.aws.services.rds.rds_service import DBInstance
arn = "*"
instance.db_instances = {
f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1": DBInstance(
id="db-master-1",
arn=f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1",
endpoint={
"Address": "db-master-1.c9akciq32.rds.amazonaws.com",
"Port": 5432,
},
engine_version="13.3",
status="available",
public=False,
encrypted=True,
deletion_protection=False,
auto_minor_version_upgrade=True,
multi_az=False,
username="admin",
iam_auth=False,
name="db-master-1",
region="us-east-1",
instance_class="db.m1.small",
engine="postgres",
db_name="staging-postgres",
allocated_storage=10,
tags=[],
)
}
backup = mock.MagicMock()
backup.protected_resources = [arn]
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_protected_by_backup_plan.rds_instance_protected_by_backup_plan.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_client.rds_client",
new=instance,
), mock.patch(
"prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan.backup_client",
new=backup,
), mock.patch(
"prowler.providers.aws.services.backup.backup_client.backup_client",
new=backup,
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_protected_by_backup_plan.rds_instance_protected_by_backup_plan import (
rds_instance_protected_by_backup_plan,
)
check = rds_instance_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Instance db-master-1 is protected by a backup plan."
)
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 == []