Simpline Blog[AWS]AWS Configでルール違反検知　その３

# Description: Check that security groups do not have an inbound rule
#              with port except permitted.
#
# Trigger Type: Change Triggered
# Scope of Changes: EC2:SecurityGroup
# Accepted Parameters: examplePort1, examplePort2, ...
# Example Values: 22,80,443, ...
#
# You can designate some ports permitted by fill in okPort.
# note: It doesn't accept by Ranges.


import boto3
import json

APPLICABLE_RESOURCES = ["AWS::EC2::SecurityGroup"]

def evaluate_compliance(configuration_item, rule_parameters):

    # Check if resource was deleted
    if configuration_item['configurationItemStatus'] == "ResourceDeleted":
        compliance_type = 'NOT_APPLICABLE'
        annotation = "This resource was deleted."

    # Check the resource for applicability
    elif configuration_item["resourceType"] not in APPLICABLE_RESOURCES:
        compliance_type = 'NOT_APPLICABLE'
        annotation = "The rule apply to only resources of type SecurityGroup."

    else:
        # Check IP violation
        for ipsg in configuration_item['configuration']['ipPermissions']:
            if 'fromPort' not in ipsg:
                compliance_type = 'NON_COMPLIANT'
                annotation = 'Exposed ports range is ALL.'
                break
            
            else:
                fp = ipsg['fromPort']
                rp = []
                rp = rule_parameters['okPort'].split(',')
                if str(fp) not in rp:
                    compliance_type = 'NON_COMPLIANT'
                    annotation = 'A forbidden port is exposed.'
                    break
                
                else:
                    compliance_type = 'COMPLIANT'
                    annotation = 'Security group is compliant.'

    return {
        "compliance_type": compliance_type,
        "annotation": annotation
    }


def lambda_handler(event, context):

    invoking_event = json.loads(event['invokingEvent'])
    rule_parameters = json.loads(event['ruleParameters'])
    configuration_item = invoking_event["configurationItem"]
    evaluation = evaluate_compliance(configuration_item, rule_parameters)
    config = boto3.client('config')

    print('Compliance evaluation for %s: %s' % (configuration_item['resourceId'], evaluation["compliance_type"]))
    print('Annotation: %s' % (evaluation["annotation"]))

    response = config.put_evaluations(
       Evaluations=[
           {
               'ComplianceResourceType': invoking_event['configurationItem']['resourceType'],
               'ComplianceResourceId':   invoking_event['configurationItem']['resourceId'],
               'ComplianceType':         evaluation["compliance_type"],
               "Annotation":             evaluation["annotation"],
               'OrderingTimestamp':      invoking_event['configurationItem']['configurationItemCaptureTime']
           },
       ],
       ResultToken=event['resultToken'])

★「fromPort」の例

80番ポートを開けた場合
→「80」

22番〜80番ポートを開けた場合
→「22」

セキュリティグループの変更をトリガーとして発行されるイベントには、セキュリティグループの既存ルールも含めてデータが格納されています。