Lambda function for Config service. Watches an EC2 security group for changes and evaluates.

evaluate-compliance.py

import json
import boto3
from datetime import datetime

# Use as a starter/stub -- can and should be modified for production
FORMAT = '%Y%m%d-%H:%M:%S'
now = datetime.now().strftime(FORMAT)

def lambda_handler(event, context):
    invoking_event = json.loads(event['invokingEvent'])
    resourceId = invoking_event['configurationItem']['resourceId']
    captureTime = invoking_event['configurationItem']['configurationItemCaptureTime']
    for item in invoking_event['configurationItem']['configuration']['ipPermissions']:
        toPort = item['toPort']
        secgrp = invoking_event['configurationItem']['configuration']['groupId']
        if toPort == 22:
            for cidr in item['ipv4Ranges']:
                bcidr = cidr['cidrIp']
                if bcidr == '0.0.0.0/0':
                    arn = invoking_event['configurationItem']['ARN']
                    print("CIDR Violation ARN: " + invoking_event['configurationItem']['ARN'])
                    sendmsg(arn)
                    dbinsert(secgrp, '1')
                    update_config(resourceId, 'NON_COMPLIANT', captureTime, event)
                else:
                    print("CIDR compliance met")
                    update_config(resourceId, 'COMPLIANT', captureTime, event)
        else:
            print("SG compliance met")
            update_config(resourceId, 'COMPLIANT', captureTime, event)
                  
def sendmsg(arn):
    client = boto3.client('sns')
    response = client.publish(
        TopicArn='arn:aws:sns:us-east-1:474683445819:cidr-ssh',
        Message='Port 22 violation for ' + str(arn),
        Subject='CIDR Violation'
    )
    
def dbinsert(secgrp, status):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('cidr-status')
    table.put_item(
    Item={
        'security-group': secgrp,
        'status': status,
        'when': now,
    }
)
    
def update_config(resourceId, compliance_type, captureTime, event):
    result_token = "No token found."
    if "resultToken" in event:
        result_token = event["resultToken"]
        config = boto3.client("config")
        config.put_evaluations(
            Evaluations=[
                {
                    "ComplianceResourceType": "AWS::EC2::SecurityGroup",
                    "ComplianceResourceId": resourceId,
                    "ComplianceType": compliance_type,
                    "Annotation": "A simple annotation",
                    "OrderingTimestamp": captureTime
                },
            ],
        ResultToken=result_token
    )