MyCheck.py

from cfnlint import CloudFormationLintRule
from cfnlint import RuleMatch
import yaml
import jmespath

class MyCheck(CloudFormationLintRule):

    """MyCheck"""
    id = 'E737'
    shortdesc = 'Check Something'
    description = 'Blah, blah, blah'
    source_url = 'https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/'
    tags = ['resources']

    def match(self, cfn):
        """interface for cfn-lint to use this rule"""
        matches = []
        resources = cfn.get_resources(['AWS::EC2::Instance'])
        for resource_name, resource in resources.items():
            violations = self.apply_rules(resource_name, resource)
            for violation in violations:
                path = ['Resources', resource_name]
                matches.append(RuleMatch(path, resource_name + ": " +  violation['RuleMessage']))
        return matches

    def apply_rules(self, resource_name, resource):
        """Check the resource using a set of built in rules"""
        violations = []
        for rule in self.getRules()['rules']:
            rule_violations = self.apply_rule(rule, resource_name, resource)
            violations = violations + rule_violations
        return violations

    def apply_rule(self, rule, resource_name, resource):
        """Check a resource using a single rule"""
        #print resource
        violations = []
        if resource['Type'] == rule['resource']:
            for assertion in rule['assertions']:
                if 'some' in assertion:
                    if not self.some(rule, assertion['some'], resource_name, resource):
                        violations += [
                            {
                                'Status': 'FAILURE',
                                'ResourceId': resource_name,
                                'RuleMessage': rule['message']
                            }
                        ]
                else:
                    violations += self.expr(rule, assertion, resource_name, resource['Properties'])

        return violations

    def some(self, rule, specification, resource_name, resource):
        """Rule passes if any of the expressions return true"""
        values = jmespath.search(specification['key'], resource['Properties'])
        if not values:
            return False
        for value in values:
            for expression in specification['expressions']:
                if len(self.expr(rule, expression, resource_name, value)) == 0:
                    return True
        return False

    def expr(self, rule, assertion, resource_name, properties):
        """Evaluate the expression for a rule"""
        funcs = {
            'eq': self.eq,
            'in': self.inList,
            'present': self.present
        }
        violations = []
        value = None
        if 'key' in assertion:
            value = jmespath.search(assertion['key'], properties)
        #print a['key'], value
        operation = assertion.get('op', 'Unknown')
        if operation in funcs:
            if not funcs[assertion['op']](assertion.get('value', None), value):
                violations += [
                    {
                        'Status':'FAILURE',
                        'ResourceId': resource_name,
                        'RuleMessage':rule['message']
                    }
                ]
        else:
            violations += [
                {
                    'Status': 'FAILURE',
                    'ResourceId': resource_name,
                    'RuleMessage': 'Unknown op: ' + operation
                }
            ]

        return violations

    def eq(self, rule_value, resource_value):
        """Test for equality"""
        if rule_value != resource_value:
            return False
        return True

    def inList(self, rule_value, resource_value):
        """Test for element in a list"""
        if resource_value in rule_value.split(','):
            return True
        return False

    def present(self, rule_value, resource_value):
        """Test that attribute is present"""
        return resource_value != None

    def getRules(self):
        rule_set_string = """version: 1.0
description: Rules for use in cfn-lint
type: CloudFormation
rules:
  - id: NAME_TAG
    message: EC2 instance should have a Name tag
    resource: AWS::EC2::Instance
    assertions:
        - some:
            key: Tags[]
            expressions:
              - key: Key
                op: eq
                value: "Name"
  - id: ENV_TAG
    message: EC2 instance should have a Env tag
    resource: AWS::EC2::Instance
    assertions:
        - some:
            key: Tags[]
            expressions:
              - key: Key
                op: eq
                value: "Env"
  - id: INSTANCE_TYPE
    message: Invalid instance type
    resource: AWS::EC2::Instance
    assertions:
        - key: InstanceType
          op: in
          value: "t2.small,c3.medium"
"""
        return yaml.load(rule_set_string)