fix: reject refresh tokens used as Bearer access tokens (#3524)

src/fastmcp/server/auth/jwt_issuer.py

@@ -207,13 +207,19 @@ class JWTIssuer:
 
         return token
 
-    def verify_token(self, token: str) -> dict[str, Any]:
+    def verify_token(
+        self,
+        token: str,
+        expected_token_use: str = "access",
+    ) -> dict[str, Any]:
         """Verify and decode a FastMCP token.
 
-        Validates JWT signature, expiration, issuer, and audience.
+        Validates JWT signature, expiration, issuer, audience, and token type.
 
         Args:
             token: JWT token to verify
+            expected_token_use: Expected token type ("access" or "refresh").
+                Defaults to "access", which rejects refresh tokens.
 
         Returns:
             Decoded token payload
@@ -225,6 +231,19 @@ class JWTIssuer:
             # Decode and verify signature
             payload = self._jwt.decode(token, self._signing_key)
 
+            # Validate token type
+            token_use = payload.get("token_use", "access")
+            if token_use != expected_token_use:
+                logger.debug(
+                    "Token type mismatch: expected %s, got %s",
+                    expected_token_use,
+                    token_use,
+                )
+                raise JoseError(
+                    f"Token type mismatch: expected {expected_token_use}, "
+                    f"got {token_use}"
+                )
+
             # Validate expiration
             exp = payload.get("exp")
             if exp and exp < time.time():

src/fastmcp/server/auth/oauth_proxy/proxy.py

@@ -1208,7 +1208,9 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
         """
         # Verify FastMCP refresh token
         try:
-            refresh_payload = self.jwt_issuer.verify_token(refresh_token.token)
+            refresh_payload = self.jwt_issuer.verify_token(
+                refresh_token.token, expected_token_use="refresh"
+            )
             refresh_jti = refresh_payload["jti"]
         except Exception as e:
             logger.debug("FastMCP refresh token validation failed: %s", e)

tests/server/auth/oauth_proxy/test_tokens.py

@@ -421,7 +421,9 @@ class TestUpstreamTokenStorageTTL:
         # by checking that we can still look up the tokens for refresh purposes.
         #
         # Extract the JTI from the refresh token to look up the mapping
-        refresh_payload = proxy.jwt_issuer.verify_token(result.refresh_token)
+        refresh_payload = proxy.jwt_issuer.verify_token(
+            result.refresh_token, expected_token_use="refresh"
+        )
         refresh_jti = refresh_payload["jti"]
 
         # The JTI mapping should exist
@@ -492,7 +494,9 @@ class TestUpstreamTokenStorageTTL:
         assert result.refresh_token is not None
 
         # Verify upstream tokens are accessible
-        refresh_payload = proxy.jwt_issuer.verify_token(result.refresh_token)
+        refresh_payload = proxy.jwt_issuer.verify_token(
+            result.refresh_token, expected_token_use="refresh"
+        )
         refresh_jti = refresh_payload["jti"]
 
         jti_mapping = await proxy._jti_mapping_store.get(key=refresh_jti)

tests/server/auth/test_jwt_issuer.py

@@ -132,7 +132,7 @@ class TestJWTIssuer:
             expires_in=60 * 60 * 24 * 30,  # 30 days
         )
 
-        payload = issuer.verify_token(token)
+        payload = issuer.verify_token(token, expected_token_use="refresh")
         assert payload["client_id"] == "client-abc"
         assert payload["token_use"] == "refresh"
         assert payload["jti"] == "refresh-token-id"
@@ -280,8 +280,31 @@ class TestJWTIssuer:
             upstream_claims=upstream_claims,
         )
 
-        payload = issuer.verify_token(token)
+        payload = issuer.verify_token(token, expected_token_use="refresh")
         assert "upstream_claims" in payload
         assert payload["upstream_claims"]["sub"] == "user-123"
         assert payload["upstream_claims"]["name"] == "Test User"
         assert payload["token_use"] == "refresh"
+
+    def test_verify_token_rejects_refresh_token_as_access(self, issuer):
+        """Refresh tokens must not be accepted when expecting access tokens."""
+        token = issuer.issue_refresh_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="refresh-token-id",
+            expires_in=60 * 60 * 24 * 30,
+        )
+
+        with pytest.raises(JoseError, match="Token type mismatch"):
+            issuer.verify_token(token)
+
+    def test_verify_token_rejects_access_token_as_refresh(self, issuer):
+        """Access tokens must not be accepted when expecting refresh tokens."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+        )
+
+        with pytest.raises(JoseError, match="Token type mismatch"):
+            issuer.verify_token(token, expected_token_use="refresh")