siftlode/backend/app/auth.py

import logging
import re
from datetime import datetime, timezone

from authlib.integrations.starlette_client import OAuth, OAuthError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse
from sqlalchemy import select
from sqlalchemy.orm import Session

from app import email as email_mod
from app.config import settings
from app.db import get_db
from app.models import DemoWhitelist, Invite, OAuthToken, User
from app.ratelimit import RateLimiter
from app.security import encrypt

_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

# 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("subfeed.auth")

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

oauth = OAuth()
oauth.register(
    name="google",
    client_id=settings.google_client_id,
    client_secret=settings.google_client_secret,
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    client_kwargs={"scope": BASE_SCOPES},
)


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 _client_ip(request: Request) -> str:
    """Best-effort client IP for rate limiting. Behind our reverse proxy (Caddy/NPM) the
    real client is the first X-Forwarded-For hop; fall back to the socket peer otherwise."""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else "unknown"


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 _EMAIL_RE.match(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):
    # 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).
    return await oauth.google.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),
):
    try:
        token = await oauth.google.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")

    email = (userinfo.get("email") or "").lower()
    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 == userinfo["sub"]).one_or_none()
    if user is None:
        user = User(google_sub=userinfo["sub"], email=email)
        db.add(user)
    user.email = email
    user.display_name = userinfo.get("name")
    user.avatar_url = userinfo.get("picture")
    user.role = "admin" if email in settings.admin_email_set else "user"
    # 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()

    tok = user.token or OAuthToken(user=user)
    # Google only returns a refresh_token on (re)consent; keep the previous one otherwise.
    if token.get("refresh_token"):
        tok.refresh_token_enc = encrypt(token["refresh_token"])
    tok.access_token = token.get("access_token")
    expires_at = token.get("expires_at")
    tok.expiry = (
        datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None
    )
    # include_granted_scopes=true means Google returns the union of all scopes the user
    # has ever granted this app, so this correctly reflects read/write upgrades too.
    tok.scopes = token.get("scope") or BASE_SCOPES
    db.add(tok)
    db.commit()

    # Multi-session: remember every account that has authenticated in this browser so the
    # user can switch between them without going through Google again. Only accounts that
    # actually completed OAuth here land in this list (it's the proof they may be switched to).
    accounts = [a for a in (request.session.get("account_ids") or []) if a != user.id]
    accounts.append(user.id)
    request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    request.session["user_id"] = user.id
    log.info("Login: %s (id=%s, role=%s)", email, user.id, user.role)
    return RedirectResponse(url="/")


@router.post("/logout")
async def logout(request: Request):
    """Sign out the active account. If other accounts have authenticated in this browser,
    switch to the most recent one instead of fully logging out."""
    active = request.session.get("user_id")
    remaining = [a for a in (request.session.get("account_ids") or []) if a != active]
    if remaining:
        request.session["account_ids"] = remaining
        request.session["user_id"] = remaining[-1]
        return JSONResponse({"ok": True, "switched": True})
    request.session.clear()
    return JSONResponse({"ok": True, "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 _EMAIL_RE.match(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 _EMAIL_RE.match(email) and is_demo_allowed(db, email):
        demo = get_or_create_demo_user(db)
        request.session["user_id"] = demo.id
        log.info("Demo login (key=%s, demo_id=%s)", email, demo.id)
        return {"authenticated": True}
    return {"authenticated": False}


def current_user(request: Request, db: Session = Depends(get_db)) -> User:
    user_id = request.session.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Not authenticated")
    user = db.get(User, user_id)
    if user is None:
        request.session.clear()
        raise HTTPException(status_code=401, detail="Not authenticated")
    # Always keep the active account in the switchable list — covers sessions created before
    # multi-session existed (their account_ids was never seeded), so the switcher isn't empty.
    accounts = request.session.get("account_ids") or []
    if user_id not in accounts:
        accounts.append(user_id)
        request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    return user


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


@router.get("/upgrade")
async def upgrade(
    request: Request, access: str = "read", user: User = Depends(require_human)
):
    """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."""
    scope = WRITE_SCOPES if access == "write" else READ_SCOPES
    return await oauth.google.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,
    }