feat(azure/vm): add new check vm_jit_access_enabled (#8202)

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

prowler/CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 ### Added
 - Certificate authentication for M365 provider [(#8404)](https://github.com/prowler-cloud/prowler/pull/8404)
 - `vm_sufficient_daily_backup_retention_period` check for Azure provider [(#8200)](https://github.com/prowler-cloud/prowler/pull/8200)
+- `vm_jit_access_enabled` check for Azure provider [(#8202)](https://github.com/prowler-cloud/prowler/pull/8202)
 
 ### Changed
 

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

@@ -25,6 +25,7 @@ class Defender(AzureService):
             ).token
         )
         self.iot_security_solutions = self._get_iot_security_solutions()
+        self.jit_policies = self._get_jit_policies()
 
     def _get_pricings(self):
         logger.info("Defender - Getting pricings...")
@@ -246,6 +247,44 @@ class Defender(AzureService):
                 )
         return iot_security_solutions
 
+    def _get_jit_policies(self) -> dict[str, dict]:
+        """
+        Get all JIT policies for all subscriptions.
+
+        Returns:
+            A dictionary of JIT policies for each subscription. The format will be:
+            {
+                "subscription_name": {
+                    "jit_policy_id": JITPolicy
+                }
+            }
+        """
+        logger.info("Defender - Getting JIT policies...")
+        jit_policies = {}
+        for subscription_name, client in self.clients.items():
+            try:
+                jit_policies[subscription_name] = {}
+                policies = client.jit_network_access_policies.list()
+                for policy in policies:
+                    vm_ids = set()
+                    for vm in getattr(policy, "virtual_machines", []):
+                        vm_ids.add(vm.id)
+                    jit_policies[subscription_name].update(
+                        {
+                            policy.id: JITPolicy(
+                                id=policy.id,
+                                name=policy.name,
+                                location=getattr(policy, "location", "Global"),
+                                vm_ids=vm_ids,
+                            ),
+                        }
+                    )
+            except Exception as error:
+                logger.error(
+                    f"Subscription name: {subscription_name} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        return jit_policies
+
 
 class Pricing(BaseModel):
     resource_id: str
@@ -317,3 +356,10 @@ class IoTSecuritySolution(BaseModel):
     resource_id: str
     name: str
     status: str
+
+
+class JITPolicy(BaseModel):
+    id: str
+    name: str
+    location: str = ""
+    vm_ids: list[str] = []

prowler/providers/azure/services/vm/vm_jit_access_enabled/vm_jit_access_enabled.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "azure",
+  "CheckID": "vm_jit_access_enabled",
+  "CheckTitle": "Enable Just-In-Time Access for Virtual Machines",
+  "CheckType": [],
+  "ServiceName": "vm",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}",
+  "Severity": "high",
+  "ResourceType": "Microsoft.Compute/virtualMachines",
+  "Description": "Ensure that Microsoft Azure virtual machines are configured to use Just-in-Time (JIT) access.",
+  "Risk": "Without JIT access, management ports such as 22 (SSH) and 3389 (RDP) may be exposed, increasing the risk of brute-force and DDoS attacks.",
+  "RelatedUrl": "https://docs.microsoft.com/en-us/azure/security-center/security-center-just-in-time?tabs=jit-config-asc%2Cjit-request-asc",
+  "Remediation": {
+    "Code": {
+      "CLI": "az security jit-policy list --query '[*].virtualMachines[*].id | []'",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable Just-in-Time (JIT) network access for your Microsoft Azure virtual machines using the Azure Portal under Security Center > Just-in-time VM access.",
+      "Url": "https://docs.microsoft.com/en-us/azure/security-center/security-center-just-in-time?tabs=jit-config-asc%2Cjit-request-asc"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "JIT access can only be enabled via the Azure Portal. Ensure Security Center standard pricing tier for servers is enabled."
+}

prowler/providers/azure/services/vm/vm_jit_access_enabled/vm_jit_access_enabled.py

@@ -0,0 +1,33 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.defender.defender_client import defender_client
+from prowler.providers.azure.services.vm.vm_client import vm_client
+
+
+class vm_jit_access_enabled(Check):
+    """
+    Ensure that Microsoft Azure virtual machines are configured to use Just-in-Time (JIT) access.
+
+    This check evaluates whether JIT access is enabled for each VM to reduce the attack surface.
+    - PASS: VM has JIT access enabled.
+    - FAIL: VM does not have JIT access enabled.
+    """
+
+    def execute(self):
+        findings = []
+        jit_enabled_vms = set()
+        for subscription_name, vms in vm_client.virtual_machines.items():
+            for jit_policy in defender_client.jit_policies[subscription_name].values():
+                jit_enabled_vms.update(jit_policy.vm_ids)
+            for vm in vms.values():
+                report = Check_Report_Azure(metadata=self.metadata(), resource=vm)
+                report.subscription = subscription_name
+                if vm.resource_id.lower() in {
+                    vm_id.lower() for vm_id in jit_enabled_vms
+                }:
+                    report.status = "PASS"
+                    report.status_extended = f"VM {vm.resource_name} in subscription {subscription_name} has JIT (Just-in-Time) access enabled."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"VM {vm.resource_name} in subscription {subscription_name} does not have JIT (Just-in-Time) access enabled."
+                findings.append(report)
+        return findings

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

@@ -6,6 +6,7 @@ from prowler.providers.azure.services.defender.defender_service import (
     AutoProvisioningSetting,
     Defender,
     IoTSecuritySolution,
+    JITPolicy,
     Pricing,
     SecurityContactConfiguration,
     Setting,
@@ -103,6 +104,19 @@ def mock_defender_get_iot_security_solutions(_):
     }
 
 
+def mock_defender_get_jit_policies(_):
+    return {
+        AZURE_SUBSCRIPTION_ID: {
+            "policy-1": JITPolicy(
+                id="policy-1",
+                name="JITPolicy1",
+                location="eastus",
+                vm_ids=["vm-1", "vm-2"],
+            )
+        }
+    }
+
+
 @patch(
     "prowler.providers.azure.services.defender.defender_service.Defender._get_pricings",
     new=mock_defender_get_pricings,
@@ -127,6 +141,10 @@ def mock_defender_get_iot_security_solutions(_):
     "prowler.providers.azure.services.defender.defender_service.Defender._get_iot_security_solutions",
     new=mock_defender_get_iot_security_solutions,
 )
+@patch(
+    "prowler.providers.azure.services.defender.defender_service.Defender._get_jit_policies",
+    new=mock_defender_get_jit_policies,
+)
 class Test_Defender_Service:
     def test_get_client(self):
         defender = Defender(set_mocked_azure_provider())
@@ -255,3 +273,13 @@ class Test_Defender_Service:
             ].status
             == "Enabled"
         )
+
+    def test_get_jit_policies(self):
+        defender = Defender(set_mocked_azure_provider())
+        assert AZURE_SUBSCRIPTION_ID in defender.jit_policies
+        assert "policy-1" in defender.jit_policies[AZURE_SUBSCRIPTION_ID]
+        policy1 = defender.jit_policies[AZURE_SUBSCRIPTION_ID]["policy-1"]
+        assert policy1.id == "policy-1"
+        assert policy1.name == "JITPolicy1"
+        assert policy1.location == "eastus"
+        assert set(policy1.vm_ids) == {"vm-1", "vm-2"}

tests/providers/azure/services/vm/vm_jit_access_enabled/vm_jit_access_enabled_test.py

@@ -0,0 +1,290 @@
+from unittest import mock
+from uuid import uuid4
+
+from prowler.providers.azure.services.defender.defender_service import JITPolicy
+from prowler.providers.azure.services.vm.vm_service import VirtualMachine
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+
+class Test_vm_jit_access_enabled:
+    def test_no_subscriptions(self):
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {}
+        defender_client = mock.MagicMock()
+        defender_client.jit_policies = {}
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert result == []
+
+    def test_no_vms(self):
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {}}
+        defender_client = mock.MagicMock()
+        defender_client.jit_policies = {AZURE_SUBSCRIPTION_ID: {}}
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert result == []
+
+    def test_vm_with_jit_enabled(self):
+        vm_id = str(uuid4())
+        vm_name = "TestVM"
+        vm_location = "eastus"
+        vm = VirtualMachine(
+            resource_id=vm_id,
+            resource_name=vm_name,
+            location=vm_location,
+            security_profile=None,
+            extensions=[],
+            storage_profile=None,
+        )
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {vm_id: vm}}
+        defender_client = mock.MagicMock()
+        jit_policy = JITPolicy(
+            id="policy1",
+            name="JITPolicy1",
+            location="eastus",
+            vm_ids={vm_id},
+        )
+        defender_client.jit_policies = {
+            AZURE_SUBSCRIPTION_ID: {jit_policy.id: jit_policy}
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_id == vm_id
+            assert result[0].resource_name == vm_name
+            assert "has JIT (Just-in-Time) access enabled" in result[0].status_extended
+
+    def test_vm_with_jit_disabled(self):
+        vm_id = str(uuid4())
+        vm_name = "TestVM"
+        vm_location = "eastus"
+        vm = VirtualMachine(
+            resource_id=vm_id,
+            resource_name=vm_name,
+            location=vm_location,
+            security_profile=None,
+            extensions=[],
+            storage_profile=None,
+        )
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {vm_id: vm}}
+        defender_client = mock.MagicMock()
+        # JIT policy does not include this VM
+        jit_policy = JITPolicy(
+            id="policy1",
+            name="JITPolicy1",
+            location="eastus",
+            vm_ids={"some-other-id"},
+        )
+        defender_client.jit_policies = {
+            AZURE_SUBSCRIPTION_ID: {jit_policy.id: jit_policy}
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_id == vm_id
+            assert result[0].resource_name == vm_name
+            assert (
+                "does not have JIT (Just-in-Time) access enabled"
+                in result[0].status_extended
+            )
+
+    def test_vm_id_case_insensitivity(self):
+        vm_id = str(uuid4())
+        vm_name = "TestVM"
+        vm_location = "eastus"
+        upper_vm_id = vm_id.upper()
+        vm = VirtualMachine(
+            resource_id=upper_vm_id,
+            resource_name=vm_name,
+            location=vm_location,
+            security_profile=None,
+            extensions=[],
+            storage_profile=None,
+        )
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {AZURE_SUBSCRIPTION_ID: {upper_vm_id: vm}}
+        defender_client = mock.MagicMock()
+        jit_policy = JITPolicy(
+            id="policy1",
+            name="JITPolicy1",
+            location="eastus",
+            vm_ids={vm_id.lower()},
+        )
+        defender_client.jit_policies = {
+            AZURE_SUBSCRIPTION_ID: {jit_policy.id: jit_policy}
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == upper_vm_id
+            assert "has JIT (Just-in-Time) access enabled" in result[0].status_extended
+
+    def test_multiple_vms_and_policies(self):
+        vm_id_1 = str(uuid4())
+        vm_id_2 = str(uuid4())
+        vm1 = VirtualMachine(
+            resource_id=vm_id_1,
+            resource_name="VM1",
+            location="eastus",
+            security_profile=None,
+            extensions=[],
+            storage_profile=None,
+        )
+        vm2 = VirtualMachine(
+            resource_id=vm_id_2,
+            resource_name="VM2",
+            location="eastus",
+            security_profile=None,
+            extensions=[],
+            storage_profile=None,
+        )
+        vm_client = mock.MagicMock()
+        vm_client.virtual_machines = {
+            AZURE_SUBSCRIPTION_ID: {vm_id_1: vm1, vm_id_2: vm2}
+        }
+        defender_client = mock.MagicMock()
+        jit_policy_1 = JITPolicy(
+            id="policy1",
+            name="JITPolicy1",
+            location="eastus",
+            vm_ids={vm_id_1},
+        )
+        jit_policy_2 = JITPolicy(
+            id="policy2",
+            name="JITPolicy2",
+            location="eastus",
+            vm_ids=set(),
+        )
+        defender_client.jit_policies = {
+            AZURE_SUBSCRIPTION_ID: {
+                jit_policy_1.id: jit_policy_1,
+                jit_policy_2.id: jit_policy_2,
+            }
+        }
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.vm_client",
+                new=vm_client,
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled.defender_client",
+                new=defender_client,
+            ),
+        ):
+            from prowler.providers.azure.services.vm.vm_jit_access_enabled.vm_jit_access_enabled import (
+                vm_jit_access_enabled,
+            )
+
+            check = vm_jit_access_enabled()
+            result = check.execute()
+            assert len(result) == 2
+            for r in result:
+                if r.resource_id == vm_id_1:
+                    assert r.status == "PASS"
+                elif r.resource_id == vm_id_2:
+                    assert r.status == "FAIL"