|
| 1 | +# Copyright (C) 2026 Percona LLC |
| 2 | +# |
| 3 | +# This program is free software: you can redistribute it and/or modify |
| 4 | +# it under the terms of the GNU Affero General Public License as published by |
| 5 | +# the Free Software Foundation, either version 3 of the License, or |
| 6 | +# (at your option) any later version. |
| 7 | +# |
| 8 | +# This program is distributed in the hope that it will be useful, |
| 9 | +# but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 10 | +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| 11 | +# GNU Affero General Public License for more details. |
| 12 | +# |
| 13 | +# You should have received a copy of the GNU Affero General Public License |
| 14 | +# along with this program. If not, see <https://www.gnu.org/licenses/>. |
| 15 | + |
| 16 | +"""Encrypt and decrypt values SEP stores at rest, keyed by ``ENCRYPTION_KEY``. |
| 17 | +
|
| 18 | +The ciphertext is Fernet: authenticated AES-128-CBC carrying its own version |
| 19 | +marker, timestamp and HMAC, rendered as URL-safe base64 text that any ``str`` |
| 20 | +or JSON column stores unchanged. Encryption is **not** deterministic: each call |
| 21 | +derives a fresh IV, so two encryptions of one plaintext differ and ciphertext |
| 22 | +can never be compared for equality. |
| 23 | +
|
| 24 | +Use :func:`is_encrypted`, never a caught :class:`DecryptionError`, to decide |
| 25 | +whether a stored value still needs encrypting. |
| 26 | +""" |
| 27 | + |
| 28 | +__all__ = ["DecryptionError", "decrypt", "encrypt", "is_encrypted"] |
| 29 | + |
| 30 | +import base64 |
| 31 | +from functools import lru_cache |
| 32 | + |
| 33 | +from cryptography.fernet import Fernet, InvalidToken |
| 34 | + |
| 35 | +from app.core.config import settings |
| 36 | + |
| 37 | +_FERNET_VERSION = 0x80 |
| 38 | +"""The first byte of every decoded Fernet token, which is its version marker.""" |
| 39 | + |
| 40 | +_MIN_TOKEN_BYTES = 73 |
| 41 | +"""The shortest decodable Fernet token: version, timestamp, IV, one block, HMAC. |
| 42 | +
|
| 43 | +CBC pads even an empty plaintext to a full 16-byte block, so no shorter value is |
| 44 | +decryptable. Accepting one would make a migration *skip* a value it can never |
| 45 | +decrypt, leaving it in the clear for good. |
| 46 | +""" |
| 47 | + |
| 48 | + |
| 49 | +class DecryptionError(ValueError): |
| 50 | + """Define exception raised when a value cannot be decrypted with the configured key.""" |
| 51 | + |
| 52 | + |
| 53 | +@lru_cache(maxsize=1) |
| 54 | +def _get_fernet() -> Fernet: |
| 55 | + """Return the process-wide cipher built from ``settings.ENCRYPTION_KEY``. |
| 56 | +
|
| 57 | + Cached so the key is resolved once per process; ``cache_clear()`` resets it |
| 58 | + between tests. Deferred behind an accessor rather than built at module |
| 59 | + scope so importing this module resolves no settings. |
| 60 | +
|
| 61 | + :return: The cached cipher. |
| 62 | + :raises RuntimeError: If ``ENCRYPTION_KEY`` is unset. |
| 63 | + :raises ValueError: Propagates from ``Fernet`` if ``ENCRYPTION_KEY`` is set |
| 64 | + but malformed. |
| 65 | + :meth:`~app.core.config.Settings.validate_encryption_key` refuses to |
| 66 | + construct settings in either case, so both mean a patched environment. |
| 67 | + """ |
| 68 | + key = settings.ENCRYPTION_KEY |
| 69 | + if key is None: |
| 70 | + raise RuntimeError("ENCRYPTION_KEY must be configured.") |
| 71 | + return Fernet(key.get_secret_value().encode()) |
| 72 | + |
| 73 | + |
| 74 | +def encrypt(value: str) -> str: |
| 75 | + """Return ``value`` encrypted as URL-safe base64 ciphertext text. |
| 76 | +
|
| 77 | + :param value: The plaintext to encrypt. |
| 78 | + :return: The ciphertext, storable in any text or JSON column. |
| 79 | + :raises RuntimeError: Propagates from :func:`_get_fernet` when |
| 80 | + ``ENCRYPTION_KEY`` is unset. |
| 81 | + """ |
| 82 | + return _get_fernet().encrypt(value.encode()).decode("ascii") |
| 83 | + |
| 84 | + |
| 85 | +def decrypt(value: str) -> str: |
| 86 | + """Return the plaintext behind ``value``. |
| 87 | +
|
| 88 | + ``value`` is encoded here rather than handed over as text: Fernet narrows a |
| 89 | + ``str`` token with ``ascii``, and :mod:`base64` turns that failure into a |
| 90 | + plain ``ValueError`` its ``binascii.Error`` handler does not catch, so a |
| 91 | + non-ASCII stored value would escape uncaught rather than as the failure this |
| 92 | + function documents. |
| 93 | +
|
| 94 | + :param value: The ciphertext to decrypt. |
| 95 | + :return: The decrypted plaintext. |
| 96 | + :raises DecryptionError: If ``value`` is not ciphertext this key produced, |
| 97 | + which covers a legacy plaintext value, a corrupt one, and one encrypted |
| 98 | + under a different key alike. Use :func:`is_encrypted` to tell those |
| 99 | + apart; this exception does not. |
| 100 | + :raises RuntimeError: Propagates from :func:`_get_fernet` when |
| 101 | + ``ENCRYPTION_KEY`` is unset. |
| 102 | + """ |
| 103 | + try: |
| 104 | + return _get_fernet().decrypt(value.encode()).decode() |
| 105 | + except InvalidToken as exc: |
| 106 | + raise DecryptionError( |
| 107 | + "Value could not be decrypted: it is malformed, or was encrypted " |
| 108 | + "with a different ENCRYPTION_KEY." |
| 109 | + ) from exc |
| 110 | + |
| 111 | + |
| 112 | +def is_encrypted(value: str) -> bool: |
| 113 | + """Return whether ``value`` is structurally a Fernet token. |
| 114 | +
|
| 115 | + Reads the token's own version marker instead of attempting a decrypt, so a |
| 116 | + token written under a *different* key still reports ``True``. That is the |
| 117 | + property a migration needs: a caught :class:`DecryptionError` cannot |
| 118 | + separate "never encrypted" from "encrypted with a key this process does not |
| 119 | + hold", and encrypting the latter again destroys the only copy of its |
| 120 | + plaintext. |
| 121 | +
|
| 122 | + :param value: The stored value to classify. |
| 123 | + :return: ``True`` when ``value`` is shaped like a Fernet token, ``False`` |
| 124 | + for anything else, including input that is not valid base64 at all. |
| 125 | + """ |
| 126 | + try: |
| 127 | + raw = base64.urlsafe_b64decode(value) |
| 128 | + except ValueError: |
| 129 | + return False |
| 130 | + return len(raw) >= _MIN_TOKEN_BYTES and raw[0] == _FERNET_VERSION |
0 commit comments