Fix versioned auth middleware checks (#4401)

fastmcp_slim/fastmcp/server/middleware/authorization.py

@@ -44,10 +44,37 @@ from fastmcp.server.middleware.middleware import (
     MiddlewareContext,
 )
 from fastmcp.tools.base import Tool, ToolResult
+from fastmcp.utilities.versions import VersionSpec
 
 logger = logging.getLogger(__name__)
 
 
+def _requested_version(meta: mt.RequestParams.Meta | None) -> VersionSpec | None:
+    if meta is None:
+        return None
+
+    meta_dict = meta.model_dump(exclude_none=True)
+    fastmcp_meta = meta_dict.get("fastmcp")
+    if not isinstance(fastmcp_meta, dict):
+        return None
+
+    version = fastmcp_meta.get("version")
+    if isinstance(version, str):
+        return VersionSpec(eq=version)
+
+    if isinstance(version, dict):
+        gte = version.get("gte")
+        lt = version.get("lt")
+        eq = version.get("eq")
+
+        if not all(value is None or isinstance(value, str) for value in (gte, lt, eq)):
+            return None
+
+        return VersionSpec(gte=gte, lt=lt, eq=eq)
+
+    return None
+
+
 class AuthMiddleware(Middleware):
     """Global authorization middleware using callable checks.
 
@@ -140,7 +167,8 @@ class AuthMiddleware(Middleware):
         # component-level auth denied access, so the two cases are
         # indistinguishable here. Keep the message ambiguous to avoid
         # disclosing existence of tools the caller is not authorized to see.
-        tool = await fastmcp.fastmcp.get_tool(tool_name)
+        version = _requested_version(context.message.meta)
+        tool = await fastmcp.fastmcp.get_tool(tool_name, version=version)
         if tool is None:
             raise AuthorizationError(
                 f"Authorization failed for tool '{tool_name}': "
@@ -212,9 +240,13 @@ class AuthMiddleware(Middleware):
         # does not exist and when component-level auth denied access, so the two
         # cases are indistinguishable here. Keep the message ambiguous to avoid
         # disclosing existence of resources the caller is not authorized to see.
-        component = await fastmcp.fastmcp.get_resource(str(uri))
+        version = _requested_version(context.message.meta)
+        component = await fastmcp.fastmcp.get_resource(str(uri), version=version)
         if component is None:
-            component = await fastmcp.fastmcp.get_resource_template(str(uri))
+            component = await fastmcp.fastmcp.get_resource_template(
+                str(uri),
+                version=version,
+            )
         if component is None:
             raise AuthorizationError(
                 f"Authorization failed for resource '{uri}': "
@@ -315,7 +347,8 @@ class AuthMiddleware(Middleware):
         # component-level auth denied access, so the two cases are
         # indistinguishable here. Keep the message ambiguous to avoid
         # disclosing existence of prompts the caller is not authorized to see.
-        prompt = await fastmcp.fastmcp.get_prompt(prompt_name)
+        version = _requested_version(context.message.meta)
+        prompt = await fastmcp.fastmcp.get_prompt(prompt_name, version=version)
         if prompt is None:
             raise AuthorizationError(
                 f"Authorization failed for prompt '{prompt_name}': "

fastmcp_slim/fastmcp/server/server.py

@@ -101,6 +101,33 @@ if TYPE_CHECKING:
 logger = get_logger(__name__)
 
 
+def _version_request_meta(
+    version: VersionSpec | None,
+) -> mcp.types.RequestParams.Meta | None:
+    if version is None:
+        return None
+
+    if version.eq is not None and version.gte is None and version.lt is None:
+        version_value: str | dict[str, str] = version.eq
+    else:
+        version_value = {
+            key: value
+            for key, value in {
+                "gte": version.gte,
+                "lt": version.lt,
+                "eq": version.eq,
+            }.items()
+            if value is not None
+        }
+
+    if not version_value:
+        return None
+
+    return mcp.types.RequestParams.Meta.model_validate(
+        {"fastmcp": {"version": version_value}}
+    )
+
+
 # The MCP SDK warns "Tool X not listed, no validation will be performed"
 # for every call to app-only tools (hidden from list_tools by design).
 # This fires even when validate_input=False. Suppress it.
@@ -1222,7 +1249,9 @@ class FastMCP(
             if run_middleware:
                 mw_context = MiddlewareContext[CallToolRequestParams](
                     message=mcp.types.CallToolRequestParams(
-                        name=name, arguments=arguments or {}
+                        name=name,
+                        arguments=arguments or {},
+                        _meta=_version_request_meta(version),  # type: ignore[unknown-argument]  # pydantic alias
                     ),
                     source="client",
                     type="request",
@@ -1389,7 +1418,10 @@ class FastMCP(
             if run_middleware:
                 uri_param = AnyUrl(uri)
                 mw_context = MiddlewareContext(
-                    message=mcp.types.ReadResourceRequestParams(uri=uri_param),
+                    message=mcp.types.ReadResourceRequestParams(
+                        uri=uri_param,
+                        _meta=_version_request_meta(version),  # type: ignore[unknown-argument]  # pydantic alias
+                    ),
                     source="client",
                     type="request",
                     method="resources/read",
@@ -1563,7 +1595,9 @@ class FastMCP(
             if run_middleware:
                 mw_context = MiddlewareContext(
                     message=mcp.types.GetPromptRequestParams(
-                        name=name, arguments=arguments
+                        name=name,
+                        arguments=arguments,
+                        _meta=_version_request_meta(version),  # type: ignore[unknown-argument]  # pydantic alias
                     ),
                     source="client",
                     type="request",

tests/server/auth/test_authorization.py

@@ -20,6 +20,7 @@ from fastmcp.server.auth import (
 from fastmcp.server.middleware import AuthMiddleware
 from fastmcp.server.transforms import ToolTransform
 from fastmcp.tools.tool_transform import ToolTransformConfig, TransformedTool
+from fastmcp.utilities.versions import VersionSpec
 
 # =============================================================================
 # Test helpers
@@ -44,6 +45,12 @@ def make_tool() -> Mock:
     return tool
 
 
+def make_restricted_tag_server() -> FastMCP:
+    return FastMCP(
+        middleware=[AuthMiddleware(auth=restrict_tag("admin", scopes=["admin"]))]
+    )
+
+
 # =============================================================================
 # Tests for require_scopes
 # =============================================================================
@@ -812,6 +819,107 @@ class TestAuthMiddlewareCallTool:
             auth_context_var.reset(tok)
 
 
+class TestAuthMiddlewareVersionedRequests:
+    async def test_middleware_blocks_explicit_restricted_tool_version(self):
+        """AuthMiddleware should check the requested tool version."""
+        mcp = make_restricted_tag_server()
+
+        @mcp.tool(name="calc", version="1.0", tags={"admin"})
+        def calc_v1() -> str:
+            return "restricted"
+
+        @mcp.tool(name="calc", version="2.0")
+        def calc_v2() -> str:
+            return "public"
+
+        tok = set_token(make_token(scopes=["read"]))
+        try:
+            async with Client(mcp) as client:
+                with pytest.raises(Exception, match="authorization|insufficient"):
+                    await client.call_tool("calc", {}, version="1.0")
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_middleware_blocks_restricted_tool_version_selected_by_range(self):
+        """AuthMiddleware should check non-exact direct server version specs."""
+        mcp = make_restricted_tag_server()
+
+        @mcp.tool(name="calc", version="1.0", tags={"admin"})
+        def calc_v1() -> str:
+            return "restricted"
+
+        @mcp.tool(name="calc", version="2.0")
+        def calc_v2() -> str:
+            return "public"
+
+        tok = set_token(make_token(scopes=["read"]))
+        try:
+            with pytest.raises(AuthorizationError):
+                await mcp.call_tool("calc", {}, version=VersionSpec(lt="2.0"))
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_middleware_blocks_explicit_restricted_resource_version(self):
+        """AuthMiddleware should check the requested resource version."""
+        mcp = make_restricted_tag_server()
+
+        @mcp.resource("data://info", version="1.0", tags={"admin"})
+        def info_v1() -> str:
+            return "restricted"
+
+        @mcp.resource("data://info", version="2.0")
+        def info_v2() -> str:
+            return "public"
+
+        tok = set_token(make_token(scopes=["read"]))
+        try:
+            async with Client(mcp) as client:
+                with pytest.raises(Exception, match="authorization|insufficient"):
+                    await client.read_resource("data://info", version="1.0")
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_middleware_blocks_explicit_restricted_template_version(self):
+        """AuthMiddleware should check the requested resource template version."""
+        mcp = make_restricted_tag_server()
+
+        @mcp.resource("data://items/{item_id}", version="1.0", tags={"admin"})
+        def item_v1(item_id: str) -> str:
+            return f"restricted {item_id}"
+
+        @mcp.resource("data://items/{item_id}", version="2.0")
+        def item_v2(item_id: str) -> str:
+            return f"public {item_id}"
+
+        tok = set_token(make_token(scopes=["read"]))
+        try:
+            async with Client(mcp) as client:
+                with pytest.raises(Exception, match="authorization|insufficient"):
+                    await client.read_resource("data://items/123", version="1.0")
+        finally:
+            auth_context_var.reset(tok)
+
+    async def test_middleware_blocks_explicit_restricted_prompt_version(self):
+        """AuthMiddleware should check the requested prompt version."""
+        mcp = make_restricted_tag_server()
+
+        @mcp.prompt(name="greet", version="1.0", tags={"admin"})
+        def greet_v1() -> str:
+            return "restricted"
+
+        @mcp.prompt(name="greet", version="2.0")
+        def greet_v2() -> str:
+            return "public"
+
+        tok = set_token(make_token(scopes=["read"]))
+        try:
+            async with Client(mcp) as client:
+                with pytest.raises(Exception, match="authorization|insufficient"):
+                    await client.get_prompt("greet", version="1.0")
+        finally:
+            auth_context_var.reset(tok)
+
+
 # =============================================================================
 # Tests for component-level auth denial messaging (issue #4054 bug 1)
 # =============================================================================