diff --git a/scanner/rules/az_db_002.py b/scanner/rules/az_db_002.py index 67b697b3..79cc4a2b 100644 --- a/scanner/rules/az_db_002.py +++ b/scanner/rules/az_db_002.py @@ -22,6 +22,25 @@ ) PLAYBOOK = "playbooks/cli/fix_az_db_002.sh" +# A failed auditing-policy lookup is reported as an unknown scan result, not +# a confirmed violation: it says nothing about whether auditing is actually +# enabled, so it must not carry the standard MEDIUM severity or remediation +# (which could send someone to re-configure auditing that was already +# compliant). This mirrors the indeterminate-finding convention used by +# AZ-CMP-002 for unreadable disks. +INDETERMINATE_SEVERITY = "LOW" +INDETERMINATE_DESCRIPTION = ( + "The auditing policy for this Azure SQL Server could not be retrieved, so its " + "auditing configuration could not be verified. This is not a confirmed " + "violation — the scanning principal could not resolve the server's auditing " + "policy (missing permissions, transient API failure, or throttling)." +) +INDETERMINATE_REMEDIATION = ( + "Grant the scanning principal permission to read the SQL Server's auditing " + "policy (Microsoft.Sql/servers/auditingSettings/read) and re-run the scan " + "to determine the actual auditing state." +) + logger = logging.getLogger(__name__) @@ -43,13 +62,32 @@ 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 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. + # We don't know the actual auditing state, so this is reported as + # an indeterminate LOW finding rather than a confirmed violation + # or a silent skip — the gap stays visible in scan output. logger.warning( - "Skipping AZ-DB-002 check for %s: could not retrieve auditing policy", + "AZ-DB-002: could not retrieve auditing policy for %s, marking indeterminate", server.name, ) + findings.append( + { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": INDETERMINATE_SEVERITY, + "category": CATEGORY, + "resource_id": server.id, + "resource_name": server.name, + "resource_type": "Microsoft.Sql/servers", + "description": INDETERMINATE_DESCRIPTION, + "remediation": INDETERMINATE_REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": resource_group, + "determination": "indeterminate", + }, + } + ) continue state = enum_str(getattr(policy, "state", None), default="Disabled") @@ -72,6 +110,7 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: "metadata": { "resource_group": resource_group, "auditing_state": state, + "determination": "non_compliant", }, } ) diff --git a/scanner/rules/az_net_003.py b/scanner/rules/az_net_003.py index f6a042b5..7d1a5800 100644 --- a/scanner/rules/az_net_003.py +++ b/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,21 +40,40 @@ 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"} + + 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), None, ) + source_matches = single_prefix.lower() in allowed_sources or matched_plural_prefix is not None - if ( - direction.lower() == "inbound" - and access.lower() == "allow" - and source_matches - and getattr(rule, "destination_port_range", "") in ("443", "*") - ): + # 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 port_matches: findings.append( { "rule_id": RULE_ID, @@ -67,7 +89,7 @@ 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, }, } diff --git a/tests/test_rules_database.py b/tests/test_rules_database.py index a89d2400..e2fd01b4 100644 --- a/tests/test_rules_database.py +++ b/tests/test_rules_database.py @@ -103,17 +103,20 @@ def test_db_002_enabled_policy_returns_no_findings(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. +def test_db_002_api_failure_returns_indeterminate_finding(mock_azure, subscription_id): + """TFT444 review: a failed policy lookup (None) must produce an + indeterminate LOW finding (metadata.determination == "indeterminate"), + matching the AZ-CMP-002 convention — not a silent skip and not a + confirmed MEDIUM violation, since the actual auditing state is unknown. """ 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 == [] + assert len(findings) == 1 + assert findings[0]["rule_id"] == "AZ-DB-002" + assert findings[0]["severity"] == "LOW" + assert findings[0]["metadata"]["determination"] == "indeterminate" def test_db_002_malformed_arm_id_does_not_raise(mock_azure, subscription_id): diff --git a/tests/test_rules_identity.py b/tests/test_rules_identity.py index 3c88f83d..e27293cf 100644 --- a/tests/test_rules_identity.py +++ b/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 ────────────────── diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index 8a541f15..cc4e4d85 100644 --- a/tests/test_rules_network.py +++ b/tests/test_rules_network.py @@ -160,7 +160,15 @@ 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, @@ -168,6 +176,7 @@ def _net_003_rule(name, direction="Inbound", access="Allow", source="0.0.0.0/0", 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,