fix(arm): consider encryption property in CKV_AZURE_2

checkov/arm/base_resource_check.py

@@ -45,7 +45,12 @@ def scan_entity_conf(self, conf: dict[str, Any], entity_type: str) -> CheckResul
             self.api_version = conf["api_version"]
             conf["config"]["apiVersion"] = conf["api_version"]  # set for better reusability of existing ARM checks
 
-            return self.scan_resource_conf(conf["config"], entity_type)  # type:ignore[no-any-return]  # issue with multi_signature annotation
+            resource_conf = conf["config"]
+            if "loop_type" in resource_conf:
+                # this means the whole resource block is surrounded by a for loop
+                resource_conf = resource_conf["config"]
+
+            return self.scan_resource_conf(resource_conf, entity_type)  # type:ignore[no-any-return]  # issue with multi_signature annotation
 
         self.api_version = None
 

checkov/arm/checks/resource/AzureManagedDiscEncryption.py

@@ -4,6 +4,7 @@
 
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.arm.base_resource_check import BaseResourceCheck
+from checkov.common.util.data_structures_utils import find_in_dict
 
 
 class AzureManagedDiscEncryption(BaseResourceCheck):
@@ -15,15 +16,21 @@ def __init__(self) -> None:
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
     def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
-        if "properties" in conf:
-            if "encryptionSettingsCollection" in conf["properties"]:
-                if "enabled" in conf["properties"]["encryptionSettingsCollection"]:
-                    if str(conf["properties"]["encryptionSettingsCollection"]["enabled"]).lower() == "true":
-                        return CheckResult.PASSED
-            elif "encryptionSettings" in conf["properties"]:
-                if "enabled" in conf["properties"]["encryptionSettings"]:
-                    if str(conf["properties"]["encryptionSettings"]["enabled"]).lower() == "true":
-                        return CheckResult.PASSED
+        properties = conf.get("properties")
+        if properties:
+            encryption = properties.get("encryption")
+            if encryption:
+                # if the block exists, then it is enabled
+                return CheckResult.PASSED
+
+            encryption_enabled = find_in_dict(input_dict=properties, key_path="encryptionSettingsCollection/enabled")
+            if str(encryption_enabled).lower() == "true":
+                return CheckResult.PASSED
+
+            encryption_enabled = find_in_dict(input_dict=properties, key_path="encryptionSettings/enabled")
+            if str(encryption_enabled).lower() == "true":
+                return CheckResult.PASSED
+
         return CheckResult.FAILED
 
 

tests/arm/checks/resource/example_AzureManagedDiscEncryption/azureManagedDiscEncryption-PASSED_3.json

@@ -0,0 +1,76 @@
+{
+  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+  "contentVersion": "1.0.0.0",
+  "parameters": {
+    "vhdUri": {
+      "type": "string",
+      "metadata": {
+        "description": "Storage VHD Uri"
+      }
+    },
+    "managedDiskName": {
+      "type": "string",
+      "metadata": {
+        "description": "Name of the managed disk to be copied"
+      }
+    },
+    "keyVaultResourceID": {
+      "type": "string",
+      "metadata": {
+        "description": "KeyVault resource id. Ex: /subscriptions/subscriptionid/resourceGroups/contosorg/providers/Microsoft.KeyVault/vaults/contosovault"
+      }
+    },
+    "keyVaultSecretUrl": {
+      "type": "string",
+      "metadata": {
+        "description": "KeyVault secret Url. Ex: https://contosovault.vault.azure.net/secrets/contososecret/e088818e865e48488cf363af16dea596"
+      }
+    },
+    "kekUrl": {
+      "type": "string",
+      "defaultValue": "",
+      "metadata": {
+        "description": "KeyVault key encryption key Url. Ex: https://contosovault.vault.azure.net/keys/contosokek/562a4bb76b524a1493a6afe8e536ee78"
+      }
+    },
+    "kekVaultResourceID": {
+      "type": "string",
+      "defaultValue": "",
+      "metadata": {
+        "description": "KekVault resource id. Ex: /subscriptions/subscriptionid/resourceGroups/contosorg/providers/Microsoft.KeyVault/vaults/contosovault"
+      }
+    },
+    "location": {
+      "type": "string",
+      "defaultValue": "[resourceGroup().location]",
+      "metadata": {
+        "description": "Location for all resources."
+      }
+    }
+  },
+  "variables": {
+    "location": "[parameters('location')]",
+    "storageAccountType": "Standard_LRS",
+    "diskSzie": "128"
+  },
+  "resources": [
+    {
+      "apiVersion": "2021-12-01",
+      "type": "Microsoft.Compute/disks",
+      "name": "encryptionBlock",
+      "location": "[variables('location')]",
+      "properties": {
+        "creationData": {
+          "createOption": "Import",
+          "sourceUri": "[parameters('vhdUri')]"
+        },
+        "accountType": "[variables('storageAccountType')]",
+        "diskSizeGB": "[variables('diskSzie')]",
+        "encryption": {
+          "diskEncryptionSetId": "exampleSetId",
+          "type": "EncryptionAtRestWithCustomerKey"
+        }
+      }
+    }
+  ]
+}

tests/arm/checks/resource/test_AzureManagedDiscEncryption.py

@@ -20,6 +20,7 @@ def test_summary(self):
         passing_resources = {
             "Microsoft.Compute/disks.enabled",
             "Microsoft.Compute/disks.collectionEnabled",
+            "Microsoft.Compute/disks.encryptionBlock",
         }
 
         failing_resources = {
@@ -29,8 +30,8 @@ def test_summary(self):
         passed_check_resources = {c.resource for c in report.passed_checks}
         failed_check_resources = {c.resource for c in report.failed_checks}
 
-        self.assertEqual(summary["passed"], 2)
-        self.assertEqual(summary["failed"], 1)
+        self.assertEqual(summary["passed"], len(passing_resources))
+        self.assertEqual(summary["failed"], len(failing_resources))
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 

tests/bicep/examples/loop.bicep

@@ -0,0 +1,21 @@
+resource Disks 'Microsoft.Compute/disks@2022-07-02' = [for (disk, i) in dataDisks: {
+  name: disk.diskName
+  location: location
+  tags: tags
+  sku: {
+    name: disk.storageAccountType
+  }
+  zones: [
+    avZone
+  ]
+  properties: {
+    creationData: {
+      createOption: 'Empty'
+    }
+    diskSizeGB: disk.diskSizeGB
+    encryption: {
+      type: 'EncryptionAtRestWithCustomerKey'
+      diskEncryptionSetId: diskEncryptionSetId
+    }
+  }
+}]

tests/bicep/test_graph_manager.py

@@ -13,16 +13,17 @@ def test_build_graph_from_source_directory():
     existing_file = EXAMPLES_DIR / "existing.bicep"
     playground_file = EXAMPLES_DIR / "playground.bicep"
     graph_file = EXAMPLES_DIR / "graph.bicep"
+    loop_file = EXAMPLES_DIR / "loop.bicep"
     graph_manager = BicepGraphManager(db_connector=NetworkxConnector())
 
     # when
     local_graph, definitions = graph_manager.build_graph_from_source_directory(source_dir=str(EXAMPLES_DIR))
 
     # then
-    assert set(definitions.keys()) == {existing_file, playground_file, graph_file}  # should no include 'malformed.bicep' file
+    assert set(definitions.keys()) == {existing_file, playground_file, graph_file, loop_file}  # should not include 'malformed.bicep' file
 
-    assert len(local_graph.vertices) == 46
-    assert len(local_graph.edges) == 41
+    assert len(local_graph.vertices) == 48
+    assert len(local_graph.edges) == 42
 
     storage_account_idx = local_graph.vertices_by_name["diagsAccount"]  # vertices_by_name exists for BicepGraphManager
     storage_account = local_graph.vertices[storage_account_idx]

tests/bicep/test_runner.py

@@ -5,7 +5,6 @@
 from checkov.bicep.runner import Runner
 from checkov.arm.runner import Runner as ArmRunner
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.bridgecrew.code_categories import CodeCategoryConfiguration
 from checkov.common.bridgecrew.severities import Severities, BcSeverities
 from checkov.common.graph.db_connectors.igraph.igraph_db_connector import IgraphConnector
 from checkov.common.graph.db_connectors.networkx.networkx_db_connector import NetworkxConnector
@@ -200,3 +199,18 @@ def test_runner_extra_resources(graph_connector):
     assert extra_resource.file_abs_path == str(test_file)
     assert extra_resource.file_path.endswith("playground.bicep")
 
+
+def test_runner_loop_resource():
+    # given
+    test_file = EXAMPLES_DIR / "loop.bicep"
+
+    # when
+    report = Runner().run(root_folder="", files=[str(test_file)], runner_filter=RunnerFilter(checks=["CKV_AZURE_2"]))
+
+    # then
+    summary = report.get_summary()
+
+    assert summary["passed"] == 1
+    assert summary["failed"] == 0
+    assert summary["skipped"] == 0
+    assert summary["parsing_errors"] == 0