Refresh token

tests/integration/services/test_authentication_service.py

@@ -1,6 +1,7 @@
 from requests_mock import Mocker
 
 from trailblazer.services.authentication_service.authentication_service import AuthenticationService
+from trailblazer.services.encryption_service.encryption_service import EncryptionService
 from trailblazer.store.models import User
 
 
@@ -16,3 +17,22 @@ def test_exchange_code(authentication_service: AuthenticationService, user_email
     # THEN the refresh token is stored on the user
     user: User = authentication_service.store.get_user(user_email)
     assert user.refresh_token
+
+
+def test_refresh_access_token(
+    authentication_service: AuthenticationService,
+    encryption_service: EncryptionService,
+    user_email: str,
+):
+    # GIVEN an encrypted refresh token
+    encrypted_refresh_token: str = encryption_service.encrypt("refresh_token")
+
+    # GIVEN an existing user with the refresh token
+    user: User = authentication_service.store.get_user(user_email)
+    user.refresh_token = encrypted_refresh_token
+
+    # WHEN refreshing the access token
+    token: str = authentication_service.refresh_access_token(user.id)
+
+    # THEN an access token is returned
+    assert token

trailblazer/clients/authentication_client/dtos/refresh_token_request.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class RefreshAccessTokenRequest(BaseModel):
+    client_id: str
+    client_secret: str
+    refresh_token: str
+    grant_type: str = "refresh_token"

trailblazer/clients/authentication_client/google_oauth_client.py

@@ -1,5 +1,8 @@
 import requests
 
+from trailblazer.clients.authentication_client.dtos.refresh_token_request import (
+    RefreshAccessTokenRequest,
+)
 from trailblazer.clients.authentication_client.dtos.tokens_request import GetTokensRequest
 from trailblazer.clients.authentication_client.dtos.tokens_response import TokensResponse
 from trailblazer.clients.authentication_client.exceptions import GoogleOAuthClientError
@@ -29,3 +32,19 @@ def get_tokens(self, authorization_code: str) -> TokensResponse:
             raise GoogleOAuthClientError(response.text)
 
         return TokensResponse.model_validate(response.json())
+
+    def get_access_token(self, refresh_token: str) -> str:
+        """Use refresh token to get a new access token."""
+        request = RefreshAccessTokenRequest(
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            refresh_token=refresh_token,
+        )
+        data: str = request.model_dump_json()
+
+        response = requests.post(self.token_uri, data=data)
+
+        if not response.ok:
+            raise GoogleOAuthClientError(response.text)
+
+        return response.json()["access_token"]

trailblazer/server/api.py

@@ -42,7 +42,7 @@
 from trailblazer.services.authentication_service.authentication_service import AuthenticationService
 from trailblazer.services.authentication_service.exceptions import AuthenticationError
 from trailblazer.services.job_service import JobService
-from trailblazer.store.models import Info
+from trailblazer.store.models import Info, User
 
 blueprint = Blueprint("api", __name__, url_prefix="/api/v1")
 
@@ -82,6 +82,18 @@ def authenticate(auth_service: AuthenticationService = Provide[Container.auth_se
         return jsonify("User not allowed"), HTTPStatus.FORBIDDEN
 
 
+@blueprint.route("/auth/refresh", methods=["GET"])
+@inject
+def refresh_token(auth_service: AuthenticationService = Provide[Container.auth_service]):
+    """Refresh access token."""
+    user: User = g.current_user
+    try:
+        token: str = auth_service.refresh_access_token(user.id)
+        return jsonify({"access_token": token}), HTTPStatus.OK
+    except AuthenticationError:
+        return jsonify("User not allowed"), HTTPStatus.FORBIDDEN
+
+
 @blueprint.route("/analyses", methods=["GET"])
 @inject
 def get_analyses(analysis_service: AnalysisService = Provide[Container.analysis_service]):

trailblazer/services/authentication_service/authentication_service.py

@@ -34,3 +34,10 @@ def authenticate(self, authorization_code: str) -> str:
         self.store.update_user_token(user_id=user.id, refresh_token=encrypted_token)
 
         return tokens.access_token
+
+    def refresh_access_token(self, user_id: int) -> str:
+        """Refresh the users access token."""
+        user: User = self.store.get_user_by_id(user_id)
+        refresh_token: str = self.encryption_service.decrypt(user.refresh_token)
+        access_token: str = self.google_oauth_client.get_access_token(refresh_token)
+        return access_token