fix(cloudformation): render variables in cfn vertices config (#7423)

* render cfn vertices config varaibles

* mypy

* export common function

* export function

* lint

* move unit tests

* rename file

Author: omriyoffe-panw

checkov/cloudformation/graph_builder/local_graph.py

@@ -16,8 +16,11 @@
 from checkov.common.graph.graph_builder import Edge
 from checkov.common.graph.graph_builder.local_graph import LocalGraph
 from checkov.common.util.consts import START_LINE, END_LINE
-from checkov.common.util.data_structures_utils import search_deep_keys
+from checkov.common.util.data_structures_utils import search_deep_keys, pickle_deepcopy
 from checkov.cloudformation.graph_builder.graph_components.generic_resource_encryption import ENCRYPTION_BY_RESOURCE_TYPE
+from checkov.common.graph.graph_builder.utils import filter_sub_keys
+from checkov.terraform.graph_builder.local_graph import update_dictionary_attribute
+
 
 if TYPE_CHECKING:
     from checkov.common.graph.graph_builder.graph_components.blocks import Block
@@ -56,6 +59,7 @@ def build_graph(self, render_variables: bool) -> None:
             logging.info(f"Rendering variables, graph has {len(self.vertices)} vertices and {len(self.edges)} edges")
             renderer = CloudformationVariableRenderer(self)
             renderer.render_variables_from_local_graph()
+            self.update_vertices_configs()
             self.update_vertices_breadcrumbs()
         self.calculate_encryption_attribute(ENCRYPTION_BY_RESOURCE_TYPE)
 
@@ -391,15 +395,37 @@ def _is_of_type(cfndict: dict[str, Any], identifier: Any, *template_sections: Te
         return False
 
     def update_vertices_configs(self) -> None:
-        # not used
-        pass
+        for vertex in self.vertices:
+            changed_attributes = list(vertex.changed_attributes.keys())
+            if changed_attributes:
+                self.update_vertex_config(vertex, changed_attributes)
 
     @staticmethod
     def update_vertex_config(
         vertex: Block, changed_attributes: list[str] | dict[str, Any], dynamic_blocks: bool = False
     ) -> None:
-        # not used
-        pass
+        if not changed_attributes:
+            return
+
+        if not isinstance(vertex.config, dict):
+            return
+
+        updated_config = pickle_deepcopy(vertex.config)
+        if isinstance(changed_attributes, dict):
+            attributes_to_update = list(changed_attributes.keys())
+        else:
+            attributes_to_update = changed_attributes
+
+        attributes_to_update = filter_sub_keys(attributes_to_update)
+
+        for attribute in attributes_to_update:
+            if attribute not in vertex.attributes:
+                continue
+
+            new_value = vertex.attributes[attribute]
+            update_dictionary_attribute(updated_config, attribute, new_value)
+
+        vertex.config = updated_config
 
 
 def get_only_dict_items(origin_dict: Union[Dict[str, Any], Any]) -> Dict[str, Dict[str, Any]]:

checkov/common/graph/graph_builder/utils.py

@@ -2,9 +2,12 @@
 
 import concurrent
 import hashlib
-from typing import Any, Callable
+from typing import Any, Callable, overload, Union, List, Dict
 import concurrent.futures
 
+from checkov.common.util.data_structures_utils import pickle_deepcopy
+from checkov.common.util.type_forcers import force_int
+
 
 def calculate_hash(data: Any) -> str:
     sha256 = hashlib.sha256(str(data).encode("utf-8"))
@@ -67,3 +70,127 @@ def adjust_value(element_name: str, value: Any) -> Any:
         return adjust_value(".".join(key_parts[1:]), new_value)
 
     return value
+
+
+def to_list(data: Any) -> list[Any] | dict[str, Any]:
+    if isinstance(data, list) and len(data) == 1 and (isinstance(data[0], str) or isinstance(data[0], int)):
+        return data
+    elif isinstance(data, list):
+        return [to_list(x) for x in data]
+    elif isinstance(data, dict):
+        return {key: to_list(val) for key, val in data.items()}
+    else:
+        return [data]
+
+
+@overload
+def update_dictionary_attribute(
+        config: dict[str, Any], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
+) -> dict[str, Any]:
+    ...
+
+
+@overload
+def update_dictionary_attribute(
+        config: list[Any], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
+) -> list[Any]:
+    ...
+
+
+def update_dictionary_attribute(
+    config: Union[List[Any], Dict[str, Any]], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
+) -> Union[List[Any], Dict[str, Any]]:
+    key_parts = key_to_update.split(".")
+    if '"' in key_to_update:
+        key_parts = join_double_quote_surrounded_dot_split(str_parts=key_parts)
+
+    if isinstance(config, dict) and isinstance(key_parts, list):
+        key = key_parts[0]
+        inner_config = config.get(key)
+
+        if inner_config is not None:
+            if len(key_parts) == 1:
+                if isinstance(inner_config, list) and not isinstance(new_value, list):
+                    new_value = [new_value]
+                config[key] = to_list(new_value) if dynamic_blocks else new_value
+                return config
+            else:
+                config[key] = update_dictionary_attribute(
+                    inner_config, ".".join(key_parts[1:]), new_value, dynamic_blocks=dynamic_blocks
+                )
+        else:
+            for key in config:
+                config[key] = update_dictionary_attribute(
+                    config[key], key_to_update, new_value, dynamic_blocks=dynamic_blocks
+                )
+    if isinstance(config, list):
+        return update_list_attribute(
+            config=config,
+            key_parts=key_parts,
+            key_to_update=key_to_update,
+            new_value=new_value,
+            dynamic_blocks=dynamic_blocks,
+        )
+    return config
+
+
+def update_list_attribute(
+    config: list[Any], key_parts: list[str], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
+) -> list[Any] | dict[str, Any]:
+    """Updates a list attribute in the given config"""
+
+    if not config:
+        # happens when we can't correctly evaluate something, because of strange defaults or 'for_each' blocks
+        return config
+
+    if len(key_parts) == 1 and len(config) == 1:
+        idx = force_int(key_parts[0])
+        # Avoid changing the config and cause side effects
+        inner_config = pickle_deepcopy(config[0])
+
+        if idx is not None and isinstance(inner_config, list):
+            if not inner_config:
+                # happens when config = [[]]
+                return config
+
+            inner_config[idx] = new_value
+            return [inner_config]
+    entry_to_update = int(key_parts[0]) if key_parts[0].isnumeric() else -1
+    for i, config_value in enumerate(config):
+        if entry_to_update == -1:
+            config[i] = update_dictionary_attribute(config=config_value, key_to_update=key_to_update, new_value=new_value, dynamic_blocks=dynamic_blocks)
+        elif entry_to_update == i:
+            config[i] = update_dictionary_attribute(config=config_value, key_to_update=".".join(key_parts[1:]), new_value=new_value, dynamic_blocks=dynamic_blocks)
+
+    return config
+
+
+def join_double_quote_surrounded_dot_split(str_parts: list[str]) -> list[str]:
+    """Joins back split strings which enclosed a dot by double quotes
+
+    ex.
+
+    ['google_project_iam_binding', 'role["roles/logging', 'admin"]'] -> ['google_project_iam_binding', 'role["roles/logging.admin"]']
+
+    If someone finds a better solution feel free to replace it!
+    """
+
+    new_str_parts = []
+    joined_str_parts: list[str] = []
+    for part in str_parts:
+        if not joined_str_parts:
+            if '"' not in part:
+                new_str_parts.append(part)
+            elif part.count('"') >= 2:
+                new_str_parts.append(part)
+            else:
+                joined_str_parts.append(part)
+            continue
+
+        joined_str_parts.append(part)
+
+        if '"' in part:
+            new_str_parts.append(".".join(joined_str_parts))
+            joined_str_parts = []
+
+    return new_str_parts

checkov/terraform/graph_builder/local_graph.py

@@ -6,18 +6,18 @@
 from collections import defaultdict
 from functools import partial
 from pathlib import Path
-from typing import List, Optional, Union, Any, Dict, overload, TypedDict, cast
+from typing import List, Optional, Union, Any, Dict, TypedDict, cast
 
 import checkov.terraform.graph_builder.foreach.consts
 from checkov.common.graph.graph_builder import Edge
 from checkov.common.graph.graph_builder import reserved_attribute_names
 from checkov.common.graph.graph_builder.graph_components.attribute_names import CustomAttributes
 from checkov.common.graph.graph_builder.local_graph import LocalGraph
-from checkov.common.graph.graph_builder.utils import calculate_hash, join_trimmed_strings, filter_sub_keys
+from checkov.common.graph.graph_builder.utils import calculate_hash, join_trimmed_strings, filter_sub_keys, \
+    update_dictionary_attribute
 from checkov.common.runners.base_runner import strtobool
 from checkov.common.typing import TFDefinitionKeyType
 from checkov.common.util.data_structures_utils import pickle_deepcopy
-from checkov.common.util.type_forcers import force_int
 from checkov.terraform.graph_builder.foreach.builder import ForeachBuilder
 from checkov.terraform.graph_builder.foreach.consts import VIRTUAL_RESOURCE
 from checkov.terraform.graph_builder.variable_rendering.vertex_reference import TerraformVertexReference
@@ -31,8 +31,7 @@
     get_attribute_is_leaf,
     get_referenced_vertices_in_value,
     attribute_has_nested_attributes,
-    remove_index_pattern_from_str,
-    join_double_quote_surrounded_dot_split, )
+    remove_index_pattern_from_str, )
 from checkov.terraform.graph_builder.foreach.utils import get_terraform_foreach_or_count_key, \
     get_sanitized_terraform_resource_id
 from checkov.terraform.graph_builder.utils import is_local_path
@@ -857,99 +856,6 @@ def _build_virtual_resources_edges(self, origin_node_index: int, vertex: Terrafo
                     self.create_edge(i, origin_node_index, VIRTUAL_RESOURCE)
 
 
-def to_list(data: Any) -> list[Any] | dict[str, Any]:
-    if isinstance(data, list) and len(data) == 1 and (isinstance(data[0], str) or isinstance(data[0], int)):
-        return data
-    elif isinstance(data, list):
-        return [to_list(x) for x in data]
-    elif isinstance(data, dict):
-        return {key: to_list(val) for key, val in data.items()}
-    else:
-        return [data]
-
-
-@overload
-def update_dictionary_attribute(
-        config: dict[str, Any], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
-) -> dict[str, Any]:
-    ...
-
-
-@overload
-def update_dictionary_attribute(
-        config: list[Any], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
-) -> list[Any]:
-    ...
-
-
-def update_dictionary_attribute(
-    config: Union[List[Any], Dict[str, Any]], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
-) -> Union[List[Any], Dict[str, Any]]:
-    key_parts = key_to_update.split(".")
-    if '"' in key_to_update:
-        key_parts = join_double_quote_surrounded_dot_split(str_parts=key_parts)
-
-    if isinstance(config, dict) and isinstance(key_parts, list):
-        key = key_parts[0]
-        inner_config = config.get(key)
-
-        if inner_config is not None:
-            if len(key_parts) == 1:
-                if isinstance(inner_config, list) and not isinstance(new_value, list):
-                    new_value = [new_value]
-                config[key] = to_list(new_value) if dynamic_blocks else new_value
-                return config
-            else:
-                config[key] = update_dictionary_attribute(
-                    inner_config, ".".join(key_parts[1:]), new_value, dynamic_blocks=dynamic_blocks
-                )
-        else:
-            for key in config:
-                config[key] = update_dictionary_attribute(
-                    config[key], key_to_update, new_value, dynamic_blocks=dynamic_blocks
-                )
-    if isinstance(config, list):
-        return update_list_attribute(
-            config=config,
-            key_parts=key_parts,
-            key_to_update=key_to_update,
-            new_value=new_value,
-            dynamic_blocks=dynamic_blocks,
-        )
-    return config
-
-
-def update_list_attribute(
-    config: list[Any], key_parts: list[str], key_to_update: str, new_value: Any, dynamic_blocks: bool = False
-) -> list[Any] | dict[str, Any]:
-    """Updates a list attribute in the given config"""
-
-    if not config:
-        # happens when we can't correctly evaluate something, because of strange defaults or 'for_each' blocks
-        return config
-
-    if len(key_parts) == 1 and len(config) == 1:
-        idx = force_int(key_parts[0])
-        # Avoid changing the config and cause side effects
-        inner_config = pickle_deepcopy(config[0])
-
-        if idx is not None and isinstance(inner_config, list):
-            if not inner_config:
-                # happens when config = [[]]
-                return config
-
-            inner_config[idx] = new_value
-            return [inner_config]
-    entry_to_update = int(key_parts[0]) if key_parts[0].isnumeric() else -1
-    for i, config_value in enumerate(config):
-        if entry_to_update == -1:
-            config[i] = update_dictionary_attribute(config=config_value, key_to_update=key_to_update, new_value=new_value, dynamic_blocks=dynamic_blocks)
-        elif entry_to_update == i:
-            config[i] = update_dictionary_attribute(config=config_value, key_to_update=".".join(key_parts[1:]), new_value=new_value, dynamic_blocks=dynamic_blocks)
-
-    return config
-
-
 def get_vertex_as_tf_module(block: TerraformBlock) -> TFModule:
     block_name = get_sanitized_terraform_resource_id(block.name)
     return TFModule(path=block.path, name=block_name, nested_tf_module=block.source_module_object, foreach_idx=block.for_each_index)

checkov/terraform/graph_builder/utils.py

@@ -382,34 +382,3 @@ def get_attribute_is_leaf(vertex: TerraformBlock) -> Dict[str, bool]:
         if other in attribute_is_leaf:
             attribute_is_leaf[other] = False
     return attribute_is_leaf
-
-
-def join_double_quote_surrounded_dot_split(str_parts: list[str]) -> list[str]:
-    """Joins back split strings which enclosed a dot by double quotes
-
-    ex.
-
-    ['google_project_iam_binding', 'role["roles/logging', 'admin"]'] -> ['google_project_iam_binding', 'role["roles/logging.admin"]']
-
-    If someone finds a better solution feel free to replace it!
-    """
-
-    new_str_parts = []
-    joined_str_parts: list[str] = []
-    for part in str_parts:
-        if not joined_str_parts:
-            if '"' not in part:
-                new_str_parts.append(part)
-            elif part.count('"') >= 2:
-                new_str_parts.append(part)
-            else:
-                joined_str_parts.append(part)
-            continue
-
-        joined_str_parts.append(part)
-
-        if '"' in part:
-            new_str_parts.append(".".join(joined_str_parts))
-            joined_str_parts = []
-
-    return new_str_parts