feat: add JWT refresh token support with automatic token renewal and session expiration handling
CI / backend (push) Failing after 3s
CI / frontend (push) Failing after 26s

Add /auth/refresh endpoint to issue new access tokens using refresh tokens with token type validation and user activity checks, implement automatic token refresh on 401 responses with single retry logic in frontend API client, add authorizedFetch helper for non-JSON endpoints with refresh support, store both access and refresh tokens in localStorage with clearTokens cleanup helper, add nexafabric.authExpired event
This commit is contained in:
2026-07-09 14:27:46 +02:00
parent 7818158a9b
commit 45baa6ae7a
4 changed files with 95 additions and 10 deletions
+22 -1
View File
@@ -5,7 +5,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from app.api.deps import CurrentUser
from app.core.security import create_access_token, create_refresh_token, verify_password
from app.core.security import create_access_token, create_refresh_token, decode_token, verify_password
from app.db.session import get_db
from app.models.domain import User
from app.schemas.domain import LoginRequest, TokenPair, UserRead
@@ -33,6 +33,27 @@ def login(payload: LoginRequest, db: Session = Depends(get_db)) -> TokenPair:
)
@router.post("/refresh", response_model=TokenPair)
def refresh(payload: dict[str, str], db: Session = Depends(get_db)) -> TokenPair:
token = payload.get("refresh_token")
if not token:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing refresh token")
try:
claims = decode_token(token)
except Exception as exc:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token") from exc
if claims.get("typ") != "refresh":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type")
user = db.scalar(select(User).where(User.id == claims.get("sub"), User.is_active.is_(True)))
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive or missing user")
permissions = sorted({permission for role in user.roles for permission in role.permissions})
return TokenPair(
access_token=create_access_token(user.id, permissions),
refresh_token=create_refresh_token(user.id),
)
@router.get("/me", response_model=UserRead)
def me(user: CurrentUser) -> User:
return user