musehub_webhook_crypto.py
python
sha256:cdfe164af70b640f93370907508ae9066c78c19dfd97f36e14aab34c3d539496
docs: add production-readiness checklist
Sonnet 5
33 days ago
| 1 | """Webhook secret encryption — AES-256 envelope encryption for musehub_webhooks.secret. |
| 2 | |
| 3 | Webhook signing secrets must be recoverable at delivery time (so we can compute the |
| 4 | HMAC-SHA256 header for subscribers). One-way hashing (bcrypt/SHA256) is therefore |
| 5 | not an option. Instead we use Fernet symmetric encryption (AES-128-CBC + HMAC-SHA256 |
| 6 | under the hood, equivalent security to AES-256 for the threat model here) keyed with |
| 7 | MUSE_WEBHOOK_SECRET_KEY from the environment. |
| 8 | |
| 9 | Encryption contract |
| 10 | ------------------- |
| 11 | - ``encrypt_secret(plaintext)`` → base64url-encoded Fernet token (str). |
| 12 | - ``decrypt_secret(ciphertext)`` → original plaintext str. |
| 13 | - Both functions are pure (no I/O) and synchronous. |
| 14 | - When MUSE_WEBHOOK_SECRET_KEY is not configured, the functions are transparent |
| 15 | pass-throughs so local dev works without extra setup (see warning in decrypt). |
| 16 | |
| 17 | Key management |
| 18 | -------------- |
| 19 | Generate a key once and store it in the environment: |
| 20 | |
| 21 | python3 -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" |
| 22 | |
| 23 | Set MUSE_WEBHOOK_SECRET_KEY to that value in your .env or secret manager. |
| 24 | Rotate keys by re-encrypting all secrets and updating the env var; Fernet tokens |
| 25 | carry the key version so future decryption needs the matching key. |
| 26 | """ |
| 27 | |
| 28 | import logging |
| 29 | |
| 30 | from cryptography.fernet import Fernet, InvalidToken |
| 31 | |
| 32 | from musehub.config import settings |
| 33 | |
| 34 | logger = logging.getLogger(__name__) |
| 35 | |
| 36 | # Lazily initialised Fernet instance — None when the key is not configured. |
| 37 | _fernet: Fernet | None = None |
| 38 | _fernet_initialised = False |
| 39 | |
| 40 | |
| 41 | def _get_fernet() -> Fernet | None: |
| 42 | """Return the singleton Fernet instance, initialising it on first call. |
| 43 | |
| 44 | Returns None when MUSE_WEBHOOK_SECRET_KEY is not set (local dev fallback). |
| 45 | """ |
| 46 | global _fernet, _fernet_initialised |
| 47 | if _fernet_initialised: |
| 48 | return _fernet |
| 49 | _fernet_initialised = True |
| 50 | key = settings.webhook_secret_key |
| 51 | if not key: |
| 52 | logger.warning( |
| 53 | "⚠️ MUSE_WEBHOOK_SECRET_KEY is not set — webhook secrets stored as plaintext. " |
| 54 | "Set this key in production to encrypt secrets at rest." |
| 55 | ) |
| 56 | return None |
| 57 | _fernet = Fernet(key.encode()) |
| 58 | return _fernet |
| 59 | |
| 60 | |
| 61 | def encrypt_secret(plaintext: str) -> str: |
| 62 | """Encrypt a webhook signing secret for storage in the database. |
| 63 | |
| 64 | Returns a Fernet token (base64url string) when a key is configured, or the |
| 65 | original plaintext when MUSE_WEBHOOK_SECRET_KEY is absent (dev fallback). |
| 66 | Empty secrets are returned as-is regardless of key configuration. |
| 67 | """ |
| 68 | if not plaintext: |
| 69 | return plaintext |
| 70 | fernet = _get_fernet() |
| 71 | if fernet is None: |
| 72 | return plaintext |
| 73 | token: bytes = fernet.encrypt(plaintext.encode()) |
| 74 | return token.decode() |
| 75 | |
| 76 | |
| 77 | _FERNET_TOKEN_PREFIX = "gAAAAAB" |
| 78 | |
| 79 | |
| 80 | def is_fernet_token(value: str) -> bool: |
| 81 | """Return True if *value* looks like a Fernet token. |
| 82 | |
| 83 | Fernet tokens are base64url-encoded and always begin with "gAAAAAB" (the |
| 84 | binary magic bytes 0x80 encoded in URL-safe base64). We use this prefix |
| 85 | to distinguish already-encrypted values from legacy plaintext secrets. |
| 86 | """ |
| 87 | return value.startswith(_FERNET_TOKEN_PREFIX) |
| 88 | |
| 89 | |
| 90 | def decrypt_secret(ciphertext: str) -> str: |
| 91 | """Decrypt a webhook signing secret retrieved from the database. |
| 92 | |
| 93 | Accepts a Fernet token produced by ``encrypt_secret``. Returns the original |
| 94 | plaintext when a key is configured, or the value unchanged when no key is set |
| 95 | (matching the dev fallback in ``encrypt_secret``). |
| 96 | |
| 97 | Raises ``ValueError`` when decryption fails — the value may have been |
| 98 | encrypted with a different key or is corrupt. |
| 99 | Empty values are returned as-is. |
| 100 | """ |
| 101 | if not ciphertext: |
| 102 | return ciphertext |
| 103 | if not is_fernet_token(ciphertext): |
| 104 | return ciphertext |
| 105 | fernet = _get_fernet() |
| 106 | if fernet is None: |
| 107 | return ciphertext |
| 108 | try: |
| 109 | plaintext: bytes = fernet.decrypt(ciphertext.encode()) |
| 110 | return plaintext.decode() |
| 111 | except InvalidToken as exc: |
| 112 | raise ValueError( |
| 113 | "Failed to decrypt webhook secret — the value may have been encrypted " |
| 114 | "with a different key or is corrupt. Check MUSE_WEBHOOK_SECRET_KEY." |
| 115 | ) from exc |
File History
1 commit
sha256:cdfe164af70b640f93370907508ae9066c78c19dfd97f36e14aab34c3d539496
docs: add production-readiness checklist
Sonnet 5
33 days ago