pdf-quiz-generator/backend/app/routers/auth.py

import logging

import httpx
import secrets
from urllib.parse import quote, urlencode
from datetime import datetime, timedelta

from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, status, Request
from sqlalchemy.orm import Session

from app.services import refresh_tokens, site_settings
from app.database import get_db
from app.models.user import User
from app.models.email_verification import EmailVerification
from app.models.password_reset import PasswordReset
from app.schemas.auth import (
    UserResponse, Token, LoginRequest, LogoutRequest, RefreshRequest,
    UserUpdateMe, SsoExchangeRequest,
)
from app.services import email_service
from app.utils.auth import (
    check_rate_limit, verify_password, create_access_token, get_current_user,
)

logger = logging.getLogger(__name__)
router = APIRouter()


def _login_key(client_ip: str) -> str:
    return f"login_attempts:{client_ip}"


def _check_login_rate_limit(client_ip: str):
    """How many times an address may guess, before it has to wait.

    Counted per address and cleared by a success, because what this is for is
    guessing — and somebody who signs in correctly has not guessed. Counting
    successes too would lock out a hospital: one public address, a ward full of
    people, eleven of whom happened to open the app this afternoon.
    """
    try:
        import redis as redis_lib
        from app.config import settings
        r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=1)
        key = _login_key(client_ip)
        count = r.incr(key)
        if count == 1:
            r.expire(key, settings.LOGIN_WINDOW_MINUTES * 60)
        if count > settings.LOGIN_MAX_ATTEMPTS:
            raise HTTPException(
                status_code=429,
                detail=f"Too many login attempts. Try again in {settings.LOGIN_WINDOW_MINUTES} minutes.")
    except HTTPException:
        raise
    except Exception as e:
        import logging; logging.getLogger(__name__).warning(f"Redis rate limit unavailable (failing open): {e}")


def _clear_login_rate_limit(client_ip: str):
    """A correct password is the end of the matter."""
    try:
        import redis as redis_lib
        from app.config import settings
        redis_lib.from_url(settings.REDIS_URL, decode_responses=True,
                           socket_connect_timeout=1).delete(_login_key(client_ip))
    except Exception:
        pass


# Rate limit: max 3 reset requests per email per hour
RESET_LIMIT = 3
RESET_WINDOW_HOURS = 1


def _check_reset_rate_limit(db: Session, email: str):
    """Refuse a fourth request in an hour — for any address, not only a real one.

    Returning early for an unknown address made this the oracle the careful
    wording below exists to avoid: ask four times and a registered address
    answers 429 while an unknown one answers 200 for ever. The limit is now
    counted against the address as typed, so both answer the same.
    """
    email_normalized = email.lower().strip()
    user = db.query(User).filter(User.email == email_normalized).first()
    window_start = datetime.utcnow() - timedelta(hours=RESET_WINDOW_HOURS)
    if user is None:
        # No rows to count for an address with no account, so the attempts are
        # counted in Redis instead — keyed by a fingerprint, because a list of
        # addresses somebody tried is itself worth not keeping.
        import hashlib

        from app.utils.auth import check_rate_limit
        check_rate_limit(
            key=f"pwreset:{hashlib.sha256(email_normalized.encode()).hexdigest()}",
            max_calls=RESET_LIMIT, window_seconds=RESET_WINDOW_HOURS * 3600,
            detail="Too many reset requests. Please wait before trying again.")
        return
    count = db.query(PasswordReset).filter(
        PasswordReset.user_id == user.id,
        PasswordReset.created_at >= window_start,
    ).count()
    if count >= RESET_LIMIT:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"Too many reset requests. Please wait before trying again.",
        )


@router.post("/login", response_model=Token)
async def login(login_data: LoginRequest, db: Session = Depends(get_db), request: Request = None):
    # Block password login if SSO-only mode
    sso_settings = _get_sso_settings()
    if sso_settings["sso_only"]:
        raise HTTPException(status_code=403, detail="Password login is disabled. Please use SSO.")

    client_ip = (request.client.host if request and request.client else "unknown")
    if request:
        _check_login_rate_limit(client_ip)

    email_normalized = login_data.email.lower().strip()
    user = db.query(User).filter(User.email == email_normalized).first()
    # An account with no password is not an account with the wrong password,
    # but it is told the same thing: which accounts have one is not a question
    # this endpoint answers.
    if not user or not user.hashed_password or not verify_password(
            login_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid email or password")

    # Check email verification — skip for users without any verification record (legacy/seeded)
    verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
    if verification and verification.verified_at is None:
        raise HTTPException(
            status_code=403,
            detail="Email not verified. Please check your inbox and verify your email before logging in.",
        )

    _clear_login_rate_limit(client_ip)
    return _signed_in(db, user, login_data, request)


def _signed_in(db: Session, user: User, login_data: LoginRequest, request: Request | None) -> Token:
    """What a successful sign-in hands back.

    A browser gets what it always got. A client that says it wants a refresh
    token also gets one, because it has nowhere safe to keep a password and no
    person sitting in front of it to ask again.
    """
    # Imported here, as everywhere else in this file: a module-level `settings`
    # plus these function-level ones would make the name local to each of them
    # and blow up on first use.
    from app.config import settings
    token = Token(
        access_token=create_access_token(data={"sub": user.email}),
        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
    )
    if login_data.refresh:
        token.refresh_token = refresh_tokens.issue(
            db, user, label=login_data.device,
            ip=request.client.host if request and request.client else None)
    return token


@router.post("/refresh", response_model=Token)
def refresh(data: RefreshRequest, db: Session = Depends(get_db), request: Request = None):
    """Trade a refresh token for a new pair.

    The old one is spent by this call. Presenting a spent token ends the whole
    session it belongs to: either it was copied or a client is replaying, and
    from here those look the same.
    """
    from app.config import settings
    client_ip = request.client.host if request and request.client else "unknown"
    # Flood protection, not a security control: the security here is that a
    # refresh token is 256 unguessable bits and spending one twice ends the
    # session. Generous, because an address can be a whole hospital behind one
    # NAT and every app launch refreshes.
    check_rate_limit(f"refresh:{client_ip}", settings.REFRESH_MAX_PER_HOUR, 3600,
                     "Too many refresh attempts. Try again later.")
    spent = refresh_tokens.spend(db, data.refresh_token, ip=client_ip)
    if spent is None:
        raise HTTPException(401, "That sign-in has expired. Sign in again.")
    user, rotated = spent
    verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
    if verification and verification.verified_at is None:
        raise HTTPException(403, "Email not verified.")
    return Token(
        access_token=create_access_token(data={"sub": user.email}),
        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
        refresh_token=rotated,
    )


@router.post("/logout", status_code=204)
def logout(data: LogoutRequest, db: Session = Depends(get_db),
           current_user: User = Depends(get_current_user)):
    """End this session, or every session.

    An access token already issued is a signature and cannot be recalled; it
    dies of old age within the day. What this ends is the ability to get
    another one, which is what a lost phone actually needs.
    """
    if data.everywhere:
        refresh_tokens.revoke_all(db, current_user)
    elif data.refresh_token:
        refresh_tokens.revoke_one(db, data.refresh_token)


@router.get("/sessions")
def list_sessions(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
    """Where this account is signed in, so a person can see and end them."""
    return refresh_tokens.sessions(db, current_user)


@router.delete("/sessions/{family}", status_code=204)
def end_session(family: str, db: Session = Depends(get_db),
                current_user: User = Depends(get_current_user)):
    refresh_tokens.revoke_family(db, current_user, family)


# ── No password lifecycle ─────────────────────────────────────────────
#
# Verifying an address, resending that mail, forgetting a password and
# resetting one all lived here. Accounts are at the identity provider now: it
# owns the address, the passkey, the second factor and any password there is,
# so a reset link minted here would set a credential nothing checks — and an
# account with no password cannot forget one.
#
# /login stays, refused while `sso_only` is set. It is the way back in if the
# provider is ever unreachable, together with the DEFAULT_ADMIN_EMAIL seed at
# startup, and taking it out would leave no door at all on a bad day.


@router.get("/me/settings")
def get_user_settings(
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """Get user settings. Most settings live in Redis (Nextcloud config etc.);
    reminders_disabled is canonical in Postgres and overrides Redis."""
    data = {}
    try:
        import redis as redis_lib, json
        from app.config import settings as cfg
        r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True)
        raw = r.get(f"user_settings:{current_user.id}")
        if raw:
            data = json.loads(raw)
    except Exception as e:
        import logging; logging.getLogger(__name__).warning(f"Failed to load user settings: {e}")

    # Canonical preferences come from the DB. Redis holds the rest, and a cache
    # is not where a choice somebody made once should live.
    data["reminders_disabled"] = bool(current_user.reminders_disabled)
    data["tts_voice"] = current_user.tts_voice
    return data


@router.put("/me/settings")
def save_user_settings(
    settings_data: dict,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """Save user settings. reminders_disabled is persisted to Postgres;
    other keys go to Redis."""
    # Persist opt-out preference to DB (canonical source for the scheduler)
    changed = False
    if "reminders_disabled" in settings_data:
        current_user.reminders_disabled = bool(settings_data.get("reminders_disabled"))
        changed = True
    if "tts_voice" in settings_data:
        voice = (settings_data.get("tts_voice") or "").strip()
        current_user.tts_voice = voice or None
        changed = True
    if changed:
        db.add(current_user)
        db.commit()

    # Keep the full blob in Redis so other fields (Nextcloud config etc.) persist
    try:
        import redis as redis_lib, json
        from app.config import settings as cfg
        r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True)
        r.set(f"user_settings:{current_user.id}", json.dumps(settings_data))
    except Exception as e:
        import logging; logging.getLogger(__name__).warning(f"Failed to save user settings to Redis: {e}")
    return {"saved": True}


@router.get("/me", response_model=UserResponse)
def get_me(current_user: User = Depends(get_current_user)):
    return UserResponse.of(current_user)


@router.put("/me")
def update_me(data: UserUpdateMe, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
    """Your name. Your sign-in belongs to the provider.

    This used to set a password as well. There is nowhere for one to be used
    that the provider does not own, so writing one here would store a
    credential nothing checks — and the settings page that offered it invited
    somebody to secure their account by a route that secures nothing.
    """
    if data.name:
        current_user.name = data.name

    db.commit()
    db.refresh(current_user)
    return {
        "id": current_user.id, "email": current_user.email,
        "name": current_user.name, "role": current_user.role,
    }


# ── SSO / OIDC ─────────────────────────────────────────────────────

def _get_sso_settings():
    """Return SSO admin settings from Redis."""
    try:
        import redis as redis_lib
        from app.config import settings as cfg
        r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True, socket_connect_timeout=1)
        return {
            "sso_only": r.get("settings:sso_only") == "true",
        }
    except Exception:
        return {"sso_only": False}


#: How long a one-time code is worth anything. The browser is mid-redirect and
#: spends it immediately; a minute is generous for that and short enough that a
#: code seen in a log is already dead.
SSO_CODE_TTL = 60


def _sso_store():
    import redis as redis_lib
    from app.config import settings as cfg
    return redis_lib.from_url(cfg.REDIS_URL, decode_responses=True, socket_connect_timeout=1)


def _stash_sso_token(token: str) -> str | None:
    """Park a freshly minted token behind a random code. None if Redis is away."""
    code = secrets.token_urlsafe(32)
    try:
        _sso_store().setex(f"sso:exchange:{code}", SSO_CODE_TTL, token)
        return code
    except Exception:
        return None


@router.post("/sso/exchange", response_model=Token)
def sso_exchange(data: SsoExchangeRequest):
    """Trade a one-time code for the token it stands for.

    Once. The key is deleted as it is read, so a code replayed from a log, a
    history entry or a Referer header buys nothing.
    """
    from app.config import settings as cfg

    key = f"sso:exchange:{data.code}"
    try:
        store = _sso_store()
        # GETDEL where the server has it, which is atomic and settles the race
        # between two tabs; the pipeline is the same thing for an older Redis.
        try:
            token = store.getdel(key)
        except Exception:
            pipe = store.pipeline()
            pipe.get(key)
            pipe.delete(key)
            token = pipe.execute()[0]
    except Exception:
        raise HTTPException(status_code=503, detail="Could not complete sign-in. Please try again.")
    if not token:
        raise HTTPException(status_code=400, detail="That sign-in link has expired. Please sign in again.")
    return Token(access_token=token, expires_in=cfg.ACCESS_TOKEN_EXPIRE_MINUTES * 60)


def _refuse_when_sso_only(what: str = "Password login is disabled. Please use SSO.") -> None:
    """No password door while the site is single sign-on.

    Login checked this; forgot-password and reset-password did not. Accounts
    are made at the provider now, so what is left to guard is the password
    somebody already has: resetting it, or setting a new one.
    """
    if _get_sso_settings()["sso_only"]:
        raise HTTPException(status_code=403, detail=what)


@router.get("/sso/config")
def sso_config():
    """Public endpoint — tells frontend whether SSO is available and login mode."""
    from app.config import settings as cfg
    sso_enabled = bool(cfg.OIDC_PROVIDER_URL and cfg.OIDC_CLIENT_ID)
    sso_settings = _get_sso_settings()
    from app.services import sso_roles

    return {
        "sso_enabled": sso_enabled,
        "sso_only": sso_settings["sso_only"],
        "provider_name": cfg.OIDC_PROVIDER_NAME if sso_enabled else None,
        # Whether roles come from the provider's groups. Says only that they
        # do, never which groups — enough for the access page to stop offering
        # a control that can now only answer 409.
        "roles_from_provider": sso_roles.is_configured(cfg),
    }


@router.get("/sso/login")
async def sso_login(request: Request):
    """Redirect user to the OIDC provider for login.

    `async`, and the redirect awaited. Authlib's Starlette client is the async
    one — `authorize_redirect` hands back a coroutine — so a sync endpoint
    returned that coroutine to FastAPI, which tried to serialise it as a
    response body and answered 500: "'coroutine' object is not iterable". The
    button had never worked; nothing found it because nothing had SSO
    configured to click it with.
    """
    from authlib.integrations.starlette_client import OAuth
    from starlette.responses import RedirectResponse
    from app.config import settings as cfg

    if not cfg.OIDC_PROVIDER_URL or not cfg.OIDC_CLIENT_ID:
        raise HTTPException(status_code=400, detail="SSO is not configured")

    oauth = OAuth()
    oauth.register(
        name="oidc",
        server_metadata_url=f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration",
        client_id=cfg.OIDC_CLIENT_ID,
        client_secret=cfg.OIDC_CLIENT_SECRET,
        client_kwargs={"scope": cfg.OIDC_SCOPES},
    )

    redirect_uri = f"{cfg.APP_URL}/api/auth/sso/callback"

    # Where to come back to, and whether this attempt is allowed to show a
    # login screen. Both live in the session rather than the URL: the callback
    # needs them and neither is the provider's business.
    #
    # `prompt=none` is the whole of "sign me in if you already know me".
    # Somebody who signed in at the provider for the other app is signed in
    # here too, without a screen, without a click — and somebody who is not
    # gets a quiet refusal we can act on rather than a login page they did not
    # ask for.
    request.session["sso_next"] = _safe_next(request.query_params.get("next"))
    silent = request.query_params.get("prompt") == "none"
    request.session["sso_silent"] = silent
    extra = {"prompt": "none"} if silent else {}
    return await oauth.oidc.authorize_redirect(request, redirect_uri, **extra)


#: The provider's way of saying "nobody is signed in here" to a silent
#: attempt. None of the three is a failure; they are the answer.
QUIET_REFUSALS = {"login_required", "interaction_required", "consent_required",
                  "account_selection_required"}


def _safe_next(raw: str | None) -> str:
    """A path within this app, or the front door.

    Anything else — a scheme, a host, a protocol-relative `//evil` — is an
    open redirect, which is exactly the thing a sign-in round trip must not
    become.
    """
    path = (raw or "").strip()
    if not path.startswith("/") or path.startswith("//"):
        return "/"
    return path[:500]


@router.get("/sso/logout")
async def sso_logout(request: Request):
    """End the session at the provider, not only here.

    Signing out used to mean "this app forgets you": the token went, and the
    provider's session did not — so pressing Sign in put you straight back in
    with no code. On a shared machine that is the wrong default and the one
    nobody expects.

    So the browser is sent to the provider's end-session endpoint, which ends
    the session behind both apps. That is the cost and it is deliberate: there
    is one session, and "sign out" should mean the same word in both places.

    No id_token_hint: neither app keeps the id token, the endpoint is per
    application, and the redirect is checked against the provider's own list.
    If the provider will not take our redirect it shows its own "you have
    logged out" page, which still ends the session — a worse landing, not a
    failure.
    """
    from starlette.responses import RedirectResponse
    from app.config import settings as cfg

    home = f"{cfg.APP_URL}/home"
    if not (cfg.OIDC_PROVIDER_URL and cfg.OIDC_CLIENT_ID):
        return RedirectResponse(url=home)
    try:
        async with httpx.AsyncClient(timeout=8) as client:
            found = await client.get(
                f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration")
            end_session = (found.json() or {}).get("end_session_endpoint")
    except Exception:
        logger.warning("Could not read the provider's logout endpoint", exc_info=True)
        end_session = None
    if not end_session:
        return RedirectResponse(url=home)
    query = urlencode({"post_logout_redirect_uri": home, "client_id": cfg.OIDC_CLIENT_ID})
    return RedirectResponse(url=f"{end_session}?{query}")


@router.get("/sso/callback")
async def sso_callback(request: Request, db: Session = Depends(get_db)):
    """Handle OIDC provider callback — create or login user."""
    from authlib.integrations.starlette_client import OAuth
    from starlette.responses import RedirectResponse
    from app.config import settings as cfg
    from app.services import sso_roles

    if not cfg.OIDC_PROVIDER_URL or not cfg.OIDC_CLIENT_ID:
        raise HTTPException(status_code=400, detail="SSO is not configured")

    oauth = OAuth()
    oauth.register(
        name="oidc",
        server_metadata_url=f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration",
        client_id=cfg.OIDC_CLIENT_ID,
        client_secret=cfg.OIDC_CLIENT_SECRET,
        client_kwargs={"scope": cfg.OIDC_SCOPES},
    )

    silent = bool(request.session.pop("sso_silent", False))
    landing = _safe_next(request.session.pop("sso_next", None))

    # A silent attempt that finds nobody signed in is not a failure. The
    # provider says so in the query string, and the visitor should land on the
    # page they asked for with no message and no sign of having been anywhere.
    refusal = request.query_params.get("error")
    if refusal:
        if silent and refusal in QUIET_REFUSALS:
            return RedirectResponse(url=f"{cfg.APP_URL}{landing}")
        return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed")

    try:
        token = await oauth.oidc.authorize_access_token(request)
    except Exception:
        if silent:
            # Anything at all going wrong on an attempt nobody asked for is
            # still not something to interrupt them with.
            return RedirectResponse(url=f"{cfg.APP_URL}{landing}")
        return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed")

    userinfo = token.get("userinfo") or {}
    email = userinfo.get("email", "").lower().strip()
    name = userinfo.get("name") or userinfo.get("preferred_username") or email.split("@")[0]

    if not email:
        return RedirectResponse(url=f"{cfg.APP_URL}/login?error=no_email")

    # An address the provider will not vouch for is not an identity. Matching
    # on email means whoever proves an address owns the account that already
    # uses it, so an unverified claim would hand over an existing account to
    # anybody who typed the address into a provider that does not check. Only
    # refused when the provider says so explicitly: a provider that omits the
    # claim is not asserting the address is unverified.
    if userinfo.get("email_verified") is False:
        return RedirectResponse(url=f"{cfg.APP_URL}/login?error=email_unverified")

    # Find or create user
    user = db.query(User).filter(User.email == email).first()
    if not user:
        user = User(
            email=email,
            # No password rather than one nobody knows. A random string here
            # reads as "has a password" everywhere that asks.
            hashed_password=None,
            name=name,
            role="user",
        )
        db.add(user)
        db.flush()
        # Auto-verify SSO users
        verification = EmailVerification(
            user_id=user.id,
            token=secrets.token_urlsafe(32),
            expires_at=datetime.utcnow() + timedelta(hours=1),
            verified_at=datetime.utcnow(),
        )
        db.add(verification)
        db.commit()
        db.refresh(user)

    # The provider's directory decides what they are here, when it has been
    # told to. Applied on every sign-in rather than only at creation: a list
    # that can add somebody to the educators group and never take them out is
    # not a list anybody can rely on. Off unless configured — see
    # services/sso_roles.
    if sso_roles.apply(db, user, userinfo, cfg):
        db.commit()
        db.refresh(user)

    # A code in the address bar, never the token itself.
    #
    # This redirect is a page load: the browser asks nginx for it, and nginx
    # logs `"$request"` — so every sign-in wrote a live bearer token, good for
    # a day, into the frontend container's access log, and into the browser's
    # history, and into the Referer of whatever the page loaded next. The code
    # that goes there instead is worth one exchange, within a minute, and is
    # gone the moment it is spent.
    code = _stash_sso_token(create_access_token(data={"sub": user.email}))
    if code is None:
        logger.error("SSO succeeded but the exchange store is unreachable")
        return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed")
    # Back to whatever was being opened when the round trip started, so a link
    # to one article lands on that article rather than the dashboard.
    where = f"&next={quote(landing, safe='')}" if landing and landing != "/" else ""
    return RedirectResponse(url=f"{cfg.APP_URL}/sso-callback?code={code}{where}")