To prevent client use v2 patch api to handle file and swift location

The change will be used to restrict client to download and delete any
file in glance-api server. The same resone and logic as what we did in
v1:
https://github.com/openstack/glance/blob/master/glance/api/v1/images.py#L429

Closes-Bug: bug/1400966
DocImpact

Note: Even this change could fully resolve the problem for Glance, but
we still need to fix this issue from glance_store perspective
separatelly due to other projects can use the lib directly.

Change-Id: I72dbead3cb2dcb87f52658ddb880e26880cc229b
Signed-off-by: Zhi Yan Liu <zhiyanl@cn.ibm.com>

glance/api/v1/images.py

@@ -25,7 +25,6 @@ import glance_store.location
 from oslo.config import cfg
 from oslo.utils import excutils
 from oslo.utils import strutils
-import six.moves.urllib.parse as urlparse
 from webob.exc import HTTPBadRequest
 from webob.exc import HTTPConflict
 from webob.exc import HTTPForbidden
@@ -42,6 +41,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 import i18n
@@ -416,26 +416,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):

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

glance/location.py

@@ -26,6 +26,7 @@ import glance.domain.proxy
 from glance import i18n
 import glance.openstack.common.log as logging
 
+
 _ = i18n._
 _LE = i18n._LE
 
@@ -66,18 +67,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 +88,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 +125,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 +135,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 +176,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 +223,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 +314,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:

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(400, response.status, 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(400, response.status, 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)
 

glance/tests/functional/v2/test_images.py

@@ -16,7 +16,6 @@
 import BaseHTTPServer
 import os
 import signal
-import tempfile
 import uuid
 
 from oslo.serialization import jsonutils
@@ -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)
@@ -335,21 +346,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)
@@ -2137,17 +2142,14 @@ class TestImages(functional.FunctionalTest):
         self.assertIsNone(image['size'])
         self.assertIsNone(image['virtual_size'])
 
-        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)
 
@@ -2156,7 +2158,42 @@ class TestImages(functional.FunctionalTest):
         response = requests.get(path, headers=headers)
         self.assertEqual(200, response.status_code)
         image = jsonutils.loads(response.text)
-        self.assertEqual(6, image['size'])
+        self.assertEqual(10, image['size'])
+
+    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.assertIsNone(image['size'])
+        self.assertIsNone(image['virtual_size'])
+
+        # 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):
@@ -2382,16 +2419,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()
 
@@ -2444,10 +2481,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'}}]
+        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}]
@@ -2468,67 +2505,6 @@ class TestImageLocationSelectionStrategy(functional.FunctionalTest):
 
         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'}}]
-        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)
-
-        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'})
-        response = requests.get(path, headers=headers)
-        self.assertEqual(200, response.status_code)
-        image = jsonutils.loads(response.text)
-        self.assertTrue('locations' in image)
-        self.assertEqual(values, image['locations'])
-        self.assertTrue('direct_url' in image)
-        self.assertEqual(values[0]['url'], image['direct_url'])
-
-        self.stop_servers()
-
 
 class TestImageLocationSelectionStrategyWithRegistry(
         TestImageLocationSelectionStrategy):

glance/tests/unit/test_store_image.py

@@ -17,6 +17,7 @@ import mox
 
 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([TENANT2], acls['read'])
 
 
-class TestImageFactory(utils.BaseTestCase):
+class TestImageFactory(unit_test_base.StoreClearingUnitTest):
 
     def setUp(self):
         super(TestImageFactory, self).setUp()

glance/tests/unit/test_store_location.py

@@ -15,6 +15,8 @@
 import glance_store
 import mock
 
+from glance.common import exception
+from glance.common import store_utils
 import glance.location
 from glance.tests.unit import base
 
@@ -30,11 +32,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
@@ -47,14 +51,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(1, image1.size)
 
             # 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(1, image2.size)
+
+    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)

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):

glance/tests/unit/v1/test_api.py

@@ -418,7 +418,7 @@ class TestGlanceAPI(base.IsolatedUnitTest):
 
         res = req.get_response(self.api)
         self.assertEqual(400, res.status_int)
-        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 = {
@@ -1005,7 +1005,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',
@@ -1022,6 +1022,36 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEqual(400, res.status_int)
 
+    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',
@@ -1038,6 +1068,34 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEqual(400, res.status_int)
 
+    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": '@'}