From 29f691731c07a79c59dda9c30f121f019c19f612 Mon Sep 17 00:00:00 2001
From: Jiang <huarch97@outlook.com>
Date: Wed, 5 Aug 2026 17:10:59 +0800
Subject: [PATCH] fix(auth): enforce Keycloak access token age

---
 .env.example                             |  1 +
 app/auth/keycloak_dependencies.py        | 14 ++++++--
 app/core/config.py                       |  1 +
 tests/auth/test_keycloak_dependencies.py | 45 +++++++++++++++++++++++-
 4 files changed, 58 insertions(+), 3 deletions(-)

diff --git a/.env.example b/.env.example
index 0f63d17..75f7c1b 100644
--- a/.env.example
+++ b/.env.example
@@ -45,6 +45,7 @@ METADATA_DB_PASSWORD="password"
 KEYCLOAK_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----"
 KEYCLOAK_ALGORITHM=RS256
 KEYCLOAK_AUDIENCE="account"
+KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS=900
 
 
 # ============================================
diff --git a/app/auth/keycloak_dependencies.py b/app/auth/keycloak_dependencies.py
index f99a358..2ddf5e9 100644
--- a/app/auth/keycloak_dependencies.py
+++ b/app/auth/keycloak_dependencies.py
@@ -1,4 +1,4 @@
-# import logging
+import time
 from uuid import UUID
 
 from fastapi import Depends, HTTPException, status
@@ -23,12 +23,22 @@ def _decode_keycloak_token(token: str) -> dict:
 
     key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
 
-    return jwt.decode(
+    payload = jwt.decode(
         token,
         key,
         algorithms=[settings.KEYCLOAK_ALGORITHM],
         audience=settings.KEYCLOAK_AUDIENCE or None,
     )
+    if settings.KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS <= 0:
+        return payload
+
+    issued_at = payload.get("iat")
+    if not isinstance(issued_at, (int, float)) or (
+        time.time() >= issued_at + settings.KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS
+    ):
+        raise JWTError("Keycloak access token is older than the allowed maximum age")
+
+    return payload
 
 
 async def get_current_keycloak_payload(
diff --git a/app/core/config.py b/app/core/config.py
index 521252a..9a521bf 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -53,6 +53,7 @@ class Settings(BaseSettings):
     KEYCLOAK_PUBLIC_KEY: str = ""
     KEYCLOAK_ALGORITHM: str = "RS256"
     KEYCLOAK_AUDIENCE: str = ""
+    KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS: int = 900
 
     # Bocha Web Search API
     BOCHA_API_KEY: str = ""
diff --git a/tests/auth/test_keycloak_dependencies.py b/tests/auth/test_keycloak_dependencies.py
index b7c6a24..18ba63d 100644
--- a/tests/auth/test_keycloak_dependencies.py
+++ b/tests/auth/test_keycloak_dependencies.py
@@ -1,7 +1,11 @@
 import pytest
 from fastapi import HTTPException
 
-from app.auth.keycloak_dependencies import get_current_keycloak_username
+from app.auth import keycloak_dependencies
+from app.auth.keycloak_dependencies import (
+    _decode_keycloak_token,
+    get_current_keycloak_username,
+)
 
 
 @pytest.fixture
@@ -28,3 +32,42 @@ async def test_current_username_rejects_username_fallback():
 
     assert exc.value.status_code == 401
     assert exc.value.detail == "Missing preferred_username claim"
+
+
+def test_decode_keycloak_token_rejects_a_token_older_than_the_configured_limit(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_PUBLIC_KEY", "public-key")
+    monkeypatch.setattr(
+        keycloak_dependencies.settings,
+        "KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS",
+        900,
+    )
+    monkeypatch.setattr(keycloak_dependencies.time, "time", lambda: 2_000)
+    monkeypatch.setattr(
+        keycloak_dependencies.jwt,
+        "decode",
+        lambda *args, **kwargs: {"iat": 1_000},
+    )
+
+    with pytest.raises(keycloak_dependencies.JWTError):
+        _decode_keycloak_token("expired-by-policy")
+
+
+def test_decode_keycloak_token_accepts_a_recent_token(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_PUBLIC_KEY", "public-key")
+    monkeypatch.setattr(
+        keycloak_dependencies.settings,
+        "KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS",
+        900,
+    )
+    monkeypatch.setattr(keycloak_dependencies.time, "time", lambda: 1_500)
+    monkeypatch.setattr(
+        keycloak_dependencies.jwt,
+        "decode",
+        lambda *args, **kwargs: {"iat": 1_000, "sub": "subject"},
+    )
+
+    assert _decode_keycloak_token("recent") == {"iat": 1_000, "sub": "subject"}