from __future__ import unicode_literals from moto.core.responses import BaseResponse from moto.ec2.utils import filters_from_querystring, \ network_acl_ids_from_querystring class NetworkACLs(BaseResponse): def create_network_acl(self): vpc_id = self.querystring.get('VpcId')[0] network_acl = self.ec2_backend.create_network_acl(vpc_id) template = self.response_template(CREATE_NETWORK_ACL_RESPONSE) return template.render(network_acl=network_acl) def create_network_acl_entry(self): network_acl_id = self.querystring.get('NetworkAclId')[0] rule_number = self.querystring.get('RuleNumber')[0] protocol = self.querystring.get('Protocol')[0] rule_action = self.querystring.get('RuleAction')[0] egress = self.querystring.get('Egress')[0] cidr_block = self.querystring.get('CidrBlock')[0] icmp_code = self.querystring.get('Icmp.Code', [None])[0] icmp_type = self.querystring.get('Icmp.Type', [None])[0] port_range_from = self.querystring.get('PortRange.From')[0] port_range_to = self.querystring.get('PortRange.To')[0] network_acl_entry = self.ec2_backend.create_network_acl_entry( network_acl_id, rule_number, protocol, rule_action, egress, cidr_block, icmp_code, icmp_type, port_range_from, port_range_to) template = self.response_template(CREATE_NETWORK_ACL_ENTRY_RESPONSE) return template.render(network_acl_entry=network_acl_entry) def delete_network_acl(self): network_acl_id = self.querystring.get('NetworkAclId')[0] self.ec2_backend.delete_network_acl(network_acl_id) template = self.response_template(DELETE_NETWORK_ACL_ASSOCIATION) return template.render() def delete_network_acl_entry(self): raise NotImplementedError( 'NetworkACLs(AmazonVPC).delete_network_acl_entry is not yet implemented') def describe_network_acls(self): network_acl_ids = network_acl_ids_from_querystring(self.querystring) filters = filters_from_querystring(self.querystring) network_acls = self.ec2_backend.get_all_network_acls(network_acl_ids, filters) template = self.response_template(DESCRIBE_NETWORK_ACL_RESPONSE) return template.render(network_acls=network_acls) def replace_network_acl_association(self): association_id = self.querystring.get('AssociationId')[0] network_acl_id = self.querystring.get('NetworkAclId')[0] association = self.ec2_backend.replace_network_acl_association( association_id, network_acl_id ) template = self.response_template(REPLACE_NETWORK_ACL_ASSOCIATION) return template.render(association=association) def replace_network_acl_entry(self): raise NotImplementedError( 'NetworkACLs(AmazonVPC).replace_network_acl_entry is not yet implemented') CREATE_NETWORK_ACL_RESPONSE = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE {{ network_acl.id }} {{ network_acl.vpc_id }} false {% for tag in network_acl.get_tags() %} {{ tag.resource_id }} {{ tag.resource_type }} {{ tag.key }} {{ tag.value }} {% endfor %} """ DESCRIBE_NETWORK_ACL_RESPONSE = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE {% for network_acl in network_acls %} {{ network_acl.id }} {{ network_acl.vpc_id }} true {% for entry in network_acl.network_acl_entries %} {{ entry.rule_number }} {{ entry.protocol }} {{ entry.rule_action }} {{ entry.egress.lower() }} {{ entry.cidr_block }} {% if entry.port_range_from or entry.port_range_to %} {{ entry.port_range_from }} {{ entry.port_range_to }} {% endif %} {% endfor %} {% for association in network_acl.associations.values() %} {{ association.id }} {{ association.network_acl_id }} {{ association.subnet_id }} {% endfor %} {% for tag in network_acl.get_tags() %} {{ tag.resource_id }} {{ tag.resource_type }} {{ tag.key }} {{ tag.value }} {% endfor %} {% endfor %} """ CREATE_NETWORK_ACL_ENTRY_RESPONSE = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE true """ REPLACE_NETWORK_ACL_ASSOCIATION = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE {{ association.new_association_id }} """ DELETE_NETWORK_ACL_ASSOCIATION = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE true """