Skip to content

Instantly share code, notes, and snippets.

@nivleshc
Last active October 9, 2019 11:31
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save nivleshc/f9b32a14d9e662701c3abcbb8f264306 to your computer and use it in GitHub Desktop.
Save nivleshc/f9b32a14d9e662701c3abcbb8f264306 to your computer and use it in GitHub Desktop.
This is the code for an AWS Lambda function. Amazon SES will invoke this lambda function to manage Amazon EC2 instances
import boto3
def provideHelp(params):
message = "Set the subject of the email to one of the following\n"
message += "help - provides this help\n"
message += "status - provides the status of all ec2 instances in the region\n"
message += "start {instance-id} - starts the ec2 instance with with the specified instance-id\n"
message += "stop {instance-id} - stops the ec2 instance with the specified instance id"
return message
def getStatus(params):
#get a list of all ec2 instances
print("getStatus:Params:",params)
#if a parameter is provided, it is the aws region to check
if (len(params) >= 1):
ec2 = boto3.client("ec2", params[0])
else:
ec2 = boto3.client("ec2")
response = ec2.describe_instances()
instances = response['Reservations']
number_instances = len(instances)
print("NumInstances:",number_instances)
message = ""
for instance in instances:
print("Instance:",str(instance))
instance_id = instance['Instances'][0]['InstanceId']
message += instance_id + "\t"
try:
if (instance['Instances'][0]['Tags'][0]['Key'] == 'Name'):
instance_name = instance['Instances'][0]['Tags'][0]['Value']
if (instance_name == ""):
message += "[No Name Found]\t"
else:
message += instance_name + "\t"
except:
message += "[No Name Found]\t"
instance_state = instance['Instances'][0]['State']['Name']
message += instance_state + "\t"
try:
instance_privateip = instance['Instances'][0]['PrivateIpAddress']
message += instance_privateip + "\t"
except:
message += "[No Private IP]\t"
try:
instance_publicip = instance['Instances'][0]['PublicIpAddress']
message += instance_publicip + "\t"
except:
message += "[No Public IP]\t"
message += "\n"
return message
def startInstance(params):
if (len(params) > 1):
instanceId = params[0]
aws_region = params[1]
ec2 = boto3.client("ec2", aws_region)
else:
instanceId = params[0]
ec2 = boto3.client("ec2")
try:
response = ec2.start_instances(
InstanceIds=[instanceId]
)
message = "Starting instance " + str(instanceId) + " Please check status in a couple of minutes\n" + str(response)
except Exception as e:
message = "Error starting instance(s) :" + str(instanceId) + " " + str(e)
print(message)
return message
def stopInstance(params):
if (len(params) > 1):
instanceId = params[0]
aws_region = params[1]
ec2 = boto3.client("ec2", aws_region)
else:
instanceId = params[0]
ec2 = boto3.client("ec2")
try:
response = ec2.stop_instances(
InstanceIds=[instanceId]
)
message = "Stopping instance " + str(instanceId) + " Please check status in a couple of minutes\n" + str(response)
except Exception as e:
message = "Error stopping instance(s) " + str(instanceId) + " " + str(e)
print(message)
return message
def sendEmail(fromAddress,recipientAddress,subject,body):
#sending email
print("sending email")
client = boto3.client('ses')
response = client.send_email(
Destination={
'ToAddresses': [recipientAddress],
},
Message={
'Body': {
'Text': {
'Charset': 'UTF-8',
'Data': body
},
},
'Subject': {
'Charset': 'UTF-8',
'Data': subject
},
},
ReplyToAddresses=[fromAddress],
ReturnPath=fromAddress,
Source=fromAddress
)
print("Response from SES:",response)
def lambda_handler(event, context):
approvedSenders = ['john@example.com','tom@example.com','jane@example']
print("Event:",event)
print("Context",context)
emailSender = event["Records"][0]["ses"]["mail"]["source"]
emailSubject = event["Records"][0]["ses"]["mail"]["commonHeaders"]["subject"]
command_split = emailSubject.split(" ")
command = command_split[0].lower()
if (len(command_split) > 1):
commandParams = command_split[1:len(command_split)]
else:
commandParams = "" #if there are no command params, then just set it to blank
print("From:",emailSender)
print("Subject:",emailSubject)
print("Command:",command)
#authenticate the sender based on fromAddress
if emailSender in approvedSenders:
switcher = {
"help": provideHelp,
"status": getStatus,
"start": startInstance,
"stop": stopInstance
}
#get the command that was specified in the email
functionToRun = switcher.get(command, lambda params: "invalid command")
body = functionToRun(commandParams)
sendEmail("admin@managedinstances.com",emailSender,"Execution result for command:"+ emailSubject, str(body))
else:
print("Sender ",emailSender," not approved for executing commands. Ignore")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment