Support falling back to OIDC metadata for auth (#1061)
Some checks failed
Main branch checks / checks (push) Has been cancelled
Some checks failed
Main branch checks / checks (push) Has been cancelled
This commit is contained in:
Luca Chang
GitHub
parent
41184ba6c8
commit
99c4f3c906
2 changed files with 100 additions and 202 deletions
|
|
@ -235,60 +235,13 @@ class TestOAuthFlow:
|
|||
assert "mcp-protocol-version" in request.headers
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_discover_oauth_metadata_request(self, oauth_provider):
|
||||
def test_create_oauth_metadata_request(self, oauth_provider):
|
||||
"""Test OAuth metadata discovery request building."""
|
||||
request = await oauth_provider._discover_oauth_metadata()
|
||||
request = oauth_provider._create_oauth_metadata_request("https://example.com")
|
||||
|
||||
# Ensure correct method and headers, and that the URL is unmodified
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://api.example.com/.well-known/oauth-authorization-server/v1/mcp"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_discover_oauth_metadata_request_no_path(self, client_metadata, mock_storage):
|
||||
"""Test OAuth metadata discovery request building when server has no path."""
|
||||
|
||||
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",
|
||||
client_metadata=client_metadata,
|
||||
storage=mock_storage,
|
||||
redirect_handler=redirect_handler,
|
||||
callback_handler=callback_handler,
|
||||
)
|
||||
|
||||
request = await provider._discover_oauth_metadata()
|
||||
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://api.example.com/.well-known/oauth-authorization-server"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_discover_oauth_metadata_request_trailing_slash(self, client_metadata, mock_storage):
|
||||
"""Test OAuth metadata discovery request building when server path has trailing slash."""
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
request = await provider._discover_oauth_metadata()
|
||||
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://api.example.com/.well-known/oauth-authorization-server/v1/mcp"
|
||||
assert str(request.url) == "https://example.com"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
|
||||
|
||||
|
|
@ -296,46 +249,16 @@ class TestOAuthFallback:
|
|||
"""Test OAuth discovery fallback behavior for legacy (act as AS not RS) servers."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_fallback_discovery_request(self, client_metadata, mock_storage):
|
||||
"""Test fallback discovery request building."""
|
||||
async def test_oauth_discovery_fallback_order(self, oauth_provider):
|
||||
"""Test fallback URL construction order."""
|
||||
discovery_urls = oauth_provider._get_discovery_urls()
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
# Set up discovery state manually as if path-aware discovery was attempted
|
||||
provider.context.discovery_base_url = "https://api.example.com"
|
||||
provider.context.discovery_pathname = "/v1/mcp"
|
||||
|
||||
# Test fallback request building
|
||||
request = await provider._discover_oauth_metadata_fallback()
|
||||
|
||||
assert request.method == "GET"
|
||||
assert str(request.url) == "https://api.example.com/.well-known/oauth-authorization-server"
|
||||
assert "mcp-protocol-version" in request.headers
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_should_attempt_fallback(self, oauth_provider):
|
||||
"""Test fallback decision logic."""
|
||||
# Should attempt fallback on 404 with non-root path
|
||||
assert oauth_provider._should_attempt_fallback(404, "/v1/mcp")
|
||||
|
||||
# Should NOT attempt fallback on 404 with root path
|
||||
assert not oauth_provider._should_attempt_fallback(404, "/")
|
||||
|
||||
# Should NOT attempt fallback on other status codes
|
||||
assert not oauth_provider._should_attempt_fallback(200, "/v1/mcp")
|
||||
assert not oauth_provider._should_attempt_fallback(500, "/v1/mcp")
|
||||
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",
|
||||
]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_handle_metadata_response_success(self, oauth_provider):
|
||||
|
|
@ -348,50 +271,11 @@ class TestOAuthFallback:
|
|||
}"""
|
||||
response = httpx.Response(200, content=content)
|
||||
|
||||
# Should return True (success) and set metadata
|
||||
result = await oauth_provider._handle_oauth_metadata_response(response, is_fallback=False)
|
||||
assert result is True
|
||||
# Should set metadata
|
||||
await oauth_provider._handle_oauth_metadata_response(response)
|
||||
assert oauth_provider.context.oauth_metadata is not None
|
||||
assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_handle_metadata_response_404_needs_fallback(self, oauth_provider):
|
||||
"""Test 404 response handling that should trigger fallback."""
|
||||
# Set up discovery state for non-root path
|
||||
oauth_provider.context.discovery_base_url = "https://api.example.com"
|
||||
oauth_provider.context.discovery_pathname = "/v1/mcp"
|
||||
|
||||
# Mock 404 response
|
||||
response = httpx.Response(404)
|
||||
|
||||
# Should return False (needs fallback)
|
||||
result = await oauth_provider._handle_oauth_metadata_response(response, is_fallback=False)
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_handle_metadata_response_404_no_fallback_needed(self, oauth_provider):
|
||||
"""Test 404 response handling when no fallback is needed."""
|
||||
# Set up discovery state for root path
|
||||
oauth_provider.context.discovery_base_url = "https://api.example.com"
|
||||
oauth_provider.context.discovery_pathname = "/"
|
||||
|
||||
# Mock 404 response
|
||||
response = httpx.Response(404)
|
||||
|
||||
# Should return True (no fallback needed)
|
||||
result = await oauth_provider._handle_oauth_metadata_response(response, is_fallback=False)
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_handle_metadata_response_404_fallback_attempt(self, oauth_provider):
|
||||
"""Test 404 response handling during fallback attempt."""
|
||||
# Mock 404 response during fallback
|
||||
response = httpx.Response(404)
|
||||
|
||||
# Should return True (fallback attempt complete, no further action needed)
|
||||
result = await oauth_provider._handle_oauth_metadata_response(response, is_fallback=True)
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_register_client_request(self, oauth_provider):
|
||||
"""Test client registration request building."""
|
||||
|
|
|
|||
39
tests/shared/test_auth.py
Normal file
39
tests/shared/test_auth.py
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
"""Tests for OAuth 2.0 shared code."""
|
||||
|
||||
from mcp.shared.auth import OAuthMetadata
|
||||
|
||||
|
||||
class TestOAuthMetadata:
|
||||
"""Tests for OAuthMetadata parsing."""
|
||||
|
||||
def test_oauth(self):
|
||||
"""Should not throw when parsing OAuth metadata."""
|
||||
OAuthMetadata.model_validate(
|
||||
{
|
||||
"issuer": "https://example.com",
|
||||
"authorization_endpoint": "https://example.com/oauth2/authorize",
|
||||
"token_endpoint": "https://example.com/oauth2/token",
|
||||
"scopes_supported": ["read", "write"],
|
||||
"response_types_supported": ["code", "token"],
|
||||
"token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post"],
|
||||
}
|
||||
)
|
||||
|
||||
def test_oidc(self):
|
||||
"""Should not throw when parsing OIDC metadata."""
|
||||
OAuthMetadata.model_validate(
|
||||
{
|
||||
"issuer": "https://example.com",
|
||||
"authorization_endpoint": "https://example.com/oauth2/authorize",
|
||||
"token_endpoint": "https://example.com/oauth2/token",
|
||||
"end_session_endpoint": "https://example.com/logout",
|
||||
"id_token_signing_alg_values_supported": ["RS256"],
|
||||
"jwks_uri": "https://example.com/.well-known/jwks.json",
|
||||
"response_types_supported": ["code", "token"],
|
||||
"revocation_endpoint": "https://example.com/oauth2/revoke",
|
||||
"scopes_supported": ["openid", "read", "write"],
|
||||
"subject_types_supported": ["public"],
|
||||
"token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post"],
|
||||
"userinfo_endpoint": "https://example.com/oauth2/userInfo",
|
||||
}
|
||||
)
|
||||
Loading…
Reference in a new issue