From the auth /security-review (contained fixes; architectural SA3 proxy-trust + SA4 session-revocation deferred to the user): - SB1: encrypt the OAuth access_token at rest (was plaintext while refresh_token was encrypted) — a ~1h Google bearer credential. New security.decrypt_optional() falls back to a refresh for legacy plaintext tokens; column is unbounded String. E2E-verified: token refreshed → stored as Fernet ciphertext → 319-subscription YouTube sync succeeded. - SA5: password_login is no longer a timing/enumeration oracle — it always runs argon2 (against a decoy hash for unknown/passwordless emails), so response time can't reveal whether an account exists. - SB2: Google login only ADOPTS+activates a pre-existing password account when Google actually attests email_verified (defense-in-depth against takeover); and the email sync won't overwrite with a value another account owns (avoids a 500). - demo_login clears the wallet first (demo can't switch to / act as a real account via the multi-account header); switch_account rejects a suspended target (which would otherwise clear the whole session on the next request). Re-review clean; ruff clean; localdev boots; YouTube auth path E2E-verified.
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
import hashlib
|
|
|
|
from argon2 import PasswordHasher
|
|
from cryptography.fernet import Fernet
|
|
|
|
from app.config import settings
|
|
|
|
# argon2id with the library defaults (sensible memory/time cost). One shared hasher instance.
|
|
_ph = PasswordHasher()
|
|
|
|
|
|
def hash_token(raw: str) -> str:
|
|
"""SHA-256 hex digest for storing single-use tokens (email-link tokens, the setup token)
|
|
by hash rather than in cleartext. Not a password hash — these are high-entropy randoms."""
|
|
return hashlib.sha256(raw.encode()).hexdigest()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _ph.hash(password)
|
|
|
|
|
|
def verify_password(password: str, hashed: str | None) -> bool:
|
|
if not hashed:
|
|
return False
|
|
try:
|
|
return _ph.verify(hashed, password)
|
|
except Exception:
|
|
# Any failure (mismatch, malformed hash) is a non-match — never raise to the caller.
|
|
return False
|
|
|
|
_fernet: Fernet | None = (
|
|
Fernet(settings.token_encryption_key.encode()) if settings.token_encryption_key else None
|
|
)
|
|
|
|
|
|
def _require_fernet() -> Fernet:
|
|
if _fernet is None:
|
|
raise RuntimeError("TOKEN_ENCRYPTION_KEY is not configured")
|
|
return _fernet
|
|
|
|
|
|
def encrypt(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return _require_fernet().encrypt(value.encode()).decode()
|
|
|
|
|
|
def decrypt(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return _require_fernet().decrypt(value.encode()).decode()
|
|
|
|
|
|
def decrypt_optional(value: str | None) -> str | None:
|
|
"""Like decrypt() but returns None instead of raising when the value is absent or not a valid
|
|
ciphertext (e.g. a legacy plaintext token stored before encryption) — so callers fall back to
|
|
a refresh rather than erroring."""
|
|
if not value:
|
|
return None
|
|
try:
|
|
return _require_fernet().decrypt(value.encode()).decode()
|
|
except Exception:
|
|
return None
|