mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 13:53:15 +00:00
148efeb269
Threat model T6: Fernet token-encryption key had no rotation path. Rotating TOKEN_ENCRYPTION_KEY immediately invalidated every github_token_encrypted value in thread metadata, forcing re-auth or bot-token re-resolution. Switch to cryptography.fernet.MultiFernet and parse TOKEN_ENCRYPTION_KEY as a comma- or newline-separated ordered list (most-recent-first). New writes encrypt under the first key; reads try every key. Deployers can prepend a new key, let active threads roll over, then drop the old key. Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> Co-authored-by: Johannes du Plessis <51395795+johannes117@users.noreply.github.com>
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""Encryption utilities for sensitive data like tokens."""
|
|
|
|
import logging
|
|
import os
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken, MultiFernet
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EncryptionKeyMissingError(ValueError):
|
|
"""Raised when TOKEN_ENCRYPTION_KEY environment variable is not set."""
|
|
|
|
|
|
def _parse_encryption_keys(raw: str) -> list[bytes]:
|
|
"""Split TOKEN_ENCRYPTION_KEY into one or more keys (most-recent-first)."""
|
|
keys: list[bytes] = []
|
|
for part in raw.replace("\n", ",").split(","):
|
|
stripped = part.strip()
|
|
if stripped:
|
|
keys.append(stripped.encode())
|
|
return keys
|
|
|
|
|
|
def _get_encryption_keys() -> list[bytes]:
|
|
"""Read the ordered key list from TOKEN_ENCRYPTION_KEY (most-recent-first).
|
|
|
|
Accepts a single key or a comma/newline separated list. The first key is
|
|
used for encryption; every key is tried for decryption.
|
|
|
|
Raises:
|
|
EncryptionKeyMissingError: If TOKEN_ENCRYPTION_KEY is unset or empty.
|
|
"""
|
|
explicit_key = os.environ.get("TOKEN_ENCRYPTION_KEY")
|
|
if not explicit_key:
|
|
raise EncryptionKeyMissingError
|
|
|
|
keys = _parse_encryption_keys(explicit_key)
|
|
if not keys:
|
|
raise EncryptionKeyMissingError
|
|
return keys
|
|
|
|
|
|
def _get_fernet() -> MultiFernet:
|
|
"""Build a MultiFernet from the configured key list."""
|
|
return MultiFernet([Fernet(k) for k in _get_encryption_keys()])
|
|
|
|
|
|
def encrypt_token(token: str) -> str:
|
|
"""Encrypt a token under the newest configured key."""
|
|
if not token:
|
|
return ""
|
|
|
|
encrypted = _get_fernet().encrypt(token.encode())
|
|
return encrypted.decode()
|
|
|
|
|
|
def decrypt_token(encrypted_token: str) -> str:
|
|
"""Decrypt a token, trying each configured key in order."""
|
|
if not encrypted_token:
|
|
return ""
|
|
|
|
try:
|
|
decrypted = _get_fernet().decrypt(encrypted_token.encode())
|
|
return decrypted.decode()
|
|
except InvalidToken:
|
|
logger.warning("Failed to decrypt token: invalid token")
|
|
return ""
|
|
except EncryptionKeyMissingError:
|
|
logger.warning("Failed to decrypt token: encryption key not set")
|
|
return ""
|