Partition ResponseCachingMiddleware cache by access token (#4041)

Co-authored-by: Claude <noreply@anthropic.com>

src/fastmcp/server/middleware/caching.py

@@ -20,6 +20,7 @@ from typing_extensions import NotRequired, Self, override
 
 from fastmcp.prompts.base import Message, Prompt, PromptResult
 from fastmcp.resources.base import Resource, ResourceContent, ResourceResult
+from fastmcp.server.dependencies import get_access_token
 from fastmcp.server.middleware.middleware import CallNext, Middleware, MiddlewareContext
 from fastmcp.tools.base import Tool, ToolResult
 from fastmcp.utilities.logging import get_logger
@@ -33,7 +34,7 @@ FIVE_MINUTES_IN_SECONDS = 300
 
 ONE_MB_IN_BYTES = 1024 * 1024
 
-GLOBAL_KEY = "__global__"
+ANONYMOUS_AUTH_KEY = "__anonymous__"
 
 
 class CachableResourceContent(FastMCPBaseModel):
@@ -194,7 +195,11 @@ class ResponseCachingMiddleware(Middleware):
 
     Notes:
     - Caches `tools/call`, `resources/read`, `prompts/get`, `tools/list`, `resources/list`, and `prompts/list` requests.
-    - Cache keys are derived from method name and arguments.
+    - Cache keys are derived from the method name, arguments, and the caller's
+      access token. Entries are partitioned per-token so that responses filtered
+      by per-component authorization (e.g. `auth=require_scopes(...)`) cannot
+      leak across users with different permissions. Unauthenticated callers
+      (including STDIO) share a single anonymous partition.
     """
 
     def __init__(
@@ -298,7 +303,9 @@ class ResponseCachingMiddleware(Middleware):
         if self._list_tools_settings.get("enabled") is False:
             return await call_next(context)
 
-        if cached_value := await self._list_tools_cache.get(key=GLOBAL_KEY):
+        cache_key: str = _get_auth_partition_key()
+
+        if cached_value := await self._list_tools_cache.get(key=cache_key):
             return cached_value
 
         tools: Sequence[Tool] = await call_next(context=context)
@@ -319,7 +326,7 @@ class ResponseCachingMiddleware(Middleware):
         ]
 
         await self._list_tools_cache.put(
-            key=GLOBAL_KEY,
+            key=cache_key,
             value=cachable_tools,
             ttl=self._list_tools_settings.get("ttl", FIVE_MINUTES_IN_SECONDS),
         )
@@ -337,7 +344,9 @@ class ResponseCachingMiddleware(Middleware):
         if self._list_resources_settings.get("enabled") is False:
             return await call_next(context)
 
-        if cached_value := await self._list_resources_cache.get(key=GLOBAL_KEY):
+        cache_key: str = _get_auth_partition_key()
+
+        if cached_value := await self._list_resources_cache.get(key=cache_key):
             return cached_value
 
         resources: Sequence[Resource] = await call_next(context=context)
@@ -358,7 +367,7 @@ class ResponseCachingMiddleware(Middleware):
         ]
 
         await self._list_resources_cache.put(
-            key=GLOBAL_KEY,
+            key=cache_key,
             value=cachable_resources,
             ttl=self._list_resources_settings.get("ttl", FIVE_MINUTES_IN_SECONDS),
         )
@@ -376,7 +385,9 @@ class ResponseCachingMiddleware(Middleware):
         if self._list_prompts_settings.get("enabled") is False:
             return await call_next(context)
 
-        if cached_value := await self._list_prompts_cache.get(key=GLOBAL_KEY):
+        cache_key: str = _get_auth_partition_key()
+
+        if cached_value := await self._list_prompts_cache.get(key=cache_key):
             return cached_value
 
         prompts: Sequence[Prompt] = await call_next(context=context)
@@ -395,7 +406,7 @@ class ResponseCachingMiddleware(Middleware):
         ]
 
         await self._list_prompts_cache.put(
-            key=GLOBAL_KEY,
+            key=cache_key,
             value=cachable_prompts,
             ttl=self._list_prompts_settings.get("ttl", FIVE_MINUTES_IN_SECONDS),
         )
@@ -417,7 +428,9 @@ class ResponseCachingMiddleware(Middleware):
         ) is False or not self._matches_tool_cache_settings(tool_name=tool_name):
             return await call_next(context=context)
 
-        cache_key: str = _make_call_tool_cache_key(msg=context.message)
+        cache_key: str = _make_call_tool_cache_key(
+            msg=context.message, auth_key=_get_auth_partition_key()
+        )
 
         if cached_value := await self._call_tool_cache.get(key=cache_key):
             return cached_value.unwrap()
@@ -446,7 +459,9 @@ class ResponseCachingMiddleware(Middleware):
         if self._read_resource_settings.get("enabled") is False:
             return await call_next(context=context)
 
-        cache_key: str = _make_read_resource_cache_key(msg=context.message)
+        cache_key: str = _make_read_resource_cache_key(
+            msg=context.message, auth_key=_get_auth_partition_key()
+        )
         cached_value: CachableResourceResult | None
 
         if cached_value := await self._read_resource_cache.get(key=cache_key):
@@ -474,7 +489,9 @@ class ResponseCachingMiddleware(Middleware):
         if self._get_prompt_settings.get("enabled") is False:
             return await call_next(context=context)
 
-        cache_key: str = _make_get_prompt_cache_key(msg=context.message)
+        cache_key: str = _make_get_prompt_cache_key(
+            msg=context.message, auth_key=_get_auth_partition_key()
+        )
 
         if cached_value := await self._get_prompt_cache.get(key=cache_key):
             return cached_value.unwrap()
@@ -534,19 +551,40 @@ def _hash_cache_key(value: str) -> str:
     return hashlib.sha256(value.encode()).hexdigest()
 
 
-def _make_call_tool_cache_key(msg: mcp.types.CallToolRequestParams) -> str:
+def _get_auth_partition_key() -> str:
+    """Return a stable, hashed identifier for the current access token.
+
+    Cache entries are partitioned by access token so that responses filtered
+    by per-component authorization (e.g. `auth=require_scopes(...)`) are not
+    leaked across users with different permissions. Unauthenticated callers
+    (including STDIO) share a single anonymous partition.
+    """
+
+    token = get_access_token()
+    if token is None:
+        return ANONYMOUS_AUTH_KEY
+    return _hash_cache_key(token.token)
+
+
+def _make_call_tool_cache_key(
+    msg: mcp.types.CallToolRequestParams, auth_key: str = ANONYMOUS_AUTH_KEY
+) -> str:
     """Make a cache key for a tool call using a stable hash of name and arguments."""
 
-    return _hash_cache_key(f"{msg.name}:{_get_arguments_str(msg.arguments)}")
+    return _hash_cache_key(f"{auth_key}:{msg.name}:{_get_arguments_str(msg.arguments)}")
 
 
-def _make_read_resource_cache_key(msg: mcp.types.ReadResourceRequestParams) -> str:
+def _make_read_resource_cache_key(
+    msg: mcp.types.ReadResourceRequestParams, auth_key: str = ANONYMOUS_AUTH_KEY
+) -> str:
     """Make a cache key for a resource read using a stable hash of URI."""
 
-    return _hash_cache_key(str(msg.uri))
+    return _hash_cache_key(f"{auth_key}:{msg.uri}")
 
 
-def _make_get_prompt_cache_key(msg: mcp.types.GetPromptRequestParams) -> str:
+def _make_get_prompt_cache_key(
+    msg: mcp.types.GetPromptRequestParams, auth_key: str = ANONYMOUS_AUTH_KEY
+) -> str:
     """Make a cache key for a prompt get using a stable hash of name and arguments."""
 
-    return _hash_cache_key(f"{msg.name}:{_get_arguments_str(msg.arguments)}")
+    return _hash_cache_key(f"{auth_key}:{msg.name}:{_get_arguments_str(msg.arguments)}")

tests/server/middleware/test_caching.py

@@ -31,6 +31,7 @@ from fastmcp.prompts.base import Message, Prompt
 from fastmcp.prompts.function_prompt import FunctionPrompt
 from fastmcp.resources.base import Resource
 from fastmcp.server.middleware.caching import (
+    ANONYMOUS_AUTH_KEY,
     CachableToolResult,
     CallToolSettings,
     ResponseCachingMiddleware,
@@ -671,3 +672,170 @@ class TestCacheKeyGeneration:
         assert len(key) == 64
         assert "ABC123" not in key
         assert key == _make_get_prompt_cache_key(msg)
+
+    def test_call_tool_key_partitions_by_auth(self):
+        msg = mcp.types.CallToolRequestParams(name="t", arguments={"a": 1})
+
+        anon = _make_call_tool_cache_key(msg)
+        user_a = _make_call_tool_cache_key(msg, auth_key="user_a")
+        user_b = _make_call_tool_cache_key(msg, auth_key="user_b")
+
+        assert anon == _make_call_tool_cache_key(msg, auth_key=ANONYMOUS_AUTH_KEY)
+        assert user_a != user_b
+        assert user_a != anon
+
+    def test_read_resource_key_partitions_by_auth(self):
+        msg = mcp.types.ReadResourceRequestParams(uri=AnyUrl("file:///tmp/x"))
+
+        user_a = _make_read_resource_cache_key(msg, auth_key="user_a")
+        user_b = _make_read_resource_cache_key(msg, auth_key="user_b")
+
+        assert user_a != user_b
+
+    def test_get_prompt_key_partitions_by_auth(self):
+        msg = mcp.types.GetPromptRequestParams(name="p", arguments={"a": "1"})
+
+        user_a = _make_get_prompt_cache_key(msg, auth_key="user_a")
+        user_b = _make_get_prompt_cache_key(msg, auth_key="user_b")
+
+        assert user_a != user_b
+
+
+class TestAuthAwareCaching:
+    """Cached responses must not leak across users with different auth tokens.
+
+    Regression tests for issue #4037: ResponseCachingMiddleware was caching
+    list/call responses with a global key, so a list filtered by per-component
+    auth checks for one user was served back to other users.
+    """
+
+    @staticmethod
+    def _make_token(scopes: list[str]):
+        from fastmcp.server.auth import AccessToken
+
+        return AccessToken(
+            token=f"token-{'-'.join(scopes) or 'none'}",
+            client_id="test-client",
+            scopes=scopes,
+            expires_at=None,
+            claims={},
+        )
+
+    @staticmethod
+    def _set_token(token):
+        from mcp.server.auth.middleware.auth_context import auth_context_var
+        from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+
+        if token is None:
+            return auth_context_var.set(None)
+        return auth_context_var.set(AuthenticatedUser(token))
+
+    async def test_list_tools_cache_does_not_leak_across_tokens(self):
+        from fastmcp.server.auth import require_scopes
+
+        mcp_server = FastMCP("test")
+        mcp_server.add_middleware(ResponseCachingMiddleware())
+
+        @mcp_server.tool(auth=require_scopes("read"))
+        def reader() -> str:
+            return "ok"
+
+        @mcp_server.tool(auth=require_scopes("read", "write"))
+        def writer() -> str:
+            return "ok"
+
+        from mcp.server.auth.middleware.auth_context import auth_context_var
+
+        # Privileged user lists tools first - both visible, gets cached.
+        privileged = self._make_token(["read", "write"])
+        tok = self._set_token(privileged)
+        try:
+            tools = await mcp_server.list_tools()
+            names = {t.name for t in tools}
+            assert names == {"reader", "writer"}
+        finally:
+            auth_context_var.reset(tok)
+
+        # Lower-privileged user must not see the cached privileged list.
+        limited = self._make_token(["read"])
+        tok = self._set_token(limited)
+        try:
+            tools = await mcp_server.list_tools()
+            names = {t.name for t in tools}
+            assert names == {"reader"}
+        finally:
+            auth_context_var.reset(tok)
+
+        # And same-token repeats still hit the cache (sanity check).
+        tok = self._set_token(limited)
+        try:
+            tools = await mcp_server.list_tools()
+            assert {t.name for t in tools} == {"reader"}
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_list_resources_cache_does_not_leak_across_tokens(self):
+        from fastmcp.server.auth import require_scopes
+
+        mcp_server = FastMCP("test")
+        mcp_server.add_middleware(ResponseCachingMiddleware())
+
+        @mcp_server.resource("data://public", auth=require_scopes("read"))
+        def public() -> str:
+            return "public"
+
+        @mcp_server.resource("data://secret", auth=require_scopes("read", "admin"))
+        def secret() -> str:
+            return "secret"
+
+        from mcp.server.auth.middleware.auth_context import auth_context_var
+
+        privileged = self._make_token(["read", "admin"])
+        tok = self._set_token(privileged)
+        try:
+            resources = await mcp_server.list_resources()
+            uris = {str(r.uri) for r in resources}
+            assert uris == {"data://public", "data://secret"}
+        finally:
+            auth_context_var.reset(tok)
+
+        limited = self._make_token(["read"])
+        tok = self._set_token(limited)
+        try:
+            resources = await mcp_server.list_resources()
+            uris = {str(r.uri) for r in resources}
+            assert uris == {"data://public"}
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_list_prompts_cache_does_not_leak_across_tokens(self):
+        from fastmcp.server.auth import require_scopes
+
+        mcp_server = FastMCP("test")
+        mcp_server.add_middleware(ResponseCachingMiddleware())
+
+        @mcp_server.prompt(auth=require_scopes("read"))
+        def public_prompt() -> str:
+            return "public"
+
+        @mcp_server.prompt(auth=require_scopes("read", "admin"))
+        def admin_prompt() -> str:
+            return "admin"
+
+        from mcp.server.auth.middleware.auth_context import auth_context_var
+
+        privileged = self._make_token(["read", "admin"])
+        tok = self._set_token(privileged)
+        try:
+            prompts = await mcp_server.list_prompts()
+            assert {p.name for p in prompts} == {"public_prompt", "admin_prompt"}
+        finally:
+            auth_context_var.reset(tok)
+
+        limited = self._make_token(["read"])
+        tok = self._set_token(limited)
+        try:
+            prompts = await mcp_server.list_prompts()
+            assert {p.name for p in prompts} == {"public_prompt"}
+        finally:
+            auth_context_var.reset(tok)