tessl/pypi-motor
Non-blocking MongoDB driver for Python asyncio and Tornado applications
—
Pending
client-encryption.mddocs/
Client-Side Field Level Encryption
Client-side field level encryption (CSFLE) support for Motor. Enables encryption and decryption of sensitive document fields before they are sent to or received from MongoDB, providing end-to-end security for sensitive data.
Capabilities
Client Encryption
The main encryption client for managing encryption keys and performing field-level encryption operations.
class AsyncIOMotorClientEncryption:
def __init__(
self,
kms_providers: Dict[str, Any],
key_vault_namespace: str,
key_vault_client: AsyncIOMotorClient,
codec_options: Optional[CodecOptions] = None,
kms_tls_options: Optional[Dict[str, Any]] = None
):
"""
Create a new client encryption instance.
Parameters:
- kms_providers: Configuration for key management systems (AWS KMS, Azure, GCP, local)
- key_vault_namespace: Database and collection name for the key vault (format: "db.collection")
- key_vault_client: Motor client connected to the key vault database
- codec_options: Options for encoding/decoding BSON documents
- kms_tls_options: TLS configuration for KMS providers
"""
async def create_data_key(
self,
kms_provider: str,
master_key: Optional[Dict[str, Any]] = None,
key_alt_names: Optional[List[str]] = None,
key_material: Optional[bytes] = None
) -> Binary:
"""
Create a new data encryption key.
Parameters:
- kms_provider: The KMS provider to use ('aws', 'azure', 'gcp', 'kmip', 'local')
- master_key: Master key configuration specific to the KMS provider
- key_alt_names: Alternative names for the key
- key_material: Custom key material (for local KMS only)
Returns:
Binary: The _id of the created data key
"""
async def encrypt(
self,
value: Any,
algorithm: str,
key_id: Optional[Binary] = None,
key_alt_name: Optional[str] = None,
query_type: Optional[str] = None,
contention_factor: Optional[int] = None,
range_opts: Optional[RangeOpts] = None
) -> Binary:
"""
Encrypt a value using client-side field level encryption.
Parameters:
- value: The value to encrypt
- algorithm: Encryption algorithm ('AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic' or 'AEAD_AES_256_CBC_HMAC_SHA_512-Random')
- key_id: The data key _id to use for encryption
- key_alt_name: Alternative name of the data key
- query_type: Query type for queryable encryption ('equality', 'range')
- contention_factor: Contention factor for queryable encryption
- range_opts: Range options for range queries
Returns:
Binary: The encrypted value
"""
async def decrypt(self, value: Binary) -> Any:
"""
Decrypt an encrypted value.
Parameters:
- value: The encrypted Binary value to decrypt
Returns:
Any: The decrypted value
"""
async def encrypt_expression(
self,
expression: Dict[str, Any],
algorithm: str,
key_id: Optional[Binary] = None,
key_alt_name: Optional[str] = None,
query_type: Optional[str] = None,
contention_factor: Optional[int] = None,
range_opts: Optional[RangeOpts] = None
) -> RawBSONDocument:
"""
Encrypt a MongoDB expression for queryable encryption.
Parameters:
- expression: MongoDB query expression to encrypt
- algorithm: Encryption algorithm
- key_id: The data key _id to use for encryption
- key_alt_name: Alternative name of the data key
- query_type: Query type for queryable encryption
- contention_factor: Contention factor for queryable encryption
- range_opts: Range options for range queries
Returns:
RawBSONDocument: The encrypted expression
"""
async def rewrap_many_data_key(
self,
filter: Dict[str, Any],
provider: Optional[str] = None,
master_key: Optional[Dict[str, Any]] = None
) -> RewrapManyDataKeyResult:
"""
Rewrap multiple data keys with a new master key.
Parameters:
- filter: Query filter to select data keys to rewrap
- provider: New KMS provider (if changing providers)
- master_key: New master key configuration
Returns:
RewrapManyDataKeyResult: Result of the rewrap operation
"""
async def delete_key(self, id: Binary) -> DeleteResult:
"""
Delete a data key from the key vault.
Parameters:
- id: The _id of the data key to delete
Returns:
DeleteResult: Result of the delete operation
"""
async def get_key(self, id: Binary) -> Optional[RawBSONDocument]:
"""
Get a data key from the key vault.
Parameters:
- id: The _id of the data key to retrieve
Returns:
Optional[RawBSONDocument]: The data key document or None if not found
"""
async def add_key_alt_name(self, id: Binary, key_alt_name: str) -> Optional[RawBSONDocument]:
"""
Add an alternative name to a data key.
Parameters:
- id: The _id of the data key
- key_alt_name: The alternative name to add
Returns:
Optional[RawBSONDocument]: The updated data key document
"""
async def get_key_by_alt_name(self, key_alt_name: str) -> Optional[RawBSONDocument]:
"""
Get a data key by its alternative name.
Parameters:
- key_alt_name: The alternative name to search for
Returns:
Optional[RawBSONDocument]: The data key document or None if not found
"""
async def remove_key_alt_name(self, id: Binary, key_alt_name: str) -> Optional[RawBSONDocument]:
"""
Remove an alternative name from a data key.
Parameters:
- id: The _id of the data key
- key_alt_name: The alternative name to remove
Returns:
Optional[RawBSONDocument]: The updated data key document
"""
async def close(self) -> None:
"""Close the client encryption instance and release resources."""Usage Examples
Basic Encryption Setup
import motor.motor_asyncio
from pymongo.encryption import ClientEncryption
async def setup_encryption():
# Configure KMS providers
kms_providers = {
"local": {
"key": b"your-96-byte-local-master-key-here" * 4 # 96 bytes
}
}
# Connect to MongoDB
key_vault_client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')
client_encryption = motor.motor_asyncio.AsyncIOMotorClientEncryption(
kms_providers=kms_providers,
key_vault_namespace="encryption.__keyVault",
key_vault_client=key_vault_client
)
# Create a data key
data_key_id = await client_encryption.create_data_key(
"local",
key_alt_names=["example-key"]
)
# Encrypt a value
encrypted_value = await client_encryption.encrypt(
"sensitive data",
algorithm="AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic",
key_id=data_key_id
)
# Decrypt the value
decrypted_value = await client_encryption.decrypt(encrypted_value)
print(f"Decrypted: {decrypted_value}")
await client_encryption.close()
key_vault_client.close()AWS KMS Integration
async def setup_aws_kms_encryption():
kms_providers = {
"aws": {
"accessKeyId": "your-access-key-id",
"secretAccessKey": "your-secret-access-key",
"region": "us-east-1"
}
}
key_vault_client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')
client_encryption = motor.motor_asyncio.AsyncIOMotorClientEncryption(
kms_providers=kms_providers,
key_vault_namespace="encryption.__keyVault",
key_vault_client=key_vault_client
)
# Create data key with AWS KMS
data_key_id = await client_encryption.create_data_key(
"aws",
master_key={
"region": "us-east-1",
"key": "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012"
},
key_alt_names=["aws-example-key"]
)
return client_encryption, data_key_idQueryable Encryption
async def queryable_encryption_example():
# Setup encryption client (same as above)
client_encryption, data_key_id = await setup_aws_kms_encryption()
# Encrypt for equality queries
encrypted_ssn = await client_encryption.encrypt(
"123-45-6789",
algorithm="Indexed",
key_id=data_key_id,
query_type="equality",
contention_factor=1
)
# Encrypt expression for range queries
range_expression = {"$gte": 1000, "$lte": 9999}
encrypted_expression = await client_encryption.encrypt_expression(
range_expression,
algorithm="Range",
key_id=data_key_id,
query_type="range",
range_opts={"min": 0, "max": 10000, "sparsity": 1}
)
await client_encryption.close()Types
from typing import Dict, List, Optional, Any
from bson import Binary
from bson.raw_bson import RawBSONDocument
from pymongo.results import DeleteResult
class RewrapManyDataKeyResult:
bulk_write_result: Optional[Any]
class RangeOpts:
min: Optional[Any]
max: Optional[Any]
sparsity: int
precision: Optional[int]Install with Tessl CLI
npx tessl i tessl/pypi-motor