Tag: python3

I’m using an Amazon Web Services Security Group as a way to allow traffic into an EC2 instance for the instance’s users. The users can give themselves access through a web interface that I wrote for them. Maybe I’ll cover that in a different post.

I found recently that the Security Group was nearing its maximum rule list. So I decided to start purging rules which would ultimately force them to re-add their IP addresses to the group.

Going in and manually removing rules is rather time-consuming. I figured I could write a script that would handle it for me. The first step was to update my previous script that inserts the rule to add a tag to the rule. The function below takes input of Security Group Id’s as a list and returns all of the rules.

def get_sg_rules(sg_id):
    client = boto3.client('ec2')
    response = client.describe_security_group_rules(
    Filters=[
        {
            'Name': 'group-id',
            'Values': sg_id
        }
    ],
    )
    return response

The script below iterates through each of the rules returned and will append the tag of “dateAdded” and a stringified date code.

for sg_rule in get_sg_rules(sg_list)['SecurityGroupRules']:
        try:
            client = boto3.client('ec2')
            response = client.create_tags(
            DryRun=False,
            Resources=[
                sg_rule['SecurityGroupRuleId'],
            ],
            Tags=[
                {
                    'Key': 'dateAdded',
                    'Value': '2022-11-05'
                },
            ]
        )
        except ClientError as e:
            print(e)

I then wrote the following Lambda function that runs every day and checks for any expired rules. The schedule is set up by a Cloudwatch Event’s rule.

import boto3
from datetime import datetime, timedelta
from botocore.exceptions import ClientError
def return_today():
    now = datetime.now()
    return now
def get_sg_rules(sg_id, old_date):
    client = boto3.client('ec2')
    response = client.describe_security_group_rules(
    Filters=[
        {
            'Name': 'group-id',
            'Values': sg_id
        },
        {
            'Name': 'tag:dateAdded',
            'Values': [old_date]
        }
    ],
    )
    
    return response

def lambda_handler(event, context):
    sg_list = ["xxxx", "xxx"]
    old_date = datetime.strftime(return_today() - timedelta(days=30), "%Y-%m-%d")
    print(old_date)
    for sg_rule in get_sg_rules(sg_list, old_date)['SecurityGroupRules']:
        try:
            client = boto3.client("ec2")
            response = client.revoke_security_group_ingress(
                GroupId=sg_rule['GroupId'],
                SecurityGroupRuleIds=[sg_rule['SecurityGroupRuleId']]
                )
            print(response)
            print("Successfully deleted the rule")
        except ClientError as e:
            print(e)
            print("Failed to delete rule")

You’ll see that the code has a list of Security Groups to check. It compares the current date to that of 30 days previous. If the tag of “dateAdded” matches that previous date then we will go ahead and remove the rule.

I hope this helps you automate your AWS Accounts. Below are links to the code repository so you can edit the code as needed. Please share it with your friends if this helps you!

Github

I took a brief break from my Lambda function creation journey to go on vacation but, now i’m back!

This function will notify a Slack channel of your choosing when an EC2 instance enters “Starting, Stopping, Stopped, or Shutting-Down” status. I thought this might be useful for instances that reside under a load balancer. It would be useful to see when your load balancer is scaling up or down in real-time via Slack notification.

In order to use this function, you will need to create a Slack Application with an OAuth key and set that key as an environment variable in your Lambda function. If you are unsure of how to do this I can walk you through it!

Please review the function below

import logging
import requests
import boto3
import os
from urllib.parse import unquote_plus
from slack import WebClient
from slack.errors import SlackApiError
logging.basicConfig(level=logging.DEBUG)

# Check EC2 Status
def lambda_handler(event, context):
    detail = event['detail']
    ids = detail['instance-id']
    eventname = detail['state']
    ec2 = boto3.resource('ec2')
# Slack Variables
    slack_token = os.environ["slackBot"]
    client = WebClient(token=slack_token)
    channel_string = "XXXXXXXXXXXXXXXXXXXX"

# Post to slack that the instance is running
    if eventname == 'running':
        try:
          instance = ids
          response_string = f"The instance: {instance} has started"
          response = client.chat_postMessage(
            channel= channel_string,
          	text="An Instance has started",
           	blocks = [{"type": "section", "text": {"type": "plain_text", "text": response_string}}]
	        	)
        except SlackApiError as e:
          assert e.response["error"]  

		#Post to slack that instance is shutting down
    elif eventname == 'shutting-down':
    	try:
	        instance = ids
	        response_string = f"The instance: {instance} is shutting down"
	        response = client.chat_postMessage(
	        	channel= channel_string,
	        	text="An Instance is Shutting Down",
	        	blocks = [{"type": "section", "text": {"type": "plain_text", "text": response_string}}]
	        	)
    	except SlackApiError as e:
           assert e.response["error"]
	      	
    elif eventname == 'stopped':
    	try:
	        instance = ids
	        response_string = f"The instance: {instance} has stopped"
	        response = client.chat_postMessage(
	        	channel= channel_string,
	        	text="An Instance has stopped",
	        	blocks = [{"type": "section", "text": {"type": "plain_text", "text": response_string}}]
	        	)
    	except SlackApiError as e:
    		assert e.response["error"]
	      	
    elif eventname == 'stopping':
    	try:
	        instance = ids
	        response_string = f"The instance: {instance} is stopping"
	        response = client.chat_postMessage(
	        	channel= channel_string,
	        	text="An Instance is stopping",
	        	blocks = [{"type": "section", "text": {"type": "plain_text", "text": response_string}}]
	        	)
    	except SlackApiError as e:
    		assert e.response["error"]

As always the function is available on GitHub as well:
https://github.com/avansledright/ec2ActionPostToSlack

If you find this function helpful please share it with your friends or repost it on your favorite social media platform!