security.py
Rama master · réplica de lectura del repositorio original.
"""Password hashing, signed session tokens, and credential encryption.
Passwords use scrypt (hashlib), session cookies are HMAC-signed stateless
tokens, and third-party credentials (Mastodon/Bluesky/Matrix/micro.blog/
Discord/Telegram tokens, SMTP passwords, vision API keys, OAuth client
secrets) are encrypted at rest with Fernet (cryptography) whenever
FEEDECHO_CREDENTIAL_KEY is set. Multi mode requires that key at startup;
single mode treats it as optional — without it credentials stay plaintext
(the operator owns the database), with it copied DB backups stop being
credential dumps.
"""
import base64
import hashlib
import hmac
import json
import secrets
import threading
import time
import settings
try:
from cryptography.fernet import Fernet
except ImportError: # pragma: no cover - cryptography is a declared dependency
Fernet = None
# scrypt cost parameters: N=2**17, r=8, p=1 (OWASP-recommended minimum,
# ~128 MiB per hash). verify_password only accepts hashes made with
# EXACTLY these parameters — attacker-supplied work factors are rejected
# before any computation, which closes the login-DoS vector.
_SCRYPT_N = 2**17
_SCRYPT_R = 8
_SCRYPT_P = 1
_SCRYPT_MAXMEM = 256 * 1024 * 1024
# scrypt releases the GIL, so N concurrent hashes genuinely run in parallel and
# stack N × ~128 MiB. Bound concurrency so a login/register/reset burst cannot
# OOM the box. Default 4 (≈512 MiB peak); tune via FEEDECHO_SCRYPT_CONCURRENCY.
_scrypt_semaphore = threading.BoundedSemaphore(settings.SCRYPT_CONCURRENCY)
SESSION_TTL_SECONDS = 30 * 24 * 60 * 60 # 30 days
_SESSION_PURPOSE = b"feedecho-session:v1"
_SESSION_AUD = "feedecho-session"
def _b64(data: bytes) -> str:
return base64.urlsafe_b64encode(data).decode()
def _unb64(text: str) -> bytes:
return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
def _scrypt(password: bytes, salt: bytes, n: int, r: int, p: int) -> bytes:
"""Run hashlib.scrypt under the global concurrency cap.
scrypt releases the GIL, so concurrent calls genuinely run in parallel and
each holds ~128 MiB. The semaphore bounds total simultaneous hashes.
"""
with _scrypt_semaphore:
return hashlib.scrypt(
password, salt=salt, n=n, r=r, p=p, maxmem=_SCRYPT_MAXMEM
)
def hash_password(password: str) -> str:
"""scrypt-hash a password. Format: scrypt$N$r$p$salt_b64$digest_b64."""
salt = secrets.token_bytes(16)
digest = _scrypt(password.encode(), salt, _SCRYPT_N, _SCRYPT_R, _SCRYPT_P)
return "scrypt${}${}${}${}${}".format(
_SCRYPT_N, _SCRYPT_R, _SCRYPT_P, _b64(salt), _b64(digest)
)
def verify_password(password: str, stored: str) -> bool:
"""Constant-time verification; False on any malformed or unknown-format
input. Stored work factors must match the current constants exactly —
anything else (forged, rotated, or corrupt) fails closed without
allocating attacker-chosen memory."""
try:
scheme, n, r, p, salt_b64, digest_b64 = stored.split("$")
if scheme != "scrypt":
return False
if (int(n), int(r), int(p)) != (_SCRYPT_N, _SCRYPT_R, _SCRYPT_P):
return False
salt = _unb64(salt_b64)
expected = _unb64(digest_b64)
actual = _scrypt(password.encode(), salt, int(n), int(r), int(p))
return hmac.compare_digest(actual, expected)
except (ValueError, TypeError):
return False
def new_token() -> str:
"""A high-entropy single-use token for email flows (verification, reset)."""
return secrets.token_urlsafe(32)
def token_hash(token: str) -> str:
"""Hash a token for at-rest storage.
Tokens are high-entropy (256 bits), so a plain SHA-256 digest is
sufficient — scrypt would add nothing against offline attacks on a
token with this much entropy.
"""
return hashlib.sha256(token.encode()).hexdigest()
def session_secret() -> bytes:
"""The master HMAC key for session cookies.
Multi mode requires FEEDECHO_SESSION_SECRET explicitly — the gate
fires even when FEEDECHO_AUTH_TOKEN is set, so a carried-over
single-mode env can never mint multi-tenant sessions. Single mode
falls back to FEEDECHO_AUTH_TOKEN (sessions are unused there).
"""
if settings.MULTI and not settings.SESSION_SECRET:
raise RuntimeError(
"FEEDECHO_SESSION_SECRET must be set when FEEDECHO_MODE=multi"
)
key = settings.SESSION_SECRET or (settings.AUTH_TOKEN or "")
return key.encode()
def _session_key() -> bytes:
"""Purpose-derived HMAC key so session tokens share no key material
with any other signed artifact (OAuth state, etc.)."""
return hmac.new(
session_secret(), _SESSION_PURPOSE, hashlib.sha256
).digest()
def sign_session(user_id: int, email: str, epoch: int = 0) -> str:
"""Return a signed session token: base64url(claims).hexsig."""
payload = _b64(
json.dumps(
{
"aud": _SESSION_AUD,
"uid": user_id,
"email": email,
"ep": epoch,
"exp": int(time.time()) + SESSION_TTL_SECONDS,
}
).encode()
).rstrip("=")
sig = hmac.new(_session_key(), payload.encode(), hashlib.sha256).hexdigest()
return f"{payload}.{sig}"
def read_session(token: str | bytes, ignore_expiry: bool = False) -> dict | None:
"""Verify a signed session token; return claims or None.
Returns None for EVERY malformed input — never raises: garbage
payloads, non-ASCII or truncated signatures, bytes tokens, wrong
audience, expired tokens, and coerced claim types all fail closed.
``ignore_expiry`` (logout only): verify the signature and claims but do
NOT reject an expired token. Logout must still bump the user's
session_epoch even when their cookie has expired, otherwise a stored/
fresher session elsewhere survives a logout the user believes happened.
"""
try:
if not isinstance(token, str) or "." not in token:
return None
payload, sig = token.rsplit(".", 1)
sig_bytes = sig.encode("ascii")
expected = hmac.new(
_session_key(), payload.encode(), hashlib.sha256
).hexdigest()
if not hmac.compare_digest(sig_bytes, expected.encode("ascii")):
return None
data = json.loads(_unb64(payload))
if not isinstance(data, dict):
return None
if data.get("aud") != _SESSION_AUD:
return None
if not ignore_expiry and int(data.get("exp", 0)) < time.time():
return None
uid = data.get("uid")
if isinstance(uid, bool) or not isinstance(uid, int):
return None
epoch = data.get("ep", 0)
if isinstance(epoch, bool) or not isinstance(epoch, int):
return None
return {
"user_id": uid,
"email": str(data.get("email", "")),
"epoch": epoch,
}
except (ValueError, TypeError, UnicodeError, json.JSONDecodeError):
return None
# ── Credential encryption (Fernet; required in multi, optional in single) ────
_credential_fernet = None
_credential_fernet_key = None
def _fernet():
"""Lazy-initialised Fernet instance, keyed from settings.CREDENTIAL_KEY.
Returns None when the key is unset or cryptography is not installed,
so callers silently fall back to plaintext.
"""
global _credential_fernet, _credential_fernet_key
key = settings.CREDENTIAL_KEY or ""
if _credential_fernet is not None and key == _credential_fernet_key:
return _credential_fernet
_credential_fernet_key = key
if not key or Fernet is None:
_credential_fernet = None
else:
_credential_fernet = Fernet(key.encode("utf-8"))
return _credential_fernet
def encrypt_secret(value: str) -> str:
"""Encrypt a third-party credential for at-rest storage.
Encrypts whenever FEEDECHO_CREDENTIAL_KEY is set, in either mode: multi
mode requires the key at startup, single mode treats it as optional
hardening for copied backups. An unset key returns the value unchanged
(multi mode without a real database warns at startup through
settings.validate_config; single mode stores plaintext by default).
"""
if not value:
return value
fernet = _fernet()
if fernet is None:
return value
token = fernet.encrypt(value.encode("utf-8"))
return token.decode("ascii")
def decrypt_secret(value: str) -> str:
"""Decrypt a stored credential, tolerating legacy plaintext rows.
Returns the plaintext credential. Values that are not valid Fernet
tokens (rows written before encryption shipped, or a key rotation that
renders the token unreadable) are returned unchanged, so a mixed
plaintext/encrypted DB keeps working. Re-encrypts on the next write.
With no key configured the value passes through untouched — an operator
who REMOVES the key after encrypting rows gets ciphertext handed to the
destination (confusing 401s), so keep the key stable once set.
"""
if not value:
return value
fernet = _fernet()
if fernet is None:
return value
try:
return fernet.decrypt(value.encode("ascii")).decode("utf-8")
except Exception:
if value.startswith("gAAAA"):
# A well-formed Fernet token that failed to decrypt is a wrong key
# or tampered value — NOT a legacy plaintext row. Returning the
# ciphertext as a "credential" would hand garbage to Mastodon/etc.
# and surface as a confusing 401. Fail loud instead.
raise ValueError(
"credential decryption failed — FEEDECHO_CREDENTIAL_KEY is "
"wrong or the stored value was tampered with"
)
# Legacy plaintext (written before encryption shipped): return as-is.
return value
def hash_secret(value: str) -> str:
"""Deterministic SHA-256 digest of a secret, for dedup of an encrypted
credential that also serves as a row's natural key (Discord webhook URL).
Fernet is non-deterministic, so an encrypted column can't be an
ON CONFLICT / WHERE-equality key; the digest of the plaintext can.
"""
return hashlib.sha256(value.encode("utf-8")).hexdigest()