From e53e4c59610d11f7a1e2b880f07fa368dd58a23b Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Sun, 5 Jul 2026 19:23:31 +0100 Subject: [PATCH 1/9] fix(scanner): resolve COR-001-004 scanner correctness issues (#151) --- scanner/rules/az_net_003.py | 2 +- tests/helpers/mock_azure.py | 32 ++++++++++++++++++++++++++++++++ tests/test_rules_database.py | 2 +- tests/test_rules_network.py | 2 +- 4 files changed, 35 insertions(+), 3 deletions(-) diff --git a/scanner/rules/az_net_003.py b/scanner/rules/az_net_003.py index f6a042b5..65b7479a 100644 --- a/scanner/rules/az_net_003.py +++ b/scanner/rules/az_net_003.py @@ -74,4 +74,4 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: ) break - return findings + return findings \ No newline at end of file diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index de3c6be0..58541f00 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -75,6 +75,7 @@ def __init__(self) -> None: self._kv_certificates: Dict[str, List[Any]] = {} self._kv_keys: Dict[str, List[Any]] = {} self._diagnostic_settings: Dict[str, Optional[bool]] = {} +<<<<<<< HEAD self._diagnostic_default: Optional[bool] = False self._conditional_access_policies: List[Any] = [] # Credential stub for Graph/SDK-based rules (idn_003..009). @@ -97,6 +98,10 @@ def __init__(self) -> None: # Some rules read azure_client.subscription_id when constructing an # SDK management client inside scan() (e.g. AZ-NET-007..010). self.subscription_id = "00000000-0000-0000-0000-000000000001" +======= + self._key_vault_certificates: Dict[str, List[Any]] = {} + self._sql_auditing_policies: Dict[Tuple[str, str], Any] = {} +>>>>>>> de986c9 (fix(scanner): resolve COR-001-004 scanner correctness issues (#151)) def set_storage_accounts(self, accounts: List[Any]) -> "MockAzureClient": self._storage_accounts = accounts @@ -207,6 +212,28 @@ def set_sql_server_firewall_rules( self._sql_firewall_rules[(resource_group, server_name)] = rules return self +<<<<<<< HEAD +======= + def set_diagnostic_settings(self, resource_id: str, status: Optional[bool]) -> "MockAzureClient": + self._diagnostic_settings[resource_id] = status + return self + + def set_key_vault_certificates(self, vault_name: str, certificates: List[Any]) -> "MockAzureClient": + 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 + +>>>>>>> de986c9 (fix(scanner): resolve COR-001-004 scanner correctness issues (#151)) def get_storage_accounts(self) -> List[Any]: return self._storage_accounts @@ -434,6 +461,11 @@ def set_web_apps(self, apps: List[Any]) -> "MockAzureClient": def get_web_apps(self) -> List[Any]: return self._web_apps + 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. diff --git a/tests/test_rules_database.py b/tests/test_rules_database.py index a89d2400..4100eb8d 100644 --- a/tests/test_rules_database.py +++ b/tests/test_rules_database.py @@ -251,4 +251,4 @@ def test_db_003_noncompliant_returns_one_finding(mock_azure, subscription_id): assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-DB-003" assert findings[0]["severity"] == "HIGH" - assert findings[0]["resource_name"] == "pgflex-nossl" + assert findings[0]["resource_name"] == "pgflex-nossl" \ No newline at end of file diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index 8a541f15..f73f77a8 100644 --- a/tests/test_rules_network.py +++ b/tests/test_rules_network.py @@ -811,4 +811,4 @@ def test_net_017_direct_internet_default_creates_finding(mock_azure, subscriptio findings = az_net_017.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-NET-017" - assert findings[0]["metadata"]["address_prefix"] == prefix + assert findings[0]["metadata"]["address_prefix"] == prefix \ No newline at end of file From 3f1beaeb1a33b25826764c7e41bd5869e18226a5 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Fri, 10 Jul 2026 22:07:21 +0000 Subject: [PATCH 2/9] fix(scanner): normalise Azure SDK enum fields in AZ-NET-003 and AZ-DB-002 Addresses review feedback on PR #163: - Add a shared enum_str() helper in azure_client.py that safely unwraps Azure SDK enum fields via .value, since str(enum_member) yields e.g. 'SecurityRuleDirection.INBOUND' rather than 'Inbound' and silently breaks naive string comparisons against real SDK objects. - AZ-NET-003: normalise direction, access, and source_address_prefix through enum_str() so real SecurityRuleDirection/SecurityRuleAccess enum values are detected correctly, not just plain-string mocks. - AZ-DB-002: normalise the auditing policy state through enum_str() so a real BlobAuditingPolicyState.ENABLED value is not mistaken for disabled (false positive) or vice versa. - AZ-DB-002: malformed ARM IDs are now logged explicitly instead of silently skipped. - AZ-NET-003: the matched plural source_address_prefixes entry is now included in finding metadata. - Add regression tests using real azure-mgmt-network / azure-mgmt-sql SDK model classes (SecurityRule, SecurityRuleDirection, SecurityRuleAccess, ServerBlobAuditingPolicy, BlobAuditingPolicyState) rather than only SimpleNamespace/string-backed mocks, per SHAURYAKSHARMA24's review. - Sync branch with upstream dev (v0.3.0) and apply current ruff format gate, per ritiksah141's review. --- scanner/azure_client.py | 16 ++++++++++++++++ tests/helpers/mock_azure.py | 4 +--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index d3b1bb21..2366056e 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -23,6 +23,22 @@ _UNSET = object() +def enum_str(value: Any, default: str = "") -> str: + """Safely coerce an Azure SDK field to its plain string form. + + Azure SDK models often return fields typed as enums (e.g. + SecurityRuleDirection, BlobAuditingPolicyState) rather than plain + strings. ``str(enum_member)`` yields something like + "SecurityRuleDirection.INBOUND", not the underlying value "Inbound", + which breaks naive string comparisons. This prefers ``.value`` when + present (covers real SDK enums and enum-like objects) and falls back + to ``str()`` for plain strings, None, or anything else. + """ + if value is None: + return default + return str(getattr(value, "value", value)) + + def enum_str(value: Any, default: str = "") -> str: """Safely coerce an Azure SDK field to its plain string form. diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 58541f00..43ea0954 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -461,9 +461,7 @@ def set_web_apps(self, apps: List[Any]) -> "MockAzureClient": def get_web_apps(self) -> List[Any]: return self._web_apps - def get_sql_server_auditing_policy( - self, resource_group: str, server_name: str - ) -> Optional[Any]: + 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 From b87b603d51155be04aafcf4a5e374f6d65773d68 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Mon, 13 Jul 2026 21:29:51 +0100 Subject: [PATCH 3/9] fix: remove duplicate _diagnostic_settings init line --- tests/helpers/mock_azure.py | 32 +------------------------------- 1 file changed, 1 insertion(+), 31 deletions(-) diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 43ea0954..a255aae5 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -75,7 +75,6 @@ def __init__(self) -> None: self._kv_certificates: Dict[str, List[Any]] = {} self._kv_keys: Dict[str, List[Any]] = {} self._diagnostic_settings: Dict[str, Optional[bool]] = {} -<<<<<<< HEAD self._diagnostic_default: Optional[bool] = False self._conditional_access_policies: List[Any] = [] # Credential stub for Graph/SDK-based rules (idn_003..009). @@ -98,10 +97,6 @@ def __init__(self) -> None: # Some rules read azure_client.subscription_id when constructing an # SDK management client inside scan() (e.g. AZ-NET-007..010). self.subscription_id = "00000000-0000-0000-0000-000000000001" -======= - self._key_vault_certificates: Dict[str, List[Any]] = {} - self._sql_auditing_policies: Dict[Tuple[str, str], Any] = {} ->>>>>>> de986c9 (fix(scanner): resolve COR-001-004 scanner correctness issues (#151)) def set_storage_accounts(self, accounts: List[Any]) -> "MockAzureClient": self._storage_accounts = accounts @@ -212,28 +207,6 @@ def set_sql_server_firewall_rules( self._sql_firewall_rules[(resource_group, server_name)] = rules return self -<<<<<<< HEAD -======= - def set_diagnostic_settings(self, resource_id: str, status: Optional[bool]) -> "MockAzureClient": - self._diagnostic_settings[resource_id] = status - return self - - def set_key_vault_certificates(self, vault_name: str, certificates: List[Any]) -> "MockAzureClient": - 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 - ->>>>>>> de986c9 (fix(scanner): resolve COR-001-004 scanner correctness issues (#151)) def get_storage_accounts(self) -> List[Any]: return self._storage_accounts @@ -461,9 +434,6 @@ def set_web_apps(self, apps: List[Any]) -> "MockAzureClient": def get_web_apps(self) -> List[Any]: return self._web_apps - 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. @@ -480,4 +450,4 @@ def parse_resource_id(resource_id: str) -> Dict[str, str]: for idx, segment in enumerate(parts): if segment.lower() == "resourcegroups" and idx + 1 < len(parts): result["resource_group"] = parts[idx + 1] - return result + return result \ No newline at end of file From dbc8688f3ae8c53fce821dda0798081fdd794dce Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Wed, 19 Aug 2026 20:09:03 +0100 Subject: [PATCH 4/9] test: add identity rule regression coverage Signed-off-by: safidnadaf --- tests/test_rules_identity.py | 529 ++++++++++++++++++++++++++++++----- 1 file changed, 454 insertions(+), 75 deletions(-) diff --git a/tests/test_rules_identity.py b/tests/test_rules_identity.py index 3c88f83d..7e3ca321 100644 --- a/tests/test_rules_identity.py +++ b/tests/test_rules_identity.py @@ -72,33 +72,67 @@ def fake_get(url, *args, **kwargs): _SUB = "00000000-0000-0000-0000-000000000001" _OWNER_ROLE_GUID = "8e3af657-a8ff-443c-a75c-2fe8c4bcb635" _CONTRIBUTOR_ROLE_GUID = "b24988ac-6180-42a0-ab88-20f7382dd24c" -_ROLE_DEF_BASE = f"/subscriptions/{_SUB}/providers/Microsoft.Authorization/roleDefinitions" +_ROLE_DEF_BASE = ( + f"/subscriptions/{_SUB}/providers/" + "Microsoft.Authorization/roleDefinitions" +) def _assignment(role_guid, principal_id, assign_id): return make_resource( - id=f"/subscriptions/{_SUB}/providers/Microsoft.Authorization/roleAssignments/{assign_id}", + id=( + f"/subscriptions/{_SUB}/providers/" + f"Microsoft.Authorization/roleAssignments/{assign_id}" + ), role_definition_id=f"{_ROLE_DEF_BASE}/{role_guid}", principal_id=principal_id, scope=f"/subscriptions/{_SUB}", ) -def test_idn_001_compliant_returns_no_findings(mock_azure, subscription_id): +def test_idn_001_compliant_returns_no_findings( + mock_azure, + subscription_id, +): """A service principal with a non-Owner role must produce no findings.""" - assignment = _assignment(_CONTRIBUTOR_ROLE_GUID, "sp-contributor-abc123", "assign-001") + assignment = _assignment( + _CONTRIBUTOR_ROLE_GUID, + "sp-contributor-abc123", + "assign-001", + ) + mock_azure.set_service_principals([assignment]) - findings = az_idn_001.scan(mock_azure, subscription_id) + + findings = az_idn_001.scan( + mock_azure, + subscription_id, + ) + assert findings == [] -def test_idn_001_noncompliant_returns_one_finding(mock_azure, subscription_id): +def test_idn_001_noncompliant_returns_one_finding( + mock_azure, + subscription_id, +): """A service principal holding the Owner role must produce exactly one finding.""" - assignment = _assignment(_OWNER_ROLE_GUID, "sp-owner-def456", "assign-002") + assignment = _assignment( + _OWNER_ROLE_GUID, + "sp-owner-def456", + "assign-002", + ) + mock_azure.set_service_principals([assignment]) - findings = az_idn_001.scan(mock_azure, subscription_id) + + findings = az_idn_001.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 + finding = findings[0] + assert _REQUIRED_FIELDS.issubset(finding.keys()) assert finding["rule_id"] == "AZ-IDN-001" assert finding["severity"] == "HIGH" @@ -113,21 +147,43 @@ def test_idn_001_noncompliant_returns_one_finding(mock_azure, subscription_id): # ── AZ-IDN-002: MFA enforced on admins via Conditional Access ─────────────── -def test_idn_002_compliant_policy_enforces_mfa_returns_no_findings(mock_azure, subscription_id): +def test_idn_002_compliant_policy_enforces_mfa_returns_no_findings( + mock_azure, + subscription_id, +): """A CA policy that is enabled, requires MFA, and covers all users is compliant.""" policy = { "state": "enabled", - "grantControls": {"builtInControls": ["mfa"]}, - "conditions": {"users": {"includeUsers": ["All"]}}, + "grantControls": { + "builtInControls": ["mfa"], + }, + "conditions": { + "users": { + "includeUsers": ["All"], + }, + }, } + mock_azure.set_conditional_access_policies([policy]) - assert az_idn_002.scan(mock_azure, subscription_id) == [] + + assert az_idn_002.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_002_noncompliant_no_policies_returns_one_finding(mock_azure, subscription_id): +def test_idn_002_noncompliant_no_policies_returns_one_finding( + mock_azure, + subscription_id, +): """No Conditional Access policies at all must produce exactly one finding.""" mock_azure.set_conditional_access_policies([]) - findings = az_idn_002.scan(mock_azure, subscription_id) + + findings = az_idn_002.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-002" assert findings[0]["severity"] == "HIGH" @@ -136,24 +192,57 @@ def test_idn_002_noncompliant_no_policies_returns_one_finding(mock_azure, subscr # ── AZ-IDN-003: guest invitations not restricted ─────────────────────────── -def test_idn_003_compliant_restricted_invites_returns_no_findings(mock_azure, subscription_id, monkeypatch): +def test_idn_003_compliant_restricted_invites_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): _install_router( monkeypatch, [ - ("authorizationPolicy", _Resp({"id": "authPol", "allowInvitesFrom": "adminsAndGuestInviters"})), + ( + "authorizationPolicy", + _Resp( + { + "id": "authPol", + "allowInvitesFrom": "adminsAndGuestInviters", + } + ), + ), ], ) - assert az_idn_003.scan(mock_azure, subscription_id) == [] + assert az_idn_003.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding(mock_azure, subscription_id, monkeypatch): + +def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding( + mock_azure, + subscription_id, + monkeypatch, +): _install_router( monkeypatch, [ - ("authorizationPolicy", _Resp({"id": "authPol", "allowInvitesFrom": "everyone"})), + ( + "authorizationPolicy", + _Resp( + { + "id": "authPol", + "allowInvitesFrom": "everyone", + } + ), + ), ], ) - findings = az_idn_003.scan(mock_azure, subscription_id) + + findings = az_idn_003.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-003" assert findings[0]["severity"] == "MEDIUM" @@ -162,9 +251,28 @@ def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding(mock_azure # ── AZ-IDN-004: no PIM for admin roles ────────────────────────────────────── -def test_idn_004_compliant_role_has_pim_returns_no_findings(mock_azure, subscription_id, monkeypatch): - role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} - schedules = {"value": [{"roleDefinitionId": "rd-ga"}]} +def test_idn_004_compliant_role_has_pim_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): + role_defs = { + "value": [ + { + "id": "rd-ga", + "displayName": "Global Administrator", + } + ] + } + + schedules = { + "value": [ + { + "roleDefinitionId": "rd-ga", + } + ] + } + _install_router( monkeypatch, [ @@ -172,20 +280,50 @@ def test_idn_004_compliant_role_has_pim_returns_no_findings(mock_azure, subscrip ("roleEligibilitySchedules", _Resp(schedules)), ], ) - assert az_idn_004.scan(mock_azure, subscription_id) == [] + + assert az_idn_004.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_004_noncompliant_role_without_pim_returns_finding(mock_azure, subscription_id, monkeypatch): - role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} - schedules = {"value": []} +def test_idn_004_noncompliant_role_without_pim_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): + role_defs = { + "value": [ + { + "id": "rd-ga", + "displayName": "Global Administrator", + } + ] + } + + schedules = { + "value": [] + } + _install_router( monkeypatch, [ - ("roleEligibilitySchedules", _Resp(schedules)), # check more specific first - ("roleDefinitions", _Resp(role_defs)), + ( + "roleEligibilitySchedules", + _Resp(schedules), + ), + ( + "roleDefinitions", + _Resp(role_defs), + ), ], ) - findings = az_idn_004.scan(mock_azure, subscription_id) + + findings = az_idn_004.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-004" assert findings[0]["severity"] == "HIGH" @@ -195,23 +333,84 @@ def test_idn_004_noncompliant_role_without_pim_returns_finding(mock_azure, subsc # ── AZ-IDN-005: guest with high-privilege role ────────────────────────────── -def test_idn_005_compliant_member_user_returns_no_findings(mock_azure, subscription_id, monkeypatch): - role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} - assignments = {"value": [{"id": "a1", "roleDefinitionId": "rd-ga", "principalId": "u1"}]} +def test_idn_005_compliant_member_user_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): + role_defs = { + "value": [ + { + "id": "rd-ga", + "displayName": "Global Administrator", + } + ] + } + + assignments = { + "value": [ + { + "id": "a1", + "roleDefinitionId": "rd-ga", + "principalId": "u1", + } + ] + } + _install_router( monkeypatch, [ - ("/users/", _Resp({"id": "u1", "displayName": "Member User", "userType": "Member"})), - ("roleDefinitions", _Resp(role_defs)), - ("roleAssignments", _Resp(assignments)), + ( + "/users/", + _Resp( + { + "id": "u1", + "displayName": "Member User", + "userType": "Member", + } + ), + ), + ( + "roleDefinitions", + _Resp(role_defs), + ), + ( + "roleAssignments", + _Resp(assignments), + ), ], ) - assert az_idn_005.scan(mock_azure, subscription_id) == [] + assert az_idn_005.scan( + mock_azure, + subscription_id, + ) == [] + + +def test_idn_005_noncompliant_guest_admin_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): + role_defs = { + "value": [ + { + "id": "rd-ga", + "displayName": "Global Administrator", + } + ] + } + + assignments = { + "value": [ + { + "id": "a1", + "roleDefinitionId": "rd-ga", + "principalId": "g1", + } + ] + } -def test_idn_005_noncompliant_guest_admin_returns_finding(mock_azure, subscription_id, monkeypatch): - role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} - assignments = {"value": [{"id": "a1", "roleDefinitionId": "rd-ga", "principalId": "g1"}]} _install_router( monkeypatch, [ @@ -226,11 +425,22 @@ def test_idn_005_noncompliant_guest_admin_returns_finding(mock_azure, subscripti } ), ), - ("roleDefinitions", _Resp(role_defs)), - ("roleAssignments", _Resp(assignments)), + ( + "roleDefinitions", + _Resp(role_defs), + ), + ( + "roleAssignments", + _Resp(assignments), + ), ], ) - findings = az_idn_005.scan(mock_azure, subscription_id) + + findings = az_idn_005.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-005" assert findings[0]["severity"] == "HIGH" @@ -240,7 +450,11 @@ def test_idn_005_noncompliant_guest_admin_returns_finding(mock_azure, subscripti # ── AZ-IDN-006: stale / non-expiring SP client secret ─────────────────────── -def test_idn_006_compliant_fresh_secret_returns_no_findings(mock_azure, subscription_id, monkeypatch): +def test_idn_006_compliant_fresh_secret_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): apps = { "value": [ { @@ -262,7 +476,11 @@ def test_idn_006_compliant_fresh_secret_returns_no_findings(mock_azure, subscrip assert az_idn_006.scan(mock_azure, subscription_id) == [] -def test_idn_006_noncompliant_secret_no_expiry_returns_finding(mock_azure, subscription_id, monkeypatch): +def test_idn_006_noncompliant_secret_no_expiry_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): apps = { "value": [ { @@ -325,13 +543,43 @@ def test_idn_006_malformed_end_date_time_does_not_log_key_id(mock_azure, subscri # ── AZ-IDN-007: active user with no MFA registered ────────────────────────── -def test_idn_007_compliant_user_with_mfa_returns_no_findings(mock_azure, subscription_id, monkeypatch): - regs = {"value": [{"id": "u1", "userPrincipalName": "a@x.com", "isEnabled": True, "isMfaRegistered": True}]} - _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))]) - assert az_idn_007.scan(mock_azure, subscription_id) == [] +def test_idn_007_compliant_user_with_mfa_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): + regs = { + "value": [ + { + "id": "u1", + "userPrincipalName": "a@x.com", + "isEnabled": True, + "isMfaRegistered": True, + } + ] + } + + _install_router( + monkeypatch, + [ + ( + "credentialUserRegistrationDetails", + _Resp(regs), + ) + ], + ) + + assert az_idn_007.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_007_noncompliant_user_without_mfa_returns_finding(mock_azure, subscription_id, monkeypatch): +def test_idn_007_noncompliant_user_without_mfa_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): regs = { "value": [ { @@ -343,48 +591,139 @@ def test_idn_007_noncompliant_user_without_mfa_returns_finding(mock_azure, subsc } ] } - _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))]) - findings = az_idn_007.scan(mock_azure, subscription_id) + + _install_router( + monkeypatch, + [ + ( + "credentialUserRegistrationDetails", + _Resp(regs), + ) + ], + ) + + findings = az_idn_007.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-007" assert findings[0]["severity"] == "HIGH" assert findings[0]["resource_name"] == "No MFA User" +def test_idn_007_disabled_user_without_mfa_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): + 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 ────────────────── def _install_fake_auth_client(monkeypatch, roles): fake_module = SimpleNamespace( AuthorizationManagementClient=lambda *a, **k: SimpleNamespace( - role_definitions=SimpleNamespace(list=lambda **kw: list(roles)) + role_definitions=SimpleNamespace( + list=lambda **kw: list(roles) + ) ) ) - monkeypatch.setitem(sys.modules, "azure.mgmt.authorization", fake_module) + + monkeypatch.setitem( + sys.modules, + "azure.mgmt.authorization", + fake_module, + ) -def test_idn_008_compliant_specific_actions_returns_no_findings(mock_azure, subscription_id, monkeypatch): +def test_idn_008_compliant_specific_actions_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): role = make_resource( role_name="ReaderPlus", name="rd1", id="/rd1", - permissions=[make_resource(actions=["Microsoft.Storage/*/read"])], - assignable_scopes=[f"/subscriptions/{_SUB}"], + permissions=[ + make_resource( + actions=["Microsoft.Storage/*/read"] + ) + ], + assignable_scopes=[ + f"/subscriptions/{_SUB}" + ], ) - _install_fake_auth_client(monkeypatch, [role]) - assert az_idn_008.scan(mock_azure, subscription_id) == [] + _install_fake_auth_client( + monkeypatch, + [role], + ) + + assert az_idn_008.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_008_noncompliant_wildcard_action_returns_finding(mock_azure, subscription_id, monkeypatch): + +def test_idn_008_noncompliant_wildcard_action_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): role = make_resource( role_name="GodMode", name="rd2", id="/rd2", - permissions=[make_resource(actions=["*"])], - assignable_scopes=[f"/subscriptions/{_SUB}"], + permissions=[ + make_resource( + actions=["*"] + ) + ], + assignable_scopes=[ + f"/subscriptions/{_SUB}" + ], + ) + + _install_fake_auth_client( + monkeypatch, + [role], + ) + + findings = az_idn_008.scan( + mock_azure, + subscription_id, ) - _install_fake_auth_client(monkeypatch, [role]) - findings = az_idn_008.scan(mock_azure, subscription_id) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-008" assert findings[0]["severity"] == "HIGH" @@ -397,22 +736,62 @@ def test_idn_008_noncompliant_wildcard_action_returns_finding(mock_azure, subscr def _install_fake_monitor_client(monkeypatch, alerts): fake_module = SimpleNamespace( MonitorManagementClient=lambda *a, **k: SimpleNamespace( - activity_log_alerts=SimpleNamespace(list_by_subscription_id=lambda: list(alerts)) + activity_log_alerts=SimpleNamespace( + list_by_subscription_id=lambda: list(alerts) + ) ) ) - monkeypatch.setitem(sys.modules, "azure.mgmt.monitor", fake_module) + monkeypatch.setitem( + sys.modules, + "azure.mgmt.monitor", + fake_module, + ) + + +def test_idn_009_compliant_alert_present_returns_no_findings( + mock_azure, + subscription_id, + monkeypatch, +): + leaf = make_resource( + field="operationName", + equals="Microsoft.Authorization/roleAssignments/write", + ) + + alert = make_resource( + enabled=True, + condition=make_resource( + all_of=[leaf] + ), + ) + + _install_fake_monitor_client( + monkeypatch, + [alert], + ) -def test_idn_009_compliant_alert_present_returns_no_findings(mock_azure, subscription_id, monkeypatch): - leaf = make_resource(field="operationName", equals="Microsoft.Authorization/roleAssignments/write") - alert = make_resource(enabled=True, condition=make_resource(all_of=[leaf])) - _install_fake_monitor_client(monkeypatch, [alert]) - assert az_idn_009.scan(mock_azure, subscription_id) == [] + assert az_idn_009.scan( + mock_azure, + subscription_id, + ) == [] -def test_idn_009_noncompliant_no_alert_returns_finding(mock_azure, subscription_id, monkeypatch): - _install_fake_monitor_client(monkeypatch, []) - findings = az_idn_009.scan(mock_azure, subscription_id) +def test_idn_009_noncompliant_no_alert_returns_finding( + mock_azure, + subscription_id, + monkeypatch, +): + _install_fake_monitor_client( + monkeypatch, + [], + ) + + findings = az_idn_009.scan( + mock_azure, + subscription_id, + ) + assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-009" - assert findings[0]["severity"] == "MEDIUM" + assert findings[0]["severity"] == "MEDIUM" \ No newline at end of file From 4e15c7a12937435c57a6d8f116c47a19f0c1d098 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Sun, 23 Aug 2026 01:35:39 +0100 Subject: [PATCH 5/9] fix: improve AZ-NET-003 scanner correctness Signed-off-by: safidnadaf --- scanner/rules/az_net_003.py | 66 +++++++++++++++++++++++++++++++++---- tests/test_rules_network.py | 24 +++++++++++++- 2 files changed, 82 insertions(+), 8 deletions(-) diff --git a/scanner/rules/az_net_003.py b/scanner/rules/az_net_003.py index 65b7479a..82229c45 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,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, }, } diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index f73f77a8..369e799a 100644 --- a/tests/test_rules_network.py +++ b/tests/test_rules_network.py @@ -160,7 +160,7 @@ 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 +168,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 +228,27 @@ 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" + + @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, From 7a44d61acf5799cba3ee10dc67dba6137c5abcfb Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Mon, 24 Aug 2026 09:20:25 +0100 Subject: [PATCH 6/9] fix: resolve ruff lint errors Signed-off-by: safidnadaf --- scanner/azure_client.py | 16 ---------------- scanner/rules/az_net_003.py | 2 +- tests/test_rules_identity.py | 2 +- tests/test_rules_network.py | 3 ++- 4 files changed, 4 insertions(+), 19 deletions(-) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index 2366056e..d3b1bb21 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -23,22 +23,6 @@ _UNSET = object() -def enum_str(value: Any, default: str = "") -> str: - """Safely coerce an Azure SDK field to its plain string form. - - Azure SDK models often return fields typed as enums (e.g. - SecurityRuleDirection, BlobAuditingPolicyState) rather than plain - strings. ``str(enum_member)`` yields something like - "SecurityRuleDirection.INBOUND", not the underlying value "Inbound", - which breaks naive string comparisons. This prefers ``.value`` when - present (covers real SDK enums and enum-like objects) and falls back - to ``str()`` for plain strings, None, or anything else. - """ - if value is None: - return default - return str(getattr(value, "value", value)) - - def enum_str(value: Any, default: str = "") -> str: """Safely coerce an Azure SDK field to its plain string form. diff --git a/scanner/rules/az_net_003.py b/scanner/rules/az_net_003.py index 82229c45..a32a9fb8 100644 --- a/scanner/rules/az_net_003.py +++ b/scanner/rules/az_net_003.py @@ -126,4 +126,4 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: ) break - return findings \ No newline at end of file + return findings diff --git a/tests/test_rules_identity.py b/tests/test_rules_identity.py index 7e3ca321..e625e57a 100644 --- a/tests/test_rules_identity.py +++ b/tests/test_rules_identity.py @@ -794,4 +794,4 @@ def test_idn_009_noncompliant_no_alert_returns_finding( assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-009" - assert findings[0]["severity"] == "MEDIUM" \ No newline at end of file + assert findings[0]["severity"] == "MEDIUM" diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index 369e799a..bb0abc50 100644 --- a/tests/test_rules_network.py +++ b/tests/test_rules_network.py @@ -833,4 +833,5 @@ def test_net_017_direct_internet_default_creates_finding(mock_azure, subscriptio findings = az_net_017.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-NET-017" - assert findings[0]["metadata"]["address_prefix"] == prefix \ No newline at end of file + assert findings[0]["metadata"]["address_prefix"] == prefix + From 61a7b0aaa54ba149b45ce56e2a19d032b7d58041 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Fri, 28 Aug 2026 00:52:48 +0100 Subject: [PATCH 7/9] test: restore az_net_016/017 imports dropped in merge Signed-off-by: safidnadaf --- tests/test_rules_network.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index bb0abc50..9455c3cc 100644 --- a/tests/test_rules_network.py +++ b/tests/test_rules_network.py @@ -835,3 +835,4 @@ def test_net_017_direct_internet_default_creates_finding(mock_azure, subscriptio assert findings[0]["rule_id"] == "AZ-NET-017" assert findings[0]["metadata"]["address_prefix"] == prefix + From 78392d2f930ecb92532b774f9b824967ce836675 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Sun, 30 Aug 2026 14:59:23 +0100 Subject: [PATCH 8/9] chore: trigger CI Signed-off-by: safidnadaf From 2b8d72599890dfe3645a5927516675934be04879 Mon Sep 17 00:00:00 2001 From: safidnadaf Date: Mon, 31 Aug 2026 10:08:10 +0100 Subject: [PATCH 9/9] fix: add compliant plural-port test, drop unrelated identity test reformatting Addresses parthrohit22's review: adds a compliant-case regression test for the destination_port_ranges fix in AZ-NET-003, and resets tests/test_rules_identity.py to dev's formatting, keeping only the one genuine new test (test_idn_007_disabled_user_without_mfa_returns_no_findings). Signed-off-by: safidnadaf --- tests/helpers/mock_azure.py | 2 +- tests/test_rules_database.py | 2 +- tests/test_rules_identity.py | 520 ++++++----------------------------- tests/test_rules_network.py | 33 ++- 4 files changed, 112 insertions(+), 445 deletions(-) diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index a255aae5..de3c6be0 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -450,4 +450,4 @@ def parse_resource_id(resource_id: str) -> Dict[str, str]: for idx, segment in enumerate(parts): if segment.lower() == "resourcegroups" and idx + 1 < len(parts): result["resource_group"] = parts[idx + 1] - return result \ No newline at end of file + return result diff --git a/tests/test_rules_database.py b/tests/test_rules_database.py index 4100eb8d..a89d2400 100644 --- a/tests/test_rules_database.py +++ b/tests/test_rules_database.py @@ -251,4 +251,4 @@ def test_db_003_noncompliant_returns_one_finding(mock_azure, subscription_id): assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-DB-003" assert findings[0]["severity"] == "HIGH" - assert findings[0]["resource_name"] == "pgflex-nossl" \ No newline at end of file + assert findings[0]["resource_name"] == "pgflex-nossl" diff --git a/tests/test_rules_identity.py b/tests/test_rules_identity.py index e625e57a..e27293cf 100644 --- a/tests/test_rules_identity.py +++ b/tests/test_rules_identity.py @@ -72,67 +72,33 @@ def fake_get(url, *args, **kwargs): _SUB = "00000000-0000-0000-0000-000000000001" _OWNER_ROLE_GUID = "8e3af657-a8ff-443c-a75c-2fe8c4bcb635" _CONTRIBUTOR_ROLE_GUID = "b24988ac-6180-42a0-ab88-20f7382dd24c" -_ROLE_DEF_BASE = ( - f"/subscriptions/{_SUB}/providers/" - "Microsoft.Authorization/roleDefinitions" -) +_ROLE_DEF_BASE = f"/subscriptions/{_SUB}/providers/Microsoft.Authorization/roleDefinitions" def _assignment(role_guid, principal_id, assign_id): return make_resource( - id=( - f"/subscriptions/{_SUB}/providers/" - f"Microsoft.Authorization/roleAssignments/{assign_id}" - ), + id=f"/subscriptions/{_SUB}/providers/Microsoft.Authorization/roleAssignments/{assign_id}", role_definition_id=f"{_ROLE_DEF_BASE}/{role_guid}", principal_id=principal_id, scope=f"/subscriptions/{_SUB}", ) -def test_idn_001_compliant_returns_no_findings( - mock_azure, - subscription_id, -): +def test_idn_001_compliant_returns_no_findings(mock_azure, subscription_id): """A service principal with a non-Owner role must produce no findings.""" - assignment = _assignment( - _CONTRIBUTOR_ROLE_GUID, - "sp-contributor-abc123", - "assign-001", - ) - + assignment = _assignment(_CONTRIBUTOR_ROLE_GUID, "sp-contributor-abc123", "assign-001") mock_azure.set_service_principals([assignment]) - - findings = az_idn_001.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_001.scan(mock_azure, subscription_id) assert findings == [] -def test_idn_001_noncompliant_returns_one_finding( - mock_azure, - subscription_id, -): +def test_idn_001_noncompliant_returns_one_finding(mock_azure, subscription_id): """A service principal holding the Owner role must produce exactly one finding.""" - assignment = _assignment( - _OWNER_ROLE_GUID, - "sp-owner-def456", - "assign-002", - ) - + assignment = _assignment(_OWNER_ROLE_GUID, "sp-owner-def456", "assign-002") mock_azure.set_service_principals([assignment]) - - findings = az_idn_001.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_001.scan(mock_azure, subscription_id) assert len(findings) == 1 - finding = findings[0] - assert _REQUIRED_FIELDS.issubset(finding.keys()) assert finding["rule_id"] == "AZ-IDN-001" assert finding["severity"] == "HIGH" @@ -147,43 +113,21 @@ def test_idn_001_noncompliant_returns_one_finding( # ── AZ-IDN-002: MFA enforced on admins via Conditional Access ─────────────── -def test_idn_002_compliant_policy_enforces_mfa_returns_no_findings( - mock_azure, - subscription_id, -): +def test_idn_002_compliant_policy_enforces_mfa_returns_no_findings(mock_azure, subscription_id): """A CA policy that is enabled, requires MFA, and covers all users is compliant.""" policy = { "state": "enabled", - "grantControls": { - "builtInControls": ["mfa"], - }, - "conditions": { - "users": { - "includeUsers": ["All"], - }, - }, + "grantControls": {"builtInControls": ["mfa"]}, + "conditions": {"users": {"includeUsers": ["All"]}}, } - mock_azure.set_conditional_access_policies([policy]) - - assert az_idn_002.scan( - mock_azure, - subscription_id, - ) == [] + assert az_idn_002.scan(mock_azure, subscription_id) == [] -def test_idn_002_noncompliant_no_policies_returns_one_finding( - mock_azure, - subscription_id, -): +def test_idn_002_noncompliant_no_policies_returns_one_finding(mock_azure, subscription_id): """No Conditional Access policies at all must produce exactly one finding.""" mock_azure.set_conditional_access_policies([]) - - findings = az_idn_002.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_002.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-002" assert findings[0]["severity"] == "HIGH" @@ -192,57 +136,24 @@ def test_idn_002_noncompliant_no_policies_returns_one_finding( # ── AZ-IDN-003: guest invitations not restricted ─────────────────────────── -def test_idn_003_compliant_restricted_invites_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_003_compliant_restricted_invites_returns_no_findings(mock_azure, subscription_id, monkeypatch): _install_router( monkeypatch, [ - ( - "authorizationPolicy", - _Resp( - { - "id": "authPol", - "allowInvitesFrom": "adminsAndGuestInviters", - } - ), - ), + ("authorizationPolicy", _Resp({"id": "authPol", "allowInvitesFrom": "adminsAndGuestInviters"})), ], ) - - assert az_idn_003.scan( - mock_azure, - subscription_id, - ) == [] + assert az_idn_003.scan(mock_azure, subscription_id) == [] -def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding(mock_azure, subscription_id, monkeypatch): _install_router( monkeypatch, [ - ( - "authorizationPolicy", - _Resp( - { - "id": "authPol", - "allowInvitesFrom": "everyone", - } - ), - ), + ("authorizationPolicy", _Resp({"id": "authPol", "allowInvitesFrom": "everyone"})), ], ) - - findings = az_idn_003.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_003.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-003" assert findings[0]["severity"] == "MEDIUM" @@ -251,28 +162,9 @@ def test_idn_003_noncompliant_everyone_can_invite_returns_one_finding( # ── AZ-IDN-004: no PIM for admin roles ────────────────────────────────────── -def test_idn_004_compliant_role_has_pim_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): - role_defs = { - "value": [ - { - "id": "rd-ga", - "displayName": "Global Administrator", - } - ] - } - - schedules = { - "value": [ - { - "roleDefinitionId": "rd-ga", - } - ] - } - +def test_idn_004_compliant_role_has_pim_returns_no_findings(mock_azure, subscription_id, monkeypatch): + role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} + schedules = {"value": [{"roleDefinitionId": "rd-ga"}]} _install_router( monkeypatch, [ @@ -280,50 +172,20 @@ def test_idn_004_compliant_role_has_pim_returns_no_findings( ("roleEligibilitySchedules", _Resp(schedules)), ], ) + assert az_idn_004.scan(mock_azure, subscription_id) == [] - assert az_idn_004.scan( - mock_azure, - subscription_id, - ) == [] - - -def test_idn_004_noncompliant_role_without_pim_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): - role_defs = { - "value": [ - { - "id": "rd-ga", - "displayName": "Global Administrator", - } - ] - } - - schedules = { - "value": [] - } +def test_idn_004_noncompliant_role_without_pim_returns_finding(mock_azure, subscription_id, monkeypatch): + role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} + schedules = {"value": []} _install_router( monkeypatch, [ - ( - "roleEligibilitySchedules", - _Resp(schedules), - ), - ( - "roleDefinitions", - _Resp(role_defs), - ), + ("roleEligibilitySchedules", _Resp(schedules)), # check more specific first + ("roleDefinitions", _Resp(role_defs)), ], ) - - findings = az_idn_004.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_004.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-004" assert findings[0]["severity"] == "HIGH" @@ -333,84 +195,23 @@ def test_idn_004_noncompliant_role_without_pim_returns_finding( # ── AZ-IDN-005: guest with high-privilege role ────────────────────────────── -def test_idn_005_compliant_member_user_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): - role_defs = { - "value": [ - { - "id": "rd-ga", - "displayName": "Global Administrator", - } - ] - } - - assignments = { - "value": [ - { - "id": "a1", - "roleDefinitionId": "rd-ga", - "principalId": "u1", - } - ] - } - +def test_idn_005_compliant_member_user_returns_no_findings(mock_azure, subscription_id, monkeypatch): + role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} + assignments = {"value": [{"id": "a1", "roleDefinitionId": "rd-ga", "principalId": "u1"}]} _install_router( monkeypatch, [ - ( - "/users/", - _Resp( - { - "id": "u1", - "displayName": "Member User", - "userType": "Member", - } - ), - ), - ( - "roleDefinitions", - _Resp(role_defs), - ), - ( - "roleAssignments", - _Resp(assignments), - ), + ("/users/", _Resp({"id": "u1", "displayName": "Member User", "userType": "Member"})), + ("roleDefinitions", _Resp(role_defs)), + ("roleAssignments", _Resp(assignments)), ], ) + assert az_idn_005.scan(mock_azure, subscription_id) == [] - assert az_idn_005.scan( - mock_azure, - subscription_id, - ) == [] - - -def test_idn_005_noncompliant_guest_admin_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): - role_defs = { - "value": [ - { - "id": "rd-ga", - "displayName": "Global Administrator", - } - ] - } - - assignments = { - "value": [ - { - "id": "a1", - "roleDefinitionId": "rd-ga", - "principalId": "g1", - } - ] - } +def test_idn_005_noncompliant_guest_admin_returns_finding(mock_azure, subscription_id, monkeypatch): + role_defs = {"value": [{"id": "rd-ga", "displayName": "Global Administrator"}]} + assignments = {"value": [{"id": "a1", "roleDefinitionId": "rd-ga", "principalId": "g1"}]} _install_router( monkeypatch, [ @@ -425,22 +226,11 @@ def test_idn_005_noncompliant_guest_admin_returns_finding( } ), ), - ( - "roleDefinitions", - _Resp(role_defs), - ), - ( - "roleAssignments", - _Resp(assignments), - ), + ("roleDefinitions", _Resp(role_defs)), + ("roleAssignments", _Resp(assignments)), ], ) - - findings = az_idn_005.scan( - mock_azure, - subscription_id, - ) - + findings = az_idn_005.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-005" assert findings[0]["severity"] == "HIGH" @@ -450,11 +240,7 @@ def test_idn_005_noncompliant_guest_admin_returns_finding( # ── AZ-IDN-006: stale / non-expiring SP client secret ─────────────────────── -def test_idn_006_compliant_fresh_secret_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_006_compliant_fresh_secret_returns_no_findings(mock_azure, subscription_id, monkeypatch): apps = { "value": [ { @@ -476,11 +262,7 @@ def test_idn_006_compliant_fresh_secret_returns_no_findings( assert az_idn_006.scan(mock_azure, subscription_id) == [] -def test_idn_006_noncompliant_secret_no_expiry_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_006_noncompliant_secret_no_expiry_returns_finding(mock_azure, subscription_id, monkeypatch): apps = { "value": [ { @@ -543,43 +325,13 @@ def test_idn_006_malformed_end_date_time_does_not_log_key_id(mock_azure, subscri # ── AZ-IDN-007: active user with no MFA registered ────────────────────────── -def test_idn_007_compliant_user_with_mfa_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): - regs = { - "value": [ - { - "id": "u1", - "userPrincipalName": "a@x.com", - "isEnabled": True, - "isMfaRegistered": True, - } - ] - } - - _install_router( - monkeypatch, - [ - ( - "credentialUserRegistrationDetails", - _Resp(regs), - ) - ], - ) - - assert az_idn_007.scan( - mock_azure, - subscription_id, - ) == [] +def test_idn_007_compliant_user_with_mfa_returns_no_findings(mock_azure, subscription_id, monkeypatch): + regs = {"value": [{"id": "u1", "userPrincipalName": "a@x.com", "isEnabled": True, "isMfaRegistered": True}]} + _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))]) + assert az_idn_007.scan(mock_azure, subscription_id) == [] -def test_idn_007_noncompliant_user_without_mfa_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_007_noncompliant_user_without_mfa_returns_finding(mock_azure, subscription_id, monkeypatch): regs = { "value": [ { @@ -591,33 +343,18 @@ def test_idn_007_noncompliant_user_without_mfa_returns_finding( } ] } - - _install_router( - monkeypatch, - [ - ( - "credentialUserRegistrationDetails", - _Resp(regs), - ) - ], - ) - - findings = az_idn_007.scan( - mock_azure, - subscription_id, - ) - + _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))]) + findings = az_idn_007.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-007" assert findings[0]["severity"] == "HIGH" assert findings[0]["resource_name"] == "No MFA User" -def test_idn_007_disabled_user_without_mfa_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): +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": [ { @@ -629,21 +366,8 @@ def test_idn_007_disabled_user_without_mfa_returns_no_findings( } ] } - - _install_router( - monkeypatch, - [ - ( - "credentialUserRegistrationDetails", - _Resp(regs), - ) - ], - ) - - assert az_idn_007.scan( - mock_azure, - subscription_id, - ) == [] + _install_router(monkeypatch, [("credentialUserRegistrationDetails", _Resp(regs))]) + assert az_idn_007.scan(mock_azure, subscription_id) == [] # ── AZ-IDN-008: custom RBAC role with wildcard permissions ────────────────── @@ -652,78 +376,34 @@ def test_idn_007_disabled_user_without_mfa_returns_no_findings( def _install_fake_auth_client(monkeypatch, roles): fake_module = SimpleNamespace( AuthorizationManagementClient=lambda *a, **k: SimpleNamespace( - role_definitions=SimpleNamespace( - list=lambda **kw: list(roles) - ) + role_definitions=SimpleNamespace(list=lambda **kw: list(roles)) ) ) - - monkeypatch.setitem( - sys.modules, - "azure.mgmt.authorization", - fake_module, - ) + monkeypatch.setitem(sys.modules, "azure.mgmt.authorization", fake_module) -def test_idn_008_compliant_specific_actions_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_008_compliant_specific_actions_returns_no_findings(mock_azure, subscription_id, monkeypatch): role = make_resource( role_name="ReaderPlus", name="rd1", id="/rd1", - permissions=[ - make_resource( - actions=["Microsoft.Storage/*/read"] - ) - ], - assignable_scopes=[ - f"/subscriptions/{_SUB}" - ], + permissions=[make_resource(actions=["Microsoft.Storage/*/read"])], + assignable_scopes=[f"/subscriptions/{_SUB}"], ) + _install_fake_auth_client(monkeypatch, [role]) + assert az_idn_008.scan(mock_azure, subscription_id) == [] - _install_fake_auth_client( - monkeypatch, - [role], - ) - - assert az_idn_008.scan( - mock_azure, - subscription_id, - ) == [] - -def test_idn_008_noncompliant_wildcard_action_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): +def test_idn_008_noncompliant_wildcard_action_returns_finding(mock_azure, subscription_id, monkeypatch): role = make_resource( role_name="GodMode", name="rd2", id="/rd2", - permissions=[ - make_resource( - actions=["*"] - ) - ], - assignable_scopes=[ - f"/subscriptions/{_SUB}" - ], - ) - - _install_fake_auth_client( - monkeypatch, - [role], - ) - - findings = az_idn_008.scan( - mock_azure, - subscription_id, + permissions=[make_resource(actions=["*"])], + assignable_scopes=[f"/subscriptions/{_SUB}"], ) - + _install_fake_auth_client(monkeypatch, [role]) + findings = az_idn_008.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-008" assert findings[0]["severity"] == "HIGH" @@ -736,62 +416,22 @@ def test_idn_008_noncompliant_wildcard_action_returns_finding( def _install_fake_monitor_client(monkeypatch, alerts): fake_module = SimpleNamespace( MonitorManagementClient=lambda *a, **k: SimpleNamespace( - activity_log_alerts=SimpleNamespace( - list_by_subscription_id=lambda: list(alerts) - ) + activity_log_alerts=SimpleNamespace(list_by_subscription_id=lambda: list(alerts)) ) ) + monkeypatch.setitem(sys.modules, "azure.mgmt.monitor", fake_module) - monkeypatch.setitem( - sys.modules, - "azure.mgmt.monitor", - fake_module, - ) +def test_idn_009_compliant_alert_present_returns_no_findings(mock_azure, subscription_id, monkeypatch): + leaf = make_resource(field="operationName", equals="Microsoft.Authorization/roleAssignments/write") + alert = make_resource(enabled=True, condition=make_resource(all_of=[leaf])) + _install_fake_monitor_client(monkeypatch, [alert]) + assert az_idn_009.scan(mock_azure, subscription_id) == [] -def test_idn_009_compliant_alert_present_returns_no_findings( - mock_azure, - subscription_id, - monkeypatch, -): - leaf = make_resource( - field="operationName", - equals="Microsoft.Authorization/roleAssignments/write", - ) - - alert = make_resource( - enabled=True, - condition=make_resource( - all_of=[leaf] - ), - ) - - _install_fake_monitor_client( - monkeypatch, - [alert], - ) - - assert az_idn_009.scan( - mock_azure, - subscription_id, - ) == [] - - -def test_idn_009_noncompliant_no_alert_returns_finding( - mock_azure, - subscription_id, - monkeypatch, -): - _install_fake_monitor_client( - monkeypatch, - [], - ) - - findings = az_idn_009.scan( - mock_azure, - subscription_id, - ) +def test_idn_009_noncompliant_no_alert_returns_finding(mock_azure, subscription_id, monkeypatch): + _install_fake_monitor_client(monkeypatch, []) + findings = az_idn_009.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-IDN-009" assert findings[0]["severity"] == "MEDIUM" diff --git a/tests/test_rules_network.py b/tests/test_rules_network.py index 9455c3cc..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", port_list=None): +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, @@ -249,6 +257,27 @@ def test_net_003_detects_plural_destination_port_ranges(mock_azure, subscription 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, @@ -834,5 +863,3 @@ def test_net_017_direct_internet_default_creates_finding(mock_azure, subscriptio assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-NET-017" assert findings[0]["metadata"]["address_prefix"] == prefix - -