refresh tokens 0.1.3

src/service/auth/auth.py

@@ -8,6 +8,8 @@ from src.database.auth.refresh_tokens import JwtCrudActions
 from src.errors.http_errors.errors import Errors
 from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
 from src.models.configs_read.env import env_settings
+
+
 class CurrentUser:
     
     def __init__(self) -> None:
@@ -52,19 +54,14 @@ class CurrentUser:
     
 
 
-    def create_access_token(self, form_data_email:str, form_data_password:str)->str:
-        '''check user info'''
-        user = self._check(form_data_email, form_data_password)
-        
+    def create_access_token(self, user_id:UUID)->str:       
         '''create new access token if all the checks are successful'''
-        return self.jwt_service.create_access_token({"sub":str(user.id)})
+        return self.jwt_service.create_access_token({"sub":str(user_id)})
 
     
     
-    def create_refresh_token(self,form_data_email:str, form_data_password:str, request:Request)->str:
-        '''check user info'''
-        user=self._check(form_data_email, form_data_password)
-        token, jti=self.jwt_service.create_refresh_token({"sub":str(user.id)})
+    def create_refresh_token(self,user_id:UUID, request:Request)->str:
+        token, jti=self.jwt_service.create_refresh_token({"sub":str(user_id)})
         
         try:
             jti=UUID(jti)
@@ -74,7 +71,7 @@ class CurrentUser:
         '''create new refresh token if all the checks are successful'''
         token_record=RefreshTokensCreate(
             id=jti,
-            user_id=user.id,
+            user_id=user_id,
             token_hash=self.hash.token_to_hash(token),
             device_info=request.headers.get("user-agent", "unknown"),
             ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
@@ -85,7 +82,7 @@ class CurrentUser:
         return token
     
     
-    def refresh_token(self, refresh_token:str, expires_delta:timedelta, request:Request)->str:
+    def refresh_token(self, refresh_token:str, request:Request)->tuple[str, str]:
         '''decode old refresh token'''
         old_refresh_token=self.jwt_service.jwt_decode(refresh_token)
         if (sub:=old_refresh_token.get("sub")) is None or (old_jti:=old_refresh_token.get("jti")) is None:
@@ -105,7 +102,13 @@ class CurrentUser:
         if old_record.is_revoked:
             self.jwt_db_actions.revoke_all(old_record.user_id)
             raise self.error.credentials_error(detail="Reuse token detected")
-        if old_record.expires_at<datetime.now(timezone.utc):
+        
+        
+        '''sqlite constraints about timezone'''
+        expires_at=old_record.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=timezone.utc)
+        if expires_at<datetime.now(timezone.utc):
             raise self.error.credentials_error(detail="Token expired")      
         
         '''user check'''    
@@ -118,7 +121,7 @@ class CurrentUser:
 
         '''create new refresh token if all the checks are successful'''
         new_refresh_token, new_jti=self.jwt_service.create_refresh_token({"sub":str(sub)})
-        
+        new_access_token=self.create_access_token(user_id=sub)
         try:
             new_jti=UUID(new_jti)
         except (ValueError, TypeError):
@@ -132,14 +135,14 @@ class CurrentUser:
             token_hash=self.hash.token_to_hash(new_refresh_token),
             device_info=request.headers.get("user-agent", "unknown"),
             ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
-            expires_at=datetime.now(timezone.utc)+expires_delta,
+            expires_at=datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
         )
         self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(new_token_record))
         
         '''update old token to deactivate it and assign replaced_by'''
         self.jwt_db_actions.update_token(old_jti, new_jti)
         
-        return new_refresh_token
+        return (new_access_token,new_refresh_token)
     
     
     
@@ -165,10 +168,13 @@ class CurrentUser:
     
     
     def login(self, form_data_email:str, form_data_password:str, request:Request)->tuple[str, str]:
+        '''revoke all the old refresh tokens'''
+        user = self._check(form_data_email, form_data_password)
+        self.jwt_db_actions.revoke_all(user_id=user.id)
         
         '''create access and refresh tokens'''
-        access_token=self.create_access_token(form_data_email=form_data_email, form_data_password=form_data_password)
-        refresh_token=self.create_refresh_token(form_data_password=form_data_password, form_data_email=form_data_email,request=request)
+        access_token=self.create_access_token(user_id=user.id)
+        refresh_token=self.create_refresh_token(user_id=user.id,request=request)
         
         return (access_token, refresh_token)
     

src/service/auth/jwt.py

@@ -34,7 +34,7 @@ class Jwt:
         user_info=data.copy()
         user_info.update({"exp": datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
                           "token_type":"access"})
-        
+        print(f"DEBUG: expires at {datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES)}, minutes={env_settings.ACCESS_TOKEN_EXPIRE_MINUTES}")
         return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM)