diff options
author | Jenkins <jenkins@review.openstack.org> | 2014-12-18 12:40:42 +0000 |
---|---|---|
committer | Gerrit Code Review <review@openstack.org> | 2014-12-18 12:40:42 +0000 |
commit | 42564be7ea133313a52639c27172d035390da1f2 (patch) | |
tree | 6610e9c69de264f81d550bb6de5728e5ceae178a | |
parent | 749187a75af02989f1a3bf1f6a44004776a7e303 (diff) | |
parent | d9a928eac360add67477e29f516af868adfe0d5e (diff) | |
download | glance-42564be7ea133313a52639c27172d035390da1f2.tar.gz |
Merge "To prevent client use v2 patch api to handle file and swift location" into stable/juno
-rw-r--r-- | glance/api/v1/images.py | 31 | ||||
-rw-r--r-- | glance/common/store_utils.py | 22 | ||||
-rw-r--r-- | glance/location.py | 30 | ||||
-rw-r--r-- | glance/tests/functional/v1/test_copy_to_file.py | 4 | ||||
-rw-r--r-- | glance/tests/functional/v2/test_images.py | 158 | ||||
-rw-r--r-- | glance/tests/unit/test_store_image.py | 3 | ||||
-rw-r--r-- | glance/tests/unit/test_store_location.py | 33 | ||||
-rw-r--r-- | glance/tests/unit/utils.py | 9 | ||||
-rw-r--r-- | glance/tests/unit/v1/test_api.py | 62 |
9 files changed, 221 insertions, 131 deletions
diff --git a/glance/api/v1/images.py b/glance/api/v1/images.py index c85b30131..746f8cdb4 100644 --- a/glance/api/v1/images.py +++ b/glance/api/v1/images.py @@ -23,7 +23,6 @@ import eventlet import glance_store as store import glance_store.location from oslo.config import cfg -import six.moves.urllib.parse as urlparse from webob.exc import HTTPBadRequest from webob.exc import HTTPConflict from webob.exc import HTTPForbidden @@ -40,6 +39,7 @@ from glance.api.v1 import filters from glance.api.v1 import upload_utils from glance.common import exception from glance.common import property_utils +from glance.common import store_utils from glance.common import utils from glance.common import wsgi from glance.i18n import _LE @@ -415,26 +415,19 @@ class Controller(controller.BaseController): @staticmethod def _validate_source(source, req): """ - External sources (as specified via the location or copy-from headers) - are supported only over non-local store types, i.e. S3, Swift, HTTP. - Note the absence of 'file://' for security reasons, see LP bug #942118. - 'swift+config://' is also absent for security reasons, see LP bug - #1334196. - If the above constraint is violated, we reject with 400 "Bad Request". + To validate if external sources (as specified via the location + or copy-from headers) are supported. Otherwise we reject + with 400 "Bad Request". """ if source: - pieces = urlparse.urlparse(source) - schemes = [scheme for scheme in store.get_known_schemes() - if scheme != 'file' and scheme != 'swift+config'] - for scheme in schemes: - if pieces.scheme == scheme: - return source - msg = ("External sourcing not supported for " - "store '%s'" % pieces.scheme) - LOG.debug(msg) - raise HTTPBadRequest(explanation=msg, - request=req, - content_type="text/plain") + if store_utils.validate_external_location(source): + return source + else: + msg = _("External source are not supported: '%s'") % source + LOG.debug(msg) + raise HTTPBadRequest(explanation=msg, + request=req, + content_type="text/plain") @staticmethod def _copy_from(req): diff --git a/glance/common/store_utils.py b/glance/common/store_utils.py index 8f04d39a9..b7537ce01 100644 --- a/glance/common/store_utils.py +++ b/glance/common/store_utils.py @@ -16,6 +16,7 @@ import sys import glance_store as store_api from oslo.config import cfg +import six.moves.urllib.parse as urlparse from glance.common import utils import glance.db as db_api @@ -119,3 +120,24 @@ def delete_image_location_from_backend(context, image_id, location): # such as uploading process failure then we can't use # location status mechanism to support image pending delete. safe_delete_from_backend(context, image_id, location) + + +def validate_external_location(uri): + """ + Validate if URI of external location are supported. + + Only over non-local store types are OK, i.e. S3, Swift, + HTTP. Note the absence of 'file://' for security reasons, + see LP bug #942118, 1400966, 'swift+config://' is also + absent for security reasons, see LP bug #1334196. + + :param uri: The URI of external image location. + :return: Whether given URI of external image location are OK. + """ + + # TODO(zhiyan): This function could be moved to glance_store. + + pieces = urlparse.urlparse(uri) + valid_schemes = [scheme for scheme in store_api.get_known_schemes() + if scheme != 'file' and scheme != 'swift+config'] + return pieces.scheme in valid_schemes diff --git a/glance/location.py b/glance/location.py index fcdba0a36..f83fa7ab7 100644 --- a/glance/location.py +++ b/glance/location.py @@ -66,18 +66,20 @@ class ImageRepoProxy(glance.domain.proxy.Repo): return result -def _check_location_uri(context, store_api, uri): +def _check_location_uri(context, store_api, store_utils, uri): """Check if an image location is valid. :param context: Glance request context :param store_api: store API module + :param store_utils: store utils module :param uri: location's uri string """ + is_ok = True try: - size = store_api.get_size_from_backend(uri, context=context) # NOTE(zhiyan): Some stores return zero when it catch exception - is_ok = size > 0 + is_ok = (store_utils.validate_external_location(uri) and + store_api.get_size_from_backend(uri, context=context) > 0) except (store.UnknownScheme, store.NotFound): is_ok = False if not is_ok: @@ -85,8 +87,8 @@ def _check_location_uri(context, store_api, uri): raise exception.BadStoreUri(message=reason) -def _check_image_location(context, store_api, location): - _check_location_uri(context, store_api, location['url']) +def _check_image_location(context, store_api, store_utils, location): + _check_location_uri(context, store_api, store_utils, location['url']) store_api.check_location_metadata(location['metadata']) @@ -122,6 +124,7 @@ class ImageFactoryProxy(glance.domain.proxy.ImageFactory): def __init__(self, factory, context, store_api, store_utils): self.context = context self.store_api = store_api + self.store_utils = store_utils proxy_kwargs = {'context': context, 'store_api': store_api, 'store_utils': store_utils} super(ImageFactoryProxy, self).__init__(factory, @@ -131,7 +134,10 @@ class ImageFactoryProxy(glance.domain.proxy.ImageFactory): def new_image(self, **kwargs): locations = kwargs.get('locations', []) for loc in locations: - _check_image_location(self.context, self.store_api, loc) + _check_image_location(self.context, + self.store_api, + self.store_utils, + loc) loc['status'] = 'active' if _count_duplicated_locations(locations, loc) > 1: raise exception.DuplicateLocation(location=loc['url']) @@ -169,7 +175,9 @@ class StoreLocations(collections.MutableSequence): def insert(self, i, location): _check_image_location(self.image_proxy.context, - self.image_proxy.store_api, location) + self.image_proxy.store_api, + self.image_proxy.store_utils, + location) location['status'] = 'active' if _count_duplicated_locations(self.value, location) > 0: raise exception.DuplicateLocation(location=location['url']) @@ -214,7 +222,9 @@ class StoreLocations(collections.MutableSequence): def __setitem__(self, i, location): _check_image_location(self.image_proxy.context, - self.image_proxy.store_api, location) + self.image_proxy.store_api, + self.image_proxy.store_utils, + location) location['status'] = 'active' self.value.__setitem__(i, location) _set_image_size(self.image_proxy.context, @@ -303,7 +313,9 @@ def _locations_proxy(target, attr): '%s') % ori_value) # NOTE(zhiyan): Check locations are all valid. for location in value: - _check_image_location(self.context, self.store_api, + _check_image_location(self.context, + self.store_api, + self.store_utils, location) location['status'] = 'active' if _count_duplicated_locations(value, location) > 1: diff --git a/glance/tests/functional/v1/test_copy_to_file.py b/glance/tests/functional/v1/test_copy_to_file.py index 15bb7081c..b64eac65c 100644 --- a/glance/tests/functional/v1/test_copy_to_file.py +++ b/glance/tests/functional/v1/test_copy_to_file.py @@ -250,7 +250,7 @@ class TestCopyToFile(functional.FunctionalTest): response, content = http.request(path, 'POST', headers=headers) self.assertEqual(response.status, 400, content) - expected = 'External sourcing not supported for store \'file\'' + expected = 'External source are not supported: \'%s\'' % copy_from msg = 'expected "%s" in "%s"' % (expected, content) self.assertTrue(expected in content, msg) @@ -276,7 +276,7 @@ class TestCopyToFile(functional.FunctionalTest): response, content = http.request(path, 'POST', headers=headers) self.assertEqual(response.status, 400, content) - expected = 'External sourcing not supported for store \'swift+config\'' + expected = 'External source are not supported: \'swift+config://xxx\'' msg = 'expected "%s" in "%s"' % (expected, content) self.assertTrue(expected in content, msg) diff --git a/glance/tests/functional/v2/test_images.py b/glance/tests/functional/v2/test_images.py index 14fe3c7ff..4c32375fd 100644 --- a/glance/tests/functional/v2/test_images.py +++ b/glance/tests/functional/v2/test_images.py @@ -16,7 +16,6 @@ import BaseHTTPServer import os import signal -import tempfile import uuid import requests @@ -47,7 +46,7 @@ def get_handler_class(fixture): self.end_headers() return - def log_message(*args, **kwargs): + def log_message(self, *args, **kwargs): # Override this method to prevent debug output from going # to stderr during testing return @@ -75,6 +74,18 @@ class TestImages(functional.FunctionalTest): self.cleanup() self.api_server.deployment_flavor = 'noauth' self.api_server.data_api = 'glance.db.sqlalchemy.api' + for i in range(3): + ret = http_server("foo_image_id%d" % i, "foo_image%d" % i) + setattr(self, 'http_server%d_pid' % i, ret[0]) + setattr(self, 'http_port%d' % i, ret[1]) + + def tearDown(self): + for i in range(3): + pid = getattr(self, 'http_server%d_pid' % i, None) + if pid: + os.kill(pid, signal.SIGKILL) + + super(TestImages, self).tearDown() def _url(self, path): return 'http://127.0.0.1:%d%s' % (self.api_port, path) @@ -329,21 +340,15 @@ class TestImages(functional.FunctionalTest): self.assertEqual(413, response.status_code, response.text) # Adding 3 image locations should fail since configured limit is 2 - for i in range(3): - file_path = os.path.join(self.test_dir, 'fake_image_%i' % i) - with open(file_path, 'w') as fap: - fap.write('glance') - path = self._url('/v2/images/%s' % image_id) media_type = 'application/openstack-images-v2.1-json-patch' headers = self._headers({'content-type': media_type}) changes = [] for i in range(3): + url = ('http://127.0.0.1:%s/foo_image' % + getattr(self, 'http_port%d' % i)) changes.append({'op': 'add', 'path': '/locations/-', - 'value': {'url': 'file://{0}'.format( - os.path.join(self.test_dir, - 'fake_image_%i' % i)), - 'metadata': {}}, + 'value': {'url': url, 'metadata': {}}, }) data = jsonutils.dumps(changes) @@ -2176,17 +2181,14 @@ class TestImages(functional.FunctionalTest): self.assertNotIn('size', image) self.assertNotIn('virtual_size', image) - file_path = os.path.join(self.test_dir, 'fake_image') - with open(file_path, 'w') as fap: - fap.write('glance') - # Update locations for the queued image path = self._url('/v2/images/%s' % image_id) media_type = 'application/openstack-images-v2.1-json-patch' headers = self._headers({'content-type': media_type}) + url = 'http://127.0.0.1:%s/foo_image' % self.http_port0 data = jsonutils.dumps([{'op': 'replace', 'path': '/locations', - 'value': [{'url': 'file://' + file_path, - 'metadata': {}}]}]) + 'value': [{'url': url, 'metadata': {}}] + }]) response = requests.patch(path, headers=headers, data=data) self.assertEqual(200, response.status_code, response.text) @@ -2195,7 +2197,42 @@ class TestImages(functional.FunctionalTest): response = requests.get(path, headers=headers) self.assertEqual(200, response.status_code) image = jsonutils.loads(response.text) - self.assertEqual(image['size'], 6) + self.assertEqual(image['size'], 10) + + def test_update_locations_with_restricted_sources(self): + self.start_servers(**self.__dict__.copy()) + # Create an image + path = self._url('/v2/images') + headers = self._headers({'content-type': 'application/json'}) + data = jsonutils.dumps({'name': 'image-1', 'disk_format': 'aki', + 'container_format': 'aki'}) + response = requests.post(path, headers=headers, data=data) + self.assertEqual(201, response.status_code) + + # Returned image entity should have a generated id and status + image = jsonutils.loads(response.text) + image_id = image['id'] + self.assertEqual('queued', image['status']) + self.assertNotIn('size', image) + self.assertNotIn('virtual_size', image) + + # Update locations for the queued image + path = self._url('/v2/images/%s' % image_id) + media_type = 'application/openstack-images-v2.1-json-patch' + headers = self._headers({'content-type': media_type}) + data = jsonutils.dumps([{'op': 'replace', 'path': '/locations', + 'value': [{'url': 'file:///foo_image', + 'metadata': {}}] + }]) + response = requests.patch(path, headers=headers, data=data) + self.assertEqual(400, response.status_code, response.text) + + data = jsonutils.dumps([{'op': 'replace', 'path': '/locations', + 'value': [{'url': 'swift+config:///foo_image', + 'metadata': {}}] + }]) + response = requests.patch(path, headers=headers, data=data) + self.assertEqual(400, response.status_code, response.text) class TestImagesWithRegistry(TestImages): @@ -2421,16 +2458,16 @@ class TestImageLocationSelectionStrategy(functional.FunctionalTest): super(TestImageLocationSelectionStrategy, self).setUp() self.cleanup() self.api_server.deployment_flavor = 'noauth' - self.foo_image_file = tempfile.NamedTemporaryFile() - self.foo_image_file.write("foo image file") - self.foo_image_file.flush() - self.addCleanup(self.foo_image_file.close) - ret = http_server("foo_image_id", "foo_image") - self.http_server_pid, self.http_port = ret + for i in range(3): + ret = http_server("foo_image_id%d" % i, "foo_image%d" % i) + setattr(self, 'http_server%d_pid' % i, ret[0]) + setattr(self, 'http_port%d' % i, ret[1]) def tearDown(self): - if self.http_server_pid is not None: - os.kill(self.http_server_pid, signal.SIGKILL) + for i in range(3): + pid = getattr(self, 'http_server%d_pid' % i, None) + if pid: + os.kill(pid, signal.SIGKILL) super(TestImageLocationSelectionStrategy, self).tearDown() @@ -2483,69 +2520,10 @@ class TestImageLocationSelectionStrategy(functional.FunctionalTest): path = self._url('/v2/images/%s' % image_id) media_type = 'application/openstack-images-v2.1-json-patch' headers = self._headers({'content-type': media_type}) - values = [{'url': 'file://%s' % self.foo_image_file.name, - 'metadata': {'idx': '1'}}, - {'url': 'http://127.0.0.1:%s/foo_image' % self.http_port, - 'metadata': {'idx': '0'}}] - doc = [{'op': 'replace', - 'path': '/locations', - 'value': values}] - data = jsonutils.dumps(doc) - response = requests.patch(path, headers=headers, data=data) - self.assertEqual(200, response.status_code) - - # Image locations should be visible - path = self._url('/v2/images/%s' % image_id) - headers = self._headers({'Content-Type': 'application/json'}) - response = requests.get(path, headers=headers) - self.assertEqual(200, response.status_code) - image = jsonutils.loads(response.text) - self.assertTrue('locations' in image) - self.assertEqual(image['locations'], values) - self.assertTrue('direct_url' in image) - self.assertEqual(image['direct_url'], values[0]['url']) - - self.stop_servers() - - def test_image_locatons_with_store_type_strategy(self): - self.api_server.show_image_direct_url = True - self.api_server.show_multiple_locations = True - self.image_location_quota = 10 - self.api_server.location_strategy = 'store_type' - preference = "http, swift, filesystem" - self.api_server.store_type_location_strategy_preference = preference - self.start_servers(**self.__dict__.copy()) - - # Create an image - path = self._url('/v2/images') - headers = self._headers({'content-type': 'application/json'}) - data = jsonutils.dumps({'name': 'image-1', 'type': 'kernel', - 'foo': 'bar', 'disk_format': 'aki', - 'container_format': 'aki'}) - response = requests.post(path, headers=headers, data=data) - self.assertEqual(201, response.status_code) - - # Get the image id - image = jsonutils.loads(response.text) - image_id = image['id'] - - # Image locations should not be visible before location is set - path = self._url('/v2/images/%s' % image_id) - headers = self._headers({'Content-Type': 'application/json'}) - response = requests.get(path, headers=headers) - self.assertEqual(200, response.status_code) - image = jsonutils.loads(response.text) - self.assertTrue('locations' in image) - self.assertTrue(image["locations"] == []) - - # Update image locations via PATCH - path = self._url('/v2/images/%s' % image_id) - media_type = 'application/openstack-images-v2.1-json-patch' - headers = self._headers({'content-type': media_type}) - values = [{'url': 'file://%s' % self.foo_image_file.name, - 'metadata': {'idx': '1'}}, - {'url': 'http://127.0.0.1:%s/foo_image' % self.http_port, - 'metadata': {'idx': '0'}}] + values = [{'url': 'http://127.0.0.1:%s/foo_image' % self.http_port0, + 'metadata': {}}, + {'url': 'http://127.0.0.1:%s/foo_image' % self.http_port1, + 'metadata': {}}] doc = [{'op': 'replace', 'path': '/locations', 'value': values}] @@ -2553,8 +2531,6 @@ class TestImageLocationSelectionStrategy(functional.FunctionalTest): response = requests.patch(path, headers=headers, data=data) self.assertEqual(200, response.status_code) - values.sort(key=lambda loc: int(loc['metadata']['idx'])) - # Image locations should be visible path = self._url('/v2/images/%s' % image_id) headers = self._headers({'Content-Type': 'application/json'}) diff --git a/glance/tests/unit/test_store_image.py b/glance/tests/unit/test_store_image.py index 665f126ae..8b334ab1f 100644 --- a/glance/tests/unit/test_store_image.py +++ b/glance/tests/unit/test_store_image.py @@ -18,6 +18,7 @@ import glance_store from glance.common import exception import glance.location +from glance.tests.unit import base as unit_test_base from glance.tests.unit import utils as unit_test_utils from glance.tests import utils @@ -759,7 +760,7 @@ class TestStoreImageRepo(utils.BaseTestCase): self.assertEqual(acls['read'], [TENANT2]) -class TestImageFactory(utils.BaseTestCase): +class TestImageFactory(unit_test_base.StoreClearingUnitTest): def setUp(self): super(TestImageFactory, self).setUp() diff --git a/glance/tests/unit/test_store_location.py b/glance/tests/unit/test_store_location.py index 884221bd9..c9ee44cd6 100644 --- a/glance/tests/unit/test_store_location.py +++ b/glance/tests/unit/test_store_location.py @@ -17,6 +17,8 @@ import mock import glance_store +from glance.common import exception +from glance.common import store_utils import glance.location from glance.tests.unit import base @@ -32,11 +34,13 @@ CONF = {'default_store': 'file', class TestStoreLocation(base.StoreClearingUnitTest): + class FakeImageProxy(): + size = None + context = None + store_api = mock.Mock() + store_utils = store_utils + def test_add_location_for_image_without_size(self): - class FakeImageProxy(): - size = None - context = None - store_api = mock.Mock() def fake_get_size_from_backend(uri, context=None): return 1 @@ -49,14 +53,31 @@ class TestStoreLocation(base.StoreClearingUnitTest): loc2 = {'url': 'file:///fake2.img.tar.gz', 'metadata': {}} # Test for insert location - image1 = FakeImageProxy() + image1 = TestStoreLocation.FakeImageProxy() locations = glance.location.StoreLocations(image1, []) locations.insert(0, loc2) self.assertEqual(image1.size, 1) # Test for set_attr of _locations_proxy - image2 = FakeImageProxy() + image2 = TestStoreLocation.FakeImageProxy() locations = glance.location.StoreLocations(image2, [loc1]) locations[0] = loc2 self.assertIn(loc2, locations) self.assertEqual(image2.size, 1) + + def test_add_location_with_restricted_sources(self): + + loc1 = {'url': 'file:///fake1.img.tar.gz', 'metadata': {}} + loc2 = {'url': 'swift+config:///xxx', 'metadata': {}} + + # Test for insert location + image1 = TestStoreLocation.FakeImageProxy() + locations = glance.location.StoreLocations(image1, []) + self.assertRaises(exception.BadStoreUri, locations.insert, 0, loc1) + self.assertNotIn(loc1, locations) + + # Test for set_attr of _locations_proxy + image2 = TestStoreLocation.FakeImageProxy() + locations = glance.location.StoreLocations(image2, [loc1]) + self.assertRaises(exception.BadStoreUri, locations.insert, 0, loc2) + self.assertNotIn(loc2, locations) diff --git a/glance/tests/unit/utils.py b/glance/tests/unit/utils.py index df591605a..f7c8d568c 100644 --- a/glance/tests/unit/utils.py +++ b/glance/tests/unit/utils.py @@ -14,12 +14,13 @@ # under the License. import urllib -import urlparse import glance_store as store from oslo.config import cfg +import six.moves.urllib.parse as urlparse from glance.common import exception +from glance.common import store_utils from glance.common import wsgi import glance.context import glance.db.simple.api as simple_db @@ -135,6 +136,12 @@ class FakeStoreUtils(object): else: self.safe_delete_from_backend(context, image_id, location) + def validate_external_location(self, uri): + if uri and urlparse.urlparse(uri).scheme: + return store_utils.validate_external_location(uri) + else: + return True + class FakeStoreAPI(object): def __init__(self, store_metadata=None): diff --git a/glance/tests/unit/v1/test_api.py b/glance/tests/unit/v1/test_api.py index bd2182ea8..4ec136d86 100644 --- a/glance/tests/unit/v1/test_api.py +++ b/glance/tests/unit/v1/test_api.py @@ -419,7 +419,7 @@ class TestGlanceAPI(base.IsolatedUnitTest): res = req.get_response(self.api) self.assertEqual(res.status_int, 400) - self.assertIn('External sourcing not supported', res.body) + self.assertIn('External source are not supported', res.body) def test_create_with_location_bad_store_uri(self): fixture_headers = { @@ -1006,7 +1006,7 @@ class TestGlanceAPI(base.IsolatedUnitTest): res = req.get_response(self.api) self.assertEqual(res.status_int, 409) - def test_add_location_with_invalid_location(self): + def test_add_location_with_invalid_location_on_conflict_image_size(self): """Tests creates an image from location and conflict image size""" fixture_headers = {'x-image-meta-store': 'file', 'x-image-meta-disk-format': 'vhd', @@ -1023,6 +1023,36 @@ class TestGlanceAPI(base.IsolatedUnitTest): res = req.get_response(self.api) self.assertEqual(res.status_int, 400) + def test_add_location_with_invalid_location_on_restricted_sources(self): + """Tests creates an image from location and restricted sources""" + fixture_headers = {'x-image-meta-store': 'file', + 'x-image-meta-disk-format': 'vhd', + 'x-image-meta-location': 'file:///etc/passwd', + 'x-image-meta-container-format': 'ovf', + 'x-image-meta-name': 'fake image #F'} + + req = webob.Request.blank("/images") + req.headers['Content-Type'] = 'application/octet-stream' + req.method = 'POST' + for k, v in fixture_headers.iteritems(): + req.headers[k] = v + res = req.get_response(self.api) + self.assertEqual(400, res.status_int) + + fixture_headers = {'x-image-meta-store': 'file', + 'x-image-meta-disk-format': 'vhd', + 'x-image-meta-location': 'swift+config://xxx', + 'x-image-meta-container-format': 'ovf', + 'x-image-meta-name': 'fake image #F'} + + req = webob.Request.blank("/images") + req.headers['Content-Type'] = 'application/octet-stream' + req.method = 'POST' + for k, v in fixture_headers.iteritems(): + req.headers[k] = v + res = req.get_response(self.api) + self.assertEqual(400, res.status_int) + def test_add_copy_from_with_location(self): """Tests creates an image from copy-from and location""" fixture_headers = {'x-image-meta-store': 'file', @@ -1039,6 +1069,34 @@ class TestGlanceAPI(base.IsolatedUnitTest): res = req.get_response(self.api) self.assertEqual(res.status_int, 400) + def test_add_copy_from_with_restricted_sources(self): + """Tests creates an image from copy-from with restricted sources""" + fixture_headers = {'x-image-meta-store': 'file', + 'x-image-meta-disk-format': 'vhd', + 'x-glance-api-copy-from': 'file:///etc/passwd', + 'x-image-meta-container-format': 'ovf', + 'x-image-meta-name': 'fake image #F'} + + req = webob.Request.blank("/images") + req.method = 'POST' + for k, v in six.iteritems(fixture_headers): + req.headers[k] = v + res = req.get_response(self.api) + self.assertEqual(400, res.status_int) + + fixture_headers = {'x-image-meta-store': 'file', + 'x-image-meta-disk-format': 'vhd', + 'x-glance-api-copy-from': 'swift+config://xxx', + 'x-image-meta-container-format': 'ovf', + 'x-image-meta-name': 'fake image #F'} + + req = webob.Request.blank("/images") + req.method = 'POST' + for k, v in six.iteritems(fixture_headers): + req.headers[k] = v + res = req.get_response(self.api) + self.assertEqual(400, res.status_int) + def test_add_copy_from_upload_image_unauthorized_with_body(self): rules = {"upload_image": '!', "modify_image": '@', "add_image": '@'} |