From b3deba5465a0580bc71226518a4caadbddcb2afa Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Thu, 13 Aug 2026 13:21:14 +0100 Subject: [PATCH 1/9] fix(scanner): remove compute-rule false positives/negatives (AZ-CMP-001/003) AZ-CMP-001: a VM's NIC lacking its own NSG is no longer flagged if the NIC's subnet carries a protecting NSG instead (a valid, common Azure pattern). Added AzureClient.get_subnet() to resolve the subnet referenced by a NIC's ip_configuration; an unresolvable subnet is still treated as unprotected so this only removes false positives, never introduces false negatives. AZ-CMP-003: a recognised endpoint-protection extension whose provisioning_state is present and not "Succeeded" is no longer read as a silent pass. It now surfaces as an indeterminate/LOW finding instead, mirroring AZ-CMP-002's existing determination convention. Extensions with no provisioning_state data (not exposed by the API) still fall back to the prior name-based check to avoid inventing new false positives. AZ-CMP-004: verified against current code -- it already checks patch_mode == "AutomaticByPlatform" for both Windows and Linux, so the weak-signal claim in #268 does not apply here. No change made. Fixes #268 Signed-off-by: PARTH J ROHIT --- scanner/azure_client.py | 39 +++++++++ scanner/rules/az_cmp_001.py | 35 +++++++- scanner/rules/az_cmp_003.py | 68 ++++++++++++++- tests/helpers/mock_azure.py | 9 ++ tests/test_azure_client_management.py | 13 +++ tests/test_rules_compute.py | 120 ++++++++++++++++++++++++++ 6 files changed, 277 insertions(+), 7 deletions(-) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index d3b1bb21..ca6dfc87 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -287,6 +287,45 @@ def get_route_tables(self) -> Optional[List[Any]]: logger.error("get_route_tables failed: %s", exc) return None + def get_subnet(self, subnet_id: str) -> Optional[Any]: + """Resolve the Subnet resource referenced by a NIC's IP configuration. + + A NIC's ip_configuration only embeds the subnet's resource ID + (.../virtualNetworks/{vnet}/subnets/{subnet}) - whether that subnet + carries its own NSG must be fetched separately via the Network + management client. + + Returns: + The Subnet resource, or ``None`` when the ID is missing/malformed + or Azure cannot return it (permissions, deletion, SDK error). + Callers must never interpret ``None`` as "subnet has no NSG". + """ + if not subnet_id: + return None + try: + resource_group = "" + vnet_name = "" + subnet_name = "" + parts = subnet_id.split("/") + for idx, segment in enumerate(parts): + lowered = segment.lower() + if lowered == "resourcegroups" and idx + 1 < len(parts): + resource_group = parts[idx + 1] + elif lowered == "virtualnetworks" and idx + 1 < len(parts): + vnet_name = parts[idx + 1] + elif lowered == "subnets" and idx + 1 < len(parts): + subnet_name = parts[idx + 1] + + if not (resource_group and vnet_name and subnet_name): + logger.error("get_subnet failed: could not parse %s", subnet_id) + return None + + client = NetworkManagementClient(self.credential, self.subscription_id) + return client.subnets.get(resource_group, vnet_name, subnet_name) + except Exception as exc: + logger.error("get_subnet failed for %s: %s", subnet_id, exc) + return None + def get_virtual_networks(self) -> List[Any]: """List all virtual networks in the subscription.""" try: diff --git a/scanner/rules/az_cmp_001.py b/scanner/rules/az_cmp_001.py index e47ebee1..215058d1 100644 --- a/scanner/rules/az_cmp_001.py +++ b/scanner/rules/az_cmp_001.py @@ -24,8 +24,31 @@ logger = logging.getLogger(__name__) +def _subnet_has_nsg(azure_client: Any, nic: Any) -> bool: + """Return whether any subnet backing this NIC's IP configurations carries its own NSG. + + A NIC without a NIC-level NSG can still be protected by an NSG attached + to its subnet - that is a common, valid Azure pattern, not a + misconfiguration. An unresolvable subnet (missing ID, permissions, SDK + error) is treated the same as "no subnet NSG" - it must never be read as + protection, only as "could not confirm", which keeps this check from + introducing new false negatives. + """ + for ip_cfg in getattr(nic, "ip_configurations", []) or []: + subnet_ref = getattr(ip_cfg, "subnet", None) + subnet_id = getattr(subnet_ref, "id", None) + if not subnet_id: + continue + + subnet = azure_client.get_subnet(subnet_id) + if subnet is not None and getattr(subnet, "network_security_group", None): + return True + + return False + + def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect VMs whose NIC has a public IP but no NSG attached.""" + """Detect VMs whose NIC has a public IP but no NSG at the NIC or subnet level.""" findings: List[Dict[str, Any]] = [] for vm in azure_client.get_virtual_machines(): @@ -51,9 +74,13 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: has_public_ip = any( getattr(ip_cfg, "public_ip_address", None) for ip_cfg in (getattr(nic, "ip_configurations", []) or []) ) - has_nsg = bool(getattr(nic, "network_security_group", None)) + has_nic_nsg = bool(getattr(nic, "network_security_group", None)) + + has_subnet_nsg = False + if has_public_ip and not has_nic_nsg: + has_subnet_nsg = _subnet_has_nsg(azure_client, nic) - if has_public_ip and not has_nsg: + if has_public_ip and not has_nic_nsg and not has_subnet_nsg: findings.append( { "rule_id": RULE_ID, @@ -70,6 +97,8 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "metadata": { "nic_id": nic_id, "nic_name": nic_name, + "nic_nsg_attached": has_nic_nsg, + "subnet_nsg_attached": has_subnet_nsg, }, } ) diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index 5f7cf0ba..a5dbeb81 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -21,6 +21,22 @@ REMEDIATION = "Install IaaSAntimalware or onboard to MDE (MDE.Windows / MDE.Linux) depending on the OS." PLAYBOOK = "playbooks/cli/fix_az_cmp_003.sh" +# A recognised EP extension whose provisioning_state is present and is not +# "Succeeded" is not actually protecting the VM - name presence alone was +# the previous (weak) signal. This is surfaced as an indeterminate result, +# not a confirmed absence of endpoint protection, since a transient/failed +# provisioning state does not prove malware protection is truly off. +INDETERMINATE_SEVERITY = "LOW" +INDETERMINATE_DESCRIPTION = ( + "A recognised endpoint protection extension is installed but its provisioning state " + "indicates it did not complete successfully, so effective protection cannot be " + "confirmed. This is not a confirmed absence of endpoint protection." +) +INDETERMINATE_REMEDIATION = ( + "Check the extension's status in the Azure Portal (VM > Extensions) and, if failed or " + "stuck, remove and reinstall it, then re-run the scan to confirm successful provisioning." +) + KNOWN_EP_EXTENSIONS = { "microsoftmonitoringagent", "mde.linux", @@ -45,7 +61,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: if exts is None: continue - installed = set() + installed: Dict[str, Any] = {} for e in exts: t = ( getattr(e, "type_properties_type", None) @@ -53,9 +69,11 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: or getattr(e, "type", "") ) if t: - installed.add(t.lower()) + installed[t.lower()] = e - if not installed.intersection(KNOWN_EP_EXTENSIONS): + matched = {name: ext for name, ext in installed.items() if name in KNOWN_EP_EXTENSIONS} + + if not matched: findings.append( { "rule_id": RULE_ID, @@ -71,9 +89,51 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "frameworks": FRAMEWORKS, "metadata": { "resource_group": rg, - "installed_extensions": sorted(installed), + "installed_extensions": sorted(installed.keys()), + "determination": "non_compliant", }, } ) + continue + + # A recognised EP extension is installed - name presence alone is not + # enough. Where the API exposes provisioning_state, an extension is + # only treated as healthy when it is unset/unknown (data doesn't + # support the check - do not invent a new false positive) or equals + # "Succeeded". Anything else (Failed, Canceled, ...) is surfaced as + # indeterminate rather than silently passed. + unhealthy_names = [] + confirmed_healthy = False + for name, ext in matched.items(): + provisioning_state = (getattr(ext, "provisioning_state", None) or "").lower() + if not provisioning_state or provisioning_state == "succeeded": + confirmed_healthy = True + break + unhealthy_names.append(name) + + if confirmed_healthy: + continue + + findings.append( + { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": INDETERMINATE_SEVERITY, + "category": CATEGORY, + "resource_id": vm.id, + "resource_name": vm_name, + "resource_type": "Microsoft.Compute/virtualMachines", + "description": INDETERMINATE_DESCRIPTION, + "remediation": INDETERMINATE_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": rg, + "installed_extensions": sorted(installed.keys()), + "unhealthy_extensions": sorted(unhealthy_names), + "determination": "indeterminate", + }, + } + ) return findings diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index de3c6be0..0403db17 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -58,6 +58,7 @@ def __init__(self) -> None: self._network_interfaces: Dict[Tuple[str, str], Any] = {} self._all_network_interfaces: Optional[List[Any]] = [] self._route_tables: Optional[List[Any]] = [] + self._subnets: Dict[str, Optional[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]] = {} @@ -239,6 +240,14 @@ def set_network_interface(self, resource_group: str, nic_name: str, nic: Any) -> def get_network_interface(self, resource_group: str, nic_name: str) -> Optional[Any]: return self._network_interfaces.get((resource_group, nic_name)) + def set_subnet(self, subnet_id: str, subnet: Optional[Any]) -> "MockAzureClient": + """Configure the Subnet resource returned for a subnet ID; ``None`` represents an unreadable subnet.""" + self._subnets[subnet_id] = subnet + return self + + def get_subnet(self, subnet_id: str) -> Optional[Any]: + return self._subnets.get(subnet_id) + def set_vm_extensions( self, resource_group: str, vm_name: str, extensions: Optional[List[Any]] ) -> "MockAzureClient": diff --git a/tests/test_azure_client_management.py b/tests/test_azure_client_management.py index b541f1d6..8df2a0ca 100644 --- a/tests/test_azure_client_management.py +++ b/tests/test_azure_client_management.py @@ -59,6 +59,19 @@ def test_single_resource_and_policy_wrappers(client): constructor.return_value.network_interfaces.get.side_effect = RuntimeError("denied") assert client.get_network_interface("rg", "nic") is None + subnet_id = "/subscriptions/s/resourceGroups/RG/providers/Microsoft.Network/virtualNetworks/vnet1/subnets/subnet1" + with patch("scanner.azure_client.NetworkManagementClient") as constructor: + constructor.return_value.subnets.get.return_value = SimpleNamespace(name="subnet1") + result = client.get_subnet(subnet_id) + assert result.name == "subnet1" + constructor.return_value.subnets.get.assert_called_once_with("RG", "vnet1", "subnet1") + + constructor.return_value.subnets.get.side_effect = RuntimeError("denied") + assert client.get_subnet(subnet_id) is None + + assert client.get_subnet("") is None + assert client.get_subnet("/not/a/valid/subnet/id") is None + with patch("scanner.azure_client.SqlManagementClient") as constructor: policy = SimpleNamespace(state="Enabled") constructor.return_value.server_blob_auditing_policies.get.return_value = policy diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index af07c4c7..175fb491 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -47,6 +47,13 @@ def _nic_id(name): return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/networkInterfaces/{name}" +def _subnet_id(vnet_name, subnet_name): + return ( + f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/" + f"virtualNetworks/{vnet_name}/subnets/{subnet_name}" + ) + + def _disk_id(name): return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Compute/disks/{name}" @@ -92,6 +99,76 @@ def test_cmp_001_noncompliant_public_ip_no_nsg_returns_one_finding(mock_azure, s assert f["resource_name"] == "vm-exposed" +def test_cmp_001_compliant_subnet_nsg_returns_no_findings(mock_azure, subscription_id): + """No NIC-level NSG, but the NIC's subnet carries one - must NOT be flagged.""" + subnet_id = _subnet_id("vnet1", "subnet1") + nic = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=subnet_id)) + ], + network_security_group=None, + ) + subnet = make_resource(id=subnet_id, network_security_group=make_resource(id="subnet-nsg")) + vm = make_resource( + id=_vm_id("vm-subnet-protected"), + name="vm-subnet-protected", + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id("nic1"))]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic1", nic) + mock_azure.set_subnet(subnet_id, subnet) + assert az_cmp_001.scan(mock_azure, subscription_id) == [] + + +def test_cmp_001_noncompliant_no_nic_nsg_no_subnet_nsg_returns_one_finding(mock_azure, subscription_id): + """Neither the NIC nor its subnet has an NSG - must still produce exactly one finding.""" + subnet_id = _subnet_id("vnet1", "subnet1") + nic = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=subnet_id)) + ], + network_security_group=None, + ) + subnet = make_resource(id=subnet_id, network_security_group=None) + vm = make_resource( + id=_vm_id("vm-fully-exposed"), + name="vm-fully-exposed", + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id("nic1"))]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic1", nic) + mock_azure.set_subnet(subnet_id, subnet) + findings = az_cmp_001.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["rule_id"] == "AZ-CMP-001" + assert f["severity"] == "HIGH" + assert f["metadata"]["nic_nsg_attached"] is False + assert f["metadata"]["subnet_nsg_attached"] is False + + +def test_cmp_001_unresolvable_subnet_still_flags(mock_azure, subscription_id): + """A subnet that fails to resolve (permissions/deleted) must not be read as protection.""" + subnet_id = _subnet_id("vnet1", "subnet1") + nic = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=subnet_id)) + ], + network_security_group=None, + ) + vm = make_resource( + id=_vm_id("vm-unresolvable-subnet"), + name="vm-unresolvable-subnet", + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id("nic1"))]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic1", nic) + # No set_subnet() call -> get_subnet() returns None, simulating an unreadable subnet. + findings = az_cmp_001.scan(mock_azure, subscription_id) + assert len(findings) == 1 + + # ── AZ-CMP-002: disk using platform-managed encryption only ───────────────── # # ManagedDiskParameters (the object actually embedded in a VM's @@ -345,6 +422,49 @@ def test_cmp_003_extensions_none_skips_without_finding(mock_azure, subscription_ assert az_cmp_003.scan(mock_azure, subscription_id) == [] +def test_cmp_003_extension_present_and_confirmed_healthy_returns_no_findings(mock_azure, subscription_id): + """A recognised EP extension with provisioning_state 'Succeeded' is compliant.""" + vm = make_resource(id=_vm_id("vm-healthy"), name="vm-healthy") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions( + _RG, + "vm-healthy", + [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")], + ) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + +def test_cmp_003_extension_present_but_unhealthy_returns_indeterminate_finding(mock_azure, subscription_id): + """A recognised EP extension that failed to provision must not be a silent pass.""" + vm = make_resource(id=_vm_id("vm-degraded"), name="vm-degraded") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions( + _RG, + "vm-degraded", + [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Failed")], + ) + findings = az_cmp_003.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["rule_id"] == "AZ-CMP-003" + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["unhealthy_extensions"] == ["iaasantimalware"] + + +def test_cmp_003_missing_provisioning_state_falls_back_to_name_based_pass(mock_azure, subscription_id): + """When provisioning_state isn't exposed by the API, name presence alone remains sufficient.""" + vm = make_resource(id=_vm_id("vm-no-state-data"), name="vm-no-state-data") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions( + _RG, + "vm-no-state-data", + [make_resource(type_properties_type="IaaSAntimalware")], + ) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + # ── AZ-CMP-004: VM without automatic OS patching ──────────────────────────── From d036ad4698c59b15b8d7285a72f3da3606b28313 Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Sun, 16 Aug 2026 14:18:39 +0100 Subject: [PATCH 2/9] fix(scanner): treat unresolvable AZ-CMP-001 subnet as indeterminate, not confirmed HIGH An unresolvable subnet (permissions gap, transient API error, deleted resource) said nothing about whether it actually has an NSG, but was being folded into "no subnet NSG" and reported as a confirmed HIGH finding. Split subnet resolution into a tri-state result (protected / confirmed unprotected / unknown) and report the unknown case as a LOW indeterminate finding instead, matching the confirmed/indeterminate convention already used by AZ-CMP-002. Signed-off-by: PARTH J ROHIT --- scanner/rules/az_cmp_001.py | 63 +++++++++++++++++++++++++++---------- tests/test_rules_compute.py | 39 +++++++++++++++++++++-- 2 files changed, 84 insertions(+), 18 deletions(-) diff --git a/scanner/rules/az_cmp_001.py b/scanner/rules/az_cmp_001.py index 215058d1..73e8a7fe 100644 --- a/scanner/rules/az_cmp_001.py +++ b/scanner/rules/az_cmp_001.py @@ -1,7 +1,7 @@ """AZ-CMP-001: Virtual machine has a public IP with no associated NSG.""" import logging -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional RULE_ID = "AZ-CMP-001" RULE_NAME = "VM with Public IP and No Associated NSG on Network Interface" @@ -21,19 +21,44 @@ ) PLAYBOOK = "playbooks/cli/fix_az_cmp_001.sh" +# An unresolvable subnet says nothing about whether it actually has an NSG, so +# it must not be read as "confirmed unprotected" and carry the same HIGH +# severity as a real finding -- that would let a permissions gap or a +# transient API failure masquerade as a genuine misconfiguration. Mirrors the +# same confirmed/indeterminate split already used by AZ-CMP-002. +INDETERMINATE_SEVERITY = "LOW" +INDETERMINATE_DESCRIPTION = ( + "A virtual machine has a public IP address assigned to its network interface with no " + "NSG on the interface itself, and the subnet backing at least one IP configuration " + "could not be resolved, so subnet-level protection could not be verified. This is not " + "a confirmed violation -- the scanning principal could not resolve the Subnet resource " + "(missing Microsoft.Network/virtualNetworks/subnets/read, transient API failure, or the " + "subnet no longer exists)." +) +INDETERMINATE_REMEDIATION = ( + "Grant the scanning principal Microsoft.Network/virtualNetworks/subnets/read on the " + "affected subnet(s) and re-run the scan to determine whether the subnet actually has " + "an NSG attached." +) + logger = logging.getLogger(__name__) -def _subnet_has_nsg(azure_client: Any, nic: Any) -> bool: - """Return whether any subnet backing this NIC's IP configurations carries its own NSG. +def _subnet_nsg_status(azure_client: Any, nic: Any) -> Optional[bool]: + """Resolve whether any subnet backing this NIC's IP configurations carries its own NSG. A NIC without a NIC-level NSG can still be protected by an NSG attached to its subnet - that is a common, valid Azure pattern, not a - misconfiguration. An unresolvable subnet (missing ID, permissions, SDK - error) is treated the same as "no subnet NSG" - it must never be read as - protection, only as "could not confirm", which keeps this check from - introducing new false negatives. + misconfiguration. + + Returns: + True - at least one backing subnet was resolved and has an NSG (protected). + False - every backing subnet was resolved and none has an NSG (confirmed unprotected). + None - at least one backing subnet could not be resolved (missing ID, permissions, + SDK error) and none of the resolvable ones confirmed protection, so + subnet-level protection cannot be confirmed either way. """ + unresolved = False for ip_cfg in getattr(nic, "ip_configurations", []) or []: subnet_ref = getattr(ip_cfg, "subnet", None) subnet_id = getattr(subnet_ref, "id", None) @@ -41,10 +66,14 @@ def _subnet_has_nsg(azure_client: Any, nic: Any) -> bool: continue subnet = azure_client.get_subnet(subnet_id) - if subnet is not None and getattr(subnet, "network_security_group", None): + if subnet is None: + unresolved = True + continue + + if getattr(subnet, "network_security_group", None): return True - return False + return None if unresolved else False def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: @@ -76,29 +105,31 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: ) has_nic_nsg = bool(getattr(nic, "network_security_group", None)) - has_subnet_nsg = False + subnet_status: Optional[bool] = None if has_public_ip and not has_nic_nsg: - has_subnet_nsg = _subnet_has_nsg(azure_client, nic) + subnet_status = _subnet_nsg_status(azure_client, nic) - if has_public_ip and not has_nic_nsg and not has_subnet_nsg: + if has_public_ip and not has_nic_nsg and subnet_status is not True: + confirmed = subnet_status is False 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": { "nic_id": nic_id, "nic_name": nic_name, "nic_nsg_attached": has_nic_nsg, - "subnet_nsg_attached": has_subnet_nsg, + "subnet_nsg_attached": False, + "determination": "non_compliant" if confirmed else "indeterminate", }, } ) diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 175fb491..f347694e 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -146,10 +146,13 @@ def test_cmp_001_noncompliant_no_nic_nsg_no_subnet_nsg_returns_one_finding(mock_ assert f["severity"] == "HIGH" assert f["metadata"]["nic_nsg_attached"] is False assert f["metadata"]["subnet_nsg_attached"] is False + assert f["metadata"]["determination"] == "non_compliant" -def test_cmp_001_unresolvable_subnet_still_flags(mock_azure, subscription_id): - """A subnet that fails to resolve (permissions/deleted) must not be read as protection.""" +def test_cmp_001_unresolvable_subnet_is_indeterminate_not_confirmed_high(mock_azure, subscription_id): + """A subnet that fails to resolve (permissions/deleted) must not be read as a confirmed + HIGH violation — the scanning principal simply couldn't verify subnet-level protection, + which is a different, lower-confidence result than a real misconfiguration.""" subnet_id = _subnet_id("vnet1", "subnet1") nic = make_resource( ip_configurations=[ @@ -167,6 +170,38 @@ def test_cmp_001_unresolvable_subnet_still_flags(mock_azure, subscription_id): # No set_subnet() call -> get_subnet() returns None, simulating an unreadable subnet. findings = az_cmp_001.scan(mock_azure, subscription_id) assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + + +def test_cmp_001_mixed_resolvable_and_unresolvable_subnets_is_indeterminate(mock_azure, subscription_id): + """When one IP config's subnet resolves with no NSG but another IP config's subnet can't + be read at all, the unresolved one might have had an NSG — so the overall result must stay + indeterminate rather than being reported as a confirmed HIGH violation.""" + resolvable_subnet_id = _subnet_id("vnet1", "subnet-resolvable") + unresolvable_subnet_id = _subnet_id("vnet1", "subnet-unresolvable") + nic = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=resolvable_subnet_id)), + make_resource(public_ip_address=make_resource(id="pip2"), subnet=make_resource(id=unresolvable_subnet_id)), + ], + network_security_group=None, + ) + subnet = make_resource(id=resolvable_subnet_id, network_security_group=None) + vm = make_resource( + id=_vm_id("vm-mixed-subnets"), + name="vm-mixed-subnets", + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id("nic1"))]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic1", nic) + mock_azure.set_subnet(resolvable_subnet_id, subnet) + # unresolvable_subnet_id is intentionally never registered via set_subnet(). + findings = az_cmp_001.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["determination"] == "indeterminate" # ── AZ-CMP-002: disk using platform-managed encryption only ───────────────── From cfe110ab0533c3d151bc3b406ea381eb080a3ff2 Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Sun, 16 Aug 2026 14:34:24 +0100 Subject: [PATCH 3/9] feat(scanner): add Defender health and real patch-assessment signals (AZ-CMP-003/004) Completes the remaining scope of #268: both rules previously relied on weak signals that could produce false negatives. AZ-CMP-003 only checked whether a named endpoint-protection extension was installed, with no health check - an installed-but-broken AV/EDR agent still read as protected. It now queries Microsoft Defender for Cloud's "Endpoint protection" security assessment as the primary signal (real agent health telemetry) and falls back to the existing extension-name/provisioning-state check only when Defender data is unavailable (not onboarded, no assessment yet, API failure). AZ-CMP-004 only read config flags (enable_automatic_updates / patch_mode), never actual patch compliance - a VM could look compliant by config while being months behind on real patches. It now also fetches the VM's live patch assessment (Azure Update Manager / Microsoft.Maintenance data, via the instance view's patch_status) and flags a confirmed violation when a completed assessment shows pending critical/security patches, even if config looks fine. A clean or unavailable assessment never suppresses an existing config-based finding, since disabled auto-patching is itself a real drift risk regardless of today's point-in-time patch level. Adds AzureClient.get_security_assessments() (Microsoft Defender for Cloud, new azure-mgmt-security dependency) and AzureClient.get_vm_patch_status() (azure-mgmt-compute instance view, no new dependency), both fail-closed (None) on API/permission errors so callers never mistake "signal unavailable" for compliant. Signed-off-by: PARTH J ROHIT --- requirements.txt | 1 + scanner/azure_client.py | 63 +++++++++ scanner/rules/az_cmp_003.py | 94 +++++++++++- scanner/rules/az_cmp_004.py | 62 ++++++++ tests/helpers/mock_azure.py | 24 ++++ tests/test_azure_client_management.py | 56 ++++++++ tests/test_rules_compute.py | 196 ++++++++++++++++++++++++++ 7 files changed, 494 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 425acc9b..7dd74051 100644 --- a/requirements.txt +++ b/requirements.txt @@ -33,6 +33,7 @@ azure-mgmt-postgresqlflexibleservers==1.0.0b1 azure-keyvault-certificates==4.8.0 azure-keyvault-keys==4.9.0 azure-mgmt-containerregistry==15.0.0 +azure-mgmt-security==7.0.0 azure-devops==7.1.0b4 prometheus-client>=0.19.0 python-json-logger>=2.0.7 diff --git a/scanner/azure_client.py b/scanner/azure_client.py index ca6dfc87..4bbda0de 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -59,6 +59,7 @@ def __init__(self, subscription_id: str, credential: Optional[Any] = None) -> No self._subscription_role_assignments_cache: Any = _UNSET self._container_registries_cache: Any = _UNSET self._disks_cache: Dict[str, Any] = {} + self._security_assessments_cache: Any = _UNSET self.devops_client = self._build_devops_client() def _build_devops_client(self) -> Optional[Any]: @@ -651,6 +652,32 @@ 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_vm_patch_status(self, resource_group: str, vm_name: str) -> Optional[Any]: + """Fetch a VM's live patch assessment from its runtime instance view. + + This surfaces Azure's actual patch-assessment evidence (populated by + Azure Update Manager / Microsoft.Maintenance whenever patch + orchestration has run) rather than the VM's config-only patch_mode + setting - a VM can be configured for automatic patching yet still be + months behind if the platform hasn't actually applied anything. + + Returns: + The nested ``AvailablePatchSummary`` (instance_view.patch_status. + available_patch_summary), or ``None`` when the instance view + could not be fetched, no ``patch_status`` is present, or no + assessment has ever run for this VM. Callers must never + interpret ``None`` as "confirmed no missing patches" - only as + "no real assessment evidence available". + """ + try: + client = ComputeManagementClient(self.credential, self.subscription_id) + instance_view = client.virtual_machines.instance_view(resource_group, vm_name) + patch_status = getattr(instance_view, "patch_status", None) + return getattr(patch_status, "available_patch_summary", None) + except Exception as exc: + logger.error("get_vm_patch_status(%s/%s) failed: %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. @@ -688,6 +715,42 @@ def get_disk(self, disk_id: str) -> Optional[Any]: self._disks_cache[disk_id] = disk return disk + # ------------------------------------------------------------------ # + # Microsoft Defender for Cloud # + # ------------------------------------------------------------------ # + + def get_security_assessments(self) -> Optional[List[Any]]: + """List Microsoft Defender for Cloud security assessments for the subscription. + + Cached for the lifetime of this client because every rule that + consults Defender health (e.g. AZ-CMP-003) re-evaluates the same + subscription-wide collection rather than querying per resource - + the assessments API only supports subscription/management-group + scope, not a single resource ID. + + Returns: + A list (including an empty list, e.g. Defender for Cloud was + never onboarded) when Azure responds successfully, or ``None`` + when permissions (Microsoft.Security/assessments/read) or an + API failure prevent the collection from being evaluated. + Callers must never interpret ``None`` as "no assessment exists" + - only as "Defender health signal unavailable". + """ + if self._security_assessments_cache is not _UNSET: + return self._security_assessments_cache + + try: + from azure.mgmt.security import SecurityCenter + + client = SecurityCenter(self.credential, self.subscription_id) + scope = f"/subscriptions/{self.subscription_id}" + self._security_assessments_cache = list(client.assessments.list(scope)) + except Exception as exc: + logger.error("get_security_assessments failed: %s", exc) + self._security_assessments_cache = None + + return self._security_assessments_cache + # ------------------------------------------------------------------ # # Databases # # ------------------------------------------------------------------ # diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index a5dbeb81..96b3cff1 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -1,7 +1,7 @@ """AZ-CMP-003: VM without endpoint protection installed.""" import logging -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional RULE_ID = "AZ-CMP-003" RULE_NAME = "VM Without Endpoint Protection Installed" @@ -21,6 +21,22 @@ REMEDIATION = "Install IaaSAntimalware or onboard to MDE (MDE.Windows / MDE.Linux) depending on the OS." PLAYBOOK = "playbooks/cli/fix_az_cmp_003.sh" +# Microsoft Defender for Cloud's "Endpoint protection" assessment reports real +# runtime health of the installed agent - a stronger signal than checking +# whether a named extension is merely present. When Defender confirms +# Unhealthy, that is a confirmed violation even if a recognised extension is +# installed (the earlier, weaker signal this rule used to rely on alone). +DEFENDER_UNHEALTHY_DESCRIPTION = ( + "Microsoft Defender for Cloud reports the 'Endpoint protection' security assessment for " + "this VM as Unhealthy. This is real agent health telemetry from Defender for Cloud, not " + "just extension-name presence, so it overrides an otherwise-installed recognised extension." +) +DEFENDER_UNHEALTHY_REMEDIATION = ( + "Open Defender for Cloud > Recommendations > 'Endpoint protection should be installed on " + "your machines', review why the agent is reporting unhealthy on this VM, and remediate " + "(reinstall/repair the AV/EDR agent, resolve conflicting security products)." +) + # A recognised EP extension whose provisioning_state is present and is not # "Succeeded" is not actually protecting the VM - name presence alone was # the previous (weak) signal. This is surfaced as an indeterminate result, @@ -47,16 +63,88 @@ logger = logging.getLogger(__name__) +def _defender_endpoint_protection_status(azure_client: Any, vm_id: str) -> Optional[bool]: + """Look up the Defender for Cloud 'Endpoint protection' assessment for a VM. + + The assessments API only supports subscription/management-group scope, + so this filters the subscription-wide collection down to the one + matching this VM's resource ID. + + Returns: + True - assessment found, status code "Healthy" (confirmed protected). + False - assessment found, status code "Unhealthy" (confirmed unprotected). + None - assessments could not be listed, no assessment matched this + VM, or its status code was "NotApplicable"/unrecognised. + Callers must treat this as "Defender signal unavailable" and + fall back to the extension-based check, never as compliant. + """ + assessments = azure_client.get_security_assessments() + if not assessments: + return None + + vm_id_lower = (vm_id or "").lower() + for assessment in assessments: + resource_details = getattr(assessment, "resource_details", None) + resource_id = (getattr(resource_details, "id", "") or "").lower() + if resource_id != vm_id_lower: + continue + + display_name = (getattr(assessment, "display_name", "") or "").lower() + if "endpoint protection" not in display_name: + continue + + status = getattr(assessment, "status", None) + code = (getattr(status, "code", "") or "").lower() + if code == "healthy": + return True + if code == "unhealthy": + return False + return None # NotApplicable or an unrecognised status code + + return None + + def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: findings: List[Dict[str, Any]] = [] for vm in azure_client.get_virtual_machines(): - parsed = azure_client.parse_resource_id(getattr(vm, "id", "")) + vm_id = getattr(vm, "id", "") + parsed = azure_client.parse_resource_id(vm_id) rg = parsed.get("resource_group", "") vm_name = parsed.get("name", "") if not rg or not vm_name: continue + defender_status = _defender_endpoint_protection_status(azure_client, vm_id) + + if defender_status is True: + continue # Defender confirms protected - the strongest available signal. + + if defender_status is False: + findings.append( + { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": vm.id, + "resource_name": vm_name, + "resource_type": "Microsoft.Compute/virtualMachines", + "description": DEFENDER_UNHEALTHY_DESCRIPTION, + "remediation": DEFENDER_UNHEALTHY_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": rg, + "signal": "defender_assessment", + "determination": "non_compliant", + }, + } + ) + continue + + # Defender signal unavailable (not onboarded, no assessment yet, API + # failure) - fall back to the extension-name/provisioning-state check. exts = azure_client.get_vm_extensions(rg, vm_name) if exts is None: continue @@ -90,6 +178,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "metadata": { "resource_group": rg, "installed_extensions": sorted(installed.keys()), + "signal": "extension_fallback", "determination": "non_compliant", }, } @@ -131,6 +220,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "resource_group": rg, "installed_extensions": sorted(installed.keys()), "unhealthy_extensions": sorted(unhealthy_names), + "signal": "extension_fallback", "determination": "indeterminate", }, } diff --git a/scanner/rules/az_cmp_004.py b/scanner/rules/az_cmp_004.py index caf62cc0..c23b33c3 100644 --- a/scanner/rules/az_cmp_004.py +++ b/scanner/rules/az_cmp_004.py @@ -25,6 +25,31 @@ ) PLAYBOOK = "playbooks/cli/fix_az_cmp_004.sh" +# A VM can look compliant by config (auto-updates/AutomaticByPlatform set) +# while still being months behind on real patches, if the platform simply +# hasn't applied anything yet. Real assessment evidence (Azure Update +# Manager / Microsoft.Maintenance, surfaced through the VM's instance view) +# can override a config-only pass into a confirmed finding. It never +# suppresses a config-based finding: config with auto-patching disabled is +# itself an unmanaged-drift risk regardless of today's patch snapshot, so +# the config-flag check always remains the fallback/baseline signal. +ASSESSMENT_OVERRIDE_DESCRIPTION = ( + "VM is configured for automatic OS patching, but its latest Azure Update Manager patch " + "assessment shows critical or security patches are still pending installation. Config " + "alone does not prove patches have actually been applied - this is real assessment " + "evidence that the VM is currently unpatched." +) +ASSESSMENT_OVERRIDE_REMEDIATION = ( + "Trigger an on-demand patch installation (Update Manager > Install now) or review why " + "the scheduled automatic patching run has not applied the pending critical/security " + "patches, then re-run the scan to confirm the assessment clears." +) + +# An assessment run whose status confirms it actually completed and produced +# real counts. Anything else (in progress, failed, unknown) is not reliable +# enough evidence to override a config-based pass. +_CONCLUSIVE_ASSESSMENT_STATUSES = {"succeeded", "completedwithwarnings"} + logger = logging.getLogger(__name__) @@ -75,6 +100,43 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "frameworks": FRAMEWORKS, "metadata": { "resource_group": rg, + "signal": "config_flags", + "determination": "non_compliant", + }, + } + ) + continue + + # Config says patching is enabled - check real assessment evidence + # for the false-negative case: config correct, platform hasn't + # actually applied the pending critical/security patches. + patch_summary = azure_client.get_vm_patch_status(rg, vm_name) + if patch_summary is None: + continue + + status = (getattr(patch_summary, "status", "") or "").lower() + critical_count = getattr(patch_summary, "critical_and_security_patch_count", None) + if status in _CONCLUSIVE_ASSESSMENT_STATUSES and critical_count is not None and critical_count > 0: + findings.append( + { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": vm.id, + "resource_name": vm_name, + "resource_type": "Microsoft.Compute/virtualMachines", + "description": ASSESSMENT_OVERRIDE_DESCRIPTION, + "remediation": ASSESSMENT_OVERRIDE_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": rg, + "signal": "patch_assessment_override", + "determination": "non_compliant", + "critical_and_security_patch_count": critical_count, + "other_patch_count": getattr(patch_summary, "other_patch_count", None), + "assessment_status": status, }, } ) diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 0403db17..50decbe7 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -93,6 +93,8 @@ def __init__(self) -> None: self._container_registries: Optional[List[Any]] = [] self._blob_containers: Dict[Tuple[str, str], Optional[List[Any]]] = {} self._blob_service_properties: Dict[Tuple[str, str], Optional[Any]] = {} + self._security_assessments: Optional[List[Any]] = None + self._vm_patch_status: Dict[Tuple[str, str], Optional[Any]] = {} # None by default, matching AzureClient.devops_client's "not configured" state. self.devops_client: Optional[Any] = None # Some rules read azure_client.subscription_id when constructing an @@ -266,6 +268,28 @@ def set_disk(self, disk_id: str, disk: Optional[Any]) -> "MockAzureClient": def get_disk(self, disk_id: str) -> Optional[Any]: return self._disks.get(disk_id) + def set_vm_patch_status(self, resource_group: str, vm_name: str, summary: Optional[Any]) -> "MockAzureClient": + """Configure the AvailablePatchSummary returned for a VM; ``None`` means no real + assessment evidence is available.""" + self._vm_patch_status[(resource_group, vm_name)] = summary + return self + + def get_vm_patch_status(self, resource_group: str, vm_name: str) -> Optional[Any]: + return self._vm_patch_status.get((resource_group, vm_name)) + + # ------------------------------------------------------------------ # + # Microsoft Defender for Cloud # + # ------------------------------------------------------------------ # + + def set_security_assessments(self, assessments: Optional[List[Any]]) -> "MockAzureClient": + """Configure Defender for Cloud assessments; ``None`` represents an API failure + or a subscription that was never onboarded to Defender for Cloud.""" + self._security_assessments = assessments + return self + + def get_security_assessments(self) -> Optional[List[Any]]: + return self._security_assessments + # ------------------------------------------------------------------ # # Storage — lifecycle & service logging (three-state: True/False/None) # # ------------------------------------------------------------------ # diff --git a/tests/test_azure_client_management.py b/tests/test_azure_client_management.py index 8df2a0ca..9c164c3e 100644 --- a/tests/test_azure_client_management.py +++ b/tests/test_azure_client_management.py @@ -253,6 +253,62 @@ def get_configuration(resource_group, name): assert names == {"good"} +def test_get_vm_patch_status_returns_summary_and_fails_closed(client): + with patch("scanner.azure_client.ComputeManagementClient") as constructor: + summary = SimpleNamespace(status="Succeeded", critical_and_security_patch_count=2) + constructor.return_value.virtual_machines.instance_view.return_value = SimpleNamespace( + patch_status=SimpleNamespace(available_patch_summary=summary) + ) + assert client.get_vm_patch_status("rg", "vm") is summary + + constructor.return_value.virtual_machines.instance_view.side_effect = RuntimeError("denied") + assert client.get_vm_patch_status("rg", "vm") is None + + +def test_get_vm_patch_status_missing_patch_status_returns_none(client): + with patch("scanner.azure_client.ComputeManagementClient") as constructor: + constructor.return_value.virtual_machines.instance_view.return_value = SimpleNamespace(patch_status=None) + assert client.get_vm_patch_status("rg", "vm") is None + + +def test_get_security_assessments_returns_results_and_caches(client): + with patch("azure.mgmt.security.SecurityCenter") as constructor: + constructor.return_value.assessments.list.return_value = [SimpleNamespace(display_name="a")] + result = client.get_security_assessments() + assert result is not None + assert [a.display_name for a in result] == ["a"] + + # cached: second call must not hit the SDK again + constructor.return_value.assessments.list.side_effect = RuntimeError("should not be called") + assert [a.display_name for a in client.get_security_assessments()] == ["a"] + + +def test_get_security_assessments_failure_returns_none(client): + with patch("azure.mgmt.security.SecurityCenter") as constructor: + constructor.return_value.assessments.list.side_effect = RuntimeError("denied") + assert client.get_security_assessments() is None + + +def test_patch_summary_and_assessment_real_sdk_models_have_expected_fields(): + """SDK-shape guard: az_cmp_004/az_cmp_003 read attributes (status, + critical_and_security_patch_count, resource_details.id, display_name, + status.code) off real SDK models. This fails loudly if a future SDK bump + ever renames or drops one of them, instead of the rule silently treating + every VM as having no real evidence (the exact class of bug fixed for + AZ-CMP-002's ManagedDiskParameters).""" + from azure.mgmt.compute.models import AvailablePatchSummary + from azure.mgmt.security.v2021_06_01.models import AzureResourceDetails, SecurityAssessmentResponse + + assert "critical_and_security_patch_count" in AvailablePatchSummary._attribute_map + assert "status" in AvailablePatchSummary._attribute_map + assert "other_patch_count" in AvailablePatchSummary._attribute_map + + assert "resource_details" in SecurityAssessmentResponse._attribute_map + assert "display_name" in SecurityAssessmentResponse._attribute_map + assert "status" in SecurityAssessmentResponse._attribute_map + assert "id" in AzureResourceDetails._attribute_map + + def test_get_container_registries_returns_results_and_caches(client): with patch("azure.mgmt.containerregistry.ContainerRegistryManagementClient") as constructor: constructor.return_value.registries.list.return_value = [SimpleNamespace(name="acr1")] diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index f347694e..63089d12 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -500,6 +500,88 @@ def test_cmp_003_missing_provisioning_state_falls_back_to_name_based_pass(mock_a assert az_cmp_003.scan(mock_azure, subscription_id) == [] +# ── AZ-CMP-003: Defender for Cloud endpoint-protection assessment ─────────── + + +def _assessment( + resource_id, display_name="Endpoint protection should be installed on virtual machines", status_code="Healthy" +): + """A SecurityAssessmentResponse-shaped stub, as returned by AzureClient.get_security_assessments().""" + return make_resource( + resource_details=make_resource(id=resource_id), + display_name=display_name, + status=make_resource(code=status_code), + ) + + +def test_cmp_003_defender_healthy_overrides_missing_extension_returns_no_findings(mock_azure, subscription_id): + """Defender for Cloud confirming Healthy is authoritative, even with no matching extension + installed - it is real agent telemetry, stronger than extension-name presence.""" + vm_id = _vm_id("vm-defender-healthy") + vm = make_resource(id=vm_id, name="vm-defender-healthy") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-defender-healthy", []) + mock_azure.set_security_assessments([_assessment(vm_id, status_code="Healthy")]) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + +def test_cmp_003_defender_unhealthy_returns_confirmed_high_finding_even_with_extension(mock_azure, subscription_id): + """Defender reporting Unhealthy is a confirmed violation, overriding a merely-installed, + successfully-provisioned extension - name presence never proved effective protection.""" + vm_id = _vm_id("vm-defender-unhealthy") + vm = make_resource(id=vm_id, name="vm-defender-unhealthy") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions( + _RG, + "vm-defender-unhealthy", + [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")], + ) + mock_azure.set_security_assessments([_assessment(vm_id, status_code="Unhealthy")]) + findings = az_cmp_003.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["severity"] == "HIGH" + assert f["metadata"]["signal"] == "defender_assessment" + assert f["metadata"]["determination"] == "non_compliant" + + +def test_cmp_003_defender_not_applicable_falls_back_to_extension_check(mock_azure, subscription_id): + """A NotApplicable Defender status carries no usable signal - the extension-based check + still governs the result.""" + vm_id = _vm_id("vm-defender-na") + vm = make_resource(id=vm_id, name="vm-defender-na") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-defender-na", [make_resource(type_properties_type="IaaSAntimalware")]) + mock_azure.set_security_assessments([_assessment(vm_id, status_code="NotApplicable")]) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + +def test_cmp_003_defender_unavailable_falls_back_to_extension_check_with_signal_metadata(mock_azure, subscription_id): + """No Defender data at all (subscription never onboarded, or the assessments API failed) - + the fallback path is explicitly tagged in metadata so callers can see which signal fired.""" + vm = make_resource(id=_vm_id("vm-no-defender"), name="vm-no-defender") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-no-defender", [make_resource(type_properties_type="CustomScript")]) + # set_security_assessments not called -> None, matching an unonboarded subscription. + findings = az_cmp_003.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["signal"] == "extension_fallback" + assert findings[0]["metadata"]["determination"] == "non_compliant" + + +def test_cmp_003_defender_assessment_for_different_resource_is_ignored(mock_azure, subscription_id): + """An assessment for a different resource ID must not be mistaken for this VM's signal - + the subscription-wide assessments list has to be filtered by resource_details.id.""" + vm_id = _vm_id("vm-target") + vm = make_resource(id=vm_id, name="vm-target") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-target", [make_resource(type_properties_type="IaaSAntimalware")]) + mock_azure.set_security_assessments([_assessment(_vm_id("vm-other"), status_code="Unhealthy")]) + # Falls back to the extension check (which passes) rather than picking up the other VM's Unhealthy status. + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + # ── AZ-CMP-004: VM without automatic OS patching ──────────────────────────── @@ -535,3 +617,117 @@ def test_cmp_004_noncompliant_no_patching_returns_one_finding(mock_azure, subscr assert f["rule_id"] == "AZ-CMP-004" assert f["severity"] == "HIGH" assert f["resource_name"] == "vm-stale" + assert f["metadata"]["signal"] == "config_flags" + assert f["metadata"]["determination"] == "non_compliant" + + +# ── AZ-CMP-004: real patch-assessment evidence override ───────────────────── + + +def _patch_summary(status="Succeeded", critical_and_security_patch_count=0, other_patch_count=0): + """An AvailablePatchSummary-shaped stub, as returned by AzureClient.get_vm_patch_status().""" + return make_resource( + status=status, + critical_and_security_patch_count=critical_and_security_patch_count, + other_patch_count=other_patch_count, + ) + + +def test_cmp_004_config_ok_but_assessment_shows_pending_critical_patches_returns_finding(mock_azure, subscription_id): + """Config says auto-patching is on, but the real Update Manager assessment shows pending + critical/security patches - config alone never proved patches were actually applied.""" + vm = make_resource( + id=_vm_id("vm-config-ok-but-behind"), + name="vm-config-ok-but-behind", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, "vm-config-ok-but-behind", _patch_summary(status="Succeeded", critical_and_security_patch_count=3) + ) + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["severity"] == "HIGH" + assert f["metadata"]["signal"] == "patch_assessment_override" + assert f["metadata"]["determination"] == "non_compliant" + assert f["metadata"]["critical_and_security_patch_count"] == 3 + + +def test_cmp_004_config_ok_and_assessment_clean_returns_no_findings(mock_azure, subscription_id): + """Config OK and a completed assessment showing zero pending critical/security patches + is a genuinely compliant VM.""" + vm = make_resource( + id=_vm_id("vm-config-ok-and-clean"), + name="vm-config-ok-and-clean", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, "vm-config-ok-and-clean", _patch_summary(status="Succeeded", critical_and_security_patch_count=0) + ) + assert az_cmp_004.scan(mock_azure, subscription_id) == [] + + +def test_cmp_004_config_ok_but_assessment_in_progress_does_not_override(mock_azure, subscription_id): + """An assessment that hasn't conclusively finished is not reliable evidence - it must not + override a config-based pass even if it happens to carry a nonzero patch count so far.""" + vm = make_resource( + id=_vm_id("vm-assessment-in-progress"), + name="vm-assessment-in-progress", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, "vm-assessment-in-progress", _patch_summary(status="InProgress", critical_and_security_patch_count=5) + ) + assert az_cmp_004.scan(mock_azure, subscription_id) == [] + + +def test_cmp_004_config_ok_and_no_assessment_data_returns_no_findings(mock_azure, subscription_id): + """No real assessment evidence available (never run, API failure) - behaves exactly like + the pre-existing config-only check with nothing to override.""" + vm = make_resource( + id=_vm_id("vm-no-assessment-data"), + name="vm-no-assessment-data", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + # set_vm_patch_status not called -> None, matching "no assessment has ever run". + assert az_cmp_004.scan(mock_azure, subscription_id) == [] + + +def test_cmp_004_config_disabled_finding_unaffected_by_clean_assessment(mock_azure, subscription_id): + """Config-disabled auto-patching is itself an unmanaged-drift risk - a clean point-in-time + assessment must not suppress the config-based finding.""" + vm = make_resource( + id=_vm_id("vm-config-disabled-but-currently-clean"), + name="vm-config-disabled-but-currently-clean", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=False, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, + "vm-config-disabled-but-currently-clean", + _patch_summary(status="Succeeded", critical_and_security_patch_count=0), + ) + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["signal"] == "config_flags" + assert findings[0]["metadata"]["determination"] == "non_compliant" From a56a076e9403bebac5f6b6dfa4960903e2354b9c Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Tue, 18 Aug 2026 11:54:59 +0100 Subject: [PATCH 4/9] fix(scanner): resolve compute rule review findings Signed-off-by: PARTH J ROHIT --- scanner/azure_client.py | 18 +++++++--- scanner/rules/az_cmp_001.py | 3 +- scanner/rules/az_cmp_003.py | 17 ++++----- tests/test_azure_client_management.py | 8 ++++- tests/test_rules_compute.py | 50 ++++++++++++++++++++++++++- 5 files changed, 81 insertions(+), 15 deletions(-) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index 4bbda0de..583b5159 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -59,6 +59,7 @@ def __init__(self, subscription_id: str, credential: Optional[Any] = None) -> No self._subscription_role_assignments_cache: Any = _UNSET self._container_registries_cache: Any = _UNSET self._disks_cache: Dict[str, Any] = {} + self._subnets_cache: Dict[str, Any] = {} self._security_assessments_cache: Any = _UNSET self.devops_client = self._build_devops_client() @@ -296,6 +297,9 @@ def get_subnet(self, subnet_id: str) -> Optional[Any]: carries its own NSG must be fetched separately via the Network management client. + The result is cached for the lifetime of this client because the same + subnet can back IP configurations on many NICs in a scan. + Returns: The Subnet resource, or ``None`` when the ID is missing/malformed or Azure cannot return it (permissions, deletion, SDK error). @@ -303,6 +307,11 @@ def get_subnet(self, subnet_id: str) -> Optional[Any]: """ if not subnet_id: return None + + if subnet_id in self._subnets_cache: + return self._subnets_cache[subnet_id] + + subnet = None try: resource_group = "" vnet_name = "" @@ -319,13 +328,14 @@ def get_subnet(self, subnet_id: str) -> Optional[Any]: if not (resource_group and vnet_name and subnet_name): logger.error("get_subnet failed: could not parse %s", subnet_id) - return None + else: + client = NetworkManagementClient(self.credential, self.subscription_id) + subnet = client.subnets.get(resource_group, vnet_name, subnet_name) - client = NetworkManagementClient(self.credential, self.subscription_id) - return client.subnets.get(resource_group, vnet_name, subnet_name) except Exception as exc: logger.error("get_subnet failed for %s: %s", subnet_id, exc) - return None + self._subnets_cache[subnet_id] = subnet + return subnet def get_virtual_networks(self) -> List[Any]: """List all virtual networks in the subscription.""" diff --git a/scanner/rules/az_cmp_001.py b/scanner/rules/az_cmp_001.py index 73e8a7fe..cf7912aa 100644 --- a/scanner/rules/az_cmp_001.py +++ b/scanner/rules/az_cmp_001.py @@ -63,6 +63,7 @@ def _subnet_nsg_status(azure_client: Any, nic: Any) -> Optional[bool]: subnet_ref = getattr(ip_cfg, "subnet", None) subnet_id = getattr(subnet_ref, "id", None) if not subnet_id: + unresolved = True continue subnet = azure_client.get_subnet(subnet_id) @@ -128,7 +129,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "nic_id": nic_id, "nic_name": nic_name, "nic_nsg_attached": has_nic_nsg, - "subnet_nsg_attached": False, + "subnet_nsg_attached": False if confirmed else None, "determination": "non_compliant" if confirmed else "indeterminate", }, } diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index 96b3cff1..768086cc 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -1,7 +1,7 @@ """AZ-CMP-003: VM without endpoint protection installed.""" import logging -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple RULE_ID = "AZ-CMP-003" RULE_NAME = "VM Without Endpoint Protection Installed" @@ -149,7 +149,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: if exts is None: continue - installed: Dict[str, Any] = {} + installed: List[Tuple[str, Any]] = [] for e in exts: t = ( getattr(e, "type_properties_type", None) @@ -157,9 +157,10 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: or getattr(e, "type", "") ) if t: - installed[t.lower()] = e + installed.append((t.lower(), e)) - matched = {name: ext for name, ext in installed.items() if name in KNOWN_EP_EXTENSIONS} + installed_names = sorted({name for name, _ in installed}) + matched = [(name, ext) for name, ext in installed if name in KNOWN_EP_EXTENSIONS] if not matched: findings.append( @@ -177,7 +178,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "frameworks": FRAMEWORKS, "metadata": { "resource_group": rg, - "installed_extensions": sorted(installed.keys()), + "installed_extensions": installed_names, "signal": "extension_fallback", "determination": "non_compliant", }, @@ -193,7 +194,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: # indeterminate rather than silently passed. unhealthy_names = [] confirmed_healthy = False - for name, ext in matched.items(): + for name, ext in matched: provisioning_state = (getattr(ext, "provisioning_state", None) or "").lower() if not provisioning_state or provisioning_state == "succeeded": confirmed_healthy = True @@ -218,8 +219,8 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "frameworks": FRAMEWORKS, "metadata": { "resource_group": rg, - "installed_extensions": sorted(installed.keys()), - "unhealthy_extensions": sorted(unhealthy_names), + "installed_extensions": installed_names, + "unhealthy_extensions": sorted(set(unhealthy_names)), "signal": "extension_fallback", "determination": "indeterminate", }, diff --git a/tests/test_azure_client_management.py b/tests/test_azure_client_management.py index 9c164c3e..de7118b9 100644 --- a/tests/test_azure_client_management.py +++ b/tests/test_azure_client_management.py @@ -65,9 +65,15 @@ def test_single_resource_and_policy_wrappers(client): result = client.get_subnet(subnet_id) assert result.name == "subnet1" constructor.return_value.subnets.get.assert_called_once_with("RG", "vnet1", "subnet1") + assert client.get_subnet(subnet_id) is result + constructor.return_value.subnets.get.assert_called_once_with("RG", "vnet1", "subnet1") + failed_client = AzureClient("sub-1", credential=MagicMock()) + constructor.return_value.subnets.get.reset_mock() constructor.return_value.subnets.get.side_effect = RuntimeError("denied") - assert client.get_subnet(subnet_id) is None + assert failed_client.get_subnet(subnet_id) is None + assert failed_client.get_subnet(subnet_id) is None + constructor.return_value.subnets.get.assert_called_once_with("RG", "vnet1", "subnet1") assert client.get_subnet("") is None assert client.get_subnet("/not/a/valid/subnet/id") is None diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 63089d12..e6d9ec65 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -79,8 +79,11 @@ def test_cmp_001_compliant_nic_with_nsg_returns_no_findings(mock_azure, subscrip def test_cmp_001_noncompliant_public_ip_no_nsg_returns_one_finding(mock_azure, subscription_id): """A NIC with a public IP and no NSG must produce exactly one finding.""" + subnet_id = _subnet_id("vnet1", "subnet1") nic = make_resource( - ip_configurations=[make_resource(public_ip_address=make_resource(id="pip1"))], + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=subnet_id)) + ], network_security_group=None, ) vm = make_resource( @@ -90,6 +93,7 @@ def test_cmp_001_noncompliant_public_ip_no_nsg_returns_one_finding(mock_azure, s ) mock_azure.set_virtual_machines([vm]) mock_azure.set_network_interface(_RG, "nic1", nic) + mock_azure.set_subnet(subnet_id, make_resource(id=subnet_id, network_security_group=None)) findings = az_cmp_001.scan(mock_azure, subscription_id) assert len(findings) == 1 f = findings[0] @@ -174,6 +178,31 @@ def test_cmp_001_unresolvable_subnet_is_indeterminate_not_confirmed_high(mock_az assert _REQUIRED_FIELDS.issubset(f.keys()) assert f["severity"] == "LOW" assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["subnet_nsg_attached"] is None + + +def test_cmp_001_missing_subnet_id_is_indeterminate_not_confirmed_high(mock_azure, subscription_id): + """A subnet reference with no ID cannot confirm that subnet protection is absent.""" + nic = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id="")) + ], + network_security_group=None, + ) + vm = make_resource( + id=_vm_id("vm-missing-subnet-id"), + name="vm-missing-subnet-id", + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id("nic1"))]), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic1", nic) + + findings = az_cmp_001.scan(mock_azure, subscription_id) + + assert len(findings) == 1 + assert findings[0]["severity"] == "LOW" + assert findings[0]["metadata"]["determination"] == "indeterminate" + assert findings[0]["metadata"]["subnet_nsg_attached"] is None def test_cmp_001_mixed_resolvable_and_unresolvable_subnets_is_indeterminate(mock_azure, subscription_id): @@ -488,6 +517,25 @@ def test_cmp_003_extension_present_but_unhealthy_returns_indeterminate_finding(m assert f["metadata"]["unhealthy_extensions"] == ["iaasantimalware"] +def test_cmp_003_duplicate_extension_types_use_healthy_record_regardless_of_order(mock_azure, subscription_id): + """Duplicate API records must not make the result depend on dict overwrite order.""" + vm = make_resource(id=_vm_id("vm-duplicate-extensions"), name="vm-duplicate-extensions") + mock_azure.set_virtual_machines([vm]) + + for extensions in ( + [ + make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded"), + make_resource(type_properties_type="iaasantimalware", provisioning_state="Failed"), + ], + [ + make_resource(type_properties_type="iaasantimalware", provisioning_state="Failed"), + make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded"), + ], + ): + mock_azure.set_vm_extensions(_RG, "vm-duplicate-extensions", extensions) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + def test_cmp_003_missing_provisioning_state_falls_back_to_name_based_pass(mock_azure, subscription_id): """When provisioning_state isn't exposed by the API, name presence alone remains sufficient.""" vm = make_resource(id=_vm_id("vm-no-state-data"), name="vm-no-state-data") From 66c0ddf5b60c4634017488c0f70cff7bb8b9bec9 Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Tue, 18 Aug 2026 12:13:59 +0100 Subject: [PATCH 5/9] style(tests): format compute rule tests Signed-off-by: PARTH J ROHIT --- tests/test_rules_compute.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index e6d9ec65..52eae4bf 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -184,9 +184,7 @@ def test_cmp_001_unresolvable_subnet_is_indeterminate_not_confirmed_high(mock_az def test_cmp_001_missing_subnet_id_is_indeterminate_not_confirmed_high(mock_azure, subscription_id): """A subnet reference with no ID cannot confirm that subnet protection is absent.""" nic = make_resource( - ip_configurations=[ - make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id="")) - ], + ip_configurations=[make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=""))], network_security_group=None, ) vm = make_resource( From 80241b5405092dd66ea4b414bc5c838860feca52 Mon Sep 17 00:00:00 2001 From: PARTH J ROHIT Date: Fri, 21 Aug 2026 12:24:06 +0100 Subject: [PATCH 6/9] perf(scanner): index endpoint-protection assessments once per scan (AZ-CMP-003) _defender_endpoint_protection_status previously rescanned the full subscription-wide assessments list for every VM. Build a resource-ID index once per scan() call instead, and make the match deterministic: when a resource has more than one "endpoint protection" assessment, an Unhealthy code always wins regardless of API response order. Signed-off-by: PARTH J ROHIT --- scanner/rules/az_cmp_003.py | 77 ++++++++++++++++++++++--------------- tests/test_rules_compute.py | 41 ++++++++++++++++++++ 2 files changed, 87 insertions(+), 31 deletions(-) diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index 768086cc..b5392ff3 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -63,49 +63,64 @@ logger = logging.getLogger(__name__) -def _defender_endpoint_protection_status(azure_client: Any, vm_id: str) -> Optional[bool]: - """Look up the Defender for Cloud 'Endpoint protection' assessment for a VM. - - The assessments API only supports subscription/management-group scope, - so this filters the subscription-wide collection down to the one - matching this VM's resource ID. +def _index_endpoint_protection_assessments(assessments: Optional[List[Any]]) -> Dict[str, List[Any]]: + """Group the subscription-wide assessments list by resource ID, once per scan. - Returns: - True - assessment found, status code "Healthy" (confirmed protected). - False - assessment found, status code "Unhealthy" (confirmed unprotected). - None - assessments could not be listed, no assessment matched this - VM, or its status code was "NotApplicable"/unrecognised. - Callers must treat this as "Defender signal unavailable" and - fall back to the extension-based check, never as compliant. + The assessments API only supports subscription/management-group scope, so + every VM's lookup would otherwise re-scan the full list. Building this + index up front makes each VM's lookup O(1) instead of O(assessment count). """ - assessments = azure_client.get_security_assessments() - if not assessments: - return None - - vm_id_lower = (vm_id or "").lower() - for assessment in assessments: + index: Dict[str, List[Any]] = {} + for assessment in assessments or []: + display_name = (getattr(assessment, "display_name", "") or "").lower() + if "endpoint protection" not in display_name: + continue resource_details = getattr(assessment, "resource_details", None) resource_id = (getattr(resource_details, "id", "") or "").lower() - if resource_id != vm_id_lower: + if not resource_id: continue + index.setdefault(resource_id, []).append(assessment) + return index - display_name = (getattr(assessment, "display_name", "") or "").lower() - if "endpoint protection" not in display_name: - continue +def _defender_endpoint_protection_status(assessments_by_resource: Dict[str, List[Any]], vm_id: str) -> Optional[bool]: + """Look up the Defender for Cloud 'Endpoint protection' assessment for a VM. + + A resource can have more than one assessment whose display name contains + "endpoint protection" (e.g. an installation check and a separate health + check). Resolving by "first match in the list" would make the result + depend on API response order, so instead every matching assessment for + the resource is considered and an "Unhealthy" code always wins - a real + unhealthy signal must never be masked by iteration order. + + Returns: + True - all matching assessments are "Healthy" (confirmed protected). + False - at least one matching assessment is "Unhealthy" (confirmed + unprotected). + None - no assessment matched this VM, or none had a status code of + "Healthy"/"Unhealthy" (e.g. only "NotApplicable"). Callers + must treat this as "Defender signal unavailable" and fall + back to the extension-based check, never as compliant. + """ + matches = assessments_by_resource.get((vm_id or "").lower()) + if not matches: + return None + + codes = set() + for assessment in matches: status = getattr(assessment, "status", None) - code = (getattr(status, "code", "") or "").lower() - if code == "healthy": - return True - if code == "unhealthy": - return False - return None # NotApplicable or an unrecognised status code + codes.add((getattr(status, "code", "") or "").lower()) - return None + if "unhealthy" in codes: + return False + if "healthy" in codes: + return True + return None # Only NotApplicable or unrecognised status codes. def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: findings: List[Dict[str, Any]] = [] + assessments_by_resource = _index_endpoint_protection_assessments(azure_client.get_security_assessments()) for vm in azure_client.get_virtual_machines(): vm_id = getattr(vm, "id", "") @@ -115,7 +130,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: if not rg or not vm_name: continue - defender_status = _defender_endpoint_protection_status(azure_client, vm_id) + defender_status = _defender_endpoint_protection_status(assessments_by_resource, vm_id) if defender_status is True: continue # Defender confirms protected - the strongest available signal. diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 52eae4bf..cef4cc62 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -628,6 +628,47 @@ def test_cmp_003_defender_assessment_for_different_resource_is_ignored(mock_azur assert az_cmp_003.scan(mock_azure, subscription_id) == [] +def test_cmp_003_defender_unhealthy_wins_over_healthy_regardless_of_assessment_order(mock_azure, subscription_id): + """A resource can have more than one 'endpoint protection' assessment (e.g. an + installation check and a separate health check). The result must not depend on + which one the API happened to list first - an Unhealthy code always wins.""" + vm_id = _vm_id("vm-mixed-assessments") + vm = make_resource(id=vm_id, name="vm-mixed-assessments") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-mixed-assessments", []) + + # Healthy listed before Unhealthy. + mock_azure.set_security_assessments( + [ + _assessment( + vm_id, display_name="Endpoint protection should be installed on virtual machines", status_code="Healthy" + ), + _assessment( + vm_id, display_name="Endpoint protection health issues should be resolved", status_code="Unhealthy" + ), + ] + ) + findings_order_a = az_cmp_003.scan(mock_azure, subscription_id) + + # Same two assessments, reversed order. + mock_azure.set_security_assessments( + [ + _assessment( + vm_id, display_name="Endpoint protection health issues should be resolved", status_code="Unhealthy" + ), + _assessment( + vm_id, display_name="Endpoint protection should be installed on virtual machines", status_code="Healthy" + ), + ] + ) + findings_order_b = az_cmp_003.scan(mock_azure, subscription_id) + + assert findings_order_a == findings_order_b + assert len(findings_order_a) == 1 + assert findings_order_a[0]["metadata"]["signal"] == "defender_assessment" + assert findings_order_a[0]["metadata"]["determination"] == "non_compliant" + + # ── AZ-CMP-004: VM without automatic OS patching ──────────────────────────── From 47403e51136e45bdbd1b72a6c81a5369df782469 Mon Sep 17 00:00:00 2001 From: parthrohit22 Date: Fri, 28 Aug 2026 00:38:49 +0100 Subject: [PATCH 7/9] fix(scanner): resolve four compute-rule correctness gaps from review AZ-CMP-001: an indeterminate result on one NIC caused an immediate break, so a later NIC on the same VM with a confirmed public-IP/no-NSG exposure was never evaluated - a real HIGH could be silently downgraded to LOW just because of NIC iteration order. scan() now keeps the worst evaluated result across all of a VM's NICs, only stopping early once a confirmed violation is found (nothing can outrank it). AZ-CMP-003: - Defender's "Endpoint protection should be installed" recommendation was renamed to "EDR solution should be installed on virtual machines" when Microsoft moved from the deprecated Log Analytics agent to agentless EDR scanning. Matching only "endpoint protection" meant the index silently matched nothing against a current subscription's real data, so Defender's signal was never found and every VM fell back to the weaker extension check. Now matches either display name. - An extension with a missing provisioning_state was treated the same as "Succeeded" (confirmed healthy). Missing state is unknown evidence, not proof of success - it's now folded into the same indeterminate path as Failed/Canceled instead of silently passing. AZ-CMP-004: unavailable, non-conclusive (e.g. InProgress), and stale patch assessments were all treated as a clean pass whenever config-based patching was enabled - config alone was silently treated as sufficient even with no real, current evidence backing it. The assessment check now requires a conclusive status AND a last_modified_time within a 30-day freshness threshold to count as a genuine clean signal; anything short of that (unavailable, non-conclusive, or stale) surfaces as an indeterminate LOW finding instead of a silent pass, mirroring the LOW/indeterminate split AZ-CMP-001/003 already use for their own unresolvable evidence. Added regression coverage for all four: two NIC-ordering cases for AZ-CMP-001, the current EDR display name and missing-provisioning-state cases for AZ-CMP-003, and unavailable/non-conclusive/stale/fresh/missing- timestamp cases for AZ-CMP-004. Signed-off-by: parthrohit22 --- scanner/rules/az_cmp_001.py | 62 ++++++---- scanner/rules/az_cmp_003.py | 36 ++++-- scanner/rules/az_cmp_004.py | 96 +++++++++++++- tests/test_rules_compute.py | 241 +++++++++++++++++++++++++++++++++--- 4 files changed, 378 insertions(+), 57 deletions(-) diff --git a/scanner/rules/az_cmp_001.py b/scanner/rules/az_cmp_001.py index cf7912aa..bd696ac1 100644 --- a/scanner/rules/az_cmp_001.py +++ b/scanner/rules/az_cmp_001.py @@ -86,6 +86,14 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: if not network_profile: continue + # One finding per VM is enough, but an indeterminate result on an + # earlier NIC must never suppress evaluation of a later NIC - a + # confirmed HIGH elsewhere on the same VM must not be downgraded to + # LOW just because it was reached second. Keep the worst evaluated + # result across all of this VM's NICs and only stop early once a + # confirmed violation is found (nothing can outrank it). + vm_finding: Optional[Dict[str, Any]] = None + for nic_ref in getattr(network_profile, "network_interfaces", []) or []: nic_id = getattr(nic_ref, "id", "") if not nic_id: @@ -112,28 +120,36 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: if has_public_ip and not has_nic_nsg and subnet_status is not True: confirmed = subnet_status is False - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY if confirmed else INDETERMINATE_SEVERITY, - "category": CATEGORY, - "resource_id": vm.id, - "resource_name": vm.name, - "resource_type": "Microsoft.Compute/virtualMachines", - "description": DESCRIPTION if confirmed else INDETERMINATE_DESCRIPTION, - "remediation": REMEDIATION if confirmed else INDETERMINATE_REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "nic_id": nic_id, - "nic_name": nic_name, - "nic_nsg_attached": has_nic_nsg, - "subnet_nsg_attached": False if confirmed else None, - "determination": "non_compliant" if confirmed else "indeterminate", - }, - } - ) - break # one finding per VM is sufficient + if vm_finding is not None and not confirmed: + # Already have a finding for this VM (confirmed or + # indeterminate) and this NIC only adds another + # indeterminate one - it can't raise the severity, so + # keep the existing finding rather than overwrite it. + continue + vm_finding = { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY if confirmed else INDETERMINATE_SEVERITY, + "category": CATEGORY, + "resource_id": vm.id, + "resource_name": vm.name, + "resource_type": "Microsoft.Compute/virtualMachines", + "description": DESCRIPTION if confirmed else INDETERMINATE_DESCRIPTION, + "remediation": REMEDIATION if confirmed else INDETERMINATE_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "nic_id": nic_id, + "nic_name": nic_name, + "nic_nsg_attached": has_nic_nsg, + "subnet_nsg_attached": False if confirmed else None, + "determination": "non_compliant" if confirmed else "indeterminate", + }, + } + if confirmed: + break # a confirmed HIGH can't be outranked by another NIC on this VM + + if vm_finding is not None: + findings.append(vm_finding) return findings diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index b5392ff3..94d16e14 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -44,8 +44,8 @@ # provisioning state does not prove malware protection is truly off. INDETERMINATE_SEVERITY = "LOW" INDETERMINATE_DESCRIPTION = ( - "A recognised endpoint protection extension is installed but its provisioning state " - "indicates it did not complete successfully, so effective protection cannot be " + "A recognised endpoint protection extension is installed but its provisioning state is " + "missing or does not confirm successful completion, so effective protection cannot be " "confirmed. This is not a confirmed absence of endpoint protection." ) INDETERMINATE_REMEDIATION = ( @@ -63,6 +63,16 @@ logger = logging.getLogger(__name__) +# Microsoft renamed this Defender for Cloud recommendation from "Endpoint +# protection should be installed..." to "EDR solution should be installed +# on virtual machines" when it moved from the deprecated Log Analytics +# agent to agentless EDR scanning. Matching only "endpoint protection" +# means the index silently matches nothing against a current subscription's +# real assessment data, so Defender's signal is never found and every VM +# falls back to the weaker extension-name check - accept either name. +_ENDPOINT_PROTECTION_DISPLAY_NAME_MARKERS = ("endpoint protection", "edr solution") + + def _index_endpoint_protection_assessments(assessments: Optional[List[Any]]) -> Dict[str, List[Any]]: """Group the subscription-wide assessments list by resource ID, once per scan. @@ -73,7 +83,7 @@ def _index_endpoint_protection_assessments(assessments: Optional[List[Any]]) -> index: Dict[str, List[Any]] = {} for assessment in assessments or []: display_name = (getattr(assessment, "display_name", "") or "").lower() - if "endpoint protection" not in display_name: + if not any(marker in display_name for marker in _ENDPOINT_PROTECTION_DISPLAY_NAME_MARKERS): continue resource_details = getattr(assessment, "resource_details", None) resource_id = (getattr(resource_details, "id", "") or "").lower() @@ -202,19 +212,21 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: continue # A recognised EP extension is installed - name presence alone is not - # enough. Where the API exposes provisioning_state, an extension is - # only treated as healthy when it is unset/unknown (data doesn't - # support the check - do not invent a new false positive) or equals - # "Succeeded". Anything else (Failed, Canceled, ...) is surfaced as - # indeterminate rather than silently passed. - unhealthy_names = [] + # enough. An extension is only treated as confirmed healthy when its + # provisioning_state is exactly "Succeeded". A missing/unexposed + # provisioning_state does not prove the extension succeeded any more + # than it proves it failed - it's unknown evidence, not a pass - so + # it's surfaced as indeterminate together with any other + # non-"Succeeded" state (Failed, Canceled, ...) rather than silently + # passed. + unconfirmed_names = [] confirmed_healthy = False for name, ext in matched: provisioning_state = (getattr(ext, "provisioning_state", None) or "").lower() - if not provisioning_state or provisioning_state == "succeeded": + if provisioning_state == "succeeded": confirmed_healthy = True break - unhealthy_names.append(name) + unconfirmed_names.append(name) if confirmed_healthy: continue @@ -235,7 +247,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "metadata": { "resource_group": rg, "installed_extensions": installed_names, - "unhealthy_extensions": sorted(set(unhealthy_names)), + "unconfirmed_extensions": sorted(set(unconfirmed_names)), "signal": "extension_fallback", "determination": "indeterminate", }, diff --git a/scanner/rules/az_cmp_004.py b/scanner/rules/az_cmp_004.py index c23b33c3..2d7fd6f5 100644 --- a/scanner/rules/az_cmp_004.py +++ b/scanner/rules/az_cmp_004.py @@ -1,6 +1,7 @@ """AZ-CMP-004: VM without automatic OS patching enabled.""" import logging +from datetime import datetime, timezone from typing import Any, Dict, List RULE_ID = "AZ-CMP-004" @@ -50,9 +51,51 @@ # enough evidence to override a config-based pass. _CONCLUSIVE_ASSESSMENT_STATUSES = {"succeeded", "completedwithwarnings"} +# A "clean" assessment (zero pending critical/security patches) only counts +# as real evidence of the VM's *current* state while it's recent - Azure +# doesn't re-run this automatically on a fixed schedule, so an old clean +# result proves nothing about patches that have become available since. +STALE_ASSESSMENT_THRESHOLD_DAYS = 30 + +# An unavailable, non-conclusive, or stale assessment means config alone is +# the only signal - which is real evidence config is correctly set, but not +# proof patches have actually landed. Surfaced as indeterminate rather than +# silently treated as a clean pass, the same LOW/indeterminate split used by +# AZ-CMP-001/003 for their own unresolvable evidence. +INDETERMINATE_SEVERITY = "LOW" +INDETERMINATE_DESCRIPTION = ( + "VM is configured for automatic OS patching, but its real Azure Update Manager patch " + "assessment is unavailable, did not complete successfully, or is older than " + f"{STALE_ASSESSMENT_THRESHOLD_DAYS} days, so the VM's actual current patch state cannot " + "be confirmed. Config alone is not proof patches have actually been applied." +) +INDETERMINATE_REMEDIATION = ( + "Trigger an on-demand patch assessment (Update Manager > Check for updates) so a current " + "result exists, then re-run the scan to confirm the VM's real patch state." +) + logger = logging.getLogger(__name__) +def _is_fresh(last_modified_time: Any) -> bool: + """Return True only when last_modified_time parses to a UTC-aware timestamp + within the staleness threshold. Missing or unparseable data is not fresh - + absence of a usable timestamp must never be read as "recent enough".""" + if isinstance(last_modified_time, datetime): + observed = last_modified_time + elif isinstance(last_modified_time, str): + try: + observed = datetime.fromisoformat(last_modified_time.replace("Z", "+00:00")) + except ValueError: + return False + else: + return False + if observed.tzinfo is None: + return False + age = datetime.now(timezone.utc) - observed + return age.days <= STALE_ASSESSMENT_THRESHOLD_DAYS + + def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: findings: List[Dict[str, Any]] = [] @@ -107,16 +150,55 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: ) continue - # Config says patching is enabled - check real assessment evidence - # for the false-negative case: config correct, platform hasn't - # actually applied the pending critical/security patches. + # Config says patching is enabled - check real assessment evidence. + # This can raise the result two ways: a conclusive, fresh assessment + # with pending critical/security patches overrides the config-based + # pass into a confirmed finding (the false-negative case: config + # correct, platform hasn't actually applied anything yet). Anything + # short of that - unavailable, non-conclusive, or stale evidence - + # is not proof patches were applied either, so it's surfaced as + # indeterminate rather than silently left as a clean pass. + def _indeterminate_finding(reason: str, patch_summary: Any = None) -> Dict[str, Any]: + metadata: Dict[str, Any] = { + "resource_group": rg, + "signal": "patch_assessment_inconclusive", + "determination": "indeterminate", + "reason": reason, + } + if patch_summary is not None: + metadata["assessment_status"] = (getattr(patch_summary, "status", "") or "").lower() + metadata["last_modified_time"] = str(getattr(patch_summary, "last_modified_time", "") or "") or None + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": INDETERMINATE_SEVERITY, + "category": CATEGORY, + "resource_id": vm.id, + "resource_name": vm_name, + "resource_type": "Microsoft.Compute/virtualMachines", + "description": INDETERMINATE_DESCRIPTION, + "remediation": INDETERMINATE_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": metadata, + } + patch_summary = azure_client.get_vm_patch_status(rg, vm_name) if patch_summary is None: + findings.append(_indeterminate_finding("assessment_unavailable")) continue status = (getattr(patch_summary, "status", "") or "").lower() + if status not in _CONCLUSIVE_ASSESSMENT_STATUSES: + findings.append(_indeterminate_finding("assessment_not_conclusive", patch_summary)) + continue + critical_count = getattr(patch_summary, "critical_and_security_patch_count", None) - if status in _CONCLUSIVE_ASSESSMENT_STATUSES and critical_count is not None and critical_count > 0: + if critical_count is None: + findings.append(_indeterminate_finding("patch_count_unavailable", patch_summary)) + continue + + if critical_count > 0: findings.append( { "rule_id": RULE_ID, @@ -140,5 +222,11 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: }, } ) + continue + + # Conclusive assessment says zero pending critical/security patches - + # only trust that as a real clean signal while it's recent. + if not _is_fresh(getattr(patch_summary, "last_modified_time", None)): + findings.append(_indeterminate_finding("assessment_stale", patch_summary)) return findings diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index cef4cc62..ddb17795 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -6,6 +6,8 @@ helper accessors from tests/helpers/mock_azure.py. """ +from datetime import datetime, timedelta, timezone + import pytest import scanner.rules.az_cmp_001 as az_cmp_001 @@ -231,6 +233,86 @@ def test_cmp_001_mixed_resolvable_and_unresolvable_subnets_is_indeterminate(mock assert findings[0]["metadata"]["determination"] == "indeterminate" +def test_cmp_001_confirmed_nic_after_indeterminate_nic_is_not_downgraded(mock_azure, subscription_id): + """A VM can have more than one exposed NIC. If the first NIC evaluated is + only indeterminate (unresolvable subnet) but a second NIC on the same VM + is a real, confirmed violation, the VM's single reported finding must be + the confirmed HIGH one - not the indeterminate LOW that happened to be + evaluated first.""" + indeterminate_subnet_id = _subnet_id("vnet1", "subnet-unresolvable") + confirmed_subnet_id = _subnet_id("vnet1", "subnet-no-nsg") + nic_indeterminate = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=indeterminate_subnet_id)) + ], + network_security_group=None, + ) + nic_confirmed = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip2"), subnet=make_resource(id=confirmed_subnet_id)) + ], + network_security_group=None, + ) + vm = make_resource( + id=_vm_id("vm-two-nics"), + name="vm-two-nics", + network_profile=make_resource( + network_interfaces=[ + make_resource(id=_nic_id("nic-indeterminate")), + make_resource(id=_nic_id("nic-confirmed")), + ] + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic-indeterminate", nic_indeterminate) + mock_azure.set_network_interface(_RG, "nic-confirmed", nic_confirmed) + mock_azure.set_subnet(confirmed_subnet_id, make_resource(id=confirmed_subnet_id, network_security_group=None)) + # indeterminate_subnet_id is intentionally never registered via set_subnet(). + findings = az_cmp_001.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "HIGH" + assert f["metadata"]["determination"] == "non_compliant" + assert f["metadata"]["nic_name"] == "nic-confirmed" + + +def test_cmp_001_indeterminate_nic_after_confirmed_nic_does_not_downgrade(mock_azure, subscription_id): + """Same scenario in the opposite NIC order - a confirmed violation found + first must not be replaced by a later indeterminate one either.""" + confirmed_subnet_id = _subnet_id("vnet1", "subnet-no-nsg") + indeterminate_subnet_id = _subnet_id("vnet1", "subnet-unresolvable") + nic_confirmed = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip1"), subnet=make_resource(id=confirmed_subnet_id)) + ], + network_security_group=None, + ) + nic_indeterminate = make_resource( + ip_configurations=[ + make_resource(public_ip_address=make_resource(id="pip2"), subnet=make_resource(id=indeterminate_subnet_id)) + ], + network_security_group=None, + ) + vm = make_resource( + id=_vm_id("vm-two-nics-reversed"), + name="vm-two-nics-reversed", + network_profile=make_resource( + network_interfaces=[ + make_resource(id=_nic_id("nic-confirmed")), + make_resource(id=_nic_id("nic-indeterminate")), + ] + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, "nic-confirmed", nic_confirmed) + mock_azure.set_network_interface(_RG, "nic-indeterminate", nic_indeterminate) + mock_azure.set_subnet(confirmed_subnet_id, make_resource(id=confirmed_subnet_id, network_security_group=None)) + findings = az_cmp_001.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["severity"] == "HIGH" + assert findings[0]["metadata"]["determination"] == "non_compliant" + + # ── AZ-CMP-002: disk using platform-managed encryption only ───────────────── # # ManagedDiskParameters (the object actually embedded in a VM's @@ -447,13 +529,13 @@ def test_cmp_002_compliant_with_real_sdk_models_returns_no_findings(mock_azure, def test_cmp_003_compliant_with_ep_extension_returns_no_findings(mock_azure, subscription_id): - """A VM with a recognised endpoint-protection extension is compliant.""" + """A VM with a recognised, successfully-provisioned endpoint-protection extension is compliant.""" vm = make_resource(id=_vm_id("vm-protected"), name="vm-protected") mock_azure.set_virtual_machines([vm]) mock_azure.set_vm_extensions( _RG, "vm-protected", - [make_resource(type_properties_type="IaaSAntimalware")], + [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")], ) assert az_cmp_003.scan(mock_azure, subscription_id) == [] @@ -512,7 +594,7 @@ def test_cmp_003_extension_present_but_unhealthy_returns_indeterminate_finding(m assert f["rule_id"] == "AZ-CMP-003" assert f["severity"] == "LOW" assert f["metadata"]["determination"] == "indeterminate" - assert f["metadata"]["unhealthy_extensions"] == ["iaasantimalware"] + assert f["metadata"]["unconfirmed_extensions"] == ["iaasantimalware"] def test_cmp_003_duplicate_extension_types_use_healthy_record_regardless_of_order(mock_azure, subscription_id): @@ -534,8 +616,9 @@ def test_cmp_003_duplicate_extension_types_use_healthy_record_regardless_of_orde assert az_cmp_003.scan(mock_azure, subscription_id) == [] -def test_cmp_003_missing_provisioning_state_falls_back_to_name_based_pass(mock_azure, subscription_id): - """When provisioning_state isn't exposed by the API, name presence alone remains sufficient.""" +def test_cmp_003_missing_provisioning_state_is_indeterminate_not_a_pass(mock_azure, subscription_id): + """When provisioning_state isn't exposed by the API, that's unknown evidence, not + confirmation the extension actually succeeded - name presence alone must not pass.""" vm = make_resource(id=_vm_id("vm-no-state-data"), name="vm-no-state-data") mock_azure.set_virtual_machines([vm]) mock_azure.set_vm_extensions( @@ -543,7 +626,12 @@ def test_cmp_003_missing_provisioning_state_falls_back_to_name_based_pass(mock_a "vm-no-state-data", [make_resource(type_properties_type="IaaSAntimalware")], ) - assert az_cmp_003.scan(mock_azure, subscription_id) == [] + findings = az_cmp_003.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["unconfirmed_extensions"] == ["iaasantimalware"] # ── AZ-CMP-003: Defender for Cloud endpoint-protection assessment ─────────── @@ -592,13 +680,31 @@ def test_cmp_003_defender_unhealthy_returns_confirmed_high_finding_even_with_ext assert f["metadata"]["determination"] == "non_compliant" +def test_cmp_003_recognises_current_edr_recommendation_display_name(mock_azure, subscription_id): + """Microsoft renamed this recommendation from 'Endpoint protection should be + installed...' to 'EDR solution should be installed on virtual machines' when it + moved to agentless EDR scanning. A current subscription's real assessment data + uses the new name - it must still be recognised as Defender's Healthy signal, + not silently ignored and left to fall back to the weaker extension check.""" + vm_id = _vm_id("vm-edr-name") + vm = make_resource(id=vm_id, name="vm-edr-name") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-edr-name", []) + mock_azure.set_security_assessments( + [_assessment(vm_id, display_name="EDR solution should be installed on virtual machines", status_code="Healthy")] + ) + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + def test_cmp_003_defender_not_applicable_falls_back_to_extension_check(mock_azure, subscription_id): """A NotApplicable Defender status carries no usable signal - the extension-based check still governs the result.""" vm_id = _vm_id("vm-defender-na") vm = make_resource(id=vm_id, name="vm-defender-na") mock_azure.set_virtual_machines([vm]) - mock_azure.set_vm_extensions(_RG, "vm-defender-na", [make_resource(type_properties_type="IaaSAntimalware")]) + mock_azure.set_vm_extensions( + _RG, "vm-defender-na", [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")] + ) mock_azure.set_security_assessments([_assessment(vm_id, status_code="NotApplicable")]) assert az_cmp_003.scan(mock_azure, subscription_id) == [] @@ -622,7 +728,9 @@ def test_cmp_003_defender_assessment_for_different_resource_is_ignored(mock_azur vm_id = _vm_id("vm-target") vm = make_resource(id=vm_id, name="vm-target") mock_azure.set_virtual_machines([vm]) - mock_azure.set_vm_extensions(_RG, "vm-target", [make_resource(type_properties_type="IaaSAntimalware")]) + mock_azure.set_vm_extensions( + _RG, "vm-target", [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")] + ) mock_azure.set_security_assessments([_assessment(_vm_id("vm-other"), status_code="Unhealthy")]) # Falls back to the extension check (which passes) rather than picking up the other VM's Unhealthy status. assert az_cmp_003.scan(mock_azure, subscription_id) == [] @@ -673,7 +781,9 @@ def test_cmp_003_defender_unhealthy_wins_over_healthy_regardless_of_assessment_o def test_cmp_004_compliant_auto_updates_returns_no_findings(mock_azure, subscription_id): - """A Windows VM with automatic updates enabled is compliant.""" + """A Windows VM with automatic updates enabled AND a fresh, conclusive, clean patch + assessment is genuinely compliant - config alone is no longer sufficient on its own, + since it doesn't prove patches have actually been applied.""" vm = make_resource( id=_vm_id("vm-patched"), name="vm-patched", @@ -683,6 +793,9 @@ def test_cmp_004_compliant_auto_updates_returns_no_findings(mock_azure, subscrip ), ) mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, "vm-patched", _patch_summary(status="Succeeded", critical_and_security_patch_count=0) + ) assert az_cmp_004.scan(mock_azure, subscription_id) == [] @@ -711,12 +824,21 @@ def test_cmp_004_noncompliant_no_patching_returns_one_finding(mock_azure, subscr # ── AZ-CMP-004: real patch-assessment evidence override ───────────────────── -def _patch_summary(status="Succeeded", critical_and_security_patch_count=0, other_patch_count=0): - """An AvailablePatchSummary-shaped stub, as returned by AzureClient.get_vm_patch_status().""" +def _patch_summary( + status="Succeeded", critical_and_security_patch_count=0, other_patch_count=0, last_modified_time=None +): + """An AvailablePatchSummary-shaped stub, as returned by AzureClient.get_vm_patch_status(). + + Defaults last_modified_time to "just now" so tests that aren't specifically + about staleness don't need to think about the freshness threshold. + """ + if last_modified_time is None: + last_modified_time = datetime.now(timezone.utc) return make_resource( status=status, critical_and_security_patch_count=critical_and_security_patch_count, other_patch_count=other_patch_count, + last_modified_time=last_modified_time, ) @@ -763,9 +885,10 @@ def test_cmp_004_config_ok_and_assessment_clean_returns_no_findings(mock_azure, assert az_cmp_004.scan(mock_azure, subscription_id) == [] -def test_cmp_004_config_ok_but_assessment_in_progress_does_not_override(mock_azure, subscription_id): - """An assessment that hasn't conclusively finished is not reliable evidence - it must not - override a config-based pass even if it happens to carry a nonzero patch count so far.""" +def test_cmp_004_config_ok_but_assessment_in_progress_is_indeterminate(mock_azure, subscription_id): + """An assessment that hasn't conclusively finished is not reliable evidence either way - + it must not become a HIGH override (the nonzero patch count so far isn't final) and must + not silently pass either (it doesn't confirm patches were applied). Indeterminate LOW.""" vm = make_resource( id=_vm_id("vm-assessment-in-progress"), name="vm-assessment-in-progress", @@ -778,12 +901,17 @@ def test_cmp_004_config_ok_but_assessment_in_progress_does_not_override(mock_azu mock_azure.set_vm_patch_status( _RG, "vm-assessment-in-progress", _patch_summary(status="InProgress", critical_and_security_patch_count=5) ) - assert az_cmp_004.scan(mock_azure, subscription_id) == [] + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["reason"] == "assessment_not_conclusive" -def test_cmp_004_config_ok_and_no_assessment_data_returns_no_findings(mock_azure, subscription_id): - """No real assessment evidence available (never run, API failure) - behaves exactly like - the pre-existing config-only check with nothing to override.""" +def test_cmp_004_config_ok_and_no_assessment_data_is_indeterminate(mock_azure, subscription_id): + """No real assessment evidence available (never run, API failure) - config alone is not + proof patches were applied, so this must surface as indeterminate, not a silent pass.""" vm = make_resource( id=_vm_id("vm-no-assessment-data"), name="vm-no-assessment-data", @@ -794,9 +922,86 @@ def test_cmp_004_config_ok_and_no_assessment_data_returns_no_findings(mock_azure ) mock_azure.set_virtual_machines([vm]) # set_vm_patch_status not called -> None, matching "no assessment has ever run". + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["reason"] == "assessment_unavailable" + + +def test_cmp_004_config_ok_but_stale_clean_assessment_is_indeterminate(mock_azure, subscription_id): + """A conclusive, clean (zero pending patches) assessment only counts as real evidence of + the VM's *current* state while it's recent. An old clean result proves nothing about + patches that have become available since - must not be a silent pass.""" + vm = make_resource( + id=_vm_id("vm-stale-clean-assessment"), + name="vm-stale-clean-assessment", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + stale_time = datetime.now(timezone.utc) - timedelta(days=az_cmp_004.STALE_ASSESSMENT_THRESHOLD_DAYS + 1) + mock_azure.set_vm_patch_status( + _RG, + "vm-stale-clean-assessment", + _patch_summary(status="Succeeded", critical_and_security_patch_count=0, last_modified_time=stale_time), + ) + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["reason"] == "assessment_stale" + + +def test_cmp_004_config_ok_and_fresh_clean_assessment_returns_no_findings(mock_azure, subscription_id): + """A conclusive, clean assessment within the freshness threshold is genuine evidence of + current compliance - must not be flagged.""" + vm = make_resource( + id=_vm_id("vm-fresh-clean-assessment"), + name="vm-fresh-clean-assessment", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + recent_time = datetime.now(timezone.utc) - timedelta(days=1) + mock_azure.set_vm_patch_status( + _RG, + "vm-fresh-clean-assessment", + _patch_summary(status="Succeeded", critical_and_security_patch_count=0, last_modified_time=recent_time), + ) assert az_cmp_004.scan(mock_azure, subscription_id) == [] +def test_cmp_004_config_ok_and_missing_last_modified_time_is_indeterminate(mock_azure, subscription_id): + """A conclusive clean assessment with no usable timestamp can't be proven fresh - absence + of a timestamp must never be read as 'recent enough'.""" + vm = make_resource( + id=_vm_id("vm-no-timestamp"), + name="vm-no-timestamp", + os_profile=make_resource( + windows_configuration=make_resource(enable_automatic_updates=True, patch_settings=None), + linux_configuration=None, + ), + ) + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_patch_status( + _RG, + "vm-no-timestamp", + make_resource( + status="Succeeded", critical_and_security_patch_count=0, other_patch_count=0, last_modified_time=None + ), + ) + findings = az_cmp_004.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["reason"] == "assessment_stale" + + def test_cmp_004_config_disabled_finding_unaffected_by_clean_assessment(mock_azure, subscription_id): """Config-disabled auto-patching is itself an unmanaged-drift risk - a clean point-in-time assessment must not suppress the config-based finding.""" From 1833747c86709c3a77a54e0706d923e84139ee2a Mon Sep 17 00:00:00 2001 From: parthrohit22 Date: Sat, 29 Aug 2026 14:10:35 +0100 Subject: [PATCH 8/9] fix(tests): resolve _subnet_id name collision from the dev merge The dev-merge commit (6aab017) resolved a *textual* conflict in tests/test_rules_compute.py cleanly but left a *semantic* one: this branch's own _subnet_id(vnet_name, subnet_name) helper (10 call sites, used throughout the AZ-CMP-001 tests) and dev's newly-added _subnet_id(name) helper (added for the AZ-CMP-007 JIT tests, 1 call site) share a name in two non-adjacent parts of the file, so git saw no textual overlap - but Python resolves both definitions to whichever one appears last in the module, silently shadowing this branch's 2-arg version for every one of its 10 call sites. Renamed dev's newer, single-use helper to _jit_subnet_id instead of touching the 10 existing call sites. Caught by actually running the test suite after the merge, not just checking for a clean git merge. Signed-off-by: Parth J Rohit Signed-off-by: parthrohit22 --- tests/test_rules_compute.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index 4c5f0e25..b5f4d6ee 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -1169,7 +1169,7 @@ def test_cmp_007_indeterminate_jit_is_not_flagged(mock_azure, subscription_id): assert az_cmp_007.scan(mock_azure, subscription_id) == [] -def _subnet_id(name): +def _jit_subnet_id(name): return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/virtualNetworks/vnet/subnets/{name}" @@ -1214,7 +1214,7 @@ def test_cmp_007_port_range_in_ranges_list_covering_rdp(mock_azure, subscription def test_cmp_007_subnet_level_nsg_exposure_is_flagged(mock_azure, subscription_id): """A VM with no NIC-level NSG is still exposed if the NSG on its subnet opens SSH.""" - subnet_id = _subnet_id("subnet1") + subnet_id = _jit_subnet_id("subnet1") nic = make_resource( network_security_group=None, ip_configurations=[make_resource(subnet=make_resource(id=subnet_id))], From 792ef98b9f10cc0240dcb1960325db5dc9436646 Mon Sep 17 00:00:00 2001 From: parthrohit22 Date: Tue, 1 Sep 2026 15:45:23 +0100 Subject: [PATCH 9/9] fix(scanner): AZ-CMP-003 check all matched extensions, tighten EDR marker - Stop breaking out of the extension loop on the first Succeeded record. A VM with one healthy and one genuinely failed recognised extension (e.g. IaaSAntimalware succeeded, MDE.Linux failed) was silently stamped compliant, with the failed extension never appearing in finding metadata. Records are now grouped by extension name so duplicate reports of the same extension still resolve on 'any succeeded', while distinct extensions follow the same unconfirmed-wins precedent already used for Defender assessments above. - Tighten the EDR display-name marker from the bare substring 'edr solution' to the full recommendation title 'edr solution should be installed', so an unrelated future recommendation containing those two words can't be mistaken for this rule's Defender signal. Addresses TFT444's review on #272. Signed-off-by: parthrohit22 --- scanner/rules/az_cmp_003.py | 33 ++++++++++++++++++++------- tests/test_rules_compute.py | 45 +++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 8 deletions(-) diff --git a/scanner/rules/az_cmp_003.py b/scanner/rules/az_cmp_003.py index 94d16e14..538bc230 100644 --- a/scanner/rules/az_cmp_003.py +++ b/scanner/rules/az_cmp_003.py @@ -70,7 +70,13 @@ # means the index silently matches nothing against a current subscription's # real assessment data, so Defender's signal is never found and every VM # falls back to the weaker extension-name check - accept either name. -_ENDPOINT_PROTECTION_DISPLAY_NAME_MARKERS = ("endpoint protection", "edr solution") +# +# The second marker is the full recommendation title, not the bare +# substring "edr solution": a bare substring match risks silently pulling +# in a future, unrelated Defender recommendation that happens to contain +# those two words, misattributing its status to this rule's endpoint- +# protection determination. +_ENDPOINT_PROTECTION_DISPLAY_NAME_MARKERS = ("endpoint protection", "edr solution should be installed") def _index_endpoint_protection_assessments(assessments: Optional[List[Any]]) -> Dict[str, List[Any]]: @@ -219,16 +225,27 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: # it's surfaced as indeterminate together with any other # non-"Succeeded" state (Failed, Canceled, ...) rather than silently # passed. - unconfirmed_names = [] - confirmed_healthy = False + # + # Every matched extension is checked - not just the first. Records + # are grouped by extension name first: two API records for the same + # extension (e.g. a transient re-sync reporting Failed then + # Succeeded) are one signal, and that name is healthy if any record + # for it succeeded. But two *different* recognised extensions are + # separate signals, and the same "unconfirmed wins" precedent as the + # Defender-assessment branch above applies across them - a VM with + # IaaSAntimalware Succeeded and MDE.Linux Failed is not compliant + # just because one of the two came up healthy. Stopping at the first + # Succeeded record would silently drop the failed extension from + # both the verdict and unconfirmed_names. + states_by_name: Dict[str, List[str]] = {} for name, ext in matched: provisioning_state = (getattr(ext, "provisioning_state", None) or "").lower() - if provisioning_state == "succeeded": - confirmed_healthy = True - break - unconfirmed_names.append(name) + states_by_name.setdefault(name, []).append(provisioning_state) - if confirmed_healthy: + unconfirmed_names = [name for name, states in states_by_name.items() if "succeeded" not in states] + any_confirmed_healthy = any("succeeded" in states for states in states_by_name.values()) + + if any_confirmed_healthy and not unconfirmed_names: continue findings.append( diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index b5f4d6ee..e0b7c57a 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -617,6 +617,31 @@ def test_cmp_003_duplicate_extension_types_use_healthy_record_regardless_of_orde assert az_cmp_003.scan(mock_azure, subscription_id) == [] +def test_cmp_003_one_succeeded_and_one_failed_extension_is_indeterminate_not_a_pass(mock_azure, subscription_id): + """Two *different* recognised EP extensions, one Succeeded and one Failed, must not be + silently stamped compliant just because one of them came up healthy - the failed one + has to surface in unconfirmed_extensions, regardless of which record is checked first.""" + for extensions in ( + [ + make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded"), + make_resource(type_properties_type="MDE.Linux", provisioning_state="Failed"), + ], + [ + make_resource(type_properties_type="MDE.Linux", provisioning_state="Failed"), + make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded"), + ], + ): + vm = make_resource(id=_vm_id("vm-mixed-extensions"), name="vm-mixed-extensions") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions(_RG, "vm-mixed-extensions", extensions) + findings = az_cmp_003.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert f["severity"] == "LOW" + assert f["metadata"]["determination"] == "indeterminate" + assert f["metadata"]["unconfirmed_extensions"] == ["mde.linux"] + + def test_cmp_003_missing_provisioning_state_is_indeterminate_not_a_pass(mock_azure, subscription_id): """When provisioning_state isn't exposed by the API, that's unknown evidence, not confirmation the extension actually succeeded - name presence alone must not pass.""" @@ -697,6 +722,26 @@ def test_cmp_003_recognises_current_edr_recommendation_display_name(mock_azure, assert az_cmp_003.scan(mock_azure, subscription_id) == [] +def test_cmp_003_edr_solution_substring_alone_does_not_match_unrelated_recommendation(mock_azure, subscription_id): + """The marker is the full 'edr solution should be installed' recommendation title, not + the bare substring 'edr solution' - an unrelated recommendation that happens to contain + those two words must not be mistaken for this rule's Defender signal.""" + vm_id = _vm_id("vm-unrelated-edr-recommendation") + vm = make_resource(id=vm_id, name="vm-unrelated-edr-recommendation") + mock_azure.set_virtual_machines([vm]) + mock_azure.set_vm_extensions( + _RG, + "vm-unrelated-edr-recommendation", + [make_resource(type_properties_type="IaaSAntimalware", provisioning_state="Succeeded")], + ) + mock_azure.set_security_assessments( + [_assessment(vm_id, display_name="Review edr solution licensing costs", status_code="Unhealthy")] + ) + # The unrelated assessment must not be picked up as the Defender signal - falls back to + # the extension check, which passes. + assert az_cmp_003.scan(mock_azure, subscription_id) == [] + + def test_cmp_003_defender_not_applicable_falls_back_to_extension_check(mock_azure, subscription_id): """A NotApplicable Defender status carries no usable signal - the extension-based check still governs the result."""