openstack/glance
Code Issues Proposed changes

Merge "Prevent user to remove last location of the image"

Browse Source
This commit is contained in:
Jenkins 2016-02-04 16:01:01 +00:00 committed by Gerrit Code Review
commit 8781f64db2
4 changed files with 44 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
glance/api/v2/images.py
View File

@ -173,7 +173,10 @@ class ImagesController(object):
path = change['path']
path_root = path[0]
value = change['value']
if path_root == 'locations':
if path_root == 'locations' and value == []:
msg = _("Cannot set locations to empty list.")
raise webob.exc.HTTPForbidden(msg)
elif path_root == 'locations' and value != []:
self._do_replace_locations(image, value)
elif path_root == 'owner' and req.context.is_admin == False:
msg = _("Owner can't be updated by non admin.")
@ -209,7 +212,10 @@ class ImagesController(object):
path = change['path']
path_root = path[0]
if path_root == 'locations':
self._do_remove_locations(image, path[1])
try:
self._do_remove_locations(image, path[1])
except exception.Forbidden as e:
raise webob.exc.HTTPForbidden(e.msg)
else:
if hasattr(image, path_root):
msg = _("Property %s may not be removed.")
@ -298,6 +304,11 @@ class ImagesController(object):
explanation=encodeutils.exception_to_unicode(e))
def _do_remove_locations(self, image, path_pos):
if len(image.locations) == 1:
LOG.debug("User forbidden to remove last location of image %s",
image.image_id)
msg = _("Cannot remove last location in the image.")
raise exception.Forbidden(msg)
pos = self._get_locations_op_pos(path_pos,
len(image.locations), False)
if pos is None:
@ -307,11 +318,11 @@ class ImagesController(object):
# NOTE(zhiyan): this actually deletes the location
# from the backend store.
image.locations.pop(pos)
# TODO(jokke): Fix this, we should catch what store throws and
# provide definitely something else than IternalServerError to user.
except Exception as e:
raise webob.exc.HTTPInternalServerError(
explanation=encodeutils.exception_to_unicode(e))
if len(image.locations) == 0 and image.status == 'active':
image.status = 'queued'
class RequestDeserializer(wsgi.JSONRequestDeserializer):

14
glance/tests/functional/v2/test_images.py
View File

@ -561,20 +561,6 @@ class TestImages(functional.FunctionalTest):
response = requests.patch(path, headers=headers, data=data)
self.assertEqual(200, response.status_code, response.text)
# Remove all locations of the image then the image size shouldn't be
# able to access
path = self._url('/v2/images/%s' % image2_id)
media_type = 'application/openstack-images-v2.1-json-patch'
headers = self._headers({'content-type': media_type})
doc = [{'op': 'replace', 'path': '/locations', 'value': []}]
data = jsonutils.dumps(doc)
response = requests.patch(path, headers=headers, data=data)
self.assertEqual(200, response.status_code, response.text)
image = jsonutils.loads(response.text)
self.assertIsNone(image['size'])
self.assertIsNone(image['virtual_size'])
self.assertEqual('queued', image['status'])
# Deletion should work. Deleting image-1
path = self._url('/v2/images/%s' % image_id)
response = requests.delete(path, headers=self._headers())

122
glance/tests/unit/v2/test_images_resource.py
View File

@ -1441,26 +1441,6 @@ class TestImagesController(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
another_request, created_image.image_id, changes)
def test_update_replace_locations(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.assertIsNone(output.size)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertEqual(new_location, output.locations[0])
self.assertEqual('active', output.status)
def test_update_replace_locations_non_empty(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
@ -1472,35 +1452,9 @@ class TestImagesController(base.IsolatedUnitTest):
def test_update_replace_locations_invalid(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'],
'value': [{'url': 'unknow://foo', 'metadata': {}}]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
request, UUID1, changes)
def test_update_replace_locations_status_exception(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID2, changes)
self.assertEqual(UUID2, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.db.image_update(None, UUID2, {'disk_format': None})
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID2, changes)
def test_update_add_property(self):
request = unit_test_utils.get_fake_request()
@ -1624,24 +1578,6 @@ class TestImagesController(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_add_locations_status_exception(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID2, changes)
self.assertEqual(UUID2, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.db.image_update(None, UUID2, {'disk_format': None})
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID2, changes)
def test_update_add_duplicate_locations(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
@ -1655,23 +1591,6 @@ class TestImagesController(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_replace_duplicate_locations(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location, new_location]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_add_too_many_locations(self):
self.config(image_location_quota=1)
request = unit_test_utils.get_fake_request()
@ -1772,9 +1691,12 @@ class TestImagesController(base.IsolatedUnitTest):
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_2' % BASE_URI,
'metadata': {}}},
]
self.controller.update(request, UUID1, changes)
self.config(image_location_quota=1)
self.config(image_location_quota=2)
# We must remove two properties to avoid being
# over the limit of 1 property
@ -1787,8 +1709,8 @@ class TestImagesController(base.IsolatedUnitTest):
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertIn('fake_location_3', output.locations[0]['url'])
self.assertEqual(2, len(output.locations))
self.assertIn('fake_location_3', output.locations[1]['url'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_remove_base_property(self):
@ -1829,24 +1751,23 @@ class TestImagesController(base.IsolatedUnitTest):
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'remove', 'path': ['locations', '0']}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.assertIsNone(output.size)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
self.controller.update(request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', '0']}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertEqual(new_location, output.locations[0])
self.assertEqual('active', output.status)
def test_update_remove_location_invalid_pos(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location' % BASE_URI,
'metadata': {}}}]
self.controller.update(request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', None]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
@ -1868,6 +1789,11 @@ class TestImagesController(base.IsolatedUnitTest):
fake_delete_image_location_from_backend)
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location' % BASE_URI,
'metadata': {}}}]
self.controller.update(request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', '0']}]
self.assertRaises(webob.exc.HTTPInternalServerError,
self.controller.update, request, UUID1, changes)
@ -2161,16 +2087,6 @@ class TestImagesControllerPolicies(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
request, UUID1, changes)
self.stubs.Set(self.store_utils, 'delete_image_location_from_backend',
fake_delete_image_location_from_backend)
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
self.controller.update(request, UUID1, changes)
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
request, UUID1, changes)
def test_update_delete_image_location_unauthorized(self):
rules = {"delete_image_location": False}
self.policy.set_rules(rules)

10
releasenotes/notes/Prevent-removing-last-image-location-d5ee3e00efe14f34.yaml Normal file
View File

@ -0,0 +1,10 @@
---
security:
- Fixing bug 1525915; image might be transitioning
from active to queued by regular user by removing
last location of image (or replacing locations
with empty list). This allows user to re-upload
data to the image breaking Glance's promise of
image data immutability. From now on, last
location cannot be removed and locations cannot
be replaced with empty list.