Document symmetric key JWT verification support (#1586)

2026-08-23 05:54:19 +02:00 · 2025-08-22 19:35:08 -04:00 · 2025-08-22 19:35:08 -04:00 · c730aaa6b0
commit c730aaa6b0
parent ade9dbdba4
3 changed files with 338 additions and 14 deletions
--- a/docs/servers/auth/token-verification.mdx
+++ b/docs/servers/auth/token-verification.mdx
@ -80,9 +80,48 @@ This configuration creates a server that validates JWTs issued by `auth.yourcomp

 The `issuer` parameter ensures tokens come from your trusted authentication system, while `audience` validation prevents tokens intended for other services from being accepted by your MCP server.

+#### Symmetric Key Verification (HMAC)
+
+Symmetric key verification uses a shared secret for both signing and validation, making it ideal for internal microservices and trusted environments where the same secret can be securely distributed to both token issuers and validators.
+
+This approach is commonly used in microservices architectures where services share a secret key, or when your authentication service and MCP server are both managed by the same organization. The HMAC algorithms (HS256, HS384, HS512) provide strong security when the shared secret is properly managed.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+# Use a shared secret for symmetric key verification
+verifier = JWTVerifier(
+    public_key="your-shared-secret-key-minimum-32-chars",  # Despite the name, this accepts symmetric secrets
+    issuer="internal-auth-service",
+    audience="mcp-internal-api",
+    algorithm="HS256"  # or HS384, HS512 for stronger security
+)
+
+mcp = FastMCP(name="Internal API", auth=verifier)
+```
+
+The verifier will validate tokens signed with the same secret using the specified HMAC algorithm. This approach offers several advantages for internal systems:
+
+- **Simplicity**: No key pair management or certificate distribution
+- **Performance**: HMAC operations are typically faster than RSA
+- **Compatibility**: Works well with existing microservice authentication patterns
+
+<Note>
+The parameter is named `public_key` for backwards compatibility, but when using HMAC algorithms (HS256/384/512), it accepts the symmetric secret string.
+</Note>
+
+<Warning>
+**Security Considerations for Symmetric Keys:**
+- Use a strong, randomly generated secret (minimum 32 characters recommended)
+- Never expose the secret in logs, error messages, or version control
+- Implement secure key distribution and rotation mechanisms
+- Consider using asymmetric keys (RSA/ECDSA) for external-facing APIs
+</Warning>
+
 #### Static Public Key Verification

-Static public key verification works when you have a fixed signing key and don't need automatic key rotation. This approach simplifies deployment in environments where JWKS endpoints aren't available.
+Static public key verification works when you have a fixed RSA or ECDSA signing key and don't need automatic key rotation. This approach is primarily useful for development environments or controlled deployments where JWKS endpoints aren't available.

 ```python
 from fastmcp import FastMCP
@ -102,7 +141,7 @@ verifier = JWTVerifier(
 mcp = FastMCP(name="Protected API", auth=verifier)
 ```

-This configuration validates tokens using a specific public key. The key must correspond to the private key used by your token issuer. While less flexible than JWKS endpoints, this approach works well for controlled environments or when using dedicated signing keys.
+This configuration validates tokens using a specific RSA or ECDSA public key. The key must correspond to the private key used by your token issuer. While less flexible than JWKS endpoints, this approach can be useful in development environments or when testing with fixed keys.

 ### Development and Testing

@ -180,11 +219,17 @@ Environment-based configuration separates authentication settings from applicati
 # Enable JWT verification
 export FASTMCP_SERVER_AUTH=JWT

-# Configure JWT verification parameters  
+# For asymmetric verification with JWKS endpoint:
 export FASTMCP_SERVER_AUTH_JWT_JWKS_URI="https://auth.company.com/.well-known/jwks.json"
 export FASTMCP_SERVER_AUTH_JWT_ISSUER="https://auth.company.com"
 export FASTMCP_SERVER_AUTH_JWT_AUDIENCE="mcp-production-api"
 export FASTMCP_SERVER_AUTH_JWT_REQUIRED_SCOPES="read:data,write:data"
+
+# OR for symmetric key verification (HMAC):
+export FASTMCP_SERVER_AUTH_JWT_PUBLIC_KEY="your-shared-secret-key-minimum-32-chars"
+export FASTMCP_SERVER_AUTH_JWT_ALGORITHM="HS256"  # or HS384, HS512
+export FASTMCP_SERVER_AUTH_JWT_ISSUER="internal-auth-service"
+export FASTMCP_SERVER_AUTH_JWT_AUDIENCE="mcp-internal-api"
 ```

 With these environment variables configured, your FastMCP server automatically enables JWT verification:
--- a/src/fastmcp/server/auth/providers/jwt.py
+++ b/src/fastmcp/server/auth/providers/jwt.py
@ -159,17 +159,20 @@ class JWTVerifierSettings(BaseSettings):
@register_provider("JWT")
 class JWTVerifier(TokenVerifier):
    """
-    JWT token verifier using public key or JWKS.
+    JWT token verifier supporting both asymmetric (RSA/ECDSA) and symmetric (HMAC) algorithms.

-    This verifier validates JWT tokens signed by an external issuer. It's ideal for
-    scenarios where you have a centralized identity provider (like Auth0, Okta, or
-    your own OAuth server) that issues JWTs, and your FastMCP server acts as a
-    resource server validating those tokens.
+    This verifier validates JWT tokens using various signing algorithms:
+    - **Asymmetric algorithms** (RS256/384/512, ES256/384/512, PS256/384/512):
+      Uses public/private key pairs. Ideal for external clients and services where
+      only the authorization server has the private key.
+    - **Symmetric algorithms** (HS256/384/512): Uses a shared secret for both
+      signing and verification. Perfect for internal microservices and trusted
+      environments where the secret can be securely shared.

    Use this when:
-    - You have JWT tokens issued by an external service
-    - You want asymmetric key verification (public/private key pairs)
-    - You need JWKS support for automatic key rotation
+    - You have JWT tokens issued by an external service (asymmetric)
+    - You need JWKS support for automatic key rotation (asymmetric)
+    - You have internal microservices sharing a secret key (symmetric)
    - Your tokens contain standard OAuth scopes and claims
    """

@ -188,11 +191,14 @@ class JWTVerifier(TokenVerifier):
        Initialize the JWT token verifier.

        Args:
-            public_key: PEM-encoded public key for verification
-            jwks_uri: URI to fetch JSON Web Key Set
+            public_key: For asymmetric algorithms (RS256, ES256, etc.): PEM-encoded public key.
+                       For symmetric algorithms (HS256, HS384, HS512): The shared secret string.
+            jwks_uri: URI to fetch JSON Web Key Set (only for asymmetric algorithms)
            issuer: Expected issuer claim
            audience: Expected audience claim(s)
-            algorithm: JWT signing algorithm (default: RS256)
+            algorithm: JWT signing algorithm. Supported algorithms:
+                      - Asymmetric: RS256/384/512, ES256/384/512, PS256/384/512 (default: RS256)
+                      - Symmetric: HS256, HS384, HS512
            required_scopes: Required scopes for all tokens
            resource_server_url: Resource server URL for TokenVerifier protocol
        """
--- a/tests/server/auth/test_jwt_provider.py
+++ b/tests/server/auth/test_jwt_provider.py
@ -11,11 +11,77 @@ from fastmcp.server.auth.providers.jwt import JWKData, JWKSData, JWTVerifier, RS
 from fastmcp.utilities.tests import run_server_in_process


+class SymmetricKeyHelper:
+    """Helper class for generating symmetric key JWT tokens for testing."""
+
+    def __init__(self, secret: str):
+        """Initialize with a secret key."""
+        self.secret = secret
+
+    def create_token(
+        self,
+        subject: str = "fastmcp-user",
+        issuer: str = "https://fastmcp.example.com",
+        audience: str | list[str] | None = None,
+        scopes: list[str] | None = None,
+        expires_in_seconds: int = 3600,
+        additional_claims: dict[str, Any] | None = None,
+        algorithm: str = "HS256",
+    ) -> str:
+        """
+        Generate a test JWT token using symmetric key for testing purposes.
+
+        Args:
+            subject: Subject claim (usually user ID)
+            issuer: Issuer claim
+            audience: Audience claim - can be a string or list of strings (optional)
+            scopes: List of scopes to include
+            expires_in_seconds: Token expiration time in seconds
+            additional_claims: Any additional claims to include
+            algorithm: JWT signing algorithm (HS256, HS384, or HS512)
+        """
+        import time
+
+        from authlib.jose import JsonWebToken
+
+        # Create header
+        header = {"alg": algorithm}
+
+        # Create payload
+        payload = {
+            "sub": subject,
+            "iss": issuer,
+            "iat": int(time.time()),
+            "exp": int(time.time()) + expires_in_seconds,
+        }
+
+        if audience:
+            payload["aud"] = audience
+
+        if scopes:
+            payload["scope"] = " ".join(scopes)
+
+        if additional_claims:
+            payload.update(additional_claims)
+
+        # Create JWT
+        jwt_lib = JsonWebToken([algorithm])
+        token_bytes = jwt_lib.encode(header, payload, self.secret)
+
+        return token_bytes.decode("utf-8")
+
+
@pytest.fixture(scope="module")
 def rsa_key_pair() -> RSAKeyPair:
    return RSAKeyPair.generate()


+@pytest.fixture(scope="module")
+def symmetric_key_helper() -> SymmetricKeyHelper:
+    """Generate a symmetric key helper for testing."""
+    return SymmetricKeyHelper("test-secret-key-for-hmac-signing")
+
+
@pytest.fixture(scope="module")
 def bearer_token(rsa_key_pair: RSAKeyPair) -> str:
    return rsa_key_pair.create_token(
@ -34,6 +100,17 @@ def bearer_provider(rsa_key_pair: RSAKeyPair) -> JWTVerifier:
    )


+@pytest.fixture
+def symmetric_provider(symmetric_key_helper: SymmetricKeyHelper) -> JWTVerifier:
+    """Create JWTVerifier configured for symmetric key verification."""
+    return JWTVerifier(
+        public_key=symmetric_key_helper.secret,
+        issuer="https://test.example.com",
+        audience="https://api.example.com",
+        algorithm="HS256",
+    )
+
+
 def run_mcp_server(
    public_key: str,
    host: str,
@ -104,6 +181,202 @@ class TestRSAKeyPair:
        # We'll validate the scopes in the BearerToken tests


+class TestSymmetricKeyJWT:
+    """Tests for JWT verification using symmetric keys (HMAC algorithms)."""
+
+    def test_initialization_with_symmetric_key(
+        self, symmetric_key_helper: SymmetricKeyHelper
+    ):
+        """Test JWTVerifier initialization with symmetric key."""
+        provider = JWTVerifier(
+            public_key=symmetric_key_helper.secret,
+            issuer="https://test.example.com",
+            algorithm="HS256",
+        )
+
+        assert provider.issuer == "https://test.example.com"
+        assert provider.public_key == symmetric_key_helper.secret
+        assert provider.algorithm == "HS256"
+        assert provider.jwks_uri is None
+
+    def test_initialization_with_different_symmetric_algorithms(
+        self, symmetric_key_helper: SymmetricKeyHelper
+    ):
+        """Test JWTVerifier initialization with different HMAC algorithms."""
+        algorithms = ["HS256", "HS384", "HS512"]
+
+        for algorithm in algorithms:
+            provider = JWTVerifier(
+                public_key=symmetric_key_helper.secret,
+                issuer="https://test.example.com",
+                algorithm=algorithm,
+            )
+            assert provider.algorithm == algorithm
+
+    async def test_valid_symmetric_token_validation(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test validation of a valid token signed with symmetric key."""
+        token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            scopes=["read", "write"],
+            algorithm="HS256",
+        )
+
+        access_token = await symmetric_provider.load_access_token(token)
+
+        assert access_token is not None
+        assert access_token.client_id == "test-user"
+        assert "read" in access_token.scopes
+        assert "write" in access_token.scopes
+        assert access_token.expires_at is not None
+
+    async def test_symmetric_token_with_different_algorithms(
+        self, symmetric_key_helper: SymmetricKeyHelper
+    ):
+        """Test that different HMAC algorithms work correctly."""
+        algorithms = ["HS256", "HS384", "HS512"]
+
+        for algorithm in algorithms:
+            provider = JWTVerifier(
+                public_key=symmetric_key_helper.secret,
+                issuer="https://test.example.com",
+                algorithm=algorithm,
+            )
+
+            token = symmetric_key_helper.create_token(
+                subject="test-user",
+                issuer="https://test.example.com",
+                algorithm=algorithm,
+            )
+
+            access_token = await provider.load_access_token(token)
+            assert access_token is not None
+            assert access_token.client_id == "test-user"
+
+    async def test_symmetric_token_issuer_validation(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test issuer validation with symmetric key tokens."""
+        # Valid issuer
+        valid_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+        access_token = await symmetric_provider.load_access_token(valid_token)
+        assert access_token is not None
+
+        # Invalid issuer
+        invalid_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://evil.example.com",
+            audience="https://api.example.com",
+        )
+        access_token = await symmetric_provider.load_access_token(invalid_token)
+        assert access_token is None
+
+    async def test_symmetric_token_audience_validation(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test audience validation with symmetric key tokens."""
+        # Valid audience
+        valid_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+        access_token = await symmetric_provider.load_access_token(valid_token)
+        assert access_token is not None
+
+        # Invalid audience
+        invalid_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://wrong-api.example.com",
+        )
+        access_token = await symmetric_provider.load_access_token(invalid_token)
+        assert access_token is None
+
+    async def test_symmetric_token_scope_extraction(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test scope extraction from symmetric key tokens."""
+        token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            scopes=["read", "write", "admin"],
+        )
+
+        access_token = await symmetric_provider.load_access_token(token)
+        assert access_token is not None
+        assert set(access_token.scopes) == {"read", "write", "admin"}
+
+    async def test_symmetric_token_expiration(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test expiration validation with symmetric key tokens."""
+        # Valid token
+        valid_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            expires_in_seconds=3600,  # 1 hour from now
+        )
+        access_token = await symmetric_provider.load_access_token(valid_token)
+        assert access_token is not None
+
+        # Expired token
+        expired_token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            expires_in_seconds=-3600,  # 1 hour ago
+        )
+        access_token = await symmetric_provider.load_access_token(expired_token)
+        assert access_token is None
+
+    async def test_symmetric_token_invalid_signature(
+        self, symmetric_key_helper: SymmetricKeyHelper, symmetric_provider: JWTVerifier
+    ):
+        """Test rejection of tokens with invalid signatures."""
+        # Create a token with a different secret
+        other_helper = SymmetricKeyHelper("different-secret-key")
+        token = other_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await symmetric_provider.load_access_token(token)
+        assert access_token is None
+
+    async def test_symmetric_token_algorithm_mismatch(
+        self, symmetric_key_helper: SymmetricKeyHelper
+    ):
+        """Test that tokens with mismatched algorithms are rejected."""
+        # Create provider expecting HS256
+        provider = JWTVerifier(
+            public_key=symmetric_key_helper.secret,
+            issuer="https://test.example.com",
+            algorithm="HS256",
+        )
+
+        # Create token with HS512
+        token = symmetric_key_helper.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            algorithm="HS512",
+        )
+
+        # Should fail because provider expects HS256
+        access_token = await provider.load_access_token(token)
+        assert access_token is None
+
+
 class TestBearerTokenJWKS:
    """Tests for JWKS URI functionality."""