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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
|
import abc
import base64
import contextlib
import os
from typing import TYPE_CHECKING, Any, Callable, Optional, Union
from sqlalchemy import String, Text, TypeDecorator
from sqlalchemy import func as sql_func
from advanced_alchemy.exceptions import IntegrityError
if TYPE_CHECKING:
from sqlalchemy.engine import Dialect
cryptography = None # type: ignore[var-annotated,unused-ignore]
with contextlib.suppress(ImportError):
from cryptography.fernet import Fernet
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
__all__ = ("EncryptedString", "EncryptedText", "EncryptionBackend", "FernetBackend", "PGCryptoBackend")
class EncryptionBackend(abc.ABC):
"""Abstract base class for encryption backends.
This class defines the interface that all encryption backends must implement.
Concrete implementations should provide the actual encryption/decryption logic.
Attributes:
passphrase (bytes): The encryption passphrase used by the backend.
"""
def mount_vault(self, key: "Union[str, bytes]") -> None:
"""Mounts the vault with the provided encryption key.
Args:
key (str | bytes): The encryption key used to initialize the backend.
"""
if isinstance(key, str):
key = key.encode()
@abc.abstractmethod
def init_engine(self, key: "Union[bytes, str]") -> None: # pragma: no cover
"""Initializes the encryption engine with the provided key.
Args:
key (bytes | str): The encryption key.
Raises:
NotImplementedError: If the method is not implemented by the subclass.
"""
@abc.abstractmethod
def encrypt(self, value: Any) -> str: # pragma: no cover
"""Encrypts the given value.
Args:
value (Any): The value to encrypt.
Returns:
str: The encrypted value.
Raises:
NotImplementedError: If the method is not implemented by the subclass.
"""
@abc.abstractmethod
def decrypt(self, value: Any) -> str: # pragma: no cover
"""Decrypts the given value.
Args:
value (Any): The value to decrypt.
Returns:
str: The decrypted value.
Raises:
NotImplementedError: If the method is not implemented by the subclass.
"""
class PGCryptoBackend(EncryptionBackend):
"""PostgreSQL pgcrypto-based encryption backend.
This backend uses PostgreSQL's pgcrypto extension for encryption/decryption operations.
Requires the pgcrypto extension to be installed in the database.
Attributes:
passphrase (bytes): The base64-encoded passphrase used for encryption and decryption.
"""
def init_engine(self, key: "Union[bytes, str]") -> None:
"""Initializes the pgcrypto engine with the provided key.
Args:
key (bytes | str): The encryption key.
"""
if isinstance(key, str):
key = key.encode()
self.passphrase = base64.urlsafe_b64encode(key)
def encrypt(self, value: Any) -> str:
"""Encrypts the given value using pgcrypto.
Args:
value (Any): The value to encrypt.
Returns:
str: The encrypted value.
"""
if not isinstance(value, str): # pragma: no cover
value = repr(value)
value = value.encode()
return sql_func.pgp_sym_encrypt(value, self.passphrase) # type: ignore[return-value]
def decrypt(self, value: Any) -> str:
"""Decrypts the given value using pgcrypto.
Args:
value (Any): The value to decrypt.
Returns:
str: The decrypted value.
"""
if not isinstance(value, str): # pragma: no cover
value = str(value)
return sql_func.pgp_sym_decrypt(value, self.passphrase) # type: ignore[return-value]
class FernetBackend(EncryptionBackend):
"""Fernet-based encryption backend.
This backend uses the Python cryptography library's Fernet implementation
for encryption/decryption operations. Provides symmetric encryption with
built-in rotation support.
Attributes:
key (bytes): The base64-encoded key used for encryption and decryption.
fernet (cryptography.fernet.Fernet): The Fernet instance used for encryption/decryption.
"""
def mount_vault(self, key: "Union[str, bytes]") -> None:
"""Mounts the vault with the provided encryption key.
This method hashes the key using SHA256 before initializing the engine.
Args:
key (str | bytes): The encryption key.
"""
if isinstance(key, str):
key = key.encode()
digest = hashes.Hash(hashes.SHA256(), backend=default_backend()) # pyright: ignore[reportPossiblyUnboundVariable]
digest.update(key)
engine_key = digest.finalize()
self.init_engine(engine_key)
def init_engine(self, key: "Union[bytes, str]") -> None:
"""Initializes the Fernet engine with the provided key.
Args:
key (bytes | str): The encryption key.
"""
if isinstance(key, str):
key = key.encode()
self.key = base64.urlsafe_b64encode(key)
self.fernet = Fernet(self.key) # pyright: ignore[reportPossiblyUnboundVariable]
def encrypt(self, value: Any) -> str:
"""Encrypts the given value using Fernet.
Args:
value (Any): The value to encrypt.
Returns:
str: The encrypted value.
"""
if not isinstance(value, str):
value = repr(value)
value = value.encode()
encrypted = self.fernet.encrypt(value)
return encrypted.decode("utf-8")
def decrypt(self, value: Any) -> str:
"""Decrypts the given value using Fernet.
Args:
value (Any): The value to decrypt.
Returns:
str: The decrypted value.
"""
if not isinstance(value, str): # pragma: no cover
value = str(value)
decrypted: Union[str, bytes] = self.fernet.decrypt(value.encode())
if not isinstance(decrypted, str):
decrypted = decrypted.decode("utf-8") # pyright: ignore[reportAttributeAccessIssue]
return decrypted
DEFAULT_ENCRYPTION_KEY = os.urandom(32)
class EncryptedString(TypeDecorator[str]):
"""SQLAlchemy TypeDecorator for storing encrypted string values in a database.
This type provides transparent encryption/decryption of string values using the specified backend.
It extends :class:`sqlalchemy.types.TypeDecorator` and implements String as its underlying type.
Args:
key (str | bytes | Callable[[], str | bytes] | None): The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).
backend (Type[EncryptionBackend] | None): The encryption backend class to use. Defaults to FernetBackend.
length (int | None): The length of the unencrypted string. This is used for documentation and validation purposes only, as encrypted strings will be longer.
**kwargs (Any | None): Additional arguments passed to the underlying String type.
Attributes:
key (str | bytes | Callable[[], str | bytes]): The encryption key.
backend (EncryptionBackend): The encryption backend instance.
length (int | None): The unencrypted string length.
"""
impl = String
cache_ok = True
def __init__(
self,
key: "Union[str, bytes, Callable[[], Union[str, bytes]]]" = DEFAULT_ENCRYPTION_KEY,
backend: "type[EncryptionBackend]" = FernetBackend,
length: "Optional[int]" = None,
**kwargs: Any,
) -> None:
"""Initializes the EncryptedString TypeDecorator.
Args:
key (str | bytes | Callable[[], str | bytes] | None): The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).
backend (Type[EncryptionBackend] | None): The encryption backend class to use. Defaults to FernetBackend.
length (int | None): The length of the unencrypted string. This is used for documentation and validation purposes only.
**kwargs (Any | None): Additional arguments passed to the underlying String type.
"""
super().__init__()
self.key = key
self.backend = backend()
self.length = length
@property
def python_type(self) -> type[str]:
"""Returns the Python type for this type decorator.
Returns:
Type[str]: The Python string type.
"""
return str
def load_dialect_impl(self, dialect: "Dialect") -> Any:
"""Loads the appropriate dialect implementation based on the database dialect.
Note: The actual column length will be larger than the specified length due to encryption overhead.
For most encryption methods, the encrypted string will be approximately 1.35x longer than the original.
Args:
dialect (Dialect): The SQLAlchemy dialect.
Returns:
Any: The dialect-specific type descriptor.
"""
if dialect.name in {"mysql", "mariadb"}:
# For MySQL/MariaDB, always use Text to avoid length limitations
return dialect.type_descriptor(Text())
if dialect.name == "oracle":
# Oracle has a 4000-byte limit for VARCHAR2 (by default)
return dialect.type_descriptor(String(length=4000))
return dialect.type_descriptor(String())
def process_bind_param(self, value: Any, dialect: "Dialect") -> "Union[str, None]":
"""Processes the value before binding it to the SQL statement.
This method encrypts the value using the specified backend and validates length if specified.
Args:
value (Any): The value to process.
dialect (Dialect): The SQLAlchemy dialect.
Raises:
IntegrityError: If the unencrypted value exceeds the maximum length.
Returns:
str | None: The encrypted value or None if the input is None.
"""
if value is None:
return value
# Validate length if specified
if self.length is not None and len(str(value)) > self.length:
msg = f"Unencrypted value exceeds maximum unencrypted length of {self.length}"
raise IntegrityError(msg)
self.mount_vault()
return self.backend.encrypt(value)
def process_result_value(self, value: Any, dialect: "Dialect") -> "Union[str, None]":
"""Processes the value after retrieving it from the database.
This method decrypts the value using the specified backend.
Args:
value (Any): The value to process.
dialect (Dialect): The SQLAlchemy dialect.
Returns:
str | None: The decrypted value or None if the input is None.
"""
if value is None:
return value
self.mount_vault()
return self.backend.decrypt(value)
def mount_vault(self) -> None:
"""Mounts the vault with the encryption key.
If the key is callable, it is called to retrieve the key. Otherwise, the key is used directly.
"""
key = self.key() if callable(self.key) else self.key
self.backend.mount_vault(key)
class EncryptedText(EncryptedString):
"""SQLAlchemy TypeDecorator for storing encrypted text/CLOB values in a database.
This type provides transparent encryption/decryption of text values using the specified backend.
It extends :class:`EncryptedString` and implements Text as its underlying type.
This is suitable for storing larger encrypted text content compared to EncryptedString.
Args:
key (str | bytes | Callable[[], str | bytes] | None): The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).
backend (Type[EncryptionBackend] | None): The encryption backend class to use. Defaults to FernetBackend.
**kwargs (Any | None): Additional arguments passed to the underlying String type.
"""
impl = Text
cache_ok = True
def load_dialect_impl(self, dialect: "Dialect") -> Any:
"""Loads the appropriate dialect implementation for Text type.
Args:
dialect (Dialect): The SQLAlchemy dialect.
Returns:
Any: The dialect-specific Text type descriptor.
"""
return dialect.type_descriptor(Text())
|
