davideandClaude Sonnet 5
739fc9fed2
Neither self-service password change nor the admin reset invalidated already-issued JWTs — a 24h-lifetime token stayed valid regardless, so a stolen token (or an attacker who already had the old password) kept working past a password change meant to lock them out. The admin reset exists precisely for the "account compromised" case and didn't evict the attacker at all. Add User.token_version (migration 943dbd74d983), embedded in every JWT as a "tv" claim and checked against the DB on every request in get_current_user/get_optional_user; a mismatch reads as session_expired. Both change-password and the admin reset bump it. change-password hands back a freshly minted token so the caller's own session keeps working instead of being logged out by its own request; the admin reset does not, since that session isn't the one making the call. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
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
|