from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
from xml.sax.saxutils import escape
class VPNConnections(EC2BaseResponse):
def create_vpn_connection(self):
vpn_conn_type = self._get_param("Type")
cgw_id = self._get_param("CustomerGatewayId")
vgw_id = self._get_param("VpnGatewayId")
tgw_id = self._get_param("TransitGatewayId")
static_routes = self._get_param("StaticRoutesOnly")
tags = add_tag_specification(self._get_multi_param("TagSpecification"))
vpn_connection = self.ec2_backend.create_vpn_connection(
vpn_conn_type,
cgw_id,
vpn_gateway_id=vgw_id,
transit_gateway_id=tgw_id,
static_routes_only=static_routes,
tags=tags,
)
if vpn_connection.transit_gateway_id:
self.ec2_backend.create_transit_gateway_vpn_attachment(
vpn_id=vpn_connection.id, transit_gateway_id=tgw_id
)
template = self.response_template(CREATE_VPN_CONNECTION_RESPONSE)
return template.render(vpn_connection=vpn_connection)
def delete_vpn_connection(self):
vpn_connection_id = self._get_param("VpnConnectionId")
vpn_connection = self.ec2_backend.delete_vpn_connection(vpn_connection_id)
if vpn_connection.transit_gateway_id:
transit_gateway_attachments = (
self.ec2_backend.describe_transit_gateway_attachments()
)
for attachment in transit_gateway_attachments:
if attachment.resource_id == vpn_connection.id:
attachment.state = "deleted"
template = self.response_template(DELETE_VPN_CONNECTION_RESPONSE)
return template.render(vpn_connection=vpn_connection)
def describe_vpn_connections(self):
vpn_connection_ids = self._get_multi_param("VpnConnectionId")
filters = self._filters_from_querystring()
vpn_connections = self.ec2_backend.get_all_vpn_connections(
vpn_connection_ids=vpn_connection_ids, filters=filters
)
template = self.response_template(DESCRIBE_VPN_CONNECTION_RESPONSE)
return template.render(vpn_connections=vpn_connections)
CUSTOMER_GATEWAY_CONFIGURATION_TEMPLATE = """
{{ vpn_connection.customer_gateway_id }}
{{ vpn_connection.vpn_gateway_id if vpn_connection.vpn_gateway_id is not none }}
{{ vpn_connection.type }}
12.1.2.3
169.254.44.42
255.255.255.252
30
65000
30
52.2.144.13
169.254.44.41
255.255.255.252
30
7224
30
sha1
aes-128-cbc
28800
group2
main
Iw2IAN9XUsQeYUrkMGP3kP59ugFDkfHg
esp
hmac-sha1-96
aes-128-cbc
3600
group2
tunnel
true
true
1387
10
3
12.1.2.3
169.254.44.42
255.255.255.252
30
65000
30
52.2.144.13
169.254.44.41
255.255.255.252
30
7224
30
sha1
aes-128-cbc
28800
group2
main
Iw2IAN9XUsQeYUrkMGP3kP59ugFDkfHg
esp
hmac-sha1-96
aes-128-cbc
3600
group2
tunnel
true
true
1387
10
3
"""
CREATE_VPN_CONNECTION_RESPONSE = (
"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
{{ vpn_connection.id }}
{{ vpn_connection.state }}
"""
+ escape(CUSTOMER_GATEWAY_CONFIGURATION_TEMPLATE)
+ """
ipsec.1
{{ vpn_connection.customer_gateway_id }}
{{ vpn_connection.vpn_gateway_id or '' }}
{% if vpn_connection.transit_gateway_id %}
{{ vpn_connection.transit_gateway_id }}
{% endif %}
{% for tag in vpn_connection.get_tags() %}
-
{{ tag.key }}
{{ tag.value }}
{% endfor %}
"""
)
CREATE_VPN_CONNECTION_ROUTE_RESPONSE = """
4f35a1b2-c2c3-4093-b51f-abb9d7311990
true
"""
DELETE_VPN_CONNECTION_RESPONSE = """
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
true
"""
DELETE_VPN_CONNECTION_ROUTE_RESPONSE = """
4f35a1b2-c2c3-4093-b51f-abb9d7311990
true
"""
DESCRIBE_VPN_CONNECTION_RESPONSE = (
"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
{% for vpn_connection in vpn_connections %}
-
{{ vpn_connection.id }}
{{ vpn_connection.state }}
"""
+ escape(CUSTOMER_GATEWAY_CONFIGURATION_TEMPLATE)
+ """
ipsec.1
{{ vpn_connection.customer_gateway_id }}
{{ vpn_connection.vpn_gateway_id or '' }}
{% if vpn_connection.transit_gateway_id %}
{{ vpn_connection.transit_gateway_id }}
{% endif %}
{% for tag in vpn_connection.get_tags() %}
-
{{ tag.key }}
{{ tag.value }}
{% endfor %}
{% endfor %}
"""
)