Check usernames in tokens match before refreshing access token #129

scigateway_auth/common/exceptions.py

@@ -45,6 +45,12 @@ class MaintenanceFileWriteError(Exception):
     """
 
 
+class UsernameMismatchError(Exception):
+    """
+    Exception raised when the usernames in the access and refresh tokens do not match.
+    """
+
+
 class UserNotAdminError(Exception):
     """
     Exception raised when a non-admin user performs an action that requires the user to be an admin.

scigateway_auth/routers/authentication.py

@@ -13,6 +13,7 @@
     ICATAuthenticationError,
     InvalidJWTError,
     JWTRefreshError,
+    UsernameMismatchError,
 )
 from scigateway_auth.common.schemas import LoginDetailsPostRequestSchema
 from scigateway_auth.src.authentication import ICATAuthenticator
@@ -97,7 +98,7 @@ def refresh_access_token(
     try:
         access_token = jwt_handler.refresh_access_token(token, refresh_token)
         return JSONResponse(content=access_token)
-    except (BlacklistedJWTError, InvalidJWTError, JWTRefreshError) as exc:
+    except (BlacklistedJWTError, InvalidJWTError, JWTRefreshError, UsernameMismatchError) as exc:
         message = "Unable to refresh access token"
         logger.exception(message)
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=message) from exc

scigateway_auth/src/jwt_handler.py

@@ -15,6 +15,7 @@
     BlacklistedJWTError,
     InvalidJWTError,
     JWTRefreshError,
+    UsernameMismatchError,
 )
 from scigateway_auth.src.authentication import ICATAuthenticator
 
@@ -52,7 +53,10 @@ def get_refresh_token(self, icat_username: str) -> str:
         """
         logger.info("Getting a refresh token")
         return self._pack_jwt(
-            {"exp": datetime.now(timezone.utc) + timedelta(days=config.authentication.refresh_token_validity_days)},
+            {
+                "username": icat_username,
+                "exp": datetime.now(timezone.utc) + timedelta(days=config.authentication.refresh_token_validity_days),
+            },
         )
 
     def refresh_access_token(self, access_token: str, refresh_token: str):
@@ -62,16 +66,20 @@ def refresh_access_token(self, access_token: str, refresh_token: str):
         :param access_token: The JWT access token to refresh.
         :param refresh_token: The JWT refresh token.
         :raises JWTRefreshError: If the JWT access token cannot be refreshed.
+        :raises UsernameMismatchError: If the usernames in the access and refresh tokens do not match
         :return: JWT access token with an updated expiry time.
         """
         logger.info("Refreshing access token")
 
         if self._is_refresh_token_blacklisted(refresh_token):
             raise BlacklistedJWTError(f"Attempted refresh from token in blacklist: {refresh_token}")
 
-        self.verify_token(refresh_token)
+        refresh_token_payload = self.verify_token(refresh_token)
         try:
             access_token_payload = self._get_jwt_payload(access_token, {"verify_exp": False})
+            if access_token_payload["username"] != refresh_token_payload["username"]:
+                raise UsernameMismatchError("The usernames in the access and refresh tokens do not match")
+
             access_token_payload["exp"] = datetime.now(timezone.utc) + timedelta(
                 minutes=config.authentication.access_token_validity_minutes,
             )