Encryption

Introduction

Expanse supports application-level encryption using the AES-256-GCM algorithm. This encryption exists to protect sensitive information in your application such as personally identifiable information (PII) from users.

This provides an additional layer of security to your application, even if your database is encrypted at rest, if an attacker were to gain access to the database or a snapshot of it, for example.

Configuration

Before you can use encryption in your application, you need to specify an encryption key using the APP_SECRET_KEY environment variable.

The key must be a 32-byte string encoded in base64. You can generate a key using the following command:

./beam encryption key generate

This will generate random, secure keys and store them in your .env file:

• the APP_SECRET_KEY variable from which the encryption keys used to encrypt will be derived
• the ENCRYPTION_SALT variable which will be used to derive the encryption keys

If you only want to generate either the APP_SECRET_KEY or the ENCRYPTION_SALT variable, you can use the --key or --salt option respectively:

./beam encryption key generate --key
./beam encryption key generate --salt

If you prefer to only display the keys without storing them, you can use the --show option:

./beam encryption key generate --show

Rotating encryption keys

If you need to change the encryption key, you can specify the previous encryption keys in the APP_PREVIOUS_KEYS environment variable, separated by a comma.

APP_SECRET_KEY=base64:_iFScL34AlXtSRs04WGFTz08J9zYxnOEGm5aMH-6FzE=
APP_PREVIOUS_KEYS=base64:YFw3lot60UF146X5Al7jz2Y-fD55XwmOcRXrbgtJ7bE=

When decrypting data, Expanse will try to decrypt the data first using the current key and, if it fails, will try to use the previous keys. That way you can gracefully change your encryption key without interruption.

Encrypting data

To encrypt data, you can use the encrypt() method from an Encryptor instance:

from sqlalchemy import select
from expanse.database.session import Session
from expanse.contracts.encryption.encryptor import Encryptor
from expanse.http.response import Response

from app.models.user import User


class UserController:

    def store(
        self, user_id: int, encryptor: Encryptor, session: Session
    ) -> Response:
        user = session.scalar(select(User).where(User.id == user_id))
        user.email = encryptor.encrypt(user.email)

        session.commit()
        # ...

Before encrypting the data, the encryptor will first derive a key from the configured secret key — using the derivation salt configured via the ENCRYPTION_SALT environment variable — and use it to encrypt the data.

The encrypted Message

The encrypt() method returns an instance of the Message class. It holds the encrypted data and some additional metadata that will be used to decrypt the data.

message = encryptor.encrypt("sensitive data")

You can serialize the Message instance to various format to JSON using the dump() method:

• dict (default): Serialize the message to a dictionary

{
    "p": "Q//+7F0pGd4n+bSDDz4zaCpQMGhKxg==",
    "h": {
        "iv": "RDTowbVAwRvVdYPQ",
        "at": "k4Ifm7zlKcJ1oYXVetBYcA==",
        "z": 1
    }
}

• json: Serialize the message to a JSON string

'{"p": "Q//+7F0pGd4n+bSDDz4zaCpQMGhKxg==", "h": {"iv": "RDTowbVAwRvVdYPQ", "at": "k4Ifm7zlKcJ1oYXVetBYcA==", "z": 1}}'

• base64: Serialize the message to a base64 string

"eyJwIjogIlEvLys3RjBwR2Q0bitiU0REejR6YUNwUU1HaEt4Zz09IiwgImgiOiB7Iml2IjogIlJEVG93YlZBd1J2VmRZUFEiLCAiYXQiOiAiazRJZm03emxLY0oxb1lYVmV0QlljQT09IiwgInoiOiAxfX0"

When the format is serialized, any string or bytes data will be encoded in base64.

Decrypting data

To decrypt data, you can use the decrypt() method – which accepts a Message instance – from an Encryptor instance:

from sqlalchemy import select
from expanse.database.session import Session
from expanse.contracts.encryption.encryptor import Encryptor
from expanse.encryption.message import Message
from expanse.http.response import Response

from app.models.user import User


class UserController:

    def store(
        self, user_id: int, encryptor: Encryptor, session: Session
    ) -> Response:
        user = session.scalar(select(User).where(User.id == user_id))
        clear_email = encryptor.decrypt(Message.load(user.email, "json"))
        # ...

You can instantiate a Message instance from a serialized message using the load() method. It accepts the same formats as the dump() method.

Message.load(message.dump())
Message.load(message.dump("json"), "json")
Message.load(message.dump("base64"), "base64")

If the decryption fails, a DecryptionError exception will be raised.

try:
    clear_email = encryptor.decrypt(user.email)
except DecryptionError:
    # Handle the error
    ...