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

Merge pull request #2028 from acsbendi/network-acl-fixes

Browse Source 
Network ACL fixes
This commit is contained in:
Mike Grima 2019-02-01 11:46:25 -08:00 committed by GitHub
commit 8d17271ee4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 59 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
moto/ec2/models.py
View File

@ -2464,7 +2464,7 @@ class SubnetBackend(object):
default_for_az, map_public_ip_on_launch)
# AWS associates a new subnet with the default Network ACL
self.associate_default_network_acl_with_subnet(subnet_id)
self.associate_default_network_acl_with_subnet(subnet_id, vpc_id)
self.subnets[availability_zone][subnet_id] = subnet
return subnet
@ -3560,8 +3560,22 @@ class NetworkAclBackend(object):
self.get_vpc(vpc_id)
network_acl = NetworkAcl(self, network_acl_id, vpc_id, default)
self.network_acls[network_acl_id] = network_acl
if default:
self.add_default_entries(network_acl_id)
return network_acl
def add_default_entries(self, network_acl_id):
default_acl_entries = [
{'rule_number': 100, 'rule_action': 'allow', 'egress': 'true'},
{'rule_number': 32767, 'rule_action': 'deny', 'egress': 'true'},
{'rule_number': 100, 'rule_action': 'allow', 'egress': 'false'},
{'rule_number': 32767, 'rule_action': 'deny', 'egress': 'false'}
]
for entry in default_acl_entries:
self.create_network_acl_entry(network_acl_id=network_acl_id, rule_number=entry['rule_number'], protocol='-1',
rule_action=entry['rule_action'], egress=entry['egress'], cidr_block='0.0.0.0/0',
icmp_code=None, icmp_type=None, port_range_from=None, port_range_to=None)
def get_all_network_acls(self, network_acl_ids=None, filters=None):
network_acls = self.network_acls.values()
@ -3636,9 +3650,9 @@ class NetworkAclBackend(object):
new_acl.associations[new_assoc_id] = association
return association
def associate_default_network_acl_with_subnet(self, subnet_id):
def associate_default_network_acl_with_subnet(self, subnet_id, vpc_id):
association_id = random_network_acl_subnet_association_id()
acl = next(acl for acl in self.network_acls.values() if acl.default)
acl = next(acl for acl in self.network_acls.values() if acl.default and acl.vpc_id == vpc_id)
acl.associations[association_id] = NetworkAclAssociation(self, association_id,
subnet_id, acl.id)

43
tests/test_ec2/test_network_acls.py
View File

@ -1,8 +1,9 @@
from __future__ import unicode_literals
import boto
import boto3
import sure # noqa
from moto import mock_ec2_deprecated
from moto import mock_ec2_deprecated, mock_ec2
@mock_ec2_deprecated
@ -173,3 +174,43 @@ def test_network_acl_tagging():
if na.id == network_acl.id)
test_network_acl.tags.should.have.length_of(1)
test_network_acl.tags["a key"].should.equal("some value")
@mock_ec2
def test_new_subnet_in_new_vpc_associates_with_default_network_acl():
ec2 = boto3.resource('ec2', region_name='us-west-1')
new_vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
new_vpc.reload()
subnet = ec2.create_subnet(VpcId=new_vpc.id, CidrBlock='10.0.0.0/24')
subnet.reload()
new_vpcs_default_network_acl = next(iter(new_vpc.network_acls.all()), None)
new_vpcs_default_network_acl.reload()
new_vpcs_default_network_acl.vpc_id.should.equal(new_vpc.id)
new_vpcs_default_network_acl.associations.should.have.length_of(1)
new_vpcs_default_network_acl.associations[0]['SubnetId'].should.equal(subnet.id)
@mock_ec2
def test_default_network_acl_default_entries():
ec2 = boto3.resource('ec2', region_name='us-west-1')
default_network_acl = next(iter(ec2.network_acls.all()), None)
default_network_acl.is_default.should.be.ok
default_network_acl.entries.should.have.length_of(4)
unique_entries = []
for entry in default_network_acl.entries:
entry['CidrBlock'].should.equal('0.0.0.0/0')
entry['Protocol'].should.equal('-1')
entry['RuleNumber'].should.be.within([100, 32767])
entry['RuleAction'].should.be.within(['allow', 'deny'])
assert type(entry['Egress']) is bool
if entry['RuleAction'] == 'allow':
entry['RuleNumber'].should.be.equal(100)
else:
entry['RuleNumber'].should.be.equal(32767)
if entry not in unique_entries:
unique_entries.append(entry)
unique_entries.should.have.length_of(4)