78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""Token issuance / rotation shared by the JSON API and the HTMX UI."""
|
|
|
|
from datetime import timedelta
|
|
|
|
import jwt
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import (
|
|
REFRESH_TOKEN_TYPE,
|
|
create_access_token,
|
|
create_refresh_token,
|
|
decode_token,
|
|
)
|
|
from app.crud import token as token_crud
|
|
from app.models.user import User
|
|
|
|
|
|
async def issue_token_pair(
|
|
db: AsyncSession, user: User, access_expires_delta: timedelta | None = None
|
|
) -> tuple[str, str]:
|
|
"""Create an access + refresh token and persist the refresh token for revocation.
|
|
|
|
``access_expires_delta`` overrides the access token's default lifetime
|
|
(``settings.access_token_expire_minutes``) — used by the web login flow to
|
|
issue a longer-lived session cookie without affecting API bearer tokens.
|
|
"""
|
|
access = create_access_token(user.id, access_expires_delta)
|
|
refresh, payload = create_refresh_token(user.id)
|
|
await token_crud.store(
|
|
db,
|
|
user_id=user.id,
|
|
jti=payload["jti"],
|
|
raw_token=refresh,
|
|
expires_at=payload["exp"], # timezone-aware UTC datetime
|
|
)
|
|
return access, refresh
|
|
|
|
|
|
class RefreshError(Exception):
|
|
pass
|
|
|
|
|
|
async def rotate_refresh_token(
|
|
db: AsyncSession, raw_refresh: str
|
|
) -> tuple[str, str, User]:
|
|
"""Validate a refresh token, revoke it, and issue a fresh pair.
|
|
|
|
Raises ``RefreshError`` if the token is invalid, expired, or already revoked.
|
|
"""
|
|
try:
|
|
payload = decode_token(raw_refresh)
|
|
except jwt.PyJWTError as exc:
|
|
raise RefreshError("Invalid refresh token") from exc
|
|
|
|
if payload.get("type") != REFRESH_TOKEN_TYPE:
|
|
raise RefreshError("Not a refresh token")
|
|
|
|
jti = payload.get("jti")
|
|
sub = payload.get("sub")
|
|
if not jti or sub is None:
|
|
raise RefreshError("Malformed refresh token")
|
|
|
|
record = await token_crud.get_active(db, jti=jti, raw_token=raw_refresh)
|
|
if record is None:
|
|
raise RefreshError("Refresh token is revoked or expired")
|
|
|
|
# Local import avoids a circular import at module load time.
|
|
from app.crud import user as user_crud
|
|
|
|
user = await user_crud.get(db, int(sub))
|
|
if user is None or not user.is_active:
|
|
raise RefreshError("User is inactive or missing")
|
|
|
|
# Rotate: revoke the presented token, then mint a new pair.
|
|
await token_crud.revoke(db, record)
|
|
access, refresh = await issue_token_pair(db, user)
|
|
return access, refresh, user
|