fix(kubernetes): report unparsable YAML files as parsing errors

checkov/kubernetes/kubernetes_utils.py

@@ -41,7 +41,7 @@ def should_include_path(full_path: str, ignore_hidden_dir: bool) -> bool:
 
 
 def get_folder_definitions(
-        root_folder: str, excluded_paths: list[str] | None
+        root_folder: str, excluded_paths: list[str] | None, out_parsing_errors: dict[str, str] | None = None
 ) -> tuple[dict[str, list[dict[str, Any]]], dict[str, list[tuple[int, str]]]]:
     files_list = []
     for root, d_names, f_names in os.walk(root_folder):
@@ -55,24 +55,33 @@ def get_folder_definitions(
                 if should_include_path(full_path, env_vars_config.IGNORE_HIDDEN_DIRECTORIES):
                     # skip temp directories
                     files_list.append(full_path)
-    return get_files_definitions(files_list)
+    return get_files_definitions(files_list, out_parsing_errors)
 
 
-def get_files_definitions(files: list[str]) -> tuple[dict[str, list[dict[str, Any]]], dict[str, list[tuple[int, str]]]]:
+def get_files_definitions(
+    files: list[str], out_parsing_errors: dict[str, str] | None = None
+) -> tuple[dict[str, list[dict[str, Any]]], dict[str, list[tuple[int, str]]]]:
     definitions = {}
     definitions_raw = {}
     results = parallel_runner.run_function(_parse_file, files)
     for result in results:
         if result:
-            path, parse_result = result
+            path, parse_result, file_parsing_errors = result
+            if file_parsing_errors and out_parsing_errors is not None:
+                out_parsing_errors.update(file_parsing_errors)
             if parse_result:
                 definitions[path], definitions_raw[path] = parse_result
     return definitions, definitions_raw
 
 
-def _parse_file(filename: str) -> tuple[str, tuple[list[dict[str, Any]], list[tuple[int, str]]] | None] | None:
+def _parse_file(
+    filename: str,
+) -> tuple[str, tuple[list[dict[str, Any]], list[tuple[int, str]]] | None, dict[str, str]] | None:
+    # the parsing errors are collected per file and merged by the caller,
+    # because the parsing itself may run in a separate process
+    parsing_errors: dict[str, str] = {}
     try:
-        return filename, parse(filename)
+        return filename, parse(filename, out_parsing_errors=parsing_errors), parsing_errors
     except (TypeError, ValueError):
         logging.warning(f"Kubernetes skipping {filename} as it is not a valid Kubernetes template", exc_info=True)
 
@@ -117,15 +126,21 @@ def create_definitions(
     root_folder: str | None,
     files: list[str] | None = None,
     runner_filter: RunnerFilter | None = None,
+    out_parsing_errors: dict[str, str] | None = None,
 ) -> tuple[dict[str, list[dict[str, Any]]], dict[str, list[tuple[int, str]]]]:
     runner_filter = runner_filter or RunnerFilter()
     definitions: dict[str, list[dict[str, Any]]] = {}
     definitions_raw: dict[str, list[tuple[int, str]]] = {}
     if files:
-        definitions, definitions_raw = get_files_definitions(files)
+        definitions, definitions_raw = get_files_definitions(files, out_parsing_errors)
 
     if root_folder:
-        definitions, definitions_raw = get_folder_definitions(root_folder, runner_filter.excluded_paths)
+        definitions, definitions_raw = get_folder_definitions(
+            root_folder, runner_filter.excluded_paths, out_parsing_errors
+        )
+
+    if out_parsing_errors:
+        logging.warning(f"[kubernetes] found errors while parsing definitions: {list(out_parsing_errors.keys())}")
 
     return definitions, definitions_raw
 

checkov/kubernetes/parser/parser.py

@@ -14,7 +14,9 @@
 add_resource_code_filter_to_logger(logger)
 
 
-def parse(filename: str) -> tuple[list[dict[str, Any]], list[tuple[int, str]]] | None:
+def parse(
+    filename: str, out_parsing_errors: dict[str, str] | None = None
+) -> tuple[list[dict[str, Any]], list[tuple[int, str]]] | None:
     template = None
     template_lines: "list[tuple[int, str]]" = []
     valid_templates = []
@@ -50,9 +52,11 @@ def parse(filename: str) -> tuple[list[dict[str, Any]], list[tuple[int, str]]] |
     except UnicodeDecodeError:
         logger.error('Cannot read file contents: %s', filename)
         return None
-    except YAMLError:
+    except YAMLError as e:
         if filename.endswith(".yaml") or filename.endswith(".yml"):
             logger.debug('Cannot read file contents: %s - is it a yaml?', filename)
+            if out_parsing_errors is not None:
+                out_parsing_errors[filename] = str(e)
         return None
 
     return valid_templates, template_lines

checkov/kubernetes/runner.py

@@ -95,7 +95,10 @@ def run(
         report = Report(self.check_type)
         if self.context is None or self.definitions is None:
             if files or root_folder:
-                self.definitions, self.definitions_raw = create_definitions(root_folder, files, runner_filter)
+                parsing_errors: dict[str, str] = {}
+                self.definitions, self.definitions_raw = create_definitions(root_folder, files, runner_filter,
+                                                                            parsing_errors)
+                report.add_parsing_errors(parsing_errors.keys())
             else:
                 return report
             if external_checks_dir:

tests/kubernetes/runner/test_runner.py

@@ -1,9 +1,11 @@
 import dis
 import inspect
 import os
+import tempfile
 import unittest
 from collections import defaultdict
 from pathlib import Path
+from unittest import mock
 
 from parameterized import parameterized_class
 
@@ -13,6 +15,7 @@
 from checkov.common.graph.db_connectors.networkx.networkx_db_connector import NetworkxConnector
 from checkov.common.graph.db_connectors.rustworkx.rustworkx_db_connector import RustworkxConnector
 from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.common.util.consts import PARSE_ERROR_FAIL_FLAG
 from checkov.kubernetes.checks.resource.base_spec_check import BaseK8Check
 from checkov.runner_filter import RunnerFilter
 from checkov.kubernetes.runner import Runner
@@ -178,6 +181,44 @@ def test_parse_with_empty_blocks(self):
         except Exception:
             self.assertTrue(False, "Could not run K8 runner on configuration")
 
+    def test_unparsable_file_is_reported_as_parsing_error(self):
+        # a single '=' is a valid YAML value, but the safe loader can't construct it,
+        # so the file used to be skipped without any indication in the report
+        invalid_template = (
+            "apiVersion: v1\n"
+            "kind: Service\n"
+            "metadata:\n"
+            "  name: test\n"
+            "  labels:\n"
+            "    test: =\n"
+            "spec:\n"
+            "  selector:\n"
+            "    app: test\n"
+            "  ports:\n"
+            "  - port: 8080\n"
+        )
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            scan_file_path = os.path.join(tmp_dir, "service.yaml")
+            with open(scan_file_path, "w") as f:
+                f.write(invalid_template)
+
+            runner = Runner(db_connector=self.db_connector())
+            report = runner.run(root_folder=None, external_checks_dir=None, files=[scan_file_path],
+                                runner_filter=RunnerFilter(framework=['kubernetes']))
+
+        self.assertEqual(report.parsing_errors, [scan_file_path])
+        self.assertEqual(report.get_summary()["parsing_errors"], 1)
+        self.assertEqual(len(report.passed_checks), 0)
+        self.assertEqual(len(report.failed_checks), 0)
+
+        # the file is now part of the report, so 'CKV_PARSE_ERROR_FAIL' can act on it
+        exit_code_thresholds = {'soft_fail': False, 'soft_fail_checks': [], 'soft_fail_threshold': None,
+                                'hard_fail_checks': [], 'hard_fail_threshold': None}
+        self.assertEqual(report.get_exit_code(exit_code_thresholds), 0)
+        with mock.patch.dict(os.environ, {PARSE_ERROR_FAIL_FLAG: "true"}):
+            self.assertEqual(report.get_exit_code(exit_code_thresholds), 1)
+
     def test_record_includes_severity(self):
         custom_check_id = "CKV_MY_CUSTOM_CHECK"