siftlode/backend/app/auth.py

import httpx
from fastapi import BackgroundTasks, Depends, HTTPException, Request
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
from starlette.requests import HTTPConnection

from app import debugsession
from app import email as email_mod
from app import logs
from app import mailtext
from app import sysconfig
from app.config import settings
from app.db import get_db
from app.models import Invite, LogEvent, User
from app.security import decrypt_optional
from app.utils import canonical_email
from app.refusal import refused

PASSWORD_MIN_LEN = 10


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)


MAX_SESSION_ACCOUNTS = 6

# OAuth scope vocabulary — the atoms + the request bundles, kept together as core. has_*_scope check
# the atoms against a user's stored grant; the *_SCOPES bundles are what the login/callback endpoints
# request (imported by app/routes/auth.py).
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}"
_YOUTUBE_SCOPES = frozenset({READ_SCOPE, WRITE_SCOPE})

log = logs.get_logger(__name__)


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 env_allowed_canonicals() -> set[str]:
    """The env bootstrap allow-list, in CANONICAL form — the only shape it may be compared in.

    S3 canonicalised the Invite branch of `is_allowed` and left this one matching the literal string,
    which is the 2026-09-03 lockout with a different source of truth: an operator who writes
    `first.last@gmail.com` into ALLOWED_EMAILS is refused when Google answers with the undotted
    primary, and a `+tag`ged entry can never match at all, because a Google account cannot contain a
    `+`. The addresses themselves stay literal wherever we MAIL them (`admin_notify_groups`); only
    the comparison is canonical.
    """
    return env_admin_canonicals() | {canonical_email(e) for e in settings.allowed_email_set}


def env_admin_canonicals() -> set[str]:
    """The env ADMIN_EMAILS bootstrap list, canonical — for the role check only. Kept separate from
    `env_allowed_canonicals` because being listed as an admin also grants access, but being allowed
    does not make anyone an admin."""
    return {canonical_email(e) for e in settings.admin_email_set}


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).

    Matched on the CANONICAL form (utils.canonical_email), not the literal string. Gmail delivers
    every dotted / `+tagged` spelling to one mailbox, so approving one spelling always meant approving
    that mailbox — matching literally only produced the 2026-09-03 lockout, where an approved user's
    Google sign-in became a silent second access request because OAuth returned the undotted primary.

    One canonical form can match SEVERAL invite rows (prod carries approved pairs for one mailbox,
    the oldest from 2026-06-16). An approved row among them grants access: the API cannot flip an
    approved invite to denied (`_decide` only accepts a pending one), so approved-beside-denied means
    the admin approved that mailbox under a different spelling, not that they revoked it. Revoking is
    deleting the invite row or suspending the account — both still work, since a deleted row is no
    longer here to match and suspension is enforced at sign-in.
    """
    if not email:
        return False
    canon = canonical_email(email)
    statuses = (
        db.execute(select(Invite.status).where(Invite.email_canonical == canon)).scalars().all()
    )
    if statuses:
        return "approved" in statuses
    return canon in env_allowed_canonicals()


def can_activate(db: Session, user: User) -> bool:
    """Is this account merely waiting for an admin to finish letting it in?

    True when it is inactive but its address is ALREADY allowed — the shape roadmap UX-25 describes:
    the admin approved the address, the person signed up afterwards, and nothing connects the two
    (the Access-requests queue only offers Approve on a still-pending invite).

    Gated on `is_allowed`, the same function the sign-in paths consult, because activating an address
    that is NOT allowed manufactures an inconsistent account: password sign-in only checks
    `is_active` and would let them in, while their Google sign-in would be turned away and silently
    reopened as a fresh access request. Letting in someone with no approved invite is the whitelist
    box's job — that action says so out loud.

    Lives in core so the admin surface and the /api/me badge count answer from ONE definition; a
    button that appears and a number that counts must never disagree about who is stuck.
    """
    if user.is_suspended or user.is_active:
        return False
    return is_allowed(db, user.email)


def count_pending_activations(db: Session) -> int:
    """How many sign-ups are waiting on an admin click (see can_activate). Drives the admin badge.

    Deliberately NOT "everyone who cannot sign in": an account that has simply not confirmed its
    address yet is waiting on its owner, not on the admin, and badging that would nag about something
    the admin did not cause. It is still listed on the admin screen — with a re-send — because the
    mail may never have arrived.
    """
    # Narrowed in SQL, then decided by can_activate itself — so the number the badge shows and the
    # button the admin sees cannot drift apart by one of them re-deriving the rule.
    rows = (
        db.execute(
            select(User).where(
                User.is_suspended.is_(False),
                User.is_active.is_(False),
            )
        )
        .scalars()
        .all()
    )
    return sum(1 for u in rows if can_activate(db, u))


def find_user_by_email(db: Session, email: str) -> User | None:
    """The account someone means when they type an address at sign-in, password-reset or
    verify-resend. None when there isn't exactly one answer.

    Exact spelling wins. Failing that, the address is resolved to its mailbox
    (utils.canonical_email) — so a password chosen at registration still works when the person types
    another spelling of their own Gmail, and so the three flows that look an account up by typed
    address stop being dead ends. Before this, requesting a reset for `xy@gmail.com` on an account
    registered as `x.y@gmail.com` issued nothing and said nothing: the endpoint answers uniformly by
    design, so the user got silence and no way in.

    SEVERAL canonical candidates return None rather than a guess. That case is real — prod carries two
    accounts for one mailbox — and picking one of them for a password sign-in would hand someone the
    wrong account. The callers all treat None as "no such account", which is the uniform,
    non-enumerating answer they already gave.
    """
    if not email:
        return None
    email = email.strip().lower()
    exact = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
    if exact is not None:
        return exact
    candidates = (
        db.execute(select(User).where(User.email_canonical == canonical_email(email)))
        .scalars()
        .all()
    )
    return candidates[0] if len(candidates) == 1 else None


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:  # noqa: BLE001 — a revoke we cannot make must not fail the sign-out
        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
    # Read the language off the row while it still exists — the confirmation below is queued as a
    # background task that runs after this session (and this user) are gone.
    lang = mailtext.lang_for_user(user)
    tok = user.token
    # Both are stored encrypted — decrypt for the revoke call (the access-token fallback was passing
    # ciphertext, so a token row with no refresh token silently failed to revoke). decrypt_optional so a
    # TOKEN_ENCRYPTION_KEY mismatch can't 500 the purge — a best-effort revoke degrades to "skip". (B-core-2)
    google_token = (
        (decrypt_optional(tok.refresh_token_enc) or decrypt_optional(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))
    # Stored log events name the account by id with no foreign key to cascade from (models.LogEvent),
    # so they go explicitly — they are request history, i.e. personal data. Forgotten FIRST: this very
    # request's access line, a queued row, a later background task would otherwise put rows back
    # under the id after this DELETE (logs.forget_user).
    logs.forget_user(user_id)
    db.execute(delete(LogEvent).where(LogEvent.user_id == user_id))
    # Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the
    # canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose
    # account was just erased — and prod really does carry approved pairs for one mailbox.
    #
    # Unless ANOTHER account still lives on that mailbox — prod carries that too (two user rows for
    # one Gmail mailbox). Then NO invite goes, not even the literal one: is_allowed matches
    # canonically, so every row here is what keeps the survivor allowed, and removing any of them
    # would refuse their next Google sign-in and reopen it as a fresh access request — this epic's
    # founding bug, caused by its own cleanup. `delete_invite` refuses for exactly this reason; the
    # user row is already gone by now, so this query sees only the others.
    canon = canonical_email(email)
    if db.execute(select(User.id).where(User.email_canonical == canon)).first() is None:
        db.execute(delete(Invite).where(Invite.email_canonical == canon))
    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, lang, operator_contact())


ACTIVE_ACCOUNT_HEADER = "x-siftlode-account"


def resolved_user_id(conn: HTTPConnection, *, strict: bool = True) -> tuple[int | None, bool]:
    """Which account this connection acts as, and whether that's the session's default account.
    Takes an HTTPConnection rather than a Request: it reads only the session and the headers.

    The signed cookie holds the *wallet* (`account_ids`, every account signed into this browser)
    plus a default `user_id`. A caller may name its account with the X-Siftlode-Account header (or
    `?account=`), but ONLY an account in the wallet — so a tab can't impersonate an account that
    never authenticated here. A connection that names no account acts as the cookie default.

    A connection naming an account NOT in the wallet acts as nobody (`None`) — typically a tab still
    pinned to an account that was signed out in another tab. It must not fall back to the default:
    that tab would go on acting as an account it never showed, its player checkpoints and pref saves
    landing there and its Sign out signing THAT account out. Only the bootstrap /api/me probe passes
    `strict=False` and gets the default, which is how such a tab learns who it now is (the frontend
    re-pins it and reloads). An account whose session ended while it stayed in the wallet (deleted,
    suspended, revoked) leaves the wallet when it is refused (`_session_user`), so a tab pinned to it
    takes the same way back.
    """
    default_id = conn.session.get("user_id")
    wallet = conn.session.get("account_ids") or []
    # Header for normal XHR; ?account= for a context that can't set headers (a plain <a> file
    # download). Both are wallet-gated below.
    hdr = conn.headers.get(ACTIVE_ACCOUNT_HEADER) or conn.query_params.get("account")
    if hdr:
        try:
            hid = int(hdr)
        except ValueError:
            # Under `strict` this becomes a 401 and the tab authenticates as nobody, which reads
            # in the browser exactly like a session that expired. The value itself stays out —
            # it came from a request, and `wallet` says which accounts were actually on offer.
            log.debug(
                "the active-account header could not be read as an id",
                extra=logs.event("auth.account_header.invalid", accounts=len(wallet)),
            )
            hid = None
        if hid is not None and hid in wallet:
            return hid, hid == default_id
        if strict:
            return None, False
    return default_id, True


def drop_from_wallet(session: dict, user_id: int) -> None:
    """Take one account out of this browser's wallet (`account_ids`); the session default is untouched."""
    session["account_ids"] = [a for a in (session.get("account_ids") or []) if a != user_id]


def _end_account_session(request: Request, user_id: int, is_default: bool) -> None:
    """An account whose session here is no longer valid. Only nuke the whole session when it is the
    SESSION DEFAULT; a per-tab account just leaves the wallet (other tabs' default may still be valid).
    Leaving is what lets a tab pinned to it recover: its next /api/me names an account no longer signed
    in here and gets the default. Left in, it answered "signed out" — the landing page, with the default
    still signed in, and no way off it but closing the tab."""
    if is_default:
        request.session.clear()
    else:
        drop_from_wallet(request.session, user_id)


def current_user(request: Request, db: Session = Depends(get_db)) -> User:
    return _session_user(request, db, *resolved_user_id(request))


def _session_user(request: Request, db: Session, user_id: int | None, is_default: bool) -> User:
    if not user_id:
        raise refused(401, "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 → end it so the block takes effect at once.
        _end_account_session(request, user_id, is_default)
        raise refused(401, "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):
        _end_account_session(request, user_id, is_default)
        raise refused(401, "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:]
    # Attribute this request's log lines to the account (E22). Here, not in the request middleware:
    # this is the one place a session becomes an account, and the middleware runs outside both the
    # session middleware and this task's context. `request.state` is shared across middleware layers
    # (it lives on the ASGI scope), so the access-log line can read it back. The user's consented
    # diagnostics level rides along: their requests are logged — and stored — in that detail (S5).
    # So does a debug session the request names (E16): only the caller's own open one counts, and a
    # header naming anything else is ignored rather than refused — a stale tab must keep working.
    session = debugsession.from_header(db, user, request.headers.get(debugsession.HEADER))
    level, until = debugsession.level_for(user, session)
    logs.set_user(user.id, level, until, session.id if session else None)
    request.state.user_id = user.id
    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).
    The one lenient resolution (`strict=False`): a tab naming an account no longer signed in here gets
    the default account, which is how it learns who it now is — see `resolved_user_id`. That includes an
    account refused right here, which `_session_user` has just taken out of the wallet."""
    user_id, is_default = resolved_user_id(request, strict=False)
    try:
        return _session_user(request, db, user_id, is_default)
    # silent-ok: this is the anonymous case as much as the refused one — every landing page hit
    # comes through here. The second attempt below is the handling, and it reports.
    except HTTPException:
        if is_default:
            return None
    try:
        return _session_user(request, db, *resolved_user_id(request, strict=False))
    except HTTPException:
        # Both the pinned account AND the default one were refused, so the tab is signed out
        # without being told why — and it got here from a session that named a real account.
        log.info(
            "a session was resolved to nobody: neither the named nor the default account held",
            extra=logs.event("auth.session.probe_rejected", user_id=user_id),
        )
        return None


def require_write_scope(user: User = Depends(current_user)) -> User:
    """Dependency: require the YouTube write scope (subscribe/unsubscribe, playlist export/edit).
    Single source of truth for the write gate — replaces the inline `if not has_write_scope(user)`
    checks that had drifted into three different 403 messages across channels/playlists."""
    if not has_write_scope(user):
        raise refused(403, "youtube_editing_off")
    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 refused(403, "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