Merge branch 'main' into new-checks-dec24

checkov/serverless/graph_builder/definition_context.py

@@ -1,14 +1,71 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import cast, Any
+
+from checkov.common.parsers.node import StrNode, ListNode
+from checkov.common.util.consts import START_LINE, END_LINE, LINE_FIELD_NAMES
+from checkov.common.util.suppression import collect_suppressions_for_report
+from checkov.serverless.utils import ServerlessElements
 
 
 def build_definitions_context(definitions: dict[str, dict[str, Any]], definitions_raw: dict[str, list[tuple[int, str]]]
                               ) -> dict[str, dict[str, Any]]:
-    return {}
+    definitions_context: dict[str, dict[str, Any]] = {}
+    for file_path, file_definitions in definitions.items():
+        definitions_context[file_path] = {}
+        for definition_attribute, definition_value in file_definitions.items():
+            if definition_attribute not in [str(e) for e in ServerlessElements]:
+                continue
+            definitions_context[file_path][definition_attribute] = {}
+            if isinstance(definition_value, dict):
+                for resource_key, resource_attributes in definition_value.items():
+                    add_resource_to_definitions_context(definitions_context, resource_key, resource_attributes,
+                                                        definition_attribute, definitions_raw, file_path)
+            elif isinstance(definition_value, list):
+                for resource in definition_value:
+                    add_resource_to_definitions_context(definitions_context, '', resource, definition_attribute,
+                                                        definitions_raw, file_path)
+
+            elif isinstance(definition_value, StrNode):
+                add_resource_to_definitions_context(definitions_context, definition_attribute, definition_value,
+                                                    definition_attribute,
+                                                    definitions_raw, file_path)
+    return definitions_context
 
 
 def add_resource_to_definitions_context(definitions_context: dict[str, dict[str, Any]], resource_key: str,
-                                        resource_attributes: dict[str, Any], definition_attribute: str,
+                                        resource_attributes: dict[str, Any] | ListNode | StrNode, definition_attribute: str,
                                         definitions_raw: dict[str, Any], file_path: str) -> None:
-    pass
+    if resource_key in LINE_FIELD_NAMES:
+        return
+
+    if isinstance(resource_attributes, dict):
+        start_line = resource_attributes[START_LINE] - 1
+        end_line = resource_attributes[END_LINE] - 1
+
+    elif isinstance(resource_attributes, ListNode):
+        start_line = resource_attributes.start_mark.line
+        end_line = resource_attributes.end_mark.line
+
+    elif isinstance(resource_attributes, StrNode):
+        start_line = resource_attributes.start_mark.line + 1
+        end_line = resource_attributes.end_mark.line + 1
+
+    else:
+        return
+
+    definition_resource = {"start_line": start_line, "end_line": end_line}
+
+    if resource_key is None and isinstance(resource_attributes, dict):
+        resource_key = f"{resource_attributes.get('type')}.{resource_attributes.get('name')}"
+    int_start_line = cast(int, definition_resource["start_line"])
+    int_end_line = cast(int, definition_resource["end_line"])
+    code_lines_for_suppressions_check = definitions_raw[file_path][int_start_line: int_end_line]
+    definition_resource['skipped_checks'] = collect_suppressions_for_report(
+        code_lines=code_lines_for_suppressions_check)
+    if isinstance(resource_attributes, dict) and 'type' in resource_attributes:
+        definition_resource["type"] = resource_attributes.get('type')
+
+    definition_resource["code_lines"] = definitions_raw[file_path][start_line - 1: end_line]
+
+    definitions_context[file_path][definition_attribute][resource_key] = definition_resource

checkov/serverless/graph_manager.py

@@ -12,7 +12,7 @@
 
 
 class ServerlessGraphManager(GraphManager[ServerlessLocalGraph, "dict[str, dict[str, Any]]"]):
-    def __init__(self, db_connector: LibraryGraphConnector, source: str = GraphSource.ARM) -> None:
+    def __init__(self, db_connector: LibraryGraphConnector, source: str = GraphSource.SERVERLESS) -> None:
         super().__init__(db_connector=db_connector, parser=None, source=source)
 
     def build_graph_from_source_directory(

checkov/serverless/runner.py

@@ -355,3 +355,6 @@ def cfn_resources_checks(self,
 
     def extract_file_path_from_abs_path(self, path: Path) -> str:
         return f"{os.path.sep}{os.path.relpath(path, self.root_folder)}"
+
+    def set_definitions_raw(self, definitions_raw: dict[str, list[tuple[int, str]]]) -> None:
+        self.definitions_raw = definitions_raw

checkov/serverless/utils.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
 import os
+from collections.abc import Collection
 from enum import Enum
 from typing import Callable, Any
+from pathlib import Path
 
 from checkov.common.parallelizer.parallel_runner import parallel_runner
+from checkov.runner_filter import RunnerFilter
 from checkov.serverless.parsers.parser import parse
 from checkov.common.runners.base_runner import filter_ignored_paths
 
@@ -28,6 +31,22 @@ def __str__(self) -> str:
         return self.value
 
 
+def create_definitions(
+    root_folder: str,
+    files: Collection[Path] | None = None,
+    runner_filter: RunnerFilter | None = None,
+) -> tuple[dict[str, dict[str, Any]], dict[str, list[tuple[int, str]]]]:
+    definitions: dict[str, dict[str, Any]] = {}
+    definitions_raw: dict[str, list[tuple[int, str]]] = {}
+    runner_filter = runner_filter or RunnerFilter()
+
+    if root_folder:
+        file_paths = get_scannable_file_paths(root_folder, runner_filter.excluded_paths)
+        definitions, definitions_raw = get_files_definitions(files=file_paths)
+
+    return definitions, definitions_raw
+
+
 def get_scannable_file_paths(root_folder: str | None = None, excluded_paths: list[str] | None = None) -> list[str]:
     files_list: list[str] = []
 

checkov/version.py

@@ -1 +1 @@
-version = '3.2.339'
+version = '3.2.341'

kubernetes/requirements.txt

@@ -1 +1 @@
-checkov==3.2.339
+checkov==3.2.341