mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Enhance create_cimd_document with confidential client support
- Add jwks_uri and jwks parameters for private_key_jwt clients - Add client_uri, logo_uri, contacts optional metadata fields - Add strict URL validation per IETF draft (no fragments, credentials, dot segments) - Make redirect_uris required (security: no defaults) Note: private_key_jwt requires SDK server-side support (not yet available)
This commit is contained in:
Jeremiah Lowin
parent
ed5d672418
commit
e04b00b2f4
2 changed files with 299 additions and 77 deletions
|
|
@ -4,91 +4,150 @@ Utility for creating CIMD (Client ID Metadata Documents).
|
|||
CIMD allows OAuth clients to be identified by a URL pointing to their
|
||||
metadata document, eliminating the need for Dynamic Client Registration (DCR).
|
||||
|
||||
See SEP-991: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/991
|
||||
See:
|
||||
- SEP-991: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/991
|
||||
- IETF Draft: https://www.ietf.org/archive/id/draft-ietf-oauth-client-id-metadata-document-00.html
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from mcp.shared.auth import OAuthClientInformationFull
|
||||
from pydantic import AnyUrl
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pydantic import AnyHttpUrl
|
||||
|
||||
__all__ = ["create_cimd_document"]
|
||||
|
||||
# Common localhost redirect URIs for OAuth callbacks
|
||||
DEFAULT_REDIRECT_URIS = [
|
||||
"http://localhost:8080/callback",
|
||||
"http://localhost:8888/callback",
|
||||
"http://localhost:9000/callback",
|
||||
"http://127.0.0.1:8080/callback",
|
||||
"http://127.0.0.1:8888/callback",
|
||||
"http://127.0.0.1:9000/callback",
|
||||
]
|
||||
|
||||
def _validate_cimd_url(url: str) -> None:
|
||||
"""Validate URL meets CIMD requirements per IETF draft."""
|
||||
parsed = urlparse(url)
|
||||
|
||||
if parsed.scheme != "https":
|
||||
raise ValueError("CIMD URL must use HTTPS")
|
||||
|
||||
if parsed.path in ("", "/"):
|
||||
raise ValueError("CIMD URL must have a non-root path")
|
||||
|
||||
if parsed.fragment:
|
||||
raise ValueError("CIMD URL must not contain a fragment")
|
||||
|
||||
if parsed.username or parsed.password:
|
||||
raise ValueError("CIMD URL must not contain credentials")
|
||||
|
||||
# Check for dot segments (. or ..) in path
|
||||
path_segments = parsed.path.split("/")
|
||||
if any(seg in (".", "..") for seg in path_segments):
|
||||
raise ValueError("CIMD URL path must not contain dot segments")
|
||||
|
||||
|
||||
def create_cimd_document(
|
||||
url: str,
|
||||
*,
|
||||
redirect_uris: list[str],
|
||||
client_name: str = "FastMCP Client",
|
||||
redirect_uris: list[str] | None = None,
|
||||
scopes: list[str] | None = None,
|
||||
) -> dict:
|
||||
jwks_uri: str | None = None,
|
||||
jwks: dict[str, Any] | None = None,
|
||||
client_uri: str | None = None,
|
||||
logo_uri: str | None = None,
|
||||
contacts: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Create a CIMD (Client ID Metadata Document) for hosting.
|
||||
|
||||
The returned dict should be serialized as JSON and served at the given URL.
|
||||
The URL itself becomes the OAuth client_id.
|
||||
|
||||
For **public clients** (CLI apps, mobile apps), omit jwks_uri/jwks.
|
||||
These clients use PKCE for security but cannot prove client identity.
|
||||
|
||||
For **confidential clients**, provide jwks_uri or jwks containing your
|
||||
public key(s). The client must sign JWTs with the corresponding private
|
||||
key when calling the token endpoint, proving ownership of the client_id.
|
||||
|
||||
Args:
|
||||
url: The HTTPS URL where this document will be hosted.
|
||||
Must use HTTPS and have a non-root path.
|
||||
Must use HTTPS, have a non-root path, no fragment, no credentials.
|
||||
This exact URL becomes the client_id.
|
||||
client_name: Human-readable name for the client.
|
||||
redirect_uris: OAuth callback URIs. Defaults to common localhost variants
|
||||
for development use.
|
||||
redirect_uris: OAuth callback URIs. These must exactly match where your
|
||||
OAuth client will receive callbacks.
|
||||
client_name: Human-readable name for the client (displayed during consent).
|
||||
scopes: OAuth scopes to request (e.g., ["openid", "profile", "email"]).
|
||||
jwks_uri: URL to JSON Web Key Set for confidential clients.
|
||||
When provided, token_endpoint_auth_method is set to "private_key_jwt".
|
||||
jwks: Inline JSON Web Key Set (alternative to jwks_uri).
|
||||
When provided, token_endpoint_auth_method is set to "private_key_jwt".
|
||||
client_uri: URL to client's homepage (displayed during consent).
|
||||
logo_uri: URL to client's logo image (displayed during consent).
|
||||
contacts: List of contact emails for the client developer.
|
||||
|
||||
Returns:
|
||||
Dict ready to serialize as JSON and host at the URL.
|
||||
|
||||
Raises:
|
||||
ValueError: If URL doesn't use HTTPS or has no path.
|
||||
ValueError: If URL is invalid or both jwks_uri and jwks are provided.
|
||||
|
||||
Example:
|
||||
>>> import json
|
||||
Example (public client):
|
||||
>>> doc = create_cimd_document(
|
||||
... "https://example.com/.well-known/oauth-client.json",
|
||||
... client_name="My App",
|
||||
... scopes=["openid", "profile"],
|
||||
... redirect_uris=["http://localhost:8080/callback"],
|
||||
... client_name="My CLI App",
|
||||
... )
|
||||
>>> print(json.dumps(doc, indent=2))
|
||||
{
|
||||
"client_id": "https://example.com/.well-known/oauth-client.json",
|
||||
"client_name": "My App",
|
||||
"redirect_uris": ["http://localhost:8080/callback", ...],
|
||||
"grant_types": ["authorization_code", "refresh_token"],
|
||||
"response_types": ["code"],
|
||||
"token_endpoint_auth_method": "none",
|
||||
"scope": "openid profile"
|
||||
}
|
||||
>>> doc["token_endpoint_auth_method"]
|
||||
'none'
|
||||
|
||||
Example (confidential client):
|
||||
>>> doc = create_cimd_document(
|
||||
... "https://example.com/.well-known/oauth-client.json",
|
||||
... redirect_uris=["https://example.com/callback"],
|
||||
... client_name="My Web App",
|
||||
... jwks_uri="https://example.com/.well-known/jwks.json",
|
||||
... )
|
||||
>>> doc["token_endpoint_auth_method"]
|
||||
'private_key_jwt'
|
||||
"""
|
||||
if not url.startswith("https://"):
|
||||
raise ValueError("CIMD URL must use HTTPS")
|
||||
_validate_cimd_url(url)
|
||||
|
||||
# Check for non-root path
|
||||
from urllib.parse import urlparse
|
||||
if jwks_uri and jwks:
|
||||
raise ValueError("Provide either jwks_uri or jwks, not both")
|
||||
|
||||
parsed = urlparse(url)
|
||||
if parsed.path in ("", "/"):
|
||||
raise ValueError("CIMD URL must have a non-root path")
|
||||
# Determine auth method based on whether keys are provided
|
||||
is_confidential = jwks_uri is not None or jwks is not None
|
||||
token_endpoint_auth_method = "private_key_jwt" if is_confidential else "none"
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id=url,
|
||||
client_name=client_name,
|
||||
redirect_uris=[AnyUrl(u) for u in (redirect_uris or DEFAULT_REDIRECT_URIS)],
|
||||
grant_types=["authorization_code", "refresh_token"],
|
||||
response_types=["code"],
|
||||
token_endpoint_auth_method="none",
|
||||
scope=" ".join(scopes) if scopes else None,
|
||||
)
|
||||
# Build the document
|
||||
# Using dict directly instead of OAuthClientInformationFull to include jwks fields
|
||||
doc: dict[str, Any] = {
|
||||
"client_id": url,
|
||||
"client_name": client_name,
|
||||
"redirect_uris": redirect_uris,
|
||||
"grant_types": ["authorization_code", "refresh_token"],
|
||||
"response_types": ["code"],
|
||||
"token_endpoint_auth_method": token_endpoint_auth_method,
|
||||
}
|
||||
|
||||
return client.model_dump(mode="json", exclude_none=True)
|
||||
# Add optional fields
|
||||
if scopes:
|
||||
doc["scope"] = " ".join(scopes)
|
||||
|
||||
if jwks_uri:
|
||||
doc["jwks_uri"] = jwks_uri
|
||||
|
||||
if jwks:
|
||||
doc["jwks"] = jwks
|
||||
|
||||
if client_uri:
|
||||
# Validate it's a valid URL
|
||||
AnyHttpUrl(client_uri)
|
||||
doc["client_uri"] = client_uri
|
||||
|
||||
if logo_uri:
|
||||
# Validate it's a valid URL
|
||||
AnyHttpUrl(logo_uri)
|
||||
doc["logo_uri"] = logo_uri
|
||||
|
||||
if contacts:
|
||||
doc["contacts"] = contacts
|
||||
|
||||
return doc
|
||||
|
|
|
|||
|
|
@ -6,69 +6,232 @@ from fastmcp.client.auth import create_cimd_document
|
|||
|
||||
|
||||
class TestCreateCimdDocument:
|
||||
def test_basic_creation(self):
|
||||
doc = create_cimd_document("https://example.com/oauth/client.json")
|
||||
"""Tests for create_cimd_document utility."""
|
||||
|
||||
def test_basic_public_client(self):
|
||||
"""Public client with minimal required fields."""
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
assert doc["client_id"] == "https://example.com/oauth/client.json"
|
||||
assert doc["client_name"] == "FastMCP Client"
|
||||
assert doc["token_endpoint_auth_method"] == "none"
|
||||
assert doc["redirect_uris"] == ["http://localhost:8080/callback"]
|
||||
assert doc["grant_types"] == ["authorization_code", "refresh_token"]
|
||||
assert doc["response_types"] == ["code"]
|
||||
assert "redirect_uris" in doc
|
||||
assert len(doc["redirect_uris"]) > 0
|
||||
assert doc["token_endpoint_auth_method"] == "none"
|
||||
|
||||
def test_custom_client_name(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
client_name="My Custom Client",
|
||||
)
|
||||
|
||||
assert doc["client_name"] == "My Custom Client"
|
||||
|
||||
def test_custom_redirect_uris(self):
|
||||
def test_multiple_redirect_uris(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:3000/callback"],
|
||||
redirect_uris=[
|
||||
"http://localhost:3000/callback",
|
||||
"http://localhost:8080/callback",
|
||||
],
|
||||
)
|
||||
|
||||
assert doc["redirect_uris"] == ["http://localhost:3000/callback"]
|
||||
assert doc["redirect_uris"] == [
|
||||
"http://localhost:3000/callback",
|
||||
"http://localhost:8080/callback",
|
||||
]
|
||||
|
||||
def test_scopes(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
scopes=["openid", "profile", "email"],
|
||||
)
|
||||
|
||||
assert doc["scope"] == "openid profile email"
|
||||
|
||||
def test_no_scopes_excludes_field(self):
|
||||
doc = create_cimd_document("https://example.com/oauth/client.json")
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
assert "scope" not in doc
|
||||
|
||||
|
||||
class TestConfidentialClients:
|
||||
"""Tests for confidential client support with private_key_jwt."""
|
||||
|
||||
def test_jwks_uri_sets_private_key_jwt(self):
|
||||
"""Providing jwks_uri makes it a confidential client."""
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["https://example.com/callback"],
|
||||
jwks_uri="https://example.com/.well-known/jwks.json",
|
||||
)
|
||||
|
||||
assert doc["token_endpoint_auth_method"] == "private_key_jwt"
|
||||
assert doc["jwks_uri"] == "https://example.com/.well-known/jwks.json"
|
||||
assert "jwks" not in doc
|
||||
|
||||
def test_inline_jwks_sets_private_key_jwt(self):
|
||||
"""Providing inline jwks makes it a confidential client."""
|
||||
jwks = {
|
||||
"keys": [
|
||||
{
|
||||
"kty": "RSA",
|
||||
"use": "sig",
|
||||
"kid": "test-key-1",
|
||||
"n": "0vx7agoebGcQSuuPiLJXZptN9...",
|
||||
"e": "AQAB",
|
||||
}
|
||||
]
|
||||
}
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["https://example.com/callback"],
|
||||
jwks=jwks,
|
||||
)
|
||||
|
||||
assert doc["token_endpoint_auth_method"] == "private_key_jwt"
|
||||
assert doc["jwks"] == jwks
|
||||
assert "jwks_uri" not in doc
|
||||
|
||||
def test_cannot_provide_both_jwks_uri_and_jwks(self):
|
||||
"""Cannot provide both jwks_uri and inline jwks."""
|
||||
with pytest.raises(ValueError, match="either jwks_uri or jwks"):
|
||||
create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["https://example.com/callback"],
|
||||
jwks_uri="https://example.com/.well-known/jwks.json",
|
||||
jwks={"keys": []},
|
||||
)
|
||||
|
||||
|
||||
class TestOptionalMetadataFields:
|
||||
"""Tests for optional metadata fields."""
|
||||
|
||||
def test_client_uri(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
client_uri="https://example.com",
|
||||
)
|
||||
|
||||
assert doc["client_uri"] == "https://example.com"
|
||||
|
||||
def test_logo_uri(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
logo_uri="https://example.com/logo.png",
|
||||
)
|
||||
|
||||
assert doc["logo_uri"] == "https://example.com/logo.png"
|
||||
|
||||
def test_contacts(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
contacts=["admin@example.com", "security@example.com"],
|
||||
)
|
||||
|
||||
assert doc["contacts"] == ["admin@example.com", "security@example.com"]
|
||||
|
||||
def test_invalid_client_uri_rejected(self):
|
||||
with pytest.raises(Exception): # Pydantic validation error
|
||||
create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
client_uri="not-a-valid-url",
|
||||
)
|
||||
|
||||
def test_invalid_logo_uri_rejected(self):
|
||||
with pytest.raises(Exception): # Pydantic validation error
|
||||
create_cimd_document(
|
||||
"https://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
logo_uri="not-a-valid-url",
|
||||
)
|
||||
|
||||
|
||||
class TestUrlValidation:
|
||||
"""Tests for CIMD URL validation per IETF draft."""
|
||||
|
||||
def test_rejects_http_url(self):
|
||||
with pytest.raises(ValueError, match="must use HTTPS"):
|
||||
create_cimd_document("http://example.com/oauth/client.json")
|
||||
create_cimd_document(
|
||||
"http://example.com/oauth/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_rejects_root_path(self):
|
||||
with pytest.raises(ValueError, match="non-root path"):
|
||||
create_cimd_document("https://example.com/")
|
||||
create_cimd_document(
|
||||
"https://example.com/",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_rejects_no_path(self):
|
||||
with pytest.raises(ValueError, match="non-root path"):
|
||||
create_cimd_document("https://example.com")
|
||||
create_cimd_document(
|
||||
"https://example.com",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_well_known_path(self):
|
||||
doc = create_cimd_document("https://example.com/.well-known/oauth-client.json")
|
||||
def test_rejects_fragment(self):
|
||||
with pytest.raises(ValueError, match="fragment"):
|
||||
create_cimd_document(
|
||||
"https://example.com/client.json#section",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_rejects_credentials(self):
|
||||
with pytest.raises(ValueError, match="credentials"):
|
||||
create_cimd_document(
|
||||
"https://user:pass@example.com/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_rejects_dot_segments(self):
|
||||
with pytest.raises(ValueError, match="dot segments"):
|
||||
create_cimd_document(
|
||||
"https://example.com/../client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_rejects_single_dot_segment(self):
|
||||
with pytest.raises(ValueError, match="dot segments"):
|
||||
create_cimd_document(
|
||||
"https://example.com/./client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
def test_well_known_path_allowed(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/.well-known/oauth-client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
assert doc["client_id"] == "https://example.com/.well-known/oauth-client.json"
|
||||
|
||||
def test_excludes_none_values(self):
|
||||
doc = create_cimd_document("https://example.com/oauth/client.json")
|
||||
def test_query_string_allowed(self):
|
||||
"""Query strings are discouraged but permitted."""
|
||||
doc = create_cimd_document(
|
||||
"https://example.com/client.json?version=1",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
# These fields should not be present when None
|
||||
assert "client_secret" not in doc
|
||||
assert "client_id_issued_at" not in doc
|
||||
assert "client_secret_expires_at" not in doc
|
||||
assert "client_uri" not in doc
|
||||
assert "logo_uri" not in doc
|
||||
assert doc["client_id"] == "https://example.com/client.json?version=1"
|
||||
|
||||
def test_port_allowed(self):
|
||||
doc = create_cimd_document(
|
||||
"https://example.com:8443/client.json",
|
||||
redirect_uris=["http://localhost:8080/callback"],
|
||||
)
|
||||
|
||||
assert doc["client_id"] == "https://example.com:8443/client.json"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue