diff --git a/glance/api/v2/images.py b/glance/api/v2/images.py
index 282069cec0..95cfe7b9c3 100644
--- a/glance/api/v2/images.py
+++ b/glance/api/v2/images.py
@@ -18,6 +18,7 @@ import http.client as http
 import os
 import re
 import urllib.parse as urlparse
+import uuid
 
 from castellan.common import exception as castellan_exception
 from castellan import key_manager
@@ -332,9 +333,10 @@ class ImagesController(object):
                 msg = _("Only images with status active can be targeted for "
                         "copying")
                 raise exception.Conflict(msg)
-            if image.status != 'queued' and import_method == 'web-download':
+            if (image.status != 'queued' and
+                    import_method in ['web-download', 'glance-download']):
                 msg = _("Image needs to be in 'queued' state to use "
-                        "'web-download' method")
+                        "'%s' method") % import_method
                 raise exception.Conflict(msg)
             if (image.status != 'uploading' and
                     import_method == 'glance-direct'):
@@ -347,6 +349,23 @@ class ImagesController(object):
             if not getattr(image, 'disk_format', None):
                 msg = _("'disk_format' needs to be set before import")
                 raise exception.Conflict(msg)
+            if import_method == 'glance-download':
+                if 'glance_region' not in body.get('method'):
+                    msg = _("'glance_region' needs to be set for "
+                            "glance-download import method")
+                    raise webob.exc.HTTPBadRequest(explanation=msg)
+                if 'glance_image_id' not in body.get('method'):
+                    msg = _("'glance_image_id' needs to be set for "
+                            "glance-download import method")
+                    raise webob.exc.HTTPBadRequest(explanation=msg)
+                try:
+                    uuid.UUID(body['method']['glance_image_id'])
+                except ValueError:
+                    msg = (_("Remote image id does not look like a UUID: %s")
+                           % body['method']['glance_image_id'])
+                    raise webob.exc.HTTPBadRequest(explanation=msg)
+                if 'glance_service_interface' not in body.get('method'):
+                    body.get('method')['glance_service_interface'] = 'public'
 
             # NOTE(danms): For copy-image only, we check policy to decide
             # if the user should be able to do this. Otherwise, we forbid
diff --git a/glance/async_/flows/_internal_plugins/base_download.py b/glance/async_/flows/_internal_plugins/base_download.py
new file mode 100644
index 0000000000..a8f90b9263
--- /dev/null
+++ b/glance/async_/flows/_internal_plugins/base_download.py
@@ -0,0 +1,131 @@
+# Copyright 2018 Red Hat, Inc.
+# Copyright 2022 OVHCloud
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import abc
+
+import glance_store as store_api
+from glance_store import backend
+from oslo_config import cfg
+from oslo_log import log as logging
+import six
+from taskflow import task
+
+from glance.common import exception
+from glance.i18n import _, _LE
+
+LOG = logging.getLogger(__name__)
+
+CONF = cfg.CONF
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseDownload(task.Task):
+
+    default_provides = 'file_uri'
+
+    def __init__(self, task_id, task_type, action_wrapper, stores,
+                 plugin_name):
+        self.task_id = task_id
+        self.task_type = task_type
+        self.image_id = action_wrapper.image_id
+        self.action_wrapper = action_wrapper
+        self.stores = stores
+        self._path = None
+        self.plugin_name = plugin_name or 'Download'
+        super(BaseDownload, self).__init__(
+            name='%s-%s-%s' % (task_type, self.plugin_name, task_id))
+
+        # NOTE(abhishekk): Use reserved 'os_glance_staging_store' for
+        # staging the data, the else part will be removed once old way
+        # of configuring store is deprecated.
+        if CONF.enabled_backends:
+            self.store = store_api.get_store_from_store_identifier(
+                'os_glance_staging_store')
+        else:
+            if CONF.node_staging_uri is None:
+                msg = (_("%(task_id)s of %(task_type)s not configured "
+                         "properly. Missing node_staging_uri: %(work_dir)s") %
+                       {'task_id': self.task_id,
+                        'task_type': self.task_type,
+                        'work_dir': CONF.node_staging_uri})
+                raise exception.BadTaskConfiguration(msg)
+
+            self.store = self._build_store()
+
+    def _build_store(self):
+        # NOTE(flaper87): Due to the nice glance_store api (#sarcasm), we're
+        # forced to build our own config object, register the required options
+        # (and by required I mean *ALL* of them, even the ones we don't want),
+        # and create our own store instance by calling a private function.
+        # This is certainly unfortunate but it's the best we can do until the
+        # glance_store refactor is done. A good thing is that glance_store is
+        # under our team's management and it gates on Glance so changes to
+        # this API will (should?) break task's tests.
+        # TODO(abhishekk): After removal of backend module from glance_store
+        # need to change this to use multi_backend module.
+        conf = cfg.ConfigOpts()
+        try:
+            backend.register_opts(conf)
+        except cfg.DuplicateOptError:
+            pass
+
+        conf.set_override('filesystem_store_datadir',
+                          CONF.node_staging_uri[7:],
+                          group='glance_store')
+
+        # NOTE(flaper87): Do not even try to judge me for this... :(
+        # With the glance_store refactor, this code will change, until
+        # that happens, we don't have a better option and this is the
+        # least worst one, IMHO.
+        store = store_api.backend._load_store(conf, 'file')
+
+        if store is None:
+            msg = (_("%(task_id)s of %(task_type)s not configured "
+                     "properly. Could not load the filesystem store") %
+                   {'task_id': self.task_id, 'task_type': self.task_type})
+            raise exception.BadTaskConfiguration(msg)
+
+        store.configure()
+        return store
+
+    def revert(self, result, **kwargs):
+        LOG.error(_LE('Task: %(task_id)s failed to import image '
+                      '%(image_id)s to the filesystem.'),
+                  {'task_id': self.task_id,
+                   'image_id': self.image_id})
+        # NOTE(abhishekk): Revert image state back to 'queued' as
+        # something went wrong.
+        # NOTE(danms): If we failed to stage the image, then none
+        # of the _ImportToStore() tasks could have run, so we need
+        # to move all stores out of "importing" and into "failed".
+        with self.action_wrapper as action:
+            action.set_image_attribute(status='queued')
+            action.remove_importing_stores(self.stores)
+            action.add_failed_stores(self.stores)
+
+        # NOTE(abhishekk): Deleting partial image data from staging area
+        if self._path is not None:
+            LOG.debug(('Deleting image %(image_id)s from staging '
+                       'area.'), {'image_id': self.image_id})
+            try:
+                if CONF.enabled_backends:
+                    store_api.delete(self._path, None)
+                else:
+                    store_api.delete_from_backend(self._path)
+            except Exception:
+                LOG.exception(_LE("Error reverting web/glance download "
+                                  "task: %(task_id)s"), {
+                    'task_id': self.task_id})
diff --git a/glance/async_/flows/_internal_plugins/glance_download.py b/glance/async_/flows/_internal_plugins/glance_download.py
new file mode 100644
index 0000000000..3eefe31455
--- /dev/null
+++ b/glance/async_/flows/_internal_plugins/glance_download.py
@@ -0,0 +1,115 @@
+# Copyright 2022 OVHCloud
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import urllib.request
+
+from oslo_config import cfg
+from oslo_log import log as logging
+from oslo_utils import encodeutils
+from oslo_utils import excutils
+from taskflow.patterns import linear_flow as lf
+
+from glance.async_.flows._internal_plugins import base_download
+from glance.async_ import utils
+from glance.common import exception
+from glance.common import utils as common_utils
+from glance.i18n import _, _LI, _LE
+
+LOG = logging.getLogger(__name__)
+
+CONF = cfg.CONF
+
+
+class _DownloadGlanceImage(base_download.BaseDownload):
+
+    def __init__(self, context, task_id, task_type, action_wrapper, stores,
+                 glance_region, glance_image_id, glance_service_interface):
+        self.context = context
+        self.glance_region = glance_region
+        self.glance_image_id = glance_image_id
+        self.glance_service_interface = glance_service_interface
+        super(_DownloadGlanceImage,
+              self).__init__(task_id, task_type, action_wrapper, stores,
+                             'GlanceDownload')
+
+    def execute(self, image_size):
+        """Create temp file into store and return path to it
+
+        :param image_size: Glance Image Size retrieved from ImportMetadata task
+        """
+        try:
+            glance_endpoint = utils.get_glance_endpoint(
+                self.context,
+                self.glance_region,
+                self.glance_service_interface)
+            image_download_url = '%s/v2/images/%s/file' % (
+                glance_endpoint, self.glance_image_id)
+            if not common_utils.validate_import_uri(image_download_url):
+                LOG.debug("Processed URI for glance-download does not pass "
+                          "filtering: %s", image_download_url)
+                msg = (_("Processed URI for glance-download does not pass "
+                         "filtering: %s") % image_download_url)
+                raise exception.ImportTaskError(msg)
+            LOG.info(_LI("Downloading glance image %s"), image_download_url)
+            token = self.context.auth_token
+            request = urllib.request.Request(image_download_url,
+                                             headers={'X-Auth-Token': token})
+            data = urllib.request.urlopen(request)
+        except Exception as e:
+            with excutils.save_and_reraise_exception():
+                LOG.error(
+                    _LE("Task %(task_id)s failed with exception %(error)s"), {
+                        "error": encodeutils.exception_to_unicode(e),
+                        "task_id": self.task_id
+                    })
+
+        self._path, bytes_written = self.store.add(self.image_id, data, 0)[0:2]
+        if bytes_written != image_size:
+            msg = (_("Task %(task_id)s failed because downloaded data "
+                     "size %(data_size)i is different from expected %("
+                     "expected)i") %
+                   {"task_id": self.task_id, "data_size": bytes_written,
+                    "expected": image_size})
+            raise exception.ImportTaskError(msg)
+        return self._path
+
+
+def get_flow(**kwargs):
+    """Return task flow for no-op.
+
+    :param context: request context
+    :param task_id: Task ID.
+    :param task_type: Type of the task.
+    :param image_repo: Image repository used.
+    :param image_id: Image ID
+    :param source_region: Source region name
+    """
+    context = kwargs.get('context')
+    task_id = kwargs.get('task_id')
+    task_type = kwargs.get('task_type')
+    action_wrapper = kwargs.get('action_wrapper')
+    stores = kwargs.get('backend', [None])
+    # glance-download parameters
+    import_req = kwargs.get('import_req')
+    method = import_req.get('method')
+    glance_region = method.get('glance_region')
+    glance_image_id = method.get('glance_image_id')
+    glance_service_interface = method.get('glance_service_interface')
+
+    return lf.Flow(task_type).add(
+        _DownloadGlanceImage(context, task_id, task_type, action_wrapper,
+                             stores, glance_region, glance_image_id,
+                             glance_service_interface),
+    )
diff --git a/glance/async_/flows/_internal_plugins/web_download.py b/glance/async_/flows/_internal_plugins/web_download.py
index 98e497c3c3..e14e4431f0 100644
--- a/glance/async_/flows/_internal_plugins/web_download.py
+++ b/glance/async_/flows/_internal_plugins/web_download.py
@@ -1,4 +1,5 @@
 # Copyright 2018 Red Hat, Inc.
+# Copyright 2022 OVHCloud
 # All Rights Reserved.
 #
 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
@@ -12,92 +13,29 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
-import glance_store as store_api
-from glance_store import backend
+
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import encodeutils
 from oslo_utils import excutils
 from taskflow.patterns import linear_flow as lf
-from taskflow import task
-from taskflow.types import failure
 
+from glance.async_.flows._internal_plugins import base_download
 from glance.common import exception
 from glance.common.scripts import utils as script_utils
-from glance.i18n import _, _LE
+from glance.i18n import _
 
 LOG = logging.getLogger(__name__)
 
 CONF = cfg.CONF
 
 
-class _WebDownload(task.Task):
-
-    default_provides = 'file_uri'
+class _WebDownload(base_download.BaseDownload):
 
     def __init__(self, task_id, task_type, uri, action_wrapper, stores):
-        self.task_id = task_id
-        self.task_type = task_type
-        self.image_id = action_wrapper.image_id
         self.uri = uri
-        self.action_wrapper = action_wrapper
-        self.stores = stores
-        self._path = None
-        super(_WebDownload, self).__init__(
-            name='%s-WebDownload-%s' % (task_type, task_id))
-
-        # NOTE(abhishekk): Use reserved 'os_glance_staging_store' for
-        # staging the data, the else part will be removed once old way
-        # of configuring store is deprecated.
-        if CONF.enabled_backends:
-            self.store = store_api.get_store_from_store_identifier(
-                'os_glance_staging_store')
-        else:
-            if CONF.node_staging_uri is None:
-                msg = (_("%(task_id)s of %(task_type)s not configured "
-                         "properly. Missing node_staging_uri: %(work_dir)s") %
-                       {'task_id': self.task_id,
-                        'task_type': self.task_type,
-                        'work_dir': CONF.node_staging_uri})
-                raise exception.BadTaskConfiguration(msg)
-
-            self.store = self._build_store()
-
-    def _build_store(self):
-        # NOTE(flaper87): Due to the nice glance_store api (#sarcasm), we're
-        # forced to build our own config object, register the required options
-        # (and by required I mean *ALL* of them, even the ones we don't want),
-        # and create our own store instance by calling a private function.
-        # This is certainly unfortunate but it's the best we can do until the
-        # glance_store refactor is done. A good thing is that glance_store is
-        # under our team's management and it gates on Glance so changes to
-        # this API will (should?) break task's tests.
-        # TODO(abhishekk): After removal of backend module from glance_store
-        # need to change this to use multi_backend module.
-        conf = cfg.ConfigOpts()
-        try:
-            backend.register_opts(conf)
-        except cfg.DuplicateOptError:
-            pass
-
-        conf.set_override('filesystem_store_datadir',
-                          CONF.node_staging_uri[7:],
-                          group='glance_store')
-
-        # NOTE(flaper87): Do not even try to judge me for this... :(
-        # With the glance_store refactor, this code will change, until
-        # that happens, we don't have a better option and this is the
-        # least worst one, IMHO.
-        store = backend._load_store(conf, 'file')
-
-        if store is None:
-            msg = (_("%(task_id)s of %(task_type)s not configured "
-                     "properly. Could not load the filesystem store") %
-                   {'task_id': self.task_id, 'task_type': self.task_type})
-            raise exception.BadTaskConfiguration(msg)
-
-        store.configure()
-        return store
+        super(_WebDownload, self).__init__(task_id, task_type, action_wrapper,
+                                           stores, 'WebDownload')
 
     def execute(self):
         """Create temp file into store and return path to it
@@ -133,36 +71,6 @@ class _WebDownload(task.Task):
             pass
         return self._path
 
-    def revert(self, result, **kwargs):
-        if isinstance(result, failure.Failure):
-            LOG.error(_LE('Task: %(task_id)s failed to import image '
-                          '%(image_id)s to the filesystem.'),
-                      {'task_id': self.task_id,
-                       'image_id': self.image_id})
-            # NOTE(abhishekk): Revert image state back to 'queued' as
-            # something went wrong.
-            # NOTE(danms): If we failed to stage the image, then none
-            # of the _ImportToStore() tasks could have run, so we need
-            # to move all stores out of "importing" and into "failed".
-            with self.action_wrapper as action:
-                action.set_image_attribute(status='queued')
-                action.remove_importing_stores(self.stores)
-                action.add_failed_stores(self.stores)
-
-        # NOTE(abhishekk): Deleting partial image data from staging area
-        if self._path is not None:
-            LOG.debug(('Deleting image %(image_id)s from staging '
-                       'area.'), {'image_id': self.image_id})
-            try:
-                if CONF.enabled_backends:
-                    store_api.delete(self._path, None)
-                else:
-                    store_api.delete_from_backend(self._path)
-            except Exception:
-                LOG.exception(_LE("Error reverting web-download "
-                                  "task: %(task_id)s"), {
-                    'task_id': self.task_id})
-
 
 def get_flow(**kwargs):
     """Return task flow for web-download.
diff --git a/glance/async_/flows/api_image_import.py b/glance/async_/flows/api_image_import.py
index 2f52d0ac28..ee72fbaeda 100644
--- a/glance/async_/flows/api_image_import.py
+++ b/glance/async_/flows/api_image_import.py
@@ -14,7 +14,9 @@
 #    under the License.
 import copy
 import functools
+import json
 import os
+import urllib.request
 
 import glance_store as store_api
 from glance_store import backend
@@ -33,6 +35,7 @@ from taskflow import task
 from glance.api import common as api_common
 import glance.async_.flows._internal_plugins as internal_plugins
 import glance.async_.flows.plugins as import_plugins
+from glance.async_ import utils
 from glance.common import exception
 from glance.common.scripts.image_import import main as image_import
 from glance.common.scripts import utils as script_utils
@@ -79,6 +82,30 @@ Possible values:
 
 CONF.register_opts(api_import_opts, group='image_import_opts')
 
+glance_download_opts = [
+    cfg.ListOpt('extra_properties',
+                item_type=cfg.types.String(quotes=True),
+                bounds=True,
+                default=[
+                    'hw_', 'trait:', 'os_distro', 'os_secure_boot',
+                    'os_type'],
+                help=_("""
+Specify metadata prefix to be set on the target image when using
+glance-download. All other properties coming from the source image won't be set
+on the target image. If specified metadata does not exist on the source image
+it won't be set on the target image. Note you can't set the os_glance prefix
+as it is reserved by glance, so the related properties won't be set on the
+target image.
+
+Possible values:
+    * List containing extra_properties prefixes: ['os_', 'architecture']
+
+""")),
+]
+
+CONF.register_opts(glance_download_opts, group='glance_download_properties')
+
+
 # TODO(jokke): We should refactor the task implementations so that we do not
 # need to duplicate what we have already for example in base_import.py.
 
@@ -89,6 +116,12 @@ class _NoStoresSucceeded(exception.GlanceException):
         super(_NoStoresSucceeded, self).__init__(message)
 
 
+class _InvalidGlanceDownloadImageStatus(exception.GlanceException):
+
+    def __init__(self, message):
+        super(_InvalidGlanceDownloadImageStatus, self).__init__(message)
+
+
 class ImportActionWrapper(object):
     """Wrapper for all the image metadata operations we do during an import.
 
@@ -207,6 +240,18 @@ class _ImportActions(object):
         # should have moderated access like all the other things here.
         return copy.deepcopy(self._image.locations)
 
+    @property
+    def image_disk_format(self):
+        return self._image.disk_format
+
+    @property
+    def image_container_format(self):
+        return self._image.container_format
+
+    @property
+    def image_extra_properties(self):
+        return dict(self._image.extra_properties)
+
     @property
     def image_status(self):
         return self._image.status
@@ -739,6 +784,93 @@ class _CompleteTask(task.Task):
                  {'task_id': self.task_id, 'task_type': self.task_type})
 
 
+class _ImportMetadata(task.Task):
+
+    default_provides = 'image_size'
+
+    def __init__(self, task_id, task_type, context, action_wrapper,
+                 import_req):
+        self.task_id = task_id
+        self.task_type = task_type
+        self.context = context
+        self.action_wrapper = action_wrapper
+        self.import_req = import_req
+        self.props_to_copy = CONF.glance_download_properties.extra_properties
+        # We store the properties that will be set in case we are reverting
+        self.properties = {}
+        self.old_properties = {}
+        self.old_attributes = {}
+        super(_ImportMetadata, self).__init__(
+            name='%s-ImportMetdata-%s' % (task_type, task_id))
+
+    def execute(self):
+        try:
+            glance_endpoint = utils.get_glance_endpoint(
+                self.context,
+                self.import_req['method']['glance_region'],
+                self.import_req['method']['glance_service_interface'])
+            glance_image_id = self.import_req['method']['glance_image_id']
+            image_download_metadata_url = '%s/v2/images/%s' % (
+                glance_endpoint, glance_image_id)
+            LOG.info(_LI("Fetching glance image metadata from remote host %s"),
+                     image_download_metadata_url)
+            token = self.context.auth_token
+            request = urllib.request.Request(image_download_metadata_url,
+                                             headers={'X-Auth-Token': token})
+            with urllib.request.urlopen(request) as payload:
+                data = json.loads(payload.read().decode('utf-8'))
+
+            if data.get('status') != 'active':
+                raise _InvalidGlanceDownloadImageStatus(
+                    _('Source image status should be active instead of %s')
+                    % data['status'])
+
+            for key, value in data.items():
+                for metadata in self.props_to_copy:
+                    if key.startswith(metadata):
+                        self.properties[key] = value
+
+            with self.action_wrapper as action:
+                # Save the old properties in case we need to revert
+                self.old_properties = action.image_extra_properties
+                self.old_attributes = {
+                    'container_format': action.image_container_format,
+                    'disk_format': action.image_disk_format,
+                }
+
+                # Set disk_format and container_format attributes
+                action.set_image_attribute(
+                    disk_format=data['disk_format'],
+                    container_format=data['container_format'])
+
+                # Set extra propoerties
+                if self.properties:
+                    action.set_image_extra_properties(self.properties)
+            try:
+                return int(data['size'])
+            except (ValueError, KeyError):
+                raise exception.ImportTaskError(
+                    _('Size attribute of remote image %s could not be '
+                      'determined.' % glance_image_id))
+        except Exception as e:
+            with excutils.save_and_reraise_exception():
+                LOG.error(
+                    "Task %(task_id)s failed with exception %(error)s", {
+                        "error": encodeutils.exception_to_unicode(e),
+                        "task_id": self.task_id
+                    })
+
+    def revert(self, result, **kwargs):
+        """Revert the extra properties set and set the image in queued"""
+        with self.action_wrapper as action:
+            for image_property in self.properties:
+                if image_property not in self.old_properties:
+                    action.pop_extra_property(image_property)
+            action.set_image_extra_properties(self.old_properties)
+            action.set_image_attribute(status='queued',
+                                       **self.old_attributes)
+
+
 def assert_quota(context, task_repo, task_id, stores,
                  action_wrapper, enforce_quota_fn,
                  **enforce_kwargs):
@@ -776,11 +908,13 @@ def get_flow(**kwargs):
     image_repo = kwargs.get('image_repo')
     admin_repo = kwargs.get('admin_repo')
     image_id = kwargs.get('image_id')
-    import_method = kwargs.get('import_req')['method']['name']
-    uri = kwargs.get('import_req')['method'].get('uri')
+    import_req = kwargs.get('import_req')
+    import_method = import_req['method']['name']
+    uri = import_req['method'].get('uri')
     stores = kwargs.get('backend', [None])
-    all_stores_must_succeed = kwargs.get('import_req').get(
+    all_stores_must_succeed = import_req.get(
         'all_stores_must_succeed', True)
+    context = kwargs.get('context')
 
     separator = ''
     if not CONF.enabled_backends and not CONF.node_staging_uri.endswith('/'):
@@ -803,7 +937,10 @@ def get_flow(**kwargs):
 
     flow.add(_ImageLock(task_id, task_type, action_wrapper))
 
-    if import_method in ['web-download', 'copy-image']:
+    if import_method in ['web-download', 'copy-image', 'glance-download']:
+        if import_method == 'glance-download':
+            flow.add(_ImportMetadata(task_id, task_type,
+                                     context, action_wrapper, import_req))
         internal_plugin = internal_plugins.get_import_plugin(**kwargs)
         flow.add(internal_plugin)
         if CONF.enabled_backends:
@@ -874,9 +1011,9 @@ def get_flow(**kwargs):
                      stores, action_wrapper,
                      ks_quota.enforce_image_size_total,
                      delta=image_size)
-    elif import_method in ('copy-image', 'web-download'):
-        # The copy-image and web-download methods will use staging space to
-        # do their work, so check that quota.
+    elif import_method in ('copy-image', 'web-download', 'glance-download'):
+        # The copy-image, web-download and glance-download methods will use
+        # staging space to do their work, so check that quota.
         assert_quota(kwargs['context'], task_repo, task_id,
                      stores, action_wrapper,
                      ks_quota.enforce_image_staging_total,
diff --git a/glance/common/config.py b/glance/common/config.py
index 59091422b2..07345978a7 100644
--- a/glance/common/config.py
+++ b/glance/common/config.py
@@ -572,7 +572,7 @@ Related options:
                 item_type=cfg.types.String(quotes=True),
                 bounds=True,
                 default=['glance-direct', 'web-download',
-                         'copy-image'],
+                         'copy-image', 'glance-download'],
                 help=_("""
     List of enabled Image Import Methods
 
diff --git a/glance/tests/unit/async_/flows/test_api_image_import.py b/glance/tests/unit/async_/flows/test_api_image_import.py
index 937c7484f4..65384555f3 100644
--- a/glance/tests/unit/async_/flows/test_api_image_import.py
+++ b/glance/tests/unit/async_/flows/test_api_image_import.py
@@ -15,6 +15,7 @@
 
 import sys
 from unittest import mock
+import urllib.error
 
 from glance_store import exceptions as store_exceptions
 from oslo_config import cfg
@@ -25,6 +26,7 @@ import glance.async_.flows.api_image_import as import_flow
 from glance.common import exception
 from glance.common.scripts.image_import import main as image_import
 from glance import context
+from glance.domain import ExtraProperties
 from glance import gateway
 import glance.tests.utils as test_utils
 
@@ -890,7 +892,9 @@ class TestImportActions(test_utils.BaseTestCase):
         self.image = mock.MagicMock()
         self.image.image_id = IMAGE_ID1
         self.image.status = 'active'
-        self.image.extra_properties = {'speed': '88mph'}
+        self.image.disk_format = 'raw'
+        self.image.container_format = 'bare'
+        self.image.extra_properties = ExtraProperties({'speed': '88mph'})
         self.image.checksum = mock.sentinel.checksum
         self.image.os_hash_algo = mock.sentinel.hash_algo
         self.image.os_hash_value = mock.sentinel.hash_value
@@ -900,6 +904,10 @@ class TestImportActions(test_utils.BaseTestCase):
     def test_image_property_proxies(self):
         self.assertEqual(IMAGE_ID1, self.actions.image_id)
         self.assertEqual('active', self.actions.image_status)
+        self.assertEqual('raw', self.actions.image_disk_format)
+        self.assertEqual('bare', self.actions.image_container_format)
+        self.assertEqual({'speed': '88mph'},
+                         self.actions.image_extra_properties)
 
     def test_merge_store_list(self):
         # Addition with no existing property works
@@ -1131,3 +1139,152 @@ class TestCompleteTask(test_utils.BaseTestCase):
                                                 {'image': IMAGE_ID1,
                                                  'task': TASK_ID1})
         self.task.succeed.assert_called_once_with({'image_id': IMAGE_ID1})
+
+
+class TestImportMetadata(test_utils.BaseTestCase):
+    def setUp(self):
+        super(TestImportMetadata, self).setUp()
+        self.config(extra_properties=[],
+                    group="glance_download_properties")
+        self.wrapper = mock.MagicMock(image_id=IMAGE_ID1)
+        self.context = context.RequestContext(user_id=TENANT1,
+                                              project_id=TENANT1,
+                                              overwrite=False)
+        self.import_req = {
+            'method': {
+                'glance_region': 'RegionTwo',
+                'glance_service_interface': 'public',
+                'glance_image_id': IMAGE_ID1
+            }
+        }
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.flows.api_image_import.json')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_execute_return_image_size(self, mock_gge, mock_json,
+                                       mock_request):
+        self.config(extra_properties=['hw:numa_nodes', 'os_hash'],
+                    group="glance_download_properties")
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        action = self.wrapper.__enter__.return_value
+        mock_json.loads.return_value = {
+            'status': 'active',
+            'disk_format': 'qcow2',
+            'container_format': 'bare',
+            'hw:numa_nodes': '2',
+            'os_hash': 'hash',
+            'extra_metadata': 'hello',
+            'size': '12345'
+        }
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        self.assertEqual(12345, task.execute())
+        mock_request.Request.assert_called_once_with(
+            'https://other.cloud.foo/image/v2/images/%s' % (
+                IMAGE_ID1),
+            headers={'X-Auth-Token': self.context.auth_token})
+        mock_gge.assert_called_once_with(self.context, 'RegionTwo', 'public')
+        action.set_image_attribute.assert_called_once_with(
+            disk_format='qcow2',
+            container_format='bare')
+        action.set_image_extra_properties.assert_called_once_with({
+            'hw:numa_nodes': '2',
+            'os_hash': 'hash'
+        })
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_execute_fail_no_glance_endpoint(self, mock_gge, mock_request):
+        action = self.wrapper.__enter__.return_value
+        mock_gge.side_effect = exception.GlanceEndpointNotFound(
+            region='RegionTwo',
+            interface='public')
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        self.assertRaises(exception.GlanceEndpointNotFound,
+                          task.execute)
+        action.assert_not_called()
+        mock_request.assert_not_called()
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_execute_fail_remote_glance_unreachable(self, mock_gge, mock_r):
+        action = self.wrapper.__enter__.return_value
+        mock_r.urlopen.side_effect = urllib.error.HTTPError(
+            '/file', 400, 'Test Fail', {}, None)
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        self.assertRaises(urllib.error.HTTPError,
+                          task.execute)
+        action.assert_not_called()
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.flows.api_image_import.json')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_execute_invalid_remote_image_state(self, mock_gge, mock_json,
+                                                mock_request):
+        action = self.wrapper.__enter__.return_value
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        mock_json.loads.return_value = {
+            'status': 'queued',
+        }
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        self.assertRaises(import_flow._InvalidGlanceDownloadImageStatus,
+                          task.execute)
+        action.assert_not_called()
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.flows.api_image_import.json')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_execute_raise_if_no_size(self, mock_gge, mock_json, mock_request):
+        self.config(extra_properties=['hw:numa_nodes', 'os_hash'],
+                    group="glance_download_properties")
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        action = self.wrapper.__enter__.return_value
+        mock_json.loads.return_value = {
+            'status': 'active',
+            'disk_format': 'qcow2',
+            'container_format': 'bare',
+            'hw:numa_nodes': '2',
+            'os_hash': 'hash',
+            'extra_metadata': 'hello',
+        }
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        self.assertRaises(exception.ImportTaskError, task.execute)
+        mock_request.Request.assert_called_once_with(
+            'https://other.cloud.foo/image/v2/images/%s' % (
+                IMAGE_ID1),
+            headers={'X-Auth-Token': self.context.auth_token})
+        mock_gge.assert_called_once_with(self.context, 'RegionTwo', 'public')
+        action.set_image_attribute.assert_called_once_with(
+            disk_format='qcow2',
+            container_format='bare')
+        action.set_image_extra_properties.assert_called_once_with({
+            'hw:numa_nodes': '2',
+            'os_hash': 'hash'
+        })
+
+    def test_revert_rollback_metadata_value(self):
+        action = self.wrapper.__enter__.return_value
+        task = import_flow._ImportMetadata(TASK_ID1, TASK_TYPE,
+                                           self.context, self.wrapper,
+                                           self.import_req)
+        task.properties = {'prop1': 'value1', 'prop2': 'value2'}
+        task.old_properties = {'prop1': 'orig_val', 'old_prop': 'old_value'}
+        task.old_attributes = {'container_format': 'bare',
+                               'disk_format': 'qcow2'}
+        task.revert(None)
+        action.set_image_attribute.assert_called_once_with(
+            status='queued',
+            container_format='bare',
+            disk_format='qcow2')
+        action.pop_extra_property.assert_called_once_with('prop2')
+        action.set_image_extra_properties.assert_called_once_with(
+            task.old_properties)
diff --git a/glance/tests/unit/async_/flows/test_base_download.py b/glance/tests/unit/async_/flows/test_base_download.py
new file mode 100644
index 0000000000..32fbd5989a
--- /dev/null
+++ b/glance/tests/unit/async_/flows/test_base_download.py
@@ -0,0 +1,207 @@
+# Copyright 2022 OVHCloud
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+from unittest import mock
+
+from glance_store import backend
+from oslo_config import cfg
+from taskflow.types import failure
+
+from glance.async_.flows import api_image_import
+import glance.common.exception
+from glance import domain
+import glance.tests.unit.utils as unit_test_utils
+import glance.tests.utils as test_utils
+
+CONF = cfg.CONF
+
+
+TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
+
+
+class TestBaseDownloadTask(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestBaseDownloadTask, self).setUp()
+
+        self.config(node_staging_uri='/tmp/staging')
+        self.image_repo = mock.MagicMock()
+        self.image_id = mock.MagicMock()
+        self.uri = mock.MagicMock()
+        self.plugin_name = 'FakeBaseDownload'
+        self.task_factory = domain.TaskFactory()
+
+        task_input = {
+            "import_req": {
+                'method': {
+                    'name': 'web_download',
+                    'uri': 'http://cloud.foo/image.qcow2'
+                }
+            }
+        }
+        task_ttl = CONF.task.task_time_to_live
+
+        self.task_type = 'import'
+        request_id = 'fake_request_id'
+        user_id = 'fake_user'
+        self.task = self.task_factory.new_task(self.task_type, TENANT1,
+                                               self.image_id, user_id,
+                                               request_id,
+                                               task_time_to_live=task_ttl,
+                                               task_input=task_input)
+
+        self.task_id = self.task.task_id
+        self.action_wrapper = api_image_import.ImportActionWrapper(
+            self.image_repo, self.image_id, self.task_id)
+        self.image_repo.get.return_value = mock.MagicMock(
+            extra_properties={'os_glance_import_task': self.task_id})
+        self.base_download_task = unit_test_utils.FakeBaseDownloadPlugin(
+            self.task.task_id, self.task_type, self.action_wrapper,
+            ['foo'], self.plugin_name)
+        self.base_download_task._path = "/path/to_downloaded_data"
+
+    def test_base_download_node_staging_uri_is_none(self):
+        self.config(node_staging_uri=None)
+        self.assertRaises(glance.common.exception.BadTaskConfiguration,
+                          unit_test_utils.FakeBaseDownloadPlugin,
+                          self.task.task_id, self.task_type, self.uri,
+                          self.action_wrapper, ['foo'])
+
+    @mock.patch.object(cfg.ConfigOpts, "set_override")
+    def test_base_download_node_store_initialization_failed(
+            self, mock_override):
+        with mock.patch.object(backend, '_load_store') as mock_load_store:
+            mock_load_store.return_value = None
+            self.assertRaises(glance.common.exception.BadTaskConfiguration,
+                              unit_test_utils.FakeBaseDownloadPlugin,
+                              self.task.task_id, self.task_type, self.uri,
+                              self.action_wrapper, ['foo'])
+            mock_override.assert_called()
+
+    def test_base_download_delete_staging_image_not_exist(self):
+        staging_path = "file:///tmp/staging/temp-image"
+        delete_from_fs_task = api_image_import._DeleteFromFS(
+            self.task.task_id, self.task_type)
+        with mock.patch.object(os.path, "exists") as mock_exists:
+            mock_exists.return_value = False
+            with mock.patch.object(os, "unlink") as mock_unlik:
+                delete_from_fs_task.execute(staging_path)
+
+                self.assertEqual(1, mock_exists.call_count)
+                self.assertEqual(0, mock_unlik.call_count)
+
+    @mock.patch.object(os.path, "exists")
+    def test_base_download_delete_staging_image_failed(self, mock_exists):
+        mock_exists.return_value = True
+        staging_path = "file:///tmp/staging/temp-image"
+        delete_from_fs_task = api_image_import._DeleteFromFS(
+            self.task.task_id, self.task_type)
+        with mock.patch.object(os, "unlink") as mock_unlink:
+            try:
+                delete_from_fs_task.execute(staging_path)
+            except OSError:
+                self.assertEqual(1, mock_unlink.call_count)
+
+            self.assertEqual(1, mock_exists.call_count)
+
+    @mock.patch.object(os.path, "exists")
+    def test_base_download_delete_staging_image_succeed(self, mock_exists):
+        mock_exists.return_value = True
+        staging_path = "file:///tmp/staging/temp-image"
+        delete_from_fs_task = api_image_import._DeleteFromFS(
+            self.task.task_id, self.task_type)
+        with mock.patch.object(os, "unlink") as mock_unlik:
+            delete_from_fs_task.execute(staging_path)
+            self.assertEqual(1, mock_exists.call_count)
+            self.assertEqual(1, mock_unlik.call_count)
+
+    @mock.patch(
+        "glance.async_.flows._internal_plugins.base_download.store_api")
+    def test_base_download_revert_with_failure(self, mock_store_api):
+        image = self.image_repo.get.return_value
+        image.extra_properties['os_glance_importing_to_stores'] = 'foo'
+        image.extra_properties['os_glance_failed_import'] = ''
+
+        self.base_download_task.execute = mock.MagicMock(
+            side_effect=glance.common.exception.ImportTaskError)
+        self.base_download_task.revert(None)
+        mock_store_api.delete_from_backend.assert_called_once_with(
+            "/path/to_downloaded_data")
+        self.assertEqual(1, self.image_repo.save.call_count)
+        self.assertEqual(
+            '', image.extra_properties['os_glance_importing_to_stores'])
+        self.assertEqual(
+            'foo', image.extra_properties['os_glance_failed_import'])
+
+    @mock.patch(
+        "glance.async_.flows._internal_plugins.base_download.store_api")
+    def test_base_download_revert_without_failure_multi_store(self,
+                                                              mock_store_api):
+        enabled_backends = {
+            'fast': 'file',
+            'cheap': 'file'
+        }
+        self.config(enabled_backends=enabled_backends)
+
+        self.base_download_task.revert("/path/to_downloaded_data")
+        mock_store_api.delete.assert_called_once_with(
+            "/path/to_downloaded_data", None)
+
+    @mock.patch(
+        "glance.async_.flows._internal_plugins.base_download.store_api")
+    def test_base_download_revert_with_failure_without_path(self,
+                                                            mock_store_api):
+        image = self.image_repo.get.return_value
+        image.status = 'importing'
+        image.extra_properties['os_glance_importing_to_stores'] = 'foo'
+        image.extra_properties['os_glance_failed_import'] = ''
+        result = failure.Failure.from_exception(
+            glance.common.exception.ImportTaskError())
+
+        self.base_download_task._path = None
+        self.base_download_task.revert(result)
+        mock_store_api.delete_from_backend.assert_not_called()
+
+        # NOTE(danms): Since we told revert that we were the problem,
+        # we should have updated the image status and moved the stores
+        # to the failed list.
+        self.image_repo.save.assert_called_once_with(image, 'importing')
+        self.assertEqual('queued', image.status)
+        self.assertEqual(
+            '', image.extra_properties['os_glance_importing_to_stores'])
+        self.assertEqual(
+            'foo', image.extra_properties['os_glance_failed_import'])
+
+    @mock.patch(
+        "glance.async_.flows._internal_plugins.base_download.store_api")
+    def test_base_download_revert_with_failure_with_path(self, mock_store_api):
+        result = failure.Failure.from_exception(
+            glance.common.exception.ImportTaskError())
+
+        self.base_download_task.revert(result)
+        mock_store_api.delete_from_backend.assert_called_once_with(
+            "/path/to_downloaded_data")
+
+    @mock.patch(
+        "glance.async_.flows._internal_plugins.base_download.store_api")
+    def test_base_download_delete_fails_on_revert(self, mock_store_api):
+        result = failure.Failure.from_exception(
+            glance.common.exception.ImportTaskError())
+        mock_store_api.delete_from_backend.side_effect = Exception
+
+        # this will verify that revert does not break because of failure
+        # while deleting data in staging area
+        self.base_download_task.revert(result)
diff --git a/glance/tests/unit/async_/flows/test_glance_download.py b/glance/tests/unit/async_/flows/test_glance_download.py
new file mode 100644
index 0000000000..5a391e50d9
--- /dev/null
+++ b/glance/tests/unit/async_/flows/test_glance_download.py
@@ -0,0 +1,167 @@
+# Copyright 2022 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+import urllib.error
+
+from glance_store._drivers import filesystem
+from oslo_config import cfg
+from oslo_utils.fixture import uuidsentinel
+
+from glance.async_.flows._internal_plugins import glance_download
+from glance.async_.flows import api_image_import
+import glance.common.exception
+import glance.context
+from glance import domain
+import glance.tests.utils as test_utils
+
+CONF = cfg.CONF
+
+
+TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
+
+
+class TestGlanceDownloadTask(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestGlanceDownloadTask, self).setUp()
+
+        self.config(node_staging_uri='/tmp/staging')
+        self.image_repo = mock.MagicMock()
+        self.image_id = mock.MagicMock()
+        self.uri = mock.MagicMock()
+        self.task_factory = domain.TaskFactory()
+        self.context = glance.context.RequestContext(tenant=TENANT1,
+                                                     auth_token='token')
+        task_input = {
+            "import_req": {
+                'method': {
+                    'name': 'glance-download',
+                    'glance_image_id': uuidsentinel.remote_image,
+                    'glance_region': 'RegionTwo',
+                    'glance_service_interface': 'public',
+                }
+            }
+        }
+        task_ttl = CONF.task.task_time_to_live
+
+        self.task_type = 'import'
+        request_id = 'fake_request_id'
+        user_id = 'fake_user'
+        self.task = self.task_factory.new_task(self.task_type, TENANT1,
+                                               self.image_id, user_id,
+                                               request_id,
+                                               task_time_to_live=task_ttl,
+                                               task_input=task_input)
+
+        self.task_id = self.task.task_id
+        self.action_wrapper = api_image_import.ImportActionWrapper(
+            self.image_repo, self.image_id, self.task_id)
+        self.image_repo.get.return_value = mock.MagicMock(
+            extra_properties={'os_glance_import_task': self.task_id})
+
+    @mock.patch.object(filesystem.Store, 'add')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_glance_download(self, mock_gge, mock_add):
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        glance_download_task = glance_download._DownloadGlanceImage(
+            self.context, self.task.task_id, self.task_type,
+            self.action_wrapper, ['foo'],
+            'RegionTwo', uuidsentinel.remote_image, 'public')
+        with mock.patch('urllib.request') as mock_request:
+            mock_add.return_value = ["path", 12345]
+            self.assertEqual(glance_download_task.execute(12345), "path")
+            mock_add.assert_called_once_with(
+                self.image_id,
+                mock_request.urlopen.return_value, 0)
+            mock_request.Request.assert_called_once_with(
+                'https://other.cloud.foo/image/v2/images/%s/file' % (
+                    uuidsentinel.remote_image),
+                headers={'X-Auth-Token': self.context.auth_token})
+        mock_gge.assert_called_once_with(self.context, 'RegionTwo', 'public')
+
+    @mock.patch.object(filesystem.Store, 'add')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_glance_download_failed(self, mock_gge, mock_add):
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        glance_download_task = glance_download._DownloadGlanceImage(
+            self.context, self.task.task_id, self.task_type,
+            self.action_wrapper, ['foo'],
+            'RegionTwo', uuidsentinel.remote_image, 'public')
+        with mock.patch('urllib.request') as mock_request:
+            mock_request.urlopen.side_effect = urllib.error.HTTPError(
+                '/file', 400, 'Test Fail', {}, None)
+            self.assertRaises(urllib.error.HTTPError,
+                              glance_download_task.execute,
+                              12345)
+            mock_add.assert_not_called()
+            mock_request.Request.assert_called_once_with(
+                'https://other.cloud.foo/image/v2/images/%s/file' % (
+                    uuidsentinel.remote_image),
+                headers={'X-Auth-Token': self.context.auth_token})
+        mock_gge.assert_called_once_with(self.context, 'RegionTwo', 'public')
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_glance_download_no_glance_endpoint(self, mock_gge, mock_request):
+        mock_gge.side_effect = glance.common.exception.GlanceEndpointNotFound(
+            region='RegionTwo',
+            interface='public')
+        glance_download_task = glance_download._DownloadGlanceImage(
+            self.context, self.task.task_id, self.task_type,
+            self.action_wrapper, ['foo'],
+            'RegionTwo', uuidsentinel.remote_image, 'public')
+        self.assertRaises(glance.common.exception.GlanceEndpointNotFound,
+                          glance_download_task.execute, 12345)
+        mock_request.assert_not_called()
+
+    @mock.patch.object(filesystem.Store, 'add')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_glance_download_size_mismatch(self, mock_gge, mock_add):
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        glance_download_task = glance_download._DownloadGlanceImage(
+            self.context, self.task.task_id, self.task_type,
+            self.action_wrapper, ['foo'],
+            'RegionTwo', uuidsentinel.remote_image, 'public')
+        with mock.patch('urllib.request') as mock_request:
+            mock_add.return_value = ["path", 1]
+            self.assertRaises(glance.common.exception.ImportTaskError,
+                              glance_download_task.execute, 12345)
+            mock_add.assert_called_once_with(
+                self.image_id,
+                mock_request.urlopen.return_value, 0)
+            mock_request.Request.assert_called_once_with(
+                'https://other.cloud.foo/image/v2/images/%s/file' % (
+                    uuidsentinel.remote_image),
+                headers={'X-Auth-Token': self.context.auth_token})
+        mock_gge.assert_called_once_with(self.context, 'RegionTwo', 'public')
+
+    @mock.patch('urllib.request')
+    @mock.patch('glance.common.utils.validate_import_uri')
+    @mock.patch('glance.async_.utils.get_glance_endpoint')
+    def test_glance_download_wrong_download_url(self, mock_gge, mock_validate,
+                                                mock_request):
+        mock_validate.return_value = False
+        mock_gge.return_value = 'https://other.cloud.foo/image'
+        glance_download_task = glance_download._DownloadGlanceImage(
+            self.context, self.task.task_id, self.task_type,
+            self.action_wrapper, ['foo'],
+            'RegionTwo', uuidsentinel.remote_image, 'public')
+        self.assertRaises(glance.common.exception.ImportTaskError,
+                          glance_download_task.execute, 12345)
+        mock_request.assert_not_called()
+        mock_validate.assert_called_once_with(
+            'https://other.cloud.foo/image/v2/images/%s/file' % (
+                uuidsentinel.remote_image))
diff --git a/glance/tests/unit/async_/flows/test_web_download.py b/glance/tests/unit/async_/flows/test_web_download.py
index c03992d874..997fcf90a7 100644
--- a/glance/tests/unit/async_/flows/test_web_download.py
+++ b/glance/tests/unit/async_/flows/test_web_download.py
@@ -13,13 +13,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import os
 from unittest import mock
 
 from glance_store._drivers import filesystem
-from glance_store import backend
 from oslo_config import cfg
-from taskflow.types import failure
 
 from glance.async_.flows._internal_plugins import web_download
 from glance.async_.flows import api_image_import
@@ -67,217 +64,84 @@ class TestWebDownloadTask(test_utils.BaseTestCase):
         self.task_id = self.task.task_id
         self.action_wrapper = api_image_import.ImportActionWrapper(
             self.image_repo, self.image_id, self.task_id)
+        self.web_download_task = web_download._WebDownload(
+            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
+            ['foo'])
         self.image_repo.get.return_value = mock.MagicMock(
             extra_properties={'os_glance_import_task': self.task_id})
 
     @mock.patch.object(filesystem.Store, 'add')
     def test_web_download(self, mock_add):
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
         with mock.patch.object(script_utils,
                                'get_image_data_iter') as mock_iter:
             mock_add.return_value = ["path", 4]
             mock_iter.return_value.headers = {}
-            self.assertEqual(web_download_task.execute(), "path")
+            self.assertEqual(self.web_download_task.execute(), "path")
             mock_add.assert_called_once_with(self.image_id,
                                              mock_iter.return_value, 0)
 
     @mock.patch.object(filesystem.Store, 'add')
     def test_web_download_with_content_length(self, mock_add):
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
         with mock.patch.object(script_utils,
                                'get_image_data_iter') as mock_iter:
             mock_iter.return_value.headers = {'content-length': '4'}
             mock_add.return_value = ["path", 4]
-            self.assertEqual(web_download_task.execute(), "path")
+            self.assertEqual(self.web_download_task.execute(), "path")
             mock_add.assert_called_once_with(self.image_id,
                                              mock_iter.return_value, 0)
 
     @mock.patch.object(filesystem.Store, 'add')
     def test_web_download_with_invalid_content_length(self, mock_add):
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
         with mock.patch.object(script_utils,
                                'get_image_data_iter') as mock_iter:
             mock_iter.return_value.headers = {'content-length': "not_valid"}
             mock_add.return_value = ["path", 4]
-            self.assertEqual(web_download_task.execute(), "path")
+            self.assertEqual(self.web_download_task.execute(), "path")
             mock_add.assert_called_once_with(self.image_id,
                                              mock_iter.return_value, 0)
 
     @mock.patch.object(filesystem.Store, 'add')
     def test_web_download_fails_when_data_size_different(self, mock_add):
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
         with mock.patch.object(script_utils,
                                'get_image_data_iter') as mock_iter:
             mock_iter.return_value.headers = {'content-length': '4'}
             mock_add.return_value = ["path", 3]
             self.assertRaises(
                 glance.common.exception.ImportTaskError,
-                web_download_task.execute)
-
-    def test_web_download_node_staging_uri_is_none(self):
-        self.config(node_staging_uri=None)
-        self.assertRaises(glance.common.exception.BadTaskConfiguration,
-                          web_download._WebDownload, self.task.task_id,
-                          self.task_type, self.uri, self.action_wrapper,
-                          ['foo'])
-
-    @mock.patch.object(cfg.ConfigOpts, "set_override")
-    def test_web_download_node_store_initialization_failed(self,
-                                                           mock_override):
-        with mock.patch.object(backend, '_load_store') as mock_load_store:
-            mock_load_store.return_value = None
-            self.assertRaises(glance.common.exception.BadTaskConfiguration,
-                              web_download._WebDownload, self.task.task_id,
-                              self.task_type, self.uri, self.action_wrapper,
-                              ['foo'])
-            mock_override.assert_called()
+                self.web_download_task.execute)
 
     def test_web_download_failed(self):
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
         with mock.patch.object(script_utils,
                                "get_image_data_iter") as mock_iter:
             mock_iter.side_effect = glance.common.exception.NotFound
             self.assertRaises(glance.common.exception.NotFound,
-                              web_download_task.execute)
-
-    def test_web_download_delete_staging_image_not_exist(self):
-        staging_path = "file:///tmp/staging/temp-image"
-        delete_from_fs_task = api_image_import._DeleteFromFS(
-            self.task.task_id, self.task_type)
-        with mock.patch.object(os.path, "exists") as mock_exists:
-            mock_exists.return_value = False
-            with mock.patch.object(os, "unlink") as mock_unlik:
-                delete_from_fs_task.execute(staging_path)
-
-                self.assertEqual(1, mock_exists.call_count)
-                self.assertEqual(0, mock_unlik.call_count)
-
-    @mock.patch.object(os.path, "exists")
-    def test_web_download_delete_staging_image_failed(self, mock_exists):
-        mock_exists.return_value = True
-        staging_path = "file:///tmp/staging/temp-image"
-        delete_from_fs_task = api_image_import._DeleteFromFS(
-            self.task.task_id, self.task_type)
-        with mock.patch.object(os, "unlink") as mock_unlink:
-            try:
-                delete_from_fs_task.execute(staging_path)
-            except OSError:
-                self.assertEqual(1, mock_unlink.call_count)
-
-            self.assertEqual(1, mock_exists.call_count)
-
-    @mock.patch.object(os.path, "exists")
-    def test_web_download_delete_staging_image_succeed(self, mock_exists):
-        mock_exists.return_value = True
-        staging_path = "file:///tmp/staging/temp-image"
-        delete_from_fs_task = api_image_import._DeleteFromFS(
-            self.task.task_id, self.task_type)
-        with mock.patch.object(os, "unlink") as mock_unlik:
-            delete_from_fs_task.execute(staging_path)
-            self.assertEqual(1, mock_exists.call_count)
-            self.assertEqual(1, mock_unlik.call_count)
+                              self.web_download_task.execute)
 
     @mock.patch.object(filesystem.Store, 'add')
-    @mock.patch("glance.async_.flows._internal_plugins.web_download.store_api")
-    def test_web_download_revert_with_failure(self, mock_store_api,
-                                              mock_add):
-        image = self.image_repo.get.return_value
-        image.extra_properties['os_glance_importing_to_stores'] = 'foo'
-        image.extra_properties['os_glance_failed_import'] = ''
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
+    def test_web_download_check_content_length(self, mock_add):
         with mock.patch.object(script_utils,
                                'get_image_data_iter') as mock_iter:
+            mock_add.return_value = ["path", 4]
             mock_iter.return_value.headers = {'content-length': '4'}
-            mock_add.return_value = "/path/to_downloaded_data", 3
-            self.assertRaises(
-                glance.common.exception.ImportTaskError,
-                web_download_task.execute)
+            self.assertEqual(self.web_download_task.execute(), "path")
+            mock_add.assert_called_once_with(self.image_id,
+                                             mock_iter.return_value, 0)
 
-        web_download_task.revert(None)
-        mock_store_api.delete_from_backend.assert_called_once_with(
-            "/path/to_downloaded_data")
-        # NOTE(danms): Since we told revert that we were not at fault,
-        # we should not have updated the image.
-        self.image_repo.save.assert_not_called()
-        self.assertEqual(
-            'foo', image.extra_properties['os_glance_importing_to_stores'])
-        self.assertEqual(
-            '', image.extra_properties['os_glance_failed_import'])
+    @mock.patch.object(filesystem.Store, 'add')
+    def test_web_download_invalid_content_length(self, mock_add):
+        with mock.patch.object(script_utils,
+                               'get_image_data_iter') as mock_iter:
+            mock_add.return_value = ["path", 4]
+            mock_iter.return_value.headers = {'content-length': 'not_valid'}
+            self.assertEqual(self.web_download_task.execute(), "path")
+            mock_add.assert_called_once_with(self.image_id,
+                                             mock_iter.return_value, 0)
 
-    @mock.patch("glance.async_.flows._internal_plugins.web_download.store_api")
-    def test_web_download_revert_without_failure_multi_store(self,
-                                                             mock_store_api):
-        enabled_backends = {
-            'fast': 'file',
-            'cheap': 'file'
-        }
-        self.config(enabled_backends=enabled_backends)
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
-        web_download_task._path = "/path/to_downloaded_data"
-        web_download_task.revert("/path/to_downloaded_data")
-        mock_store_api.delete.assert_called_once_with(
-            "/path/to_downloaded_data", None)
-
-    @mock.patch("glance.async_.flows._internal_plugins.web_download.store_api")
-    def test_web_download_revert_with_failure_without_path(self,
-                                                           mock_store_api):
-        image = self.image_repo.get.return_value
-        image.status = 'importing'
-        image.extra_properties['os_glance_importing_to_stores'] = 'foo'
-        image.extra_properties['os_glance_failed_import'] = ''
-        result = failure.Failure.from_exception(
-            glance.common.exception.ImportTaskError())
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
-        web_download_task.revert(result)
-        mock_store_api.delete_from_backend.assert_not_called()
-
-        # NOTE(danms): Since we told revert that we were the problem,
-        # we should have updated the image status and moved the stores
-        # to the failed list.
-        self.image_repo.save.assert_called_once_with(image, 'importing')
-        self.assertEqual('queued', image.status)
-        self.assertEqual(
-            '', image.extra_properties['os_glance_importing_to_stores'])
-        self.assertEqual(
-            'foo', image.extra_properties['os_glance_failed_import'])
-
-    @mock.patch("glance.async_.flows._internal_plugins.web_download.store_api")
-    def test_web_download_revert_with_failure_with_path(self, mock_store_api):
-        result = failure.Failure.from_exception(
-            glance.common.exception.ImportTaskError())
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
-        web_download_task._path = "/path/to_downloaded_data"
-        web_download_task.revert(result)
-        mock_store_api.delete_from_backend.assert_called_once_with(
-            "/path/to_downloaded_data")
-
-    @mock.patch("glance.async_.flows._internal_plugins.web_download.store_api")
-    def test_web_download_delete_fails_on_revert(self, mock_store_api):
-        result = failure.Failure.from_exception(
-            glance.common.exception.ImportTaskError())
-        mock_store_api.delete_from_backend.side_effect = Exception
-        web_download_task = web_download._WebDownload(
-            self.task.task_id, self.task_type, self.uri, self.action_wrapper,
-            ['foo'])
-        web_download_task._path = "/path/to_downloaded_data"
-        # this will verify that revert does not break because of failure
-        # while deleting data in staging area
-        web_download_task.revert(result)
+    @mock.patch.object(filesystem.Store, 'add')
+    def test_web_download_wrong_content_length(self, mock_add):
+        with mock.patch.object(script_utils,
+                               'get_image_data_iter') as mock_iter:
+            mock_add.return_value = ["path", 2]
+            mock_iter.return_value.headers = {'content-length': '4'}
+            self.assertRaises(glance.common.exception.ImportTaskError,
+                              self.web_download_task.execute)
diff --git a/glance/tests/unit/utils.py b/glance/tests/unit/utils.py
index ccf5107b33..3cbb51dbc2 100644
--- a/glance/tests/unit/utils.py
+++ b/glance/tests/unit/utils.py
@@ -21,6 +21,7 @@ import urllib
 
 from oslo_config import cfg
 
+from glance.async_.flows._internal_plugins import base_download
 from glance.common import exception
 from glance.common import store_utils
 from glance.common import wsgi
@@ -373,3 +374,8 @@ class FakeTask(object):
     def fail(self, message):
         self.message = message
         self._status = 'failure'
+
+
+class FakeBaseDownloadPlugin(base_download.BaseDownload):
+    def execute(self):
+        pass
diff --git a/glance/tests/unit/v2/test_discovery_image_import.py b/glance/tests/unit/v2/test_discovery_image_import.py
index 14b68fbd84..949da8c077 100644
--- a/glance/tests/unit/v2/test_discovery_image_import.py
+++ b/glance/tests/unit/v2/test_discovery_image_import.py
@@ -36,7 +36,7 @@ class TestInfoControllers(test_utils.BaseTestCase):
         # TODO(rosmaita): change this when import methods are
         # listed in the config file
         import_methods = ['glance-direct', 'web-download',
-                          'copy-image']
+                          'copy-image', 'glance-download']
 
         req = unit_test_utils.get_fake_request()
         output = self.controller.get_image_import(req)
diff --git a/glance/tests/unit/v2/test_images_resource.py b/glance/tests/unit/v2/test_images_resource.py
index 67a6f36492..203092554a 100644
--- a/glance/tests/unit/v2/test_images_resource.py
+++ b/glance/tests/unit/v2/test_images_resource.py
@@ -911,6 +911,45 @@ class TestImagesController(base.IsolatedUnitTest):
                               {'method': {'name': 'web-download',
                                           'uri': 'fake_uri'}})
 
+    def test_image_import_raises_bad_request_for_glance_download_missing_input(
+            self):
+        request = unit_test_utils.get_fake_request()
+        with mock.patch.object(
+                glance.notifier.ImageRepoProxy, 'get') as mock_get:
+            mock_get.return_value = FakeImage(status='queued')
+            self.assertRaises(webob.exc.HTTPBadRequest,
+                              self.controller.import_image, request, UUID4,
+                              {'method': {'name': 'glance-download'}})
+
+    def test_image_import_raise_bad_request_wrong_id_for_glance_download(
+            self):
+        request = unit_test_utils.get_fake_request()
+        with mock.patch.object(
+                glance.notifier.ImageRepoProxy, 'get') as mock_get:
+            mock_get.return_value = FakeImage(status='queued')
+            self.assertRaises(webob.exc.HTTPBadRequest,
+                              self.controller.import_image, request, UUID4,
+                              {'method': {'name': 'glance-download',
+                                          'glance_image_id': 'fake_id',
+                                          'glance_region': 'REGION4'}})
+
+    @mock.patch.object(glance.domain.TaskFactory, 'new_task')
+    @mock.patch.object(glance.notifier.ImageRepoProxy, 'get')
+    def test_image_import_add_default_service_endpoint_for_glance_download(
+            self, mock_get, mock_nt):
+        request = unit_test_utils.get_fake_request()
+        mock_get.return_value = FakeImage(status='queued')
+        body = {'method': {'name': 'glance-download',
+                           'glance_image_id': UUID4,
+                           'glance_region': 'REGION2'}}
+        self.controller.import_image(request, UUID4, body)
+        expected_req = {'method': {'name': 'glance-download',
+                                   'glance_image_id': UUID4,
+                                   'glance_region': 'REGION2',
+                                   'glance_service_interface': 'public'}}
+        self.assertEqual(expected_req,
+                         mock_nt.call_args.kwargs['task_input']['import_req'])
+
     @mock.patch('glance.context.get_ksa_client')
     def test_image_import_proxies(self, mock_client):
         # Make sure that we proxy to the remote side when we need to
@@ -4865,7 +4904,7 @@ class TestImagesDeserializer(test_utils.BaseTestCase):
         request.body = jsonutils.dump_as_bytes(import_body)
         return request
 
-    KNOWN_IMPORT_METHODS = ['glance-direct', 'web-download']
+    KNOWN_IMPORT_METHODS = ['glance-direct', 'web-download', 'glance-download']
 
     def test_import_image_invalid_import_method(self):
         # Bug 1754634: make sure that what's considered valid
diff --git a/releasenotes/notes/add-glance-download-method-be6d9e927b8b0a43.yaml b/releasenotes/notes/add-glance-download-method-be6d9e927b8b0a43.yaml
new file mode 100644
index 0000000000..7dcb1e79fa
--- /dev/null
+++ b/releasenotes/notes/add-glance-download-method-be6d9e927b8b0a43.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Glance to glance image import plugin. With this release users can import
+    an image from an other glance server from an other opensatck region.
+    The two glance services must use the same keystone service. The feature
+    use the same keystone authentication token on both glance services and
+    copy by default container_format, disk_format and customizable properties
+    from source image ``['hw_', 'trait:', 'os_distro', 'os_secure_boot',
+    'os_type']``
diff --git a/setup.cfg b/setup.cfg
index 611f5adc2c..ae28080563 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -82,6 +82,7 @@ glance.image_import.plugins =
 glance.image_import.internal_plugins =
     web_download = glance.async_.flows._internal_plugins.web_download:get_flow
     copy_image = glance.async_.flows._internal_plugins.copy_image:get_flow
+    glance_download = glance.async_.flows._internal_plugins.glance_download:get_flow
 
 
 [egg_info]