upload resource_subgraph_map to s3 using the platform_integration

checkov/common/bridgecrew/platform_integration.py

@@ -33,7 +33,7 @@
 from checkov.common.bridgecrew.platform_key import read_key, persist_key, bridgecrew_file
 from checkov.common.bridgecrew.wrapper import reduce_scan_reports, persist_checks_results, \
     enrich_and_persist_checks_metadata, checkov_results_prefix, persist_run_metadata, _put_json_object, \
-    persist_logs_stream, persist_graphs
+    persist_logs_stream, persist_graphs, persist_resource_subgraph_maps
 from checkov.common.models.consts import SUPPORTED_FILE_EXTENSIONS, SUPPORTED_FILES, SCANNABLE_PACKAGE_FILES
 from checkov.common.bridgecrew.check_type import CheckType
 from checkov.common.runners.base_runner import filter_ignored_paths
@@ -565,6 +565,14 @@ def persist_graphs(self, graphs: dict[str, list[tuple[LibraryGraph, Optional[str
         persist_graphs(graphs, self.s3_client, self.bucket, self.repo_path, self.persist_graphs_timeout,
                        absolute_root_folder=absolute_root_folder)
 
+    def persist_resource_subgraph_maps(self, resource_subgraph_maps: dict[str, dict[str, str]]) -> None:
+        if not self.use_s3_integration or not self.s3_client:
+            return
+        if not self.bucket or not self.repo_path:
+            logging.error(f"Something went wrong: bucket {self.bucket}, repo path {self.repo_path}")
+            return
+        persist_resource_subgraph_maps(resource_subgraph_maps, self.s3_client, self.bucket, self.repo_path, self.persist_graphs_timeout)
+
     def commit_repository(self, branch: str) -> str | None:
         """
         :param branch: branch to be persisted

checkov/common/bridgecrew/wrapper.py

@@ -190,3 +190,28 @@ def _upload_graph(check_type: str, graph: LibraryGraph, _absolute_root_folder: s
             timeout=timeout
         )
     logging.info(f"Done persisting {len(list(itertools.chain(*graphs.values())))} graphs")
+
+
+def persist_resource_subgraph_maps(
+        resource_subgraph_maps: dict[str, dict[str, str]],
+        s3_client: S3Client,
+        bucket: str,
+        full_repo_object_key: str,
+        timeout: int
+) -> None:
+    def _upload_resource_subgraph_map(check_type: str, resource_subgraph_map: dict[str, str]):
+        s3_key = os.path.join(graphs_repo_object_key, check_type, "multi-graph/resource_subgraph_maps")
+        try:
+            _put_json_object(s3_client, resource_subgraph_map, bucket, s3_key)
+        except Exception:
+            logging.error(f'failed to upload resource_subgraph_map from framework {check_type} to platform', exc_info=True)
+
+    graphs_repo_object_key = full_repo_object_key.replace('checkov', 'graphs')[:-4]
+    with futures.ThreadPoolExecutor() as executor:
+        futures.wait(
+            [executor.submit(_upload_resource_subgraph_map, check_type, resource_subgraph_map) for
+             check_type, resource_subgraph_map in resource_subgraph_maps.items()],
+            return_when=futures.FIRST_EXCEPTION,
+            timeout=timeout
+        )
+    logging.info(f"Done persisting {len(resource_subgraph_maps)} resource_subgraph_maps")

checkov/common/runners/runner_registry.py

@@ -96,6 +96,7 @@ def __init__(
         self.licensing_integration = licensing_integration  # can be maniuplated by unit tests
         self.secrets_omitter_class = secrets_omitter_class
         self.check_type_to_graph: dict[str, list[tuple[LibraryGraph, Optional[str]]]] = {}
+        self.check_type_to_resource_subgraph_map: dict[str, dict[str, str]] = {}
         for runner in runners:
             if isinstance(runner, image_runner):
                 runner.image_referencers = self.image_referencing_runners
@@ -124,7 +125,7 @@ def run(
                 # This is the only runner, so raise a clear indication of failure
                 raise ModuleNotEnabledError(f'The framework "{runner_check_type}" is part of the "{self.licensing_integration.get_subscription_for_runner(runner_check_type).name}" module, which is not enabled in the platform')
         else:
-            def _parallel_run(runner: _BaseRunner) -> tuple[Report | list[Report], str | None, Optional[list[tuple[LibraryGraph, Optional[str]]]]]:
+            def _parallel_run(runner: _BaseRunner) -> tuple[Report | list[Report], str | None, Optional[list[tuple[LibraryGraph, Optional[str]]]], Optional[dict[str, str]]]:
                 report = runner.run(
                     root_folder=root_folder,
                     external_checks_dir=external_checks_dir,
@@ -138,8 +139,9 @@ def _parallel_run(runner: _BaseRunner) -> tuple[Report | list[Report], str | Non
                     report = Report(check_type=runner.check_type)
 
                 if runner.graph_manager:
-                    return report, runner.check_type, self.extract_graphs_from_runner(runner)
-                return report, None, None
+                    return report, runner.check_type, self.extract_graphs_from_runner(runner), \
+                        self.extract_resource_subgraph_map_from_runner(runner)
+                return report, None, None, None
 
             valid_runners = []
             invalid_runners = []
@@ -170,13 +172,18 @@ def _parallel_run(runner: _BaseRunner) -> tuple[Report | list[Report], str | Non
                                                                    group_size=1)
             reports = []
             full_check_type_to_graph = {}
+            full_check_type_to_resource_subgraph_map = {}
             for result in parallel_runner_results:
                 if result is not None:
-                    report, check_type, graphs = result
+                    report, check_type, graphs, resource_subgraph_map = result
                     reports.append(report)
-                    if check_type is not None and graphs is not None:
-                        full_check_type_to_graph[check_type] = graphs
+                    if check_type is not None:
+                        if graphs is not None:
+                            full_check_type_to_graph[check_type] = graphs
+                        if resource_subgraph_map is not None:
+                            full_check_type_to_resource_subgraph_map[check_type] = resource_subgraph_map
             self.check_type_to_graph = full_check_type_to_graph
+            self.check_type_to_resource_subgraph_map = full_check_type_to_resource_subgraph_map
 
         merged_reports = self._merge_reports(reports)
         if bc_integration.bc_api_key:
@@ -192,6 +199,12 @@ def _parallel_run(runner: _BaseRunner) -> tuple[Report | list[Report], str | Non
         if not self.check_type_to_graph:
             self.check_type_to_graph = {runner.check_type: self.extract_graphs_from_runner(runner) for runner
                                         in self.runners if runner.graph_manager}
+        if not self.check_type_to_resource_subgraph_map:
+            self.check_type_to_resource_subgraph_map = {}
+            for runner in self.runners:
+                resource_subgraph_map = self.extract_resource_subgraph_map_from_runner(runner)
+                if resource_subgraph_map is not None:
+                    self.check_type_to_resource_subgraph_map[runner.check_type] = resource_subgraph_map
         return self.scan_reports
 
     def _merge_reports(self, reports: Iterable[Report | list[Report]]) -> list[Report]:
@@ -758,3 +771,8 @@ def extract_graphs_from_runner(runner: _BaseRunner) -> list[tuple[LibraryGraph,
         elif runner.graph_manager:
             return [(runner.graph_manager.get_reader_endpoint(), None)]
         return []
+
+    @staticmethod
+    def extract_resource_subgraph_map_from_runner(runner: _BaseRunner) -> Optional[dict[str, str]]:
+        # exist only for terraform
+        return getattr(runner, 'resource_subgraph_map', None)   # type:ignore[no-any-return]

checkov/main.py

@@ -472,6 +472,7 @@ def run(self, banner: str = checkov_banner, tool: str = checkov_tool, source_typ
                         files=file,
                     )
                     self.graphs = runner_registry.check_type_to_graph
+                    self.resource_subgraph_maps = runner_registry.check_type_to_resource_subgraph_map
                     if runner_registry.is_error_in_reports(self.scan_reports):
                         self.exit_run()
                     if baseline:
@@ -555,6 +556,7 @@ def run(self, banner: str = checkov_banner, tool: str = checkov_tool, source_typ
                     bc_integration.persist_run_metadata(self.run_metadata)
                     if bc_integration.enable_persist_graphs:
                         bc_integration.persist_graphs(self.graphs)
+                        bc_integration.persist_resource_subgraph_maps(self.resource_subgraph_maps)
                     self.url = self.commit_repository()
 
                 should_run_contributor_metrics = bc_integration.bc_api_key and self.config.repo_id and self.config.prisma_api_url
@@ -575,6 +577,7 @@ def run(self, banner: str = checkov_banner, tool: str = checkov_tool, source_typ
                     repo_root_for_plan_enrichment=self.config.repo_root_for_plan_enrichment,
                 )
                 self.graphs = runner_registry.check_type_to_graph
+                self.resource_subgraph_maps = runner_registry.check_type_to_resource_subgraph_map
                 if runner_registry.is_error_in_reports(self.scan_reports):
                     self.exit_run()
                 if baseline:
@@ -680,6 +683,7 @@ def upload_results(
         bc_integration.persist_run_metadata(self.run_metadata)
         if bc_integration.enable_persist_graphs:
             bc_integration.persist_graphs(self.graphs, absolute_root_folder=absolute_root_folder)
+            bc_integration.persist_resource_subgraph_maps(self.resource_subgraph_maps)
         self.url = self.commit_repository()
 
     def print_results(