mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Add server-side CIMD support (SEP-991)
Implements OAuth Client ID Metadata Documents support, allowing servers to accept URL-based client_id values where the URL points to a JSON metadata document. This eliminates the need for Dynamic Client Registration (DCR) in many cases. - Private _cimd.py module (can swap for SDK when available) - SSRF protection for URL validation - Size-limited fetch with TTL caching - CIMD enabled by default (opt-out via cimd_enabled=False) - Advertises support via client_id_metadata_document_supported metadata - Client-side client_metadata_url parameter
This commit is contained in:
Jeremiah Lowin
parent
07750efaab
commit
006a284bf4
6 changed files with 827 additions and 3 deletions
|
|
@ -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:
|
||||
|
|
|
|||
269
src/fastmcp/server/auth/_cimd.py
Normal file
269
src/fastmcp/server/auth/_cimd.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
461
tests/server/auth/test_cimd.py
Normal file
461
tests/server/auth/test_cimd.py
Normal file
|
|
@ -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__
|
||||
Loading…
Add table
Add a link
Reference in a new issue