feat: add Key Committing AES-GCM and AES-CBC support

compliance_exceptions/decryption_exceptions.md

@@ -0,0 +1,49 @@
+# Compliance Exceptions for Decryption Implementation
+
+## Summary
+
+The Python S3 Encryption Client does not currently support Ranged Gets.
+Ranged Gets allow downloading and decrypting a subset of bytes from an encrypted S3 object.
+This is an optional feature per the specification ("MAY support") and is planned for a future release.
+
+## Ranged Gets
+
+##= specification/s3-encryption/decryption.md#ranged-gets
+##= type=exception
+##% The S3EC MAY support the "range" parameter on GetObject which specifies a subset of bytes to download and decrypt.
+
+Justification: Ranged Gets are not yet implemented in the Python S3 Encryption Client. The specification uses MAY, making this an optional feature. This is planned for a future release.
+
+---
+
+##= specification/s3-encryption/decryption.md#ranged-gets
+##= type=exception
+##% If the S3EC supports Ranged Gets, the S3EC MUST adjust the customer-provided range to include the beginning and end of the cipher blocks for the given range.
+
+Justification: Not applicable since Ranged Gets are not yet supported. When Ranged Gets are implemented, this requirement will be fulfilled.
+
+---
+
+##= specification/s3-encryption/decryption.md#ranged-gets
+##= type=exception
+##% If the object was encrypted with ALG_AES_256_GCM_IV12_TAG16_NO_KDF, then ALG_AES_256_CTR_IV16_TAG16_NO_KDF MUST be used to decrypt the range of the object.
+
+Justification: Not applicable since Ranged Gets are not yet supported. When Ranged Gets are implemented, the correct CTR-mode algorithm suite will be used for GCM-encrypted objects.
+
+---
+
+##= specification/s3-encryption/decryption.md#ranged-gets
+##= type=exception
+##% If the object was encrypted with ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY, then ALG_AES_256_CTR_HKDF_SHA512_COMMIT_KEY MUST be used to decrypt the range of the object.
+
+Justification: Not applicable since Ranged Gets are not yet supported. When Ranged Gets are implemented, the correct CTR-mode algorithm suite will be used for key-committing objects.
+
+---
+
+##= specification/s3-encryption/decryption.md#ranged-gets
+##= type=exception
+##% If the GetObject response contains a range, but the GetObject request does not contain a range, the S3EC MUST throw an exception.
+
+Justification: Not applicable since Ranged Gets are not yet supported. When Ranged Gets are implemented, this validation will be added to detect unexpected range responses.
+
+---

compliance_exceptions/encryption_exceptions.md

@@ -0,0 +1,63 @@
+# Compliance Exceptions for Encryption Implementation
+
+## Summary
+
+The Python S3 Encryption Client does not implement AES-CTR algorithm suites (used only for ranged-get decryption),
+does not yet validate IV/Message ID for zero values, does not validate maximum plaintext length,
+and relies on Python's `cryptography` library to automatically append GCM auth tags.
+
+## AES-CTR Algorithm Suites
+
+##= specification/s3-encryption/encryption.md#alg-aes-256-ctr-hkdf-sha512-commit-key
+##= type=exception
+##% Attempts to encrypt using key committing AES-CTR MUST fail.
+
+Justification: The AES-CTR algorithm suites are only used for ranged-get decryption. Since ranged gets are not yet implemented, these algorithm suites are not defined in the `AlgorithmSuite` enum and cannot be selected for encryption. The constraint is satisfied structurally.
+
+---
+
+##= specification/s3-encryption/encryption.md#alg-aes-256-ctr-iv16-tag16-no-kdf
+##= type=exception
+##% Attempts to encrypt using AES-CTR MUST fail.
+
+Justification: Same as above. AES-CTR is not available as an algorithm suite option, so it cannot be used for encryption.
+
+---
+
+## GCM Auth Tag Appending
+
+##= specification/s3-encryption/encryption.md#alg-aes-256-gcm-hkdf-sha512-commit-key
+##= type=exception
+##% The client MUST append the GCM auth tag to the ciphertext if the underlying crypto provider does not do so automatically.
+
+Justification: Python's `cryptography` library (`AESGCM.encrypt`) automatically appends the GCM authentication tag to the ciphertext. No manual appending is needed.
+
+---
+
+##= specification/s3-encryption/encryption.md#alg-aes-256-gcm-iv12-tag16-no-kdf
+##= type=exception
+##% The client MUST append the GCM auth tag to the ciphertext if the underlying crypto provider does not do so automatically.
+
+Justification: Python's `cryptography` library (`AESGCM.encrypt`) automatically appends the GCM authentication tag to the ciphertext. No manual appending is needed.
+
+---
+
+## Cipher Initialization Validation
+
+##= specification/s3-encryption/encryption.md#cipher-initialization
+##= type=exception
+##% The client SHOULD validate that the generated IV or Message ID is not zeros.
+
+Justification: This SHOULD-level validation is not yet implemented. The IV and Message ID are generated using `os.urandom()`, which is cryptographically secure and extremely unlikely to produce all-zero output. This validation is planned for a future release.
+
+---
+
+## Plaintext Length Validation
+
+##= specification/s3-encryption/encryption.md#content-encryption
+##= type=exception
+##% The client MUST validate that the length of the plaintext bytes does not exceed the algorithm suite's cipher's maximum content length in bytes.
+
+Justification: Maximum plaintext length validation is not yet implemented. For AES-GCM with a 12-byte IV, the maximum plaintext size is approximately 64 GiB, which exceeds practical S3 single-object upload limits. This validation is planned for a future release.
+
+---

src/s3_encryption/__init__.py

@@ -15,16 +15,39 @@
     DefaultCryptoMaterialsManager,
 )
 from .materials.keyring import AbstractKeyring
+from .materials.materials import AlgorithmSuite, CommitmentPolicy
 from .pipelines import GetEncryptedObjectPipeline, PutEncryptedObjectPipeline
 
 S3_METADATA_PREFIX = "x-amz-meta-"
 
+# Thread-local context attribute names
+_CTX_ENCRYPTION_CONTEXT = "encryption_context"
+_CTX_BUCKET = "bucket"
+_CTX_KEY = "key"
+_CTX_S3_CLIENT = "s3_client"
+_CTX_INSTRUCTION_FILE_MODE = "instruction_file_mode"
+
+# Attributes to clean up after get_object completes
+# (s3_client is intentionally excluded — it is not request-scoped)
+_GET_OBJECT_CLEANUP_ATTRS = (_CTX_ENCRYPTION_CONTEXT, _CTX_BUCKET, _CTX_KEY)
+
 
 @define
 class S3EncryptionClientConfig:
     """Configuration object for the S3 Encryption Client."""
 
     keyring: AbstractKeyring
+    encryption_algorithm: AlgorithmSuite = field(
+        default=AlgorithmSuite.ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY
+    )
+    commitment_policy: CommitmentPolicy = field(
+        default=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
+    )
+    ##= specification/s3-encryption/client.md#enable-legacy-unauthenticated-modes
+    ##% The S3EC MUST support the option to enable or disable legacy unauthenticated modes (content encryption algorithms).
+    ##= specification/s3-encryption/client.md#enable-legacy-unauthenticated-modes
+    ##% The option to enable legacy unauthenticated modes MUST be set to false by default.
+    enable_legacy_unauthenticated_modes: bool = field(default=False)
     cmm: AbstractCryptoMaterialsManager = field()
     ##= specification/s3-encryption/data-format/metadata-strategy.md#instruction-file
     ##= type=implementation
@@ -41,6 +64,51 @@ class S3EncryptionClientConfig:
     def _default_cmm_for_keyring(self):
         return DefaultCryptoMaterialsManager(self.keyring)
 
+    ##= specification/s3-encryption/client.md#encryption-algorithm
+    ##% The S3EC MUST validate that the configured encryption algorithm is not legacy.
+    ##= specification/s3-encryption/client.md#encryption-algorithm
+    ##% If the configured encryption algorithm is legacy, then the S3EC MUST throw an exception.
+    ##= specification/s3-encryption/client.md#key-commitment
+    ##% The S3EC MUST validate the configured Encryption Algorithm against the provided key commitment policy.
+    ##= specification/s3-encryption/client.md#key-commitment
+    ##% If the configured Encryption Algorithm is incompatible with the key commitment policy, then it MUST throw an exception.
+    def __attrs_post_init__(self):
+        """Validate algorithm suite and commitment policy configuration."""
+        if self.encryption_algorithm.is_legacy:
+            raise S3EncryptionClientError(
+                f"Cannot configure S3 Encryption Client with legacy algorithm suite "
+                f"{self.encryption_algorithm.name}. Legacy algorithm suites are only "
+                f"supported for decryption (and enable_legacy_unauthenticated_modes is True)."
+            )
+
+        ##= specification/s3-encryption/key-commitment.md#commitment-policy
+        ##% When the commitment policy is REQUIRE_ENCRYPT_ALLOW_DECRYPT, the S3EC MUST only encrypt using an algorithm suite which supports key commitment.
+        ##= specification/s3-encryption/key-commitment.md#commitment-policy
+        ##% When the commitment policy is REQUIRE_ENCRYPT_REQUIRE_DECRYPT, the S3EC MUST only encrypt using an algorithm suite which supports key commitment.
+        if (
+            self.commitment_policy
+            in (
+                CommitmentPolicy.REQUIRE_ENCRYPT_ALLOW_DECRYPT,
+                CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT,
+            )
+            and not self.encryption_algorithm.supports_key_commitment
+        ):
+            raise S3EncryptionClientError(
+                f"Commitment policy {self.commitment_policy.name} requires a key-committing "
+                f"algorithm suite, but {self.encryption_algorithm.name} does not support key commitment."
+            )
+
+        ##= specification/s3-encryption/key-commitment.md#commitment-policy
+        ##% When the commitment policy is FORBID_ENCRYPT_ALLOW_DECRYPT, the S3EC MUST NOT encrypt using an algorithm suite which supports key commitment.
+        if (
+            self.commitment_policy == CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT
+            and self.encryption_algorithm.supports_key_commitment
+        ):
+            raise S3EncryptionClientError(
+                f"Commitment policy {self.commitment_policy.name} forbids key-committing "
+                f"algorithm suites, but {self.encryption_algorithm.name} supports key commitment."
+            )
+
 
 class S3EncryptionClientPlugin:
     """Plugin that adds encryption/decryption capabilities to a boto3 S3 client.
@@ -67,7 +135,7 @@ def on_put_object_before_call(self, params, **kwargs):
             params: Dictionary of parameters for the PutObject call (after serialization)
             **kwargs: Additional event arguments
         """
-        if getattr(self._context, "instruction_file_mode", False):
+        if getattr(self._context, _CTX_INSTRUCTION_FILE_MODE, False):
             raise S3EncryptionClientError(
                 "Instruction file mode is exclusively for reading instruction files "
                 "and not supported in put_object!"
@@ -88,11 +156,12 @@ def on_put_object_before_call(self, params, **kwargs):
             # Unexpected body type - should not happen as boto3 validates before this point
             raise S3EncryptionClientError("Unexpected type of body parameter!")
 
-        encryption_context = getattr(self._context, "encryption_context", None)
+        encryption_context = getattr(self._context, _CTX_ENCRYPTION_CONTEXT, None)
 
-        pipeline = PutEncryptedObjectPipeline(self.config.cmm)
+        pipeline = PutEncryptedObjectPipeline(self.config.cmm, self.config.encryption_algorithm)
         encrypted_data, encryption_metadata = pipeline.encrypt(
-            body_bytes, encryption_context=encryption_context
+            body_bytes,
+            encryption_context=encryption_context,
         )
 
         params["body"] = encrypted_data
@@ -118,12 +187,12 @@ def on_get_object_after_call(self, parsed, **kwargs):
             **kwargs: Additional event arguments (includes 'params' with request parameters)
         """
         # Check if plaintext mode is enabled via thread-local flag
-        if getattr(self._context, "instruction_file_mode", False):
+        if getattr(self._context, _CTX_INSTRUCTION_FILE_MODE, False):
             self.process_instruction_file(parsed)
             return
 
         # Get encryption context from thread-local storage (set by get_object wrapper)
-        encryption_context = getattr(self._context, "encryption_context", None)
+        encryption_context = getattr(self._context, _CTX_ENCRYPTION_CONTEXT, None)
 
         # The parsed response already has the Body as a StreamingBody
         # We need to read it, decrypt it, and replace it
@@ -137,13 +206,15 @@ def on_get_object_after_call(self, parsed, **kwargs):
         # Create a pipeline and decrypt the data
         pipeline = GetEncryptedObjectPipeline(
             self.config.cmm,
-            s3_client=getattr(self._context, "s3_client", None),
+            commitment_policy=self.config.commitment_policy,
+            s3_client=getattr(self._context, _CTX_S3_CLIENT, None),
+            enable_legacy_unauthenticated_modes=self.config.enable_legacy_unauthenticated_modes,
         )
         decrypted_data = pipeline.decrypt(
             response,
             encryption_context,
-            bucket=getattr(self._context, "bucket", None),
-            key=getattr(self._context, "key", None),
+            bucket=getattr(self._context, _CTX_BUCKET, None),
+            key=getattr(self._context, _CTX_KEY, None),
             instruction_suffix=self.config.instruction_file_suffix,
         )
 
@@ -163,7 +234,7 @@ def process_instruction_file(self, parsed):
         Args:
             parsed: Dictionary containing the parsed response
         """
-        instruction_key = getattr(self._context, "key", None)
+        instruction_key = getattr(self._context, _CTX_KEY, None)
 
         # In plaintext mode, parse instruction file and append to metadata
         existing_metadata = parsed.get("Metadata", {})
@@ -242,8 +313,8 @@ def put_object(self, **kwargs):
             raise S3EncryptionClientError(f"Failed to encrypt object: {str(e)}") from e
         finally:
             # Clean up thread-local storage
-            if hasattr(self._plugin._context, "encryption_context"):
-                delattr(self._plugin._context, "encryption_context")
+            if hasattr(self._plugin._context, _CTX_ENCRYPTION_CONTEXT):
+                delattr(self._plugin._context, _CTX_ENCRYPTION_CONTEXT)
 
     def get_object(self, **kwargs):
         """Download and decrypt an object from S3.
@@ -266,12 +337,12 @@ def get_object(self, **kwargs):
         encryption_context = kwargs.pop("EncryptionContext", None)
 
         # Store encryption context in thread-local storage for the event handler
-        self._plugin._context.encryption_context = encryption_context
+        setattr(self._plugin._context, _CTX_ENCRYPTION_CONTEXT, encryption_context)
         # Store wrapped client in thread-local storage for
         # the event handler to fetch instruction files
-        self._plugin._context.s3_client = self.wrapped_s3_client
-        self._plugin._context.bucket = kwargs.get("Bucket")
-        self._plugin._context.key = kwargs.get("Key")
+        setattr(self._plugin._context, _CTX_S3_CLIENT, self.wrapped_s3_client)
+        setattr(self._plugin._context, _CTX_BUCKET, kwargs.get("Bucket"))
+        setattr(self._plugin._context, _CTX_KEY, kwargs.get("Key"))
 
         try:
             return self.wrapped_s3_client.get_object(**kwargs)
@@ -284,7 +355,6 @@ def get_object(self, **kwargs):
         finally:
             # Clean up thread-local storage;
             # do not clean up the client as it is not thread local only
-            attrs = ["encryption_context", "Bucket", "Key"]
-            for attr in attrs:
+            for attr in _GET_OBJECT_CLEANUP_ATTRS:
                 if hasattr(self._plugin._context, attr):
                     delattr(self._plugin._context, attr)