feat(auth): N3 — server-side multi-session account switch

Track every account that completes OAuth in a browser session (session 'account_ids',
~14-day cookie), so the user can switch between them without another Google round-trip.
New GET /api/me/accounts (switchable accounts, active flagged) and POST /api/me/switch
(guarded: target must be in the session list — proof it signed in here — and re-checked
against is_allowed in case the invite was revoked). Logout now signs out the active
account and falls back to the most recent remaining one, else clears the session. UI: the
account popover lists the other signed-in accounts (click to switch, reloads) plus an
'Add another account' action (-> /auth/login, which uses prompt=select_account).
Trilingual. Third step of Epic N.
This commit is contained in:
npeter83 2026-06-16 02:05:38 +02:00
parent 8c076f7a75
commit 583e003c23
7 changed files with 142 additions and 8 deletions

View file

@ -16,6 +16,9 @@ from app.security import encrypt
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
# How many recently-used accounts to keep switchable in one browser session.
MAX_SESSION_ACCOUNTS = 6
# Base sign-in requests only the non-sensitive OpenID scopes (profile/email). These do
# NOT trigger Google's "unverified app" warning and do NOT expire after 7 days, so a new
# user gets a clean, familiar consent. YouTube access is granted later, one step at a
@ -175,6 +178,12 @@ async def callback(
db.add(tok)
db.commit()
# Multi-session: remember every account that has authenticated in this browser so the
# user can switch between them without going through Google again. Only accounts that
# actually completed OAuth here land in this list (it's the proof they may be switched to).
accounts = [a for a in (request.session.get("account_ids") or []) if a != user.id]
accounts.append(user.id)
request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
request.session["user_id"] = user.id
log.info("Login: %s (id=%s, role=%s)", email, user.id, user.role)
return RedirectResponse(url="/")
@ -182,8 +191,16 @@ async def callback(
@router.post("/logout")
async def logout(request: Request):
"""Sign out the active account. If other accounts have authenticated in this browser,
switch to the most recent one instead of fully logging out."""
active = request.session.get("user_id")
remaining = [a for a in (request.session.get("account_ids") or []) if a != active]
if remaining:
request.session["account_ids"] = remaining
request.session["user_id"] = remaining[-1]
return JSONResponse({"ok": True, "switched": True})
request.session.clear()
return JSONResponse({"ok": True})
return JSONResponse({"ok": True, "switched": False})
@router.post("/request-access")

View file

@ -1,14 +1,62 @@
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.auth import current_user, has_read_scope, has_write_scope
from app.auth import current_user, has_read_scope, has_write_scope, is_allowed
from app.db import get_db
from app.models import Invite, User
router = APIRouter(prefix="/api/me", tags=["me"])
@router.get("/accounts")
def my_accounts(
request: Request,
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> list[dict]:
"""The accounts that have authenticated in this browser session (switchable without a
new Google sign-in). The active one is flagged."""
ids = request.session.get("account_ids") or [user.id]
rows = {u.id: u for u in db.query(User).filter(User.id.in_(ids)).all()}
out = []
for uid in ids: # preserve recency order from the session
u = rows.get(uid)
if u is not None:
out.append(
{
"id": u.id,
"email": u.email,
"display_name": u.display_name,
"avatar_url": u.avatar_url,
"active": u.id == user.id,
}
)
return out
@router.post("/switch")
def switch_account(
payload: dict,
request: Request,
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> dict:
"""Switch the active account to another one already authenticated in this browser. No
Google round-trip but only accounts present in the session list (proof they signed in
here) are allowed, and access is re-checked in case the invite was revoked since."""
target = payload.get("user_id")
if target not in (request.session.get("account_ids") or []):
raise HTTPException(status_code=403, detail="Not an account you've signed into here.")
u = db.get(User, target)
if u is None:
raise HTTPException(status_code=404, detail="That account no longer exists.")
if not is_allowed(db, u.email):
raise HTTPException(status_code=403, detail="That account no longer has access.")
request.session["user_id"] = target
return {"ok": True, "user_id": target}
@router.get("")
def get_me(
user: User = Depends(current_user), db: Session = Depends(get_db)