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

Merge pull request #512 from kevgliss/sgEgressPermissions

Browse Source 
Adds support for authorizing and revoking egress permissions
This commit is contained in:
Steve Pulec 2016-01-17 15:46:53 -05:00
commit ea097791cb
3 changed files with 108 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
moto/ec2/models.py
View File

@ -1240,7 +1240,6 @@ class SecurityGroupBackend(object):
def get_security_group_from_id(self, group_id):
# 2 levels of chaining necessary since it's a complex structure
all_groups = itertools.chain.from_iterable([x.values() for x in self.groups.values()])
for group in all_groups:
if group.id == group_id:
return group
@ -1267,7 +1266,6 @@ class SecurityGroupBackend(object):
source_group_ids=None,
vpc_id=None):
group = self.get_security_group_by_name_or_id(group_name_or_id, vpc_id)
if ip_ranges and not isinstance(ip_ranges, list):
ip_ranges = [ip_ranges]
if ip_ranges:
@ -1323,6 +1321,55 @@ class SecurityGroupBackend(object):
raise InvalidPermissionNotFoundError()
def authorize_security_group_egress(self,
group_name_or_id,
ip_protocol,
from_port,
to_port,
ip_ranges,
src_group_id=None,
cidr_ip=None,
vpc_id=None):
group = self.get_security_group_by_name_or_id(group_name_or_id, vpc_id)
if ip_ranges:
for cidr in ip_ranges:
if not is_valid_cidr(cidr):
raise InvalidCIDRSubnetError(cidr=cidr)
# for VPCs
source_groups = []
source_group = self.get_security_group_from_id(src_group_id)
if source_group:
source_groups.append(source_group)
security_rule = SecurityRule(ip_protocol, from_port, to_port, ip_ranges, source_groups)
group.egress_rules.append(security_rule)
def revoke_security_group_egress(self,
group_name_or_id,
ip_protocol,
from_port,
to_port,
ip_ranges,
source_group_names=None,
source_group_ids=None,
vpc_id=None):
group = self.get_security_group_by_name_or_id(group_name_or_id, vpc_id)
source_groups = []
for source_group_name in source_group_names:
source_group = self.get_security_group_from_name(source_group_name, vpc_id)
if source_group:
source_groups.append(source_group)
security_rule = SecurityRule(ip_protocol, from_port, to_port, ip_ranges, source_groups)
if security_rule in group.egress_rules:
group.egress_rules.remove(security_rule)
return security_rule
raise InvalidPermissionNotFoundError()
class SecurityGroupIngress(object):

45
moto/ec2/responses/security_groups.py
View File

@ -31,7 +31,8 @@ def process_rules_from_querystring(querystring):
class SecurityGroups(BaseResponse):
def authorize_security_group_egress(self):
raise NotImplementedError('SecurityGroups.authorize_security_group_egress is not yet implemented')
self.ec2_backend.authorize_security_group_egress(*process_rules_from_querystring(self.querystring))
return AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE
def authorize_security_group_ingress(self):
self.ec2_backend.authorize_security_group_ingress(*process_rules_from_querystring(self.querystring))
@ -73,7 +74,10 @@ class SecurityGroups(BaseResponse):
return template.render(groups=groups)
def revoke_security_group_egress(self):
raise NotImplementedError('SecurityGroups.revoke_security_group_egress is not yet implemented')
success = self.ec2_backend.revoke_security_group_egress(*process_rules_from_querystring(self.querystring))
if not success:
return "Could not find a matching egress rule", dict(status=404)
return REVOKE_SECURITY_GROUP_EGRESS_RESPONSE
def revoke_security_group_ingress(self):
self.ec2_backend.revoke_security_group_ingress(*process_rules_from_querystring(self.querystring))
@ -128,7 +132,31 @@ DESCRIBE_SECURITY_GROUPS_RESPONSE = """<DescribeSecurityGroupsResponse xmlns="ht
</item>
{% endfor %}
</ipPermissions>
<ipPermissionsEgress/>
<ipPermissionsEgress>
{% for rule in group.egress_rules %}
<item>
<ipProtocol>{{ rule.ip_protocol }}</ipProtocol>
<fromPort>{{ rule.from_port }}</fromPort>
<toPort>{{ rule.to_port }}</toPort>
<groups>
{% for source_group in rule.source_groups %}
<item>
<userId>111122223333</userId>
<groupId>{{ source_group.id }}</groupId>
<groupName>{{ source_group.name }}</groupName>
</item>
{% endfor %}
</groups>
<ipRanges>
{% for ip_range in rule.ip_ranges %}
<item>
<cidrIp>{{ ip_range }}</cidrIp>
</item>
{% endfor %}
</ipRanges>
</item>
{% endfor %}
</ipPermissionsEgress>
<tagSet>
{% for tag in group.get_tags() %}
<item>
@ -153,3 +181,14 @@ REVOKE_SECURITY_GROUP_INGRESS_REPONSE = """<RevokeSecurityGroupIngressResponse x
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</RevokeSecurityGroupIngressResponse>"""
AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE = """
<AuthorizeSecurityGroupEgressResponse xmlns="http://ec2.amazonaws.com/doc/2015-10-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</AuthorizeSecurityGroupEgressResponse>"""
REVOKE_SECURITY_GROUP_EGRESS_RESPONSE = """<RevokeSecurityGroupEgressResponse xmlns="http://ec2.amazonaws.com/doc/2012-12-01/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</RevokeSecurityGroupEgressResponse>"""

17
tests/test_ec2/test_security_groups.py
View File

@ -152,6 +152,23 @@ def test_authorize_ip_range_and_revoke():
security_group = conn.get_all_security_groups()[0]
security_group.rules.should.have.length_of(0)
# Test for egress as well
egress_security_group = conn.create_security_group('testegress', 'testegress', vpc_id='vpc-3432589')
success = conn.authorize_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
assert success.should.be.true
egress_security_group = conn.get_all_security_groups(groupnames='testegress')[0]
int(egress_security_group.rules_egress[0].to_port).should.equal(2222)
egress_security_group.rules_egress[0].grants[0].cidr_ip.should.equal("123.123.123.123/32")
# Wrong Cidr should throw error
egress_security_group.revoke.when.called_with(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32").should.throw(EC2ResponseError)
# Actually revoke
conn.revoke_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
egress_security_group = conn.get_all_security_groups()[0]
egress_security_group.rules_egress.should.have.length_of(0)
@mock_ec2
def test_authorize_other_group_and_revoke():