fix(serverless): report unparsable serverless files as parsing errors

checkov/serverless/parsers/parser.py

@@ -35,7 +35,9 @@
 FILE_LOCATION_PATTERN = re.compile(r'^file\(([^?%*:|"<>]+?)\)')
 
 
-def parse(filename: str) -> tuple[dict[str, Any], list[tuple[int, str]]] | None:
+def parse(
+    filename: str, out_parsing_errors: dict[str, str] | None = None
+) -> tuple[dict[str, Any], list[tuple[int, str]]] | None:
     template = None
     template_lines = None
 
@@ -57,6 +59,8 @@ def parse(filename: str) -> tuple[dict[str, Any], list[tuple[int, str]]] | None:
         return None
     except CfnParseError as e:
         logger.warning(f"Failed to parse file {e.filename} because it isn't valid yaml")
+        if out_parsing_errors is not None:
+            out_parsing_errors[filename] = str(e)
         return None
 
     if template is None or template_lines is None:

checkov/serverless/runner.py

@@ -108,7 +108,9 @@ def run(
             if self.root_folder:
                 files_list = get_scannable_file_paths(self.root_folder, runner_filter.excluded_paths)
 
-            definitions, definitions_raw = get_files_definitions(files_list, filepath_fn)
+            parsing_errors: dict[str, str] = {}
+            definitions, definitions_raw = get_files_definitions(files_list, filepath_fn, parsing_errors)
+            report.add_parsing_errors(parsing_errors.keys())
 
             # Filter out empty files that have not been parsed successfully
             self.definitions = {k: v for k, v in definitions.items() if v}

checkov/serverless/utils.py

@@ -73,22 +73,30 @@ def get_scannable_file_paths(root_folder: str | None = None, excluded_paths: lis
 
 
 def get_files_definitions(
-        files: list[str], filepath_fn: Callable[[str], str] | None = None
+        files: list[str], filepath_fn: Callable[[str], str] | None = None,
+        out_parsing_errors: dict[str, str] | None = None
 ) -> tuple[dict[str, dict[str, Any]], dict[str, list[tuple[int, str]]]]:
     results = parallel_runner.run_function(_parallel_parse, files)
     definitions = {}
     definitions_raw = {}
-    for file, result in results:
+    for file, result, file_parsing_errors in results:
+        if file_parsing_errors and out_parsing_errors is not None:
+            out_parsing_errors.update(file_parsing_errors)
         if result:
             path = filepath_fn(file) if filepath_fn else file
             definitions[path], definitions_raw[path] = result
 
     return definitions, definitions_raw
 
 
-def _parallel_parse(f: str) -> tuple[str, tuple[dict[str, Any], list[tuple[int, str]]] | None]:
+def _parallel_parse(
+    f: str,
+) -> tuple[str, tuple[dict[str, Any], list[tuple[int, str]]] | None, dict[str, str]]:
     """Thin wrapper to return filename with parsed content"""
-    return f, parse(f)
+    # 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] = {}
+    return f, parse(f, out_parsing_errors=parsing_errors), parsing_errors
 
 
 def get_resource_tags(entity: EntityDetails, registry: ServerlessRegistry = sls_registry) -> Optional[dict[str, str]]:

tests/serverless/runner/test_runner.py

@@ -1,15 +1,18 @@
 import dis
 import inspect
 import os
+import tempfile
 import unittest
 from collections import defaultdict
 from pathlib import Path
 from typing import Dict, Any
+from unittest import mock
 
 from checkov.cloudformation.checks.resource.aws import *  # noqa - prevent circular import
 from checkov.common.bridgecrew.check_type import CheckType
 from checkov.common.bridgecrew.severities import Severities, BcSeverities
 from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.common.util.consts import PARSE_ERROR_FAIL_FLAG
 from checkov.runner_filter import RunnerFilter
 from checkov.serverless.checks.function.base_function_check import BaseFunctionCheck
 from checkov.serverless.runner import Runner
@@ -144,6 +147,41 @@ def test_record_relative_path_with_abs_file(self):
             # no need to join with a '/' because the CFN runner adds it to the start of the file path
             self.assertEqual(record.repo_file_path, f'/{file_rel_path}')
 
+    def test_unparsable_file_is_reported_as_parsing_error(self):
+        # a single '=' is a valid YAML value, but the loader can't construct it,
+        # so the file used to be skipped without any indication in the report
+        invalid_template = (
+            "provider:\n"
+            "  name: aws\n"
+            "  tags:\n"
+            "    test: =\n"
+            "functions:\n"
+            "  hello:\n"
+            "    handler: handler.hello\n"
+        )
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            scan_file_path = os.path.join(tmp_dir, "serverless.yml")
+            with open(scan_file_path, "w") as f:
+                f.write(invalid_template)
+
+            report = Runner().run(
+                root_folder=None, files=[scan_file_path], external_checks_dir=None,
+                runner_filter=RunnerFilter(framework=['serverless'])
+            )
+
+        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_wrong_check_imports(self):
         wrong_imports = ["arm", "cloudformation", "dockerfile", "helm", "kubernetes", "terraform"]
         check_imports = []