feat: Session token rotation(M2-11012)

.env.default

@@ -41,6 +41,9 @@ AUTHENTICATION__REFRESH_TOKEN__EXPIRATION=540
 # per environment if needed.
 #AUTHENTICATION__ACCESS_TOKEN__WEB_ADMIN_EXPIRATION=15
 #AUTHENTICATION__REFRESH_TOKEN__WEB_ADMIN_EXPIRATION=30
+# Grace period (seconds) after a web/admin refresh token rotates during which the old
+# token still redeems for the same replacement pair (default 60 in code).
+#AUTHENTICATION__REFRESH_TOKEN__ROTATION_GRACE_SECONDS=60
 AUTHENTICATION__ALGORITHM="HS256"
 AUTHENTICATION__TOKEN_TYPE="Bearer"
 AUTHENTICATION__PASSWORD_RECOVER__EXPIRATION=900

src/apps/authentication/api/auth.py

@@ -35,6 +35,7 @@
 from apps.authentication.services.mfa_notifications import MFANotificationService
 from apps.authentication.services.mfa_session import MFASessionService
 from apps.authentication.services.recovery_codes import send_recovery_code_notifications, verify_recovery_code_service
+from apps.authentication.services.rotation import TokenRotationService
 from apps.authentication.services.security import AuthenticationService
 from apps.shared.domain.response import Response
 from apps.shared.exception import BaseError
@@ -57,6 +58,16 @@ def client_token_claims(content_source: MindloggerContentSource | None) -> dict:
     return {JWTClaim.client: content_source} if content_source else {}
 
 
+async def revoke_token_family_if_web_admin(session, token: InternalToken) -> None:
+    """On logout of a rotating (web/admin) token, revoke its whole family so a superseded
+    refresh token in the same chain cannot keep the session alive."""
+    if token.payload.family and token.payload.client in (
+        MindloggerContentSource.web,
+        MindloggerContentSource.admin,
+    ):
+        await TokenRotationService(session).revoke_family(token.payload.family, token.payload.sub)
+
+
 async def get_token(
     request: Request,
     user_login_schema: UserLoginRequest = Body(...),
@@ -113,13 +124,14 @@ async def get_token(
 
     rjti = str(uuid.uuid4())
     refresh_token = AuthenticationService.create_refresh_token(
-        {JWTClaim.sub: str(user.id), JWTClaim.jti: rjti, **client_token_claims(content_source)}
+        {JWTClaim.sub: str(user.id), JWTClaim.jti: rjti, JWTClaim.family: rjti, **client_token_claims(content_source)}
     )
 
     access_token = AuthenticationService.create_access_token(
         {
             JWTClaim.sub: str(user.id),
             JWTClaim.rjti: rjti,
+            JWTClaim.family: rjti,
             **client_token_claims(content_source),
         }
     )
@@ -308,13 +320,19 @@ async def verify_mfa_totp(
             # Issue refresh and access tokens
             rjti = str(uuid.uuid4())
             refresh_token = AuthenticationService.create_refresh_token(
-                {JWTClaim.sub: str(user.id), JWTClaim.jti: rjti, **client_token_claims(content_source)}
+                {
+                    JWTClaim.sub: str(user.id),
+                    JWTClaim.jti: rjti,
+                    JWTClaim.family: rjti,
+                    **client_token_claims(content_source),
+                }
             )
 
             access_token = AuthenticationService.create_access_token(
                 {
                     JWTClaim.sub: str(user.id),
                     JWTClaim.rjti: rjti,
+                    JWTClaim.family: rjti,
                     **client_token_claims(content_source),
                 }
             )
@@ -585,13 +603,19 @@ async def verify_mfa_recovery_code(
             # Step 6: Issue refresh and access tokens
             rjti = str(uuid.uuid4())
             refresh_token = AuthenticationService.create_refresh_token(
-                {JWTClaim.sub: str(user_id), JWTClaim.jti: rjti, **client_token_claims(content_source)}
+                {
+                    JWTClaim.sub: str(user_id),
+                    JWTClaim.jti: rjti,
+                    JWTClaim.family: rjti,
+                    **client_token_claims(content_source),
+                }
             )
 
             access_token = AuthenticationService.create_access_token(
                 {
                     JWTClaim.sub: str(user_id),
                     JWTClaim.rjti: rjti,
+                    JWTClaim.family: rjti,
                     **client_token_claims(content_source),
                 }
             )
@@ -640,6 +664,8 @@ async def refresh_access_token(
 ) -> Response[Token]:
     """Refresh access token."""
     user_id: uuid.UUID | None = None
+    reuse_family: str | None = None
+    refresh_outcome = "reused"
     try:
         async with atomic(session):
             try:
@@ -671,37 +697,94 @@ async def refresh_access_token(
                 raise InvalidRefreshToken() from e
 
             user_id = token_data.sub
+            family = token_data.family or token_data.jti
+            is_web_admin = token_data.client in (MindloggerContentSource.web, MindloggerContentSource.admin)
+
+            if is_web_admin:
+                # Rotating clients: slide the refresh window by issuing a fresh token each time,
+                # with a grace window that idempotently redeems the old token, and reuse detection
+                # that revokes the whole family.
+                rotation = TokenRotationService(session)
+
+                if await rotation.is_family_revoked(family):
+                    raise AuthenticationError
+
+                replacement = await rotation.get_rotation_replacement(token_data.jti)
+                if replacement is not None:
+                    # Within the grace window: hand back the same replacement pair.
+                    access_token = replacement.access_token
+                    refresh_token = replacement.refresh_token
+                    refresh_outcome = "grace_redeemed"
+                elif await AuthenticationService(session).is_revoked(InternalToken(payload=token_data)):
+                    # Old token replayed after its grace window -> treat as theft. Defer the
+                    # family revocation to its own committed transaction (raising here would roll
+                    # back this atomic block and undo it).
+                    reuse_family = family
+                else:
+                    new_rjti = str(uuid.uuid4())
+                    refresh_token = AuthenticationService.create_refresh_token(
+                        {
+                            JWTClaim.sub: str(user_id),
+                            JWTClaim.jti: new_rjti,
+                            JWTClaim.family: family,
+                            **client_token_claims(token_data.client),
+                        }
+                    )
+                    access_token = AuthenticationService.create_access_token(
+                        {
+                            JWTClaim.sub: str(user_id),
+                            JWTClaim.rjti: new_rjti,
+                            JWTClaim.family: family,
+                            **client_token_claims(token_data.client),
+                        }
+                    )
+                    # Mark the old refresh token used, and record the replacement for the grace window.
+                    await AuthenticationService(session).revoke_token(
+                        InternalToken(payload=token_data), TokenPurpose.REFRESH
+                    )
+                    await rotation.store_rotation_record(
+                        token_data.jti,
+                        Token(access_token=access_token, refresh_token=refresh_token),
+                    )
+                    refresh_outcome = "rotated"
+            else:
+                # Mobile / unknown / legacy: reuse the same refresh token (unchanged behavior).
+                revoked = await AuthenticationService(session).is_revoked(InternalToken(payload=token_data))
+                if revoked:
+                    raise AuthenticationError
+
+                rjti = token_data.jti
+                refresh_token = schema.refresh_token
+                if regenerate_refresh_token:
+                    # blacklist current refresh token
+                    await AuthenticationService(session).revoke_token(
+                        InternalToken(payload=token_data), TokenPurpose.REFRESH
+                    )
 
-            # Check if the token is in the blacklist
-            revoked = await AuthenticationService(session).is_revoked(InternalToken(payload=token_data))
-            if revoked:
-                raise AuthenticationError
-
-            rjti = token_data.jti
-            refresh_token = schema.refresh_token
-            if regenerate_refresh_token:
-                # blacklist current refresh token
-                await AuthenticationService(session).revoke_token(
-                    InternalToken(payload=token_data), TokenPurpose.REFRESH
-                )
+                    rjti = str(uuid.uuid4())
+                    refresh_token = AuthenticationService.create_refresh_token(
+                        {
+                            JWTClaim.sub: str(user_id),
+                            JWTClaim.jti: rjti,
+                            JWTClaim.exp: token_data.exp,
+                            **client_token_claims(token_data.client),
+                        }
+                    )
 
-                rjti = str(uuid.uuid4())
-                refresh_token = AuthenticationService.create_refresh_token(
+                access_token = AuthenticationService.create_access_token(
                     {
                         JWTClaim.sub: str(user_id),
-                        JWTClaim.jti: rjti,
-                        JWTClaim.exp: token_data.exp,
+                        JWTClaim.rjti: rjti,
                         **client_token_claims(token_data.client),
                     }
                 )
 
-            access_token = AuthenticationService.create_access_token(
-                {
-                    JWTClaim.sub: str(user_id),
-                    JWTClaim.rjti: rjti,
-                    **client_token_claims(token_data.client),
-                }
-            )
+        if reuse_family is not None:
+            # Commit the family revocation in its own transaction, then reject the request.
+            logger.warning(f"Refresh token reuse detected; revoking family user_id={user_id} family={reuse_family}")
+            async with atomic(session):
+                await TokenRotationService(session).revoke_family(reuse_family, user_id)
+            raise AuthenticationError
     except BaseError as e:
         await log(
             AuditEvent(
@@ -712,6 +795,7 @@ async def refresh_access_token(
         )
         raise
 
+    logger.info(f"Token refresh succeeded user_id={user_id} outcome={refresh_outcome}")
     await log(
         AuditEvent(
             event_action=EventAction.USER_SESSION_REFRESH,
@@ -733,6 +817,7 @@ async def delete_access_token(
     try:
         async with atomic(session):
             await AuthenticationService(session).revoke_token(token, TokenPurpose.ACCESS)
+            await revoke_token_family_if_web_admin(session, token)
         async with atomic(session):
             if schema and schema.device_id:
                 await UserDeviceService(session, user.id).remove_device(schema.device_id)
@@ -764,6 +849,7 @@ async def delete_refresh_token(
     """Add token to the blacklist."""
     async with atomic(session):
         await AuthenticationService(session).revoke_token(token, TokenPurpose.REFRESH)
+        await revoke_token_family_if_web_admin(session, token)
     if schema and schema.device_id:
         async with atomic(session):
             await UserDeviceService(session, token.payload.sub).remove_device(schema.device_id)

src/apps/authentication/domain/token/internal.py

@@ -23,6 +23,7 @@ class JWTClaim(StrEnum):
     rjti = "rjti"
     mfa_session_id = "mfa_session_id"
     client = "client"
+    family = "family"
 
 
 class TokenPayload(InternalModel):
@@ -34,6 +35,10 @@ class TokenPayload(InternalModel):
     # None for tokens issued before the claim existed or to clients that do not
     # send the header
     client: MindloggerContentSource | None = None
+    # Token family (the login refresh token's jti). Shared by every access/refresh
+    # token descended from one login, so a whole rotated chain can be revoked at once.
+    # None for tokens issued before the claim existed.
+    family: str | None = None
 
 
 class InternalToken(InternalModel):

src/apps/authentication/services/rotation.py

@@ -0,0 +1,63 @@
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from apps.authentication.crud import TokenBlacklistCRUD
+from apps.authentication.domain.token import InternalToken, Token, TokenPayload, TokenPurpose
+from apps.authentication.services.core import TokensService
+from config import settings
+from infrastructure.utility.redis_client import RedisCache
+
+__all__ = ["TokenRotationService"]
+
+
+class TokenRotationService:
+    """Refresh-token rotation support for web/admin clients.
+
+    Two responsibilities:
+    - a short-lived Redis "grace record" mapping a just-rotated refresh token's jti to
+      the replacement token pair, so the old token redeems idempotently for a brief window;
+    - family revocation via the existing token blacklist, keyed by the family id, so a whole
+      rotated chain can be killed at once (on reuse detection or logout).
+    """
+
+    def __init__(self, session) -> None:
+        self.session = session
+        self.redis_client = RedisCache()
+
+    @staticmethod
+    def _grace_key(old_jti: str) -> str:
+        return f"token_rotation:{old_jti}"
+
+    @staticmethod
+    def _family_blacklist_jti(family_id: str) -> str:
+        # Namespaced so a family-revocation row never collides with a real token jti
+        # (the family id equals the login refresh token's jti, which itself gets
+        # blacklisted on its first rotation).
+        return f"family:{family_id}"
+
+    async def get_rotation_replacement(self, old_jti: str) -> Token | None:
+        raw = await self.redis_client.get(self._grace_key(old_jti))
+        if not raw:
+            return None
+        data = json.loads(raw)
+        return Token(access_token=data["access_token"], refresh_token=data["refresh_token"])
+
+    async def store_rotation_record(self, old_jti: str, token: Token) -> None:
+        await self.redis_client.set(
+            self._grace_key(old_jti),
+            json.dumps({"access_token": token.access_token, "refresh_token": token.refresh_token}),
+            ex=settings.authentication.refresh_token.rotation_grace_seconds,
+        )
+
+    async def is_family_revoked(self, family_id: str) -> bool:
+        return await TokenBlacklistCRUD(self.session).exist_by_key("jti", self._family_blacklist_jti(family_id))
+
+    async def revoke_family(self, family_id: str, user_id: uuid.UUID) -> None:
+        # Blacklist a synthetic row under the namespaced family jti, retained comfortably past
+        # any live token in the family (refresh lifetime is the longest a token can survive).
+        expire = datetime.now(timezone.utc) + timedelta(minutes=settings.authentication.refresh_token.expiration)
+        family_token = InternalToken(
+            payload=TokenPayload(sub=user_id, exp=int(expire.timestamp()), jti=self._family_blacklist_jti(family_id))
+        )
+        await TokensService(self.session).revoke(family_token, TokenPurpose.REFRESH)