feat(arm): arm custom policy support (#6769)

* arm custom policy support

checkov/arm/graph_builder/context_definitions.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List, Tuple, Dict, Any
+
+
+def build_definitions_context(definitions: Dict[Path, Any], definitions_raw: Dict[Path, List[Tuple[int, str]]]
+                              ) -> Dict[str, Dict[str, Any]]:
+    # TODO
+    return {}

checkov/arm/graph_builder/graph_to_definitions.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from checkov.arm.graph_builder.graph_components.blocks import ArmBlock
+
+
+def convert_graph_vertices_to_definitions(
+    vertices: list[ArmBlock], root_folder: str | Path | None
+) -> tuple[dict[Path, Any], dict[str, dict[str, Any]]]:
+    # TODO
+    return ({}, {})

checkov/arm/graph_builder/local_graph.py

@@ -6,8 +6,9 @@
 
 from checkov.arm.graph_builder.graph_components.block_types import BlockType
 from checkov.arm.graph_builder.graph_components.blocks import ArmBlock
+from checkov.arm.utils import ArmElements, extract_resource_name_from_resource_id_func, \
+    extract_resource_name_from_reference_func
 from checkov.arm.graph_builder.variable_rendering.renderer import ArmVariableRenderer
-from checkov.arm.utils import ArmElements
 from checkov.common.graph.graph_builder import CustomAttributes, Edge
 from checkov.common.graph.graph_builder.local_graph import LocalGraph
 from checkov.common.util.consts import START_LINE, END_LINE
@@ -16,6 +17,10 @@
 if TYPE_CHECKING:
     from checkov.common.graph.graph_builder.local_graph import _Block
 
+DEPENDS_ON_FIELD = 'dependsOn'
+RESOURCE_ID_FUNC = 'resourceId('
+REFERENCE_FUNC = 'reference('
+
 
 class ArmLocalGraph(LocalGraph[ArmBlock]):
     def __init__(self, definitions: dict[str, dict[str, Any]]) -> None:
@@ -27,14 +32,21 @@ def __init__(self, definitions: dict[str, dict[str, Any]]) -> None:
 
     def build_graph(self, render_variables: bool = False) -> None:
         self._create_vertices()
-        logging.debug(f"[ArmLocalGraph] created {len(self.vertices)} vertices")
+        logging.warning(f"[ArmLocalGraph] created {len(self.vertices)} vertices")
 
-        self._create_edges()
-        logging.debug(f"[ArmLocalGraph] created {len(self.edges)} edges")
+        '''
+            In order to resolve the resources names for the dependencies we need to render the variables first
+            Examples: https://learn.microsoft.com/en-us/azure/azure-resource-manager/templates/resource-dependency
+        '''
+
+        self._create_vars_and_parameters_edges()
         if render_variables:
             renderer = ArmVariableRenderer(self)
             renderer.render_variables_from_local_graph()
 
+        self._create_edges()
+        logging.warning(f"[ArmLocalGraph] created {len(self.edges)} edges")
+
     def _create_vertices(self) -> None:
         for file_path, definition in self.definitions.items():
             self._create_parameter_vertices(file_path=file_path, parameters=definition.get(ArmElements.PARAMETERS))
@@ -83,7 +95,7 @@ def _create_parameter_vertices(self, file_path: str, parameters: dict[str, dict[
             if name in (START_LINE, END_LINE):
                 continue
             if not isinstance(config, dict):
-                logging.debug(f"[ArmLocalGraph] parameter {name} has wrong type {type(config)}")
+                logging.warning(f"[ArmLocalGraph] parameter {name} has wrong type {type(config)}")
                 continue
 
             attributes = pickle_deepcopy(config)
@@ -121,28 +133,30 @@ def _create_resource_vertices(self, file_path: str, resources: list[dict[str, An
                     path=file_path,
                     block_type=BlockType.RESOURCE,
                     attributes=attributes,
-                    id=f"{resource_type}.{resource_name}",
+                    id=f"{resource_type}.{resource_name}"
                 )
             )
 
     def _create_edges(self) -> None:
-        self._create_vars_and_parameters_edges()
-        # todo add explicit references edges
-
-    def _create_edge(self, element_name: str, origin_vertex_index: int, label: str) -> None:
-        vertex_name = element_name
-        if "." in vertex_name:
-            # special case for bicep and arm elements, when properties are accessed
-            vertex_name = vertex_name.split(".")[0]
-
-        dest_vertex_index = self.vertices_by_name.get(vertex_name)
-        if dest_vertex_index or dest_vertex_index == 0:
-            if origin_vertex_index == dest_vertex_index:
-                return
-            edge = Edge(origin_vertex_index, dest_vertex_index, label)
-            self.edges.append(edge)
-            self.out_edges[origin_vertex_index].append(edge)
-            self.in_edges[dest_vertex_index].append(edge)
+        for origin_vertex_index, vertex in enumerate(self.vertices):
+            if DEPENDS_ON_FIELD in vertex.attributes:
+                self._create_explicit_edge(origin_vertex_index, vertex.name, vertex.attributes['dependsOn'])
+            self._create_implicit_edges(origin_vertex_index, vertex.name, vertex.attributes)
+
+    def _create_explicit_edge(self, origin_vertex_index: int, resource_name: str, deps: list[str]) -> None:
+        for dep in deps:
+            if RESOURCE_ID_FUNC in dep:
+                processed_dep = extract_resource_name_from_resource_id_func(dep)
+            else:
+                processed_dep = dep.split('/')[-1]
+            # Check if the processed dependency exists in the map
+            if processed_dep in self.vertices_by_name:
+                self._create_edge(processed_dep, origin_vertex_index, f'{resource_name}->{processed_dep}')
+            else:
+                # Dependency not found
+                logging.warning(f"[ArmLocalGraph] resource dependency {processed_dep} defined in {dep} for resource"
+                                f" {resource_name} not found")
+                continue
 
     def _create_vars_and_parameters_edges(self) -> None:
         pattern = r"(variables|parameters)\('(\w+)'\)"
@@ -156,13 +170,32 @@ def _create_vars_and_parameters_edges(self) -> None:
                         var_name = match[1]
                         self._create_edge(var_name, origin_vertex_index, attr_key)
 
+    def _create_edge(self, element_name: str, origin_vertex_index: int, label: str) -> None:
+        dest_vertex_index = self.vertices_by_name.get(element_name)
+        if origin_vertex_index == dest_vertex_index or dest_vertex_index is None:
+            return
+        edge = Edge(origin_vertex_index, dest_vertex_index, label)
+        self.edges.append(edge)
+        self.out_edges[origin_vertex_index].append(edge)
+        self.in_edges[dest_vertex_index].append(edge)
+
+    def _create_implicit_edges(self, origin_vertex_index: int, resource_name: str, resource: dict[str, Any]) -> None:
+        for value in resource.values():
+            if isinstance(value, str):
+                if REFERENCE_FUNC in value:
+                    self._create_implicit_edge(origin_vertex_index, resource_name, value)
+
+    def _create_implicit_edge(self, origin_vertex_index: int, resource_name: str, reference_string: str) -> None:
+        dep_name = extract_resource_name_from_reference_func(reference_string)
+        self._create_edge(dep_name, origin_vertex_index, f'{resource_name}->{dep_name}')
+
     def update_vertices_configs(self) -> None:
         # not used
         pass
 
     @staticmethod
     def update_vertex_config(
-        vertex: _Block, changed_attributes: list[str] | dict[str, Any], has_dynamic_blocks: bool = False
+            vertex: _Block, changed_attributes: list[str] | dict[str, Any], has_dynamic_blocks: bool = False
     ) -> None:
         # not used
         pass

checkov/arm/utils.py

@@ -40,7 +40,7 @@ def get_scannable_file_paths(root_folder: str | None = None, excluded_paths: lis
 
 
 def get_files_definitions(
-    files: Iterable[str], filepath_fn: Callable[[str], str] | None = None
+        files: Iterable[str], filepath_fn: Callable[[str], str] | None = None
 ) -> tuple[dict[str, dict[str, Any]], dict[str, list[tuple[int, str]]], list[str]]:
     """Parses ARM files into its definitions and raw data"""
 
@@ -60,3 +60,32 @@ def get_files_definitions(
             parsing_errors.append(os.path.normpath(file))
 
     return definitions, definitions_raw, parsing_errors
+
+
+def extract_resource_name_from_resource_id_func(resource_id: str) -> str:
+    '''
+        Examples:
+            resourceId('Microsoft.Network/virtualNetworks/', virtualNetworkName) -> virtualNetworkName
+    '''
+    return clean_string(resource_id.split(',')[-1].split(')')[0])
+
+
+def extract_resource_name_from_reference_func(reference: str) -> str:
+    '''
+        Examples:
+                reference('storageAccountName') -> storageAccountName
+                reference('myStorage').primaryEndpoints -> myStorage
+                reference('myStorage', '2022-09-01', 'Full').location -> myStorage
+                reference(resourceId('storageResourceGroup', 'Microsoft.Storage/storageAccounts', 'storageAccountName')), '2022-09-01') -> storageAccountName
+                reference(resourceId('Microsoft.Network/publicIPAddresses', 'ipAddressName')) -> ipAddressName
+    '''
+    resource_name = ')'.join(reference.split('reference(', 1)[1].split(')')[:-1])
+    if 'resourceId' in resource_name:
+        return clean_string(
+            ''.join(resource_name.split('resourceId(', 1)[1].split(')')[0]).split(',')[-1])
+    else:
+        return clean_string(resource_name.split(',')[0].split('/')[-1])
+
+
+def clean_string(input: str) -> str:
+    return input.replace("'", '').replace(" ", "")

checkov/common/bridgecrew/integration_features/features/custom_policies_integration.py

@@ -86,6 +86,8 @@ def pre_scan(self) -> None:
                                 get_graph_checks_registry("kubernetes").checks.append(check)
                             elif f.lower() == "bicep":
                                 get_graph_checks_registry("bicep").checks.append(check)
+                            elif f.lower() == "arm":
+                                get_graph_checks_registry("arm").checks.append(check)
                     else:
                         for registry in get_all_graph_checks_registries():
                             registry.checks.append(check)

checkov/common/checks_infra/registry.py

@@ -23,7 +23,7 @@
 
 GraphSupportedIACFrameworks = [GraphSource.TERRAFORM, GraphSource.CLOUDFORMATION, GraphSource.KUBERNETES,
                                GraphSource.TERRAFORM_PLAN, GraphSource.KUSTOMIZE, GraphSource.BICEP,
-                               GraphSource.GITHUB_ACTION, GraphSource.HELM, GraphSource.ANSIBLE]
+                               GraphSource.GITHUB_ACTION, GraphSource.HELM, GraphSource.ANSIBLE, GraphSource.ARM]
 
 
 class Registry(BaseRegistry):

tests/arm/examples/ExplicitDepsResources/interface.json

@@ -0,0 +1,41 @@
+{
+    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+    "contentVersion": "1.0.0.0",
+    "resources": [
+      {
+        "type": "Microsoft.Network/publicIPAddresses",
+        "apiVersion": "2020-06-01",
+        "name": "PublicIP1",
+        "location": "westus",
+        "properties": {
+          "publicIPAllocationMethod": "Dynamic"
+        }
+      },
+      {
+        "type": "Microsoft.Network/networkInterfaces",
+        "apiVersion": "2020-06-01",
+        "name": "NIC1",
+        "location": "westus",
+        "dependsOn": [
+          "Microsoft.Network/virtualNetworks/VNet1/subnets/Subnet1",
+          "Microsoft.Network/publicIPAddresses/PublicIP1"
+        ],
+        "properties": {
+          "ipConfigurations": [
+            {
+              "name": "ipconfig1",
+              "properties": {
+                "subnet": {
+                  "id": "[resourceId('Microsoft.Network/virtualNetworks/subnets', 'VNet1', 'Subnet1')]"
+                },
+                "privateIPAllocationMethod": "Dynamic",
+                "publicIPAddress": {
+                  "id": "[resourceId('Microsoft.Network/publicIPAddresses', 'PublicIP1')]"
+                }
+              }
+            }
+          ]
+        }
+      }
+    ]
+  }

tests/arm/examples/ExplicitDepsResources/storage.json

@@ -0,0 +1,56 @@
+{
+    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+    "contentVersion": "1.0.0.0",
+    "resources": [
+      {
+        "type": "Microsoft.Network/storageAccounts",
+        "apiVersion": "2019-06-01",
+        "name": "storaccount1",
+        "location": "westus",
+        "sku": {
+          "name": "Standard_LRS"
+        },
+        "kind": "StorageV2",
+        "properties": {}
+      },
+      {
+        "type": "Microsoft.Compute/virtualMachines",
+        "apiVersion": "2019-12-01",
+        "name": "VM1",
+        "location": "westus",
+        "dependsOn": [
+          "Microsoft.Network/virtualNetworks/VNet1/subnets/Subnet1",
+          "Microsoft.Network/storageAccounts/storaccount1"
+        ],
+        "properties": {
+          "hardwareProfile": {
+            "vmSize": "Standard_D2s_v3"
+          },
+          "storageProfile": {
+            "imageReference": {
+              "publisher": "MicrosoftWindowsServer",
+              "offer": "WindowsServer",
+              "sku": "2019-Datacenter",
+              "version": "latest"
+            },
+            "osDisk": {
+              "createOption": "FromImage"
+            },
+            "dataDisks": []
+          },
+          "osProfile": {
+            "computerName": "vm1",
+            "adminUsername": "adminuser",
+            "adminPassword": "Password123!"
+          },
+          "networkProfile": {
+            "networkInterfaces": [
+              {
+                "id": "[resourceId('Microsoft.Network/networkInterfaces', 'nic1')]"
+              }
+            ]
+          }
+        }
+      }
+    ]
+  }

tests/arm/examples/ExplicitDepsResources/subnet.json

@@ -0,0 +1,36 @@
+{
+    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+    "contentVersion": "1.0.0.0",
+    "resources": [
+      {
+        "type": "Microsoft.Network/virtualNetworks",
+        "apiVersion": "2020-06-01",
+        "name": "VNet1",
+        "location": "westus",
+        "properties": {
+          "addressSpace": {
+            "addressPrefixes": ["10.0.0.0/16"]
+          },
+          "subnets": [
+            {
+              "name": "Subnet1",
+              "properties": {
+                "addressPrefix": "10.0.1.0/24"
+              }
+            }
+          ]
+        }
+      },
+      {
+        "type": "Microsoft.Network/virtualNetworks/subnets",
+        "apiVersion": "2020-06-01",
+        "name": "Subnet1",
+        "dependsOn": [
+          "Microsoft.Network/virtualNetworks/VNet1"
+        ],
+        "properties": {
+          "addressPrefix": "10.0.1.0/24"
+        }
+      }
+    ]
+  }