diff --git a/glance/api/v2/images.py b/glance/api/v2/images.py index bcaeecc3..3aa40465 100644 --- a/glance/api/v2/images.py +++ b/glance/api/v2/images.py @@ -162,11 +162,16 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer): except exception.InvalidObject as e: raise webob.exc.HTTPBadRequest(explanation=unicode(e)) + # Ensure all specified properties are allowed + self._check_readonly(body) + self._check_reserved(body) + # Create a dict of base image properties, with user- and deployer- # defined properties contained in a 'properties' dictionary image = {'properties': body} - for key in ['id', 'name', 'visibility', 'created_at', 'updated_at', - 'tags', 'status']: + for key in ['checksum', 'created_at', 'container_format', + 'disk_format', 'id', 'min_disk', 'min_ram', 'name', 'size', + 'status', 'tags', 'updated_at', 'visibility']: try: image[key] = image['properties'].pop(key) except KeyError: @@ -175,16 +180,24 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer): if 'visibility' in image: image['is_public'] = image.pop('visibility') == 'public' - self._check_readonly(image) return {'image': image} @staticmethod def _check_readonly(image): - for key in ['created_at', 'updated_at', 'status']: + for key in ['created_at', 'updated_at', 'status', 'checksum', 'size', + 'direct_url', 'self', 'file', 'schema']: if key in image: msg = "Attribute \'%s\' is read-only." % key raise webob.exc.HTTPForbidden(explanation=unicode(msg)) + @staticmethod + def _check_reserved(image): + for key in ['owner', 'protected', 'is_public', 'location', + 'deleted', 'deleted_at']: + if key in image: + msg = "Attribute \'%s\' is reserved." % key + raise webob.exc.HTTPForbidden(explanation=unicode(msg)) + def create(self, request): return self._parse_image(request) diff --git a/glance/tests/unit/v2/test_images_resource.py b/glance/tests/unit/v2/test_images_resource.py index ad238ff1..8ceae3cd 100644 --- a/glance/tests/unit/v2/test_images_resource.py +++ b/glance/tests/unit/v2/test_images_resource.py @@ -338,81 +338,105 @@ class TestImagesDeserializer(test_utils.BaseTestCase): super(TestImagesDeserializer, self).setUp() self.deserializer = glance.api.v2.images.RequestDeserializer() - def test_create_with_id(self): + def test_create_minimal(self): request = unit_test_utils.get_fake_request() - image_id = utils.generate_uuid() - request.body = json.dumps({'id': image_id}) + request.body = json.dumps({}) output = self.deserializer.create(request) - expected = {'image': {'id': image_id, 'properties': {}}} + expected = {'image': {'properties': {}}} self.assertEqual(expected, output) - def test_create_with_name(self): + def test_create_full(self): request = unit_test_utils.get_fake_request() - request.body = json.dumps({'name': 'image-1'}) + request.body = json.dumps({ + 'id': UUID3, + 'name': 'image-1', + 'visibility': 'public', + 'tags': ['one', 'two'], + 'container_format': 'ami', + 'disk_format': 'ami', + 'min_ram': 128, + 'min_disk': 10, + 'foo': 'bar', + }) output = self.deserializer.create(request) - expected = {'image': {'name': 'image-1', 'properties': {}}} - self.assertEqual(expected, output) - - def test_create_public(self): - request = unit_test_utils.get_fake_request() - request.body = json.dumps({'visibility': 'public'}) - output = self.deserializer.create(request) - expected = {'image': {'is_public': True, 'properties': {}}} - self.assertEqual(expected, output) - - def test_create_private(self): - request = unit_test_utils.get_fake_request() - request.body = json.dumps({'visibility': 'private'}) - output = self.deserializer.create(request) - expected = {'image': {'is_public': False, 'properties': {}}} + expected = {'image': { + 'id': UUID3, + 'name': 'image-1', + 'is_public': True, + 'tags': ['one', 'two'], + 'container_format': 'ami', + 'disk_format': 'ami', + 'min_ram': 128, + 'min_disk': 10, + 'properties': {'foo': 'bar'}, + }} self.assertEqual(expected, output) def test_create_readonly_attributes_forbidden(self): - for key in ['created_at', 'updated_at']: + bodies = [ + {'created_at': ISOTIME}, + {'updated_at': ISOTIME}, + {'status': 'saving'}, + {'direct_url': 'http://example.com'}, + {'size': 10}, + {'checksum': 'asdf'}, + {'self': 'http://example.com'}, + {'file': 'http://example.com'}, + {'schema': 'http://example.com'}, + ] + + for body in bodies: request = unit_test_utils.get_fake_request() - request.body = json.dumps({key: ISOTIME}) + request.body = json.dumps(body) self.assertRaises(webob.exc.HTTPForbidden, - self.deserializer.update, request) - - def test_create_status_attribute_forbidden(self): - request = unit_test_utils.get_fake_request() - request.body = json.dumps({'status': 'saving'}) - self.assertRaises(webob.exc.HTTPForbidden, - self.deserializer.update, request) - - def test_create_with_tags(self): - request = unit_test_utils.get_fake_request() - request.body = json.dumps({'tags': ['one', 'two']}) - output = self.deserializer.create(request) - expected = {'image': {'tags': ['one', 'two'], 'properties': {}}} - self.assertEqual(expected, output) + self.deserializer.create, request) def test_update(self): request = unit_test_utils.get_fake_request() - request.body = json.dumps({'name': 'image-1', 'visibility': 'public'}) + request.body = json.dumps({ + 'id': UUID3, + 'name': 'image-1', + 'visibility': 'public', + 'tags': ['one', 'two'], + 'container_format': 'ami', + 'disk_format': 'ami', + 'min_ram': 128, + 'min_disk': 10, + 'foo': 'bar', + }) output = self.deserializer.update(request) - expected = { - 'image': { - 'name': 'image-1', - 'is_public': True, - 'properties': {}, - }, - } + expected = {'image': { + 'id': UUID3, + 'name': 'image-1', + 'is_public': True, + 'tags': ['one', 'two'], + 'container_format': 'ami', + 'disk_format': 'ami', + 'min_ram': 128, + 'min_disk': 10, + 'properties': {'foo': 'bar'}, + }} self.assertEqual(expected, output) def test_update_readonly_attributes_forbidden(self): - for key in ['created_at', 'updated_at']: + bodies = [ + {'created_at': ISOTIME}, + {'updated_at': ISOTIME}, + {'status': 'saving'}, + {'direct_url': 'http://example.com'}, + {'size': 10}, + {'checksum': 'asdf'}, + {'self': 'http://example.com'}, + {'file': 'http://example.com'}, + {'schema': 'http://example.com'}, + ] + + for body in bodies: request = unit_test_utils.get_fake_request() - request.body = json.dumps({key: ISOTIME}) + request.body = json.dumps(body) self.assertRaises(webob.exc.HTTPForbidden, self.deserializer.update, request) - def test_update_status_attribute_forbidden(self): - request = unit_test_utils.get_fake_request() - request.body = json.dumps({'status': 'saving'}) - self.assertRaises(webob.exc.HTTPForbidden, - self.deserializer.update, request) - def test_index(self): marker = utils.generate_uuid() path = '/images?limit=1&marker=%s' % marker