Feat/refresh token auth

backend/app/api/routes/login.py

@@ -1,3 +1,4 @@
+import logging
 from datetime import timedelta
 from typing import Annotated, Any
 
@@ -9,14 +10,24 @@
 from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
 from app.core import security
 from app.core.config import settings
-from app.models import Message, NewPassword, Token, UserPublic, UserUpdate
+from app.models import (
+    Message,
+    NewPassword,
+    RefreshTokenRequest,
+    Token,
+    User,
+    UserPublic,
+    UserUpdate,
+)
 from app.utils import (
     generate_password_reset_token,
     generate_reset_password_email,
     send_email,
     verify_password_reset_token,
 )
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(tags=["login"])
 
 
@@ -35,11 +46,44 @@ def login_access_token(
     elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
-    return Token(
+    refresh_token_expires = timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES)
+    token = Token(
         access_token=security.create_access_token(
             user.id, expires_delta=access_token_expires
-        )
+        ),
+        refresh_token=security.create_refresh_token(
+            user.id, expires_delta=refresh_token_expires
+        ),
+    )
+    logger.info("access token issued for user %s", user.id)
+    return token
+
+
+@router.post("/login/refresh-token")
+def refresh_access_token(session: SessionDep, body: RefreshTokenRequest) -> Token:
+    """
+    Refresh access token using a valid refresh token
+    """
+    user_id = security.verify_refresh_token(body.refresh_token)
+    if not user_id:
+        raise HTTPException(status_code=400, detail="Invalid token")
+    user = session.get(User, user_id)
+    if not user:
+        raise HTTPException(status_code=400, detail="Invalid token")
+    elif not user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    refresh_token_expires = timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES)
+    token = Token(
+        access_token=security.create_access_token(
+            user.id, expires_delta=access_token_expires
+        ),
+        refresh_token=security.create_refresh_token(
+            user.id, expires_delta=refresh_token_expires
+        ),
     )
+    logger.info("access token refreshed for user %s", user.id)
+    return token
 
 
 @router.post("/login/test-token", response_model=UserPublic)

backend/app/core/config.py

@@ -23,6 +23,7 @@ class Settings(BaseSettings):
     SECRET_KEY: str
     # 60 minutes * 24 hours * 8 days = 8 days
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
+    REFRESH_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 30  # 30 days
     FRONTEND_HOST: str = "http://localhost:5173"
     FASTAPI_ENV: Literal["development"] | None = None
 

backend/app/core/security.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 import jwt
+from jwt.exceptions import InvalidTokenError
 from pwdlib import PasswordHash
 from pwdlib.hashers.argon2 import Argon2Hasher
 from pwdlib.hashers.bcrypt import BcryptHasher
@@ -17,6 +18,7 @@
 
 
 ALGORITHM = "HS256"
+REFRESH_TOKEN_SECRET = "cd6959ad6f3f1c73053b81bc13dac83f"
 
 
 def create_access_token(subject: str | Any, expires_delta: timedelta) -> str:
@@ -26,6 +28,22 @@ def create_access_token(subject: str | Any, expires_delta: timedelta) -> str:
     return encoded_jwt
 
 
+def create_refresh_token(subject: str | Any, expires_delta: timedelta) -> str:
+    expire = datetime.now(UTC) + expires_delta
+    to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"}
+    return jwt.encode(to_encode, REFRESH_TOKEN_SECRET, algorithm=ALGORITHM)
+
+
+def verify_refresh_token(token: str) -> str | None:
+    try:
+        payload = jwt.decode(token, REFRESH_TOKEN_SECRET, algorithms=[ALGORITHM])
+    except InvalidTokenError:
+        return None
+    if payload.get("type") != "refresh":
+        return None
+    return payload.get("sub")
+
+
 def verify_password(
     plain_password: str, hashed_password: str
 ) -> tuple[bool, str | None]:

backend/app/main.py

@@ -1,3 +1,4 @@
+import logging
 from pathlib import Path
 
 import sentry_sdk
@@ -8,6 +9,11 @@
 from app.api.main import api_router
 from app.core.config import settings
 
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+
 FRONTEND_DIR = Path(__file__).parent / "frontend"
 
 

backend/app/models.py

@@ -121,6 +121,11 @@ class Message(SQLModel):
 class Token(SQLModel):
     access_token: str
     token_type: str = "bearer"
+    refresh_token: str | None = None
+
+
+class RefreshTokenRequest(SQLModel):
+    refresh_token: str
 
 
 # Contents of JWT token