feat(backup): add new check backup_recovery_point_encrypted (#5426)

Co-authored-by: Sergio <sergio@prowler.com>

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 <backup_vault_name> --encryption-key-arn <kms_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": ""
+}

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

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]

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": "",

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

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"

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

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