break(graph): remove CHECKOV_CREATE_GRAPH env var to control graph cr…

…eation (#5606)

* remove CHECKOV_CREATE_GRAPH env var

* fix linting

* fix typing

checkov/arm/runner.py

@@ -18,7 +18,7 @@
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.runners.base_runner import BaseRunner, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner
 from checkov.common.util.consts import START_LINE, END_LINE
 from checkov.common.util.secrets import omit_secret_value_from_checks
 from checkov.runner_filter import RunnerFilter
@@ -80,7 +80,7 @@ def run(
             for directory in external_checks_dir:
                 arm_resource_registry.load_external_checks(directory)
 
-                if CHECKOV_CREATE_GRAPH and self.graph_registry:
+                if self.graph_registry:
                     self.graph_registry.load_external_checks(directory)
 
         if files:
@@ -96,7 +96,7 @@ def run(
 
         report.add_parsing_errors(parsing_errors)
 
-        if CHECKOV_CREATE_GRAPH and self.graph_registry and self.graph_manager:
+        if self.graph_registry and self.graph_manager:
             logging.info("Creating ARM graph")
             local_graph = self.graph_manager.build_graph_from_definitions(definitions=self.definitions)
             logging.info("Successfully created ARM graph")
@@ -109,7 +109,7 @@ def run(
         self.add_python_check_results(report=report, runner_filter=runner_filter, root_folder=root_folder)
 
         # run graph checks
-        if CHECKOV_CREATE_GRAPH and self.graph_registry:
+        if self.graph_registry:
             self.add_graph_check_results(report=report, runner_filter=runner_filter)
 
         return report

checkov/bicep/runner.py

@@ -27,7 +27,7 @@
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.runners.base_runner import BaseRunner, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner
 from checkov.common.typing import _CheckResult
 from checkov.common.util.secrets import omit_secret_value_from_checks
 from checkov.common.util.suppression import collect_suppressions_for_report
@@ -105,30 +105,26 @@ def run(
             if external_checks_dir:
                 for directory in external_checks_dir:
                     resource_registry.load_external_checks(directory)
-
-                    if CHECKOV_CREATE_GRAPH:
-                        self.graph_registry.load_external_checks(directory)
+                    self.graph_registry.load_external_checks(directory)
 
             self.context = build_definitions_context(definitions=self.definitions, definitions_raw=self.definitions_raw)
 
-            if CHECKOV_CREATE_GRAPH:
-                logging.info("Creating Bicep graph")
-                local_graph = self.graph_manager.build_graph_from_definitions(self.definitions)
-                logging.info("Successfully created Bicep graph")
+            logging.info("Creating Bicep graph")
+            local_graph = self.graph_manager.build_graph_from_definitions(self.definitions)
+            logging.info("Successfully created Bicep graph")
 
-                self.graph_manager.save_graph(local_graph)
-                self.definitions, self.breadcrumbs = convert_graph_vertices_to_tf_definitions(
-                    vertices=local_graph.vertices, root_folder=root_folder
-                )
+            self.graph_manager.save_graph(local_graph)
+            self.definitions, self.breadcrumbs = convert_graph_vertices_to_tf_definitions(
+                vertices=local_graph.vertices, root_folder=root_folder
+            )
 
         self.pbar.initiate(len(self.definitions))
 
         # run Python checks
         self.add_python_check_results(report=report, runner_filter=runner_filter, root_folder=root_folder)
 
         # run graph checks
-        if CHECKOV_CREATE_GRAPH:
-            self.add_graph_check_results(report=report, runner_filter=runner_filter)
+        self.add_graph_check_results(report=report, runner_filter=runner_filter)
 
         if runner_filter.run_image_referencer:
             if files:

checkov/cloudformation/runner.py

@@ -28,7 +28,7 @@
 from checkov.common.output.graph_record import GraphRecord
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report, merge_reports, CheckType
-from checkov.common.runners.base_runner import BaseRunner, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner
 from checkov.common.util.secrets import omit_secret_value_from_checks
 from checkov.runner_filter import RunnerFilter
 
@@ -89,25 +89,22 @@ def run(
             if external_checks_dir:
                 for directory in external_checks_dir:
                     cfn_registry.load_external_checks(directory)
-
-                    if CHECKOV_CREATE_GRAPH:
-                        self.graph_registry.load_external_checks(directory)
+                    self.graph_registry.load_external_checks(directory)
 
             self.context = build_definitions_context(self.definitions, self.definitions_raw)
 
-            if CHECKOV_CREATE_GRAPH:
-                logging.info("creating CloudFormation graph")
-                local_graph = self.graph_manager.build_graph_from_definitions(self.definitions)
-                logging.info("Successfully created CloudFormation graph")
-
-                for vertex in local_graph.vertices:
-                    if vertex.block_type == BlockType.RESOURCE:
-                        report.add_resource(f'{vertex.path}:{vertex.id}')
-                self.graph_manager.save_graph(local_graph)
-                self.definitions, self.breadcrumbs = convert_graph_vertices_to_definitions(
-                    vertices=local_graph.vertices,
-                    root_folder=root_folder,
-                )
+            logging.info("creating CloudFormation graph")
+            local_graph = self.graph_manager.build_graph_from_definitions(self.definitions)
+            logging.info("Successfully created CloudFormation graph")
+
+            for vertex in local_graph.vertices:
+                if vertex.block_type == BlockType.RESOURCE:
+                    report.add_resource(f'{vertex.path}:{vertex.id}')
+            self.graph_manager.save_graph(local_graph)
+            self.definitions, self.breadcrumbs = convert_graph_vertices_to_definitions(
+                vertices=local_graph.vertices,
+                root_folder=root_folder,
+            )
 
         # TODO: replace with real graph rendering
         for cf_file in self.definitions.keys():
@@ -126,9 +123,8 @@ def run(
         self.check_definitions(root_folder, runner_filter, report)
 
         # run graph checks
-        if CHECKOV_CREATE_GRAPH:
-            graph_report = self.get_graph_checks_report(root_folder, runner_filter)
-            merge_reports(report, graph_report)
+        graph_report = self.get_graph_checks_report(root_folder, runner_filter)
+        merge_reports(report, graph_report)
 
         if runner_filter.run_image_referencer:
             if files:
@@ -191,7 +187,7 @@ def check_definitions(self, root_folder: str | None, runner_filter: RunnerFilter
                                         severity=check.severity
                                     )
 
-                                    if CHECKOV_CREATE_GRAPH and self.breadcrumbs:
+                                    if self.breadcrumbs:
                                         breadcrumb = self.breadcrumbs.get(record.file_path, {}).get(record.resource)
                                         if breadcrumb:
                                             record = GraphRecord(record, breadcrumb)

checkov/common/runners/base_runner.py

@@ -14,7 +14,6 @@
 
 from checkov.common.graph.checks_infra.base_check import BaseGraphCheck
 from checkov.common.output.report import Report
-from checkov.common.util.type_forcers import convert_str_to_bool
 from checkov.runner_filter import RunnerFilter
 from checkov.common.graph.graph_manager import GraphManager  # noqa
 
@@ -45,7 +44,6 @@ def strtobool(val: str) -> int:
         raise ValueError("invalid boolean value %r for environment variable CKV_IGNORE_HIDDEN_DIRECTORIES" % (val,))
 
 
-CHECKOV_CREATE_GRAPH = convert_str_to_bool(os.getenv("CHECKOV_CREATE_GRAPH", "True"))
 IGNORED_DIRECTORIES_ENV = os.getenv("CKV_IGNORED_DIRECTORIES", "node_modules,.terraform,.serverless")
 IGNORE_HIDDEN_DIRECTORY_ENV = strtobool(os.getenv("CKV_IGNORE_HIDDEN_DIRECTORIES", "True"))
 

checkov/common/runners/object_runner.py

@@ -18,7 +18,7 @@
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report, CheckType
 from checkov.common.parallelizer.parallel_runner import parallel_runner
-from checkov.common.runners.base_runner import BaseRunner, filter_ignored_paths, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner, filter_ignored_paths
 from checkov.common.runners.graph_manager import ObjectGraphManager
 from checkov.common.typing import _CheckResult
 from checkov.common.util.consts import START_LINE, END_LINE
@@ -121,7 +121,7 @@ def run(
             for directory in external_checks_dir:
                 registry.load_external_checks(directory)
 
-                if CHECKOV_CREATE_GRAPH and self.graph_registry:
+                if self.graph_registry:
                     self.graph_registry.load_external_checks(directory)
 
         if not self.context or not self.definitions:
@@ -139,7 +139,7 @@ def run(
 
             self.context = self.build_definitions_context(definitions=self.definitions, definitions_raw=self.definitions_raw)
 
-            if CHECKOV_CREATE_GRAPH and self.graph_registry and self.graph_manager:
+            if self.graph_registry and self.graph_manager:
                 logging.info(f"Creating {self.source} graph")
                 local_graph = self.graph_manager.build_graph_from_definitions(
                     definitions=self.definitions, graph_class=self.graph_class  # type:ignore[arg-type]  # the paths are just `str`
@@ -158,7 +158,7 @@ def run(
         self.add_python_check_results(report=report, registry=registry, runner_filter=runner_filter, root_folder=root_folder)
 
         # run graph checks
-        if CHECKOV_CREATE_GRAPH and self.graph_registry:
+        if self.graph_registry:
             self.add_graph_check_results(report=report, runner_filter=runner_filter)
 
         return report

checkov/common/util/env_vars_config.py

@@ -22,7 +22,6 @@ def __init__(self) -> None:
         self.CHECK_FAIL_LEVEL = os.getenv("CHECKOV_CHECK_FAIL_LEVEL", CheckFailLevel.ERROR)
         self.CREATE_COMPLEX_VERTICES = convert_str_to_bool(os.getenv("CREATE_COMPLEX_VERTICES", True))
         self.CREATE_EDGES = convert_str_to_bool(os.getenv("CREATE_EDGES", True))
-        self.CREATE_GRAPH = convert_str_to_bool(os.getenv("CHECKOV_CREATE_GRAPH", True))
         self.CREATE_MARKDOWN_HYPERLINKS = convert_str_to_bool(os.getenv("CHECKOV_CREATE_MARKDOWN_HYPERLINKS", False))
         self.CREATE_SCA_IMAGE_REPORTS_FOR_IR = convert_str_to_bool(
             os.getenv("CHECKOV_CREATE_SCA_IMAGE_REPORTS_FOR_IR", True)

checkov/dockerfile/runner.py

@@ -17,7 +17,7 @@
 from checkov.common.output.report import Report
 from checkov.common.output.extra_resource import ExtraResource
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.runners.base_runner import BaseRunner, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner
 from checkov.common.util.consts import START_LINE, END_LINE
 from checkov.common.util.dockerfile import is_dockerfile
 from checkov.common.typing import _CheckResult
@@ -95,7 +95,7 @@ def run(
                 for directory in external_checks_dir:
                     registry.load_external_checks(directory)
 
-                    if CHECKOV_CREATE_GRAPH and self.graph_registry:
+                    if self.graph_registry:
                         self.graph_registry.load_external_checks(directory)
 
             if files:
@@ -111,7 +111,7 @@ def run(
 
             self.context = build_definitions_context(definitions=self.definitions, definitions_raw=self.definitions_raw)
 
-            if CHECKOV_CREATE_GRAPH and self.graph_registry and self.graph_manager:
+            if self.graph_registry and self.graph_manager:
                 logging.info("Creating Dockerfile graph")
                 local_graph = self.graph_manager.build_graph_from_definitions(definitions=self.definitions)
                 logging.info("Successfully created Dockerfile graph")
@@ -124,7 +124,7 @@ def run(
         self.add_python_check_results(report=report, runner_filter=runner_filter, root_folder=root_folder)
 
         # run graph checks
-        if CHECKOV_CREATE_GRAPH and self.graph_registry:
+        if self.graph_registry:
             self.add_graph_check_results(report=report, runner_filter=runner_filter, root_folder=root_folder)
 
         if runner_filter.run_image_referencer:

checkov/kubernetes/runner.py

@@ -16,7 +16,7 @@
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report, merge_reports
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.runners.base_runner import BaseRunner, CHECKOV_CREATE_GRAPH
+from checkov.common.runners.base_runner import BaseRunner
 from checkov.common.util.data_structures_utils import pickle_deepcopy
 from checkov.kubernetes.checks.resource.registry import registry
 from checkov.kubernetes.graph_builder.local_graph import KubernetesLocalGraph
@@ -102,13 +102,13 @@ def run(
                 for directory in external_checks_dir:
                     registry.load_external_checks(directory)
 
-                    if CHECKOV_CREATE_GRAPH and self.graph_registry:
+                    if self.graph_registry:
                         self.graph_registry.load_external_checks(directory)
 
             self.context = build_definitions_context(self.definitions, self.definitions_raw)
             self.spread_list_items()
 
-            if CHECKOV_CREATE_GRAPH and self.graph_manager:
+            if self.graph_manager:
                 logging.info("creating Kubernetes graph")
                 local_graph = self.graph_manager.build_graph_from_definitions(pickle_deepcopy(self.definitions))
                 logging.info("Successfully created Kubernetes graph")
@@ -120,7 +120,7 @@ def run(
         self.pbar.initiate(len(self.definitions))
         report = self.check_definitions(root_folder, runner_filter, report, collect_skip_comments=collect_skip_comments)
 
-        if CHECKOV_CREATE_GRAPH and self.graph_manager:
+        if self.graph_manager:
             graph_report = self.get_graph_checks_report(root_folder, runner_filter)
             merge_reports(report, graph_report)
 

checkov/terraform/graph_manager.py

@@ -32,8 +32,7 @@ def build_multi_graph_from_source_directory(
         excluded_paths: list[str] | None = None,
         external_modules_download_path: str = DEFAULT_EXTERNAL_MODULES_DIR,
         vars_files: list[str] | None = None,
-        create_graph: bool = True,
-    ) -> tuple[list[tuple[TerraformLocalGraph | None, list[dict[TFDefinitionKey, dict[str, Any]]], str]], dict[str, str]]:
+    ) -> tuple[list[tuple[TerraformLocalGraph, list[dict[TFDefinitionKey, dict[str, Any]]], str]], dict[str, str]]:
         logging.info("Parsing HCL files in source dir to multi graph")
         modules_with_definitions = self.parser.parse_multi_graph_hcl_module(
             source_dir=source_dir,
@@ -43,20 +42,18 @@ def build_multi_graph_from_source_directory(
             parsing_errors=parsing_errors,
             excluded_paths=excluded_paths,
             vars_files=vars_files,
-            create_graph=create_graph,
         )
 
-        graphs: list[tuple[TerraformLocalGraph | None, list[dict[TFDefinitionKey, dict[str, Any]]], str]] = []
+        graphs: list[tuple[TerraformLocalGraph, list[dict[TFDefinitionKey, dict[str, Any]]], str]] = []
         resource_subgraph_map: dict[str, str] = {}
         for module, tf_definitions in modules_with_definitions:
-            if create_graph and module:
-                logging.info("Building graph from parsed module")
-                local_graph = local_graph_class(module)
-                local_graph.build_graph(render_variables=render_variables)
-                subgraph_abs_path = module.source_dir
-                subgraph_path = subgraph_abs_path[subgraph_abs_path.rindex(source_dir) + len(source_dir) + 1:]
-                graphs.append((local_graph, tf_definitions, subgraph_path))
-                self.update_resource_subgraph_map(local_graph, subgraph_path, resource_subgraph_map, source_dir)
+            logging.info("Building graph from parsed module")
+            local_graph = local_graph_class(module)
+            local_graph.build_graph(render_variables=render_variables)
+            subgraph_abs_path = module.source_dir
+            subgraph_path = subgraph_abs_path[subgraph_abs_path.rindex(source_dir) + len(source_dir) + 1:]
+            graphs.append((local_graph, tf_definitions, subgraph_path))
+            self.update_resource_subgraph_map(local_graph, subgraph_path, resource_subgraph_map, source_dir)
         return graphs, resource_subgraph_map
 
     def build_graph_from_source_directory(
@@ -69,8 +66,7 @@ def build_graph_from_source_directory(
         excluded_paths: list[str] | None = None,
         external_modules_download_path: str = DEFAULT_EXTERNAL_MODULES_DIR,
         vars_files: list[str] | None = None,
-        create_graph: bool = True,
-    ) -> tuple[TerraformLocalGraph | None, dict[TFDefinitionKey, dict[str, Any]]]:
+    ) -> tuple[TerraformLocalGraph, dict[TFDefinitionKey, dict[str, Any]]]:
         logging.info("Parsing HCL files in source dir to graph")
         module, tf_definitions = self.parser.parse_hcl_module(
             source_dir=source_dir,
@@ -80,14 +76,11 @@ def build_graph_from_source_directory(
             parsing_errors=parsing_errors,
             excluded_paths=excluded_paths,
             vars_files=vars_files,
-            create_graph=create_graph,
         )
 
-        local_graph = None
-        if create_graph and module:
-            logging.info("Building graph from parsed module")
-            local_graph = local_graph_class(module)
-            local_graph.build_graph(render_variables=render_variables)
+        logging.info("Building graph from parsed module")
+        local_graph = local_graph_class(module)
+        local_graph.build_graph(render_variables=render_variables)
 
         return local_graph, tf_definitions
 
@@ -114,12 +107,15 @@ def build_graph_from_definitions(
 
         return local_graph
 
-    def build_multi_graph_from_definitions(self, definitions: dict[TFDefinitionKey, dict[str, Any]],
-                                           render_variables: bool = True) -> list[tuple[str, TerraformLocalGraph]]:
+    def build_multi_graph_from_definitions(
+        self,
+        definitions: dict[TFDefinitionKey, dict[str, Any]],
+        render_variables: bool = True,
+    ) -> list[tuple[str | None, TerraformLocalGraph]]:
         module, tf_definitions = self.parser.parse_hcl_module_from_tf_definitions(definitions, "", self.source)
         dirs_to_definitions = self.parser.create_definition_by_dirs(tf_definitions)
 
-        graphs: list[tuple[str, TerraformLocalGraph]] = []
+        graphs: list[tuple[str | None, TerraformLocalGraph]] = []
         for source_path, dir_definitions in dirs_to_definitions.items():
             module, parsed_tf_definitions = self.parser.parse_hcl_module_from_multi_tf_definitions(dir_definitions, source_path, self.source)
             local_graph = TerraformLocalGraph(module)