Add DebugTokenVerifier with custom sync/async validation (#2296)

jlowin · web-flow · commit de58bb0e6ca2 · 2025-10-31T10:38:01.000-04:00
* Add DebugTokenVerifier with custom sync/async validation

* move import
diff --git a/docs/servers/auth/token-verification.mdx b/docs/servers/auth/token-verification.mdx
@@ -210,6 +210,67 @@ Static token verification stores tokens as plain text and should never be used i
 </Warning>
 
 
+### Debug/Custom Token Verification
+
+The `DebugTokenVerifier` provides maximum flexibility for testing and special cases where standard token verification isn't applicable. It delegates validation to a user-provided callable, making it useful for prototyping, testing scenarios, or handling opaque tokens without introspection endpoints.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Accept all tokens (useful for rapid development)
+verifier = DebugTokenVerifier()
+
+mcp = FastMCP(name="Development Server", auth=verifier)
+```
+
+By default, `DebugTokenVerifier` accepts any non-empty token as valid. This eliminates authentication barriers during early development, allowing you to focus on core functionality before adding security.
+
+For more controlled testing, provide custom validation logic:
+
+```python
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Synchronous validation - check token prefix
+verifier = DebugTokenVerifier(
+    validate=lambda token: token.startswith("dev-"),
+    client_id="development-client",
+    scopes=["read", "write"]
+)
+
+mcp = FastMCP(name="Development Server", auth=verifier)
+```
+
+The validation callable can also be async, enabling database lookups or external service calls:
+
+```python
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Asynchronous validation - check against cache
+async def validate_token(token: str) -> bool:
+    # Check if token exists in Redis, database, etc.
+    return await redis.exists(f"valid_tokens:{token}")
+
+verifier = DebugTokenVerifier(
+    validate=validate_token,
+    client_id="api-client",
+    scopes=["api:access"]
+)
+
+mcp = FastMCP(name="Custom API", auth=verifier)
+```
+
+**Use Cases:**
+
+- **Testing**: Accept any token during integration tests without setting up token infrastructure
+- **Prototyping**: Quickly validate concepts without authentication complexity
+- **Opaque tokens without introspection**: When you have tokens from an IDP that provides no introspection endpoint, and you're willing to accept tokens without validation (validation happens later at the upstream service)
+- **Custom token formats**: Implement validation for non-standard token formats or legacy systems
+
+<Warning>
+`DebugTokenVerifier` bypasses standard security checks. Only use in controlled environments (development, testing) or when you fully understand the security implications. For production, use proper JWT or introspection-based verification.
+</Warning>
+
 ### Test Token Generation
 
 Test token generation helps when you need to test JWT verification without setting up complete identity infrastructure. FastMCP includes utilities for generating test key pairs and signed tokens.
diff --git a/src/fastmcp/server/auth/__init__.py b/src/fastmcp/server/auth/__init__.py
@@ -5,6 +5,7 @@
     AccessToken,
     AuthProvider,
 )
+from .providers.debug import DebugTokenVerifier
 from .providers.jwt import JWTVerifier, StaticTokenVerifier
 from .oauth_proxy import OAuthProxy
 from .oidc_proxy import OIDCProxy
@@ -13,6 +14,7 @@
 __all__ = [
     "AccessToken",
     "AuthProvider",
+    "DebugTokenVerifier",
     "JWTVerifier",
     "OAuthProvider",
     "OAuthProxy",
diff --git a/src/fastmcp/server/auth/providers/debug.py b/src/fastmcp/server/auth/providers/debug.py
@@ -0,0 +1,114 @@
+"""Debug token verifier for testing and special cases.
+
+This module provides a flexible token verifier that delegates validation
+to a custom callable. Useful for testing, development, or scenarios where
+standard verification isn't possible (like opaque tokens without introspection).
+
+Example:
+    ```python
+    from fastmcp import FastMCP
+    from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+    # Accept all tokens (default - useful for testing)
+    auth = DebugTokenVerifier()
+
+    # Custom sync validation logic
+    auth = DebugTokenVerifier(validate=lambda token: token.startswith("valid-"))
+
+    # Custom async validation logic
+    async def check_cache(token: str) -> bool:
+        return await redis.exists(f"token:{token}")
+
+    auth = DebugTokenVerifier(validate=check_cache)
+
+    mcp = FastMCP("My Server", auth=auth)
+    ```
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Awaitable, Callable
+
+from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class DebugTokenVerifier(TokenVerifier):
+    """Token verifier with custom validation logic.
+
+    This verifier delegates token validation to a user-provided callable.
+    By default, it accepts all non-empty tokens (useful for testing).
+
+    Use cases:
+    - Testing: Accept any token without real verification
+    - Development: Custom validation logic for prototyping
+    - Opaque tokens: When you have tokens with no introspection endpoint
+
+    WARNING: This bypasses standard security checks. Only use in controlled
+    environments or when you understand the security implications.
+    """
+
+    def __init__(
+        self,
+        validate: Callable[[str], bool]
+        | Callable[[str], Awaitable[bool]] = lambda token: True,
+        client_id: str = "debug-client",
+        scopes: list[str] | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        """Initialize the debug token verifier.
+
+        Args:
+            validate: Callable that takes a token string and returns True if valid.
+                Can be sync or async. Default accepts all tokens.
+            client_id: Client ID to assign to validated tokens
+            scopes: Scopes to assign to validated tokens
+            required_scopes: Required scopes (inherited from TokenVerifier base class)
+        """
+        super().__init__(required_scopes=required_scopes)
+        self.validate = validate
+        self.client_id = client_id
+        self.scopes = scopes or []
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token using custom validation logic.
+
+        Args:
+            token: The token string to validate
+
+        Returns:
+            AccessToken if validation succeeds, None otherwise
+        """
+        # Reject empty tokens
+        if not token or not token.strip():
+            logger.debug("Rejecting empty token")
+            return None
+
+        try:
+            # Call validation function and await if result is awaitable
+            result = self.validate(token)
+            if inspect.isawaitable(result):
+                is_valid = await result
+            else:
+                is_valid = result
+
+            if not is_valid:
+                logger.debug("Token validation failed: callable returned False")
+                return None
+
+            # Return valid AccessToken
+            return AccessToken(
+                token=token,
+                client_id=self.client_id,
+                scopes=self.scopes,
+                expires_at=None,  # No expiration
+                claims={"token": token},  # Store original token in claims
+            )
+
+        except Exception as e:
+            logger.debug("Token validation error: %s", e, exc_info=True)
+            return None
diff --git a/tests/server/auth/test_debug_verifier.py b/tests/server/auth/test_debug_verifier.py
@@ -0,0 +1,169 @@
+"""Unit tests for DebugTokenVerifier."""
+
+import re
+
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+
+class TestDebugTokenVerifier:
+    """Test DebugTokenVerifier initialization and validation."""
+
+    def test_init_defaults(self):
+        """Test initialization with default parameters."""
+        verifier = DebugTokenVerifier()
+
+        assert verifier.client_id == "debug-client"
+        assert verifier.scopes == []
+        assert verifier.required_scopes == []
+        assert callable(verifier.validate)
+
+    def test_init_custom_parameters(self):
+        """Test initialization with custom parameters."""
+        verifier = DebugTokenVerifier(
+            validate=lambda t: t.startswith("valid-"),
+            client_id="custom-client",
+            scopes=["read", "write"],
+            required_scopes=["admin"],
+        )
+
+        assert verifier.client_id == "custom-client"
+        assert verifier.scopes == ["read", "write"]
+        assert verifier.required_scopes == ["admin"]
+
+    async def test_verify_token_default_accepts_all(self):
+        """Test that default verifier accepts all non-empty tokens."""
+        verifier = DebugTokenVerifier()
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is not None
+        assert result.token == "any-token"
+        assert result.client_id == "debug-client"
+        assert result.scopes == []
+        assert result.expires_at is None
+        assert result.claims == {"token": "any-token"}
+
+    async def test_verify_token_rejects_empty(self):
+        """Test that empty tokens are rejected even with default verifier."""
+        verifier = DebugTokenVerifier()
+
+        # Empty string
+        assert await verifier.verify_token("") is None
+
+        # Whitespace only
+        assert await verifier.verify_token("   ") is None
+
+    async def test_verify_token_sync_callable_success(self):
+        """Test token verification with custom sync callable that passes."""
+        verifier = DebugTokenVerifier(
+            validate=lambda t: t.startswith("valid-"),
+            client_id="test-client",
+            scopes=["read"],
+        )
+
+        result = await verifier.verify_token("valid-token-123")
+
+        assert result is not None
+        assert result.token == "valid-token-123"
+        assert result.client_id == "test-client"
+        assert result.scopes == ["read"]
+        assert result.expires_at is None
+        assert result.claims == {"token": "valid-token-123"}
+
+    async def test_verify_token_sync_callable_failure(self):
+        """Test token verification with custom sync callable that fails."""
+        verifier = DebugTokenVerifier(validate=lambda t: t.startswith("valid-"))
+
+        result = await verifier.verify_token("invalid-token")
+
+        assert result is None
+
+    async def test_verify_token_async_callable_success(self):
+        """Test token verification with custom async callable that passes."""
+
+        async def async_validator(token: str) -> bool:
+            # Simulate async operation (e.g., database check)
+            return token in {"token1", "token2", "token3"}
+
+        verifier = DebugTokenVerifier(
+            validate=async_validator,
+            client_id="async-client",
+            scopes=["admin"],
+        )
+
+        result = await verifier.verify_token("token2")
+
+        assert result is not None
+        assert result.token == "token2"
+        assert result.client_id == "async-client"
+        assert result.scopes == ["admin"]
+
+    async def test_verify_token_async_callable_failure(self):
+        """Test token verification with custom async callable that fails."""
+
+        async def async_validator(token: str) -> bool:
+            return token in {"token1", "token2", "token3"}
+
+        verifier = DebugTokenVerifier(validate=async_validator)
+
+        result = await verifier.verify_token("token99")
+
+        assert result is None
+
+    async def test_verify_token_callable_exception(self):
+        """Test that exceptions in validate callable are handled gracefully."""
+
+        def failing_validator(token: str) -> bool:
+            raise ValueError("Something went wrong")
+
+        verifier = DebugTokenVerifier(validate=failing_validator)
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is None
+
+    async def test_verify_token_async_callable_exception(self):
+        """Test that exceptions in async validate callable are handled gracefully."""
+
+        async def failing_async_validator(token: str) -> bool:
+            raise ValueError("Async validation failed")
+
+        verifier = DebugTokenVerifier(validate=failing_async_validator)
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is None
+
+    async def test_verify_token_whitelist_pattern(self):
+        """Test using verifier with a whitelist of allowed tokens."""
+        allowed_tokens = {"secret-token-1", "secret-token-2", "admin-token"}
+
+        verifier = DebugTokenVerifier(validate=lambda t: t in allowed_tokens)
+
+        # Allowed tokens
+        assert await verifier.verify_token("secret-token-1") is not None
+        assert await verifier.verify_token("admin-token") is not None
+
+        # Disallowed tokens
+        assert await verifier.verify_token("unknown-token") is None
+        assert await verifier.verify_token("hacker-token") is None
+
+    async def test_verify_token_pattern_matching(self):
+        """Test using verifier with regex-like pattern matching."""
+
+        pattern = re.compile(r"^[A-Z]{3}-\d{4}-[a-z]{2}$")
+
+        verifier = DebugTokenVerifier(
+            validate=lambda t: bool(pattern.match(t)),
+            client_id="pattern-client",
+        )
+
+        # Valid patterns
+        result = await verifier.verify_token("ABC-1234-xy")
+        assert result is not None
+        assert result.client_id == "pattern-client"
+
+        # Invalid patterns
+        assert await verifier.verify_token("abc-1234-xy") is None  # Wrong case
+        assert await verifier.verify_token("ABC-123-xy") is None  # Wrong digits
+        assert await verifier.verify_token("ABC-1234-xyz") is None  # Too many chars
