A newer version of this documentation is available.

View Latest

Field Level Encryption from the Python SDK

    Field Level Encryption is available in Couchbase Data Platform 5.5, from Python SDK version 2.4.0.

    Having examined the concepts behind Field Level Encryption, let’s follow through how we encrypt some data with the Python SDK.


    The following example starts with importing the InMemoryKeyStore module, and creating the insecure keystore (only use this one in testing!). Then registering the provider:

    from cbencryption import AES256CryptoProvider
    from couchbase.bucket import Bucket
    from couchbase.crypto import InMemoryKeyStore
    # create insecure key store and register both public and private keys
    keystore = InMemoryKeyStore()
    keystore.set_key('mypublickey', b'!mysecretkey#9^5usdk39d&dlf)03sL')
    keystore.set_key('myprivatekey', b'myauthpassword')
    # create and register provider
    provider = AES256CryptoProvider.AES256CryptoProvider(keystore, 'mypublickey', 'myprivatekey')
    bucket = Bucket("couchbase://",password='password')
    bucket.register_crypto_provider('AES-256-HMAC-SHA256', provider)


    To encrypt a document, the alg name must match the provider name, and the kid (key ID) must match a key in the keystore:

    prefix = '__crypt_'
    document = {'message': 'The old grey goose jumped over the wrickety gate.'}
    fieldspec = [{'alg': 'AES-256-HMAC-SHA256', 'name': 'message'}]
    encrypted_document = bucket.encrypt_fields(document,
    expected = {
        "__crypt_message": {"alg": "AES-256-HMAC-SHA256",
                            "kid": "mypublickey",
                            "ciphertext": "sR6AFEIGWS5Fy9QObNOhbCgfg3vXH4NHVRK1qkhKLQqjkByg2n69lot89qFEJuBsVNTXR77PZR6RjN4h4M9evg=="

    Decryption & Checking

    We can filter the signature/iv-independent fields, for comparison, to check that the original and the decrypted values are the same:

    def filter_encrypted(encrypted_dict):
        return {k:v for k,v in encrypted_dict.items() if k in {"alg","kid","ciphertext"}}
    subset_expected = filter_encrypted(expected)
    subset_actual = filter_encrypted(encrypted_document)
    assert subset_expected == subset_actual
    # decrypt document using registered provider
    decrypted_document = bucket.decrypt_fields(encrypted_document, fieldspec, prefix)
    assert decrypted_document==document

    The complete code sample is available in our devguide examples.