feat(examples): add usage examples with integration tests

.github/workflows/python-integ.yml

@@ -56,6 +56,9 @@ jobs:
           CI_S3_BUCKET: ${{ vars.CI_S3_BUCKET }}
           CI_KMS_KEY_ALIAS: ${{ vars.CI_KMS_KEY_ALIAS }}
 
+      - name: Run examples
+        run: make test-examples
+
       - name: Generate coverage HTML report
         if: always()
         run: uv run coverage html -d coverage-report

Makefile

@@ -21,7 +21,7 @@ format:
 	uv run ruff check --fix src/ test/
 
 # Run all tests with combined coverage
-test: test-unit test-integration
+test: test-unit test-integration test-examples
 
 # Run unit tests (creates .coverage report)
 test-unit:
@@ -31,6 +31,9 @@ test-unit:
 test-integration:
 	uv run pytest test/integration/ --verbose --cov=src/s3_encryption --cov-append --cov-report=term-missing
 
+test-examples:
+	uv run pytest examples/test/ -v
+
 # Clean up cache files
 clean:
 	find . -type d -name __pycache__ -exec rm -rf {} +

examples/__init__.py

@@ -0,0 +1,2 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0

examples/src/__init__.py

@@ -0,0 +1,2 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0

examples/src/delayed_auth_streaming_example.py

@@ -0,0 +1,89 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""
+This example demonstrates streaming decryption with delayed authentication
+using the S3 Encryption Client.
+
+By default, the S3 Encryption Client buffers the entire ciphertext and verifies
+the authentication tag before releasing any plaintext. This is the safest mode,
+but requires holding the entire object in memory.
+
+With delayed authentication enabled, plaintext is released incrementally as it
+is decrypted, before the authentication tag has been verified. This allows
+processing large files without buffering the entire object in memory.
+
+Your application should still read the stream to completion. In the event that
+an error is thrown in the final read due to an invalid authentication tag,
+your application must be able to invalidate the associated data.
+
+WARNING: With delayed authentication, plaintext is released before it has been
+authenticated. An attacker could modify the ciphertext and the client would
+release tampered plaintext before detecting the modification. Only use this
+mode when you need to process files too large to buffer in memory and you
+understand the security implications.
+
+This example:
+1. Creates a KMS Keyring
+2. Configures the S3 Encryption Client with delayed authentication enabled
+3. Encrypts and uploads a large object to S3
+4. Streams the decrypted object back, reading it in chunks
+5. Verifies the decrypted content matches the original
+"""
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig
+from s3_encryption.exceptions import S3EncryptionClientSecurityError
+from s3_encryption.materials.kms_keyring import KmsKeyring
+
+# 10 MB of example data
+EXAMPLE_DATA: bytes = b"A" * (10 * 1024 * 1024)
+CHUNK_SIZE = 1024 * 1024  # 1 MB
+
+
+def delayed_auth_streaming_decrypt(
+    s3_client, kms_client, kms_key_id: str, bucket: str, key: str
+):
+    """Demonstrate streaming decryption with delayed authentication.
+
+    Args:
+        s3_client: boto3 S3 client.
+        kms_client: boto3 KMS client.
+        kms_key_id: KMS key ARN or alias to use for encryption/decryption.
+        bucket: S3 bucket name.
+        key: S3 object key.
+    """
+    # 1. Create a KMS Keyring.
+    keyring = KmsKeyring(kms_client=kms_client, kms_key_id=kms_key_id)
+
+    # 2. Configure the S3 Encryption Client with delayed authentication.
+    config = S3EncryptionClientConfig(
+        keyring=keyring,
+        enable_delayed_authentication=True,
+    )
+    s3ec = S3EncryptionClient(wrapped_s3_client=s3_client, config=config)
+
+    # 3. Encrypt and upload the object.
+    s3ec.put_object(Bucket=bucket, Key=key, Body=EXAMPLE_DATA)
+
+    # 4. Stream the decrypted object back in chunks.
+    # With delayed authentication, plaintext is released incrementally
+    # without buffering the entire object in memory.
+    response = s3ec.get_object(Bucket=bucket, Key=key)
+    body = response["Body"]
+
+    chunks = []
+    try:
+        while True:
+            chunk = body.read(CHUNK_SIZE)
+            if not chunk:
+                break
+            chunks.append(chunk)
+
+        plaintext = b"".join(chunks)
+
+    except S3EncryptionClientSecurityError:
+        # Authentication tag verification failed.
+        # Discard any plaintext released before the error.
+        raise
+
+    # 5. Verify the decrypted content matches the original.
+    assert plaintext == EXAMPLE_DATA, "Decrypted plaintext does not match original data"

examples/src/instruction_file_example.py

@@ -0,0 +1,59 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""
+This example demonstrates decrypting S3 objects that store their encryption
+metadata in instruction files rather than S3 object metadata.
+
+An instruction file is a companion S3 object that contains the encryption
+metadata (encrypted data key, IV, algorithm, etc.) as JSON. By default,
+the instruction file has the same key as the encrypted object with a
+".instruction" suffix appended.
+
+You can also use a custom instruction file suffix. This requires configuring
+the S3 Encryption Client with the matching suffix.
+
+NOTE: At this time, the S3 Encryption Client in Python ONLY supports decrypting
+(reading) with instruction files; encrypting with instruction files is not supported
+at this time.
+
+This example:
+1. Decrypts an object using the default instruction file suffix (".instruction")
+2. Decrypts the same object using a custom instruction file suffix
+"""
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig
+from s3_encryption.materials.kms_keyring import KmsKeyring
+
+
+def instruction_file_get(
+    s3_client, kms_client, kms_key_id: str, bucket: str, key: str, expected_plaintext: bytes
+):
+    """Demonstrate decrypting objects with default and custom instruction file suffixes.
+
+    Args:
+        s3_client: boto3 S3 client.
+        kms_client: boto3 KMS client.
+        kms_key_id: KMS key ARN or alias used to encrypt the object.
+        bucket: S3 bucket containing the encrypted object and instruction files.
+        key: S3 object key of the encrypted object.
+        expected_plaintext: Expected plaintext content for verification.
+    """
+    keyring = KmsKeyring(kms_client=kms_client, kms_key_id=kms_key_id)
+
+    # 1. Decrypt using the default instruction file suffix (".instruction").
+    # The client will fetch "<key>.instruction" for the encryption metadata.
+    config = S3EncryptionClientConfig(keyring=keyring)
+    s3ec = S3EncryptionClient(wrapped_s3_client=s3_client, config=config)
+
+    response = s3ec.get_object(Bucket=bucket, Key=key)
+    plaintext = response["Body"].read()
+    assert plaintext == expected_plaintext, "Default suffix: decrypted plaintext does not match"
+
+    # 2. Decrypt while specifying the Instruction File Suffix
+    # InstructionFileSuffix is a per-request keyword argument on get_object,
+    # so the same client can use different suffixes per request.
+    response = s3ec.get_object(
+        Bucket=bucket, Key=key, InstructionFileSuffix=".custom-suffix-instruction"
+    )
+    plaintext = response["Body"].read()
+    assert plaintext == expected_plaintext, "Custom suffix: decrypted plaintext does not match"

examples/src/kms_keyring_put_get_example.py

@@ -0,0 +1,95 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""
+This example demonstrates a basic put/get roundtrip using the S3 Encryption Client
+with a KMS Keyring.
+
+The KMS Keyring uses a symmetric KMS key to generate and decrypt data keys.
+The S3 Encryption Client encrypts the object before uploading to S3 and decrypts
+it on download, so the data is protected at rest.
+
+This example:
+1. Creates a KMS Keyring with the provided KMS key ID
+2. Wraps a boto3 S3 client with the S3 Encryption Client
+3. Creates an encryption context bound to the S3 bucket and key
+4. Puts an encrypted object to S3
+5. Gets and decrypts the object from S3
+6. Verifies the decrypted plaintext matches the original
+
+Here is an example KMS Key Policy statement that would validate the
+Encryption Context used in this example::
+
+    Sid: RestrictToEncryptionContextBucket
+    Effect: Allow
+    Principal:
+      AWS: "arn:aws:iam::<account-id>:role/<role-name>"
+    Action:
+      - kms:GenerateDataKey
+      - kms:Decrypt
+    Resource: "*"
+    Condition:
+      StringEquals:
+        "kms:EncryptionContext:aws-s3-bucket": "<bucket>"
+"""
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig, S3EncryptionClientError
+from s3_encryption.materials.kms_keyring import KmsKeyring
+
+EXAMPLE_DATA: bytes = b"Hello, S3 Encryption Client!"
+
+
+def kms_keyring_put_get(s3_client, kms_client, kms_key_id: str, bucket: str, key: str):
+    """Demonstrate an encrypt/decrypt cycle using a KMS Keyring with S3.
+
+    Args:
+        s3_client: boto3 S3 client.
+        kms_client: boto3 KMS client.
+        kms_key_id: KMS key ARN or alias to use for encryption/decryption.
+        bucket: S3 bucket name.
+        key: S3 object key.
+    """
+    # 1. Create a KMS Keyring.
+    keyring = KmsKeyring(kms_client=kms_client, kms_key_id=kms_key_id)
+
+    # 2. Wrap the S3 client with the S3 Encryption Client.
+    # The default commitment policy is REQUIRE_ENCRYPT_REQUIRE_DECRYPT,
+    # which enforces key-committing algorithm suites on both encrypt and decrypt.
+    config = S3EncryptionClientConfig(keyring=keyring)
+    s3ec = S3EncryptionClient(wrapped_s3_client=s3_client, config=config)
+
+    # 3. Create an encryption context.
+    # The encryption context is a set of key-value pairs that are bound to the ciphertext.
+    # Including the bucket and key ensures the ciphertext is tied to this specific S3 object.
+    # This will also be visible to KMS when evaluating key policies.
+    # See the example KMS Key Policy in this module's docstring.
+    # The encryption context is optional, but strongly recommended.
+    encryption_context = {
+        "aws-s3-bucket": bucket,
+        "aws-s3-key": key,
+    }
+
+    # 4. Put an encrypted object.
+    s3ec.put_object(Bucket=bucket, Key=key, Body=EXAMPLE_DATA, EncryptionContext=encryption_context)
+
+    # 5. Get and decrypt the object.
+    # If you specified an encryption context during encryption,
+    # you must provide the same encryption context during decryption.
+    response = s3ec.get_object(Bucket=bucket, Key=key, EncryptionContext=encryption_context)
+    plaintext = response["Body"].read()
+
+    # 6. Optional Verify the decrypted plaintext matches the original.
+    assert plaintext == EXAMPLE_DATA, "Decrypted plaintext does not match original data"
+
+    # However, if the Encryption Context is not present at decryption time, then decryption will fail
+    failed = False
+    try:
+        _ = s3ec.get_object(
+            Bucket=bucket, Key=key,
+            # Incomplete Encryption Context
+            EncryptionContext={"aws-s3-bucket": bucket})
+    except S3EncryptionClientError as e:
+        failed = True
+        assert hasattr(e, "kwargs")
+        assert e.kwargs.get("msg") is not None
+        assert e.kwargs.get("msg") == "Provided encryption context does not match information retrieved from S3"
+    assert failed

examples/src/legacy_decrypt_example.py

@@ -0,0 +1,60 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""
+This example demonstrates how to decrypt legacy S3 objects that were encrypted
+using older versions of the S3 Encryption Client (V1 or V2).
+
+Legacy objects use the KmsV1 wrapping algorithm and may use unauthenticated
+content encryption (AES-CBC). To decrypt these objects, you must:
+1. Enable legacy wrapping algorithms on the KMS Keyring
+2. Enable legacy unauthenticated modes on the S3 Encryption Client config
+3. Use a commitment policy that allows non-key-committing algorithm suites
+
+This example:
+1. Creates a KMS Keyring with legacy wrapping algorithms enabled
+2. Configures the S3 Encryption Client with legacy decryption support
+3. Decrypts a legacy V1 object from S3
+4. Verifies the decrypted plaintext matches the expected content
+"""
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig
+from s3_encryption.materials.kms_keyring import KmsKeyring
+from s3_encryption.materials.materials import CommitmentPolicy
+
+
+def decrypt_legacy_object(s3_client, kms_client, kms_key_id: str, bucket: str, key: str):
+    """Decrypt a legacy S3 object encrypted by an older S3 Encryption Client.
+
+    Args:
+        s3_client: boto3 S3 client.
+        kms_client: boto3 KMS client.
+        kms_key_id: KMS key ARN or alias used to encrypt the object.
+        bucket: S3 bucket name.
+        key: S3 object key.
+
+    Returns:
+        Decrypted plaintext bytes.
+    """
+    # 1. Create a KMS Keyring with legacy wrapping algorithms enabled.
+    # This allows the keyring to decrypt data keys wrapped using the KmsV1 mode,
+    # which older S3 Encryption Clients used.
+    keyring = KmsKeyring(
+        kms_client=kms_client,
+        kms_key_id=kms_key_id,
+        enable_legacy_wrapping_algorithms=True,
+    )
+
+    # 2. Configure the S3 Encryption Client for legacy decryption.
+    # - enable_legacy_unauthenticated_modes: allows decryption of AES-CBC content
+    # - REQUIRE_ENCRYPT_ALLOW_DECRYPT: new objects are encrypted with key-committing
+    #   algorithm suites, while still allowing decryption of legacy objects
+    config = S3EncryptionClientConfig(
+        keyring=keyring,
+        enable_legacy_unauthenticated_modes=True,
+        commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_ALLOW_DECRYPT,
+    )
+    s3ec = S3EncryptionClient(wrapped_s3_client=s3_client, config=config)
+
+    # 3. Decrypt the legacy object.
+    response = s3ec.get_object(Bucket=bucket, Key=key)
+    return response["Body"].read()

examples/test/__init__.py

@@ -0,0 +1,2 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0

examples/test/test_i_delayed_auth_streaming_example.py

@@ -0,0 +1,27 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Test suite for the delayed auth streaming decrypt example."""
+import boto3
+import pytest
+
+from ..src.delayed_auth_streaming_example import delayed_auth_streaming_decrypt
+
+pytestmark = [pytest.mark.examples]
+
+BUCKET = "s3ec-python-github-test-bucket"
+KEY = "examples/delayed-auth-streaming"
+KMS_KEY_ID = "arn:aws:kms:us-west-2:370957321024:alias/S3EC-Python-Github-KMS-Key"
+
+
+def test_delayed_auth_streaming_decrypt():
+    s3_client = boto3.client("s3", region_name="us-west-2")
+    kms_client = boto3.client("kms", region_name="us-west-2")
+    delayed_auth_streaming_decrypt(
+        s3_client=s3_client,
+        kms_client=kms_client,
+        kms_key_id=KMS_KEY_ID,
+        bucket=BUCKET,
+        key=KEY,
+    )
+    # Clean up
+    s3_client.delete_object(Bucket=BUCKET, Key=KEY)