feat(decryption): streaming decryption with cipher-agnostic BufferedDecryptingStream and DelayedAuthDecryptingStream

src/s3_encryption/__init__.py

@@ -34,7 +34,29 @@
 
 @define
 class S3EncryptionClientConfig:
-    """Configuration object for the S3 Encryption Client."""
+    """Configuration for the S3 Encryption Client.
+
+    Attributes:
+        keyring: Keyring used for encrypting/decrypting data keys.
+        encryption_algorithm: Algorithm suite for encryption. Defaults to
+            ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY (V3 key-committing).
+        commitment_policy: Key commitment policy for encryption and decryption.
+            Defaults to REQUIRE_ENCRYPT_REQUIRE_DECRYPT.
+        enable_legacy_unauthenticated_modes: If True, allow decryption of objects
+            encrypted with legacy CBC algorithm suites. Defaults to False.
+        cmm: Crypto materials manager. Defaults to a DefaultCryptoMaterialsManager
+            wrapping the provided keyring.
+        instruction_file_suffix: Suffix appended to the S3 object key when
+            fetching instruction files. Defaults to ".instruction".
+        enable_delayed_authentication: If True, release plaintext from streams
+            before GCM tag verification. Defaults to False. Has no effect for
+            CBC encrypted ciphertext, which is always streamed as there is no
+            authentication tag.
+
+    Raises:
+        S3EncryptionClientError: If the encryption algorithm is legacy, or if
+            the algorithm suite is incompatible with the commitment policy.
+    """
 
     keyring: AbstractKeyring
     encryption_algorithm: AlgorithmSuite = field(
@@ -60,6 +82,15 @@ class S3EncryptionClientConfig:
     ##% as its associated object suffixed with ".instruction".
     instruction_file_suffix: str = field(default=".instruction")
 
+    ##= specification/s3-encryption/client.md#enable-delayed-authentication
+    ##= type=implementation
+    ##% The S3EC MUST support the option to enable or disable Delayed Authentication mode.
+
+    ##= specification/s3-encryption/client.md#enable-delayed-authentication
+    ##= type=implication
+    ##% Delayed Authentication mode MUST be set to false by default.
+    enable_delayed_authentication: bool = field(default=False)
+
     @cmm.default
     def _default_cmm_for_keyring(self):
         return DefaultCryptoMaterialsManager(self.keyring)
@@ -197,10 +228,18 @@ def on_get_object_after_call(self, parsed, **kwargs):
         # The parsed response already has the Body as a StreamingBody
         # We need to read it, decrypt it, and replace it
 
+        # content_length is going to the cipher-text's content length
+        content_length = parsed.get("ContentLength")
+        if content_length is None:
+            obj_key = getattr(self._context, _CTX_KEY, None)
+            raise S3EncryptionClientError(
+                f"S3 response is missing ContentLength and is invalid. Key: {obj_key}"
+            )
         # Create a response dict that matches what the pipeline expects
         response = {
             "Body": parsed.get("Body"),
             "Metadata": parsed.get("Metadata", {}),
+            "ContentLength": content_length,
         }
 
         # Create a pipeline and decrypt the data
@@ -212,18 +251,15 @@ def on_get_object_after_call(self, parsed, **kwargs):
         )
         decrypted_data = pipeline.decrypt(
             response,
-            encryption_context,
+            instruction_suffix=self.config.instruction_file_suffix,
+            enable_delayed_authentication=self.config.enable_delayed_authentication,
+            encryption_context=encryption_context,
             bucket=getattr(self._context, _CTX_BUCKET, None),
             key=getattr(self._context, _CTX_KEY, None),
-            instruction_suffix=self.config.instruction_file_suffix,
         )
 
-        # Create a new streaming body with the decrypted data
-        stream = io.BytesIO(decrypted_data)
-        streaming_body = StreamingBody(stream, len(decrypted_data))
-
-        # Replace body with decrypted data
-        parsed["Body"] = streaming_body
+        # Replace body with decrypting stream
+        parsed["Body"] = decrypted_data
 
     def process_instruction_file(self, parsed):
         """Process instruction file in plaintext mode.

src/s3_encryption/buffered_decrypt.py

@@ -0,0 +1,20 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""One Shot decryption into a buffer."""
+
+from io import BytesIO
+
+from botocore.response import StreamingBody
+
+from s3_encryption.decryptor import Decryptor
+
+
+def one_shot_decrypt(streaming_body: StreamingBody, decryptor: Decryptor):
+    """Decrypt a streaming object.
+
+    Args:
+        streaming_body (object): A streaming object.
+        decryptor (Decryptor): Decryptor object.
+    """
+    plaintext = decryptor.finalize(streaming_body.read())
+    return StreamingBody(BytesIO(plaintext), len(plaintext))

src/s3_encryption/decryptor.py

@@ -0,0 +1,141 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Decryptor abstractions for S3 Encryption Client."""
+
+from abc import ABC, abstractmethod
+
+from attrs import define, field
+from cryptography.exceptions import InvalidTag
+
+from .exceptions import S3EncryptionClientError, S3EncryptionClientSecurityError
+
+
+class Decryptor(ABC):
+    """Abstract base class for content decryption.
+
+    Implementations own all cipher and padding state, presenting a uniform
+    streaming interface to the decrypting stream classes.
+    """
+
+    @property
+    @abstractmethod
+    def content_length(self) -> int:
+        """Total byte length of the encrypted content (ciphertext + any trailing tag)."""
+
+    @property
+    @abstractmethod
+    def amount_read(self) -> int:
+        """Number of ciphertext bytes consumed so far."""
+
+    @abstractmethod
+    def update(self, data: bytes) -> bytes:
+        """Process a chunk of ciphertext, returning any available plaintext."""
+
+    @abstractmethod
+    def finalize(self, data: bytes) -> bytes:
+        """Process the final chunk of ciphertext and finalize decryption."""
+
+
+@define
+class AesCbcDecryptor(Decryptor):
+    """AES-CBC decryptor that owns both the cipher and PKCS7 unpadder.
+
+    Args:
+        decryptor: A cryptography CBC cipher decryptor context.
+        unpadder: A cryptography PKCS7 unpadding context.
+        content_length: Total byte length of the CBC ciphertext.
+    """
+
+    _decryptor: object = field()
+    _unpadder: object = field()
+    _content_length: int = field()
+    _amount_read: int = field(init=False, default=0)
+
+    @property
+    def content_length(self) -> int:  # noqa: D102
+        return self._content_length
+
+    @property
+    def amount_read(self) -> int:  # noqa: D102
+        return self._amount_read
+
+    def update(self, data: bytes) -> bytes:
+        """Decrypt a chunk and unpad incrementally."""
+        self._amount_read += len(data)
+        plaintext = self._decryptor.update(data)
+        return self._unpadder.update(plaintext)
+
+    def finalize(self, data: bytes) -> bytes:
+        """Finalize CBC decryption and flush the unpadder."""
+        try:
+            self._amount_read += len(data)
+            plaintext = self._decryptor.update(data) if data else b""
+            plaintext += self._decryptor.finalize()
+            return self._unpadder.update(plaintext) + self._unpadder.finalize()
+        except Exception as e:
+            raise S3EncryptionClientSecurityError(f"Failed to decrypt CBC content: {e}") from e
+
+
+@define
+class AesGcmDecryptor(Decryptor):
+    """AES-GCM decryptor that handles trailing auth tag verification.
+
+    Args:
+        decryptor: A cryptography GCM cipher decryptor context.
+        tag_length: Length of the GCM authentication tag in bytes.
+        content_length: Total byte length of the encrypted content (ciphertext + tag).
+    """
+
+    _decryptor: object = field()
+    _tag_length: int = field()
+    _content_length: int = field()
+    _amount_read: int = field(init=False, default=0)
+    _tail: bytes = field(init=False, default=b"")
+
+    @property
+    def content_length(self) -> int:  # noqa: D102
+        return self._content_length
+
+    @property
+    def amount_read(self) -> int:  # noqa: D102
+        return self._amount_read
+
+    @property
+    def tag_length(self) -> int:
+        """Length of the GCM authentication tag in bytes."""
+        return self._tag_length
+
+    def update(self, data: bytes) -> bytes:
+        """Decrypt a chunk, holding back the last tag_length bytes.
+
+        A rolling _tail buffer always retains the last tag_length bytes
+        so the GCM tag is never passed to the cipher's update().
+        """
+        self._amount_read += len(data)
+        buf = self._tail + data
+        if len(buf) <= self._tag_length:
+            self._tail = buf
+            return b""
+        self._tail = buf[-self._tag_length :]
+        return self._decryptor.update(buf[: -self._tag_length])
+
+    def finalize(self, data: bytes) -> bytes:
+        """Finalize decryption using the buffered tag."""
+        try:
+            self._amount_read += len(data)
+            buf = self._tail + data
+            if len(buf) < self._tag_length:
+                raise S3EncryptionClientError(
+                    f"Incomplete GCM data: expected at least {self._tag_length} "
+                    f"tag bytes, got {len(buf)} total remaining bytes."
+                )
+            tag = buf[-self._tag_length :]
+            ciphertext = buf[: -self._tag_length]
+            plaintext = self._decryptor.update(ciphertext) if ciphertext else b""
+            return plaintext + self._decryptor.finalize_with_tag(tag)
+        except S3EncryptionClientError:
+            raise
+        except InvalidTag as e:
+            raise S3EncryptionClientSecurityError(f"Failed to decrypt Object: {e}") from e
+        except Exception as e:
+            raise S3EncryptionClientError(f"Failed to decrypt Object: {e}") from e

src/s3_encryption/materials/materials.py

@@ -172,6 +172,26 @@ def kc_gcm_iv(self) -> bytes:
         ##% the IV used in the AES-GCM content encryption/decryption MUST consist entirely of bytes with the value 0x01.
         return b"\x01" * self.cipher_iv_length_bytes
 
+    @property
+    def cipher_block_size_bits(self) -> int:
+        """Block size of the cipher in bits."""
+        return self._cipher_block_size_bits
+
+    @property
+    def cipher_block_size_bytes(self) -> int:
+        """Block size of the cipher in bytes."""
+        return self._cipher_block_size_bits // 8
+
+    @property
+    def cipher_tag_length_bits(self) -> int:
+        """Authentication tag length of the cipher in bits."""
+        return self._cipher_tag_length_bits
+
+    @property
+    def cipher_tag_length_bytes(self) -> int:
+        """Authentication tag length of the cipher in bytes."""
+        return self._cipher_tag_length_bits // 8
+
 
 class CommitmentPolicy(Enum):
     """Commitment policies controlling key-commitment behavior."""