feat(serverless): Serverless graph vertices (#6894)

* add vertices implementation

checkov/serverless/graph_builder/definition_context.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def build_definitions_context(definitions: dict[str, dict[str, Any]], definitions_raw: dict[str, list[tuple[int, str]]]
+                              ) -> dict[str, dict[str, Any]]:
+    return {}
+
+
+def add_resource_to_definitions_context(definitions_context: dict[str, dict[str, Any]], resource_key: str,
+                                        resource_attributes: dict[str, Any], definition_attribute: str,
+                                        definitions_raw: dict[str, Any], file_path: str) -> None:
+    pass

checkov/serverless/graph_builder/graph_to_definitions.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any, TYPE_CHECKING
+
+from checkov.serverless.utils import ServerlessElements
+
+if TYPE_CHECKING:
+    from checkov.serverless.graph_builder.graph_components.blocks import ServerlessBlock
+
+
+def convert_graph_vertices_to_definitions(vertices: list[ServerlessBlock], root_folder: str | Path | None) \
+        -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
+    serverless_definitions: dict[str, dict[str, Any]] = {}
+    breadcrumbs: dict[str, dict[str, Any]] = {}
+    for vertex in vertices:
+        block_path = vertex.path
+        element_name = vertex.name.split('.')[-1]
+        # Plugins section is formatted as a list
+        if vertex.block_type == ServerlessElements.PLUGINS:
+            serverless_definitions.setdefault(block_path, {}).setdefault(vertex.block_type, []).append(element_name)
+
+        # If there is a ket named 'value' in the config it means that
+        # this vertex's config contains only a single string
+        elif 'value' in vertex.config:
+            # If the vertex is provider or service and it only contains a string the section should look like:
+            # provider: <value>
+            # service: <value>
+            if element_name == ServerlessElements.PROVIDER or element_name == ServerlessElements.SERVICE:
+                serverless_definitions.setdefault(block_path, {})[vertex.block_type] = vertex.config['value']
+
+            # Otherwise it's a vertex of a specific nested attribute and need to include the full path
+            # Examples:
+            # provider:
+            #   runtime: nodejs20.x
+            # custom:
+            #   myCustomVar: value
+            else:
+                serverless_definitions.setdefault(block_path, {}).setdefault(vertex.block_type, {})[element_name] = \
+                    vertex.config['value']
+
+        # Otherwise, the vertex config is a dict
+        else:
+            serverless_definitions.setdefault(block_path, {}).setdefault(vertex.block_type, {})[
+                element_name] = vertex.config
+
+        if vertex.breadcrumbs:
+            relative_block_path = f"/{os.path.relpath(block_path, root_folder)}"
+            add_breadcrumbs(vertex, breadcrumbs, relative_block_path)
+    return serverless_definitions, breadcrumbs
+
+
+def add_breadcrumbs(vertex: ServerlessBlock, breadcrumbs: dict[str, dict[str, Any]], relative_block_path: str) -> None:
+    breadcrumbs.setdefault(relative_block_path, {})[vertex.name] = vertex.breadcrumbs

checkov/serverless/graph_builder/local_graph.py

@@ -2,17 +2,99 @@
 
 from typing import Any
 
+from checkov.common.graph.graph_builder import CustomAttributes
 from checkov.common.graph.graph_builder.local_graph import LocalGraph, _Block
+from checkov.common.util.consts import LINE_FIELD_NAMES
+from checkov.common.util.data_structures_utils import pickle_deepcopy
 from checkov.serverless.graph_builder.graph_components.blocks import ServerlessBlock
+from checkov.serverless.utils import ServerlessElements
 
 
 class ServerlessLocalGraph(LocalGraph[ServerlessBlock]):
     def __init__(self, definitions: dict[str, dict[str, Any]]) -> None:
         super().__init__()
         self.vertices: list[ServerlessBlock] = []
         self.definitions = definitions
-        self.vertices_by_path_and_id: dict[tuple[str, str], int] = {}
-        self.vertices_by_name: dict[str, int] = {}
+        self.vertices_by_path_and_name: dict[tuple[str, str], int] = {}
+
+    def build_graph(self, render_variables: bool = True) -> None:
+        self._create_vertices()
+
+    def _create_vertices(self) -> None:
+        for file_path, definition in self.definitions.items():
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.FUNCTIONS)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.PARAMS)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.PROVIDER)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.LAYERS)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.CUSTOM)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.PACKAGE)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.PLUGINS)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.SERVICE)
+            self._create_vertex(file_path=file_path, definition=definition, element_type=ServerlessElements.RESOURCES)
+
+        for i, vertex in enumerate(self.vertices):
+            self.vertices_by_block_type[vertex.block_type].append(i)
+            self.vertices_by_path_and_name[(vertex.path, vertex.name)] = i
+
+            self.in_edges[i] = []
+            self.out_edges[i] = []
+
+    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)
+        if not resources:
+            return
+
+        elif isinstance(resources, str):
+            self.vertices.append(ServerlessBlock(
+                name=f'{element_type}',
+                config={"value": pickle_deepcopy(resources)},
+                path=file_path,
+                block_type=element_type,
+                attributes={"value": pickle_deepcopy(resources)},
+                id=f"{file_path}:{element_type}"
+            ))
+
+        else:
+            for attribute in resources:
+                if attribute in LINE_FIELD_NAMES:
+                    continue
+
+                if isinstance(resources, list):
+                    full_conf = {"value": pickle_deepcopy(attribute)}
+                    self.vertices.append(ServerlessBlock(
+                        name=f'{element_type}.{attribute}',
+                        config=full_conf,
+                        path=file_path,
+                        block_type=element_type,
+                        attributes=full_conf,
+                        id=f"{file_path}:{element_type}.{attribute}"
+                    ))
+
+                else:
+                    attribute_value = resources[attribute]
+                    if not isinstance(attribute_value, dict):
+                        full_conf = {"value": pickle_deepcopy(attribute_value)}
+                    else:
+                        full_conf = attribute_value
+
+                    config = pickle_deepcopy(full_conf)
+
+                    resource_type = element_type
+
+                    attributes = pickle_deepcopy(config)
+                    attributes[CustomAttributes.RESOURCE_TYPE] = resource_type
+
+                    self.vertices.append(ServerlessBlock(
+                        name=f'{resource_type}.{attribute}',
+                        config=config,
+                        path=file_path,
+                        block_type=resource_type,
+                        attributes=attributes,
+                        id=f"{file_path}:{resource_type}.{attribute}"
+                    ))
 
     def get_resources_types_in_graph(self) -> list[str]:
         # not used
@@ -25,6 +107,3 @@ def update_vertex_config(vertex: _Block, changed_attributes: list[str] | dict[st
 
     def update_vertices_configs(self) -> None:
         pass
-
-    def build_graph(self, render_variables: bool) -> None:
-        pass

checkov/serverless/runner.py

@@ -11,6 +11,8 @@
 from checkov.common.bridgecrew.check_type import CheckType
 from checkov.common.util.secrets import omit_secret_value_from_checks
 from checkov.serverless.base_registry import EntityDetails
+from checkov.serverless.graph_builder.definition_context import build_definitions_context
+from checkov.serverless.graph_builder.graph_to_definitions import convert_graph_vertices_to_definitions
 from checkov.serverless.graph_builder.local_graph import ServerlessLocalGraph
 from checkov.serverless.graph_manager import ServerlessGraphManager
 from checkov.serverless.parsers.context_parser import ContextParser as SlsContextParser, ContextParser
@@ -111,6 +113,17 @@ def run(
             # Filter out empty files that have not been parsed successfully
             self.definitions = {k: v for k, v in definitions.items() if v}
             self.definitions_raw = {k: v for k, v in definitions_raw.items() if k in definitions.keys()}
+            self.context = build_definitions_context(definitions=self.definitions, definitions_raw=self.definitions_raw)
+
+            logging.info("Creating Serverless graph")
+            local_graph = self.graph_manager.build_graph_from_definitions(definitions=self.definitions)
+            logging.info("Successfully created Serverless graph")
+
+            self.graph_manager.save_graph(local_graph)
+            self.definitions, self.breadcrumbs = convert_graph_vertices_to_definitions(
+                vertices=local_graph.vertices,
+                root_folder=root_folder,
+            )
 
         self.pbar.initiate(len(self.definitions))
 
@@ -140,34 +153,34 @@ def complete_python_checks(self,
         # "Complete" checks
         # NOTE: Ignore code content, no point in showing (could be long)
         file_abs_path = Path(sls_file).absolute()
-        entity_lines_range, entity_code_lines = sls_context_parser.extract_code_lines(sls_file_data)
-        if entity_lines_range:
-            skipped_checks = CfnContextParser.collect_skip_comments(entity_code_lines or [])
-            variable_evaluations: dict[str, Any] = {}
-            entity = EntityDetails(sls_context_parser.provider_type, sls_file_data)
-            results = complete_registry.scan(sls_file, entity, skipped_checks, runner_filter)
-            tags = cfn_utils.get_resource_tags(entity, complete_registry)  # type:ignore[arg-type]
-            if results:
-                for check, check_result in results.items():
-                    record = Record(check_id=check.id, check_name=check.name, check_result=check_result,
-                                    code_block=[],  # Don't show, could be large
-                                    file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
-                                    file_line_range=entity_lines_range,
-                                    resource="complete",  # Weird, not sure what to put where
-                                    evaluations=variable_evaluations,
-                                    check_class=check.__class__.__module__,
-                                    file_abs_path=str(file_abs_path),
-                                    entity_tags=tags, severity=check.severity)
-                    record.set_guideline(check.guideline)
-                    report.add_record(record=record)
-            else:
-                report.extra_resources.add(
-                    ExtraResource(
-                        file_abs_path=str(file_abs_path),
-                        file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
-                        resource="complete",
-                    )
+        entity_code_lines = self.definitions_raw[sls_file]
+        entity_lines_range = [1, len(entity_code_lines) - 1]
+        skipped_checks = CfnContextParser.collect_skip_comments(entity_code_lines or [])
+        variable_evaluations: dict[str, Any] = {}
+        entity = EntityDetails(sls_context_parser.provider_type, sls_file_data)
+        results = complete_registry.scan(sls_file, entity, skipped_checks, runner_filter)
+        tags = cfn_utils.get_resource_tags(entity, complete_registry)  # type:ignore[arg-type]
+        if results:
+            for check, check_result in results.items():
+                record = Record(check_id=check.id, check_name=check.name, check_result=check_result,
+                                code_block=[],  # Don't show, could be large
+                                file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
+                                file_line_range=entity_lines_range,
+                                resource="complete",  # Weird, not sure what to put where
+                                evaluations=variable_evaluations,
+                                check_class=check.__class__.__module__,
+                                file_abs_path=str(file_abs_path),
+                                entity_tags=tags, severity=check.severity)
+                record.set_guideline(check.guideline)
+                report.add_record(record=record)
+        else:
+            report.extra_resources.add(
+                ExtraResource(
+                    file_abs_path=str(file_abs_path),
+                    file_path=self.extract_file_path_from_abs_path(Path(sls_file)),
+                    resource="complete",
                 )
+            )
 
     def single_item_sections_checks(self,
                                     sls_file: str,
@@ -183,7 +196,8 @@ def single_item_sections_checks(self,
                 continue
             entity_lines_range, entity_code_lines = sls_context_parser.extract_code_lines(item_content)
             if not entity_lines_range:
-                entity_lines_range, entity_code_lines = sls_context_parser.extract_code_lines(sls_file_data)
+                entity_code_lines = self.definitions_raw[sls_file]
+                entity_lines_range = [1, len(entity_code_lines) - 1]
 
             skipped_checks = CfnContextParser.collect_skip_comments(entity_code_lines or [])
             variable_evaluations: dict[str, Any] = {}

checkov/serverless/utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from enum import Enum
 from typing import Callable, Any
 
 from checkov.common.parallelizer.parallel_runner import parallel_runner
@@ -11,6 +12,22 @@
     "CKV_SLS_FILE_MASK", "serverless.yml,serverless.yaml").split(",")
 
 
+class ServerlessElements(str, Enum):
+    PARAMS = "params"
+    FUNCTIONS = "functions"
+    PROVIDER = "provider"
+    LAYERS = "layers"
+    CUSTOM = "custom"
+    PACKAGE = "package"
+    PLUGINS = "plugins"
+    SERVICE = "service"
+    RESOURCES = "resources"
+
+    def __str__(self) -> str:
+        # needed, because of a Python 3.11 change
+        return self.value
+
+
 def get_scannable_file_paths(root_folder: str | None = None, excluded_paths: list[str] | None = None) -> list[str]:
     files_list: list[str] = []
 

tests/serverless/graph_builder/resources/serverless.yaml

@@ -0,0 +1,62 @@
+service: acme-service
+
+frameworkVersion: "^2.30.0"
+
+plugins:
+  - serverless-vpc-discovery
+
+custom:
+  resources: ${file(./serverless.${opt:stage}.yml)}
+  vpc:
+    vpcName: acme-vpc
+    subnetNames:
+      - private-subnet-a
+      - private-subnet-b
+    securityGroupNames:
+      - allow_out_to_vpc_endpoints
+      - lambda_sg
+provider:
+  region: us-east-1
+  name: aws
+  runtime: python3.7
+  tracing:
+    lambda: true
+  deploymentBucket: ${self:custom.resources.deploymentBucket}
+  environment: ${self:custom.resources.environment}
+  iamManagedPolicies:
+    - arn:aws:iam::aws:policy/ReadOnlyAccess
+  iamRoleStatements: ${self:custom.resources.iamRoleStatements}
+  ecr:
+    images:
+      base:
+        path: ../../
+        file: ./path/to/Dockerfile
+
+functions:
+  acmeFunc:
+    image: base
+    timeout: 120
+    memorySize: 3000
+    environment:
+      EFS_MOUNT_PATH: ${self:custom.localMountPath}
+    fileSystemConfig:
+      localMountPath: ${self:custom.localMountPath}
+      arn: 'arn:aws:elasticfilesystem:${self:provider.region}:#{AWS::AccountId}:access-point/${self:custom.resources.efsAccessPoint}'
+    events:
+      - sqs:
+          arn: arn:aws:sqs:#{AWS::Region}:#{AWS::AccountId}:job_queue
+          batchSize: 10
+          maximumBatchingWindow: 0
+  acmeFunc2:
+    image: base
+    environment:
+      EFS_MOUNT_PATH: ${self:custom.localMountPath}
+      CLEAN_UP_BEFORE_PROCESS: 'true'
+    timeout: 900
+    memorySize: 9000
+    fileSystemConfig:
+      localMountPath: ${self:custom.localMountPath}
+      arn: 'arn:aws:elasticfilesystem:${self:provider.region}:#{AWS::AccountId}:access-point/${self:custom.resources.efsAccessPoint}'
+
+resources:
+  Resources: ${file(./serverless.${opt:stage}.yml)}  # just shouldn't raise an exception