Repository: bridgecrewio/checkov
Pull request: support check suppression via CFN Metadata
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Fixes #2194
Fixes #1931

Extended the graph to also store the CloudFormation `Metadata` and added the possibility to actually add the skip comment that section. Added an example to the docs. 

Problem statement:
Allow suppression to be done in Metadata properties of the AWS CloudFormation template and/or resources
Allow suppression to be done in Metadata properties of the template and/or resources

CKV_AWS_116: DLQ applies only to async invocations
**Describe the issue**
CKV_AWS_116: "Ensure that AWS Lambda function is configured for a Dead Letter Queue(DLQ)"
DLQ applies only to synchronous Lambda invocations, so if you have a REST API for example it won't have any effect and thus is unnecessary. The rule fires up for all Lambdas.

I understand that unfortunately in a general case it is impossible to figure out from a CloudFormation template with a Lambda function whether it is synchronous or asynchronous, because you can call any function in either way with AWS CLI or API, without any event mapping.

Hints from issue discussion:
hi, actually the DLQ is for asynchronous Lambda invokes not synchronous. We already have an open issue talking about extending the check to allow different configuration via event invoke config #1795
Since we don't know, how a user will use the Lambda, we give recommendation what should be configured, but if it doesn't apply to some of the Lambdas, then feel free to add a skip comment and details about not being useful for that case.
@gruebel sorry, you are right, I had a brain zap, it is other way around indeed. I would add a comment but I pipe autogenerated CFN from CDK so no easy way to add comments. Any chance checkov could support metadata section of CFN for suppression? This way you can link a suppress statement with a resource without hand-writing CFN.
not yet, but an issue exists 😉 just add your 👍 here #1931 I also thought this would be helpful for CDK setups

Commit message: add CDK example

Patch to review:
diff --git a/checkov/cloudformation/cfn_utils.py b/checkov/cloudformation/cfn_utils.py
index d976c6bb3d..e4ec933e55 100644
--- a/checkov/cloudformation/cfn_utils.py
+++ b/checkov/cloudformation/cfn_utils.py
@@ -152,7 +152,10 @@ def build_definitions_context(
                             {"start_line": start_line, "end_line": end_line, "code_lines": code_lines},
                         )
                         if file_path_definition.upper() == TemplateSections.RESOURCES.value.upper():
-                            skipped_checks = ContextParser.collect_skip_comments(code_lines)
+                            skipped_checks = ContextParser.collect_skip_comments(
+                                entity_code_lines=code_lines,
+                                resource_config=attr_value,
+                            )
                             dpath.new(
                                 definitions_context,
                                 [file_path, str(file_path_definition), str(attribute), "skipped_checks"],
diff --git a/checkov/cloudformation/context_parser.py b/checkov/cloudformation/context_parser.py
index c64cf45e98..f3f53d92ac 100644
--- a/checkov/cloudformation/context_parser.py
+++ b/checkov/cloudformation/context_parser.py
@@ -1,3 +1,4 @@
+import itertools
 import logging
 import operator
 import re
@@ -111,7 +112,7 @@ def find_lines(node: Union[ListNode, DictNode], kv: str) -> Generator[int, None,
                 yield node[kv]
 
     @staticmethod
-    def collect_skip_comments(entity_code_lines: List[Tuple[int, str]]) -> List[_SkippedCheck]:
+    def collect_skip_comments(entity_code_lines: List[Tuple[int, str]], resource_config: Optional[DictNode] = None) -> List[_SkippedCheck]:
         skipped_checks = []
         bc_id_mapping = bc_integration.get_id_mapping()
         ckv_to_bc_id_mapping = bc_integration.get_ckv_to_bc_id_mapping()
@@ -130,6 +131,28 @@ def collect_skip_comments(entity_code_lines: List[Tuple[int, str]]) -> List[_Ski
                     skipped_check["bc_id"] = ckv_to_bc_id_mapping.get(skipped_check["id"])
 
                 skipped_checks.append(skipped_check)
+        if resource_config:
+            metadata = resource_config.get("Metadata")
+            if metadata:
+                ckv_skip = metadata.get("checkov", {}).get("skip", [])
+                bc_skip = metadata.get("bridgecrew", {}).get("skip", [])
+                if ckv_skip or bc_skip:
+                    for skip in itertools.chain(ckv_skip, bc_skip):
+                        skip_id = skip.get("id")
+                        skip_comment = skip.get("comment", "No comment provided")
+                        if skip_id is None:
+                            logging.debug("Check suppression is missing key 'id'")
+                            continue
+
+                        skipped_check = {"id": skip_id, "suppress_comment": skip_comment}
+                        if bc_id_mapping and skipped_check["id"] in bc_id_mapping:
+                            skipped_check["bc_id"] = skipped_check["id"]
+                            skipped_check["id"] = bc_id_mapping[skipped_check["id"]]
+                        elif ckv_to_bc_id_mapping:
+                            skipped_check["bc_id"] = ckv_to_bc_id_mapping.get(skipped_check["id"])
+
+                        skipped_checks.append(skipped_check)
+
         return skipped_checks
 
     @staticmethod
diff --git a/checkov/cloudformation/graph_builder/graph_components/blocks.py b/checkov/cloudformation/graph_builder/graph_components/blocks.py
index fd9e3b355a..eba2161a84 100644
--- a/checkov/cloudformation/graph_builder/graph_components/blocks.py
+++ b/checkov/cloudformation/graph_builder/graph_components/blocks.py
@@ -7,15 +7,16 @@
 
 class CloudformationBlock(Block):
     def __init__(
-            self,
-            name: str,
-            config: Dict[str, Any],
-            path: str,
-            block_type: str,
-            attributes: Dict[str, Any],
-            id: str = "",
-            source: str = "",
-            condition: bool = True,
+        self,
+        name: str,
+        config: Dict[str, Any],
+        path: str,
+        block_type: str,
+        attributes: Dict[str, Any],
+        id: str = "",
+        source: str = "",
+        condition: bool = True,
+        metadata: Optional[Dict[str, Any]] = None
     ) -> None:
         """
             :param name: unique name given to the terraform block, for example: 'aws_vpc.example_name'
@@ -26,6 +27,7 @@ def __init__(
         """
         super().__init__(name, config, path, block_type, attributes, id, source)
         self.condition = condition
+        self.metadata = metadata
 
     def update_attribute(
             self, attribute_key: str, attribute_value: Any, change_origin_id: int,
diff --git a/checkov/cloudformation/graph_builder/graph_to_definitions.py b/checkov/cloudformation/graph_builder/graph_to_definitions.py
index 1e65a40dfe..c942ae9b59 100644
--- a/checkov/cloudformation/graph_builder/graph_to_definitions.py
+++ b/checkov/cloudformation/graph_builder/graph_to_definitions.py
@@ -20,9 +20,13 @@ def convert_graph_vertices_to_definitions(
         block_name = vertex.name.split('.')[-1]  # vertex.name is "type.name" so type.name -> [type, name]
 
         definition = {
-            'Type': vertex.attributes['resource_type'] if vertex.block_type == BlockType.RESOURCE else vertex.block_type,
-            'Properties': vertex.config
+            "Type": vertex.attributes["resource_type"] if vertex.block_type == BlockType.RESOURCE else vertex.block_type,
+            "Properties": vertex.config,
         }
+
+        if vertex.metadata:
+            definition["Metadata"] = vertex.metadata
+
         definitions.setdefault(block_path, {}).setdefault(block_type, {}).setdefault(block_name, definition)
 
         relative_block_path = f"/{os.path.relpath(block_path, root_folder)}"
diff --git a/checkov/cloudformation/graph_builder/local_graph.py b/checkov/cloudformation/graph_builder/local_graph.py
index 998363d290..996955030f 100644
--- a/checkov/cloudformation/graph_builder/local_graph.py
+++ b/checkov/cloudformation/graph_builder/local_graph.py
@@ -82,6 +82,7 @@ def _create_section_vertices(self, file_path: str, file_conf: dict, section: Tem
             attributes = attributes_operator(obj)
             block_name = name if not is_resources_section else f"{obj.get('Type', 'UnTyped')}.{name}"
             config = obj if not is_resources_section else obj.get("Properties")
+            metadata = obj.get("Metadata")
             id = f"{block_type}.{block_name}" if not is_resources_section else block_name
             self.vertices.append(CloudformationBlock(
                 name=block_name,
@@ -90,7 +91,8 @@ def _create_section_vertices(self, file_path: str, file_conf: dict, section: Tem
                 block_type=block_type,
                 attributes=attributes,
                 id=id,
-                source=self.source
+                source=self.source,
+                metadata=metadata
             ))
 
             if not self._vertices_indexes.get(file_path):
diff --git a/checkov/cloudformation/runner.py b/checkov/cloudformation/runner.py
index 41f781511d..c0f28c08e0 100644
--- a/checkov/cloudformation/runner.py
+++ b/checkov/cloudformation/runner.py
@@ -111,7 +111,14 @@ def check_definitions(self, root_folder, runner_filter, report):
                         if entity_lines_range and entity_code_lines:
                             # TODO - Variable Eval Message!
                             variable_evaluations = {}
-                            skipped_checks = ContextParser.collect_skip_comments(entity_code_lines)
+
+                            skipped_checks =  resource_context.get("skipped_checks")
+                            if not skipped_checks:
+                                skipped_checks = ContextParser.collect_skip_comments(
+                                    entity_code_lines=entity_code_lines,
+                                    resource_config=resource,
+                                )
+
                             entity = {resource_name: resource}
                             results = cfn_registry.scan(cf_file, entity, skipped_checks, runner_filter)
                             tags = cfn_utils.get_resource_tags(entity)
diff --git a/docs/2.Basics/Suppressing and Skipping Policies.md b/docs/2.Basics/Suppressing and Skipping Policies.md
index 1db4c07971..7a809e44bd 100644
--- a/docs/2.Basics/Suppressing and Skipping Policies.md	
+++ b/docs/2.Basics/Suppressing and Skipping Policies.md	
@@ -94,3 +94,59 @@ Resources:
       MasterUserPassword: 'password' # checkov:skip=CKV_SECRET_6 or next to it
       # checkov:skip=CKV_SECRET_6 or after it
 ```
+
+## CloudFormation Metadata
+Additionally, it is possible to suppress CloudFormation checks via the `Metadata` section inside a resource.
+```yaml
+Resources:
+  MyDB:
+    Metadata:
+      checkov:
+        skip:
+          - id: "CKV_AWS_157"
+            comment: "Ensure that RDS instances have Multi-AZ enabled"
+    Type: "AWS::RDS::DBInstance"
+    Properties:
+      DBName: "mydb"
+      DBInstanceClass: "db.t3.micro"
+      Engine: "mysql"
+      MasterUsername: "master"
+      MasterUserPassword: "password"
+```
+
+### CDK Example
+The `Metadata` section of a CDK construct can only be adjusted via the L1 resource.
+```typescript
+const bucket = new aws_s3.Bucket(this, 'MyBucket', {
+  versioned: true
+});
+const cfnBucket = bucket.node.defaultChild as aws_s3.CfnBucket;
+
+cfnBucket.cfnOptions.metadata = {
+  'checkov': {
+    'skip': [
+      {
+        'id': 'CKV_AWS_18',
+        'comment': 'Ensure the S3 bucket has access logging enabled'
+      }
+    ]
+  }
+}
+```
+Using `cdk synth` this results in a following CloudFormation template
+```yaml
+Resources:
+  MyBucketF68F3FF0:
+    Type: AWS::S3::Bucket
+    Properties:
+      VersioningConfiguration:
+        Status: Enabled
+    UpdateReplacePolicy: Retain
+    DeletionPolicy: Retain
+    Metadata:
+      checkov:
+        skip:
+          - id: CKV_AWS_18
+            comment: Ensure the S3 bucket has access logging enabled
+
+```
diff --git a/tests/cloudformation/utils/file_formats/test.yaml b/tests/cloudformation/utils/file_formats/test.yaml
index 8aa5594a62..5cde1bdaee 100644
--- a/tests/cloudformation/utils/file_formats/test.yaml
+++ b/tests/cloudformation/utils/file_formats/test.yaml
@@ -15,6 +15,16 @@ Resources:
     Properties:
       KmsMasterKeyId: !Ref KmsMasterKeyId
   MyDB:
+    # Test case for check skip via 'Metadata'
+    Metadata:
+      checkov:
+        skip:
+          - id: CKV_AWS_17
+            comment: Ensure all data stored in RDS is not publicly accessible
+      bridgecrew:
+        skip:
+          - id: CKV_AWS_157
+            comment: Ensure that RDS instances have Multi-AZ enabled
     Type: 'AWS::RDS::DBInstance'
     # Test case for check skip via comment
     # checkov:skip=CKV_AWS_16:Ensure all data stored in the RDS is securely encrypted at rest
@@ -24,6 +34,8 @@ Resources:
       Engine: 'mysql'
       MasterUsername: 'master'
       MasterUserPassword: 'password'
+      PubliclyAccessible: True
+
 Outputs:
   DBAppPublicDNS:
     Description: DB App Public DNS Name
diff --git a/tests/cloudformation/utils/test_cfn_utils.py b/tests/cloudformation/utils/test_cfn_utils.py
index 3f88e1ae62..c2d2cfb268 100644
--- a/tests/cloudformation/utils/test_cfn_utils.py
+++ b/tests/cloudformation/utils/test_cfn_utils.py
@@ -56,7 +56,7 @@ def test_resources_value(self):
         self.assertIsNotNone(yaml_resources)
         self.assertEqual(len(yaml_resources), 2)
         self.validate_definition_lines(yaml_resources["MySourceQueue"], 13, 16, 4)
-        self.validate_definition_lines(yaml_resources["MyDB"], 17, 26, 10)
+        self.validate_definition_lines(yaml_resources["MyDB"], 17, 37, 21)
         # Asserting test2.yaml file
         yaml2_resources = self.definitions_context[str(self.test_root_dir / "test2.yaml")][
             TemplateSections.RESOURCES.value
@@ -81,7 +81,7 @@ def test_outputs_value(self):
         yaml_outputs = self.definitions_context[str(self.test_root_dir / "test.yaml")][TemplateSections.OUTPUTS.value]
         self.assertIsNotNone(yaml_outputs)
         self.assertEqual(len(yaml_outputs), 1)
-        self.validate_definition_lines(yaml_outputs["DBAppPublicDNS"], 28, 30, 3)
+        self.validate_definition_lines(yaml_outputs["DBAppPublicDNS"], 40, 42, 3)
         # Asserting test2.yaml file
         yaml2_outputs = self.definitions_context[str(self.test_root_dir / "test2.yaml")][TemplateSections.OUTPUTS.value]
         self.assertIsNotNone(yaml2_outputs)
@@ -101,11 +101,26 @@ def test_skipped_check_exists(self):
         skipped_checks = self.definitions_context[str(self.test_root_dir / "test.yaml")][
             TemplateSections.RESOURCES.value
         ]["MyDB"]["skipped_checks"]
-        self.assertEqual(len(skipped_checks), 1)
-        self.assertEqual(skipped_checks[0]["id"], "CKV_AWS_16")
-        self.assertEqual(
-            skipped_checks[0]["suppress_comment"],
-            "Ensure all data stored in the RDS is securely encrypted at rest\\n')",
+        self.assertEqual(len(skipped_checks), 3)
+        self.assertCountEqual(
+            skipped_checks,
+            [
+                {
+                    "id": "CKV_AWS_16",
+                    "suppress_comment": "Ensure all data stored in the RDS is securely encrypted at rest\\n')",
+                    "bc_id": "BC_AWS_GENERAL_4",
+                },
+                {
+                    "id": "CKV_AWS_17",
+                    "suppress_comment": "Ensure all data stored in RDS is not publicly accessible",
+                    "bc_id": "BC_AWS_PUBLIC_2",
+                },
+                {
+                    "id": "CKV_AWS_157",
+                    "suppress_comment": "Ensure that RDS instances have Multi-AZ enabled",
+                    "bc_id": "BC_AWS_GENERAL_73",
+                },
+            ],
         )
 
 


Code review comments:
diff_hunk: @@ -130,6 +131,28 @@ def collect_skip_comments(entity_code_lines: List[Tuple[int, str]]) -> List[_Ski
                     skipped_check["bc_id"] = ckv_to_bc_id_mapping.get(skipped_check["id"])
 
                 skipped_checks.append(skipped_check)
+        if resource_config:
+            metadata = resource_config.get("Metadata")
+            if metadata:
+                ckv_skip = metadata.get("checkov", {}).get("skip", [])
+                bc_skip = metadata.get("bridgecrew", {}).get("skip", [])
+                if ckv_skip or bc_skip:
+                    for skip in itertools.chain(ckv_skip, bc_skip):
+                        skip_id = skip.get("id")
+                        skip_comment = skip.get("comment", "No comment provided")
+                        if skip_id is None:
+                            logging.debug("Check suppression is missing key 'id'")
path: checkov/cloudformation/context_parser.py
text: @user1:
@author Let's elevate to `info` / `warning` please, this is highly unexpected! Also add more identifiable data to the log to help the suppressor identify the problematic suppression

@author:
makes sense, let's go with `warning` and if users complain being to noisy, then we can reduce it later 😄