in src/dynamodb_encryption_sdk/material_providers/aws_kms.py [0:0]
def _decrypt_initial_material(self, encryption_context):
# type: (EncryptionContext) -> bytes
"""Decrypt an encrypted initial cryptographic material value.
:param encryption_context: Encryption context providing information about request
:type encryption_context: EncryptionContext
:returns: Plaintext of initial cryptographic material
:rtype: bytes
"""
key_id = self._select_key_id(encryption_context)
self._validate_key_id(key_id, encryption_context)
kms_encryption_context = self._kms_encryption_context(
encryption_context=encryption_context,
encryption_description=encryption_context.material_description.get(
MaterialDescriptionKeys.CONTENT_ENCRYPTION_ALGORITHM.value
),
signing_description=encryption_context.material_description.get(
MaterialDescriptionKeys.ITEM_SIGNATURE_ALGORITHM.value
),
)
encrypted_initial_material = base64.b64decode(
to_bytes(encryption_context.material_description.get(MaterialDescriptionKeys.WRAPPED_DATA_KEY.value))
)
kms_params = dict(CiphertextBlob=encrypted_initial_material, EncryptionContext=kms_encryption_context)
if self._grant_tokens:
kms_params["GrantTokens"] = self._grant_tokens
# Catch any boto3 errors and normalize to expected UnwrappingError
try:
response = self._client(key_id).decrypt(**kms_params)
return response["Plaintext"]
except (botocore.exceptions.ClientError, KeyError):
message = "Failed to unwrap AWS KMS protected materials"
_LOGGER.exception(message)
raise UnwrappingError(message)