merge branch 'master' into PRWLR-5989-ensure-that-spf-records-are-published-for-all-exchange

docs/getting-started/requirements.md

@@ -247,6 +247,7 @@ Prowler for M365 requires two types of permission scopes to be set (if you want
     - `User.Read` (IMPORTANT: this must be set as **delegated**): Required for the sign-in.
     - `Sites.Read.All`: Required for SharePoint service.
     - `SharePointTenantSettings.Read.All`: Required for SharePoint service.
+    - `Domain.Read.All`: Required for `defender_domain_spf_records_published` check.
 
 - **Powershell Modules Permissions**: These are set at the `M365_USER` level, so the user used to run Prowler must have one of the following roles:
     - `Global Reader` (recommended): this allows you to read all roles needed.

docs/tutorials/microsoft365/getting-started-m365.md

@@ -119,6 +119,7 @@ Follow these steps to assign the permissions:
 3. Search and select every permission below and once all are selected click on `Add permissions`:
 
     - `Directory.Read.All`
+    - `Domain.Read.All`
     - `Policy.Read.All`
     - `Sites.Read.All`
     - `SharePointTenantSettings.Read.All`

prowler/CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - Update the compliance list supported for each provider from docs. [(#7694)](https://github.com/prowler-cloud/prowler/pull/7694)
 - Allow setting cluster name in in-cluster mode in Kubernetes. [(#7695)](https://github.com/prowler-cloud/prowler/pull/7695)
 - Add Prowler ThreatScore for M365 provider. [(#7692)](https://github.com/prowler-cloud/prowler/pull/7692)
+- Add new check `defender_domain_spf_records_published`. [(#7724)](https://github.com/prowler-cloud/prowler/pull/7724)
 - Add new check `admincenter_organization_customer_lockbox_enabled`. [(#7732)](https://github.com/prowler-cloud/prowler/pull/7732)
 - Add new check `admincenter_external_calendar_sharing_disabled`. [(#7733)](https://github.com/prowler-cloud/prowler/pull/7733)
 - Add GitHub provider. [(#5787)](https://github.com/prowler-cloud/prowler/pull/5787)
@@ -26,6 +27,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 ### Fixed
 - Update CIS 4.0 for M365 provider. [(#7699)](https://github.com/prowler-cloud/prowler/pull/7699)
+- Enhance defender policies checks logic. [(#7719)](https://github.com/prowler-cloud/prowler/pull/7719)
 - Update and upgrade CIS for all the providers [(#7738)](https://github.com/prowler-cloud/prowler/pull/7738)
 - Cover policies with conditions with SNS endpoint in `sns_topics_not_publicly_accessible`. [(#7750)](https://github.com/prowler-cloud/prowler/pull/7750)
 - Fix `m365_powershell test_credentials` to use sanitized credentials. [(#7761)](https://github.com/prowler-cloud/prowler/pull/7761)

prowler/compliance/m365/cis_4.0_m365.json

@@ -470,7 +470,9 @@
     {
       "Id": "2.1.8",
       "Description": "For each domain that is configured in Exchange, a corresponding Sender Policy Framework (SPF) record should be created.",
-      "Checks": [],
+      "Checks": [
+        "defender_domain_spf_records_published"
+      ],
       "Attributes": [
         {
           "Section": "2 Microsoft 365 Defender",

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, Optional
 
 from pydantic import BaseModel
@@ -35,6 +36,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 = []
@@ -333,6 +344,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
@@ -424,3 +464,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,
     InboundSpamRule,
     MalwarePolicy,
     MalwareRule,
@@ -214,6 +215,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 (
@@ -554,3 +575,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
+                == ""
+            )