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.

Keystore

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://10.143.180.101:8091/default",password='password')
bucket.register_crypto_provider('AES-256-HMAC-SHA256', provider)

Encryption

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,
                                           fieldspec,
                                           prefix)
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.