使用中的加密#
客户端字段级加密#
MongoDB 4.2中的新功能是,客户端字段级加密允许应用程序在原有的MongoDB加密功能(如 Encryption at Rest 和 TLS/SSL (Transport Encryption) .
通过字段级加密,应用程序可以加密文档中的字段 先前的 通过网络将数据传输到服务器。客户端字段级加密支持应用程序必须保证未授权方(包括服务器管理员)无法读取加密数据的工作负载。
参见
上的MongoDB文档 Client Side Field Level Encryption 。
依赖关系#
要开始在项目中使用客户端字段级加密,您需要安装 pymongocrypt 和 pymongo-auth-aws 库以及驱动程序本身。安装驱动程序和依赖项的兼容版本,如下所示::
$ python -m pip install 'pymongo[encryption]'
注意,在Linux上安装需要pip19或更高版本才能支持manylinux2010控制盘。有关安装pymongocrypt的更多信息,请参见 the installation instructions on the project's PyPI page .
此外,无论是 crypt_shared 或 mongocryptd 才能使用 automatic 客户端加密。
孟加拉语#
这个 mongocryptd
二进制文件是自动客户端加密所必需的,并作为组件包含在 MongoDB Enterprise Server package . 有关详细的安装说明,请参见 the MongoDB documentation on mongocryptd .
mongocryptd
执行以下操作:
分析指定给数据库连接的自动加密规则。如果JSON架构包含无效的自动加密语法或任何文档验证语法,
mongocryptd
返回错误。使用指定的自动加密规则标记读写操作中的字段以进行加密。
拒绝对加密字段应用时可能返回意外或错误结果的读/写操作。有关支持和不支持的操作,请参阅 Read/Write Support with Automatic Field Level Encryption .
配置了自动加密的MongoClient将自动生成 mongocryptd
从应用程序的 PATH
. 作为自动加密选项的一部分,应用程序可以控制生成行为。例如,将路径设置为 mongocryptd
流程:
auto_encryption_opts = AutoEncryptionOpts(
...,
mongocryptd_spawn_path='/path/to/mongocryptd')
控制的日志输出 mongocryptd
传递选项使用 mongocryptd_spawn_args
::
auto_encryption_opts = AutoEncryptionOpts(
...,
mongocryptd_spawn_args=['--logpath=/path/to/mongocryptd.log', '--logappend'])
如果您的应用程序希望管理 mongocryptd
手动处理,可以禁用生成 mongocryptd
::
auto_encryption_opts = AutoEncryptionOpts(
...,
mongocryptd_bypass_spawn=True,
# URI of the local ``mongocryptd`` process.
mongocryptd_uri='mongodb://localhost:27020')
mongocryptd
只负责支持自动客户端字段级加密,本身不执行任何加密或解密。
自动客户端字段级加密#
通过创建一个 MongoClient
与 auto_encryption_opts
选项设置为的实例 AutoEncryptionOpts
. 下面的示例演示如何使用 ClientEncryption
创建新的加密数据密钥。
备注
自动客户端字段级加密需要mongodb4.2enterprise或mongodb4.2atlas集群。服务器的社区版本支持自动解密以及 显式加密 .
提供本地自动加密规则#
下面的示例演示如何通过 schema_map
选项。自动加密规则使用 strict subset of the JSON Schema syntax .
提供 schema_map
提供了比依赖从服务器获取的JSON模式更安全的功能。它可以防止恶意服务器发布虚假的JSON模式,这可能诱使客户端发送应加密的未加密数据。
中提供的JSON架构 schema_map
仅适用于配置自动客户端字段级加密。JSON模式中的其他验证规则将不会由驱动程序强制执行,并将导致错误。
import os
from bson.codec_options import CodecOptions
from bson import json_util
from pymongo import MongoClient
from pymongo.encryption import Algorithm, ClientEncryption
from pymongo.encryption_options import AutoEncryptionOpts
def create_json_schema_file(kms_providers, key_vault_namespace, key_vault_client):
client_encryption = ClientEncryption(
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 MongoClient, 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 = 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)
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 MongoClient used to access the key vault (key_vault_namespace).
key_vault_client = MongoClient()
key_vault = key_vault_client[key_vault_db_name][key_vault_coll_name]
# Ensure that two data keys cannot share the same keyAltName.
key_vault.drop()
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
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", "r") 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 = MongoClient(auto_encryption_opts=auto_encryption_opts)
db_name, coll_name = encrypted_namespace.split(".", 1)
coll = client[db_name][coll_name]
# Clear old data
coll.drop()
coll.insert_one({"encryptedField": "123456789"})
print("Decrypted document: %s" % (coll.find_one(),))
unencrypted_coll = MongoClient()[db_name][coll_name]
print("Encrypted document: %s" % (unencrypted_coll.find_one(),))
if __name__ == "__main__":
main()
服务器端字段级加密强制#
mongodb4.2服务器支持使用模式验证来强制对集合中的特定字段进行加密。此架构验证将阻止应用程序为标记为的任何字段插入未加密的值 "encrypt"
JSON架构关键字。
以下示例显示如何使用以下命令设置自动客户端字段级加密 ClientEncryption
创建新的加密数据密钥并使用 Automatic Encryption JSON Schema Syntax :
import os
from bson.codec_options import CodecOptions
from bson.binary import STANDARD
from pymongo import MongoClient
from pymongo.encryption import Algorithm, ClientEncryption
from pymongo.encryption_options import AutoEncryptionOpts
from pymongo.errors import OperationFailure
from pymongo.write_concern import WriteConcern
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 MongoClient used to access the key vault (key_vault_namespace).
key_vault_client = MongoClient()
key_vault = key_vault_client[key_vault_db_name][key_vault_coll_name]
# Ensure that two data keys cannot share the same keyAltName.
key_vault.drop()
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
client_encryption = ClientEncryption(
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 MongoClient, 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 = 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 = MongoClient(auto_encryption_opts=auto_encryption_opts)
db_name, coll_name = encrypted_namespace.split(".", 1)
db = client[db_name]
# Clear old data
db.drop_collection(coll_name)
# Create the collection with the encryption JSON Schema.
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]
coll.insert_one({"encryptedField": "123456789"})
print("Decrypted document: %s" % (coll.find_one(),))
unencrypted_coll = MongoClient()[db_name][coll_name]
print("Encrypted document: %s" % (unencrypted_coll.find_one(),))
try:
unencrypted_coll.insert_one({"encryptedField": "123456789"})
except OperationFailure as exc:
print("Unencrypted insert failed: %s" % (exc.details,))
if __name__ == "__main__":
main()
显式加密#
Explicit encryption is a MongoDB community feature and does not use the
mongocryptd
process. Explicit encryption is provided by the
ClientEncryption
class, for example:
import os
from pymongo import MongoClient
from pymongo.encryption import Algorithm, ClientEncryption
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 MongoClient used to read/write application data.
client = MongoClient()
coll = client.test.coll
# Clear old data
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.
key_vault.drop()
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
client_encryption = ClientEncryption(
kms_providers,
key_vault_namespace,
# The MongoClient to use for reading/writing to the key vault.
# This can be the same MongoClient 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 MongoClient, Database, or Collection.
coll.codec_options,
)
# Create a new data key for the encryptedField.
data_key_id = client_encryption.create_data_key(
"local", key_alt_names=["pymongo_encryption_example_3"]
)
# Explicitly encrypt a field:
encrypted_field = client_encryption.encrypt(
"123456789",
Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
key_id=data_key_id,
)
coll.insert_one({"encryptedField": encrypted_field})
doc = coll.find_one()
print("Encrypted document: %s" % (doc,))
# Explicitly decrypt the field:
doc["encryptedField"] = client_encryption.decrypt(doc["encryptedField"])
print("Decrypted document: %s" % (doc,))
# Cleanup resources.
client_encryption.close()
client.close()
if __name__ == "__main__":
main()
带自动解密的显式加密#
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 os
from pymongo import MongoClient
from pymongo.encryption import Algorithm, ClientEncryption
from pymongo.encryption_options import AutoEncryptionOpts
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 = MongoClient(auto_encryption_opts=auto_encryption_opts)
coll = client.test.coll
# Clear old data
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.
key_vault.drop()
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
client_encryption = ClientEncryption(
kms_providers,
key_vault_namespace,
# The MongoClient to use for reading/writing to the key vault.
# This can be the same MongoClient 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 MongoClient, Database, or Collection.
coll.codec_options,
)
# Create a new data key for the encryptedField.
data_key_id = client_encryption.create_data_key(
"local", key_alt_names=["pymongo_encryption_example_4"]
)
# Explicitly encrypt a field:
encrypted_field = client_encryption.encrypt(
"123456789",
Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
key_alt_name="pymongo_encryption_example_4",
)
coll.insert_one({"encryptedField": encrypted_field})
# Automatically decrypts any encrypted fields.
doc = coll.find_one()
print("Decrypted document: %s" % (doc,))
unencrypted_coll = MongoClient().test.coll
print("Encrypted document: %s" % (unencrypted_coll.find_one(),))
# Cleanup resources.
client_encryption.close()
client.close()
if __name__ == "__main__":
main()
CSFLE按需凭据#
pymongocrypt
1.4添加了对获取AWS、GCP和Azure云环境的按需KMS凭据的支持。
要使驱动程序的行为能够从环境中获取凭据,请将带有空映射的相应密钥(“AWS”、“GCP”或“azure”)添加到 AutoEncryptionOpts
或 ClientEncryption
选择。
使用AWS凭据的应用程序将如下所示:
from pymongo import MongoClient
from pymongo.encryption import ClientEncryption
client = MongoClient()
client_encryption = ClientEncryption(
# The empty dictionary enables on-demand credentials.
kms_providers={"aws": {}},
key_vault_namespace="keyvault.datakeys",
key_vault_client=client,
codec_options=client.codec_options,
)
master_key = {
"region": "us-east-1",
"key": ("arn:aws:kms:us-east-1:123456789:key/89fcc2c4-08b0-4bd9-9f25-e30687b580d0"),
}
client_encryption.create_data_key("aws", master_key)
以上功能将启用从环境获取AWS凭据的相同行为 MONGODB-AWS系统 身份验证,包括缓存以避免速率限制。
使用GCP凭据的应用程序将如下所示:
from pymongo import MongoClient
from pymongo.encryption import ClientEncryption
client = MongoClient()
client_encryption = ClientEncryption(
# The empty dictionary enables on-demand credentials.
kms_providers={"gcp": {}},
key_vault_namespace="keyvault.datakeys",
key_vault_client=client,
codec_options=client.codec_options,
)
master_key = {
"projectId": "my-project",
"location": "global",
"keyRing": "key-ring-csfle",
"keyName": "key-name-csfle",
}
client_encryption.create_data_key("gcp", master_key)
驱动程序将查询 VM instance metadata 以获得凭据。
使用Azure凭据的应用程序将如下所示,这一次使用 AutoEncryptionOpts
:
from pymongo import MongoClient
from pymongo.encryption_options import AutoEncryptionOpts
# The empty dictionary enables on-demand credentials.
kms_providers = ({"azure": {}},)
key_vault_namespace = "keyvault.datakeys"
auto_encryption_opts = AutoEncryptionOpts(kms_providers, key_vault_namespace)
client = MongoClient(auto_encryption_opts=auto_encryption_opts)
coll = client.test.coll
coll.insert_one({"encryptedField": "123456789"})
司机将会 acquire an access token 从Azure VM。
可查询加密#
自动可查询加密#
自动可查询加密需要MongoDB 7.0+Enterprise或MongoDB 7.0+Atlas集群。
可查询加密是客户端字段级别加密的第二个版本。数据在客户端加密。可查询加密支持编入索引的加密字段,这些字段将在服务器端进一步处理。
可查询加密中的自动加密配置为 encrypted_fields
映射,如下例所示:
import os
from bson.codec_options import CodecOptions
from pymongo import MongoClient
from pymongo.encryption import Algorithm, ClientEncryption, QueryType
from pymongo.encryption_options import AutoEncryptionOpts
local_master_key = os.urandom(96)
kms_providers = {"local": {"key": local_master_key}}
key_vault_namespace = "keyvault.datakeys"
key_vault_client = MongoClient()
client_encryption = ClientEncryption(
kms_providers, key_vault_namespace, key_vault_client, CodecOptions()
)
key_vault = key_vault_client["keyvault"]["datakeys"]
key_vault.drop()
# Ensure that two data keys cannot share the same keyAltName.
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
key1_id = client_encryption.create_data_key("local", key_alt_names=["firstName"])
key2_id = client_encryption.create_data_key("local", key_alt_names=["lastName"])
encrypted_fields_map = {
"default.encryptedCollection": {
"escCollection": "encryptedCollection.esc",
"ecocCollection": "encryptedCollection.ecoc",
"fields": [
{
"path": "firstName",
"bsonType": "string",
"keyId": key1_id,
"queries": [{"queryType": "equality"}],
},
{
"path": "lastName",
"bsonType": "string",
"keyId": key2_id,
},
],
}
}
auto_encryption_opts = AutoEncryptionOpts(
kms_providers,
key_vault_namespace,
encrypted_fields_map=encrypted_fields_map,
)
client = MongoClient(auto_encryption_opts=auto_encryption_opts)
client.default.drop_collection("encryptedCollection")
coll = client.default.create_collection("encryptedCollection")
coll.insert_one({"_id": 1, "firstName": "Jane", "lastName": "Doe"})
docs = list(coll.find({"firstName": "Jane"}))
print(docs)
在上面的示例中, firstName
和 lastName
自动对字段进行加密和解密。
Explicit Queryable Encryption#
显式可查询加密需要MongoDB 7.0+。
可查询加密是客户端字段级别加密的第二个版本。数据在客户端加密。可查询加密支持编入索引的加密字段,这些字段将在服务器端进一步处理。
可查询加密中的显式加密是使用 encrypt
和 decrypt
方法:研究方法。自动加密(以允许 find_one
以自动解密)是使用 encrypted_fields
映射,如下例所示:
import os
from pymongo import MongoClient
from pymongo.encryption import (
Algorithm,
AutoEncryptionOpts,
ClientEncryption,
QueryType,
)
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)
# Set up the key vault (key_vault_namespace) for this example.
client = MongoClient()
key_vault = client[key_vault_db_name][key_vault_coll_name]
# Ensure that two data keys cannot share the same keyAltName.
key_vault.drop()
key_vault.create_index(
"keyAltNames",
unique=True,
partialFilterExpression={"keyAltNames": {"$exists": True}},
)
client_encryption = ClientEncryption(
kms_providers,
key_vault_namespace,
# The MongoClient to use for reading/writing to the key vault.
# This can be the same MongoClient 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 MongoClient, Database, or Collection.
client.codec_options,
)
# Create a new data key for the encryptedField.
indexed_key_id = client_encryption.create_data_key("local")
unindexed_key_id = client_encryption.create_data_key("local")
encrypted_fields = {
"escCollection": "enxcol_.default.esc",
"ecocCollection": "enxcol_.default.ecoc",
"fields": [
{
"keyId": indexed_key_id,
"path": "encryptedIndexed",
"bsonType": "string",
"queries": {"queryType": "equality"},
},
{
"keyId": unindexed_key_id,
"path": "encryptedUnindexed",
"bsonType": "string",
},
],
}
opts = AutoEncryptionOpts(
{"local": {"key": local_master_key}},
key_vault.full_name,
bypass_query_analysis=True,
key_vault_client=client,
)
# The MongoClient used to read/write application data.
encrypted_client = MongoClient(auto_encryption_opts=opts)
encrypted_client.drop_database("test")
db = encrypted_client.test
# Create the collection with encrypted fields.
coll = db.create_collection("coll", encryptedFields=encrypted_fields)
# Create and encrypt an indexed and unindexed value.
val = "encrypted indexed value"
unindexed_val = "encrypted unindexed value"
insert_payload_indexed = client_encryption.encrypt(
val, Algorithm.INDEXED, indexed_key_id, contention_factor=1
)
insert_payload_unindexed = client_encryption.encrypt(
unindexed_val, Algorithm.UNINDEXED, unindexed_key_id
)
# Insert the payloads.
coll.insert_one(
{
"encryptedIndexed": insert_payload_indexed,
"encryptedUnindexed": insert_payload_unindexed,
}
)
# Encrypt our find payload using QueryType.EQUALITY.
# The value of "indexed_key_id" must be the same as used to encrypt
# the values above.
find_payload = client_encryption.encrypt(
val,
Algorithm.INDEXED,
indexed_key_id,
query_type=QueryType.EQUALITY,
contention_factor=1,
)
# Find the document we inserted using the encrypted payload.
# The returned document is automatically decrypted.
doc = coll.find_one({"encryptedIndexed": find_payload})
print("Returned document: %s" % (doc,))
# Cleanup resources.
client_encryption.close()
encrypted_client.close()
client.close()
if __name__ == "__main__":
main()