diff --git a/playbooks/cli/fix_az_cmp_002.sh b/playbooks/cli/fix_az_cmp_002.sh index 927790d..94f3baf 100644 --- a/playbooks/cli/fix_az_cmp_002.sh +++ b/playbooks/cli/fix_az_cmp_002.sh @@ -3,6 +3,16 @@ # Rule: AZ-CMP-002 — Virtual machine disk not protected by CMK or ADE # Usage: ./fix_az_cmp_002.sh # Severity: HIGH +# +# This script only remediates a *confirmed* finding (metadata.determination +# == "non_compliant" — a disk resolved to Disk.encryption.type == +# EncryptionAtRestWithPlatformKey with no ADE enabled). If the finding's +# metadata.determination is "indeterminate", the scanning principal could +# not read the Disk resource (missing Microsoft.Compute/disks/read, or the +# disk was deleted) and the actual encryption state is unknown. Running this +# script against an indeterminate finding may enable ADE on a disk that was +# already compliant via CMK. Run `az disk show --ids ` to confirm +# the encryption state before proceeding in that case. set -e @@ -17,6 +27,8 @@ if [ -z "$RESOURCE_GROUP" ] || [ -z "$VM_NAME" ] || [ -z "$KEYVAULT_NAME" ]; the echo " 1. Create a Key Vault if one does not exist:" echo " az keyvault create --resource-group --name --enabled-for-disk-encryption true" echo " 2. Ensure the VM is running before enabling encryption" + echo " 3. If the finding was indeterminate, confirm the actual encryption" + echo " state with 'az disk show' before running this script." exit 1 fi diff --git a/scanner/azure_client.py b/scanner/azure_client.py index 7e85d52..bb6f598 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -55,6 +55,7 @@ def __init__(self, subscription_id: str, credential: Optional[Any] = None) -> No self._managed_identity_principals_cache: Any = _UNSET self._subscription_role_assignments_cache: Any = _UNSET self._container_registries_cache: Any = _UNSET + self._disks_cache: Dict[str, Any] = {} self.devops_client = self._build_devops_client() def _build_devops_client(self) -> Optional[Any]: @@ -406,6 +407,43 @@ def get_vm_extensions(self, resource_group: str, vm_name: str) -> Optional[List[ logger.error("get_vm_extensions failed for %s/%s: %s", resource_group, vm_name, exc) return None + def get_disk(self, disk_id: str) -> Optional[Any]: + """Resolve the Disk resource referenced by a VM's ManagedDiskParameters. + + ManagedDiskParameters (the object embedded in a VM's storage_profile) + only carries id, storage_account_type, disk_encryption_set and + security_profile — the encryption state lives on the underlying Disk + resource and must be fetched separately. Cached per subscription + (this client instance already scopes one subscription) because the + same disk may be evaluated more than once during a scan. + + Returns: + The Disk resource, or ``None`` when the ID is missing/malformed + or Azure cannot return it (permissions, deletion, SDK error). + Callers must never interpret ``None`` as a compliant result. + """ + if not disk_id: + return None + if disk_id in self._disks_cache: + return self._disks_cache[disk_id] + + disk = None + try: + parsed = self.parse_resource_id(disk_id) + resource_group = parsed.get("resource_group", "") + disk_name = parsed.get("name", "") + if resource_group and disk_name: + client = ComputeManagementClient(self.credential, self.subscription_id) + disk = client.disks.get(resource_group, disk_name) + else: + logger.error("get_disk failed: could not parse resource group/name from %s", disk_id) + except Exception as exc: + logger.error("get_disk failed for %s: %s", disk_id, exc) + disk = None + + self._disks_cache[disk_id] = disk + return disk + # ------------------------------------------------------------------ # # Databases # # ------------------------------------------------------------------ # diff --git a/scanner/rules/az_cmp_002.py b/scanner/rules/az_cmp_002.py index 73feea7..84b8b18 100644 --- a/scanner/rules/az_cmp_002.py +++ b/scanner/rules/az_cmp_002.py @@ -1,7 +1,9 @@ """AZ-CMP-002: Virtual machine OS or data disk using platform-managed encryption only.""" import logging -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional + +from scanner.azure_client import enum_str RULE_ID = "AZ-CMP-002" RULE_NAME = "Virtual machine disk not protected by customer-managed key or ADE" @@ -23,33 +25,82 @@ ) PLAYBOOK = "playbooks/cli/fix_az_cmp_002.sh" -logger = logging.getLogger(__name__) - - -def _disk_needs_flagging(managed_disk: Any) -> bool: - """Return True only if the disk uses platform-managed encryption. +# Unreadable disks are reported as an unknown scan result, not a confirmed +# violation: a missing Microsoft.Compute/disks/read grant says nothing about +# the disk's actual encryption state, so it must not carry HIGH severity or +# the standard remediation (which could send someone to re-encrypt a disk +# that was already compliant via CMK). +INDETERMINATE_SEVERITY = "LOW" +INDETERMINATE_DESCRIPTION = ( + "One or more disks attached to this virtual machine could not be read, so their " + "encryption configuration could not be verified against CIS 7.2. This is not a " + "confirmed violation — the scanning principal could not resolve the Disk resource " + "(missing Microsoft.Compute/disks/read, transient API failure, or the disk no " + "longer exists)." +) +INDETERMINATE_REMEDIATION = ( + "Grant the scanning principal Microsoft.Compute/disks/read on the affected disk(s) " + "and re-run the scan to determine the actual encryption state." +) - Azure platform-managed encryption (EncryptionAtRestWithPlatformKey) is the - default for all managed disks and does not satisfy CIS 7.2, which requires - customer-managed keys (CMK) or Azure Disk Encryption (ADE). +logger = logging.getLogger(__name__) - Disks using EncryptionAtRestWithCustomerKey or - EncryptionAtRestWithPlatformAndCustomerKeys are compliant and should not - be flagged. +_PLATFORM_KEY_ONLY = "EncryptionAtRestWithPlatformKey" +_CUSTOMER_MANAGED_TYPES = { + "EncryptionAtRestWithCustomerKey", + "EncryptionAtRestWithPlatformAndCustomerKeys", +} + + +def _classify_disk(azure_client: Any, managed_disk: Any) -> Optional[str]: + """Classify a VM's managed disk as "compliant", "non_compliant", or "indeterminate". + + Returns None when there is no managed disk to evaluate at all (e.g. an + ephemeral OS disk with no managed disk reference) — that is not the same + as an indeterminate result and must not be surfaced as either a pass or + a finding. + + A VM's storage_profile only embeds a ManagedDiskParameters reference + (id, storage_account_type, disk_encryption_set, security_profile) — the + encryption state lives on the underlying Disk resource and must be + resolved via AzureClient.get_disk(). Azure Disk Encryption (ADE) is + reported on Disk.encryption_settings_collection.enabled and satisfies + CIS 7.2 regardless of the key type. Absent ADE, Disk.encryption.type + distinguishes EncryptionAtRestWithPlatformKey (non-compliant) from + EncryptionAtRestWithCustomerKey / EncryptionAtRestWithPlatformAndCustomerKeys + (compliant). """ if managed_disk is None: - return False + return None + + disk_id = getattr(managed_disk, "id", None) + if not disk_id: + logger.warning("AZ-CMP-002: managed disk has no id, marking indeterminate") + return "indeterminate" - encryption = getattr(managed_disk, "security_profile", None) - if encryption is None: - encryption = getattr(managed_disk, "encryption", None) + disk = azure_client.get_disk(disk_id) + if disk is None: + logger.warning("AZ-CMP-002: could not resolve Disk resource %s, marking indeterminate", disk_id) + return "indeterminate" - encryption_type = getattr(encryption, "type", None) + ade_settings = getattr(disk, "encryption_settings_collection", None) + if getattr(ade_settings, "enabled", False): + return "compliant" - if encryption_type is None: - return False + encryption = getattr(disk, "encryption", None) + encryption_type = enum_str(getattr(encryption, "type", None)) - return encryption_type == "EncryptionAtRestWithPlatformKey" + if encryption_type == _PLATFORM_KEY_ONLY: + return "non_compliant" + if encryption_type in _CUSTOMER_MANAGED_TYPES: + return "compliant" + + logger.warning( + "AZ-CMP-002: disk %s has unrecognised encryption type %r, marking indeterminate", + disk_id, + encryption_type, + ) + return "indeterminate" def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: @@ -72,33 +123,47 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: continue unencrypted_disks = [] + indeterminate_disks = [] # Check OS disk os_disk = getattr(storage_profile, "os_disk", None) if os_disk: managed_disk = getattr(os_disk, "managed_disk", None) - if _disk_needs_flagging(managed_disk): - unencrypted_disks.append(getattr(os_disk, "name", "os-disk")) + status = _classify_disk(azure_client, managed_disk) + disk_name = getattr(os_disk, "name", "os-disk") + if status == "non_compliant": + unencrypted_disks.append(disk_name) + elif status == "indeterminate": + indeterminate_disks.append(disk_name) # Check data disks data_disks = getattr(storage_profile, "data_disks", []) or [] for disk in data_disks: managed_disk = getattr(disk, "managed_disk", None) - if _disk_needs_flagging(managed_disk): - unencrypted_disks.append(getattr(disk, "name", f"data-disk-{getattr(disk, 'lun', '?')}")) - - if unencrypted_disks: + status = _classify_disk(azure_client, managed_disk) + disk_name = getattr(disk, "name", f"data-disk-{getattr(disk, 'lun', '?')}") + if status == "non_compliant": + unencrypted_disks.append(disk_name) + elif status == "indeterminate": + indeterminate_disks.append(disk_name) + + if unencrypted_disks or indeterminate_disks: + # A confirmed non-compliant disk makes the whole finding a real + # HIGH violation, even if other disks on the same VM are also + # indeterminate. Only when *nothing* is confirmed non-compliant + # does this drop to an unknown/LOW scan result. + confirmed = bool(unencrypted_disks) findings.append( { "rule_id": RULE_ID, "rule_name": RULE_NAME, - "severity": SEVERITY, + "severity": SEVERITY if confirmed else INDETERMINATE_SEVERITY, "category": CATEGORY, "resource_id": vm_id, "resource_name": vm_name, "resource_type": "Microsoft.Compute/virtualMachines", - "description": DESCRIPTION, - "remediation": REMEDIATION, + "description": DESCRIPTION if confirmed else INDETERMINATE_DESCRIPTION, + "remediation": REMEDIATION if confirmed else INDETERMINATE_REMEDIATION, "playbook": PLAYBOOK, "frameworks": FRAMEWORKS, "metadata": { @@ -106,6 +171,9 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "location": location, "unencrypted_disks": unencrypted_disks, "unencrypted_disk_count": len(unencrypted_disks), + "indeterminate_disks": indeterminate_disks, + "indeterminate_disk_count": len(indeterminate_disks), + "determination": "non_compliant" if confirmed else "indeterminate", }, } ) diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 75a6cc9..4091a28 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -56,6 +56,7 @@ def __init__(self) -> None: # --- Additional state added for full rule-coverage tests ---------- # self._network_interfaces: Dict[Tuple[str, str], Any] = {} self._vm_extensions: Dict[Tuple[str, str], Optional[List[Any]]] = {} + self._disks: Dict[str, Optional[Any]] = {} self._storage_lifecycle: Dict[Tuple[str, str], Optional[bool]] = {} self._storage_logging: Dict[Tuple[str, str, str], Optional[bool]] = {} self._virtual_networks: List[Any] = [] @@ -223,6 +224,14 @@ def get_vm_extensions(self, resource_group: str, vm_name: str) -> Optional[List[ # Default to an empty list (compliant-by-default) unless configured. return self._vm_extensions.get((resource_group, vm_name), []) + def set_disk(self, disk_id: str, disk: Optional[Any]) -> "MockAzureClient": + """Configure the Disk resource returned for a managed disk ID; ``None`` represents an unreadable disk.""" + self._disks[disk_id] = disk + return self + + def get_disk(self, disk_id: str) -> Optional[Any]: + return self._disks.get(disk_id) + # ------------------------------------------------------------------ # # Storage — lifecycle & service logging (three-state: True/False/None) # # ------------------------------------------------------------------ # diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 0b8f0eb..af07c4c 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -6,12 +6,21 @@ helper accessors from tests/helpers/mock_azure.py. """ +import pytest + import scanner.rules.az_cmp_001 as az_cmp_001 import scanner.rules.az_cmp_002 as az_cmp_002 import scanner.rules.az_cmp_003 as az_cmp_003 import scanner.rules.az_cmp_004 as az_cmp_004 from tests.helpers.mock_azure import make_resource +try: + from azure.mgmt.compute.models import Disk, Encryption, EncryptionSettingsCollection, ManagedDiskParameters + + _AZURE_SDK_AVAILABLE = True +except ImportError: # pragma: no cover - exercised only when SDK isn't installed + _AZURE_SDK_AVAILABLE = False + _REQUIRED_FIELDS = { "rule_id", "rule_name", @@ -38,6 +47,10 @@ def _nic_id(name): return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/networkInterfaces/{name}" +def _disk_id(name): + return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Compute/disks/{name}" + + # ── AZ-CMP-001: VM public IP with no NSG ──────────────────────────────────── @@ -80,14 +93,34 @@ def test_cmp_001_noncompliant_public_ip_no_nsg_returns_one_finding(mock_azure, s # ── AZ-CMP-002: disk using platform-managed encryption only ───────────────── +# +# ManagedDiskParameters (the object actually embedded in a VM's +# storage_profile.os_disk/data_disks) only exposes id, storage_account_type, +# disk_encryption_set and security_profile -- it has no "encryption" +# attribute, and its security_profile (VMDiskSecurityProfile) carries +# security_encryption_type, not "type". The real encryption state lives on +# the underlying Disk resource, resolved via AzureClient.get_disk(id), whose +# Encryption.type and EncryptionSettingsCollection.enabled are the fields +# that actually distinguish compliant from non-compliant. These fixtures +# mirror that shape instead of inventing attributes the SDK does not define. + + +def _managed_disk(name): + """A ManagedDiskParameters-shaped stub: only carries an id.""" + return make_resource(id=_disk_id(name)) + + +def _disk(encryption_type=None, ade_enabled=False): + """A Disk-shaped stub as returned by AzureClient.get_disk().""" + return make_resource( + encryption=make_resource(type=encryption_type) if encryption_type else None, + encryption_settings_collection=make_resource(enabled=ade_enabled), + ) -def test_cmp_002_compliant_cmk_disk_returns_no_findings(mock_azure, subscription_id): - """OS disk encrypted with a customer-managed key is compliant.""" - os_disk = make_resource( - name="osdisk", - managed_disk=make_resource(encryption=make_resource(type="EncryptionAtRestWithCustomerKey")), - ) +def test_cmp_002_compliant_customer_key_returns_no_findings(mock_azure, subscription_id): + """OS disk encrypted with a customer-managed key only is compliant.""" + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-cmk")) vm = make_resource( id=_vm_id("vm-cmk"), name="vm-cmk", @@ -95,15 +128,49 @@ def test_cmp_002_compliant_cmk_disk_returns_no_findings(mock_azure, subscription storage_profile=make_resource(os_disk=os_disk, data_disks=[]), ) mock_azure.set_virtual_machines([vm]) + mock_azure.set_disk(_disk_id("disk-cmk"), _disk(encryption_type="EncryptionAtRestWithCustomerKey")) assert az_cmp_002.scan(mock_azure, subscription_id) == [] -def test_cmp_002_noncompliant_platform_key_returns_one_finding(mock_azure, subscription_id): - """OS disk using platform-managed encryption only must produce one finding.""" - os_disk = make_resource( - name="osdisk", - managed_disk=make_resource(encryption=make_resource(type="EncryptionAtRestWithPlatformKey")), +def test_cmp_002_compliant_platform_and_customer_key_returns_no_findings(mock_azure, subscription_id): + """OS disk encrypted with platform-and-customer keys is compliant.""" + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-both")) + vm = make_resource( + id=_vm_id("vm-both"), + name="vm-both", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[]), ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_disk(_disk_id("disk-both"), _disk(encryption_type="EncryptionAtRestWithPlatformAndCustomerKeys")) + assert az_cmp_002.scan(mock_azure, subscription_id) == [] + + +def test_cmp_002_compliant_ade_enabled_returns_no_findings(mock_azure, subscription_id): + """A platform-key disk with Azure Disk Encryption enabled is compliant.""" + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-ade")) + vm = make_resource( + id=_vm_id("vm-ade"), + name="vm-ade", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_disk( + _disk_id("disk-ade"), _disk(encryption_type="EncryptionAtRestWithPlatformKey", ade_enabled=True) + ) + assert az_cmp_002.scan(mock_azure, subscription_id) == [] + + +def test_cmp_002_noncompliant_platform_key_returns_one_finding(mock_azure, subscription_id): + """OS disk using platform-managed encryption only, with no ADE, must be flagged. + + This is a regression test: the previous implementation read + managed_disk.security_profile / managed_disk.encryption, neither of + which exist on ManagedDiskParameters, so every branch resolved to None + and this case incorrectly returned no findings. + """ + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-pmk")) vm = make_resource( id=_vm_id("vm-pmk"), name="vm-pmk", @@ -111,6 +178,7 @@ def test_cmp_002_noncompliant_platform_key_returns_one_finding(mock_azure, subsc storage_profile=make_resource(os_disk=os_disk, data_disks=[]), ) mock_azure.set_virtual_machines([vm]) + mock_azure.set_disk(_disk_id("disk-pmk"), _disk(encryption_type="EncryptionAtRestWithPlatformKey")) findings = az_cmp_002.scan(mock_azure, subscription_id) assert len(findings) == 1 f = findings[0] @@ -118,6 +186,122 @@ def test_cmp_002_noncompliant_platform_key_returns_one_finding(mock_azure, subsc assert f["rule_id"] == "AZ-CMP-002" assert f["severity"] == "HIGH" assert f["resource_name"] == "vm-pmk" + assert f["metadata"]["unencrypted_disks"] == ["osdisk"] + assert f["metadata"]["indeterminate_disks"] == [] + assert f["metadata"]["determination"] == "non_compliant" + + +def test_cmp_002_indeterminate_unreadable_disk_returns_low_severity_unknown_finding(mock_azure, subscription_id): + """A Disk resource that cannot be read must not be treated as compliant, but it is + also not a confirmed violation, so it must not carry the same HIGH severity as an + actual platform-key-only disk — it surfaces as a distinct, lower-severity unknown + scan result instead.""" + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-unreadable")) + vm = make_resource( + id=_vm_id("vm-unreadable"), + name="vm-unreadable", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[]), + ) + mock_azure.set_virtual_machines([vm]) + # No mock_azure.set_disk() call: get_disk() returns None, as it would if + # Azure denied access or the disk had been deleted. + findings = az_cmp_002.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["unencrypted_disks"] == [] + assert f["metadata"]["indeterminate_disks"] == ["osdisk"] + assert f["metadata"]["determination"] == "indeterminate" + + +def test_cmp_002_confirmed_violation_outweighs_indeterminate_sibling_disk(mock_azure, subscription_id): + """A VM with one confirmed platform-key-only disk and one unreadable disk is a real + HIGH finding, not an unknown one — the confirmed violation must not be diluted by an + indeterminate sibling on the same VM.""" + os_disk = make_resource(name="osdisk", managed_disk=_managed_disk("disk-pmk-2")) + data_disk = make_resource(name="datadisk", lun=0, managed_disk=_managed_disk("disk-unreadable-2")) + vm = make_resource( + id=_vm_id("vm-mixed"), + name="vm-mixed", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[data_disk]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_disk(_disk_id("disk-pmk-2"), _disk(encryption_type="EncryptionAtRestWithPlatformKey")) + # disk-unreadable-2 is left unconfigured on the mock: get_disk() returns None. + findings = az_cmp_002.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "HIGH" + assert f["metadata"]["unencrypted_disks"] == ["osdisk"] + assert f["metadata"]["indeterminate_disks"] == ["datadisk"] + assert f["metadata"]["determination"] == "non_compliant" + + +@pytest.mark.skipif(not _AZURE_SDK_AVAILABLE, reason="azure-mgmt-compute not installed") +def test_cmp_002_managed_disk_parameters_has_no_encryption_attribute(): + """SDK-shape guard: the original bug read managed_disk.encryption / + managed_disk.security_profile.type, attributes ManagedDiskParameters has never + exposed. If a future SDK bump ever added them, this rule's real-model tests below + would silently stop testing anything — this test fails loudly instead if the SDK's + actual attribute surface ever drifts from what the fix assumes.""" + managed_disk = ManagedDiskParameters(id="disk-1") + assert not hasattr(managed_disk, "encryption") + + +@pytest.mark.skipif(not _AZURE_SDK_AVAILABLE, reason="azure-mgmt-compute not installed") +def test_cmp_002_noncompliant_with_real_sdk_models_returns_one_finding(mock_azure, subscription_id): + """Regression test using genuine azure.mgmt.compute.models instances (not + make_resource stand-ins, which accept arbitrary kwargs and would have silently + accepted the original bug's invented attribute shape). ManagedDiskParameters only + carries an id; Disk.encryption.type is the real source of truth for the platform-key + determination.""" + managed_disk = ManagedDiskParameters(id=_disk_id("disk-pmk-real")) + os_disk = make_resource(name="osdisk", managed_disk=managed_disk) + vm = make_resource( + id=_vm_id("vm-pmk-real"), + name="vm-pmk-real", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[]), + ) + mock_azure.set_virtual_machines([vm]) + real_disk = Disk( + location="eastus", + encryption=Encryption(type="EncryptionAtRestWithPlatformKey"), + encryption_settings_collection=EncryptionSettingsCollection(enabled=False), + ) + mock_azure.set_disk(_disk_id("disk-pmk-real"), real_disk) + + findings = az_cmp_002.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "HIGH" + assert f["metadata"]["unencrypted_disks"] == ["osdisk"] + assert f["metadata"]["determination"] == "non_compliant" + + +@pytest.mark.skipif(not _AZURE_SDK_AVAILABLE, reason="azure-mgmt-compute not installed") +def test_cmp_002_compliant_with_real_sdk_models_returns_no_findings(mock_azure, subscription_id): + """Real SDK model counterpart: a customer-managed key disk (genuine Disk/Encryption + instances) must not be flagged.""" + managed_disk = ManagedDiskParameters(id=_disk_id("disk-cmk-real")) + os_disk = make_resource(name="osdisk", managed_disk=managed_disk) + vm = make_resource( + id=_vm_id("vm-cmk-real"), + name="vm-cmk-real", + location="eastus", + storage_profile=make_resource(os_disk=os_disk, data_disks=[]), + ) + mock_azure.set_virtual_machines([vm]) + real_disk = Disk( + location="eastus", + encryption=Encryption(type="EncryptionAtRestWithCustomerKey"), + encryption_settings_collection=EncryptionSettingsCollection(enabled=False), + ) + mock_azure.set_disk(_disk_id("disk-cmk-real"), real_disk) + + assert az_cmp_002.scan(mock_azure, subscription_id) == [] # ── AZ-CMP-003: VM without endpoint protection ──────────────────────────────