modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix OAuth discovery fallback and URL ordering (#1624)

Browse source
This commit is contained in:
Max Isbey 2025-11-13 19:37:24 +00:00 • committed by GitHub
commit 91ccdb3d65
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 319 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

30
src/mcp/client/auth/oauth2.py
View file

@ -21,14 +21,14 @@ from pydantic import BaseModel, Field, ValidationError
from mcp.client.auth import OAuthFlowError, OAuthTokenError
from mcp.client.auth.utils import (
build_protected_resource_discovery_urls,
build_oauth_authorization_server_metadata_discovery_urls,
build_protected_resource_metadata_discovery_urls,
create_client_registration_request,
create_oauth_metadata_request,
extract_field_from_www_auth,
extract_resource_metadata_from_www_auth,
extract_scope_from_www_auth,
get_client_metadata_scopes,
get_discovery_urls,
handle_auth_metadata_response,
handle_protected_resource_response,
handle_registration_response,
@ -463,10 +463,10 @@ class OAuthClientProvider(httpx.Auth):
www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
# Step 1: Discover protected resource metadata (SEP-985 with fallback support)
prm_discovery_urls = build_protected_resource_discovery_urls(
prm_discovery_urls = build_protected_resource_metadata_discovery_urls(
www_auth_resource_metadata_url, self.context.server_url
)
prm_discovery_success = False
for url in prm_discovery_urls: # pragma: no branch
discovery_request = create_oauth_metadata_request(url)
@ -474,23 +474,23 @@ class OAuthClientProvider(httpx.Auth):
prm = await handle_protected_resource_response(discovery_response)
if prm:
prm_discovery_success = True
# saving the response metadata
self.context.protected_resource_metadata = prm
if prm.authorization_servers: # pragma: no branch
self.context.auth_server_url = str(prm.authorization_servers[0])
# todo: try all authorization_servers to find the OASM
assert (
len(prm.authorization_servers) > 0
) # this is always true as authorization_servers has a min length of 1
self.context.auth_server_url = str(prm.authorization_servers[0])
break
else:
logger.debug(f"Protected resource metadata discovery failed: {url}")
if not prm_discovery_success:
raise OAuthFlowError(
"Protected resource metadata discovery failed: no valid metadata found"
) # pragma: no cover
# Step 2: Discover OAuth metadata (with fallback for legacy servers)
asm_discovery_urls = get_discovery_urls(self.context.auth_server_url or self.context.server_url)
asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
self.context.auth_server_url, self.context.server_url
)
# Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
for url in asm_discovery_urls: # pragma: no cover
oauth_metadata_request = create_oauth_metadata_request(url)
oauth_metadata_response = yield oauth_metadata_request

39
src/mcp/client/auth/utils.py
View file

@ -64,7 +64,7 @@ def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
return extract_field_from_www_auth(response, "resource_metadata")
def build_protected_resource_discovery_urls(www_auth_url: str | None, server_url: str) -> list[str]:
def build_protected_resource_metadata_discovery_urls(www_auth_url: str | None, server_url: str) -> list[str]:
"""
Build ordered list of URLs to try for protected resource metadata discovery.
@ -126,8 +126,21 @@ def get_client_metadata_scopes(
return None
def get_discovery_urls(auth_server_url: str) -> list[str]:
"""Generate ordered list of (url, type) tuples for discovery attempts."""
def build_oauth_authorization_server_metadata_discovery_urls(auth_server_url: str | None, server_url: str) -> list[str]:
"""
Generate ordered list of (url, type) tuples for discovery attempts.
Args:
auth_server_url: URL for the OAuth Authorization Metadata URL if found, otherwise None
server_url: URL for the MCP server, used as a fallback if auth_server_url is None
"""
if not auth_server_url:
# Legacy path using the 2025-03-26 spec:
# link: https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization
parsed = urlparse(server_url)
return [f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-authorization-server"]
urls: list[str] = []
parsed = urlparse(auth_server_url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
@ -137,18 +150,22 @@ def get_discovery_urls(auth_server_url: str) -> list[str]:
oauth_path = f"/.well-known/oauth-authorization-server{parsed.path.rstrip('/')}"
urls.append(urljoin(base_url, oauth_path))
# OAuth root fallback
urls.append(urljoin(base_url, "/.well-known/oauth-authorization-server"))
# RFC 8414 section 5: Path-aware OIDC discovery
# See https://www.rfc-editor.org/rfc/rfc8414.html#section-5
if parsed.path and parsed.path != "/":
# RFC 8414 section 5: Path-aware OIDC discovery
# See https://www.rfc-editor.org/rfc/rfc8414.html#section-5
oidc_path = f"/.well-known/openid-configuration{parsed.path.rstrip('/')}"
urls.append(urljoin(base_url, oidc_path))
# https://openid.net/specs/openid-connect-discovery-1_0.html
oidc_path = f"{parsed.path.rstrip('/')}/.well-known/openid-configuration"
urls.append(urljoin(base_url, oidc_path))
return urls
# OAuth root
urls.append(urljoin(base_url, "/.well-known/oauth-authorization-server"))
# OIDC 1.0 fallback (appends to full URL per OIDC spec)
oidc_fallback = f"{auth_server_url.rstrip('/')}/.well-known/openid-configuration"
urls.append(oidc_fallback)
# https://openid.net/specs/openid-connect-discovery-1_0.html
urls.append(urljoin(base_url, "/.well-known/openid-configuration"))
return urls

291
tests/client/test_auth.py
View file

@ -12,13 +12,13 @@ from pydantic import AnyHttpUrl, AnyUrl
from mcp.client.auth import OAuthClientProvider, PKCEParameters
from mcp.client.auth.utils import (
build_protected_resource_discovery_urls,
build_oauth_authorization_server_metadata_discovery_urls,
build_protected_resource_metadata_discovery_urls,
create_oauth_metadata_request,
extract_field_from_www_auth,
extract_resource_metadata_from_www_auth,
extract_scope_from_www_auth,
get_client_metadata_scopes,
get_discovery_urls,
handle_registration_response,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken, ProtectedResourceMetadata
@ -275,7 +275,7 @@ class TestOAuthFlow:
status_code=401, headers={}, request=httpx.Request("GET", "https://request-api.example.com")
)
urls = build_protected_resource_discovery_urls(
urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
assert len(urls) == 1
@ -286,7 +286,7 @@ class TestOAuthFlow:
'Bearer resource_metadata="https://prm.example.com/.well-known/oauth-protected-resource/path"'
)
urls = build_protected_resource_discovery_urls(
urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
assert len(urls) == 2
@ -307,14 +307,69 @@ class TestOAuthFlow:
class TestOAuthFallback:
"""Test OAuth discovery fallback behavior for legacy (act as AS not RS) servers."""
@pytest.mark.anyio
async def test_oauth_discovery_legacy_fallback_when_no_prm(self):
"""Test that when PRM discovery fails, only root OAuth URL is tried (March 2025 spec)."""
# When auth_server_url is None (PRM failed), we use server_url and only try root
discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(None, "https://mcp.linear.app/sse")
# Should only try the root URL (legacy behavior)
assert discovery_urls == [
"https://mcp.linear.app/.well-known/oauth-authorization-server",
]
@pytest.mark.anyio
async def test_oauth_discovery_path_aware_when_auth_server_has_path(self):
"""Test that when auth server URL has a path, only path-based URLs are tried."""
discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
"https://auth.example.com/tenant1", "https://api.example.com/mcp"
)
# Should try path-based URLs only (no root URLs)
assert discovery_urls == [
"https://auth.example.com/.well-known/oauth-authorization-server/tenant1",
"https://auth.example.com/.well-known/openid-configuration/tenant1",
"https://auth.example.com/tenant1/.well-known/openid-configuration",
]
@pytest.mark.anyio
async def test_oauth_discovery_root_when_auth_server_has_no_path(self):
"""Test that when auth server URL has no path, only root URLs are tried."""
discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
"https://auth.example.com", "https://api.example.com/mcp"
)
# Should try root URLs only
assert discovery_urls == [
"https://auth.example.com/.well-known/oauth-authorization-server",
"https://auth.example.com/.well-known/openid-configuration",
]
@pytest.mark.anyio
async def test_oauth_discovery_root_when_auth_server_has_only_slash(self):
"""Test that when auth server URL has only trailing slash, treated as root."""
discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
"https://auth.example.com/", "https://api.example.com/mcp"
)
# Should try root URLs only
assert discovery_urls == [
"https://auth.example.com/.well-known/oauth-authorization-server",
"https://auth.example.com/.well-known/openid-configuration",
]
@pytest.mark.anyio
async def test_oauth_discovery_fallback_order(self, oauth_provider: OAuthClientProvider):
"""Test fallback URL construction order."""
discovery_urls = get_discovery_urls(oauth_provider.context.auth_server_url or oauth_provider.context.server_url)
"""Test fallback URL construction order when auth server URL has a path."""
# Simulate PRM discovery returning an auth server URL with a path
oauth_provider.context.auth_server_url = oauth_provider.context.server_url
discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
oauth_provider.context.auth_server_url, oauth_provider.context.server_url
)
assert discovery_urls == [
"https://api.example.com/.well-known/oauth-authorization-server/v1/mcp",
"https://api.example.com/.well-known/oauth-authorization-server",
"https://api.example.com/.well-known/openid-configuration/v1/mcp",
"https://api.example.com/v1/mcp/.well-known/openid-configuration",
]
@ -358,13 +413,14 @@ class TestOAuthFallback:
assert discovery_request.method == "GET"
# Send a successful discovery response with minimal protected resource metadata
# Note: auth server URL has a path (/v1/mcp), so only path-based URLs will be tried
discovery_response = httpx.Response(
200,
content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com/v1/mcp"]}',
request=discovery_request,
)
# Next request should be to discover OAuth metadata
# Next request should be to discover OAuth metadata at path-aware OAuth URL
oauth_metadata_request_1 = await auth_flow.asend(discovery_response)
assert (
str(oauth_metadata_request_1.url)
@ -379,9 +435,9 @@ class TestOAuthFallback:
request=oauth_metadata_request_1,
)
# Next request should be to discover OAuth metadata at the next endpoint
# Next request should be path-aware OIDC URL (not root URL since auth server has path)
oauth_metadata_request_2 = await auth_flow.asend(oauth_metadata_response_1)
assert str(oauth_metadata_request_2.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
assert str(oauth_metadata_request_2.url) == "https://auth.example.com/.well-known/openid-configuration/v1/mcp"
assert oauth_metadata_request_2.method == "GET"
# Send a 400 response
@ -391,9 +447,9 @@ class TestOAuthFallback:
request=oauth_metadata_request_2,
)
# Next request should be to discover OAuth metadata at the next endpoint
# Next request should be OIDC path-appended URL
oauth_metadata_request_3 = await auth_flow.asend(oauth_metadata_response_2)
assert str(oauth_metadata_request_3.url) == "https://auth.example.com/.well-known/openid-configuration/v1/mcp"
assert str(oauth_metadata_request_3.url) == "https://auth.example.com/v1/mcp/.well-known/openid-configuration"
assert oauth_metadata_request_3.method == "GET"
# Send a 500 response
@ -408,7 +464,8 @@ class TestOAuthFallback:
return_value=("test_auth_code", "test_code_verifier")
)
# Next request should fall back to legacy behavior and auth with the RS (mocked /authorize, next is /token)
# All path-based URLs failed, flow continues with default endpoints
# Next request should be token exchange using MCP server base URL (fallback when OAuth metadata not found)
token_request = await auth_flow.asend(oauth_metadata_response_3)
assert str(token_request.url) == "https://api.example.com/token"
assert token_request.method == "POST"
@ -1055,6 +1112,210 @@ def test_build_metadata(
)
class TestLegacyServerFallback:
"""Test backward compatibility with legacy servers that don't support PRM (issue #1495)."""
@pytest.mark.anyio
async def test_legacy_server_no_prm_falls_back_to_root_oauth_discovery(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test that when PRM discovery fails completely, we fall back to root OAuth discovery (March 2025 spec)."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> tuple[str, str | None]:
return "test_auth_code", "test_state" # pragma: no cover
# Simulate a legacy server like Linear
provider = OAuthClientProvider(
server_url="https://mcp.linear.app/sse",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
# Mock client info to skip DCR
provider.context.client_info = OAuthClientInformationFull(
client_id="existing_client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx.Request("GET", "https://mcp.linear.app/sse")
auth_flow = provider.async_auth_flow(test_request)
# First request
request = await auth_flow.__anext__()
assert "Authorization" not in request.headers
# Send 401 without WWW-Authenticate header (typical legacy server)
response = httpx.Response(401, headers={}, request=test_request)
# Should try path-based PRM first
prm_request_1 = await auth_flow.asend(response)
assert str(prm_request_1.url) == "https://mcp.linear.app/.well-known/oauth-protected-resource/sse"
# PRM returns 404
prm_response_1 = httpx.Response(404, request=prm_request_1)
# Should try root-based PRM
prm_request_2 = await auth_flow.asend(prm_response_1)
assert str(prm_request_2.url) == "https://mcp.linear.app/.well-known/oauth-protected-resource"
# PRM returns 404 again - all PRM URLs failed
prm_response_2 = httpx.Response(404, request=prm_request_2)
# Should fall back to root OAuth discovery (March 2025 spec behavior)
oauth_metadata_request = await auth_flow.asend(prm_response_2)
assert str(oauth_metadata_request.url) == "https://mcp.linear.app/.well-known/oauth-authorization-server"
assert oauth_metadata_request.method == "GET"
# Send successful OAuth metadata response
oauth_metadata_response = httpx.Response(
200,
content=(
b'{"issuer": "https://mcp.linear.app", '
b'"authorization_endpoint": "https://mcp.linear.app/authorize", '
b'"token_endpoint": "https://mcp.linear.app/token"}'
),
request=oauth_metadata_request,
)
# Mock authorization
provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Next should be token exchange
token_request = await auth_flow.asend(oauth_metadata_response)
assert str(token_request.url) == "https://mcp.linear.app/token"
# Send successful token response
token_response = httpx.Response(
200,
content=b'{"access_token": "linear_token", "token_type": "Bearer", "expires_in": 3600}',
request=token_request,
)
# Final request with auth header
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer linear_token"
assert str(final_request.url) == "https://mcp.linear.app/sse"
# Complete flow
final_response = httpx.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
@pytest.mark.anyio
async def test_legacy_server_with_different_prm_and_root_urls(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test PRM fallback with different WWW-Authenticate and root URLs."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> tuple[str, str | None]:
return "test_auth_code", "test_state" # pragma: no cover
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
provider.context.client_info = OAuthClientInformationFull(
client_id="existing_client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
await auth_flow.__anext__()
# 401 with custom WWW-Authenticate PRM URL
response = httpx.Response(
401,
headers={
"WWW-Authenticate": 'Bearer resource_metadata="https://custom.prm.com/.well-known/oauth-protected-resource"'
},
request=test_request,
)
# Try custom PRM URL first
prm_request_1 = await auth_flow.asend(response)
assert str(prm_request_1.url) == "https://custom.prm.com/.well-known/oauth-protected-resource"
# Returns 500
prm_response_1 = httpx.Response(500, request=prm_request_1)
# Try path-based fallback
prm_request_2 = await auth_flow.asend(prm_response_1)
assert str(prm_request_2.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
# Returns 404
prm_response_2 = httpx.Response(404, request=prm_request_2)
# Try root fallback
prm_request_3 = await auth_flow.asend(prm_response_2)
assert str(prm_request_3.url) == "https://api.example.com/.well-known/oauth-protected-resource"
# Also returns 404 - all PRM URLs failed
prm_response_3 = httpx.Response(404, request=prm_request_3)
# Should fall back to root OAuth discovery
oauth_metadata_request = await auth_flow.asend(prm_response_3)
assert str(oauth_metadata_request.url) == "https://api.example.com/.well-known/oauth-authorization-server"
# Complete the flow
oauth_metadata_response = httpx.Response(
200,
content=(
b'{"issuer": "https://api.example.com", '
b'"authorization_endpoint": "https://api.example.com/authorize", '
b'"token_endpoint": "https://api.example.com/token"}'
),
request=oauth_metadata_request,
)
provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
token_request = await auth_flow.asend(oauth_metadata_response)
assert str(token_request.url) == "https://api.example.com/token"
token_response = httpx.Response(
200,
content=b'{"access_token": "test_token", "token_type": "Bearer", "expires_in": 3600}',
request=token_request,
)
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer test_token"
final_response = httpx.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
class TestSEP985Discovery:
"""Test SEP-985 protected resource metadata discovery with fallback."""
@ -1084,7 +1345,7 @@ class TestSEP985Discovery:
)
# Build discovery URLs
discovery_urls = build_protected_resource_discovery_urls(
discovery_urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
@ -1224,7 +1485,7 @@ class TestSEP985Discovery:
)
# Build discovery URLs
discovery_urls = build_protected_resource_discovery_urls(
discovery_urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)