feat(azure): add recovery_vault_has_protected_items check (#11048)

Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>
2026-07-24 04:51:51 +00:00 · 2026-06-23 03:59:25 +09:00
parent b6caaa4268
commit a0fdc96649
10 changed files with 483 additions and 0 deletions
@@ -262,6 +262,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `bedrock_prompt_management_exists` check for AWS provider [(#10878)](https://github.com/prowler-cloud/prowler/pull/10878)
 - 8 Gmail attachment safety and spoofing protection checks for Google Workspace provider using the Cloud Identity Policy API [(#10980)](https://github.com/prowler-cloud/prowler/pull/10980)
 - `bedrock_prompt_encrypted_with_cmk` check for AWS provider [(#10905)](https://github.com/prowler-cloud/prowler/pull/10905)
+- `recovery_vault_has_protected_items` check for Azure provider, verifying that Recovery Services vaults have at least one protected backup item [(#11048)](https://github.com/prowler-cloud/prowler/pull/11048)

 ### 🔄 Changed

@@ -0,0 +1,37 @@
+{
+  "Provider": "azure",
+  "CheckID": "recovery_vault_backup_policy_retention_adequate",
+  "CheckTitle": "Recovery Services backup policy has at least 30 days retention",
+  "CheckType": [],
+  "ServiceName": "recovery",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "NotDefined",
+  "ResourceGroup": "storage",
+  "Description": "**Azure Recovery Services** backup policies are evaluated for adequate **daily retention**. Policies with fewer than 30 days of retention may not provide sufficient recovery points for incident investigation, compliance requirements, or ransomware recovery scenarios.",
+  "Risk": "Short retention periods limit the ability to recover from **delayed-discovery incidents** such as ransomware that encrypts backups over time, data corruption that propagates across backup cycles, or compliance investigations requiring historical data. Insufficient retention may violate regulatory requirements.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://learn.microsoft.com/en-us/azure/backup/backup-azure-vm-backup-faq",
+    "https://learn.microsoft.com/en-us/azure/backup/backup-azure-backup-faq"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "az backup policy set --resource-group <resource-group> --vault-name <vault-name> --name <policy-name> --policy @policy.json",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the Azure portal\n2. Go to Recovery Services vaults and select the vault\n3. Go to Backup policies\n4. Select the policy to modify\n5. Under Retention range, set Daily backup point retention to at least 30 days\n6. Click Save",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Set daily backup retention to at least **30 days**. For production workloads, consider 90+ days. Configure **weekly, monthly, and yearly** retention points for long-term compliance. Use **immutable vaults** to prevent backup deletion by compromised admin accounts.",
+      "Url": "https://hub.prowler.com/check/recovery_vault_backup_policy_retention_adequate"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "This check evaluates daily retention from the backup policy's retention_policy.daily_schedule.retention_duration.count. Policies without a daily schedule are flagged. The 30-day minimum aligns with common compliance requirements and incident response timelines."
+}
@@ -0,0 +1,65 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.recovery.recovery_client import recovery_client
+
+MINIMUM_RETENTION_DAYS = 30
+
+
+class recovery_vault_backup_policy_retention_adequate(Check):
+    """Check if vault backup policies retain backups for at least 30 days."""
+
+    def execute(self) -> list[Check_Report_Azure]:
+        """Execute the check across Recovery Services vault backup policies."""
+
+        findings = []
+
+        for subscription_name, vaults in recovery_client.vaults.items():
+            for vault in vaults.values():
+                if not vault.backup_policies:
+                    report = Check_Report_Azure(
+                        metadata=self.metadata(), resource=vault
+                    )
+                    report.subscription = subscription_name
+                    report.resource_name = vault.name
+                    report.resource_id = vault.id
+                    report.location = vault.location
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"Recovery vault '{vault.name}' has no backup "
+                        f"policies configured."
+                    )
+                    findings.append(report)
+                    continue
+
+                for policy in vault.backup_policies.values():
+                    report = Check_Report_Azure(
+                        metadata=self.metadata(), resource=vault
+                    )
+                    report.subscription = subscription_name
+                    report.resource_name = f"{vault.name}/{policy.name}"
+                    report.resource_id = policy.id
+                    report.location = vault.location
+
+                    if policy.retention_days is None:
+                        report.status = "FAIL"
+                        report.status_extended = (
+                            f"Backup policy '{policy.name}' in vault "
+                            f"'{vault.name}' has no daily retention configured."
+                        )
+                    elif policy.retention_days < MINIMUM_RETENTION_DAYS:
+                        report.status = "FAIL"
+                        report.status_extended = (
+                            f"Backup policy '{policy.name}' in vault "
+                            f"'{vault.name}' has {policy.retention_days}-day "
+                            f"retention (minimum: {MINIMUM_RETENTION_DAYS})."
+                        )
+                    else:
+                        report.status = "PASS"
+                        report.status_extended = (
+                            f"Backup policy '{policy.name}' in vault "
+                            f"'{vault.name}' has {policy.retention_days}-day "
+                            f"retention."
+                        )
+
+                    findings.append(report)
+
+        return findings
@@ -0,0 +1,37 @@
+{
+  "Provider": "azure",
+  "CheckID": "recovery_vault_has_protected_items",
+  "CheckTitle": "Recovery Services vault has backup protected items configured",
+  "CheckType": [],
+  "ServiceName": "recovery",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "microsoft.recoveryservices/vaults",
+  "ResourceGroup": "storage",
+  "Description": "**Azure Recovery Services** vaults are evaluated for **backup protected items**. A vault with no protected items indicates that no resources (VMs, databases, file shares) are configured for backup through this vault, potentially leaving critical workloads unprotected.",
+  "Risk": "Empty vaults represent a gap in **disaster recovery** coverage. Resources without backup protection face permanent data loss from accidental deletion, ransomware, corruption, or regional outages. The vault may have been provisioned but backup was never configured.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://learn.microsoft.com/en-us/azure/backup/backup-azure-vms-first-look-arm",
+    "https://learn.microsoft.com/en-us/azure/backup/backup-overview"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "az backup protection enable-for-vm --resource-group <rg> --vault-name <vault> --vm <vm_id> --policy-name DefaultPolicy",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the Azure portal\n2. Go to Recovery Services vaults and select the vault\n3. Click + Backup\n4. Select the workload type (Azure VM, SQL, File Share, etc.)\n5. Select the backup policy\n6. Choose the resources to protect\n7. Click Enable Backup",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Configure **backup protection** for critical resources including VMs, databases, and file shares. Use **Azure Backup policies** with appropriate retention and frequency. If the vault is unused, consider deleting it to reduce resource sprawl.",
+      "Url": "https://hub.prowler.com/check/recovery_vault_has_protected_items"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "This check reports on each Recovery Services vault individually. An empty vault is not necessarily a security risk if backups are managed through a different vault or Azure Backup center."
+}
@@ -0,0 +1,36 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.recovery.recovery_client import recovery_client
+
+
+class recovery_vault_has_protected_items(Check):
+    """Check if Recovery Services vaults have protected backup items."""
+
+    def execute(self) -> list[Check_Report_Azure]:
+        """Execute the check across Recovery Services vaults."""
+
+        findings = []
+
+        for subscription_name, vaults in recovery_client.vaults.items():
+            for vault in vaults.values():
+                report = Check_Report_Azure(metadata=self.metadata(), resource=vault)
+                report.subscription = subscription_name
+                report.resource_name = vault.name
+                report.resource_id = vault.id
+                report.location = vault.location
+
+                if vault.backup_protected_items:
+                    report.status = "PASS"
+                    report.status_extended = (
+                        f"Recovery Services vault '{vault.name}' has "
+                        f"{len(vault.backup_protected_items)} protected items."
+                    )
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"Recovery Services vault '{vault.name}' has no "
+                        f"protected items configured."
+                    )
+
+                findings.append(report)
+
+        return findings
@@ -0,0 +1,191 @@
+from unittest import mock
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+VAULT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/rg1/providers/Microsoft.RecoveryServices/vaults/test-vault"
+POLICY_ID = f"{VAULT_ID}/backupPolicies/DefaultPolicy"
+
+
+class Test_recovery_vault_backup_policy_retention_adequate:
+    def test_no_subscriptions(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate import (
+                recovery_vault_backup_policy_retention_adequate,
+            )
+
+            recovery_client.vaults = {}
+
+            check = recovery_vault_backup_policy_retention_adequate()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_vault_no_policies(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate import (
+                recovery_vault_backup_policy_retention_adequate,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="test-vault",
+                location="eastus",
+                backup_policies={},
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_backup_policy_retention_adequate()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "has no backup policies configured" in result[0].status_extended
+
+    def test_policy_adequate_retention(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate import (
+                recovery_vault_backup_policy_retention_adequate,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupPolicy,
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="test-vault",
+                location="eastus",
+                backup_policies={
+                    POLICY_ID: BackupPolicy(
+                        id=POLICY_ID,
+                        name="DefaultPolicy",
+                        retention_days=90,
+                    )
+                },
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_backup_policy_retention_adequate()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "90-day" in result[0].status_extended
+
+    def test_policy_insufficient_retention(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate import (
+                recovery_vault_backup_policy_retention_adequate,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupPolicy,
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="test-vault",
+                location="eastus",
+                backup_policies={
+                    POLICY_ID: BackupPolicy(
+                        id=POLICY_ID,
+                        name="ShortPolicy",
+                        retention_days=7,
+                    )
+                },
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_backup_policy_retention_adequate()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "7-day" in result[0].status_extended
+            assert "minimum: 30" in result[0].status_extended
+
+    def test_policy_no_retention_configured(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_backup_policy_retention_adequate.recovery_vault_backup_policy_retention_adequate import (
+                recovery_vault_backup_policy_retention_adequate,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupPolicy,
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="test-vault",
+                location="eastus",
+                backup_policies={
+                    POLICY_ID: BackupPolicy(
+                        id=POLICY_ID,
+                        name="NoRetentionPolicy",
+                        retention_days=None,
+                    )
+                },
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_backup_policy_retention_adequate()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "no daily retention" in result[0].status_extended
@@ -0,0 +1,116 @@
+from unittest import mock
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+VAULT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/rg1/providers/Microsoft.RecoveryServices/vaults/test-vault"
+
+
+class Test_recovery_vault_has_protected_items:
+    def test_no_subscriptions(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items import (
+                recovery_vault_has_protected_items,
+            )
+
+            recovery_client.vaults = {}
+
+            check = recovery_vault_has_protected_items()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_vault_with_protected_items(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items import (
+                recovery_vault_has_protected_items,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupItem,
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="test-vault",
+                location="eastus",
+                backup_protected_items={
+                    "item1": BackupItem(
+                        id="item1",
+                        name="vm-backup",
+                        workload_type=None,
+                    )
+                },
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_has_protected_items()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "1 protected items" in result[0].status_extended
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_id == VAULT_ID
+            assert result[0].resource_name == "test-vault"
+            assert result[0].location == "eastus"
+
+    def test_vault_empty(self):
+        recovery_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.recovery.recovery_vault_has_protected_items.recovery_vault_has_protected_items import (
+                recovery_vault_has_protected_items,
+            )
+            from prowler.providers.azure.services.recovery.recovery_service import (
+                BackupVault,
+            )
+
+            vault = BackupVault(
+                id=VAULT_ID,
+                name="empty-vault",
+                location="westeurope",
+                backup_protected_items={},
+            )
+            recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {VAULT_ID: vault}}
+
+            check = recovery_vault_has_protected_items()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "no protected items" in result[0].status_extended
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_id == VAULT_ID
+            assert result[0].resource_name == "empty-vault"
+            assert result[0].location == "westeurope"