feat: add auth/refresh for get access token from refresh token

Back/core/config.py

@@ -7,7 +7,8 @@ class Settings(BaseSettings):
     DEBUG: bool = False
 
     SECRET_KEY: str
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+    ACCESS_TOKEN_EXPIRE_DAYS: int = 1
+    REFRESH_TOKEN_EXPIRE_WEEKS: int = 12
     ALGORITHM: str = "HS256"
     SECRET_PASS_LENGTH: int
 

Back/core/jwt.py

@@ -14,7 +14,7 @@ def create_access_token(
         expire = datetime.now(timezone.utc) + expires_delta
     else:
         expire = datetime.now(timezone.utc) + timedelta(
-            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
+            days=settings.ACCESS_TOKEN_EXPIRE_DAYS
         )
 
     payload = {
@@ -30,6 +30,33 @@ def create_access_token(
     )
 
 
+def create_refresh_token(
+    subject: str,
+    token_version: int,
+    expires_delta: timedelta | None = None,
+) -> str:
+
+    if expires_delta:
+        expire = datetime.now(timezone.utc) + expires_delta
+    else:
+        expire = datetime.now(timezone.utc) + timedelta(
+            weeks=settings.REFRESH_TOKEN_EXPIRE_WEEKS
+        )
+
+    payload = {
+        "sub": subject,
+        "token_version": token_version,
+        "exp": expire,
+        "type": "refresh"
+    }
+
+    return jwt.encode(
+        payload,
+        settings.SECRET_KEY,
+        algorithm=settings.ALGORITHM,
+    )
+
+
 def decode_token(token: str):
 
     try:

Back/domains/auth/api.py

@@ -2,14 +2,17 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.session import get_db
+from core.jwt import decode_token, create_access_token
+from domains.users.repo import get_user_by_id
 
 from domains.auth.schemas import (
     LoginRequest,
-    TokenResponse
+    TokenResponse,
+    RefreshTokenRequest
 )
 
 from domains.auth.service import login_user
-
+import uuid
 
 router = APIRouter(
     prefix="/auth",
@@ -35,4 +38,52 @@ async def login(
             detail="Invalid username or secret"
         )
 
-    return token
+    return token
+
+@router.post("/refresh", response_model=TokenResponse)
+async def refresh(
+    payload: RefreshTokenRequest,
+    db: AsyncSession = Depends(get_db)
+):
+    payload_data = decode_token(payload.refresh_token)
+    if not payload_data or payload_data.get("type") != "refresh":
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid refresh token",
+        )
+
+    user_id = payload_data.get("sub")
+    token_version = payload_data.get("token_version")
+
+    if not user_id or token_version is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid refresh token payload",
+        )
+
+    try:
+        user_uuid = uuid.UUID(user_id)
+    except ValueError:
+       raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid user ID in token",
+        )
+
+    user = await get_user_by_id(db, user_uuid)
+
+    if not user or not user.is_active or user.token_version != token_version:
+         raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid refresh token",
+        )
+
+    access_token = create_access_token(
+        subject=str(user.id),
+        token_version=user.token_version
+    )
+
+    return {
+        "access_token": access_token,
+        "refresh_token": payload.refresh_token,
+        "token_type": "bearer"
+    }

Back/domains/auth/schemas.py

@@ -9,8 +9,12 @@ class LoginRequest(BaseModel):
 
 class TokenResponse(BaseModel):
     access_token: str
+    refresh_token: str
     token_type: str = "bearer"
 
+class RefreshTokenRequest(BaseModel):
+    refresh_token: str
+
 
 class AuthUser(BaseModel):
     id: uuid.UUID

Back/domains/auth/service.py

@@ -1,7 +1,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.security import verify_password
-from core.jwt import create_access_token
+from core.jwt import create_access_token, create_refresh_token
 
 from domains.users.repo import get_user_by_username
 
@@ -48,7 +48,13 @@ async def login_user(
         token_version=user.token_version
     )
 
+    refresh_token = create_refresh_token(
+        subject=str(user.id),
+        token_version=user.token_version
+    )
+
     return {
         "access_token": token,
+        "refresh_token": refresh_token,
         "token_type": "bearer"
     }