feat(defender): add new check for spf records published

prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.metadata.json

@@ -0,0 +1,32 @@
+{
+  "Provider": "m365",
+  "CheckID": "defender_domain_spf_records_published",
+  "CheckTitle": "Ensure that SPF records are published for all Exchange Online Domains",
+  "CheckType": [],
+  "ServiceName": "defender",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "low",
+  "ResourceType": "Exchange Online Domain",
+  "Description": "Ensure that each configured Exchange domain has a corresponding SPF (Sender Policy Framework) record published in DNS to validate authorized email senders.",
+  "Risk": "Without SPF records, messages from your domain could be spoofed, increasing the risk of phishing attacks and reducing the credibility of your email communications.",
+  "RelatedUrl": "https://learn.microsoft.com/en-us/microsoft-365/security/office-365-security/email-authentication-spf-configure?view=o365-worldwide",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. If all email is sent from Exchange Online, add the following TXT DNS record to each accepted domain: v=spf1 include:spf.protection.outlook.com -all. 2. If other systems send email on your behalf, refer to the Microsoft documentation for SPF configuration guidance.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Publish SPF records for each Exchange domain to help prevent spoofing and unauthorized sending.",
+      "Url": "https://learn.microsoft.com/en-us/office365/SecurityCompliance/set-up-spf-in-office-365-to-help-prevent-spoofing"
+    }
+  },
+  "Categories": [
+    "e3"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.py

@@ -0,0 +1,43 @@
+from typing import List
+
+from prowler.lib.check.models import Check, CheckReportM365
+from prowler.providers.m365.services.defender.defender_client import defender_client
+
+
+class defender_domain_spf_records_published(Check):
+    """
+    Check if SPF records 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 SPF records are published for all domains.
+
+        This method checks the DNS configuration for each domain to determine if the SPF record is present.
+
+        Returns:
+            List[CheckReportM365]: A list of reports containing the result of the check.
+        """
+        findings = []
+        for domain_id, domain in defender_client.domain_service_configurations.items():
+            report = CheckReportM365(
+                metadata=self.metadata(),
+                resource={},
+                resource_name=domain_id,
+                resource_id=domain_id,
+            )
+            report.status = "FAIL"
+            report.status_extended = f"SPF record is not published on Exchange Online for domain with ID {domain_id}."
+
+            for config in domain.service_configuration_records:
+                if config.record_type == "Txt":
+                    if config.text == "v=spf1 include:spf.protection.outlook.com -all":
+                        report.status = "PASS"
+                        report.status_extended = f"SPF record is published on Exchange Online for domain with ID {domain_id}."
+
+            findings.append(report)
+
+        return findings

prowler/providers/m365/services/defender/defender_service.py

@@ -1,3 +1,4 @@
+from asyncio import gather, get_event_loop
 from typing import List
 
 from pydantic import BaseModel
@@ -33,6 +34,16 @@ class Defender(M365Service):
             self.report_submission_policy = self._get_report_submission_policy()
             self.powershell.close()
 
+        loop = get_event_loop()
+        self.tenant_domain = provider.identity.tenant_domain
+        attributes = loop.run_until_complete(
+            gather(
+                self._get_domain_service_configurations(),
+            )
+        )
+
+        self.domain_service_configurations = attributes[0]
+
     def _get_malware_filter_policy(self):
         logger.info("M365 - Getting Defender malware filter policy...")
         malware_policies = []
@@ -276,6 +287,35 @@ class Defender(M365Service):
             )
         return report_submission_policy
 
+    async def _get_domain_service_configurations(self):
+        logger.info("Microsoft365 - Getting domain service configurations...")
+        domains_configuration = {}
+        try:
+            domains_list = await self.client.domains.get()
+            domains_configuration.update({})
+            for domain in domains_list.value:
+                if domain:
+                    domain_configuration = await self.client.domains.by_domain_id(
+                        domain.id
+                    ).service_configuration_records.get()
+                    domains_configuration.update(
+                        {
+                            domain.id: DomainServiceConfiguration(
+                                service_configuration_records=(
+                                    domain_configuration.value
+                                    if domain_configuration.value
+                                    else None
+                                ),
+                            )
+                        }
+                    )
+
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        return domains_configuration
+
 
 class MalwarePolicy(BaseModel):
     enable_file_filter: bool
@@ -344,3 +384,7 @@ class ReportSubmissionPolicy(BaseModel):
     report_phish_addresses: list[str]
     report_chat_message_enabled: bool
     report_chat_message_to_customized_address_enabled: bool
+
+
+class DomainServiceConfiguration(BaseModel):
+    service_configuration_records: List

tests/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published_test.py

@@ -0,0 +1,132 @@
+from unittest import mock
+
+from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
+
+
+class Test_defender_domain_spf_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_service_configurations = {}
+
+        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_spf_records_published.defender_domain_spf_records_published.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_domain_spf_records_published.defender_domain_spf_records_published import (
+                defender_domain_spf_records_published,
+            )
+
+            check = defender_domain_spf_records_published()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_domain_spf_record_present(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_spf_records_published.defender_domain_spf_records_published.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_domain_spf_records_published.defender_domain_spf_records_published import (
+                defender_domain_spf_records_published,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                DomainServiceConfiguration,
+            )
+
+            domain_id = "domain1"
+
+            record = mock.MagicMock()
+            record.record_type = "Txt"
+            record.text = "v=spf1 include:spf.protection.outlook.com -all"
+
+            defender_client.domain_service_configurations = {
+                domain_id: DomainServiceConfiguration(
+                    service_configuration_records=[record]
+                )
+            }
+
+            check = defender_domain_spf_records_published()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"SPF record is 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_spf_record_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_spf_records_published.defender_domain_spf_records_published.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_domain_spf_records_published.defender_domain_spf_records_published import (
+                defender_domain_spf_records_published,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                DomainServiceConfiguration,
+            )
+
+            domain_id = "domain2"
+
+            record = mock.MagicMock()
+            record.record_type = "Txt"
+            record.text = ""
+
+            defender_client.domain_service_configurations = {
+                domain_id: DomainServiceConfiguration(
+                    service_configuration_records=[record]
+                )
+            }
+
+            check = defender_domain_spf_records_published()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"SPF 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"

tests/providers/m365/services/defender/m365_defender_service_test.py

@@ -9,6 +9,7 @@ from prowler.providers.m365.services.defender.defender_service import (
     Defender,
     DefenderInboundSpamPolicy,
     DkimConfig,
+    DomainServiceConfiguration,
     MalwarePolicy,
     MalwareRule,
     OutboundSpamPolicy,
@@ -160,6 +161,26 @@ def mock_defender_get_outbound_spam_filter_rule(_):
     }
 
 
+async def mock_defender_get_domain_service_configuration(_):
+    class Record:
+        def __init__(self, record_type, txt):
+            self.record_type = record_type
+            self.txt = txt
+
+    spf_record = Record("Txt", "v=spf1 include:spf.protection.outlook.com -all")
+    mx_record = Record("Mx", "")
+    empty_txt_record = Record("Txt", "")
+
+    return {
+        "Domain1": DomainServiceConfiguration(
+            service_configuration_records=[spf_record, mx_record]
+        ),
+        "Domain2": DomainServiceConfiguration(
+            service_configuration_records=[empty_txt_record]
+        ),
+    }
+
+
 class Test_Defender_Service:
     def test_get_client(self):
         with (
@@ -442,3 +463,48 @@ class Test_Defender_Service:
             assert report_submission_policy.report_not_junk_addresses == []
             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_service_configurations",
+        new=mock_defender_get_domain_service_configuration,
+    )
+    def test_get_domain_service_configuration(self):
+        with (
+            mock.patch(
+                "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
+            ),
+        ):
+            defender_client = Defender(set_mocked_m365_provider())
+
+            domain_service_config = defender_client.domain_service_configurations
+            assert len(domain_service_config) == 2
+            assert (
+                domain_service_config["Domain1"]
+                .service_configuration_records[0]
+                .record_type
+                == "Txt"
+            )
+            assert (
+                domain_service_config["Domain1"].service_configuration_records[0].txt
+                == "v=spf1 include:spf.protection.outlook.com -all"
+            )
+            assert (
+                domain_service_config["Domain1"]
+                .service_configuration_records[1]
+                .record_type
+                == "Mx"
+            )
+            assert (
+                domain_service_config["Domain1"].service_configuration_records[1].txt
+                == ""
+            )
+            assert (
+                domain_service_config["Domain2"]
+                .service_configuration_records[0]
+                .record_type
+                == "Txt"
+            )
+            assert (
+                domain_service_config["Domain2"].service_configuration_records[0].txt
+                == ""
+            )