feat: short-lived access & refresh tokens for web/admin clients

Author: sricharan-varanasi

src/apps/authentication/services/security.py

@@ -50,8 +50,12 @@ def token_expiration_minutes(
     @staticmethod
     def create_access_token(data: dict) -> str:
         to_encode = data.copy()
-        expires_delta = timedelta(minutes=settings.authentication.access_token.expiration)
-        expire = datetime.now(timezone.utc) + expires_delta
+        minutes = AuthenticationService.token_expiration_minutes(
+            to_encode.get(JWTClaim.client),
+            settings.authentication.access_token.expiration,
+            settings.authentication.access_token.web_admin_expiration,
+        )
+        expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
         to_encode.setdefault(JWTClaim.exp, expire)
         to_encode.setdefault(JWTClaim.jti, str(uuid.uuid4()))
         encoded_jwt = jwt.encode(
@@ -130,8 +134,12 @@ def decode_mfa_token(token: str) -> str:
     @staticmethod
     def create_refresh_token(data: dict) -> str:
         to_encode = data.copy()
-        expires_delta = timedelta(minutes=settings.authentication.refresh_token.expiration)
-        expire = datetime.now(timezone.utc) + expires_delta
+        minutes = AuthenticationService.token_expiration_minutes(
+            to_encode.get(JWTClaim.client),
+            settings.authentication.refresh_token.expiration,
+            settings.authentication.refresh_token.web_admin_expiration,
+        )
+        expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
         to_encode.setdefault(JWTClaim.exp, expire)
         to_encode.setdefault(JWTClaim.jti, str(uuid.uuid4()))
         encoded_jwt = jwt.encode(

src/apps/authentication/tests/test_auth.py

@@ -505,3 +505,89 @@ async def test_login_without_client_claim(self, client: TestClient, user: User,
         assert "client" not in access_payload
         assert "client" not in refresh_payload
         self._assert_lifetimes_unchanged(access_payload, refresh_payload, before, after)
+
+
+class TestShortLivedWebAdminTokens(BaseTest):
+    """Web/admin clients get shorter token lifetimes when configured; others are unchanged."""
+
+    get_token_url = auth_router.url_path_for("get_token")
+    refresh_access_token_url = auth_router.url_path_for("refresh_access_token")
+
+    SHORT_ACCESS = 15
+    SHORT_REFRESH = 120
+
+    @pytest.fixture
+    def short_lifetimes(self, mocker: MockerFixture):
+        mocker.patch.object(settings.authentication.access_token, "web_admin_expiration", self.SHORT_ACCESS)
+        mocker.patch.object(settings.authentication.refresh_token, "web_admin_expiration", self.SHORT_REFRESH)
+
+    @staticmethod
+    def _decode(token: str, secret: str) -> dict:
+        return jwt.decode(token, secret, algorithms=[settings.authentication.algorithm])
+
+    @staticmethod
+    def _assert_expires_in(exp: int, minutes: int, before: datetime.datetime, after: datetime.datetime):
+        delta = datetime.timedelta(minutes=minutes)
+        assert int((before + delta).timestamp()) <= exp <= int((after + delta).timestamp()) + 1
+
+    @pytest.mark.parametrize("content_source", ("web", "admin"))
+    async def test_web_admin_get_short_lifetimes(
+        self, client: TestClient, user: User, short_lifetimes, content_source: str
+    ):
+        before = datetime.datetime.now(datetime.timezone.utc)
+        resp = await client.post(
+            self.get_token_url,
+            data={"email": user.email_encrypted, "password": TEST_PASSWORD},
+            headers={"Mindlogger-Content-Source": content_source},
+        )
+        after = datetime.datetime.now(datetime.timezone.utc)
+        assert resp.status_code == http.HTTPStatus.OK
+        token = resp.json()["result"]["token"]
+        access = self._decode(token["accessToken"], settings.authentication.access_token.secret_key)
+        refresh = self._decode(token["refreshToken"], settings.authentication.refresh_token.secret_key)
+        self._assert_expires_in(access["exp"], self.SHORT_ACCESS, before, after)
+        self._assert_expires_in(refresh["exp"], self.SHORT_REFRESH, before, after)
+
+    @pytest.mark.parametrize(
+        "headers",
+        ({"Mindlogger-Content-Source": "mobile"}, None),
+        ids=("mobile", "no-header"),
+    )
+    async def test_mobile_and_unknown_keep_default_lifetimes(
+        self, client: TestClient, user: User, short_lifetimes, headers: dict | None
+    ):
+        before = datetime.datetime.now(datetime.timezone.utc)
+        resp = await client.post(
+            self.get_token_url,
+            data={"email": user.email_encrypted, "password": TEST_PASSWORD},
+            headers=headers,
+        )
+        after = datetime.datetime.now(datetime.timezone.utc)
+        assert resp.status_code == http.HTTPStatus.OK
+        token = resp.json()["result"]["token"]
+        access = self._decode(token["accessToken"], settings.authentication.access_token.secret_key)
+        refresh = self._decode(token["refreshToken"], settings.authentication.refresh_token.secret_key)
+        self._assert_expires_in(access["exp"], settings.authentication.access_token.expiration, before, after)
+        self._assert_expires_in(refresh["exp"], settings.authentication.refresh_token.expiration, before, after)
+
+    async def test_refresh_preserves_short_access_lifetime(
+        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"},
+        )
+        refresh_token = login.json()["result"]["token"]["refreshToken"]
+
+        before = datetime.datetime.now(datetime.timezone.utc)
+        resp = await client.post(
+            self.refresh_access_token_url,
+            data={"refresh_token": refresh_token},
+            headers={"Mindlogger-Content-Source": "admin"},
+        )
+        after = datetime.datetime.now(datetime.timezone.utc)
+        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)

src/apps/authentication/tests/test_mfa_flow.py

@@ -2,7 +2,7 @@
 
 import http
 import json
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from unittest.mock import AsyncMock, patch
 
 import jwt
@@ -309,6 +309,42 @@ async def test_verify_mfa_embeds_client_claim(
         assert access_payload["client"] == "admin"
         assert refresh_payload["client"] == "admin"
 
+    async def test_verify_mfa_applies_short_lifetime_for_web_admin(
+        self, client: TestClient, user_with_mfa: User, session: AsyncSession, mocker: MockerFixture
+    ):
+        """Tokens issued via MFA verification honor the short web/admin lifetime."""
+        short_access = 15
+        mocker.patch.object(settings.authentication.access_token, "web_admin_expiration", short_access)
+
+        login_response = await client.post(
+            url=self.get_token_url,
+            data=dict(email=user_with_mfa.email_encrypted, password=TEST_PASSWORD),
+        )
+        mfa_token = login_response.json()["result"]["mfaToken"]
+
+        crud = UsersCRUD(session)
+        fresh_user = await crud.get_by_id(user_with_mfa.id)
+        assert fresh_user is not None
+        assert fresh_user.mfa_secret is not None
+        valid_code = totp_service.get_current_code(totp_service.decrypt_secret(fresh_user.mfa_secret))
+
+        mocker.patch("apps.authentication.api.auth.log")
+        before = datetime.now(timezone.utc)
+        verify_response = await client.post(
+            url=self.verify_mfa_url,
+            data=dict(mfaToken=mfa_token, totpCode=valid_code),
+            headers={"Mindlogger-Content-Source": "admin"},
+        )
+        after = datetime.now(timezone.utc)
+        assert verify_response.status_code == http.HTTPStatus.OK
+        access_payload = jwt.decode(
+            verify_response.json()["result"]["token"]["accessToken"],
+            settings.authentication.access_token.secret_key,
+            algorithms=[settings.authentication.algorithm],
+        )
+        delta = timedelta(minutes=short_access)
+        assert int((before + delta).timestamp()) <= access_payload["exp"] <= int((after + delta).timestamp()) + 1
+
     async def test_verify_mfa_with_invalid_code_fails(
         self, client: TestClient, user_with_mfa: User, mocker: MockerFixture
     ):