proxywhirl.cache.crypto

Credential encryption utilities for cache storage.

Provides Fernet symmetric encryption for proxy credentials at rest (L2/L3 tiers). Uses environment variable PROXYWHIRL_CACHE_ENCRYPTION_KEY for key management. Supports key rotation via MultiFernet with PROXYWHIRL_CACHE_KEY_PREVIOUS.

Classes

CredentialEncryptor

Handles encryption/decryption of proxy credentials with key rotation support.

Functions

create_multi_fernet()

Create MultiFernet instance with all valid encryption keys.

get_encryption_keys()

Get all valid encryption keys for MultiFernet.

rotate_key(new_key)

Rotate encryption keys by setting new current key.

Module Contents

class proxywhirl.cache.crypto.CredentialEncryptor(key=None)[source]

Handles encryption/decryption of proxy credentials with key rotation support.

Uses Fernet symmetric encryption (AES-128-CBC + HMAC) to protect credentials stored in L2 (JSONL files) and L3 (SQLite database). Supports gradual key rotation via MultiFernet using both current and previous keys.

Example

>>> encryptor = CredentialEncryptor()
>>> encrypted = encryptor.encrypt(SecretStr("mypassword"))
>>> decrypted = encryptor.decrypt(encrypted)
>>> decrypted.get_secret_value()
'mypassword'

Initialize encryptor with Fernet key or MultiFernet.

Parameters:

key (bytes | None) – Optional Fernet key (32 url-safe base64-encoded bytes). If None, uses get_encryption_keys() to load current and previous keys from environment variables. If no env vars set, generates a new key (WARNING: regenerated keys cannot decrypt existing cached data).

Raises:

ValueError – If provided key is invalid for Fernet

decrypt(encrypted)[source]

Decrypt encrypted bytes back to SecretStr.

Parameters:

encrypted (bytes) – Encrypted bytes from storage

Returns:

SecretStr containing decrypted plaintext (never logs value)

Raises:

ValueError – If decryption fails (wrong key, corrupted data)

Return type:

pydantic.SecretStr

encrypt(secret)[source]

Encrypt a SecretStr to bytes.

Parameters:

secret (pydantic.SecretStr) – SecretStr containing plaintext to encrypt

Returns:

Encrypted bytes suitable for storage in BLOB fields

Raises:

ValueError – If encryption fails

Return type:

bytes

proxywhirl.cache.crypto.create_multi_fernet()[source]

Create MultiFernet instance with all valid encryption keys.

MultiFernet tries keys in order for decryption (newest first). All new encryptions use the first (current) key.

Returns:

MultiFernet instance configured with current and previous keys

Raises:

ValueError – If any key has invalid format

Return type:

cryptography.fernet.MultiFernet

Example

>>> mf = create_multi_fernet()
>>> encrypted = mf.encrypt(b"secret")
>>> mf.decrypt(encrypted)
b'secret'
proxywhirl.cache.crypto.get_encryption_keys()[source]

Get all valid encryption keys for MultiFernet.

Returns keys in priority order: current key first, then previous key. This allows decryption of data encrypted with either key while always encrypting new data with the current (first) key.

Returns:

List of Fernet keys as bytes. Always contains at least one key. First key is current, subsequent keys are for backward compatibility.

Raises:

ValueError – If any key has invalid Fernet format

Return type:

list[bytes]

Example

>>> keys = get_encryption_keys()
>>> len(keys)  # 1 or 2 depending on env vars
1
proxywhirl.cache.crypto.rotate_key(new_key)[source]

Rotate encryption keys by setting new current key.

This function updates environment variables to perform key rotation: - Current key moves to PROXYWHIRL_CACHE_KEY_PREVIOUS - New key becomes PROXYWHIRL_CACHE_ENCRYPTION_KEY

This allows gradual migration: new data uses new key, old data can still be decrypted with previous key.

Parameters:

new_key (str) – New Fernet key as base64-encoded string

Raises:

ValueError – If new_key has invalid Fernet format

Return type:

None

Example

>>> from cryptography.fernet import Fernet
>>> new_key = Fernet.generate_key().decode()
>>> rotate_key(new_key)