A script to configure routes in in a VPC peering connection

peer_routes.py

# Quick script to configure routes for a VPC peering connection
# Searches a region for all peering connection and prompts to choose one
# Configures routes between the peered networks for all routing tables
# STS/AssumeRole not implemented cross-account peering. Instead, 
# Choose accepter/requestor depending on which credentials are set in the environment
# Enter either IPv4 and IPv6 route destinations
# Example usage:
#  ( Assuming boto credentials are configured)
#   $ pip install boto3
#   $ python3.6 peer_routes.py
#  Peering connections:
#  1), Management-Production: pcx-49733720
#  2), Management-Dev: pcx-a9d2bfc0
#  3), Management-Staging: pcx-c2e8b3ab
#  Which peering connection needs routes? 3
#  Which VPC should be updated (requestor, accepter, or both)? requestor
#  Manually enter routing destinations (if requestor/accepter are in different accounts) (y/n)?y
#  Enter destination: 172.26.0.0/16
#  Enter destination: 2600:1f14:c83:de00::/56
#  Enter destination:
#  Added route to 172.26.0.0/16 in table rtb-f2a2aa95.
#  Added route to 2600:1f14:c83:de00::/56 in table rtb-f2a2aa95.
#  Added route to 172.26.0.0/16 in table rtb-f3a2aa94.

import sys
import boto3
from botocore.exceptions import ClientError

def choose_peering_connection(peering_connections):
    print("Peering connections: \n")
    for i, cnxn in enumerate(peering_connections):
        p_id = cnxn['VpcPeeringConnectionId']
        p_name = ""
        for t in cnxn['Tags']:
            if t['Key'] == "Name":
                p_name = t['Value']

        print("{idx}) {_name}: {_id}".format(idx=i+1, _name=p_name, _id=p_id))
    s = int(input('Which peering connection needs routes? '))

    return peering_connections[s-1]


def discover_peering_connection():
    peering_connections = ec2_c.describe_vpc_peering_connections()
    if len(peering_connections['VpcPeeringConnections']) > 1:
        active_connections = []
        for i, cnxn in enumerate(peering_connections['VpcPeeringConnections']):
            if cnxn['Status']['Code'] == "deleted":
                continue
            else:
                active_connections.append(cnxn)
        if len(active_connections) == 0:
            print("No peering connections!")
            sys.exit(0)
        else:
            return choose_peering_connection(active_connections)
    else:
        return peering_connections[0]

def find_route_destinations(vpc_id):
    dest_cidrs = []
    dest_vpc = ec2_c.describe_vpcs(VpcIds=[vpc_id])
    dest_vpc = dest_vpc['Vpcs'][0]
    dest_cidrs.append(dest_vpc['CidrBlock'])
    if 'Ipv6CidrBlockAssociationSet' in dest_vpc.keys():
        dest_cidrs.append(dest_vpc['Ipv6CidrBlockAssociationSet'][0]['Ipv6CidrBlock'])
    return dest_cidrs


def add_routes(rt, route_dests, gw):
    rt_id = rt['RouteTableId']
    ec2_r = boto3.resource('ec2')
    route_table = ec2_r.RouteTable(rt_id)
    for r in route_dests:
        try:
            if ":" in r:
                route_table.create_route(
                    DestinationIpv6CidrBlock=r,
                    VpcPeeringConnectionId=gw,
                    RouteTableId=rt_id)
            else:
                route_table.create_route(
                    DestinationCidrBlock=r,
                    VpcPeeringConnectionId=gw,
                    RouteTableId=rt_id)
            print("Added route to {0} in table {1}.".format(r, rt_id))
        except ClientError as e:
            if e.response['Error']['Code'] == "RouteAlreadyExists":
                print("Route already exists for {0} in table {1}. Moving on.".format(r, rt_id))
            else:
                print(e)
                sys.exit(1)


def update_route_tables(peering_cnxn_id, source_vpc_id, dest_vpc_id):
    route_dests = []
    s = ""
    while s not in ['y', 'n']:
        s = input('Manually enter routing destinations (if requestor/accepter are in different accounts) (y/n)?')
    if s == "y":
        while True:
            dest = input('Enter destination: ')
            if dest == "":
                break
            route_dests.append(dest)
    elif s == "n":
        route_dests = find_route_destinations(source_vpc_id)
    route_tables = ec2_c.describe_route_tables()
    for rt in route_tables['RouteTables']:
        if rt['VpcId'] == source_vpc_id:
            add_routes(rt, route_dests, peering_cnxn_id)


ec2_c = boto3.client('ec2')
cnxn = discover_peering_connection()
cnxn_id = cnxn['VpcPeeringConnectionId']
requestor_vpc = cnxn['RequesterVpcInfo']['VpcId']
accepter_vpc = cnxn['AccepterVpcInfo']['VpcId']
choices = [
    "both",
    "requestor",
    "accepter",
    ]
c = ""
while c not in choices:
    c = input('Which VPC should be updated (requestor, accepter, or both)? ')
    if c == "both" or c == "requestor":
        update_route_tables(cnxn_id, requestor_vpc, accepter_vpc)
    if c == "both" or c == "accepter":
        update_route_tables(cnxn_id, accepter_vpc, requestor_vpc)