github-featured/modelcontextprotocol--python-sdk
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
modelcontextprotocol--pytho…/tests/client/test_auth.py
T

3256 lines
135 KiB
Python
Raw Permalink Blame History

"""Tests for refactored OAuth client authentication implementation."""
import base64
import json
import time
from unittest import mock
from urllib.parse import parse_qs, quote, unquote, urlparse
import httpx2
import pytest
from inline_snapshot import Is, snapshot
from pydantic import AnyHttpUrl, AnyUrl
from mcp.client.auth import OAuthClientProvider, PKCEParameters
from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
from mcp.client.auth.utils import (
build_oauth_authorization_server_metadata_discovery_urls,
build_protected_resource_metadata_discovery_urls,
create_client_info_from_metadata_url,
create_client_registration_request,
create_oauth_metadata_request,
credentials_match_issuer,
extract_field_from_www_auth,
extract_resource_metadata_from_www_auth,
extract_scope_from_www_auth,
get_client_metadata_scopes,
handle_registration_response,
is_valid_client_metadata_url,
should_use_client_metadata_url,
union_scopes,
validate_authorization_response_iss,
validate_metadata_issuer,
)
from mcp.server.auth.routes import build_metadata
from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
from mcp.shared.auth import (
AuthorizationCodeResult,
OAuthClientInformationFull,
OAuthClientMetadata,
OAuthMetadata,
OAuthToken,
ProtectedResourceMetadata,
)
class MockTokenStorage:
"""Mock token storage for testing."""
def __init__(self):
self._tokens: OAuthToken | None = None
self._client_info: OAuthClientInformationFull | None = None
async def get_tokens(self) -> OAuthToken | None:
return self._tokens
async def set_tokens(self, tokens: OAuthToken) -> None:
self._tokens = tokens
async def get_client_info(self) -> OAuthClientInformationFull | None:
return self._client_info
async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
self._client_info = client_info
@pytest.fixture
def mock_storage():
return MockTokenStorage()
@pytest.fixture
def client_metadata():
return OAuthClientMetadata(
client_name="Test Client",
client_uri=AnyHttpUrl("https://example.com"),
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
scope="read write",
)
@pytest.fixture
def valid_tokens():
return OAuthToken(
access_token="test_access_token",
token_type="Bearer",
expires_in=3600,
refresh_token="test_refresh_token",
scope="read write",
)
@pytest.fixture
def oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage):
async def redirect_handler(url: str) -> None:
"""Mock redirect handler."""
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
"""Mock callback handler."""
return AuthorizationCodeResult(code="test_auth_code", state="test_state") # pragma: no cover
return OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
@pytest.fixture
def prm_metadata_response():
"""PRM metadata response with scopes."""
return httpx2.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp", '
b'"authorization_servers": ["https://auth.example.com"], '
b'"scopes_supported": ["resource:read", "resource:write"]}'
),
)
@pytest.fixture
def prm_metadata_without_scopes_response():
"""PRM metadata response without scopes."""
return httpx2.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp", '
b'"authorization_servers": ["https://auth.example.com"], '
b'"scopes_supported": null}'
),
)
@pytest.fixture
def init_response_with_www_auth_scope():
"""Initial 401 response with WWW-Authenticate header containing scope."""
return httpx2.Response(
401,
headers={"WWW-Authenticate": 'Bearer scope="special:scope from:www-authenticate"'},
request=httpx2.Request("GET", "https://api.example.com/test"),
)
@pytest.fixture
def init_response_without_www_auth_scope():
"""Initial 401 response without WWW-Authenticate scope."""
return httpx2.Response(
401,
headers={},
request=httpx2.Request("GET", "https://api.example.com/test"),
)
class TestPKCEParameters:
"""Test PKCE parameter generation."""
def test_pkce_generation(self):
"""Test PKCE parameter generation creates valid values."""
pkce = PKCEParameters.generate()
# Verify lengths
assert len(pkce.code_verifier) == 128
assert 43 <= len(pkce.code_challenge) <= 128
# Verify characters used in verifier
allowed_chars = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~")
assert all(c in allowed_chars for c in pkce.code_verifier)
# Verify base64url encoding in challenge (no padding)
assert "=" not in pkce.code_challenge
def test_pkce_uniqueness(self):
"""Test PKCE generates unique values each time."""
pkce1 = PKCEParameters.generate()
pkce2 = PKCEParameters.generate()
assert pkce1.code_verifier != pkce2.code_verifier
assert pkce1.code_challenge != pkce2.code_challenge
class TestOAuthContext:
"""Test OAuth context functionality."""
@pytest.mark.anyio
async def test_oauth_provider_initialization(
self, oauth_provider: OAuthClientProvider, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test OAuthClientProvider basic setup."""
assert oauth_provider.context.server_url == "https://api.example.com/v1/mcp"
assert oauth_provider.context.client_metadata == client_metadata
assert oauth_provider.context.storage == mock_storage
assert oauth_provider.context is not None
def test_context_url_parsing(self, oauth_provider: OAuthClientProvider):
"""Test get_authorization_base_url() extracts base URLs correctly."""
context = oauth_provider.context
# Test with path
assert context.get_authorization_base_url("https://api.example.com/v1/mcp") == "https://api.example.com"
# Test with no path
assert context.get_authorization_base_url("https://api.example.com") == "https://api.example.com"
# Test with port
assert (
context.get_authorization_base_url("https://api.example.com:8080/path/to/mcp")
== "https://api.example.com:8080"
)
# Test with query params
assert (
context.get_authorization_base_url("https://api.example.com/path?param=value") == "https://api.example.com"
)
@pytest.mark.anyio
async def test_token_validity_checking(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
"""Test is_token_valid() and can_refresh_token() logic."""
context = oauth_provider.context
# No tokens - should be invalid
assert not context.is_token_valid()
assert not context.can_refresh_token()
# Set valid tokens and client info
context.current_tokens = valid_tokens
context.token_expiry_time = time.time() + 1800 # 30 minutes from now
context.client_info = OAuthClientInformationFull(
client_id="test_client_id",
client_secret="test_client_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Should be valid
assert context.is_token_valid()
assert context.can_refresh_token() # Has refresh token and client info
# Expire the token
context.token_expiry_time = time.time() - 100 # Expired 100 seconds ago
assert not context.is_token_valid()
assert context.can_refresh_token() # Can still refresh
# Remove refresh token
context.current_tokens.refresh_token = None
assert not context.can_refresh_token()
# Remove client info
context.current_tokens.refresh_token = "test_refresh_token"
context.client_info = None
assert not context.can_refresh_token()
def test_clear_tokens(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
"""Test clear_tokens() removes token data."""
context = oauth_provider.context
context.current_tokens = valid_tokens
context.token_expiry_time = time.time() + 1800
# Clear tokens
context.clear_tokens()
# Verify cleared
assert context.current_tokens is None
assert context.token_expiry_time is None
class TestOAuthFlow:
"""Test OAuth flow methods."""
@pytest.mark.anyio
async def test_build_protected_resource_discovery_urls(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test protected resource metadata discovery URL building with fallback."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="test_state") # pragma: no cover
provider = OAuthClientProvider(
server_url="https://api.example.com",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
# Test without WWW-Authenticate (fallback)
init_response = httpx2.Response(
status_code=401, headers={}, request=httpx2.Request("GET", "https://request-api.example.com")
)
urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
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"'
)
urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
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):
"""Test OAuth metadata discovery request building."""
request = 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://example.com"
assert "mcp-protocol-version" in request.headers
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 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/openid-configuration/v1/mcp",
"https://api.example.com/v1/mcp/.well-known/openid-configuration",
]
@pytest.mark.anyio
async def test_oauth_discovery_fallback_conditions(self, oauth_provider: OAuthClientProvider):
"""Test the conditions during which an AS metadata discovery fallback will be attempted."""
# Ensure no tokens are stored
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
# Mock client info to skip DCR
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="existing_client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Create a test request
test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp")
# Mock the auth flow
auth_flow = oauth_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 to trigger the OAuth flow
response = httpx2.Response(
401,
headers={
"WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"'
},
request=test_request,
)
# Next request should be to discover protected resource metadata
discovery_request = await auth_flow.asend(response)
assert str(discovery_request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
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 = httpx2.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 at path-aware OAuth URL
oauth_metadata_request_1 = await auth_flow.asend(discovery_response)
assert (
str(oauth_metadata_request_1.url)
== "https://auth.example.com/.well-known/oauth-authorization-server/v1/mcp"
)
assert oauth_metadata_request_1.method == "GET"
# Send a 404 response
oauth_metadata_response_1 = httpx2.Response(
404,
content=b"Not Found",
request=oauth_metadata_request_1,
)
# 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/openid-configuration/v1/mcp"
assert oauth_metadata_request_2.method == "GET"
# Send a 400 response
oauth_metadata_response_2 = httpx2.Response(
400,
content=b"Bad Request",
request=oauth_metadata_request_2,
)
# 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/v1/mcp/.well-known/openid-configuration"
assert oauth_metadata_request_3.method == "GET"
# Send a 500 response
oauth_metadata_response_3 = httpx2.Response(
500,
content=b"Internal Server Error",
request=oauth_metadata_request_3,
)
# Mock the authorization process to minimize unnecessary state in this test
oauth_provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# 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"
# Send a successful token response
token_response = httpx2.Response(
200,
content=(
b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
b'"refresh_token": "new_refresh_token"}'
),
request=token_request,
)
# After OAuth flow completes, the original request is retried with auth header
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
assert final_request.method == "GET"
assert str(final_request.url) == "https://api.example.com/v1/mcp"
# Send final success response to properly close the generator
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass # Expected - generator should complete
@pytest.mark.anyio
async def test_handle_metadata_response_success(self, oauth_provider: OAuthClientProvider):
"""Test successful metadata response handling."""
# Create minimal valid OAuth metadata
content = b"""{
"issuer": "https://auth.example.com",
"authorization_endpoint": "https://auth.example.com/authorize",
"token_endpoint": "https://auth.example.com/token"
}"""
response = httpx2.Response(200, content=content)
# Should set metadata; the empty path is preserved (no trailing slash added)
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_prioritize_www_auth_scope_over_prm(
self,
oauth_provider: OAuthClientProvider,
prm_metadata_response: httpx2.Response,
init_response_with_www_auth_scope: httpx2.Response,
):
"""Test that WWW-Authenticate scope is prioritized over PRM scopes."""
# First, process PRM metadata to set protected_resource_metadata with scopes
await oauth_provider._handle_protected_resource_response(prm_metadata_response)
# Process the scope selection with WWW-Authenticate header
scopes = get_client_metadata_scopes(
extract_scope_from_www_auth(init_response_with_www_auth_scope),
oauth_provider.context.protected_resource_metadata,
)
# Verify that WWW-Authenticate scope is used (not PRM scopes)
assert scopes == "special:scope from:www-authenticate"
@pytest.mark.anyio
async def test_prioritize_prm_scopes_when_no_www_auth_scope(
self,
oauth_provider: OAuthClientProvider,
prm_metadata_response: httpx2.Response,
init_response_without_www_auth_scope: httpx2.Response,
):
"""Test that PRM scopes are prioritized when WWW-Authenticate header has no scopes."""
# Process the PRM metadata to set protected_resource_metadata with scopes
await oauth_provider._handle_protected_resource_response(prm_metadata_response)
# Process the scope selection without WWW-Authenticate scope
scopes = get_client_metadata_scopes(
extract_scope_from_www_auth(init_response_without_www_auth_scope),
oauth_provider.context.protected_resource_metadata,
)
# Verify that PRM scopes are used
assert scopes == "resource:read resource:write"
@pytest.mark.anyio
async def test_omit_scope_when_no_prm_scopes_or_www_auth(
self,
oauth_provider: OAuthClientProvider,
prm_metadata_without_scopes_response: httpx2.Response,
init_response_without_www_auth_scope: httpx2.Response,
):
"""Test that scope is omitted when PRM has no scopes and WWW-Authenticate doesn't specify scope."""
# Process the PRM metadata without scopes
await oauth_provider._handle_protected_resource_response(prm_metadata_without_scopes_response)
# Process the scope selection without WWW-Authenticate scope
scopes = get_client_metadata_scopes(
extract_scope_from_www_auth(init_response_without_www_auth_scope),
oauth_provider.context.protected_resource_metadata,
)
# Verify that scope is omitted
assert scopes is None
@pytest.mark.anyio
async def test_token_exchange_request_authorization_code(self, oauth_provider: OAuthClientProvider):
"""Test token exchange request building."""
# Set up required context
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
token_endpoint_auth_method="client_secret_post",
)
request = await oauth_provider._exchange_token_authorization_code("test_auth_code", "test_verifier")
assert request.method == "POST"
assert str(request.url) == "https://api.example.com/token"
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
# Check form data
content = request.content.decode()
assert "grant_type=authorization_code" in content
assert "code=test_auth_code" in content
assert "code_verifier=test_verifier" in content
assert "client_id=test_client" in content
assert "client_secret=test_secret" in content
@pytest.mark.anyio
async def test_refresh_token_request(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
"""Test refresh token request building."""
# Set up required context
oauth_provider.context.current_tokens = valid_tokens
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
token_endpoint_auth_method="client_secret_post",
)
request = await oauth_provider._refresh_token()
assert request.method == "POST"
assert str(request.url) == "https://api.example.com/token"
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
# Check form data
content = request.content.decode()
assert "grant_type=refresh_token" in content
assert "refresh_token=test_refresh_token" in content
assert "client_id=test_client" in content
assert "client_secret=test_secret" in content
@pytest.mark.anyio
async def test_basic_auth_token_exchange(self, oauth_provider: OAuthClientProvider):
"""Test token exchange with client_secret_basic authentication."""
# Set up OAuth metadata to support basic auth
oauth_provider.context.oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
token_endpoint_auth_methods_supported=["client_secret_basic", "client_secret_post"],
)
client_id_raw = "test@client" # Include special character to test URL encoding
client_secret_raw = "test:secret" # Include colon to test URL encoding
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id=client_id_raw,
client_secret=client_secret_raw,
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
token_endpoint_auth_method="client_secret_basic",
)
request = await oauth_provider._exchange_token_authorization_code("test_auth_code", "test_verifier")
# Should use basic auth (registered method)
assert "Authorization" in request.headers
assert request.headers["Authorization"].startswith("Basic ")
# Decode and verify credentials are properly URL-encoded
encoded_creds = request.headers["Authorization"][6:] # Remove "Basic " prefix
decoded = base64.b64decode(encoded_creds).decode()
client_id, client_secret = decoded.split(":", 1)
# Check URL encoding was applied
assert client_id == "test%40client" # @ should be encoded as %40
assert client_secret == "test%3Asecret" # : should be encoded as %3A
# Verify decoded values match original
assert unquote(client_id) == client_id_raw
assert unquote(client_secret) == client_secret_raw
# client_secret should NOT be in body for basic auth
content = request.content.decode()
assert "client_secret=" not in content
assert "client_id=test%40client" in content # client_id still in body
@pytest.mark.anyio
async def test_basic_auth_refresh_token(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
"""Test token refresh with client_secret_basic authentication."""
oauth_provider.context.current_tokens = valid_tokens
# Set up OAuth metadata to only support basic auth
oauth_provider.context.oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
token_endpoint_auth_methods_supported=["client_secret_basic"],
)
client_id = "test_client"
client_secret = "test_secret"
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id=client_id,
client_secret=client_secret,
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
token_endpoint_auth_method="client_secret_basic",
)
request = await oauth_provider._refresh_token()
assert "Authorization" in request.headers
assert request.headers["Authorization"].startswith("Basic ")
encoded_creds = request.headers["Authorization"][6:]
decoded = base64.b64decode(encoded_creds).decode()
assert decoded == f"{client_id}:{client_secret}"
# client_secret should NOT be in body
content = request.content.decode()
assert "client_secret=" not in content
@pytest.mark.anyio
async def test_none_auth_method(self, oauth_provider: OAuthClientProvider):
"""Test 'none' authentication method (public client)."""
oauth_provider.context.oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
token_endpoint_auth_methods_supported=["none"],
)
client_id = "public_client"
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id=client_id,
client_secret=None, # No secret for public client
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
token_endpoint_auth_method="none",
)
request = await oauth_provider._exchange_token_authorization_code("test_auth_code", "test_verifier")
# Should NOT have Authorization header
assert "Authorization" not in request.headers
# Should NOT have client_secret in body
content = request.content.decode()
assert "client_secret=" not in content
assert "client_id=public_client" in content
class TestProtectedResourceMetadata:
"""Test protected resource handling."""
@pytest.mark.anyio
async def test_resource_param_included_with_recent_protocol_version(self, oauth_provider: OAuthClientProvider):
"""Test resource parameter is included for protocol version >= 2025-06-18."""
# Set protocol version to 2025-06-18
oauth_provider.context.protocol_version = "2025-06-18"
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Test in token exchange
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" in content
# Check URL-encoded resource parameter
expected_resource = quote(oauth_provider.context.get_resource_url(), safe="")
assert f"resource={expected_resource}" in content
# Test in refresh token
oauth_provider.context.current_tokens = OAuthToken(
access_token="test_access",
token_type="Bearer",
refresh_token="test_refresh",
)
refresh_request = await oauth_provider._refresh_token()
refresh_content = refresh_request.content.decode()
assert "resource=" in refresh_content
@pytest.mark.anyio
async def test_resource_param_excluded_with_old_protocol_version(self, oauth_provider: OAuthClientProvider):
"""Test resource parameter is excluded for protocol version < 2025-06-18."""
# Set protocol version to older version
oauth_provider.context.protocol_version = "2025-03-26"
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Test in token exchange
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" not in content
# Test in refresh token
oauth_provider.context.current_tokens = OAuthToken(
access_token="test_access",
token_type="Bearer",
refresh_token="test_refresh",
)
refresh_request = await oauth_provider._refresh_token()
refresh_content = refresh_request.content.decode()
assert "resource=" not in refresh_content
@pytest.mark.anyio
async def test_resource_param_included_with_protected_resource_metadata(self, oauth_provider: OAuthClientProvider):
"""Test resource parameter is always included when protected resource metadata exists."""
# Set old protocol version but with protected resource metadata
oauth_provider.context.protocol_version = "2025-03-26"
oauth_provider.context.protected_resource_metadata = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
authorization_servers=[AnyHttpUrl("https://api.example.com")],
)
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Test in token exchange
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" in content
@pytest.mark.parametrize(
("protocol_version", "expected"),
[
("2025-03-26", False),
("2025-06-18", True),
("2025-11-25", True),
# Unrecognized strings gate conservatively, even ones sorting after 2025-06-18.
("zzz", False),
("9999-99-99", False),
],
)
def test_should_include_resource_param_by_protocol_version(
oauth_provider: OAuthClientProvider, protocol_version: str, expected: bool
) -> None:
"""Resource param is included only for recognized versions >= 2025-06-18."""
assert oauth_provider.context.should_include_resource_param(protocol_version) is expected
@pytest.mark.anyio
async def test_validate_resource_rejects_mismatched_resource(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""Client must reject PRM resource that doesn't match server URL."""
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
)
provider._initialized = True
prm = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://evil.example.com/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
with pytest.raises(OAuthFlowError, match="does not match expected"):
await provider._validate_resource_match(prm)
@pytest.mark.anyio
async def test_validate_resource_accepts_matching_resource(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""Client must accept PRM resource that matches server URL."""
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
)
provider._initialized = True
prm = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
# Should not raise
await provider._validate_resource_match(prm)
@pytest.mark.anyio
async def test_validate_resource_custom_callback(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""Custom callback overrides default validation."""
callback_called_with: list[tuple[str, str | None]] = []
async def custom_validate(server_url: str, prm_resource: str | None) -> None:
callback_called_with.append((server_url, prm_resource))
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
validate_resource_url=custom_validate,
)
provider._initialized = True
# This would normally fail default validation (different origin),
# but custom callback accepts it
prm = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://evil.example.com/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
await provider._validate_resource_match(prm)
assert callback_called_with == snapshot([("https://api.example.com/v1/mcp", "https://evil.example.com/mcp")])
@pytest.mark.anyio
async def test_validate_resource_accepts_root_url_with_trailing_slash(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""Root URLs with trailing slash normalization should match."""
provider = OAuthClientProvider(
server_url="https://api.example.com",
client_metadata=client_metadata,
storage=mock_storage,
)
provider._initialized = True
prm = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
# Should not raise despite trailing slash difference
await provider._validate_resource_match(prm)
@pytest.mark.anyio
async def test_validate_resource_accepts_server_url_with_trailing_slash(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""Server URL with trailing slash should match PRM resource."""
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp/",
client_metadata=client_metadata,
storage=mock_storage,
)
provider._initialized = True
prm = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
# Should not raise - both normalize to the same URL with trailing slash
await provider._validate_resource_match(prm)
@pytest.mark.anyio
async def test_get_resource_url_uses_canonical_when_prm_mismatches(
client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
"""get_resource_url falls back to canonical URL when PRM resource doesn't match."""
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
)
provider._initialized = True
# Set PRM with a resource that is NOT a parent of the server URL
provider.context.protected_resource_metadata = ProtectedResourceMetadata(
resource=AnyHttpUrl("https://other.example.com/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
# get_resource_url should return the canonical server URL, not the PRM resource
assert provider.context.get_resource_url() == snapshot("https://api.example.com/v1/mcp")
class TestRegistrationResponse:
"""Test client registration response handling."""
@pytest.mark.anyio
async def test_handle_registration_response_reads_before_accessing_text(self):
"""Test that response.aread() is called before accessing response.text."""
# Track if aread() was called
class MockResponse(httpx2.Response):
def __init__(self):
self.status_code = 400
self._aread_called = False
self._text = "Registration failed with error"
async def aread(self):
self._aread_called = True
return b"test content"
@property
def text(self):
if not self._aread_called:
raise RuntimeError("Response.text accessed before response.aread()") # pragma: no cover
return self._text
mock_response = MockResponse()
# This should call aread() before accessing text
with pytest.raises(Exception) as exc_info:
await handle_registration_response(mock_response)
# Verify aread() was called
assert mock_response._aread_called
# Verify the error message includes the response text
assert "Registration failed: 400" in str(exc_info.value)
@pytest.mark.anyio
async def test_registration_response_with_substituted_metadata_yields_the_credentials():
"""A 201 whose echoed metadata differs from the request still registers the client.
The authorization server returned an application_type outside OIDC Registration's set,
a null redirect_uris, and an auth method the SDK does not implement. RFC 7591 §3.2.1
permits the server to substitute values; the client keeps the credentials it minted.
"""
body = (
b'{"client_id": "issued-id", "client_secret": "issued-secret", '
b'"application_type": "confidential", "redirect_uris": null, '
b'"token_endpoint_auth_method": "client_secret_jwt"}'
)
response = httpx2.Response(201, content=body)
client_info = await handle_registration_response(response)
assert client_info.client_id == "issued-id"
assert client_info.client_secret == "issued-secret"
assert client_info.application_type == "confidential"
@pytest.mark.anyio
@pytest.mark.parametrize("echoed_issuer", ["https://not-the-flow.example", 12345], ids=["string", "not-a-string"])
async def test_registration_response_does_not_seed_the_issuer_binding_from_the_body(echoed_issuer: object):
"""The issuer binding (SEP-2352) is the SDK's record of which server it registered with,
stamped by the auth flow; an "issuer" member in the untrusted response body is dropped
before parsing - never populating the binding, and never failing the parse either, so a
mismatched or malformed value cannot discard the credentials on every 401."""
body = json.dumps({"client_id": "issued-id", "issuer": echoed_issuer}).encode()
client_info = await handle_registration_response(httpx2.Response(201, content=body))
assert client_info.client_id == "issued-id"
assert client_info.issuer is None
@pytest.mark.anyio
@pytest.mark.parametrize(
"content",
[b"<html>not json</html>", b'["json", "but", "not", "an", "object"]', '{"client_id": "café"}'.encode("latin-1")],
ids=["not-json", "not-an-object", "not-utf8"],
)
async def test_a_2xx_body_that_is_not_client_information_is_an_oauth_registration_error(content: bytes):
"""A success status whose body is not client information - unparseable, not an object, or
not valid UTF-8 - surfaces as OAuthRegistrationError rather than a raw parse failure, so a
single OAuthFlowError handler still covers registration."""
response = httpx2.Response(201, content=content)
with pytest.raises(OAuthRegistrationError):
await handle_registration_response(response)
@pytest.mark.anyio
async def test_token_exchange_reports_an_unimplemented_registered_auth_method(oauth_provider: OAuthClientProvider):
"""A server-assigned auth method the SDK cannot apply (RFC 7591 §3.2.1 lets the server
substitute one) is reported at the token exchange rather than sending the request
unauthenticated for the server to reject as invalid_client."""
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="registered-id",
client_secret="registered-secret",
token_endpoint_auth_method="client_secret_jwt",
)
with pytest.raises(OAuthTokenError):
await oauth_provider._exchange_token_authorization_code("test_auth_code", "test_verifier")
def test_prepare_token_auth_leaves_a_private_key_jwt_client_to_its_provider(oauth_provider: OAuthClientProvider):
"""private_key_jwt is recognized, so the base leaves the request untouched rather than
raising - PrivateKeyJWTOAuthProvider's inherited refresh path passes through here, and a
refresh the server then rejects (no assertion is signed on it) falls back to a fresh,
signed client-credentials exchange instead of aborting the flow."""
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="registered-id",
client_secret="registered-secret",
token_endpoint_auth_method="private_key_jwt",
)
data, headers = oauth_provider.context.prepare_token_auth({"grant_type": "refresh_token"}, {})
assert data == {"grant_type": "refresh_token"}
assert headers == {}
class TestCreateClientRegistrationRequest:
"""Test client registration request creation."""
def test_uses_registration_endpoint_from_metadata(self):
"""Test that registration URL comes from metadata when available."""
oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
registration_endpoint=AnyHttpUrl("https://auth.example.com/register"),
)
client_metadata = OAuthClientMetadata(redirect_uris=[AnyHttpUrl("http://localhost:3000/callback")])
request = create_client_registration_request(oauth_metadata, client_metadata, "https://auth.example.com")
assert str(request.url) == "https://auth.example.com/register"
assert request.method == "POST"
def test_falls_back_to_default_register_endpoint_when_no_metadata(self):
"""Test that registration uses fallback URL when auth_server_metadata is None."""
client_metadata = OAuthClientMetadata(redirect_uris=[AnyHttpUrl("http://localhost:3000/callback")])
request = create_client_registration_request(None, client_metadata, "https://auth.example.com")
assert str(request.url) == "https://auth.example.com/register"
assert request.method == "POST"
def test_falls_back_when_metadata_has_no_registration_endpoint(self):
"""Test fallback when metadata exists but lacks registration_endpoint."""
oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
# No registration_endpoint
)
client_metadata = OAuthClientMetadata(redirect_uris=[AnyHttpUrl("http://localhost:3000/callback")])
request = create_client_registration_request(oauth_metadata, client_metadata, "https://auth.example.com")
assert str(request.url) == "https://auth.example.com/register"
assert request.method == "POST"
def test_registration_request_sends_application_type():
"""SEP-837: the DCR body carries application_type, defaulting to native and overridable."""
redirect_uris: list[AnyUrl] = [AnyUrl("http://localhost:3000/callback")]
default_request = create_client_registration_request(
None, OAuthClientMetadata(redirect_uris=redirect_uris), "https://auth.example.com"
)
assert json.loads(default_request.content)["application_type"] == "native"
web_request = create_client_registration_request(
None, OAuthClientMetadata(redirect_uris=redirect_uris, application_type="web"), "https://auth.example.com"
)
assert json.loads(web_request.content)["application_type"] == "web"
class TestAuthFlow:
"""Test the auth flow in httpx2."""
@pytest.mark.anyio
async def test_auth_flow_with_valid_tokens(
self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
):
"""Test auth flow when tokens are already valid."""
# Pre-store valid tokens
await mock_storage.set_tokens(valid_tokens)
oauth_provider.context.current_tokens = valid_tokens
oauth_provider.context.token_expiry_time = time.time() + 1800
oauth_provider._initialized = True
# Create a test request
test_request = httpx2.Request("GET", "https://api.example.com/test")
# Mock the auth flow
auth_flow = oauth_provider.async_auth_flow(test_request)
# Should get the request with auth header added
request = await auth_flow.__anext__()
assert request.headers["Authorization"] == "Bearer test_access_token"
# Send a successful response
response = httpx2.Response(200)
try:
await auth_flow.asend(response)
except StopAsyncIteration:
pass # Expected
@pytest.mark.anyio
async def test_auth_flow_with_no_tokens(self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage):
"""Test auth flow when no tokens are available, triggering the full OAuth flow."""
# Ensure no tokens are stored
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
# Create a test request
test_request = httpx2.Request("GET", "https://api.example.com/mcp")
# Mock the auth flow
auth_flow = oauth_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 to trigger the OAuth flow
response = httpx2.Response(
401,
headers={
"WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"'
},
request=test_request,
)
# Next request should be to discover protected resource metadata
discovery_request = await auth_flow.asend(response)
assert discovery_request.method == "GET"
assert str(discovery_request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
# Send a successful discovery response with minimal protected resource metadata
discovery_response = httpx2.Response(
200,
content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
request=discovery_request,
)
# Next request should be to discover OAuth metadata
oauth_metadata_request = await auth_flow.asend(discovery_response)
assert oauth_metadata_request.method == "GET"
assert str(oauth_metadata_request.url).startswith("https://auth.example.com/")
assert "mcp-protocol-version" in oauth_metadata_request.headers
# Send a successful OAuth metadata response
oauth_metadata_response = httpx2.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", '
b'"registration_endpoint": "https://auth.example.com/register"}'
),
request=oauth_metadata_request,
)
# Next request should be to register client
registration_request = await auth_flow.asend(oauth_metadata_response)
assert registration_request.method == "POST"
assert str(registration_request.url) == "https://auth.example.com/register"
# Send a successful registration response
registration_response = httpx2.Response(
201,
content=b'{"client_id": "test_client_id", "client_secret": "test_client_secret", "redirect_uris": ["http://localhost:3030/callback"]}',
request=registration_request,
)
# Mock the authorization process
oauth_provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Next request should be to exchange token
token_request = await auth_flow.asend(registration_response)
assert token_request.method == "POST"
assert str(token_request.url) == "https://auth.example.com/token"
assert "code=test_auth_code" in token_request.content.decode()
# Send a successful token response
token_response = httpx2.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 should be the original request with auth header
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
assert final_request.method == "GET"
assert str(final_request.url) == "https://api.example.com/mcp"
# Send final success response to properly close the generator
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass # Expected - generator should complete
# Verify tokens were stored
assert oauth_provider.context.current_tokens is not None
assert oauth_provider.context.current_tokens.access_token == "new_access_token"
assert oauth_provider.context.token_expiry_time is not None
@pytest.mark.anyio
async def test_auth_flow_no_unnecessary_retry_after_oauth(
self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
):
"""Test that requests are not retried unnecessarily - the core bug that caused 2x performance degradation."""
# Pre-store valid tokens so no OAuth flow is needed
await mock_storage.set_tokens(valid_tokens)
oauth_provider.context.current_tokens = valid_tokens
oauth_provider.context.token_expiry_time = time.time() + 1800
oauth_provider._initialized = True
test_request = httpx2.Request("GET", "https://api.example.com/mcp")
auth_flow = oauth_provider.async_auth_flow(test_request)
# Count how many times the request is yielded
request_yields = 0
# First request - should have auth header already
request = await auth_flow.__anext__()
request_yields += 1
assert request.headers["Authorization"] == "Bearer test_access_token"
# Send a successful 200 response
response = httpx2.Response(200, request=request)
# In the buggy version, this would yield the request AGAIN unconditionally
# In the fixed version, this should end the generator
try:
await auth_flow.asend(response) # extra request
request_yields += 1 # pragma: no cover
# If we reach here, the bug is present
pytest.fail(
f"Unnecessary retry detected! Request was yielded {request_yields} times. "
f"This indicates the retry logic bug that caused 2x performance degradation. "
f"The request should only be yielded once for successful responses."
) # pragma: no cover
except StopAsyncIteration:
# This is the expected behavior - no unnecessary retry
pass
# Verify exactly one request was yielded (no double-sending)
assert request_yields == 1, f"Expected 1 request yield, got {request_yields}"
@pytest.mark.anyio
async def test_token_exchange_accepts_201_status(
self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""Test that token exchange accepts both 200 and 201 status codes."""
# Ensure no tokens are stored
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
# Create a test request
test_request = httpx2.Request("GET", "https://api.example.com/mcp")
# Mock the auth flow
auth_flow = oauth_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 to trigger the OAuth flow
response = httpx2.Response(
401,
headers={
"WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"'
},
request=test_request,
)
# Next request should be to discover protected resource metadata
discovery_request = await auth_flow.asend(response)
assert discovery_request.method == "GET"
assert str(discovery_request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
# Send a successful discovery response with minimal protected resource metadata
discovery_response = httpx2.Response(
200,
content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
request=discovery_request,
)
# Next request should be to discover OAuth metadata
oauth_metadata_request = await auth_flow.asend(discovery_response)
assert oauth_metadata_request.method == "GET"
assert str(oauth_metadata_request.url).startswith("https://auth.example.com/")
assert "mcp-protocol-version" in oauth_metadata_request.headers
# Send a successful OAuth metadata response
oauth_metadata_response = httpx2.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", '
b'"registration_endpoint": "https://auth.example.com/register"}'
),
request=oauth_metadata_request,
)
# Next request should be to register client
registration_request = await auth_flow.asend(oauth_metadata_response)
assert registration_request.method == "POST"
assert str(registration_request.url) == "https://auth.example.com/register"
# Send a successful registration response with 201 status
registration_response = httpx2.Response(
201,
content=b'{"client_id": "test_client_id", "client_secret": "test_client_secret", "redirect_uris": ["http://localhost:3030/callback"]}',
request=registration_request,
)
# Mock the authorization process
oauth_provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Next request should be to exchange token
token_request = await auth_flow.asend(registration_response)
assert token_request.method == "POST"
assert str(token_request.url) == "https://auth.example.com/token"
assert "code=test_auth_code" in token_request.content.decode()
# Send a successful token response with 201 status code (test both 200 and 201 are accepted)
token_response = httpx2.Response(
201,
content=(
b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
b'"refresh_token": "new_refresh_token"}'
),
request=token_request,
)
# Final request should be the original request with auth header
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
assert final_request.method == "GET"
assert str(final_request.url) == "https://api.example.com/mcp"
# Send final success response to properly close the generator
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass # Expected - generator should complete
# Verify tokens were stored
assert oauth_provider.context.current_tokens is not None
assert oauth_provider.context.current_tokens.access_token == "new_access_token"
assert oauth_provider.context.token_expiry_time is not None
@pytest.mark.anyio
async def test_403_insufficient_scope_updates_scope_from_header(
self,
oauth_provider: OAuthClientProvider,
mock_storage: MockTokenStorage,
valid_tokens: OAuthToken,
):
"""Test that 403 response correctly updates scope from WWW-Authenticate header."""
# Pre-store valid tokens and client info
client_info = OAuthClientInformationFull(
client_id="test_client_id",
client_secret="test_client_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
await mock_storage.set_tokens(valid_tokens)
await mock_storage.set_client_info(client_info)
oauth_provider.context.current_tokens = valid_tokens
oauth_provider.context.token_expiry_time = time.time() + 1800
oauth_provider.context.client_info = client_info
oauth_provider._initialized = True
# Original scope
assert oauth_provider.context.client_metadata.scope == "read write"
redirect_captured = False
captured_state = None
async def capture_redirect(url: str) -> None:
nonlocal redirect_captured, captured_state
redirect_captured = True
# SEP-2350: the authorization URL carries the union of the prior and challenged scopes
scope = parse_qs(urlparse(url).query)["scope"][0]
assert scope == "read write admin:write admin:delete"
# Extract state from redirect URL
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
oauth_provider.context.redirect_handler = capture_redirect
# Mock callback
async def mock_callback() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="auth_code", state=captured_state)
oauth_provider.context.callback_handler = mock_callback
test_request = httpx2.Request("GET", "https://api.example.com/mcp")
auth_flow = oauth_provider.async_auth_flow(test_request)
# First request
request = await auth_flow.__anext__()
# Send 403 with new scope requirement
response_403 = httpx2.Response(
403,
headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin:write admin:delete"'},
request=request,
)
# Trigger step-up - should get token exchange request
token_exchange_request = await auth_flow.asend(response_403)
# Verify scope was updated to the union of prior and challenged scopes (SEP-2350)
assert oauth_provider.context.client_metadata.scope == "read write admin:write admin:delete"
assert redirect_captured
# Complete the flow with successful token response
token_response = httpx2.Response(
200,
json={
"access_token": "new_token_with_new_scope",
"token_type": "Bearer",
"expires_in": 3600,
"scope": "admin:write admin:delete",
},
request=token_exchange_request,
)
# Should get final retry request
final_request = await auth_flow.asend(token_response)
# Send success response - flow should complete
success_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(success_response)
pytest.fail("Should have stopped after successful response") # pragma: no cover
except StopAsyncIteration:
pass # Expected
@pytest.mark.anyio
async def test_403_step_up_preserves_scope_from_stored_token(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""SEP-2350: a restart-loaded token's scope is folded into the step-up union.
On restart only the token is reloaded (not client_metadata.scope), so the stored token's
granted scope must seed the union, or the challenge would re-authorize for less.
"""
client_info = OAuthClientInformationFull(
client_id="test_client_id",
client_secret="test_client_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
# Simulate a restart: a token granted "read" is loaded, but client_metadata carries no scope.
oauth_provider.context.current_tokens = OAuthToken(access_token="t", scope="read")
oauth_provider.context.token_expiry_time = time.time() + 1800
oauth_provider.context.client_info = client_info
oauth_provider.context.client_metadata.scope = None
oauth_provider._initialized = True
captured_state: str | None = None
reauthorize_scope: str | None = None
async def capture_redirect(url: str) -> None:
nonlocal captured_state, reauthorize_scope
params = parse_qs(urlparse(url).query)
reauthorize_scope = params["scope"][0]
captured_state = params.get("state", [None])[0]
async def mock_callback() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="auth_code", state=captured_state)
oauth_provider.context.redirect_handler = capture_redirect
oauth_provider.context.callback_handler = mock_callback
auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/mcp"))
request = await auth_flow.__anext__()
response_403 = httpx2.Response(
403,
headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
request=request,
)
token_exchange_request = await auth_flow.asend(response_403)
assert reauthorize_scope == "read write"
# Drive the flow to completion so the context lock is released cleanly
token_response = httpx2.Response(
200,
json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
request=token_exchange_request,
)
final_request = await auth_flow.asend(token_response)
try:
await auth_flow.asend(httpx2.Response(200, request=final_request))
except StopAsyncIteration:
pass
@pytest.mark.parametrize(
(
"issuer_url",
"service_documentation_url",
"authorization_endpoint",
"token_endpoint",
"registration_endpoint",
"revocation_endpoint",
),
(
# Pydantic's AnyUrl incorrectly adds trailing slash to base URLs
# This is being fixed in https://github.com/pydantic/pydantic-core/pull/1719 (Pydantic 2.12+)
pytest.param(
"https://auth.example.com",
"https://auth.example.com/docs",
"https://auth.example.com/authorize",
"https://auth.example.com/token",
"https://auth.example.com/register",
"https://auth.example.com/revoke",
id="simple-url",
marks=pytest.mark.xfail(
reason="Pydantic AnyUrl adds trailing slash to base URLs - fixed in Pydantic 2.12+"
),
),
pytest.param(
"https://auth.example.com/",
"https://auth.example.com/docs",
"https://auth.example.com/authorize",
"https://auth.example.com/token",
"https://auth.example.com/register",
"https://auth.example.com/revoke",
id="with-trailing-slash",
),
pytest.param(
"https://auth.example.com/v1/mcp",
"https://auth.example.com/v1/mcp/docs",
"https://auth.example.com/v1/mcp/authorize",
"https://auth.example.com/v1/mcp/token",
"https://auth.example.com/v1/mcp/register",
"https://auth.example.com/v1/mcp/revoke",
id="with-path-param",
),
),
)
def test_build_metadata(
issuer_url: str,
service_documentation_url: str,
authorization_endpoint: str,
token_endpoint: str,
registration_endpoint: str,
revocation_endpoint: str,
):
metadata = build_metadata(
issuer_url=AnyHttpUrl(issuer_url),
service_documentation_url=AnyHttpUrl(service_documentation_url),
client_registration_options=ClientRegistrationOptions(enabled=True, valid_scopes=["read", "write", "admin"]),
revocation_options=RevocationOptions(enabled=True),
)
assert metadata.model_dump(exclude_defaults=True, mode="json") == snapshot(
{
"issuer": Is(issuer_url),
"authorization_endpoint": Is(authorization_endpoint),
"token_endpoint": Is(token_endpoint),
"registration_endpoint": Is(registration_endpoint),
"scopes_supported": ["read", "write", "admin"],
"grant_types_supported": ["authorization_code", "refresh_token"],
"token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
"service_documentation": Is(service_documentation_url),
"revocation_endpoint": Is(revocation_endpoint),
"revocation_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
"code_challenge_methods_supported": ["S256"],
}
)
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() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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."""
@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 # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
)
# Test with 401 response without WWW-Authenticate header
init_response = httpx2.Response(
status_code=401, headers={}, request=httpx2.Request("GET", "https://api.example.com/v1/mcp")
)
# Build discovery URLs
discovery_urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
# 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 # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
)
# 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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 = httpx2.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 # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
)
# Test with 401 response with WWW-Authenticate header
init_response = httpx2.Response(
status_code=401,
headers={
"WWW-Authenticate": 'Bearer resource_metadata="https://custom.example.com/.well-known/oauth-protected-resource"'
},
request=httpx2.Request("GET", "https://api.example.com/v1/mcp"),
)
# Build discovery URLs
discovery_urls = build_protected_resource_metadata_discovery_urls(
extract_resource_metadata_from_www_auth(init_response), provider.context.server_url
)
# 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."""
@pytest.mark.parametrize(
"www_auth_header,field_name,expected_value",
[
# Quoted values
('Bearer scope="read write"', "scope", "read write"),
(
'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"',
"resource_metadata",
"https://api.example.com/.well-known/oauth-protected-resource",
),
('Bearer error="insufficient_scope"', "error", "insufficient_scope"),
# Unquoted values
("Bearer scope=read", "scope", "read"),
(
"Bearer resource_metadata=https://api.example.com/.well-known/oauth-protected-resource",
"resource_metadata",
"https://api.example.com/.well-known/oauth-protected-resource",
),
("Bearer error=invalid_token", "error", "invalid_token"),
# Multiple parameters with quoted value
(
'Bearer realm="api", scope="admin:write resource:read", error="insufficient_scope"',
"scope",
"admin:write resource:read",
),
(
'Bearer realm="api", resource_metadata="https://api.example.com/.well-known/oauth-protected-resource", '
'error="insufficient_scope"',
"resource_metadata",
"https://api.example.com/.well-known/oauth-protected-resource",
),
# Multiple parameters with unquoted value
('Bearer realm="api", scope=basic', "scope", "basic"),
# Values with special characters
(
'Bearer scope="resource:read resource:write user_profile"',
"scope",
"resource:read resource:write user_profile",
),
(
'Bearer resource_metadata="https://api.example.com/auth/metadata?version=1"',
"resource_metadata",
"https://api.example.com/auth/metadata?version=1",
),
],
)
def test_extract_field_from_www_auth_valid_cases(
self,
client_metadata: OAuthClientMetadata,
mock_storage: MockTokenStorage,
www_auth_header: str,
field_name: str,
expected_value: str,
):
"""Test extraction of various fields from valid WWW-Authenticate headers."""
init_response = httpx2.Response(
status_code=401,
headers={"WWW-Authenticate": www_auth_header},
request=httpx2.Request("GET", "https://api.example.com/test"),
)
result = extract_field_from_www_auth(init_response, field_name)
assert result == expected_value
@pytest.mark.parametrize(
"www_auth_header,field_name,description",
[
# No header
(None, "scope", "no WWW-Authenticate header"),
# Empty header
("", "scope", "empty WWW-Authenticate header"),
# Header without requested field
('Bearer realm="api", error="insufficient_scope"', "scope", "no scope parameter"),
('Bearer realm="api", scope="read write"', "resource_metadata", "no resource_metadata parameter"),
# Malformed field (empty value)
("Bearer scope=", "scope", "malformed scope parameter"),
("Bearer resource_metadata=", "resource_metadata", "malformed resource_metadata parameter"),
],
)
def test_extract_field_from_www_auth_invalid_cases(
self,
client_metadata: OAuthClientMetadata,
mock_storage: MockTokenStorage,
www_auth_header: str | None,
field_name: str,
description: str,
):
"""Test extraction returns None for invalid cases."""
headers = {"WWW-Authenticate": www_auth_header} if www_auth_header is not None else {}
init_response = httpx2.Response(
status_code=401, headers=headers, request=httpx2.Request("GET", "https://api.example.com/test")
)
result = extract_field_from_www_auth(init_response, field_name)
assert result is None, f"Should return None for {description}"
class TestCIMD:
"""Test Client ID Metadata Document (CIMD) support."""
@pytest.mark.parametrize(
"url,expected",
[
# Valid CIMD URLs
("https://example.com/client", True),
("https://example.com/client-metadata.json", True),
("https://example.com/path/to/client", True),
("https://example.com:8443/client", True),
# Invalid URLs - HTTP (not HTTPS)
("http://example.com/client", False),
# Invalid URLs - root path
("https://example.com", False),
("https://example.com/", False),
# Invalid URLs - None or empty
(None, False),
("", False),
# Invalid URLs - malformed (triggers urlparse exception)
("http://[::1/foo/", False),
],
)
def test_is_valid_client_metadata_url(self, url: str | None, expected: bool):
"""Test CIMD URL validation."""
assert is_valid_client_metadata_url(url) == expected
def test_should_use_client_metadata_url_when_server_supports(self):
"""Test that CIMD is used when server supports it and URL is provided."""
oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
client_id_metadata_document_supported=True,
)
assert should_use_client_metadata_url(oauth_metadata, "https://example.com/client") is True
def test_should_not_use_client_metadata_url_when_server_does_not_support(self):
"""Test that CIMD is not used when server doesn't support it."""
oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
client_id_metadata_document_supported=False,
)
assert should_use_client_metadata_url(oauth_metadata, "https://example.com/client") is False
def test_should_not_use_client_metadata_url_when_not_provided(self):
"""Test that CIMD is not used when no URL is provided."""
oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
client_id_metadata_document_supported=True,
)
assert should_use_client_metadata_url(oauth_metadata, None) is False
def test_should_not_use_client_metadata_url_when_no_metadata(self):
"""Test that CIMD is not used when OAuth metadata is None."""
assert should_use_client_metadata_url(None, "https://example.com/client") is False
def test_create_client_info_from_metadata_url(self):
"""Test creating client info from CIMD URL."""
client_info = create_client_info_from_metadata_url(
"https://example.com/client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
assert client_info.client_id == "https://example.com/client"
assert client_info.token_endpoint_auth_method == "none"
assert client_info.redirect_uris == [AnyUrl("http://localhost:3030/callback")]
assert client_info.client_secret is None
def test_oauth_provider_with_valid_client_metadata_url(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test OAuthClientProvider initialization with valid client_metadata_url."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
client_metadata_url="https://example.com/client",
)
assert provider.context.client_metadata_url == "https://example.com/client"
def test_oauth_provider_with_invalid_client_metadata_url_raises_error(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test OAuthClientProvider raises error for invalid client_metadata_url."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="test_state") # pragma: no cover
with pytest.raises(ValueError) as exc_info:
OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
client_metadata_url="http://example.com/client", # HTTP instead of HTTPS
)
assert "HTTPS URL with a non-root pathname" in str(exc_info.value)
@pytest.mark.anyio
async def test_auth_flow_uses_cimd_when_server_supports(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test that auth flow uses CIMD URL as client_id when server supports it."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
client_metadata_url="https://example.com/client",
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
request = await auth_flow.__anext__()
assert "Authorization" not in request.headers
# Send 401 response
response = httpx2.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx2.Response(
200,
content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
request=prm_request,
)
# OAuth metadata discovery
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx2.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", '
b'"client_id_metadata_document_supported": true}'
),
request=oauth_request,
)
# Mock authorization
provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Should skip DCR and go directly to token exchange
token_request = await auth_flow.asend(oauth_response)
assert token_request.method == "POST"
assert str(token_request.url) == "https://auth.example.com/token"
# Verify client_id is the CIMD URL
content = token_request.content.decode()
assert "client_id=https%3A%2F%2Fexample.com%2Fclient" in content
# Verify client info was set correctly
assert provider.context.client_info is not None
assert provider.context.client_info.client_id == "https://example.com/client"
assert provider.context.client_info.token_endpoint_auth_method == "none"
# Complete the flow
token_response = httpx2.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 = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
@pytest.mark.anyio
async def test_auth_flow_falls_back_to_dcr_when_no_cimd_support(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""Test that auth flow falls back to DCR when server doesn't support CIMD."""
async def redirect_handler(url: str) -> None:
pass # pragma: no cover
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state="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,
client_metadata_url="https://example.com/client",
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
await auth_flow.__anext__()
# Send 401 response
response = httpx2.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx2.Response(
200,
content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
request=prm_request,
)
# OAuth metadata discovery - server does NOT support CIMD
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx2.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", '
b'"registration_endpoint": "https://auth.example.com/register"}'
),
request=oauth_request,
)
# Should proceed to DCR instead of skipping it
registration_request = await auth_flow.asend(oauth_response)
assert registration_request.method == "POST"
assert str(registration_request.url) == "https://auth.example.com/register"
# Complete the flow to avoid generator cleanup issues
registration_response = httpx2.Response(
201,
content=b'{"client_id": "dcr_client_id", "redirect_uris": ["http://localhost:3030/callback"]}',
request=registration_request,
)
# Mock authorization
provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
token_request = await auth_flow.asend(registration_response)
token_response = httpx2.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)
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
class TestSEP2207OfflineAccessScope:
"""Test SEP-2207: offline_access scope augmentation for OIDC-flavored refresh tokens."""
def _make_as_metadata(self, scopes_supported: list[str] | None = None) -> OAuthMetadata:
return OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
scopes_supported=scopes_supported,
)
def _make_prm(self, scopes_supported: list[str] | None = None) -> ProtectedResourceMetadata:
return ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
scopes_supported=scopes_supported,
)
def test_offline_access_added_when_as_supports_and_client_has_refresh_token(self):
"""offline_access is appended when AS advertises it and client supports refresh_token grant."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write offline_access"
def test_offline_access_added_with_www_authenticate_scope(self):
"""offline_access is appended even when scopes come from WWW-Authenticate header."""
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope="read write",
protected_resource_metadata=None,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write offline_access"
def test_offline_access_not_added_when_as_does_not_support(self):
"""offline_access is not added when AS does not advertise it in scopes_supported."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_client_has_no_refresh_token_grant(self):
"""offline_access is not added when client does not support refresh_token grant."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code"],
)
assert scopes == "read write"
def test_offline_access_not_duplicated_when_already_present(self):
"""offline_access is not added again if it already appears in the selected scopes."""
prm = self._make_prm(scopes_supported=["read", "offline_access", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read offline_access write"
def test_offline_access_not_added_when_no_scopes_selected(self):
"""offline_access is not added when no base scopes are available (None)."""
asm = self._make_as_metadata(scopes_supported=["offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=None,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
# When AS scopes are the only source and include offline_access,
# the base scope is "offline_access" and no duplication happens
assert scopes == "offline_access"
def test_offline_access_not_added_when_as_scopes_supported_is_none(self):
"""offline_access is not added when AS scopes_supported is None."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=None)
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_no_as_metadata(self):
"""offline_access is not added when AS metadata is not available."""
prm = self._make_prm(scopes_supported=["read", "write"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=None,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_no_grant_types_provided(self):
"""offline_access is not added when client_grant_types is None."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=None,
)
assert scopes == "read write"
def test_default_client_metadata_includes_refresh_token_grant(self):
"""Default OAuthClientMetadata includes refresh_token in grant_types (SEP-2207 guidance)."""
metadata = OAuthClientMetadata(redirect_uris=[AnyUrl("http://localhost:3030/callback")])
assert "refresh_token" in metadata.grant_types
@pytest.mark.anyio
async def test_auth_flow_adds_offline_access_when_as_advertises(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""E2E: auth flow includes offline_access in authorization request when AS advertises it."""
captured_auth_url: str | None = None
captured_state: str | None = None
async def redirect_handler(url: str) -> None:
nonlocal captured_auth_url, captured_state
captured_auth_url = url
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state=captured_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,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
# Pre-set client info to skip DCR
provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
request = await auth_flow.__anext__()
assert "Authorization" not in request.headers
# Send 401
response = httpx2.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx2.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp",'
b' "authorization_servers": ["https://auth.example.com"],'
b' "scopes_supported": ["read", "write"]}'
),
request=prm_request,
)
# OAuth metadata discovery - AS advertises offline_access
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx2.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",'
b' "scopes_supported": ["read", "write", "offline_access"]}'
),
request=oauth_request,
)
# This triggers authorization, which calls redirect_handler
token_request = await auth_flow.asend(oauth_response)
# Verify the authorization URL included offline_access in the scope
assert captured_auth_url is not None
parsed = urlparse(captured_auth_url)
params = parse_qs(parsed.query)
scope_value = params["scope"][0]
scope_parts = scope_value.split()
assert "offline_access" in scope_parts
assert "read" in scope_parts
assert "write" in scope_parts
# OIDC requires prompt=consent when offline_access is requested
assert params["prompt"][0] == "consent"
# Complete the token exchange
token_response = httpx2.Response(
200,
content=(
b'{"access_token": "new_access_token", "token_type": "Bearer",'
b' "expires_in": 3600, "refresh_token": "new_refresh_token"}'
),
request=token_request,
)
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
# Close the generator
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
@pytest.mark.anyio
async def test_auth_flow_no_offline_access_when_as_does_not_advertise(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""E2E: auth flow does NOT include offline_access when AS doesn't advertise it."""
captured_auth_url: str | None = None
captured_state: str | None = None
async def redirect_handler(url: str) -> None:
nonlocal captured_auth_url, captured_state
captured_auth_url = url
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
async def callback_handler() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="test_auth_code", state=captured_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,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
# Pre-set client info to skip DCR
provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
await auth_flow.__anext__()
# Send 401
response = httpx2.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx2.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp",'
b' "authorization_servers": ["https://auth.example.com"],'
b' "scopes_supported": ["read", "write"]}'
),
request=prm_request,
)
# OAuth metadata discovery - AS does NOT advertise offline_access
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx2.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",'
b' "scopes_supported": ["read", "write"]}'
),
request=oauth_request,
)
# This triggers authorization, which calls redirect_handler
token_request = await auth_flow.asend(oauth_response)
# Verify the authorization URL does NOT include offline_access
assert captured_auth_url is not None
parsed = urlparse(captured_auth_url)
params = parse_qs(parsed.query)
scope_value = params["scope"][0]
scope_parts = scope_value.split()
assert "offline_access" not in scope_parts
assert "read" in scope_parts
assert "write" in scope_parts
# prompt=consent should NOT be present without offline_access
assert "prompt" not in params
# Complete the token exchange
token_response = httpx2.Response(
200,
content=b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600}',
request=token_request,
)
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
# Close the generator
final_response = httpx2.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
_ISSUER = "https://as.example.com"
def _issuer_metadata(*, issuer: str = _ISSUER, iss_supported: bool | None = None) -> OAuthMetadata:
# Validate from string inputs so url_preserve_empty_path keeps the issuer as transmitted,
# matching the wire path (model_validate_json) rather than normalizing a bare authority.
return OAuthMetadata.model_validate(
{
"issuer": issuer,
"authorization_endpoint": f"{issuer}/authorize",
"token_endpoint": f"{issuer}/token",
"authorization_response_iss_parameter_supported": iss_supported,
}
)
@pytest.mark.parametrize(
("issuer", "iss", "iss_supported"),
[
pytest.param(_ISSUER, _ISSUER, True, id="advertised-and-correct"),
pytest.param(_ISSUER, None, None, id="not-advertised-and-omitted"),
pytest.param(_ISSUER, _ISSUER, None, id="not-advertised-but-correct"),
# An issuer that genuinely ends in a slash (e.g. Auth0) must match its own iss.
pytest.param("https://as.example.com/", "https://as.example.com/", True, id="trailing-slash-issuer"),
],
)
def test_validate_authorization_response_iss_accepts(issuer: str, iss: str | None, iss_supported: bool | None):
"""RFC 9207: a matching or legitimately absent iss is accepted."""
validate_authorization_response_iss(iss, _issuer_metadata(issuer=issuer, iss_supported=iss_supported))
@pytest.mark.parametrize(
("iss", "iss_supported", "match"),
[
pytest.param(None, True, "missing iss", id="advertised-but-omitted"),
pytest.param("https://evil.example.com", True, "iss mismatch", id="wrong-issuer"),
pytest.param("https://evil.example.com", None, "iss mismatch", id="unexpected-when-not-advertised"),
pytest.param(f"{_ISSUER}/", True, "iss mismatch", id="trailing-slash-not-normalized"),
],
)
def test_validate_authorization_response_iss_rejects(iss: str | None, iss_supported: bool | None, match: str):
"""RFC 9207: a mismatched iss, or one missing when advertised, is rejected via simple string compare."""
with pytest.raises(OAuthFlowError, match=match):
validate_authorization_response_iss(iss, _issuer_metadata(iss_supported=iss_supported))
def test_validate_authorization_response_iss_without_metadata():
"""With no AS metadata, a present iss is rejected and an absent one is accepted."""
validate_authorization_response_iss(None, None)
with pytest.raises(OAuthFlowError, match="iss mismatch"):
validate_authorization_response_iss(_ISSUER, None)
def test_validate_metadata_issuer_accepts_match():
validate_metadata_issuer(_issuer_metadata(issuer=_ISSUER), _ISSUER)
def test_validate_metadata_issuer_rejects_mismatch():
with pytest.raises(OAuthFlowError, match="metadata issuer mismatch"):
validate_metadata_issuer(_issuer_metadata(issuer="https://attacker.example.com"), _ISSUER)
@pytest.mark.parametrize(
("previous", "new", "expected"),
[
pytest.param("mcp:basic", "mcp:write", "mcp:basic mcp:write", id="disjoint-union-order"),
pytest.param(
"mcp:basic offline_access", "mcp:write mcp:basic", "mcp:basic offline_access mcp:write", id="dedup"
),
pytest.param(None, "mcp:write", "mcp:write", id="no-previous"),
pytest.param("mcp:basic", None, "mcp:basic", id="no-new"),
pytest.param(None, None, None, id="both-empty"),
],
)
def test_union_scopes(previous: str | None, new: str | None, expected: str | None):
"""SEP-2350: union merges previous and new scopes, dedups, and preserves order."""
assert union_scopes(previous, new) == expected
def test_credentials_match_issuer_same_issuer():
info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as")
assert credentials_match_issuer(info, "https://as", None) is True
def test_credentials_match_issuer_different_issuer():
info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as")
assert credentials_match_issuer(info, "https://other", None) is False
def test_credentials_match_issuer_no_recorded_issuer_is_left_alone():
"""Credentials with no bound issuer (pre-registered / legacy) carry no binding to enforce."""
info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")])
assert credentials_match_issuer(info, "https://as", None) is True
def test_credentials_match_issuer_cimd_is_portable():
"""A client_id equal to the configured client_metadata_url (CIMD) is portable across servers."""
cimd_url = "https://client.example/metadata.json"
info = OAuthClientInformationFull(
client_id=cimd_url,
redirect_uris=[AnyUrl("http://localhost/cb")],
token_endpoint_auth_method="none",
issuer="https://as",
)
assert credentials_match_issuer(info, "https://other", cimd_url) is True
def test_credentials_match_issuer_url_shaped_dcr_id_is_not_portable():
"""A URL-shaped client_id from DCR (not the configured CIMD URL) stays bound to its issuer."""
info = OAuthClientInformationFull(
client_id="https://as.example.com/clients/123",
redirect_uris=[AnyUrl("http://localhost/cb")],
issuer="https://as.example.com",
)
assert credentials_match_issuer(info, "https://other", "https://client.example/metadata.json") is False
@pytest.mark.anyio
async def test_handle_token_response_backfills_omitted_scope_from_request(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""RFC 6749 §5.1: an omitted token-response scope means granted == requested.
The token is stored with the requested scope filled in so it remains self-describing
after a restart, when the SEP-2350 step-up union reads it but ``client_metadata.scope``
has reverted to its constructor value.
"""
oauth_provider.context.client_metadata.scope = "read admin"
response = httpx2.Response(
200,
json={"access_token": "t", "token_type": "Bearer", "expires_in": 3600},
request=httpx2.Request("POST", "https://auth.example.com/token"),
)
await oauth_provider._handle_token_response(response)
assert oauth_provider.context.current_tokens is not None
assert oauth_provider.context.current_tokens.scope == "read admin"
stored = await mock_storage.get_tokens()
assert stored is not None
assert stored.scope == "read admin"
@pytest.mark.anyio
async def test_handle_token_response_raises_on_non_2xx_with_body(oauth_provider: OAuthClientProvider):
response = httpx2.Response(
400,
json={"error": "invalid_grant"},
request=httpx2.Request("POST", "https://auth.example.com/token"),
)
with pytest.raises(OAuthTokenError, match=r"Token exchange failed \(400\).*invalid_grant"):
await oauth_provider._handle_token_response(response)
@pytest.mark.anyio
async def test_handle_refresh_response_carries_prior_scope_and_refresh_token_when_omitted(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""RFC 6749 §6: omitted refresh-response scope and refresh_token are carried forward.
Omitted scope means it is unchanged from the prior access token. Omitted refresh_token
means the AS does not rotate refresh tokens; the client keeps using the previously
issued one so the next expiry can refresh instead of forcing a full re-authorization.
"""
oauth_provider.context.current_tokens = OAuthToken(
access_token="old", scope="read write", refresh_token="prior-refresh"
)
response = httpx2.Response(
200,
json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600},
request=httpx2.Request("POST", "https://auth.example.com/token"),
)
ok = await oauth_provider._handle_refresh_response(response)
assert ok is True
assert oauth_provider.context.current_tokens is not None
assert oauth_provider.context.current_tokens.access_token == "new"
assert oauth_provider.context.current_tokens.scope == "read write"
assert oauth_provider.context.current_tokens.refresh_token == "prior-refresh"
stored = await mock_storage.get_tokens()
assert stored is not None
assert stored.scope == "read write"
assert stored.refresh_token == "prior-refresh"
@pytest.mark.anyio
async def test_handle_refresh_response_adopts_rotated_refresh_token_when_returned(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""A refresh response that includes ``refresh_token`` replaces the prior one (rotation)."""
oauth_provider.context.current_tokens = OAuthToken(
access_token="old", scope="read write", refresh_token="prior-refresh"
)
response = httpx2.Response(
200,
json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "rotated"},
request=httpx2.Request("POST", "https://auth.example.com/token"),
)
ok = await oauth_provider._handle_refresh_response(response)
assert ok is True
stored = await mock_storage.get_tokens()
assert stored is not None
assert stored.refresh_token == "rotated"
@pytest.mark.anyio
async def test_issuer_binding_re_evaluated_after_asm_when_prm_discovery_failed(
oauth_provider: OAuthClientProvider,
):
"""SEP-2352: on the legacy no-PRM path the binding check uses the ASM-discovered issuer.
PRM discovery fails (404) so ``auth_server_url`` stays ``None`` and the post-PRM check is
skipped; when ASM discovery then succeeds via the root well-known fallback, the discovered
metadata's issuer is compared against the stored credentials' bound issuer and a mismatch
triggers re-registration.
"""
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="stale-client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
issuer="https://old-as.example.com",
)
auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
request = await auth_flow.__anext__()
response_401 = httpx2.Response(401, request=request)
# PRM discovery: path-based then root, both 404.
prm_req = await auth_flow.asend(response_401)
assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource"
# ASM discovery via root fallback (no auth_server_url) succeeds with a different issuer.
asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
asm_response = httpx2.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", '
b'"registration_endpoint": "https://api.example.com/register"}'
),
request=asm_req,
)
# The stale bound credentials are discarded, so the next yield is a DCR request
# rather than the authorize redirect.
next_req = await auth_flow.asend(asm_response)
assert oauth_provider.context.auth_server_url is None
assert next_req.method == "POST"
assert str(next_req.url) == "https://api.example.com/register"
await auth_flow.aclose()
@pytest.mark.anyio
@pytest.mark.parametrize(
"asm_responses",
[
pytest.param(
[httpx2.Response(404), httpx2.Response(404)],
id="asm-discovery-failed",
),
pytest.param(
[
httpx2.Response(
200,
content=(
b'{"issuer": "https://new-as.example.com", '
b'"authorization_endpoint": "https://new-as.example.com/authorize", '
b'"token_endpoint": "https://new-as.example.com/token"}'
),
)
],
id="asm-metadata-without-registration-endpoint",
),
],
)
async def test_issuer_is_not_stamped_when_registration_falls_back_to_the_resource_origin(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, asm_responses: list[httpx2.Response]
):
"""SEP-2352: a fallback registration is not recorded as bound to the PRM-advertised AS.
PRM advertises a new authorization server, so the stored credentials (bound to the old
issuer) are discarded. DCR then falls back to the resource-server origin's ``/register``
because the new AS's metadata either could not be discovered or omits
``registration_endpoint``. That registration was not derived from the new AS's metadata,
so persisting it as bound to the new AS would wedge the binding check on later flows;
instead the issuer is left unset.
"""
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="stale-client",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
issuer="https://api.example.com/",
)
captured_state: str | None = None
async def capture_redirect(url: str) -> None:
nonlocal captured_state
captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
async def echo_callback() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="auth_code", state=captured_state)
oauth_provider.context.redirect_handler = capture_redirect
oauth_provider.context.callback_handler = echo_callback
auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
request = await auth_flow.__anext__()
response_401 = httpx2.Response(
401,
headers={
"WWW-Authenticate": (
'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"'
)
},
request=request,
)
# PRM succeeds and advertises a new AS — the discard block fires.
prm_req = await auth_flow.asend(response_401)
assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource"
prm_response = httpx2.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://new-as.example.com"]}'
),
request=prm_req,
)
# ASM discovery for the new AS yields no usable registration_endpoint — either every
# well-known URL 404s, or metadata is returned without one.
next_req = await auth_flow.asend(prm_response)
assert oauth_provider.context.client_info is None
assert oauth_provider.context.oauth_metadata is None
assert str(next_req.url) == "https://new-as.example.com/.well-known/oauth-authorization-server"
for asm_response in asm_responses:
asm_response.request = next_req
next_req = await auth_flow.asend(asm_response)
# Step 4 falls back to the resource-server origin's /register.
dcr_req = next_req
assert dcr_req.method == "POST"
assert str(dcr_req.url) == "https://api.example.com/register"
dcr_response = httpx2.Response(
201,
json={"client_id": "fallback-client", "redirect_uris": ["http://localhost:3030/callback"]},
request=dcr_req,
)
token_req = await auth_flow.asend(dcr_response)
# The persisted record carries no issuer binding — not the PRM-advertised AS we never reached.
stored = await mock_storage.get_client_info()
assert stored is not None
assert stored.client_id == "fallback-client"
assert stored.issuer is None
# Drive the flow to completion so the context lock is released cleanly.
token_response = httpx2.Response(
200, json={"access_token": "t", "token_type": "Bearer", "expires_in": 3600}, request=token_req
)
final_req = await auth_flow.asend(token_response)
try:
await auth_flow.asend(httpx2.Response(200, request=final_req))
except StopAsyncIteration:
pass
@pytest.mark.anyio
async def test_issuer_is_stamped_when_same_origin_fallback_register_is_on_the_discovered_issuer(
oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
"""SEP-2352: a fallback registration on the discovered issuer's own host is still bound.
Legacy same-origin embedded AS: PRM is absent, root ASM discovery succeeds with
``issuer`` equal to the resource origin and no ``registration_endpoint``. DCR falls
back to ``<resource-origin>/register`` — the issuer's own host — so the binding was
established and is recorded, preserving auto-recovery on a later AS migration.
"""
oauth_provider.context.current_tokens = None
oauth_provider.context.token_expiry_time = None
oauth_provider._initialized = True
oauth_provider.context.client_info = None
captured_state: str | None = None
async def capture_redirect(url: str) -> None:
nonlocal captured_state
captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
async def echo_callback() -> AuthorizationCodeResult:
return AuthorizationCodeResult(code="auth_code", state=captured_state)
oauth_provider.context.redirect_handler = capture_redirect
oauth_provider.context.callback_handler = echo_callback
auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
request = await auth_flow.__anext__()
# PRM discovery 404s on both well-known URLs.
prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource"
# Root ASM discovery succeeds with the resource origin as issuer and no registration_endpoint.
asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
asm_response = httpx2.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=asm_req,
)
# DCR falls back to the resource origin's /register — the issuer's own host.
dcr_req = await auth_flow.asend(asm_response)
assert dcr_req.method == "POST"
assert str(dcr_req.url) == "https://api.example.com/register"
dcr_response = httpx2.Response(
201,
json={"client_id": "embedded-client", "redirect_uris": ["http://localhost:3030/callback"]},
request=dcr_req,
)
token_req = await auth_flow.asend(dcr_response)
stored = await mock_storage.get_client_info()
assert stored is not None
assert oauth_provider.context.oauth_metadata is not None
assert stored.client_id == "embedded-client"
assert stored.issuer == str(oauth_provider.context.oauth_metadata.issuer)
assert urlparse(stored.issuer).netloc == "api.example.com"
token_response = httpx2.Response(
200, json={"access_token": "t", "token_type": "Bearer", "expires_in": 3600}, request=token_req
)
final_req = await auth_flow.asend(token_response)
try:
await auth_flow.asend(httpx2.Response(200, request=final_req))
except StopAsyncIteration:
pass
Reference in New Issue View Git Blame Copy Permalink