Python 100 project #42: Slack Bot – AWS EC2 list

Following up the previous project, I created Slack bot to get EC2 instance list (of all regions) in one shot.

So now no need to open the terminal to invoke the command every time. Just need to ask Slack “/100p ec2 list” and the result is posted.

I used AWS API Gateway to receive the slash command from Slack. So it is easy to add functions.

Output Example:

Here is the code:

This is the receiver code which is invoked when Slack slach command post request to API Gateway.

from base64 import b64decode
import json
import os
from urllib.parse import parse_qs
import logging

import boto3


ENCRYPTED_EXPECTED_TOKEN = "kms_base64encodedkey="

kms = boto3.client('kms')
expected_token = str(kms.decrypt(CiphertextBlob = b64decode(ENCRYPTED_EXPECTED_TOKEN))['Plaintext'], 'utf-8')


logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    req_body = event['body']
    params = parse_qs(req_body)
    print("received data...", params)
    token = params['token'][0]
    if token != expected_token:
        logger.error("Request token (%s) does not match exptected", token)
        raise Exception("Invalid request token")

    user = params['user_name'][0]
    command = params['command'][0]
    channel = params['channel_name'][0]
    if 'text' in params.keys():
        command_text = params['text'][0]
    else:
        command_text = ''
    response_url = params['response_url'][0]
    arg = command_text.split(' ')
    
    sns = boto3.client('sns')
    SNS_CHANNEL = os.environ['SNS_CHANNEL']
    topic_arn = sns.create_topic(Name=SNS_CHANNEL)['TopicArn']
    message={"user_name": user, "command": command, "channel": channel, "command_text": command_text, "response_url": response_url}
    message=json.dumps(message)
    message=json.dumps({'default': message, 'lambda': message})
    response = sns.publish(
        TopicArn=topic_arn,
        Subject='/100p',
        MessageStructure='json',
        Message=message
    )
    return { "text": "received command - %s . Please wait for a few seconds for the reply to be posted." % (command_text) }

from base64 import b64decode

import json

import os

from urllib.parse import parse_qs

import logging

import boto3

ENCRYPTED_EXPECTED_TOKEN = "kms_base64encodedkey="

kms = boto3.client('kms')

expected_token = str(kms.decrypt(CiphertextBlob = b64decode(ENCRYPTED_EXPECTED_TOKEN))['Plaintext'], 'utf-8')

logger = logging.getLogger()

logger.setLevel(logging.INFO)

def lambda_handler(event, context):

req_body = event['body']

params = parse_qs(req_body)

print("received data...", params)

token = params['token'][0]

if token != expected_token:

logger.error("Request token (%s) does not match exptected", token)

raise Exception("Invalid request token")

user = params['user_name'][0]

command = params['command'][0]

channel = params['channel_name'][0]

if 'text' in params.keys():

command_text = params['text'][0]

else:

command_text = ''

response_url = params['response_url'][0]

arg = command_text.split(' ')

sns = boto3.client('sns')

SNS_CHANNEL = os.environ['SNS_CHANNEL']

topic_arn = sns.create_topic(Name=SNS_CHANNEL)['TopicArn']

message={"user_name": user, "command": command, "channel": channel, "command_text": command_text, "response_url": response_url}

message=json.dumps(message)

message=json.dumps({'default': message, 'lambda': message})

response = sns.publish(

TopicArn=topic_arn,

Subject='/100p',

MessageStructure='json',

Message=message

)

return { "text": "received command - %s . Please wait for a few seconds for the reply to be posted." % (command_text) }

And this is the actual code to post the result to the Slack.

import json
import sys

import boto3
import requests

def get_regions(service):
    credential = boto3.session.Session()
    return credential.get_available_regions(service)


def list_ec2_servers(region):
    credential = boto3.session.Session()
    ec2 = credential.client('ec2', region_name=region)
    instances = ec2.describe_instances()
    servers_list = []
    for reservations in instances['Reservations']:
        for instance in reservations['Instances']:
            tags = parse_keyvalue_sets(instance['Tags'])
            state = instance['State']['Name']
            servers_list.append([region, instance['InstanceId'], tags['Name'], state])
    return servers_list


def parse_keyvalue_sets(tags):
    result = {}
    for tag in tags:
        key = tag['Key']
        val = tag['Value']
        result[key] = val
    return result


def lambda_handler(event, context):
    message = event['Records'][0]['Sns']['Message']
    try:
        message = json.loads(message)
        user_name = message['user_name']
        command = message['command']
        command_text = message['command_text']
        response_url = message['response_url']
        arg = command_text.split(' ')

        if arg[0] == 'ec2':
            resp = ec2_helper(arg[1:])
        # TODO else: statement for other functions

        # if response_type is not specified, act as the same as ephemeral
        # ephemeral, response message will be visible only to the user
        slack_message = {
            'channel': '@%s' % user_name,
            # 'response_type': 'in_channel',
            'response_type': 'ephemeral',
            'isDelayedResponse': 'true',
            'text': resp
        }
        print("Send message to %s %s" % (response_url, slack_message))
        header = {'Content-Type': 'application/json'}
        response = requests.post(response_url, headers=header, data=json.dumps(slack_message))
        if response.status_code == 200:
            print("Message posted to %s" % slack_message['channel'])
    except requests.exceptions.RequestException as e:
        print(e)
    except:
        e = sys.exc_info()[0]
        print("Something wrong happened...", e)

def ec2_helper(command):
    regions = get_regions('ec2')

    if command[0] == 'list':
        region_servers = []
        for region in regions:
            servers = list_ec2_servers(region)
            region_servers.extend(servers)

        msg = ""
        for server in region_servers:
            msg += '\t'.join(server)
            msg += "\n"
    # TODO else for other functions

    return msg

import json

import sys

import boto3

import requests

def get_regions(service):

credential = boto3.session.Session()

return credential.get_available_regions(service)

def list_ec2_servers(region):

credential = boto3.session.Session()

ec2 = credential.client('ec2', region_name=region)

instances = ec2.describe_instances()

servers_list = []

for reservations in instances['Reservations']:

for instance in reservations['Instances']:

tags = parse_keyvalue_sets(instance['Tags'])

state = instance['State']['Name']

servers_list.append([region, instance['InstanceId'], tags['Name'], state])

return servers_list

def parse_keyvalue_sets(tags):

result = {}

for tag in tags:

key = tag['Key']

val = tag['Value']

result[key] = val

return result

def lambda_handler(event, context):

message = event['Records'][0]['Sns']['Message']

try:

message = json.loads(message)

user_name = message['user_name']

command = message['command']

command_text = message['command_text']

response_url = message['response_url']

arg = command_text.split(' ')

if arg[0] == 'ec2':

resp = ec2_helper(arg[1:])

# TODO else: statement for other functions

# if response_type is not specified, act as the same as ephemeral

# ephemeral, response message will be visible only to the user

slack_message = {

'channel': '@%s' % user_name,

# 'response_type': 'in_channel',

'response_type': 'ephemeral',

'isDelayedResponse': 'true',

'text': resp

}

print("Send message to %s %s" % (response_url, slack_message))

header = {'Content-Type': 'application/json'}

response = requests.post(response_url, headers=header, data=json.dumps(slack_message))

if response.status_code == 200:

print("Message posted to %s" % slack_message['channel'])

except requests.exceptions.RequestException as e:

print(e)

except:

e = sys.exc_info()[0]

print("Something wrong happened...", e)

def ec2_helper(command):

regions = get_regions('ec2')

if command[0] == 'list':

region_servers = []

for region in regions:

servers = list_ec2_servers(region)

region_servers.extend(servers)

msg = ""

for server in region_servers:

msg += '\t'.join(server)

msg += "\n"

# TODO else for other functions

return msg

Share this:

Related