Client-Side Field Level Encryption

Starting in MongoDB 4.2, client-side field level encryption allows an application to encrypt specific data fields in addition to pre-existing MongoDB encryption features such as Encryption at Rest and TLS/SSL (Transport Encryption).

With field level encryption, applications can encrypt fields in documents prior to transmitting data over the wire to the server. Client-side field level encryption supports workloads where applications must guarantee that unauthorized parties, including server administrators, cannot read the encrypted data.

See also

The MongoDB documentation on

client-side-field-level-encryption

Dependencies

To get started using client-side field level encryption in your project, you will need to install the pymongocrypt library as well as the driver itself. Install both the driver and a compatible version of pymongocrypt like this:

$ python -m pip install 'motor[encryption]'

Note that installing on Linux requires pip 19 or later for manylinux2010 wheel support. For more information about installing pymongocrypt see the installation instructions on the project’s PyPI page.

mongocryptd

The mongocryptd binary is required for automatic client-side encryption and is included as a component in the MongoDB Enterprise Server package. For more information on this binary, see the PyMongo documentation on mongocryptd.

Automatic Client-Side Field Level Encryption

Automatic client-side field level encryption is enabled by creating a AsyncIOMotorClient with the auto_encryption_opts option set to an instance of AutoEncryptionOpts. The following examples show how to setup automatic client-side field level encryption using AsyncIOMotorClientEncryption to create a new encryption data key.

Note

Automatic client-side field level encryption requires MongoDB 4.2+ enterprise or a MongoDB 4.2+ Atlas cluster. The community version of the server supports automatic decryption as well as Explicit Encryption.

Providing Local Automatic Encryption Rules

The following example shows how to specify automatic encryption rules via the schema_map option. The automatic encryption rules are expressed using a strict subset of the JSON Schema syntax.

Supplying a schema_map provides more security than relying on JSON Schemas obtained from the server. It protects against a malicious server advertising a false JSON Schema, which could trick the client into sending unencrypted data that should be encrypted.

JSON Schemas supplied in the schema_map only apply to configuring automatic client-side field level encryption. Other validation rules in the JSON schema will not be enforced by the driver and will result in an error.

import asyncio
import os

from bson import json_util
from bson.codec_options import CodecOptions
from pymongo.encryption import Algorithm
from pymongo.encryption_options import AutoEncryptionOpts

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorClientEncryption


async def create_json_schema_file(kms_providers, key_vault_namespace, key_vault_client):
    client_encryption = AsyncIOMotorClientEncryption(
        kms_providers,
        key_vault_namespace,
        key_vault_client,
        # The CodecOptions class used for encrypting and decrypting.
        # This should be the same CodecOptions instance you have configured
        # on MotorClient, Database, or Collection. We will not be calling
        # encrypt() or decrypt() in this example so we can use any
        # CodecOptions.
        CodecOptions(),
    )

    # Create a new data key and json schema for the encryptedField.
    # https://dochub.mongodb.org/core/client-side-field-level-encryption-automatic-encryption-rules
    data_key_id = await client_encryption.create_data_key(
        "local", key_alt_names=["pymongo_encryption_example_1"]
    )
    schema = {
        "properties": {
            "encryptedField": {
                "encrypt": {
                    "keyId": [data_key_id],
                    "bsonType": "string",
                    "algorithm": Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
                }
            }
        },
        "bsonType": "object",
    }
    # Use CANONICAL_JSON_OPTIONS so that other drivers and tools will be
    # able to parse the MongoDB extended JSON file.
    json_schema_string = json_util.dumps(schema, json_options=json_util.CANONICAL_JSON_OPTIONS)

    with open("jsonSchema.json", "w") as file:
        file.write(json_schema_string)


async def main():
    # The MongoDB namespace (db.collection) used to store the
    # encrypted documents in this example.
    encrypted_namespace = "test.coll"

    # This must be the same master key that was used to create
    # the encryption key.
    local_master_key = os.urandom(96)
    kms_providers = {"local": {"key": local_master_key}}

    # The MongoDB namespace (db.collection) used to store
    # the encryption data keys.
    key_vault_namespace = "encryption.__pymongoTestKeyVault"
    key_vault_db_name, key_vault_coll_name = key_vault_namespace.split(".", 1)

    # The MotorClient used to access the key vault (key_vault_namespace).
    key_vault_client = AsyncIOMotorClient()
    key_vault = key_vault_client[key_vault_db_name][key_vault_coll_name]
    # Ensure that two data keys cannot share the same keyAltName.
    await key_vault.drop()
    await key_vault.create_index(
        "keyAltNames", unique=True, partialFilterExpression={"keyAltNames": {"$exists": True}}
    )

    await create_json_schema_file(kms_providers, key_vault_namespace, key_vault_client)

    # Load the JSON Schema and construct the local schema_map option.
    with open("jsonSchema.json") as file:
        json_schema_string = file.read()
    json_schema = json_util.loads(json_schema_string)
    schema_map = {encrypted_namespace: json_schema}

    auto_encryption_opts = AutoEncryptionOpts(
        kms_providers, key_vault_namespace, schema_map=schema_map
    )

    client = AsyncIOMotorClient(auto_encryption_opts=auto_encryption_opts)
    db_name, coll_name = encrypted_namespace.split(".", 1)
    coll = client[db_name][coll_name]
    # Clear old data
    await coll.drop()

    await coll.insert_one({"encryptedField": "123456789"})
    decrypted_doc = await coll.find_one()
    print(f"Decrypted document: {decrypted_doc}")
    unencrypted_coll = AsyncIOMotorClient()[db_name][coll_name]
    encrypted_doc = await unencrypted_coll.find_one()
    print(f"Encrypted document: {encrypted_doc}")


if __name__ == "__main__":
    asyncio.run(main())

Server-Side Field Level Encryption Enforcement

The MongoDB 4.2+ server supports using schema validation to enforce encryption of specific fields in a collection. This schema validation will prevent an application from inserting unencrypted values for any fields marked with the "encrypt" JSON schema keyword.

The following example shows how to setup automatic client-side field level encryption using AsyncIOMotorClientEncryption to create a new encryption data key and create a collection with the Automatic Encryption JSON Schema Syntax.

import asyncio
import os

from bson.binary import STANDARD
from bson.codec_options import CodecOptions
from pymongo.encryption import Algorithm
from pymongo.encryption_options import AutoEncryptionOpts
from pymongo.errors import OperationFailure
from pymongo.write_concern import WriteConcern

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorClientEncryption


async def main():
    # The MongoDB namespace (db.collection) used to store the
    # encrypted documents in this example.
    encrypted_namespace = "test.coll"

    # This must be the same master key that was used to create
    # the encryption key.
    local_master_key = os.urandom(96)
    kms_providers = {"local": {"key": local_master_key}}

    # The MongoDB namespace (db.collection) used to store
    # the encryption data keys.
    key_vault_namespace = "encryption.__pymongoTestKeyVault"
    key_vault_db_name, key_vault_coll_name = key_vault_namespace.split(".", 1)

    # The MotorClient used to access the key vault (key_vault_namespace).
    key_vault_client = AsyncIOMotorClient()
    key_vault = key_vault_client[key_vault_db_name][key_vault_coll_name]
    # Ensure that two data keys cannot share the same keyAltName.
    await key_vault.drop()
    await key_vault.create_index(
        "keyAltNames", unique=True, partialFilterExpression={"keyAltNames": {"$exists": True}}
    )

    client_encryption = AsyncIOMotorClientEncryption(
        kms_providers,
        key_vault_namespace,
        key_vault_client,
        # The CodecOptions class used for encrypting and decrypting.
        # This should be the same CodecOptions instance you have configured
        # on MotorClient, Database, or Collection. We will not be calling
        # encrypt() or decrypt() in this example so we can use any
        # CodecOptions.
        CodecOptions(),
    )

    # Create a new data key and json schema for the encryptedField.
    data_key_id = await client_encryption.create_data_key(
        "local", key_alt_names=["pymongo_encryption_example_2"]
    )
    json_schema = {
        "properties": {
            "encryptedField": {
                "encrypt": {
                    "keyId": [data_key_id],
                    "bsonType": "string",
                    "algorithm": Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
                }
            }
        },
        "bsonType": "object",
    }

    auto_encryption_opts = AutoEncryptionOpts(kms_providers, key_vault_namespace)
    client = AsyncIOMotorClient(auto_encryption_opts=auto_encryption_opts)
    db_name, coll_name = encrypted_namespace.split(".", 1)
    db = client[db_name]
    # Clear old data
    await db.drop_collection(coll_name)
    # Create the collection with the encryption JSON Schema.
    await db.create_collection(
        coll_name,
        # uuid_representation=STANDARD is required to ensure that any
        # UUIDs in the $jsonSchema document are encoded to BSON Binary
        # with the standard UUID subtype 4. This is only needed when
        # running the "create" collection command with an encryption
        # JSON Schema.
        codec_options=CodecOptions(uuid_representation=STANDARD),
        write_concern=WriteConcern(w="majority"),
        validator={"$jsonSchema": json_schema},
    )
    coll = client[db_name][coll_name]

    await coll.insert_one({"encryptedField": "123456789"})
    decrypted_doc = await coll.find_one()
    print(f"Decrypted document: {decrypted_doc}")
    unencrypted_coll = AsyncIOMotorClient()[db_name][coll_name]
    encrypted_doc = await unencrypted_coll.find_one()
    print(f"Encrypted document: {encrypted_doc}")
    try:
        await unencrypted_coll.insert_one({"encryptedField": "123456789"})
    except OperationFailure as exc:
        print(f"Unencrypted insert failed: {exc.details}")


if __name__ == "__main__":
    asyncio.run(main())

Explicit Encryption

Explicit encryption is a MongoDB community feature and does not use the mongocryptd process. Explicit encryption is provided by the AsyncIOMotorClientEncryption class, for example:

import asyncio
import os

from pymongo.encryption import Algorithm

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorClientEncryption


async def main():
    # This must be the same master key that was used to create
    # the encryption key.
    local_master_key = os.urandom(96)
    kms_providers = {"local": {"key": local_master_key}}

    # The MongoDB namespace (db.collection) used to store
    # the encryption data keys.
    key_vault_namespace = "encryption.__pymongoTestKeyVault"
    key_vault_db_name, key_vault_coll_name = key_vault_namespace.split(".", 1)

    # The MotorClient used to read/write application data.
    client = AsyncIOMotorClient()
    coll = client.test.coll
    # Clear old data
    await coll.drop()

    # Set up the key vault (key_vault_namespace) for this example.
    key_vault = client[key_vault_db_name][key_vault_coll_name]
    # Ensure that two data keys cannot share the same keyAltName.
    await key_vault.drop()
    await key_vault.create_index(
        "keyAltNames", unique=True, partialFilterExpression={"keyAltNames": {"$exists": True}}
    )

    client_encryption = AsyncIOMotorClientEncryption(
        kms_providers,
        key_vault_namespace,
        # The Motorlient to use for reading/writing to the key vault.
        # This can be the same MotorClient used by the main application.
        client,
        # The CodecOptions class used for encrypting and decrypting.
        # This should be the same CodecOptions instance you have configured
        # on MotorClient, Database, or Collection.
        coll.codec_options,
    )

    # Create a new data key for the encryptedField.
    data_key_id = await client_encryption.create_data_key(
        "local", key_alt_names=["pymongo_encryption_example_3"]
    )

    # Explicitly encrypt a field:
    encrypted_field = await client_encryption.encrypt(
        "123456789", Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic, key_id=data_key_id
    )
    await coll.insert_one({"encryptedField": encrypted_field})
    doc = await coll.find_one()
    print(f"Encrypted document: {doc}")

    # Explicitly decrypt the field:
    doc["encryptedField"] = await client_encryption.decrypt(doc["encryptedField"])
    print(f"Decrypted document: {doc}")

    # Cleanup resources.
    await client_encryption.close()


if __name__ == "__main__":
    asyncio.run(main())

Explicit Encryption with Automatic Decryption

Although automatic encryption requires MongoDB 4.2 enterprise or a MongoDB 4.2 Atlas cluster, automatic decryption is supported for all users. To configure automatic decryption without automatic encryption set bypass_auto_encryption=True in AutoEncryptionOpts:

import asyncio
import os

from pymongo.encryption import Algorithm
from pymongo.encryption_options import AutoEncryptionOpts

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorClientEncryption


async def main():
    # This must be the same master key that was used to create
    # the encryption key.
    local_master_key = os.urandom(96)
    kms_providers = {"local": {"key": local_master_key}}

    # The MongoDB namespace (db.collection) used to store
    # the encryption data keys.
    key_vault_namespace = "encryption.__pymongoTestKeyVault"
    key_vault_db_name, key_vault_coll_name = key_vault_namespace.split(".", 1)

    # bypass_auto_encryption=True disable automatic encryption but keeps
    # the automatic _decryption_ behavior. bypass_auto_encryption will
    # also disable spawning mongocryptd.
    auto_encryption_opts = AutoEncryptionOpts(
        kms_providers, key_vault_namespace, bypass_auto_encryption=True
    )

    client = AsyncIOMotorClient(auto_encryption_opts=auto_encryption_opts)
    coll = client.test.coll
    # Clear old data
    await coll.drop()

    # Set up the key vault (key_vault_namespace) for this example.
    key_vault = client[key_vault_db_name][key_vault_coll_name]
    # Ensure that two data keys cannot share the same keyAltName.
    await key_vault.drop()
    await key_vault.create_index(
        "keyAltNames", unique=True, partialFilterExpression={"keyAltNames": {"$exists": True}}
    )

    client_encryption = AsyncIOMotorClientEncryption(
        kms_providers,
        key_vault_namespace,
        # The MotorClient to use for reading/writing to the key vault.
        # This can be the same MotorClient used by the main application.
        client,
        # The CodecOptions class used for encrypting and decrypting.
        # This should be the same CodecOptions instance you have configured
        # on MotorClient, Database, or Collection.
        coll.codec_options,
    )

    # Create a new data key for the encryptedField.
    _ = await client_encryption.create_data_key(
        "local", key_alt_names=["pymongo_encryption_example_4"]
    )

    # Explicitly encrypt a field:
    encrypted_field = await client_encryption.encrypt(
        "123456789",
        Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
        key_alt_name="pymongo_encryption_example_4",
    )
    await coll.insert_one({"encryptedField": encrypted_field})
    # Automatically decrypts any encrypted fields.
    doc = await coll.find_one()
    print(f"Decrypted document: {doc}")
    unencrypted_coll = AsyncIOMotorClient().test.coll
    print(f"Encrypted document: {await unencrypted_coll.find_one()}")

    # Cleanup resources.
    await client_encryption.close()


if __name__ == "__main__":
    asyncio.run(main())