[python] Functions for aws automation

This post is a documentation of the functions/methods in the python script that will help to automate the AWS.

Declaration

import boto3
from ipaddress import ip_network
from pprint import pprint


filter = [{'Name':'tag:Name', 'Values':['*']}]

The filter can be found in Automated Ramblings.

Resource connection
This function connects to the resource based on the one of your choosing, can be EC2 or S3. It uses the default connection setup by awscli. Which is why I do not need to put in region name, access key id and secret key id.

def get_client(resource_name):
    return boto3.client(resource_name)

Extracting values from VPC
This function looks for the value you are interested such as the vpcid and cidr block. This function returns a list.

def get_attribute_from_vpc(attrib_id, vpc_list):
    # resets the collector.
    collector = []
    for vpc in vpc_list:
        collector.append(vpc[attrib_id])
    return collector

Suggesting a list of subnets based on the cidr block in VPC
This function uses ipaddress module’s ip_network method to list the possible subnets.This function returns a list.

def suggest_subnets(supernet, prefix_len):
    # resets the subnets list
    subnets = []
    for subnet in ip_network(supernet).subnets(new_prefix=prefix_len):
        subnets.append(str(subnet))
    return subnets

Create subnet
This function helps to call the boto3 create_subnet method, I created this function for myself to see the creation requirement easier. It also helps to make my code easier to read and understand.

def create_subnet(client, vpc_id, subnet):
    return client.create_subnet(VpcId=vpc_id, CidrBlock=subnet)

Create tag base on user’s input
This tag is for use in create_tag method in boto3 to name my items in VPC.

def name_resource(chosen_name):
    tag = [
        {
            'Key': 'Name',
            'Value': chosen_name
        }
    ]
    return tag

The menu generator
This function not only generates menu, it’s main purpose is to organise the list into a dictionary, hence this function returns a dictionary. User’s input will be integer, and based on the integer typed in the value which is either a cidrblock or vpcid will be chosen, the value returned can also be used for methods that require cidrblock or vpcid.

def resource_menu(resource_list):
    collector = []
    for resource in resource_list:
        collector.append(resource)
    menu_dict = {}
    for i in range(len(collector)):
        j = i + 1
        menu_dict[j] = collector[i]
    return menu_dict

Create routing table
This function is for explicitly creating a routing table, AWS actually creates routing table after subnet is created. A newly created routing table is just a container, this routing table has to be associated with subnet in order to use, you can add other routes in this table such as a default route.

def create_rtbl(client, vpc_id):
    return client.create_route_table(VpcId=vpc_id)

Create route
This function is for adding new routes, it can be adding a destination subnet of 0.0.0.0/0 which matches all ip addresses usually for default gateway, and also to add the next hop. This function can be used to create default route to the internet gateway.

def create_route(client, rtbl_id, igw_id, dst_net):
    return client.create_route(RouteTableId=rtbl_id, DestinationCidrBlock=dst_net, GatewayId=igw_id)

Associate routing table to subnet
This function associates the created routing table to a subnet.

def associate_route_table(client, rtbl_id, subnet_id):
    return client.associate_route_table(RouteTableId=rtbl_id, SubnetId=subnet_id)

Find gateway id
This function helps to find the internet gateway id that is associated with your vpc id. I will document how the code works on next post.

def find_internet_gateway_id(client, vpc_id):
    igw_id = ""
    igw_responses = client.describe_internet_gateways(Filters=filter)
    for i in igw_responses['InternetGateways']:
        for j in i['Attachments']:
            if vpc_id in j['VpcId']:
                igw_id = i['InternetGatewayId']
    return igw_id

Testing codes: create subnet from suggested list
My code is tested under main, the purpose is to test each functions which I created and see if the expected results can be yielded.

if __name__ == '__main__':
    ec2 = get_client('ec2')
    """:type pyboto3.ec2"""
    vpc_ids = get_attribute_from_vpc('VpcId', ec2.describe_vpcs(Filters=filter)['Vpcs'])
    menu = resource_menu(vpc_ids)
    print(menu)
    choice = int(input("Choose an option:"))
    if choice not in menu.keys():
        print("Invalid choice!")
        print(menu)
    else:
        cidr_ids = get_attribute_from_vpc('CidrBlock', ec2.describe_vpcs(Filters=filter)['Vpcs'])
        cidr_menu = resource_menu(cidr_ids)
        prefix_len = int(input("Choose a prefix length:"))
        cidr_suggestions = suggest_subnets(cidr_menu[choice], prefix_len)
        cidr_suggestions_menu = resource_menu(cidr_suggestions)
        print(cidr_suggestions_menu)
        cidr_choice = int(input("Choose one of the suggestion:"))
        if cidr_choice not in cidr_suggestions_menu.keys():
            print("Invalid subnet! Choose only from the suggested ones.")
        else:
            create_subnet(ec2, menu[choice], cidr_suggestions_menu[cidr_choice])

This code when triggered will look like below.

Testing code: create subnet from suggested list, create routing table, put in a default route into routing table and associate the routing table to a subnet.

if __name__ == '__main__':
    ec2 = get_client('ec2')
    """:type : pyboto3.ec2"""
    vpc_ids = get_attribute_from_vpc('VpcId', ec2.describe_vpcs(Filters=filter)['Vpcs'])
    menu = resource_menu(vpc_ids)
    print(menu)
    choice = int(input("Choose an option:"))
    if choice not in menu.keys():
        print("Invalid choice!")
        print(menu)
    else:
        cidr_ids = get_attribute_from_vpc('CidrBlock', ec2.describe_vpcs(Filters=filter)['Vpcs'])
        cidr_menu = resource_menu(cidr_ids)
        prefix_len = int(input("Choose a prefix length:"))
        cidr_suggestions = suggest_subnets(cidr_menu[choice], prefix_len)
        cidr_suggestions_menu = resource_menu(cidr_suggestions)
        print(cidr_suggestions_menu)
        cidr_choice = int(input("Choose one of the suggestion:"))
        if cidr_choice not in cidr_suggestions_menu.keys():
            print("Invalid subnet! Choose only from the suggested ones.")
        else:
            create_subnet_response = create_subnet(ec2, menu[choice], cidr_suggestions_menu[cidr_choice])
            pprint("{} is created for {}".format(create_subnet_response['Subnet']['SubnetId'],
                                                 create_subnet_response['Subnet']['VpcId']))
            create_rtbl_response = create_rtbl(ec2, menu[choice])
            pprint("{} is created for {}".format(create_rtbl_response['RouteTable']['RouteTableId'],
                                                 create_rtbl_response['RouteTable']['VpcId']))
            create_route(ec2, create_rtbl_response['RouteTable']['RouteTableId'],
                         find_internet_gateway_id(ec2, menu[choice]),
                                                            '0.0.0.0/0')
            pprint("Default route table to IGW is created")
            associate_route_table(ec2, create_rtbl_response['RouteTable']['RouteTableId'],
                                  create_subnet_response['Subnet']['SubnetId'])
            pprint("{} is associated with {}".format(create_rtbl_response['RouteTable']['RouteTableId'],
                                  create_subnet_response['Subnet']['SubnetId']))

Results