mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
Hugo Pereira Brito
545 lines
20 KiB
Python
545 lines
20 KiB
Python
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import botocore
|
|
from boto3 import client
|
|
from moto import mock_aws
|
|
|
|
from prowler.providers.aws.services.backup.backup_service import (
|
|
Backup,
|
|
BackupVault,
|
|
RecoveryPoint,
|
|
)
|
|
from tests.providers.aws.utils import (
|
|
AWS_ACCOUNT_NUMBER,
|
|
AWS_REGION_EU_WEST_1,
|
|
set_mocked_aws_provider,
|
|
)
|
|
|
|
# Mocking Backup Calls
|
|
make_api_call = botocore.client.BaseClient._make_api_call
|
|
|
|
|
|
def mock_make_api_call(self, operation_name, kwarg):
|
|
"""
|
|
Mock every AWS API call
|
|
"""
|
|
if operation_name == "ListBackupVaults":
|
|
return {
|
|
"BackupVaultList": [
|
|
{
|
|
"BackupVaultArn": "ARN",
|
|
"BackupVaultName": "Test Vault",
|
|
"EncryptionKeyArn": "",
|
|
"NumberOfRecoveryPoints": 0,
|
|
"Locked": True,
|
|
"MinRetentionDays": 1,
|
|
"MaxRetentionDays": 2,
|
|
}
|
|
]
|
|
}
|
|
if operation_name == "ListBackupPlans":
|
|
return {
|
|
"BackupPlansList": [
|
|
{
|
|
"BackupPlanArn": "ARN",
|
|
"BackupPlanId": "ID",
|
|
"BackupPlanName": "Test Plan",
|
|
"VersionId": "test_version_id",
|
|
"LastExecutionDate": datetime(2015, 1, 1),
|
|
"AdvancedBackupSettings": [],
|
|
}
|
|
]
|
|
}
|
|
if operation_name == "ListReportPlans":
|
|
return {
|
|
"ReportPlans": [
|
|
{
|
|
"ReportPlanArn": "ARN",
|
|
"ReportPlanName": "Test Report Plan",
|
|
"LastAttemptedExecutionTime": datetime(2015, 1, 1),
|
|
"LastSuccessfulExecutionTime": datetime(2015, 1, 1),
|
|
}
|
|
]
|
|
}
|
|
if operation_name == "ListBackupSelections":
|
|
return {
|
|
"BackupSelectionsList": [
|
|
{
|
|
"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",
|
|
}
|
|
if operation_name == "ListRecoveryPointsByBackupVault":
|
|
return {
|
|
"RecoveryPoints": [
|
|
{
|
|
"RecoveryPointArn": "arn:aws:backup:eu-west-1:123456789012:recovery-point:1",
|
|
"BackupVaultName": "Test Vault",
|
|
"BackupVaultArn": "arn:aws:backup:eu-west-1:123456789012:backup-vault:Test Vault",
|
|
"BackupVaultRegion": "eu-west-1",
|
|
"CreationDate": datetime(2015, 1, 1),
|
|
"Status": "COMPLETED",
|
|
"EncryptionKeyArn": "",
|
|
"ResourceArn": "arn:aws:dynamodb:eu-west-1:123456789012:table/MyDynamoDBTable",
|
|
"ResourceType": "DynamoDB",
|
|
"BackupPlanId": "ID-TestBackupPlan",
|
|
"VersionId": "test_version_id",
|
|
"IsEncrypted": True,
|
|
}
|
|
]
|
|
}
|
|
return make_api_call(self, operation_name, kwarg)
|
|
|
|
|
|
def mock_generate_regional_clients(provider, service):
|
|
regional_client = provider._session.current_session.client(
|
|
service, region_name=AWS_REGION_EU_WEST_1
|
|
)
|
|
regional_client.region = AWS_REGION_EU_WEST_1
|
|
return {AWS_REGION_EU_WEST_1: regional_client}
|
|
|
|
|
|
class TestBackupService:
|
|
# Test Backup Client
|
|
@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_get_client(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert (
|
|
backup.regional_clients[AWS_REGION_EU_WEST_1].__class__.__name__ == "Backup"
|
|
)
|
|
|
|
# Test Backup Session
|
|
@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__get_session__(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
access_analyzer = Backup(aws_provider)
|
|
assert access_analyzer.session.__class__.__name__ == "Session"
|
|
|
|
# Test Backup Serviceç
|
|
@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__get_service__(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
access_analyzer = Backup(aws_provider)
|
|
assert access_analyzer.service == "backup"
|
|
|
|
# Test Backup List Backup Vaults
|
|
@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_backup_vaults(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert len(backup.backup_vaults) == 1
|
|
assert backup.backup_vaults[0].arn == "ARN"
|
|
assert backup.backup_vaults[0].name == "Test Vault"
|
|
assert backup.backup_vaults[0].region == AWS_REGION_EU_WEST_1
|
|
assert backup.backup_vaults[0].encryption == ""
|
|
assert backup.backup_vaults[0].recovery_points == 0
|
|
assert backup.backup_vaults[0].locked is True
|
|
assert backup.backup_vaults[0].min_retention_days == 1
|
|
assert backup.backup_vaults[0].max_retention_days == 2
|
|
|
|
# Test Backup List Backup Plans
|
|
@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_backup_plans(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert len(backup.backup_plans) == 1
|
|
assert backup.backup_plans[0].arn == "ARN"
|
|
assert backup.backup_plans[0].id == "ID"
|
|
assert backup.backup_plans[0].region == AWS_REGION_EU_WEST_1
|
|
assert backup.backup_plans[0].name == "Test Plan"
|
|
assert backup.backup_plans[0].version_id == "test_version_id"
|
|
assert backup.backup_plans[0].last_execution_date == datetime(2015, 1, 1)
|
|
assert backup.backup_plans[0].advanced_settings == []
|
|
|
|
# Test Backup List Report Plans
|
|
@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_backup_report_plans(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert len(backup.backup_report_plans) == 1
|
|
assert backup.backup_report_plans[0].arn == "ARN"
|
|
assert backup.backup_report_plans[0].region == AWS_REGION_EU_WEST_1
|
|
assert backup.backup_report_plans[0].name == "Test Report Plan"
|
|
assert backup.backup_report_plans[0].last_attempted_execution_date == datetime(
|
|
2015, 1, 1
|
|
)
|
|
assert backup.backup_report_plans[0].last_successful_execution_date == datetime(
|
|
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_backup_selections(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert len(backup.protected_resources) == 1
|
|
assert (
|
|
"arn:aws:dynamodb:eu-west-1:123456789012:table/MyDynamoDBTable"
|
|
in backup.protected_resources
|
|
)
|
|
|
|
@mock_aws
|
|
def test_list_tags(self):
|
|
backup_client = client("backup", region_name=AWS_REGION_EU_WEST_1)
|
|
|
|
# Create necessary resources and tags
|
|
backup_vault = backup_client.create_backup_vault(
|
|
BackupVaultName="TestVault",
|
|
EncryptionKeyArn=f"arn:aws:kms:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:key/1234abcd-12ab-34cd-56ef-123456789012",
|
|
)
|
|
|
|
tags = {"TestKey": "TestValue"}
|
|
|
|
backup_client.tag_resource(
|
|
ResourceArn=backup_vault["BackupVaultArn"], Tags=tags
|
|
)
|
|
|
|
# Create a backup plan
|
|
backup_plan = backup_client.create_backup_plan(
|
|
BackupPlan={
|
|
"BackupPlanName": "TestPlan",
|
|
"Rules": [
|
|
{
|
|
"RuleName": "TestRule",
|
|
"TargetBackupVaultName": "TestVault", # Match the vault name
|
|
"ScheduleExpression": "cron(0 12 * * ? *)",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
backup_client.tag_resource(ResourceArn=backup_plan["BackupPlanArn"], Tags=tags)
|
|
|
|
# Test list_tags
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
|
|
assert len(backup.backup_vaults) == 1
|
|
assert len(backup.backup_vaults[0].tags) == 1
|
|
assert backup.backup_vaults[0].tags[0]["TestKey"] == "TestValue"
|
|
assert len(backup.backup_plans) == 1
|
|
assert len(backup.backup_plans[0].tags) == 1
|
|
assert backup.backup_plans[0].tags[0]["TestKey"] == "TestValue"
|
|
|
|
# Test Backup List Recovery Points
|
|
@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_recovery_points(self):
|
|
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
|
backup = Backup(aws_provider)
|
|
assert len(backup.recovery_points) == 1
|
|
assert (
|
|
backup.recovery_points[0].arn
|
|
== "arn:aws:backup:eu-west-1:123456789012:recovery-point:1"
|
|
)
|
|
assert backup.recovery_points[0].backup_vault_name == "Test Vault"
|
|
assert backup.recovery_points[0].backup_vault_region == "eu-west-1"
|
|
assert backup.recovery_points[0].tags == []
|
|
assert backup.recovery_points[0].encrypted is True
|
|
|
|
def test_recovery_point_limit_bounds_tag_calls_to_selected_points(self):
|
|
class FakePaginator:
|
|
def paginate(self, **kwargs):
|
|
return [
|
|
{
|
|
"RecoveryPoints": [
|
|
{
|
|
"RecoveryPointArn": "arn:aws:backup:eu-west-1:123456789012:recovery-point:new",
|
|
"IsEncrypted": True,
|
|
"CreationDate": datetime(2024, 1, 2),
|
|
},
|
|
{
|
|
"RecoveryPointArn": "arn:aws:backup:eu-west-1:123456789012:recovery-point:old",
|
|
"IsEncrypted": True,
|
|
"CreationDate": datetime(2024, 1, 1),
|
|
},
|
|
]
|
|
}
|
|
]
|
|
|
|
class FakeBackupClient:
|
|
def __init__(self):
|
|
self.tag_calls = []
|
|
|
|
def get_paginator(self, name):
|
|
assert name == "list_recovery_points_by_backup_vault"
|
|
return FakePaginator()
|
|
|
|
def list_tags(self, **kwargs):
|
|
self.tag_calls.append(kwargs["ResourceArn"])
|
|
return {"Tags": {}}
|
|
|
|
regional_client = FakeBackupClient()
|
|
backup = Backup.__new__(Backup)
|
|
backup.backup_vaults = [
|
|
BackupVault(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:backup-vault:vault",
|
|
name="vault",
|
|
region=AWS_REGION_EU_WEST_1,
|
|
encryption="",
|
|
recovery_points=2,
|
|
locked=False,
|
|
)
|
|
]
|
|
backup.recovery_points = []
|
|
backup.recovery_point_limit = 1
|
|
backup.regional_clients = {AWS_REGION_EU_WEST_1: regional_client}
|
|
|
|
backup._list_recovery_points()
|
|
backup._select_recovery_points_for_analysis()
|
|
for recovery_point in backup.recovery_points:
|
|
backup._list_tags(recovery_point)
|
|
|
|
assert [rp.id for rp in backup.recovery_points] == ["new"]
|
|
assert regional_client.tag_calls == [
|
|
"arn:aws:backup:eu-west-1:123456789012:recovery-point:new"
|
|
]
|
|
|
|
def test_recovery_point_limit_selects_global_newest_across_vaults(self):
|
|
class FakePaginator:
|
|
def __init__(self, recovery_points_by_vault):
|
|
self.recovery_points_by_vault = recovery_points_by_vault
|
|
|
|
def paginate(self, **kwargs):
|
|
assert "PageSize" not in kwargs
|
|
return [
|
|
{
|
|
"RecoveryPoints": self.recovery_points_by_vault[
|
|
kwargs["BackupVaultName"]
|
|
]
|
|
}
|
|
]
|
|
|
|
class FakeBackupClient:
|
|
def __init__(self, recovery_points_by_vault):
|
|
self.recovery_points_by_vault = recovery_points_by_vault
|
|
|
|
def get_paginator(self, name):
|
|
assert name == "list_recovery_points_by_backup_vault"
|
|
return FakePaginator(self.recovery_points_by_vault)
|
|
|
|
backup = Backup.__new__(Backup)
|
|
backup.recovery_point_limit = 1
|
|
backup.recovery_points = []
|
|
backup.backup_vaults = [
|
|
BackupVault(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:backup-vault:old-vault",
|
|
name="old-vault",
|
|
region=AWS_REGION_EU_WEST_1,
|
|
encryption="",
|
|
recovery_points=1,
|
|
locked=False,
|
|
),
|
|
BackupVault(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:backup-vault:new-vault",
|
|
name="new-vault",
|
|
region=AWS_REGION_EU_WEST_1,
|
|
encryption="",
|
|
recovery_points=1,
|
|
locked=False,
|
|
),
|
|
]
|
|
backup.regional_clients = {
|
|
AWS_REGION_EU_WEST_1: FakeBackupClient(
|
|
{
|
|
"old-vault": [
|
|
{
|
|
"RecoveryPointArn": "arn:aws:backup:eu-west-1:123456789012:recovery-point:old",
|
|
"IsEncrypted": True,
|
|
"CreationDate": datetime(2024, 1, 1),
|
|
}
|
|
],
|
|
"new-vault": [
|
|
{
|
|
"RecoveryPointArn": "arn:aws:backup:eu-west-1:123456789012:recovery-point:new",
|
|
"IsEncrypted": True,
|
|
"CreationDate": datetime(2024, 1, 2),
|
|
}
|
|
],
|
|
}
|
|
)
|
|
}
|
|
|
|
backup._list_recovery_points()
|
|
backup._select_recovery_points_for_analysis()
|
|
|
|
assert [rp.id for rp in backup.recovery_points] == ["new"]
|
|
|
|
def test_recovery_point_limit_exposes_only_selected_resources(self):
|
|
backup = Backup.__new__(Backup)
|
|
backup.recovery_point_limit = 2
|
|
backup.recovery_points = []
|
|
backup.backup_vaults = [
|
|
BackupVault(
|
|
arn="arn",
|
|
name="vault",
|
|
region="eu-west-1",
|
|
encryption="",
|
|
recovery_points=3,
|
|
locked=False,
|
|
)
|
|
]
|
|
|
|
class Paginator:
|
|
def paginate(self, **_kwargs):
|
|
return [
|
|
{
|
|
"RecoveryPoints": [
|
|
{
|
|
"RecoveryPointArn": f"arn:aws:backup:eu-west-1:123456789012:recovery-point:{i}",
|
|
"IsEncrypted": True,
|
|
}
|
|
for i in range(3)
|
|
]
|
|
}
|
|
]
|
|
|
|
backup.regional_clients = {
|
|
"eu-west-1": SimpleNamespace(get_paginator=lambda _: Paginator())
|
|
}
|
|
tagged = []
|
|
|
|
def list_tags(recovery_point):
|
|
tagged.append(recovery_point.arn)
|
|
|
|
backup._list_tags = list_tags
|
|
|
|
backup._list_recovery_points()
|
|
backup._select_recovery_points_for_analysis()
|
|
for recovery_point in backup.recovery_points:
|
|
backup._list_tags(recovery_point)
|
|
|
|
assert len(backup.recovery_points) == 2
|
|
assert len(tagged) == 2
|
|
|
|
def test_recovery_point_limit_uses_deterministic_tie_breaker(self):
|
|
backup = Backup.__new__(Backup)
|
|
backup.recovery_point_limit = 2
|
|
backup.recovery_points = [
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:us-east-1:123456789012:recovery-point:z",
|
|
id="z",
|
|
region="us-east-1",
|
|
backup_vault_name="vault-b",
|
|
encrypted=True,
|
|
backup_vault_region="us-east-1",
|
|
),
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:recovery-point:b",
|
|
id="b",
|
|
region="eu-west-1",
|
|
backup_vault_name="vault-b",
|
|
encrypted=True,
|
|
backup_vault_region="eu-west-1",
|
|
),
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:recovery-point:a",
|
|
id="a",
|
|
region="eu-west-1",
|
|
backup_vault_name="vault-a",
|
|
encrypted=True,
|
|
backup_vault_region="eu-west-1",
|
|
),
|
|
]
|
|
|
|
backup._select_recovery_points_for_analysis()
|
|
|
|
assert [rp.id for rp in backup.recovery_points] == ["a", "b"]
|
|
|
|
def test_recovery_point_limit_keeps_newest_before_tie_breaker(self):
|
|
backup = Backup.__new__(Backup)
|
|
backup.recovery_point_limit = 2
|
|
backup.recovery_points = [
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:recovery-point:older-a",
|
|
id="older-a",
|
|
region="eu-west-1",
|
|
backup_vault_name="vault-a",
|
|
encrypted=True,
|
|
backup_vault_region="eu-west-1",
|
|
creation_date=datetime(2024, 1, 1),
|
|
),
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:us-east-1:123456789012:recovery-point:newer-z",
|
|
id="newer-z",
|
|
region="us-east-1",
|
|
backup_vault_name="vault-z",
|
|
encrypted=True,
|
|
backup_vault_region="us-east-1",
|
|
creation_date=datetime(2024, 1, 2),
|
|
),
|
|
RecoveryPoint(
|
|
arn="arn:aws:backup:eu-west-1:123456789012:recovery-point:missing-date",
|
|
id="missing-date",
|
|
region="eu-west-1",
|
|
backup_vault_name="vault-a",
|
|
encrypted=True,
|
|
backup_vault_region="eu-west-1",
|
|
),
|
|
]
|
|
|
|
backup._select_recovery_points_for_analysis()
|
|
|
|
assert [rp.id for rp in backup.recovery_points] == ["newer-z", "older-a"]
|