fix: keep refresh-token revocation exp correct under per-client lifetimes

Author: sricharan-varanasi

src/apps/authentication/services/security.py

@@ -176,9 +176,27 @@ def _get_refresh_token_by_access(self, token: InternalToken) -> InternalToken |
         if not token.payload.rjti:
             return None
 
+        # Reconstruct the paired refresh token's exp from the access token's exp. The
+        # deltas must match the per-client lifetimes the tokens were actually minted with
+        # (via the `client` claim), otherwise a short web/admin token would under-estimate
+        # the refresh exp and its blacklist entry could be purged before the refresh token
+        # truly expires — letting a revoked token be used again.
+        client = token.payload.client
         access_exp = datetime.fromtimestamp(token.payload.exp, timezone.utc)
-        refresh_expires_delta = timedelta(minutes=settings.authentication.refresh_token.expiration)
-        access_expires_delta = timedelta(minutes=settings.authentication.access_token.expiration)
+        refresh_expires_delta = timedelta(
+            minutes=self.token_expiration_minutes(
+                client,
+                settings.authentication.refresh_token.expiration,
+                settings.authentication.refresh_token.web_admin_expiration,
+            )
+        )
+        access_expires_delta = timedelta(
+            minutes=self.token_expiration_minutes(
+                client,
+                settings.authentication.access_token.expiration,
+                settings.authentication.access_token.web_admin_expiration,
+            )
+        )
         expire = access_exp - access_expires_delta + refresh_expires_delta
         refresh_token = InternalToken(
             payload=TokenPayload(

src/apps/authentication/tests/test_auth.py

@@ -591,3 +591,28 @@ async def test_refresh_preserves_short_access_lifetime(
         assert resp.status_code == http.HTTPStatus.OK
         access = self._decode(resp.json()["result"]["accessToken"], settings.authentication.access_token.secret_key)
         self._assert_expires_in(access["exp"], self.SHORT_ACCESS, before, after)
+
+    async def test_logout_revokes_paired_refresh_token_under_short_lifetimes(
+        self, client: TestClient, user: User, short_lifetimes, mocker: MockerFixture
+    ):
+        mocker.patch("apps.authentication.api.auth.log")
+        login = await client.post(
+            self.get_token_url,
+            data={"email": user.email_encrypted, "password": TEST_PASSWORD},
+            headers={"Mindlogger-Content-Source": "admin"},
+        )
+        token = login.json()["result"]["token"]
+
+        logout = await client.post(
+            auth_router.url_path_for("delete_access_token"),
+            headers={"Authorization": f"Bearer {token['accessToken']}"},
+        )
+        assert logout.status_code == http.HTTPStatus.OK
+
+        # Revoking the access token must also revoke its paired refresh token.
+        resp = await client.post(
+            self.refresh_access_token_url,
+            data={"refresh_token": token["refreshToken"]},
+        )
+        assert resp.status_code == http.HTTPStatus.UNAUTHORIZED
+        assert resp.json()["result"][0]["message"] == AuthenticationError.message

src/apps/authentication/tests/unit/test_auth_service.py

@@ -1,4 +1,5 @@
 import uuid
+from datetime import datetime, timedelta, timezone
 
 import pytest
 from pytest import FixtureRequest
@@ -7,13 +8,15 @@
 
 from apps.authentication.domain.login import UserLoginRequest
 from apps.authentication.domain.token import InternalToken
-from apps.authentication.domain.token.internal import TokenPurpose
+from apps.authentication.domain.token.internal import TokenPayload, TokenPurpose
 from apps.authentication.errors import BadCredentials, InvalidCredentials
 from apps.authentication.services import AuthenticationService
 from apps.authentication.services.core import TokensService
 from apps.users.cruds.user import UsersCRUD
 from apps.users.domain import User
 from apps.users.errors import UserIsDeletedError, UserNotFound
+from config import settings
+from infrastructure.http.domain import MindloggerContentSource
 
 TEST_PASSWORD = "Test12345!"
 RJTI = str(uuid.uuid4())
@@ -154,3 +157,47 @@ async def test_token_revoke__ttl_less_than_one(
         await token_blacklist_service.revoke(access_token_internal, TokenPurpose.ACCESS)
         is_revoked = await token_blacklist_service.is_revoked(access_token_internal)
         assert not is_revoked
+
+
+class TestRefreshTokenExpDerivation:
+    """`_get_refresh_token_by_access` must derive the paired refresh exp using the
+    same per-client lifetimes the tokens were minted with (keyed on the `client` claim)."""
+
+    @staticmethod
+    def _access_token(access_exp: datetime, client: MindloggerContentSource | None) -> InternalToken:
+        return InternalToken(
+            payload=TokenPayload(
+                sub=uuid.uuid4(),
+                exp=int(access_exp.timestamp()),
+                jti=str(uuid.uuid4()),
+                rjti=RJTI,
+                client=client,
+            )
+        )
+
+    def test_web_admin_uses_short_deltas(self, auth_service: AuthenticationService, mocker: MockerFixture):
+        mocker.patch.object(settings.authentication.access_token, "web_admin_expiration", 15)
+        mocker.patch.object(settings.authentication.refresh_token, "web_admin_expiration", 120)
+        access_exp = datetime.now(timezone.utc) + timedelta(minutes=15)
+        derived = auth_service._get_refresh_token_by_access(
+            self._access_token(access_exp, MindloggerContentSource.admin)
+        )
+        assert derived is not None
+        expected = int((access_exp - timedelta(minutes=15) + timedelta(minutes=120)).timestamp())
+        assert derived.payload.exp == expected
+        assert derived.payload.jti == RJTI
+
+    def test_legacy_client_none_uses_defaults(self, auth_service: AuthenticationService, mocker: MockerFixture):
+        mocker.patch.object(settings.authentication.access_token, "web_admin_expiration", 15)
+        mocker.patch.object(settings.authentication.refresh_token, "web_admin_expiration", 120)
+        access_exp = datetime.now(timezone.utc) + timedelta(minutes=settings.authentication.access_token.expiration)
+        derived = auth_service._get_refresh_token_by_access(self._access_token(access_exp, None))
+        assert derived is not None
+        expected = int(
+            (
+                access_exp
+                - timedelta(minutes=settings.authentication.access_token.expiration)
+                + timedelta(minutes=settings.authentication.refresh_token.expiration)
+            ).timestamp()
+        )
+        assert derived.payload.exp == expected