added rsa token and adjusted logic accordingly

VaibhavSingh8 · VaibhavSingh8 · commit 8a133033ac1b · 2025-07-06T22:28:41.000+05:30
diff --git a/.env.example b/.env.example
@@ -7,8 +7,10 @@ RDS_BACKEND_BASE_URL='http://localhost:3000'
 RDS_PUBLIC_KEY="public-key-here"
 GOOGLE_OAUTH_CLIENT_ID="google-client-id"
 GOOGLE_OAUTH_CLIENT_SECRET="client-secret"
-GOOGLE_OAUTH_REDIRECT_URI="environment-url/auth/google/callback"
-GOOGLE_JWT_SECRET_KEY=generate-secret-key
+# Google JWT RSA Keys
+GOOGLE_JWT_PRIVATE_KEY="generate keys and paste here"
+GOOGLE_JWT_PUBLIC_KEY="generate keys and paste here"
 
-GOOGLE_JWT_SECRET_KEY="7i=aje)abyu!8m3jc9&vlmo@e-o__65_jw_=g^ktw7@+saa1z("
-GOOGLE_OAUTH_REDIRECT_URI="http://localhost:8000/v1/auth/google/callback"
+# use if required
+# GOOGLE_JWT_ACCESS_LIFETIME="20"
+# GOOGLE_JWT_REFRESH_LIFETIME="30"
diff --git a/todo/middlewares/jwt_auth.py b/todo/middlewares/jwt_auth.py
@@ -3,9 +3,17 @@
 from django.http import JsonResponse
 
 from todo.utils.jwt_utils import verify_jwt_token
-from todo.utils.google_jwt_utils import validate_google_access_token
+from todo.utils.google_jwt_utils import (
+    validate_google_access_token,
+    validate_google_refresh_token,
+    generate_google_access_token,
+)
 from todo.exceptions.auth_exceptions import TokenMissingError, TokenExpiredError, TokenInvalidError
-from todo.exceptions.google_auth_exceptions import GoogleTokenExpiredError, GoogleTokenInvalidError
+from todo.exceptions.google_auth_exceptions import (
+    GoogleTokenExpiredError,
+    GoogleTokenInvalidError,
+    GoogleRefreshTokenExpiredError,
+)
 from todo.constants.messages import AuthErrorMessages, ApiErrors
 from todo.dto.responses.error_response import ApiErrorResponse, ApiErrorDetail
 
@@ -25,7 +33,8 @@ def __call__(self, request):
             auth_success = self._try_authentication(request)
 
             if auth_success:
-                return self.get_response(request)
+                response = self.get_response(request)
+                return self._process_response(request, response)
             else:
                 error_response = ApiErrorResponse(
                     statusCode=status.HTTP_401_UNAUTHORIZED,
@@ -38,7 +47,8 @@ def __call__(self, request):
                     ],
                 )
                 return JsonResponse(
-                    data=error_response.model_dump(mode="json", exclude_none=True), status=status.HTTP_401_UNAUTHORIZED
+                    data=error_response.model_dump(mode="json", exclude_none=True), 
+                    status=status.HTTP_401_UNAUTHORIZED
                 )
 
         except (TokenMissingError, TokenExpiredError, TokenInvalidError) as e:
@@ -57,7 +67,8 @@ def __call__(self, request):
                 ],
             )
             return JsonResponse(
-                data=error_response.model_dump(mode="json", exclude_none=True), status=status.HTTP_401_UNAUTHORIZED
+                data=error_response.model_dump(mode="json", exclude_none=True), 
+                status=status.HTTP_401_UNAUTHORIZED
             )
 
     def _try_authentication(self, request) -> bool:
@@ -73,25 +84,61 @@ def _try_google_auth(self, request) -> bool:
         try:
             google_token = request.COOKIES.get("ext-access")
 
-            if not google_token:
-                return False
-
-            payload = validate_google_access_token(google_token)
+            if google_token:
+                try:
+                    payload = validate_google_access_token(google_token)
+                    self._set_google_user_data(request, payload)
+                    return True
+                except (GoogleTokenExpiredError, GoogleTokenInvalidError):
+                    pass
 
-            request.auth_type = "google"
-            request.user_id = payload["user_id"]
-            request.google_id = payload["google_id"]
-            request.user_email = payload["email"]
-            request.user_name = payload["name"]
-            request.user_role = "external_user"
-
-            return True
+            return self._try_google_refresh(request)
 
         except (GoogleTokenExpiredError, GoogleTokenInvalidError) as e:
             raise e
         except Exception:
             return False
 
+    def _try_google_refresh(self, request) -> bool:
+        """Try to refresh Google access token"""
+        try:
+            refresh_token = request.COOKIES.get("ext-refresh")
+            
+            if not refresh_token:
+                return False
+                
+            payload = validate_google_refresh_token(refresh_token)
+            
+            user_data = {
+                "user_id": payload["user_id"],
+                "google_id": payload["google_id"],
+                "email": payload["email"],
+                "name": payload.get("name", ""),
+            }
+            
+            new_access_token = generate_google_access_token(user_data)
+            
+            self._set_google_user_data(request, payload)
+
+            request._new_access_token = new_access_token
+            request._access_token_expires = settings.GOOGLE_JWT["ACCESS_TOKEN_LIFETIME"]
+            
+            return True
+            
+        except (GoogleRefreshTokenExpiredError, GoogleTokenInvalidError):
+            return False
+        except Exception:
+            return False
+
+    def _set_google_user_data(self, request, payload):
+        """Set Google user data on request"""
+        request.auth_type = "google"
+        request.user_id = payload["user_id"]
+        request.google_id = payload["google_id"]
+        request.user_email = payload["email"]
+        request.user_name = payload.get("name", "")
+        request.user_role = "external_user"
+
     def _try_rds_auth(self, request) -> bool:
         try:
             rds_token = request.COOKIES.get(self.rds_cookie_name)
@@ -112,6 +159,28 @@ def _try_rds_auth(self, request) -> bool:
         except Exception:
             return False
 
+    def _process_response(self, request, response):
+        """Process response and set new cookies if Google token was refreshed"""
+        if hasattr(request, '_new_access_token'):
+            config = self._get_cookie_config()
+            response.set_cookie(
+                "ext-access",
+                request._new_access_token,
+                max_age=request._access_token_expires,
+                **config
+            )
+        return response
+
+    def _get_cookie_config(self):
+        """Get Google cookie configuration"""
+        return {
+            "path": "/",
+            "domain": settings.GOOGLE_COOKIE_SETTINGS.get("COOKIE_DOMAIN"),
+            "secure": settings.GOOGLE_COOKIE_SETTINGS.get("COOKIE_SECURE", False),
+            "httponly": True,
+            "samesite": settings.GOOGLE_COOKIE_SETTINGS.get("COOKIE_SAMESITE", "Lax"),
+        }
+
     def _is_public_path(self, path: str) -> bool:
         return any(path.startswith(public_path) for public_path in settings.PUBLIC_PATHS)
 
@@ -122,7 +191,8 @@ def _handle_rds_auth_error(self, exception):
             errors=[ApiErrorDetail(title=ApiErrors.AUTHENTICATION_FAILED, detail=str(exception))],
         )
         return JsonResponse(
-            data=error_response.model_dump(mode="json", exclude_none=True), status=status.HTTP_401_UNAUTHORIZED
+            data=error_response.model_dump(mode="json", exclude_none=True), 
+            status=status.HTTP_401_UNAUTHORIZED
         )
 
     def _handle_google_auth_error(self, exception):
@@ -132,7 +202,8 @@ def _handle_google_auth_error(self, exception):
             errors=[ApiErrorDetail(title=ApiErrors.AUTHENTICATION_FAILED, detail=str(exception))],
         )
         return JsonResponse(
-            data=error_response.model_dump(mode="json", exclude_none=True), status=status.HTTP_401_UNAUTHORIZED
+            data=error_response.model_dump(mode="json", exclude_none=True), 
+            status=status.HTTP_401_UNAUTHORIZED
         )
 
 
@@ -169,4 +240,4 @@ def get_current_user_info(request) -> dict:
             }
         )
 
-    return user_info
+    return user_info
diff --git a/todo/urls.py b/todo/urls.py
@@ -5,7 +5,6 @@
 from todo.views.auth import (
     GoogleLoginView,
     GoogleCallbackView,
-    GoogleRefreshView,
     GoogleLogoutView,
 )
 
@@ -16,6 +15,5 @@
     path("labels", LabelListView.as_view(), name="labels"),
     path("auth/google/login/", GoogleLoginView.as_view(), name="google_login"),
     path("auth/google/callback/", GoogleCallbackView.as_view(), name="google_callback"),
-    path("auth/google/refresh/", GoogleRefreshView.as_view(), name="google_refresh"),
     path("auth/google/logout/", GoogleLogoutView.as_view(), name="google_logout"),
 ]
diff --git a/todo/utils/google_jwt_utils.py b/todo/utils/google_jwt_utils.py
@@ -8,7 +8,7 @@
     GoogleRefreshTokenExpiredError,
 )
 
-from todo.constants.messages import AuthErrorMessages, ApiErrors
+from todo.constants.messages import AuthErrorMessages
 
 
 def generate_google_access_token(user_data: dict) -> str:
@@ -28,14 +28,11 @@ def generate_google_access_token(user_data: dict) -> str:
             "token_type": "access",
         }
 
-        token = jwt.encode(
-            payload=payload, key=settings.GOOGLE_JWT["SECRET_KEY"], algorithm=settings.GOOGLE_JWT["ALGORITHM"]
-        )
-
+        token = jwt.encode(payload=payload, key=settings.GOOGLE_JWT["PRIVATE_KEY"], algorithm=settings.GOOGLE_JWT["ALGORITHM"])
         return token
 
-    except Exception:
-        raise GoogleTokenInvalidError(ApiErrors.GOOGLE_API_ERROR)
+    except Exception as e:
+        raise GoogleTokenInvalidError(f"Token generation failed: {str(e)}")
 
 
 def generate_google_refresh_token(user_data: dict) -> str:
@@ -53,22 +50,17 @@ def generate_google_refresh_token(user_data: dict) -> str:
             "email": user_data["email"],
             "token_type": "refresh",
         }
-
-        token = jwt.encode(
-            payload=payload, key=settings.GOOGLE_JWT["SECRET_KEY"], algorithm=settings.GOOGLE_JWT["ALGORITHM"]
-        )
+        token = jwt.encode(payload=payload, key=settings.GOOGLE_JWT["PRIVATE_KEY"], algorithm=settings.GOOGLE_JWT["ALGORITHM"])
 
         return token
 
-    except Exception:
-        raise GoogleTokenInvalidError(ApiErrors.GOOGLE_API_ERROR)
+    except Exception as e:
+        raise GoogleTokenInvalidError(f"Refresh token generation failed: {str(e)}")
 
 
 def validate_google_access_token(token: str) -> dict:
     try:
-        payload = jwt.decode(
-            jwt=token, key=settings.GOOGLE_JWT["SECRET_KEY"], algorithms=[settings.GOOGLE_JWT["ALGORITHM"]]
-        )
+        payload = jwt.decode(jwt=token, key=settings.GOOGLE_JWT["PUBLIC_KEY"], algorithms=[settings.GOOGLE_JWT["ALGORITHM"]])
 
         if payload.get("token_type") != "access":
             raise GoogleTokenInvalidError(AuthErrorMessages.GOOGLE_TOKEN_INVALID)
@@ -77,25 +69,26 @@ def validate_google_access_token(token: str) -> dict:
 
     except jwt.ExpiredSignatureError:
         raise GoogleTokenExpiredError()
-    except jwt.InvalidTokenError:
-        raise GoogleTokenInvalidError(AuthErrorMessages.GOOGLE_TOKEN_INVALID)
+    except jwt.InvalidTokenError as e:
+        raise GoogleTokenInvalidError(f"Invalid token: {str(e)}")
+    except Exception as e:
+        raise GoogleTokenInvalidError(f"Token validation failed: {str(e)}")
 
 
 def validate_google_refresh_token(token: str) -> dict:
     try:
-        payload = jwt.decode(
-            jwt=token, key=settings.GOOGLE_JWT["SECRET_KEY"], algorithms=[settings.GOOGLE_JWT["ALGORITHM"]]
-        )
-
+        payload = jwt.decode(jwt=token, key=settings.GOOGLE_JWT["PUBLIC_KEY"], algorithms=[settings.GOOGLE_JWT["ALGORITHM"]])
         if payload.get("token_type") != "refresh":
             raise GoogleTokenInvalidError(AuthErrorMessages.GOOGLE_TOKEN_INVALID)
 
         return payload
 
     except jwt.ExpiredSignatureError:
         raise GoogleRefreshTokenExpiredError()
-    except jwt.InvalidTokenError:
-        raise GoogleTokenInvalidError(AuthErrorMessages.GOOGLE_TOKEN_INVALID)
+    except jwt.InvalidTokenError as e:
+        raise GoogleTokenInvalidError(f"Invalid refresh token: {str(e)}")
+    except Exception as e:
+        raise GoogleTokenInvalidError(f"Refresh token validation failed: {str(e)}")
 
 
 def generate_google_token_pair(user_data: dict) -> dict:
diff --git a/todo/views/auth.py b/todo/views/auth.py
diff --git a/todo_project/settings/base.py b/todo_project/settings/base.py

Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,6 @@`
`5`	`5`	`from todo.views.auth import (`
`6`	`6`	`GoogleLoginView,`
`7`	`7`	`GoogleCallbackView,`
`8`		`- GoogleRefreshView,`
`9`	`8`	`GoogleLogoutView,`
`10`	`9`	`)`
`11`	`10`
`@@ -16,6 +15,5 @@`
`16`	`15`	`path("labels", LabelListView.as_view(), name="labels"),`
`17`	`16`	`path("auth/google/login/", GoogleLoginView.as_view(), name="google_login"),`
`18`	`17`	`path("auth/google/callback/", GoogleCallbackView.as_view(), name="google_callback"),`
`19`		`- path("auth/google/refresh/", GoogleRefreshView.as_view(), name="google_refresh"),`
`20`	`18`	`path("auth/google/logout/", GoogleLogoutView.as_view(), name="google_logout"),`
`21`	`19`	`]`