Skip to content

Instantly share code, notes, and snippets.

@af-inet
Created February 24, 2018 20:33
Show Gist options
  • Select an option

  • Save af-inet/09115fae01a64307014c30dd1c6fdb75 to your computer and use it in GitHub Desktop.

Select an option

Save af-inet/09115fae01a64307014c30dd1c6fdb75 to your computer and use it in GitHub Desktop.
AWS create routes for a peering connection between 2 VPCs
#!/usr/bin/env python
#
# https://boto3.readthedocs.io/en/latest/reference/services/ec2.html
#
import argparse
import boto3
client = boto3.client("ec2")
def parse_args():
parser = argparse.ArgumentParser(
description='create routes for a peering connection between 2 VPCs')
parser.add_argument('VPC_1_ID',
type=str,
help='VPC id you want to add routes for')
parser.add_argument('VPC_2_ID',
type=str,
help='VPC id you want to add routes for')
parser.add_argument('PEERING_ID',
type=str,
help='peering connection id you want to add routes for')
return parser.parse_args()
def get_cidr_by_vpc_id(vpc_id):
vpc_list = client.describe_vpcs(Filters=[
{
"Name": "vpc-id",
"Values": [vpc_id]
}
])['Vpcs']
if len(vpc_list) == 1:
return vpc_list[0]['CidrBlock']
elif len(vpc_list) > 1:
raise Exception("More than one VPC with ID: %s" % vpc_list)
else:
raise Exception("No VPC found with ID: %s" % vpc_id)
def get_routetables_by_vpc(vpc_id):
route_list = client.describe_route_tables(Filters=[
{
"Name": "vpc-id",
"Values": [vpc_id]
}
])
if route_list.get('RouteTables'):
return route_list['RouteTables']
else:
raise Exception("No Route Tables Found for VPC: %s" % vpc_id)
def main():
args = parse_args()
tables_1 = get_routetables_by_vpc(args.VPC_1_ID)
tables_2 = get_routetables_by_vpc(args.VPC_2_ID)
cidr_1 = get_cidr_by_vpc_id(args.VPC_1_ID)
cidr_2 = get_cidr_by_vpc_id(args.VPC_2_ID)
print("[*] creating routes...")
# routes VPC 1 to VPC 2
for table in tables_1:
response = client.create_route(
DestinationCidrBlock=cidr_2,
DryRun=False,
RouteTableId=table["RouteTableId"],
VpcPeeringConnectionId=args.PEERING_ID)
print(response)
# routes VPC 2 to VPC 1
for table in tables_2:
response = client.create_route(
DestinationCidrBlock=cidr_1,
DryRun=False,
RouteTableId=table["RouteTableId"],
VpcPeeringConnectionId=args.PEERING_ID)
print(response)
print("[*] done!")
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment