From 763a4fc5fd33c10dd31a4b7166fda06f7a4990a6 Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Fri, 7 Aug 2026 19:32:22 +0100 Subject: [PATCH 1/2] fix: resolve real Disk resource for AZ-CMP-002 encryption detection _disk_needs_flagging() read managed_disk.security_profile.type and managed_disk.encryption.type, neither of which exist on azure.mgmt.compute.models.ManagedDiskParameters (id, storage_account_type, disk_encryption_set, security_profile) or VMDiskSecurityProfile (security_encryption_type, disk_encryption_set). Every branch resolved to None, so the rule never flagged a disk under any configuration. Add AzureClient.get_disk() to resolve the underlying Disk resource from a managed disk's id, cached per subscription, returning None (never treated as compliant) on failure. Classify each disk from Disk.encryption.type (platform key vs customer key vs platform-and-customer keys) and Disk.encryption_settings_collection.enabled (ADE). Disks that cannot be read are reported as indeterminate in finding metadata rather than silently passing. Rebuild the AZ-CMP-002 test fixtures against the real SDK attribute surface and add compliant/non-compliant/indeterminate coverage, including a regression test for the platform-key-only case that the old implementation always missed. Fixes #236 Signed-off-by: PARTH J ROHIT --- playbooks/cli/fix_az_cmp_002.sh | 12 ++++ scanner/azure_client.py | 38 +++++++++++ scanner/rules/az_cmp_002.py | 109 +++++++++++++++++++++++--------- tests/helpers/mock_azure.py | 9 +++ tests/test_rules_compute.py | 104 ++++++++++++++++++++++++++---- 5 files changed, 230 insertions(+), 42 deletions(-) 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..0258809 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" @@ -10,46 +12,79 @@ FRAMEWORKS = {"CIS": "7.2", "NIST": "PR.DS-1", "ISO27001": "A.10.1.1", "SOC2": "CC6.7"} DESCRIPTION = ( "One or more disks attached to this virtual machine are using platform-managed " - "encryption only (EncryptionAtRestWithPlatformKey). CIS 7.2 requires disks to be " - "protected using either Azure Disk Encryption (ADE) or server-side encryption with " - "a customer-managed key (CMK). Platform-managed encryption does not give the " - "organisation control over the encryption keys." + "encryption only (EncryptionAtRestWithPlatformKey), or their encryption state could " + "not be verified. CIS 7.2 requires disks to be protected using either Azure Disk " + "Encryption (ADE) or server-side encryption with a customer-managed key (CMK). " + "Platform-managed encryption does not give the organisation control over the " + "encryption keys." ) REMEDIATION = ( "Configure server-side encryption with a customer-managed key via a Disk Encryption " "Set, or enable Azure Disk Encryption on all OS and data disks. Navigate to: " "Virtual Machine > Disks > Additional settings > Disk encryption set, or use " - "az vm encryption enable with a Key Vault." + "az vm encryption enable with a Key Vault. If any disks were reported indeterminate, " + "grant the scanning principal Microsoft.Compute/disks/read and re-run the scan." ) 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. - - 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). - - 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 - encryption = getattr(managed_disk, "security_profile", None) - if encryption is None: - encryption = getattr(managed_disk, "encryption", 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_type = getattr(encryption, "type", 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" - if encryption_type is None: - return False + ade_settings = getattr(disk, "encryption_settings_collection", None) + if getattr(ade_settings, "enabled", False): + return "compliant" - return encryption_type == "EncryptionAtRestWithPlatformKey" + encryption = getattr(disk, "encryption", None) + encryption_type = enum_str(getattr(encryption, "type", None)) + + 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,22 +107,31 @@ 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: findings.append( { "rule_id": RULE_ID, @@ -106,6 +150,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 unencrypted_disks 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..6863e64 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -38,6 +38,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 +84,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 +119,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 +169,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 +177,29 @@ 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_one_finding(mock_azure, subscription_id): + """A Disk resource that cannot be read must not be treated as compliant.""" + 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["metadata"]["unencrypted_disks"] == [] + assert f["metadata"]["indeterminate_disks"] == ["osdisk"] + assert f["metadata"]["determination"] == "indeterminate" # ── AZ-CMP-003: VM without endpoint protection ────────────────────────────── From c0252adadb95412f85b1a7a366320075e0e58e7c Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Fri, 7 Aug 2026 20:14:50 +0100 Subject: [PATCH 2/2] fix: downgrade indeterminate AZ-CMP-002 disks to a LOW unknown result Per review on #237: an unreadable Disk resource is not a confirmed CIS 7.2 violation, so it must not carry the same HIGH severity and remediation text as a genuine platform-key-only disk. Indeterminate disks now produce a distinct LOW-severity finding with its own description/remediation pointing at the missing Microsoft.Compute/disks/read grant, while a confirmed non-compliant disk still outweighs any indeterminate sibling on the same VM and keeps HIGH severity. Add regression tests built from genuine azure.mgmt.compute.models.ManagedDiskParameters/Disk/Encryption instances (not the make_resource stand-in, which accepts arbitrary kwargs and would have silently accepted the original bug's invented attribute shape) so the SDK-shape mismatch that caused the original fail-open bug cannot recur unnoticed. One test asserts ManagedDiskParameters has no encryption attribute directly, as a canary against a future SDK shape drift. Refs #236 Signed-off-by: PARTH J ROHIT --- scanner/rules/az_cmp_002.py | 43 +++++++++++---- tests/test_rules_compute.py | 106 +++++++++++++++++++++++++++++++++++- 2 files changed, 136 insertions(+), 13 deletions(-) diff --git a/scanner/rules/az_cmp_002.py b/scanner/rules/az_cmp_002.py index 0258809..84b8b18 100644 --- a/scanner/rules/az_cmp_002.py +++ b/scanner/rules/az_cmp_002.py @@ -12,21 +12,37 @@ FRAMEWORKS = {"CIS": "7.2", "NIST": "PR.DS-1", "ISO27001": "A.10.1.1", "SOC2": "CC6.7"} DESCRIPTION = ( "One or more disks attached to this virtual machine are using platform-managed " - "encryption only (EncryptionAtRestWithPlatformKey), or their encryption state could " - "not be verified. CIS 7.2 requires disks to be protected using either Azure Disk " - "Encryption (ADE) or server-side encryption with a customer-managed key (CMK). " - "Platform-managed encryption does not give the organisation control over the " - "encryption keys." + "encryption only (EncryptionAtRestWithPlatformKey). CIS 7.2 requires disks to be " + "protected using either Azure Disk Encryption (ADE) or server-side encryption with " + "a customer-managed key (CMK). Platform-managed encryption does not give the " + "organisation control over the encryption keys." ) REMEDIATION = ( "Configure server-side encryption with a customer-managed key via a Disk Encryption " "Set, or enable Azure Disk Encryption on all OS and data disks. Navigate to: " "Virtual Machine > Disks > Additional settings > Disk encryption set, or use " - "az vm encryption enable with a Key Vault. If any disks were reported indeterminate, " - "grant the scanning principal Microsoft.Compute/disks/read and re-run the scan." + "az vm encryption enable with a Key Vault." ) PLAYBOOK = "playbooks/cli/fix_az_cmp_002.sh" +# 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." +) + logger = logging.getLogger(__name__) _PLATFORM_KEY_ONLY = "EncryptionAtRestWithPlatformKey" @@ -132,17 +148,22 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: 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": { @@ -152,7 +173,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "unencrypted_disk_count": len(unencrypted_disks), "indeterminate_disks": indeterminate_disks, "indeterminate_disk_count": len(indeterminate_disks), - "determination": "non_compliant" if unencrypted_disks else "indeterminate", + "determination": "non_compliant" if confirmed else "indeterminate", }, } ) diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 6863e64..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", @@ -182,8 +191,11 @@ def test_cmp_002_noncompliant_platform_key_returns_one_finding(mock_azure, subsc assert f["metadata"]["determination"] == "non_compliant" -def test_cmp_002_indeterminate_unreadable_disk_returns_one_finding(mock_azure, subscription_id): - """A Disk resource that cannot be read must not be treated as 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"), @@ -197,11 +209,101 @@ def test_cmp_002_indeterminate_unreadable_disk_returns_one_finding(mock_azure, s 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 ──────────────────────────────