sqlalchemy_encrypted

EncryptedString

Bases: types.TypeDecorator

Used to store encrypted values in a database

Source code in ormar\fields\sqlalchemy_encrypted.py

class EncryptedString(types.TypeDecorator):
    """
    Used to store encrypted values in a database
    """

    impl = types.TypeEngine

    def __init__(
        self,
        encrypt_secret: Union[str, Callable],
        encrypt_backend: EncryptBackends = EncryptBackends.FERNET,
        encrypt_custom_backend: Type[EncryptBackend] = None,
        **kwargs: Any,
    ) -> None:
        _field_type = kwargs.pop("_field_type")
        super().__init__()
        if not cryptography:  # pragma: nocover
            raise ModelDefinitionError(
                "In order to encrypt a column 'cryptography' is required!"
            )
        backend = BACKENDS_MAP.get(encrypt_backend, encrypt_custom_backend)
        if not backend or not lenient_issubclass(backend, EncryptBackend):
            raise ModelDefinitionError("Wrong or no encrypt backend provided!")

        self.backend: EncryptBackend = backend()
        self._field_type: "BaseField" = _field_type
        self._underlying_type: Any = _field_type.column_type
        self._key: Union[str, Callable] = encrypt_secret
        type_ = self._field_type.__type__
        if type_ is None:  # pragma: nocover
            raise ModelDefinitionError(
                f"Improperly configured field " f"{self._field_type.name}"
            )
        self.type_: Any = type_

    def __repr__(self) -> str:  # pragma: nocover
        return "TEXT()"

    def load_dialect_impl(self, dialect: Dialect) -> Any:
        return dialect.type_descriptor(types.TEXT())

    def _refresh(self) -> None:
        key = self._key() if callable(self._key) else self._key
        self.backend._refresh(key)

    def process_bind_param(self, value: Any, dialect: Dialect) -> Optional[str]:
        if value is None:
            return value
        self._refresh()
        try:
            value = self._underlying_type.process_bind_param(value, dialect)
        except AttributeError:
            encoder = ormar.SQL_ENCODERS_MAP.get(self.type_, None)
            if encoder:
                value = encoder(value)  # type: ignore

        encrypted_value = self.backend.encrypt(value)
        return encrypted_value

    def process_result_value(self, value: Any, dialect: Dialect) -> Any:
        if value is None:
            return value
        self._refresh()
        decrypted_value = self.backend.decrypt(value)
        try:
            return self._underlying_type.process_result_value(decrypted_value, dialect)
        except AttributeError:
            decoder = ormar.DECODERS_MAP.get(self.type_, None)
            if decoder:
                return decoder(decrypted_value)  # type: ignore

            return self._field_type.__type__(decrypted_value)  # type: ignore

FernetBackend

Bases: EncryptBackend

Two-way encryption, data stored in db are encrypted but decrypted during query.

Source code in ormar\fields\sqlalchemy_encrypted.py

class FernetBackend(EncryptBackend):
    """
    Two-way encryption, data stored in db are encrypted but decrypted during query.
    """

    def _initialize_backend(self, secret_key: bytes) -> None:
        self.secret_key = base64.urlsafe_b64encode(secret_key)
        self.fernet = Fernet(self.secret_key)

    def encrypt(self, value: Any) -> str:
        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:
        if not isinstance(value, str):  # pragma: nocover
            value = str(value)
        decrypted: Union[str, bytes] = self.fernet.decrypt(value.encode())
        if not isinstance(decrypted, str):
            decrypted = decrypted.decode("utf-8")
        return decrypted

HashBackend

Bases: EncryptBackend

One-way hashing - in example for passwords, no way to decrypt the value!

Source code in ormar\fields\sqlalchemy_encrypted.py

class HashBackend(EncryptBackend):
    """
    One-way hashing - in example for passwords, no way to decrypt the value!
    """

    def _initialize_backend(self, secret_key: bytes) -> None:
        self.secret_key = base64.urlsafe_b64encode(secret_key)

    def encrypt(self, value: Any) -> str:
        if not isinstance(value, str):  # pragma: nocover
            value = repr(value)
        value = value.encode()
        digest = hashes.Hash(hashes.SHA512(), backend=default_backend())
        digest.update(self.secret_key)
        digest.update(value)
        hashed_value = digest.finalize()
        return hashed_value.hex()

    def decrypt(self, value: Any) -> str:
        if not isinstance(value, str):  # pragma: nocover
            value = str(value)
        return value