fix(scanner): resolve COR-001-004 scanner correctness issues (#151)

scanner/azure_client.py

@@ -41,9 +41,17 @@ def __init__(
 
     @staticmethod
     def parse_resource_id(resource_id: str) -> Dict[str, str]:
-        """Return resource_group and name parsed from an Azure resource ID."""
-        parts = resource_id.split("/")
-        result: Dict[str, str] = {"name": parts[-1] if parts else ""}
+        """Return resource_group and name parsed from an Azure resource ID.
+
+        Always returns both keys, even for malformed or empty IDs, so
+        callers can safely use parsed["resource_group"] without risking
+        a KeyError.
+        """
+        parts = (resource_id or "").split("/")
+        result: Dict[str, str] = {
+            "name": parts[-1] if parts else "",
+            "resource_group": "",
+        }
         for idx, segment in enumerate(parts):
             if segment.lower() == "resourcegroups" and idx + 1 < len(parts):
                 result["resource_group"] = parts[idx + 1]

scanner/rules/az_db_002.py

@@ -1,5 +1,6 @@
 """AZ-DB-002: Azure SQL server has no auditing configured."""
 
+import logging
 from typing import Any, Dict, List
 
 RULE_ID = "AZ-DB-002"
@@ -19,6 +20,8 @@
 )
 PLAYBOOK = "playbooks/cli/fix_az_db_002.sh"
 
+logger = logging.getLogger(__name__)
+
 
 def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
     """Detect SQL servers where server-level blob auditing is disabled."""
@@ -32,11 +35,18 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
 
         policy = azure_client.get_sql_server_auditing_policy(resource_group, server.name)
         if policy is None:
-            # Could not retrieve policy — treat as unaudited
-            is_disabled = True
-        else:
-            state = str(getattr(policy, "state", "Disabled"))
-            is_disabled = state.lower() != "enabled"
+            # Could not retrieve the policy (API/auth failure, throttling, etc).
+            # Skip this resource rather than flagging it — we don't actually
+            # know its auditing state, so treating a failed call as
+            # "disabled" produces a false positive.
+            logger.warning(
+                "Skipping AZ-DB-002 check for %s: could not retrieve auditing policy",
+                server.name,
+            )
+            continue
+
+        state = str(getattr(policy, "state", "Disabled"))
+        is_disabled = state.lower() != "enabled"
 
         if is_disabled:
             findings.append({

scanner/rules/az_net_003.py

@@ -33,10 +33,19 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
 
     for nsg in azure_client.get_network_security_groups():
         for rule in getattr(nsg, "security_rules", []) or []:
+            direction = str(getattr(rule, "direction", "") or "")
+            access = str(getattr(rule, "access", "") or "")
+            allowed_sources = {"*", "0.0.0.0/0", "internet", "any"}
+            single_prefix = str(getattr(rule, "source_address_prefix", "") or "")
+            plural_prefixes = getattr(rule, "source_address_prefixes", None) or []
+            source_matches = single_prefix.lower() in allowed_sources or any(
+                str(prefix or "").lower() in allowed_sources for prefix in plural_prefixes
+            )
+
             if (
-                getattr(rule, "direction", "") == "Inbound"
-                and getattr(rule, "access", "") == "Allow"
-                and getattr(rule, "source_address_prefix", "") in ("*", "0.0.0.0/0", "Internet", "Any")
+                direction.lower() == "inbound"
+                and access.lower() == "allow"
+                and source_matches
                 and getattr(rule, "destination_port_range", "") in ("443", "*")
             ):
                 findings.append({

tests/helpers/mock_azure.py

@@ -35,6 +35,7 @@ def __init__(self) -> None:
         self._sql_firewall_rules: Dict[Tuple[str, str], List[Any]] = {}
         self._diagnostic_settings: Dict[str, Optional[bool]] = {}
         self._key_vault_certificates: Dict[str, List[Any]] = {}
+        self._sql_auditing_policies: Dict[Tuple[str, str], Any] = {}
 
     def set_storage_accounts(self, accounts: List[Any]) -> "MockAzureClient":
         self._storage_accounts = accounts
@@ -74,6 +75,17 @@ def set_key_vault_certificates(self, vault_name: str, certificates: List[Any]) -
         self._key_vault_certificates[vault_name] = certificates
         return self
 
+    def set_sql_server_auditing_policy(
+        self, resource_group: str, server_name: str, policy: Optional[Any]
+    ) -> "MockAzureClient":
+        """Configure the auditing policy returned for a server.
+
+        Pass None to simulate an API failure (e.g. throttling, auth error),
+        distinct from a real policy object with state="Disabled".
+        """
+        self._sql_auditing_policies[(resource_group, server_name)] = policy
+        return self
+
     def get_storage_accounts(self) -> List[Any]:
         return self._storage_accounts
 
@@ -103,11 +115,24 @@ def get_diagnostic_settings(self, resource_id: str) -> Optional[bool]:
     def get_key_vault_certificates(self, vault_name: str) -> List[Any]:
         return self._key_vault_certificates.get(vault_name, [])
 
+    def get_sql_server_auditing_policy(
+        self, resource_group: str, server_name: str
+    ) -> Optional[Any]:
+        return self._sql_auditing_policies.get((resource_group, server_name))
+
     @staticmethod
     def parse_resource_id(resource_id: str) -> Dict[str, str]:
-        """Parse an Azure resource ID into a dict with name and resource_group."""
-        parts = resource_id.split("/")
-        result: Dict[str, str] = {"name": parts[-1] if parts else ""}
+        """Parse an Azure resource ID into a dict with name and resource_group.
+
+        Always returns both keys, even for malformed or empty IDs, so
+        callers can safely use parsed["resource_group"] without risking
+        a KeyError.
+        """
+        parts = (resource_id or "").split("/")
+        result: Dict[str, str] = {
+            "name": parts[-1] if parts else "",
+            "resource_group": "",
+        }
         for idx, segment in enumerate(parts):
             if segment.lower() == "resourcegroups" and idx + 1 < len(parts):
                 result["resource_group"] = parts[idx + 1]

tests/test_rules_database.py

@@ -1,5 +1,6 @@
-"""Rule regression tests for AZ-DB-004."""
+"""Rule regression tests for AZ-DB-002 and AZ-DB-004."""
 
+import scanner.rules.az_db_002 as az_db_002
 import scanner.rules.az_db_004 as az_db_004
 from tests.helpers.mock_azure import make_resource
 
@@ -62,3 +63,57 @@ def test_db_004_no_firewall_rules_returns_no_findings(mock_azure, subscription_i
     mock_azure.set_sql_server_firewall_rules(_RG, "sql-no-rules", [])
     findings = az_db_004.scan(mock_azure, subscription_id)
     assert findings == []
+
+
+def test_db_002_disabled_policy_returns_one_finding(mock_azure, subscription_id):
+    """A SQL Server with auditing explicitly disabled must produce one finding."""
+    server = make_resource(id=_sql_id("sql-unaudited"), name="sql-unaudited")
+    policy = make_resource(state="Disabled")
+    mock_azure.set_sql_servers([server])
+    mock_azure.set_sql_server_auditing_policy(_RG, "sql-unaudited", policy)
+    findings = az_db_002.scan(mock_azure, subscription_id)
+    assert len(findings) == 1
+    assert findings[0]["rule_id"] == "AZ-DB-002"
+
+
+def test_db_002_enabled_policy_returns_no_findings(mock_azure, subscription_id):
+    """A SQL Server with auditing enabled must produce no findings."""
+    server = make_resource(id=_sql_id("sql-audited"), name="sql-audited")
+    policy = make_resource(state="Enabled")
+    mock_azure.set_sql_servers([server])
+    mock_azure.set_sql_server_auditing_policy(_RG, "sql-audited", policy)
+    findings = az_db_002.scan(mock_azure, subscription_id)
+    assert findings == []
+
+
+def test_db_002_api_failure_returns_no_findings(mock_azure, subscription_id):
+    """COR-003: a failed policy lookup (None) must be skipped, not flagged.
+
+    Previously, a None policy (API/auth failure) was treated the same as an
+    explicitly disabled policy, producing a false positive.
+    """
+    server = make_resource(id=_sql_id("sql-api-failed"), name="sql-api-failed")
+    mock_azure.set_sql_servers([server])
+    mock_azure.set_sql_server_auditing_policy(_RG, "sql-api-failed", None)
+    findings = az_db_002.scan(mock_azure, subscription_id)
+    assert findings == []
+
+
+def test_db_002_malformed_arm_id_does_not_raise(mock_azure, subscription_id):
+    """COR-004: a server with a malformed ARM ID must not raise an error."""
+    server = make_resource(id="not-a-valid-arm-id", name="sql-malformed")
+    mock_azure.set_sql_servers([server])
+    findings = az_db_002.scan(mock_azure, subscription_id)
+    assert findings == []
+
+
+def test_db_004_malformed_arm_id_does_not_raise_keyerror(mock_azure, subscription_id):
+    """COR-004: az_db_004 indexes parsed["resource_group"] directly, so a
+    malformed ARM ID with no resourceGroups segment previously raised a
+    KeyError. parse_resource_id must always include the key.
+    """
+    server = make_resource(id="not-a-valid-arm-id", name="sql-malformed")
+    mock_azure.set_sql_servers([server])
+    mock_azure.set_sql_server_firewall_rules("", "sql-malformed", [])
+    findings = az_db_004.scan(mock_azure, subscription_id)
+    assert findings == []

tests/test_rules_network.py

@@ -1,7 +1,8 @@
-"""Rule regression tests for AZ-NET-001 and AZ-NET-002."""
+"""Rule regression tests for AZ-NET-001, AZ-NET-002, and AZ-NET-003."""
 
 import scanner.rules.az_net_001 as az_net_001
 import scanner.rules.az_net_002 as az_net_002
+import scanner.rules.az_net_003 as az_net_003
 from tests.helpers.mock_azure import make_resource
 
 _REQUIRED_FIELDS = {
@@ -103,3 +104,70 @@ def test_net_002_noncompliant_returns_one_finding(mock_azure, subscription_id):
     assert finding["severity"] == "HIGH"
     assert finding["category"] == "Network"
     assert finding["resource_name"] == "nsg-rdp-open"
+
+
+def _net_003_rule(name, direction="Inbound", access="Allow",
+                   source="0.0.0.0/0", source_list=None, port="443"):
+    return make_resource(
+        name=name,
+        direction=direction,
+        access=access,
+        source_address_prefix=source,
+        source_address_prefixes=source_list or [],
+        destination_port_range=port,
+    )
+
+
+def test_net_003_noncompliant_returns_one_finding(mock_azure, subscription_id):
+    """An NSG with Allow-inbound-443-from-any must produce exactly one finding."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-443-open"),
+        name="nsg-443-open",
+        security_rules=[_net_003_rule("AllowHTTPSFromInternet")],
+    )
+    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"
+
+
+def test_net_003_compliant_returns_no_findings(mock_azure, subscription_id):
+    """An NSG restricting 443 to a trusted IP range must produce no findings."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-443-restricted"),
+        name="nsg-443-restricted",
+        security_rules=[_net_003_rule("AllowHTTPSFromTrusted", source="10.0.0.0/24")],
+    )
+    mock_azure.set_network_security_groups([nsg])
+    findings = az_net_003.scan(mock_azure, subscription_id)
+    assert findings == []
+
+
+def test_net_003_direction_is_case_insensitive(mock_azure, subscription_id):
+    """COR-001: lowercase/mixed-case direction values must still be detected."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-lowercase-direction"),
+        name="nsg-lowercase-direction",
+        security_rules=[_net_003_rule("AllowHTTPSLower", direction="inbound")],
+    )
+    mock_azure.set_network_security_groups([nsg])
+    findings = az_net_003.scan(mock_azure, subscription_id)
+    assert len(findings) == 1
+
+
+def test_net_003_detects_plural_source_prefixes(mock_azure, subscription_id):
+    """COR-002: an open source listed only in source_address_prefixes must be detected."""
+    nsg = make_resource(
+        id=_nsg_id("nsg-plural-prefix"),
+        name="nsg-plural-prefix",
+        security_rules=[
+            _net_003_rule(
+                "AllowHTTPSPluralOpen",
+                source="",
+                source_list=["0.0.0.0/0"],
+            )
+        ],
+    )
+    mock_azure.set_network_security_groups([nsg])
+    findings = az_net_003.scan(mock_azure, subscription_id)
+    assert len(findings) == 1