Merge pull request #261 from tsanders/master

Added support for mocking VPC Network ACLs
This commit is contained in:
Steve Pulec 2014-11-17 20:15:50 -05:00
commit 63ae707828
5 changed files with 399 additions and 12 deletions

View File

@ -72,6 +72,14 @@ class InvalidSubnetIdError(EC2ClientError):
.format(subnet_id))
class InvalidNetworkAclIdError(EC2ClientError):
def __init__(self, network_acl_id):
super(InvalidNetworkAclIdError, self).__init__(
"InvalidNetworkAclID.NotFound",
"The network acl ID '{0}' does not exist"
.format(network_acl_id))
class InvalidNetworkInterfaceIdError(EC2ClientError):
def __init__(self, eni_id):
super(InvalidNetworkInterfaceIdError, self).__init__(

View File

@ -52,7 +52,8 @@ from .exceptions import (
InvalidVPCPeeringConnectionStateTransitionError,
TagLimitExceeded,
InvalidID,
InvalidCIDRSubnetError
InvalidCIDRSubnetError,
InvalidNetworkAclIdError
)
from .utils import (
EC2_RESOURCE_TO_PREFIX,
@ -87,7 +88,8 @@ from .utils import (
is_valid_cidr,
filter_internet_gateways,
filter_reservations,
)
random_network_acl_id,
random_network_acl_subnet_association_id)
def validate_resource_ids(resource_ids):
@ -1408,6 +1410,9 @@ class VPCBackend(object):
# AWS creates a default main route table and security group.
self.create_route_table(vpc_id, main=True)
# AWS creates a default Network ACL
default_network_acl = self.create_network_acl(vpc_id, default=True)
default = self.get_security_group_from_name('default', vpc_id=vpc_id)
if not default:
self.create_security_group('default', 'default VPC security group', vpc_id=vpc_id)
@ -1602,6 +1607,9 @@ class SubnetBackend(object):
subnet_id = random_subnet_id()
subnet = Subnet(self, subnet_id, vpc_id, cidr_block)
self.get_vpc(vpc_id) # Validate VPC exists
# AWS associates a new subnet with the default Network ACL
self.associate_default_network_acl_with_subnet(subnet_id)
self.subnets[subnet_id] = subnet
return subnet
@ -2274,6 +2282,139 @@ class DHCPOptionsSetBackend(object):
return True
class NetworkAclBackend(object):
def __init__(self):
self.network_acls = {}
super(NetworkAclBackend, self).__init__()
def get_network_acl(self, network_acl_id):
network_acl = self.network_acls.get(network_acl_id, None)
if not network_acl:
raise InvalidNetworkAclIdError(network_acl_id)
return network_acl
def create_network_acl(self, vpc_id, default=False):
network_acl_id = random_network_acl_id()
vpc = self.get_vpc(vpc_id)
network_acl = NetworkAcl(self, network_acl_id, vpc_id, default)
self.network_acls[network_acl_id] = network_acl
return network_acl
def get_all_network_acls(self, network_acl_ids=None, filters=None):
network_acls = self.network_acls.values()
if network_acl_ids:
network_acls = [network_acl for network_acl in network_acls
if network_acl.id in network_acl_ids ]
if len(network_acls) != len(network_acl_ids):
invalid_id = list(set(network_acl_ids).difference(set([network_acl.id for network_acl in network_acls])))[0]
raise InvalidRouteTableIdError(invalid_id)
return generic_filter(filters, network_acls)
def delete_network_acl(self, network_acl_id):
deleted = self.network_acls.pop(network_acl_id, None)
if not deleted:
raise InvalidNetworkAclIdError(network_acl_id)
return deleted
def create_network_acl_entry(self, network_acl_id, rule_number,
protocol, rule_action, egress, cidr_block,
icmp_code, icmp_type, port_range_from,
port_range_to):
network_acl_entry = NetworkAclEntry(self, network_acl_id, rule_number,
protocol, rule_action, egress,
cidr_block, icmp_code, icmp_type,
port_range_from, port_range_to)
network_acl = self.get_network_acl(network_acl_id)
network_acl.network_acl_entries.append(network_acl_entry)
return network_acl_entry
def replace_network_acl_association(self, association_id,
network_acl_id):
# lookup existing association for subnet and delete it
default_acl = next(value for key, value in self.network_acls.items()
if association_id in value.associations.keys())
subnet_id = None
for key, value in default_acl.associations.items():
if key == association_id:
subnet_id = default_acl.associations[key].subnet_id
del default_acl.associations[key]
break
new_assoc_id = random_network_acl_subnet_association_id()
association = NetworkAclAssociation(self,
new_assoc_id,
subnet_id,
network_acl_id)
new_acl = self.get_network_acl(network_acl_id)
new_acl.associations[new_assoc_id] = association
return association
def associate_default_network_acl_with_subnet(self, subnet_id):
association_id = random_network_acl_subnet_association_id()
acl = next(acl for acl in self.network_acls.values() if acl.default)
acl.associations[association_id] = NetworkAclAssociation(self, association_id,
subnet_id, acl.id)
class NetworkAclAssociation(object):
def __init__(self, ec2_backend, new_association_id,
subnet_id, network_acl_id):
self.ec2_backend = ec2_backend
self.id = new_association_id
self.subnet_id = subnet_id
self.network_acl_id = network_acl_id
super(NetworkAclAssociation, self).__init__()
class NetworkAcl(TaggedEC2Resource):
def __init__(self, ec2_backend, network_acl_id, vpc_id, default=False):
self.ec2_backend = ec2_backend
self.id = network_acl_id
self.vpc_id = vpc_id
self.network_acl_entries = []
self.associations = {}
self.default = default
def get_filter_value(self, filter_name):
if filter_name == "vpc-id":
return self.vpc_id
elif filter_name == "association.network-acl-id":
return self.id
elif filter_name == "association.subnet-id":
return [assoc.subnet_id for assoc in self.associations.values()]
filter_value = super(NetworkAcl, self).get_filter_value(filter_name)
if filter_value is None:
self.ec2_backend.raise_not_implemented_error("The filter '{0}' for DescribeNetworkAcls".format(filter_name))
return filter_value
class NetworkAclEntry(TaggedEC2Resource):
def __init__(self, ec2_backend, network_acl_id, rule_number,
protocol, rule_action, egress, cidr_block,
icmp_code, icmp_type, port_range_from,
port_range_to):
self.ec2_backend = ec2_backend
self.network_acl_id = network_acl_id
self.rule_number = rule_number
self.protocol = protocol
self.rule_action = rule_action
self.egress = egress
self.cidr_block = cidr_block
self.icmp_code = icmp_code
self.icmp_type = icmp_type
self.port_range_from = port_range_from
self.port_range_to = port_range_to
class EC2Backend(BaseBackend, InstanceBackend, TagBackend, AmiBackend,
RegionsAndZonesBackend, SecurityGroupBackend, EBSBackend,
VPCBackend, SubnetBackend, SubnetRouteTableAssociationBackend,
@ -2281,7 +2422,8 @@ class EC2Backend(BaseBackend, InstanceBackend, TagBackend, AmiBackend,
VPCPeeringConnectionBackend,
RouteTableBackend, RouteBackend, InternetGatewayBackend,
VPCGatewayAttachmentBackend, SpotRequestBackend,
ElasticAddressBackend, KeyPairBackend, DHCPOptionsSetBackend):
ElasticAddressBackend, KeyPairBackend, DHCPOptionsSetBackend,
NetworkAclBackend):
# Use this to generate a proper error template response when in a response handler.
def raise_error(self, code, message):
@ -2307,7 +2449,7 @@ class EC2Backend(BaseBackend, InstanceBackend, TagBackend, AmiBackend,
elif resource_prefix == EC2_RESOURCE_TO_PREFIX['internet-gateway']:
self.describe_internet_gateways(internet_gateway_ids=[resource_id])
elif resource_prefix == EC2_RESOURCE_TO_PREFIX['network-acl']:
self.raise_not_implemented_error('DescribeNetworkAcls')
self.get_all_network_acls()
elif resource_prefix == EC2_RESOURCE_TO_PREFIX['network-interface']:
self.describe_network_interfaces(filters={'network-interface-id': resource_id})
elif resource_prefix == EC2_RESOURCE_TO_PREFIX['reserved-instance']:

View File

@ -1,25 +1,144 @@
from __future__ import unicode_literals
from jinja2 import Template
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):
raise NotImplementedError('NetworkACLs(AmazonVPC).create_network_acl is not yet implemented')
vpc_id = self.querystring.get('VpcId')[0]
network_acl = self.ec2_backend.create_network_acl(vpc_id)
template = Template(CREATE_NETWORK_ACL_RESPONSE)
return template.render(network_acl=network_acl)
def create_network_acl_entry(self):
raise NotImplementedError('NetworkACLs(AmazonVPC).create_network_acl_entry is not yet implemented')
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 = Template(CREATE_NETWORK_ACL_ENTRY_RESPONSE)
return template.render(network_acl_entry=network_acl_entry)
def delete_network_acl(self):
raise NotImplementedError('NetworkACLs(AmazonVPC).delete_network_acl is not yet implemented')
network_acl_id = self.querystring.get('NetworkAclId')[0]
self.ec2_backend.delete_network_acl(network_acl_id)
template = 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')
raise NotImplementedError(
'NetworkACLs(AmazonVPC).delete_network_acl_entry is not yet implemented')
def describe_network_acls(self):
raise NotImplementedError('NetworkACLs(AmazonVPC).describe_network_acls is not yet implemented')
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 = Template(DESCRIBE_NETWORK_ACL_RESPONSE)
return template.render(network_acls=network_acls)
def replace_network_acl_association(self):
raise NotImplementedError('NetworkACLs(AmazonVPC).replace_network_acl_association is not yet implemented')
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 = 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')
raise NotImplementedError(
'NetworkACLs(AmazonVPC).replace_network_acl_entry is not yet implemented')
CREATE_NETWORK_ACL_RESPONSE = """
<CreateNetworkAclResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<networkAcl>
<networkAclId>{{ network_acl.id }}</networkAclId>
<vpcId>{{ network_acl.vpc_id }}</vpcId>
<default>false</default>
<entrySet/>
<associationSet/>
<tagSet/>
</networkAcl>
</CreateNetworkAclResponse>
"""
DESCRIBE_NETWORK_ACL_RESPONSE = """
<DescribeNetworkAclsResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<networkAclSet>
{% for network_acl in network_acls %}
<item>
<networkAclId>{{ network_acl.id }}</networkAclId>
<vpcId>{{ network_acl.vpc_id }}</vpcId>
<default>true</default>
<entrySet>
{% for entry in network_acl.network_acl_entries %}
<item>
<ruleNumber>{{ entry.rule_number }}</ruleNumber>
<protocol>{{ entry.protocol }}</protocol>
<ruleAction>{{ entry.rule_action }}</ruleAction>
<egress>{{ entry.egress.lower() }}</egress>
<cidrBlock>{{ entry.cidr_block }}</cidrBlock>
{% if entry.port_range_from or entry.port_range_to %}
<portRange>
<from>{{ entry.port_range_from }}</from>
<to>{{ entry.port_range_to }}</to>
</portRange>
{% endif %}
</item>
{% endfor %}
</entrySet>
<associationSet>
{% for association in network_acl.associations.values() %}
<item>
<networkAclAssociationId>{{ association.id }}</networkAclAssociationId>
<networkAclId>{{ association.network_acl_id }}</networkAclId>
<subnetId>{{ association.subnet_id }}</subnetId>
</item>
{% endfor %}
</associationSet>
<tagSet/>
</item>
{% endfor %}
</networkAclSet>
</DescribeNetworkAclsResponse>
"""
CREATE_NETWORK_ACL_ENTRY_RESPONSE = """
<CreateNetworkAclEntryResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</CreateNetworkAclEntryResponse>
"""
REPLACE_NETWORK_ACL_ASSOCIATION = """
<ReplaceNetworkAclAssociationResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<newAssociationId>{{ association.new_association_id }}</newAssociationId>
</ReplaceNetworkAclAssociationResponse>
"""
DELETE_NETWORK_ACL_ASSOCIATION = """
<DeleteNetworkAclResponse xmlns="http://ec2.amazonaws.com/doc/2014-10-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</DeleteNetworkAclResponse>
"""

View File

@ -10,6 +10,7 @@ EC2_RESOURCE_TO_PREFIX = {
'instance': 'i',
'internet-gateway': 'igw',
'network-acl': 'acl',
'network-acl-subnet-assoc': 'aclassoc',
'network-interface': 'eni',
'network-interface-attachment': 'eni-attach',
'reserved-instance': 'uuid4',
@ -72,6 +73,14 @@ def random_subnet_association_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['route-table-association'])
def random_network_acl_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['network-acl'])
def random_network_acl_subnet_association_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['network-acl-subnet-assoc'])
def random_volume_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['volume'])
@ -158,6 +167,14 @@ def route_table_ids_from_querystring(querystring_dict):
return route_table_ids
def network_acl_ids_from_querystring(querystring_dict):
network_acl_ids = []
for key, value in querystring_dict.items():
if 'NetworkAclId' in key:
network_acl_ids.append(value[0])
return network_acl_ids
def vpc_ids_from_querystring(querystring_dict):
vpc_ids = []
for key, value in querystring_dict.items():

View File

@ -5,6 +5,107 @@ import sure # noqa
from moto import mock_ec2
@mock_ec2
def test_default_network_acl_created_with_vpc():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(1)
@mock_ec2
def test_network_acls():
pass
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
network_acl = conn.create_network_acl(vpc.id)
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(2)
@mock_ec2
def test_new_subnet_associates_with_default_network_acl():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
subnet = conn.create_subnet(vpc.id, "10.0.0.0/18")
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(1)
acl = all_network_acls[0]
acl.associations.should.have.length_of(1)
acl.associations[0].subnet_id.should.equal(subnet.id)
@mock_ec2
def test_network_acl_entries():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
network_acl = conn.create_network_acl(vpc.id)
network_acl_entry = conn.create_network_acl_entry(
network_acl.id, 110, 6,
'ALLOW', '0.0.0.0/0', False,
port_range_from='443',
port_range_to='443'
)
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(2)
test_network_acl = next(na for na in all_network_acls
if na.id == network_acl.id)
entries = test_network_acl.network_acl_entries
entries.should.have.length_of(1)
entries[0].rule_number.should.equal('110')
entries[0].protocol.should.equal('6')
entries[0].rule_action.should.equal('ALLOW')
@mock_ec2
def test_associate_new_network_acl_with_subnet():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
subnet = conn.create_subnet(vpc.id, "10.0.0.0/18")
network_acl = conn.create_network_acl(vpc.id)
conn.associate_network_acl(network_acl.id, subnet.id)
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(2)
test_network_acl = next(na for na in all_network_acls
if na.id == network_acl.id)
test_network_acl.associations.should.have.length_of(1)
test_network_acl.associations[0].subnet_id.should.equal(subnet.id)
@mock_ec2
def test_delete_network_acl():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.create_vpc("10.0.0.0/16")
subnet = conn.create_subnet(vpc.id, "10.0.0.0/18")
network_acl = conn.create_network_acl(vpc.id)
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(2)
any(acl.id == network_acl.id for acl in all_network_acls).should.be.ok
conn.delete_network_acl(network_acl.id)
updated_network_acls = conn.get_all_network_acls()
updated_network_acls.should.have.length_of(1)
any(acl.id == network_acl.id for acl in updated_network_acls).shouldnt.be.ok