Tag: ec2

  • Updating AWS Managed Prefix Lists

    I was working with a customer the other day trying to come up with a way to import a bunch of IP addresses into a white list on AWS. We came up with the approach of using Managed Prefix Lists in VPC. I wrote some Python in order to grab some code from an API and then automatically put it into a prefix list.

    The code takes input from an API that is managed by a 3rd party. We first use that and parse the returned values into meaningful lists. After that, we pass the IPs to the function which it will check if the entry exists or not. If it does, it will pass the IP. If it doesn’t exist it will automatically add it.

    import requests
    import json
    import os
    import boto3
    from botocore.exceptions import ClientError
    import ipaddress
    
    def check_for_existing(list_id, ip):
        client = boto3.client("ec2", region_name="us-west-2")
        try:
            response = client.get_managed_prefix_list_entries(
                PrefixListId=list_id,
                MaxResults=100,
            )
            for entry in response['Entries']:
                if entry['Cidr'] == ip:
                    return True
                else:
                    pass
            return False
        except ClientError as e:
            print(e)
    
    
    
    def get_prefix_list_id(list_name):
        client = boto3.client("ec2", region_name="us-west-2")
        response = client.describe_managed_prefix_lists(
            MaxResults=100,
            Filters=[
                {
                    "Name": "prefix-list-name",
                    "Values": [list_name]
                }
            ]
        )
        for p_list in response['PrefixLists']:
            return {"ID": p_list['PrefixListId'], "VERSION": p_list['Version']}
    
    def update_managed_prefix_list(list_name, ip):
        client = boto3.client("ec2", region_name="us-west-2")
        if check_for_existing(get_prefix_list_id(list_name)['ID'], ip) == True:
            print("Rule already exists")
            return False
        else:
            try:
                response = client.modify_managed_prefix_list(
                            DryRun=False,
                            PrefixListId=get_prefix_list_id(list_name)['ID'],
                            CurrentVersion=get_prefix_list_id(list_name)['VERSION'],
                            AddEntries=[
                                {
                                    "Cidr": ip
                                }
                            ]
                        )
                return True
            except ClientError as e:
                print(e)
                print("Failed to update list")
    
    if __name__ == "__main__":
        url = "https://<my IP address URL>"
        headers = {}
        r = requests.get(url, headers=headers)
        json_ips = json.loads(r.content)
        ip = ""
        list_name = ""
        result = update_managed_prefix_list(list_name, ip)
        if result == True:
            print("Successfully Updates lists")
        else:
            print("Failed to update lists")

    If you are going to use this code it will need some modifications. I ultimately did not deploy this code but I had plans to run it as a Lambda function on a schedule so the lists would always be up to date.

    If this code is helpful to you please share it with your friends!

    Github

  • EC2 Reservation Notification

    I realized today that I haven’t updated my EC2 reservations recently. Wondering why I never did this I came to understand that there was no way that I was getting notified that the reservations were expiring. I spent the day putting together a script that would look through my reservations, assess the time of their expiration, and then notify me if it was nearing my threshold of 3 weeks.

    I put this together as a local script but it can also be adapted to run as a lambda function which is what I have it set up to do. As always, you can view my code below and on GitHub.

    import boto3
    from datetime import datetime, timezone, timedelta
    from botocore.exceptions import ClientError
    import os
    import json
    ec2_client = boto3.client("ec2", region_name="us-west-2")
    
    def get_reserved_instances():
        response = ec2_client.describe_reserved_instances()
        reserved_instances = {}
        for reservedInstances in response['ReservedInstances']:
            reserved_instances.update({
                reservedInstances['ReservedInstancesId']: {
                    "ExpireDate": reservedInstances['End'],
                    "Type": reservedInstances['InstanceType']
                }
            })
        return reserved_instances
    def determine_expirery(expirery_date):
        now = datetime.now(timezone.utc)
        delta_min = timedelta(days=21)
        delta_max = timedelta(days=22)
        if expirery_date - now >= delta_min and expirery_date - now < delta_max:
            return True
        else:
            return False
    #Send Result to SNS
    def sendToSNS(messages):
        sns = boto3.client('sns')
        try:
            send_message = sns.publish(
                TargetArn=os.environ['SNS_TOPIC'],
                Subject='EC2-Reservation',
                Message=messages,
                )
            return send_message
        except ClientError as e:
            print("Failed to send message to SNS")
            print(e)
    
    
    if __name__ == "__main__":
    
        for reservation, res_details in get_reserved_instances().items():
            if determine_expirery(res_details['ExpireDate']) == True:
                sns_message = {"reservation": reservation, "expires": res_details['ExpireDate'].strftime("%m/%d/%Y, %H:%M:%S")}
                sendToSNS(json.dumps(sns_message))
    #  

    I have an SNS topic setup that is set to send messages to a Lambda function in the backend so I can format my messages and send them to a Slack channel for notifications.

    If you have any questions, feel free to comment or message me on Twitter!

    GitHub