fix(terraform_plan): add terraform plan vertices to terraform graph if not exist

checkov/terraform/deep_analysis_plan_graph_manager.py

@@ -5,45 +5,70 @@
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
 from checkov.common.output.report import Report
 from checkov.terraform.plan_parser import TF_PLAN_RESOURCE_ADDRESS
-from typing import Dict
+from typing import Dict, Tuple
 
 
 class DeepAnalysisGraphManager:
     def __init__(self, tf_graph: TerraformLocalGraph, tf_plan_graph: TerraformLocalGraph) -> None:
         self.tf_graph: TerraformLocalGraph = tf_graph
         self.tf_plan_graph: TerraformLocalGraph = tf_plan_graph
-        self._address_to_tf_vertex_map: Dict[str, TerraformBlock] = {}
-        self._address_to_tf_plan_vertex_map: Dict[str, TerraformBlock] = {}
+        self._address_to_tf_idx_and_vertex_map: Dict[str, Tuple[TerraformBlock, int]] = {}
+        self._address_to_tf_plan_idx_and_vertex_map: Dict[str, Tuple[TerraformBlock, int]] = {}
         self._apply_address_mapping()
 
     def _apply_address_mapping(self) -> None:
-        self._address_to_tf_vertex_map = {
-            vertex.attributes[TF_PLAN_RESOURCE_ADDRESS]: vertex
-            for vertex in self.tf_graph.vertices
+        self._address_to_tf_idx_and_vertex_map = {
+            vertex.attributes[TF_PLAN_RESOURCE_ADDRESS]: (i, vertex)
+            for i, vertex in enumerate(self.tf_graph.vertices)
             if vertex.block_type == BlockType.RESOURCE
         }
-        self._address_to_tf_plan_vertex_map = {
-            vertex.attributes[TF_PLAN_RESOURCE_ADDRESS]: vertex
-            for vertex in self.tf_plan_graph.vertices
+        self._address_to_tf_plan_idx_and_vertex_map = {
+            vertex.attributes[TF_PLAN_RESOURCE_ADDRESS]: (i, vertex)
+            for i, vertex in enumerate(self.tf_plan_graph.vertices)
             if vertex.block_type == BlockType.RESOURCE
         }
 
+    def append_vertex_to_terraform_graph(self, tf_plan_vertex, tf_plan_vertex_index, address):
+        new_vertex_idx = len(self.tf_graph.vertices)
+        self.tf_graph.vertices.append(tf_plan_vertex)
+        self._address_to_tf_idx_and_vertex_map[address] = (new_vertex_idx, tf_plan_vertex)
+
+        def __get_tf_vertex_idx_from_tf_plan_vertex(v):
+            v = self._address_to_tf_idx_and_vertex_map.get(v.attributes.get('__address__'))
+            if not v:
+                return None
+            return v[1]
+
+        for edge in self.tf_plan_graph.out_edges[tf_plan_vertex_index]:
+            dest = self.tf_plan_graph.vertices[edge.dest]
+            dest_index = __get_tf_vertex_idx_from_tf_plan_vertex(dest)
+            if dest_index:
+                self.tf_graph.create_edge(new_vertex_idx, dest_index, edge.label)
+        for edge in self.tf_plan_graph.in_edges[tf_plan_vertex_index]:
+            origin = self.tf_plan_graph.vertices[edge.origin]
+            origin_index = __get_tf_vertex_idx_from_tf_plan_vertex(origin)
+            if origin_index:
+                self.tf_graph.create_edge(origin_index, new_vertex_idx, edge.label)
+
     def enrich_tf_graph_attributes(self) -> None:
-        for address, tf_plan_vertex in self._address_to_tf_plan_vertex_map.items():
-            tf_vertex = self._address_to_tf_vertex_map.get(address)
-            if not tf_vertex:
-                logging.info(f'Cant find this address: {address} in tf graph')
+        for address, tf_plan_idx_and_vertex in self._address_to_tf_plan_idx_and_vertex_map.items():
+            tf_plan_vertex_index, tf_plan_vertex = tf_plan_idx_and_vertex
+            tf_idx_and_vertex = self._address_to_tf_idx_and_vertex_map.get(address)
+            if not tf_idx_and_vertex:
+                logging.info(f'Cant find this address: {address} in tf graph, adding it')
+                self.append_vertex_to_terraform_graph(tf_plan_vertex, tf_plan_vertex_index, address)
                 continue
+            _, tf_vertex = tf_idx_and_vertex
             tf_vertex.attributes = {**tf_vertex.attributes, **tf_plan_vertex.attributes}
             tf_vertex.path = tf_plan_vertex.path
 
     def filter_report(self, report: Report) -> None:
         report.failed_checks = [check for check in report.failed_checks if
-                                check.resource_address in self._address_to_tf_plan_vertex_map]
+                                check.resource_address in self._address_to_tf_plan_idx_and_vertex_map]
         report.passed_checks = [check for check in report.passed_checks if
-                                check.resource_address in self._address_to_tf_plan_vertex_map]
+                                check.resource_address in self._address_to_tf_plan_idx_and_vertex_map]
         report.skipped_checks = [check for check in report.skipped_checks if
-                                 check.resource_address in self._address_to_tf_plan_vertex_map]
+                                 check.resource_address in self._address_to_tf_plan_idx_and_vertex_map]
         # No need to filter other fields for now
         report.resources = set()
         report.extra_resources = set()

checkov/terraform/graph_builder/local_graph.py

@@ -285,7 +285,7 @@ def _build_edges_for_vertex(self, origin_node_index: int, vertex: TerraformBlock
                     continue
                 target_variable = next((v for v in target_variables if self.vertices[v].name == attribute), None)
                 if target_variable is not None:
-                    self._create_edge(target_variable, origin_node_index, "default", cross_variable_edges)
+                    self.create_edge(target_variable, origin_node_index, "default", cross_variable_edges)
         elif vertex.block_type == BlockType.TF_VARIABLE:
             # Assuming the tfvars file is in the same directory as the variables file (best practice)
             target_variable = 0
@@ -294,7 +294,7 @@ def _build_edges_for_vertex(self, origin_node_index: int, vertex: TerraformBlock
                     target_variable = index
                     break
             if target_variable:
-                self._create_edge(target_variable, origin_node_index, "default", cross_variable_edges)
+                self.create_edge(target_variable, origin_node_index, "default", cross_variable_edges)
 
     def _create_edge_from_reference(self, attribute_key: Any, origin_node_index: int, dest_node_index: int,
                                     sub_values: List[Any], vertex_reference: TerraformVertexReference,
@@ -312,8 +312,8 @@ def _create_edge_from_reference(self, attribute_key: Any, origin_node_index: int
                         f"Module {self.vertices[dest_node_index]} does not have source attribute, skipping"
                     )
             else:
-                self._create_edge(origin_node_index, dest_node_index, attribute_key,
-                                  cross_variable_edges)
+                self.create_edge(origin_node_index, dest_node_index, attribute_key,
+                                 cross_variable_edges)
 
     def _get_target_variables(self, vertex: TerraformBlock, dest_module_path: str) -> list[int]:
         if self.use_new_tf_parser:
@@ -343,8 +343,8 @@ def _build_cross_variable_edges(self):
                 modules = vertex.breadcrumbs.get(CustomAttributes.SOURCE_MODULE, [])
                 self._build_edges_for_vertex(origin_node_index, vertex, aliases, resources_types, True, modules)
 
-    def _create_edge(self, origin_vertex_index: int, dest_vertex_index: int, label: str,
-                     cross_variable_edges: bool = False) -> bool:
+    def create_edge(self, origin_vertex_index: int, dest_vertex_index: int, label: str,
+                    cross_variable_edges: bool = False) -> bool:
         if origin_vertex_index == dest_vertex_index:
             return False
         edge = Edge(origin_vertex_index, dest_vertex_index, label)
@@ -392,7 +392,7 @@ def _connect_module(
             for vertex_index in output_blocks_with_name:
                 vertex = self.vertices[vertex_index]
                 if self._should_add_edge(vertex, dest_module_path, module_node):
-                    added_edge = self._create_edge(origin_node_index, vertex_index, attribute_key, cross_variable_edges)
+                    added_edge = self.create_edge(origin_node_index, vertex_index, attribute_key, cross_variable_edges)
                     if added_edge:
                         self.vertices[origin_node_index].add_module_connection(attribute_key, vertex_index)
                     break

checkov/terraform/plan_runner.py

@@ -128,7 +128,7 @@ def run(
                             report.add_resource(f'{vertex.path}:{resource_id}')
                 self.graph_manager.save_graph(self.tf_plan_local_graph)
                 if self._should_run_deep_analysis:
-                    tf_local_graph = self._create_terraform_graph()
+                    tf_local_graph = self._create_terraform_graph(runner_filter)
 
         if external_checks_dir:
             for directory in external_checks_dir:
@@ -181,11 +181,12 @@ def _get_graph_report(
             return graph_report
         return self.get_graph_checks_report(root_folder, runner_filter)
 
-    def _create_terraform_graph(self) -> TerraformLocalGraph:
+    def _create_terraform_graph(self, runner_filter) -> TerraformLocalGraph:
         graph_manager = TerraformGraphManager(db_connector=self.db_connector)
         tf_local_graph, _ = graph_manager.build_graph_from_source_directory(
             self.repo_root_for_plan_enrichment,
-            render_variables=True
+            render_variables=True,
+            download_external_modules=runner_filter.download_external_modules
         )
         self.graph_manager = graph_manager
         return tf_local_graph

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/source/main.tf

@@ -0,0 +1,14 @@
+provider "aws" {
+  region  = "us-west-2"
+  profile = "dev8"
+}
+
+module "s3_module" {
+  source = "./module"
+
+  bucket = aws_s3_bucket.example.id
+}
+
+resource "aws_s3_bucket" "example" {
+  bucket = "example"
+}

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/source/module/main.tf

@@ -0,0 +1,4 @@
+module "inner_s3_module" {
+  source = "../module2"
+  bucket2 = var.bucket
+}

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/source/module/variable.tf

@@ -0,0 +1,3 @@
+variable "bucket" {
+  type = string
+}

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/source/module2/main.tf

@@ -0,0 +1,11 @@
+locals {
+  bucket2 = var.bucket2
+}
+
+resource "aws_s3_bucket_public_access_block" "var_bucket" {
+  bucket                  = local.bucket2
+  block_public_acls       = true
+  block_public_policy     = true
+  ignore_public_acls      = true
+  restrict_public_buckets = true
+}

tests/terraform/runner/resources/plan_and_tf_combine_graph_with_missing_resources/source/module2/variable.tf

@@ -0,0 +1,3 @@
+variable "bucket2" {
+  type = string
+}