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 ab2bacdd..caaa4d20 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -59,6 +59,8 @@ 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() def _build_devops_client(self) -> Optional[Any]: @@ -287,6 +289,54 @@ 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. + + 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). + Callers must never interpret ``None`` as "subnet has no NSG". + """ + 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 = "" + 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) + else: + client = NetworkManagementClient(self.credential, self.subscription_id) + subnet = client.subnets.get(resource_group, vnet_name, subnet_name) + + except Exception as exc: + logger.error("get_subnet failed for %s: %s", subnet_id, exc) + self._subnets_cache[subnet_id] = subnet + return subnet + def get_virtual_networks(self) -> List[Any]: """List all virtual networks in the subscription.""" try: @@ -629,6 +679,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. @@ -666,6 +742,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_001.py b/scanner/rules/az_cmp_001.py index e47ebee1..bd696ac1 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,11 +21,64 @@ ) 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_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. + + 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) + if not subnet_id: + unresolved = True + continue + + subnet = azure_client.get_subnet(subnet_id) + if subnet is None: + unresolved = True + continue + + if getattr(subnet, "network_security_group", None): + return True + + return None if unresolved else 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(): @@ -33,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: @@ -51,28 +112,44 @@ 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)) - - if has_public_ip and not has_nsg: - 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": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "nic_id": nic_id, - "nic_name": nic_name, - }, - } - ) - break # one finding per VM is sufficient + has_nic_nsg = bool(getattr(nic, "network_security_group", None)) + + subnet_status: Optional[bool] = None + if has_public_ip and not has_nic_nsg: + subnet_status = _subnet_nsg_status(azure_client, nic) + + if has_public_ip and not has_nic_nsg and subnet_status is not True: + confirmed = subnet_status is False + 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 5f7cf0ba..538bc230 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, Tuple RULE_ID = "AZ-CMP-003" RULE_NAME = "VM Without Endpoint Protection Installed" @@ -21,6 +21,38 @@ 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, +# 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 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 = ( + "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", @@ -31,21 +63,124 @@ 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. +# +# 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]]: + """Group the subscription-wide assessments list by resource ID, once per scan. + + 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). + """ + index: Dict[str, List[Any]] = {} + for assessment in assessments or []: + display_name = (getattr(assessment, "display_name", "") or "").lower() + 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() + if not resource_id: + continue + index.setdefault(resource_id, []).append(assessment) + return index + + +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) + codes.add((getattr(status, "code", "") or "").lower()) + + 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(): - 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(assessments_by_resource, 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 - installed = set() + installed: List[Tuple[str, Any]] = [] for e in exts: t = ( getattr(e, "type_properties_type", None) @@ -53,9 +188,12 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: or getattr(e, "type", "") ) if t: - installed.add(t.lower()) + installed.append((t.lower(), e)) + + installed_names = sorted({name for name, _ in installed}) + matched = [(name, ext) for name, ext in installed if name in KNOWN_EP_EXTENSIONS] - if not installed.intersection(KNOWN_EP_EXTENSIONS): + if not matched: findings.append( { "rule_id": RULE_ID, @@ -71,9 +209,66 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "frameworks": FRAMEWORKS, "metadata": { "resource_group": rg, - "installed_extensions": sorted(installed), + "installed_extensions": installed_names, + "signal": "extension_fallback", + "determination": "non_compliant", }, } ) + continue + + # A recognised EP extension is installed - name presence alone is not + # 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. + # + # 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() + states_by_name.setdefault(name, []).append(provisioning_state) + + 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( + { + "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": installed_names, + "unconfirmed_extensions": sorted(set(unconfirmed_names)), + "signal": "extension_fallback", + "determination": "indeterminate", + }, + } + ) return findings diff --git a/scanner/rules/az_cmp_004.py b/scanner/rules/az_cmp_004.py index caf62cc0..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" @@ -25,9 +26,76 @@ ) 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"} + +# 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]] = [] @@ -75,8 +143,90 @@ 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. + # 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 critical_count is None: + findings.append(_indeterminate_finding("patch_count_unavailable", patch_summary)) + continue + + if 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, + }, + } + ) + 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/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 5b3c3ebb..70ea4e94 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]] = {} @@ -93,6 +94,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 @@ -240,6 +243,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": @@ -258,6 +269,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 + def set_jit_policies(self, policies: Optional[List[Any]]) -> "MockAzureClient": """Configure the Defender for Cloud JIT policies; ``None`` represents an unreadable/indeterminate result.""" self._jit_policies = policies diff --git a/tests/test_azure_client_management.py b/tests/test_azure_client_management.py index b541f1d6..de7118b9 100644 --- a/tests/test_azure_client_management.py +++ b/tests/test_azure_client_management.py @@ -59,6 +59,25 @@ 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") + 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 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 + with patch("scanner.azure_client.SqlManagementClient") as constructor: policy = SimpleNamespace(state="Enabled") constructor.return_value.server_blob_auditing_policies.get.return_value = policy @@ -240,6 +259,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 751bcb77..e0b7c57a 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 @@ -48,6 +50,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}" @@ -73,8 +82,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( @@ -84,6 +96,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] @@ -93,6 +106,214 @@ 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 + assert f["metadata"]["determination"] == "non_compliant" + + +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=[ + 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 + f = findings[0] + 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): + """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" + + +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 @@ -309,13 +530,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) == [] @@ -346,11 +567,269 @@ 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"]["unconfirmed_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_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.""" + 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")], + ) + 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 ─────────── + + +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_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_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.""" + 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", provisioning_state="Succeeded")] + ) + 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", 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) == [] + + +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 ──────────────────────────── 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", @@ -360,6 +839,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) == [] @@ -381,6 +863,212 @@ 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, 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, + ) + + +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_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", + 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) + ) + 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_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", + 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". + 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.""" + 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" # ── AZ-CMP-007: management ports open without Just-In-Time (JIT) access ────── @@ -526,7 +1214,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}" @@ -571,7 +1259,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))],