from moto.core import ACCOUNT_ID from moto.core.utils import camelcase_to_underscores from moto.ec2.utils import add_tag_specification from ._base_response import EC2BaseResponse class VPCs(EC2BaseResponse): def _get_doc_date(self): return ( "2013-10-15" if "Boto/" in self.headers.get("user-agent", "") else "2016-11-15" ) def create_vpc(self): cidr_block = self._get_param("CidrBlock") tags = self._get_multi_param("TagSpecification") instance_tenancy = self._get_param("InstanceTenancy", if_none="default") amazon_provided_ipv6_cidr_block = self._get_param( "AmazonProvidedIpv6CidrBlock" ) in ["true", "True"] if tags: tags = tags[0].get("Tag") vpc = self.ec2_backend.create_vpc( cidr_block, instance_tenancy, amazon_provided_ipv6_cidr_block=amazon_provided_ipv6_cidr_block, tags=tags, ) doc_date = self._get_doc_date() template = self.response_template(CREATE_VPC_RESPONSE) return template.render(vpc=vpc, doc_date=doc_date) def delete_vpc(self): vpc_id = self._get_param("VpcId") vpc = self.ec2_backend.delete_vpc(vpc_id) template = self.response_template(DELETE_VPC_RESPONSE) return template.render(vpc=vpc) def describe_vpcs(self): self.error_on_dryrun() vpc_ids = self._get_multi_param("VpcId") filters = self._filters_from_querystring() vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters) doc_date = ( "2013-10-15" if "Boto/" in self.headers.get("user-agent", "") else "2016-11-15" ) template = self.response_template(DESCRIBE_VPCS_RESPONSE) return template.render(vpcs=vpcs, doc_date=doc_date) def modify_vpc_tenancy(self): vpc_id = self._get_param("VpcId") tenancy = self._get_param("InstanceTenancy") value = self.ec2_backend.modify_vpc_tenancy(vpc_id, tenancy) template = self.response_template(MODIFY_VPC_TENANCY_RESPONSE) return template.render(value=value) def describe_vpc_attribute(self): vpc_id = self._get_param("VpcId") attribute = self._get_param("Attribute") attr_name = camelcase_to_underscores(attribute) value = self.ec2_backend.describe_vpc_attribute(vpc_id, attr_name) template = self.response_template(DESCRIBE_VPC_ATTRIBUTE_RESPONSE) return template.render(vpc_id=vpc_id, attribute=attribute, value=value) def describe_vpc_classic_link_dns_support(self): vpc_ids = self._get_multi_param("VpcIds") filters = self._filters_from_querystring() vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters) doc_date = self._get_doc_date() template = self.response_template( DESCRIBE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE ) return template.render(vpcs=vpcs, doc_date=doc_date) def enable_vpc_classic_link_dns_support(self): vpc_id = self._get_param("VpcId") classic_link_dns_supported = ( self.ec2_backend.enable_vpc_classic_link_dns_support(vpc_id=vpc_id) ) doc_date = self._get_doc_date() template = self.response_template(ENABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE) return template.render( classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date ) def disable_vpc_classic_link_dns_support(self): vpc_id = self._get_param("VpcId") classic_link_dns_supported = ( self.ec2_backend.disable_vpc_classic_link_dns_support(vpc_id=vpc_id) ) doc_date = self._get_doc_date() template = self.response_template(DISABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE) return template.render( classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date ) def describe_vpc_classic_link(self): vpc_ids = self._get_multi_param("VpcId") filters = self._filters_from_querystring() vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters) doc_date = self._get_doc_date() template = self.response_template(DESCRIBE_VPC_CLASSIC_LINK_RESPONSE) return template.render(vpcs=vpcs, doc_date=doc_date) def enable_vpc_classic_link(self): vpc_id = self._get_param("VpcId") classic_link_enabled = self.ec2_backend.enable_vpc_classic_link(vpc_id=vpc_id) doc_date = self._get_doc_date() template = self.response_template(ENABLE_VPC_CLASSIC_LINK_RESPONSE) return template.render( classic_link_enabled=classic_link_enabled, doc_date=doc_date ) def disable_vpc_classic_link(self): vpc_id = self._get_param("VpcId") classic_link_enabled = self.ec2_backend.disable_vpc_classic_link(vpc_id=vpc_id) doc_date = self._get_doc_date() template = self.response_template(DISABLE_VPC_CLASSIC_LINK_RESPONSE) return template.render( classic_link_enabled=classic_link_enabled, doc_date=doc_date ) def modify_vpc_attribute(self): vpc_id = self._get_param("VpcId") for attribute in ("EnableDnsSupport", "EnableDnsHostnames"): if self.querystring.get("%s.Value" % attribute): attr_name = camelcase_to_underscores(attribute) attr_value = self.querystring.get("%s.Value" % attribute)[0] self.ec2_backend.modify_vpc_attribute(vpc_id, attr_name, attr_value) return MODIFY_VPC_ATTRIBUTE_RESPONSE return None def associate_vpc_cidr_block(self): vpc_id = self._get_param("VpcId") amazon_provided_ipv6_cidr_blocks = self._get_param( "AmazonProvidedIpv6CidrBlock" ) # todo test on AWS if can create an association for IPV4 and IPV6 in the same call? cidr_block = ( self._get_param("CidrBlock") if not amazon_provided_ipv6_cidr_blocks else None ) value = self.ec2_backend.associate_vpc_cidr_block( vpc_id, cidr_block, amazon_provided_ipv6_cidr_blocks ) if not amazon_provided_ipv6_cidr_blocks: render_template = ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE else: render_template = IPV6_ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE template = self.response_template(render_template) return template.render( vpc_id=vpc_id, value=value, cidr_block=value["cidr_block"], association_id=value["association_id"], cidr_block_state="associating", ) def disassociate_vpc_cidr_block(self): association_id = self._get_param("AssociationId") value = self.ec2_backend.disassociate_vpc_cidr_block(association_id) if "::" in value.get("cidr_block", ""): render_template = IPV6_DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE else: render_template = DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE template = self.response_template(render_template) return template.render( vpc_id=value["vpc_id"], cidr_block=value["cidr_block"], association_id=value["association_id"], cidr_block_state="disassociating", ) def create_vpc_endpoint(self): vpc_id = self._get_param("VpcId") service_name = self._get_param("ServiceName") route_table_ids = self._get_multi_param("RouteTableId") subnet_ids = self._get_multi_param("SubnetId") endpoint_type = self._get_param("VpcEndpointType") policy_document = self._get_param("PolicyDocument") client_token = self._get_param("ClientToken") tags = self._get_multi_param("TagSpecification") private_dns_enabled = self._get_bool_param("PrivateDnsEnabled", if_none=True) security_group_ids = self._get_multi_param("SecurityGroupId") tags = add_tag_specification(tags) vpc_end_point = self.ec2_backend.create_vpc_endpoint( vpc_id=vpc_id, service_name=service_name, endpoint_type=endpoint_type, policy_document=policy_document, route_table_ids=route_table_ids, subnet_ids=subnet_ids, client_token=client_token, security_group_ids=security_group_ids, tags=tags, private_dns_enabled=private_dns_enabled, ) template = self.response_template(CREATE_VPC_END_POINT) return template.render(vpc_end_point=vpc_end_point) def describe_vpc_endpoint_services(self): vpc_end_point_services = self.ec2_backend.describe_vpc_endpoint_services( dry_run=self._get_bool_param("DryRun"), service_names=self._get_multi_param("ServiceName"), filters=self._get_multi_param("Filter"), max_results=self._get_int_param("MaxResults"), next_token=self._get_param("NextToken"), region=self.region, ) template = self.response_template(DESCRIBE_VPC_ENDPOINT_SERVICES_RESPONSE) return template.render(vpc_end_points=vpc_end_point_services) def describe_vpc_endpoints(self): vpc_end_points_ids = self._get_multi_param("VpcEndpointId") filters = self._filters_from_querystring() vpc_end_points = self.ec2_backend.describe_vpc_endpoints( vpc_end_point_ids=vpc_end_points_ids, filters=filters ) template = self.response_template(DESCRIBE_VPC_ENDPOINT_RESPONSE) return template.render(vpc_end_points=vpc_end_points, account_id=ACCOUNT_ID) def delete_vpc_endpoints(self): vpc_end_points_ids = self._get_multi_param("VpcEndpointId") response = self.ec2_backend.delete_vpc_endpoints(vpce_ids=vpc_end_points_ids) template = self.response_template(DELETE_VPC_ENDPOINT_RESPONSE) return template.render(response=response) def create_managed_prefix_list(self): address_family = self._get_param("AddressFamily") max_entries = self._get_param("MaxEntries") prefix_list_name = self._get_param("PrefixListName") entry = self._get_multi_param("Entry") tags = self._get_multi_param("TagSpecification") tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags tags = (tags or {}).get("Tag", []) tags = {t["Key"]: t["Value"] for t in tags} managed_prefix_list = self.ec2_backend.create_managed_prefix_list( address_family=address_family, entry=entry, max_entries=max_entries, prefix_list_name=prefix_list_name, tags=tags, ) template = self.response_template(CREATE_MANAGED_PREFIX_LIST) return template.render(managed_prefix_list=managed_prefix_list) def describe_managed_prefix_lists(self): prefix_list_ids = self._get_multi_param("PrefixListId") filters = self._filters_from_querystring() managed_prefix_lists = self.ec2_backend.describe_managed_prefix_lists( prefix_list_ids=prefix_list_ids, filters=filters ) template = self.response_template(DESCRIBE_MANAGED_PREFIX_LIST) return template.render(managed_prefix_lists=managed_prefix_lists) def get_managed_prefix_list_entries(self): prefix_list_id = self._get_param("PrefixListId") target_version = self._get_param("TargetVersion") managed_prefix_list = self.ec2_backend.get_managed_prefix_list_entries( prefix_list_id=prefix_list_id ) entries = [] if managed_prefix_list: entries = ( list(managed_prefix_list.entries.values())[-1] if managed_prefix_list.entries.values() else [] ) if target_version: target_version = int(target_version) entries = managed_prefix_list.entries.get(target_version) template = self.response_template(GET_MANAGED_PREFIX_LIST_ENTRIES) return template.render(entries=entries) def delete_managed_prefix_list(self): prefix_list_id = self._get_param("PrefixListId") managed_prefix_list = self.ec2_backend.delete_managed_prefix_list( prefix_list_id ) template = self.response_template(DELETE_MANAGED_PREFIX_LIST) return template.render(managed_prefix_list=managed_prefix_list) def describe_prefix_lists(self): prefix_list_ids = self._get_multi_param("PrefixListId") filters = self._filters_from_querystring() managed_pls = self.ec2_backend.describe_managed_prefix_lists( prefix_list_ids=prefix_list_ids, filters=filters ) template = self.response_template(DESCRIBE_PREFIX_LIST) return template.render(managed_pls=managed_pls) def modify_managed_prefix_list(self): add_entry = self._get_multi_param("AddEntry") prefix_list_id = self._get_param("PrefixListId") current_version = self._get_param("CurrentVersion") prefix_list_name = self._get_param("PrefixListName") remove_entry = self._get_multi_param("RemoveEntry") current_version = int(current_version) if current_version else None managed_prefix_list = self.ec2_backend.modify_managed_prefix_list( add_entry=add_entry, prefix_list_id=prefix_list_id, current_version=current_version, prefix_list_name=prefix_list_name, remove_entry=remove_entry, ) template = self.response_template(MODIFY_PREFIX_LIST) return template.render(managed_prefix_list=managed_prefix_list) CREATE_VPC_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE {{ vpc.id }} pending {{ vpc.cidr_block }} {% if doc_date == "2016-11-15" %} {% for assoc in vpc.get_cidr_block_association_set() %} {{assoc.cidr_block}} {{ assoc.association_id }} {{assoc.cidr_block_state.state}} {% endfor %} {% for assoc in vpc.get_cidr_block_association_set(ipv6=True) %} {{assoc.cidr_block}} {{ assoc.association_id }} {{assoc.cidr_block_state.state}} {% endfor %} {% endif %} {% if vpc.dhcp_options %}{{ vpc.dhcp_options.id }}{% else %}dopt-1a2b3c4d2{% endif %} {{ vpc.instance_tenancy }} {{ vpc.owner_id }} {% for tag in vpc.get_tags() %} {{ tag.resource_id }} {{ tag.resource_type }} {{ tag.key }} {{ tag.value }} {% endfor %} """ DESCRIBE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {% for vpc in vpcs %} {{ vpc.id }} {{ vpc.classic_link_dns_supported }} {% endfor %} """ ENABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {{ classic_link_dns_supported }} """ DISABLE_VPC_CLASSIC_LINK_DNS_SUPPORT_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {{ classic_link_dns_supported }} """ DESCRIBE_VPC_CLASSIC_LINK_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {% for vpc in vpcs %} {{ vpc.id }} {{ vpc.classic_link_enabled }} {% endfor %} """ ENABLE_VPC_CLASSIC_LINK_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {{ classic_link_enabled }} """ DISABLE_VPC_CLASSIC_LINK_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {{ classic_link_enabled }} """ DESCRIBE_VPCS_RESPONSE = """ 7a62c442-3484-4f42-9342-6942EXAMPLE {% for vpc in vpcs %} {{ vpc.id }} {{ vpc.state }} {{ vpc.cidr_block }} {% if doc_date == "2016-11-15" %} {% for assoc in vpc.get_cidr_block_association_set() %} {{assoc.cidr_block}} {{ assoc.association_id }} {{assoc.cidr_block_state.state}} {% endfor %} {% for assoc in vpc.get_cidr_block_association_set(ipv6=True) %} {{assoc.cidr_block}} {{ assoc.association_id }} {{assoc.cidr_block_state.state}} {% endfor %} {% endif %} {% if vpc.dhcp_options %}{{ vpc.dhcp_options.id }}{% else %}dopt-7a8b9c2d{% endif %} {{ vpc.instance_tenancy }} {{ vpc.is_default }} {{ vpc.owner_id }} {% for tag in vpc.get_tags() %} {{ tag.resource_id }} {{ tag.resource_type }} {{ tag.key }} {{ tag.value }} {% endfor %} {% endfor %} """ DELETE_VPC_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE true """ MODIFY_VPC_TENANCY_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE true """ DESCRIBE_VPC_ATTRIBUTE_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE {{ vpc_id }} <{{ attribute }}> {{ value }} """ MODIFY_VPC_ATTRIBUTE_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE true """ ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE {{vpc_id}} {{association_id}} {{cidr_block}} {{cidr_block_state}} """ DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """ 7a62c49f-347e-4fc4-9331-6e8eEXAMPLE {{vpc_id}} {{association_id}} {{cidr_block}} {{cidr_block_state}} """ IPV6_ASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """ 33af6c54-1139-4d50-b4f7-15a8example {{vpc_id}} {{association_id}} {{cidr_block}} {{cidr_block_state}} """ IPV6_DISASSOCIATE_VPC_CIDR_BLOCK_RESPONSE = """ 33af6c54-1139-4d50-b4f7-15a8example {{vpc_id}} {{association_id}} {{cidr_block}} {{cidr_block_state}} """ CREATE_VPC_END_POINT = """ {{ vpc_end_point.policy_document }} {{ vpc_end_point.state }} false {{ vpc_end_point.service_name }} {{ vpc_end_point.vpc_id }} {{ vpc_end_point.id }} {% for routeid in vpc_end_point.route_table_ids %} {{ routeid }} {% endfor %} {% for network_interface_id in vpc_end_point.network_interface_ids %} {{ network_interface_id }} {% endfor %} {% for subnetId in vpc_end_point.subnet_ids %} {{ subnetId }} {% endfor %} {{ 'true' if vpc_end_point.private_dns_enabled else 'false' }} {% if vpc_end_point.dns_entries %} {% for entry in vpc_end_point.dns_entries %} {{ entry["hosted_zone_id"] }} {{ entry["dns_name"] }} {% endfor %} {% endif %} {% for tag in vpc_end_point.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {{ vpc_end_point.created_at }} """ DESCRIBE_VPC_ENDPOINT_SERVICES_RESPONSE = """ 19a9ff46-7df6-49b8-9726-3df27527089d {% for serviceName in vpc_end_points.serviceNames %} {{ serviceName }} {% endfor %} {% for service in vpc_end_points.servicesDetails %} {{ 'true' if service.AcceptanceRequired else 'false' }} {% for zone in service.AvailabilityZones %} {{ zone }} {% endfor %} {% for endpoint in service.BaseEndpointDnsNames %} {{ endpoint }} {% endfor %} {{ 'true' if service.ManagesVpcEndpoints else 'false' }} {{ service.Owner }} {% if service.PrivateDnsName is defined %} {{ service.PrivateDnsName }} {% for dns_name in service.PrivateDnsNames %} {{ dns_name.PrivateDnsName }} {% endfor %} {{ service.PrivateDnsNameVerificationState }} {% endif %} {{ service.ServiceId }} {{ service.ServiceName }} {% for service_type in service.ServiceType %} {{ service_type.ServiceType }} {% endfor %} {% for tag in service.Tags %} {% for key, value in tag.items() %} {{ key }} {{ value }} {% endfor %} {% endfor %} {{ 'true' if service.VpcEndpointPolicySupported else 'false' }} {% endfor %} {% if vpc_end_points.nextToken|length %} {{ vpc_end_points.nextToken }} {% endif %} """ DESCRIBE_VPC_ENDPOINT_RESPONSE = """ 19a9ff46-7df6-49b8-9726-3df27527089d {% for vpc_end_point in vpc_end_points %} {% if vpc_end_point.policy_document %} {{ vpc_end_point.policy_document }} {% endif %} {{ vpc_end_point.state }} {{ 'true' if vpc_end_point.private_dns_enabled else 'false' }} {{ vpc_end_point.service_name }} {{ vpc_end_point.vpc_id }} {{ vpc_end_point.id }} {{ vpc_end_point.endpoint_type }} {% if vpc_end_point.subnet_ids %} {% for subnet_id in vpc_end_point.subnet_ids %} {{ subnet_id }} {% endfor %} {% endif %} {% if vpc_end_point.route_table_ids %} {% for route_table_id in vpc_end_point.route_table_ids %} {{ route_table_id }} {% endfor %} {% endif %} {% if vpc_end_point.network_interface_ids %} {% for network_interface_id in vpc_end_point.network_interface_ids %} {{ network_interface_id }} {% endfor %} {% endif %} {% if vpc_end_point.dns_entries %} {% for entry in vpc_end_point.dns_entries %} {{ entry["hosted_zone_id"] }} {{ entry["dns_name"] }} {% endfor %} {% endif %} {% if vpc_end_point.security_group_ids %} {% for group_id in vpc_end_point.security_group_ids %} {{ group_id }} TODO {% endfor %} {% endif %} {% for tag in vpc_end_point.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {{ account_id }} {{ vpc_end_point.created_at }} {% endfor %} """ DELETE_VPC_ENDPOINT_RESPONSE = """ 19a9ff46-7df6-49b8-9726-3df27527089d {{ 'Error' if not response else '' }} """ CREATE_MANAGED_PREFIX_LIST = """ {{ managed_prefix_list.address_family }} {{ managed_prefix_list.max_entries }} {{ managed_prefix_list.owner_id }} {{ managed_prefix_list.prefix_list_arn }} {{ managed_prefix_list.id }} {{ managed_prefix_list.prefix_list_name }} {{ managed_prefix_list.state }} {% for tag in managed_prefix_list.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {{ managed_prefix_list.version }} """ DESCRIBE_MANAGED_PREFIX_LIST = """ 934214d3-4501-4797-b896-13e8fc7ec256 {% for managed_prefix_list in managed_prefix_lists %} {{ managed_prefix_list.address_family }} {% if managed_prefix_list.max_entries %} {{ managed_prefix_list.max_entries }} {% endif %} {{ managed_prefix_list.owner_id }} {{ managed_prefix_list.prefix_list_arn }} {{ managed_prefix_list.id }} {{ managed_prefix_list.prefix_list_name }} {{ managed_prefix_list.state }} {% for tag in managed_prefix_list.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {% if managed_prefix_list.version %} {{ managed_prefix_list.version }} {% endif %} {% endfor %} """ GET_MANAGED_PREFIX_LIST_ENTRIES = """ 39a3c79f-846f-4382-a592-example {% for entry in entries %} {{ entry.Cidr or ''}} {{ entry.Description or ''}} {% endfor %} """ DELETE_MANAGED_PREFIX_LIST = """ 39a3c79f-846f-4382-a592-example {{ managed_prefix_list.address_family }} {{ managed_prefix_list.max_entries or '' }} {{ managed_prefix_list.owner_id }} {{ managed_prefix_list.prefix_list_arn }} {{ managed_prefix_list.id }} {{ managed_prefix_list.prefix_list_name }} {{ managed_prefix_list.state }} {% for tag in managed_prefix_list.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {{ managed_prefix_list.version or ''}} """ DESCRIBE_PREFIX_LIST = """ 8a2ec0e2-6918-4270-ae45-58e61971e97d {% for pl in managed_pls %} {% if pl.prefix_list_name and pl.prefix_list_name.startswith("com.amazonaws.") %} {% for entry in pl.entries.1 %} {{ entry.Cidr }} {% endfor %} {{ pl.id }} {{ pl.prefix_list_name }} {% endif %} {% endfor %} """ MODIFY_PREFIX_LIST = """ 602f3752-c348-4b14-81e2-example {{ managed_prefix_list.address_family }} {{ managed_prefix_list.max_entries or '' }} {{ managed_prefix_list.owner_id }} {{ managed_prefix_list.prefix_list_arn }} {{ managed_prefix_list.id }} {{ managed_prefix_list.prefix_list_name }} {{ managed_prefix_list.state }} {% for tag in managed_prefix_list.get_tags() %} {{ tag.key }} {{ tag.value }} {% endfor %} {{ managed_prefix_list.version or ''}} """