path: root/.venv/lib/python3.12/site-packages/azure/storage/blob/_encryption.py

# pylint: disable=too-many-lines
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------

import math
import os
import sys
import warnings
from collections import OrderedDict
from io import BytesIO
from json import (
    dumps,
    loads,
)
from typing import Any, Callable, Dict, IO, Optional, Tuple, TYPE_CHECKING
from typing import OrderedDict as TypedOrderedDict
from typing_extensions import Protocol

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.ciphers.algorithms import AES
from cryptography.hazmat.primitives.ciphers.modes import CBC
from cryptography.hazmat.primitives.padding import PKCS7

from azure.core.exceptions import HttpResponseError
from azure.core.utils import CaseInsensitiveDict

from ._version import VERSION
from ._shared import decode_base64_to_bytes, encode_base64

if TYPE_CHECKING:
    from azure.core.pipeline import PipelineResponse
    from cryptography.hazmat.primitives.ciphers import AEADEncryptionContext
    from cryptography.hazmat.primitives.padding import PaddingContext


_ENCRYPTION_PROTOCOL_V1 = '1.0'
_ENCRYPTION_PROTOCOL_V2 = '2.0'
_ENCRYPTION_PROTOCOL_V2_1 = '2.1'
_VALID_ENCRYPTION_PROTOCOLS = [_ENCRYPTION_PROTOCOL_V1, _ENCRYPTION_PROTOCOL_V2, _ENCRYPTION_PROTOCOL_V2_1]
_ENCRYPTION_V2_PROTOCOLS = [_ENCRYPTION_PROTOCOL_V2, _ENCRYPTION_PROTOCOL_V2_1]
_GCM_REGION_DATA_LENGTH = 4 * 1024 * 1024
_GCM_NONCE_LENGTH = 12
_GCM_TAG_LENGTH = 16

_ERROR_OBJECT_INVALID = \
    '{0} does not define a complete interface. Value of {1} is either missing or invalid.'

_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION = (
    'The require_encryption flag is set, but encryption is not supported'
    ' for this method.')


class KeyEncryptionKey(Protocol):

    def wrap_key(self, key: bytes) -> bytes:
        ...

    def unwrap_key(self, key: bytes, algorithm: str) -> bytes:
        ...

    def get_kid(self) -> str:
        ...

    def get_key_wrap_algorithm(self) -> str:
        ...


def _validate_not_none(param_name: str, param: Any):
    if param is None:
        raise ValueError(f'{param_name} should not be None.')


def _validate_key_encryption_key_wrap(kek: KeyEncryptionKey):
    # Note that None is not callable and so will fail the second clause of each check.
    if not hasattr(kek, 'wrap_key') or not callable(kek.wrap_key):
        raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'wrap_key'))
    if not hasattr(kek, 'get_kid') or not callable(kek.get_kid):
        raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'get_kid'))
    if not hasattr(kek, 'get_key_wrap_algorithm') or not callable(kek.get_key_wrap_algorithm):
        raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'get_key_wrap_algorithm'))


class StorageEncryptionMixin(object):
    def _configure_encryption(self, kwargs: Dict[str, Any]):
        self.require_encryption = kwargs.get("require_encryption", False)
        self.encryption_version = kwargs.get("encryption_version", "1.0")
        self.key_encryption_key = kwargs.get("key_encryption_key")
        self.key_resolver_function = kwargs.get("key_resolver_function")
        if self.key_encryption_key and self.encryption_version == '1.0':
            warnings.warn("This client has been configured to use encryption with version 1.0. " +
                          "Version 1.0 is deprecated and no longer considered secure. It is highly " +
                          "recommended that you switch to using version 2.0. The version can be " +
                          "specified using the 'encryption_version' keyword.")


class _EncryptionAlgorithm(object):
    """
    Specifies which client encryption algorithm is used.
    """
    AES_CBC_256 = 'AES_CBC_256'
    AES_GCM_256 = 'AES_GCM_256'


class _WrappedContentKey:
    """
    Represents the envelope key details stored on the service.
    """

    def __init__(self, algorithm: str, encrypted_key: bytes, key_id: str) -> None:
        """
        :param str algorithm:
            The algorithm used for wrapping.
        :param bytes encrypted_key:
            The encrypted content-encryption-key.
        :param str key_id:
            The key-encryption-key identifier string.
        """
        _validate_not_none('algorithm', algorithm)
        _validate_not_none('encrypted_key', encrypted_key)
        _validate_not_none('key_id', key_id)

        self.algorithm = algorithm
        self.encrypted_key = encrypted_key
        self.key_id = key_id


class _EncryptedRegionInfo:
    """
    Represents the length of encryption elements.
    This is only used for Encryption V2.
    """

    def __init__(self, data_length: int, nonce_length: int, tag_length: int) -> None:
        """
        :param int data_length:
            The length of the encryption region data (not including nonce + tag).
        :param int nonce_length:
            The length of nonce used when encrypting.
        :param int tag_length:
            The length of the encryption tag.
        """
        _validate_not_none('data_length', data_length)
        _validate_not_none('nonce_length', nonce_length)
        _validate_not_none('tag_length', tag_length)

        self.data_length = data_length
        self.nonce_length = nonce_length
        self.tag_length = tag_length


class _EncryptionAgent:
    """
    Represents the encryption agent stored on the service.
    It consists of the encryption protocol version and encryption algorithm used.
    """

    def __init__(self, encryption_algorithm: _EncryptionAlgorithm, protocol: str) -> None:
        """
        :param _EncryptionAlgorithm encryption_algorithm:
            The algorithm used for encrypting the message contents.
        :param str protocol:
            The protocol version used for encryption.
        """
        _validate_not_none('encryption_algorithm', encryption_algorithm)
        _validate_not_none('protocol', protocol)

        self.encryption_algorithm = str(encryption_algorithm)
        self.protocol = protocol


class _EncryptionData:
    """
    Represents the encryption data that is stored on the service.
    """

    def __init__(
        self, content_encryption_IV: Optional[bytes],
        encrypted_region_info: Optional[_EncryptedRegionInfo],
        encryption_agent: _EncryptionAgent,
        wrapped_content_key: _WrappedContentKey,
        key_wrapping_metadata: Dict[str, Any]
    ) -> None:
        """
        :param Optional[bytes] content_encryption_IV:
            The content encryption initialization vector.
            Required for AES-CBC (V1).
        :param Optional[_EncryptedRegionInfo] encrypted_region_info:
            The info about the autenticated block sizes.
            Required for AES-GCM (V2).
        :param _EncryptionAgent encryption_agent:
            The encryption agent.
        :param _WrappedContentKey wrapped_content_key:
            An object that stores the wrapping algorithm, the key identifier,
            and the encrypted key bytes.
        :param Dict[str, Any] key_wrapping_metadata:
            A dict containing metadata related to the key wrapping.
        """
        _validate_not_none('encryption_agent', encryption_agent)
        _validate_not_none('wrapped_content_key', wrapped_content_key)

        # Validate we have the right matching optional parameter for the specified algorithm
        if encryption_agent.encryption_algorithm == _EncryptionAlgorithm.AES_CBC_256:
            _validate_not_none('content_encryption_IV', content_encryption_IV)
        elif encryption_agent.encryption_algorithm == _EncryptionAlgorithm.AES_GCM_256:
            _validate_not_none('encrypted_region_info', encrypted_region_info)
        else:
            raise ValueError("Invalid encryption algorithm.")

        self.content_encryption_IV = content_encryption_IV
        self.encrypted_region_info = encrypted_region_info
        self.encryption_agent = encryption_agent
        self.wrapped_content_key = wrapped_content_key
        self.key_wrapping_metadata = key_wrapping_metadata


class GCMBlobEncryptionStream:
    """
    A stream that performs AES-GCM encryption on the given data as
    it's streamed. Data is read and encrypted in regions. The stream
    will use the same encryption key and will generate a guaranteed unique
    nonce for each encryption region.
    """
    def __init__(
        self, content_encryption_key: bytes,
        data_stream: IO[bytes],
    ) -> None:
        """
        :param bytes content_encryption_key: The encryption key to use.
        :param IO[bytes] data_stream: The data stream to read data from.
        """
        self.content_encryption_key = content_encryption_key
        self.data_stream = data_stream

        self.offset = 0
        self.current = b''
        self.nonce_counter = 0

    def read(self, size: int = -1) -> bytes:
        """
        Read data from the stream. Specify -1 to read all available data.

        :param int size: The amount of data to read. Defaults to -1 for all data.
        :return: The bytes read.
        :rtype: bytes
        """
        result = BytesIO()
        remaining = sys.maxsize if size == -1 else size

        while remaining > 0:
            # Start by reading from current
            if len(self.current) > 0:
                read = min(remaining, len(self.current))
                result.write(self.current[:read])

                self.current = self.current[read:]
                self.offset += read
                remaining -= read

            if remaining > 0:
                # Read one region of data and encrypt it
                data = self.data_stream.read(_GCM_REGION_DATA_LENGTH)
                if len(data) == 0:
                    # No more data to read
                    break

                self.current = encrypt_data_v2(data, self.nonce_counter, self.content_encryption_key)
                # IMPORTANT: Must increment the nonce each time.
                self.nonce_counter += 1

        return result.getvalue()


def encrypt_data_v2(data: bytes, nonce: int, key: bytes) -> bytes:
    """
    Encrypts the given data using the given nonce and key using AES-GCM.
    The result includes the data in the form: nonce + ciphertext + tag.

    :param bytes data: The raw data to encrypt.
    :param int nonce: The nonce to use for encryption.
    :param bytes key: The encryption key to use for encryption.
    :return: The encrypted bytes in the form: nonce + ciphertext + tag.
    :rtype: bytes
    """
    nonce_bytes = nonce.to_bytes(_GCM_NONCE_LENGTH, 'big')
    aesgcm = AESGCM(key)

    # Returns ciphertext + tag
    ciphertext_with_tag = aesgcm.encrypt(nonce_bytes, data, None)
    return nonce_bytes + ciphertext_with_tag


def is_encryption_v2(encryption_data: Optional[_EncryptionData]) -> bool:
    """
    Determine whether the given encryption data signifies version 2.0 or 2.1.

    :param Optional[_EncryptionData] encryption_data: The encryption data. Will return False if this is None.
    :return: True, if the encryption data indicates encryption V2, false otherwise.
    :rtype: bool
    """
    # If encryption_data is None, assume no encryption
    return bool(encryption_data and (encryption_data.encryption_agent.protocol in _ENCRYPTION_V2_PROTOCOLS))


def modify_user_agent_for_encryption(
        user_agent: str,
        moniker: str,
        encryption_version: str,
        request_options: Dict[str, Any]
    ) -> None:
    """
    Modifies the request options to contain a user agent string updated with encryption information.
    Adds azstorage-clientsideencryption/<version> immediately proceeding the SDK descriptor.

    :param str user_agent: The existing User Agent to modify.
    :param str moniker: The specific SDK moniker. The modification will immediately proceed azsdk-python-{moniker}.
    :param str encryption_version: The version of encryption being used.
    :param Dict[str, Any] request_options: The reuqest options to add the user agent override to.
    """
    # If the user has specified user_agent_overwrite=True, don't make any modifications
    if request_options.get('user_agent_overwrite'):
        return

    # If the feature flag is already present, don't add it again
    feature_flag = f"azstorage-clientsideencryption/{encryption_version}"
    if feature_flag in user_agent:
        return

    index = user_agent.find(f"azsdk-python-{moniker}")
    user_agent = f"{user_agent[:index]}{feature_flag} {user_agent[index:]}"
    # Since we are using user_agent_overwrite=True, we must prepend the user's user_agent if there is one
    if request_options.get('user_agent'):
        user_agent = f"{request_options.get('user_agent')} {user_agent}"

    request_options['user_agent'] = user_agent
    request_options['user_agent_overwrite'] = True


def get_adjusted_upload_size(length: int, encryption_version: str) -> int:
    """
    Get the adjusted size of the blob upload which accounts for
    extra encryption data (padding OR nonce + tag).

    :param int length: The plaintext data length.
    :param str encryption_version: The version of encryption being used.
    :return: The new upload size to use.
    :rtype: int
    """
    if encryption_version == _ENCRYPTION_PROTOCOL_V1:
        return length + (16 - (length % 16))

    if encryption_version == _ENCRYPTION_PROTOCOL_V2:
        encryption_data_length = _GCM_NONCE_LENGTH + _GCM_TAG_LENGTH
        regions = math.ceil(length / _GCM_REGION_DATA_LENGTH)
        return length + (regions * encryption_data_length)

    raise ValueError("Invalid encryption version specified.")


def get_adjusted_download_range_and_offset(
        start: int,
        end: int,
        length: Optional[int],
        encryption_data: Optional[_EncryptionData]) -> Tuple[Tuple[int, int], Tuple[int, int]]:
    """
    Gets the new download range and offsets into the decrypted data for
    the given user-specified range. The new download range will include all
    the data needed to decrypt the user-provided range and will include only
    full encryption regions.

    The offsets returned will be the offsets needed to fetch the user-requested
    data out of the full decrypted data. The end offset is different based on the
    encryption version. For V1, the end offset is offset from the end whereas for
    V2, the end offset is the ending index into the stream.
    V1: decrypted_data[start_offset : len(decrypted_data) - end_offset]
    V2: decrypted_data[start_offset : end_offset]

    :param int start: The user-requested start index.
    :param int end: The user-requested end index.
    :param Optional[int] length: The user-requested length. Only used for V1.
    :param Optional[_EncryptionData] encryption_data: The encryption data to determine version and sizes.
    :return: (new start, new end), (start offset, end offset)
    :rtype: Tuple[Tuple[int, int], Tuple[int, int]]
    """
    start_offset, end_offset = 0, 0
    if encryption_data is None:
        return (start, end), (start_offset, end_offset)

    if encryption_data.encryption_agent.protocol == _ENCRYPTION_PROTOCOL_V1:
        if start is not None:
            # Align the start of the range along a 16 byte block
            start_offset = start % 16
            start -= start_offset

            # Include an extra 16 bytes for the IV if necessary
            # Because of the previous offsetting, start_range will always
            # be a multiple of 16.
            if start > 0:
                start_offset += 16
                start -= 16

        if length is not None:
            # Align the end of the range along a 16 byte block
            end_offset = 15 - (end % 16)
            end += end_offset

    elif encryption_data.encryption_agent.protocol in _ENCRYPTION_V2_PROTOCOLS:
        start_offset, end_offset = 0, end

        if encryption_data.encrypted_region_info is None:
            raise ValueError("Missing required metadata for Encryption V2")

        nonce_length = encryption_data.encrypted_region_info.nonce_length
        data_length = encryption_data.encrypted_region_info.data_length
        tag_length = encryption_data.encrypted_region_info.tag_length
        region_length = nonce_length + data_length + tag_length
        requested_length = end - start

        if start is not None:
            # Find which data region the start is in
            region_num = start // data_length
            # The start of the data region is different from the start of the encryption region
            data_start = region_num * data_length
            region_start = region_num * region_length
            # Offset is based on data region
            start_offset = start - data_start
            # New start is the start of the encryption region
            start = region_start

        if end is not None:
            # Find which data region the end is in
            region_num = end // data_length
            end_offset = start_offset + requested_length + 1
            # New end is the end of the encryption region
            end = (region_num * region_length) + region_length - 1

    return (start, end), (start_offset, end_offset)


def parse_encryption_data(metadata: Dict[str, Any]) -> Optional[_EncryptionData]:
    """
    Parses the encryption data out of the given blob metadata. If metadata does
    not exist or there are parsing errors, this function will just return None.

    :param Dict[str, Any] metadata: The blob metadata parsed from the response.
    :return: The encryption data or None
    :rtype: Optional[_EncryptionData]
    """
    try:
        # Use case insensitive dict as key needs to be case-insensitive
        case_insensitive_metadata = CaseInsensitiveDict(metadata)
        return _dict_to_encryption_data(loads(case_insensitive_metadata['encryptiondata']))
    except:  # pylint: disable=bare-except
        return None


def adjust_blob_size_for_encryption(size: int, encryption_data: Optional[_EncryptionData]) -> int:
    """
    Adjusts the given blob size for encryption by subtracting the size of
    the encryption data (nonce + tag). This only has an affect for encryption V2.

    :param int size: The original blob size.
    :param Optional[_EncryptionData] encryption_data: The encryption data to determine version and sizes.
    :return: The new blob size.
    :rtype: int
    """
    if (encryption_data is not None and
        encryption_data.encrypted_region_info is not None and
        is_encryption_v2(encryption_data)):

        nonce_length = encryption_data.encrypted_region_info.nonce_length
        data_length = encryption_data.encrypted_region_info.data_length
        tag_length = encryption_data.encrypted_region_info.tag_length
        region_length = nonce_length + data_length + tag_length

        num_regions = math.ceil(size / region_length)
        metadata_size = num_regions * (nonce_length + tag_length)
        return size - metadata_size

    return size


def _generate_encryption_data_dict(
        kek: KeyEncryptionKey,
        cek: bytes,
        iv: Optional[bytes],
        version: str
    ) -> TypedOrderedDict[str, Any]:
    """
    Generates and returns the encryption metadata as a dict.

    :param KeyEncryptionKey kek: The key encryption key. See calling functions for more information.
    :param bytes cek: The content encryption key.
    :param Optional[bytes] iv: The initialization vector. Only required for AES-CBC.
    :param str version: The client encryption version used.
    :return: A dict containing all the encryption metadata.
    :rtype: Dict[str, Any]
    """
    # Encrypt the cek.
    if version == _ENCRYPTION_PROTOCOL_V1:
        wrapped_cek = kek.wrap_key(cek)
    # For V2, we include the encryption version in the wrapped key.
    elif version == _ENCRYPTION_PROTOCOL_V2:
        # We must pad the version to 8 bytes for AES Keywrap algorithms
        to_wrap = _ENCRYPTION_PROTOCOL_V2.encode().ljust(8, b'\0') + cek
        wrapped_cek = kek.wrap_key(to_wrap)
    else:
        raise ValueError("Invalid encryption version specified.")

    # Build the encryption_data dict.
    # Use OrderedDict to comply with Java's ordering requirement.
    wrapped_content_key = OrderedDict()
    wrapped_content_key['KeyId'] = kek.get_kid()
    wrapped_content_key['EncryptedKey'] = encode_base64(wrapped_cek)
    wrapped_content_key['Algorithm'] = kek.get_key_wrap_algorithm()

    encryption_agent = OrderedDict()
    encryption_agent['Protocol'] = version

    if version == _ENCRYPTION_PROTOCOL_V1:
        encryption_agent['EncryptionAlgorithm'] = _EncryptionAlgorithm.AES_CBC_256

    elif version == _ENCRYPTION_PROTOCOL_V2:
        encryption_agent['EncryptionAlgorithm'] = _EncryptionAlgorithm.AES_GCM_256

        encrypted_region_info = OrderedDict()
        encrypted_region_info['DataLength'] = _GCM_REGION_DATA_LENGTH
        encrypted_region_info['NonceLength'] = _GCM_NONCE_LENGTH

    encryption_data_dict: TypedOrderedDict[str, Any] = OrderedDict()
    encryption_data_dict['WrappedContentKey'] = wrapped_content_key
    encryption_data_dict['EncryptionAgent'] = encryption_agent
    if version == _ENCRYPTION_PROTOCOL_V1:
        encryption_data_dict['ContentEncryptionIV'] = encode_base64(iv)
    elif version == _ENCRYPTION_PROTOCOL_V2:
        encryption_data_dict['EncryptedRegionInfo'] = encrypted_region_info
    encryption_data_dict['KeyWrappingMetadata'] = OrderedDict({'EncryptionLibrary': 'Python ' + VERSION})

    return encryption_data_dict


def _dict_to_encryption_data(encryption_data_dict: Dict[str, Any]) -> _EncryptionData:
    """
    Converts the specified dictionary to an EncryptionData object for
    eventual use in decryption.

    :param dict encryption_data_dict:
        The dictionary containing the encryption data.
    :return: an _EncryptionData object built from the dictionary.
    :rtype: _EncryptionData
    """
    try:
        protocol = encryption_data_dict['EncryptionAgent']['Protocol']
        if protocol not in _VALID_ENCRYPTION_PROTOCOLS:
            raise ValueError("Unsupported encryption version.")
    except KeyError as exc:
        raise ValueError("Unsupported encryption version.") from exc
    wrapped_content_key = encryption_data_dict['WrappedContentKey']
    wrapped_content_key = _WrappedContentKey(wrapped_content_key['Algorithm'],
                                             decode_base64_to_bytes(wrapped_content_key['EncryptedKey']),
                                             wrapped_content_key['KeyId'])

    encryption_agent = encryption_data_dict['EncryptionAgent']
    encryption_agent = _EncryptionAgent(encryption_agent['EncryptionAlgorithm'],
                                        encryption_agent['Protocol'])

    if 'KeyWrappingMetadata' in encryption_data_dict:
        key_wrapping_metadata = encryption_data_dict['KeyWrappingMetadata']
    else:
        key_wrapping_metadata = None

    # AES-CBC only
    encryption_iv = None
    if 'ContentEncryptionIV' in encryption_data_dict:
        encryption_iv = decode_base64_to_bytes(encryption_data_dict['ContentEncryptionIV'])

    # AES-GCM only
    region_info = None
    if 'EncryptedRegionInfo' in encryption_data_dict:
        encrypted_region_info = encryption_data_dict['EncryptedRegionInfo']
        region_info = _EncryptedRegionInfo(encrypted_region_info['DataLength'],
                                           encrypted_region_info['NonceLength'],
                                           _GCM_TAG_LENGTH)

    encryption_data = _EncryptionData(encryption_iv,
                                      region_info,
                                      encryption_agent,
                                      wrapped_content_key,
                                      key_wrapping_metadata)

    return encryption_data


def _generate_AES_CBC_cipher(cek: bytes, iv: bytes) -> Cipher:
    """
    Generates and returns an encryption cipher for AES CBC using the given cek and iv.

    :param bytes[] cek: The content encryption key for the cipher.
    :param bytes[] iv: The initialization vector for the cipher.
    :return: A cipher for encrypting in AES256 CBC.
    :rtype: ~cryptography.hazmat.primitives.ciphers.Cipher
    """

    backend = default_backend()
    algorithm = AES(cek)
    mode = CBC(iv)
    return Cipher(algorithm, mode, backend)


def _validate_and_unwrap_cek(
    encryption_data: _EncryptionData,
    key_encryption_key: Optional[KeyEncryptionKey] = None,
    key_resolver: Optional[Callable[[str], KeyEncryptionKey]] = None
) -> bytes:
    """
    Extracts and returns the content_encryption_key stored in the encryption_data object
    and performs necessary validation on all parameters.
    :param _EncryptionData encryption_data:
        The encryption metadata of the retrieved value.
    :param Optional[KeyEncryptionKey] key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param Optional[Callable[[str], KeyEncryptionKey]] key_resolver:
        A function used that, given a key_id, will return a key_encryption_key. Please refer
        to high-level service object instance variables for more details.
    :return: The content_encryption_key stored in the encryption_data object.
    :rtype: bytes
    """

    _validate_not_none('encrypted_key', encryption_data.wrapped_content_key.encrypted_key)

    # Validate we have the right info for the specified version
    if encryption_data.encryption_agent.protocol == _ENCRYPTION_PROTOCOL_V1:
        _validate_not_none('content_encryption_IV', encryption_data.content_encryption_IV)
    elif encryption_data.encryption_agent.protocol in _ENCRYPTION_V2_PROTOCOLS:
        _validate_not_none('encrypted_region_info', encryption_data.encrypted_region_info)
    else:
        raise ValueError('Specified encryption version is not supported.')

    content_encryption_key: Optional[bytes] = None

    # If the resolver exists, give priority to the key it finds.
    if key_resolver is not None:
        key_encryption_key = key_resolver(encryption_data.wrapped_content_key.key_id)

    if key_encryption_key is None:
        raise ValueError("Unable to decrypt. key_resolver and key_encryption_key cannot both be None.")
    if not hasattr(key_encryption_key, 'get_kid') or not callable(key_encryption_key.get_kid):
        raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'get_kid'))
    if not hasattr(key_encryption_key, 'unwrap_key') or not callable(key_encryption_key.unwrap_key):
        raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'unwrap_key'))
    if encryption_data.wrapped_content_key.key_id != key_encryption_key.get_kid():
        raise ValueError('Provided or resolved key-encryption-key does not match the id of key used to encrypt.')
    # Will throw an exception if the specified algorithm is not supported.
    content_encryption_key = key_encryption_key.unwrap_key(
        encryption_data.wrapped_content_key.encrypted_key,
        encryption_data.wrapped_content_key.algorithm)

    # For V2, the version is included with the cek. We need to validate it
    # and remove it from the actual cek.
    if encryption_data.encryption_agent.protocol in _ENCRYPTION_V2_PROTOCOLS:
        version_2_bytes = encryption_data.encryption_agent.protocol.encode().ljust(8, b'\0')
        cek_version_bytes = content_encryption_key[:len(version_2_bytes)]
        if cek_version_bytes != version_2_bytes:
            raise ValueError('The encryption metadata is not valid and may have been modified.')

        # Remove version from the start of the cek.
        content_encryption_key = content_encryption_key[len(version_2_bytes):]

    _validate_not_none('content_encryption_key', content_encryption_key)

    return content_encryption_key


def _decrypt_message(
    message: bytes,
    encryption_data: _EncryptionData,
    key_encryption_key: Optional[KeyEncryptionKey] = None,
    resolver: Optional[Callable[[str], KeyEncryptionKey]] = None
) -> bytes:
    """
    Decrypts the given ciphertext using AES256 in CBC mode with 128 bit padding.
    Unwraps the content-encryption-key using the user-provided or resolved key-encryption-key (kek).
    Returns the original plaintext.

    :param bytes message:
        The ciphertext to be decrypted.
    :param _EncryptionData encryption_data:
        The metadata associated with this ciphertext.
    :param Optional[KeyEncryptionKey] key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param Optional[Callable[[str], KeyEncryptionKey]] resolver:
        The user-provided key resolver. Uses the kid string to return a key-encryption-key
        implementing the interface defined above.
    :return: The decrypted plaintext.
    :rtype: bytes
    """
    _validate_not_none('message', message)
    content_encryption_key = _validate_and_unwrap_cek(encryption_data, key_encryption_key, resolver)

    if encryption_data.encryption_agent.protocol == _ENCRYPTION_PROTOCOL_V1:
        if not encryption_data.content_encryption_IV:
            raise ValueError("Missing required metadata for decryption.")

        cipher = _generate_AES_CBC_cipher(content_encryption_key, encryption_data.content_encryption_IV)

        # decrypt data
        decryptor = cipher.decryptor()
        decrypted_data = (decryptor.update(message) + decryptor.finalize())

        # unpad data
        unpadder = PKCS7(128).unpadder()
        decrypted_data = (unpadder.update(decrypted_data) + unpadder.finalize())

    elif encryption_data.encryption_agent.protocol in _ENCRYPTION_V2_PROTOCOLS:
        block_info = encryption_data.encrypted_region_info
        if not block_info or not block_info.nonce_length:
            raise ValueError("Missing required metadata for decryption.")

        if encryption_data.encrypted_region_info is None:
            raise ValueError("Missing required metadata for Encryption V2")

        nonce_length = int(encryption_data.encrypted_region_info.nonce_length)

        # First bytes are the nonce
        nonce = message[:nonce_length]
        ciphertext_with_tag = message[nonce_length:]

        aesgcm = AESGCM(content_encryption_key)
        decrypted_data = aesgcm.decrypt(nonce, ciphertext_with_tag, None)

    else:
        raise ValueError('Specified encryption version is not supported.')

    return decrypted_data


def encrypt_blob(blob: bytes, key_encryption_key: KeyEncryptionKey, version: str) -> Tuple[str, bytes]:
    """
    Encrypts the given blob using the given encryption protocol version.
    Wraps the generated content-encryption-key using the user-provided key-encryption-key (kek).
    Returns a json-formatted string containing the encryption metadata. This method should
    only be used when a blob is small enough for single shot upload. Encrypting larger blobs
    is done as a part of the upload_data_chunks method.

    :param bytes blob:
        The blob to be encrypted.
    :param KeyEncryptionKey key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param str version: The client encryption version to use.
    :return: A tuple of json-formatted string containing the encryption metadata and the encrypted blob data.
    :rtype: (str, bytes)
    """

    _validate_not_none('blob', blob)
    _validate_not_none('key_encryption_key', key_encryption_key)
    _validate_key_encryption_key_wrap(key_encryption_key)

    if version == _ENCRYPTION_PROTOCOL_V1:
        # AES256 uses 256 bit (32 byte) keys and always with 16 byte blocks
        content_encryption_key = os.urandom(32)
        initialization_vector = os.urandom(16)

        cipher = _generate_AES_CBC_cipher(content_encryption_key, initialization_vector)

        # PKCS7 with 16 byte blocks ensures compatibility with AES.
        padder = PKCS7(128).padder()
        padded_data = padder.update(blob) + padder.finalize()

        # Encrypt the data.
        encryptor = cipher.encryptor()
        encrypted_data = encryptor.update(padded_data) + encryptor.finalize()

    elif version == _ENCRYPTION_PROTOCOL_V2:
        # AES256 GCM uses 256 bit (32 byte) keys and a 12 byte nonce.
        content_encryption_key = os.urandom(32)
        initialization_vector = None

        data = BytesIO(blob)
        encryption_stream = GCMBlobEncryptionStream(content_encryption_key, data)

        encrypted_data = encryption_stream.read()

    else:
        raise ValueError("Invalid encryption version specified.")

    encryption_data = _generate_encryption_data_dict(key_encryption_key, content_encryption_key,
                                                     initialization_vector, version)
    encryption_data['EncryptionMode'] = 'FullBlob'

    return dumps(encryption_data), encrypted_data


def generate_blob_encryption_data(
    key_encryption_key: Optional[KeyEncryptionKey],
    version: str
) -> Tuple[Optional[bytes], Optional[bytes], Optional[str]]:
    """
    Generates the encryption_metadata for the blob.

    :param Optional[KeyEncryptionKey] key_encryption_key:
        The key-encryption-key used to wrap the cek associate with this blob.
    :param str version: The client encryption version to use.
    :return: A tuple containing the cek and iv for this blob as well as the
        serialized encryption metadata for the blob.
    :rtype: (Optional[bytes], Optional[bytes], Optional[str])
    """

    encryption_data = None
    content_encryption_key = None
    initialization_vector = None
    if key_encryption_key:
        _validate_key_encryption_key_wrap(key_encryption_key)
        content_encryption_key = os.urandom(32)
        # Initialization vector only needed for V1
        if version == _ENCRYPTION_PROTOCOL_V1:
            initialization_vector = os.urandom(16)
        encryption_data_dict = _generate_encryption_data_dict(key_encryption_key,
                                                         content_encryption_key,
                                                         initialization_vector,
                                                         version)
        encryption_data_dict['EncryptionMode'] = 'FullBlob'
        encryption_data = dumps(encryption_data_dict)

    return content_encryption_key, initialization_vector, encryption_data


def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
        require_encryption: bool,
        key_encryption_key: Optional[KeyEncryptionKey],
        key_resolver: Optional[Callable[[str], KeyEncryptionKey]],
        content: bytes,
        start_offset: int,
        end_offset: int,
        response_headers: Dict[str, Any]
) -> bytes:
    """
    Decrypts the given blob contents and returns only the requested range.

    :param bool require_encryption:
        Whether the calling blob service requires objects to be decrypted.
    :param Optional[KeyEncryptionKey] key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param key_resolver:
        The user-provided key resolver. Uses the kid string to return a key-encryption-key
        implementing the interface defined above.
    :type key_resolver: Optional[Callable[[str], KeyEncryptionKey]]
    :param bytes content:
        The encrypted blob content.
    :param int start_offset:
        The adjusted offset from the beginning of the *decrypted* content for the caller's data.
    :param int end_offset:
        The adjusted offset from the end of the *decrypted* content for the caller's data.
    :param Dict[str, Any] response_headers:
        A dictionary of response headers from the download request. Expected to include the
        'x-ms-meta-encryptiondata' header if the blob was encrypted.
    :return: The decrypted blob content.
    :rtype: bytes
    """
    try:
        encryption_data = _dict_to_encryption_data(loads(response_headers['x-ms-meta-encryptiondata']))
    except Exception as exc:  # pylint: disable=broad-except
        if require_encryption:
            raise ValueError(
                'Encryption required, but received data does not contain appropriate metadata.' + \
                'Data was either not encrypted or metadata has been lost.') from exc

        return content

    algorithm = encryption_data.encryption_agent.encryption_algorithm
    if algorithm not in(_EncryptionAlgorithm.AES_CBC_256, _EncryptionAlgorithm.AES_GCM_256):
        raise ValueError('Specified encryption algorithm is not supported.')

    version = encryption_data.encryption_agent.protocol
    if version not in _VALID_ENCRYPTION_PROTOCOLS:
        raise ValueError('Specified encryption version is not supported.')

    content_encryption_key = _validate_and_unwrap_cek(encryption_data, key_encryption_key, key_resolver)

    if version == _ENCRYPTION_PROTOCOL_V1:
        blob_type = response_headers['x-ms-blob-type']

        iv: Optional[bytes] = None
        unpad = False
        if 'content-range' in response_headers:
            content_range = response_headers['content-range']
            # Format: 'bytes x-y/size'

            # Ignore the word 'bytes'
            content_range = content_range.split(' ')

            content_range = content_range[1].split('-')
            content_range = content_range[1].split('/')
            end_range = int(content_range[0])
            blob_size = int(content_range[1])

            if start_offset >= 16:
                iv = content[:16]
                content = content[16:]
                start_offset -= 16
            else:
                iv = encryption_data.content_encryption_IV

            if end_range == blob_size - 1:
                unpad = True
        else:
            unpad = True
            iv = encryption_data.content_encryption_IV

        if blob_type == 'PageBlob':
            unpad = False

        if iv is None:
            raise ValueError("Missing required metadata for Encryption V1")

        cipher = _generate_AES_CBC_cipher(content_encryption_key, iv)
        decryptor = cipher.decryptor()

        content = decryptor.update(content) + decryptor.finalize()
        if unpad:
            unpadder = PKCS7(128).unpadder()
            content = unpadder.update(content) + unpadder.finalize()

        return content[start_offset: len(content) - end_offset]

    if version in _ENCRYPTION_V2_PROTOCOLS:
        # We assume the content contains only full encryption regions
        total_size = len(content)
        offset = 0

        if encryption_data.encrypted_region_info is None:
            raise ValueError("Missing required metadata for Encryption V2")

        nonce_length = encryption_data.encrypted_region_info.nonce_length
        data_length = encryption_data.encrypted_region_info.data_length
        tag_length = encryption_data.encrypted_region_info.tag_length
        region_length = nonce_length + data_length + tag_length

        decrypted_content = bytearray()
        while offset < total_size:
            # Process one encryption region at a time
            process_size = min(region_length, total_size)
            encrypted_region = content[offset:offset + process_size]

            # First bytes are the nonce
            nonce = encrypted_region[:nonce_length]
            ciphertext_with_tag = encrypted_region[nonce_length:]

            aesgcm = AESGCM(content_encryption_key)
            decrypted_data = aesgcm.decrypt(nonce, ciphertext_with_tag, None)
            decrypted_content.extend(decrypted_data)

            offset += process_size

        # Read the caller requested data from the decrypted content
        return decrypted_content[start_offset:end_offset]

    raise ValueError('Specified encryption version is not supported.')


def get_blob_encryptor_and_padder(
    cek: Optional[bytes],
    iv: Optional[bytes],
    should_pad: bool
) -> Tuple[Optional["AEADEncryptionContext"], Optional["PaddingContext"]]:
    encryptor = None
    padder = None

    if cek is not None and iv is not None:
        cipher = _generate_AES_CBC_cipher(cek, iv)
        encryptor = cipher.encryptor()
        padder = PKCS7(128).padder() if should_pad else None

    return encryptor, padder


def encrypt_queue_message(message: str, key_encryption_key: KeyEncryptionKey, version: str) -> str:
    """
    Encrypts the given plain text message using the given protocol version.
    Wraps the generated content-encryption-key using the user-provided key-encryption-key (kek).
    Returns a json-formatted string containing the encrypted message and the encryption metadata.

    :param str message:
        The plain text message to be encrypted.
    :param KeyEncryptionKey key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param str version: The client encryption version to use.
    :return: A json-formatted string containing the encrypted message and the encryption metadata.
    :rtype: str
    """

    _validate_not_none('message', message)
    _validate_not_none('key_encryption_key', key_encryption_key)
    _validate_key_encryption_key_wrap(key_encryption_key)

    # Queue encoding functions all return unicode strings, and encryption should
    # operate on binary strings.
    message_as_bytes: bytes = message.encode('utf-8')

    if version == _ENCRYPTION_PROTOCOL_V1:
        # AES256 CBC uses 256 bit (32 byte) keys and always with 16 byte blocks
        content_encryption_key = os.urandom(32)
        initialization_vector = os.urandom(16)

        cipher = _generate_AES_CBC_cipher(content_encryption_key, initialization_vector)

        # PKCS7 with 16 byte blocks ensures compatibility with AES.
        padder = PKCS7(128).padder()
        padded_data = padder.update(message_as_bytes) + padder.finalize()

        # Encrypt the data.
        encryptor = cipher.encryptor()
        encrypted_data = encryptor.update(padded_data) + encryptor.finalize()

    elif version == _ENCRYPTION_PROTOCOL_V2:
        # AES256 GCM uses 256 bit (32 byte) keys and a 12 byte nonce.
        content_encryption_key = os.urandom(32)
        initialization_vector = None

        # The nonce MUST be different for each key
        nonce = os.urandom(12)
        aesgcm = AESGCM(content_encryption_key)

        # Returns ciphertext + tag
        cipertext_with_tag = aesgcm.encrypt(nonce, message_as_bytes, None)
        encrypted_data = nonce + cipertext_with_tag

    else:
        raise ValueError("Invalid encryption version specified.")

    # Build the dictionary structure.
    queue_message = {'EncryptedMessageContents': encode_base64(encrypted_data),
                     'EncryptionData': _generate_encryption_data_dict(key_encryption_key,
                                                                      content_encryption_key,
                                                                      initialization_vector,
                                                                      version)}

    return dumps(queue_message)


def decrypt_queue_message(
    message: str,
    response: "PipelineResponse",
    require_encryption: bool,
    key_encryption_key: Optional[KeyEncryptionKey],
    resolver: Optional[Callable[[str], KeyEncryptionKey]]
) -> str:
    """
    Returns the decrypted message contents from an EncryptedQueueMessage.
    If no encryption metadata is present, will return the unaltered message.
    :param str message:
        The JSON formatted QueueEncryptedMessage contents with all associated metadata.
    :param Any response:
        The pipeline response used to generate an error with.
    :param bool require_encryption:
        If set, will enforce that the retrieved messages are encrypted and decrypt them.
    :param Optional[KeyEncryptionKey] key_encryption_key:
        The user-provided key-encryption-key. Must implement the following methods:
        wrap_key(key)
            - Wraps the specified key using an algorithm of the user's choice.
        get_key_wrap_algorithm()
            - Returns the algorithm used to wrap the specified symmetric key.
        get_kid()
            - Returns a string key id for this key-encryption-key.
    :param Optional[Callable[[str], KeyEncryptionKey]] resolver:
        The user-provided key resolver. Uses the kid string to return a key-encryption-key
        implementing the interface defined above.
    :return: The plain text message from the queue message.
    :rtype: str
    """
    response = response.http_response

    try:
        deserialized_message: Dict[str, Any] = loads(message)

        encryption_data = _dict_to_encryption_data(deserialized_message['EncryptionData'])
        decoded_data = decode_base64_to_bytes(deserialized_message['EncryptedMessageContents'])
    except (KeyError, ValueError) as exc:
        # Message was not json formatted and so was not encrypted
        # or the user provided a json formatted message
        # or the metadata was malformed.
        if require_encryption:
            raise ValueError(
                'Encryption required, but received message does not contain appropriate metatadata. ' + \
                'Message was either not encrypted or metadata was incorrect.') from exc

        return message
    try:
        return _decrypt_message(decoded_data, encryption_data, key_encryption_key, resolver).decode('utf-8')
    except Exception as error:
        raise HttpResponseError(
            message="Decryption failed.",
            response=response, #type: ignore [arg-type]
            error=error) from error