Compare commits

...
Author SHA1 Message Date
Daniel Barranquero e29978bde6 test(m365/entra): cover CA deleted-object-reference resolver and report-only message
Add service-layer tests for the directory-object resolver (404/Request_ResourceNotFound flags as deleted; 5xx/throttling and successful resolutions do not; roles resolve via the roleDefinitions endpoint; sentinels skipped and identifiers deduped across policies).

Surface report-only mode in the check's failure message, link the related directory-sync-account check via RelatedTo metadata, and hoist the shared sentinel set to a single CONDITIONAL_ACCESS_SENTINEL_IDS constant in entra_service.
2026-06-17 14:19:13 +02:00
Ernest Provo 83f8c0bfee feat(m365/entra): wire CA-references resolver into entra_service + CHANGELOG
Completes #11063 by:
- Adding `_resolve_directory_object_references` and helper
  `_resolve_identifiers_for_type` to `entra_service.py`.
- Hooking the resolver into `Entra.__init__` as a second-phase
  `loop.run_until_complete` after the main asyncio.gather, populating
  `self.unresolved_directory_object_references: Set[Tuple[str, str]]`.
- Adding the CHANGELOG entry for the new check.

Fix #11063
2026-05-19 12:59:40 -04:00
Ernest Provo c172b59834 feat(m365/entra): add entra_conditional_access_policy_no_deleted_object_references check
Audits every Conditional Access policy for references to users, groups, or
directory roles that no longer resolve in Microsoft Entra ID. Stale references
silently change runtime policy behavior — include* references shrink
enforcement scope, exclude* references can alter exemption logic — which is
a known root cause of "we thought MFA was required but it wasn't" incidents.

The implementation follows the eager-init pattern from #11232:

- A new resolver method `_resolve_directory_object_references` runs at service
  init time after the main `asyncio.gather`, walks every loaded Conditional
  Access policy, deduplicates identifiers per type, and queries Graph for each
  one (`/users/{id}`, `/groups/{id}`,
  `/roleManagement/directory/roleDefinitions/{id}`). Sentinel values (`All`,
  `None`, `GuestsOrExternalUsers`) are skipped before any Graph call.
- Only HTTP 404 responses (or `Request_ResourceNotFound` error code) flag an
  identifier as deleted. Transient errors (5xx, throttling) are logged and
  skipped so the check never produces false positives on network blips.
- Result is cached on the entra client as
  `unresolved_directory_object_references: Set[Tuple[str, str]]`, which the
  new pure-sync check reads to emit findings.
- The check runs against all policies regardless of state — disabled-policy
  references are a misconfiguration that becomes live the moment the policy
  is re-enabled.

Fix #11063
2026-05-19 12:59:40 -04:00
8 changed files with 853 additions and 1 deletions
+1
View File
@@ -6,6 +6,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
### 🚀 Added
- `entra_conditional_access_policy_no_deleted_object_references` check for M365 provider [(#11063)](https://github.com/prowler-cloud/prowler/issues/11063)
- `entra_app_registration_client_secret_unused` check for M365 provider [(#11232)](https://github.com/prowler-cloud/prowler/pull/11232)
- `cloudsql_instance_cmek_encryption_enabled` check for GCP provider [(#11023)](https://github.com/prowler-cloud/prowler/pull/11023)
@@ -0,0 +1,44 @@
{
"Provider": "m365",
"CheckID": "entra_conditional_access_policy_no_deleted_object_references",
"CheckTitle": "Conditional Access policies must not reference deleted users, groups, or roles",
"CheckType": [],
"ServiceName": "entra",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Every object identifier referenced by any Conditional Access policy under conditions.users (includeUsers, excludeUsers, includeGroups, excludeGroups, includeRoles, excludeRoles) must resolve to an existing Microsoft Entra object. This check audits all Conditional Access policies regardless of state and reports any whose user, group, or role references no longer resolve in the directory.",
"Risk": "When a user, group, or directory role referenced by a Conditional Access policy stops resolving (account or group deleted, role template removed), the reference becomes orphaned. include* references silently shrink the policy's enforcement scope; exclude* references can cause the policy to evaluate unexpectedly. This is a common root cause of MFA-not-applied incidents.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://learn.microsoft.com/en-us/graph/api/resources/conditionalaccesspolicy?view=graph-rest-1.0",
"https://learn.microsoft.com/en-us/graph/api/resources/conditionalaccessusers?view=graph-rest-1.0",
"https://learn.microsoft.com/en-us/graph/api/user-get?view=graph-rest-1.0",
"https://learn.microsoft.com/en-us/graph/api/group-get?view=graph-rest-1.0",
"https://learn.microsoft.com/en-us/graph/api/unifiedroledefinition-get?view=graph-rest-1.0",
"https://learn.microsoft.com/en-us/entra/identity/conditional-access/concept-conditional-access-users-groups"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Sign in to the Microsoft Entra admin center (https://entra.microsoft.com)\n2. Navigate to Protection > Conditional Access > Policies\n3. Open each policy reported by this check\n4. Under Assignments > Users, remove every user, group, or role identifier reported as deleted\n5. Save the policy and re-run the audit\n6. For ongoing hygiene, audit Conditional Access policies after any bulk user/group/role cleanup",
"Terraform": ""
},
"Recommendation": {
"Text": "Audit each Conditional Access policy quarterly and remove references to deleted users, groups, or directory roles. Stale references in include collections silently shrink enforcement scope; stale references in exclude collections can cause policies to behave unexpectedly. Treat both as misconfigurations regardless of policy state.",
"Url": "https://hub.prowler.com/check/entra_conditional_access_policy_no_deleted_object_references"
}
},
"Categories": [
"identity-access",
"e3"
],
"DependsOn": [],
"RelatedTo": [
"entra_conditional_access_policy_directory_sync_account_excluded"
],
"Notes": "The check runs against all Conditional Access policies regardless of state (enabled, disabled, enabledForReportingButNotEnforced) — stale references in disabled policies are a misconfiguration that becomes live the moment the policy is re-enabled. Transient Graph errors (5xx, throttling) are not treated as deletions; only HTTP 404 responses flag an identifier as orphaned."
}
@@ -0,0 +1,115 @@
from prowler.lib.check.models import Check, CheckReportM365
from prowler.providers.m365.services.entra.entra_client import entra_client
from prowler.providers.m365.services.entra.entra_service import (
CONDITIONAL_ACCESS_SENTINEL_IDS,
ConditionalAccessPolicyState,
)
class entra_conditional_access_policy_no_deleted_object_references(Check):
"""
Ensure Conditional Access policies do not reference deleted directory objects.
Stale references to deleted users, groups, or directory roles silently change
the runtime behavior of a Conditional Access policy: include* references
shrink enforcement scope, exclude* references can change exemption logic.
Either way, the policy stops behaving the way the operator believes it does.
The directory-object existence check runs once at service init time and is
cached on the entra client. This check reads from that cache and reports any
policy whose users/groups/roles inclusion or exclusion collections name an
identifier that no longer resolves in Microsoft Entra ID.
- PASS: The policy references no deleted users, groups, or roles.
- FAIL: The policy references at least one deleted user, group, or role.
"""
def execute(self) -> list[CheckReportM365]:
findings = []
unresolved = entra_client.unresolved_directory_object_references
for policy_id, policy in entra_client.conditional_access_policies.items():
report = CheckReportM365(
metadata=self.metadata(),
resource=policy,
resource_name=policy.display_name,
resource_id=policy_id,
)
orphans = self._collect_orphans(policy, unresolved)
if not orphans:
report.status = "PASS"
report.status_extended = (
f"Conditional Access policy {policy.display_name} references no "
f"deleted directory objects."
)
else:
report.status = "FAIL"
report.status_extended = self._format_failure(
policy.display_name, orphans, policy.state
)
findings.append(report)
return findings
@staticmethod
def _collect_orphans(policy, unresolved):
"""Walk the six identifier collections on a policy and return orphans.
Returns:
list[tuple[str, str, str]]: ``(type, id, side)`` tuples where
``type`` is one of ``user|group|role``, ``id`` is the Graph
identifier that failed to resolve, and ``side`` is one of
``include|exclude``.
"""
if not policy.conditions or not policy.conditions.user_conditions:
return []
uc = policy.conditions.user_conditions
collections = (
("user", "include", uc.included_users),
("user", "exclude", uc.excluded_users),
("group", "include", uc.included_groups),
("group", "exclude", uc.excluded_groups),
("role", "include", uc.included_roles),
("role", "exclude", uc.excluded_roles),
)
orphans = []
for type_, side, identifiers in collections:
for identifier in identifiers:
if identifier in CONDITIONAL_ACCESS_SENTINEL_IDS:
continue
if (type_, identifier) in unresolved:
orphans.append((type_, identifier, side))
return orphans
@staticmethod
def _format_failure(display_name, orphans, state=None):
# Group orphans by type for a readable, deterministic message.
by_type = {"user": [], "group": [], "role": []}
for type_, identifier, side in orphans:
by_type[type_].append(f"{identifier} ({side})")
parts = []
for type_ in ("user", "group", "role"):
if by_type[type_]:
joined = ", ".join(sorted(by_type[type_]))
parts.append(f"{type_}s: {joined}")
# Surface report-only mode explicitly: the stale references are not yet
# enforced, but become live the moment the policy is turned on.
report_only = (
" The policy is in report-only mode, so these references are not "
"enforced yet but will take effect once it is enabled."
if state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING
else ""
)
return (
f"Conditional Access policy {display_name} references "
f"{len(orphans)} deleted directory object(s) — {'; '.join(parts)}."
f"{report_only}"
)
@@ -3,7 +3,7 @@ import json
from asyncio import gather
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional, Set, Tuple
from uuid import UUID
from kiota_abstractions.base_request_configuration import RequestConfiguration
@@ -18,6 +18,12 @@ from prowler.lib.logger import logger
from prowler.providers.m365.lib.service.service import M365Service
from prowler.providers.m365.m365_provider import M365Provider
# Sentinel identifiers used in Conditional Access ``conditions.users``
# collections that do not correspond to real directory objects and must not be
# resolved against Graph. Shared by the resolver below and the check that reads
# its result.
CONDITIONAL_ACCESS_SENTINEL_IDS = {"All", "None", "GuestsOrExternalUsers"}
class Entra(M365Service):
"""
@@ -110,6 +116,19 @@ class Entra(M365Service):
self.app_registrations: Dict[str, "AppRegistration"] = attributes[11]
self.user_accounts_status = {}
# Resolve directory-object identifiers referenced by Conditional Access
# policies. This runs as a separate phase because it depends on the
# main gather having populated ``conditional_access_policies`` first.
# The result is cached on the instance so sync checks can read it
# without issuing Graph calls of their own.
self.unresolved_directory_object_references: Set[Tuple[str, str]] = (
loop.run_until_complete(
self._resolve_directory_object_references(
self.conditional_access_policies
)
)
)
if created_loop:
asyncio.set_event_loop(None)
loop.close()
@@ -1316,6 +1335,117 @@ OAuthAppInfo
)
return app_registrations
async def _resolve_directory_object_references(
self,
policies: Dict[str, "ConditionalAccessPolicy"],
) -> Set[Tuple[str, str]]:
"""Resolve every user/group/role identifier referenced by CA policies.
Walks the inclusion/exclusion collections of every loaded Conditional
Access policy, deduplicates the resulting identifiers per type, and
queries Microsoft Graph for each one. Identifiers that return HTTP 404
are reported as deleted. Non-404 errors (5xx, throttling, transient
network failures) are logged and skipped — they must not be flagged as
deletions per the related check's specification.
The sentinel values ``All``, ``None``, and ``GuestsOrExternalUsers``
are not directory identifiers and are excluded before any Graph call.
Args:
policies: Conditional Access policies keyed by policy ID.
Returns:
Set[Tuple[str, str]]: ``(type, identifier)`` pairs where ``type``
is one of ``user|group|role`` and ``identifier`` failed to
resolve via Graph with HTTP 404.
"""
logger.info(
"Entra - Resolving directory-object references in Conditional "
"Access policies..."
)
ids_by_type: Dict[str, Set[str]] = {
"user": set(),
"group": set(),
"role": set(),
}
for policy in policies.values():
if not getattr(policy, "conditions", None):
continue
user_conditions = getattr(policy.conditions, "user_conditions", None)
if user_conditions is None:
continue
for ident in (user_conditions.included_users or []) + (
user_conditions.excluded_users or []
):
if ident and ident not in CONDITIONAL_ACCESS_SENTINEL_IDS:
ids_by_type["user"].add(ident)
for ident in (user_conditions.included_groups or []) + (
user_conditions.excluded_groups or []
):
if ident and ident not in CONDITIONAL_ACCESS_SENTINEL_IDS:
ids_by_type["group"].add(ident)
for ident in (user_conditions.included_roles or []) + (
user_conditions.excluded_roles or []
):
if ident and ident not in CONDITIONAL_ACCESS_SENTINEL_IDS:
ids_by_type["role"].add(ident)
unresolved: Set[Tuple[str, str]] = set()
# Resolve types in parallel; within a type, walk identifiers serially
# to keep concurrent Graph calls bounded and avoid throttling.
await gather(
self._resolve_identifiers_for_type("user", ids_by_type["user"], unresolved),
self._resolve_identifiers_for_type(
"group", ids_by_type["group"], unresolved
),
self._resolve_identifiers_for_type("role", ids_by_type["role"], unresolved),
)
return unresolved
async def _resolve_identifiers_for_type(
self,
type_: str,
identifiers: Set[str],
unresolved: Set[Tuple[str, str]],
) -> None:
"""Resolve a set of identifiers of a given type, mutating ``unresolved``.
Only HTTP 404 responses add to the unresolved set; every other error
is logged and treated as 'unknown' (the identifier is not flagged).
"""
for identifier in identifiers:
try:
if type_ == "user":
await self.client.users.by_user_id(identifier).get()
elif type_ == "group":
await self.client.groups.by_group_id(identifier).get()
elif type_ == "role":
await self.client.role_management.directory.role_definitions.by_unified_role_definition_id(
identifier
).get()
else:
continue
except ODataError as error:
status_code = getattr(error, "response_status_code", None)
error_code = getattr(error.error, "code", None) if error.error else None
if status_code == 404 or error_code == "Request_ResourceNotFound":
unresolved.add((type_, identifier))
else:
logger.warning(
f"Entra - Could not resolve {type_} '{identifier}' for "
f"Conditional Access reference check: "
f"{error.__class__.__name__}: {error}"
)
except Exception as error:
logger.warning(
f"Entra - Unexpected error resolving {type_} '{identifier}' "
f"for Conditional Access reference check: "
f"{error.__class__.__name__}: {error}"
)
class ConditionalAccessPolicyState(Enum):
ENABLED = "enabled"
@@ -0,0 +1,390 @@
from unittest import mock
from uuid import uuid4
from prowler.providers.m365.services.entra.entra_service import (
ApplicationsConditions,
ConditionalAccessPolicy,
ConditionalAccessPolicyState,
Conditions,
GrantControlOperator,
GrantControls,
PersistentBrowser,
SessionControls,
SignInFrequency,
UsersConditions,
)
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
def _make_policy(
*,
display_name="Test Policy",
state=ConditionalAccessPolicyState.ENABLED,
included_users=None,
excluded_users=None,
included_groups=None,
excluded_groups=None,
included_roles=None,
excluded_roles=None,
):
"""Build a ConditionalAccessPolicy with the minimum fields required by the model."""
policy_id = str(uuid4())
policy = ConditionalAccessPolicy(
id=policy_id,
display_name=display_name,
conditions=Conditions(
application_conditions=ApplicationsConditions(
included_applications=[],
excluded_applications=[],
included_user_actions=[],
),
user_conditions=UsersConditions(
included_users=included_users or [],
excluded_users=excluded_users or [],
included_groups=included_groups or [],
excluded_groups=excluded_groups or [],
included_roles=included_roles or [],
excluded_roles=excluded_roles or [],
),
client_app_types=[],
),
grant_controls=GrantControls(
built_in_controls=[],
operator=GrantControlOperator.OR,
authentication_strength=None,
),
session_controls=SessionControls(
persistent_browser=PersistentBrowser(is_enabled=False, mode=""),
sign_in_frequency=SignInFrequency(
is_enabled=False, frequency=None, type=None, interval=None
),
),
state=state,
)
return policy_id, policy
def _entra_client_mock():
client = mock.MagicMock
client.audited_tenant = "audited_tenant"
client.audited_domain = DOMAIN
return client
CHECK_MODULE = (
"prowler.providers.m365.services.entra."
"entra_conditional_access_policy_no_deleted_object_references."
"entra_conditional_access_policy_no_deleted_object_references.entra_client"
)
class Test_entra_conditional_access_policy_no_deleted_object_references:
def test_no_policies(self):
"""No Conditional Access policies in tenant: no findings."""
entra_client = _entra_client_mock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {}
entra_client.unresolved_directory_object_references = set()
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 0
def test_sentinel_only_references_pass(self):
"""Policy with only sentinel values ('All', 'GuestsOrExternalUsers') passes."""
entra_client = _entra_client_mock()
policy_id, policy = _make_policy(
display_name="MFA For All",
included_users=["All"],
excluded_users=["GuestsOrExternalUsers"],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = set()
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
"references no deleted directory objects" in result[0].status_extended
)
assert result[0].resource_id == policy_id
assert result[0].resource_name == "MFA For All"
def test_all_references_resolve_pass(self):
"""Policy with real identifiers, none in the unresolved set: PASS."""
entra_client = _entra_client_mock()
live_user = str(uuid4())
live_group = str(uuid4())
policy_id, policy = _make_policy(
display_name="Targeted Policy",
included_users=[live_user],
included_groups=[live_group],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = set()
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
def test_deleted_user_in_include_fails(self):
"""Policy referencing a deleted user in includeUsers fails with type+side reported."""
entra_client = _entra_client_mock()
deleted_user = str(uuid4())
policy_id, policy = _make_policy(
display_name="Require MFA",
included_users=[deleted_user],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = {
("user", deleted_user)
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "1 deleted directory object(s)" in result[0].status_extended
assert "users:" in result[0].status_extended
assert deleted_user in result[0].status_extended
assert "(include)" in result[0].status_extended
def test_deleted_group_in_exclude_fails(self):
"""Policy referencing a deleted group in excludeGroups fails with exclude side reported."""
entra_client = _entra_client_mock()
deleted_group = str(uuid4())
policy_id, policy = _make_policy(
display_name="Block Legacy Auth",
included_users=["All"],
excluded_groups=[deleted_group],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = {
("group", deleted_group)
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "groups:" in result[0].status_extended
assert "(exclude)" in result[0].status_extended
def test_deleted_role_in_disabled_policy_still_fails(self):
"""Disabled policy with a stale role reference still FAILs (per spec)."""
entra_client = _entra_client_mock()
deleted_role = str(uuid4())
policy_id, policy = _make_policy(
display_name="Legacy Admin Policy",
state=ConditionalAccessPolicyState.DISABLED,
included_roles=[deleted_role],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = {
("role", deleted_role)
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "roles:" in result[0].status_extended
assert deleted_role in result[0].status_extended
def test_orphans_grouped_by_type_across_collections(self):
"""A single policy with orphans of every type aggregates them grouped by type."""
entra_client = _entra_client_mock()
deleted_user = str(uuid4())
deleted_group = str(uuid4())
deleted_role = str(uuid4())
policy_id, policy = _make_policy(
display_name="Composite Policy",
included_users=[deleted_user],
excluded_groups=[deleted_group],
included_roles=[deleted_role],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = {
("user", deleted_user),
("group", deleted_group),
("role", deleted_role),
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "3 deleted directory object(s)" in result[0].status_extended
assert "users:" in result[0].status_extended
assert "groups:" in result[0].status_extended
assert "roles:" in result[0].status_extended
def test_report_only_policy_failure_notes_mode(self):
"""A report-only policy with an orphan FAILs and flags the not-yet-enforced state."""
entra_client = _entra_client_mock()
deleted_user = str(uuid4())
policy_id, policy = _make_policy(
display_name="Report Only MFA",
state=ConditionalAccessPolicyState.ENABLED_FOR_REPORTING,
included_users=[deleted_user],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {policy_id: policy}
entra_client.unresolved_directory_object_references = {
("user", deleted_user)
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "report-only mode" in result[0].status_extended
def test_multiple_policies_mixed(self):
"""Two policies: one clean, one with an orphan. Distinct PASS/FAIL findings."""
entra_client = _entra_client_mock()
deleted_user = str(uuid4())
clean_id, clean_policy = _make_policy(
display_name="Clean Policy",
included_users=["All"],
)
dirty_id, dirty_policy = _make_policy(
display_name="Stale Reference Policy",
excluded_users=[deleted_user],
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(CHECK_MODULE, new=entra_client),
):
from prowler.providers.m365.services.entra.entra_conditional_access_policy_no_deleted_object_references.entra_conditional_access_policy_no_deleted_object_references import (
entra_conditional_access_policy_no_deleted_object_references,
)
entra_client.conditional_access_policies = {
clean_id: clean_policy,
dirty_id: dirty_policy,
}
entra_client.unresolved_directory_object_references = {
("user", deleted_user)
}
check = entra_conditional_access_policy_no_deleted_object_references()
result = check.execute()
assert len(result) == 2
clean_result = next(r for r in result if r.resource_id == clean_id)
dirty_result = next(r for r in result if r.resource_id == dirty_id)
assert clean_result.status == "PASS"
assert dirty_result.status == "FAIL"
assert "(exclude)" in dirty_result.status_extended
@@ -873,3 +873,175 @@ class Test_Entra_Service:
assert merged.password_credentials[0].key_id == "cred-app"
assert merged.password_credentials[0].display_name == "app-level-secret"
assert merged.password_credentials[0].is_active()
def test__resolve_identifiers_for_type_flags_only_404(self):
"""Only HTTP 404 / Request_ResourceNotFound mark an id as deleted.
Transient errors (5xx, throttling) and successful resolutions must
never be added to the unresolved set — that is the contract the check
relies on to avoid false positives during Graph outages.
"""
from msgraph.generated.models.o_data_errors.main_error import MainError
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
deleted_by_status = "deleted-status-404"
deleted_by_code = "deleted-code-rnf"
transient = "transient-503"
live = "live-user"
error_404 = ODataError()
error_404.response_status_code = 404
error_404.error = None # status code alone is enough
error_rnf = ODataError()
error_rnf.response_status_code = None
error_rnf.error = MainError()
error_rnf.error.code = "Request_ResourceNotFound"
error_503 = ODataError()
error_503.response_status_code = 503
error_503.error = MainError()
error_503.error.code = "ServiceUnavailable"
user_builders = {
deleted_by_status: SimpleNamespace(get=AsyncMock(side_effect=error_404)),
deleted_by_code: SimpleNamespace(get=AsyncMock(side_effect=error_rnf)),
transient: SimpleNamespace(get=AsyncMock(side_effect=error_503)),
live: SimpleNamespace(get=AsyncMock(return_value=SimpleNamespace(id=live))),
}
entra_service = Entra.__new__(Entra)
entra_service.client = SimpleNamespace(
users=SimpleNamespace(
by_user_id=MagicMock(side_effect=lambda uid: user_builders[uid])
)
)
unresolved = set()
asyncio.run(
entra_service._resolve_identifiers_for_type(
"user", set(user_builders), unresolved
)
)
assert unresolved == {
("user", deleted_by_status),
("user", deleted_by_code),
}
def test__resolve_identifiers_for_type_role_uses_role_definitions_endpoint(self):
"""A deleted role is resolved against the roleDefinitions endpoint."""
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
deleted_role = "deleted-role-id"
error_404 = ODataError()
error_404.response_status_code = 404
error_404.error = None
by_role_id = MagicMock(
return_value=SimpleNamespace(get=AsyncMock(side_effect=error_404))
)
entra_service = Entra.__new__(Entra)
entra_service.client = SimpleNamespace(
role_management=SimpleNamespace(
directory=SimpleNamespace(
role_definitions=SimpleNamespace(
by_unified_role_definition_id=by_role_id
)
)
)
)
unresolved = set()
asyncio.run(
entra_service._resolve_identifiers_for_type(
"role", {deleted_role}, unresolved
)
)
assert unresolved == {("role", deleted_role)}
by_role_id.assert_called_once_with(deleted_role)
def test__resolve_directory_object_references_skips_sentinels_and_dedups(self):
"""End-to-end resolver: sentinels are never queried, ids are deduped
across policies, and only deleted ids land in the unresolved set."""
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
deleted_user = "deleted-user-id"
live_user = "live-user-id"
deleted_group = "deleted-group-id"
def _user_conditions(**kwargs):
base = {
"included_users": [],
"excluded_users": [],
"included_groups": [],
"excluded_groups": [],
"included_roles": [],
"excluded_roles": [],
}
base.update(kwargs)
return SimpleNamespace(**base)
def _policy(user_conditions):
return SimpleNamespace(
conditions=SimpleNamespace(user_conditions=user_conditions)
)
policies = {
"policy-a": _policy(
_user_conditions(
included_users=["All", deleted_user, live_user],
excluded_groups=[deleted_group],
)
),
# Same deleted_user referenced again — must be resolved only once.
"policy-b": _policy(
_user_conditions(
included_users=[deleted_user],
excluded_users=["GuestsOrExternalUsers"],
)
),
# Policy without user conditions must be skipped without error.
"policy-c": SimpleNamespace(
conditions=SimpleNamespace(user_conditions=None)
),
}
error_404 = ODataError()
error_404.response_status_code = 404
error_404.error = None
user_builders = {
deleted_user: SimpleNamespace(get=AsyncMock(side_effect=error_404)),
live_user: SimpleNamespace(
get=AsyncMock(return_value=SimpleNamespace(id=live_user))
),
}
group_builders = {
deleted_group: SimpleNamespace(get=AsyncMock(side_effect=error_404)),
}
by_user_id = MagicMock(side_effect=lambda uid: user_builders[uid])
by_group_id = MagicMock(side_effect=lambda gid: group_builders[gid])
entra_service = Entra.__new__(Entra)
entra_service.client = SimpleNamespace(
users=SimpleNamespace(by_user_id=by_user_id),
groups=SimpleNamespace(by_group_id=by_group_id),
)
unresolved = asyncio.run(
entra_service._resolve_directory_object_references(policies)
)
assert unresolved == {
("user", deleted_user),
("group", deleted_group),
}
# Sentinels are filtered before any Graph call; only the two real user
# ids are queried, and the deduped deleted_user is queried exactly once.
queried_users = {call.args[0] for call in by_user_id.call_args_list}
assert queried_users == {deleted_user, live_user}
assert user_builders[deleted_user].get.await_count == 1