siftlode/backend/app/auth.py

import logging

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 email as email_mod
from app import mailtext
from app import sysconfig
from app.config import settings
from app.db import get_db
from app.models import Invite, User
from app.security import decrypt_optional
from app.utils import canonical_email

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)


DEMO_GOOGLE_SUB = "__demo__"
DEMO_EMAIL = "demo@siftlode.local"
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 = logging.getLogger("siftlode.auth")


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

    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
    email = email.lower()
    return email in settings.allowed_email_set or email in settings.admin_email_set


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_demo or 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_demo.is_(False),
                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 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 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
    # 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))
    # Every canonically-equal invite goes, not just the literal spelling: is_allowed now 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.
    db.execute(delete(Invite).where(Invite.email_canonical == canonical_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, lang, operator_contact())


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


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 require_write_scope(user: User = Depends(require_human)) -> 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. Layers on
    `require_human` because writing to YouTube needs a real, non-demo identity."""
    if not has_write_scope(user):
        raise HTTPException(
            status_code=403,
            detail="Enable YouTube editing in Settings first.",
        )
    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