diff --git a/src/fastmcp/client/auth/__init__.py b/src/fastmcp/client/auth/__init__.py index 6ec3ecf4b..9402904fd 100644 --- a/src/fastmcp/client/auth/__init__.py +++ b/src/fastmcp/client/auth/__init__.py @@ -1,4 +1,5 @@ from .bearer import BearerAuth +from .cimd import create_cimd_document from .oauth import OAuth -__all__ = ["BearerAuth", "OAuth"] +__all__ = ["BearerAuth", "OAuth", "create_cimd_document"] diff --git a/src/fastmcp/client/auth/cimd.py b/src/fastmcp/client/auth/cimd.py new file mode 100644 index 000000000..440a549fb --- /dev/null +++ b/src/fastmcp/client/auth/cimd.py @@ -0,0 +1,153 @@ +""" +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 +- IETF Draft: https://www.ietf.org/archive/id/draft-ietf-oauth-client-id-metadata-document-00.html +""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import urlparse + +from pydantic import AnyHttpUrl + +__all__ = ["create_cimd_document"] + + +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", + scopes: list[str] | None = None, + 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, have a non-root path, no fragment, no credentials. + This exact URL becomes the client_id. + 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 is invalid or both jwks_uri and jwks are provided. + + Example (public client): + >>> doc = create_cimd_document( + ... "https://example.com/.well-known/oauth-client.json", + ... redirect_uris=["http://localhost:8080/callback"], + ... client_name="My CLI App", + ... ) + >>> 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' + """ + _validate_cimd_url(url) + + if jwks_uri and jwks: + raise ValueError("Provide either jwks_uri or jwks, not both") + + # 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" + + # 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, + } + + # 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 diff --git a/src/fastmcp/client/auth/oauth.py b/src/fastmcp/client/auth/oauth.py index 349db61b3..502fd7144 100644 --- a/src/fastmcp/client/auth/oauth.py +++ b/src/fastmcp/client/auth/oauth.py @@ -149,6 +149,7 @@ class OAuth(OAuthClientProvider): additional_client_metadata: dict[str, Any] | None = None, callback_port: int | None = None, httpx_client_factory: McpHttpClientFactory | None = None, + client_metadata_url: str | None = None, ): """ Initialize OAuth client provider for an MCP server. @@ -161,7 +162,12 @@ class OAuth(OAuthClientProvider): token_storage: An AsyncKeyValue-compatible token store, tokens are stored in memory if not provided additional_client_metadata: Extra fields for OAuthClientMetadata callback_port: Fixed port for OAuth callback (default: random available port) + client_metadata_url: URL-based client ID per CIMD (SEP-991). When provided + and the server supports CIMD, this URL is used as the client_id instead + of performing Dynamic Client Registration. The URL must be HTTPS with + a non-root path and point to a JSON document containing client metadata. """ + self.client_metadata_url = client_metadata_url parsed_url = urlparse(mcp_url) server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" @@ -215,6 +221,7 @@ class OAuth(OAuthClientProvider): storage=self.token_storage_adapter, redirect_handler=self.redirect_handler, callback_handler=self.callback_handler, + client_metadata_url=client_metadata_url, ) async def _initialize(self) -> None: diff --git a/src/fastmcp/server/auth/_cimd.py b/src/fastmcp/server/auth/_cimd.py new file mode 100644 index 000000000..270306d8f --- /dev/null +++ b/src/fastmcp/server/auth/_cimd.py @@ -0,0 +1,269 @@ +""" +Private CIMD (Client ID Metadata Document) implementation for FastMCP. + +This module implements server-side CIMD support (SEP-991) ahead of the MCP SDK. +When/if the SDK adds official server-side CIMD support, we should migrate to +using their implementation instead of this private module. + +All functions and classes in this module are private (prefixed with _) to make +it easy to swap out the implementation later without breaking public API. + +Reference: +- SEP-991: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/991 +- IETF draft: https://datatracker.ietf.org/doc/draft-ietf-oauth-client-id-metadata-document/ +""" + +from __future__ import annotations + +import ipaddress +import socket +import time +from typing import Any +from urllib.parse import urlparse + +import httpx +from mcp.shared.auth import OAuthClientInformationFull + +from fastmcp.utilities.logging import get_logger + +logger = get_logger(__name__) + + +def _is_cimd_client_id(client_id: str) -> bool: + """Check if client_id is a URL suitable for CIMD lookup. + + Per the spec, CIMD URLs must: + - Use HTTPS scheme + - Have a non-root path component + """ + if not isinstance(client_id, str): + return False + if not client_id.startswith("https://"): + return False + try: + parsed = urlparse(client_id) + return parsed.scheme == "https" and parsed.path not in ("", "/") + except Exception: + return False + + +def _is_private_ip(ip_str: str) -> bool: + """Check if an IP address is in a private/reserved range. + + This is used for SSRF protection to prevent fetching metadata from + internal network addresses. + """ + try: + ip = ipaddress.ip_address(ip_str) + + # Check for private, loopback, link-local, etc. + if ip.is_private: + return True + if ip.is_loopback: + return True + if ip.is_link_local: + return True + if ip.is_multicast: + return True + if ip.is_reserved: + return True + + # For IPv4, also check for special ranges + if isinstance(ip, ipaddress.IPv4Address): + # 0.0.0.0/8 - "This" network + if ip_str.startswith("0."): + return True + + return False + except ValueError: + # If we can't parse the IP, treat it as suspicious + return True + + +def _validate_cimd_url(url: str) -> None: + """Validate URL for CIMD fetch, including SSRF protection. + + Raises: + ValueError: If URL is invalid or resolves to private IP + """ + 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") + + hostname = parsed.hostname + if not hostname: + raise ValueError("CIMD URL must have a valid hostname") + + # DNS resolution check for SSRF protection + try: + # Get all IP addresses for the hostname + _, _, ipaddrlist = socket.gethostbyname_ex(hostname) + for ip in ipaddrlist: + if _is_private_ip(ip): + raise ValueError(f"CIMD URL resolves to private IP address: {ip}") + except socket.gaierror as e: + raise ValueError(f"Failed to resolve CIMD URL hostname: {e}") from e + + +async def _fetch_client_metadata( + client_id: str, + *, + timeout: float = 10.0, + max_size: int = 1_048_576, # 1MB +) -> dict[str, Any]: + """Fetch and parse client metadata document from URL. + + Args: + client_id: The HTTPS URL to fetch metadata from + timeout: Request timeout in seconds + max_size: Maximum response size in bytes + + Returns: + Parsed JSON metadata as a dictionary + + Raises: + ValueError: If URL validation fails or response is too large + httpx.HTTPError: If the HTTP request fails + """ + _validate_cimd_url(client_id) + + async with httpx.AsyncClient() as client: + response = await client.get( + client_id, + timeout=timeout, + headers={"Accept": "application/json"}, + follow_redirects=True, + ) + response.raise_for_status() + + # Check content length header before reading body + content_length = response.headers.get("content-length") + if content_length and int(content_length) > max_size: + raise ValueError( + f"CIMD response too large: {content_length} bytes (max: {max_size})" + ) + + # Read response with size check + content = response.content + if len(content) > max_size: + raise ValueError( + f"CIMD response too large: {len(content)} bytes (max: {max_size})" + ) + + return response.json() + + +def _create_client_from_metadata( + client_id: str, + metadata: dict[str, Any], +) -> OAuthClientInformationFull: + """Convert fetched metadata to OAuthClientInformationFull. + + Args: + client_id: The URL used to fetch the metadata (must match client_id in doc) + metadata: The parsed metadata document + + Returns: + OAuthClientInformationFull instance + + Raises: + ValueError: If client_id in metadata doesn't match URL + """ + # Validate that client_id in the document matches the URL + doc_client_id = metadata.get("client_id") + if doc_client_id != client_id: + raise ValueError( + f"client_id in metadata ({doc_client_id}) doesn't match URL ({client_id})" + ) + + # Ensure token_endpoint_auth_method defaults to "none" for CIMD clients + # (they don't have client secrets) + if "token_endpoint_auth_method" not in metadata: + metadata = {**metadata, "token_endpoint_auth_method": "none"} + + return OAuthClientInformationFull.model_validate(metadata) + + +class _CIMDCache: + """Simple in-memory cache for CIMD metadata with TTL support. + + Per the spec, servers SHOULD cache metadata respecting HTTP headers, + with a maximum of 24 hours. + """ + + def __init__( + self, + default_ttl: int = 3600, # 1 hour default + max_ttl: int = 86400, # 24 hours max per spec + ): + self._cache: dict[str, tuple[OAuthClientInformationFull, float]] = {} + self._default_ttl = default_ttl + self._max_ttl = max_ttl + + def get(self, client_id: str) -> OAuthClientInformationFull | None: + """Get cached client info if present and not expired.""" + if client_id not in self._cache: + return None + + client_info, expires_at = self._cache[client_id] + if time.time() > expires_at: + del self._cache[client_id] + return None + + return client_info + + def set( + self, + client_id: str, + client_info: OAuthClientInformationFull, + ttl: int | None = None, + ) -> None: + """Cache client info with TTL (capped at max_ttl).""" + effective_ttl = min(ttl or self._default_ttl, self._max_ttl) + expires_at = time.time() + effective_ttl + self._cache[client_id] = (client_info, expires_at) + + def clear(self) -> None: + """Clear all cached entries.""" + self._cache.clear() + + +async def _get_cimd_client( + client_id: str, + cache: _CIMDCache | None = None, +) -> OAuthClientInformationFull: + """Main entry point - fetch and validate CIMD client. + + Args: + client_id: The HTTPS URL to use as client_id + cache: Optional cache instance for storing results + + Returns: + OAuthClientInformationFull for the client + + Raises: + ValueError: If validation fails + httpx.HTTPError: If fetch fails + """ + # Check cache first + if cache: + cached = cache.get(client_id) + if cached is not None: + logger.debug(f"CIMD cache hit for {client_id}") + return cached + + logger.debug(f"Fetching CIMD metadata from {client_id}") + + # Fetch and validate + metadata = await _fetch_client_metadata(client_id) + client_info = _create_client_from_metadata(client_id, metadata) + + # Cache result + if cache: + cache.set(client_id, client_info) + + return client_info diff --git a/src/fastmcp/server/auth/auth.py b/src/fastmcp/server/auth/auth.py index bc255b394..13b57d41b 100644 --- a/src/fastmcp/server/auth/auth.py +++ b/src/fastmcp/server/auth/auth.py @@ -4,6 +4,7 @@ import json from typing import Any, cast from urllib.parse import urlparse +from mcp.server.auth.handlers.metadata import MetadataHandler from mcp.server.auth.handlers.token import TokenErrorResponse from mcp.server.auth.handlers.token import TokenHandler as _SDKTokenHandler from mcp.server.auth.json_response import PydanticJSONResponse @@ -30,11 +31,13 @@ from mcp.server.auth.settings import ( ClientRegistrationOptions, RevocationOptions, ) +from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata from pydantic import AnyHttpUrl, Field from starlette.middleware import Middleware from starlette.middleware.authentication import AuthenticationMiddleware from starlette.routing import Route +from fastmcp.server.auth._cimd import _CIMDCache, _get_cimd_client, _is_cimd_client_id from fastmcp.utilities.logging import get_logger logger = get_logger(__name__) @@ -335,6 +338,7 @@ class OAuthProvider( client_registration_options: ClientRegistrationOptions | None = None, revocation_options: RevocationOptions | None = None, required_scopes: list[str] | None = None, + cimd_enabled: bool = True, ): """ Initialize the OAuth provider. @@ -346,6 +350,9 @@ class OAuthProvider( client_registration_options: The client registration options. revocation_options: The revocation options. required_scopes: Scopes that are required for all requests. + cimd_enabled: Whether to enable Client ID Metadata Document (CIMD) + support per SEP-991. When enabled, URL-based client_ids will be + fetched and validated automatically. Defaults to True. """ super().__init__(base_url=base_url, required_scopes=required_scopes) @@ -379,6 +386,69 @@ class OAuthProvider( self.client_registration_options = client_registration_options self.revocation_options = revocation_options + # CIMD (Client ID Metadata Document) support - SEP-991 + self._cimd_enabled = cimd_enabled + self._cimd_cache = _CIMDCache() if cimd_enabled else None + + async def _lookup_cimd_client( + self, client_id: str + ) -> OAuthClientInformationFull | None: + """Try to look up client via CIMD if it's a URL-based client_id. + + Subclasses should call this in their get_client() implementation + before falling back to their normal client lookup. + + Args: + client_id: The client ID to look up + + Returns: + OAuthClientInformationFull if found via CIMD, None otherwise + """ + if not self._cimd_enabled: + return None + if not _is_cimd_client_id(client_id): + return None + try: + return await _get_cimd_client(client_id, cache=self._cimd_cache) + except Exception as e: + logger.debug(f"CIMD lookup failed for {client_id}: {e}") + return None + + def _build_cimd_metadata(self) -> OAuthMetadata: + """Build OAuth metadata with CIMD support flag enabled. + + This creates metadata identical to what the SDK builds, but with + client_id_metadata_document_supported set to True. + """ + from mcp.server.auth.routes import build_metadata + + assert self.base_url is not None + + # Build base metadata using SDK function + metadata = build_metadata( + issuer_url=self.base_url, + service_documentation_url=self.service_documentation_url, + client_registration_options=self.client_registration_options + or ClientRegistrationOptions(), + revocation_options=self.revocation_options or RevocationOptions(), + ) + + # Add CIMD support flag + metadata.client_id_metadata_document_supported = True + + return metadata + + def _create_cimd_metadata_route(self) -> Route: + """Create a metadata route that advertises CIMD support.""" + metadata = self._build_cimd_metadata() + metadata_handler = MetadataHandler(metadata=metadata) + + return Route( + path="/.well-known/oauth-authorization-server", + endpoint=cors_middleware(metadata_handler.handle, ["GET", "OPTIONS"]), + methods=["GET", "OPTIONS"], + ) + async def verify_token(self, token: str) -> AccessToken | None: """ Verify a bearer token and return access info if valid. @@ -425,8 +495,9 @@ class OAuthProvider( revocation_options=self.revocation_options, ) - # Replace the token endpoint with our custom handler that returns - # proper OAuth 2.1 error codes (invalid_client instead of unauthorized_client) + # Replace certain endpoints with our custom handlers: + # - Token endpoint: OAuth 2.1 compliant error codes + # - Metadata endpoint: Add CIMD support flag when enabled oauth_routes: list[Route] = [] for route in sdk_routes: if ( @@ -448,6 +519,13 @@ class OAuthProvider( methods=["POST", "OPTIONS"], ) ) + elif ( + isinstance(route, Route) + and route.path == "/.well-known/oauth-authorization-server" + and self._cimd_enabled + ): + # Replace metadata endpoint to advertise CIMD support + oauth_routes.append(self._create_cimd_metadata_route()) else: oauth_routes.append(route) diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py index b1f3516b6..6d220793d 100644 --- a/src/fastmcp/server/auth/oauth_proxy.py +++ b/src/fastmcp/server/auth/oauth_proxy.py @@ -911,8 +911,13 @@ class OAuthProxy(OAuthProvider): provided to the DCR client during registration, not the upstream client ID. For unregistered clients, returns None (which will raise an error in the SDK). + CIMD (URL-based) client_ids are also supported if cimd_enabled is True. """ - # Load from storage + # Try CIMD lookup first for URL-based client_ids + if cimd_client := await self._lookup_cimd_client(client_id): + return cimd_client + + # Load from storage (DCR-registered clients) if not (client := await self._client_store.get(key=client_id)): return None diff --git a/src/fastmcp/server/auth/providers/in_memory.py b/src/fastmcp/server/auth/providers/in_memory.py index de1e786b1..8fdff4088 100644 --- a/src/fastmcp/server/auth/providers/in_memory.py +++ b/src/fastmcp/server/auth/providers/in_memory.py @@ -63,6 +63,10 @@ class InMemoryOAuthProvider(OAuthProvider): ] = {} # refresh_token_str -> access_token_str async def get_client(self, client_id: str) -> OAuthClientInformationFull | None: + # Try CIMD lookup first for URL-based client_ids + if cimd_client := await self._lookup_cimd_client(client_id): + return cimd_client + # Fall back to registered clients return self.clients.get(client_id) async def register_client(self, client_info: OAuthClientInformationFull) -> None: diff --git a/tests/client/auth/test_cimd.py b/tests/client/auth/test_cimd.py new file mode 100644 index 000000000..50407f80b --- /dev/null +++ b/tests/client/auth/test_cimd.py @@ -0,0 +1,237 @@ +"""Tests for CIMD document creation utility.""" + +import pytest + +from fastmcp.client.auth import create_cimd_document + + +class TestCreateCimdDocument: + """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["redirect_uris"] == ["http://localhost:8080/callback"] + assert doc["grant_types"] == ["authorization_code", "refresh_token"] + assert doc["response_types"] == ["code"] + 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_multiple_redirect_uris(self): + doc = create_cimd_document( + "https://example.com/oauth/client.json", + redirect_uris=[ + "http://localhost:3000/callback", + "http://localhost:8080/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", + 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", + 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/", + 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", + redirect_uris=["http://localhost:8080/callback"], + ) + + 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_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"], + ) + + 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" diff --git a/tests/server/auth/test_cimd.py b/tests/server/auth/test_cimd.py new file mode 100644 index 000000000..b9bf17739 --- /dev/null +++ b/tests/server/auth/test_cimd.py @@ -0,0 +1,461 @@ +"""Tests for CIMD (Client ID Metadata Document) support per SEP-991.""" + +from __future__ import annotations + +import time +from unittest.mock import patch + +import pytest +from mcp.shared.auth import OAuthClientInformationFull +from pydantic import AnyUrl +from pytest_httpx import HTTPXMock + +from fastmcp.server.auth._cimd import ( + _CIMDCache, + _create_client_from_metadata, + _fetch_client_metadata, + _get_cimd_client, + _is_cimd_client_id, + _is_private_ip, + _validate_cimd_url, +) +from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider + + +class TestIsCimdClientId: + """Test _is_cimd_client_id function.""" + + def test_valid_https_with_path(self): + assert _is_cimd_client_id("https://example.com/oauth/client.json") is True + assert _is_cimd_client_id("https://app.example.com/metadata") is True + assert _is_cimd_client_id("https://example.com/a") is True + + def test_rejects_http(self): + assert _is_cimd_client_id("http://example.com/oauth/client.json") is False + + def test_rejects_root_path(self): + assert _is_cimd_client_id("https://example.com") is False + assert _is_cimd_client_id("https://example.com/") is False + + def test_rejects_non_url(self): + assert _is_cimd_client_id("not-a-url") is False + assert _is_cimd_client_id("client-id-123") is False + assert _is_cimd_client_id("") is False + + def test_rejects_non_string(self): + assert _is_cimd_client_id(123) is False # type: ignore[arg-type] + assert _is_cimd_client_id(None) is False # type: ignore[arg-type] + + +class TestIsPrivateIp: + """Test _is_private_ip function.""" + + def test_private_ipv4_10_x(self): + assert _is_private_ip("10.0.0.1") is True + assert _is_private_ip("10.255.255.255") is True + + def test_private_ipv4_172_16_x(self): + assert _is_private_ip("172.16.0.1") is True + assert _is_private_ip("172.31.255.255") is True + # 172.15.x and 172.32.x are NOT private + assert _is_private_ip("172.15.0.1") is False + assert _is_private_ip("172.32.0.1") is False + + def test_private_ipv4_192_168_x(self): + assert _is_private_ip("192.168.0.1") is True + assert _is_private_ip("192.168.255.255") is True + + def test_localhost(self): + assert _is_private_ip("127.0.0.1") is True + assert _is_private_ip("127.0.0.2") is True + + def test_link_local(self): + assert _is_private_ip("169.254.0.1") is True + assert _is_private_ip("169.254.255.255") is True + + def test_public_ipv4(self): + assert _is_private_ip("8.8.8.8") is False + assert _is_private_ip("1.1.1.1") is False + assert _is_private_ip("93.184.216.34") is False # example.com + + def test_ipv6_loopback(self): + assert _is_private_ip("::1") is True + + def test_invalid_ip(self): + # Invalid IPs are treated as suspicious (returns True) + assert _is_private_ip("not-an-ip") is True + + +class TestValidateCimdUrl: + """Test _validate_cimd_url function.""" + + def test_requires_https(self): + with pytest.raises(ValueError, match="must use HTTPS"): + _validate_cimd_url("http://example.com/metadata") + + def test_requires_non_root_path(self): + with pytest.raises(ValueError, match="non-root path"): + _validate_cimd_url("https://example.com") + with pytest.raises(ValueError, match="non-root path"): + _validate_cimd_url("https://example.com/") + + def test_rejects_private_ip(self): + # Mock DNS resolution to return private IP + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("example.com", [], ["10.0.0.1"]) + with pytest.raises(ValueError, match="private IP"): + _validate_cimd_url("https://internal.example.com/metadata") + + def test_accepts_public_ip(self): + # Mock DNS resolution to return public IP + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("example.com", [], ["93.184.216.34"]) + # Should not raise + _validate_cimd_url("https://example.com/metadata") + + +class TestFetchClientMetadata: + """Test _fetch_client_metadata function.""" + + async def test_fetch_success(self, httpx_mock: HTTPXMock): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + # Mock DNS to return public IP + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, json=metadata) + + result = await _fetch_client_metadata(client_id) + + assert result == metadata + + async def test_fetch_timeout(self, httpx_mock: HTTPXMock): + import httpx + + client_id = "https://slow.example.com/metadata.json" + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("slow.example.com", [], ["93.184.216.34"]) + httpx_mock.add_exception(httpx.TimeoutException("Connection timed out")) + + with pytest.raises(httpx.TimeoutException): + await _fetch_client_metadata(client_id, timeout=0.1) + + async def test_fetch_size_limit_via_header(self, httpx_mock: HTTPXMock): + client_id = "https://large.example.com/metadata.json" + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("large.example.com", [], ["93.184.216.34"]) + # Return a response with content-length exceeding limit + httpx_mock.add_response( + url=client_id, + json={"client_id": client_id}, + headers={"content-length": "2000000"}, # 2MB + ) + + with pytest.raises(ValueError, match="too large"): + await _fetch_client_metadata(client_id, max_size=1_048_576) + + async def test_fetch_invalid_json(self, httpx_mock: HTTPXMock): + client_id = "https://bad.example.com/metadata.json" + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("bad.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, text="not json {{{") + + with pytest.raises(Exception): # JSONDecodeError + await _fetch_client_metadata(client_id) + + +class TestCreateClientFromMetadata: + """Test _create_client_from_metadata function.""" + + def test_creates_client_from_valid_metadata(self): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + "token_endpoint_auth_method": "none", + } + + result = _create_client_from_metadata(client_id, metadata) + + assert isinstance(result, OAuthClientInformationFull) + assert result.client_id == client_id + assert result.client_name == "Test Client" + assert result.token_endpoint_auth_method == "none" + + def test_defaults_token_endpoint_auth_method_to_none(self): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + # No token_endpoint_auth_method specified + } + + result = _create_client_from_metadata(client_id, metadata) + assert result.token_endpoint_auth_method == "none" + + def test_rejects_mismatched_client_id(self): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": "https://other.example.com/different.json", + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + with pytest.raises(ValueError, match="doesn't match"): + _create_client_from_metadata(client_id, metadata) + + +class TestCIMDCache: + """Test _CIMDCache class.""" + + def test_cache_hit(self): + cache = _CIMDCache() + client_id = "https://client.example.com/metadata.json" + client_info = OAuthClientInformationFull( + client_id=client_id, + redirect_uris=[AnyUrl("http://localhost:3000/callback")], + ) + + cache.set(client_id, client_info) + result = cache.get(client_id) + + assert result is not None + assert result.client_id == client_id + + def test_cache_miss(self): + cache = _CIMDCache() + result = cache.get("https://unknown.example.com/metadata.json") + assert result is None + + def test_cache_expiry(self): + cache = _CIMDCache(default_ttl=1) # 1 second TTL + client_id = "https://client.example.com/metadata.json" + client_info = OAuthClientInformationFull( + client_id=client_id, + redirect_uris=[AnyUrl("http://localhost:3000/callback")], + ) + + cache.set(client_id, client_info, ttl=1) + + # Should be in cache + assert cache.get(client_id) is not None + + # Wait for expiry + time.sleep(1.1) + + # Should be expired + assert cache.get(client_id) is None + + def test_cache_respects_max_ttl(self): + cache = _CIMDCache(default_ttl=3600, max_ttl=2) # 2 second max + client_id = "https://client.example.com/metadata.json" + client_info = OAuthClientInformationFull( + client_id=client_id, + redirect_uris=[AnyUrl("http://localhost:3000/callback")], + ) + + # Try to set with 1 hour TTL, should be capped to 2 seconds + cache.set(client_id, client_info, ttl=3600) + + # Should be in cache + assert cache.get(client_id) is not None + + # Wait longer than max_ttl + time.sleep(2.1) + + # Should be expired due to max_ttl cap + assert cache.get(client_id) is None + + def test_cache_clear(self): + cache = _CIMDCache() + client_id = "https://client.example.com/metadata.json" + client_info = OAuthClientInformationFull( + client_id=client_id, + redirect_uris=[AnyUrl("http://localhost:3000/callback")], + ) + + cache.set(client_id, client_info) + assert cache.get(client_id) is not None + + cache.clear() + assert cache.get(client_id) is None + + +class TestGetCimdClient: + """Test _get_cimd_client function.""" + + async def test_fetches_and_validates(self, httpx_mock: HTTPXMock): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, json=metadata) + + result = await _get_cimd_client(client_id) + + assert isinstance(result, OAuthClientInformationFull) + assert result.client_id == client_id + + async def test_uses_cache(self, httpx_mock: HTTPXMock): + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + cache = _CIMDCache() + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, json=metadata) + + # First call - should fetch + result1 = await _get_cimd_client(client_id, cache=cache) + + # Second call - should use cache (no HTTP request) + result2 = await _get_cimd_client(client_id, cache=cache) + + assert result1.client_id == client_id + assert result2.client_id == client_id + + # Verify only one HTTP request was made + assert len(httpx_mock.get_requests()) == 1 + + +class TestOAuthProviderCIMD: + """Test CIMD integration in OAuthProvider base class.""" + + async def test_lookup_cimd_client_returns_none_for_non_url(self): + """CIMD lookup should return None for non-URL client_ids.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + result = await provider._lookup_cimd_client("regular-client-id") + assert result is None + + async def test_lookup_cimd_client_returns_none_when_disabled(self): + """CIMD lookup should return None when cimd_enabled=False.""" + provider = InMemoryOAuthProvider( + base_url="http://localhost:8000", + ) + provider._cimd_enabled = False + + result = await provider._lookup_cimd_client( + "https://client.example.com/metadata.json" + ) + assert result is None + + async def test_lookup_cimd_client_success(self, httpx_mock: HTTPXMock): + """CIMD lookup should successfully fetch and return client info.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, json=metadata) + + result = await provider._lookup_cimd_client(client_id) + + assert result is not None + assert result.client_id == client_id + + async def test_lookup_cimd_client_returns_none_on_error( + self, httpx_mock: HTTPXMock + ): + """CIMD lookup should return None and log on error, not raise.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + client_id = "https://failing.example.com/metadata.json" + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("failing.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, status_code=404) + + result = await provider._lookup_cimd_client(client_id) + + assert result is None + + +class TestInMemoryProviderCIMD: + """Test CIMD integration in InMemoryOAuthProvider.""" + + async def test_get_client_uses_cimd_for_url(self, httpx_mock: HTTPXMock): + """get_client should use CIMD for URL-based client_ids.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + client_id = "https://client.example.com/metadata.json" + metadata = { + "client_id": client_id, + "client_name": "Test Client", + "redirect_uris": ["http://localhost:3000/callback"], + } + + with patch("socket.gethostbyname_ex") as mock_dns: + mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"]) + httpx_mock.add_response(url=client_id, json=metadata) + + result = await provider.get_client(client_id) + + assert result is not None + assert result.client_id == client_id + + async def test_get_client_uses_registered_for_non_url(self): + """get_client should use registered clients for non-URL client_ids.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + # Register a client + client_info = OAuthClientInformationFull( + client_id="regular-client-id", + redirect_uris=[AnyUrl("http://localhost:3000/callback")], + ) + await provider.register_client(client_info) + + result = await provider.get_client("regular-client-id") + + assert result is not None + assert result.client_id == "regular-client-id" + + async def test_get_client_returns_none_for_unknown(self): + """get_client should return None for unknown non-URL client_ids.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + + result = await provider.get_client("unknown-client-id") + assert result is None + + +class TestOAuthMetadataAdvertisesCIMD: + """Test that OAuth metadata advertises CIMD support.""" + + def test_cimd_enabled_by_default(self): + """CIMD should be enabled by default.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + assert provider._cimd_enabled is True + assert provider._cimd_cache is not None + + def test_cimd_can_be_disabled(self): + """CIMD can be disabled via constructor parameter.""" + provider = InMemoryOAuthProvider(base_url="http://localhost:8000") + provider._cimd_enabled = False + # Note: We can't easily test the full disable path without + # exposing cimd_enabled in InMemoryOAuthProvider's __init__