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¶
Handles encryption/decryption of proxy credentials with key rotation support. |
Functions¶
Create MultiFernet instance with all valid encryption keys. |
|
Get all valid encryption keys for MultiFernet. |
|
|
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:
- 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:
- 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:
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)