fix(serverless): Enhance yaml parsing, better support for file expansions

- Improve yaml parsing wrt file() expansion
- Add tests for file() expansion
- Add a '__file__' marker attribute to yaml nodes
- Utilize the '__file__' marker when generating reports for serverless
- Raise CfnParseError on circular inclusions
- Added a ``logger.error`` that logs specifics on pyyaml parse errors
- Updated the serverless graph builder to cope with file() expansion

Fixes #7006

Author: thentenaar

checkov/cloudformation/cfn_utils.py

@@ -260,6 +260,9 @@ def enrich_resources_with_globals(original_template: dict[str, Any]) -> dict[str
 
         # Iterate over the resources in the template copy
         for _resource_name, resource_details in new_template.get('Resources', {}).items():
+            if _resource_name == '__file__':
+                continue
+
             resource_type = resource_details.get('Type', '')
             if (resource_type not in supported_types_and_globals):
                 continue

checkov/cloudformation/parser/__init__.py

@@ -30,16 +30,15 @@ def parse(
 
     try:
         (template, template_lines) = cfn_yaml.load(filename, cfn_yaml.ContentType.CFN)
-    except IOError as err:
-        if err.errno == 2:
-            error = f"Template file not found: {filename} - {err}"
-            LOGGER.error(error)
-        elif err.errno == 21:
-            error = f"Template references a directory, not a file: {filename} - {err}"
-            LOGGER.error(error)
-        elif err.errno == 13:
-            error = f"Permission denied when accessing template file: {filename} - {err}"
-            LOGGER.error(error)
+    except FileNotFoundError as e:
+        error = f'Template file not found: {e.filename}'
+        LOGGER.error(error)
+    except IsADirectoryError as e:
+        error = f'Template references a directory, not a file: {e.filename}'
+        LOGGER.error(error)
+    except PermissionError as e:
+        error = f'Permission denied when accessing {e.filename}'
+        LOGGER.error(error)
     except UnicodeDecodeError as err:
         error = f"Cannot read file contents: {filename} - {err}"
         LOGGER.error(error)
@@ -81,6 +80,8 @@ def parse(
     if isinstance(template, dict):
         resources = template.get(TemplateSections.RESOURCES.value, None)
         if resources and isinstance(resources, dict):
+            if '__file__' in resources:
+                del resources['__file__']
             if "__startline__" in resources:
                 del resources["__startline__"]
             if "__endline__" in resources:

checkov/cloudformation/parser/cfn_yaml.py

@@ -7,20 +7,19 @@
 import json
 import logging
 import platform
+import re
 from collections.abc import Hashable
 from enum import Enum
 from pathlib import Path
-from typing import Any, TYPE_CHECKING, NoReturn, Callable
+from typing import Any, TYPE_CHECKING, NoReturn, Callable, cast
 
-from yaml import MappingNode
-from yaml import ScalarNode
-from yaml import SequenceNode
+from yaml import MappingNode, ScalarNode, SequenceNode
 from yaml.composer import Composer
-from yaml.constructor import ConstructorError
-from yaml.constructor import SafeConstructor
+from yaml.constructor import ConstructorError, SafeConstructor
 from yaml.reader import Reader
 from yaml.resolver import Resolver
 from yaml.scanner import Scanner
+from yaml.error import MarkedYAMLError, YAMLError
 from charset_normalizer import from_path
 
 from checkov.common.parsers.json.decoder import SimpleDecoder
@@ -98,6 +97,7 @@ def __init__(self, filename: str, content_type: ContentType | None = None) -> No
                 NodeConstructor.construct_yaml_null_error,
             )
         self.filename = filename
+        self.files_loaded: dict[Path, bool] = {}
 
     # To support lazy loading, the original constructors first yield
     # an empty object, then fill them in when iterated. Due to
@@ -142,15 +142,52 @@ def construct_yaml_str(self, node: ScalarNode) -> StrNode:
         assert isinstance(obj, str)  # nosec
         return StrNode(obj, node.start_mark, node.end_mark)
 
+    def mark_with_filename(self, root: Node | None, filename: str) -> None:
+        if not root:
+            return
+
+        setattr(root, 'filename', filename)  # noqa: B010
+        if isinstance(root, SequenceNode):
+            for v in root.value:
+                self.mark_with_filename(v, filename)
+        if isinstance(root, MappingNode):
+            for k, v in root.value:
+                self.mark_with_filename(k, filename)
+                self.mark_with_filename(v, filename)
+
     def construct_yaml_seq(self, node: SequenceNode) -> ListNode:
+        # Handle serverless file() expansions on SequenceNode
+        if isinstance(node.value, list) and len(node.value) > 0:
+            for i, v in enumerate(node.value):
+                if not isinstance(v, ScalarNode) or not isinstance(node.value[i].value, str):
+                    continue
+
+                m = re.match(r'\$\{file\((.+\.ya?ml)\)\}$', v.value)
+                if m is None:
+                    continue
+
+                path = (Path(self.filename).parent / m[1]).resolve()
+                if path in self.files_loaded:
+                    raise CfnParseError(
+                        filename=node.filename if hasattr(node, 'filename') else self.filename,
+                        message=f'Circular include of {m[1]}',
+                        line_number=node.start_mark.line,
+                        column_number=node.start_mark.column
+                    )
+                else:
+                    self.files_loaded[path] = True
+                    content = read_file_with_any_encoding(file_path=path)
+                    node.value[i] = MarkedLoader(content, m[1], None).get_single_node()
+                    self.mark_with_filename(node.value[i], m[1])
+
         obj, = SafeConstructor.construct_yaml_seq(self, node)  # type:ignore[no-untyped-call]
         assert isinstance(obj, list)  # nosec
         return ListNode(obj, node.start_mark, node.end_mark)  # nosec
 
     def construct_yaml_null_error(self, node: Node) -> NoReturn:
         """Throw a null error"""
         raise CfnParseError(
-            filename=self.filename,
+            filename=node.filename if hasattr(node, 'filename') else self.filename,
             message=f"Null value at line {node.start_mark.line + 1} column {node.start_mark.column + 1}",
             line_number=node.start_mark.line,
             column_number=node.start_mark.column,
@@ -179,7 +216,7 @@ def __init__(self, stream: str, filename: str, content_type: ContentType | None
     def construct_mapping(self, node: MappingNode, deep: bool = False) -> dict[Hashable, Any]:
         mapping = super(MarkedLoader, self).construct_mapping(node, deep=deep)
         # Add 1 so line numbering starts at 1
-        # mapping['__line__'] = node.start_mark.line + 1
+        mapping['__file__'] = node.filename if hasattr(node, 'filename') else self.filename
         mapping['__startline__'] = node.start_mark.line + 1
         mapping['__endline__'] = node.end_mark.line + 1
         return mapping
@@ -199,7 +236,7 @@ def multi_constructor(loader: MarkedLoader, tag_suffix: str, node: ScalarNode) -
         constructor = construct_getatt
     elif tag_suffix == "Ref" and (isinstance(node.value, list) or isinstance(node.value, dict)):
         raise CfnParseError(
-            filename="",
+            filename=node.filename if hasattr(node, 'filename') else loader.filename,
             message='Invalid !Ref: {}'.format(node.value),
             line_number=0,
             column_number=0)
@@ -232,18 +269,30 @@ def loads(yaml_string: str, fname: str, content_type: ContentType | None = None)
     """
     Load the given YAML string
     """
+    if len(yaml_string) == 0:
+        return {}
+
     loader = MarkedLoader(yaml_string, fname, content_type)
     loader.add_multi_constructor('!', multi_constructor)  # type:ignore[no-untyped-call]
 
-    template: "DictNode | dict[str, Any]" = loader.get_single_data()
-    # Convert an empty file to an empty dict
-    if template is None:
-        template = {}
-
-    return template
+    try:
+        return cast(DictNode | dict[str, Any], loader.get_single_data())
+    except MarkedYAMLError as e:
+        logging.error(f'YAML error parsing {fname}: {e}')
+        if e.problem and e.problem_mark:
+            raise CfnParseError(
+                filename=fname,
+                message=e.problem,
+                line_number=e.problem_mark.line,
+                column_number=e.problem_mark.column) from e
+        else:
+            raise CfnParseError(filename=fname, message=str(e), line_number=0, column_number=0) from e
+    except YAMLError as e:
+        logging.error(f'YAML error parsing {fname}: {e}')
+        raise CfnParseError(filename=fname, message=str(e), line_number=0, column_number=0) from e
 
 
-def load(filename: str | Path, content_type: ContentType) -> tuple[dict[str, Any], list[tuple[int, str]]]:
+def load(filename: str | Path, content_type: ContentType | None) -> tuple[dict[str, Any], list[tuple[int, str]]]:
     """
     Load the given YAML file
     """

checkov/common/util/consts.py

@@ -4,7 +4,8 @@
 RESOLVED_MODULE_ENTRY_NAME = "__resolved__"
 START_LINE = '__startline__'
 END_LINE = '__endline__'
-LINE_FIELD_NAMES = {START_LINE, END_LINE}
+FILE = '__file__'
+LINE_FIELD_NAMES = {START_LINE, END_LINE, FILE}
 TRUE_AFTER_UNKNOWN = 'true_after_unknown'
 
 DEV_API_GET_HEADERS = {

checkov/serverless/graph_builder/local_graph.py

@@ -43,7 +43,20 @@ def _create_vertex(self, file_path: str, definition: dict[str, Any] | None,
                        element_type: ServerlessElements) -> None:
         if not definition:
             return
+
         resources = definition.get(element_type)
+
+        # resources -> Resources
+        if element_type == ServerlessElements.RESOURCES and resources is None:
+            resources = definition.get('Resources')
+
+        if isinstance(resources, list) and len(resources) > 0 and \
+           isinstance(resources[0], dict) and resources[0]['__file__'] != file_path:
+            for r in resources:
+                if isinstance(r, dict):
+                    self._create_vertex(file_path, {element_type: r}, element_type)
+            return
+
         if not resources:
             return
 

checkov/serverless/parsers/context_parser.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+from pathlib import Path
 from typing import Any
 
 from checkov.serverless.parsers.parser import FUNCTIONS_TOKEN, PROVIDER_TOKEN, IAM_ROLE_STATEMENTS_TOKEN, \
     ENVIRONMENT_TOKEN, STACK_TAGS_TOKEN, TAGS_TOKEN
 from checkov.cloudformation.context_parser import ContextParser as CfnContextParser, STARTLINE, ENDLINE
+from checkov.common.util.file_utils import read_file_with_any_encoding
 
 
 class ContextParser(object):
@@ -30,6 +32,11 @@ def __init__(self, sls_file: str, sls_template: dict[str, Any], sls_template_lin
         self.functions_conf = sls_template.get(FUNCTIONS_TOKEN) or {}
         self.provider_type = self._infer_provider_type()
 
+    def file(self, content: dict[str, Any]) -> str:
+        if isinstance(content, dict):
+            return str(content.get('__file__', self.sls_file))
+        return self.sls_file
+
     def extract_code_lines(
         self, content: dict[str, Any]
     ) -> tuple[list[int], list[tuple[int, str]]] | tuple[None, None]:
@@ -40,7 +47,14 @@ def extract_code_lines(
 
             entity_lines_range = [start_line, end_line - 1]
 
-            entity_code_lines = self.sls_template_lines[start_line - 1: end_line - 1]
+            fname = self.file(content)
+            lines = self.sls_template_lines
+            if fname != self.sls_file:
+                lines = []
+                text = read_file_with_any_encoding(Path(self.sls_file).parent / fname)
+                for i, ln in enumerate(text.splitlines(True)):
+                    lines.append((i + 1, ln))
+            entity_code_lines = lines[start_line - 1: end_line - 1]
             return entity_lines_range, entity_code_lines
         return None, None
 

checkov/serverless/parsers/parser.py

@@ -11,7 +11,6 @@
 import re
 
 import yaml
-from yaml import YAMLError
 
 from checkov.cloudformation.parser import cfn_yaml
 from checkov.cloudformation.context_parser import ContextParser
@@ -39,37 +38,31 @@
 def parse(filename: str) -> tuple[dict[str, Any], list[tuple[int, str]]] | None:
     template = None
     template_lines = None
+
     try:
         (template, template_lines) = cfn_yaml.load(filename, cfn_yaml.ContentType.SLS)
         if not template or not is_checked_sls_template(template):
             return None
-    except IOError as e:
-        if e.errno == 2:
-            logger.error('Template file not found: %s', filename)
-            return None
-        elif e.errno == 21:
-            logger.error('Template references a directory, not a file: %s',
-                         filename)
-            return None
-        elif e.errno == 13:
-            logger.error('Permission denied when accessing template file: %s',
-                         filename)
-            return None
+    except FileNotFoundError as e:
+        logger.error(f'Template file not found: {e.filename}')
+        return None
+    except IsADirectoryError as e:
+        logger.error(f'Template references a directory, not a file: {e.filename}')
+        return None
+    except PermissionError as e:
+        logger.error(f'Permission denied when accessing {e.filename}')
+        return None
     except UnicodeDecodeError:
         logger.error('Cannot read file contents: %s', filename)
         return None
-    except CfnParseError:
-        logger.warning(f"Failed to parse file {filename} because it isn't a valid template")
-        return None
-    except YAMLError:
-        logger.warning(f"Failed to parse file {filename} as a yaml")
+    except CfnParseError as e:
+        logger.warning(f"Failed to parse file {e.filename} because it isn't valid yaml")
         return None
 
     if template is None or template_lines is None:
         return None
 
     process_variables(template, filename)
-
     return template, template_lines
 
 

checkov/serverless/runner.py

@@ -117,7 +117,7 @@ def run(
 
             logging.info("Creating Serverless graph")
             local_graph = self.graph_manager.build_graph_from_definitions(definitions=self.definitions)
-            logging.info("Successfully created Serverless graph")
+            logging.info(f'Successfully created Serverless graph ({len(local_graph.vertices)} vertices)')
 
             self.graph_manager.save_graph(local_graph)
             self.definitions, self.breadcrumbs = convert_graph_vertices_to_definitions(
@@ -204,6 +204,8 @@ def single_item_sections_checks(self,
             entity = EntityDetails(sls_context_parser.provider_type, item_content)
             results = registry.scan(sls_file, entity, skipped_checks, runner_filter)
             tags = get_resource_tags(entity, registry)
+            fname = Path(sls_context_parser.file(item_content)).resolve()
+
             if results:
                 for check, check_result in results.items():
                     censored_code_lines = omit_secret_value_from_checks(
@@ -218,7 +220,7 @@ def single_item_sections_checks(self,
                         check_name=check.name,
                         check_result=check_result,
                         code_block=censored_code_lines,
-                        file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
+                        file_path=self.extract_file_path_from_abs_path(fname),
                         file_line_range=entity_lines_range or [0, 0],
                         resource=token,
                         evaluations=variable_evaluations,
@@ -265,6 +267,7 @@ def multi_item_sections_checks(self,
                     entity = EntityDetails(sls_context_parser.provider_type, item_content)
                     results = registry.scan(sls_file, entity, skipped_checks, runner_filter)
                     tags = get_resource_tags(entity, registry)
+                    fname = Path(sls_context_parser.file(item_content)).resolve()
                     if results:
                         for check, check_result in results.items():
                             censored_code_lines = omit_secret_value_from_checks(
@@ -276,7 +279,7 @@ def multi_item_sections_checks(self,
                             )
                             record = Record(check_id=check.id, check_name=check.name, check_result=check_result,
                                             code_block=censored_code_lines,
-                                            file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
+                                            file_path=self.extract_file_path_from_abs_path(fname),
                                             file_line_range=entity_lines_range,
                                             resource=item_name, evaluations=variable_evaluations,
                                             check_class=check.__class__.__module__,

tests/cloudformation/parser/cfn_file.yaml

@@ -0,0 +1,13 @@
+---
+service: api-services
+provider:
+  name: aws
+  stage: ${sls:stage}
+  runtime: nodejs20.x
+  region: 'us-east-1'
+  iamManagedPolicies:
+    - 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
+
+resources:
+  - ${file(./cfn_file_resources.yaml)}
+

tests/cloudformation/parser/cfn_file_circular.yaml

@@ -0,0 +1,13 @@
+---
+service: api-services
+provider:
+  name: aws
+  stage: ${sls:stage}
+  runtime: nodejs20.x
+  region: 'us-east-1'
+  iamManagedPolicies:
+    - 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
+
+resources:
+  - ${file(./cfn_file_circular.yaml)}
+

tests/cloudformation/parser/cfn_file_resources.yaml

@@ -0,0 +1,8 @@
+---
+Resources:
+  MyBucket:
+    Type: AWS::S3::Bucket
+    Properties:
+      BucketName: my-bucket
+      AccessControl: PublicRead
+