feat(defender): add new check defender_malware_policy_notifications_internal_users_malware_enabled (#7435)

Co-authored-by: HugoPBrito <hugopbrit@gmail.com> Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
2026-07-23 12:31:54 +00:00 · 2025-04-18 17:08:05 +02:00
parent 56d7431d56
commit d09a680aaa
9 changed files with 329 additions and 42 deletions
@@ -10,6 +10,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - Add check for unused Service Accounts in GCP [(#7419)](https://github.com/prowler-cloud/prowler/pull/7419).
 - Add Powershell to Microsoft365 [(#7331)](https://github.com/prowler-cloud/prowler/pull/7331).
 - Add service Defender to Microsoft365 with one check for Common Attachments filter enabled in Malware Policies [(#7425)](https://github.com/prowler-cloud/prowler/pull/7425).
+- Add check for Notifications for Internal users enabled in Malware Policies from service Defender in M365 [(#7435)](https://github.com/prowler-cloud/prowler/pull/7435).
 - Support CLOUDSDK_AUTH_ACCESS_TOKEN in GCP [(#7495)](https://github.com/prowler-cloud/prowler/pull/7495).
 - Add service Exchange to Microsoft365 with one check for Organizations Mailbox Auditing enabled [(#7408)](https://github.com/prowler-cloud/prowler/pull/7408)
 - Add check for Bypass Disable in every Mailbox for service Defender in M365 [(#7418)](https://github.com/prowler-cloud/prowler/pull/7418)
@@ -23,19 +23,31 @@ class defender_malware_policy_common_attachments_filter_enabled(Check):
            List[CheckReportM365]: A list of reports containing the result of the check.
        """
        findings = []
-        for policy in defender_client.malware_policies:
+        if not defender_client.malware_policies:
            report = CheckReportM365(
                metadata=self.metadata(),
-                resource=policy,
+                resource={},
                resource_name="Defender Malware Policy",
                resource_id="defenderMalwarePolicy",
            )
            report.status = "FAIL"
-            report.status_extended = f"Common Attachment Types Filter is not enabled in the Defender anti-malware policy {policy.identity}."
+            report.status_extended = "Common Attachment Types Filter is not enabled."
+            findings.append(report)
+        else:
+            for policy in defender_client.malware_policies:
+                report = CheckReportM365(
+                    metadata=self.metadata(),
+                    resource=policy,
+                    resource_name="Defender Malware Policy",
+                    resource_id="defenderMalwarePolicy",
+                )
+                report.status = "FAIL"
+                report.status_extended = f"Common Attachment Types Filter is not enabled in anti-malware policy {policy.identity}."

-            if policy.enable_file_filter:
-                report.status = "PASS"
-                report.status_extended = f"Common Attachment Types Filter is enabled in the Defender anti-malware policy {policy.identity}."
+                if policy.enable_file_filter:
+                    report.status = "PASS"
+                    report.status_extended = f"Common Attachment Types Filter is enabled in anti-malware policy {policy.identity}."
+                    break

            findings.append(report)

@@ -0,0 +1,30 @@
+{
+  "Provider": "m365",
+  "CheckID": "defender_malware_policy_notifications_internal_users_malware_enabled",
+  "CheckTitle": "Ensure notifications for internal users sending malware is Enabled",
+  "CheckType": [],
+  "ServiceName": "defender",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "Defender Malware Policy",
+  "Description": "Verify that Exchange Online Protection (EOP) is configured to notify admins of malicious activity from internal users.",
+  "Risk": "If notifications for internal users sending malware are not enabled, administrators may not be aware of potential threats originating from within the organization, increasing the risk of undetected malicious activities.",
+  "RelatedUrl": "https://learn.microsoft.com/en-us/defender-office-365/anti-malware-protection-about",
+  "Remediation": {
+    "Code": {
+      "CLI": "Set-MalwareFilterPolicy -Identity Default -EnableInternalSenderAdminNotifications $true -InternalSenderAdminAddress 'admin@example.com'",
+      "NativeIaC": "",
+      "Other": "1. Connect to Exchange Online using Connect-ExchangeOnline. 2. Execute the command: Get-MalwareFilterPolicy | fl Identity, EnableInternalSenderAdminNotifications, InternalSenderAdminAddress. 3. Ensure 'Notify an admin about undelivered messages from internal senders' is set to On and that at least one email address is listed under Administrator email address.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable notifications for internal users sending malware in your Defender Malware Policy to ensure admins are alerted of potential threats.",
+      "Url": "https://learn.microsoft.com/en-us/powershell/module/exchange/set-malwarefilterpolicy?view=exchange-ps"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,62 @@
+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_malware_policy_notifications_internal_users_malware_enabled(Check):
+    """
+    Check if notifications for internal users sending malware are enabled in the Defender anti-malware policy.
+
+    Attributes:
+        metadata: Metadata associated with the check (inherited from Check).
+    """
+
+    def execute(self) -> List[CheckReportM365]:
+        """
+        Execute the check to verify if notifications for internal users sending malware are enabled.
+
+        This method checks the Defender anti-malware policy to determine if notifications for internal users
+        sending malware are enabled.
+
+        Returns:
+            List[CheckReportM365]: A list of reports containing the result of the check.
+        """
+        findings = []
+        if not defender_client.malware_policies:
+            report = CheckReportM365(
+                metadata=self.metadata(),
+                resource={},
+                resource_name="Defender Malware Policy",
+                resource_id="defenderMalwarePolicy",
+            )
+            report.status = "FAIL"
+            report.status_extended = (
+                "Notifications for internal users sending malware are not enabled."
+            )
+            findings.append(report)
+        else:
+            for policy in defender_client.malware_policies:
+                report = CheckReportM365(
+                    metadata=self.metadata(),
+                    resource=policy,
+                    resource_name="Defender Malware Policy",
+                    resource_id="defenderMalwarePolicy",
+                )
+                report.status = "FAIL"
+                report.status_extended = (
+                    "Notifications for internal users sending malware are not enabled."
+                )
+
+                if policy.enable_internal_sender_admin_notifications:
+                    if policy.internal_sender_admin_address:
+                        report.status = "PASS"
+                        report.status_extended = "Notifications for internal users sending malware are enabled."
+                        break
+                    else:
+                        report.status = "FAIL"
+                        report.status_extended = "Notifications for internal users sending malware are enabled, but no email addresses are configured."
+
+            findings.append(report)
+
+        return findings
@@ -20,12 +20,19 @@ class Defender(M365Service):
            if isinstance(malware_policy, dict):
                malware_policy = [malware_policy]
            for policy in malware_policy:
-                malware_policies.append(
-                    DefenderMalwarePolicy(
-                        enable_file_filter=policy.get("EnableFileFilter", True),
-                        identity=policy.get("Identity", ""),
+                if policy:
+                    malware_policies.append(
+                        DefenderMalwarePolicy(
+                            enable_file_filter=policy.get("EnableFileFilter", True),
+                            identity=policy.get("Identity", ""),
+                            enable_internal_sender_admin_notifications=policy.get(
+                                "EnableInternalSenderAdminNotifications", False
+                            ),
+                            internal_sender_admin_address=policy.get(
+                                "InternalSenderAdminAddress", ""
+                            ),
+                        )
                    )
-                )
        except Exception as error:
            logger.error(
                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -36,3 +43,5 @@ class Defender(M365Service):
 class DefenderMalwarePolicy(BaseModel):
    enable_file_filter: bool
    identity: str
+    enable_internal_sender_admin_notifications: bool
+    internal_sender_admin_address: str
@@ -30,34 +30,28 @@ class Test_defender_malware_policy_common_attachments_filter_enabled:
            )

            defender_client.malware_policies = [
-                DefenderMalwarePolicy(enable_file_filter=False, identity="Policy1"),
-                DefenderMalwarePolicy(enable_file_filter=False, identity="Policy2"),
+                DefenderMalwarePolicy(
+                    enable_file_filter=False,
+                    identity="Policy1",
+                    enable_internal_sender_admin_notifications=False,
+                    internal_sender_admin_address="",
+                ),
            ]

            check = defender_malware_policy_common_attachments_filter_enabled()
            result = check.execute()
-            assert len(result) == 2
+            assert len(result) == 1

            assert result[0].status == "FAIL"
            assert (
                result[0].status_extended
-                == "Common Attachment Types Filter is not enabled in the Defender anti-malware policy Policy1."
+                == "Common Attachment Types Filter is not enabled in anti-malware policy Policy1."
            )
            assert result[0].resource == defender_client.malware_policies[0].dict()
            assert result[0].resource_name == "Defender Malware Policy"
            assert result[0].resource_id == "defenderMalwarePolicy"
            assert result[0].location == "global"

-            assert result[1].status == "FAIL"
-            assert (
-                result[1].status_extended
-                == "Common Attachment Types Filter is not enabled in the Defender anti-malware policy Policy2."
-            )
-            assert result[1].resource == defender_client.malware_policies[1].dict()
-            assert result[1].resource_name == "Defender Malware Policy"
-            assert result[1].resource_id == "defenderMalwarePolicy"
-            assert result[1].location == "global"
-
    def test_enable_file_filter_enabled(self):
        defender_client = mock.MagicMock()
        defender_client.audited_tenant = "audited_tenant"
@@ -84,34 +78,28 @@ class Test_defender_malware_policy_common_attachments_filter_enabled:
            )

            defender_client.malware_policies = [
-                DefenderMalwarePolicy(enable_file_filter=True, identity="Policy1"),
-                DefenderMalwarePolicy(enable_file_filter=True, identity="Policy2"),
+                DefenderMalwarePolicy(
+                    enable_file_filter=True,
+                    identity="Policy1",
+                    enable_internal_sender_admin_notifications=False,
+                    internal_sender_admin_address="",
+                ),
            ]

            check = defender_malware_policy_common_attachments_filter_enabled()
            result = check.execute()
-            assert len(result) == 2
+            assert len(result) == 1

            assert result[0].status == "PASS"
            assert (
                result[0].status_extended
-                == "Common Attachment Types Filter is enabled in the Defender anti-malware policy Policy1."
+                == "Common Attachment Types Filter is enabled in anti-malware policy Policy1."
            )
            assert result[0].resource == defender_client.malware_policies[0].dict()
            assert result[0].resource_name == "Defender Malware Policy"
            assert result[0].resource_id == "defenderMalwarePolicy"
            assert result[0].location == "global"

-            assert result[1].status == "PASS"
-            assert (
-                result[1].status_extended
-                == "Common Attachment Types Filter is enabled in the Defender anti-malware policy Policy2."
-            )
-            assert result[1].resource == defender_client.malware_policies[1].dict()
-            assert result[1].resource_name == "Defender Malware Policy"
-            assert result[1].resource_id == "defenderMalwarePolicy"
-            assert result[1].location == "global"
-
    def test_no_policy(self):
        defender_client = mock.MagicMock()
        defender_client.audited_tenant = "audited_tenant"
@@ -138,4 +126,13 @@ class Test_defender_malware_policy_common_attachments_filter_enabled:

            check = defender_malware_policy_common_attachments_filter_enabled()
            result = check.execute()
-            assert len(result) == 0
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Common Attachment Types Filter is not enabled."
+            )
+            assert result[0].resource == {}
+            assert result[0].resource_name == "Defender Malware Policy"
+            assert result[0].resource_id == "defenderMalwarePolicy"
+            assert result[0].location == "global"
@@ -0,0 +1,155 @@
+from unittest import mock
+
+from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
+
+
+class Test_defender_malware_policy_notifications_internal_users_malware_enabled:
+    def test_notifications_disabled(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_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled import (
+                defender_malware_policy_notifications_internal_users_malware_enabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                DefenderMalwarePolicy,
+            )
+
+            defender_client.malware_policies = [
+                DefenderMalwarePolicy(
+                    enable_file_filter=True,
+                    identity="Default",
+                    enable_internal_sender_admin_notifications=False,
+                    internal_sender_admin_address="",
+                )
+            ]
+
+            check = (
+                defender_malware_policy_notifications_internal_users_malware_enabled()
+            )
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Notifications for internal users sending malware are not enabled."
+            )
+            assert result[0].resource == defender_client.malware_policies[0].dict()
+            assert result[0].resource_name == "Defender Malware Policy"
+            assert result[0].resource_id == "defenderMalwarePolicy"
+            assert result[0].location == "global"
+
+    def test_notifications_enabled_without_email(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_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled import (
+                defender_malware_policy_notifications_internal_users_malware_enabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                DefenderMalwarePolicy,
+            )
+
+            defender_client.malware_policies = [
+                DefenderMalwarePolicy(
+                    enable_file_filter=True,
+                    identity="Default",
+                    enable_internal_sender_admin_notifications=True,
+                    internal_sender_admin_address="",
+                )
+            ]
+
+            check = (
+                defender_malware_policy_notifications_internal_users_malware_enabled()
+            )
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Notifications for internal users sending malware are enabled, but no email addresses are configured."
+            )
+            assert result[0].resource == defender_client.malware_policies[0].dict()
+            assert result[0].resource_name == "Defender Malware Policy"
+            assert result[0].resource_id == "defenderMalwarePolicy"
+            assert result[0].location == "global"
+
+    def test_notifications_enabled_with_email(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_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_malware_policy_notifications_internal_users_malware_enabled.defender_malware_policy_notifications_internal_users_malware_enabled import (
+                defender_malware_policy_notifications_internal_users_malware_enabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                DefenderMalwarePolicy,
+            )
+
+            defender_client.malware_policies = [
+                DefenderMalwarePolicy(
+                    enable_file_filter=True,
+                    identity="Default",
+                    enable_internal_sender_admin_notifications=True,
+                    internal_sender_admin_address="security@example.com",
+                )
+            ]
+
+            check = (
+                defender_malware_policy_notifications_internal_users_malware_enabled()
+            )
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Notifications for internal users sending malware are enabled."
+            )
+            assert result[0].resource == defender_client.malware_policies[0].dict()
+            assert result[0].resource_name == "Defender Malware Policy"
+            assert result[0].resource_id == "defenderMalwarePolicy"
+            assert result[0].location == "global"
@@ -11,8 +11,18 @@ from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider

 def mock_defender_get_malware_filter_policy(_):
    return [
-        DefenderMalwarePolicy(enable_file_filter=False, identity="Policy1"),
-        DefenderMalwarePolicy(enable_file_filter=True, identity="Policy2"),
+        DefenderMalwarePolicy(
+            enable_file_filter=False,
+            identity="Policy1",
+            enable_internal_sender_admin_notifications=False,
+            internal_sender_admin_address="",
+        ),
+        DefenderMalwarePolicy(
+            enable_file_filter=True,
+            identity="Policy2",
+            enable_internal_sender_admin_notifications=True,
+            internal_sender_admin_address="security@example.com",
+        ),
    ]


@@ -50,5 +60,16 @@ class Test_Defender_Service:
            malware_policies = defender_client.malware_policies
            assert malware_policies[0].enable_file_filter is False
            assert malware_policies[0].identity == "Policy1"
+            assert (
+                malware_policies[0].enable_internal_sender_admin_notifications is False
+            )
+            assert malware_policies[0].internal_sender_admin_address == ""
            assert malware_policies[1].enable_file_filter is True
            assert malware_policies[1].identity == "Policy2"
+            assert (
+                malware_policies[1].enable_internal_sender_admin_notifications is True
+            )
+            assert (
+                malware_policies[1].internal_sender_admin_address
+                == "security@example.com"
+            )