implement custom rule IS DEFINED (#2656)

Co-authored-by: Kevin DeJong <[email protected]>

Author: kftsehk

docs/custom_rules.md

@@ -45,13 +45,16 @@ The specified operator to be used for this rule. The supported values are define
 | NOT_IN | Checks the specified property is not equal to or not contained by the array value |
 | \>= | Checks the specified property is greater than or equal to the value given |
 | <= | Checks the specified property is less than or equal to the value given |
+| IS | Checks the specified property is defined or not defined, the value must be one of DEFINED or NOT_DEFINED |
 
 #### Value
 
 The value which the operator is comparing against (e.g `CompareMe`).
 
 Multi-word inputs are accepted  (e.g `Compare Me`). Array inputs are also accepted for set operations (e.g `[Apples, Oranges, Pears]`).
 
+For operator `IS`, the value must be one of `DEFINED` or `NOT_DEFINED`.
+
 #### Error Level (Optional)
 
 To specify the error level any breach of this rule is categorized. The supported values include all existing error levels (e.g `ERROR` or `WARN`)
@@ -71,6 +74,12 @@ This rule validates all EC2 instances in a template aren’t using the instance
 AWS::EC2::Instance InstanceType != "p3.2xlarge"
 ```
 
+This rule specify all lambda function in a template must specify environment variable `NODE_ENV`.
+
+```
+AWS::Lambda::Function Environment.Variables.NODE_ENV IS DEFINED
+```
+
 To include this rules, include your custom rules text file using the `-z custom_rules.txt` argument when running cfn-lint.
 
 

src/cfnlint/rules/custom/Operators.py

@@ -2,10 +2,22 @@
 Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 SPDX-License-Identifier: MIT-0
 """
+
 # pylint: disable=cyclic-import
 import cfnlint.rules
 
-OPERATOR = ["EQUALS", "NOT_EQUALS", "==", "!=", "IN", "NOT_IN", ">=", "<="]
+OPERATOR = [
+    "EQUALS",
+    "NOT_EQUALS",
+    "==",
+    "!=",
+    "IN",
+    "NOT_IN",
+    ">=",
+    "<=",
+    "IS DEFINED",
+    "IS NOT_DEFINED",
+]
 
 
 def CreateCustomRule(
@@ -81,6 +93,109 @@ def match_resource_properties(self, properties, _, path, cfn):
     )
 
 
+def CreateCustomIsDefinedRule(rule_id, resourceType, prop, value, error_message):
+    class CustomIsDefinedRule(cfnlint.rules.CloudFormationLintRule):
+        def __init__(
+            self,
+            rule_id,
+            resourceType,
+            prop,
+            value,
+            error_message,
+            description,
+            shortdesc,
+        ):
+            super().__init__()
+            self.id = rule_id
+            self.resource_property_types.append(resourceType)
+            self.property_chain = prop.split(".")
+            if value == "DEFINED":
+                self.is_defined = True
+            elif value == "NOT_DEFINED":
+                self.is_defined = False
+            else:
+                raise ValueError("IS must follow either DEFINED or NOT_DEFINED")
+            self.error_message = error_message
+            self.description = description
+            self.shortdesc = shortdesc
+
+        def _split_inset_properties(self, property_chain):
+            if property_chain:
+                if len(property_chain) > 1:
+                    return property_chain[0], property_chain[1:]
+                return property_chain[0], []
+
+            return None, []
+
+        def _check_value(self, value, path, property_chain, cfn):
+            matches = []
+            child_property, new_property_chain = self._split_inset_properties(
+                property_chain
+            )
+            if self.is_defined and (
+                value is None or value.get(child_property, None) is None
+            ):
+                matches.append(
+                    cfnlint.rules.RuleMatch(
+                        path, error_message or f"{path} must be defined"
+                    )
+                )
+            if child_property is not None:
+                matches.extend(
+                    cfn.check_value(
+                        value,
+                        child_property,
+                        path,
+                        check_value=self._check_value,
+                        property_chain=new_property_chain,
+                        cfn=cfn,
+                    )
+                )
+                return matches
+            if not self.is_defined and value is not None:
+                matches.append(
+                    cfnlint.rules.RuleMatch(
+                        path, error_message or f"{path} must not be defined"
+                    )
+                )
+            return matches
+
+        def match_resource_properties(self, properties, _, path, cfn):
+            child_property, new_property_chain = self._split_inset_properties(
+                self.property_chain
+            )
+            matches = []
+            # here does nothing when the value is not defined, this is checked separately below
+            matches.extend(
+                cfn.check_value(
+                    properties,
+                    child_property,
+                    path,
+                    check_value=self._check_value,
+                    property_chain=new_property_chain,
+                    cfn=cfn,
+                )
+            )
+            # check child exists separately when checking is_defined
+            if self.is_defined and properties.get(child_property, None) is None:
+                matches.append(
+                    cfnlint.rules.RuleMatch(
+                        path, error_message or f"{path} must be defined"
+                    )
+                )
+            return matches
+
+    return CustomIsDefinedRule(
+        rule_id,
+        resourceType,
+        prop,
+        value,
+        error_message,
+        shortdesc=f"Custom rule to check for value is {value}",
+        description=f"Created from the custom rules parameter. This rule will check if a property value is {value}",
+    )
+
+
 def CreateEqualsRule(rule_id, resourceType, prop, value, error_message):
     def rule_func(value, expected_value, path):
         matches = []

src/cfnlint/rules/custom/__init__.py

@@ -88,6 +88,14 @@ def process_sets(raw_value):
             return cfnlint.rules.custom.Operators.CreateLesserRule(
                 error_level + str(rule_id), resourceType, prop, value, error_message
             )
+        if operator == "IS":
+            if value in ["DEFINED", "NOT_DEFINED"]:
+                return cfnlint.rules.custom.Operators.CreateCustomIsDefinedRule(
+                    error_level + str(rule_id), resourceType, prop, value, error_message
+                )
+            return cfnlint.rules.custom.Operators.CreateInvalidRule(
+                "E" + str(rule_id), f"{operator} {value}"
+            )
 
     return cfnlint.rules.custom.Operators.CreateInvalidRule(
         "E" + str(rule_id), operator

test/fixtures/custom_rules/good/custom_rule_perfect.txt

@@ -5,4 +5,6 @@ AWS::IAM::Role AssumeRolePolicyDocument.Version NOT_IN [2012-10-16,2012-11-20,20
 AWS::IAM::Policy PolicyName EQUALS "root" WARN ABC
 AWS::IAM::Policy PolicyName IN [2012-10-16,root,2012-10-18] ERROR ABC
 AWS::IAM::Policy PolicyName NOT_EQUALS "user" WARN ABC
-AWS::IAM::Policy PolicyName NOT_IN [2012-10-16,2012-11-20,2012-10-18] ERROR ABC
+AWS::IAM::Policy PolicyName NOT_IN [2012-10-16,2012-11-20,2012-10-18] ERROR ABC
+AWS::Lambda::Function Environment.Variables.NODE_ENV IS DEFINED
+AWS::Lambda::Function Environment.Variables.PRIVATE_KEY IS NOT_DEFINED

test/fixtures/templates/good/custom/is-defined.yaml

@@ -0,0 +1,29 @@
+---
+AWSTemplateFormatVersion: "2010-09-09"
+Description: >
+  Testing for IS DEFINED rule
+Resources:
+  LambdaExecutionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: "2012-10-17"
+        Statement:
+          - Effect: Allow
+            Principal:
+              Service:
+                - lambda.amazonaws.com
+                - ec2.amazonaws.com
+            Action:
+              - sts:AssumeRole
+      Path: "/"
+  LambdaFunctionTestDefined:
+    Type: AWS::Lambda::Function
+    Properties:
+      Handler: index.handler
+      Role: !GetAtt LambdaExecutionRole.Arn
+      Environment:
+        Variables:
+          NODE_ENV: ""
+      Code: ./
+      Runtime: nodejs18.x

test/fixtures/templates/good/custom/is-not-defined.yaml

@@ -0,0 +1,36 @@
+---
+AWSTemplateFormatVersion: "2010-09-09"
+Description: >
+  Testing for IS NOT_DEFINED rule
+Resources:
+  LambdaExecutionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: "2012-10-17"
+        Statement:
+          - Effect: Allow
+            Principal:
+              Service:
+                - lambda.amazonaws.com
+                - ec2.amazonaws.com
+            Action:
+              - sts:AssumeRole
+      Path: "/"
+  LambdaFunctionTestNotDefined1:
+    Type: AWS::Lambda::Function
+    Properties:
+      Handler: index.handler
+      Role: !GetAtt LambdaExecutionRole.Arn
+      Code: ./
+      Runtime: nodejs18.x
+  LambdaFunctionTestNotDefined2:
+    Type: AWS::Lambda::Function
+    Properties:
+      Handler: index.handler
+      Role: !GetAtt LambdaExecutionRole.Arn
+      Environment:
+        Variables:
+          OTHER_VARS: ""
+      Code: ./
+      Runtime: nodejs18.x

test/fixtures/templates/good/generic.yaml

@@ -157,6 +157,21 @@ Resources:
       ResourceSignal:
         Timeout: PT15M
         Count: 1
+  LambdaFunction:
+    Type: AWS::Lambda::Function
+    Properties:
+      Code:
+        ZipFile: |
+          exports.handler = function(event, context) {
+            console.log('Hello World');
+            context.done(null, 'Hello World');
+          };
+      Environment:
+        Variables:
+          NODE_ENV: !Ref pEnvironment
+      Handler: index.handler
+      Role: "arn:aws:iam::123456789012:role/lambda_basic_execution"
+      Runtime: nodejs18.x
 Outputs:
   ElasticIP:
     Value: !Sub "${ElasticIP}/32"

test/unit/module/test_template.py

@@ -49,6 +49,7 @@ def test_build_graph(self):
 IamPipeline [color=black, label="IamPipeline\\n<AWS::CloudFormation::Stack>", shape=ellipse, type=Resource];
 CustomResource [color=black, label="CustomResource\\n<Custom::Function>", shape=ellipse, type=Resource];
 WaitCondition [color=black, label="WaitCondition\\n<AWS::CloudFormation::WaitCondition>", shape=ellipse, type=Resource];
+LambdaFunction [color=black, label="LambdaFunction\\n<AWS::Lambda::Function>", shape=ellipse, type=Resource];
 RolePolicies -> RootRole  [color=black, key=0, label=Ref, source_paths="['Properties', 'Roles', 0]"];
 RootInstanceProfile -> RootRole  [color=black, key=0, label=Ref, source_paths="['Properties', 'Roles', 0]"];
 MyEC2Instance -> RootInstanceProfile  [color=black, key=0, label=Ref, source_paths="['Properties', 'IamInstanceProfile']"];
@@ -71,7 +72,7 @@ def test_build_graph(self):
 
     def test_get_resources_success(self):
         """Test Success on Get Resources"""
-        valid_resource_count = 12
+        valid_resource_count = 13
         resources = self.template.get_resources()
         assert (
             len(resources) == valid_resource_count
@@ -121,7 +122,7 @@ def test_get_parameter_names(self):
 
     def test_get_valid_refs(self):
         """Get Valid REFs"""
-        valid_ref_count = 27
+        valid_ref_count = 28
         refs = self.template.get_valid_refs()
         assert len(refs) == valid_ref_count, "Expected {} refs, got {}".format(
             valid_ref_count, len(refs)

test/unit/rules/custom/test_is_defined.py

@@ -0,0 +1,36 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+from test.unit.rules import BaseRuleTestCase
+
+from cfnlint.rules.custom.Operators import CreateCustomIsDefinedRule
+
+
+class TestIsDefinedRule(BaseRuleTestCase):
+    """Test template mapping configurations"""
+
+    def setUp(self):
+        """Setup"""
+        super(TestIsDefinedRule, self).setUp()
+        self.collection.register(
+            CreateCustomIsDefinedRule(
+                "E9001",
+                "AWS::Lambda::Function",
+                "Environment.Variables.NODE_ENV",
+                "DEFINED",
+                "NODE_ENV should be defined",
+            )
+        )
+
+    success_templates = ["test/fixtures/templates/good/custom/is-defined.yaml"]
+
+    def test_file_positive(self):
+        """Test Positive"""
+        self.helper_file_positive()
+
+    def test_file_negative(self):
+        """Test failure"""
+        self.helper_file_negative(
+            "test/fixtures/templates/good/custom/is-not-defined.yaml", 2
+        )

test/unit/rules/custom/test_is_not_defined.py

@@ -0,0 +1,36 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+from test.unit.rules import BaseRuleTestCase
+
+from cfnlint.rules.custom.Operators import CreateCustomIsDefinedRule
+
+
+class TestIsDefinedRule(BaseRuleTestCase):
+    """Test template mapping configurations"""
+
+    def setUp(self):
+        """Setup"""
+        super(TestIsDefinedRule, self).setUp()
+        self.collection.register(
+            CreateCustomIsDefinedRule(
+                "E9001",
+                "AWS::Lambda::Function",
+                "Environment.Variables.NODE_ENV",
+                "NOT_DEFINED",
+                "NODE_ENV should not be defined",
+            )
+        )
+
+    success_templates = ["test/fixtures/templates/good/custom/is-not-defined.yaml"]
+
+    def test_file_positive(self):
+        """Test Positive"""
+        self.helper_file_positive()
+
+    def test_file_negative(self):
+        """Test failure"""
+        self.helper_file_negative(
+            "test/fixtures/templates/good/custom/is-defined.yaml", 1
+        )