Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add basic Network ACL create, describe support

Browse Source
This commit is contained in:
Tyler Sanders 2014-11-12 15:33:31 -06:00
parent 79cd93119d
commit 0f18dd597c
5 changed files with 214 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/ec2/exceptions.py
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__(

76
moto/ec2/models.py
View File

@ -51,7 +51,8 @@ from .exceptions import (
InvalidVPCPeeringConnectionStateTransitionError,
TagLimitExceeded,
InvalidID,
InvalidCIDRSubnetError
InvalidCIDRSubnetError,
InvalidNetworkAclIdError
)
from .utils import (
EC2_RESOURCE_TO_PREFIX,
@ -86,6 +87,7 @@ from .utils import (
is_valid_cidr,
filter_internet_gateways,
filter_reservations,
random_network_acl_id,
)
@ -2268,6 +2270,75 @@ 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):
network_acl_id = random_network_acl_id()
network_acl = NetworkAcl(self, network_acl_id, vpc_id)
vpc = self.get_vpc(vpc_id)
self.network_acls[network_acl_id] = network_acl
return network_acl
def get_all_network_acls(self, filters=None):
network_acls = self.network_acls.values()
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
class NetworkAcl(TaggedEC2Resource):
def __init__(self, ec2_backend, network_acl_id, vpc_id):
self.ec2_backend = ec2_backend
self.id = network_acl_id
self.vpc_id = vpc_id
self.network_acl_entries = []
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,
@ -2275,7 +2346,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):

100
moto/ec2/responses/network_acls.py
View File

@ -1,25 +1,111 @@
from __future__ import unicode_literals
from jinja2 import Template
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_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')
raise NotImplementedError(
'NetworkACLs(AmazonVPC).delete_network_acl is not yet implemented')
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')
filters = filters_from_querystring(self.querystring)
network_acls = self.ec2_backend.get_all_network_acls(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')
raise NotImplementedError(
'NetworkACLs(AmazonVPC).replace_network_acl_association is not yet implemented')
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/>
<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>
"""

4
moto/ec2/utils.py
View File

@ -72,6 +72,10 @@ 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_volume_id():
return random_id(prefix=EC2_RESOURCE_TO_PREFIX['volume'])

36
tests/test_ec2/test_network_acls.py
View File

@ -7,4 +7,38 @@ from moto import mock_ec2
@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(1)
@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(1)
entries = all_network_acls[0].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')