feat: implement token expiration handling and refresh mechanism

- Added functions to check if the access token is expired and to refresh the token using the refresh token.
- Updated dependencies to utilize the new token management functions in the authentication flow.
- Enhanced the login endpoint to support session management with token state.

Author: atyrode

src/backend/config.py

@@ -1,7 +1,10 @@
 import os
 import json
+import time
+import httpx
 import redis
-from typing import Optional, Dict, Any
+import jwt
+from typing import Optional, Dict, Any, Tuple
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -60,3 +63,72 @@ def get_auth_url() -> str:
 def get_token_url() -> str:
     """Get the token endpoint URL"""
     return f"{OIDC_CONFIG['server_url']}/realms/{OIDC_CONFIG['realm']}/protocol/openid-connect/token"
+
+def is_token_expired(token_data: Dict[str, Any], buffer_seconds: int = 30) -> bool:
+    """
+    Check if the access token is expired or about to expire
+    
+    Args:
+        token_data: The token data containing the access token
+        buffer_seconds: Buffer time in seconds to refresh token before it actually expires
+        
+    Returns:
+        bool: True if token is expired or about to expire, False otherwise
+    """
+    if not token_data or 'access_token' not in token_data:
+        return True
+        
+    try:
+        # Decode the JWT token without verification to get expiration time
+        decoded = jwt.decode(token_data['access_token'], options={"verify_signature": False})
+        
+        # Get expiration time from token
+        exp_time = decoded.get('exp', 0)
+        
+        # Check if token is expired or about to expire (with buffer)
+        current_time = time.time()
+        return current_time + buffer_seconds >= exp_time
+    except Exception as e:
+        print(f"Error checking token expiration: {str(e)}")
+        return True
+
+async def refresh_token(session_id: str, token_data: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]:
+    """
+    Refresh the access token using the refresh token
+    
+    Args:
+        session_id: The session ID
+        token_data: The current token data containing the refresh token
+        
+    Returns:
+        Tuple[bool, Dict[str, Any]]: Success status and updated token data
+    """
+    if not token_data or 'refresh_token' not in token_data:
+        return False, token_data
+        
+    try:
+        async with httpx.AsyncClient() as client:
+            refresh_response = await client.post(
+                get_token_url(),
+                data={
+                    'grant_type': 'refresh_token',
+                    'client_id': OIDC_CONFIG['client_id'],
+                    'client_secret': OIDC_CONFIG['client_secret'],
+                    'refresh_token': token_data['refresh_token']
+                }
+            )
+            
+            if refresh_response.status_code != 200:
+                print(f"Token refresh failed: {refresh_response.text}")
+                return False, token_data
+                
+            # Get new token data
+            new_token_data = refresh_response.json()
+            
+            # Update session with new tokens
+            set_session(session_id, new_token_data)
+            
+            return True, new_token_data
+    except Exception as e:
+        print(f"Error refreshing token: {str(e)}")
+        return False, token_data

src/backend/dependencies.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from fastapi import Request, HTTPException, Depends
 
-from config import get_session
+from config import get_session, is_token_expired, refresh_token
 
 class SessionData:
     def __init__(self, access_token: str, token_data: dict):
@@ -33,6 +33,23 @@ async def __call__(self, request: Request) -> Optional[SessionData]:
                     headers={"WWW-Authenticate": "Bearer"},
                 )
             return None
+            
+        # Check if token is expired and refresh if needed
+        if is_token_expired(session):
+            # Try to refresh the token
+            success, new_session = await refresh_token(session_id, session)
+            if not success:
+                # Token refresh failed, user needs to re-authenticate
+                if self.auto_error:
+                    raise HTTPException(
+                        status_code=401,
+                        detail="Session expired",
+                        headers={"WWW-Authenticate": "Bearer"},
+                    )
+                return None
+            # Use the refreshed token data
+            session = new_session
+            
         return SessionData(
             access_token=session.get('access_token'),
             token_data=session

src/backend/routers/auth.py

@@ -14,13 +14,18 @@
 
 @auth_router.get("/login")
 async def login(request: Request, kc_idp_hint: str = None, popup: str = None):
+    
     session_id = secrets.token_urlsafe(32)
+
     auth_url = get_auth_url()
     state = "popup" if popup == "1" else "default"
+
     if kc_idp_hint:
         auth_url = f"{auth_url}&kc_idp_hint={kc_idp_hint}"
+
     # Add state param to OIDC URL
     auth_url = f"{auth_url}&state={state}"
+
     response = RedirectResponse(auth_url)
     response.set_cookie('session_id', session_id)