feat(azure): add network_subnet_nsg_associated check (#11043)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
2026-07-04 19:21:51 +00:00 · 2026-06-22 21:03:54 +09:00
parent 6dda1ae485
commit 869f0726f5
9 changed files with 339 additions and 2 deletions
@@ -28,6 +28,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `postgresql_flexible_server_high_availability_enabled` check for Azure provider, verifying PostgreSQL Flexible Servers have high availability enabled for automatic failover to a standby replica [(#11046)](https://github.com/prowler-cloud/prowler/pull/11046)
 - `aks_cluster_azure_monitor_enabled` check for Azure provider, verifying AKS clusters have Azure Monitor (Container Insights) enabled for metrics, logs, and alerting [(#11029)](https://github.com/prowler-cloud/prowler/pull/11029)
 - `aks_cluster_local_accounts_disabled` check for Azure provider, verifying AKS clusters have local accounts disabled so authentication is forced through Microsoft Entra ID [(#11030)](https://github.com/prowler-cloud/prowler/pull/11030)
+- `network_subnet_nsg_associated` check for Azure provider, verifying virtual network subnets have a network security group associated to enforce traffic filtering [(#11043)](https://github.com/prowler-cloud/prowler/pull/11043)
 - `aks_cluster_auto_upgrade_enabled` check for Azure provider [(#11027)](https://github.com/prowler-cloud/prowler/pull/11027)
 - Public `Provider.get_class()` method that resolves a provider class by name for both built-in and external (entry-point) providers [(#11398)](https://github.com/prowler-cloud/prowler/pull/11398)
 - Jira timeout preventing the calls from hanging indefinitely when the Jira endpoint is unreachable or slow [(#11602)](https://github.com/prowler-cloud/prowler/pull/11602)
@@ -1974,7 +1974,9 @@
    {
      "Id": "7.11",
      "Description": "Ensure subnets are associated with network security groups",
-      "Checks": [],
+      "Checks": [
+        "network_subnet_nsg_associated"
+      ],
      "Attributes": [
        {
          "Section": "7 Networking Services",
@@ -1429,6 +1429,7 @@
        "network_public_ip_shodan",
        "network_rdp_internet_access_restricted",
        "network_ssh_internet_access_restricted",
+        "network_subnet_nsg_associated",
        "network_udp_internet_access_restricted",
        "network_watcher_enabled"
      ]
@@ -1483,6 +1484,7 @@
        "network_public_ip_shodan",
        "network_rdp_internet_access_restricted",
        "network_ssh_internet_access_restricted",
+        "network_subnet_nsg_associated",
        "network_udp_internet_access_restricted",
        "network_watcher_enabled"
      ]
@@ -307,7 +307,8 @@
        "app_minimum_tls_version_12",
        "mysql_flexible_server_minimum_tls_version_12",
        "sqlserver_recommended_minimal_tls_version",
-        "storage_ensure_minimum_tls_version_12"
+        "storage_ensure_minimum_tls_version_12",
+        "network_subnet_nsg_associated"
      ]
    },
    {
@@ -17,6 +17,7 @@ class Network(AzureService):
        self.bastion_hosts = self._get_bastion_hosts()
        self.network_watchers = self._get_network_watchers()
        self.public_ip_addresses = self._get_public_ip_addresses()
+        self.virtual_networks = self._get_virtual_networks()

    def _get_security_groups(self):
        logger.info("Network - Getting Network Security Groups...")
@@ -203,6 +204,38 @@ class Network(AzureService):
                )
        return public_ip_addresses

+    def _get_virtual_networks(self):
+        logger.info("Network - Getting Virtual Networks...")
+        virtual_networks = {}
+        for subscription, client in self.clients.items():
+            try:
+                virtual_networks[subscription] = []
+                vnet_list = client.virtual_networks.list_all()
+                for vnet in vnet_list:
+                    subnets = []
+                    for subnet in getattr(vnet, "subnets", []) or []:
+                        nsg = getattr(subnet, "network_security_group", None)
+                        subnets.append(
+                            VNetSubnet(
+                                id=subnet.id,
+                                name=subnet.name,
+                                nsg_id=getattr(nsg, "id", None) if nsg else None,
+                            )
+                        )
+                    virtual_networks[subscription].append(
+                        VirtualNetwork(
+                            id=vnet.id,
+                            name=vnet.name,
+                            location=vnet.location,
+                            subnets=subnets,
+                        )
+                    )
+            except Exception as error:
+                logger.error(
+                    f"Subscription name: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        return virtual_networks
+

@dataclass
 class BastionHost:
@@ -261,3 +294,22 @@ class PublicIp:
    name: str
    location: str
    ip_address: str
+
+
+@dataclass
+class VNetSubnet:
+    id: str
+    name: str
+    nsg_id: Optional[str] = None
+
+
+@dataclass
+class VirtualNetwork:
+    id: str
+    name: str
+    location: str
+    subnets: List[VNetSubnet] = None
+
+    def __post_init__(self):
+        if self.subnets is None:
+            self.subnets = []
@@ -0,0 +1,37 @@
+{
+  "Provider": "azure",
+  "CheckID": "network_subnet_nsg_associated",
+  "CheckTitle": "Subnet has a network security group associated",
+  "CheckType": [],
+  "ServiceName": "network",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "microsoft.network/virtualnetworks/subnets",
+  "ResourceGroup": "network",
+  "Description": "**Azure Virtual Network** subnets are evaluated for **Network Security Group (NSG)** association. Each subnet should have an NSG to enforce inbound and outbound traffic filtering rules. Subnets without NSGs allow all traffic by default.",
+  "Risk": "Subnets without NSGs have **no network-level access control**. All inbound and outbound traffic is allowed by default, enabling **lateral movement**, **unauthorized access** to resources, and **data exfiltration** across the virtual network.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://learn.microsoft.com/en-us/azure/virtual-network/network-security-groups-overview",
+    "https://learn.microsoft.com/en-us/azure/virtual-network/tutorial-filter-network-traffic"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "az network vnet subnet update --resource-group <rg> --vnet-name <vnet> --name <subnet> --network-security-group <nsg_name>",
+      "NativeIaC": "```bicep\nresource vnet 'Microsoft.Network/virtualNetworks@2023-09-01' existing = {\n  name: '<example_resource_name>'\n}\n\nresource subnet 'Microsoft.Network/virtualNetworks/subnets@2023-09-01' = {\n  name: '<example_resource_name>'\n  parent: vnet\n  properties: {\n    addressPrefix: '10.0.1.0/24'\n    networkSecurityGroup: {\n      id: '<example_resource_id>' // Critical: associates NSG with subnet\n    }\n  }\n}\n```",
+      "Other": "1. Sign in to Azure portal\n2. Go to Virtual networks and select the VNet\n3. Click on Subnets\n4. Select the subnet without an NSG\n5. Under Network security group, select an existing NSG or create a new one\n6. Click Save",
+      "Terraform": "```hcl\nresource \"azurerm_subnet_network_security_group_association\" \"<example_resource_name>\" {\n  subnet_id                 = \"<example_resource_id>\"  # Critical: associates NSG with subnet\n  network_security_group_id = \"<example_resource_id>\"\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Associate a **Network Security Group** with every subnet. Create NSG rules following **least privilege** \u2014 deny all by default and allow only required traffic. Exclude Azure-managed subnets (GatewaySubnet, AzureFirewallSubnet, AzureBastionSubnet) which have their own security controls.",
+      "Url": "https://hub.prowler.com/check/network_subnet_nsg_associated"
+    }
+  },
+  "Categories": [
+    "internet-exposed"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "This check excludes Azure-managed subnets (GatewaySubnet, AzureFirewallSubnet, AzureFirewallManagementSubnet, AzureBastionSubnet, RouteServerSubnet) which should not have custom NSGs."
+}
@@ -0,0 +1,54 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.network.network_client import network_client
+
+# Subnets that are managed by Azure and should not have custom NSGs
+EXCLUDED_SUBNET_NAMES = {
+    "GatewaySubnet",
+    "AzureFirewallSubnet",
+    "AzureFirewallManagementSubnet",
+    "AzureBastionSubnet",
+    "RouteServerSubnet",
+}
+
+
+class network_subnet_nsg_associated(Check):
+    """
+    Ensure every subnet has a Network Security Group (NSG) associated.
+
+    This check evaluates whether each subnet in every virtual network has an NSG associated to enforce inbound and outbound traffic filtering. Azure-managed subnets (e.g. GatewaySubnet, AzureFirewallSubnet, AzureBastionSubnet) are excluded because they must not have custom NSGs.
+
+    - PASS: The subnet has an NSG associated.
+    - FAIL: The subnet does not have an NSG associated.
+    """
+
+    def execute(self) -> Check_Report_Azure:
+        findings = []
+
+        for subscription_name, vnets in network_client.virtual_networks.items():
+            for vnet in vnets:
+                for subnet in vnet.subnets:
+                    if subnet.name in EXCLUDED_SUBNET_NAMES:
+                        continue
+
+                    report = Check_Report_Azure(metadata=self.metadata(), resource=vnet)
+                    report.subscription = subscription_name
+                    report.resource_name = f"{vnet.name}/{subnet.name}"
+                    report.resource_id = subnet.id
+                    report.location = vnet.location
+
+                    if subnet.nsg_id:
+                        report.status = "PASS"
+                        report.status_extended = (
+                            f"Subnet '{subnet.name}' in VNet '{vnet.name}' "
+                            f"has an NSG associated."
+                        )
+                    else:
+                        report.status = "FAIL"
+                        report.status_extended = (
+                            f"Subnet '{subnet.name}' in VNet '{vnet.name}' "
+                            f"does not have an NSG associated."
+                        )
+
+                    findings.append(report)
+
+        return findings
@@ -0,0 +1,188 @@
+from unittest import mock
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+VNET_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/rg1/providers/Microsoft.Network/virtualNetworks/test-vnet"
+SUBNET_ID = f"{VNET_ID}/subnets/test-subnet"
+NSG_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/rg1/providers/Microsoft.Network/networkSecurityGroups/test-nsg"
+
+
+class Test_network_subnet_nsg_associated:
+    def test_no_subscriptions(self):
+        network_client = mock.MagicMock
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated.network_client",
+                new=network_client,
+            ),
+        ):
+            from prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated import (
+                network_subnet_nsg_associated,
+            )
+
+            network_client.virtual_networks = {}
+
+            check = network_subnet_nsg_associated()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_subnet_with_nsg(self):
+        network_client = mock.MagicMock
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated.network_client",
+                new=network_client,
+            ),
+        ):
+            from prowler.providers.azure.services.network.network_service import (
+                VirtualNetwork,
+                VNetSubnet,
+            )
+            from prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated import (
+                network_subnet_nsg_associated,
+            )
+
+            vnet = VirtualNetwork(
+                id=VNET_ID,
+                name="test-vnet",
+                location="eastus",
+                subnets=[VNetSubnet(id=SUBNET_ID, name="test-subnet", nsg_id=NSG_ID)],
+            )
+            network_client.virtual_networks = {AZURE_SUBSCRIPTION_ID: [vnet]}
+
+            check = network_subnet_nsg_associated()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+
+    def test_subnet_without_nsg(self):
+        network_client = mock.MagicMock
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated.network_client",
+                new=network_client,
+            ),
+        ):
+            from prowler.providers.azure.services.network.network_service import (
+                VirtualNetwork,
+                VNetSubnet,
+            )
+            from prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated import (
+                network_subnet_nsg_associated,
+            )
+
+            vnet = VirtualNetwork(
+                id=VNET_ID,
+                name="test-vnet",
+                location="eastus",
+                subnets=[VNetSubnet(id=SUBNET_ID, name="app-subnet", nsg_id=None)],
+            )
+            network_client.virtual_networks = {AZURE_SUBSCRIPTION_ID: [vnet]}
+
+            check = network_subnet_nsg_associated()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+
+    def test_gateway_subnet_excluded(self):
+        network_client = mock.MagicMock
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated.network_client",
+                new=network_client,
+            ),
+        ):
+            from prowler.providers.azure.services.network.network_service import (
+                VirtualNetwork,
+                VNetSubnet,
+            )
+            from prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated import (
+                network_subnet_nsg_associated,
+            )
+
+            vnet = VirtualNetwork(
+                id=VNET_ID,
+                name="test-vnet",
+                location="eastus",
+                subnets=[
+                    VNetSubnet(
+                        id=f"{VNET_ID}/subnets/GatewaySubnet",
+                        name="GatewaySubnet",
+                        nsg_id=None,
+                    )
+                ],
+            )
+            network_client.virtual_networks = {AZURE_SUBSCRIPTION_ID: [vnet]}
+
+            check = network_subnet_nsg_associated()
+            result = check.execute()
+            # GatewaySubnet should be excluded
+            assert len(result) == 0
+
+    def test_mixed_subnets(self):
+        network_client = mock.MagicMock
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated.network_client",
+                new=network_client,
+            ),
+        ):
+            from prowler.providers.azure.services.network.network_service import (
+                VirtualNetwork,
+                VNetSubnet,
+            )
+            from prowler.providers.azure.services.network.network_subnet_nsg_associated.network_subnet_nsg_associated import (
+                network_subnet_nsg_associated,
+            )
+
+            vnet = VirtualNetwork(
+                id=VNET_ID,
+                name="test-vnet",
+                location="eastus",
+                subnets=[
+                    VNetSubnet(id=f"{VNET_ID}/subnets/app", name="app", nsg_id=NSG_ID),
+                    VNetSubnet(id=f"{VNET_ID}/subnets/db", name="db", nsg_id=None),
+                    VNetSubnet(
+                        id=f"{VNET_ID}/subnets/GatewaySubnet",
+                        name="GatewaySubnet",
+                        nsg_id=None,
+                    ),
+                ],
+            )
+            network_client.virtual_networks = {AZURE_SUBSCRIPTION_ID: [vnet]}
+
+            check = network_subnet_nsg_associated()
+            result = check.execute()
+            # 2 results: app (PASS) + db (FAIL), GatewaySubnet excluded
+            assert len(result) == 2
+            statuses = {r.status for r in result}
+            assert "PASS" in statuses
+            assert "FAIL" in statuses