1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
# Copyright 2024-present MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from pymongocrypt.mongocrypt import MongoCrypt
from pymongocrypt.options import DataKeyOpts, ExplicitEncryptOpts
from pymongocrypt.synchronous.state_machine import run_state_machine
class ExplicitEncrypter:
def __init__(self, callback, mongo_crypt_opts):
"""Encrypts and decrypts BSON values.
This class is used by a driver to support explicit encryption and
decryption of individual fields in a BSON document.
:Parameters:
- `callback`: A :class:`MongoCryptCallback`.
- `mongo_crypt_opts`: A :class:`MongoCryptOptions`.
"""
self.callback = callback
if mongo_crypt_opts.schema_map is not None:
raise ValueError("mongo_crypt_opts.schema_map must be None")
self.mongocrypt = MongoCrypt(mongo_crypt_opts, callback)
def create_data_key(
self, kms_provider, master_key=None, key_alt_names=None, key_material=None
):
"""Creates a data key used for explicit encryption.
:Parameters:
- `kms_provider`: The KMS provider to use. Supported values are
"aws", "azure", "gcp", "kmip", "local", or a named provider like
"kmip:name".
- `master_key`: See class:`DataKeyOpts`.
- `key_alt_names` (optional): An optional list of string alternate
names used to reference a key. If a key is created with alternate
names, then encryption may refer to the key by the unique
alternate name instead of by ``_id``.
- `key_material`: (optional) See class:`DataKeyOpts`.
:Returns:
The _id of the created data key document.
"""
# CDRIVER-3275 each key_alt_name needs to be wrapped in a bson
# document.
encoded_names = []
if key_alt_names is not None:
for name in key_alt_names:
encoded_names.append(self.callback.bson_encode({"keyAltName": name}))
if key_material is not None:
key_material = self.callback.bson_encode({"keyMaterial": key_material})
opts = DataKeyOpts(master_key, encoded_names, key_material)
with self.mongocrypt.data_key_context(kms_provider, opts) as ctx:
key = run_state_machine(ctx, self.callback)
return self.callback.insert_data_key(key)
def rewrap_many_data_key(self, filter, provider=None, master_key=None):
"""Decrypts and encrypts all matching data keys with a possibly new `master_key` value.
:Parameters:
- `filter`: A document used to filter the data keys.
- `provider`: (optional) The name of a different kms provider.
- `master_key`: Optional document for the given provider.
:Returns:
A binary document with the rewrap data.
"""
with self.mongocrypt.rewrap_many_data_key_context(
filter, provider, master_key
) as ctx:
return run_state_machine(ctx, self.callback)
def encrypt(
self,
value,
algorithm,
key_id=None,
key_alt_name=None,
query_type=None,
contention_factor=None,
range_opts=None,
is_expression=False,
text_opts=None,
):
"""Encrypts a BSON value.
Note that exactly one of ``key_id`` or ``key_alt_name`` must be
provided.
:Parameters:
- `value` (bytes): The BSON value to encrypt.
- `algorithm` (string): The encryption algorithm to use. See
:class:`Algorithm` for some valid options.
- `key_id` (bytes): The bytes of the binary subtype 4 ``_id`` data
key. For example, ``uuid.bytes`` or ``bytes(bson_binary)``.
- `key_alt_name` (string): Identifies a key vault document by
'keyAltName'.
- `query_type` (str): The query type to execute.
- `contention_factor` (int): The contention factor to use
when the algorithm is "Indexed".
- `range_opts` (bytes): Options for explicit encryption
with the "range" algorithm encoded as a BSON document.
- `is_expression` (boolean): True if this is an encryptExpression()
context. Defaults to False.
- `text_opts` (bytes): Options for explicit encryption
with the "textPreview" algorithm encoded as a BSON document.
:Returns:
The encrypted BSON value.
.. versionchanged:: 1.3
Added the `query_type` and `contention_factor` parameters.
.. versionchanged:: 1.5
Added the `range_opts` and `is_expression` parameters.
.. versionchanged:: 1.16
Added the `text_opts` parameter.
"""
# CDRIVER-3275 key_alt_name needs to be wrapped in a bson document.
if key_alt_name is not None:
key_alt_name = self.callback.bson_encode({"keyAltName": key_alt_name})
opts = ExplicitEncryptOpts(
algorithm,
key_id,
key_alt_name,
query_type,
contention_factor,
range_opts,
is_expression,
text_opts,
)
with self.mongocrypt.explicit_encryption_context(value, opts) as ctx:
return run_state_machine(ctx, self.callback)
def decrypt(self, value):
"""Decrypts a BSON value.
:Parameters:
- `value`: The encoded document to decrypt, which must be in the
form { "v" : encrypted BSON value }}.
:Returns:
The decrypted BSON value.
"""
with self.mongocrypt.explicit_decryption_context(value) as ctx:
return run_state_machine(ctx, self.callback)
def close(self):
"""Cleanup resources."""
self.mongocrypt.close()
|
