feat(m365): add DMARC domain check

prowler/providers/m365/services/defender/defender_domain_dmarc_records_published/__init__.py

@@ -0,0 +1 @@
+

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.<domain>` 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.<domain> 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.<domain> 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": ""
+}

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<policy>[^;\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('"')

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