diff --git a/src/s3_encryption/materials/__init__.py b/src/s3_encryption/materials/__init__.py index c5cc7d6d..0a2bcde3 100644 --- a/src/s3_encryption/materials/__init__.py +++ b/src/s3_encryption/materials/__init__.py @@ -11,10 +11,12 @@ from .keyring import AbstractKeyring from .kms_keyring import KmsKeyring from .materials import AlgorithmSuite, CommitmentPolicy, EncryptionMaterials +from .raw_aes_keyring import RawAesKeyring __all__ = [ "AbstractKeyring", "KmsKeyring", + "RawAesKeyring", "AbstractCryptoMaterialsManager", "DefaultCryptoMaterialsManager", "EncryptedDataKey", diff --git a/src/s3_encryption/materials/raw_aes_keyring.py b/src/s3_encryption/materials/raw_aes_keyring.py new file mode 100644 index 00000000..7925407d --- /dev/null +++ b/src/s3_encryption/materials/raw_aes_keyring.py @@ -0,0 +1,174 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Raw AES keyring module for S3 Encryption Client. + +This module provides a keyring implementation that wraps data keys using a +locally-held, caller-provided AES key ("bring your own key"), with no +dependency on a remote key management service such as AWS KMS. +""" + +import os + +from attrs import define, field +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from ..exceptions import S3EncryptionClientError +from .encrypted_data_key import EncryptedDataKey +from .keyring import S3Keyring + +# Canonical wrapping-algorithm identifier for this keyring. This is the value +# stored in object metadata (x-amz-wrap-alg / V3 code "02") so that decrypt +# can recognize which keyring produced an EncryptedDataKey. +AES_WRAP_ALGORITHM = "AES/GCM" + +# Reserved encryption-context key used to bind an EncryptedDataKey to the +# wrapping key that produced it, so decrypt can fail fast on a key mismatch +# instead of only failing during (or after) the AES-GCM unwrap. +RAW_AES_KEY_NAME_CONTEXT_KEY = "aws:x-amz-raw-aes-key-name" + +_VALID_WRAPPING_KEY_LENGTHS = (16, 24, 32) +_NONCE_LENGTH_BYTES = 12 + + +def _serialize_encryption_context(encryption_context: dict) -> bytes: + """Serialize an encryption context into deterministic bytes for use as AAD. + + Args: + encryption_context: The encryption context to serialize. + + Returns: + bytes: A deterministic byte representation of the encryption context. + """ + serialized = b"" + for key in sorted(encryption_context): + value = encryption_context[key] + serialized += key.encode("utf-8") + b"\x00" + value.encode("utf-8") + b"\x00" + return serialized + + +@define +class RawAesKeyring(S3Keyring): + """AES implementation of the S3 keyring for caller-supplied ("bring your own") keys. + + This keyring wraps and unwraps data keys locally using AES-GCM with a + wrapping key supplied by the caller, rather than delegating to a remote + key management service such as AWS KMS. + + Attributes: + key_name (str): A caller-chosen identifier for the wrapping key. Stored + in the encryption context / material description alongside the + encrypted data key and validated on decrypt to ensure the correct + wrapping key is used. + wrapping_key (bytes): The raw AES wrapping key. Must be 16, 24, or 32 + bytes (AES-128, AES-192, or AES-256). + """ + + key_name: str = field() + wrapping_key: bytes = field() + + @wrapping_key.validator + def _check_wrapping_key(self, attribute, value): + if ( + not isinstance(value, (bytes, bytearray)) + or len(value) not in _VALID_WRAPPING_KEY_LENGTHS + ): + raise S3EncryptionClientError( + "RawAesKeyring wrapping_key must be 16, 24, or 32 bytes " + f"(AES-128, AES-192, or AES-256), got " + f"{len(value) if isinstance(value, (bytes, bytearray)) else type(value)}." + ) + + def on_encrypt(self, enc_materials): + """Generate a plaintext data key and wrap it with the local AES wrapping key. + + Args: + enc_materials (EncryptionMaterials or dict): Encryption materials to process + + Returns: + EncryptionMaterials: The processed encryption materials with a locally + wrapped data key + """ + enc_materials = super().on_encrypt(enc_materials) + + encryption_context = enc_materials.encryption_context + if RAW_AES_KEY_NAME_CONTEXT_KEY in encryption_context: + raise S3EncryptionClientError( + f"{RAW_AES_KEY_NAME_CONTEXT_KEY} is a reserved key for the S3 encryption client" + ) + encryption_context[RAW_AES_KEY_NAME_CONTEXT_KEY] = self.key_name + + plaintext_data_key = os.urandom(enc_materials.encryption_algorithm.data_key_length_bytes) + nonce = os.urandom(_NONCE_LENGTH_BYTES) + aad = _serialize_encryption_context(encryption_context) + + aesgcm = AESGCM(self.wrapping_key) + wrapped_key = aesgcm.encrypt(nonce, plaintext_data_key, aad) + + enc_materials.encrypted_data_key = EncryptedDataKey( + key_provider_id=b"S3Keyring", + key_provider_info=AES_WRAP_ALGORITHM, + encrypted_data_key=nonce + wrapped_key, + ) + enc_materials.plaintext_data_key = plaintext_data_key + return enc_materials + + def on_decrypt(self, dec_materials, encrypted_data_keys=None): + """Unwrap the encrypted data key using the local AES wrapping key. + + Args: + dec_materials (DecryptionMaterials): A DecryptionMaterials instance containing + decryption materials + encrypted_data_keys (List[EncryptedDataKey], optional): A list of encrypted data + keys to try. + + Returns: + DecryptionMaterials: The updated dec_materials with the plaintext data key + """ + dec_materials = super().on_decrypt(dec_materials, encrypted_data_keys) + + edks = ( + encrypted_data_keys + if encrypted_data_keys is not None + else dec_materials.encrypted_data_keys + ) + edk = edks[0] + + if edk.key_provider_info != AES_WRAP_ALGORITHM: + raise S3EncryptionClientError( + f"{edk.key_provider_info} is not a valid key wrapping algorithm for RawAesKeyring!" + ) + + encryption_context_from_request = dec_materials.encryption_context_from_request + encryption_context_stored = dec_materials.encryption_context_stored + + if RAW_AES_KEY_NAME_CONTEXT_KEY in encryption_context_from_request: + raise S3EncryptionClientError( + f"{RAW_AES_KEY_NAME_CONTEXT_KEY} is a reserved key for the S3 encryption client" + ) + + encryption_context_stored_copy = encryption_context_stored.copy() + stored_key_name = encryption_context_stored_copy.pop(RAW_AES_KEY_NAME_CONTEXT_KEY, None) + if stored_key_name != self.key_name: + raise S3EncryptionClientError( + f"Encrypted data key was wrapped with key_name={stored_key_name!r}, " + f"but this RawAesKeyring is configured with key_name={self.key_name!r}." + ) + + if encryption_context_stored_copy != encryption_context_from_request: + raise S3EncryptionClientError( + "Provided encryption context does not match information retrieved from S3" + ) + + nonce = edk.encrypted_data_key[:_NONCE_LENGTH_BYTES] + wrapped_key = edk.encrypted_data_key[_NONCE_LENGTH_BYTES:] + aad = _serialize_encryption_context(dec_materials.encryption_context_stored) + + aesgcm = AESGCM(self.wrapping_key) + try: + dec_materials.plaintext_data_key = aesgcm.decrypt(nonce, wrapped_key, aad) + except Exception as e: + raise S3EncryptionClientError( + f"Failed to unwrap data key with RawAesKeyring: {e}" + ) from e + + return dec_materials diff --git a/src/s3_encryption/pipelines.py b/src/s3_encryption/pipelines.py index ca200a7f..1c87580c 100644 --- a/src/s3_encryption/pipelines.py +++ b/src/s3_encryption/pipelines.py @@ -35,6 +35,23 @@ from .metadata import ObjectMetadata from .stream import DecryptingStream +##= specification/s3-encryption/data-format/content-metadata.md#v3-only +##% The V3 format uses compression here such that each wrapping algorithm is represented by a two digit string. +##= specification/s3-encryption/data-format/content-metadata.md#v3-only +##% - The wrapping algorithm value "02" MUST be translated to AES/GCM upon retrieval, and vice versa on write. +##= specification/s3-encryption/data-format/content-metadata.md#v3-only +##% - The wrapping algorithm value "12" MUST be translated to kms+context upon retrieval, and vice versa on write. +##= specification/s3-encryption/data-format/content-metadata.md#v3-only +##% - The wrapping algorithm value "22" MUST be translated to RSA-OAEP-SHA1 upon retrieval, and vice versa on write. +# Maps a keyring's canonical wrapping-algorithm identifier (EncryptedDataKey.key_provider_info) +# to its two-digit V3 compressed code, and vice versa. +_V3_WRAP_ALG_TO_CODE = { + "AES/GCM": "02", + "kms+context": "12", + "RSA-OAEP-SHA1": "22", +} +_V3_CODE_TO_WRAP_ALG = {code: alg for alg, code in _V3_WRAP_ALG_TO_CODE.items()} + @define class PutEncryptedObjectPipeline: @@ -107,7 +124,7 @@ def _encrypt_gcm(self, plaintext, enc_mats, edk_bytes): ##% The generated IV or Message ID MUST be set or returned from the encryption metadata = ObjectMetadata( encrypted_data_key_v2=b64_edk, - encrypted_data_key_algorithm="kms+context", + encrypted_data_key_algorithm=enc_mats.encrypted_data_key.key_provider_info, content_iv=b64_iv, content_cipher="AES/GCM/NoPadding", encrypted_data_key_context=enc_mats.encryption_context, @@ -150,6 +167,14 @@ def _encrypt_kc_gcm(self, plaintext, enc_mats, edk_bytes): b64_message_id = base64.b64encode(message_id).decode("utf-8") b64_commit_key = base64.b64encode(commit_key).decode("utf-8") + wrap_alg = enc_mats.encrypted_data_key.key_provider_info + wrap_alg_code = _V3_WRAP_ALG_TO_CODE.get(wrap_alg) + if wrap_alg_code is None: + raise S3EncryptionClientError( + f"Unknown wrapping algorithm '{wrap_alg}' returned by keyring. " + f"Valid values are: {list(_V3_WRAP_ALG_TO_CODE)}." + ) + ##= specification/s3-encryption/encryption.md#content-encryption ##= type=implementation ##% The generated IV or Message ID MUST be set or returned from the encryption @@ -158,15 +183,18 @@ def _encrypt_kc_gcm(self, plaintext, enc_mats, edk_bytes): ##= type=implementation ##% The derived key commitment value MUST be set or returned from the encryption ##% process such that it can be included in the content metadata. + ##= specification/s3-encryption/data-format/content-metadata.md#v3-only + ##% The Encryption Context value MUST be used for wrapping algorithm `kms+context` or `12`. + ##% The Material Description MUST be used for wrapping algorithms `AES/GCM` (`02`) and `RSA-OAEP-SHA1` (`22`). + metadata_context = enc_mats.encryption_context if enc_mats.encryption_context else None metadata = ObjectMetadata( content_cipher_v3=str(algorithm_suite.suite_id), - encrypted_data_key_algorithm_v3="12", + encrypted_data_key_algorithm_v3=wrap_alg_code, encrypted_data_key_v3=b64_edk, message_id_v3=b64_message_id, key_commitment_v3=b64_commit_key, - encryption_context_v3=( - enc_mats.encryption_context if enc_mats.encryption_context else None - ), + encryption_context_v3=metadata_context if wrap_alg == "kms+context" else None, + mat_desc_v3=metadata_context if wrap_alg != "kms+context" else None, ) return encrypted_data, metadata.to_dict() @@ -227,7 +255,7 @@ def _init_gcm(self, enc_mats, edk_bytes): self._encryptor = cipher.encryptor() self._metadata = ObjectMetadata( encrypted_data_key_v2=base64.b64encode(edk_bytes).decode("utf-8"), - encrypted_data_key_algorithm="kms+context", + encrypted_data_key_algorithm=enc_mats.encrypted_data_key.key_provider_info, content_iv=base64.b64encode(iv).decode("utf-8"), content_cipher="AES/GCM/NoPadding", encrypted_data_key_context=enc_mats.encryption_context, @@ -244,15 +272,24 @@ def _init_kc_gcm(self, enc_mats, edk_bytes): ) self._encryptor = cipher.encryptor() self._encryptor.authenticate_additional_data(algorithm_suite.suite_id_bytes) + + wrap_alg = enc_mats.encrypted_data_key.key_provider_info + wrap_alg_code = _V3_WRAP_ALG_TO_CODE.get(wrap_alg) + if wrap_alg_code is None: + raise S3EncryptionClientError( + f"Unknown wrapping algorithm '{wrap_alg}' returned by keyring. " + f"Valid values are: {list(_V3_WRAP_ALG_TO_CODE)}." + ) + + metadata_context = enc_mats.encryption_context if enc_mats.encryption_context else None self._metadata = ObjectMetadata( content_cipher_v3=str(algorithm_suite.suite_id), - encrypted_data_key_algorithm_v3="12", + encrypted_data_key_algorithm_v3=wrap_alg_code, encrypted_data_key_v3=base64.b64encode(edk_bytes).decode("utf-8"), message_id_v3=base64.b64encode(message_id).decode("utf-8"), key_commitment_v3=base64.b64encode(commit_key).decode("utf-8"), - encryption_context_v3=( - enc_mats.encryption_context if enc_mats.encryption_context else None - ), + encryption_context_v3=metadata_context if wrap_alg == "kms+context" else None, + mat_desc_v3=metadata_context if wrap_alg != "kms+context" else None, ).to_dict() @property @@ -739,31 +776,17 @@ def _decrypt_v1_v2( return self.cmm.decrypt_materials(dec_materials) - ##= specification/s3-encryption/data-format/content-metadata.md#v3-only - ##% The V3 format uses compression here such that each wrapping algorithm is represented by a two digit string. - ##= specification/s3-encryption/data-format/content-metadata.md#v3-only - ##% - The wrapping algorithm value "02" MUST be translated to AES/GCM upon retrieval, and vice versa on write. - ##= specification/s3-encryption/data-format/content-metadata.md#v3-only - ##% - The wrapping algorithm value "12" MUST be translated to kms+context upon retrieval, and vice versa on write. - ##= specification/s3-encryption/data-format/content-metadata.md#v3-only - ##% - The wrapping algorithm value "22" MUST be translated to RSA-OAEP-SHA1 upon retrieval, and vice versa on write. - _V3_WRAP_ALG_MAP = { - "02": "AES/GCM", - "12": "kms+context", - "22": "RSA-OAEP-SHA1", - } - def _decrypt_v3(self, metadata, encryption_context) -> DecryptionMaterials: """Prepare V3 decryption materials.""" edk_bytes = base64.b64decode(metadata.encrypted_data_key_v3) # Map V3 compressed wrapping algorithm to canonical key_provider_info raw_wrap_alg = metadata.encrypted_data_key_algorithm_v3 or "12" - wrap_alg = self._V3_WRAP_ALG_MAP.get(raw_wrap_alg) + wrap_alg = _V3_CODE_TO_WRAP_ALG.get(raw_wrap_alg) if wrap_alg is None: raise S3EncryptionClientError( f"Unknown V3 wrapping algorithm: '{raw_wrap_alg}'. " - f"Valid values are: {list(self._V3_WRAP_ALG_MAP.keys())}. " + f"Valid values are: {list(_V3_CODE_TO_WRAP_ALG.keys())}. " f"The object metadata may have been tampered with." ) diff --git a/test/test_raw_aes_keyring.py b/test/test_raw_aes_keyring.py new file mode 100644 index 00000000..b778908d --- /dev/null +++ b/test/test_raw_aes_keyring.py @@ -0,0 +1,215 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for RawAesKeyring implementation.""" + +import os + +import pytest + +from src.s3_encryption.exceptions import S3EncryptionClientError +from src.s3_encryption.materials.encrypted_data_key import EncryptedDataKey +from src.s3_encryption.materials.materials import DecryptionMaterials, EncryptionMaterials +from src.s3_encryption.materials.raw_aes_keyring import ( + AES_WRAP_ALGORITHM, + RAW_AES_KEY_NAME_CONTEXT_KEY, + RawAesKeyring, +) + + +def _wrapping_key(length=32): + return os.urandom(length) + + +class TestRawAesKeyringInitialization: + """Tests for RawAesKeyring initialization.""" + + def test_initialization_with_required_parameters(self): + wrapping_key = _wrapping_key() + keyring = RawAesKeyring(key_name="my-key", wrapping_key=wrapping_key) + + assert keyring.key_name == "my-key" + assert keyring.wrapping_key == wrapping_key + + @pytest.mark.parametrize("length", [16, 24, 32]) + def test_initialization_accepts_valid_aes_key_lengths(self, length): + wrapping_key = _wrapping_key(length) + keyring = RawAesKeyring(key_name="my-key", wrapping_key=wrapping_key) + + assert keyring.wrapping_key == wrapping_key + + @pytest.mark.parametrize("length", [0, 8, 15, 31, 33, 64]) + def test_initialization_rejects_invalid_key_lengths(self, length): + with pytest.raises(S3EncryptionClientError, match="16, 24, or 32 bytes"): + RawAesKeyring(key_name="my-key", wrapping_key=os.urandom(length)) + + def test_initialization_rejects_non_bytes_wrapping_key(self): + with pytest.raises(S3EncryptionClientError, match="16, 24, or 32 bytes"): + RawAesKeyring(key_name="my-key", wrapping_key="not-bytes") + + +class TestRawAesKeyringOnEncrypt: + """Tests for RawAesKeyring encryption operations.""" + + def test_on_encrypt_returns_encryption_materials(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_materials = EncryptionMaterials(encryption_context={"key": "value"}) + + result = keyring.on_encrypt(enc_materials) + + assert isinstance(result, EncryptionMaterials) + + def test_on_encrypt_sets_plaintext_and_encrypted_data_key(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_materials = EncryptionMaterials(encryption_context={}) + + result = keyring.on_encrypt(enc_materials) + + assert result.plaintext_data_key is not None + assert len(result.plaintext_data_key) == result.encryption_algorithm.data_key_length_bytes + assert result.encrypted_data_key is not None + assert result.encrypted_data_key.key_provider_info == AES_WRAP_ALGORITHM + assert result.encrypted_data_key.key_provider_id == b"S3Keyring" + + def test_on_encrypt_generates_unique_plaintext_keys(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + + first = keyring.on_encrypt(EncryptionMaterials(encryption_context={})) + second = keyring.on_encrypt(EncryptionMaterials(encryption_context={})) + + assert first.plaintext_data_key != second.plaintext_data_key + assert ( + first.encrypted_data_key.encrypted_data_key + != second.encrypted_data_key.encrypted_data_key + ) + + def test_on_encrypt_adds_key_name_to_encryption_context(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_materials = EncryptionMaterials(encryption_context={"app": "demo"}) + + result = keyring.on_encrypt(enc_materials) + + assert result.encryption_context[RAW_AES_KEY_NAME_CONTEXT_KEY] == "my-key" + assert result.encryption_context["app"] == "demo" + + def test_on_encrypt_rejects_reserved_key_in_context(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_materials = EncryptionMaterials( + encryption_context={RAW_AES_KEY_NAME_CONTEXT_KEY: "spoofed"} + ) + + with pytest.raises(S3EncryptionClientError, match="reserved key"): + keyring.on_encrypt(enc_materials) + + +class TestRawAesKeyringOnDecrypt: + """Tests for RawAesKeyring decryption operations.""" + + def _encrypt(self, keyring, encryption_context=None): + enc_materials = EncryptionMaterials(encryption_context=encryption_context or {}) + return keyring.on_encrypt(enc_materials) + + def test_on_decrypt_round_trips_plaintext_data_key(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_result = self._encrypt(keyring, {"app": "demo"}) + + dec_materials = DecryptionMaterials( + encrypted_data_keys=[enc_result.encrypted_data_key], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={"app": "demo"}, + ) + + result = keyring.on_decrypt(dec_materials) + + assert result.plaintext_data_key == enc_result.plaintext_data_key + + def test_on_decrypt_fails_with_wrong_wrapping_key(self): + encrypting_keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_result = self._encrypt(encrypting_keyring, {"app": "demo"}) + + decrypting_keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + dec_materials = DecryptionMaterials( + encrypted_data_keys=[enc_result.encrypted_data_key], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={"app": "demo"}, + ) + + with pytest.raises(S3EncryptionClientError, match="Failed to unwrap data key"): + decrypting_keyring.on_decrypt(dec_materials) + + def test_on_decrypt_fails_with_mismatched_key_name(self): + encrypting_keyring = RawAesKeyring(key_name="key-a", wrapping_key=_wrapping_key()) + wrapping_key = encrypting_keyring.wrapping_key + enc_result = self._encrypt(encrypting_keyring, {"app": "demo"}) + + decrypting_keyring = RawAesKeyring(key_name="key-b", wrapping_key=wrapping_key) + dec_materials = DecryptionMaterials( + encrypted_data_keys=[enc_result.encrypted_data_key], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={"app": "demo"}, + ) + + with pytest.raises(S3EncryptionClientError, match="key_name"): + decrypting_keyring.on_decrypt(dec_materials) + + def test_on_decrypt_fails_with_mismatched_encryption_context(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_result = self._encrypt(keyring, {"app": "demo"}) + + dec_materials = DecryptionMaterials( + encrypted_data_keys=[enc_result.encrypted_data_key], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={"app": "different"}, + ) + + with pytest.raises(S3EncryptionClientError, match="does not match"): + keyring.on_decrypt(dec_materials) + + def test_on_decrypt_rejects_reserved_key_in_request_context(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_result = self._encrypt(keyring, {}) + + dec_materials = DecryptionMaterials( + encrypted_data_keys=[enc_result.encrypted_data_key], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={RAW_AES_KEY_NAME_CONTEXT_KEY: "my-key"}, + ) + + with pytest.raises(S3EncryptionClientError, match="reserved key"): + keyring.on_decrypt(dec_materials) + + def test_on_decrypt_rejects_invalid_wrap_algorithm(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + edk = EncryptedDataKey( + key_provider_id=b"S3Keyring", + key_provider_info="kms+context", + encrypted_data_key=b"not-really-encrypted", + ) + dec_materials = DecryptionMaterials( + encrypted_data_keys=[edk], + encryption_context_stored={}, + encryption_context_from_request={}, + ) + + with pytest.raises(S3EncryptionClientError, match="not a valid key wrapping algorithm"): + keyring.on_decrypt(dec_materials) + + def test_on_decrypt_fails_when_ciphertext_tampered(self): + keyring = RawAesKeyring(key_name="my-key", wrapping_key=_wrapping_key()) + enc_result = self._encrypt(keyring, {}) + + tampered_bytes = bytearray(enc_result.encrypted_data_key.encrypted_data_key) + tampered_bytes[-1] ^= 0xFF + tampered_edk = EncryptedDataKey( + key_provider_id=enc_result.encrypted_data_key.key_provider_id, + key_provider_info=enc_result.encrypted_data_key.key_provider_info, + encrypted_data_key=bytes(tampered_bytes), + ) + + dec_materials = DecryptionMaterials( + encrypted_data_keys=[tampered_edk], + encryption_context_stored=enc_result.encryption_context, + encryption_context_from_request={}, + ) + + with pytest.raises(S3EncryptionClientError, match="Failed to unwrap data key"): + keyring.on_decrypt(dec_materials)