Implement SEP-985: OAuth Protected Resource Metadata discovery fallback (#1548)
Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Paul Carleton <paulc@anthropic.com>
This commit is contained in:
Chris Coutinho
GitHub
parent
de2289dc93
commit
3390e49c01
1 changed files with 252 additions and 30 deletions
|
|
@ -241,10 +241,10 @@ class TestOAuthFlow:
|
|||
"""Test OAuth flow methods."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_discover_protected_resource_request(
|
||||
async def test_build_protected_resource_discovery_urls(
|
||||
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
|
||||
):
|
||||
"""Test protected resource discovery request building maintains backward compatibility."""
|
||||
"""Test protected resource metadata discovery URL building with fallback."""
|
||||
|
||||
async def redirect_handler(url: str) -> None:
|
||||
pass
|
||||
|
|
@ -265,20 +265,19 @@ class TestOAuthFlow:
|
|||
status_code=401, headers={}, request=httpx.Request("GET", "https://request-api.example.com")
|
||||
)
|
||||
|
||||
request = await provider._discover_protected_resource(init_response)
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
urls = provider._build_protected_resource_discovery_urls(init_response)
|
||||
assert len(urls) == 1
|
||||
assert urls[0] == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
# Test with WWW-Authenticate header
|
||||
init_response.headers["WWW-Authenticate"] = (
|
||||
'Bearer resource_metadata="https://prm.example.com/.well-known/oauth-protected-resource/path"'
|
||||
)
|
||||
|
||||
request = await provider._discover_protected_resource(init_response)
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://prm.example.com/.well-known/oauth-protected-resource/path"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
urls = provider._build_protected_resource_discovery_urls(init_response)
|
||||
assert len(urls) == 2
|
||||
assert urls[0] == "https://prm.example.com/.well-known/oauth-protected-resource/path"
|
||||
assert urls[1] == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
@pytest.mark.anyio
|
||||
def test_create_oauth_metadata_request(self, oauth_provider: OAuthClientProvider):
|
||||
|
|
@ -1034,6 +1033,182 @@ def test_build_metadata(
|
|||
)
|
||||
|
||||
|
||||
class TestSEP985Discovery:
|
||||
"""Test SEP-985 protected resource metadata discovery with fallback."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_path_based_fallback_when_no_www_authenticate(
|
||||
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
|
||||
):
|
||||
"""Test that client falls back to path-based well-known URI when WWW-Authenticate is absent."""
|
||||
|
||||
async def redirect_handler(url: str) -> None:
|
||||
pass
|
||||
|
||||
async def callback_handler() -> tuple[str, str | None]:
|
||||
return "test_auth_code", "test_state"
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
# Test with 401 response without WWW-Authenticate header
|
||||
init_response = httpx.Response(
|
||||
status_code=401, headers={}, request=httpx.Request("GET", "https://api.example.com/v1/mcp")
|
||||
)
|
||||
|
||||
# Build discovery URLs
|
||||
discovery_urls = provider._build_protected_resource_discovery_urls(init_response)
|
||||
|
||||
# Should have path-based URL first, then root-based URL
|
||||
assert len(discovery_urls) == 2
|
||||
assert discovery_urls[0] == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
|
||||
assert discovery_urls[1] == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_root_based_fallback_after_path_based_404(
|
||||
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
|
||||
):
|
||||
"""Test that client falls back to root-based URI when path-based returns 404."""
|
||||
|
||||
async def redirect_handler(url: str) -> None:
|
||||
pass
|
||||
|
||||
async def callback_handler() -> tuple[str, str | None]:
|
||||
return "test_auth_code", "test_state"
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
# Ensure no tokens are stored
|
||||
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")],
|
||||
)
|
||||
|
||||
# Create a test request
|
||||
test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")
|
||||
|
||||
# Mock the auth flow
|
||||
auth_flow = provider.async_auth_flow(test_request)
|
||||
|
||||
# First request should be the original request without auth header
|
||||
request = await auth_flow.__anext__()
|
||||
assert "Authorization" not in request.headers
|
||||
|
||||
# Send a 401 response without WWW-Authenticate header
|
||||
response = httpx.Response(401, headers={}, request=test_request)
|
||||
|
||||
# Next request should be to discover protected resource metadata (path-based)
|
||||
discovery_request_1 = await auth_flow.asend(response)
|
||||
assert str(discovery_request_1.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
|
||||
assert discovery_request_1.method == "GET"
|
||||
|
||||
# Send 404 response for path-based discovery
|
||||
discovery_response_1 = httpx.Response(404, request=discovery_request_1)
|
||||
|
||||
# Next request should be to root-based well-known URI
|
||||
discovery_request_2 = await auth_flow.asend(discovery_response_1)
|
||||
assert str(discovery_request_2.url) == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
assert discovery_request_2.method == "GET"
|
||||
|
||||
# Send successful discovery response
|
||||
discovery_response_2 = httpx.Response(
|
||||
200,
|
||||
content=(
|
||||
b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
|
||||
),
|
||||
request=discovery_request_2,
|
||||
)
|
||||
|
||||
# Mock the rest of the OAuth flow
|
||||
provider._perform_authorization = mock.AsyncMock(return_value=("test_auth_code", "test_code_verifier"))
|
||||
|
||||
# Next should be OAuth metadata discovery
|
||||
oauth_metadata_request = await auth_flow.asend(discovery_response_2)
|
||||
assert oauth_metadata_request.method == "GET"
|
||||
|
||||
# Complete the flow
|
||||
oauth_metadata_response = httpx.Response(
|
||||
200,
|
||||
content=(
|
||||
b'{"issuer": "https://auth.example.com", '
|
||||
b'"authorization_endpoint": "https://auth.example.com/authorize", '
|
||||
b'"token_endpoint": "https://auth.example.com/token"}'
|
||||
),
|
||||
request=oauth_metadata_request,
|
||||
)
|
||||
|
||||
token_request = await auth_flow.asend(oauth_metadata_response)
|
||||
token_response = httpx.Response(
|
||||
200,
|
||||
content=(
|
||||
b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
|
||||
b'"refresh_token": "new_refresh_token"}'
|
||||
),
|
||||
request=token_request,
|
||||
)
|
||||
|
||||
final_request = await auth_flow.asend(token_response)
|
||||
final_response = httpx.Response(200, request=final_request)
|
||||
try:
|
||||
await auth_flow.asend(final_response)
|
||||
except StopAsyncIteration:
|
||||
pass
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_www_authenticate_takes_priority_over_well_known(
|
||||
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
|
||||
):
|
||||
"""Test that WWW-Authenticate header resource_metadata takes priority over well-known URIs."""
|
||||
|
||||
async def redirect_handler(url: str) -> None:
|
||||
pass
|
||||
|
||||
async def callback_handler() -> tuple[str, str | None]:
|
||||
return "test_auth_code", "test_state"
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
# Test with 401 response with WWW-Authenticate header
|
||||
init_response = httpx.Response(
|
||||
status_code=401,
|
||||
headers={
|
||||
"WWW-Authenticate": 'Bearer resource_metadata="https://custom.example.com/.well-known/oauth-protected-resource"'
|
||||
},
|
||||
request=httpx.Request("GET", "https://api.example.com/v1/mcp"),
|
||||
)
|
||||
|
||||
# Build discovery URLs
|
||||
discovery_urls = provider._build_protected_resource_discovery_urls(init_response)
|
||||
|
||||
# Should have WWW-Authenticate URL first, then fallback URLs
|
||||
assert len(discovery_urls) == 3
|
||||
assert discovery_urls[0] == "https://custom.example.com/.well-known/oauth-protected-resource"
|
||||
assert discovery_urls[1] == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
|
||||
assert discovery_urls[2] == "https://api.example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
|
||||
class TestWWWAuthenticate:
|
||||
"""Test WWW-Authenticate header parsing functionality."""
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue