ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Support EdDSA verification in JWTVerifier (#4752)

Browse source
This commit is contained in:
nate nowack 2026-08-06 13:09:37 -05:00 • committed by GitHub
commit 75fb116e36
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 195 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

18
docs/servers/auth/token-verification.mdx
View file

@ -80,6 +80,19 @@ 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.
`JWTVerifier` accepts RSA (`RS*` and `PS*`), ECDSA (`ES*`), and Edwards-curve (`Ed25519` and `Ed448`) signatures from JWKS endpoints. Set `algorithm` when your issuer does not use the default `RS256`:
```python
verifier = JWTVerifier(
jwks_uri="https://auth.yourcompany.com/.well-known/jwks.json",
issuer="https://auth.yourcompany.com",
audience="mcp-production-api",
algorithm="Ed25519",
)
```
The legacy `EdDSA` identifier is also accepted for compatibility with identity providers that have not yet adopted the fully specified identifiers from RFC 9864.
### 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.
@ -121,7 +134,7 @@ The parameter is named `public_key` for backwards compatibility, but when using
### Static Public Key Verification
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.
Static public key verification works when you have a fixed RSA, ECDSA, or EdDSA 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
@ -141,7 +154,7 @@ verifier = JWTVerifier(
mcp = FastMCP(name="Protected API", auth=verifier)
```
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.
This configuration validates tokens using a specific RSA, ECDSA, or EdDSA 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.
## Opaque Token Verification
Many authorization servers issue opaque tokens rather than self-contained JWTs. Opaque tokens are random strings that carry no information themselves - the authorization server maintains their state and validation requires querying the server. FastMCP supports opaque token validation through OAuth 2.0 Token Introspection (RFC 7662).
@ -425,4 +438,3 @@ mcp = FastMCP(name="Production API", auth=verifier)
This keeps configuration out of your codebase while maintaining explicit setup.
This approach enables the same codebase to run across development, staging, and production environments with different authentication requirements. Development might use static tokens while production uses JWT verification, all controlled through environment configuration.

54
fastmcp_slim/fastmcp/server/auth/providers/jwt.py
View file

@ -6,7 +6,7 @@ import contextlib
import json
import time
from dataclasses import dataclass
from typing import Any, TypeAlias, cast
from typing import Any, Literal, TypeAlias, cast
import httpx2
from cryptography.hazmat.primitives import serialization
@ -29,22 +29,30 @@ JWKKeyData: TypeAlias = dict[str, str | list[str]]
SUPPORTED_JWS_HEADER_FIELDS = frozenset(JWS_HEADER_REGISTRY)
def _import_key_for_algorithm(key: str | bytes | JWKKeyData, algorithm: str):
def _key_type_for_algorithm(algorithm: str) -> Literal["oct", "RSA", "EC", "OKP"]:
if algorithm.startswith("HS"):
return jwk.import_key(key, "oct")
return "oct"
if algorithm.startswith(("RS", "PS")):
return jwk.import_key(key, "RSA")
return "RSA"
if algorithm.startswith("ES"):
return jwk.import_key(key, "EC")
return "EC"
if algorithm in {"EdDSA", "Ed25519", "Ed448"}:
return "OKP"
raise ValueError(f"Unsupported algorithm: {algorithm}.")
def _import_key_for_algorithm(key: str | bytes | JWKKeyData, algorithm: str):
return jwk.import_key(key, _key_type_for_algorithm(algorithm))
def _jwk_to_pem(key_data: JWKKeyData) -> str:
key_type = key_data.get("kty")
if key_type == "RSA":
return jwk.import_key(key_data, "RSA").as_pem().decode("utf-8")
if key_type == "EC":
return jwk.import_key(key_data, "EC").as_pem().decode("utf-8")
if key_type == "OKP":
return jwk.import_key(key_data, "OKP").as_pem().decode("utf-8")
raise ValueError(f"Unsupported JWK key type: {key_type!r}")
@ -72,6 +80,8 @@ class JWKData(TypedDict, total=False):
alg: str # Algorithm (e.g., "RS256")
n: str # Modulus (for RSA keys)
e: str # Exponent (for RSA keys)
crv: str # Curve name (for EC and OKP keys)
x: str # Public key coordinate (for EC and OKP keys)
x5c: list[str] # X.509 certificate chain (for JWKs)
x5t: str # X.509 certificate thumbprint (for JWKs)
@ -194,10 +204,11 @@ def _looks_like_pem_public_key(key: str | bytes) -> bool:
class JWTVerifier(TokenVerifier):
"""
JWT token verifier supporting both asymmetric (RSA/ECDSA) and symmetric (HMAC) algorithms.
JWT token verifier supporting asymmetric (RSA/ECDSA/EdDSA) and symmetric (HMAC) algorithms.
This verifier validates JWT tokens using various signing algorithms:
- **Asymmetric algorithms** (RS256/384/512, ES256/384/512, PS256/384/512):
- **Asymmetric algorithms** (RS256/384/512, ES256/384/512, PS256/384/512,
Ed25519, Ed448, and legacy EdDSA):
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
@ -232,7 +243,7 @@ class JWTVerifier(TokenVerifier):
jwks_uri: URI to fetch a JSON Web Key Set; used when verifying tokens with remote JWKS.
issuer: Expected issuer claim value or list of allowed issuer values.
audience: Expected audience claim value or list of allowed audience values.
algorithm: JWT signing algorithm to accept (default: "RS256"). Supported: HS256/384/512, RS256/384/512, ES256/384/512, PS256/384/512.
algorithm: JWT signing algorithm to accept (default: "RS256"). Supported: HS256/384/512, RS256/384/512, ES256/384/512, PS256/384/512, Ed25519, Ed448, and legacy EdDSA.
required_scopes: Scopes that must be present in validated tokens.
base_url: Base URL passed to the parent TokenVerifier.
ssrf_safe: If True, JWKS fetches use SSRF protection (HTTPS-only,
@ -275,6 +286,9 @@ class JWTVerifier(TokenVerifier):
"PS256",
"PS384",
"PS512",
"EdDSA",
"Ed25519",
"Ed448",
}:
raise ValueError(f"Unsupported algorithm: {algorithm}.")
@ -347,19 +361,31 @@ class JWTVerifier(TokenVerifier):
try:
jwks_data = await self._fetch_jwks()
# Cache all usable keys. A key that cannot be converted (e.g. an
# unsupported kty like OKP/Ed25519) is skipped rather than failing
# the whole set — per RFC 7517 §5, clients should ignore JWKs they
# don't understand. Otherwise one exotic key published by the
# authorization server would reject every token, including ones
# signed by supported keys in the same set (#4515).
# Cache all usable keys. A key that cannot be converted is skipped
# rather than failing the whole set — per RFC 7517 §5, clients
# should ignore JWKs they don't understand. Otherwise one exotic
# key published by the authorization server would reject every
# token, including ones signed by supported keys in the same set
# (#4515).
self._jwks_cache = {}
skipped_kids: set[str] = set()
expected_key_type = _key_type_for_algorithm(self.algorithm)
for key_data in jwks_data.get("keys", []):
if not isinstance(key_data, dict):
self.logger.debug("Skipping non-object JWKS entry: %r", key_data)
continue
key_kid = key_data.get("kid")
if key_data.get("kty") != expected_key_type:
self.logger.debug(
"Skipping JWKS key %r: key type %r is incompatible "
"with algorithm %s",
key_kid,
key_data.get("kty"),
self.algorithm,
)
if key_kid:
skipped_kids.add(key_kid)
continue
try:
public_key = _jwk_to_pem(key_data)
except (JoseError, TypeError, KeyError, ValueError) as e:

2
fastmcp_slim/pyproject.toml
View file

@ -96,7 +96,7 @@ server = [
"griffelib>=2.0.0",
"jsonref>=1.1.0",
"jsonschema-path>=0.3.4",
"joserfc>=1.1.0",
"joserfc>=1.5.0",
"openapi-pydantic>=0.5.1",
"packaging>=24.0",
"py-key-value-aio[filetree,keyring,memory]>=0.4.4,<0.5.0",

164
tests/server/auth/test_jwt_provider.py
View file

@ -4,6 +4,9 @@ from typing import Any, cast
from unittest.mock import MagicMock, patch
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed448 import Ed448PrivateKey
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from joserfc import jwk as jose_jwk
from joserfc import jwt
from joserfc.jws import JWSRegistry
@ -81,6 +84,49 @@ class SymmetricKeyHelper:
return token
def create_okp_key_pair(
private_key: Ed25519PrivateKey | Ed448PrivateKey,
) -> tuple[str, str]:
"""Serialize an EdDSA key pair as PEM strings."""
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
).decode()
public_pem = (
private_key.public_key()
.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
.decode()
)
return private_pem, public_pem
def create_okp_token(
private_key: str,
algorithm: str,
*,
kid: str | None = None,
) -> str:
"""Create a JWT signed by an OKP key."""
header = {"alg": algorithm}
if kid is not None:
header["kid"] = kid
return jwt.encode(
header,
{
"sub": "test-user",
"iss": "https://test.example.com",
"aud": "https://api.example.com",
"exp": int(time.time()) + 3600,
},
jose_jwk.import_key(private_key, "OKP"),
algorithms=[algorithm],
)
@pytest.fixture(scope="module")
def symmetric_key_helper() -> SymmetricKeyHelper:
"""Generate a symmetric key helper for testing."""
@ -485,6 +531,58 @@ class TestSymmetricKeyJWT:
assert access_token is None
class TestEdDSAJWT:
"""Tests for JWT verification using Edwards-curve keys."""
@pytest.mark.parametrize("algorithm", ["Ed25519", "Ed448"])
async def test_static_public_key(self, algorithm: str):
"""Fully specified EdDSA algorithms verify with a static public key."""
if algorithm == "Ed25519":
private_key = Ed25519PrivateKey.generate()
else:
private_key = Ed448PrivateKey.generate()
private_pem, public_pem = create_okp_key_pair(private_key)
verifier = JWTVerifier(
public_key=public_pem,
issuer="https://test.example.com",
audience="https://api.example.com",
algorithm=algorithm,
)
access_token = await verifier.load_access_token(
create_okp_token(private_pem, algorithm)
)
assert access_token is not None
assert access_token.client_id == "test-user"
@pytest.mark.filterwarnings(
"ignore:EdDSA is deprecated via RFC 9864:joserfc.errors.SecurityWarning"
)
async def test_legacy_eddsa_jwks(
self,
httpx_mock: HTTPXMock,
):
"""Legacy EdDSA tokens verify against an Ed25519 JWKS entry."""
private_pem, public_pem = create_okp_key_pair(Ed25519PrivateKey.generate())
public_jwk = jose_jwk.import_key(public_pem, "OKP").as_dict()
public_jwk.update(kid="ed25519-key", alg="EdDSA", use="sig")
httpx_mock.add_response(json={"keys": [public_jwk]})
verifier = JWTVerifier(
jwks_uri="https://test.example.com/.well-known/jwks.json",
issuer="https://test.example.com",
audience="https://api.example.com",
algorithm="EdDSA",
)
access_token = await verifier.load_access_token(
create_okp_token(private_pem, "EdDSA", kid="ed25519-key")
)
assert access_token is not None
assert access_token.client_id == "test-user"
def _create_token_without_sub(
rsa_key_pair: RSAKeyPair,
*,
@ -662,7 +760,7 @@ class TestBearerTokenJWKS:
assert access_token.claims.get("iss") == issuer
assert access_token.claims.get("aud") == audience
async def test_jwks_skips_unsupported_key_types(
async def test_jwks_skips_unusable_keys(
self,
rsa_key_pair: RSAKeyPair,
jwks_provider: JWTVerifier,
@ -670,28 +768,19 @@ class TestBearerTokenJWKS:
httpx_mock: HTTPXMock,
mock_dns,
):
"""An unsupported key type in the JWKS (e.g. OKP/Ed25519) must be
skipped, not poison the whole key set - #4515.
Some authorization servers (e.g. Rauthy, Ory Hydra) publish an
Ed25519 key alongside RSA keys; tokens signed by the RSA keys must
still verify.
"""
okp_key = cast(
"""An unusable key must not poison the whole key set - #4515."""
malformed_key = cast(
"JWKData",
{
"kty": "OKP",
"crv": "Ed25519",
"kid": "ed25519-key",
"alg": "EdDSA",
"kty": "RSA",
"kid": "malformed-key",
"use": "sig",
"x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo",
},
)
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
# Unsupported key FIRST, so an unguarded conversion loop would
# Malformed key FIRST, so an unguarded conversion loop would
# abort before reaching the RSA key the token needs
mock_jwks_data["keys"].insert(0, okp_key)
mock_jwks_data["keys"].insert(0, malformed_key)
httpx_mock.add_response(json=mock_jwks_data)
token = rsa_key_pair.create_token(
@ -705,37 +794,60 @@ class TestBearerTokenJWKS:
assert access_token is not None
assert access_token.client_id == "test-user"
async def test_jwks_with_only_unsupported_keys_rejects_cleanly(
async def test_jwks_ignores_other_algorithm_key_types_without_kid(
self,
rsa_key_pair: RSAKeyPair,
jwks_provider: JWTVerifier,
mock_jwks_data: JWKSData,
httpx_mock: HTTPXMock,
mock_dns,
):
"""Unrelated key types do not make a no-kid lookup ambiguous."""
_, public_pem = create_okp_key_pair(Ed25519PrivateKey.generate())
okp_key = jose_jwk.import_key(public_pem, "OKP").as_dict()
okp_key.update(kid="ed25519-key", alg="Ed25519", use="sig")
mock_jwks_data["keys"].append(cast("JWKData", okp_key))
httpx_mock.add_response(json=mock_jwks_data)
token = rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
)
access_token = await jwks_provider.load_access_token(token)
assert access_token is not None
assert access_token.client_id == "test-user"
async def test_jwks_with_only_unusable_keys_rejects_cleanly(
self,
rsa_key_pair: RSAKeyPair,
jwks_provider: JWTVerifier,
httpx_mock: HTTPXMock,
mock_dns,
):
"""If every key in the JWKS is unsupported, verification fails
"""If every key in the JWKS is unusable, verification fails
cleanly (returns None) rather than crashing - #4515."""
okp_only = {
unusable_only = {
"keys": [
cast(
"JWKData",
{
"kty": "OKP",
"crv": "Ed25519",
"kid": "ed25519-key",
"alg": "EdDSA",
"kty": "RSA",
"kid": "malformed-key",
"use": "sig",
"x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo",
},
)
]
}
httpx_mock.add_response(json=okp_only)
httpx_mock.add_response(json=unusable_only)
token = rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
kid="ed25519-key",
kid="malformed-key",
)
access_token = await jwks_provider.load_access_token(token)

2
uv.lock generated
View file

@ -1046,7 +1046,7 @@ requires-dist = [
{ name = "httpx2", marker = "extra == 'client'", specifier = ">=2.5.0" },
{ name = "httpx2", marker = "extra == 'mcp'", specifier = ">=2.5.0" },
{ name = "httpx2", marker = "extra == 'server'", specifier = ">=2.5.0" },
{ name = "joserfc", marker = "extra == 'server'", specifier = ">=1.1.0" },
{ name = "joserfc", marker = "extra == 'server'", specifier = ">=1.5.0" },
{ name = "jsonref", marker = "extra == 'gemini'", specifier = ">=1.1.0" },
{ name = "jsonref", marker = "extra == 'server'", specifier = ">=1.1.0" },
{ name = "jsonschema-path", marker = "extra == 'server'", specifier = ">=0.3.4" },