fix: resolve real Disk resource for AZ-CMP-002 encryption detection

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 <resource-group> <vm-name> <keyvault-name>
 # 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 <disk-id>` 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 <rg> --name <kv-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
 

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                                                             #
     # ------------------------------------------------------------------ #

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,40 +123,57 @@ 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": {
                         "resource_group": resource_group,
                         "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",
                     },
                 }
             )

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) #
     # ------------------------------------------------------------------ #