feat(azure/vm): add new check vm_sufficient_daily_backup_retention_period (#8200)

Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
2026-07-23 20:42:02 +00:00 · 2025-08-11 11:44:45 +02:00
parent 5572d476ad
commit d238050065
9 changed files with 478 additions and 4 deletions
@@ -80,6 +80,7 @@ The following list includes all the Azure checks with configurable variables tha
 | `app_ensure_python_version_is_latest`                         | `python_latest_version`                          | String          |
 | `app_ensure_java_version_is_latest`                           | `java_latest_version`                            | String          |
 | `sqlserver_recommended_minimal_tls_version`                   | `recommended_minimal_tls_versions`               | List of Strings |
+| `vm_sufficient_daily_backup_retention_period`                 | `vm_backup_min_daily_retention_days`             | Integer         |
 | `vm_desired_sku_size`                                         | `desired_vm_sku_sizes`                           | List of Strings |
 | `defender_attack_path_notifications_properly_configured`      | `defender_attack_path_minimal_risk_level`        | String          |

@@ -6,6 +6,7 @@ All notable changes to the **Prowler SDK** are documented in this file.

 ### Added
 - Certificate authentication for M365 provider [(#8404)](https://github.com/prowler-cloud/prowler/pull/8404)
+- `vm_sufficient_daily_backup_retention_period` check for Azure provider [(#8200)](https://github.com/prowler-cloud/prowler/pull/8200)

 ### Changed

@@ -459,6 +459,9 @@ azure:
      "Standard_DS3_v2",
      "Standard_D4s_v3",
    ]
+  # Azure VM Backup Configuration
+  # azure.vm_sufficient_daily_backup_retention_period
+  vm_backup_min_daily_retention_days: 7

 # GCP Configuration
 gcp:
@@ -11,20 +11,30 @@ from prowler.providers.azure.lib.service.service import AzureService


 class BackupItem(BaseModel):
-    """Minimal BackupItem: only essential identifying and descriptive fields."""
+    """Model that represents a backup item."""

    id: str
    name: str
    workload_type: Optional[DataSourceType]
+    backup_policy_id: Optional[str] = None
+
+
+class BackupPolicy(BaseModel):
+    """Model that represents a backup policy."""
+
+    id: str
+    name: str
+    retention_days: Optional[int] = None


 class BackupVault(BaseModel):
-    """Minimal BackupVault: only essential identifying fields and its backup items."""
+    """Model that represents a backup vault."""

    id: str
    name: str
    location: str
    backup_protected_items: dict[str, BackupItem] = Field(default_factory=dict)
+    backup_policies: dict[str, BackupPolicy] = Field(default_factory=dict)


 class Recovery(AzureService):
@@ -71,6 +81,9 @@ class RecoveryBackup(AzureService):
                vault.backup_protected_items = self._get_backup_protected_items(
                    subscription_name=subscription_name, vault=vault
                )
+                vault.backup_policies = self._get_backup_policies(
+                    subscription_name=subscription_name, vault=vault
+                )

    def _get_backup_protected_items(
        self, subscription_name: str, vault: BackupVault
@@ -95,7 +108,58 @@ class RecoveryBackup(AzureService):
                    workload_type=(
                        item_properties.workload_type if item_properties else None
                    ),
+                    backup_policy_id=(
+                        item_properties.policy_id if item_properties else None
+                    ),
                )
-        except Exception as e:
-            logger.error(f"Recovery - Error getting backup protected items: {e}")
+        except Exception as error:
+            logger.error(
+                f"Subscription name: {subscription_name} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
        return backup_protected_items_dict
+
+    def _get_backup_policies(
+        self, subscription_name: str, vault: BackupVault
+    ) -> dict[str, BackupPolicy]:
+        """
+        Retrieve all backup policies for a given vault.
+        """
+        logger.info("Recovery - Getting backup policies...")
+        backup_policies_dict: dict[str, BackupPolicy] = {}
+        unique_backup_policies: set[str] = set()
+        try:
+            for item in vault.backup_protected_items.values():
+                if item.backup_policy_id:
+                    unique_backup_policies.add(item.backup_policy_id)
+            for policy_id in unique_backup_policies:
+                policy = self.clients[subscription_name].protection_policies.get(
+                    vault_name=vault.name,
+                    resource_group_name=vault.id.split("/")[4],
+                    policy_name=policy_id.split("/")[-1],
+                )
+                backup_policies_dict[policy_id] = BackupPolicy(
+                    id=policy.id,
+                    name=policy.name,
+                    retention_days=getattr(
+                        getattr(
+                            getattr(
+                                getattr(
+                                    getattr(policy, "properties", None),
+                                    "retention_policy",
+                                    None,
+                                ),
+                                "daily_schedule",
+                                None,
+                            ),
+                            "retention_duration",
+                            None,
+                        ),
+                        "count",
+                        None,
+                    ),
+                )
+        except Exception as error:
+            logger.error(
+                f"Subscription name: {subscription_name} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        return backup_policies_dict
@@ -0,0 +1,30 @@
+{
+  "Provider": "azure",
+  "CheckID": "vm_sufficient_daily_backup_retention_period",
+  "CheckTitle": "Ensure there is a sufficient daily backup retention period configured for Azure virtual machines.",
+  "CheckType": [],
+  "ServiceName": "vm",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "Microsoft.Compute/virtualMachines",
+  "Description": "Ensure there is a sufficient daily backup retention period configured for Azure virtual machines.",
+  "Risk": "Having an optimal daily backup retention period for your Azure virtual machines will enforce your backup strategy to follow the best practices as specified in the compliance regulations promoted by your organization. Retaining VM backups for a longer period of time will allow you to handle more efficiently your data restoration process in the event of a failure.",
+  "RelatedUrl": "https://docs.microsoft.com/en-us/azure/backup/backup-azure-vms-introduction",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/azure/VirtualMachines/sufficient-backup-retention-period.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Set the daily backup retention period for each VM's backup policy to meet or exceed your organization's minimum requirement.",
+      "Url": "https://docs.microsoft.com/en-us/azure/backup/backup-azure-vms-introduction"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,51 @@
+from azure.mgmt.recoveryservicesbackup.activestamp.models import DataSourceType
+
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.recovery.recovery_client import recovery_client
+from prowler.providers.azure.services.vm.vm_client import vm_client
+
+
+class vm_sufficient_daily_backup_retention_period(Check):
+    """
+    Ensure there is a sufficient daily backup retention period configured for Azure virtual machines.
+    - PASS: The VM has a backup policy with sufficient daily retention period.
+    - FAIL: The VM does not have a backup policy or the retention period is insufficient.
+    """
+
+    def execute(self) -> list[Check_Report_Azure]:
+        findings = []
+        min_retention_days = getattr(vm_client, "audit_config", {}).get(
+            "vm_backup_min_daily_retention_days", 7
+        )
+
+        for subscription, vms in vm_client.virtual_machines.items():
+            vaults = recovery_client.vaults.get(subscription, {})
+            for vm in vms.values():
+                backup_found = False
+                retention_days = None
+                for vault in vaults.values():
+                    for backup_item in vault.backup_protected_items.values():
+                        if (
+                            backup_item.workload_type == DataSourceType.VM
+                            and backup_item.name.split(";")[-1] == vm.resource_name
+                        ):
+                            backup_found = True
+                            policy_id = backup_item.backup_policy_id
+                            if policy_id and policy_id in vault.backup_policies:
+                                retention_days = vault.backup_policies[
+                                    policy_id
+                                ].retention_days
+                            break
+                    if backup_found:
+                        break
+                if backup_found and retention_days:
+                    report = Check_Report_Azure(metadata=self.metadata(), resource=vm)
+                    report.subscription = subscription
+                    if retention_days >= min_retention_days:
+                        report.status = "PASS"
+                        report.status_extended = f"VM {vm.resource_name} in subscription {subscription} has a daily backup retention period of {retention_days} days (minimum required: {min_retention_days})."
+                    else:
+                        report.status = "FAIL"
+                        report.status_extended = f"VM {vm.resource_name} in subscription {subscription} has insufficient daily backup retention period of {retention_days} days (minimum required: {min_retention_days})."
+                    findings.append(report)
+        return findings
@@ -85,6 +85,7 @@ class TestAzureProvider:
                "python_latest_version": "3.12",
                "java_latest_version": "17",
                "recommended_minimal_tls_versions": ["1.2", "1.3"],
+                "vm_backup_min_daily_retention_days": 7,
                "desired_vm_sku_sizes": [
                    "Standard_A8_v2",
                    "Standard_DS3_v2",
@@ -0,0 +1,323 @@
+from unittest import mock
+from uuid import uuid4
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+
+class Test_vm_sufficient_daily_backup_retention_period:
+    def test_no_subscriptions(self):
+        vm_client = mock.MagicMock()
+        recovery_client = mock.MagicMock()
+        vm_client.virtual_machines = {}
+        recovery_client.vaults = {}
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period import (
+                vm_sufficient_daily_backup_retention_period,
+            )
+
+            check = vm_sufficient_daily_backup_retention_period()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_no_vms(self):
+        vm_client = mock.MagicMock()
+        recovery_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {}}
+        recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {}}
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period import (
+                vm_sufficient_daily_backup_retention_period,
+            )
+
+            check = vm_sufficient_daily_backup_retention_period()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_vm_with_sufficient_retention(self):
+        from azure.mgmt.recoveryservicesbackup.activestamp.models import DataSourceType
+
+        from prowler.providers.azure.services.recovery.recovery_service import (
+            BackupItem,
+            BackupPolicy,
+            BackupVault,
+        )
+        from prowler.providers.azure.services.vm.vm_service import (
+            ManagedDiskParameters,
+            OSDisk,
+            StorageProfile,
+            VirtualMachine,
+        )
+
+        vm_id = str(uuid4())
+        vm_name = "VMTest"
+        vault_id = str(uuid4())
+        policy_id = str(uuid4())
+        retention_days = 14
+        min_retention_days = 7
+
+        vm = VirtualMachine(
+            resource_id=vm_id,
+            resource_name=vm_name,
+            location="eastus",
+            security_profile=None,
+            extensions=[],
+            storage_profile=StorageProfile(
+                os_disk=OSDisk(
+                    name="os_disk_name",
+                    operating_system_type="Linux",
+                    managed_disk=ManagedDiskParameters(id="managed_disk_id"),
+                ),
+                data_disks=[],
+            ),
+        )
+        backup_item = BackupItem(
+            id=str(uuid4()),
+            name=f"someprefix;{vm_name}",
+            workload_type=DataSourceType.VM,
+            backup_policy_id=policy_id,
+        )
+        backup_policy = BackupPolicy(
+            id=policy_id,
+            name="policy1",
+            retention_days=retention_days,
+        )
+        vault = BackupVault(
+            id=vault_id,
+            name="vault1",
+            location="eastus",
+            backup_protected_items={backup_item.id: backup_item},
+            backup_policies={policy_id: backup_policy},
+        )
+        vm_client = mock.MagicMock()
+        recovery_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {vm_id: vm}}
+        recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {vault_id: vault}}
+        vm_client.audit_config = {
+            "vm_backup_min_daily_retention_days": min_retention_days
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(
+                    audit_config=vm_client.audit_config
+                ),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period import (
+                vm_sufficient_daily_backup_retention_period,
+            )
+
+            check = vm_sufficient_daily_backup_retention_period()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_name == vm_name
+            assert result[0].resource_id == vm_id
+            assert (
+                f"has a daily backup retention period of {retention_days} days"
+                in result[0].status_extended
+            )
+
+    def test_vm_with_insufficient_retention(self):
+        from azure.mgmt.recoveryservicesbackup.activestamp.models import DataSourceType
+
+        from prowler.providers.azure.services.recovery.recovery_service import (
+            BackupItem,
+            BackupPolicy,
+            BackupVault,
+        )
+        from prowler.providers.azure.services.vm.vm_service import (
+            ManagedDiskParameters,
+            OSDisk,
+            StorageProfile,
+            VirtualMachine,
+        )
+
+        vm_id = str(uuid4())
+        vm_name = "VMTest"
+        vault_id = str(uuid4())
+        policy_id = str(uuid4())
+        retention_days = 3
+        min_retention_days = 7
+
+        vm = VirtualMachine(
+            resource_id=vm_id,
+            resource_name=vm_name,
+            location="eastus",
+            security_profile=None,
+            extensions=[],
+            storage_profile=StorageProfile(
+                os_disk=OSDisk(
+                    name="os_disk_name",
+                    operating_system_type="Linux",
+                    managed_disk=ManagedDiskParameters(id="managed_disk_id"),
+                ),
+                data_disks=[],
+            ),
+        )
+        backup_item = BackupItem(
+            id=str(uuid4()),
+            name=f"someprefix;{vm_name}",
+            workload_type=DataSourceType.VM,
+            backup_policy_id=policy_id,
+        )
+        backup_policy = BackupPolicy(
+            id=policy_id,
+            name="policy1",
+            retention_days=retention_days,
+        )
+        vault = BackupVault(
+            id=vault_id,
+            name="vault1",
+            location="eastus",
+            backup_protected_items={backup_item.id: backup_item},
+            backup_policies={policy_id: backup_policy},
+        )
+        vm_client = mock.MagicMock()
+        recovery_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {vm_id: vm}}
+        recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {vault_id: vault}}
+        vm_client.audit_config = {
+            "vm_backup_min_daily_retention_days": min_retention_days
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(
+                    audit_config=vm_client.audit_config
+                ),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period import (
+                vm_sufficient_daily_backup_retention_period,
+            )
+
+            check = vm_sufficient_daily_backup_retention_period()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_name == vm_name
+            assert result[0].resource_id == vm_id
+            assert (
+                f"has insufficient daily backup retention period of {retention_days} days"
+                in result[0].status_extended
+            )
+
+    def test_vm_with_no_backup_policy(self):
+        from azure.mgmt.recoveryservicesbackup.activestamp.models import DataSourceType
+
+        from prowler.providers.azure.services.recovery.recovery_service import (
+            BackupItem,
+            BackupVault,
+        )
+        from prowler.providers.azure.services.vm.vm_service import (
+            ManagedDiskParameters,
+            OSDisk,
+            StorageProfile,
+            VirtualMachine,
+        )
+
+        vm_id = str(uuid4())
+        vm_name = "VMTest"
+        vault_id = str(uuid4())
+
+        vm = VirtualMachine(
+            resource_id=vm_id,
+            resource_name=vm_name,
+            location="eastus",
+            security_profile=None,
+            extensions=[],
+            storage_profile=StorageProfile(
+                os_disk=OSDisk(
+                    name="os_disk_name",
+                    operating_system_type="Linux",
+                    managed_disk=ManagedDiskParameters(id="managed_disk_id"),
+                ),
+                data_disks=[],
+            ),
+        )
+        backup_item = BackupItem(
+            id=str(uuid4()),
+            name=f"someprefix;{vm_name}",
+            workload_type=DataSourceType.VM,
+            backup_policy_id=None,
+        )
+        vault = BackupVault(
+            id=vault_id,
+            name="vault1",
+            location="eastus",
+            backup_protected_items={backup_item.id: backup_item},
+            backup_policies={},
+        )
+        vm_client = mock.MagicMock()
+        recovery_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {vm_id: vm}}
+        recovery_client.vaults = {AZURE_SUBSCRIPTION_ID: {vault_id: vault}}
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period.recovery_client",
+                new=recovery_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_sufficient_daily_backup_retention_period.vm_sufficient_daily_backup_retention_period import (
+                vm_sufficient_daily_backup_retention_period,
+            )
+
+            check = vm_sufficient_daily_backup_retention_period()
+            result = check.execute()
+            assert len(result) == 0