We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
checkov silently returns [] when an `--external-checks-dir` doesn't have an init file
**Describe the bug**
When I use `--external-checks-dir` with a directory without an `__init__.py`, all `checkov` does is output `[]` when doing --output json.


**Expected behavior**
Yell at me for giving a directory without any `__init__.py` file in it.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `checkov/common/runners/runner_registry.py`
Content:
```
1 import argparse
2 import itertools
3 import json
4 import logging
5 import os
6 from abc import abstractmethod
7 from typing import List, Union, Dict, Any, Tuple, Optional
8 
9 from typing_extensions import Literal
10 import platform
11 import multiprocessing
12 from multiprocessing import Pipe
13 
14 from cyclonedx.output import get_instance as get_cyclonedx_outputter
15 
16 from checkov.common.bridgecrew.integration_features.integration_feature_registry import integration_feature_registry
17 from checkov.common.output.baseline import Baseline
18 from checkov.common.output.report import Report, report_to_cyclonedx
19 from checkov.common.runners.base_runner import BaseRunner
20 from checkov.common.util import data_structures_utils
21 from checkov.runner_filter import RunnerFilter
22 from checkov.terraform.context_parsers.registry import parser_registry
23 from checkov.terraform.runner import Runner as tf_runner
24 from checkov.terraform.parser import Parser
25 
26 
27 CHECK_BLOCK_TYPES = frozenset(["resource", "data", "provider", "module"])
28 OUTPUT_CHOICES = ["cli", "cyclonedx", "json", "junitxml", "github_failed_only", "sarif"]
29 OUTPUT_DELIMITER = "\n--- OUTPUT DELIMITER ---\n"
30 
31 
32 class RunnerRegistry:
33     runners: List[BaseRunner] = []
34     scan_reports: List[Report] = []
35     banner = ""
36 
37     def __init__(self, banner: str, runner_filter: RunnerFilter, *runners: BaseRunner) -> None:
38         self.logger = logging.getLogger(__name__)
39         self.runner_filter = runner_filter
40         self.runners = list(runners)
41         self.banner = banner
42         self.scan_reports = []
43         self.filter_runner_framework()
44 
45     @abstractmethod
46     def extract_entity_details(self, entity: Dict[str, Any]) -> Tuple[str, str, Dict[str, Any]]:
47         raise NotImplementedError()
48 
49     def run(
50         self,
51         root_folder: Optional[str] = None,
52         external_checks_dir: Optional[List[str]] = None,
53         files: Optional[List[str]] = None,
54         guidelines: Optional[Dict[str, str]] = None,
55         collect_skip_comments: bool = True,
56         repo_root_for_plan_enrichment: Optional[List[Union[str, os.PathLike]]] = None,
57     ) -> List[Report]:
58         if platform.system() == 'Windows':
59             integration_feature_registry.run_pre_runner()
60             for runner in self.runners:
61                 report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,
62                                     runner_filter=self.runner_filter, collect_skip_comments=collect_skip_comments)
63                 self._handle_report(report, guidelines, repo_root_for_plan_enrichment)
64             return self.scan_reports
65 
66         # use multiprocessing for unix os
67         logging.info("Running the runners using multiprocessing")
68         processes = []
69         integration_feature_registry.run_pre_runner()
70         for runner in self.runners:
71             parent_conn, child_conn = Pipe(duplex=False)
72             process = multiprocessing.get_context("fork").Process(target=RunnerRegistry._run_runner,
73                                                                   args=(runner, root_folder, external_checks_dir, files,
74                                                                   self.runner_filter, collect_skip_comments, child_conn))
75             processes.append((process, parent_conn))
76             process.start()
77 
78         for process, parent_conn in processes:
79             scan_report = parent_conn.recv()
80             self._handle_report(scan_report, guidelines, repo_root_for_plan_enrichment)
81         return self.scan_reports
82 
83     def _handle_report(self, scan_report, guidelines, repo_root_for_plan_enrichment):
84         integration_feature_registry.run_post_runner(scan_report)
85         if guidelines:
86             RunnerRegistry.enrich_report_with_guidelines(scan_report, guidelines)
87         if repo_root_for_plan_enrichment:
88             enriched_resources = RunnerRegistry.get_enriched_resources(repo_root_for_plan_enrichment)
89             scan_report = Report("terraform_plan").enrich_plan_report(scan_report, enriched_resources)
90             scan_report = Report("terraform_plan").handle_skipped_checks(scan_report, enriched_resources)
91         self.scan_reports.append(scan_report)
92 
93     @staticmethod
94     def _run_runner(runner, root_folder, external_checks_dir, files, runner_filter, collect_skip_comments,
95                     child_conn):
96         report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,
97                             runner_filter=runner_filter, collect_skip_comments=collect_skip_comments)
98         child_conn.send(report)
99         child_conn.close()
100 
101     def print_reports(
102         self,
103         scan_reports: List[Report],
104         config: argparse.Namespace,
105         url: Optional[str] = None,
106         created_baseline_path: Optional[str] = None,
107         baseline: Optional[Baseline] = None,
108     ) -> Literal[0, 1]:
109         output_formats = set(config.output)
110 
111         if "cli" in config.output and not config.quiet:
112             print(f"{self.banner}\n")
113         exit_codes = []
114         report_jsons = []
115         sarif_reports = []
116         junit_reports = []
117         cyclonedx_reports = []
118         for report in scan_reports:
119             if not report.is_empty():
120                 if "json" in config.output:
121                     report_jsons.append(report.get_dict(is_quiet=config.quiet, url=url))
122                 if "junitxml" in config.output:
123                     junit_reports.append(report)
124                     # report.print_junit_xml()
125                 if "github_failed_only" in config.output:
126                     report.print_failed_github_md(use_bc_ids=config.output_bc_ids)
127                 if "sarif" in config.output:
128                     sarif_reports.append(report)
129                 if "cli" in config.output:
130                     report.print_console(
131                         is_quiet=config.quiet,
132                         is_compact=config.compact,
133                         created_baseline_path=created_baseline_path,
134                         baseline=baseline,
135                         use_bc_ids=config.output_bc_ids,
136                     )
137                     if url:
138                         print("More details: {}".format(url))
139                     output_formats.discard("cli")
140                     if output_formats:
141                         print(OUTPUT_DELIMITER)
142                 if "cyclonedx" in config.output:
143                     cyclonedx_reports.append(report)
144             exit_codes.append(report.get_exit_code(config.soft_fail, config.soft_fail_on, config.hard_fail_on))
145 
146         if "sarif" in config.output:
147             master_report = Report(None)
148             for report in sarif_reports:
149                 master_report.failed_checks += report.failed_checks
150             master_report.print_sarif_report()
151             output_formats.remove("sarif")
152             if output_formats:
153                 print(OUTPUT_DELIMITER)
154         if "json" in config.output:
155             if len(report_jsons) == 1:
156                 print(json.dumps(report_jsons[0], indent=4))
157             else:
158                 print(json.dumps(report_jsons, indent=4))
159             output_formats.remove("json")
160             if output_formats:
161                 print(OUTPUT_DELIMITER)
162         if "junitxml" in config.output:
163             if len(junit_reports) == 1:
164                 junit_reports[0].print_junit_xml(use_bc_ids=config.output_bc_ids)
165             else:
166                 master_report = Report(None)
167                 for report in junit_reports:
168                     master_report.skipped_checks += report.skipped_checks
169                     master_report.passed_checks += report.passed_checks
170                     master_report.failed_checks += report.failed_checks
171                 master_report.print_junit_xml(use_bc_ids=config.output_bc_ids)
172             output_formats.remove("junitxml")
173             if output_formats:
174                 print(OUTPUT_DELIMITER)
175 
176         if "cyclonedx" in config.output:
177             if cyclonedx_reports:
178                 # More than one Report - combine Reports first
179                 report = Report(None)
180                 for r in cyclonedx_reports:
181                     report.passed_checks += r.passed_checks
182                     report.skipped_checks += r.skipped_checks
183                     report.failed_checks += r.failed_checks
184             else:
185                 report = cyclonedx_reports[0]
186             cyclonedx_output = get_cyclonedx_outputter(
187                 bom=report.get_cyclonedx_bom()
188             )
189             print(cyclonedx_output.output_as_string())
190             output_formats.remove("cyclonedx")
191             if output_formats:
192                 print(OUTPUT_DELIMITER)
193 
194         exit_code = 1 if 1 in exit_codes else 0
195         return exit_code
196 
197     def filter_runner_framework(self) -> None:
198         if not self.runner_filter:
199             return
200         if self.runner_filter.framework is None:
201             return
202         if self.runner_filter.framework == "all":
203             return
204         self.runners = [runner for runner in self.runners if runner.check_type in self.runner_filter.framework]
205 
206     def remove_runner(self, runner: BaseRunner) -> None:
207         if runner in self.runners:
208             self.runners.remove(runner)
209 
210     @staticmethod
211     def enrich_report_with_guidelines(scan_report: Report, guidelines: Dict[str, str]) -> None:
212         for record in itertools.chain(scan_report.failed_checks, scan_report.passed_checks, scan_report.skipped_checks):
213             if record.check_id in guidelines:
214                 record.set_guideline(guidelines[record.check_id])
215 
216     @staticmethod
217     def get_enriched_resources(repo_roots: List[Union[str, os.PathLike]]) -> Dict[str, Dict[str, Any]]:
218         repo_definitions = {}
219         for repo_root in repo_roots:
220             tf_definitions = {}
221             parsing_errors = {}
222             Parser().parse_directory(
223                 directory=repo_root,  # assume plan file is in the repo-root
224                 out_definitions=tf_definitions,
225                 out_parsing_errors=parsing_errors,
226             )
227             repo_definitions[repo_root] = { 'tf_definitions': tf_definitions, 'parsing_errors': parsing_errors }
228 
229         enriched_resources = {}
230         for repo_root, parse_results in repo_definitions.items():
231             for full_file_path, definition in parse_results['tf_definitions'].items():
232                 definitions_context = parser_registry.enrich_definitions_context((full_file_path, definition))
233                 abs_scanned_file, _ = tf_runner._strip_module_referrer(full_file_path)
234                 scanned_file = os.path.relpath(abs_scanned_file, repo_root)
235                 for block_type, block_value in definition.items():
236                     if block_type in CHECK_BLOCK_TYPES:
237                         for entity in block_value:
238                             context_parser = parser_registry.context_parsers[block_type]
239                             definition_path = context_parser.get_entity_context_path(entity)
240                             entity_id = ".".join(definition_path)
241                             entity_context_path = [block_type] + definition_path
242                             entity_context = data_structures_utils.get_inner_dict(
243                                 definitions_context[full_file_path], entity_context_path
244                             )
245                             entity_lines_range = [
246                                 entity_context.get("start_line"),
247                                 entity_context.get("end_line"),
248                             ]
249                             entity_code_lines = entity_context.get("code_lines")
250                             skipped_checks = entity_context.get("skipped_checks")
251                             enriched_resources[entity_id] = {
252                                 "entity_code_lines": entity_code_lines,
253                                 "entity_lines_range": entity_lines_range,
254                                 "scanned_file": scanned_file,
255                                 "skipped_checks": skipped_checks,
256                             }
257         return enriched_resources
258
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/checkov/common/runners/runner_registry.py b/checkov/common/runners/runner_registry.py
--- a/checkov/common/runners/runner_registry.py
+++ b/checkov/common/runners/runner_registry.py
@@ -152,7 +152,9 @@
             if output_formats:
                 print(OUTPUT_DELIMITER)
         if "json" in config.output:
-            if len(report_jsons) == 1:
+            if not report_jsons:
+                print(json.dumps(Report(None).get_summary(), indent=4))
+            elif len(report_jsons) == 1:
                 print(json.dumps(report_jsons[0], indent=4))
             else:
                 print(json.dumps(report_jsons, indent=4))

{"golden_diff": "diff --git a/checkov/common/runners/runner_registry.py b/checkov/common/runners/runner_registry.py\n--- a/checkov/common/runners/runner_registry.py\n+++ b/checkov/common/runners/runner_registry.py\n@@ -152,7 +152,9 @@\n             if output_formats:\n                 print(OUTPUT_DELIMITER)\n         if \"json\" in config.output:\n-            if len(report_jsons) == 1:\n+            if not report_jsons:\n+                print(json.dumps(Report(None).get_summary(), indent=4))\n+            elif len(report_jsons) == 1:\n                 print(json.dumps(report_jsons[0], indent=4))\n             else:\n                 print(json.dumps(report_jsons, indent=4))\n", "issue": "checkov silently returns [] when an `--external-checks-dir` doesn't have an init file\n**Describe the bug**\r\nWhen I use `--external-checks-dir` with a directory without an `__init__.py`, all `checkov` does is output `[]` when doing --output json.\r\n\r\n\r\n**Expected behavior**\r\nYell at me for giving a directory without any `__init__.py` file in it.\r\n\n", "before_files": [{"content": "import argparse\nimport itertools\nimport json\nimport logging\nimport os\nfrom abc import abstractmethod\nfrom typing import List, Union, Dict, Any, Tuple, Optional\n\nfrom typing_extensions import Literal\nimport platform\nimport multiprocessing\nfrom multiprocessing import Pipe\n\nfrom cyclonedx.output import get_instance as get_cyclonedx_outputter\n\nfrom checkov.common.bridgecrew.integration_features.integration_feature_registry import integration_feature_registry\nfrom checkov.common.output.baseline import Baseline\nfrom checkov.common.output.report import Report, report_to_cyclonedx\nfrom checkov.common.runners.base_runner import BaseRunner\nfrom checkov.common.util import data_structures_utils\nfrom checkov.runner_filter import RunnerFilter\nfrom checkov.terraform.context_parsers.registry import parser_registry\nfrom checkov.terraform.runner import Runner as tf_runner\nfrom checkov.terraform.parser import Parser\n\n\nCHECK_BLOCK_TYPES = frozenset([\"resource\", \"data\", \"provider\", \"module\"])\nOUTPUT_CHOICES = [\"cli\", \"cyclonedx\", \"json\", \"junitxml\", \"github_failed_only\", \"sarif\"]\nOUTPUT_DELIMITER = \"\\n--- OUTPUT DELIMITER ---\\n\"\n\n\nclass RunnerRegistry:\n    runners: List[BaseRunner] = []\n    scan_reports: List[Report] = []\n    banner = \"\"\n\n    def __init__(self, banner: str, runner_filter: RunnerFilter, *runners: BaseRunner) -> None:\n        self.logger = logging.getLogger(__name__)\n        self.runner_filter = runner_filter\n        self.runners = list(runners)\n        self.banner = banner\n        self.scan_reports = []\n        self.filter_runner_framework()\n\n    @abstractmethod\n    def extract_entity_details(self, entity: Dict[str, Any]) -> Tuple[str, str, Dict[str, Any]]:\n        raise NotImplementedError()\n\n    def run(\n        self,\n        root_folder: Optional[str] = None,\n        external_checks_dir: Optional[List[str]] = None,\n        files: Optional[List[str]] = None,\n        guidelines: Optional[Dict[str, str]] = None,\n        collect_skip_comments: bool = True,\n        repo_root_for_plan_enrichment: Optional[List[Union[str, os.PathLike]]] = None,\n    ) -> List[Report]:\n        if platform.system() == 'Windows':\n            integration_feature_registry.run_pre_runner()\n            for runner in self.runners:\n                report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,\n                                    runner_filter=self.runner_filter, collect_skip_comments=collect_skip_comments)\n                self._handle_report(report, guidelines, repo_root_for_plan_enrichment)\n            return self.scan_reports\n\n        # use multiprocessing for unix os\n        logging.info(\"Running the runners using multiprocessing\")\n        processes = []\n        integration_feature_registry.run_pre_runner()\n        for runner in self.runners:\n            parent_conn, child_conn = Pipe(duplex=False)\n            process = multiprocessing.get_context(\"fork\").Process(target=RunnerRegistry._run_runner,\n                                                                  args=(runner, root_folder, external_checks_dir, files,\n                                                                  self.runner_filter, collect_skip_comments, child_conn))\n            processes.append((process, parent_conn))\n            process.start()\n\n        for process, parent_conn in processes:\n            scan_report = parent_conn.recv()\n            self._handle_report(scan_report, guidelines, repo_root_for_plan_enrichment)\n        return self.scan_reports\n\n    def _handle_report(self, scan_report, guidelines, repo_root_for_plan_enrichment):\n        integration_feature_registry.run_post_runner(scan_report)\n        if guidelines:\n            RunnerRegistry.enrich_report_with_guidelines(scan_report, guidelines)\n        if repo_root_for_plan_enrichment:\n            enriched_resources = RunnerRegistry.get_enriched_resources(repo_root_for_plan_enrichment)\n            scan_report = Report(\"terraform_plan\").enrich_plan_report(scan_report, enriched_resources)\n            scan_report = Report(\"terraform_plan\").handle_skipped_checks(scan_report, enriched_resources)\n        self.scan_reports.append(scan_report)\n\n    @staticmethod\n    def _run_runner(runner, root_folder, external_checks_dir, files, runner_filter, collect_skip_comments,\n                    child_conn):\n        report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,\n                            runner_filter=runner_filter, collect_skip_comments=collect_skip_comments)\n        child_conn.send(report)\n        child_conn.close()\n\n    def print_reports(\n        self,\n        scan_reports: List[Report],\n        config: argparse.Namespace,\n        url: Optional[str] = None,\n        created_baseline_path: Optional[str] = None,\n        baseline: Optional[Baseline] = None,\n    ) -> Literal[0, 1]:\n        output_formats = set(config.output)\n\n        if \"cli\" in config.output and not config.quiet:\n            print(f\"{self.banner}\\n\")\n        exit_codes = []\n        report_jsons = []\n        sarif_reports = []\n        junit_reports = []\n        cyclonedx_reports = []\n        for report in scan_reports:\n            if not report.is_empty():\n                if \"json\" in config.output:\n                    report_jsons.append(report.get_dict(is_quiet=config.quiet, url=url))\n                if \"junitxml\" in config.output:\n                    junit_reports.append(report)\n                    # report.print_junit_xml()\n                if \"github_failed_only\" in config.output:\n                    report.print_failed_github_md(use_bc_ids=config.output_bc_ids)\n                if \"sarif\" in config.output:\n                    sarif_reports.append(report)\n                if \"cli\" in config.output:\n                    report.print_console(\n                        is_quiet=config.quiet,\n                        is_compact=config.compact,\n                        created_baseline_path=created_baseline_path,\n                        baseline=baseline,\n                        use_bc_ids=config.output_bc_ids,\n                    )\n                    if url:\n                        print(\"More details: {}\".format(url))\n                    output_formats.discard(\"cli\")\n                    if output_formats:\n                        print(OUTPUT_DELIMITER)\n                if \"cyclonedx\" in config.output:\n                    cyclonedx_reports.append(report)\n            exit_codes.append(report.get_exit_code(config.soft_fail, config.soft_fail_on, config.hard_fail_on))\n\n        if \"sarif\" in config.output:\n            master_report = Report(None)\n            for report in sarif_reports:\n                master_report.failed_checks += report.failed_checks\n            master_report.print_sarif_report()\n            output_formats.remove(\"sarif\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n        if \"json\" in config.output:\n            if len(report_jsons) == 1:\n                print(json.dumps(report_jsons[0], indent=4))\n            else:\n                print(json.dumps(report_jsons, indent=4))\n            output_formats.remove(\"json\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n        if \"junitxml\" in config.output:\n            if len(junit_reports) == 1:\n                junit_reports[0].print_junit_xml(use_bc_ids=config.output_bc_ids)\n            else:\n                master_report = Report(None)\n                for report in junit_reports:\n                    master_report.skipped_checks += report.skipped_checks\n                    master_report.passed_checks += report.passed_checks\n                    master_report.failed_checks += report.failed_checks\n                master_report.print_junit_xml(use_bc_ids=config.output_bc_ids)\n            output_formats.remove(\"junitxml\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n\n        if \"cyclonedx\" in config.output:\n            if cyclonedx_reports:\n                # More than one Report - combine Reports first\n                report = Report(None)\n                for r in cyclonedx_reports:\n                    report.passed_checks += r.passed_checks\n                    report.skipped_checks += r.skipped_checks\n                    report.failed_checks += r.failed_checks\n            else:\n                report = cyclonedx_reports[0]\n            cyclonedx_output = get_cyclonedx_outputter(\n                bom=report.get_cyclonedx_bom()\n            )\n            print(cyclonedx_output.output_as_string())\n            output_formats.remove(\"cyclonedx\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n\n        exit_code = 1 if 1 in exit_codes else 0\n        return exit_code\n\n    def filter_runner_framework(self) -> None:\n        if not self.runner_filter:\n            return\n        if self.runner_filter.framework is None:\n            return\n        if self.runner_filter.framework == \"all\":\n            return\n        self.runners = [runner for runner in self.runners if runner.check_type in self.runner_filter.framework]\n\n    def remove_runner(self, runner: BaseRunner) -> None:\n        if runner in self.runners:\n            self.runners.remove(runner)\n\n    @staticmethod\n    def enrich_report_with_guidelines(scan_report: Report, guidelines: Dict[str, str]) -> None:\n        for record in itertools.chain(scan_report.failed_checks, scan_report.passed_checks, scan_report.skipped_checks):\n            if record.check_id in guidelines:\n                record.set_guideline(guidelines[record.check_id])\n\n    @staticmethod\n    def get_enriched_resources(repo_roots: List[Union[str, os.PathLike]]) -> Dict[str, Dict[str, Any]]:\n        repo_definitions = {}\n        for repo_root in repo_roots:\n            tf_definitions = {}\n            parsing_errors = {}\n            Parser().parse_directory(\n                directory=repo_root,  # assume plan file is in the repo-root\n                out_definitions=tf_definitions,\n                out_parsing_errors=parsing_errors,\n            )\n            repo_definitions[repo_root] = { 'tf_definitions': tf_definitions, 'parsing_errors': parsing_errors }\n\n        enriched_resources = {}\n        for repo_root, parse_results in repo_definitions.items():\n            for full_file_path, definition in parse_results['tf_definitions'].items():\n                definitions_context = parser_registry.enrich_definitions_context((full_file_path, definition))\n                abs_scanned_file, _ = tf_runner._strip_module_referrer(full_file_path)\n                scanned_file = os.path.relpath(abs_scanned_file, repo_root)\n                for block_type, block_value in definition.items():\n                    if block_type in CHECK_BLOCK_TYPES:\n                        for entity in block_value:\n                            context_parser = parser_registry.context_parsers[block_type]\n                            definition_path = context_parser.get_entity_context_path(entity)\n                            entity_id = \".\".join(definition_path)\n                            entity_context_path = [block_type] + definition_path\n                            entity_context = data_structures_utils.get_inner_dict(\n                                definitions_context[full_file_path], entity_context_path\n                            )\n                            entity_lines_range = [\n                                entity_context.get(\"start_line\"),\n                                entity_context.get(\"end_line\"),\n                            ]\n                            entity_code_lines = entity_context.get(\"code_lines\")\n                            skipped_checks = entity_context.get(\"skipped_checks\")\n                            enriched_resources[entity_id] = {\n                                \"entity_code_lines\": entity_code_lines,\n                                \"entity_lines_range\": entity_lines_range,\n                                \"scanned_file\": scanned_file,\n                                \"skipped_checks\": skipped_checks,\n                            }\n        return enriched_resources\n", "path": "checkov/common/runners/runner_registry.py"}], "after_files": [{"content": "import argparse\nimport itertools\nimport json\nimport logging\nimport os\nfrom abc import abstractmethod\nfrom typing import List, Union, Dict, Any, Tuple, Optional\n\nfrom typing_extensions import Literal\nimport platform\nimport multiprocessing\nfrom multiprocessing import Pipe\n\nfrom cyclonedx.output import get_instance as get_cyclonedx_outputter\n\nfrom checkov.common.bridgecrew.integration_features.integration_feature_registry import integration_feature_registry\nfrom checkov.common.output.baseline import Baseline\nfrom checkov.common.output.report import Report, report_to_cyclonedx\nfrom checkov.common.runners.base_runner import BaseRunner\nfrom checkov.common.util import data_structures_utils\nfrom checkov.runner_filter import RunnerFilter\nfrom checkov.terraform.context_parsers.registry import parser_registry\nfrom checkov.terraform.runner import Runner as tf_runner\nfrom checkov.terraform.parser import Parser\n\n\nCHECK_BLOCK_TYPES = frozenset([\"resource\", \"data\", \"provider\", \"module\"])\nOUTPUT_CHOICES = [\"cli\", \"cyclonedx\", \"json\", \"junitxml\", \"github_failed_only\", \"sarif\"]\nOUTPUT_DELIMITER = \"\\n--- OUTPUT DELIMITER ---\\n\"\n\n\nclass RunnerRegistry:\n    runners: List[BaseRunner] = []\n    scan_reports: List[Report] = []\n    banner = \"\"\n\n    def __init__(self, banner: str, runner_filter: RunnerFilter, *runners: BaseRunner) -> None:\n        self.logger = logging.getLogger(__name__)\n        self.runner_filter = runner_filter\n        self.runners = list(runners)\n        self.banner = banner\n        self.scan_reports = []\n        self.filter_runner_framework()\n\n    @abstractmethod\n    def extract_entity_details(self, entity: Dict[str, Any]) -> Tuple[str, str, Dict[str, Any]]:\n        raise NotImplementedError()\n\n    def run(\n        self,\n        root_folder: Optional[str] = None,\n        external_checks_dir: Optional[List[str]] = None,\n        files: Optional[List[str]] = None,\n        guidelines: Optional[Dict[str, str]] = None,\n        collect_skip_comments: bool = True,\n        repo_root_for_plan_enrichment: Optional[List[Union[str, os.PathLike]]] = None,\n    ) -> List[Report]:\n        if platform.system() == 'Windows':\n            integration_feature_registry.run_pre_runner()\n            for runner in self.runners:\n                report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,\n                                    runner_filter=self.runner_filter, collect_skip_comments=collect_skip_comments)\n                self._handle_report(report, guidelines, repo_root_for_plan_enrichment)\n            return self.scan_reports\n\n        # use multiprocessing for unix os\n        logging.info(\"Running the runners using multiprocessing\")\n        processes = []\n        integration_feature_registry.run_pre_runner()\n        for runner in self.runners:\n            parent_conn, child_conn = Pipe(duplex=False)\n            process = multiprocessing.get_context(\"fork\").Process(target=RunnerRegistry._run_runner,\n                                                                  args=(runner, root_folder, external_checks_dir, files,\n                                                                  self.runner_filter, collect_skip_comments, child_conn))\n            processes.append((process, parent_conn))\n            process.start()\n\n        for process, parent_conn in processes:\n            scan_report = parent_conn.recv()\n            self._handle_report(scan_report, guidelines, repo_root_for_plan_enrichment)\n        return self.scan_reports\n\n    def _handle_report(self, scan_report, guidelines, repo_root_for_plan_enrichment):\n        integration_feature_registry.run_post_runner(scan_report)\n        if guidelines:\n            RunnerRegistry.enrich_report_with_guidelines(scan_report, guidelines)\n        if repo_root_for_plan_enrichment:\n            enriched_resources = RunnerRegistry.get_enriched_resources(repo_root_for_plan_enrichment)\n            scan_report = Report(\"terraform_plan\").enrich_plan_report(scan_report, enriched_resources)\n            scan_report = Report(\"terraform_plan\").handle_skipped_checks(scan_report, enriched_resources)\n        self.scan_reports.append(scan_report)\n\n    @staticmethod\n    def _run_runner(runner, root_folder, external_checks_dir, files, runner_filter, collect_skip_comments,\n                    child_conn):\n        report = runner.run(root_folder, external_checks_dir=external_checks_dir, files=files,\n                            runner_filter=runner_filter, collect_skip_comments=collect_skip_comments)\n        child_conn.send(report)\n        child_conn.close()\n\n    def print_reports(\n        self,\n        scan_reports: List[Report],\n        config: argparse.Namespace,\n        url: Optional[str] = None,\n        created_baseline_path: Optional[str] = None,\n        baseline: Optional[Baseline] = None,\n    ) -> Literal[0, 1]:\n        output_formats = set(config.output)\n\n        if \"cli\" in config.output and not config.quiet:\n            print(f\"{self.banner}\\n\")\n        exit_codes = []\n        report_jsons = []\n        sarif_reports = []\n        junit_reports = []\n        cyclonedx_reports = []\n        for report in scan_reports:\n            if not report.is_empty():\n                if \"json\" in config.output:\n                    report_jsons.append(report.get_dict(is_quiet=config.quiet, url=url))\n                if \"junitxml\" in config.output:\n                    junit_reports.append(report)\n                    # report.print_junit_xml()\n                if \"github_failed_only\" in config.output:\n                    report.print_failed_github_md(use_bc_ids=config.output_bc_ids)\n                if \"sarif\" in config.output:\n                    sarif_reports.append(report)\n                if \"cli\" in config.output:\n                    report.print_console(\n                        is_quiet=config.quiet,\n                        is_compact=config.compact,\n                        created_baseline_path=created_baseline_path,\n                        baseline=baseline,\n                        use_bc_ids=config.output_bc_ids,\n                    )\n                    if url:\n                        print(\"More details: {}\".format(url))\n                    output_formats.discard(\"cli\")\n                    if output_formats:\n                        print(OUTPUT_DELIMITER)\n                if \"cyclonedx\" in config.output:\n                    cyclonedx_reports.append(report)\n            exit_codes.append(report.get_exit_code(config.soft_fail, config.soft_fail_on, config.hard_fail_on))\n\n        if \"sarif\" in config.output:\n            master_report = Report(None)\n            for report in sarif_reports:\n                master_report.failed_checks += report.failed_checks\n            master_report.print_sarif_report()\n            output_formats.remove(\"sarif\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n        if \"json\" in config.output:\n            if not report_jsons:\n                print(json.dumps(Report(None).get_summary(), indent=4))\n            elif len(report_jsons) == 1:\n                print(json.dumps(report_jsons[0], indent=4))\n            else:\n                print(json.dumps(report_jsons, indent=4))\n            output_formats.remove(\"json\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n        if \"junitxml\" in config.output:\n            if len(junit_reports) == 1:\n                junit_reports[0].print_junit_xml(use_bc_ids=config.output_bc_ids)\n            else:\n                master_report = Report(None)\n                for report in junit_reports:\n                    master_report.skipped_checks += report.skipped_checks\n                    master_report.passed_checks += report.passed_checks\n                    master_report.failed_checks += report.failed_checks\n                master_report.print_junit_xml(use_bc_ids=config.output_bc_ids)\n            output_formats.remove(\"junitxml\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n\n        if \"cyclonedx\" in config.output:\n            if cyclonedx_reports:\n                # More than one Report - combine Reports first\n                report = Report(None)\n                for r in cyclonedx_reports:\n                    report.passed_checks += r.passed_checks\n                    report.skipped_checks += r.skipped_checks\n                    report.failed_checks += r.failed_checks\n            else:\n                report = cyclonedx_reports[0]\n            cyclonedx_output = get_cyclonedx_outputter(\n                bom=report.get_cyclonedx_bom()\n            )\n            print(cyclonedx_output.output_as_string())\n            output_formats.remove(\"cyclonedx\")\n            if output_formats:\n                print(OUTPUT_DELIMITER)\n\n        exit_code = 1 if 1 in exit_codes else 0\n        return exit_code\n\n    def filter_runner_framework(self) -> None:\n        if not self.runner_filter:\n            return\n        if self.runner_filter.framework is None:\n            return\n        if self.runner_filter.framework == \"all\":\n            return\n        self.runners = [runner for runner in self.runners if runner.check_type in self.runner_filter.framework]\n\n    def remove_runner(self, runner: BaseRunner) -> None:\n        if runner in self.runners:\n            self.runners.remove(runner)\n\n    @staticmethod\n    def enrich_report_with_guidelines(scan_report: Report, guidelines: Dict[str, str]) -> None:\n        for record in itertools.chain(scan_report.failed_checks, scan_report.passed_checks, scan_report.skipped_checks):\n            if record.check_id in guidelines:\n                record.set_guideline(guidelines[record.check_id])\n\n    @staticmethod\n    def get_enriched_resources(repo_roots: List[Union[str, os.PathLike]]) -> Dict[str, Dict[str, Any]]:\n        repo_definitions = {}\n        for repo_root in repo_roots:\n            tf_definitions = {}\n            parsing_errors = {}\n            Parser().parse_directory(\n                directory=repo_root,  # assume plan file is in the repo-root\n                out_definitions=tf_definitions,\n                out_parsing_errors=parsing_errors,\n            )\n            repo_definitions[repo_root] = { 'tf_definitions': tf_definitions, 'parsing_errors': parsing_errors }\n\n        enriched_resources = {}\n        for repo_root, parse_results in repo_definitions.items():\n            for full_file_path, definition in parse_results['tf_definitions'].items():\n                definitions_context = parser_registry.enrich_definitions_context((full_file_path, definition))\n                abs_scanned_file, _ = tf_runner._strip_module_referrer(full_file_path)\n                scanned_file = os.path.relpath(abs_scanned_file, repo_root)\n                for block_type, block_value in definition.items():\n                    if block_type in CHECK_BLOCK_TYPES:\n                        for entity in block_value:\n                            context_parser = parser_registry.context_parsers[block_type]\n                            definition_path = context_parser.get_entity_context_path(entity)\n                            entity_id = \".\".join(definition_path)\n                            entity_context_path = [block_type] + definition_path\n                            entity_context = data_structures_utils.get_inner_dict(\n                                definitions_context[full_file_path], entity_context_path\n                            )\n                            entity_lines_range = [\n                                entity_context.get(\"start_line\"),\n                                entity_context.get(\"end_line\"),\n                            ]\n                            entity_code_lines = entity_context.get(\"code_lines\")\n                            skipped_checks = entity_context.get(\"skipped_checks\")\n                            enriched_resources[entity_id] = {\n                                \"entity_code_lines\": entity_code_lines,\n                                \"entity_lines_range\": entity_lines_range,\n                                \"scanned_file\": scanned_file,\n                                \"skipped_checks\": skipped_checks,\n                            }\n        return enriched_resources\n", "path": "checkov/common/runners/runner_registry.py"}]}