# 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/ 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