Merge pull request #2 from YuriFontella/fix/exp

add: expiration access token

Author: YuriFontella

.env.example

@@ -22,3 +22,5 @@ DATABASE_MAX_INACTIVE_CONNECTION_LIFETIME=300.0
 
 # JWT Settings
 JWT_ALGORITHM=HS256
+ACCESS_TOKEN_EXPIRE_MINUTES=15
+REFRESH_TOKEN_EXPIRE_DAYS=7

src/config/base.py

@@ -38,6 +38,8 @@ class AppSettings:
     PBKDF2_ALGORITHM: str = "sha256"
     MAX_FINGERPRINT_VALUE: int = 100_000_000
     BCRYPT_GENSALT: int = 12
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 15  # 15 minutes
+    REFRESH_TOKEN_EXPIRE_DAYS: int = 7  # 7 days
 
     def __post_init__(self):
         self.SECRET_KEY = self.SECRET_KEY or os.getenv("SECRET_KEY")
@@ -46,6 +48,8 @@ def __post_init__(self):
         self.PBKDF2_ALGORITHM = os.getenv("PBKDF2_ALGORITHM", self.PBKDF2_ALGORITHM)
         self.MAX_FINGERPRINT_VALUE = int(os.getenv("MAX_FINGERPRINT_VALUE", self.MAX_FINGERPRINT_VALUE))
         self.BCRYPT_GENSALT = int(os.getenv("BCRYPT_GENSALT", self.BCRYPT_GENSALT))
+        self.ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", self.ACCESS_TOKEN_EXPIRE_MINUTES))
+        self.REFRESH_TOKEN_EXPIRE_DAYS = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", self.REFRESH_TOKEN_EXPIRE_DAYS))
 
         if not self.ALLOWED_CORS_ORIGINS:
             cors_origins = os.getenv("ALLOWED_CORS_ORIGINS")

src/db/migrations/003_create_sessions.sql

@@ -6,7 +6,8 @@ CREATE TABLE IF NOT EXISTS sessions (
     revoked bool DEFAULT false,
     user_uuid uuid NOT NULL REFERENCES users (uuid) ON DELETE CASCADE,
     type varchar NOT NULL DEFAULT 'manual',
-    date timestamp with time zone DEFAULT current_timestamp
+    create timestamp with time zone DEFAULT current_timestamp,
+    update timestamp with time zone DEFAULT current_timestamp
 );
 
 ALTER TABLE sessions ADD CONSTRAINT uq_sessions_access_token UNIQUE (access_token);

src/domain/users/repositories/session.py

@@ -41,11 +41,17 @@ async def revoke_session(self, session_uuid: UUID) -> bool:
         await self.connection.execute(query, str(session_uuid))
         return True
 
-    async def update_access_token(self, session_uuid: str, access_token: str) -> None:
-        """Atualiza apenas o access_token de uma sessão existente"""
+    async def update_access_token(
+        self,
+        session_uuid: str,
+        access_token: str,
+        user_agent: Optional[str],
+        ip: Optional[str],
+    ) -> None:
+        """Atualiza o access_token, user_agent e ip de uma sessão existente"""
         query = """
             UPDATE sessions
-            SET access_token = $1, date = NOW()
-            WHERE uuid = $2 AND revoked = false
+            SET access_token = $1, user_agent = $2, ip = $3, update = NOW()
+            WHERE uuid = $4 AND revoked = false
         """
-        await self.connection.execute(query, access_token, session_uuid)
+        await self.connection.execute(query, access_token, user_agent, ip, session_uuid)

src/domain/users/services.py

@@ -3,6 +3,8 @@
 import bcrypt
 import jwt
 
+from datetime import datetime, timezone, timedelta
+
 from dataclasses import dataclass, field
 from typing import Optional
 from asyncpg import Connection
@@ -86,14 +88,26 @@ async def authenticate(
         if not session:
             raise ValueError("Something went wrong creating the session")
 
+        # Calculate expiration times
+        access_token_exp = datetime.now(timezone.utc) + timedelta(minutes=self.settings.app.ACCESS_TOKEN_EXPIRE_MINUTES)
+        refresh_token_exp = datetime.now(timezone.utc) + timedelta(days=self.settings.app.REFRESH_TOKEN_EXPIRE_DAYS)
+
         access_token_jwt = jwt.encode(
-            {"uuid": str(user_uuid), "access_token": random_access_token},
+            {
+                "uuid": str(user_uuid),
+                "access_token": random_access_token,
+                "exp": access_token_exp
+            },
             key=self.settings.app.SECRET_KEY,
             algorithm=self.settings.app.JWT_ALGORITHM,
         )
 
         refresh_token_jwt = jwt.encode(
-            {"uuid": str(user_uuid), "refresh_token": random_refresh_token},
+            {
+                "uuid": str(user_uuid),
+                "refresh_token": random_refresh_token,
+                "exp": refresh_token_exp
+            },
             key=self.settings.app.SECRET_KEY,
             algorithm=self.settings.app.JWT_ALGORITHM,
         )
@@ -105,7 +119,6 @@ async def refresh_access_token(
     ) -> Token:
         """Refreshes the access_token using a valid refresh_token"""
         try:
-            # Decode the refresh token JWT
             decoded = jwt.decode(
                 jwt=refresh_token,
                 key=self.settings.app.SECRET_KEY,
@@ -146,18 +159,29 @@ async def refresh_access_token(
             # Update only the access token in the existing session
             await self.session_repository.update_access_token(
                 session_uuid=session["uuid"],
-                access_token=access_token_hash.hex()
+                access_token=access_token_hash.hex(),
+                user_agent=user_agent,
+                ip=ip
             )
 
+            # Calculate expiration time for new access token
+            access_token_exp = datetime.now(timezone.utc) + timedelta(minutes=self.settings.app.ACCESS_TOKEN_EXPIRE_MINUTES)
+
             # Generate new access token JWT
             access_token_jwt = jwt.encode(
-                {"uuid": user_uuid, "access_token": random_access_token},
+                {
+                    "uuid": user_uuid,
+                    "access_token": random_access_token,
+                    "exp": access_token_exp
+                },
                 key=self.settings.app.SECRET_KEY,
                 algorithm=self.settings.app.JWT_ALGORITHM,
             )
 
             # Return the new access token and keep the same refresh token
             return Token(access_token=access_token_jwt, refresh_token=refresh_token)
 
+        except jwt.ExpiredSignatureError:
+            raise ValueError("Refresh token expired")
         except jwt.PyJWTError:
             raise ValueError("Invalid refresh token format")

src/server/auth.py

@@ -1,6 +1,6 @@
 import hashlib
 
-from jwt import PyJWTError, decode
+from jwt import PyJWTError, ExpiredSignatureError, decode
 
 from litestar.connection import ASGIConnection
 from litestar.exceptions import NotAuthorizedException
@@ -44,8 +44,10 @@ async def authenticate_request(
             if not user:
                 raise NotAuthorizedException()
 
+        except ExpiredSignatureError:
+            raise NotAuthorizedException(detail="Token expired")
         except PyJWTError:
-            raise NotAuthorizedException()
+            raise NotAuthorizedException(detail="Invalid token")
 
         else:
             return AuthenticationResult(user=user, auth=auth)