1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / libcloud / storage / drivers /
Filename : azure_blobs.py
back
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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 __future__ import with_statement

import base64
import hashlib
import hmac
import os
import binascii
from datetime import datetime, timedelta

from libcloud.utils.py3 import ET
from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import urlencode
from libcloud.utils.py3 import urlquote
from libcloud.utils.py3 import tostring
from libcloud.utils.py3 import b

from libcloud.utils.xml import fixxpath
from libcloud.utils.files import read_in_chunks
from libcloud.common.types import LibcloudError
from libcloud.common.azure import AzureConnection

from libcloud.storage.base import Object, Container, StorageDriver
from libcloud.storage.types import ContainerIsNotEmptyError
from libcloud.storage.types import ContainerAlreadyExistsError
from libcloud.storage.types import InvalidContainerNameError
from libcloud.storage.types import ContainerDoesNotExistError
from libcloud.storage.types import ObjectDoesNotExistError
from libcloud.storage.types import ObjectHashMismatchError

# Desired number of items in each response inside a paginated request
RESPONSES_PER_REQUEST = 100

# According to the Azure Docs:
# > The block must be less than or equal to 100 MB in size for version
# > 2016-05-31 and later (4 MB for older versions).
# However for performance reasons, using a lower upload chunk size
# usually leads to fewer dropped requests and retries.
AZURE_UPLOAD_CHUNK_SIZE = int(
    os.getenv('LIBCLOUD_AZURE_UPLOAD_CHUNK_SIZE_MB', '4')
) * 1024 * 1024

AZURE_DOWNLOAD_CHUNK_SIZE = int(
    os.getenv('LIBCLOUD_AZURE_DOWNLOAD_CHUNK_SIZE_MB', '4')
) * 1024 * 1024

# The time period (in seconds) for which a lease must be obtained.
# If set as -1, we get an infinite lease, but that is a bad idea. If
# after getting an infinite lease, there was an issue in releasing the
# lease, the object will remain 'locked' forever, unless the lease is
# released using the lease_id (which is not exposed to the user)
AZURE_LEASE_PERIOD = int(
    os.getenv('LIBCLOUD_AZURE_LEASE_PERIOD_SECONDS', '60')
)

AZURE_STORAGE_HOST_SUFFIX = 'blob.core.windows.net'

AZURE_STORAGE_CDN_URL_DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ'

AZURE_STORAGE_CDN_URL_START_MINUTES = float(
    os.getenv('LIBCLOUD_AZURE_STORAGE_CDN_URL_START_MINUTES', '5')
)

AZURE_STORAGE_CDN_URL_EXPIRY_HOURS = float(
    os.getenv('LIBCLOUD_AZURE_STORAGE_CDN_URL_EXPIRY_HOURS', '24')
)


class AzureBlobLease(object):
    """
    A class to help in leasing an azure blob and renewing the lease
    """
    def __init__(self, driver, object_path, use_lease):
        """
        :param driver: The Azure storage driver that is being used
        :type driver: :class:`AzureStorageDriver`

        :param object_path: The path of the object we need to lease
        :type object_path: ``str``

        :param use_lease: Indicates if we must take a lease or not
        :type use_lease: ``bool``
        """
        self.object_path = object_path
        self.driver = driver
        self.use_lease = use_lease
        self.lease_id = None
        self.params = {'comp': 'lease'}

    def renew(self):
        """
        Renew the lease if it is older than a predefined time period
        """
        if self.lease_id is None:
            return

        headers = {'x-ms-lease-action': 'renew',
                   'x-ms-lease-id': self.lease_id,
                   'x-ms-lease-duration': '60'}

        response = self.driver.connection.request(self.object_path,
                                                  headers=headers,
                                                  params=self.params,
                                                  method='PUT')

        if response.status != httplib.OK:
            raise LibcloudError('Unable to obtain lease', driver=self)

    def update_headers(self, headers):
        """
        Update the lease id in the headers
        """
        if self.lease_id:
            headers['x-ms-lease-id'] = self.lease_id

    def __enter__(self):
        if not self.use_lease:
            return self

        headers = {'x-ms-lease-action': 'acquire',
                   'x-ms-lease-duration': '60'}

        response = self.driver.connection.request(self.object_path,
                                                  headers=headers,
                                                  params=self.params,
                                                  method='PUT')

        if response.status == httplib.NOT_FOUND:
            return self
        elif response.status != httplib.CREATED:
            raise LibcloudError('Unable to obtain lease', driver=self)

        self.lease_id = response.headers['x-ms-lease-id']
        return self

    def __exit__(self, type, value, traceback):
        if self.lease_id is None:
            return

        headers = {'x-ms-lease-action': 'release',
                   'x-ms-lease-id': self.lease_id}
        response = self.driver.connection.request(self.object_path,
                                                  headers=headers,
                                                  params=self.params,
                                                  method='PUT')

        if response.status != httplib.OK:
            raise LibcloudError('Unable to release lease', driver=self)


class AzureBlobsConnection(AzureConnection):
    """
    Represents a single connection to Azure Blobs.

    The main Azure Blob Storage service uses a prefix in the hostname to
    distinguish between accounts, e.g. ``theaccount.blob.core.windows.net``.
    However, some custom deployments of the service, such as the Azurite
    emulator, instead use a URL prefix such as ``/theaccount``. To support
    these deployments, the parameter ``account_prefix`` must be set on the
    connection. This is done by instantiating the driver with arguments such
    as ``host='somewhere.tld'`` and ``key='theaccount'``. To specify a custom
    host without an account prefix, e.g. for use-cases where the custom host
    implements an auditing proxy or similar, the driver can be instantiated
    with ``host='theaccount.somewhere.tld'`` and ``key=''``.

    :param account_prefix: Optional prefix identifying the sotrage account.
                           Used when connecting to a custom deployment of the
                           storage service like Azurite or IoT Edge Storage.
    :type account_prefix: ``str``
    """
    def __init__(self, *args, **kwargs):
        self.account_prefix = kwargs.pop('account_prefix', None)
        super(AzureBlobsConnection, self).__init__(*args, **kwargs)

    def morph_action_hook(self, action):
        action = super(AzureBlobsConnection, self).morph_action_hook(action)

        if self.account_prefix is not None:
            action = '/%s%s' % (self.account_prefix, action)

        return action

    API_VERSION = '2018-11-09'


class AzureBlobsStorageDriver(StorageDriver):
    name = 'Microsoft Azure (blobs)'
    website = 'http://windows.azure.com/'
    connectionCls = AzureBlobsConnection
    hash_type = 'md5'
    supports_chunked_encoding = False

    def __init__(self, key, secret=None, secure=True, host=None, port=None,
                 **kwargs):
        self._host_argument_set = bool(host)

        # B64decode() this key and keep it, so that we don't have to do
        # so for every request. Minor performance improvement
        secret = base64.b64decode(b(secret))

        super(AzureBlobsStorageDriver, self).__init__(key=key, secret=secret,
                                                      secure=secure, host=host,
                                                      port=port, **kwargs)

    def _ex_connection_class_kwargs(self):
        result = {}

        # host argument has precedence
        if not self._host_argument_set:
            result['host'] = '%s.%s' % (self.key, AZURE_STORAGE_HOST_SUFFIX)
        else:
            result['account_prefix'] = self.key

        return result

    def _xml_to_container(self, node):
        """
        Converts a container XML node to a container instance

        :param node: XML info of the container
        :type node: :class:`xml.etree.ElementTree.Element`

        :return: A container instance
        :rtype: :class:`Container`
        """

        name = node.findtext(fixxpath(xpath='Name'))
        props = node.find(fixxpath(xpath='Properties'))
        metadata = node.find(fixxpath(xpath='Metadata'))

        extra = {
            'url': node.findtext(fixxpath(xpath='Url')),
            'last_modified': node.findtext(fixxpath(xpath='Last-Modified')),
            'etag': props.findtext(fixxpath(xpath='Etag')),
            'lease': {
                'status': props.findtext(fixxpath(xpath='LeaseStatus')),
                'state': props.findtext(fixxpath(xpath='LeaseState')),
                'duration': props.findtext(fixxpath(xpath='LeaseDuration')),
            },
            'meta_data': {}
        }

        if metadata is not None:
            for meta in list(metadata):
                extra['meta_data'][meta.tag] = meta.text

        return Container(name=name, extra=extra, driver=self)

    def _response_to_container(self, container_name, response):
        """
        Converts a HTTP response to a container instance

        :param container_name: Name of the container
        :type container_name: ``str``

        :param response: HTTP Response
        :type node: L{}

        :return: A container instance
        :rtype: :class:`Container`
        """

        headers = response.headers
        extra = {
            'url': 'http://%s%s' % (response.connection.host,
                                    response.connection.action),
            'etag': headers['etag'],
            'last_modified': headers['last-modified'],
            'lease': {
                'status': headers.get('x-ms-lease-status', None),
                'state': headers.get('x-ms-lease-state', None),
                'duration': headers.get('x-ms-lease-duration', None),
            },
            'meta_data': {}
        }

        for key, value in response.headers.items():
            if key.startswith('x-ms-meta-'):
                key = key.split('x-ms-meta-')[1]
                extra['meta_data'][key] = value

        return Container(name=container_name, extra=extra, driver=self)

    def _xml_to_object(self, container, blob):
        """
        Converts a BLOB XML node to an object instance

        :param container: Instance of the container holding the blob
        :type: :class:`Container`

        :param blob: XML info of the blob
        :type blob: L{}

        :return: An object instance
        :rtype: :class:`Object`
        """

        name = blob.findtext(fixxpath(xpath='Name'))
        props = blob.find(fixxpath(xpath='Properties'))
        metadata = blob.find(fixxpath(xpath='Metadata'))
        etag = props.findtext(fixxpath(xpath='Etag'))
        size = int(props.findtext(fixxpath(xpath='Content-Length')))

        extra = {
            'content_type': props.findtext(fixxpath(xpath='Content-Type')),
            'etag': etag,
            'md5_hash': props.findtext(fixxpath(xpath='Content-MD5')),
            'last_modified': props.findtext(fixxpath(xpath='Last-Modified')),
            'url': blob.findtext(fixxpath(xpath='Url')),
            'hash': props.findtext(fixxpath(xpath='Etag')),
            'lease': {
                'status': props.findtext(fixxpath(xpath='LeaseStatus')),
                'state': props.findtext(fixxpath(xpath='LeaseState')),
                'duration': props.findtext(fixxpath(xpath='LeaseDuration')),
            },
            'content_encoding': props.findtext(fixxpath(
                                               xpath='Content-Encoding')),
            'content_language': props.findtext(fixxpath(
                                               xpath='Content-Language')),
            'blob_type': props.findtext(fixxpath(xpath='BlobType'))
        }

        if extra['md5_hash']:
            value = binascii.hexlify(base64.b64decode(b(extra['md5_hash'])))
            value = value.decode('ascii')
            extra['md5_hash'] = value

        meta_data = {}
        if metadata is not None:
            for meta in list(metadata):
                meta_data[meta.tag] = meta.text

        return Object(name=name, size=size, hash=etag, meta_data=meta_data,
                      extra=extra, container=container, driver=self)

    def _response_to_object(self, object_name, container, response):
        """
        Converts a HTTP response to an object (from headers)

        :param object_name: Name of the object
        :type object_name: ``str``

        :param container: Instance of the container holding the blob
        :type: :class:`Container`

        :param response: HTTP Response
        :type node: L{}

        :return: An object instance
        :rtype: :class:`Object`
        """

        headers = response.headers
        size = int(headers['content-length'])
        etag = headers['etag']

        extra = {
            'url': 'http://%s%s' % (response.connection.host,
                                    response.connection.action),
            'etag': etag,
            'md5_hash': headers.get('content-md5', None),
            'content_type': headers.get('content-type', None),
            'content_language': headers.get('content-language', None),
            'content_encoding': headers.get('content-encoding', None),
            'last_modified': headers['last-modified'],
            'lease': {
                'status': headers.get('x-ms-lease-status', None),
                'state': headers.get('x-ms-lease-state', None),
                'duration': headers.get('x-ms-lease-duration', None),
            },
            'blob_type': headers['x-ms-blob-type']
        }

        if extra['md5_hash']:
            value = binascii.hexlify(base64.b64decode(b(extra['md5_hash'])))
            value = value.decode('ascii')
            extra['md5_hash'] = value

        meta_data = {}
        for key, value in response.headers.items():
            if key.startswith('x-ms-meta-'):
                key = key.split('x-ms-meta-')[1]
                meta_data[key] = value

        return Object(name=object_name, size=size, hash=etag, extra=extra,
                      meta_data=meta_data, container=container, driver=self)

    def iterate_containers(self):
        """
        @inherits: :class:`StorageDriver.iterate_containers`
        """
        params = {'comp': 'list',
                  'maxresults': RESPONSES_PER_REQUEST,
                  'include': 'metadata'}

        while True:
            response = self.connection.request('/', params)
            if response.status != httplib.OK:
                raise LibcloudError('Unexpected status code: %s' %
                                    (response.status), driver=self)

            body = response.parse_body()
            containers = body.find(fixxpath(xpath='Containers'))
            containers = containers.findall(fixxpath(xpath='Container'))

            for container in containers:
                yield self._xml_to_container(container)

            params['marker'] = body.findtext('NextMarker')
            if not params['marker']:
                break

    def iterate_container_objects(self, container, prefix=None,
                                  ex_prefix=None):
        """
        @inherits: :class:`StorageDriver.iterate_container_objects`
        """
        prefix = self._normalize_prefix_argument(prefix, ex_prefix)

        params = {'restype': 'container',
                  'comp': 'list',
                  'maxresults': RESPONSES_PER_REQUEST,
                  'include': 'metadata'}

        if prefix:
            params['prefix'] = prefix

        container_path = self._get_container_path(container)

        while True:
            response = self.connection.request(container_path,
                                               params=params)

            if response.status == httplib.NOT_FOUND:
                raise ContainerDoesNotExistError(value=None,
                                                 driver=self,
                                                 container_name=container.name)

            elif response.status != httplib.OK:
                raise LibcloudError('Unexpected status code: %s' %
                                    (response.status), driver=self)

            body = response.parse_body()
            blobs = body.find(fixxpath(xpath='Blobs'))
            blobs = blobs.findall(fixxpath(xpath='Blob'))

            for blob in blobs:
                yield self._xml_to_object(container, blob)

            params['marker'] = body.findtext('NextMarker')
            if not params['marker']:
                break

    def get_container(self, container_name):
        """
        @inherits: :class:`StorageDriver.get_container`
        """
        params = {'restype': 'container'}

        container_path = '/%s' % (container_name)

        response = self.connection.request(container_path, params=params,
                                           method='HEAD')

        if response.status == httplib.NOT_FOUND:
            raise ContainerDoesNotExistError('Container %s does not exist' %
                                             (container_name), driver=self,
                                             container_name=container_name)
        elif response.status != httplib.OK:
            raise LibcloudError('Unexpected status code: %s' %
                                (response.status), driver=self)

        return self._response_to_container(container_name, response)

    def get_object(self, container_name, object_name):
        """
        @inherits: :class:`StorageDriver.get_object`
        """

        container = self.get_container(container_name=container_name)
        object_path = self._get_object_path(container, object_name)

        response = self.connection.request(object_path, method='HEAD')

        if response.status == httplib.OK:
            obj = self._response_to_object(object_name, container, response)
            return obj

        raise ObjectDoesNotExistError(value=None, driver=self,
                                      object_name=object_name)

    def get_object_cdn_url(self, obj,
                           ex_expiry=AZURE_STORAGE_CDN_URL_EXPIRY_HOURS):
        """
        Return a SAS URL that enables reading the given object.

        :param obj: Object instance.
        :type  obj: :class:`Object`

        :param ex_expiry: The number of hours after which the URL expires.
                          Defaults to 24 hours.
        :type  ex_expiry: ``float``

        :return: A SAS URL for the object.
        :rtype: ``str``
        """
        object_path = self._get_object_path(obj.container, obj.name)

        now = datetime.utcnow()
        start = now - timedelta(minutes=AZURE_STORAGE_CDN_URL_START_MINUTES)
        expiry = now + timedelta(hours=ex_expiry)

        params = {
            'st': start.strftime(AZURE_STORAGE_CDN_URL_DATE_FORMAT),
            'se': expiry.strftime(AZURE_STORAGE_CDN_URL_DATE_FORMAT),
            'sp': 'r',
            'spr': 'https' if self.secure else 'http,https',
            'sv': self.connectionCls.API_VERSION,
            'sr': 'b',
        }

        string_to_sign = '\n'.join((
            params['sp'],
            params['st'],
            params['se'],
            '/blob/{}{}'.format(self.key, object_path),
            '',  # signedIdentifier
            '',  # signedIP
            params['spr'],
            params['sv'],
            params['sr'],
            '',  # snapshot
            '',  # rscc
            '',  # rscd
            '',  # rsce
            '',  # rscl
            '',  # rsct
        ))

        params['sig'] = base64.b64encode(
            hmac.new(
                self.secret,
                string_to_sign.encode('utf-8'),
                hashlib.sha256
            ).digest()
        ).decode('utf-8')

        return '{scheme}://{host}:{port}{action}?{sas_token}'.format(
            scheme='https' if self.secure else 'http',
            host=self.connection.host,
            port=self.connection.port,
            action=self.connection.morph_action_hook(object_path),
            sas_token=urlencode(params),
        )

    def _get_container_path(self, container):
        """
        Return a container path

        :param container: Container instance
        :type  container: :class:`Container`

        :return: A path for this container.
        :rtype: ``str``
        """
        return '/%s' % (container.name)

    def _get_object_path(self, container, object_name):
        """
        Return an object's CDN path.

        :param container: Container instance
        :type  container: :class:`Container`

        :param object_name: Object name
        :type  object_name: :class:`str`

        :return: A  path for this object.
        :rtype: ``str``
        """
        container_url = self._get_container_path(container)
        object_name_cleaned = urlquote(object_name)
        object_path = '%s/%s' % (container_url, object_name_cleaned)
        return object_path

    def create_container(self, container_name):
        """
        @inherits: :class:`StorageDriver.create_container`
        """
        params = {'restype': 'container'}

        container_path = '/%s' % (container_name)
        response = self.connection.request(container_path, params=params,
                                           method='PUT')

        if response.status == httplib.CREATED:
            return self._response_to_container(container_name, response)
        elif response.status == httplib.CONFLICT:
            raise ContainerAlreadyExistsError(
                value='Container with this name already exists. The name must '
                      'be unique among all the containers in the system',
                container_name=container_name, driver=self)
        elif response.status == httplib.BAD_REQUEST:
            raise InvalidContainerNameError(value='Container name contains ' +
                                            'invalid characters.',
                                            container_name=container_name,
                                            driver=self)

        raise LibcloudError('Unexpected status code: %s' % (response.status),
                            driver=self)

    def delete_container(self, container):
        """
        @inherits: :class:`StorageDriver.delete_container`
        """
        # Azure does not check if the container is empty. So, we will do
        # a check to ensure that the behaviour is similar to other drivers
        for obj in container.iterate_objects():
            raise ContainerIsNotEmptyError(
                value='Container must be empty before it can be deleted.',
                container_name=container.name, driver=self)

        params = {'restype': 'container'}
        container_path = self._get_container_path(container)

        # Note: All the objects in the container must be deleted first
        response = self.connection.request(container_path, params=params,
                                           method='DELETE')

        if response.status == httplib.ACCEPTED:
            return True
        elif response.status == httplib.NOT_FOUND:
            raise ContainerDoesNotExistError(value=None,
                                             driver=self,
                                             container_name=container.name)

        return False

    def download_object(self, obj, destination_path, overwrite_existing=False,
                        delete_on_failure=True):
        """
        @inherits: :class:`StorageDriver.download_object`
        """
        obj_path = self._get_object_path(obj.container, obj.name)
        response = self.connection.request(obj_path, raw=True, data=None)

        return self._get_object(obj=obj, callback=self._save_object,
                                response=response,
                                callback_kwargs={
                                    'obj': obj,
                                    'response': response.response,
                                    'destination_path': destination_path,
                                    'overwrite_existing': overwrite_existing,
                                    'delete_on_failure': delete_on_failure},
                                success_status_code=httplib.OK)

    def download_object_as_stream(self, obj, chunk_size=None):
        """
        @inherits: :class:`StorageDriver.download_object_as_stream`
        """
        obj_path = self._get_object_path(obj.container, obj.name)
        response = self.connection.request(obj_path, method='GET',
                                           stream=True, raw=True)
        iterator = response.iter_content(AZURE_DOWNLOAD_CHUNK_SIZE)

        return self._get_object(obj=obj,
                                callback=read_in_chunks,
                                response=response,
                                callback_kwargs={'iterator': iterator,
                                                 'chunk_size': chunk_size},
                                success_status_code=httplib.OK)

    def download_object_range(self, obj, destination_path, start_bytes,
                              end_bytes=None, overwrite_existing=False,
                              delete_on_failure=True):
        self._validate_start_and_end_bytes(start_bytes=start_bytes,
                                           end_bytes=end_bytes)

        obj_path = self._get_object_path(obj.container, obj.name)
        headers = {'x-ms-range': self._get_standard_range_str(start_bytes,
                                                              end_bytes)}
        response = self.connection.request(obj_path, headers=headers,
                                           raw=True, data=None)

        # NOTE: Some Azure Blobs implementation return 200 instead of 206
        # status code, see
        # https://github.com/c-w/libcloud-tests/pull/2#issuecomment-592765323
        # for details.
        success_status_codes = [httplib.OK, httplib.PARTIAL_CONTENT]

        return self._get_object(obj=obj, callback=self._save_object,
                                response=response,
                                callback_kwargs={
                                    'obj': obj,
                                    'response': response.response,
                                    'destination_path': destination_path,
                                    'overwrite_existing': overwrite_existing,
                                    'delete_on_failure': delete_on_failure,
                                    'partial_download': True},
                                success_status_code=success_status_codes)

    def download_object_range_as_stream(self, obj, start_bytes, end_bytes=None,
                                        chunk_size=None):
        self._validate_start_and_end_bytes(start_bytes=start_bytes,
                                           end_bytes=end_bytes)

        obj_path = self._get_object_path(obj.container, obj.name)

        headers = {'x-ms-range': self._get_standard_range_str(start_bytes,
                                                              end_bytes)}
        response = self.connection.request(obj_path, method='GET',
                                           headers=headers,
                                           stream=True, raw=True)
        iterator = response.iter_content(AZURE_DOWNLOAD_CHUNK_SIZE)
        success_status_codes = [httplib.OK, httplib.PARTIAL_CONTENT]

        return self._get_object(
            obj=obj, callback=read_in_chunks,
            response=response,
            callback_kwargs={'iterator': iterator,
                             'chunk_size': chunk_size},
            success_status_code=success_status_codes)

    def _upload_in_chunks(self, stream, object_path, lease, meta_data,
                          content_type, object_name, file_path, verify_hash,
                          headers):
        """
        Uploads data from an interator in fixed sized chunks to Azure Storage
        """

        data_hash = None
        if verify_hash:
            data_hash = self._get_hash_function()

        bytes_transferred = 0
        count = 1
        chunks = []
        headers = headers or {}

        lease.update_headers(headers)

        params = {'comp': 'block'}

        # Read the input data in chunk sizes suitable for Azure
        for data in read_in_chunks(stream, AZURE_UPLOAD_CHUNK_SIZE,
                                   fill_size=True):
            data = b(data)
            content_length = len(data)
            bytes_transferred += content_length

            if verify_hash:
                data_hash.update(data)

            chunk_hash = self._get_hash_function()
            chunk_hash.update(data)
            chunk_hash = base64.b64encode(b(chunk_hash.digest()))

            headers['Content-MD5'] = chunk_hash.decode('utf-8')
            headers['Content-Length'] = str(content_length)

            # Block id can be any unique string that is base64 encoded
            # A 10 digit number can hold the max value of 50000 blocks
            # that are allowed for azure
            block_id = base64.b64encode(b('%10d' % (count)))
            block_id = block_id.decode('utf-8')
            params['blockid'] = block_id

            # Keep this data for a later commit
            chunks.append(block_id)

            # Renew lease before updating
            lease.renew()

            resp = self.connection.request(object_path, method='PUT',
                                           data=data, headers=headers,
                                           params=params)

            if resp.status != httplib.CREATED:
                resp.parse_error()
                raise LibcloudError('Error uploading chunk %d. Code: %d' %
                                    (count, resp.status), driver=self)

            count += 1

        if verify_hash:
            data_hash = base64.b64encode(b(data_hash.digest()))
            data_hash = data_hash.decode('utf-8')

        response = self._commit_blocks(object_path=object_path,
                                       chunks=chunks,
                                       lease=lease,
                                       meta_data=meta_data,
                                       content_type=content_type,
                                       data_hash=data_hash,
                                       object_name=object_name,
                                       file_path=file_path)

        # According to the Azure docs:
        # > This header refers to the content of the request, meaning, in this
        # > case, the list of blocks, and not the content of the blob itself.
        # However, the validation code assumes that the content-md5 in the
        # server response refers to the object so we must discard the value
        response.headers['content-md5'] = None

        return {
            'response': response,
            'data_hash': data_hash,
            'bytes_transferred': bytes_transferred,
        }

    def _commit_blocks(self, object_path, chunks, lease,
                       meta_data, content_type, data_hash,
                       object_name, file_path):
        """
        Makes a final commit of the data.
        """

        root = ET.Element('BlockList')

        for block_id in chunks:
            part = ET.SubElement(root, 'Uncommitted')
            part.text = str(block_id)

        data = tostring(root)
        params = {'comp': 'blocklist'}
        headers = {}

        lease.update_headers(headers)
        lease.renew()

        headers['x-ms-blob-content-type'] = self._determine_content_type(
            content_type, object_name, file_path)

        if data_hash is not None:
            headers['x-ms-blob-content-md5'] = data_hash

        self._update_metadata(headers, meta_data)

        data_hash = self._get_hash_function()
        data_hash.update(data.encode('utf-8'))
        data_hash = base64.b64encode(b(data_hash.digest()))
        headers['Content-MD5'] = data_hash.decode('utf-8')

        headers['Content-Length'] = len(data)

        response = self.connection.request(object_path, data=data,
                                           params=params, headers=headers,
                                           method='PUT')

        if response.status != httplib.CREATED:
            raise LibcloudError('Error in blocklist commit', driver=self)

        return response

    def upload_object(self, file_path, container, object_name,
                      verify_hash=True, extra=None, headers=None,
                      ex_use_lease=False,
                      **deprecated_kwargs):
        """
        Upload an object currently located on a disk.

        @inherits: :class:`StorageDriver.upload_object`

        :param ex_use_lease: Indicates if we must take a lease before upload
        :type ex_use_lease: ``bool``
        """
        if deprecated_kwargs:
            raise ValueError('Support for arguments was removed: %s'
                             % ', '.join(deprecated_kwargs.keys()))

        blob_size = os.stat(file_path).st_size

        with open(file_path, 'rb') as fobj:
            return self._put_object(container=container,
                                    object_name=object_name,
                                    extra=extra, verify_hash=verify_hash,
                                    use_lease=ex_use_lease, headers=headers,
                                    blob_size=blob_size, file_path=file_path,
                                    stream=fobj)

    def upload_object_via_stream(self, iterator, container, object_name,
                                 verify_hash=True, extra=None, headers=None,
                                 ex_use_lease=False,
                                 **deprecated_kwargs):
        """
        @inherits: :class:`StorageDriver.upload_object_via_stream`

        :param ex_use_lease: Indicates if we must take a lease before upload
        :type ex_use_lease: ``bool``
        """
        if deprecated_kwargs:
            raise ValueError('Support for arguments was removed: %s'
                             % ', '.join(deprecated_kwargs.keys()))

        return self._put_object(container=container,
                                object_name=object_name,
                                extra=extra, verify_hash=verify_hash,
                                use_lease=ex_use_lease,
                                headers=headers,
                                blob_size=None,
                                stream=iterator)

    def delete_object(self, obj):
        """
        @inherits: :class:`StorageDriver.delete_object`
        """
        object_path = self._get_object_path(obj.container, obj.name)
        response = self.connection.request(object_path, method='DELETE')

        if response.status == httplib.ACCEPTED:
            return True
        elif response.status == httplib.NOT_FOUND:
            raise ObjectDoesNotExistError(value=None, driver=self,
                                          object_name=obj.name)

        return False

    def _update_metadata(self, headers, meta_data):
        """
        Update the given metadata in the headers

        :param headers: The headers dictionary to be updated
        :type headers: ``dict``

        :param meta_data: Metadata key value pairs
        :type meta_data: ``dict``
        """
        for key, value in list(meta_data.items()):
            key = 'x-ms-meta-%s' % (key)
            headers[key] = value

    def _put_object(self, container, object_name, stream,
                    extra=None, verify_hash=True, headers=None,
                    blob_size=None, file_path=None,
                    use_lease=False):
        """
        Control function that does the real job of uploading data to a blob
        """
        extra = extra or {}
        content_type = extra.get('content_type', None)
        meta_data = extra.get('meta_data', {})

        object_path = self._get_object_path(container, object_name)

        # Get a lease if required and do the operations
        with AzureBlobLease(self, object_path, use_lease) as lease:
            if blob_size is not None and blob_size <= AZURE_UPLOAD_CHUNK_SIZE:
                result_dict = self._upload_directly(stream=stream,
                                                    object_path=object_path,
                                                    lease=lease,
                                                    blob_size=blob_size,
                                                    meta_data=meta_data,
                                                    headers=headers,
                                                    content_type=content_type,
                                                    object_name=object_name,
                                                    file_path=file_path)
            else:
                result_dict = self._upload_in_chunks(stream=stream,
                                                     object_path=object_path,
                                                     lease=lease,
                                                     meta_data=meta_data,
                                                     headers=headers,
                                                     content_type=content_type,
                                                     object_name=object_name,
                                                     file_path=file_path,
                                                     verify_hash=verify_hash)

            response = result_dict['response']
            bytes_transferred = result_dict['bytes_transferred']
            data_hash = result_dict['data_hash']
            headers = response.headers

        if response.status != httplib.CREATED:
            raise LibcloudError(
                'Unexpected status code, status_code=%s' % (response.status),
                driver=self)

        server_hash = headers.get('content-md5')

        if server_hash:
            server_hash = binascii.hexlify(base64.b64decode(b(server_hash)))
            server_hash = server_hash.decode('utf-8')
        else:
            # TODO: HACK - We could poll the object for a while and get
            # the hash
            pass

        if (verify_hash and server_hash and data_hash != server_hash):
            raise ObjectHashMismatchError(
                value='MD5 hash checksum does not match',
                object_name=object_name, driver=self)

        return Object(name=object_name, size=bytes_transferred,
                      hash=headers['etag'], extra=None,
                      meta_data=meta_data, container=container,
                      driver=self)

    def _upload_directly(self, stream, object_path, lease, blob_size,
                         meta_data, content_type, object_name, file_path,
                         headers):

        headers = headers or {}
        lease.update_headers(headers)

        self._update_metadata(headers, meta_data)

        headers['Content-Length'] = str(blob_size)
        headers['x-ms-blob-type'] = 'BlockBlob'

        return self._upload_object(object_name=object_name,
                                   file_path=file_path,
                                   content_type=content_type,
                                   request_path=object_path,
                                   stream=stream,
                                   headers=headers)

    def ex_set_object_metadata(self, obj, meta_data):
        """
        Set metadata for an object

        :param obj: The blob object
        :type obj: :class:`Object`

        :param meta_data: Metadata key value pairs
        :type meta_data: ``dict``
        """
        object_path = self._get_object_path(obj.container, obj.name)
        params = {'comp': 'metadata'}
        headers = {}

        self._update_metadata(headers, meta_data)

        response = self.connection.request(object_path, method='PUT',
                                           params=params,
                                           headers=headers)

        if response.status != httplib.OK:
            response.parse_error('Setting metadata')