Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Way Three: Automatically Configure EC2 Instances
AWSTemplateFormatVersion: '2010-09-09'
Description: Creates an event rule that detects CreateVpc and emails an administrator.
Parameters:
SecurityGroupId:
Type: String
SecurityGroupName:
Type: String
Resources:
RunInstancesRule:
Type: AWS::Events::Rule
Properties:
Description: 'Cloudwatch event required to implement this rule.'
Name: 'DetectRunInstances'
EventPattern:
source:
- aws.ec2
detail-type:
- "AWS API Call via CloudTrail"
detail:
eventSource:
- ec2.amazonaws.com
eventName:
- RunInstances
State: 'ENABLED'
Targets:
-
Arn: !GetAtt RunInstancesLambda.Arn
Id: 'RunInstancesTarget'
LambdaInvokePermission:
Type: AWS::Lambda::Permission
Properties:
FunctionName: !GetAtt RunInstancesLambda.Arn
Action: 'lambda:InvokeFunction'
Principal: events.amazonaws.com
SourceArn: !GetAtt RunInstancesRule.Arn
LambdaLogGroup:
Type: AWS::Logs::LogGroup
DependsOn: RunInstancesLambda
Properties:
LogGroupName: "/aws/lambda/run-instances-lambda"
RetentionInDays: 14
LambdaRole:
Type: AWS::IAM::Role
Properties:
RoleName: "EC2EventRuleRole"
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
- events.amazonaws.com
Action:
- sts:AssumeRole
Path: "/"
ManagedPolicyArns:
- 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
Policies:
- PolicyName: root
PolicyDocument:
Version: '2012-10-17'
Statement:
- Sid: AllowEC2ApiCalls
Effect: Allow
Action:
- ec2:ModifyInstanceAttribute
- ec2:CreateTags
- logs:CreateLogStream
- logs:PutLogEvents
Resource: "*"
RunInstancesLambda:
Type: AWS::Lambda::Function
Properties:
FunctionName: 'run-instances-lambda'
Timeout: 40
Handler: index.lambda_handler
Runtime: python3.6
MemorySize: 128
Code:
ZipFile:
!Sub |
import boto3
import uuid
import os
REQUIRED_TAGS = {'INTERNAL_ID': str(uuid.uuid4()), 'ADMIN': 'admin@example.com'}
REQUIRED_SECURITY_GROUPS = {'groupId': os.environ['ADMIN_SG_ID'],
'groupName': os.environ['ADMIN_SG_NAME']}
EC2_CLIENT = boto3.client('ec2')
class EC2Validator:
def __init__(self, event):
self.raw_event = event
self.item_set = event['detail']['responseElements']['instancesSet']['items']
@staticmethod
def create_missing_tags(missing_keys, instance_id):
tagset_to_create = []
for key in missing_keys:
tagset_to_create.append({'Key': key, 'Value': REQUIRED_TAGS[key]})
EC2_CLIENT.create_tags(Resources=[instance_id],
Tags=tagset_to_create)
@staticmethod
def collect_missing_keys(tagset):
tag_keys = []
for tag in tagset:
tag_keys.append(tag['key'])
return REQUIRED_TAGS.keys() - set(tag_keys)
def validate_tags(self):
for instance_item in self.item_set:
instance_id = instance_item['instanceId']
if instance_item['tagSet']:
missing_keys = EC2Validator.collect_missing_keys(instance_item['tagSet']['items'])
if missing_keys:
EC2Validator.create_missing_tags(missing_keys, instance_id)
else:
EC2Validator.create_missing_tags(REQUIRED_TAGS.keys(), instance_id)
@staticmethod
def collect_missing_groups(original_groupset):
group_ids = []
for group in original_groupset:
group_ids.append(group['groupId'])
return group_ids
def validate_required_security_group(self):
for instance_item in self.item_set:
current_groups = instance_item['groupSet']['items']
missing_groups = EC2Validator.collect_missing_groups(current_groups)
if REQUIRED_SECURITY_GROUPS['groupId'] not in missing_groups:
missing_groups.append(REQUIRED_SECURITY_GROUPS['groupId'])
EC2_CLIENT.modify_instance_attribute(InstanceId=instance_item['instanceId'],
Groups=missing_groups)
def lambda_handler(event, context):
ec2_validator = EC2Validator(event)
ec2_validator.validate_tags()
ec2_validator.validate_required_security_group()
Role: !GetAtt LambdaRole.Arn
Environment:
Variables:
ADMIN_SG_ID: !Ref SecurityGroupId
ADMIN_SG_NAME: !Ref SecurityGroupName
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.