diff --git a/prowler/providers/m365/services/defender/defender_domain_spf_records_published/__init__.py b/prowler/providers/m365/services/defender/defender_domain_spf_records_published/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.metadata.json b/prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.metadata.json new file mode 100644 index 0000000000..82eca41d7c --- /dev/null +++ b/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": "" +} diff --git a/prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.py b/prowler/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published.py new file mode 100644 index 0000000000..4469e23b3c --- /dev/null +++ b/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 diff --git a/prowler/providers/m365/services/defender/defender_service.py b/prowler/providers/m365/services/defender/defender_service.py index cec2e7dce4..3e665fce04 100644 --- a/prowler/providers/m365/services/defender/defender_service.py +++ b/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 diff --git a/tests/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published_test.py b/tests/providers/m365/services/defender/defender_domain_spf_records_published/defender_domain_spf_records_published_test.py new file mode 100644 index 0000000000..0f0a80a750 --- /dev/null +++ b/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" diff --git a/tests/providers/m365/services/defender/m365_defender_service_test.py b/tests/providers/m365/services/defender/m365_defender_service_test.py index 3e5a534f2c..0ed1a5f949 100644 --- a/tests/providers/m365/services/defender/m365_defender_service_test.py +++ b/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 + == "" + )