feat(azure): Defender checks related to defender settings (#3347)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>

docs/developer-guide/unit-testing.md

@@ -523,7 +523,7 @@ from unittest import mock
 from uuid import uuid4
 
 # Azure Constants
-AZURE_SUSCRIPTION = str(uuid4())
+AZURE_SUBSCRIPTION = str(uuid4())
 
 
 
@@ -542,7 +542,7 @@ class Test_defender_ensure_defender_for_arm_is_on:
 
         # Create the custom Defender object to be tested
         defender_client.pricings = {
-            AZURE_SUSCRIPTION: {
+            AZURE_SUBSCRIPTION: {
                 "Arm": Defender_Pricing(
                     resource_id=resource_id,
                     pricing_tier="Not Standard",
@@ -580,9 +580,9 @@ class Test_defender_ensure_defender_for_arm_is_on:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Defender plan Defender for ARM from subscription {AZURE_SUSCRIPTION} is set to OFF (pricing tier not standard)"
+                == f"Defender plan Defender for ARM from subscription {AZURE_SUBSCRIPTION} is set to OFF (pricing tier not standard)"
             )
-            assert result[0].subscription == AZURE_SUSCRIPTION
+            assert result[0].subscription == AZURE_SUBSCRIPTION
             assert result[0].resource_name == "Defender plan ARM"
             assert result[0].resource_id == resource_id
 ```

prowler/providers/azure/services/defender/defender_ensure_mcas_is_enabled/defender_ensure_mcas_is_enabled.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "azure",
+  "CheckID": "defender_ensure_mcas_is_enabled",
+  "CheckTitle": "Ensure that Microsoft Defender for Cloud Apps integration with Microsoft Defender for Cloud is Selected",
+  "CheckType": [],
+  "ServiceName": "defender",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "DefenderSettings",
+  "Description": "This integration setting enables Microsoft Defender for Cloud Apps (formerly 'Microsoft Cloud App Security' or 'MCAS' - see additional info) to communicate with Microsoft Defender for Cloud.",
+  "Risk": "Microsoft Defender for Cloud offers an additional layer of protection by using Azure Resource Manager events, which is considered to be the control plane for Azure. By analyzing the Azure Resource Manager records, Microsoft Defender for Cloud detects unusual or potentially harmful operations in the Azure subscription environment. Several of the preceding analytics are powered by Microsoft Defender for Cloud Apps. To benefit from these analytics, subscription must have a Cloud App Security license. Microsoft Defender for Cloud Apps works only with Standard Tier subscriptions.",
+  "RelatedUrl": "https://learn.microsoft.com/en-in/azure/defender-for-cloud/defender-for-cloud-introduction#secure-cloud-applications",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/azure/SecurityCenter/defender-cloud-apps-integration.html#",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "1. From Azure Home select the Portal Menu. 2. Select Microsoft Defender for Cloud. 3. Select Environment Settings blade. 4. Select the subscription. 5. Check App Service Defender Plan to On. 6. Select Save.",
+      "Url": "https://docs.microsoft.com/en-us/rest/api/securitycenter/settings/list"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Microsoft Defender for Cloud Apps works with Standard pricing tier Subscription. Choosing the Standard pricing tier of Microsoft Defender for Cloud incurs an additional cost per resource."
+}

prowler/providers/azure/services/defender/defender_ensure_mcas_is_enabled/defender_ensure_mcas_is_enabled.py

@@ -0,0 +1,28 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.defender.defender_client import defender_client
+
+
+class defender_ensure_mcas_is_enabled(Check):
+    def execute(self) -> Check_Report_Azure:
+        findings = []
+
+        for (
+            subscription_name,
+            settings,
+        ) in defender_client.settings.items():
+            report = Check_Report_Azure(self.metadata())
+            report.status = "FAIL"
+            report.subscription = subscription_name
+            report.resource_name = "MCAS"
+            report.resource_id = "MCAS"
+            report.status_extended = f"Microsoft Defender for Cloud Apps not exists for subscription {subscription_name}."
+            if "MCAS" in settings:
+                report.resource_id = settings["MCAS"].resource_id
+                report.status_extended = f"Microsoft Defender for Cloud Apps is disabeld for subscription {subscription_name}."
+                if settings["MCAS"].enabled:
+                    report.status = "PASS"
+                    report.status_extended = f"Microsoft Defender for Cloud Apps is enabled for subscription {subscription_name}."
+
+            findings.append(report)
+
+        return findings

prowler/providers/azure/services/defender/defender_ensure_wdatp_is_enabled/defender_ensure_wdatp_is_enabled.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "azure",
+  "CheckID": "defender_ensure_wdatp_is_enabled",
+  "CheckTitle": "Ensure that Microsoft Defender for Endpoint integration with Microsoft Defender for Cloud is selected",
+  "CheckType": [],
+  "ServiceName": "defender",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "DefenderSettings",
+  "Description": "This integration setting enables Microsoft Defender for Endpoint (formerly 'Advanced Threat Protection' or 'ATP' or 'WDATP' - see additional info) to communicate with Microsoft Defender for Cloud.",
+  "Risk": "Microsoft Defender for Endpoint integration brings comprehensive Endpoint Detection and Response (EDR) capabilities within Microsoft Defender for Cloud. This integration helps to spot abnormalities, as well as detect and respond to advanced attacks on endpoints monitored by Microsoft Defender for Cloud. MDE works only with Standard Tier subscriptions.",
+  "RelatedUrl": "https://learn.microsoft.com/en-in/azure/defender-for-cloud/integration-defender-for-endpoint?tabs=windows",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/azure/SecurityCenter/defender-endpoint-integration.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "",
+      "Url": "https://learn.microsoft.com/en-us/microsoft-365/security/defender-endpoint/azure-server-integration?view=o365-worldwide"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Microsoft Defender for Endpoint works with Standard pricing tier Subscription. Choosing the Standard pricing tier of Microsoft Defender for Cloud incurs an additional cost per resource."
+}

prowler/providers/azure/services/defender/defender_ensure_wdatp_is_enabled/defender_ensure_wdatp_is_enabled.py

@@ -0,0 +1,28 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.defender.defender_client import defender_client
+
+
+class defender_ensure_wdatp_is_enabled(Check):
+    def execute(self) -> Check_Report_Azure:
+        findings = []
+
+        for (
+            subscription_name,
+            settings,
+        ) in defender_client.settings.items():
+            report = Check_Report_Azure(self.metadata())
+            report.status = "FAIL"
+            report.subscription = subscription_name
+            report.resource_name = "WDATP"
+            report.resource_id = "WDATP"
+            report.status_extended = f"Microsoft Defender for Endpoint integration not exists for subscription {subscription_name}."
+            if "WDATP" in settings:
+                report.status_extended = f"Microsoft Defender for Endpoint integration is disabeld for subscription {subscription_name}."
+                report.resource_id = settings["WDATP"].resource_id
+                if settings["WDATP"].enabled:
+                    report.status = "PASS"
+                    report.status_extended = f"Microsoft Defender for Endpoint integration is enabled for subscription {subscription_name}."
+
+            findings.append(report)
+
+        return findings

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

@@ -15,6 +15,7 @@ class Defender(AzureService):
         self.pricings = self.__get_pricings__()
         self.auto_provisioning_settings = self.__get_auto_provisioning_settings__()
         self.assessments = self.__get_assessments__()
+        self.settings = self.__get_settings__()
         self.security_contacts = self.__get_security_contacts__()
 
     def __get_pricings__(self):
@@ -90,6 +91,30 @@ class Defender(AzureService):
                 )
         return assessments
 
+    def __get_settings__(self):
+        logger.info("Defender - Getting settings...")
+        settings = {}
+        for subscription_name, client in self.clients.items():
+            try:
+                settings_list = client.settings.list()
+                settings.update({subscription_name: {}})
+                for setting in settings_list:
+                    settings[subscription_name].update(
+                        {
+                            setting.name: Setting(
+                                resource_id=setting.id,
+                                resource_type=setting.type,
+                                kind=setting.kind,
+                                enabled=setting.enabled,
+                            )
+                        }
+                    )
+            except Exception as error:
+                logger.error(
+                    f"Subscription name: {subscription_name} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        return settings
+
     def __get_security_contacts__(self):
         logger.info("Defender - Getting security contacts...")
         security_contacts = {}
@@ -137,6 +162,13 @@ class Assesment(BaseModel):
     status: str
 
 
+class Setting(BaseModel):
+    resource_id: str
+    resource_type: str
+    kind: str
+    enabled: bool
+
+
 class SecurityContacts(BaseModel):
     resource_id: str
     emails: str

tests/providers/azure/services/defender/defender_ensure_mcas_is_enabled/defender_ensure_mcas_is_enabled_test.py

@@ -0,0 +1,115 @@
+from unittest import mock
+from uuid import uuid4
+
+from prowler.providers.azure.services.defender.defender_service import Setting
+from tests.providers.azure.azure_fixtures import AZURE_SUBSCRIPTION
+
+
+class Test_defender_ensure_mcas_is_enabled:
+    def test_defender_no_settings(self):
+        defender_client = mock.MagicMock
+        defender_client.settings = {}
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled import (
+                defender_ensure_mcas_is_enabled,
+            )
+
+            check = defender_ensure_mcas_is_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_defender_mcas_disabled(self):
+        resource_id = str(uuid4())
+        defender_client = mock.MagicMock
+        defender_client.settings = {
+            AZURE_SUBSCRIPTION: {
+                "MCAS": Setting(
+                    resource_id=resource_id,
+                    resource_type="Microsoft.Security/locations/settings",
+                    kind="DataExportSettings",
+                    enabled=False,
+                )
+            }
+        }
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled import (
+                defender_ensure_mcas_is_enabled,
+            )
+
+            check = defender_ensure_mcas_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Cloud Apps is disabeld for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "MCAS"
+            assert result[0].resource_id == resource_id
+
+    def test_defender_mcas_enabled(self):
+        resource_id = str(uuid4())
+        defender_client = mock.MagicMock
+        defender_client.settings = {
+            AZURE_SUBSCRIPTION: {
+                "MCAS": Setting(
+                    resource_id=resource_id,
+                    resource_type="Microsoft.Security/locations/settings",
+                    kind="DataExportSettings",
+                    enabled=True,
+                )
+            }
+        }
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled import (
+                defender_ensure_mcas_is_enabled,
+            )
+
+            check = defender_ensure_mcas_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Cloud Apps is enabled for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "MCAS"
+            assert result[0].resource_id == resource_id
+
+    def test_defender_mcas_no_settings(self):
+        defender_client = mock.MagicMock
+        defender_client.settings = {AZURE_SUBSCRIPTION: {}}
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_mcas_is_enabled.defender_ensure_mcas_is_enabled import (
+                defender_ensure_mcas_is_enabled,
+            )
+
+            check = defender_ensure_mcas_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Cloud Apps not exists for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "MCAS"
+            assert result[0].resource_id == "MCAS"

tests/providers/azure/services/defender/defender_ensure_wdatp_is_enabled/defender_ensure_wdatp_is_enabled_test.py

@@ -0,0 +1,115 @@
+from unittest import mock
+from uuid import uuid4
+
+from prowler.providers.azure.services.defender.defender_service import Setting
+from tests.providers.azure.azure_fixtures import AZURE_SUBSCRIPTION
+
+
+class Test_defender_ensure_wdatp_is_enabled:
+    def test_defender_no_settings(self):
+        defender_client = mock.MagicMock
+        defender_client.settings = {}
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled import (
+                defender_ensure_wdatp_is_enabled,
+            )
+
+            check = defender_ensure_wdatp_is_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_defender_wdatp_disabled(self):
+        resource_id = str(uuid4())
+        defender_client = mock.MagicMock
+        defender_client.settings = {
+            AZURE_SUBSCRIPTION: {
+                "WDATP": Setting(
+                    resource_id=resource_id,
+                    resource_type="Microsoft.Security/locations/settings",
+                    kind="DataExportSettings",
+                    enabled=False,
+                )
+            }
+        }
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled import (
+                defender_ensure_wdatp_is_enabled,
+            )
+
+            check = defender_ensure_wdatp_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Endpoint integration is disabeld for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "WDATP"
+            assert result[0].resource_id == resource_id
+
+    def test_defender_wdatp_enabled(self):
+        resource_id = str(uuid4())
+        defender_client = mock.MagicMock
+        defender_client.settings = {
+            AZURE_SUBSCRIPTION: {
+                "WDATP": Setting(
+                    resource_id=resource_id,
+                    resource_type="Microsoft.Security/locations/settings",
+                    kind="DataExportSettings",
+                    enabled=True,
+                )
+            }
+        }
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled import (
+                defender_ensure_wdatp_is_enabled,
+            )
+
+            check = defender_ensure_wdatp_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Endpoint integration is enabled for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "WDATP"
+            assert result[0].resource_id == resource_id
+
+    def test_defender_wdatp_no_settings(self):
+        defender_client = mock.MagicMock
+        defender_client.settings = {AZURE_SUBSCRIPTION: {}}
+
+        with mock.patch(
+            "prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled.defender_client",
+            new=defender_client,
+        ):
+            from prowler.providers.azure.services.defender.defender_ensure_wdatp_is_enabled.defender_ensure_wdatp_is_enabled import (
+                defender_ensure_wdatp_is_enabled,
+            )
+
+            check = defender_ensure_wdatp_is_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Microsoft Defender for Endpoint integration not exists for subscription {AZURE_SUBSCRIPTION}."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION
+            assert result[0].resource_name == "WDATP"
+            assert result[0].resource_id == "WDATP"

tests/providers/azure/services/defender/defender_service_test.py

@@ -7,6 +7,7 @@ from prowler.providers.azure.services.defender.defender_service import (
     Defender,
     Pricing,
     SecurityContacts,
+    Setting,
 )
 from tests.providers.azure.azure_fixtures import (
     AZURE_SUBSCRIPTION,
@@ -67,6 +68,19 @@ def mock_defender_get_security_contacts(_):
     }
 
 
+def mock_defender_get_settings(_):
+    return {
+        AZURE_SUBSCRIPTION: {
+            "MCAS": Setting(
+                resource_id="/subscriptions/resource_id",
+                resource_type="Microsoft.Security/locations/settings",
+                kind="DataExportSettings",
+                enabled=True,
+            )
+        }
+    }
+
+
 @patch(
     "prowler.providers.azure.services.defender.defender_service.Defender.__get_pricings__",
     new=mock_defender_get_pricings,
@@ -79,6 +93,10 @@ def mock_defender_get_security_contacts(_):
     "prowler.providers.azure.services.defender.defender_service.Defender.__get_assessments__",
     new=mock_defender_get_assessments,
 )
+@patch(
+    "prowler.providers.azure.services.defender.defender_service.Defender.__get_settings__",
+    new=mock_defender_get_settings,
+)
 @patch(
     "prowler.providers.azure.services.defender.defender_service.Defender.__get_security_contacts__",
     new=mock_defender_get_security_contacts,
@@ -151,6 +169,22 @@ class Test_Defender_Service:
         )
         assert defender.assessments[AZURE_SUBSCRIPTION]["default"].status == "Healthy"
 
+    def test__get_settings__(self):
+        defender = Defender(set_mocked_azure_audit_info())
+        assert len(defender.settings) == 1
+        assert (
+            defender.settings[AZURE_SUBSCRIPTION]["MCAS"].resource_id
+            == "/subscriptions/resource_id"
+        )
+        assert (
+            defender.settings[AZURE_SUBSCRIPTION]["MCAS"].resource_type
+            == "Microsoft.Security/locations/settings"
+        )
+        assert (
+            defender.settings[AZURE_SUBSCRIPTION]["MCAS"].kind == "DataExportSettings"
+        )
+        assert defender.settings[AZURE_SUBSCRIPTION]["MCAS"].enabled
+
     def test__get_security_contacts__(self):
         defender = Defender(set_mocked_azure_audit_info())
         assert len(defender.security_contacts) == 1