chore(revision): add missing changes

This commit is contained in:
pedrooot committed 2026-07-23 16:21:21 +02:00
1 parent 3d50bed244
commit f07cc4cad6
37 files changed
+640 -147

No files matched your search

@@ -18,6 +18,14 @@ class admincenter_forms_phishing_protection_enabled(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether Microsoft Forms internal phishing protection is enabled.
Reads the Microsoft Forms org settings and reports PASS when internal
phishing protection is enabled, and FAIL otherwise.
Returns:
List[CheckReportM365]: The findings for the check.
"""
findings = []
settings = admincenter_client.forms_settings
if not settings:
@@ -18,6 +18,15 @@ class admincenter_office_store_apps_restricted(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether Office Store access and trials are restricted.
Reads the org-wide apps and services settings and reports PASS when both
Office Store access and trials on behalf of the organization are disabled,
and FAIL otherwise.
Returns:
List[CheckReportM365]: The findings for the check.
"""
findings = []
settings = admincenter_client.apps_and_services_settings
if not settings:
@@ -87,6 +87,14 @@ class AdminCenter(M365Service):
return organization_config
def _get_mailbox_policy(self):
"""Retrieve the OWA mailbox policies via Exchange Online PowerShell.
Reads the OWA mailbox policy configuration and captures each policy's
default flag and Bookings mailbox creation setting.
Returns:
List[OwaMailboxPolicy]: The parsed OWA mailbox policies, empty on error.
"""
logger.info("Microsoft365 - Getting OWA mailbox policy configuration...")
mailbox_policies = []
try:
@@ -359,6 +367,15 @@ class Organization(BaseModel):
class OwaMailboxPolicy(BaseModel):
"""Represents an Outlook on the web (OWA) mailbox policy.
Attributes:
id: The mailbox policy identifier.
is_default: Whether the policy is the default OWA mailbox policy.
bookings_mailbox_creation_enabled: Whether users can create Bookings
mailboxes under this policy.
"""
id: str
is_default: bool = False
bookings_mailbox_creation_enabled: bool = True
@@ -371,9 +388,24 @@ class SharingPolicy(BaseModel):
class AppsAndServicesSettings(BaseModel):
"""Represents the org-wide apps and services settings.
Attributes:
office_store_enabled: Whether users can access the Office Store.
app_and_services_trial_enabled: Whether users can start trials on behalf
of the organization.
"""
office_store_enabled: bool = True
app_and_services_trial_enabled: bool = True
class FormsSettings(BaseModel):
"""Represents the Microsoft Forms org settings.
Attributes:
in_org_forms_phishing_scan_enabled: Whether internal phishing protection
scans forms for phishing keywords.
"""
in_org_forms_phishing_scan_enabled: bool = False
@@ -20,6 +20,16 @@ class defender_priority_account_protection_enabled(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the priority account protection check.
Evaluates the tenant-level ``EnablePriorityAccountProtection`` flag from
the Defender email tenant settings, producing PASS when enabled and FAIL
when disabled. Returns no findings when settings are unavailable.
Returns:
List[CheckReportM365]: A list with the check report, or empty when no
email tenant settings are available.
"""
findings = []
settings = defender_client.email_tenant_settings
if not settings:
@@ -19,6 +19,18 @@ class defender_strict_preset_security_policy_enabled(Check):
"""
def _has_enabled_strict_preset(self, rules) -> bool:
"""Check whether any rule enables the Strict Preset Security Policy.
A rule qualifies only when it is named the Strict Preset Security Policy,
is in the ``Enabled`` state, and targets at least one recipient scope.
Args:
rules: Iterable of preset security policy rules (EOP or ATP).
Returns:
bool: True if at least one rule enables the Strict Preset Security
Policy with recipients, False otherwise.
"""
return any(
rule.name == STRICT_PRESET_NAME
and rule.state == "Enabled"
@@ -27,6 +39,15 @@ class defender_strict_preset_security_policy_enabled(Check):
)
def execute(self) -> List[CheckReportM365]:
"""Execute the Strict Preset Security Policy check.
Evaluates whether the Strict Preset Security Policy is enabled for both
Exchange Online Protection (EOP) and Defender for Office 365 (ATP),
producing PASS only when both are enabled with recipients.
Returns:
List[CheckReportM365]: A single-element list with the check report.
"""
findings = []
report = CheckReportM365(
metadata=self.metadata(),
@@ -19,6 +19,15 @@ class entra_access_review_guest_users_configured(Check):
"""
def _is_fail_closed(self, definition) -> bool:
"""Determine whether an access review definition is fail-closed.
Args:
definition: The access review definition to evaluate.
Returns:
bool: True if the review denies access by default, auto-applies
decisions, and has mail notifications and reminders enabled.
"""
return (
definition.default_decision == "Deny"
and definition.auto_apply_enabled
@@ -27,6 +36,15 @@ class entra_access_review_guest_users_configured(Check):
)
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether a fail-closed access review for guest users exists.
Searches the tenant's access review definitions for an active, fail-closed
review scoped to guest users.
Returns:
List[CheckReportM365]: A single report indicating whether an active,
fail-closed access review scoped to guest users is configured.
"""
findings = []
definitions = entra_client.access_review_definitions
@@ -21,7 +21,19 @@ class entra_access_review_privileged_roles_configured(Check):
"""
def _targets_privileged_roles(self, definition) -> bool:
# For PIM role reviews the role reference lives in the resource scopes.
"""Determine whether an access review targets privileged directory roles.
For PIM role reviews the role reference lives in the resource scopes, so both
the scope query and the resource scope queries are inspected for markers that
indicate directory role assignments.
Args:
definition: The access review definition to evaluate.
Returns:
bool: True if any of the review's scope queries reference privileged
(PIM) directory roles.
"""
queries = [definition.scope_query] + definition.resource_scope_queries
return any(
marker in query.lower()
@@ -30,6 +42,15 @@ class entra_access_review_privileged_roles_configured(Check):
)
def _is_fail_closed(self, definition) -> bool:
"""Determine whether an access review definition is fail-closed.
Args:
definition: The access review definition to evaluate.
Returns:
bool: True if the review denies access by default, auto-applies
decisions, and has mail notifications and reminders enabled.
"""
return (
definition.default_decision == "Deny"
and definition.auto_apply_enabled
@@ -38,6 +59,15 @@ class entra_access_review_privileged_roles_configured(Check):
)
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether a fail-closed access review for privileged roles exists.
Searches the tenant's access review definitions for an active, fail-closed
review scoped to privileged (PIM) directory roles.
Returns:
List[CheckReportM365]: A single report indicating whether an active,
fail-closed access review scoped to privileged roles is configured.
"""
findings = []
definitions = entra_client.access_review_definitions
@@ -17,6 +17,17 @@ class entra_authentication_method_authenticator_companion_app_disabled(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the Microsoft Authenticator companion app disabled check.
Verifies that Microsoft Authenticator on companion applications (Authenticator
Lite) is disabled so that MFA remains bound to the full Microsoft Authenticator
app.
Returns:
List[CheckReportM365]: A list with a single report describing whether
Microsoft Authenticator on companion applications is disabled, or an empty
list when the authentication methods policy settings are not available.
"""
findings = []
settings = entra_client.authentication_methods_policy_settings
if not settings:
@@ -17,6 +17,18 @@ class entra_authentication_method_authenticator_show_context(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the Microsoft Authenticator show context check.
Verifies that the Microsoft Authenticator method is enabled and configured to
display both the application name and the geographic location of the sign-in in
push and passwordless notifications.
Returns:
List[CheckReportM365]: A list with a single report describing whether
Microsoft Authenticator shows application name and geographic location
context, or an empty list when the authentication methods policy settings
are not available.
"""
findings = []
settings = entra_client.authentication_methods_policy_settings
if not settings:
@@ -35,7 +47,8 @@ class entra_authentication_method_authenticator_show_context(Check):
)
if (
settings.authenticator_display_app_information_state == "enabled"
settings.authenticator_state == "enabled"
and settings.authenticator_display_app_information_state == "enabled"
and settings.authenticator_display_location_information_state == "enabled"
):
report.status = "PASS"
@@ -35,15 +35,21 @@ class entra_authentication_method_email_otp_disabled(Check):
resource_id=entra_client.tenant_domain,
)
if email_config.state == "enabled":
if email_config.state == "disabled":
report.status = "PASS"
report.status_extended = (
"Email OTP authentication method is disabled in the tenant."
)
elif email_config.state == "enabled":
report.status = "FAIL"
report.status_extended = (
"Email OTP authentication method is enabled in the tenant."
)
else:
report.status = "PASS"
report.status = "FAIL"
report.status_extended = (
"Email OTP authentication method is disabled in the tenant."
"Email OTP authentication method state could not be determined; "
"treating as enabled/non-compliant."
)
findings.append(report)
@@ -18,6 +18,8 @@ class entra_conditional_access_trusted_named_location_exists(Check):
def execute(self) -> List[CheckReportM365]:
findings = []
named_locations = entra_client.named_locations
if not named_locations:
return findings
report = CheckReportM365(
metadata=self.metadata(),
@@ -24,34 +24,41 @@ class entra_idle_session_timeout_configured(Check):
findings = []
policies = entra_client.activity_based_timeout_policies
report = CheckReportM365(
metadata=self.metadata(),
resource=policies if policies else {},
resource_name="Activity Based Timeout Policies",
resource_id="activityBasedTimeoutPolicies",
)
report.status = "FAIL"
report.status_extended = (
"No activity-based timeout policy enforces an idle session timeout of 3 "
"hours or less."
)
if not policies:
report = CheckReportM365(
metadata=self.metadata(),
resource={},
resource_name="Activity Based Timeout Policies",
resource_id="activityBasedTimeoutPolicies",
)
report.status = "FAIL"
report.status_extended = (
"No activity-based timeout policy enforces an idle session timeout of "
"3 hours or less."
)
findings.append(report)
return findings
for policy in policies:
report = CheckReportM365(
metadata=self.metadata(),
resource=policy,
resource_name=policy.display_name or "Activity Based Timeout Policy",
resource_id=policy.id,
)
timeout = policy.web_session_idle_timeout_seconds
if timeout is not None and timeout <= MAX_IDLE_TIMEOUT_SECONDS:
report = CheckReportM365(
metadata=self.metadata(),
resource=policy,
resource_name=policy.display_name
or "Activity Based Timeout Policy",
resource_id=policy.id,
)
report.status = "PASS"
report.status_extended = (
f"Activity-based timeout policy '{policy.display_name or policy.id}' "
f"enforces an idle session timeout of {timeout // 60} minutes."
)
break
else:
report.status = "FAIL"
report.status_extended = (
f"Activity-based timeout policy '{policy.display_name or policy.id}' "
"does not enforce an idle session timeout of 3 hours or less."
)
findings.append(report)
findings.append(report)
return findings
@@ -20,10 +20,22 @@ class entra_password_protection_custom_banned_list_enforced(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the custom banned password list enforcement check.
Evaluate whether the Password Rule Settings directory setting enforces a
non-empty custom banned password list. When the settings object is absent,
no finding is produced.
Returns:
List[CheckReportM365]: A list with a single report when the Password Rule
Settings exist, or an empty list when they are absent.
"""
findings = []
settings = entra_client.directory_settings.get(
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
)
if not settings:
return findings
report = CheckReportM365(
metadata=self.metadata(),
@@ -26,6 +26,8 @@ class entra_password_protection_lockout_duration_configured(Check):
settings = entra_client.directory_settings.get(
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
)
if not settings:
return findings
report = CheckReportM365(
metadata=self.metadata(),
@@ -21,10 +21,22 @@ class entra_password_protection_lockout_threshold_limited(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the smart lockout threshold check.
Evaluate whether the Password Rule Settings directory setting limits the smart
lockout threshold to the recommended maximum. When the settings object is
absent, no finding is produced.
Returns:
List[CheckReportM365]: A list with a single report when the Password Rule
Settings exist, or an empty list when they are absent.
"""
findings = []
settings = entra_client.directory_settings.get(
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
)
if not settings:
return findings
report = CheckReportM365(
metadata=self.metadata(),
@@ -20,10 +20,22 @@ class entra_password_protection_on_premises_enforced(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Execute the on-premises password protection enforcement check.
Evaluate whether the Password Rule Settings directory setting enables and
enforces banned-password protection for on-premises Active Directory. When the
settings object is absent, no finding is produced.
Returns:
List[CheckReportM365]: A list with a single report when the Password Rule
Settings exist, or an empty list when they are absent.
"""
findings = []
settings = entra_client.directory_settings.get(
PASSWORD_RULE_SETTINGS_TEMPLATE_ID
)
if not settings:
return findings
report = CheckReportM365(
metadata=self.metadata(),
@@ -19,6 +19,15 @@ class entra_pim_global_administrator_approval_required(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate PIM approval settings for the Global Administrator role.
Reports whether Privileged Identity Management requires approval to activate
the Global Administrator role and whether at least one approver is configured.
Returns:
List[CheckReportM365]: A single report for the Global Administrator PIM
role settings, or an empty list when the settings are absent.
"""
findings = []
setting = entra_client.pim_role_approval_settings.get(
GLOBAL_ADMINISTRATOR_ROLE_TEMPLATE_ID
@@ -37,7 +46,12 @@ class entra_pim_global_administrator_approval_required(Check):
"PIM does not require approval to activate the Global Administrator role."
)
if setting.is_approval_required and setting.has_approvers:
if setting.is_approval_required and not setting.has_approvers:
report.status_extended = (
"PIM requires approval to activate the Global Administrator role but "
"no approvers are configured."
)
elif setting.is_approval_required and setting.has_approvers:
report.status = "PASS"
report.status_extended = (
"PIM requires approval to activate the Global Administrator role and "
@@ -15,11 +15,21 @@ class entra_pim_privileged_role_administrator_approval_required(Check):
- PASS: Approval is required to activate Privileged Role Administrator and
approvers exist.
- FAIL: Approval is not required, no approvers are configured, or PIM settings are
unavailable.
- FAIL: Approval is not required or no approvers are configured.
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate PIM approval settings for the Privileged Role Administrator role.
Reports whether Privileged Identity Management requires approval to activate
the Privileged Role Administrator role and whether at least one approver is
configured.
Returns:
List[CheckReportM365]: A single report for the Privileged Role
Administrator PIM role settings, or an empty list when the settings are
absent.
"""
findings = []
setting = entra_client.pim_role_approval_settings.get(
PRIVILEGED_ROLE_ADMINISTRATOR_ROLE_TEMPLATE_ID
@@ -19,6 +19,16 @@ class entra_policy_default_user_cannot_create_m365_groups(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether default users can create Microsoft 365 groups.
Inspects the Group.Unified directory setting to determine whether non-admin
users are allowed to create Microsoft 365 groups. When the setting is absent
the tenant default (group creation allowed) applies.
Returns:
List[CheckReportM365]: A single report indicating whether non-admin users
are restricted from creating Microsoft 365 groups.
"""
findings = []
settings = entra_client.directory_settings.get(
GROUP_UNIFIED_SETTINGS_TEMPLATE_ID
@@ -15,6 +15,16 @@ class entra_policy_guest_invitations_restricted_to_allowed_domains(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether guest invitations are restricted to allowed domains.
Inspects the B2B collaboration policy to determine whether guest invitations
are limited to an allow-list of domains that contains at least one domain.
Returns:
List[CheckReportM365]: A single report indicating whether guest
invitations are restricted to a non-empty allow-list of domains, or an
empty list when the policy is absent.
"""
findings = []
policy = entra_client.b2b_collaboration_policy
if not policy:
@@ -31,18 +41,12 @@ class entra_policy_guest_invitations_restricted_to_allowed_domains(Check):
"Guest invitations are not restricted to an allow-list of domains."
)
if policy.invitations_restricted_to_allowed_domains:
if policy.invitations_restricted_to_allowed_domains and policy.allowed_domains:
report.status = "PASS"
if policy.allowed_domains:
report.status_extended = (
"Guest invitations are restricted to an allow-list of "
f"{len(policy.allowed_domains)} domain(s)."
)
else:
report.status_extended = (
"Guest invitations are restricted to an allow-list with no "
"domains (all external invitations are blocked)."
)
report.status_extended = (
"Guest invitations are restricted to an allow-list of "
f"{len(policy.allowed_domains)} domain(s)."
)
findings.append(report)
return findings
@@ -1313,35 +1313,49 @@ OAuthAppInfo
"roleManagementPolicyAssignments?$filter=scopeId%20eq%20'/'%20and%20"
"scopeType%20eq%20'DirectoryRole'&$expand=policy($expand=rules)"
)
builder = self.client.policies.with_url(url)
request_info = builder.to_get_request_information()
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if response:
request_info = self.client.policies.with_url(
url
).to_get_request_information()
assignments = []
while True:
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if not response:
break
data = json.loads(response)
for assignment in data.get("value", []) or []:
role_id = assignment.get("roleDefinitionId")
if not role_id:
continue
rules = (assignment.get("policy", {}) or {}).get("rules", []) or []
is_approval_required = False
has_approvers = False
for rule in rules:
if rule.get("id") == "Approval_EndUser_Assignment":
setting = rule.get("setting", {}) or {}
is_approval_required = bool(
setting.get("isApprovalRequired", False)
)
for stage in setting.get("approvalStages", []) or []:
if stage.get("primaryApprovers"):
has_approvers = True
break
settings[role_id] = PimRoleApprovalSetting(
role_definition_id=role_id,
is_approval_required=is_approval_required,
has_approvers=has_approvers,
)
page = data.get("value", []) or []
if not page:
break
assignments.extend(page)
next_link = data.get("@odata.nextLink") or data.get("nextLink")
if not next_link:
break
request_info = self.client.policies.with_url(
next_link
).to_get_request_information()
for assignment in assignments:
role_id = assignment.get("roleDefinitionId")
if not role_id:
continue
rules = (assignment.get("policy", {}) or {}).get("rules", []) or []
is_approval_required = False
has_approvers = False
for rule in rules:
if rule.get("id") == "Approval_EndUser_Assignment":
setting = rule.get("setting", {}) or {}
is_approval_required = bool(
setting.get("isApprovalRequired", False)
)
for stage in setting.get("approvalStages", []) or []:
if stage.get("primaryApprovers"):
has_approvers = True
break
settings[role_id] = PimRoleApprovalSetting(
role_definition_id=role_id,
is_approval_required=is_approval_required,
has_approvers=has_approvers,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -1364,47 +1378,61 @@ OAuthAppInfo
"https://graph.microsoft.com/v1.0/identityGovernance/"
"accessReviews/definitions"
)
builder = self.client.identity_governance.with_url(url)
request_info = builder.to_get_request_information()
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if response:
request_info = self.client.identity_governance.with_url(
url
).to_get_request_information()
raw_definitions = []
while True:
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if not response:
break
data = json.loads(response)
for definition in data.get("value", []) or []:
scope = definition.get("scope", {}) or {}
settings = definition.get("settings", {}) or {}
definitions.append(
AccessReviewDefinition(
id=definition.get("id", ""),
display_name=definition.get("displayName"),
status=definition.get("status"),
scope_query=str(scope.get("query", "")),
resource_scope_queries=[
str(resource_scope.get("query", ""))
for resource_scope in (
scope.get("resourceScopes", []) or []
)
],
principal_scope_queries=[
str(principal_scope.get("query", ""))
for principal_scope in (
scope.get("principalScopes", []) or []
)
],
default_decision=settings.get("defaultDecision"),
auto_apply_enabled=bool(
settings.get("autoApplyDecisionsEnabled", False)
),
mail_notifications_enabled=bool(
settings.get("mailNotificationsEnabled", False)
),
reminders_enabled=bool(
settings.get("reminderNotificationsEnabled", False)
),
duration_in_days=settings.get("instanceDurationInDays"),
)
page = data.get("value", []) or []
if not page:
break
raw_definitions.extend(page)
next_link = data.get("@odata.nextLink") or data.get("nextLink")
if not next_link:
break
request_info = self.client.identity_governance.with_url(
next_link
).to_get_request_information()
for definition in raw_definitions:
scope = definition.get("scope", {}) or {}
settings = definition.get("settings", {}) or {}
definitions.append(
AccessReviewDefinition(
id=definition.get("id", ""),
display_name=definition.get("displayName"),
status=definition.get("status"),
scope_query=str(scope.get("query", "")),
resource_scope_queries=[
str(resource_scope.get("query", ""))
for resource_scope in (
scope.get("resourceScopes", []) or []
)
],
principal_scope_queries=[
str(principal_scope.get("query", ""))
for principal_scope in (
scope.get("principalScopes", []) or []
)
],
default_decision=settings.get("defaultDecision"),
auto_apply_enabled=bool(
settings.get("autoApplyDecisionsEnabled", False)
),
mail_notifications_enabled=bool(
settings.get("mailNotificationsEnabled", False)
),
reminders_enabled=bool(
settings.get("reminderNotificationsEnabled", False)
),
duration_in_days=settings.get("instanceDurationInDays"),
)
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -1488,23 +1516,38 @@ OAuthAppInfo
request_info = (
self.client.identity.conditional_access.named_locations.to_get_request_information()
)
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if response:
raw_locations = []
while True:
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if not response:
break
data = json.loads(response)
for location in data.get("value", []) or []:
odata_type = location.get("@odata.type", "")
ip_ranges = location.get("ipRanges", []) or []
named_locations.append(
NamedLocation(
id=location.get("id", ""),
display_name=location.get("displayName"),
is_trusted=bool(location.get("isTrusted", False)),
is_ip_location="ipNamedLocation" in odata_type,
ip_ranges_count=len(ip_ranges),
)
page = data.get("value", []) or []
if not page:
break
raw_locations.extend(page)
next_link = data.get("@odata.nextLink") or data.get("nextLink")
if not next_link:
break
request_info = (
self.client.identity.conditional_access.named_locations.with_url(
next_link
).to_get_request_information()
)
for location in raw_locations:
odata_type = location.get("@odata.type", "")
ip_ranges = location.get("ipRanges", []) or []
named_locations.append(
NamedLocation(
id=location.get("id", ""),
display_name=location.get("displayName"),
is_trusted=bool(location.get("isTrusted", False)),
is_ip_location="ipNamedLocation" in odata_type,
ip_ranges_count=len(ip_ranges),
)
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -18,6 +18,15 @@ class intune_device_enrollment_personal_device_restricted(Check):
"""
def execute(self) -> List[CheckReportM365]:
"""Evaluate whether the default enrollment restriction blocks personal devices.
Reads the default (priority 0) device platform restriction configuration and
reports PASS when personally owned devices are blocked for all platforms, and
FAIL when they are allowed for one or more platforms.
Returns:
List[CheckReportM365]: The findings for the check.
"""
findings = []
configurations = intune_client.device_enrollment_configurations
@@ -106,17 +106,20 @@ class Intune(M365Service):
try:
# The beta endpoint exposes the full set of per-platform restrictions
# (including macOS/Android for Work) that v1.0 may omit.
builder = (
self.client.device_management.device_enrollment_configurations.with_url(
"https://graph.microsoft.com/beta/deviceManagement/"
"deviceEnrollmentConfigurations"
next_link = (
"https://graph.microsoft.com/beta/deviceManagement/"
"deviceEnrollmentConfigurations"
)
while next_link:
builder = self.client.device_management.device_enrollment_configurations.with_url(
next_link
)
)
request_info = builder.to_get_request_information()
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if response:
request_info = builder.to_get_request_information()
response = await self.client.request_adapter.send_primitive_async(
request_info, "bytes", {}
)
if not response:
break
data = json.loads(response)
for config in data.get("value", []) or []:
odata_type = config.get("@odata.type", "")
@@ -143,6 +146,9 @@ class Intune(M365Service):
platform_restrictions=platform_restrictions,
)
)
# Follow pagination using the opaque nextLink until exhausted.
next_link = data.get("@odata.nextLink") or data.get("nextLink")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -390,6 +396,17 @@ class IntuneManagedDevice(BaseModel):
class DeviceEnrollmentConfiguration(BaseModel):
"""Represents an Intune device enrollment (platform restrictions) configuration.
Attributes:
id: The device enrollment configuration identifier.
priority: The configuration priority (0 is the default configuration).
odata_type: The Graph ``@odata.type`` of the configuration.
platform_restrictions: Mapping of platform restriction name to a boolean
indicating whether personal-device enrollment is blocked (or the
platform is fully blocked) for that platform.
"""
id: str
priority: int = 0
odata_type: Optional[str] = None
@@ -31,6 +31,7 @@ class Test_entra_authentication_method_authenticator_show_context:
def test_both_enabled(self):
result = self._run(
AuthenticationMethodsPolicySettings(
authenticator_state="enabled",
authenticator_display_app_information_state="enabled",
authenticator_display_location_information_state="enabled",
)
@@ -40,8 +41,19 @@ class Test_entra_authentication_method_authenticator_show_context:
def test_location_disabled(self):
result = self._run(
AuthenticationMethodsPolicySettings(
authenticator_state="enabled",
authenticator_display_app_information_state="enabled",
authenticator_display_location_information_state="disabled",
)
)
assert result[0].status == "FAIL"
def test_context_enabled_but_authenticator_disabled(self):
result = self._run(
AuthenticationMethodsPolicySettings(
authenticator_state="disabled",
authenticator_display_app_information_state="enabled",
authenticator_display_location_information_state="enabled",
)
)
assert result[0].status == "FAIL"
@@ -101,3 +101,39 @@ class Test_entra_authentication_method_email_otp_disabled:
result[0].status_extended
== "Email OTP authentication method is disabled in the tenant."
)
def test_email_otp_unknown_state(self):
entra_client = mock.MagicMock
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled.entra_client",
new=entra_client,
),
):
from prowler.providers.m365.services.entra.entra_authentication_method_email_otp_disabled.entra_authentication_method_email_otp_disabled import (
entra_authentication_method_email_otp_disabled,
)
entra_client.authentication_method_configurations = {
"Email": AuthenticationMethodConfiguration(
id="Email",
state="unknown",
),
}
entra_client.tenant_domain = DOMAIN
check = entra_authentication_method_email_otp_disabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Email OTP authentication method state could not be determined; "
"treating as enabled/non-compliant."
)
@@ -25,8 +25,7 @@ class Test_entra_conditional_access_trusted_named_location_exists:
def test_no_locations(self):
result = self._run([])
assert len(result) == 1
assert result[0].status == "FAIL"
assert len(result) == 0
def test_trusted_ip_location(self):
result = self._run(
@@ -41,6 +41,15 @@ class Test_entra_device_registration_registering_user_not_local_admin:
assert result[0].status == "FAIL"
def test_selected_registering_users(self):
result = self._run(
DeviceRegistrationPolicy(
azure_ad_join_registering_users_type=DeviceRegistrationMembershipType.ENUMERATED.value
)
)
assert len(result) == 1
assert result[0].status == "PASS"
def test_none_registering_users(self):
result = self._run(
DeviceRegistrationPolicy(
azure_ad_join_registering_users_type=DeviceRegistrationMembershipType.NONE.value
@@ -10,7 +10,7 @@ CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protec
class Test_entra_password_protection_custom_banned_list_enforced:
def _run(self, directory_settings):
entra_client = mock.MagicMock
entra_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -27,8 +27,7 @@ class Test_entra_password_protection_custom_banned_list_enforced:
def test_template_absent(self):
result = self._run({})
assert len(result) == 1
assert result[0].status == "FAIL"
assert len(result) == 0
def test_enforced_with_list(self):
result = self._run(
@@ -10,7 +10,7 @@ CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_password_protec
class Test_entra_password_protection_lockout_duration_configured:
def _run(self, directory_settings):
entra_client = mock.MagicMock
entra_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -26,7 +26,7 @@ class Test_entra_password_protection_lockout_duration_configured:
return entra_password_protection_lockout_duration_configured().execute()
def test_template_absent(self):
assert self._run({})[0].status == "FAIL"
assert len(self._run({})) == 0
def test_at_minimum(self):
result = self._run(
@@ -26,7 +26,7 @@ class Test_entra_password_protection_lockout_threshold_limited:
return entra_password_protection_lockout_threshold_limited().execute()
def test_template_absent(self):
assert self._run({})[0].status == "FAIL"
assert len(self._run({})) == 0
def test_within_limit(self):
result = self._run(
@@ -25,8 +25,9 @@ class Test_entra_password_protection_on_premises_enforced:
entra_client.directory_settings = directory_settings
return entra_password_protection_on_premises_enforced().execute()
def test_template_absent(self):
assert self._run({})[0].status == "FAIL"
def test_no_resources(self):
result = self._run({})
assert len(result) == 0
def test_enabled_and_enforced(self):
result = self._run(
@@ -11,7 +11,7 @@ CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_pim_global_admi
class Test_entra_pim_global_administrator_approval_required:
def _run(self, settings):
entra_client = mock.MagicMock
entra_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -10,7 +10,7 @@ CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_policy_default_
class Test_entra_policy_default_user_cannot_create_m365_groups:
def _run(self, directory_settings):
entra_client = mock.MagicMock
entra_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -1,8 +1,6 @@
from unittest import mock
from prowler.providers.m365.services.entra.entra_service import (
B2BCollaborationPolicy,
)
from prowler.providers.m365.services.entra.entra_service import B2BCollaborationPolicy
from tests.providers.m365.m365_fixtures import set_mocked_m365_provider
CHECK_MODULE_PATH = "prowler.providers.m365.services.entra.entra_policy_guest_invitations_restricted_to_allowed_domains.entra_policy_guest_invitations_restricted_to_allowed_domains"
@@ -47,3 +45,12 @@ class Test_entra_policy_guest_invitations_restricted_to_allowed_domains:
)
)
assert result[0].status == "FAIL"
def test_restricted_with_empty_allowed_domains(self):
result = self._run(
B2BCollaborationPolicy(
invitations_restricted_to_allowed_domains=True,
allowed_domains=[],
)
)
assert result[0].status == "FAIL"
@@ -1,5 +1,6 @@
import asyncio
import importlib
import json
from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
@@ -1196,3 +1197,62 @@ class Test_Entra_Service:
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
def test__get_named_locations_paginates_through_next_links(self):
entra_service = Entra.__new__(Entra)
page_one = json.dumps(
{
"value": [
{
"@odata.type": "#microsoft.graph.ipNamedLocation",
"id": "loc-1",
"displayName": "Trusted IPs",
"isTrusted": True,
"ipRanges": [{"cidrAddress": "10.0.0.0/8"}],
}
],
"@odata.nextLink": "next-link",
}
)
page_two = json.dumps(
{
"value": [
{
"@odata.type": "#microsoft.graph.countryNamedLocation",
"id": "loc-2",
"displayName": "Countries",
"isTrusted": False,
}
]
}
)
send_mock = AsyncMock(side_effect=[page_one, page_two])
next_link_builder = SimpleNamespace(
to_get_request_information=MagicMock(return_value="req-info-page-two")
)
with_url_mock = MagicMock(return_value=next_link_builder)
named_locations_builder = SimpleNamespace(
to_get_request_information=MagicMock(return_value="req-info-page-one"),
with_url=with_url_mock,
)
entra_service.client = SimpleNamespace(
identity=SimpleNamespace(
conditional_access=SimpleNamespace(
named_locations=named_locations_builder
)
),
request_adapter=SimpleNamespace(send_primitive_async=send_mock),
)
named_locations = asyncio.run(entra_service._get_named_locations())
# Both pages are requested and every entry is accumulated before parsing.
assert send_mock.await_count == 2
with_url_mock.assert_called_once_with("next-link")
assert [loc.id for loc in named_locations] == ["loc-1", "loc-2"]
assert named_locations[0].is_ip_location is True
assert named_locations[0].ip_ranges_count == 1
assert named_locations[1].is_ip_location is False
assert named_locations[1].is_trusted is False
@@ -10,7 +10,7 @@ CHECK_MODULE_PATH = "prowler.providers.m365.services.intune.intune_device_enroll
class Test_intune_device_enrollment_personal_device_restricted:
def _run(self, configurations):
intune_client = mock.MagicMock
intune_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -161,6 +161,64 @@ class Test_Intune_DeviceEnrollmentConfigurations:
def test_empty_response(self):
assert self._run_getter({"value": []}) == []
def test_follows_pagination_across_pages(self):
import json
first_page = {
"value": [
{
"@odata.type": "#microsoft.graph.deviceEnrollmentPlatformRestrictionsConfiguration",
"id": "page-one",
"priority": 0,
"iosRestriction": {
"personalDeviceEnrollmentBlocked": True,
"platformBlocked": False,
},
},
],
"@odata.nextLink": "https://graph.microsoft.com/beta/deviceManagement/deviceEnrollmentConfigurations?$skiptoken=abc",
}
second_page = {
"value": [
{
"@odata.type": "#microsoft.graph.deviceEnrollmentPlatformRestrictionsConfiguration",
"id": "page-two",
"priority": 1,
"androidRestriction": {
"personalDeviceEnrollmentBlocked": False,
"platformBlocked": True,
},
},
],
}
service = Intune.__new__(Intune)
client = mock.MagicMock()
builder = (
client.device_management.device_enrollment_configurations.with_url.return_value
)
builder.to_get_request_information.return_value = mock.MagicMock()
client.request_adapter.send_primitive_async = AsyncMock(
side_effect=[
json.dumps(first_page).encode(),
json.dumps(second_page).encode(),
]
)
service.client = client
loop = asyncio.new_event_loop()
try:
result = loop.run_until_complete(
service._get_device_enrollment_configurations()
)
finally:
loop.close()
# Configurations from BOTH pages must be collected.
assert len(result) == 2
assert {config.id for config in result} == {"page-one", "page-two"}
assert client.request_adapter.send_primitive_async.await_count == 2
class Test_Intune_Service:
def test_get_settings_secure_by_default_true(self):