feat(general): enrich terraform definitions context key

checkov/common/util/parser_utils.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, List, Optional, Tuple
 
 import hcl2
 
+from checkov.common.runners.base_runner import strtobool
 from checkov.common.typing import TFDefinitionKeyType
 
 _FUNCTION_NAME_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
@@ -385,13 +387,15 @@ def get_tf_definition_key_from_module_dependency(
     return f"{path}{TERRAFORM_NESTED_MODULE_PATH_PREFIX}{module_dependency[:module_index]}{TERRAFORM_NESTED_MODULE_INDEX_SEPARATOR}{module_dependency_num}{module_dependency[module_index:]}{TERRAFORM_NESTED_MODULE_PATH_ENDING}"
 
 
-def get_module_from_full_path(file_path: TFDefinitionKeyType | None) -> Tuple[Optional[str], Optional[str]]:
+def get_module_from_full_path(file_path: TFDefinitionKeyType | None) -> Tuple[TFDefinitionKeyType | None, str | None]:
     from checkov.terraform.modules.module_objects import TFDefinitionKey
     if not file_path or not is_nested(file_path):
         return None, None
     if isinstance(file_path, TFDefinitionKey):
         if file_path.tf_source_modules is None:
             return None, None
+        if strtobool(os.getenv('ENABLE_DEFINITION_KEY', 'False')):
+            return TFDefinitionKey(file_path=file_path.tf_source_modules.path, tf_source_modules=file_path.tf_source_modules.nested_tf_module), None
         return file_path.tf_source_modules.path, None
     tmp_path = file_path[file_path.index(TERRAFORM_NESTED_MODULE_PATH_PREFIX) + TERRAFORM_NESTED_MODULE_PATH_SEPARATOR_LENGTH: -TERRAFORM_NESTED_MODULE_PATH_SEPARATOR_LENGTH]
     if is_nested(tmp_path):
@@ -403,5 +407,17 @@ def get_module_from_full_path(file_path: TFDefinitionKeyType | None) -> Tuple[Op
     return module, index
 
 
+def get_module_name(file_path: TFDefinitionKeyType) -> str | None:
+    from checkov.terraform.modules.module_objects import TFDefinitionKey
+    if isinstance(file_path, TFDefinitionKey):
+        if not file_path.tf_source_modules:
+            return None
+        module_name = file_path.tf_source_modules.name
+        if file_path.tf_source_modules.foreach_idx:
+            module_name = f'{module_name}[\"{file_path.tf_source_modules.foreach_idx}\"]'
+        return module_name
+    return None
+
+
 def get_abs_path(file_path: TFDefinitionKeyType) -> str:
     return file_path[:get_current_module_index(file_path)] if isinstance(file_path, str) else str(file_path.file_path)

checkov/terraform/context_parsers/base_parser.py

@@ -14,6 +14,7 @@
 from checkov.common.comment.enum import COMMENT_REGEX
 from checkov.common.models.enums import ContextCategories
 from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+from checkov.common.typing import TFDefinitionKeyType
 from checkov.common.util.parser_utils import get_abs_path
 from checkov.terraform.context_parsers.registry import parser_registry
 
@@ -154,7 +155,7 @@ def _compute_definition_end_line(self, start_line_num: int) -> int:
         return end_line_num
 
     def run(
-            self, tf_file: str, definition_blocks: List[Dict[str, Any]], collect_skip_comments: bool = True
+            self, tf_file: TFDefinitionKeyType, definition_blocks: List[Dict[str, Any]], collect_skip_comments: bool = True
     ) -> Dict[str, Any]:
         # TF files for loaded modules have this formation:  <file>[<referrer>#<index>]
         # Chop off everything after the file name for our purposes here

checkov/terraform/context_parsers/registry.py

@@ -1,9 +1,14 @@
+from __future__ import annotations
+
 import logging
+import os
 from typing import Dict, TYPE_CHECKING, Tuple, List, Any
 
 import dpath
 
 from checkov.common.resource_code_logger_filter import add_resource_code_filter_to_logger
+from checkov.common.runners.base_runner import strtobool
+from checkov.common.typing import TFDefinitionKeyType
 from checkov.terraform.modules.module_objects import TFDefinitionKey
 
 if TYPE_CHECKING:
@@ -12,7 +17,7 @@
 
 class ParserRegistry:
     context_parsers: Dict[str, "BaseContextParser"] = {}  # noqa: CCE003
-    definitions_context: Dict[str, Dict[str, Dict[str, Any]]] = {}  # noqa: CCE003
+    definitions_context: Dict[TFDefinitionKeyType, Dict[str, Dict[str, Any]]] = {}  # noqa: CCE003
 
     def __init__(self) -> None:
         self.logger = logging.getLogger(__name__)
@@ -26,24 +31,24 @@ def reset_definitions_context(self) -> None:
 
     def enrich_definitions_context(
         self, definitions: Tuple[str, Dict[str, List[Dict[str, Any]]]], collect_skip_comments: bool = True
-    ) -> Dict[str, Dict[str, Dict[str, Any]]]:
+    ) -> Dict[TFDefinitionKeyType, Dict[str, Dict[str, Any]]]:
         supported_definitions = [parser_type for parser_type in self.context_parsers.keys()]
         (tf_definition_key, definition_blocks_types) = definitions
+        enable_definition_key = strtobool(os.getenv('ENABLE_DEFINITION_KEY', 'False'))
         if isinstance(tf_definition_key, TFDefinitionKey):
-            tf_file = tf_definition_key.file_path
+            tf_file: TFDefinitionKeyType = tf_definition_key.file_path if not enable_definition_key else tf_definition_key
         else:
-            tf_file = tf_definition_key
+            tf_file = tf_definition_key if not enable_definition_key \
+                else TFDefinitionKey(file_path=tf_definition_key, tf_source_modules=None)
+
         if definition_blocks_types:
             definition_blocks_types = {x: definition_blocks_types[x] for x in definition_blocks_types.keys()}
             for definition_type in definition_blocks_types.keys():
                 if definition_type in supported_definitions:
                     dpath.new(self.definitions_context, [tf_file, definition_type], {})
                     context_parser = self.context_parsers[definition_type]
                     definition_blocks = definition_blocks_types[definition_type]
-                    self.definitions_context[tf_file][definition_type] = context_parser.run(
-                        tf_file, definition_blocks, collect_skip_comments
-                    )
-
+                    self.definitions_context[tf_file][definition_type] = context_parser.run(tf_file, definition_blocks, collect_skip_comments)
         return self.definitions_context
 
 

checkov/terraform/graph_builder/graph_components/blocks.py

@@ -6,14 +6,15 @@
 import re
 
 from checkov.common.runners.base_runner import strtobool
+from checkov.common.typing import TFDefinitionKeyType
 from checkov.terraform.graph_builder.utils import INTERPOLATION_EXPR
 from checkov.common.graph.graph_builder.graph_components.blocks import Block
 from checkov.common.util.consts import RESOLVED_MODULE_ENTRY_NAME
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
 from checkov.terraform.graph_builder.utils import remove_module_dependency_in_path
 
 if TYPE_CHECKING:
-    from checkov.terraform import TFModule, TFDefinitionKey
+    from checkov.terraform import TFModule
 
 
 class TerraformBlock(Block):
@@ -30,16 +31,16 @@ class TerraformBlock(Block):
     )
 
     def __init__(
-            self,
-            name: str,
-            config: Dict[str, Any],
-            path: str | TFDefinitionKey,
-            block_type: str,
-            attributes: Dict[str, Any],
-            id: str = "",
-            source: str = "",
-            has_dynamic_block: bool = False,
-            dynamic_attributes: dict[str, Any] | None = None,
+        self,
+        name: str,
+        config: Dict[str, Any],
+        path: TFDefinitionKeyType,
+        block_type: str,
+        attributes: Dict[str, Any],
+        id: str = "",
+        source: str = "",
+        has_dynamic_block: bool = False,
+        dynamic_attributes: dict[str, Any] | None = None,
     ) -> None:
         """
             when adding a new field be sure to add it to the equality function below
@@ -60,7 +61,7 @@ def __init__(
             has_dynamic_block=has_dynamic_block,
             dynamic_attributes=dynamic_attributes,
         )
-        self.module_dependency: str | None = ""
+        self.module_dependency: TFDefinitionKeyType | None = ""
         self.module_dependency_num: str | None = ""
         if path:
             if strtobool(os.getenv('CHECKOV_ENABLE_NESTED_MODULES', 'True')):

checkov/terraform/graph_builder/graph_components/module.py

@@ -11,10 +11,10 @@
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
 from checkov.terraform.graph_builder.graph_components.blocks import TerraformBlock
 from checkov.terraform.parser_functions import handle_dynamic_values
+from checkov.terraform.modules import TFDefinitionKey
 from hcl2 import START_LINE, END_LINE
 
 if TYPE_CHECKING:
-    from checkov.terraform.modules import TFDefinitionKey
     from typing_extensions import TypeAlias
 
 _AddBlockTypeCallable: TypeAlias = "Callable[[Module, list[dict[str, dict[str, Any]]], str | TFDefinitionKey], None]"

checkov/terraform/graph_builder/graph_to_tf_definitions.py

@@ -5,6 +5,7 @@
 
 from checkov.common.graph.graph_builder import CustomAttributes
 from checkov.common.runners.base_runner import strtobool
+from checkov.common.typing import TFDefinitionKeyType
 from checkov.common.util.parser_utils import get_tf_definition_key_from_module_dependency
 from checkov.terraform.modules.module_objects import TFDefinitionKey
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
@@ -13,9 +14,9 @@
 
 def convert_graph_vertices_to_tf_definitions(
     vertices: List[TerraformBlock], root_folder: str
-) -> Tuple[Dict[str | TFDefinitionKey, Dict[str, Any]], Dict[str, Dict[str, Any]]]:
+) -> Tuple[Dict[TFDefinitionKeyType, Dict[str, Any]], Dict[str, Dict[str, Any]]]:
     use_new_tf_parser = strtobool(os.getenv('CHECKOV_NEW_TF_PARSER', 'True'))
-    tf_definitions: Dict[str | TFDefinitionKey, Dict[str, Any]] = {}
+    tf_definitions: Dict[TFDefinitionKeyType, Dict[str, Any]] = {}
     breadcrumbs: Dict[str, Dict[str, Any]] = {}
     for vertex in vertices:
         block_path = vertex.path
@@ -26,7 +27,7 @@ def convert_graph_vertices_to_tf_definitions(
         if block_type == BlockType.TF_VARIABLE:
             continue
 
-        tf_path: str | TFDefinitionKey = TFDefinitionKey(file_path=block_path) if use_new_tf_parser else block_path
+        tf_path: TFDefinitionKeyType = TFDefinitionKey(file_path=block_path) if use_new_tf_parser else block_path
         if vertex.module_dependency or hasattr(vertex, "source_module_object") and vertex.source_module_object:
             if use_new_tf_parser:
                 tf_path = TFDefinitionKey(file_path=block_path, tf_source_modules=vertex.source_module_object)

checkov/terraform/modules/module_utils.py

@@ -226,12 +226,15 @@ def get_module_dependency_map_support_nested_modules(
 
 
 def get_nested_modules_data_as_list(file_path: str) -> tuple[list[tuple[str | None, str | None]], str]:
+    from checkov.terraform.modules.module_objects import TFDefinitionKey
     path = get_abs_path(file_path)
     module_path: str | None = file_path
     modules_list = []
 
     while is_nested(module_path):
         module, index = get_module_from_full_path(module_path)
+        if isinstance(module, TFDefinitionKey):
+            module = module.file_path
         modules_list.append((module, index))
         module_path = module
     modules_list.reverse()

checkov/terraform/plan_runner.py

@@ -9,7 +9,7 @@
 import pathlib
 
 from checkov.common.graph.checks_infra.registry import BaseRegistry
-from checkov.common.typing import LibraryGraphConnector
+from checkov.common.typing import LibraryGraphConnector, TFDefinitionKeyType
 from checkov.common.graph.graph_builder.consts import GraphSource
 from checkov.terraform.modules.module_objects import TFDefinitionKey
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
@@ -201,7 +201,7 @@ def check_tf_definition(self, report, root_folder, runner_filter, collect_skip_c
                                    block_type, runner_filter)
 
     @staticmethod
-    def _get_file_path(full_file_path: str | TFDefinitionKey, root_folder: str | pathlib.Path) -> tuple[str, str]:
+    def _get_file_path(full_file_path: TFDefinitionKeyType, root_folder: str | pathlib.Path) -> tuple[str, str]:
         if isinstance(full_file_path, TFDefinitionKey):
             full_file_path = full_file_path.file_path
         if platform.system() == "Windows":

checkov/terraform/runner.py

@@ -28,7 +28,7 @@
 from checkov.common.util.parser_utils import get_module_from_full_path, get_abs_path, \
     get_tf_definition_key_from_module_dependency, TERRAFORM_NESTED_MODULE_PATH_PREFIX, \
     TERRAFORM_NESTED_MODULE_PATH_ENDING, TERRAFORM_NESTED_MODULE_PATH_SEPARATOR_LENGTH, \
-    TERRAFORM_NESTED_MODULE_INDEX_SEPARATOR
+    TERRAFORM_NESTED_MODULE_INDEX_SEPARATOR, get_module_name
 from checkov.common.util.secrets import omit_secret_value_from_checks, omit_secret_value_from_graph_checks
 from checkov.common.variables.context import EvaluationContext
 from checkov.runner_filter import RunnerFilter
@@ -374,21 +374,23 @@ def run_block(self, entities,
 
             if self.enable_nested_modules:
                 entity_id = entity_config.get(CustomAttributes.TF_RESOURCE_ADDRESS)
-                module, _ = get_module_from_full_path(full_file_path)
-                if module:
-                    full_definition_path = entity_id.split('.')
-                    try:
-                        module_name_index = len(full_definition_path) - full_definition_path[::-1][1:].index(BlockType.MODULE) - 1  # the next item after the last 'module' prefix is the module name
-                    except ValueError as e:
-                        # TODO handle multiple modules with the same name in repo
-                        logging.warning(f'Failed to get module name for resource {entity_id}. {str(e)}')
-                        continue
-                    module_name = full_definition_path[module_name_index]
-                    caller_context = definition_context[module].get(BlockType.MODULE, {}).get(module_name)
+                module_full_path, _ = get_module_from_full_path(full_file_path)
+                if module_full_path:
+                    module_name = get_module_name(full_file_path)
+                    if not module_name:
+                        full_definition_path = entity_id.split('.')
+                        try:
+                            module_name_index = len(full_definition_path) - full_definition_path[::-1][1:].index(BlockType.MODULE) - 1  # the next item after the last 'module' prefix is the module name
+                        except ValueError as e:
+                            # TODO handle multiple modules with the same name in repo
+                            logging.warning(f'Failed to get module name for resource {entity_id}. {str(e)}')
+                            continue
+                        module_name = full_definition_path[module_name_index]
+                    caller_context = definition_context[module_full_path].get(BlockType.MODULE, {}).get(module_name)
                     if not caller_context:
                         continue
                     caller_file_line_range = [caller_context.get('start_line'), caller_context.get('end_line')]
-                    abs_caller_file = get_abs_path(module)
+                    abs_caller_file = get_abs_path(module_full_path)
                     caller_file_path = f"/{os.path.relpath(abs_caller_file, root_folder)}"
             elif module_referrer is not None:
                 referrer_id = self._find_id_for_referrer(full_file_path)
@@ -416,10 +418,10 @@ def run_block(self, entities,
             else:
                 entity_context_path = entity_context_path_header + block_type + definition_path
             # Entity can exist only once per dir, for file as well
-            if isinstance(full_file_path, TFDefinitionKey):
-                context_path = full_file_path.file_path
+            if not strtobool(os.getenv('ENABLE_DEFINITION_KEY', 'False')):
+                context_path = full_file_path.file_path if isinstance(full_file_path, TFDefinitionKey) else full_file_path
             else:
-                context_path = full_file_path
+                context_path = full_file_path if isinstance(full_file_path, TFDefinitionKey) else TFDefinitionKey(file_path=full_file_path, tf_source_modules=None)
             try:
                 entity_context = data_structures_utils.get_inner_dict(
                     definition_context[context_path],
@@ -569,10 +571,11 @@ def push_skipped_checks_down_old(definition_context, module_path, skipped_checks
     def push_skipped_checks_down_from_modules(self, definition_context):
         module_context_parser = parser_registry.context_parsers[BlockType.MODULE]
         for tf_definition_key, definition in self.definitions.items():
-            if isinstance(tf_definition_key, TFDefinitionKey):
-                full_file_path = tf_definition_key.file_path
+            if not strtobool(os.getenv('ENABLE_DEFINITION_KEY', 'False')):
+                full_file_path = tf_definition_key.file_path if isinstance(tf_definition_key, TFDefinitionKey) else tf_definition_key
             else:
-                full_file_path = tf_definition_key
+                full_file_path = tf_definition_key if isinstance(tf_definition_key, TFDefinitionKey)\
+                    else TFDefinitionKey(file_path=tf_definition_key, tf_source_modules=None)
             definition_modules_context = definition_context.get(full_file_path, {}).get(BlockType.MODULE, {})
             for entity in definition.get(BlockType.MODULE, []):
                 module_name = module_context_parser.get_entity_context_path(entity)[0]

checkov/terraform/tf_parser.py

@@ -9,6 +9,7 @@
 import deep_merge
 
 from checkov.common.runners.base_runner import filter_ignored_paths, IGNORE_HIDDEN_DIRECTORY_ENV
+from checkov.common.typing import TFDefinitionKeyType
 from checkov.common.util.consts import DEFAULT_EXTERNAL_MODULES_DIR, RESOLVED_MODULE_ENTRY_NAME
 from checkov.common.util.data_structures_utils import pickle_deepcopy
 from checkov.common.util.type_forcers import force_list
@@ -508,7 +509,7 @@ def add_tfvars_with_source_dir(self, module: Module, source: str, source_dir: st
                 block = [{var_name: {"default": default}}]
                 module.add_blocks(BlockType.TF_VARIABLE, block, path, source)
 
-    def get_dirname(self, path: str | TFDefinitionKey) -> str:
+    def get_dirname(self, path: TFDefinitionKeyType) -> str:
         if isinstance(path, TFDefinitionKey):
             path = path.file_path
         dirname_path = self.dirname_cache.get(path)
@@ -521,7 +522,7 @@ def should_loaded_file(self, file: TFDefinitionKey, root_dir: str) -> bool:
         return not self.get_dirname(file) != root_dir
 
     def get_module_source(
-        self, module_call_data: dict[str, Any], module_call_name: str, file: str | TFDefinitionKey
+        self, module_call_data: dict[str, Any], module_call_name: str, file: TFDefinitionKeyType
     ) -> Optional[str]:
         source = module_call_data.get("source")
         if not source or not isinstance(source, list):