ai-rpg/app/api/deps.py

"""Shared API dependencies: current user, admin guard, db session, settings.

Token resolution order:
1. `Authorization: Bearer <jwt>` header (preferred).
2. `?access_token=<jwt>` query parameter (fallback for SSE — EventSource cannot
   set custom headers).
"""

from __future__ import annotations

import uuid
from typing import Any

from fastapi import Depends, HTTPException, Query, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.core.settings_service import get_all_settings
from app.db import get_db
from app.models import User

_bearer = HTTPBearer(auto_error=False)


async def get_current_user(
    creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
    access_token: str | None = Query(default=None, description="JWT access token (fallback for SSE)"),
    db: AsyncSession = Depends(get_db),
) -> User:
    """Resolve the JWT bearer token to a User row.

    Accepts the token either in the `Authorization: Bearer <jwt>` header
    or as the `?access_token=<jwt>` query parameter (the latter is needed
    because native `EventSource` cannot send custom headers, and the
    frontend SSE client passes the token via query string).

    Raises 401 on missing/invalid/expired token.
    """
    raw_token: str | None = None
    if creds is not None and creds.scheme.lower() == "bearer":
        raw_token = creds.credentials
    elif access_token:
        raw_token = access_token

    if not raw_token:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing bearer token")

    try:
        payload = decode_token(raw_token)
    except Exception as e:  # noqa: BLE001
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"Invalid token: {e}")
    if payload.get("type") != "access":
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wrong token type")
    user_id_str = payload.get("sub")
    if not user_id_str:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Token missing sub")
    try:
        user_id = uuid.UUID(user_id_str)
    except ValueError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid user id in token")

    user = (
        await db.execute(select(User).where(User.id == user_id))
    ).scalar_one_or_none()
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "User not found")
    if not user.is_active:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Account disabled")
    return user


async def require_admin(user: User = Depends(get_current_user)) -> User:
    """Require an admin user."""
    if not user.is_admin:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Admin only")
    return user


async def get_settings_dict(db: AsyncSession = Depends(get_db)) -> dict[str, Any]:
    """FastAPI dependency: returns the full settings dict."""
    return await get_all_settings(db)