feat(exchange): add new check exchange_mailbox_audit_bypass_disabled (#7418)

Co-authored-by: HugoPBrito <hugopbrit@gmail.com>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
Daniel Barranquero
2025-04-16 20:06:32 +02:00
committed by Pepe Fagoaga
parent 311c9a41ff
commit 00e33d39bb
14 changed files with 421 additions and 100 deletions
+1
View File
@@ -11,6 +11,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Support CLOUDSDK_AUTH_ACCESS_TOKEN in GCP [(#7495)](https://github.com/prowler-cloud/prowler/pull/7495).
- Add Powershell to Microsoft365 [(#7331)](https://github.com/prowler-cloud/prowler/pull/7331)
- Add service Exchange to Microsoft365 with one check for Organizations Mailbox Auditing enabled [(#7408)](https://github.com/prowler-cloud/prowler/pull/7408)
- Add check for Bypass Disable in every Mailbox for service Defender in M365 [(#7418)](https://github.com/prowler-cloud/prowler/pull/7418)
- Add new check `teams_email_sending_to_channel_disabled` [(#7533)](https://github.com/prowler-cloud/prowler/pull/7533)
### Fixed
@@ -188,3 +188,22 @@ class M365PowerShell(PowerShellSession):
}
"""
return self.execute("Get-OrganizationConfig | ConvertTo-Json")
def get_mailbox_audit_config(self) -> dict:
"""
Get Exchange Online Mailbox Audit Configuration.
Retrieves the current mailbox audit configuration settings for Exchange Online.
Returns:
dict: Mailbox audit configuration settings in JSON format.
Example:
>>> get_mailbox_audit_config()
{
"Name": "MyMailbox",
"Id": "12345678-1234-1234-1234-123456789012",
"AuditBypassEnabled": false
}
"""
return self.execute("Get-MailboxAuditBypassAssociation | ConvertTo-Json")
@@ -1,7 +1,6 @@
from asyncio import gather, get_event_loop
from typing import List, Optional
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
from pydantic import BaseModel
from prowler.lib.logger import logger
@@ -40,20 +39,6 @@ class AdminCenter(M365Service):
license_details = await self.client.users.by_user_id(
user.id
).license_details.get()
try:
mailbox_settings = await self.client.users.by_user_id(
user.id
).mailbox_settings.get()
mailbox_settings.user_purpose
except ODataError as error:
if error.error.code == "MailboxNotEnabledForRESTAPI":
logger.warning(
f"MailboxNotEnabledForRESTAPI for user {user.id}"
)
else:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
users.update(
{
user.id: User(
@@ -0,0 +1,30 @@
{
"Provider": "m365",
"CheckID": "exchange_mailbox_audit_bypass_disabled",
"CheckTitle": "Ensure 'AuditBypassEnabled' is not enabled on any mailbox in the organization.",
"CheckType": [],
"ServiceName": "exchange",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "high",
"ResourceType": "Exchange Mailboxes",
"Description": "Ensure that no mailboxes in the organization have 'AuditBypassEnabled' set to true. This setting prevents mailbox audit logging and can allow unauthorized access without traceability.",
"Risk": "If 'AuditBypassEnabled' is set to true for any mailbox, access to those mailboxes won't be logged, creating a blind spot in forensic analysis and increasing the risk of undetected malicious activity.",
"RelatedUrl": "https://learn.microsoft.com/en-us/powershell/module/exchange/get-mailboxauditbypassassociation?view=exchange-ps",
"Remediation": {
"Code": {
"CLI": "$MBXAudit = Get-MailboxAuditBypassAssociation -ResultSize unlimited | Where-Object { $_.AuditBypassEnabled -eq $true }; foreach ($mailbox in $MBXAudit) { $mailboxName = $mailbox.Name; Set-MailboxAuditBypassAssociation -Identity $mailboxName -AuditBypassEnabled $false; Write-Host \"Audit Bypass disabled for mailbox Identity: $mailboxName\" -ForegroundColor Green }",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Ensure that no mailboxes have 'AuditBypassEnabled' enabled to guarantee full audit logging for all mailbox activities.",
"Url": "https://learn.microsoft.com/en-us/powershell/module/exchange/set-mailboxauditbypassassociation?view=exchange-ps"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,39 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportM365
from prowler.providers.m365.services.exchange.exchange_client import exchange_client
class exchange_mailbox_audit_bypass_disabled(Check):
"""Verify if Exchange mailbox auditing is enabled.
This check ensures that mailbox auditing is not bypassed and is properly enabled.
"""
def execute(self) -> List[CheckReportM365]:
"""Run the check to validate Exchange mailbox auditing.
Iterates through the mailbox configurations to determine if auditing is enabled
and generates a report for each mailbox.
Returns:
List[CheckReportM365]: A list of reports with the audit status for each mailbox.
"""
findings = []
for mailbox_config in exchange_client.mailboxes_config:
report = CheckReportM365(
metadata=self.metadata(),
resource=mailbox_config,
resource_name=mailbox_config.name,
resource_id=mailbox_config.id,
)
report.status = "FAIL"
report.status_extended = f"Exchange mailbox auditing is bypassed and not enabled for mailbox: {mailbox_config.name}."
if not mailbox_config.audit_bypass_enabled:
report.status = "PASS"
report.status_extended = f"Exchange mailbox auditing is enabled for mailbox: {mailbox_config.name}."
findings.append(report)
return findings
@@ -10,6 +10,7 @@ class Exchange(M365Service):
super().__init__(provider)
self.powershell.connect_exchange_online()
self.organization_config = self._get_organization_config()
self.mailboxes_config = self._get_mailbox_audit_config()
self.powershell.close()
def _get_organization_config(self):
@@ -17,19 +18,49 @@ class Exchange(M365Service):
organization_config = None
try:
organization_configuration = self.powershell.get_organization_config()
organization_config = Organization(
name=organization_configuration.get("Name", ""),
guid=organization_configuration.get("Guid", ""),
audit_disabled=organization_configuration.get("AuditDisabled", False),
)
if organization_configuration:
organization_config = Organization(
name=organization_configuration.get("Name", ""),
guid=organization_configuration.get("Guid", ""),
audit_disabled=organization_configuration.get(
"AuditDisabled", False
),
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return organization_config
def _get_mailbox_audit_config(self):
logger.info("Microsoft365 - Getting mailbox audit configuration...")
mailboxes_config = []
try:
mailbox_audit_data = self.powershell.get_mailbox_audit_config()
for mailbox_audit_config in mailbox_audit_data:
mailboxes_config.append(
MailboxAuditConfig(
name=mailbox_audit_config.get("Name", ""),
id=mailbox_audit_config.get("Id", ""),
audit_bypass_enabled=mailbox_audit_config.get(
"AuditBypassEnabled", True
),
)
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return mailboxes_config
class Organization(BaseModel):
name: str
guid: str
audit_disabled: bool
class MailboxAuditConfig(BaseModel):
name: str
id: str
audit_bypass_enabled: bool
@@ -21,21 +21,22 @@ class purview_audit_log_search_enabled(Check):
"""
findings = []
audit_log_config = purview_client.audit_log_config
report = CheckReportM365(
metadata=self.metadata(),
resource=audit_log_config if audit_log_config else {},
resource_name="Purview Settings",
resource_id="purviewSettings",
)
report.status = "FAIL"
report.status_extended = "Purview audit log search is not enabled."
if audit_log_config:
report = CheckReportM365(
metadata=self.metadata(),
resource=audit_log_config if audit_log_config else {},
resource_name="Purview Settings",
resource_id="purviewSettings",
)
report.status = "FAIL"
report.status_extended = "Purview audit log search is not enabled."
if purview_client.audit_log_config and getattr(
purview_client.audit_log_config, "audit_log_search", False
):
report.status = "PASS"
report.status_extended = "Purview audit log search is enabled."
if purview_client.audit_log_config and getattr(
purview_client.audit_log_config, "audit_log_search", False
):
report.status = "PASS"
report.status_extended = "Purview audit log search is enabled."
findings.append(report)
findings.append(report)
return findings
@@ -17,11 +17,12 @@ class Purview(M365Service):
audit_log_config = None
try:
audit_log_config_response = self.powershell.get_audit_log_config()
audit_log_config = AuditLogConfig(
audit_log_search=audit_log_config_response.get(
"UnifiedAuditLogIngestionEnabled", False
if audit_log_config_response:
audit_log_config = AuditLogConfig(
audit_log_search=audit_log_config_response.get(
"UnifiedAuditLogIngestionEnabled", False
)
)
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -21,21 +21,22 @@ class teams_email_sending_to_channel_disabled(Check):
"""
findings = []
teams_settings = teams_client.teams_settings
report = CheckReportM365(
metadata=self.metadata(),
resource=teams_settings if teams_settings else {},
resource_name="Teams Settings",
resource_id="teamsSettings",
)
report.status = "FAIL"
report.status_extended = "Users can send emails to channel email addresses."
if teams_settings and not teams_settings.allow_email_into_channel:
report.status = "PASS"
report.status_extended = (
"Users can not send emails to channel email addresses."
if teams_settings:
report = CheckReportM365(
metadata=self.metadata(),
resource=teams_settings if teams_settings else {},
resource_name="Teams Settings",
resource_id="teamsSettings",
)
report.status = "FAIL"
report.status_extended = "Users can send emails to channel email addresses."
findings.append(report)
if teams_settings and not teams_settings.allow_email_into_channel:
report.status = "PASS"
report.status_extended = (
"Users can not send emails to channel email addresses."
)
findings.append(report)
return findings
@@ -22,46 +22,47 @@ class teams_external_file_sharing_restricted(Check):
List[CheckReportM365]: A list of reports containing the result of the check.
"""
findings = []
cloud_storage_settings = teams_client.teams_settings.cloud_storage_settings
report = CheckReportM365(
metadata=self.metadata(),
resource=cloud_storage_settings if cloud_storage_settings else {},
resource_name="Cloud Storage Settings",
resource_id="cloudStorageSettings",
)
report.status = "FAIL"
report.status_extended = "External file sharing is not restricted to only approved cloud storage services."
if teams_client.teams_settings:
cloud_storage_settings = teams_client.teams_settings.cloud_storage_settings
report = CheckReportM365(
metadata=self.metadata(),
resource=cloud_storage_settings if cloud_storage_settings else {},
resource_name="Cloud Storage Settings",
resource_id="cloudStorageSettings",
)
report.status = "FAIL"
report.status_extended = "External file sharing is not restricted to only approved cloud storage services."
allowed_services = teams_client.audit_config.get(
"allowed_cloud_storage_services", []
)
if cloud_storage_settings:
# Get storage services from CloudStorageSettings class items
storage_services = [
attr
for attr, type in CloudStorageSettings.__annotations__.items()
if type is bool
]
allowed_services = teams_client.audit_config.get(
"allowed_cloud_storage_services", []
)
if cloud_storage_settings:
# Get storage services from CloudStorageSettings class items
storage_services = [
attr
for attr, type in CloudStorageSettings.__annotations__.items()
if type is bool
]
# Check if all services are disabled when no allowed services are specified
# or if all enabled services are in the allowed list
if (
not allowed_services
and all(
not getattr(cloud_storage_settings, service, True)
for service in storage_services
)
) or (
allowed_services
and not any(
getattr(cloud_storage_settings, service, True)
and service not in allowed_services
for service in storage_services
)
):
report.status = "PASS"
report.status_extended = "External file sharing is restricted to only approved cloud storage services."
# Check if all services are disabled when no allowed services are specified
# or if all enabled services are in the allowed list
if (
not allowed_services
and all(
not getattr(cloud_storage_settings, service, True)
for service in storage_services
)
) or (
allowed_services
and not any(
getattr(cloud_storage_settings, service, True)
and service not in allowed_services
for service in storage_services
)
):
report.status = "PASS"
report.status_extended = "External file sharing is restricted to only approved cloud storage services."
findings.append(report)
findings.append(report)
return findings
@@ -17,16 +17,19 @@ class Teams(M365Service):
teams_settings = None
try:
settings = self.powershell.get_teams_settings()
teams_settings = TeamsSettings(
cloud_storage_settings=CloudStorageSettings(
allow_box=settings.get("AllowBox", True),
allow_drop_box=settings.get("AllowDropBox", True),
allow_egnyte=settings.get("AllowEgnyte", True),
allow_google_drive=settings.get("AllowGoogleDrive", True),
allow_share_file=settings.get("AllowShareFile", True),
),
allow_email_into_channel=settings.get("AllowEmailIntoChannel", True),
)
if settings:
teams_settings = TeamsSettings(
cloud_storage_settings=CloudStorageSettings(
allow_box=settings.get("AllowBox", True),
allow_drop_box=settings.get("AllowDropBox", True),
allow_egnyte=settings.get("AllowEgnyte", True),
allow_google_drive=settings.get("AllowGoogleDrive", True),
allow_share_file=settings.get("AllowShareFile", True),
),
allow_email_into_channel=settings.get(
"AllowEmailIntoChannel", True
),
)
except Exception as error:
logger.error(
@@ -0,0 +1,175 @@
from unittest import mock
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
class Test_exchange_mailbox_audit_bypass_disabled:
def test_no_mailboxes(self):
exchange_client = mock.MagicMock()
exchange_client.audited_tenant = "audited_tenant"
exchange_client.audited_domain = DOMAIN
exchange_client.mailboxes_config = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
mock.patch(
"prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled.exchange_client",
new=exchange_client,
),
):
from prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled import (
exchange_mailbox_audit_bypass_disabled,
)
check = exchange_mailbox_audit_bypass_disabled()
result = check.execute()
assert len(result) == 0
def test_audit_bypass_disabled_and_enabled(self):
exchange_client = mock.MagicMock()
exchange_client.audited_tenant = "audited_tenant"
exchange_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
mock.patch(
"prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled.exchange_client",
new=exchange_client,
),
):
from prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled import (
exchange_mailbox_audit_bypass_disabled,
)
from prowler.providers.m365.services.exchange.exchange_service import (
MailboxAuditConfig,
)
exchange_client.mailboxes_config = [
MailboxAuditConfig(name="test", id="test", audit_bypass_enabled=True),
MailboxAuditConfig(
name="test2", id="test2", audit_bypass_enabled=False
),
]
check = exchange_mailbox_audit_bypass_disabled()
result = check.execute()
assert len(result) == 2
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Exchange mailbox auditing is bypassed and not enabled for mailbox: test."
)
assert result[0].resource == exchange_client.mailboxes_config[0].dict()
assert result[0].resource_name == "test"
assert result[0].resource_id == "test"
assert result[0].location == "global"
assert result[1].status == "PASS"
assert (
result[1].status_extended
== "Exchange mailbox auditing is enabled for mailbox: test2."
)
assert result[1].resource == exchange_client.mailboxes_config[1].dict()
assert result[1].resource_name == "test2"
assert result[1].resource_id == "test2"
assert result[1].location == "global"
def test_audit_bypass_enabled(self):
exchange_client = mock.MagicMock()
exchange_client.audited_tenant = "audited_tenant"
exchange_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
mock.patch(
"prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled.exchange_client",
new=exchange_client,
),
):
from prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled import (
exchange_mailbox_audit_bypass_disabled,
)
from prowler.providers.m365.services.exchange.exchange_service import (
MailboxAuditConfig,
)
exchange_client.mailboxes_config = [
MailboxAuditConfig(name="test", id="test", audit_bypass_enabled=True),
]
check = exchange_mailbox_audit_bypass_disabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Exchange mailbox auditing is bypassed and not enabled for mailbox: test."
)
assert result[0].resource == exchange_client.mailboxes_config[0].dict()
assert result[0].resource_name == "test"
assert result[0].resource_id == "test"
assert result[0].location == "global"
def test_audit_bypass_disabled(self):
exchange_client = mock.MagicMock()
exchange_client.audited_tenant = "audited_tenant"
exchange_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
mock.patch(
"prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled.exchange_client",
new=exchange_client,
),
):
from prowler.providers.m365.services.exchange.exchange_mailbox_audit_bypass_disabled.exchange_mailbox_audit_bypass_disabled import (
exchange_mailbox_audit_bypass_disabled,
)
from prowler.providers.m365.services.exchange.exchange_service import (
MailboxAuditConfig,
)
exchange_client.mailboxes_config = [
MailboxAuditConfig(name="test", id="test", audit_bypass_enabled=False),
]
check = exchange_mailbox_audit_bypass_disabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Exchange mailbox auditing is enabled for mailbox: test."
)
assert result[0].resource == exchange_client.mailboxes_config[0].dict()
assert result[0].resource_name == "test"
assert result[0].resource_id == "test"
assert result[0].location == "global"
@@ -4,6 +4,7 @@ from unittest.mock import patch
from prowler.providers.m365.models import M365IdentityInfo
from prowler.providers.m365.services.exchange.exchange_service import (
Exchange,
MailboxAuditConfig,
Organization,
)
from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
@@ -13,6 +14,13 @@ def mock_exchange_get_organization_config(_):
return Organization(audit_disabled=True, name="test", guid="test")
def mock_exchange_get_mailbox_audit_config(_):
return [
MailboxAuditConfig(name="test", id="test", audit_bypass_enabled=False),
MailboxAuditConfig(name="test2", id="test2", audit_bypass_enabled=True),
]
class Test_Exchange_Service:
def test_get_client(self):
with (
@@ -50,3 +58,29 @@ class Test_Exchange_Service:
assert organization_config.audit_disabled is True
exchange_client.powershell.close()
@patch(
"prowler.providers.m365.services.exchange.exchange_service.Exchange._get_mailbox_audit_config",
new=mock_exchange_get_mailbox_audit_config,
)
def test_get_mailbox_audit_config(self):
with (
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
):
exchange_client = Exchange(
set_mocked_m365_provider(
identity=M365IdentityInfo(tenant_domain=DOMAIN)
)
)
mailbox_audit_config = exchange_client.mailboxes_config
assert len(mailbox_audit_config) == 2
assert mailbox_audit_config[0].name == "test"
assert mailbox_audit_config[0].id == "test"
assert mailbox_audit_config[0].audit_bypass_enabled is False
assert mailbox_audit_config[1].name == "test2"
assert mailbox_audit_config[1].id == "test2"
assert mailbox_audit_config[1].audit_bypass_enabled is True
exchange_client.powershell.close()