fix(scanner): remove compute-rule false positives/negatives (#268)

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

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

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,18 +21,79 @@
 )
 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():
         network_profile = getattr(vm, "network_profile", None)
         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