Merge branch 'bridgecrewio:main' into ACRAnonymousPullDisabled

CHANGELOG.md

@@ -1,6 +1,17 @@
 # CHANGELOG
 
-## [Unreleased](https://github.com/bridgecrewio/checkov/compare/3.2.108...HEAD)
+## [Unreleased](https://github.com/bridgecrewio/checkov/compare/3.2.112...HEAD)
+
+## [3.2.112](https://github.com/bridgecrewio/checkov/compare/3.2.108...3.2.112) - 2024-05-29
+
+### Feature
+
+- **terraform:** Add provider address to resources - [#6266](https://github.com/bridgecrewio/checkov/pull/6266)
+- **terraform:** Support for count & for_each in data blocks - [#6359](https://github.com/bridgecrewio/checkov/pull/6359)
+
+### Bug Fix
+
+- **terraform:** Fix an issue for loading tfvars + issue in the dynamic rendering - [#6360](https://github.com/bridgecrewio/checkov/pull/6360)
 
 ## [3.2.108](https://github.com/bridgecrewio/checkov/compare/3.2.107...3.2.108) - 2024-05-26
 

checkov/cloudformation/checks/graph_checks/SageMakerIAMPolicyOverlyPermissiveToAllTraffic.yaml

@@ -0,0 +1,25 @@
+metadata:
+  id: "CKV2_AWS_68"
+  name: "Ensure SageMaker notebook instance IAM policy is not overly permissive"
+  category: "NETWORKING"
+
+definition:
+  and:
+    - cond_type: filter
+      value:
+        - AWS::SageMaker::NotebookInstance
+      operator: within
+      attribute: resource_type
+    - cond_type: connection
+      resource_types:
+        - AWS::SageMaker::NotebookInstance
+      connected_resource_types:
+        - AWS::IAM::Role
+      operator: exists
+    - cond_type: attribute
+      resource_types:
+        - AWS::IAM::Role
+      attribute: "AssumeRolePolicyDocument.Statement[?(@.Effect == Allow)].Action[*]"
+      operator: "jsonpath_not_equals"
+      value: "*"
+

checkov/common/graph/graph_builder/graph_components/attribute_names.py

@@ -23,6 +23,7 @@ class CustomAttributes:
     ENCRYPTION = "encryption_"
     ENCRYPTION_DETAILS = "encryption_details_"
     TF_RESOURCE_ADDRESS = "__address__"
+    PROVIDER_ADDRESS = "__provider_address__"
     REFERENCES = "references_"
     FOREACH_ATTRS = "foreach_attrs_"
     SOURCE_MODULE_OBJECT = "source_module_object_"

checkov/common/util/env_vars_config.py

@@ -21,6 +21,7 @@ def __init__(self) -> None:
         self.CACHE_DIR = convert_str_to_bool(os.getenv("CKV_CACHE_DIR", str(Path(tempfile.gettempdir()) / "cache")))
         self.CHECK_FAIL_LEVEL = os.getenv("CHECKOV_CHECK_FAIL_LEVEL", CheckFailLevel.ERROR)
         self.CREATE_COMPLEX_VERTICES = convert_str_to_bool(os.getenv("CREATE_COMPLEX_VERTICES", True))
+        self.CHECKOV_ENABLE_DATAS_FOREACH_HANDLING = os.getenv('CHECKOV_ENABLE_DATAS_FOREACH_HANDLING', 'False')
         self.CREATE_EDGES = convert_str_to_bool(os.getenv("CREATE_EDGES", True))
         self.CREATE_MARKDOWN_HYPERLINKS = convert_str_to_bool(os.getenv("CHECKOV_CREATE_MARKDOWN_HYPERLINKS", False))
         self.CREATE_SCA_IMAGE_REPORTS_FOR_IR = convert_str_to_bool(

checkov/terraform/checks/graph_checks/aws/SageMakerIAMPolicyOverlyPermissiveToAllTraffic.yaml

@@ -0,0 +1,25 @@
+metadata:
+  id: "CKV2_AWS_68"
+  name: "Ensure SageMaker notebook instance IAM policy is not overly permissive"
+  category: "NETWORKING"
+
+definition:
+  and:
+    - cond_type: filter
+      value:
+        - aws_sagemaker_notebook_instance
+      operator: within
+      attribute: resource_type
+    - cond_type: connection
+      resource_types:
+        - aws_sagemaker_notebook_instance
+      connected_resource_types:
+        - aws_iam_role
+      operator: exists
+    - cond_type: attribute
+      resource_types:
+        - aws_iam_role
+      attribute: "policy.Statement[?(@.Effect == Allow)].Action[*]"
+      operator: "jsonpath_not_equals"
+      value: "*"
+

checkov/terraform/checks/resource/aws/S3AllowsAnyPrincipal.py

@@ -38,7 +38,8 @@ def check_conditions(statement) -> bool:
     if any(condition_type in condition for condition_type in string_conditions):
         for condition_type in string_conditions:
             if condition_type in condition:
-                if any(source in condition[condition_type] for source in ['aws:sourceVpce', 'aws:SourceVpc']):
+                if any(source in condition[condition_type] for source in ['aws:sourceVpce', 'aws:SourceVpc',
+                                                                          'aws:PrincipalOrgPath', 'aws:userid']):
                     return True
 
     # Default fail if none of the above conditions are met

checkov/terraform/graph_builder/foreach/builder.py

@@ -2,6 +2,7 @@
 
 import typing
 
+from checkov.terraform.graph_builder.foreach.data_handler import ForeachDataHandler
 from checkov.terraform.graph_builder.foreach.module_handler import ForeachModuleHandler
 from checkov.terraform.graph_builder.foreach.resource_handler import ForeachResourceHandler
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
@@ -14,8 +15,19 @@ class ForeachBuilder:
     def __init__(self, local_graph: TerraformLocalGraph):
         self._resource_handler = ForeachResourceHandler(local_graph)
         self._module_handler = ForeachModuleHandler(local_graph)
+        self._data_handler = ForeachDataHandler(local_graph)
 
     def handle(self, foreach_blocks: dict[str, list[int]]) -> None:
+        """
+        First Data blocks that Modules can inherit from are handled.
+        Second, Module blocks are handled.
+        Last Resource blocks that can be duplicate by the Modules rendering.
+        """
+        if self._data_handler.local_graph.enable_datas_foreach_handling:
+            if foreach_blocks.get(BlockType.DATA):
+                self._data_handler.handle(foreach_blocks[BlockType.DATA])
+                self._data_handler.local_graph._arrange_graph_data()
+                self._data_handler.local_graph._build_edges()
         if self._module_handler.local_graph.enable_modules_foreach_handling:
             if foreach_blocks.get(BlockType.MODULE):
                 self._module_handler.handle(foreach_blocks[BlockType.MODULE])

checkov/terraform/graph_builder/foreach/data_handler.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from checkov.terraform.graph_builder.graph_components.block_types import BlockType
+from checkov.terraform.graph_builder.foreach.foreach_entity_handler import ForeachEntityHandler
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from checkov.terraform.graph_builder.local_graph import TerraformLocalGraph
+
+
+class ForeachDataHandler(ForeachEntityHandler):
+    def __init__(self, local_graph: TerraformLocalGraph) -> None:
+        super().__init__(local_graph, BlockType.DATA)

checkov/terraform/graph_builder/foreach/foreach_entity_handler.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional, TYPE_CHECKING
+
+from checkov.common.util.data_structures_utils import pickle_deepcopy
+from checkov.terraform.graph_builder.foreach.abstract_handler import ForeachAbstractHandler
+from checkov.terraform.graph_builder.foreach.consts import FOR_EACH_BLOCK_TYPE, FOREACH_STRING, COUNT_STRING
+from checkov.terraform.graph_builder.graph_components.blocks import TerraformBlock
+
+if TYPE_CHECKING:
+    from checkov.terraform.graph_builder.local_graph import TerraformLocalGraph
+
+
+class ForeachEntityHandler(ForeachAbstractHandler):
+    def __init__(self, local_graph: TerraformLocalGraph, block_type_to_handle: str) -> None:
+        super().__init__(local_graph)
+        self.block_type_to_handle = block_type_to_handle
+
+    def handle(self, resources_blocks: list[int]) -> None:
+        block_index_to_statement: FOR_EACH_BLOCK_TYPE = self._get_statements(resources_blocks)
+        self._create_new_resources(block_index_to_statement)
+
+    def _get_statements(self, resources_blocks: list[int]) -> FOR_EACH_BLOCK_TYPE:
+        if not resources_blocks:
+            return {}
+        block_index_to_statement: FOR_EACH_BLOCK_TYPE = {}
+        for block_index, block in enumerate(self.local_graph.vertices):
+            if block.block_type != self.block_type_to_handle or not (FOREACH_STRING in block.attributes or COUNT_STRING in block.attributes):
+                continue
+            foreach_statement = self._get_static_foreach_statement(block_index)
+            block_index_to_statement[block_index] = foreach_statement
+        blocks_to_render = [block_idx for block_idx, statement in block_index_to_statement.items() if statement is None]
+        if blocks_to_render:
+            rendered_statements: FOR_EACH_BLOCK_TYPE = self._handle_dynamic_statement(blocks_to_render)
+            block_index_to_statement.update(rendered_statements)
+        return block_index_to_statement
+
+    def _get_static_foreach_statement(self, block_index: int) -> Optional[list[str] | dict[str, Any] | int]:
+        attributes = self.local_graph.vertices[block_index].attributes
+        if not attributes.get(FOREACH_STRING) and not attributes.get(COUNT_STRING):
+            return None
+        try:
+            if self._is_static_statement(block_index):
+                return self._handle_static_statement(block_index)
+            else:
+                return None
+        except Exception as e:
+            logging.info(
+                f"Cannot get foreach statement for block: {self.local_graph.vertices[block_index]}, error: {str(e)}")
+            return None
+
+    def _handle_dynamic_statement(self, blocks_to_render: list[int]) -> FOR_EACH_BLOCK_TYPE:
+        rendered_statements_by_idx: FOR_EACH_BLOCK_TYPE = {}
+        sub_graph = self._build_sub_graph(blocks_to_render)
+        self._render_sub_graph(sub_graph, blocks_to_render)
+        for block_idx in blocks_to_render:
+            if not self._is_static_statement(block_idx, sub_graph):
+                rendered_statements_by_idx[block_idx] = None
+            else:
+                rendered_statements_by_idx[block_idx] = self._handle_static_statement(block_idx, sub_graph)
+        return rendered_statements_by_idx
+
+    def _create_new_resources_count(self, statement: int, block_idx: int) -> None:
+        main_resource = self.local_graph.vertices[block_idx]
+        for i in range(statement):
+            self._create_new_resource(main_resource, i, resource_idx=block_idx, foreach_idx=i)
+
+    def _create_new_foreach_resource(self, block_idx: int, foreach_idx: int, main_resource: TerraformBlock,
+                                     new_key: int | str, new_value: int | str) -> None:
+        self._create_new_resource(main_resource, new_value, new_key=new_key, resource_idx=block_idx, foreach_idx=foreach_idx)
+
+    def _create_new_resource(
+            self,
+            main_resource: TerraformBlock,
+            new_value: int | str,
+            resource_idx: int,
+            foreach_idx: int,
+            new_key: int | str | None = None,
+    ) -> None:
+        new_resource = pickle_deepcopy(main_resource)
+        block_type, block_name = new_resource.name.split('.')
+        key_to_val_changes = self._build_key_to_val_changes(main_resource, new_value, new_key)
+        config_attrs = new_resource.config.get(block_type, {}).get(block_name, {})
+
+        self._update_foreach_attrs(config_attrs, key_to_val_changes, new_resource)
+        idx_to_change = new_key or new_value
+        self._add_index_to_resource_block_properties(new_resource, idx_to_change)
+        if foreach_idx == 0:
+            self.local_graph.vertices[resource_idx] = new_resource
+        else:
+            self.local_graph.vertices.append(new_resource)
+
+    @staticmethod
+    def _add_index_to_resource_block_properties(block: TerraformBlock, idx: str | int) -> None:
+        block_type, block_name = block.name.split('.')
+        idx_with_separator = ForeachEntityHandler._update_block_name_and_id(block, idx)
+        if block.config.get(block_type) and block.config.get(block_type, {}).get(block_name):
+            block.config[block_type][f"{block_name}[{idx_with_separator}]"] = block.config[block_type].pop(block_name)
+
+    def _create_new_resources(self, block_index_to_statement: FOR_EACH_BLOCK_TYPE) -> None:
+        for block_idx, statement in block_index_to_statement.items():
+            if not statement:
+                continue
+            if isinstance(statement, int):
+                self._create_new_resources_count(statement, block_idx)
+            else:
+                self._create_new_resources_foreach(statement, block_idx)

checkov/terraform/graph_builder/foreach/resource_handler.py

@@ -1,110 +1,15 @@
 from __future__ import annotations
 
-import logging
-import typing
-from typing import Any, Optional
-
-from checkov.common.util.data_structures_utils import pickle_deepcopy
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
-from checkov.terraform.graph_builder.foreach.abstract_handler import ForeachAbstractHandler
-from checkov.terraform.graph_builder.foreach.consts import FOREACH_STRING, COUNT_STRING, FOR_EACH_BLOCK_TYPE
-from checkov.terraform.graph_builder.graph_components.blocks import TerraformBlock
-
-if typing.TYPE_CHECKING:
-    from checkov.terraform.graph_builder.local_graph import TerraformLocalGraph
-
-
-class ForeachResourceHandler(ForeachAbstractHandler):
-    def __init__(self, local_graph: TerraformLocalGraph) -> None:
-        super().__init__(local_graph)
-
-    def handle(self, resources_blocks: list[int]) -> None:
-        block_index_to_statement: FOR_EACH_BLOCK_TYPE = self._get_statements(resources_blocks)
-        self._create_new_resources(block_index_to_statement)
+from checkov.terraform.graph_builder.foreach.foreach_entity_handler import ForeachEntityHandler
 
-    def _get_statements(self, resources_blocks: list[int]) -> FOR_EACH_BLOCK_TYPE:
-        if not resources_blocks:
-            return {}
-        block_index_to_statement: FOR_EACH_BLOCK_TYPE = {}
-        for block_index, block in enumerate(self.local_graph.vertices):
-            if block.block_type != BlockType.RESOURCE or not (FOREACH_STRING in block.attributes or COUNT_STRING in block.attributes):
-                continue
-            foreach_statement = self._get_static_foreach_statement(block_index)
-            block_index_to_statement[block_index] = foreach_statement
-        blocks_to_render = [block_idx for block_idx, statement in block_index_to_statement.items() if statement is None]
-        if blocks_to_render:
-            rendered_statements: FOR_EACH_BLOCK_TYPE = self._handle_dynamic_statement(blocks_to_render)
-            block_index_to_statement.update(rendered_statements)
-        return block_index_to_statement
+from typing import TYPE_CHECKING
 
-    def _get_static_foreach_statement(self, block_index: int) -> Optional[list[str] | dict[str, Any] | int]:
-        attributes = self.local_graph.vertices[block_index].attributes
-        if not attributes.get(FOREACH_STRING) and not attributes.get(COUNT_STRING):
-            return None
-        try:
-            if self._is_static_statement(block_index):
-                return self._handle_static_statement(block_index)
-            else:
-                return None
-        except Exception as e:
-            logging.info(
-                f"Cant get foreach statement for block: {self.local_graph.vertices[block_index]}, error: {str(e)}")
-            return None
-
-    def _handle_dynamic_statement(self, blocks_to_render: list[int]) -> FOR_EACH_BLOCK_TYPE:
-        rendered_statements_by_idx: FOR_EACH_BLOCK_TYPE = {}
-        sub_graph = self._build_sub_graph(blocks_to_render)
-        self._render_sub_graph(sub_graph, blocks_to_render)
-        for block_idx in blocks_to_render:
-            if not self._is_static_statement(block_idx, sub_graph):
-                rendered_statements_by_idx[block_idx] = None
-            else:
-                rendered_statements_by_idx[block_idx] = self._handle_static_statement(block_idx, sub_graph)
-        return rendered_statements_by_idx
-
-    def _create_new_resources_count(self, statement: int, block_idx: int) -> None:
-        main_resource = self.local_graph.vertices[block_idx]
-        for i in range(statement):
-            self._create_new_resource(main_resource, i, resource_idx=block_idx, foreach_idx=i)
-
-    def _create_new_resource(
-            self,
-            main_resource: TerraformBlock,
-            new_value: int | str,
-            resource_idx: int,
-            foreach_idx: int,
-            new_key: int | str | None = None,
-    ) -> None:
-        new_resource = pickle_deepcopy(main_resource)
-        block_type, block_name = new_resource.name.split('.')
-        key_to_val_changes = self._build_key_to_val_changes(main_resource, new_value, new_key)
-        config_attrs = new_resource.config.get(block_type, {}).get(block_name, {})
-
-        self._update_foreach_attrs(config_attrs, key_to_val_changes, new_resource)
-        idx_to_change = new_key or new_value
-        self._add_index_to_resource_block_properties(new_resource, idx_to_change)
-        if foreach_idx == 0:
-            self.local_graph.vertices[resource_idx] = new_resource
-        else:
-            self.local_graph.vertices.append(new_resource)
+if TYPE_CHECKING:
+    from checkov.terraform.graph_builder.local_graph import TerraformLocalGraph
 
-    def _create_new_foreach_resource(self, block_idx: int, foreach_idx: int, main_resource: TerraformBlock,
-                                     new_key: int | str, new_value: int | str) -> None:
-        self._create_new_resource(main_resource, new_value, new_key=new_key, resource_idx=block_idx,
-                                  foreach_idx=foreach_idx)
 
-    @staticmethod
-    def _add_index_to_resource_block_properties(block: TerraformBlock, idx: str | int) -> None:
-        block_type, block_name = block.name.split('.')
-        idx_with_separator = ForeachResourceHandler._update_block_name_and_id(block, idx)
-        if block.config.get(block_type) and block.config.get(block_type, {}).get(block_name):
-            block.config[block_type][f"{block_name}[{idx_with_separator}]"] = block.config[block_type].pop(block_name)
+class ForeachResourceHandler(ForeachEntityHandler):
 
-    def _create_new_resources(self, block_index_to_statement: FOR_EACH_BLOCK_TYPE) -> None:
-        for block_idx, statement in block_index_to_statement.items():
-            if not statement:
-                continue
-            if isinstance(statement, int):
-                self._create_new_resources_count(statement, block_idx)
-            else:
-                self._create_new_resources_foreach(statement, block_idx)
+    def __init__(self, local_graph: TerraformLocalGraph) -> None:
+        super().__init__(local_graph, BlockType.RESOURCE)

checkov/terraform/graph_builder/graph_components/module.py

@@ -7,6 +7,7 @@
 
 from checkov.common.typing import TFDefinitionKeyType
 from checkov.common.util.data_structures_utils import pickle_deepcopy
+from checkov.terraform import TFDefinitionKey
 from checkov.terraform.graph_builder.graph_components.block_types import BlockType
 from checkov.terraform.graph_builder.graph_components.blocks import TerraformBlock
 from checkov.terraform.parser_functions import handle_dynamic_values
@@ -34,6 +35,7 @@ def __init__(
         self.resources_types: Set[str] = set()
         self.source_dir = source_dir
         self.render_dynamic_blocks_env_var = os.getenv('CHECKOV_RENDER_DYNAMIC_MODULES', 'True')
+        self.temp_tf_definition: dict[TFDefinitionKey, dict[str, Any]] = {}
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Module):