Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Dynamic Troposphere + boto3 template that creates a complete and functional VPC through AWS CloudFormation
#!/usr/bin/env python
from troposphere import Join, Output
from troposphere import Parameter, Ref, Tags, Template
from troposphere.ec2 import PortRange
from troposphere.ec2 import NetworkAcl
from troposphere.ec2 import Route
from troposphere.ec2 import VPCGatewayAttachment
from troposphere.ec2 import SubnetRouteTableAssociation
from troposphere.ec2 import Subnet
from troposphere.ec2 import CustomerGateway
from troposphere.ec2 import VPNConnectionRoute
from troposphere.ec2 import RouteTable
from troposphere.ec2 import VPC
from troposphere.ec2 import NetworkAclEntry
from troposphere.ec2 import VPNGateway
from troposphere.ec2 import InternetGateway
from troposphere.ec2 import SubnetNetworkAclAssociation
from troposphere.ec2 import VPNConnection
from troposphere.ec2 import VPNGatewayRoutePropagation
from troposphere.ec2 import SecurityGroup, SecurityGroupRule
from troposphere.ec2 import Instance
from troposphere.ec2 import EIP
import sys
import boto3
s = boto3.session.Session(profile_name = sys.argv[1])
ec2 = s.resource('ec2')
t = Template()
t.add_version('2010-09-09')
t.add_description('''Populated VPC template.''')
### Parameters
OPBgpAsn = t.add_parameter(Parameter(
'OPBgpAsn',
Type = 'Number',
Description = 'On Premise BGP ASN',
ConstraintDescription = 'Numeric value from 1 to 65535.'
))
BgpPeer = t.add_parameter(Parameter(
'BgpPeer',
Type='String',
Description='IP Address of your VPN device',
MinLength='7',
AllowedPattern='(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})',
MaxLength='15',
ConstraintDescription='must be a valid IP address of the form x.x.x.x',
))
tagname = {
'CGW': 'Customer Gateway name',
'VGW': 'Virtual Gateway name',
'VPN': 'VPN name',
'VPC': 'VPC name',
'NATInstance': 'NAT Instance Name'
}
for k in tagname:
varname = '{0}Name'.format(k)
vars()[varname] = t.add_parameter(Parameter(
'{0}Name'.format(k),
Type = 'String',
Description = tagname[k],
AllowedPattern = '[-_ a-zA-Z0-9]+',
MinLength = '4',
MaxLength = '64'
))
VPCBlock = t.add_parameter(Parameter(
'VPCBlock',
ConstraintDescription=('Firsts 2 VPC CIDR bytes, i.e. 10.20.'),
Description='Firsts 2 VPC CIDR bytes',
MinLength='4',
AllowedPattern='(\d{1,3})\.(\d{1,3})\.',
MaxLength='8',
Type='String',
))
### Resources
# VPN setup
VPNGateway = t.add_resource(VPNGateway(
'VPNGateway',
Type='ipsec.1',
Tags = Tags(
Name = Ref('VGWName'),
Application = Ref('AWS::StackName')
)
))
CustomerGateway = t.add_resource(CustomerGateway(
'CustomerGateway',
BgpAsn = Ref('OPBgpAsn'),
IpAddress=Ref('BgpPeer'),
Type='ipsec.1',
Tags=Tags(
Name = Ref('CGWName'),
VPN = Join('', ['Gateway to ', Ref('BgpPeer')]),
)
))
VPNConnection = t.add_resource(VPNConnection(
'VPNConnection',
CustomerGatewayId = Ref('CustomerGateway'),
StaticRoutesOnly = 'false',
Type = 'ipsec.1',
VpnGatewayId = Ref('VPNGateway'),
Tags = Tags(Name = Ref('VPNName'))
))
# Internet Gateway
InternetGateway = t.add_resource(InternetGateway(
'InternetGateway',
Tags=Tags(
Application = Ref('AWS::StackName'),
Name = 'vpc-inet'
)
))
# VPC setup
VPC = t.add_resource(VPC(
'VPC',
EnableDnsSupport = 'true',
CidrBlock = Join('', [Ref('VPCBlock'), '0.0/16']),
EnableDnsHostnames = 'true',
Tags=Tags(
Name = Ref('VPCName'),
Application = Ref('AWS::StackName'),
Network = 'VPN Connected VPC',
)
))
VPNGatewayAttachment = t.add_resource(VPCGatewayAttachment(
'VPNGatewayAttachment',
VpcId = Ref('VPC'),
VpnGatewayId = Ref('VPNGateway')
))
InternetGatewayAttachment = t.add_resource(VPCGatewayAttachment(
'InternetGatewayAttachment',
VpcId=Ref('VPC'),
InternetGatewayId = Ref('InternetGateway')
))
VPCId = t.add_output(Output(
'VPCId',
Description = 'VPCId of the newly created VPC',
Value = Ref('VPC'),
))
### Subnets setup
subnets = {
'private': {'Name': 'Private', 'CIDR': {'a': '0.0/24', 'b': '1.0/24'}},
'nat': {'Name': 'NAT', 'CIDR': {'a': '253.0/24', 'b': '254.0/24'}},
'public': {'Name': 'Public', 'CIDR': {'a': '100.0/24', 'b': '101.0/24'}},
}
rtids = []
for k in subnets:
# create route tables
rtname = '{0}RT'.format(k)
vars()[rtname] = t.add_resource(RouteTable(
rtname,
DependsOn = 'VPNGateway',
VpcId = Ref('VPC'),
Tags=Tags(
Application = Ref('AWS::StackName'),
Name = rtname
)
))
rtids.append(Ref(rtname))
for az in ['a', 'b']:
# create subnet
subname = '{0}SubnetAZ{1}'.format(subnets[k]['Name'], az)
vars()[subname] = t.add_resource(Subnet(
subname,
DependsOn = rtname,
VpcId = Ref('VPC'),
CidrBlock = Join('', [Ref('VPCBlock'), subnets[k]['CIDR'][az]]),
Tags = Tags(
Application = Ref('AWS::StackName'),
Name = '{0}-az{1}'.format(k, az)
)
))
# associate subnet with route table
rtassoc = '{0}RTAssoc'.format(subname)
vars()[rtassoc] = t.add_resource(SubnetRouteTableAssociation(
rtassoc,
SubnetId = Ref(subname),
RouteTableId = Ref(rtname)
))
# route table propagation through VPN
VPNRoutePropagation = t.add_resource(VPNGatewayRoutePropagation(
'VPNRoutePropagation',
DependsOn = 'VPNGatewayAttachment',
RouteTableIds = rtids,
VpnGatewayId = Ref('VPNGateway')
))
### Security Groups
sg = []
for port in ['22', '80', '443']:
sg.append(
SecurityGroupRule(
IpProtocol = 'tcp',
FromPort = port,
ToPort = port,
CidrIp = '0.0.0.0/0'
)
)
NATSecurityGroup = t.add_resource(SecurityGroup(
'NATSecurityGroup',
GroupDescription = 'Enable ports 22, 80 and 443 worldwide',
SecurityGroupIngress = sg,
VpcId = Ref('VPC'),
Tags = Tags(Name = 'ssh-http-https')
))
### Instances
image = [
i for i in ec2.images.filter(
Filters=[ {'Name': 'name', 'Values': ['amzn-ami-vpc-nat-hvm*']} ]
)
][-1]
azanatsub = subnets['nat']['CIDR']['a'].split('.')[0]
NATInstance = t.add_resource(Instance(
'NATInstance',
DependsOn = [
'InternetGatewayAttachment',
'NATSecurityGroup',
'NATSubnetAZa'
],
ImageId = image.id,
InstanceType = 't2.micro',
KeyName = 'ulic',
SecurityGroupIds = [Ref('NATSecurityGroup')],
SourceDestCheck = 'false',
SubnetId = Ref('NATSubnetAZa'),
PrivateIpAddress = Join('', [Ref('VPCBlock'), '{0}.254'.format(azanatsub)]),
Tags = Tags(Name = Ref('NATInstanceName'))
))
NATEIP = t.add_resource(EIP(
'NATEIP',
DependsOn = 'NATInstance',
Domain='vpc',
InstanceId=Ref('NATInstance')
))
print(t.to_json())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.