feat: add RawAesKeyring for bring-your-own-key encryption

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",

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

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."
             )