diff --git a/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/__init__.py b/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.metadata.json b/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.metadata.json new file mode 100644 index 0000000000..e6eab7dddb --- /dev/null +++ b/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.metadata.json @@ -0,0 +1,34 @@ +{ + "Provider": "aws", + "CheckID": "backup_recovery_point_encrypted", + "CheckTitle": "Check if AWS Backup recovery points are encrypted at rest.", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "backup", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:backup:region:account-id:recovery-point/recovery-point-id", + "Severity": "medium", + "ResourceType": "AwsBackupRecoveryPoint", + "Description": "This control checks if an AWS Backup recovery point is encrypted at rest. The control fails if the recovery point isn't encrypted at rest.", + "Risk": "Without encryption at rest, AWS Backup recovery points are vulnerable to unauthorized access, which could compromise the confidentiality and integrity of the backed-up data.", + "RelatedUrl": "https://docs.aws.amazon.com/aws-backup/latest/devguide/encryption.html", + "Remediation": { + "Code": { + "CLI": "aws backup update-backup-vault --backup-vault-name --encryption-key-arn ", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/backup-controls.html#backup-1", + "Terraform": "" + }, + "Recommendation": { + "Text": "Ensure that AWS Backup recovery points are encrypted at rest by using an AWS KMS key when creating backups.", + "Url": "https://docs.aws.amazon.com/aws-backup/latest/devguide/encryption.html" + } + }, + "Categories": [ + "encryption" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.py b/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.py new file mode 100644 index 0000000000..dd41668217 --- /dev/null +++ b/prowler/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted.py @@ -0,0 +1,22 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.backup.backup_client import backup_client + + +class backup_recovery_point_encrypted(Check): + def execute(self): + findings = [] + for recovery_point in backup_client.recovery_points: + report = Check_Report_AWS(self.metadata()) + report.region = recovery_point.backup_vault_region + report.resource_id = recovery_point.backup_vault_name + report.resource_arn = recovery_point.arn + report.resource_tags = recovery_point.tags + report.status = "FAIL" + report.status_extended = f"Backup Recovery Point {recovery_point.arn} for Backup Vault {recovery_point.backup_vault_name} is not encrypted at rest." + if recovery_point.encrypted: + report.status = "PASS" + report.status_extended = f"Backup Recovery Point {recovery_point.arn} for Backup Vault {recovery_point.backup_vault_name} is encrypted at rest." + + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/backup/backup_service.py b/prowler/providers/aws/services/backup/backup_service.py index ee18e91944..4805e36259 100644 --- a/prowler/providers/aws/services/backup/backup_service.py +++ b/prowler/providers/aws/services/backup/backup_service.py @@ -26,6 +26,8 @@ class Backup(AWSService): self.__threading_call__(self._list_backup_report_plans) self.protected_resources = [] self.__threading_call__(self._list_backup_selections) + self.recovery_points = [] + self.__threading_call__(self._list_recovery_points) def _list_backup_vaults(self, regional_client): logger.info("Backup - Listing Backup Vaults...") @@ -178,6 +180,33 @@ class Backup(AWSService): f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + def _list_recovery_points(self, regional_client): + logger.info("Backup - Listing Recovery Points...") + try: + for backup_vault in self.backup_vaults: + paginator = regional_client.get_paginator( + "list_recovery_points_by_backup_vault" + ) + for page in paginator.paginate(BackupVaultName=backup_vault.name): + for recovery_point in page.get("RecoveryPoints", []): + self.recovery_points.append( + RecoveryPoint( + arn=recovery_point.get("RecoveryPointArn"), + backup_vault_name=backup_vault.name, + encrypted=recovery_point.get("IsEncrypted", False), + backup_vault_region=backup_vault.region, + tags=[], + ) + ) + except ClientError as error: + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + class BackupVault(BaseModel): arn: str @@ -208,3 +237,11 @@ class BackupReportPlan(BaseModel): name: str last_attempted_execution_date: Optional[datetime] last_successful_execution_date: Optional[datetime] + + +class RecoveryPoint(BaseModel): + arn: str + backup_vault_name: str + encrypted: bool + backup_vault_region: str + tags: Optional[list] diff --git a/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.metadata.json b/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.metadata.json index 6d568a450c..7a77b8e543 100644 --- a/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.metadata.json +++ b/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.metadata.json @@ -3,10 +3,7 @@ "CheckID": "backup_vaults_encrypted", "CheckTitle": "Ensure that AWS Backup vaults are encrypted with AWS KMS", "CheckType": [ - "Recover", - "Resilience", - "Backup", - "Data Protection" + "Software and Configuration Checks/AWS Security Best Practices" ], "ServiceName": "backup", "SubServiceName": "", diff --git a/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.py b/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.py index d53d8e00a0..053ebbb2e9 100644 --- a/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.py +++ b/prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.py @@ -7,23 +7,20 @@ class backup_vaults_encrypted(Check): findings = [] if backup_client.backup_vaults: for backup_vault in backup_client.backup_vaults: - # By default we assume that the result is fail report = Check_Report_AWS(self.metadata()) - report.status = "FAIL" - report.status_extended = ( - f"Backup Vault {backup_vault.name} is not encrypted." - ) report.resource_arn = backup_vault.arn report.resource_id = backup_vault.name report.region = backup_vault.region report.resource_tags = backup_vault.tags - # if it is encrypted we only change the status and the status extended + report.status = "FAIL" + report.status_extended = ( + f"Backup Vault {backup_vault.name} is not encrypted at rest." + ) if backup_vault.encryption: report.status = "PASS" report.status_extended = ( - f"Backup Vault {backup_vault.name} is encrypted." + f"Backup Vault {backup_vault.name} is encrypted at rest." ) - # then we store the finding findings.append(report) return findings diff --git a/tests/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted_test.py b/tests/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted_test.py new file mode 100644 index 0000000000..dd49821133 --- /dev/null +++ b/tests/providers/aws/services/backup/backup_recovery_point_encrypted/backup_recovery_point_encrypted_test.py @@ -0,0 +1,194 @@ +from datetime import datetime +from unittest import mock + +import botocore +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider + +make_api_call = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call_encrypted(self, operation_name, kwarg): + 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, + } + ] + } + if operation_name == "ListBackupVaults": + return { + "BackupVaultList": [ + { + "BackupVaultArn": "ARN", + "BackupVaultName": "Test Vault", + "EncryptionKeyArn": "", + "NumberOfRecoveryPoints": 0, + "Locked": True, + "MinRetentionDays": 1, + "MaxRetentionDays": 2, + } + ] + } + return make_api_call(self, operation_name, kwarg) + + +def mock_make_api_call_not_encrypted(self, operation_name, kwarg): + 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": False, + } + ] + } + if operation_name == "ListBackupVaults": + return { + "BackupVaultList": [ + { + "BackupVaultArn": "ARN", + "BackupVaultName": "Test Vault", + "EncryptionKeyArn": "", + "NumberOfRecoveryPoints": 0, + "Locked": True, + "MinRetentionDays": 1, + "MaxRetentionDays": 2, + } + ] + } + return make_api_call(self, operation_name, kwarg) + + +class Test_backup_recovery_point_encrypted: + @mock_aws + def test_no_backup_recovery_points(self): + backup_client = client("backup", region_name=AWS_REGION_EU_WEST_1) + backup_client.recovery_points = [] + + from prowler.providers.aws.services.backup.backup_service import Backup + + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted.backup_client", + new=Backup(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted import ( + backup_recovery_point_encrypted, + ) + + check = backup_recovery_point_encrypted() + result = check.execute() + + assert len(result) == 0 + + @mock_aws + def test_backup_recovery_points_not_encrypted(self): + with mock.patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_not_encrypted, + ): + # backup_client = client("backup", region_name=AWS_REGION_EU_WEST_1) + # backup_client.recovery_points = [] + + from prowler.providers.aws.services.backup.backup_service import Backup + + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted.backup_client", + new=Backup(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted import ( + backup_recovery_point_encrypted, + ) + + check = backup_recovery_point_encrypted() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "Backup Recovery Point arn:aws:backup:eu-west-1:123456789012:recovery-point:1 for Backup Vault Test Vault is not encrypted at rest." + ) + assert result[0].resource_id == "Test Vault" + assert ( + result[0].resource_arn + == "arn:aws:backup:eu-west-1:123456789012:recovery-point:1" + ) + assert result[0].resource_tags == [] + assert result[0].region == "eu-west-1" + + @mock_aws + def test_backup_recovery_points_encrypted(self): + with mock.patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_encrypted, + ): + # backup_client = client("backup", region_name=AWS_REGION_EU_WEST_1) + # backup_client.recovery_points = [] + + from prowler.providers.aws.services.backup.backup_service import Backup + + aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted.backup_client", + new=Backup(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.backup.backup_recovery_point_encrypted.backup_recovery_point_encrypted import ( + backup_recovery_point_encrypted, + ) + + check = backup_recovery_point_encrypted() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + "Backup Recovery Point arn:aws:backup:eu-west-1:123456789012:recovery-point:1 for Backup Vault Test Vault is encrypted at rest." + ) + assert result[0].resource_id == "Test Vault" + assert ( + result[0].resource_arn + == "arn:aws:backup:eu-west-1:123456789012:recovery-point:1" + ) + assert result[0].resource_tags == [] + assert result[0].region == "eu-west-1" diff --git a/tests/providers/aws/services/backup/backup_service_test.py b/tests/providers/aws/services/backup/backup_service_test.py index 8482b33df7..d4a7be392f 100644 --- a/tests/providers/aws/services/backup/backup_service_test.py +++ b/tests/providers/aws/services/backup/backup_service_test.py @@ -85,6 +85,25 @@ def mock_make_api_call(self, operation_name, kwarg): "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) @@ -253,3 +272,23 @@ class TestBackupService: 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 diff --git a/tests/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted_test.py b/tests/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted_test.py index fe0223c4eb..7aac56123f 100644 --- a/tests/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted_test.py +++ b/tests/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted_test.py @@ -63,7 +63,7 @@ class Test_backup_vaults_encrypted: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"Backup Vault {result[0].resource_id} is not encrypted." + == f"Backup Vault {result[0].resource_id} is not encrypted at rest." ) assert result[0].resource_id == "MyBackupVault" assert result[0].resource_arn == backup_vault_arn @@ -106,7 +106,7 @@ class Test_backup_vaults_encrypted: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"Backup Vault {result[0].resource_id} is encrypted." + == f"Backup Vault {result[0].resource_id} is encrypted at rest." ) assert result[0].resource_id == "MyBackupVault" assert result[0].resource_arn == backup_vault_arn