From de986c98f9da2cb0a94389a1a20a61667a2a23d2 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Sun, 5 Jul 2026 19:23:31 +0100 Subject: [PATCH] fix(scanner): resolve COR-001-004 scanner correctness issues (#151) --- scanner/azure_client.py | 14 ++++++-- scanner/rules/az_db_002.py | 20 ++++++++--- scanner/rules/az_net_003.py | 15 ++++++-- tests/helpers/mock_azure.py | 31 ++++++++++++++-- tests/test_rules_database.py | 57 ++++++++++++++++++++++++++++- tests/test_rules_network.py | 70 +++++++++++++++++++++++++++++++++++- 6 files changed, 191 insertions(+), 16 deletions(-) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index 52df367..3547283 100644 --- a/scanner/azure_client.py +++ b/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] diff --git a/scanner/rules/az_db_002.py b/scanner/rules/az_db_002.py index 7edba32..cdd66df 100644 --- a/scanner/rules/az_db_002.py +++ b/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({ diff --git a/scanner/rules/az_net_003.py b/scanner/rules/az_net_003.py index a0a18e0..db5b6ab 100644 --- a/scanner/rules/az_net_003.py +++ b/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({ diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index ed18ef5..e84a813 100644 --- a/tests/helpers/mock_azure.py +++ b/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] diff --git a/tests/test_rules_database.py b/tests/test_rules_database.py index e6a214f..9016522 100644 --- a/tests/test_rules_database.py +++ b/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 == [] diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index 8215c4d..83ce2ec 100644 --- a/tests/test_rules_network.py +++ b/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