openstack
/
nova
Code Issues Proposed changes

Fix security group race condition while creating rule 

Previously, it was possible for someone to create a security group rule
where the rule references another security group. To do this nova would
first create the security group rule and then look up the referenced group
(group_id's) name in order to return that via the api. During this time
it's possible for someone to delete this security group rule and the
referenced group before the call returned resulting in a 404 error being
raised. This patch addresses this issue by looking up the group name
first and then creating the security group rule in order to avoid this
from occuring.

Note: this patch also adds a test to cover the case where an invalid group
id was passed though this does not really add any real additional coverage
to the change largely because the code is currently using global stubs. That
said, the previous coverage should hopefully be sufficient

Change-Id: If58ffa5629ba5166f260379ac47922974de31be0
Related-bug: 1262566
This commit is contained in:
Aaron Rosen 2014-05-08 12:07:27 -07:00
parent eb84cfb4ef
commit 2620a7c74e
2 changed files with 27 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
nova/api/openstack/compute/contrib/security_groups.py
View File

@ -208,7 +208,12 @@ class SecurityGroupControllerBase(object):
self.compute_api = compute.API(
security_group_api=self.security_group_api)
def _format_security_group_rule(self, context, rule):
def _format_security_group_rule(self, context, rule, group_rule_data=None):
"""Return a secuity group rule in desired API response format.
If group_rule_data is passed in that is used rather than querying
for it.
"""
sg_rule = {}
sg_rule['id'] = rule['id']
sg_rule['parent_group_id'] = rule['parent_group_id']
@ -235,6 +240,8 @@ class SecurityGroupControllerBase(object):
return
sg_rule['group'] = {'name': source_group.get('name'),
'tenant_id': source_group.get('project_id')}
elif group_rule_data:
sg_rule['group'] = group_rule_data
else:
sg_rule['ip_range'] = {'cidr': rule['cidr']}
return sg_rule
@ -394,23 +401,22 @@ class SecurityGroupRulesController(SecurityGroupControllerBase):
msg = _("Bad prefix for network in cidr %s") % new_rule['cidr']
raise exc.HTTPBadRequest(explanation=msg)
group_rule_data = None
with translate_exceptions():
if sg_rule.get('group_id'):
source_group = self.security_group_api.get(
context, id=sg_rule['group_id'])
group_rule_data = {'name': source_group.get('name'),
'tenant_id': source_group.get('project_id')}
security_group_rule = (
self.security_group_api.create_security_group_rule(
context, security_group, new_rule))
formatted_rule = self._format_security_group_rule(context,
security_group_rule)
if formatted_rule:
return {"security_group_rule": formatted_rule}
# TODO(arosen): if we first look up the security group information for
# the group_id before creating the rule we can avoid the case that
# the remote group (group_id) has been deleted when we go to look
# up it's name.
with translate_exceptions():
raise exception.SecurityGroupNotFound(
security_group_id=sg_rule['group_id'])
security_group_rule,
group_rule_data)
return {"security_group_rule": formatted_rule}
def _rule_args_to_dict(self, context, to_port=None, from_port=None,
ip_protocol=None, cidr=None, group_id=None):

10
nova/tests/api/openstack/compute/contrib/test_security_groups.py
View File

@ -982,13 +982,21 @@ class TestSecurityGroupRules(test.TestCase):
req, {'security_group_rule': rule})
def test_create_with_non_existing_parent_group_id(self):
rule = security_group_rule_template(group_id='invalid',
rule = security_group_rule_template(group_id=None,
parent_group_id=self.invalid_id)
req = fakes.HTTPRequest.blank('/v2/fake/os-security-group-rules')
self.assertRaises(webob.exc.HTTPNotFound, self.controller.create,
req, {'security_group_rule': rule})
def test_create_with_non_existing_group_id(self):
rule = security_group_rule_template(group_id='invalid',
parent_group_id=self.sg2['id'])
req = fakes.HTTPRequest.blank('/v2/fake/os-security-group-rules')
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
req, {'security_group_rule': rule})
def test_create_with_invalid_protocol(self):
rule = security_group_rule_template(ip_protocol='invalid-protocol',
cidr='10.2.2.0/24',