plm-lottery/app/auth/dependencies.py

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

from app.api.errors import http_error
from app.auth.security import decode_access_token
from app.db.models import User
from app.db.session import get_session

_bearer = HTTPBearer()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(_bearer),
    session: AsyncSession = Depends(get_session),
) -> User:
    try:
        user_id, token_version = decode_access_token(credentials.credentials)
    except Exception as exc:
        raise http_error(status.HTTP_401_UNAUTHORIZED, "session_expired", "invalid token") from exc

    user = await session.scalar(select(User).where(User.id == user_id))
    if user is None:
        raise http_error(status.HTTP_401_UNAUTHORIZED, "session_expired", "user not found")
    if user.token_version != token_version:
        # B-34: a password change (self-service or admin reset) bumps
        # token_version, so a token issued before it — including one an
        # attacker who had the old password is still holding — reads as
        # expired rather than staying valid until it naturally times out.
        raise http_error(status.HTTP_401_UNAUTHORIZED, "session_expired", "token has been superseded")
    return user


async def get_optional_user(
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> User | None:
    """Like get_current_user, but for endpoints reachable both logged-out and
    logged-in (e.g. /rounds/current) that need to personalize their response
    *if* the caller happens to be authenticated, without requiring it."""
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return None
    try:
        user_id, token_version = decode_access_token(auth_header.removeprefix("Bearer "))
    except Exception:
        return None
    user = await session.scalar(select(User).where(User.id == user_id))
    if user is None or user.token_version != token_version:
        return None
    return user