In this tutorial, we are going to write a custom AWS Config rule in Lambda (Python). Our rule will remove public accessible CIDR blocks from security groups which shouldn’t be public. When a security group is created or modified, the Config triggers the Lambda function which will then take the necessary action.
What is AWS Config?
AWS Config monitors the resources in the AWS Account. If a change is detected it can send you an email, or execute a Lambda function. At the time same it maintains a record of changes. The power comes when you start writing custom Lambda functions to make necessary changes to a resource. You could write the Lambda functions in the languages supported by AWS.
We will be writing the Lambda function in Python. The function will be triggered when a security group is modified or created. The triggering aspect is fully managed by AWS Config. The function will check if the security group has a public accessible rule and remove it. However, some security group needs to be public as a result we will configure an exclusion list of such security groups.
The Lambda
When a security group is modified, AWS Config publishes an event which passes a request to the Lambda. This request has an important attribute “invokingEvent“. Click here to download the Lambda script.
This contains the resource which caused AWS Config to trigger the Lambda. What we care about is the resourceId in the invokingEvent.
{
"version": "1.0",
"invokingEvent": "{\"configurationItemDiff\":null,\"configurationItem\":{\"relatedEvents\":[],\"relationships\":[{\"resourceId\":\"vpc-23466d45\",\"resourceName\":null,\"resourceType\":\"AWS::EC2::VPC\",\"name\":\"Is contained in Vpc\"}],\"configuration\":{\"description\":\"launch-wizard-1 created 2019-02-04T11:29:56.753+00:00\",\"groupName\":\"Office-Connection-Only\",\"ipPermissions\":[{\"ipProtocol\":\"-1\",\"ipv6Ranges\":[],\"prefixListIds\":[],\"userIdGroupPairs\":[],\"ipv4Ranges\":[{\"cidrIp\":\"109.176.80.138/32\"}],\"ipRanges\":[\"109.176.80.138/32\"]}],\"ownerId\":\"453151231029\",\"groupId\":\"sg-0802d47fff1606208\",\"ipPermissionsEgress\":[{\"ipProtocol\":\"-1\",\"ipv6Ranges\":[],\"prefixListIds\":[],\"userIdGroupPairs\":[],\"ipv4Ranges\":[{\"cidrIp\":\"0.0.0.0/0\"}],\"ipRanges\":[\"0.0.0.0/0\"]}],\"tags\":[{\"key\":\"Name\",\"value\":\"public\"}],\"vpcId\":\"vpc-23466d45\"},\"supplementaryConfiguration\":{},\"tags\":{\"Name\":\"public\"},\"configurationItemVersion\":\"1.3\",\"configurationItemCaptureTime\":\"2019-05-10T15:44:01.369Z\",\"configurationStateId\":1557535440800,\"awsAccountId\":\"453151231029\",\"configurationItemStatus\":\"ResourceDiscovered\",\"resourceType\":\"AWS::EC2::SecurityGroup\",\"resourceId\":\"sg-0802d47fff1606208\",\"resourceName\":\"Office-Connection-Only\",\"ARN\":\"arn:aws:ec2:eu-west-1:453151231029:security-group/sg-0802d47fff1606208\",\"awsRegion\":\"eu-west-1\",\"availabilityZone\":\"Not Applicable\",\"configurationStateMd5Hash\":\"\",\"resourceCreationTime\":null},\"notificationCreationTime\":\"2019-05-11T12:36:08.921Z\",\"messageType\":\"ConfigurationItemChangeNotification\",\"recordVersion\":\"1.3\"}",
"ruleParameters": "{}",
"eventLeftScope": false,
"executionRoleArn": "arn:aws:iam::453151231029:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig",
"configRuleArn": "arn:aws:config:eu-west-1:453151231029:config-rule/config-rule-lakcds",
"configRuleName": "monitor-sg",
"configRuleId": "config-rule-lakcds",
"accountId": "4531523231029"
}
Lambda handler
Our Lambda handler is plain and simple. We decode the response received and pass it to our method which will perform the logic.
The “put_evaluations” is required. Every Lambda function needs to call this method to inform AWS about the evaluation outcome. This is very important as we would like to inform it about the outcome of the script.
def lambda_handler(event, context):
print(event)
# decode the aws confing response
invoking_event = json.loads(event['invokingEvent'])
configuration_item = invoking_event["configurationItem"]
# pass the configuration item to our method
evaluation = evaluate_compliance(configuration_item)
config = boto3.client('config')
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'])
Evaluation of Compliance code
def evaluate_compliance(configuration_item):
if configuration_item["resourceType"] not in APPLICABLE_RESOURCES:
return {
"compliance_type": "NOT_APPLICABLE",
"annotation": "The rule doesn't apply to resources of type " +
configuration_item["resourceType"] + "."
}
if configuration_item["configurationItemStatus"] == "ResourceDeleted":
return {
"compliance_type": "NOT_APPLICABLE",
"annotation": "The configurationItem was deleted and therefore cannot be validated."
}
group_id = configuration_item["configuration"]["groupId"]
client = boto3.client("ec2");
try:
response = client.describe_security_groups(GroupIds=[group_id])
except botocore.exceptions.ClientError as e:
return {
"compliance_type": NON_COMPLIANT,
"annotation": "describe_security_groups failure on group " + group_id
}
protocol_all = False
if group_id not in SECURITY_GROUPS_ALLOWED_PUBLIC_ACCESS:
# lets find public accessible CIDR Blocks
for security_group_rule in response["SecurityGroups"][0]["IpPermissions"]:
# if the rule is all protocol, FromPort is missing
if "FromPort" not in security_group_rule:
protocol_all = True
for sgName, val in security_group_rule.items():
if sgName == "IpRanges":
for r in val:
if r["CidrIp"] == "0.0.0.0/0":
if not protocol_all:
client.revoke_security_group_ingress(GroupId=group_id,
IpProtocol=security_group_rule["IpProtocol"],
CidrIp=r["CidrIp"],
FromPort=security_group_rule["FromPort"],
ToPort=security_group_rule["ToPort"])
else:
client.revoke_security_group_ingress(GroupId=group_id,
IpProtocol=security_group_rule["IpProtocol"],
CidrIp=r["CidrIp"])
compliance_type = NON_COMPLIANT
annotation_message = "Permissions were modified"
else:
compliance_type = COMPLIANT
annotation_message = "Permissions are correct"
return {
"compliance_type": compliance_type,
"annotation": annotation_message
}
AWS Config vs CloudTrail
Config monitors the resources in your AWS account and can trigger alerts and events. With config, you could automatically run security compliance code on resources. Config allows this functionality by a terminology called Rules. Rules are the desired configuration settings of the AWS resources. As of now, there are 84 managed rules but you could add your custom rule if they don’t exist. The screenshot below is of the Rules section of Config.
Cloudtrail is a service which records all AWS API calls. The API calls could be from AWS SDK, CLI or Management console. For example when an EC2 instance is launched. This is recorded in Cloudtrail as an event. 
Which resources are supported by AWS Config?
AWS config supports many AWS resources.
Few of them are:
- API Gateway
- CloudFront
- CloudWatch
- DynamoDB
- EC2
- Security Groups
- Network ACL
- Redshift
- Lambda function
I’m a passionate engineer based in London.
Currently, I’m working as a Cloud Consultant at Contino.
Aside my full time job, I either work on my own startup projects or you will see me in a HIIT class 🙂