feat(m365): add defender_domain_dmarc_records_published check (#11936)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
Co-authored-by: hdy2001 <56308320+hdy2001@users.noreply.github.com>

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"

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