fix: adds RSA tokens and adjusted logic

.env.example

@@ -7,5 +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"
+
+# use if required
+# GOOGLE_JWT_ACCESS_LIFETIME="20"
+# GOOGLE_JWT_REFRESH_LIFETIME="30"

.github/workflows/test.yml

@@ -12,12 +12,10 @@ jobs:
     env:
       MONGODB_URI: mongodb://db:27017
       DB_NAME: todo-app
-      GOOGLE_JWT_SECRET_KEY: "test-secret-key-for-jwt"
       GOOGLE_JWT_ACCESS_LIFETIME: "3600"
       GOOGLE_JWT_REFRESH_LIFETIME: "604800"
       GOOGLE_OAUTH_CLIENT_ID: "test-client-id"
       GOOGLE_OAUTH_CLIENT_SECRET: "test-client-secret"
-      GOOGLE_OAUTH_REDIRECT_URI: "http://localhost:3000/auth/callback"
       COOKIE_SECURE: "False"
       COOKIE_SAMESITE: "Lax"
 

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,
@@ -73,25 +82,61 @@ def _try_google_auth(self, request) -> bool:
         try:
             google_token = request.COOKIES.get("ext-access")
 
-            if not 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
+
+            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_access_token(google_token)
+            payload = validate_google_refresh_token(refresh_token)
 
-            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"
+            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 (GoogleTokenExpiredError, GoogleTokenInvalidError) as e:
-            raise e
+        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 +157,25 @@ 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)
 

todo/tests/integration/base_mongo_test.py

@@ -20,8 +20,7 @@ def setUpClass(cls):
         cls.db = cls.mongo_client.get_database("testdb")
 
         cls.override = override_settings(
-            MONGODB_URI=cls.mongo_url,
-            DB_NAME="testdb",
+            MONGODB_URI=cls.mongo_url, DB_NAME="testdb", FRONTEND_URL="http://localhost:4000"
         )
         cls.override.enable()
         DatabaseManager.reset()

todo/tests/unit/middlewares/test_jwt_auth.py

@@ -1,7 +1,6 @@
 from unittest import TestCase
 from unittest.mock import Mock, patch
 from django.http import HttpRequest, JsonResponse
-from django.conf import settings
 from rest_framework import status
 import json
 
@@ -17,9 +16,6 @@ def setUp(self):
         self.request.path = "/v1/tasks"
         self.request.headers = {}
         self.request.COOKIES = {}
-        self._original_public_paths = settings.PUBLIC_PATHS
-        settings.PUBLIC_PATHS = ["/v1/auth/google/login"]
-        self.addCleanup(setattr, settings, "PUBLIC_PATHS", self._original_public_paths)
 
     def test_public_path_authentication_bypass(self):
         """Test that requests to public paths bypass authentication"""