tessl/pypi-aioboto3
Async boto3 wrapper providing asynchronous AWS SDK functionality
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
client-side-encryption.mddocs/
Client-Side Encryption
Advanced S3 client-side encryption capabilities for secure data storage with async support. The S3 CSE (Client-Side Encryption) functionality provides encryption and decryption of data before sending to or receiving from S3, ensuring data security at rest and in transit.
Capabilities
S3 CSE Client
Main client-side encryption client that wraps S3 operations with encryption/decryption capabilities.
class S3CSEClient:
def __init__(
self,
s3_client,
kms_key_id: str = None,
encryption_key = None,
key_wrap_algorithm: str = None
):
"""
Initialize S3 client-side encryption client.
Parameters:
- s3_client: Underlying S3 client
- kms_key_id: KMS key ID for key encryption
- encryption_key: Master encryption key
- key_wrap_algorithm: Algorithm for key wrapping
"""
async def __aenter__(self):
"""
Async context manager entry.
Returns:
self: The S3CSEClient instance
"""
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""
Async context manager exit.
Parameters:
- exc_type: Exception type if an exception occurred
- exc_val: Exception value if an exception occurred
- exc_tb: Exception traceback if an exception occurred
"""
async def put_object(self, **kwargs):
"""
Put an encrypted object to S3.
Parameters:
- Bucket: S3 bucket name
- Key: S3 object key
- Body: Object body (will be encrypted)
- **kwargs: Additional S3 put_object parameters
Returns:
Response from S3 put_object operation
"""
async def get_object(self, **kwargs):
"""
Get and decrypt an object from S3.
Parameters:
- Bucket: S3 bucket name
- Key: S3 object key
- **kwargs: Additional S3 get_object parameters
Returns:
Decrypted response with Body containing decrypted data
"""
async def upload_file(
self,
filename: str,
bucket: str,
key: str,
**kwargs
):
"""
Upload and encrypt a file to S3.
Parameters:
- filename: Local file path
- bucket: S3 bucket name
- key: S3 object key
- **kwargs: Additional upload parameters
"""
async def download_file(
self,
bucket: str,
key: str,
filename: str,
**kwargs
):
"""
Download and decrypt a file from S3.
Parameters:
- bucket: S3 bucket name
- key: S3 object key
- filename: Local file path to save decrypted data
- **kwargs: Additional download parameters
"""S3 CSE Bucket
Bucket resource wrapper with client-side encryption support.
class S3CSEBucket:
def __init__(self, bucket_resource, cse_client):
"""
Initialize S3 CSE bucket wrapper.
Parameters:
- bucket_resource: S3 bucket resource
- cse_client: S3CSEClient instance
"""
async def __aenter__(self):
"""
Async context manager entry.
Returns:
self: The S3CSEBucket instance
"""
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""
Async context manager exit.
"""
async def put_object(self, key: str, body, **kwargs):
"""
Put encrypted object to this bucket.
Parameters:
- key: S3 object key
- body: Object body (will be encrypted)
- **kwargs: Additional parameters
"""
async def get_object(self, key: str, **kwargs):
"""
Get and decrypt object from this bucket.
Parameters:
- key: S3 object key
- **kwargs: Additional parameters
Returns:
Decrypted object data
"""S3 CSE Object
Object resource wrapper with client-side encryption support.
class S3CSEObject:
def __init__(self, object_resource, cse_client):
"""
Initialize S3 CSE object wrapper.
Parameters:
- object_resource: S3 object resource
- cse_client: S3CSEClient instance
"""
async def __aenter__(self):
"""
Async context manager entry.
Returns:
self: The S3CSEObject instance
"""
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""
Async context manager exit.
"""
async def put(self, body, **kwargs):
"""
Put encrypted data to this object.
Parameters:
- body: Object body (will be encrypted)
- **kwargs: Additional parameters
"""
async def get(self, **kwargs):
"""
Get and decrypt this object's data.
Parameters:
- **kwargs: Additional parameters
Returns:
Decrypted object data
"""Encryption Utilities
Supporting classes and functions for encryption operations.
class DummyAIOFile:
def __init__(self, data: bytes):
"""
Async file wrapper for encrypted data.
Parameters:
- data: Encrypted/decrypted byte data
"""
async def read(self, n: int = -1) -> bytes:
"""
Read data from the file.
Parameters:
- n: Number of bytes to read (-1 for all)
Returns:
Byte data
"""
async def readany(self) -> bytes:
"""
Read any available data.
Returns:
Available byte data
"""
async def readexactly(self, n: int) -> bytes:
"""
Read exactly n bytes.
Parameters:
- n: Number of bytes to read
Returns:
Exactly n bytes of data
"""
async def readchunk(self) -> tuple:
"""
Read a chunk of data.
Returns:
Tuple of (data, is_eof)
"""Usage Examples
Basic Encryption Setup
import aioboto3
from aioboto3.s3.cse import S3CSEClient
async def basic_cse_usage():
session = aioboto3.Session()
# Create regular S3 client first
async with session.client('s3', region_name='us-east-1') as s3_client:
# Create CSE client with KMS key
cse_client = S3CSEClient(
s3_client=s3_client,
kms_key_id='arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012'
)
async with cse_client:
# Upload encrypted data
await cse_client.put_object(
Bucket='my-secure-bucket',
Key='sensitive-data.txt',
Body=b'This is sensitive information'
)
# Download and decrypt data
response = await cse_client.get_object(
Bucket='my-secure-bucket',
Key='sensitive-data.txt'
)
decrypted_data = await response['Body'].read()
print(f"Decrypted: {decrypted_data}")File Upload/Download with Encryption
async def file_operations_with_encryption():
session = aioboto3.Session()
async with session.client('s3', region_name='us-east-1') as s3_client:
cse_client = S3CSEClient(
s3_client=s3_client,
kms_key_id='your-kms-key-id'
)
async with cse_client:
# Upload encrypted file
await cse_client.upload_file(
'/path/to/sensitive/document.pdf',
'my-secure-bucket',
'documents/encrypted-document.pdf'
)
# Download and decrypt file
await cse_client.download_file(
'my-secure-bucket',
'documents/encrypted-document.pdf',
'/path/to/decrypted/document.pdf'
)Using Master Encryption Key
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os
async def master_key_encryption():
session = aioboto3.Session()
# Generate or load master key
master_key = AESGCM.generate_key(bit_length=256)
async with session.client('s3', region_name='us-east-1') as s3_client:
cse_client = S3CSEClient(
s3_client=s3_client,
encryption_key=master_key,
key_wrap_algorithm='AES-GCM'
)
async with cse_client:
# Encrypt and upload data
sensitive_data = "Secret business information"
await cse_client.put_object(
Bucket='my-secure-bucket',
Key='business/secret.txt',
Body=sensitive_data.encode()
)
# Decrypt and retrieve data
response = await cse_client.get_object(
Bucket='my-secure-bucket',
Key='business/secret.txt'
)
decrypted_content = await response['Body'].read()
print(f"Retrieved: {decrypted_content.decode()}")Batch Operations with Encryption
async def batch_encrypted_operations():
session = aioboto3.Session()
async with session.client('s3', region_name='us-east-1') as s3_client:
cse_client = S3CSEClient(
s3_client=s3_client,
kms_key_id='your-kms-key-id'
)
async with cse_client:
# Upload multiple encrypted files
files_to_encrypt = [
('file1.txt', b'Content of file 1'),
('file2.txt', b'Content of file 2'),
('file3.txt', b'Content of file 3')
]
for filename, content in files_to_encrypt:
await cse_client.put_object(
Bucket='my-secure-bucket',
Key=f'encrypted/{filename}',
Body=content
)
# Download and decrypt all files
for filename, _ in files_to_encrypt:
response = await cse_client.get_object(
Bucket='my-secure-bucket',
Key=f'encrypted/{filename}'
)
decrypted = await response['Body'].read()
print(f"{filename}: {decrypted}")Error Handling with CSE
import botocore.exceptions
from cryptography.exceptions import InvalidTag
async def cse_error_handling():
session = aioboto3.Session()
try:
async with session.client('s3', region_name='us-east-1') as s3_client:
cse_client = S3CSEClient(
s3_client=s3_client,
kms_key_id='your-kms-key-id'
)
async with cse_client:
response = await cse_client.get_object(
Bucket='my-secure-bucket',
Key='encrypted-file.txt'
)
data = await response['Body'].read()
except InvalidTag:
print("Decryption failed - invalid authentication tag")
except botocore.exceptions.ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'AccessDenied':
print("Access denied - check KMS key permissions")
elif error_code == 'NoSuchKey':
print("Encrypted object not found")
else:
print(f"S3 error: {error_code}")
except Exception as e:
print(f"Encryption/decryption error: {e}")Configuring Encryption Algorithms
async def advanced_encryption_config():
session = aioboto3.Session()
async with session.client('s3', region_name='us-east-1') as s3_client:
# Configure with specific encryption settings
cse_client = S3CSEClient(
s3_client=s3_client,
kms_key_id='your-kms-key-id',
key_wrap_algorithm='AES-GCM' # Specify key wrapping algorithm
)
async with cse_client:
# Upload with metadata about encryption
await cse_client.put_object(
Bucket='my-secure-bucket',
Key='metadata-encrypted.txt',
Body=b'Data with encryption metadata',
Metadata={
'encryption-info': 'client-side-encrypted',
'algorithm': 'AES-GCM'
}
)Install with Tessl CLI
npx tessl i tessl/pypi-aioboto3