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:
These are the results in AWS console:
VPC
Subnets
Subnets routes
Route table
Route association with subnet
Security groups
EC2 instance, this I was using the Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type.
The web site
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>" > /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)
cyruslab 