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.userscope import owned_by
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)


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

    Canonicalising only the Invite branch of `is_allowed` 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(owned_by(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())


def session_user_id(conn: HTTPConnection) -> int | None:
    """The account this browser session signed in as, read from the signed cookie alone (no DB).
    One account per session: signing in as another replaces it (`routes.auth._establish_session`)."""
    return conn.session.get("user_id")


def stamp_session(request: Request, user: User) -> None:
    """Record the account's current `session_epoch` in the signed cookie, so `current_user` can later
    reject the cookie once the epoch moves (password reset/change, or "log out everywhere")."""
    request.session["epoch"] = user.session_epoch or 0


# The account a tab was built for, sent with every request it makes (`identityHeaders`, the
# frontend). The cookie alone says who is asking, and one browser holds one account: a sign-in as
# someone else in another tab replaces it under every open tab. A tab still showing the previous
# account would otherwise act as the new one — its player checkpoints, pref saves, Sign out and
# Delete account landing there — until it noticed. It never selects an account; a mismatch is
# refused, and the tab reloads as whoever is signed in now.
EXPECTED_ACCOUNT_HEADER = "x-siftlode-expect-account"


def built_for_another_account(user_id: int, expected: str | None) -> bool:
    """Whether a tab naming `expected` is showing another account than the one signed in. No name is
    no claim (a caller from before this check, a script): only a mismatch is refused, and the
    session stays — it is the other tab's sign-in, and it holds. The one comparison for every way a
    tab names its account: this header on a request, `expect` on a navigation."""
    if expected is None or expected == str(user_id):
        return False
    log.info(
        "a request from a tab built for another account was refused",
        extra=logs.event("auth.session.account_changed", user_id=user_id),
    )
    return True


def current_user(request: Request, db: Session = Depends(get_db)) -> User:
    user = _session_user(request, db, session_user_id(request))
    if built_for_another_account(user.id, request.headers.get(EXPECTED_ACCOUNT_HEADER)):
        raise refused(401, "account_changed")
    return user


def _session_user(request: Request, db: Session, user_id: int | None) -> 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.
        request.session.clear()
        raise refused(401, "not_authenticated")
    # A cookie whose epoch is behind the account's (revoked by a password reset/change or "log out
    # everywhere") is refused — and so is one with no epoch at all, which only a build before the
    # one-account session wrote: its holder signs in once more.
    epoch = request.session.get("epoch")
    if epoch is None or int(epoch) != (user.session_epoch or 0):
        request.session.clear()
        raise refused(401, "not_authenticated")
    # Attribute this request's log lines to the account. 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.
    # So does a debug session the request names: 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)."""
    user_id = session_user_id(request)
    if user_id is None:
        return None
    try:
        return _session_user(request, db, user_id)
    except HTTPException:
        # A session that named an account, refused: the browser is signed out without being told
        # why, so say it here.
        log.info(
            "a session was refused: its account no longer holds",
            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