diff --git a/prowler/changelog.d/add-defender-domain-dmarc-records-published-check.added.md b/prowler/changelog.d/add-defender-domain-dmarc-records-published-check.added.md new file mode 100644 index 0000000000..63bf4e01a9 --- /dev/null +++ b/prowler/changelog.d/add-defender-domain-dmarc-records-published-check.added.md @@ -0,0 +1 @@ +`defender_domain_dmarc_records_published` checks that every Exchange Online domain publishes a DMARC record with an enforcing policy (`p=quarantine` or `p=reject`) diff --git a/prowler/compliance/m365/cis_4.0_m365.json b/prowler/compliance/m365/cis_4.0_m365.json index 35d64d358e..5ca427d0aa 100644 --- a/prowler/compliance/m365/cis_4.0_m365.json +++ b/prowler/compliance/m365/cis_4.0_m365.json @@ -526,7 +526,9 @@ { "Id": "2.1.10", "Description": "DMARC, or Domain-based Message Authentication, Reporting, and Conformance, assists recipient mail systems in determining the appropriate action to take when messages from a domain fail to meet SPF or DKIM authentication criteria.", - "Checks": [], + "Checks": [ + "defender_domain_dmarc_records_published" + ], "Attributes": [ { "Section": "2 Microsoft 365 Defender", diff --git a/prowler/compliance/m365/cis_6.0_m365.json b/prowler/compliance/m365/cis_6.0_m365.json index 5815c67ac9..33d902d6fa 100644 --- a/prowler/compliance/m365/cis_6.0_m365.json +++ b/prowler/compliance/m365/cis_6.0_m365.json @@ -543,7 +543,9 @@ { "Id": "2.1.10", "Description": "DMARC, or Domain-based Message Authentication, Reporting, and Conformance, assists recipient mail systems in determining the appropriate action to take when messages from a domain fail to meet SPF or DKIM authentication criteria.", - "Checks": [], + "Checks": [ + "defender_domain_dmarc_records_published" + ], "Attributes": [ { "Section": "2 Microsoft 365 Defender", diff --git a/prowler/compliance/m365/cis_7.0_m365.json b/prowler/compliance/m365/cis_7.0_m365.json index 7cc1315275..a39d30f54b 100644 --- a/prowler/compliance/m365/cis_7.0_m365.json +++ b/prowler/compliance/m365/cis_7.0_m365.json @@ -554,7 +554,9 @@ { "Id": "2.1.10", "Description": "DMARC, or Domain-based Message Authentication, Reporting, and Conformance, assists recipient mail systems in determining the appropriate action to take when messages from a domain fail to meet SPF or DKIM authentication criteria.", - "Checks": [], + "Checks": [ + "defender_domain_dmarc_records_published" + ], "Attributes": [ { "Section": "2 Microsoft Defender", diff --git a/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/__init__.py b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.metadata.json b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.metadata.json new file mode 100644 index 0000000000..63bff30a32 --- /dev/null +++ b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.metadata.json @@ -0,0 +1,38 @@ +{ + "Provider": "m365", + "CheckID": "defender_domain_dmarc_records_published", + "CheckTitle": "Exchange Online domain has a DMARC record published with an enforcement policy", + "CheckType": [], + "ServiceName": "defender", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "NotDefined", + "ResourceGroup": "collaboration", + "Description": "**Microsoft 365 Exchange Online domains** should publish a **DMARC** DNS TXT record at `_dmarc.` with an enforcement policy (`p=quarantine` or `p=reject`). This evaluates each accepted domain to confirm the published record and that it does not rely solely on the monitoring-only `p=none` policy.", + "Risk": "Without a **DMARC** record, or with one set to `p=none`, receiving mail systems cannot act on messages that fail **SPF**/**DKIM** alignment, enabling **domain spoofing** and **BEC phishing**.\n\nAttackers can impersonate the domain to deliver phishing or malware, and the domain owner loses enforcement over unauthorized senders, impacting **confidentiality** and **integrity**.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://www.rfc-editor.org/rfc/rfc7489", + "https://learn.microsoft.com/en-us/microsoft-365/security/office-365-security/email-authentication-dmarc-configure?view=o365-worldwide" + ], + "Remediation": { + "Code": { + "CLI": "", + "NativeIaC": "", + "Other": "1. Publish a DMARC TXT record at _dmarc. such as: v=DMARC1; p=quarantine; rua=mailto:dmarcreports@. 2. Prefer p=reject once monitoring confirms legitimate mail is aligned. 3. Avoid leaving the policy at p=none, which only monitors and does not enforce.", + "Terraform": "" + }, + "Recommendation": { + "Text": "- Publish a DMARC record for every accepted domain with an enforcement policy (`p=quarantine` or `p=reject`)\n- Combine with **SPF** and **DKIM** so messages can be evaluated for alignment\n- Start with `p=quarantine` while monitoring `rua`/`ruf` reports, then move to `p=reject`\n- Review DMARC aggregate reports regularly to catch legitimate senders before enforcing", + "Url": "https://hub.prowler.com/check/defender_domain_dmarc_records_published" + } + }, + "Categories": [ + "email-security", + "e3" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.py b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.py new file mode 100644 index 0000000000..e849c7c067 --- /dev/null +++ b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.py @@ -0,0 +1,114 @@ +from typing import List, Optional + +from prowler.lib.check.models import Check, CheckReportM365 +from prowler.providers.m365.services.defender.defender_client import defender_client + + +class defender_domain_dmarc_records_published(Check): + """ + Check if DMARC records with an enforcement policy are published for all + Exchange Online domains. + + Attributes: + metadata: Metadata associated with the check (inherited from Check). + """ + + def execute(self) -> List[CheckReportM365]: + """ + Execute the check to verify if DMARC records are published for all domains. + + This method inspects the DMARC DNS TXT record resolved for each domain + (at ``_dmarc.``) and validates that an enforcement policy + (``p=quarantine`` or ``p=reject``) is configured. + + Returns: + List[CheckReportM365]: A list of reports containing the result of the check. + """ + findings = [] + + # If the Graph domain list could not be retrieved, an empty result means + # the DMARC status is unknown for the tenant, not that it has no domains. + if ( + not defender_client.domain_dmarc_configurations + and defender_client.domain_discovery_failed + ): + report = CheckReportM365( + metadata=self.metadata(), + resource={}, + resource_name=defender_client.tenant_domain, + resource_id=defender_client.tenant_domain, + ) + report.status = "MANUAL" + report.status_extended = "DMARC records could not be verified because the Exchange Online domain list could not be retrieved; manual review is required." + findings.append(report) + return findings + + for ( + domain_id, + domain, + ) in defender_client.domain_dmarc_configurations.items(): + report = CheckReportM365( + metadata=self.metadata(), + resource={}, + resource_name=domain_id, + resource_id=domain_id, + ) + + policy = self._get_dmarc_policy(domain.dmarc_record) + + if domain.lookup_failed: + report.status = "MANUAL" + report.status_extended = f"DMARC record for domain with ID {domain_id} could not be verified because the DNS lookup did not complete; manual review is required." + elif policy in ("quarantine", "reject"): + report.status = "PASS" + report.status_extended = f"DMARC record is published on Exchange Online for domain with ID {domain_id} with enforcement policy p={policy}." + elif policy == "none": + report.status = "FAIL" + report.status_extended = f"DMARC record is published on Exchange Online for domain with ID {domain_id} but uses monitoring-only policy p=none." + elif not domain.dmarc_record: + report.status = "FAIL" + report.status_extended = f"DMARC record is not published on Exchange Online for domain with ID {domain_id}." + else: + report.status = "FAIL" + report.status_extended = f"DMARC record for domain with ID {domain_id} is malformed and does not include a valid enforcement policy." + + findings.append(report) + + return findings + + @staticmethod + def _get_dmarc_policy(record: Optional[str]) -> Optional[str]: + """ + Extract the DMARC policy (``p=``) tag value from a raw DMARC TXT record. + + Args: + record: The raw ``_dmarc.`` TXT record content, or ``None`` + if no record was found. + + Returns: + Optional[str]: The lowercase policy value (e.g. ``"reject"``), or + ``None`` if the record is missing or malformed. Per RFC 7489 the + record must start with an exact ``v=DMARC1`` version tag and the + ``p`` policy tag must immediately follow it. + """ + if not record: + return None + + tags = [tag.strip() for tag in record.split(";") if tag.strip()] + if len(tags) < 2: + return None + + # RFC 7489: the version tag MUST be first and equal to "DMARC1" exactly + # (e.g. "v=DMARC10" is not a DMARC record). + version_name, _, version_value = tags[0].partition("=") + if version_name.strip().lower() != "v" or version_value.strip().lower() != ( + "dmarc1" + ): + return None + + # RFC 7489: the policy tag MUST immediately follow the version tag. + policy_name, _, policy_value = tags[1].partition("=") + if policy_name.strip().lower() != "p": + return None + + return policy_value.strip().lower() diff --git a/prowler/providers/m365/services/defender/defender_service.py b/prowler/providers/m365/services/defender/defender_service.py index 6700bc19d6..f31ee69afa 100644 --- a/prowler/providers/m365/services/defender/defender_service.py +++ b/prowler/providers/m365/services/defender/defender_service.py @@ -1,5 +1,9 @@ +import asyncio from typing import List, Optional +import dns.asyncresolver +import dns.exception +import dns.resolver from pydantic.v1 import BaseModel from prowler.lib.logger import logger @@ -92,6 +96,38 @@ class Defender(M365Service): self.email_tenant_settings = self._get_email_tenant_settings() self.powershell.close() + self.tenant_domain = provider.identity.tenant_domain + self.domain_dmarc_configurations = {} + # Set when the Graph domain list could not be retrieved, so an empty + # result can be told apart from a tenant that genuinely has no domains. + self.domain_discovery_failed = False + + created_loop = False + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + created_loop = True + + if loop.is_closed(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + created_loop = True + + if loop.is_running(): + raise RuntimeError( + "Cannot initialize Defender service while event loop is running" + ) + + self.domain_dmarc_configurations = loop.run_until_complete( + self._get_domain_dmarc_configurations() + ) + + if created_loop: + asyncio.set_event_loop(None) + loop.close() + def _parse_protection_policy_rules(self, rules_data): """Parse preset security policy rules into PresetSecurityPolicyRule models.""" rules = [] @@ -732,6 +768,102 @@ class Defender(M365Service): ) return teams_protection_policy + async def _get_domain_dmarc_configurations(self): + """ + Get the DMARC DNS configuration for each accepted Exchange Online domain. + + Retrieves the list of verified domains from Microsoft Graph and, for each + one, performs a DNS TXT lookup at ``_dmarc.`` to capture the + published DMARC record, if any. + + Returns: + dict[str, DomainDmarcConfiguration]: DMARC configuration keyed by domain name. + """ + logger.info("M365 - Getting Defender domain DMARC configurations...") + domain_dmarc_configurations = {} + try: + verified_domain_ids = [] + domains_list = await self.client.domains.get() + + while domains_list: + for domain in getattr(domains_list, "value", []) or []: + if not domain or not getattr(domain, "is_verified", False): + continue + verified_domain_ids.append(domain.id) + + next_link = getattr(domains_list, "odata_next_link", None) + if not next_link: + break + domains_list = await self.client.domains.with_url(next_link).get() + + dmarc_records = await asyncio.gather( + *( + self._get_dmarc_txt_record(domain_id) + for domain_id in verified_domain_ids + ) + ) + for domain_id, (dmarc_record, lookup_failed) in zip( + verified_domain_ids, dmarc_records + ): + domain_dmarc_configurations[domain_id] = DomainDmarcConfiguration( + domain=domain_id, + dmarc_record=dmarc_record, + lookup_failed=lookup_failed, + ) + except Exception as error: + # The domain list could not be retrieved, so DMARC status is unknown + # for the whole tenant rather than confirmed absent. + self.domain_discovery_failed = True + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return domain_dmarc_configurations + + @staticmethod + async def _get_dmarc_txt_record(domain: str) -> tuple[Optional[str], bool]: + """ + Resolve the DMARC DNS TXT record published at ``_dmarc.``. + + Args: + domain: The domain name to look up the DMARC record for. + + Returns: + tuple[Optional[str], bool]: A ``(record, lookup_failed)`` pair. + ``record`` is the raw content of the first ``v=DMARC1`` TXT + record found, or ``None`` when the domain publishes no DMARC + record or the lookup could not be completed. ``lookup_failed`` + is ``True`` only when the DNS lookup could not be completed + (timeout / no reachable nameserver), distinguishing an + unverifiable result from a confirmed absence. + """ + try: + answers = await dns.asyncresolver.resolve(f"_dmarc.{domain}", "TXT") + for answer in answers: + record = "".join( + part.decode() if isinstance(part, bytes) else part + for part in answer.strings + ) + # RFC 7489: a DMARC record's first tag must be exactly "v=DMARC1" + # ("v=DMARC10" and other TXT records at the name are not DMARC). + name, _, value = record.split(";")[0].partition("=") + if name.strip().lower() == "v" and value.strip().lower() == "dmarc1": + return record, False + except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): + # The domain resolves but publishes no DMARC record: a confirmed absence. + return None, False + except (dns.resolver.NoNameservers, dns.exception.Timeout) as error: + # The lookup could not be completed, so the DMARC status is unknown. + logger.warning( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return None, True + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return None, True + return None, False + class MalwarePolicy(BaseModel): enable_file_filter: bool @@ -925,6 +1057,23 @@ class TeamsProtectionPolicy(BaseModel): zap_enabled: bool +class DomainDmarcConfiguration(BaseModel): + """Model for a domain's published DMARC DNS TXT record. + + Attributes: + domain: The domain name the DMARC record was looked up for. + dmarc_record: The raw ``_dmarc.`` TXT record content, or + ``None`` if no DMARC record was found. + lookup_failed: ``True`` when the DNS lookup could not be completed + (timeout / no reachable nameserver), so the DMARC status is unknown + rather than a confirmed absence. + """ + + domain: str + dmarc_record: Optional[str] = None + lookup_failed: bool = False + + class PresetSecurityPolicyRule(BaseModel): """Model for a preset security policy rule (EOP or ATP). diff --git a/tests/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published_test.py b/tests/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published_test.py new file mode 100644 index 0000000000..b1e6ea09f9 --- /dev/null +++ b/tests/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published_test.py @@ -0,0 +1,452 @@ +from unittest import mock + +from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider + + +class Test_defender_domain_dmarc_records_published: + def test_no_domains(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_client.audited_domain = DOMAIN + defender_client.domain_dmarc_configurations = {} + defender_client.domain_discovery_failed = False + + 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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + + check = defender_domain_dmarc_records_published() + result = check.execute() + assert len(result) == 0 + + def test_domain_discovery_failed(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_client.audited_domain = DOMAIN + defender_client.tenant_domain = DOMAIN + defender_client.domain_dmarc_configurations = {} + defender_client.domain_discovery_failed = True + + 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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert ( + result[0].status_extended + == "DMARC records could not be verified because the Exchange Online domain list could not be retrieved; manual review is required." + ) + assert result[0].resource_name == DOMAIN + assert result[0].resource_id == DOMAIN + + def test_domain_dmarc_invalid_version(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain8" + + # "v=DMARC10" must not be accepted as a valid "v=DMARC1" record. + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="v=DMARC10; p=reject", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"DMARC record for domain with ID {domain_id} is malformed and does not include a valid enforcement policy." + ) + + def test_domain_dmarc_reject(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain1" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="v=DMARC1; p=reject; rua=mailto:dmarc@domain1", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"DMARC record is published on Exchange Online for domain with ID {domain_id} with enforcement policy p=reject." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_quarantine(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain2" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="v=DMARC1;p=quarantine;pct=100", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"DMARC record is published on Exchange Online for domain with ID {domain_id} with enforcement policy p=quarantine." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_none_policy(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain3" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="v=DMARC1; p=none", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"DMARC record is published on Exchange Online for domain with ID {domain_id} but uses monitoring-only policy p=none." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_missing(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain4" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record=None, + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"DMARC record is not published on Exchange Online for domain with ID {domain_id}." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_malformed(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain5" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="this is not a dmarc record", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"DMARC record for domain with ID {domain_id} is malformed and does not include a valid enforcement policy." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_lookup_failed(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain7" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record=None, + lookup_failed=True, + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert ( + result[0].status_extended + == f"DMARC record for domain with ID {domain_id} could not be verified because the DNS lookup did not complete; manual review is required." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" + + def test_domain_dmarc_no_policy_tag(self): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_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.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published.defender_client", + new=defender_client, + ), + ): + from prowler.providers.m365.services.defender.defender_domain_dmarc_records_published.defender_domain_dmarc_records_published import ( + defender_domain_dmarc_records_published, + ) + from prowler.providers.m365.services.defender.defender_service import ( + DomainDmarcConfiguration, + ) + + domain_id = "domain6" + + defender_client.domain_dmarc_configurations = { + domain_id: DomainDmarcConfiguration( + domain=domain_id, + dmarc_record="v=DMARC1; rua=mailto:dmarc@domain6", + ) + } + + check = defender_domain_dmarc_records_published() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"DMARC record for domain with ID {domain_id} is malformed and does not include a valid enforcement policy." + ) + assert result[0].resource == {} + assert result[0].resource_name == domain_id + assert result[0].resource_id == domain_id + assert result[0].location == "global" diff --git a/tests/providers/m365/services/defender/m365_defender_service_test.py b/tests/providers/m365/services/defender/m365_defender_service_test.py index 8134b15f59..4006488cf2 100644 --- a/tests/providers/m365/services/defender/m365_defender_service_test.py +++ b/tests/providers/m365/services/defender/m365_defender_service_test.py @@ -1,5 +1,7 @@ +import asyncio +from types import SimpleNamespace from unittest import mock -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, patch from prowler.providers.m365.models import M365IdentityInfo from prowler.providers.m365.services.defender.defender_service import ( @@ -9,6 +11,7 @@ from prowler.providers.m365.services.defender.defender_service import ( Defender, DefenderInboundSpamPolicy, DkimConfig, + DomainDmarcConfiguration, InboundSpamRule, MalwarePolicy, MalwareRule, @@ -214,6 +217,19 @@ def mock_defender_get_outbound_spam_filter_rule(_): } +async def mock_defender_get_domain_dmarc_configurations(_): + return { + "domain1.com": DomainDmarcConfiguration( + domain="domain1.com", + dmarc_record="v=DMARC1; p=reject", + ), + "domain2.com": DomainDmarcConfiguration( + domain="domain2.com", + dmarc_record=None, + ), + } + + class Test_Defender_Service: def test_get_client(self): with ( @@ -555,6 +571,73 @@ class Test_Defender_Service: assert report_submission_policy.report_phish_addresses == [] assert report_submission_policy.report_chat_message_enabled is True + @patch( + "prowler.providers.m365.services.defender.defender_service.Defender._get_domain_dmarc_configurations", + new=mock_defender_get_domain_dmarc_configurations, + ) + def test_get_domain_dmarc_configurations(self): + with ( + mock.patch( + "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online" + ), + ): + defender_client = Defender( + set_mocked_m365_provider( + identity=M365IdentityInfo(tenant_domain=DOMAIN) + ) + ) + domain_dmarc_configurations = defender_client.domain_dmarc_configurations + assert len(domain_dmarc_configurations) == 2 + assert ( + domain_dmarc_configurations["domain1.com"].dmarc_record + == "v=DMARC1; p=reject" + ) + assert domain_dmarc_configurations["domain2.com"].dmarc_record is None + defender_client.powershell.close() + + def test_get_dmarc_txt_record_found(self): + class FakeAnswer: + def __init__(self, strings): + self.strings = strings + + with mock.patch( + "prowler.providers.m365.services.defender.defender_service.dns.asyncresolver.resolve", + new=AsyncMock(return_value=[FakeAnswer([b"v=DMARC1; p=reject"])]), + ): + record, lookup_failed = asyncio.run( + Defender._get_dmarc_txt_record("domain1.com") + ) + assert record == "v=DMARC1; p=reject" + assert lookup_failed is False + + def test_get_dmarc_txt_record_not_found(self): + import dns.resolver + + with mock.patch( + "prowler.providers.m365.services.defender.defender_service.dns.asyncresolver.resolve", + new=AsyncMock(side_effect=dns.resolver.NXDOMAIN), + ): + record, lookup_failed = asyncio.run( + Defender._get_dmarc_txt_record("domain2.com") + ) + assert record is None + # A confirmed absence, not an unverifiable lookup. + assert lookup_failed is False + + def test_get_dmarc_txt_record_lookup_failed(self): + import dns.exception + + with mock.patch( + "prowler.providers.m365.services.defender.defender_service.dns.asyncresolver.resolve", + new=AsyncMock(side_effect=dns.exception.Timeout), + ): + record, lookup_failed = asyncio.run( + Defender._get_dmarc_txt_record("domain3.com") + ) + assert record is None + # A transient failure leaves the DMARC status unknown. + assert lookup_failed is True + @patch( "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.get_eop_protection_policy_rule", return_value=[ @@ -651,3 +734,69 @@ class Test_Defender_Service: email_tenant_settings = defender_client.email_tenant_settings assert email_tenant_settings.priority_account_protection_enabled is True defender_client.powershell.close() + + +def test_defender__get_domain_dmarc_configurations_handles_pagination(): + defender_service = Defender.__new__(Defender) + + domains_page_one = [ + SimpleNamespace(id="domain1.com", is_verified=True), + SimpleNamespace(id="unverified.com", is_verified=False), + ] + domains_page_two = [ + SimpleNamespace(id="domain2.com", is_verified=True), + ] + + domains_response_page_one = SimpleNamespace( + value=domains_page_one, + odata_next_link="next-link", + ) + domains_response_page_two = SimpleNamespace( + value=domains_page_two, odata_next_link=None + ) + + domains_with_url_builder = SimpleNamespace( + get=AsyncMock(return_value=domains_response_page_two) + ) + with_url_mock = MagicMock(return_value=domains_with_url_builder) + + domains_builder = SimpleNamespace( + get=AsyncMock(return_value=domains_response_page_one), + with_url=with_url_mock, + ) + + defender_service.client = SimpleNamespace(domains=domains_builder) + + with mock.patch( + "prowler.providers.m365.services.defender.defender_service.Defender._get_dmarc_txt_record", + new=AsyncMock(return_value=("v=DMARC1; p=reject", False)), + ): + domain_dmarc_configurations = asyncio.run( + defender_service._get_domain_dmarc_configurations() + ) + + assert set(domain_dmarc_configurations) == {"domain1.com", "domain2.com"} + assert ( + domain_dmarc_configurations["domain1.com"].dmarc_record == "v=DMARC1; p=reject" + ) + assert domains_builder.get.await_count == 1 + assert domains_with_url_builder.get.await_count == 1 + with_url_mock.assert_called_once_with("next-link") + + +def test_defender__get_domain_dmarc_configurations_marks_discovery_failure(): + defender_service = Defender.__new__(Defender) + defender_service.domain_discovery_failed = False + + domains_builder = SimpleNamespace( + get=AsyncMock(side_effect=Exception("Graph domains request failed")) + ) + defender_service.client = SimpleNamespace(domains=domains_builder) + + domain_dmarc_configurations = asyncio.run( + defender_service._get_domain_dmarc_configurations() + ) + + # An empty result caused by a discovery failure must be distinguishable. + assert domain_dmarc_configurations == {} + assert defender_service.domain_discovery_failed is True