pymongocrypt

from pymongocrypt import binding

import os from pymongo import MongoClient from pymongo.encryption import ClientEncryption # A 96-byte local master key (for demonstration purposes). # In production, use a secure Key Management System (KMS). local_key = os.environ.get("MONGO_LOCAL_MASTER_KEY", "A" * 96).encode('utf-8') if len(local_key) != 96: # Generate a temporary key if not set correctly for quick local run local_key = os.urandom(96) print(f"WARNING: MONGO_LOCAL_MASTER_KEY not set or invalid length. Using a generated temporary key.") kms_providers = { "local": {"key": local_key} } # Key Vault setup connection_string = os.environ.get("MONGO_URI", "mongodb://localhost:27017/") client = MongoClient(connection_string) key_vault_db = "encryption" key_vault_coll = "__keyVault" key_vault_namespace = f"{key_vault_db}.{key_vault_coll}" # Ensure key vault collection exists and has the required index client.get_database(key_vault_db).get_collection(key_vault_coll).create_index( [('keyAltNames', 1)], unique=True, partialFilterExpression={'keyAltNames': {'$exists': True}} ) # Initialize ClientEncryption # This object implicitly uses the libmongocrypt bindings provided by pymongocrypt. # Note: kms_tls_options is required even if empty for local KMS. client_encryption = ClientEncryption( kms_providers=kms_providers, key_vault_namespace=key_vault_namespace, kms_tls_options={}, key_vault_client=client ) # Create a data key (if one doesn't exist) for encryption try: key_id = client_encryption.get_key_by_alt_name("my_data_key")["_id"] print(f"Using existing data key ID: {key_id}") except Exception: # Key not found, create one key_id = client_encryption.create_data_key( "local", key_alt_names=["my_data_key"] ) print(f"Created new data key ID: {key_id}") # Encrypt a value value_to_encrypt = "My secret data that should be encrypted" encrypted_value = client_encryption.encrypt( value_to_encrypt, key_id=key_id, algorithm="AEAD_AES256_CBC_HMAC_SHA521_Deterministic" # Or Random ) print(f"Original value: {value_to_encrypt}") print(f"Encrypted value (Binary type): {encrypted_value}") # Decrypt the value decrypted_value = client_encryption.decrypt(encrypted_value) print(f"Decrypted value: {decrypted_value}") # Clean up client.drop_database(key_vault_db) client.close()