feat: add VM Scale Set inventory collector and rule AZ-CMP-005

compliance/frameworks/cis_azure_benchmark.json

@@ -133,6 +133,11 @@
       "control_name": "Ensure that 'OS patching' is enabled for virtual machines",
       "description": "The virtual machine does not have automatic OS patching enabled. CIS 8.3 requires that OS patches are applied in a timely manner. Unpatched VMs are vulnerable to known exploits targeting unpatched OS vulnerabilities."
     },
+    "AZ-CMP-006": {
+      "control_id": "N/A-CMP-006",
+      "control_name": "VM Scale Set NSG baseline (covered by the repository's CIS 7.1 network-interface NSG rule)",
+      "description": "CIS Azure Foundations recommendation 7.1 (\"Ensure that Network Security Groups are attached to network interfaces with public IP addresses\") is assigned to AZ-CMP-001 under the repository's one-CIS-ID-per-rule convention. This is the same underlying control applied to VM Scale Set network interface configurations instead of standalone VM NICs, so it is not assigned a second numbered mapping."
+    },
     "AZ-CMP-007": {
       "control_id": "N/A-CMP-007",
       "control_name": "Just-In-Time (JIT) VM access - Defender for Cloud recommendation, no numbered CIS Azure Foundations 2.0.0 control",

compliance/frameworks/iso27001.json

@@ -128,6 +128,11 @@
       "control_name": "Management of technical vulnerabilities",
       "description": "The virtual machine does not have automatic OS patching enabled. A.12.6.1 requires that information about technical vulnerabilities is obtained and the organisation's exposure evaluated. Without automatic patching, known OS vulnerabilities remain unmitigated."
     },
+    "AZ-CMP-006": {
+      "control_id": "A.13.1.1",
+      "control_name": "Network controls",
+      "description": "VM Scale Set instances with public IPs and no NSG on their network interface configuration have unrestricted network access. Network controls should be applied to all compute resources accessible from the internet."
+    },
     "AZ-CMP-007": {
       "control_id": "A.13.1.1",
       "control_name": "Network controls",

compliance/frameworks/nist_csf.json

@@ -133,6 +133,11 @@
       "control_name": "A vulnerability management plan is developed and implemented",
       "description": "The virtual machine does not have automatic OS patching enabled. PR.IP-12 requires that a vulnerability management plan is developed and implemented. Without automatic patching, known OS vulnerabilities remain unmitigated and exploitable."
     },
+    "AZ-CMP-006": {
+      "control_id": "PR.AC-3",
+      "control_name": "Remote access is managed",
+      "description": "VM Scale Set instances with public IPs and no NSG on their network interface configuration have unrestricted network access. NSGs should be attached to control inbound and outbound traffic and manage remote access to compute resources."
+    },
     "AZ-CMP-007": {
       "control_id": "PR.AC-3",
       "control_name": "Remote access is managed",

compliance/frameworks/soc2.json

@@ -148,6 +148,11 @@
       "control_name": "System Vulnerabilities are Identified and Managed",
       "description": "The virtual machine does not have automatic OS patching enabled. CC7.1 requires that vulnerabilities in system components are identified and managed through a defined process. Without automatic patching, known OS vulnerabilities are left unmitigated and exploitable."
     },
+    "AZ-CMP-006": {
+      "control_id": "CC6.6",
+      "control_name": "Restricts Access from Outside the Network Boundary",
+      "description": "A VM Scale Set network interface configuration with a public IP and no NSG has unrestricted inbound network access from the internet with no filtering in place. CC6.6 requires that logical access from outside the network boundary is restricted and controlled."
+    },
     "AZ-CMP-007": {
       "control_id": "CC6.6",
       "control_name": "Restricts Access from Outside the Network Boundary",

playbooks/cli/fix_az_cmp_006.sh

@@ -0,0 +1,35 @@
+#!/bin/bash
+set -euo pipefail
+# AZ-CMP-006: Associate an NSG with a VM Scale Set's network interface configuration
+# Usage: ./fix_az_cmp_006.sh <resource-group> <vmss-name> <nic-config-index> <nsg-resource-id>
+#
+# Find <nic-config-index> (0-based) with:
+#   az vmss show --resource-group <rg> --name <vmss-name> \
+#     --query 'virtualMachineProfile.networkProfile.networkInterfaceConfigurations[].name'
+#
+# If the NSG does not yet exist, create it first:
+#   az network nsg create --resource-group <rg> --name <nsg-name>
+RESOURCE_GROUP="${1:-}"
+VMSS_NAME="${2:-}"
+NIC_CONFIG_INDEX="${3:-}"
+NSG_ID="${4:-}"
+
+if [ -z "$RESOURCE_GROUP" ] || [ -z "$VMSS_NAME" ] || [ -z "$NIC_CONFIG_INDEX" ] || [ -z "$NSG_ID" ]; then
+  echo "Usage: $0 <resource-group> <vmss-name> <nic-config-index> <nsg-resource-id>"
+  exit 1
+fi
+
+echo "WARNING: this changes the VMSS model and requires upgrading existing instances to take"
+echo "effect on already-running VMs, which can briefly disrupt traffic depending on your"
+echo "upgrade policy. Review the scale set's upgrade policy before proceeding."
+
+echo "Associating NSG with network interface configuration index $NIC_CONFIG_INDEX on VMSS '$VMSS_NAME'..."
+
+az vmss update \
+  --resource-group "$RESOURCE_GROUP" \
+  --name "$VMSS_NAME" \
+  --set "virtualMachineProfile.networkProfile.networkInterfaceConfigurations[$NIC_CONFIG_INDEX].networkSecurityGroup.id=$NSG_ID"
+
+echo "Model updated for $VMSS_NAME. Existing instances still need to be upgraded to pick up the"
+echo "change (Manual/Rolling upgrade policy):"
+echo "  az vmss update-instances --resource-group $RESOURCE_GROUP --name $VMSS_NAME --instance-ids '*'"

scanner/azure_client.py

@@ -671,6 +671,15 @@ def get_virtual_machines(self) -> List[Any]:
             logger.error("get_virtual_machines failed: %s", exc)
             return []
 
+    def get_virtual_machine_scale_sets(self) -> List[Any]:
+        """List all VM Scale Sets across all resource groups in the subscription."""
+        try:
+            client = ComputeManagementClient(self.credential, self.subscription_id)
+            return list(client.virtual_machine_scale_sets.list_all())
+        except Exception as exc:
+            logger.error("get_virtual_machine_scale_sets failed: %s", exc)
+            return []
+
     def get_web_apps(self) -> List[Any]:
         """List all App Services in the subscription."""
         try:

scanner/rules/az_cmp_006.py

@@ -0,0 +1,134 @@
+"""AZ-CMP-006: VM Scale Set network profile has a public IP with no associated NSG."""
+
+import logging
+from typing import Any, Dict, List, Optional
+
+RULE_ID = "AZ-CMP-006"
+RULE_NAME = "VM Scale Set with Public IP and No Associated NSG on Network Interface"
+SEVERITY = "HIGH"
+CATEGORY = "Compute"
+FRAMEWORKS = {"CIS": "N/A-CMP-006", "NIST": "PR.AC-3", "ISO27001": "A.13.1.1", "SOC2": "CC6.6"}
+DESCRIPTION = (
+    "A VM Scale Set network interface configuration provisions a public IP address "
+    "for its instances but has no Network Security Group protecting that interface, "
+    "either directly or via the subnet it deploys into. Without an NSG, all inbound "
+    "ports are open to the internet by default on every instance created from this "
+    "scale set, creating an unrestricted attack surface."
+)
+REMEDIATION = (
+    "Attach an NSG to the scale set's network interface configuration or to the "
+    "subnet it deploys into, with rules that allow only required inbound traffic. "
+    "Remove the public IP configuration if internet access is not needed and use "
+    "Azure Bastion or a load balancer for administrative/application access instead."
+)
+PLAYBOOK = "playbooks/cli/fix_az_cmp_006.sh"
+
+# A subnet reference that can't be resolved (VNet collection failure, missing
+# permissions, or an ID this scan never saw) says nothing about whether that
+# subnet actually has an NSG — it must not be treated the same as a resolved
+# subnet confirmed to have none, or a scan-visibility gap silently turns into
+# a false HIGH finding on an already-protected VMSS.
+INDETERMINATE_SEVERITY = "LOW"
+INDETERMINATE_DESCRIPTION = (
+    "A VM Scale Set network interface configuration provisions a public IP address and has no "
+    "NSG directly attached, but the NSG state of the subnet it deploys into could not be "
+    "verified (virtual network collection failed, the scanning principal lacks "
+    "Microsoft.Network/virtualNetworks/read, or the subnet reference could not be matched). "
+    "This is not a confirmed violation — the subnet may already be protected by an NSG this "
+    "scan could not see."
+)
+INDETERMINATE_REMEDIATION = (
+    "Grant the scanning principal Microsoft.Network/virtualNetworks/read on the relevant "
+    "virtual network(s) and re-run the scan to determine the actual subnet NSG state."
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _subnet_nsg_map(azure_client: Any) -> Dict[str, bool]:
+    """Map subnet resource ID (lowercased) -> whether that subnet has an NSG attached.
+
+    A VMSS network interface configuration only references its subnet by ID
+    (ApiEntityReference); the subnet's own NSG lives on the VirtualNetwork
+    resource, so it must be resolved separately to avoid flagging a VMSS that
+    is actually protected at the subnet level instead of the NIC level.
+    Azure resource IDs are case-insensitive, so keys are normalized to
+    lowercase to avoid missing a match on casing differences alone.
+    """
+    subnet_nsgs: Dict[str, bool] = {}
+    for vnet in azure_client.get_virtual_networks():
+        for subnet in getattr(vnet, "subnets", []) or []:
+            subnet_id = getattr(subnet, "id", None)
+            if subnet_id:
+                subnet_nsgs[subnet_id.lower()] = bool(getattr(subnet, "network_security_group", None))
+    return subnet_nsgs
+
+
+def _subnet_nsg_status(subnet_nsgs: Dict[str, bool], ip_cfg: Any) -> Optional[bool]:
+    """Return True/False if the ip config's subnet NSG state is known, None if unresolved."""
+    subnet_ref = getattr(ip_cfg, "subnet", None)
+    subnet_id = getattr(subnet_ref, "id", None)
+    if not subnet_id:
+        return None
+    return subnet_nsgs.get(subnet_id.lower())
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    """Detect VM Scale Sets whose network interface configuration has a public IP
+    but no NSG protecting it, at either the NIC or the subnet level."""
+    findings: List[Dict[str, Any]] = []
+    subnet_nsgs = _subnet_nsg_map(azure_client)
+
+    for vmss in azure_client.get_virtual_machine_scale_sets():
+        vmss_id = getattr(vmss, "id", "")
+        vmss_name = getattr(vmss, "name", "")
+        if not vmss_id or not vmss_name:
+            continue
+
+        vm_profile = getattr(vmss, "virtual_machine_profile", None)
+        network_profile = getattr(vm_profile, "network_profile", None)
+        if not network_profile:
+            continue
+
+        net_configs = getattr(network_profile, "network_interface_configurations", []) or []
+        for net_config in net_configs:
+            ip_configs = getattr(net_config, "ip_configurations", []) or []
+            has_public_ip = any(getattr(ip_cfg, "public_ip_address_configuration", None) for ip_cfg in ip_configs)
+            has_nic_nsg = bool(getattr(net_config, "network_security_group", None))
+
+            if not has_public_ip or has_nic_nsg:
+                continue
+
+            subnet_statuses = [_subnet_nsg_status(subnet_nsgs, ip_cfg) for ip_cfg in ip_configs]
+            has_subnet_nsg = any(status is True for status in subnet_statuses)
+            if has_subnet_nsg:
+                continue  # protected at the subnet level, compliant
+
+            has_unresolved_subnet = any(status is None for status in subnet_statuses)
+            confirmed = not has_unresolved_subnet
+
+            parsed = azure_client.parse_resource_id(vmss_id)
+            findings.append(
+                {
+                    "rule_id": RULE_ID,
+                    "rule_name": RULE_NAME,
+                    "severity": SEVERITY if confirmed else INDETERMINATE_SEVERITY,
+                    "category": CATEGORY,
+                    "resource_id": vmss_id,
+                    "resource_name": vmss_name,
+                    "resource_type": "Microsoft.Compute/virtualMachineScaleSets",
+                    "description": DESCRIPTION if confirmed else INDETERMINATE_DESCRIPTION,
+                    "remediation": REMEDIATION if confirmed else INDETERMINATE_REMEDIATION,
+                    "playbook": PLAYBOOK,
+                    "frameworks": FRAMEWORKS,
+                    "metadata": {
+                        "resource_group": parsed.get("resource_group", ""),
+                        "location": getattr(vmss, "location", ""),
+                        "network_interface_configuration": getattr(net_config, "name", ""),
+                        "determination": "non_compliant" if confirmed else "indeterminate",
+                    },
+                }
+            )
+            break  # one finding per VMSS is sufficient
+
+    return findings

tests/helpers/mock_azure.py

@@ -50,6 +50,7 @@ def __init__(self) -> None:
         self._network_security_groups: List[Any] = []
         self._express_route_ports: Optional[List[Any]] = []
         self._virtual_machines: List[Any] = []
+        self._virtual_machine_scale_sets: List[Any] = []
         self._key_vaults: List[Any] = []
         self._sql_servers: List[Any] = []
         self._service_principals: List[Any] = []
@@ -190,6 +191,10 @@ def set_virtual_machines(self, vms: List[Any]) -> "MockAzureClient":
         self._virtual_machines = vms
         return self
 
+    def set_virtual_machine_scale_sets(self, scale_sets: List[Any]) -> "MockAzureClient":
+        self._virtual_machine_scale_sets = scale_sets
+        return self
+
     def set_key_vaults(self, vaults: List[Any]) -> "MockAzureClient":
         self._key_vaults = vaults
         return self
@@ -217,6 +222,9 @@ def get_network_security_groups(self) -> List[Any]:
     def get_virtual_machines(self) -> List[Any]:
         return self._virtual_machines
 
+    def get_virtual_machine_scale_sets(self) -> List[Any]:
+        return self._virtual_machine_scale_sets
+
     def get_key_vaults(self) -> List[Any]:
         return self._key_vaults
 

tests/test_azure_client_management.py

@@ -32,6 +32,11 @@ def test_parse_resource_id_handles_full_and_short_ids():
         ("get_public_ip_addresses", "scanner.azure_client.NetworkManagementClient", "public_ip_addresses.list_all"),
         ("get_load_balancers", "scanner.azure_client.NetworkManagementClient", "load_balancers.list_all"),
         ("get_virtual_machines", "scanner.azure_client.ComputeManagementClient", "virtual_machines.list_all"),
+        (
+            "get_virtual_machine_scale_sets",
+            "scanner.azure_client.ComputeManagementClient",
+            "virtual_machine_scale_sets.list_all",
+        ),
         ("get_postgresql_servers", "scanner.azure_client.PostgreSQLManagementClient", "servers.list"),
         ("get_sql_servers", "scanner.azure_client.SqlManagementClient", "servers.list"),
         ("get_key_vaults", "scanner.azure_client.KeyVaultManagementClient", "vaults.list_by_subscription"),