Skip to content

Instantly share code, notes, and snippets.

@kamal2222ahmed
Forked from bwhaley/peer_routes.py
Created March 25, 2018 20:17
Show Gist options
  • Save kamal2222ahmed/cd45e9698cff4102636d756ddd7f79c5 to your computer and use it in GitHub Desktop.
Save kamal2222ahmed/cd45e9698cff4102636d756ddd7f79c5 to your computer and use it in GitHub Desktop.
A script to configure routes in in a VPC peering connection
# 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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment