# Copyright 2014 OpenStack Foundation
# 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.
__all__ = [
'get_task',
'unpack_task_input',
'set_base_image_properties',
'validate_location_uri',
'validate_legacy_import_from_uri',
'get_image_data_iter',
'open_external_uri',
'SafeRedirectHandler',
]
import http.client
import urllib
import urllib.error
import urllib.request
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import timeutils
from glance.common import exception
from glance.common import utils as common_utils
from glance.i18n import _, _LE
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
[docs]def get_task(task_repo, task_id):
"""Gets a TaskProxy object.
:param task_repo: TaskRepo object used to perform DB operations
:param task_id: ID of the Task
"""
task = None
try:
task = task_repo.get(task_id)
except exception.NotFound:
msg = _LE('Task not found for task_id %s') % task_id
LOG.exception(msg)
return task
[docs]def set_base_image_properties(properties=None):
"""Sets optional base properties for creating Image.
:param properties: Input dict to set some base properties
"""
if isinstance(properties, dict) and len(properties) == 0:
# TODO(nikhil): We can make these properties configurable while
# implementing the pipeline logic for the scripts. The below shown
# are placeholders to show that the scripts work on 'devstack'
# environment.
properties['disk_format'] = 'qcow2'
properties['container_format'] = 'bare'
[docs]def validate_location_uri(location):
"""Validate location uri into acceptable format.
:param location: Location uri to be validated
"""
if not location:
raise exception.BadStoreUri(_('Invalid location: %s') % location)
elif location.startswith(('http://', 'https://')):
return location
# NOTE: file type uri is being avoided for security reasons,
# see LP bug #942118 #1400966.
elif location.startswith(("file:///", "filesystem:///")):
msg = _("File based imports are not allowed. Please use a non-local "
"source of image data.")
# NOTE: raise BadStoreUri and let the encompassing block save the error
# msg in the task.message.
raise exception.BadStoreUri(msg)
else:
# TODO(nikhil): add other supported uris
supported = ['http', ]
msg = _("The given uri is not valid. Please specify a "
"valid uri from the following list of supported uri "
"%(supported)s") % {'supported': supported}
raise urllib.error.URLError(msg)
[docs]def validate_legacy_import_from_uri(location):
"""Validate legacy ``import_from`` URI (scheme + import filter)."""
uri = validate_location_uri(location)
if not common_utils.validate_uri(uri):
msg = (_("URI for legacy import task does not pass filtering: %s") %
uri)
raise exception.Invalid(msg)
return uri
[docs]class SafeRedirectHandler(urllib.request.HTTPRedirectHandler):
"""HTTP redirect handler that validates redirect destinations."""
[docs] def redirect_request(self, req, fp, code, msg, headers, newurl):
if not common_utils.validate_uri(newurl):
msg = (_("Redirect to disallowed URL: %s") % newurl)
raise exception.InvalidRedirect(msg)
return super().redirect_request(req, fp, code, msg, headers, newurl)
class _PinnedHTTPConnection(http.client.HTTPConnection):
"""HTTP connection that connects to a pre-validated IP address."""
def __init__(self, host, port=None, *, pinned_ip=None, **kwargs):
self._pinned_ip = pinned_ip
super().__init__(host, port=port, **kwargs)
def connect(self):
if self._pinned_ip:
self.sock = self._create_connection(
(self._pinned_ip, self.port), self.timeout,
self.source_address)
return
super().connect()
class _PinnedHTTPSConnection(http.client.HTTPSConnection):
"""HTTPS connection that connects to a pre-validated IP address."""
def __init__(self, host, port=None, *, pinned_ip=None, **kwargs):
self._pinned_ip = pinned_ip
super().__init__(host, port=port, **kwargs)
def connect(self):
if self._pinned_ip:
sock = self._create_connection(
(self._pinned_ip, self.port), self.timeout,
self.source_address)
if self._tunnel_host:
self.sock = sock
self._tunnel()
else:
self.sock = self._context.wrap_socket(
sock, server_hostname=self.host)
return
super().connect()
def _pinned_ip_for_request(req):
"""Return a pinned destination IP for an external HTTP(S) URI."""
try:
return common_utils.get_validated_address(req.full_url)
except ValueError as exc:
msg = (_("URI does not pass filtering: %s") % req.full_url)
LOG.debug("%s (%s)", msg, exc)
raise exception.Invalid(msg)
class _ValidatedExternalHTTPHandler(urllib.request.HTTPHandler):
"""Open HTTP URIs using validated, pinned destination addresses."""
def http_open(self, req):
pinned_ip = _pinned_ip_for_request(req)
return self.do_open(
lambda host, **kwargs: _PinnedHTTPConnection(
host, pinned_ip=pinned_ip, **kwargs),
req)
class _ValidatedExternalHTTPSHandler(urllib.request.HTTPSHandler):
"""Open HTTPS URIs using validated, pinned destination addresses."""
def https_open(self, req):
pinned_ip = _pinned_ip_for_request(req)
return self.do_open(
lambda host, **kwargs: _PinnedHTTPSConnection(
host, pinned_ip=pinned_ip, **kwargs),
req)
[docs]def open_external_uri(uri_or_request):
"""Open an external URL string or Request with validation and IP pinning.
The destination is re-checked at fetch time and the TCP connection is
pinned to the validated address so DNS cannot rebind mid-download.
Suitable for import downloads and other outbound HTTP(S) fetches that
need the same host filtering protections.
"""
opener = urllib.request.build_opener(
SafeRedirectHandler,
_ValidatedExternalHTTPHandler,
_ValidatedExternalHTTPSHandler,
)
return opener.open(uri_or_request)
[docs]def get_image_data_iter(uri):
"""Returns iterable object either for local file or uri
:param uri: uri (remote or local) to the datasource we want to iterate
Remote HTTP(S) URIs are validated and resolved at fetch time. The
connection is pinned to the validated destination address to close the
DNS rebinding window between API validation and worker download.
"""
if uri.startswith("file://"):
uri = uri.split("file://")[-1]
# NOTE(flaper87): The caller of this function expects to have
# an iterable object. FileObjects in python are iterable, therefore
# we are returning it as is.
# The file descriptor will be eventually cleaned up by the garbage
# collector once its ref-count is dropped to 0. That is, when there
# won't be any references pointing to this file.
#
# We're not using StringIO or other tools to avoid reading everything
# into memory. Some images may be quite heavy.
data = open(uri, "rb")
return _size_limited_reader(data)
return _size_limited_reader(open_external_uri(uri))
def _size_limited_reader(data):
"""Wrap image data with CooperativeReader and LimitingReader.
Preserves headers/close from the underlying object for callers that still
expect a single iterable (stable/2025.2 has not adopted the (data, size)
return shape from newer branches).
"""
limited = common_utils.LimitingReader(
common_utils.CooperativeReader(data),
CONF.image_size_cap)
if hasattr(data, 'headers'):
limited.headers = data.headers
if hasattr(data, 'close'):
limited.close = data.close
return limited
class CallbackIterator(object):
"""A proxy iterator that calls a callback function periodically
This is used to wrap a reading file object and proxy its chunks
through to another caller. Periodically, the callback function
will be called with information about the data processed so far,
allowing for status updating or cancel flag checking. The function
can be called every time we process a chunk, or only after we have
processed a certain amount of data since the last call.
:param source: A source iterator whose content will be proxied
through this object.
:param callback: A function to be called periodically while iterating.
The signature should be fn(chunk_bytes, total_bytes),
where chunk is the number of bytes since the last
call of the callback, and total_bytes is the total amount
copied thus far.
:param min_interval: Limit the calls to callback to only when this many
seconds have elapsed since the last callback (a
close() or final iteration may fire the callback in
less time to ensure completion).
"""
def __init__(self, source, callback, min_interval=None):
self._source = source
self._callback = callback
self._min_interval = min_interval
self._chunk_bytes = 0
self._total_bytes = 0
self._timer = None
@property
def callback_due(self):
"""Indicates if a callback should be made.
If no time-based limit is set, this will always be True.
If a limit is set, then this returns True exactly once,
resetting the timer when it does.
"""
if not self._min_interval:
return True
if not self._timer:
self._timer = timeutils.StopWatch(self._min_interval)
self._timer.start()
if self._timer.expired():
self._timer.restart()
return True
else:
return False
def __iter__(self):
return self
def __next__(self):
try:
chunk = next(self._source)
except StopIteration:
# NOTE(danms): Make sure we call the callback the last
# time if we have processed data since the last one.
self._call_callback(b'', is_last=True)
raise
self._call_callback(chunk)
return chunk
def close(self):
self._call_callback(b'', is_last=True)
if hasattr(self._source, 'close'):
return self._source.close()
def _call_callback(self, chunk, is_last=False):
self._total_bytes += len(chunk)
self._chunk_bytes += len(chunk)
if not self._chunk_bytes:
# NOTE(danms): Never call the callback if we haven't processed
# any data since the last time
return
if is_last or self.callback_due:
# FIXME(danms): Perhaps we should only abort the read if
# the callback raises a known abort exception, otherwise
# log and swallow. Need to figure out what exception
# read() callers would be expecting that we could raise
# from here.
self._callback(self._chunk_bytes, self._total_bytes)
self._chunk_bytes = 0
def read(self, size=None):
chunk = self._source.read(size)
self._call_callback(chunk)
return chunk
