Prevent wrongly privilege escalation of a context

Current codes in create_volume() may exists a scenario that
a context which is not elevated but after it fails in volume
creation, it becomes elevated. This patch saves original
context, so that if it fails in volume creation simply use the
original context for further scheduling.

Fix bug:1187076

Change-Id: I2822b1612ec741209b278fc65f18d0f8f3243e0a
This commit is contained in:
XueChendi 2013-07-03 22:20:32 +08:00
parent faffe83d88
commit ca0e7295f4
3 changed files with 38 additions and 1 deletions

View File

@ -135,6 +135,9 @@ class RequestContext(object):
return context
def deepcopy(self):
return copy.deepcopy(self)
# NOTE(sirp): the openstack/common version of RequestContext uses
# tenant/user whereas the Cinder version uses project_id/user_id. We need
# this shim in order to use context-aware code from openstack/common, like

View File

@ -1260,6 +1260,38 @@ class VolumeTestCase(test.TestCase):
# clean up
self.volume.delete_volume(self.context, volume['id'])
def test_create_volume_from_unelevated_context(self):
"""Test context does't change after volume creation failure."""
def fake_create_volume(context, volume_ref, snapshot_ref,
sourcevol_ref, image_service, image_id,
image_location):
raise exception.CinderException('fake exception')
def fake_reschedule_or_reraise(context, volume_id, exc_info,
snapshot_id, image_id, request_spec,
filter_properties, allow_reschedule):
self.assertFalse(context.is_admin)
self.assertFalse('admin' in context.roles)
#compare context passed in with the context we saved
self.assertDictMatch(self.saved_ctxt.__dict__,
context.__dict__)
#create context for testing
ctxt = self.context.deepcopy()
if 'admin' in ctxt.roles:
ctxt.roles.remove('admin')
ctxt.is_admin = False
#create one copy of context for future comparison
self.saved_ctxt = ctxt.deepcopy()
self.stubs.Set(self.volume, '_reschedule_or_reraise',
fake_reschedule_or_reraise)
self.stubs.Set(self.volume, '_create_volume',
fake_create_volume)
volume_src = self._create_volume()
self.volume.create_volume(ctxt, volume_src['id'])
def test_create_volume_from_sourcevol(self):
"""Test volume can be created from a source volume."""
def fake_create_cloned_volume(volume, src_vref):

View File

@ -209,6 +209,7 @@ class VolumeManager(manager.SchedulerDependentManager):
filter_properties=None, allow_reschedule=True,
snapshot_id=None, image_id=None, source_volid=None):
"""Creates and exports the volume."""
context_before_elevated = context.deepcopy()
context = context.elevated()
if filter_properties is None:
filter_properties = {}
@ -277,7 +278,8 @@ class VolumeManager(manager.SchedulerDependentManager):
{'status': sourcevol_ref['status']})
exc_info = sys.exc_info()
# try to re-schedule volume:
self._reschedule_or_reraise(context, volume_id, exc_info,
self._reschedule_or_reraise(context_before_elevated,
volume_id, exc_info,
snapshot_id, image_id,
request_spec, filter_properties,
allow_reschedule)