feat(m365): ensure all forms of mail forwarding are blocked or disabled (#7658)

Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

tests/providers/m365/services/defender/defender_antispam_outbound_policy_configured/defender_antispam_outbound_policy_configured_test.py

@@ -37,6 +37,7 @@ class Test_defender_antispam_outbound_policy_configured:
                     notify_limit_exceeded_addresses=["test@correo.com"],
                     notify_sender_blocked_addresses=["test@correo.com"],
                     default=False,
+                    auto_forwarding_mode=False,
                 )
             }
             defender_client.outbound_spam_rules = {
@@ -92,6 +93,7 @@ class Test_defender_antispam_outbound_policy_configured:
                     notify_limit_exceeded_addresses=[],
                     notify_sender_blocked_addresses=[],
                     default=False,
+                    auto_forwarding_mode=False,
                 )
             }
             defender_client.outbound_spam_rules = {
@@ -146,6 +148,7 @@ class Test_defender_antispam_outbound_policy_configured:
                     notify_limit_exceeded_addresses=["test@correo.com"],
                     notify_sender_blocked_addresses=["test@correo.com"],
                     default=True,
+                    auto_forwarding_mode=False,
                 )
             }
             defender_client.outbound_spam_rules = {}
@@ -198,6 +201,7 @@ class Test_defender_antispam_outbound_policy_configured:
                     notify_limit_exceeded_addresses=["admin@org.com"],
                     notify_sender_blocked_addresses=["admin@org.com"],
                     default=False,
+                    auto_forwarding_mode=False,
                 )
             }
             defender_client.outbound_spam_rules = {}

tests/providers/m365/services/defender/defender_antispam_outbound_policy_forwarding_disabled/defender_antispam_outbound_policy_forwarding_disabled_test.py

@@ -0,0 +1,252 @@
+from unittest import mock
+
+from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
+
+
+class Test_defender_antispam_outbound_policy_forwarding_disabled:
+    def test_no_outbound_spam_policies(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_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled import (
+                defender_antispam_outbound_policy_forwarding_disabled,
+            )
+
+            defender_client.outbound_spam_policies = {}
+            defender_client.outbound_spam_rules = {}
+
+            check = defender_antispam_outbound_policy_forwarding_disabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_forwarding_disabled_custom_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_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled import (
+                defender_antispam_outbound_policy_forwarding_disabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                OutboundSpamPolicy,
+                OutboundSpamRule,
+            )
+
+            defender_client.outbound_spam_policies = {
+                "Policy1": OutboundSpamPolicy(
+                    default=False,
+                    notify_sender_blocked=True,
+                    notify_limit_exceeded=True,
+                    notify_limit_exceeded_addresses=["test@correo.com"],
+                    notify_sender_blocked_addresses=["test@correo.com"],
+                    auto_forwarding_mode=False,
+                )
+            }
+            defender_client.outbound_spam_rules = {
+                "Policy1": OutboundSpamRule(state="Enabled")
+            }
+
+            check = defender_antispam_outbound_policy_forwarding_disabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Outbound Spam Policy Policy1 does not allow mail forwarding."
+            )
+            assert (
+                result[0].resource
+                == defender_client.outbound_spam_policies["Policy1"].dict()
+            )
+            assert result[0].resource_name == "Defender Outbound Spam Policy"
+            assert result[0].resource_id == "Policy1"
+            assert result[0].location == "global"
+
+    def test_forwarding_enabled_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_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled import (
+                defender_antispam_outbound_policy_forwarding_disabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                OutboundSpamPolicy,
+                OutboundSpamRule,
+            )
+
+            defender_client.outbound_spam_policies = {
+                "Policy2": OutboundSpamPolicy(
+                    default=False,
+                    notify_sender_blocked=True,
+                    notify_limit_exceeded=True,
+                    notify_limit_exceeded_addresses=["test@correo.com"],
+                    notify_sender_blocked_addresses=["test@correo.com"],
+                    auto_forwarding_mode=True,
+                )
+            }
+            defender_client.outbound_spam_rules = {
+                "Policy2": OutboundSpamRule(state="Enabled")
+            }
+
+            check = defender_antispam_outbound_policy_forwarding_disabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Outbound Spam Policy Policy2 does allow mail forwarding."
+            )
+            assert (
+                result[0].resource
+                == defender_client.outbound_spam_policies["Policy2"].dict()
+            )
+            assert result[0].resource_name == "Defender Outbound Spam Policy"
+            assert result[0].resource_id == "Policy2"
+            assert result[0].location == "global"
+
+    def test_forwarding_disabled_default_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_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled import (
+                defender_antispam_outbound_policy_forwarding_disabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                OutboundSpamPolicy,
+            )
+
+            defender_client.outbound_spam_policies = {
+                "Default": OutboundSpamPolicy(
+                    default=True,
+                    notify_sender_blocked=True,
+                    notify_limit_exceeded=True,
+                    notify_limit_exceeded_addresses=["test@correo.com"],
+                    notify_sender_blocked_addresses=["test@correo.com"],
+                    auto_forwarding_mode=False,
+                )
+            }
+            defender_client.outbound_spam_rules = {}
+
+            check = defender_antispam_outbound_policy_forwarding_disabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Outbound Spam Policy Default does not allow mail forwarding."
+            )
+            assert (
+                result[0].resource
+                == defender_client.outbound_spam_policies["Default"].dict()
+            )
+            assert result[0].resource_name == "Defender Outbound Spam Policy"
+            assert result[0].resource_id == "Default"
+            assert result[0].location == "global"
+
+    def test_policy_without_rule(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_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.m365.services.defender.defender_antispam_outbound_policy_forwarding_disabled.defender_antispam_outbound_policy_forwarding_disabled import (
+                defender_antispam_outbound_policy_forwarding_disabled,
+            )
+            from prowler.providers.m365.services.defender.defender_service import (
+                OutboundSpamPolicy,
+            )
+
+            defender_client.outbound_spam_policies = {
+                "PolicyX": OutboundSpamPolicy(
+                    default=False,
+                    notify_sender_blocked=True,
+                    notify_limit_exceeded=True,
+                    notify_limit_exceeded_addresses=["test@correo.com"],
+                    notify_sender_blocked_addresses=["test@correo.com"],
+                    auto_forwarding_mode=False,
+                )
+            }
+            defender_client.outbound_spam_rules = {}
+
+            check = defender_antispam_outbound_policy_forwarding_disabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Outbound Spam Policy PolicyX does allow mail forwarding."
+            )
+            assert (
+                result[0].resource
+                == defender_client.outbound_spam_policies["PolicyX"].dict()
+            )
+            assert result[0].resource_name == "Defender Outbound Spam Policy"
+            assert result[0].resource_id == "PolicyX"
+            assert result[0].location == "global"

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

@@ -106,6 +106,7 @@ def mock_defender_get_outbound_spam_filter_policy(_):
             notify_limit_exceeded=True,
             notify_limit_exceeded_addresses=["security@example.com"],
             notify_sender_blocked_addresses=["security@example.com"],
+            auto_forwarding_mode=False,
             default=False,
         ),
         "Policy2": OutboundSpamPolicy(
@@ -113,6 +114,7 @@ def mock_defender_get_outbound_spam_filter_policy(_):
             notify_limit_exceeded=False,
             notify_limit_exceeded_addresses=[],
             notify_sender_blocked_addresses=[],
+            auto_forwarding_mode=True,
             default=True,
         ),
     }
@@ -308,6 +310,7 @@ class Test_Defender_Service:
             assert outbound_spam_policies[
                 "Policy1"
             ].notify_sender_blocked_addresses == ["security@example.com"]
+            assert outbound_spam_policies["Policy1"].auto_forwarding_mode is False
             assert outbound_spam_policies["Policy1"].default is False
             assert outbound_spam_policies["Policy2"].notify_sender_blocked is False
             assert outbound_spam_policies["Policy2"].notify_limit_exceeded is False
@@ -317,6 +320,7 @@ class Test_Defender_Service:
             assert (
                 outbound_spam_policies["Policy2"].notify_sender_blocked_addresses == []
             )
+            assert outbound_spam_policies["Policy2"].auto_forwarding_mode is True
             assert outbound_spam_policies["Policy2"].default is True
 
     @patch(

tests/providers/m365/services/exchange/exchange_service_test.py

@@ -55,11 +55,13 @@ def mock_exchange_get_transport_rules(_):
             name="test",
             scl=-1,
             sender_domain_is=["example.com"],
+            redirect_message_to=None,
         ),
         TransportRule(
             name="test2",
             scl=0,
             sender_domain_is=["example.com"],
+            redirect_message_to=["test@example.com"],
         ),
     ]
 
@@ -261,9 +263,11 @@ class Test_Exchange_Service:
             assert transport_rules[0].name == "test"
             assert transport_rules[0].scl == -1
             assert transport_rules[0].sender_domain_is == ["example.com"]
+            assert transport_rules[0].redirect_message_to is None
             assert transport_rules[1].name == "test2"
             assert transport_rules[1].scl == 0
             assert transport_rules[1].sender_domain_is == ["example.com"]
+            assert transport_rules[1].redirect_message_to == ["test@example.com"]
 
             exchange_client.powershell.close()
 

tests/providers/m365/services/exchange/exchange_transport_rules_mail_forwarding_disabled/exchange_transport_rules_mail_forwarding_disabled_test.py

@@ -0,0 +1,145 @@
+from unittest import mock
+
+from tests.providers.m365.m365_fixtures import DOMAIN, set_mocked_m365_provider
+
+
+class Test_exchange_transport_rules_mail_forwarding_disabled:
+    def test_empty_rule_list(self):
+        exchange_client = mock.MagicMock()
+        exchange_client.audited_tenant = "audited_tenant"
+        exchange_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.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled.exchange_client",
+                new=exchange_client,
+            ),
+        ):
+            from prowler.providers.m365.services.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled import (
+                exchange_transport_rules_mail_forwarding_disabled,
+            )
+
+            exchange_client.transport_rules = []
+
+            check = exchange_transport_rules_mail_forwarding_disabled()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    def test_forwarding_disabled(self):
+        exchange_client = mock.MagicMock()
+        exchange_client.audited_tenant = "audited_tenant"
+        exchange_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.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled.exchange_client",
+                new=exchange_client,
+            ),
+        ):
+            from prowler.providers.m365.services.exchange.exchange_service import (
+                TransportRule,
+            )
+            from prowler.providers.m365.services.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled import (
+                exchange_transport_rules_mail_forwarding_disabled,
+            )
+
+            exchange_client.transport_rules = [
+                TransportRule(
+                    name="Rule1", redirect_message_to=[], sender_domain_is=[]
+                ),
+                TransportRule(
+                    name="Rule2", redirect_message_to=[], sender_domain_is=[]
+                ),
+            ]
+
+            check = exchange_transport_rules_mail_forwarding_disabled()
+            result = check.execute()
+
+            assert len(result) == 2
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Transport rule Rule1 does not allow forwarding mail to external domains."
+            )
+            assert result[0].resource_name == "Rule1"
+            assert result[0].resource_id == "ExchangeTransportRule"
+            assert result[1].status == "PASS"
+            assert (
+                result[1].status_extended
+                == "Transport rule Rule2 does not allow forwarding mail to external domains."
+            )
+            assert result[1].resource_name == "Rule2"
+
+    def test_forwarding_enabled(self):
+        exchange_client = mock.MagicMock()
+        exchange_client.audited_tenant = "audited_tenant"
+        exchange_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.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled.exchange_client",
+                new=exchange_client,
+            ),
+        ):
+            from prowler.providers.m365.services.exchange.exchange_service import (
+                TransportRule,
+            )
+            from prowler.providers.m365.services.exchange.exchange_transport_rules_mail_forwarding_disabled.exchange_transport_rules_mail_forwarding_disabled import (
+                exchange_transport_rules_mail_forwarding_disabled,
+            )
+
+            exchange_client.transport_rules = [
+                TransportRule(
+                    name="ForwardingRule",
+                    redirect_message_to=["external@example.com"],
+                    sender_domain_is=[],
+                ),
+                TransportRule(
+                    name="NoForwardingRule",
+                    redirect_message_to=[],
+                    sender_domain_is=[],
+                ),
+            ]
+
+            check = exchange_transport_rules_mail_forwarding_disabled()
+            result = check.execute()
+
+            assert len(result) == 2
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Transport rule ForwardingRule allows forwarding mail to external domains: external@example.com."
+            )
+            assert result[0].resource_name == "ForwardingRule"
+            assert result[0].resource_id == "ExchangeTransportRule"
+            assert result[0].location == "global"
+            assert result[1].status == "PASS"
+            assert (
+                result[1].status_extended
+                == "Transport rule NoForwardingRule does not allow forwarding mail to external domains."
+            )
+            assert result[1].resource_name == "NoForwardingRule"
+            assert result[1].resource_id == "ExchangeTransportRule"
+            assert result[1].location == "global"