[python]Start up script to create VPC to launch EC2

Use case
This is an interactive start up script to do from creating VPC to launching EC2. This is a follow up from this post – Functions for aws automation, I have added a few more functions to make it complete.

Demonstration
This is the interactive script:
Screenshot 2019-05-11 at 7.09.32 PM.png
Screenshot 2019-05-11 at 7.09.43 PM.png

These are the results in AWS console:
VPC
Screenshot 2019-05-11 at 7.21.42 PM

Subnets
Screenshot 2019-05-11 at 7.22.01 PM.png

Subnets routes
Screenshot 2019-05-11 at 7.25.49 PM.png

Route table
Screenshot 2019-05-11 at 7.27.07 PM.png

Route association with subnet
Screenshot 2019-05-11 at 7.27.44 PM.png

Security groups
Screenshot 2019-05-11 at 7.22.27 PM.png
Screenshot 2019-05-11 at 7.30.19 PM.png

EC2 instance, this I was using the Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type.
Screenshot 2019-05-11 at 7.31.42 PM

The web site
Screenshot 2019-05-11 at 8.29.01 PM

The codes of the above
The code is documented see the inline comments below.

import boto3 # AWS API
from ipaddress import ip_network # Use to test subnet and host net
import botocore.exceptions # for catching exceptions in calling AWS API
from pprint import pprint # makes printing easier for a list

# This filter works for VPC, subnets as long as the Tag is named.
filter = [{'Name':'tag:Name', 'Values':['*']}]

# Tested working on Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type
linux_startup_script = """#!/bin/bash
                    yum update -y
                    yum install httpd24 -y
                    service httpd start
                    chkconfig httpd on
                    echo "<h1>Test page</h1>" &gt; /var/www/html/index.html"""

# Network that matches all.
match_all_net = '0.0.0.0/0'

# AWS client initializer, S3 or EC2, uses boto3's client method.
def get_client(resource_name):
    return boto3.client(resource_name)

# For collecting values with a dictionary key (attrib_id)
# vpc_list can be called using describe_subnets(Filters=filter)[attrib_id] or describe_vpcs(Filters=filter)[attrib_id]
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


# base on the prefix length, ip_network method is able to list down the subnets from a VPC cidr block.
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


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


# Use for naming VPCs, subnets, internetgateways, security groups...
def create_tags(client, chosen_name, resources):
    tag = [
        {
            'Key': 'Name',
            'Value': chosen_name
        }
    ]
    return client.create_tags(Resources=[resources], Tags=tag)


# Converts the list processed by get_attribute_from_vpc into a dictionary.
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


# Creates a route table, for storing routes.
def create_rtbl(client, vpc_id):
    return client.create_route_table(VpcId=vpc_id)


# Define the routes and stores in route table, needs to attach to internet gateway if internet access is required.
def create_route(client, rtbl_id, igw_id, dst_net):
    return client.create_route(RouteTableId=rtbl_id, DestinationCidrBlock=dst_net, GatewayId=igw_id)


# Not needed, but this can find the existing internet gateway id.
# Use get_attribute_from_vpc and resource_menu instead to get any kinds of ids
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

# attach the route table to a subnet.
def associate_route_table(client, rtbl_id, subnet_id):
    return client.associate_route_table(RouteTableId=rtbl_id, SubnetId=subnet_id)

# This allows the EC2 instance assigned to this subnet to automatically gets a dynamically assigned public address.
def auto_assigned_public_ipv4_address(client, subnetid):
    return client.modify_subnet_attribute(SubnetId=subnetid,
                                          MapPublicIpOnLaunch={'Value': True})

# works like a firewall, needs to associate with a VPC.
def create_security_group(client, group_name, description, vpc_id):
    return client.create_security_group(GroupName=group_name, Description=description, VpcId=vpc_id)


# Collects ip addresses from user's input, this is used for source address in security group.
def cidrip_list_collector():
    results = []
    process_results = []
    stop = int(input("How many source ip address you want"))
    print("Press enter with empty response to quit.")
    for i in range(0,stop):
        ip = input("Source address:")
        if ip is not "":
            try:
                ip_network(ip)
                results.append(ip)
            except ValueError as e:
                print(e)
        else:
            break
    # collecting the dictionary / dictionaries in a list
    for result in results:
        results_dict = {'CidrIp': result}
        process_results.append(results_dict)
    return process_results


# This creates the security group required format, and passed as a dictionary for the IpPermission parameter
# in boto3.authorize_security_group_ingress method.
def rule_form():
    port_list = []
    protocol_response = input("Protocol (tcp/udp)?:")
    if protocol_response.lower() == 'tcp':
        protocol = protocol_response.lower()
    elif protocol_response.lower() == 'udp':
        protocol = protocol_response.lower()
    else:
        print("Invalid choice, this field cannot be empty, hence default to tcp")
        protocol = 'tcp'
    port_range_response = input("Enter your port range, if only one port example 80, write 80,80, \r\n"
                                "if it is a range like 90-100 write 90,100:").split(',')
    for index in port_range_response:
        port_list.append(index)
    ip_ranges_list = cidrip_list_collector()
    return {
            'IpProtocol': protocol,
            'FromPort': int(port_list[0]),
            'ToPort': int(port_list[1]),
            'IpRanges': ip_ranges_list
        }


# When calling this method, the ip_permission has to be passed as a list, the parameter of
# boto3.authorize_security_group_ingress IpPermissions only accepts list.
def create_inbound_rule(client, group_id, ip_permission):
    return client.authorize_security_group_ingress(GroupId=group_id, IpPermissions=ip_permission)


# for sec_group_id, subnet_id you can use the methods get_attribute_from_vpc and resource_menu to find them.
# user_data is not compulsory, it can be a startup script for an EC2 instance.
# image_id has to be obtained from aws console.
# if you want to launch one EC2 at once, min_count and max_count both are 1, if you need to launch 10,
# then min_count = 1, max_count = 10, the min_count and max_count only accept integer.
# python's input method is a string, you will need to convert to integer using int() method.
def launch_ec2_instance(client, image_id, keyname, min_count, max_count, sec_group_id, subnet_id, user_data):
    return client.run_instances(ImageId=image_id,
                                KeyName=keyname,
                                MinCount=min_count,
                                MaxCount=max_count,
                                InstanceType='t2.micro',
                                SecurityGroupIds=[sec_group_id],
                                SubnetId=subnet_id,
                                UserData=user_data)

# This key pair name is used for ssh
# returns the keyname of user's choice and the response after key pair is created.
def create_key_pair(client):
    keyname = input('enter a key name:')
    return keyname, client.create_key_pair(KeyName=keyname)

# Creates a VPC
def create_vpc(client, cidr_block):
    return client.create_vpc(CidrBlock=cidr_block)

# creates an internet gateway, and returns the internet gateway response
# After the internet gateway is created, it is attached to a VPC
def create_intenet_gateway(client, vpc_id):
    create_igw_response = client.create_internet_gateway()
    pprint("Internet gateway {} is created".format(create_igw_response['InternetGateway']['InternetGatewayId']))
    pprint("Attaching {} to {}".format(
        create_igw_response['InternetGateway']['InternetGatewayId'],
        vpc_id
    ))
    return [
        client.attach_internet_gateway(
            InternetGatewayId=create_igw_response['InternetGateway']['InternetGatewayId'],
            VpcId=vpc_id),
        create_igw_response]


if __name__ == '__main__':
    # EC2 client, can be S3 if you need to configure storage.
    ec2 = get_client('ec2')
    """:type : pyboto3.ec2"""
    cidr_block = input("The subnet block for your VPC: ")
    try:
        # Test the cidr block entered by user.
        ip_network(cidr_block)
        try:
            # always get a response after you call the method. This case is to create a vpc with cidr block specified
            # by the user.
            create_vpc_response = create_vpc(ec2, cidr_block)
            # the response is useful if you need to get the value such as VpcId, this not only applies to VPC,
            # it applies to security group, internet gateway, subnets creation....
            pprint('VPC {} is created.'.format(create_vpc_response['Vpc']['VpcId']))
            # collects the desired name of VPC by user. Naming VPC is to conveniently use the filter.
            create_vpc_name = input('Name this VPC {}: '.format(create_vpc_response['Vpc']['VpcId']))
            # Name the VPC with user's chosen name for the VPC
            create_vpc_tags_response = create_tags(ec2,
                                               create_vpc_name,
                                               create_vpc_response['Vpc']['VpcId']
                                               )
            pprint("VPC {} is named.".format(create_vpc_response['Vpc']['VpcId']))
            # Creates an internet gateway, always get the response, for later usage.
            create_internet_gateway_response = create_intenet_gateway(
                ec2, create_vpc_response['Vpc']['VpcId'])
            pprint("This section you need to choose subnets from the CIDR block...")
            # Get the prefix length from user, checks should be written such that chosen
            # prefix length must never be shorter than the prefix length of the CIDR block of VPC.
            prefix_len = int(input("Enter the prefix length you want for {}: ".format(
                create_vpc_response['Vpc']['CidrBlock'])))
            # This collects the dictionary of CIDR blocks of VPCs
            resource_menu_response = resource_menu(suggest_subnets(create_vpc_response['Vpc']['CidrBlock'],
                                                                   prefix_len))
            iteration = True
            while iteration:
                pprint(resource_menu_response)
                # This user's choice is a dictionary key.
                subnet_choice = int(input("Select the subnet you wish to create from the suggested list only: "))
                try:
                    # Creates the subnet, always get the response for later use.
                    # the user's choice is used here to reference subnet.
                    # create subnet require vpc id and subnet.
                    create_subnet_response = create_subnet(ec2,
                                  create_vpc_response['Vpc']['VpcId'],
                                  resource_menu_response[subnet_choice])
                    if input("Want to make this subnet to auto assign public address to EC2 instance?: ").lower() == 'y':
                        # Enable subnet to automatically assign public address to EC2 instance.
                        # subnet id is required to enable the auto assign pub address.
                        # as shown here again, always get response after something is created.
                        # you will use it most of the time, such as create_subnet_response is used to
                        # reference the subnet id.
                        auto_assigned_public_ipv4_address_response = auto_assigned_public_ipv4_address(ec2,
                                                          create_subnet_response['Subnet']['SubnetId'])
                        pprint("Subnet is enabled for auto assigned public ipv4 address, "
                               "this means whenever an EC2 instance is attached to this subnet,"
                               "the EC2 instance will be auto assigned a publicly routable address.")
                    else:
                        pprint("Auto assigned ipv4 public address is not enabled for {}".format(
                            create_subnet_response['Subnet']['SubnetId']
                        ))
                    create_subnet_name = input("Name this subnet {}: ".format(
                        create_subnet_response['Subnet']['SubnetId']))
                    # Name the subnet, so that the filter can be used, to only show the things you have created.
                    create_subnet_tags_response = create_tags(ec2,
                                create_subnet_name,
                                create_subnet_response['Subnet']['SubnetId'])
                    # This is to prevent user from selecting the same subnet that was configured before.
                    del resource_menu_response[subnet_choice]
                    if input("Create another subnet for {}?: ".format(
                            create_vpc_response['Vpc']['VpcId']
                    )).lower() == 'y':
                        iteration = True
                    else:
                        iteration = False
                except botocore.exceptions.ClientError as e:
                    pprint(e)
            pprint("You have created VPC and subnet(s), the next is to create route table...")
            # To create the route table, as always get the response. This line uses create_vpc_response again
            # to reference VPC ID.
            create_rtbl_response = create_rtbl(ec2, create_vpc_response['Vpc']['VpcId'])
            pprint("Route table {} is created, the next is to put in the routes...".format(
                create_rtbl_response['RouteTable']['RouteTableId']
            ))
            add_route_iteration = True
            while add_route_iteration:
                if(input("Is this a default route?: ")).lower() == 'y':
                    # creating routes, and store them in route table.
                    # attach to internet gateway is optional, only requires if your EC2 instance requires
                    # internet access, or you need visitor to access from the internet.
                    # if user says yes to default route, match_all_net is used.
                    # create_internet_gateway_response is a list that contains:
                    # response from attach_internet_gateway and create_internet_gateway.
                    # create_internet_gateway_response[0] contains response from boto3.attach_internet_gateway.
                    # create_internet_gateway_response[1] contains response from boto3.create_internet_gateway.
                    create_route_response = create_route(ec2,
                                 create_rtbl_response['RouteTable']['RouteTableId'],
                                 create_internet_gateway_response[1]['InternetGateway']['InternetGatewayId'],
                                 match_all_net)
                else:
                    dst_net_choice = (input("Specify a network: "))
                    try:
                        # checks if the destination network is valid or not.
                        ip_network(dst_net_choice)
                        # For user to specify the destination network for another route.
                        create_route_response = create_route(ec2,
                                     create_rtbl_response['RouteTable']['RouteTableId'],
                                     create_internet_gateway_response[1]['InternetGateway']['InternetGatewayId'],
                                     dst_net_choice)

                    except ValueError as e:
                        pprint(e)
                # Collects the list of subnets
                subnets = get_attribute_from_vpc('CidrBlock', ec2.describe_subnets(Filters=filter)['Subnets'])
                # Collects the list of subnet ids.
                subnet_ids = get_attribute_from_vpc('SubnetId', ec2.describe_subnets(Filters=filter)['Subnets'])
                # Converts the list of subnets to a dictionary of subnets.
                resource_menu_subnets = resource_menu(subnets)
                # User only sees the subnets and not the subnet ids. Subnet ids are not readable by user, haha.
                pprint(resource_menu_subnets)
                invalid_subnet_choice = True
                while invalid_subnet_choice:
                    # The subnets_choice has to be an integer to reference the value of subnet id.
                    subnets_choice = int(input("Choose the subnet to associate the route table:"))
                    # Converts the list of subnet ids into a dictionary of subnet ids.
                    resource_menu_subnet_ids = resource_menu(subnet_ids)
                    try:
                        associate_route_table_response = associate_route_table(
                            ec2,
                            create_rtbl_response['RouteTable']['RouteTableId'],
                            resource_menu_subnet_ids[subnets_choice]) # Gets the subnet id based on the integer from user.
                        pprint("Associated {} to subnet {}".format(
                            create_rtbl_response['RouteTable']['RouteTableId'],
                            resource_menu_subnet_ids[subnets_choice]
                        ))
                    except botocore.exceptions.ClientError as e:
                        pprint(e)
                    if subnets_choice in resource_menu_subnets.keys():
                        invalid_subnet_choice = False

                    else:
                        invalid_subnet_choice = True

                if(input("Add some more routes?: ")).lower() == 'y':
                    add_route_iteration = True
                else:
                    add_route_iteration = False
            pprint("Next is to create ingress security group...")
            sec_grp_name = input("Name the security group: ")
            sec_grp_description = input("Describe this security group: ")
            # security group requires a group name and VPC ID, description is not mandatory.
            # This creates the security group and attached to VPC.
            create_security_group_response = create_security_group(ec2,
                                                                   sec_grp_name,
                                                                   sec_grp_description,
                                                                   create_vpc_response['Vpc']['VpcId'])
            pprint("Security group {} is created and attached to VPC {}...".format(
                create_security_group_response['GroupId'],
                create_vpc_response['Vpc']['VpcId']
            ))
            want_to_add_more_rules = True
            while want_to_add_more_rules:
                # sec_grp_rule is for the IpPermissions parameter in boto3.authorize_security_group_ingress.
                # the IpPermissions parameter only accepts list.
                # hence the dictionary type returned by rule_form() has to be converted to list.
                sec_grp_rule = [rule_form()]
                # creating the rule for inbound only, rule will be attached to a security group.
                create_inbound_rule_response = create_inbound_rule(ec2,
                                    create_security_group_response['GroupId'],
                                    sec_grp_rule)
                pprint("Adding ingress rules to security group {}...".format(
                    create_security_group_response['GroupId']
                ))
                if input("Want to add more rules?: ").lower() == 'y':
                    want_to_add_more_rules = True
                else:
                    want_to_add_more_rules = False
            pprint("The next is to launch new EC2 instance(s). You need to head over "
                   "to aws console to find the image id that starts with ami-xxxxxxx")
            pprint("I need to generate a pair of keys for your EC2, tell me your keyname you would like: ")
            # create_key_pair() returns the keyname of user's choice and the response from boto3.create_key_pair.
            keyname, create_key_pair_response = create_key_pair(ec2)
            imageid = input("Your chosen image id found in aws console: ")
            ec2_count = int(input("How many EC2 instances would you want to launch at once?: "))
            subnets_for_ec2 = get_attribute_from_vpc('CidrBlock', ec2.describe_subnets(Filters=filter)['Subnets'])
            subnet_ids_for_ec2 = get_attribute_from_vpc('SubnetId', ec2.describe_subnets(Filters=filter)['Subnets'])
            resource_menu_subnets = resource_menu(subnets_for_ec2)
            resource_menu_subnets_ids = resource_menu(subnet_ids_for_ec2)
            pprint(resource_menu_subnets)
            subnets_choice = int(input("Choose the subnet to launch EC2 instance(s):"))
            launch_ec2_instance(ec2,
                                imageid, # Find out from aws console
                                keyname,
                                1, ec2_count, # min and max count, must be integer
                                create_security_group_response['GroupId'],
                                resource_menu_subnets_ids[subnets_choice],
                                linux_startup_script) # startup script uses yum, which red hat and 
                                                    # Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type use.
        except botocore.exceptions.ClientError as e:
            pprint(e.response['Error']['Message'])
    except ValueError as e:
        pprint(e)
Advertisements

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s