Fix/az net 003 az db 002 scanner correctness

scanner/rules/az_net_003.py

@@ -10,6 +10,7 @@
 SEVERITY = "HIGH"
 CATEGORY = "Network"
 FRAMEWORKS = {"CIS": "9.3", "NIST": "SC-7", "ISO27001": "A.13.1.1"}
+
 DESCRIPTION = (
     "A Network Security Group has an inbound rule allowing unrestricted access "
     "on port 443 from any source (0.0.0.0/0). While HTTPS traffic is encrypted, "
@@ -19,11 +20,13 @@
     "Review manually before remediating — do not auto-remediate without confirming "
     "the service is not meant to be publicly accessible."
 )
+
 REMEDIATION = (
     "Restrict the inbound rule on port 443 to known IP ranges or use an "
     "Application Gateway with WAF to front any public-facing HTTPS services. "
     "If the service must be public, ensure it is protected by DDoS Standard."
 )
+
 PLAYBOOK = "playbooks/cli/fix_az_net_003.sh"
 
 logger = logging.getLogger(__name__)
@@ -37,20 +40,65 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
         for rule in getattr(nsg, "security_rules", []) or []:
             direction = enum_str(getattr(rule, "direction", None))
             access = enum_str(getattr(rule, "access", None))
-            allowed_sources = {"*", "0.0.0.0/0", "internet", "any"}
-            single_prefix = enum_str(getattr(rule, "source_address_prefix", None))
-            plural_prefixes = getattr(rule, "source_address_prefixes", None) or []
+
+            allowed_sources = {
+                "*",
+                "0.0.0.0/0",
+                "internet",
+                "any",
+            }
+
+            # Azure can expose the source as either a single prefix
+            # or a list of prefixes.
+            single_prefix = enum_str(
+                getattr(rule, "source_address_prefix", None)
+            )
+
+            plural_prefixes = (
+                getattr(rule, "source_address_prefixes", None) or []
+            )
+
             matched_plural_prefix = next(
-                (prefix for prefix in plural_prefixes if enum_str(prefix).lower() in allowed_sources),
+                (
+                    prefix
+                    for prefix in plural_prefixes
+                    if enum_str(prefix).lower() in allowed_sources
+                ),
                 None,
             )
-            source_matches = single_prefix.lower() in allowed_sources or matched_plural_prefix is not None
+
+            source_matches = (
+                single_prefix.lower() in allowed_sources
+                or matched_plural_prefix is not None
+            )
+
+            # Azure can expose the destination port as either a single
+            # port/range or a list of ports/ranges.
+            destination_port_range = enum_str(
+                getattr(rule, "destination_port_range", None)
+            )
+
+            destination_port_ranges = (
+                getattr(rule, "destination_port_ranges", None) or []
+            )
+
+            destination_port_ranges = [
+                enum_str(port) for port in destination_port_ranges
+            ]
+
+            port_matches = (
+                destination_port_range in ("443", "*")
+                or any(
+                    port in ("443", "*")
+                    for port in destination_port_ranges
+                )
+            )
 
             if (
                 direction.lower() == "inbound"
                 and access.lower() == "allow"
                 and source_matches
-                and getattr(rule, "destination_port_range", "") in ("443", "*")
+                and port_matches
             ):
                 findings.append(
                     {
@@ -67,7 +115,11 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
                         "frameworks": FRAMEWORKS,
                         "metadata": {
                             "rule_name": getattr(rule, "name", ""),
-                            "source_prefix": single_prefix if single_prefix.lower() in allowed_sources else "",
+                            "source_prefix": (
+                                single_prefix
+                                if single_prefix.lower() in allowed_sources
+                                else ""
+                            ),
                             "matched_source_address_prefix": matched_plural_prefix,
                         },
                     }

tests/test_rules_identity.py

@@ -351,6 +351,25 @@ def test_idn_007_noncompliant_user_without_mfa_returns_finding(mock_azure, subsc
     assert findings[0]["resource_name"] == "No MFA User"
 
 
+def test_idn_007_disabled_user_without_mfa_returns_no_findings(mock_azure, subscription_id, monkeypatch):
+    """A disabled user account without MFA registered must not be flagged —
+    the rule targets active users only, since a disabled account cannot be
+    used to sign in regardless of its MFA state."""
+    regs = {
+        "value": [
+            {
+                "id": "u2",
+                "userDisplayName": "Disabled No MFA User",
+                "userPrincipalName": "disabled@x.com",
+                "isEnabled": False,
+                "isMfaRegistered": False,
+            }
+        ]
+    }
+    _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))])
+    assert az_idn_007.scan(mock_azure, subscription_id) == []
+
+
 # ── AZ-IDN-008: custom RBAC role with wildcard permissions ──────────────────
 
 

tests/test_rules_network.py

@@ -160,14 +160,23 @@ def _vnet_id(name):
     return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/virtualNetworks/{name}"
 
 
-def _net_003_rule(name, direction="Inbound", access="Allow", source="0.0.0.0/0", source_list=None, port="443"):
+def _net_003_rule(
+    name,
+    direction="Inbound",
+    access="Allow",
+    source="0.0.0.0/0",
+    source_list=None,
+    port="443",
+    port_list=None,
+):
     return make_resource(
         name=name,
         direction=direction,
         access=access,
         source_address_prefix=source,
         source_address_prefixes=source_list or [],
         destination_port_range=port,
+        destination_port_ranges=port_list or [],
     )
 
 
@@ -227,6 +236,48 @@ def test_net_003_detects_plural_source_prefixes(mock_azure, subscription_id):
     assert len(findings) == 1
 
 
+def test_net_003_detects_plural_destination_port_ranges(mock_azure, subscription_id):
+    """COR-003: port 443 listed only in destination_port_ranges must be detected."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-plural-port"),
+        name="nsg-plural-port",
+        security_rules=[
+            _net_003_rule(
+                "AllowHTTPSPluralPort",
+                source="0.0.0.0/0",
+                port="",
+                port_list=["443"],
+            )
+        ],
+    )
+    mock_azure.set_network_security_groups([nsg])
+    findings = az_net_003.scan(mock_azure, subscription_id)
+    assert len(findings) == 1
+    assert findings[0]["rule_id"] == "AZ-NET-003"
+    assert findings[0]["severity"] == "HIGH"
+
+
+def test_net_003_compliant_plural_destination_port_ranges(mock_azure, subscription_id):
+    """Non-blocking (parthrohit22): a rule using destination_port_ranges for
+    ports that don't include 443/* must not be flagged — pins down that the
+    plural-port fix only broadens detection for 443/*, not for any port."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-plural-port-safe"),
+        name="nsg-plural-port-safe",
+        security_rules=[
+            _net_003_rule(
+                "AllowOtherPortsPluralOpen",
+                source="0.0.0.0/0",
+                port="",
+                port_list=["80", "8080"],
+            )
+        ],
+    )
+    mock_azure.set_network_security_groups([nsg])
+    findings = az_net_003.scan(mock_azure, subscription_id)
+    assert findings == []
+
+
 @pytest.mark.skipif(not _AZURE_SDK_AVAILABLE, reason="azure-mgmt-network not installed")
 def test_net_003_detects_finding_with_real_sdk_enum_direction_and_access(mock_azure, subscription_id):
     """COR-001 (SDK model): real SecurityRuleDirection/Access enums, not plain strings,