refactor

Author: ihrpr

examples/servers/simple-auth/mcp_simple_auth/server.py

@@ -18,7 +18,7 @@
 
 from mcp.server.auth.middleware.auth_context import get_access_token
 from mcp.server.auth.settings import AuthSettings
-from mcp.server.auth.token_verifier import IntrospectionTokenVerifier
+from .token_verifier import IntrospectionTokenVerifier
 from mcp.server.fastmcp.server import FastMCP
 
 logger = logging.getLogger(__name__)

examples/servers/simple-auth/mcp_simple_auth/token_verifier.py

@@ -0,0 +1,64 @@
+"""Example token verifier implementation using OAuth 2.0 Token Introspection (RFC 7662)."""
+
+import logging
+from mcp.server.auth.provider import AccessToken
+
+logger = logging.getLogger(__name__)
+
+
+class IntrospectionTokenVerifier:
+    """Example token verifier that uses OAuth 2.0 Token Introspection (RFC 7662).
+    
+    This is a simple example implementation for demonstration purposes.
+    Production implementations should consider:
+    - Connection pooling and reuse
+    - More sophisticated error handling
+    - Rate limiting and retry logic
+    - Comprehensive configuration options
+    """
+
+    def __init__(self, introspection_endpoint: str):
+        self.introspection_endpoint = introspection_endpoint
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token via introspection endpoint."""
+        import httpx
+
+        # Validate URL to prevent SSRF attacks
+        if not self.introspection_endpoint.startswith(('https://', 'http://localhost', 'http://127.0.0.1')):
+            logger.warning(f"Rejecting introspection endpoint with unsafe scheme: {self.introspection_endpoint}")
+            return None
+
+        # Configure secure HTTP client
+        timeout = httpx.Timeout(10.0, connect=5.0)
+        limits = httpx.Limits(max_connections=10, max_keepalive_connections=5)
+        
+        async with httpx.AsyncClient(
+            timeout=timeout,
+            limits=limits,
+            verify=True,  # Enforce SSL verification
+        ) as client:
+            try:
+                response = await client.post(
+                    self.introspection_endpoint,
+                    data={"token": token},
+                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                )
+
+                if response.status_code != 200:
+                    logger.debug(f"Token introspection returned status {response.status_code}")
+                    return None
+
+                data = response.json()
+                if not data.get("active", False):
+                    return None
+
+                return AccessToken(
+                    token=token,
+                    client_id=data.get("client_id", "unknown"),
+                    scopes=data.get("scope", "").split() if data.get("scope") else [],
+                    expires_at=data.get("exp"),
+                )
+            except Exception as e:
+                logger.warning(f"Token introspection failed: {e}")
+                return None

src/mcp/server/auth/provider.py

@@ -278,3 +278,19 @@ def construct_redirect_uri(redirect_uri_base: str, **params: str | None) -> str:
 
     redirect_uri = urlunparse(parsed_uri._replace(query=urlencode(query_params)))
     return redirect_uri
+
+
+class ProviderTokenVerifier:
+    """Token verifier that uses an OAuthAuthorizationServerProvider.
+    
+    This is provided for backwards compatibility with existing auth_server_provider
+    configurations. For new implementations using AS/RS separation, consider using 
+    the TokenVerifier protocol with a dedicated implementation like IntrospectionTokenVerifier.
+    """
+
+    def __init__(self, provider: "OAuthAuthorizationServerProvider[AccessToken, RefreshToken, AuthorizationCode]"):
+        self.provider = provider
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token using the provider's load_access_token method."""
+        return await self.provider.load_access_token(token)

src/mcp/server/auth/token_verifier.py

@@ -1,8 +1,8 @@
-"""Token verification protocol and implementations."""
+"""Token verification protocol."""
 
-from typing import Any, Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable
 
-from mcp.server.auth.provider import AccessToken, OAuthAuthorizationServerProvider
+from mcp.server.auth.provider import AccessToken
 
 
 @runtime_checkable
@@ -12,49 +12,3 @@ class TokenVerifier(Protocol):
     async def verify_token(self, token: str) -> AccessToken | None:
         """Verify a bearer token and return access info if valid."""
         ...
-
-
-class ProviderTokenVerifier:
-    """Token verifier that uses an OAuthAuthorizationServerProvider."""
-
-    def __init__(self, provider: OAuthAuthorizationServerProvider[Any, Any, Any]):
-        self.provider = provider
-
-    async def verify_token(self, token: str) -> AccessToken | None:
-        """Verify token using the provider's load_access_token method."""
-        return await self.provider.load_access_token(token)
-
-
-class IntrospectionTokenVerifier:
-    """Token verifier that uses OAuth 2.0 Token Introspection (RFC 7662)."""
-
-    def __init__(self, introspection_endpoint: str):
-        self.introspection_endpoint = introspection_endpoint
-
-    async def verify_token(self, token: str) -> AccessToken | None:
-        """Verify token via introspection endpoint."""
-        import httpx
-
-        async with httpx.AsyncClient() as client:
-            try:
-                response = await client.post(
-                    self.introspection_endpoint,
-                    data={"token": token},
-                    headers={"Content-Type": "application/x-www-form-urlencoded"},
-                )
-
-                if response.status_code != 200:
-                    return None
-
-                data = response.json()
-                if not data.get("active", False):
-                    return None
-
-                return AccessToken(
-                    token=token,
-                    client_id=data.get("client_id", "unknown"),
-                    scopes=data.get("scope", "").split() if data.get("scope") else [],
-                    expires_at=data.get("exp"),
-                )
-            except Exception:
-                return None

src/mcp/server/fastmcp/server.py

@@ -30,9 +30,9 @@
     BearerAuthBackend,
     RequireAuthMiddleware,
 )
-from mcp.server.auth.provider import OAuthAuthorizationServerProvider
+from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier
 from mcp.server.auth.settings import AuthSettings
-from mcp.server.auth.token_verifier import ProviderTokenVerifier, TokenVerifier
+from mcp.server.auth.token_verifier import TokenVerifier
 from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, elicit_with_validation
 from mcp.server.fastmcp.exceptions import ResourceError
 from mcp.server.fastmcp.prompts import Prompt, PromptManager
@@ -67,6 +67,8 @@
 logger = get_logger(__name__)
 
 
+
+
 class Settings(BaseSettings, Generic[LifespanResultT]):
     """FastMCP server settings.
 
@@ -169,7 +171,7 @@ def __init__(
         self._auth_server_provider = auth_server_provider
         self._token_verifier = token_verifier
 
-        # Create token verifier from provider if needed
+        # Create token verifier from provider if needed (backwards compatibility)
         if auth_server_provider and not token_verifier:
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._event_store = event_store

tests/server/auth/middleware/test_bearer_auth.py

@@ -19,8 +19,8 @@
 from mcp.server.auth.provider import (
     AccessToken,
     OAuthAuthorizationServerProvider,
+    ProviderTokenVerifier,
 )
-from mcp.server.auth.token_verifier import ProviderTokenVerifier
 
 
 class MockOAuthProvider: