siftlode/backend/app/auth.py

import ipaddress
import logging
import secrets
from datetime import datetime, timedelta, timezone

import httpx
from authlib.integrations.starlette_client import OAuth, OAuthError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse
from starlette.requests import HTTPConnection
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session

from app import email as email_mod
from app import sysconfig
from app.config import settings
from app.db import SessionLocal, get_db
from app.models import AuthToken, DemoWhitelist, Invite, OAuthToken, User
from app.ratelimit import RateLimiter
from app.security import decrypt, encrypt, hash_password, hash_token, verify_password
from app.utils import valid_email

# Email+password auth tuning.
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)
# At most one "your account is suspended" email per address per hour, so a suspended user who
# keeps retrying (or a script with their valid credentials) can't be used to mailbomb them.
_suspend_email_limiter = RateLimiter(max_events=1, window_seconds=3600)


def operator_contact() -> str | None:
    """Who a blocked user should contact — the first configured admin email, if any."""
    return next(iter(sorted(settings.admin_email_set)), None)


def _notify_suspended(background: BackgroundTasks, email: str) -> None:
    """Schedule the one-per-hour 'account suspended' email to a user whose (otherwise valid)
    sign-in was just blocked. Gated by verified credentials at the call site, so this can't be
    triggered by someone who doesn't control the account."""
    if _suspend_email_limiter.allow(email):
        background.add_task(email_mod.send_account_suspended, email, operator_contact())

# The single shared demo account's stable identity. The row is created lazily on first
# demo login (see get_or_create_demo_user); these are just its sentinel keys.
DEMO_GOOGLE_SUB = "__demo__"
DEMO_EMAIL = "demo@siftlode.local"

# Throttle the hidden demo-login endpoint per client IP: enough for a real paste-and-wait,
# stingy enough that the field can't be hammered as an enumeration oracle or for DoS.
_demo_limiter = RateLimiter(max_events=5, window_seconds=60)

# 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
# time, by the onboarding wizard via /auth/upgrade (incremental authorization):
#   - read  -> youtube.readonly  (read subscriptions, build the feed)
#   - write -> youtube           (full: unsubscribe / playlist export)
# Both YouTube scopes are "sensitive"; the wizard explains the Google warning before
# sending the user there, so the extra permission never feels like a surprise.
WRITE_SCOPE = "https://www.googleapis.com/auth/youtube"
READ_SCOPE = f"{WRITE_SCOPE}.readonly"
BASE_SCOPES = "openid email profile"
READ_SCOPES = f"{BASE_SCOPES} {READ_SCOPE}"
WRITE_SCOPES = f"{BASE_SCOPES} {READ_SCOPE} {WRITE_SCOPE}"

log = logging.getLogger("siftlode.auth")

router = APIRouter(prefix="/auth", tags=["auth"])

# The Google OAuth client is built lazily from the CURRENT credentials (DB override via sysconfig,
# else env), so the install wizard / admin can set or change them at runtime without a restart. The
# authlib registry is rebuilt only when the creds change (cheap, first use after a change).
_oauth = OAuth()
_oauth_creds: tuple[str, str] | None = None


def google_oauth(db: Session):
    """The configured authlib Google client, or None when no credentials are set (Google sign-in
    is then disabled — email+password still works)."""
    global _oauth, _oauth_creds
    cid = sysconfig.get_str(db, "google_client_id")
    csec = sysconfig.get_str(db, "google_client_secret")
    if not cid or not csec:
        return None
    if _oauth_creds != (cid, csec):
        _oauth = OAuth()
        _oauth.register(
            name="google",
            client_id=cid,
            client_secret=csec,
            server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
            client_kwargs={"scope": BASE_SCOPES},
        )
        _oauth_creds = (cid, csec)
    return _oauth.google


def google_enabled(db: Session) -> bool:
    """Whether Sign in with Google is available (both client id + secret resolve to non-empty)."""
    return bool(
        sysconfig.get_str(db, "google_client_id")
        and sysconfig.get_str(db, "google_client_secret")
    )


def has_read_scope(user: User) -> bool:
    """Whether the user's stored grant lets us read their YouTube data (build the feed).
    The full write scope is a superset, so it implies read."""
    tok = user.token
    if tok is None or not tok.scopes:
        return False
    granted = tok.scopes.split()
    return READ_SCOPE in granted or WRITE_SCOPE in granted


def has_write_scope(user: User) -> bool:
    """Whether the user's stored grant includes YouTube write (unsubscribe / export)."""
    tok = user.token
    if tok is None or not tok.scopes:
        return False
    return WRITE_SCOPE in tok.scopes.split()


def is_allowed(db: Session, email: str) -> bool:
    """May this email sign in? An approved Invite is the source of truth; the env
    ALLOWED_EMAILS/ADMIN_EMAILS remain a bootstrap fallback (e.g. a fresh DB)."""
    if not email:
        return False
    email = email.lower()
    inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
    if inv is not None:
        return inv.status == "approved"
    return email in settings.allowed_email_set or email in settings.admin_email_set


def is_demo_allowed(db: Session, email: str) -> bool:
    """Whether this email may enter the shared demo account (no Google sign-in)."""
    if not email:
        return False
    return (
        db.execute(
            select(DemoWhitelist).where(DemoWhitelist.email == email.lower())
        ).scalar_one_or_none()
        is not None
    )


def get_or_create_demo_user(db: Session) -> User:
    """The one shared demo user, created on first use. No OAuth token / YouTube scope, so
    has_read_scope/has_write_scope are False and every YouTube-touching path is closed to it."""
    user = db.query(User).filter(User.is_demo.is_(True)).one_or_none()
    if user is None:
        user = User(
            google_sub=DEMO_GOOGLE_SUB,
            email=DEMO_EMAIL,
            display_name="Demo",
            role="user",
            is_demo=True,
            preferences={"language": "en"},
        )
        db.add(user)
        db.commit()
    return user


def _norm_ip(s: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
    """Parse an IP for comparison, unwrapping IPv4-mapped IPv6 (`::ffff:10.0.0.1` → `10.0.0.1`) so a
    plain-IPv4 allowlist still matches a peer that surfaces in mapped form. None for a non-IP string."""
    try:
        ip = ipaddress.ip_address(s.strip())
    except ValueError:
        return None
    return ip.ipv4_mapped if getattr(ip, "ipv4_mapped", None) else ip


def _client_ip(request: Request) -> str:
    """Best-effort client IP for rate limiting. X-Forwarded-For is client-controlled, so we trust it
    ONLY when the request actually arrived from a configured reverse proxy (settings.trusted_proxy_ips
    — the address the proxy connects FROM, e.g. the VPS Caddy's WireGuard peer IP). Our proxy APPENDS
    the client it saw to XFF, so the RIGHTMOST entry is the real client and can't be forged by a
    client pre-seeding the header. Any other peer (someone hitting the app port directly, bypassing the
    proxy) is untrusted — we use its real socket IP, so it can't spoof its rate-limit identity.

    SAFETY: this relies on `request.client` being the TRUE socket peer. uvicorn is started WITHOUT
    `--proxy-headers` (see entrypoint.sh) precisely so it never rewrites `request.client` from XFF —
    do NOT add that flag, or the trust check below would compare against a forgeable value."""
    peer = request.client.host if request.client else "unknown"
    trusted = {_norm_ip(p) for p in settings.trusted_proxy_ip_set} - {None}
    if _norm_ip(peer) in trusted:
        xff = request.headers.get("x-forwarded-for")
        parts = [p.strip() for p in (xff or "").split(",") if p.strip()]
        if parts:
            return parts[-1]
    return peer


def upsert_pending_invite(db: Session, email: str) -> Invite | None:
    """Idempotently record an access request. Returns the Invite if a *new* pending row
    was created (so the caller can notify admins), else None for already-known emails."""
    email = (email or "").lower()
    if not valid_email(email):
        return None
    inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
    if inv is not None:
        # Let a previously-denied person ask again; don't disturb approved/pending rows.
        if inv.status == "denied":
            inv.status = "pending"
            inv.requested_at = datetime.now(timezone.utc)
            inv.decided_at = None
            inv.decided_by = None
            db.commit()
            return inv
        return None
    inv = Invite(email=email, status="pending")
    db.add(inv)
    db.commit()
    return inv


@router.get("/login")
async def login(request: Request, db: Session = Depends(get_db)):
    # access_type=offline ensures a refresh_token on first authorization. We avoid
    # prompt=consent so returning users get a quick sign-in; the stored refresh token
    # is kept when Google doesn't re-issue one (see the callback).
    client = google_oauth(db)
    if client is None:
        # Google sign-in isn't configured on this instance — land back on the login page (the UI
        # hides the button, so this is just a safety net for a stale/direct hit).
        return RedirectResponse(url="/")
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="select_account",
        include_granted_scopes="true",
        scope=BASE_SCOPES,
    )


@router.get("/callback")
async def callback(
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
):
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/")
    try:
        token = await client.authorize_access_token(request)
    except OAuthError as exc:
        raise HTTPException(status_code=400, detail=f"OAuth error: {exc.error}")

    userinfo = token.get("userinfo")
    if not userinfo or not userinfo.get("sub"):
        raise HTTPException(status_code=400, detail="No user info returned by Google")

    sub = userinfo["sub"]
    email = (userinfo.get("email") or "").lower()

    # Linking mode: an already-authenticated account is attaching this Google identity (a
    # password account enabling YouTube or connecting SSO). The marker is set by /auth/link and
    # /auth/upgrade; when present we attach to the current user instead of identifying by sub.
    link_uid = request.session.pop("oauth_link_uid", None)
    if link_uid is not None:
        return _complete_link(request, db, link_uid, sub, userinfo, token)

    if not is_allowed(db, email):
        log.warning("Login denied (not approved): %s", email or "<no email>")
        # A denied Google login doubles as an access request: record it for the admin and
        # land the user on a friendly "request received" screen instead of a raw 403.
        if email:
            inv = upsert_pending_invite(db, email)
            if inv is not None and settings.admin_email_set:
                background.add_task(
                    email_mod.send_admin_new_request,
                    sorted(settings.admin_email_set),
                    email,
                )
            return RedirectResponse(url="/?access=requested")
        return RedirectResponse(url="/?access=denied")

    user = db.query(User).filter(User.google_sub == sub).one_or_none()
    if user is None:
        # No account carries this Google identity yet. If one already exists for this email —
        # e.g. an email+password registration — adopt the Google identity onto it instead of
        # inserting a duplicate (which would collide on the unique email and 500). Google has
        # verified the address, so this safely unifies password + Google sign-in on one account.
        existing = db.query(User).filter(User.email == email).one_or_none()
        if existing is not None:
            if existing.google_sub is not None and existing.google_sub != sub:
                # A different Google identity is already bound to this email — refuse rather than
                # 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
            # previously pending password registration (otherwise current_user would bounce it).
            user.email_verified = True
            user.is_active = True
        else:
            user = User(google_sub=sub, email=email)
            db.add(user)
    # A suspended account can't sign in by any method. Block before establishing the session and
    # tell the (verified) owner why — a successful Google auth proves they control this account.
    if user.is_suspended:
        log.info("Suspended account blocked at Google login: %s", email)
        _notify_suspended(background, user.email)
        return RedirectResponse(url="/?login=suspended")
    # 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
    # be locked out. Everyone else's role is managed in the admin UI and stored in the DB, so a
    # routine login must NOT reset it (that would clobber a UI promotion/demotion). Only seed a
    # default for a brand-new account (its role attribute is still unset before flush).
    if email in settings.admin_email_set:
        user.role = "admin"
    elif not user.role:
        user.role = "user"
    # Google attests the email, so a Google sign-in implicitly verifies it. Setting this also lets
    # a Google user who later adds a password sign in (password-login gates on email_verified).
    if userinfo.get("email_verified", True):
        user.email_verified = True
    # Default UI language from the Google-reported locale on first login (if the user hasn't
    # picked one yet); unsupported locales fall back to English.
    prefs = dict(user.preferences or {})
    if not prefs.get("language"):
        loc = (userinfo.get("locale") or "").split("-")[0].lower()
        prefs["language"] = loc if loc in {"en", "hu", "de"} else "en"
        user.preferences = prefs
    db.flush()

    _store_token(db, user, token)
    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
    _remember_epoch(request, user)
    log.info("Login: %s (id=%s, role=%s)", email, user.id, user.role)
    return RedirectResponse(url="/")


def _store_token(db: Session, user: User, token: dict) -> None:
    """Persist the OAuth grant on the user's single token row, creating it if absent. Google
    only returns a refresh_token on (re)consent, so the previous one is kept otherwise.
    include_granted_scopes=true means `scope` is the union of everything granted, so this
    reflects read/write upgrades correctly."""
    tok = user.token or OAuthToken(user=user)
    if token.get("refresh_token"):
        tok.refresh_token_enc = encrypt(token["refresh_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
    db.add(tok)


def revoke_google_token(token: str | None) -> None:
    """Best-effort revocation of a Google OAuth grant at Google's endpoint. Revoking the refresh
    token tears down the whole grant, so a later sign-in starts from a clean consent (no scopes
    silently carried over via include_granted_scopes). Fail-soft: never blocks the caller — used
    from account deletion, where our own data is already gone regardless of Google's response."""
    if not token:
        return
    try:
        resp = httpx.post(
            "https://oauth2.googleapis.com/revoke",
            data={"token": token},
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10,
        )
        log.info("Google token revoke: status=%s", resp.status_code)
    except Exception:
        log.warning("Google token revoke failed", exc_info=True)


def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
    """Hard-delete a user and ALL their personal data (subscriptions, tags, video states,
    playlists, notifications, tokens — all cascade on the users row), erase their access-request
    row, and revoke their Google grant. Shared by GDPR self-deletion and admin deletion. Does NOT
    touch any browser session — each caller handles its own."""
    email = user.email.lower()
    user_id = user.id
    tok = user.token
    google_token = (decrypt(tok.refresh_token_enc) or tok.access_token) if tok else None
    # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table.
    db.execute(delete(User).where(User.id == user_id))
    db.execute(delete(Invite).where(Invite.email == email))
    db.commit()
    if google_token:
        background.add_task(revoke_google_token, google_token)
    # Confirm the erasure to the (now former) user — GDPR good practice, and covers both the
    # self-service and admin deletion paths since both funnel through here.
    background.add_task(email_mod.send_account_deleted, email, operator_contact())


def _complete_link(
    request: Request, db: Session, link_uid: int, sub: str, userinfo: dict, token: dict
):
    """Attach the just-authorized Google identity to the already-signed-in account that started
    the link/upgrade. No is_allowed gate — they're already an active user. Refuses to hijack a
    Google identity owned by another account, or to silently swap a different linked identity."""
    # Consume the companion marker up front so it never lingers across the early returns below.
    explicit = request.session.pop("oauth_link_explicit", False)
    # The marker must match the live session user (guards a stale/forged marker).
    if link_uid != request.session.get("user_id"):
        return RedirectResponse(url="/?link=error")
    user = db.get(User, link_uid)
    if user is None or user.is_demo:
        return RedirectResponse(url="/?link=error")
    other = (
        db.query(User).filter(User.google_sub == sub, User.id != user.id).one_or_none()
    )
    if other is not None:
        return RedirectResponse(url="/?link=conflict")
    if user.google_sub is not None and user.google_sub != sub:
        return RedirectResponse(url="/?link=mismatch")
    user.google_sub = sub
    if not user.display_name:
        user.display_name = userinfo.get("name")
    if not user.avatar_url:
        user.avatar_url = userinfo.get("picture")
    _store_token(db, user, token)
    db.commit()
    log.info("Google linked: uid=%s scopes=%s", user.id, token.get("scope"))
    # An explicit "Connect Google" lands on Settings with a confirmation; a YouTube upgrade
    # (wizard / Settings access rows) returns to the app so the onboarding flow resumes as before.
    return RedirectResponse(url="/?link=ok" if explicit else "/")


@router.post("/logout")
async def logout(request: Request):
    """Sign the requesting tab's active account out of this browser (removing it from the wallet).
    Per-tab aware: the account is taken from the X-Siftlode-Account header when present, so logging
    out in one tab doesn't disturb another tab running a different account. If the account being
    removed was the session default, promote the most recent remaining account (or clear the
    session when none remain)."""
    active, is_default = resolved_user_id(request)
    remaining = [a for a in (request.session.get("account_ids") or []) if a != active]
    request.session["account_ids"] = remaining
    if is_default:
        if remaining:
            request.session["user_id"] = remaining[-1]
            return JSONResponse({"ok": True, "switched": True})
        request.session.clear()
        return JSONResponse({"ok": True, "switched": False})
    # A non-default (per-tab) account signed out: it's gone from the wallet and the session default
    # is untouched. The tab drops its own override and falls back to the default on reload.
    return JSONResponse({"ok": True, "switched": False})


@router.post("/request-access")
async def request_access(
    payload: dict,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Public: ask for access. Idempotent — repeat requests for the same email don't
    duplicate or re-spam admins (upsert returns None for an already-pending row)."""
    email = (payload.get("email") or "").strip().lower()
    if not valid_email(email):
        raise HTTPException(status_code=400, detail="Enter a valid email address.")
    if is_allowed(db, email):
        return {"status": "approved"}  # already allowed — just sign in
    inv = upsert_pending_invite(db, email)
    if inv is not None and settings.admin_email_set:
        background.add_task(
            email_mod.send_admin_new_request, sorted(settings.admin_email_set), email
        )
    return {"status": "pending"}


@router.post("/demo")
def demo_login(payload: dict, request: Request, db: Session = Depends(get_db)) -> dict:
    """Hidden demo entry: a whitelisted email logs straight into the shared demo account,
    no Google OAuth. The login page fires this quietly (debounced) as the user types/pastes,
    so there is no visible 'demo login' button. Rate-limited per IP; when blocked it answers
    exactly like a non-match (authenticated=false) so the field can't be hammered as an
    enumeration oracle. On a match it sets the session and the client reloads into the app."""
    if not _demo_limiter.allow(_client_ip(request)):
        return {"authenticated": False}
    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
        _remember_epoch(request, demo)
        log.info("Demo login (key=%s, demo_id=%s)", email, demo.id)
        return {"authenticated": True}
    return {"authenticated": False}


def _establish_session(request: Request, user: User) -> None:
    """Sign `user` in for this browser session, mirroring the Google callback's multi-account
    handling so password sign-in joins the same account switcher."""
    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
    _remember_epoch(request, user)


def _app_base() -> str:
    """Public origin of the deployed app (OAUTH_REDIRECT_URL is .../auth/callback)."""
    return settings.app_base


def _issue_token(db: Session, user: User, kind: str, ttl: timedelta) -> str:
    """Create a single-use token of `kind`; only its hash is stored. Returns the raw token
    (goes only into the emailed link)."""
    raw = secrets.token_urlsafe(32)
    db.add(
        AuthToken(
            user_id=user.id,
            kind=kind,
            token_hash=hash_token(raw),
            expires_at=datetime.now(timezone.utc) + ttl,
        )
    )
    db.commit()
    return raw


def _consume_token(db: Session, raw: str | None, kind: str) -> User | None:
    """Validate + burn a token. Returns its user, or None if missing/expired/used/wrong kind."""
    if not raw:
        return None
    row = db.execute(
        select(AuthToken).where(
            AuthToken.token_hash == hash_token(raw), AuthToken.kind == kind
        )
    ).scalar_one_or_none()
    if row is None or row.used_at is not None or row.expires_at < datetime.now(timezone.utc):
        return None
    row.used_at = datetime.now(timezone.utc)
    db.commit()
    return db.get(User, row.user_id)


@router.post("/register")
def register(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Email+password registration. Creates a pending (inactive, unverified) account, sends a
    verification link, and records an access request for admin approval. Two gates before
    sign-in: verified email AND admin approval. Responses are uniform once input is valid, so
    the endpoint can't be used to discover which emails are registered."""
    email = (payload.get("email") or "").strip().lower()
    password = payload.get("password") or ""
    # Input validation is safe to reveal (independent of whether the email exists).
    if not valid_email(email):
        raise HTTPException(status_code=400, detail="Enter a valid email address.")
    if len(password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    if not sysconfig.get_bool(db, "allow_registration"):
        raise HTTPException(status_code=403, detail="Registration is currently closed.")
    if not _register_limiter.allow(_client_ip(request)):
        return {"status": "ok"}  # silently throttle; uniform response

    # Do the existence check + account creation OFF the response path (a background task with its own
    # session). The create path hashes the password + writes several rows + schedules emails; doing it
    # inline would make a NEW email respond measurably slower than an already-registered one (which
    # skips all that) — a timing oracle for enumeration. Off-path, any valid email responds the same.
    background.add_task(_register_account, email, password)
    return {"status": "ok"}


def _register_account(email: str, password: str) -> None:
    """Background worker for /register: create the pending account (+ verification email + admin
    notice) for a genuinely new email; no-op for an already-registered one. Runs after the response
    with its own DB session, so the request's timing never reveals whether the email already exists."""
    with SessionLocal() as db:
        if db.execute(select(User).where(User.email == email)).scalar_one_or_none() is not None:
            return  # already registered — nothing to do (the uniform "ok" was already returned)
        # Without working SMTP we can't deliver a verification link, so email ownership can't gate
        # sign-in. Admin approval stays the real gate (is_active=False); mark the account verified so
        # the flow still completes on a no-SMTP self-host. See email.email_enabled().
        email_ok = email_mod.email_enabled()
        user = User(
            email=email,
            password_hash=hash_password(password),
            is_active=False,
            email_verified=not email_ok,
        )
        db.add(user)
        db.flush()
        upsert_pending_invite(db, email)  # admin-approval gate (commits)
        if email_ok:
            raw = _issue_token(db, user, "verify", VERIFY_TTL)
            # Fragment (#), not a query token: keeps the token out of proxy/access logs + Referer (SB3).
            email_mod.send_verify_email(email, f"{_app_base()}/#verify={raw}")
        if settings.admin_email_set:
            email_mod.send_admin_new_request(sorted(settings.admin_email_set), email)


@router.post("/verify")
def verify_email_confirm(payload: dict, db: Session = Depends(get_db)) -> dict:
    """Confirm an email-verification token the SPA read from the URL fragment (SB3). Uniform 400
    on any invalid/expired/used token."""
    user = _consume_token(db, payload.get("token"), "verify")
    if user is None:
        raise HTTPException(status_code=400, detail="This verification link is invalid or has expired.")
    user.email_verified = True
    db.commit()
    return {"ok": True}


@router.get("/verify")
def verify_email(token: str, db: Session = Depends(get_db)):
    """Legacy query-token verification link (pre-SB3 emails still in flight). New emails use the
    fragment + POST flow above; this bounces to a friendly status for older links until they expire."""
    user = _consume_token(db, token, "verify")
    if user is None:
        return RedirectResponse(url="/?verify=invalid")
    user.email_verified = True
    db.commit()
    return RedirectResponse(url="/?verify=ok")


@router.post("/password-login")
def password_login(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Email+password sign-in. Wrong email/password give a single uniform 401 (no enumeration).
    Once the password is proven correct, the owner gets a specific reason if their account is
    still pending or suspended — that's not an enumeration leak (they already hold the password)."""
    if not _login_limiter.allow(_client_ip(request)):
        raise HTTPException(status_code=429, detail="Too many attempts. Try again shortly.")
    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()
    # 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)
        op = operator_contact()
        detail = "Your account has been suspended."
        if op:
            detail += f" If you have questions, contact the operator at {op}."
        raise HTTPException(status_code=403, detail=detail)
    if not user.email_verified:
        raise HTTPException(status_code=403, detail="Please verify your email first (check your inbox).")
    if not user.is_active:
        raise HTTPException(status_code=403, detail="Your account is awaiting admin approval.")
    _establish_session(request, user)
    log.info("Password login: %s (id=%s, role=%s)", email, user.id, user.role)
    return {"ok": True}


@router.post("/password-reset/request")
def password_reset_request(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
) -> dict:
    """Request a password-reset link. Uniform response + timing regardless of whether the email has a
    password account (the lookup runs off the response path), so it can't probe for registered emails."""
    if not _reset_limiter.allow(_client_ip(request)):
        return {"status": "ok"}
    email = (payload.get("email") or "").strip().lower()
    # Do the account lookup + token issue + email OFF the response path (a background task with its
    # own session), so the endpoint takes the same time for any valid email whether or not it has a
    # password account — no timing-based enumeration. Any valid email schedules the same task.
    if valid_email(email):
        background.add_task(_send_reset_if_eligible, email)
    return {"status": "ok"}


def _send_reset_if_eligible(email: str) -> None:
    """Background worker for password_reset_request: issue a reset token + email the link, but ONLY
    for a real (non-demo) password account. Runs after the response with its own DB session, so the
    request's timing never reveals whether the account exists. Token rides the URL FRAGMENT (#) so it
    can't leak into proxy/access logs or a Referer (SB3)."""
    with SessionLocal() as db:
        user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
        if user is not None and user.password_hash and not user.is_demo:
            raw = _issue_token(db, user, "reset", RESET_TTL)
            email_mod.send_password_reset(email, f"{_app_base()}/#reset={raw}")


@router.post("/password-reset/confirm")
def password_reset_confirm(payload: dict, db: Session = Depends(get_db)) -> dict:
    """Set a new password from a valid reset token, then invalidate any other reset tokens."""
    token = payload.get("token")
    new_password = payload.get("password") or ""
    if len(new_password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    user = _consume_token(db, token, "reset")
    if user is None:
        raise HTTPException(status_code=400, detail="This reset link is invalid or has expired.")
    user.password_hash = hash_password(new_password)
    bump_session_epoch(user)  # SA4: a reset (often a compromise response) kills all existing sessions.
    # Burn any other outstanding reset tokens for this user.
    for row in db.execute(
        select(AuthToken).where(
            AuthToken.user_id == user.id, AuthToken.kind == "reset", AuthToken.used_at.is_(None)
        )
    ).scalars():
        row.used_at = datetime.now(timezone.utc)
    db.commit()
    return {"ok": True}


# A tab can act as a different signed-in account than the session default by sending this header
# (per-tab identity: two tabs in one browser, two accounts). Lowercased — Starlette header lookup
# is case-insensitive, but we compare explicitly below.
ACTIVE_ACCOUNT_HEADER = "x-siftlode-account"


def resolved_user_id(conn: HTTPConnection) -> tuple[int | None, bool]:
    """Which account this connection acts as, and whether that's the session's default account.
    Takes any HTTPConnection — an HTTP Request OR a WebSocket — so the WS push channel shares this
    exact per-tab resolution instead of re-implementing it.

    The signed cookie holds the *wallet* (`account_ids`, every account signed into this browser)
    plus a default `user_id`. A caller may override the default with the X-Siftlode-Account header,
    but ONLY for an account already in the wallet — so a tab can't impersonate an account that never
    authenticated here. Everything else (WebSocket, plain navigations) keeps using the cookie default.
    """
    default_id = conn.session.get("user_id")
    wallet = conn.session.get("account_ids") or []
    # Header for normal XHR; ?account= for contexts that can't set headers (WebSocket, and a
    # plain <a> file download). Both are wallet-gated below, so neither can impersonate an
    # account that never signed into this browser.
    hdr = conn.headers.get(ACTIVE_ACCOUNT_HEADER) or conn.query_params.get("account")
    if hdr:
        try:
            hid = int(hdr)
        except ValueError:
            hid = None
        if hid is not None and hid in wallet:
            return hid, hid == default_id
    return default_id, True


def _remember_epoch(request: Request, user: User) -> None:
    """Record this account's current `session_epoch` in the signed cookie (SA4), so `current_user`
    can later reject the cookie if the account's epoch was bumped (password reset/change, or
    "log out everywhere"). Per-account so one account's revocation doesn't evict the others."""
    epochs = dict(request.session.get("epochs") or {})
    epochs[str(user.id)] = user.session_epoch
    request.session["epochs"] = epochs


def bump_session_epoch(user: User) -> None:
    """Invalidate every EXISTING signed cookie for this account by advancing its epoch (SA4). Cookies
    minted before the bump carry a now-stale epoch and are rejected by `current_user`."""
    user.session_epoch = (user.session_epoch or 0) + 1


def current_user(request: Request, db: Session = Depends(get_db)) -> User:
    user_id, is_default = resolved_user_id(request)
    if not user_id:
        raise HTTPException(status_code=401, detail="Not authenticated")
    user = db.get(User, user_id)
    if user is None or not user.is_active or user.is_suspended:
        # Suspended/deactivated mid-session → drop the session so the block takes effect at once.
        # But only nuke the whole session when the SESSION DEFAULT identity is gone; a stale
        # per-tab header account just 401s that one tab (other tabs' default may still be valid).
        if is_default:
            request.session.clear()
        raise HTTPException(status_code=401, detail="Not authenticated")
    # SA4: reject a cookie whose recorded epoch is behind the account's current one (revoked by a
    # password reset/change or "log out everywhere"). A cookie with no recorded epoch (pre-SA4, or an
    # account signed in before this shipped) is treated as epoch 0 — so the FIRST bump revokes it too.
    epochs = request.session.get("epochs") or {}
    if int(epochs.get(str(user_id), 0)) != (user.session_epoch or 0):
        if is_default:
            request.session.clear()
        raise HTTPException(status_code=401, detail="Not authenticated")
    # Always keep the session default in the switchable list — covers sessions created before
    # multi-session existed (their account_ids was never seeded), so the switcher isn't empty.
    default_id = request.session.get("user_id")
    accounts = request.session.get("account_ids") or []
    if default_id and default_id not in accounts:
        accounts.append(default_id)
        request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    return user


def optional_current_user(
    request: Request, db: Session = Depends(get_db)
) -> User | None:
    """Like `current_user` but returns None instead of raising 401 when there's no valid session.
    Used by the bootstrap /api/me probe so the logged-out landing doesn't log a failed request to
    the browser console (a non-2xx fetch is a console error no matter how the app handles it)."""
    try:
        return current_user(request, db)
    except HTTPException:
        return None


@router.post("/logout-others")
def logout_others(
    request: Request, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """SA4: sign every OTHER session for this account out — bump the epoch so all existing cookies
    are rejected, then re-stamp THIS cookie so the caller stays signed in. For "I left myself logged
    in somewhere" / a suspected session compromise. The demo account (shared) is excluded."""
    if user.is_demo:
        raise HTTPException(status_code=403, detail="Not available in the demo account.")
    # Commit the bump BEFORE re-stamping this cookie, so a failed commit can't strand the current
    # session at a newer epoch than the DB (see set_password).
    bump_session_epoch(user)
    db.commit()
    _remember_epoch(request, user)
    log.info("Logout-others: uid=%s (epoch=%s)", user.id, user.session_epoch)
    return {"ok": True}


def require_human(user: User = Depends(current_user)) -> User:
    """Reject the shared demo account from actions that need a real Google/YouTube identity
    or spend the shared quota. Most YouTube paths are already closed to it implicitly (it has
    no token, so has_read_scope/has_write_scope are False); this makes the boundary explicit
    where there's no scope check to lean on."""
    if user.is_demo:
        raise HTTPException(status_code=403, detail="Not available in the demo account.")
    return user


def admin_user(user: User = Depends(current_user)) -> User:
    """Dependency: require the signed-in user to be an admin. Single source of truth for the
    admin gate — every admin route/action depends on this rather than re-checking the role."""
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    return user


def count_admins(db: Session, *, active_only: bool = False) -> int:
    """How many admins exist (optionally only un-suspended ones). Used by the 'can't remove /
    suspend / delete the last admin' guards so they don't each re-spell the count query."""
    q = select(func.count()).select_from(User).where(User.role == "admin")
    if active_only:
        q = q.where(User.is_suspended.is_(False))
    return db.scalar(q) or 0


@router.get("/link")
async def link_google(
    request: Request,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
):
    """Start linking a Google account to the signed-in (e.g. password) account. Requests only
    the identity scopes — YouTube access is granted separately via /auth/upgrade. The callback
    sees `oauth_link_uid` and attaches the identity instead of creating a new account."""
    if user.is_demo:
        return RedirectResponse(url="/")
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/?link=error")
    request.session["oauth_link_uid"] = user.id
    request.session["oauth_link_explicit"] = True
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="select_account",
        include_granted_scopes="true",
        scope=BASE_SCOPES,
    )


@router.post("/set-password")
def set_password(
    payload: dict,
    request: Request,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Set or change the signed-in account's password. Setting a first password (Google-only
    account) needs no current password — the session already proves identity. Changing an
    existing one requires the current password, so a hijacked session can't silently rotate it."""
    if user.is_demo:
        raise HTTPException(status_code=403, detail="Not available in the demo account.")
    new_password = payload.get("password") or ""
    if len(new_password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    if user.password_hash:
        if not verify_password(payload.get("current_password") or "", user.password_hash):
            raise HTTPException(status_code=403, detail="Current password is incorrect.")
    user.password_hash = hash_password(new_password)
    # SA4: a password change logs out every OTHER session; re-stamp this cookie at the new epoch so
    # the person making the change stays signed in here. Commit BEFORE re-stamping so a failed commit
    # can't leave the cookie at a newer epoch than the DB (which would wrongly 401 THIS session).
    bump_session_epoch(user)
    db.commit()
    _remember_epoch(request, user)
    log.info("Password set/changed: uid=%s", user.id)
    return {"ok": True}


@router.get("/upgrade")
async def upgrade(
    request: Request,
    access: str = "read",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
):
    """Incremental consent for the onboarding wizard. `access=read` grants YouTube
    read-only (enough to build the feed); `access=write` additionally grants management
    (unsubscribe). The shared callback stores whatever Google returns, so `can_read` /
    `can_write` flip on once granted. Anything other than an explicit `write` defaults to
    the least-privileged read scope."""
    # This is the browser-facing redirect entry point, so the shared demo account is bounced
    # straight home (no YouTube link is ever attached to it) rather than shown a raw 403.
    if user.is_demo:
        return RedirectResponse(url="/")
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/")
    # Attach the grant to THIS account in the callback. Without it a password account (no
    # google_sub) would get a brand-new, separate Google user created instead of being linked.
    request.session["oauth_link_uid"] = user.id
    scope = WRITE_SCOPES if access == "write" else READ_SCOPES
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="consent",
        include_granted_scopes="true",
        scope=scope,
    )


@router.get("/me")
async def me(user: User = Depends(current_user)) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "display_name": user.display_name,
        "avatar_url": user.avatar_url,
        "role": user.role,
    }


@router.get("/config")
def auth_config(db: Session = Depends(get_db)) -> dict:
    """Public: which sign-in options this instance offers, so the login page can hide what's off
    (e.g. no Google button when Google OAuth isn't configured)."""
    return {
        "google_enabled": google_enabled(db),
        "allow_registration": sysconfig.get_bool(db, "allow_registration"),
        # Contact shown on the public legal pages (the operator's admin email), if configured.
        "operator_contact": operator_contact(),
    }