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 00000000000..8b137891791 --- /dev/null +++ b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/__init__.py @@ -0,0 +1 @@ + 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 00000000000..b6841d4e243 --- /dev/null +++ b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.metadata.json @@ -0,0 +1,39 @@ +{ + "Provider": "m365", + "CheckID": "defender_domain_dmarc_records_published", + "CheckTitle": "Exchange Online domain has an enforcing DMARC record", + "CheckType": [], + "ServiceName": "defender", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "NotDefined", + "ResourceGroup": "collaboration", + "Description": "**Microsoft 365 Exchange Online domains** publish DMARC TXT records at `_dmarc.` with an enforcing policy. This evaluates each domain from the Defender DKIM configuration list and expects `p=quarantine` or `p=reject`.", + "Risk": "Without an enforcing **DMARC** policy, attackers can spoof the organization's mail domains more easily, increasing phishing, BEC, and impersonation risk. A `p=none` policy is useful for monitoring but does not instruct receivers to quarantine or reject unauthenticated mail.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://www.rfc-editor.org/rfc/rfc7489", + "https://learn.microsoft.com/en-us/defender-office-365/email-authentication-dmarc-configure", + "https://learn.microsoft.com/en-us/defender-office-365/email-authentication-about" + ], + "Remediation": { + "Code": { + "CLI": "Publish a TXT record at _dmarc. with v=DMARC1 and p=quarantine or p=reject.", + "NativeIaC": "", + "Other": "1. Confirm SPF and DKIM are aligned for the domain\n2. Publish a DMARC TXT record at _dmarc. using p=quarantine or p=reject\n3. Include rua/ruf reporting addresses where appropriate\n4. Validate the TXT record from an external DNS resolver", + "Terraform": "" + }, + "Recommendation": { + "Text": "- Enforce DMARC with `p=quarantine` or `p=reject` for every sending domain\n- Start with monitoring only when needed, but move to enforcement after validating legitimate mail flows\n- Combine DMARC with SPF and DKIM alignment for defense in depth\n- Monitor aggregate reports for authentication failures and spoofing attempts", + "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 00000000000..b5462ff0fa6 --- /dev/null +++ b/prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.py @@ -0,0 +1,92 @@ +import re +from typing import List + +import dns.resolver + +from prowler.lib.check.models import Check, CheckReportM365 +from prowler.providers.m365.services.defender import defender_client + +DMARC_POLICY_RE = re.compile( + r"(?:^|;)\s*p\s*=\s*(?P[^;\s]+)", + re.IGNORECASE, +) + + +class defender_domain_dmarc_records_published(Check): + """ + Check if Exchange Online domains publish enforcing DMARC records. + + Attributes: + metadata: Metadata associated with the check (inherited from Check). + """ + + def execute(self) -> List[CheckReportM365]: + """ + Execute the check to verify if DMARC is enforced for all domains. + + This method checks each domain from the DKIM signing configuration list + for a DMARC TXT record with p=quarantine or p=reject. + + Returns: + List[CheckReportM365]: A list of reports containing the result. + """ + findings = [] + client = defender_client.defender_client + for config in client.dkim_configurations: + dmarc_policy = _get_dmarc_policy(config.id) + + report = CheckReportM365( + metadata=self.metadata(), + resource=config, + resource_name=config.id, + resource_id=config.id, + ) + report.status = "FAIL" + report.status_extended = ( + f"DMARC record for domain {config.id} is not published " + "with an enforcing policy." + ) + + if dmarc_policy in ("quarantine", "reject"): + report.status = "PASS" + report.status_extended = ( + f"DMARC record for domain {config.id} is published with " + f"enforcing policy p={dmarc_policy}." + ) + + findings.append(report) + + return findings + + +def _get_dmarc_policy(domain: str) -> str: + try: + txt_records = dns.resolver.resolve(f"_dmarc.{domain}", "TXT") + except dns.exception.DNSException: + return "" + + for record in txt_records: + txt_value = _txt_record_to_string(record) + if "V=DMARC1" not in txt_value.upper(): + continue + + policy_match = DMARC_POLICY_RE.search(txt_value) + if policy_match: + return policy_match.group("policy").lower() + + return "" + + +def _txt_record_to_string(record) -> str: + strings = getattr(record, "strings", None) + if strings: + return "".join( + ( + chunk.decode("utf-8", errors="ignore") + if isinstance(chunk, bytes) + else str(chunk) + ) + for chunk in strings + ) + + return str(record).strip('"') 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 00000000000..72428edd20c --- /dev/null +++ b/tests/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published_test.py @@ -0,0 +1,190 @@ +from contextlib import contextmanager +from importlib import import_module +from unittest import mock + +import dns.resolver + +from prowler.providers.m365.services.defender import defender_service +from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider + +CHECK_MODULE = ( + "prowler.providers.m365.services.defender." + "defender_domain_dmarc_records_published." + "defender_domain_dmarc_records_published" +) +GET_PROVIDER = "prowler.providers.common.provider.Provider.get_global_provider" +CONNECT_EXCHANGE = ( + "prowler.providers.m365.lib.powershell.m365_powershell." + "M365PowerShell.connect_exchange_online" +) +DEFENDER_SERVICE = "prowler.providers.m365.services.defender.defender_service" +DEFENDER_INIT = f"{DEFENDER_SERVICE}.Defender.__init__" + + +class TXTRecord: + def __init__(self, *strings): + self.strings = strings + + +class Test_defender_domain_dmarc_records_published: + def test_dmarc_policy_reject(self): + defender_client = _mock_defender_client("example.com") + + with ( + _mock_m365_provider(), + _mock_defender_init(), + mock.patch( + f"{CHECK_MODULE}.defender_client.defender_client", + new=defender_client, + ), + mock.patch( + f"{CHECK_MODULE}.dns.resolver.resolve", + return_value=[TXTRecord(b"v=DMARC1; p=reject")], + ) as resolver_mock, + ): + check = _get_check_class()() + result = check.execute() + + resolver_mock.assert_called_once_with("_dmarc.example.com", "TXT") + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + "DMARC record for domain example.com is published with " + "enforcing policy p=reject." + ) + config = defender_client.dkim_configurations[0] + assert result[0].resource == config.dict() + assert result[0].resource_name == "example.com" + assert result[0].resource_id == "example.com" + assert result[0].location == "global" + + def test_dmarc_policy_quarantine_split_txt_chunks(self): + defender_client = _mock_defender_client("example.org") + + with ( + _mock_m365_provider(), + _mock_defender_init(), + mock.patch( + f"{CHECK_MODULE}.defender_client.defender_client", + new=defender_client, + ), + mock.patch( + f"{CHECK_MODULE}.dns.resolver.resolve", + return_value=[ + TXTRecord(b"not-a-dmarc-record"), + TXTRecord(b"v=DMARC1; ", b"p=quarantine; pct=100"), + ], + ), + ): + check = _get_check_class()() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + "DMARC record for domain example.org is published with " + "enforcing policy p=quarantine." + ) + + def test_dmarc_policy_none_fails(self): + defender_client = _mock_defender_client("example.net") + + with ( + _mock_m365_provider(), + _mock_defender_init(), + mock.patch( + f"{CHECK_MODULE}.defender_client.defender_client", + new=defender_client, + ), + mock.patch( + f"{CHECK_MODULE}.dns.resolver.resolve", + return_value=[TXTRecord(b"v=DMARC1; p=none")], + ), + ): + check = _get_check_class()() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMARC record for domain example.net is not published with an " + "enforcing policy." + ) + + def test_missing_dmarc_record_fails(self): + defender_client = _mock_defender_client("example.edu") + + with ( + _mock_m365_provider(), + _mock_defender_init(), + mock.patch( + f"{CHECK_MODULE}.defender_client.defender_client", + new=defender_client, + ), + mock.patch( + f"{CHECK_MODULE}.dns.resolver.resolve", + side_effect=dns.resolver.NXDOMAIN, + ), + ): + check = _get_check_class()() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMARC record for domain example.edu is not published with an " + "enforcing policy." + ) + + def test_no_dkim_configurations(self): + defender_client = _mock_defender_client() + + with ( + _mock_m365_provider(), + _mock_defender_init(), + mock.patch( + f"{CHECK_MODULE}.defender_client.defender_client", + new=defender_client, + ), + ): + check = _get_check_class()() + result = check.execute() + + assert len(result) == 0 + + +def _mock_defender_client(*domains): + defender_client = mock.MagicMock() + defender_client.audited_tenant = "audited_tenant" + defender_client.audited_domain = DOMAIN + defender_client.dkim_configurations = [ + defender_service.DkimConfig(dkim_signing_enabled=True, id=domain) + for domain in domains + ] + return defender_client + + +def _mock_m365_provider(): + return mock.patch( + GET_PROVIDER, + return_value=set_mocked_m365_provider(), + ) + + +def _mock_defender_init(): + @contextmanager + def _patched_defender(): + with ( + mock.patch(CONNECT_EXCHANGE), + mock.patch( + DEFENDER_INIT, + return_value=None, + ), + ): + yield + + return _patched_defender() + + +def _get_check_class(): + return import_module(CHECK_MODULE).defender_domain_dmarc_records_published