Evict user info cache entries when roles change

Author: cyrossignol

api/core/security.py

@@ -17,8 +17,10 @@
 # Set up logger for this module
 logger = get_logger(__name__)
 
-# TTL cache for token validation (1 hour TTL, max 1000 entries)
-_token_cache: cachetools.TTLCache[str, "UserInfo"] = cachetools.TTLCache(
+# TTL cache keyed by a user's OIDC subject. Evict entries when roles change. We
+# still validate the JWT signature and expiry on every request before reading a
+# cached record.
+_user_info_cache: cachetools.TTLCache[str, "UserInfo"] = cachetools.TTLCache(
     maxsize=1000, ttl=60 * 60
 )
 
@@ -41,6 +43,17 @@ async def close_tdei_client() -> None:
         _tdei_client = None
 
 
+def evict_user_from_cache(auth_uid: str) -> None:
+    """
+    Evict a user's cached UserInfo object so that their next request re-fetches
+    permissions.
+
+    Call this after modifying a user's roles in the OSM DB to ensure the change
+    takes effect on their next request rather than after the cache TTL expires.
+    """
+    _user_info_cache.pop(auth_uid, None)
+
+
 security = HTTPBearer()
 
 
@@ -151,9 +164,13 @@ async def validate_token(
     osm_db_session: AsyncSession = Depends(get_osm_db_session),
     task_db_session: AsyncSession = Depends(get_task_db_session),
 ) -> UserInfo:
-    """Dependency to get current authenticated user from TDEI/KeyCloak token and APIs.
+    """
+    Dependency that gets the current authenticated user from the TDEI/KeyCloak
+    access token and fetches permissions from TDEI APIs.
 
-    Results are cached by token for 1 hour to avoid repeated validation calls.
+    The JWT signature and expiry are validated on every request. The expensive
+    TDEI API and DB lookups are cached for 1 hour and should be evicted when a
+    user's role changes via evict_user_from_cache().
     """
     token = credentials.credentials
 
@@ -172,16 +189,16 @@ async def validate_token(
     if user_id is None:
         raise credentials_exception
 
-    # Check cache first
-    if token in _token_cache:
+    # Cache keyed by user ID so roles take effect immediately after eviction:
+    if user_id in _user_info_cache:
         logger.info("Token validation cache hit")
-        return _token_cache[token]
+        return _user_info_cache[user_id]
 
-    # Cache miss - perform full validation
+    # Cache miss: fetch TDEI roles and DB data:
     user_info = await _validate_token_uncached(
         token, user_id, payload, osm_db_session, task_db_session
     )
-    _token_cache[token] = user_info
+    _user_info_cache[user_id] = user_info
 
     return user_info
 

api/src/users/routes.py

@@ -4,7 +4,7 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from api.core.database import get_osm_session
-from api.core.security import UserInfo, validate_token
+from api.core.security import UserInfo, evict_user_from_cache, validate_token
 from api.src.users.repository import UserRepository
 from api.src.users.schemas import SetRoleRequest, WorkspaceUserRoleItem
 
@@ -48,6 +48,7 @@ async def assign_member_role(
         )
 
     await user_repo.assign_member_role(workspace_id, user_id, body.role)
+    evict_user_from_cache(str(user_id))
 
 
 @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
@@ -64,3 +65,4 @@ async def remove_member_role(
         )
 
     await user_repo.remove_member_role(workspace_id, user_id)
+    evict_user_from_cache(str(user_id))

api/src/workspaces/routes.py

@@ -5,7 +5,7 @@
 
 from api.core.database import get_osm_session, get_task_session
 from api.core.logging import get_logger
-from api.core.security import UserInfo, validate_token
+from api.core.security import UserInfo, evict_user_from_cache, validate_token
 from api.src.users.repository import UserRepository
 from api.src.users.schemas import WorkspaceUserRoleType
 from api.src.workspaces.repository import OSMRepository, WorkspaceRepository
@@ -148,6 +148,12 @@ async def create_workspace(
             WorkspaceUserRoleType.LEAD,
         )
 
+        # Evict the creator's cache so their next request reflects the new
+        # workspace and lead role rather than serving stale data for up to
+        # an hour:
+        #
+        evict_user_from_cache(str(current_user.user_uuid))
+
         return workspace
     except Exception as e:
         logger.error(f"Failed to create workspace: {str(e)}")