fix(auth): security hardening — token encryption, timing, adoption + isolation

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.
This commit is contained in:
npeter83 2026-07-11 21:26:39 +02:00
parent f5fac09833
commit e92751dbce
4 changed files with 51 additions and 8 deletions

View file

@ -22,6 +22,10 @@ from app.utils import valid_email
PASSWORD_MIN_LEN = 10
VERIFY_TTL = timedelta(hours=24)
RESET_TTL = timedelta(hours=1)
# A throwaway argon2 hash to verify against when the email is unknown / has no password, so a
# login attempt pays the same KDF cost either way and its timing can't reveal whether the account
# exists (anti-enumeration). Computed once at startup; it never matches a real password.
_DECOY_PASSWORD_HASH = hash_password(secrets.token_urlsafe(16))
_register_limiter = RateLimiter(max_events=5, window_seconds=300)
_login_limiter = RateLimiter(max_events=10, window_seconds=300)
_reset_limiter = RateLimiter(max_events=5, window_seconds=300)
@ -276,6 +280,12 @@ async def callback(
# hijack it. (Near-impossible with real Google accounts: email↔sub is stable.)
log.warning("Google login email collision (different sub): %s", email)
return RedirectResponse(url="/?access=denied")
if not userinfo.get("email_verified", False):
# Adopting (and activating) a pre-existing password account requires Google to
# actually attest the address — otherwise an unverified Google email that merely
# collides with a pending registration could seize that account. Defense-in-depth.
log.warning("Google login: unverified email, refusing to adopt account: %s", email)
return RedirectResponse(url="/?access=denied")
user = existing
user.google_sub = sub
# Google verified the email and is_allowed granted access, so finish activating a
@ -291,7 +301,13 @@ async def callback(
log.info("Suspended account blocked at Google login: %s", email)
_notify_suspended(background, user.email)
return RedirectResponse(url="/?login=suspended")
user.email = email
# Keep the email in sync with Google, but never overwrite it with one another account already
# owns — that would hit the unique constraint and 500, wedging this user's sign-in (rare: a
# Google-side email change to a value that collides with a different local account).
if user.email != email:
clash = db.query(User).filter(User.email == email, User.id != user.id).one_or_none()
if clash is None:
user.email = email
user.display_name = userinfo.get("name")
user.avatar_url = userinfo.get("picture")
# ADMIN_EMAILS (env) is the bootstrap admin list and always wins, so the configured admin can't
@ -337,7 +353,9 @@ def _store_token(db: Session, user: User, token: dict) -> None:
tok = user.token or OAuthToken(user=user)
if token.get("refresh_token"):
tok.refresh_token_enc = encrypt(token["refresh_token"])
tok.access_token = token.get("access_token")
# Encrypt the access token at rest too (it's a live ~1h bearer credential), matching the
# refresh token — a DB/backup/log leak otherwise hands out working Google API tokens.
tok.access_token = encrypt(token.get("access_token"))
expires_at = token.get("expires_at")
tok.expiry = datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None
tok.scopes = token.get("scope") or BASE_SCOPES
@ -471,6 +489,9 @@ def demo_login(payload: dict, request: Request, db: Session = Depends(get_db)) -
email = (payload.get("email") or "").strip().lower()
if valid_email(email) and is_demo_allowed(db, email):
demo = get_or_create_demo_user(db)
# Isolate the demo: drop any real-account wallet on this tab first, so the demo session
# can't switch to / act as a previously signed-in account via the multi-account header.
request.session.clear()
request.session["user_id"] = demo.id
log.info("Demo login (key=%s, demo_id=%s)", email, demo.id)
return {"authenticated": True}
@ -603,7 +624,11 @@ def password_login(
email = (payload.get("email") or "").strip().lower()
password = payload.get("password") or ""
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
if user is None or user.is_demo or not verify_password(password, user.password_hash):
# Always run the KDF (against a decoy hash for an unknown / passwordless / demo account) so the
# response time is the same whether or not the email exists — no timing-based enumeration.
candidate_hash = user.password_hash if (user and not user.is_demo) else None
password_ok = verify_password(password, candidate_hash or _DECOY_PASSWORD_HASH)
if user is None or user.is_demo or not password_ok:
raise HTTPException(status_code=401, detail="Invalid email or password.")
if user.is_suspended:
_notify_suspended(background, user.email)