feat(graph): add support for modules in graph checks

checkov/common/checks_infra/solvers/attribute_solvers/base_attribute_solver.py

@@ -19,7 +19,7 @@
 if TYPE_CHECKING:
     from networkx import DiGraph
 
-SUPPORTED_BLOCK_TYPES = {BlockType.RESOURCE, TerraformBlockType.DATA}
+SUPPORTED_BLOCK_TYPES = {BlockType.RESOURCE, TerraformBlockType.DATA, TerraformBlockType.MODULE}
 WILDCARD_PATTERN = re.compile(r"(\S+[.][*][.]*)+")
 
 OPERATION_TO_FUNC = {
@@ -31,8 +31,9 @@
 class BaseAttributeSolver(BaseSolver):
     operator = ""  # noqa: CCE003  # a static attribute
 
-    def __init__(self, resource_types: List[str], attribute: Optional[str], value: Any, is_jsonpath_check: bool = False
-                 ) -> None:
+    def __init__(
+        self, resource_types: List[str], attribute: Optional[str], value: Any, is_jsonpath_check: bool = False
+    ) -> None:
         super().__init__(SolverType.ATTRIBUTE)
         self.resource_types = resource_types
         self.attribute = attribute

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

@@ -96,6 +96,10 @@ def get_attribute_dict(self, add_hash: bool = True) -> Dict[str, Any]:
         if self.block_type == BlockType.DATA:
             base_attributes[CustomAttributes.RESOURCE_TYPE] = f'data.{self.id.split(".")[0]}'
 
+        if self.block_type == BlockType.MODULE:
+            # since module names are user defined we are just setting 'module' as resource type for easier searching
+            base_attributes[CustomAttributes.RESOURCE_TYPE] = "module"
+
         if "changed_attributes" in base_attributes:
             # removed changed attributes if it was added previously for calculating hash.
             del base_attributes["changed_attributes"]

checkov/common/output/report.py

@@ -542,6 +542,7 @@ def enrich_plan_report(
     def handle_skipped_checks(
             report: "Report", enriched_resources: Dict[str, Dict[str, Any]]
     ) -> "Report":
+        module_address_len = len("module.")
         skip_records = []
         for record in report.failed_checks:
             resource_skips = enriched_resources.get(record.resource, {}).get(
@@ -557,7 +558,7 @@ def handle_skipped_checks(
                     report.add_record(record)
 
             if record.resource_address and record.resource_address.startswith("module."):
-                module_path = record.resource_address[0:record.resource_address.index('.', len("module.") + 1)]
+                module_path = record.resource_address[module_address_len:record.resource_address.index('.', module_address_len + 1)]
                 module_enrichments = enriched_resources.get(module_path, {})
                 for module_skip in module_enrichments.get("skipped_checks", []):
                     if record.check_id in module_skip["id"]:

checkov/terraform/context_parsers/parsers/module_context_parser.py

@@ -12,12 +12,12 @@ def __init__(self) -> None:
 
     def get_entity_context_path(self, entity_block: Dict[str, Dict[str, Any]]) -> List[str]:
         entity_name = next(iter(entity_block.keys()))
-        return ["module", entity_name]
+        return [entity_name]
 
     def enrich_definition_block(self, definition_blocks: List[Dict[str, Any]]) -> Dict[str, Any]:
         for entity_block in definition_blocks:
             entity_name, entity_config = next(iter(entity_block.items()))
-            self.context["module"][entity_name] = {
+            self.context[entity_name] = {
                 "start_line": entity_config[START_LINE],
                 "end_line": entity_config[END_LINE],
                 "code_lines": self.file_lines[entity_config[START_LINE] - 1: entity_config[END_LINE]],

checkov/terraform/graph_builder/graph_components/module.py

@@ -156,13 +156,14 @@ def _add_resource(self, blocks: List[Dict[str, Dict[str, Any]]], path: str) -> N
                     if provisioner:
                         self._handle_provisioner(provisioner, attributes)
                     attributes["resource_type"] = [resource_type]
+                    block_name = f"{resource_type}.{name}"
                     resource_block = TerraformBlock(
                         block_type=BlockType.RESOURCE,
-                        name=f"{resource_type}.{name}",
+                        name=block_name,
                         config=self.clean_bad_characters(resource_dict),
                         path=path,
                         attributes=attributes,
-                        id=f"{resource_type}.{name}",
+                        id=block_name,
                         source=self.source,
                     )
                     self._add_to_blocks(resource_block)
@@ -178,13 +179,14 @@ def _add_data(self, blocks: List[Dict[str, Dict[str, Any]]], path: str) -> None:
         for data_dict in blocks:
             for data_type in data_dict:
                 for name in data_dict[data_type]:
+                    block_name = f"{data_type}.{name}"
                     data_block = TerraformBlock(
                         block_type=BlockType.DATA,
-                        name=data_type + "." + name,
+                        name=block_name,
                         config=data_dict,
                         path=path,
                         attributes=data_dict.get(data_type, {}).get(name, {}),
-                        id=data_type + "." + name,
+                        id=block_name,
                         source=self.source,
                     )
                     self._add_to_blocks(data_block)

checkov/terraform/runner.py

@@ -334,10 +334,7 @@ def run_block(self, entities,
 
                     try:
                         caller_context = definition_context[abs_caller_file]
-                        # HACK ALERT: module data is currently double-nested in
-                        #             definition context. If fixed, remove the
-                        #             addition of "module." at the beginning.
-                        for part in f"module.{referrer_id}".split("."):
+                        for part in referrer_id.split("."):
                             caller_context = caller_context[part]
                     except KeyError:
                         logging.debug("Unable to find caller context for: %s", abs_caller_file)
@@ -467,19 +464,23 @@ def push_skipped_checks_down(self, definition_context, module_path, skipped_chec
             if module_path not in mod_ref:
                 continue
 
-            for next_type in definition_context[definition]:
-                # skip if type is not a terraform resource
-                if next_type not in CHECK_BLOCK_TYPES:
+            for block_type, block_configs in definition_context[definition].items():
+                # skip if type is not a Terraform resource
+                if block_type not in CHECK_BLOCK_TYPES:
                     continue
 
-                # there may be multiple resource types - aws_bucket, etc
-                for resource_type in definition_context[definition][next_type]:
-                    # there may be multiple names for each resource type
-                    for resource_name in definition_context[definition][next_type][resource_type]:
-                        # append the skipped checks from the module to the other resources.
-                        # this could also be from a module to another module.
-                        self.context[definition][next_type][resource_type][resource_name][
-                            "skipped_checks"] += skipped_checks
+                if block_type == "module":
+                    # modules don't have a type, just a name
+                    for resource_config in block_configs.values():
+                        # append the skipped checks also from a module to another module
+                        resource_config["skipped_checks"] += skipped_checks
+                else:
+                    # there may be multiple resource types - aws_bucket, etc
+                    for resource_configs in block_configs.values():
+                        # there may be multiple names for each resource type
+                        for resource_config in resource_configs.values():
+                            # append the skipped checks from the module to the other resources.
+                            resource_config["skipped_checks"] += skipped_checks
 
     @staticmethod
     def _strip_module_referrer(file_path: str) -> Tuple[str, Optional[str]]:

docs/3.Custom Policies/Examples.md

@@ -12,16 +12,33 @@ nav_order: 4
 ```yaml
 ---
 metadata:
- name: "Check that all resources are tagged with the key - env"
- id: "CKV2_AWS_1"
- category: "GENERAL_SECURITY"
+  name: "Check that all resources are tagged with the key - env"
+  id: "CKV2_AWS_1"
+  category: "GENERAL_SECURITY"
 scope:
- provider: aws
+  provider: aws
 definition:
-       cond_type: "attribute"
-       resource_types: "all"
-       attribute: "tags.env"
-       operator: "exists"
+  cond_type: "attribute"
+  resource_types: "all"
+  attribute: "tags.env"
+  operator: "exists"
+```
+
+## Basic Query - Module block example 
+
+```yaml
+---
+metadata:
+  name: "Ensure all modules are using the official AWS ones"
+  id: "CKV2_AWS_1"
+  category: "SUPPLY_CHAIN"
+definition:
+  cond_type: attribute
+  resource_types:
+    - module
+  attribute: source
+  operator: starting_with
+  value: terraform-aws-modules
 ```
 
 ## Basic Query - Terraform plan resource not deleted

tests/terraform/runner/extra_yaml_checks/module_source.yaml

@@ -0,0 +1,11 @@
+metadata:
+  name: "Ensure all modules are using the official AWS ones"
+  id: "CUSTOM_GRAPH_AWS_2"
+  category: "SUPPLY_CHAIN"
+definition:
+  cond_type: attribute
+  resource_types:
+    - module
+  attribute: source
+  operator: starting_with
+  value: terraform-aws-modules

tests/terraform/runner/resources/module_check/main.tf

@@ -0,0 +1,25 @@
+module "pass" {
+  source = "terraform-aws-modules/ec2-instance/aws"
+
+  name = "terraform"
+
+  ami                    = "ami-0ff8a91507f77f867"
+  instance_type          = "t3.micro"
+  key_name               = "user1"
+  vpc_security_group_ids = ["sg-12345678"]
+  subnet_id              = "subnet-123456"
+}
+
+module "fail" {
+  source = "cloudposse/ec2-instance/aws"
+
+  name = "cloudposse"
+
+  ami                    = "ami-0ff8a91507f77f867"
+  ssh_key_pair           = "user1"
+  instance_type          = "t3.micro"
+  security_groups        = ["sg-12345678"]
+  subnet                 = "subnet-123456"
+  namespace              = "eg"
+  stage                  = "dev"
+}

tests/terraform/runner/test_runner.py

@@ -189,6 +189,36 @@ def test_runner_extra_yaml_check(self):
         # Remove external checks from registry.
         runner.graph_registry.checks[:] = [check for check in runner.graph_registry.checks if "CUSTOM" not in check.id]
 
+    def test_runner_yaml_module_check(self):
+        # given
+        current_dir = Path(__file__).parent
+        tf_dir_path = current_dir / "resources/module_check"
+        extra_checks_dir_path = current_dir / "extra_yaml_checks"
+        runner = Runner()
+
+        # when
+        report = runner.run(root_folder=str(tf_dir_path), external_checks_dir=[str(extra_checks_dir_path)])
+
+        # then
+        summary = report.get_summary()
+
+        passing_resources = {"pass"}
+        failing_resources = {"fail"}
+
+        passed_check_resources = {c.resource for c in report.passed_checks}
+        failed_check_resources = {c.resource for c in report.failed_checks}
+
+        self.assertEqual(summary["passed"], len(passing_resources))
+        self.assertEqual(summary["failed"], len(failing_resources))
+        self.assertEqual(summary["skipped"], 0)
+        self.assertEqual(summary["parsing_errors"], 0)
+
+        self.assertEqual(passing_resources, passed_check_resources)
+        self.assertEqual(failing_resources, failed_check_resources)
+
+        # Remove external checks from registry.
+        runner.graph_registry.checks[:] = [check for check in runner.graph_registry.checks if "CUSTOM" not in check.id]
+
     def test_runner_specific_file(self):
         current_dir = os.path.dirname(os.path.realpath(__file__))
 
@@ -391,7 +421,7 @@ def scan_module_conf(self, conf):
             module_registry.checks[resource].remove(check)
 
         self.assertEqual(len(result.passed_checks), 1)
-        self.assertIn('module.some-module', map(lambda record: record.resource, result.passed_checks))
+        self.assertIn('some-module', map(lambda record: record.resource, result.passed_checks))
 
     def test_terraform_module_checks_are_performed_even_if_supported_resources_is_omitted(self):
         check_name = "TF_M_2"
@@ -424,7 +454,7 @@ def scan_module_conf(self, conf):
             module_registry.checks[resource].remove(check)
 
         self.assertEqual(len(result.passed_checks), 1)
-        self.assertIn('module.some-module', map(lambda record: record.resource, result.passed_checks))
+        self.assertIn('some-module', map(lambda record: record.resource, result.passed_checks))
 
     def test_terraform_multiple_module_versions(self):
         # given
@@ -1070,8 +1100,6 @@ def test_module_failure_reporting_772(self):
                 assert record.file_path == "/module/module.tf"
                 self.assertEqual(record.file_line_range, [7, 13])
                 assert record.caller_file_path == "/main.tf"
-                # ATTENTION!! If this breaks, see the "HACK ALERT" comment in runner.run_block.
-                #             A bug might have been fixed.
                 self.assertEqual(record.caller_file_line_range, [6, 8])
 
         self.assertTrue(found_inside)
@@ -1086,7 +1114,7 @@ def test_loading_external_checks_yaml(self):
         current_dir = os.path.dirname(os.path.realpath(__file__))
         extra_checks_dir_path = current_dir + "/extra_yaml_checks"
         runner.load_external_checks([extra_checks_dir_path])
-        self.assertEqual(len(runner.graph_registry.checks), base_len + 2)
+        self.assertEqual(len(runner.graph_registry.checks), base_len + 3)
         runner.graph_registry.checks = runner.graph_registry.checks[:base_len]
 
     def test_loading_external_checks_yaml_multiple_times(self):
@@ -1095,11 +1123,14 @@ def test_loading_external_checks_yaml_multiple_times(self):
         runner.graph_registry.checks = []
         extra_checks_dir_path = [current_dir + "/extra_yaml_checks"]
         runner.load_external_checks(extra_checks_dir_path)
-        self.assertEqual(len(runner.graph_registry.checks), 2)
+        self.assertEqual(len(runner.graph_registry.checks), 3)
         runner.load_external_checks(extra_checks_dir_path)
-        self.assertEqual(len(runner.graph_registry.checks), 2)
-        self.assertIn('CUSTOM_GRAPH_AWS_1', [x.id for x in runner.graph_registry.checks])
-        self.assertIn('CKV2_CUSTOM_1', [x.id for x in runner.graph_registry.checks])
+        self.assertEqual(len(runner.graph_registry.checks), 3)
+
+        graph_checks = [x.id for x in runner.graph_registry.checks]
+        self.assertIn('CUSTOM_GRAPH_AWS_1', graph_checks)
+        self.assertIn('CUSTOM_GRAPH_AWS_2', graph_checks)
+        self.assertIn('CKV2_CUSTOM_1', graph_checks)
         runner.graph_registry.checks = []
 
     def test_loading_external_checks_python(self):