python-sdk/tests/client/test_auth.py

"""
Tests for refactored OAuth client authentication implementation.
"""

import base64
import json
import time
from collections.abc import AsyncGenerator
from unittest import mock
from urllib.parse import parse_qs, unquote, urlparse

import httpx
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
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_auth_metadata_response,
    handle_registration_response,
    is_valid_client_metadata_url,
    should_use_client_metadata_url,
    validate_metadata_issuer,
)
from mcp.shared.auth import (
    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  # pragma: no cover

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self._tokens = tokens

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        return self._client_info  # pragma: no cover

    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() -> tuple[str, str | None]:
        """Mock callback handler."""
        return "test_auth_code", "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 httpx.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 httpx.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 httpx.Response(
        401,
        headers={"WWW-Authenticate": 'Bearer scope="special:scope from:www-authenticate"'},
        request=httpx.Request("GET", "https://api.example.com/test"),
    )


@pytest.fixture
def init_response_without_www_auth_scope():
    """Initial 401 response without WWW-Authenticate scope."""
    return httpx.Response(
        401,
        headers={},
        request=httpx.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.timeout == 300.0
        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() -> tuple[str, str | None]:
            return "test_auth_code", "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 = httpx.Response(
            status_code=401, headers={}, request=httpx.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 = httpx.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 = httpx.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 = httpx.Response(
            200,
            content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com/v1/mcp"]}',
            request=discovery_request,
        )

        # Next request should be to discover OAuth metadata 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 = httpx.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 = httpx.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 = httpx.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 = httpx.Response(
            200,
            content=(
                b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
                b'"refresh_token": "new_refresh_token"}'
            ),
            request=token_request,
        )

        # 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 = httpx.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 = httpx.Response(200, content=content)

        # Should set metadata
        await oauth_provider._handle_oauth_metadata_response(response)
        assert oauth_provider.context.oauth_metadata is not None
        assert str(oauth_provider.context.oauth_metadata.issuer) == "https://auth.example.com/"

    @pytest.mark.anyio
    async def test_prioritize_www_auth_scope_over_prm(
        self,
        oauth_provider: OAuthClientProvider,
        prm_metadata_response: httpx.Response,
        init_response_with_www_auth_scope: httpx.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: httpx.Response,
        init_response_without_www_auth_scope: httpx.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: httpx.Response,
        init_response_without_www_auth_scope: httpx.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
        from urllib.parse import quote

        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


async def _start_discovery(
    provider: OAuthClientProvider,
) -> tuple[AsyncGenerator[httpx.Request, httpx.Response], httpx.Request]:
    """Drive `provider`'s auth flow to the point where it has sent the MCP request, seen a 401 and
    issued its first protected-resource-metadata request; returns (flow, that request)."""
    provider.context.current_tokens = None
    provider.context.token_expiry_time = None
    provider._initialized = True
    mcp_request = httpx.Request("POST", "https://api.example.com/v1/mcp")
    flow = provider.async_auth_flow(mcp_request)
    sent = await flow.__anext__()
    assert sent is mcp_request
    # No resource_metadata hint, so discovery tries the path-based well-known URL, then the root one.
    unauthorized = httpx.Response(401, request=mcp_request)
    prm_request = await flow.asend(unauthorized)
    assert (prm_request.method, str(prm_request.url)) == (
        "GET",
        "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp",
    )
    return flow, prm_request


async def _redirect(request: httpx.Request, status: int, location: str) -> httpx.Response:
    """A redirect answer to `request`, as httpx hands it back when it does not follow it."""
    transport = httpx.MockTransport(lambda r: httpx.Response(status, headers={"location": location}))
    async with httpx.AsyncClient(transport=transport) as client:
        return await client.send(request)


@pytest.mark.anyio
async def test_auth_flow_follows_a_same_origin_redirect_of_its_own_request(oauth_provider: OAuthClientProvider):
    """SDK-defined: a request the OAuth flow makes (here protected-resource metadata discovery)
    follows a redirect that stays within its origin and keeps its method, like an MCP request."""
    flow, prm_request = await _start_discovery(oauth_provider)

    follow_up = await flow.asend(await _redirect(prm_request, 307, "/.well-known/oauth-protected-resource/v1/mcp/"))

    assert (follow_up.method, str(follow_up.url)) == (
        "GET",
        "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp/",
    )
    await flow.aclose()


@pytest.mark.anyio
async def test_auth_flow_does_not_follow_a_redirect_of_its_own_request_to_another_origin(
    oauth_provider: OAuthClientProvider,
):
    """SDK-defined: a redirect of a flow request to another origin is handed to the flow unfollowed,
    which treats it as "not served here" and moves to its next discovery URL."""
    flow, prm_request = await _start_discovery(oauth_provider)

    next_request = await flow.asend(await _redirect(prm_request, 307, "https://elsewhere.example/prm"))

    assert (next_request.method, str(next_request.url)) == (
        "GET",
        "https://api.example.com/.well-known/oauth-protected-resource",
    )
    await flow.aclose()


@pytest.mark.anyio
async def test_auth_flow_stops_following_a_redirecting_request_after_a_few_hops(
    oauth_provider: OAuthClientProvider,
):
    """SDK-defined: a flow request that keeps redirecting within its origin is followed a bounded
    number of times; the redirect after that is handed to the flow unfollowed."""
    flow, request = await _start_discovery(oauth_provider)

    hops = 0
    while str(request.url) != "https://api.example.com/.well-known/oauth-protected-resource":
        request = await flow.asend(
            await _redirect(request, 307, f"/.well-known/oauth-protected-resource/v1/mcp/{hops}")
        )
        hops += 1

    assert hops == 6  # five followed, the sixth handed back and taken as "try the next URL"
    await flow.aclose()


@pytest.mark.anyio
@pytest.mark.parametrize(("status", "keep_trying"), [(404, True), (307, True), (500, False)])
async def test_auth_metadata_response_says_whether_to_try_the_next_discovery_url(
    status: int, keep_trying: bool
) -> None:
    """SDK-defined: a 4xx or a 3xx (redirects are not followed on these requests) from a discovery
    candidate means the metadata is not served there and the next well-known URL is tried; a 5xx
    stops discovery."""
    assert await handle_auth_metadata_response(httpx.Response(status)) == (keep_trying, None)


class TestRegistrationResponse:
    """Test client registration response handling."""

    @pytest.mark.anyio
    async def test_handle_registration_response_reads_before_accessing_text(self):
        """The registration error carries the response text, which for a streamed response means
        reading it first (a streamed httpx response raises ResponseNotRead otherwise)."""
        response = httpx.Response(400, stream=httpx.ByteStream(b"Registration failed with error"))

        with pytest.raises(OAuthRegistrationError) as exc_info:
            await handle_registration_response(response)

        assert str(exc_info.value) == snapshot("Registration failed: 400 Registration failed with error")

    @pytest.mark.anyio
    async def test_registration_error_names_an_unfollowed_redirect(self):
        """SDK-defined: when the registration endpoint answered with a redirect that was not followed,
        the error says where it pointed (without userinfo or query) instead of only the bare status."""
        request = httpx.Request("POST", "https://as.example/register")
        async with httpx.AsyncClient(
            transport=httpx.MockTransport(
                lambda r: httpx.Response(307, headers={"location": "https://u:p@elsewhere.example/register?state=x"})
            )
        ) as client:
            response = await client.send(request)

        with pytest.raises(OAuthRegistrationError) as exc_info:
            await handle_registration_response(response)

        assert str(exc_info.value) == snapshot(
            "Registration failed: 307 (redirected to https://elsewhere.example/register; not followed) "
        )


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"


class TestAuthFlow:
    """Test the auth flow in httpx."""

    @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 = httpx.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 = httpx.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 = httpx.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 = httpx.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 = httpx.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 = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://auth.example.com", '
                b'"authorization_endpoint": "https://auth.example.com/authorize", '
                b'"token_endpoint": "https://auth.example.com/token", '
                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 = httpx.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 = httpx.Response(
            200,
            content=(
                b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
                b'"refresh_token": "new_refresh_token"}'
            ),
            request=token_request,
        )

        # Final request 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 = httpx.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 = httpx.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 = httpx.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_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
            # Verify the new scope is included in authorization URL
            assert "scope=admin%3Awrite+admin%3Adelete" in url or "scope=admin:write+admin:delete" in url.replace(
                "%3A", ":"
            ).replace("+", " ")
            # Extract state from redirect URL
            from urllib.parse import parse_qs, urlparse

            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() -> tuple[str, str | None]:
            return "auth_code", captured_state

        oauth_provider.context.callback_handler = mock_callback

        test_request = httpx.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 = httpx.Response(
            403,
            headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin:write admin:delete"'},
            request=request,
        )

        # Trigger step-up - discovery runs first (nothing published here), then the token exchange
        prm_request = await auth_flow.asend(response_403)
        prm_request = await auth_flow.asend(httpx.Response(404, request=prm_request))
        asm_request = await auth_flow.asend(httpx.Response(404, request=prm_request))
        token_exchange_request = await auth_flow.asend(httpx.Response(404, request=asm_request))

        # Verify scope was updated
        assert oauth_provider.context.client_metadata.scope == "admin:write admin:delete"
        assert redirect_captured

        # Complete the flow with successful token response
        token_response = httpx.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 = httpx.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.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,
):
    from mcp.server.auth.routes import build_metadata
    from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions

    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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        # Simulate a legacy server like Linear
        provider = OAuthClientProvider(
            server_url="https://mcp.linear.app/sse",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
        )

        provider.context.current_tokens = None
        provider.context.token_expiry_time = None
        provider._initialized = True

        # Mock client info to skip DCR
        provider.context.client_info = OAuthClientInformationFull(
            client_id="existing_client",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        )

        test_request = httpx.Request("GET", "https://mcp.linear.app/sse")
        auth_flow = provider.async_auth_flow(test_request)

        # First request
        request = await auth_flow.__anext__()
        assert "Authorization" not in request.headers

        # Send 401 without WWW-Authenticate header (typical legacy server)
        response = httpx.Response(401, headers={}, request=test_request)

        # Should try path-based PRM first
        prm_request_1 = await auth_flow.asend(response)
        assert str(prm_request_1.url) == "https://mcp.linear.app/.well-known/oauth-protected-resource/sse"

        # PRM returns 404
        prm_response_1 = httpx.Response(404, request=prm_request_1)

        # Should try root-based PRM
        prm_request_2 = await auth_flow.asend(prm_response_1)
        assert str(prm_request_2.url) == "https://mcp.linear.app/.well-known/oauth-protected-resource"

        # PRM returns 404 again - all PRM URLs failed
        prm_response_2 = httpx.Response(404, request=prm_request_2)

        # Should fall back to root OAuth discovery (March 2025 spec behavior)
        oauth_metadata_request = await auth_flow.asend(prm_response_2)
        assert str(oauth_metadata_request.url) == "https://mcp.linear.app/.well-known/oauth-authorization-server"
        assert oauth_metadata_request.method == "GET"

        # Send successful OAuth metadata response
        oauth_metadata_response = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://mcp.linear.app", '
                b'"authorization_endpoint": "https://mcp.linear.app/authorize", '
                b'"token_endpoint": "https://mcp.linear.app/token"}'
            ),
            request=oauth_metadata_request,
        )

        # Mock authorization
        provider._perform_authorization_code_grant = mock.AsyncMock(
            return_value=("test_auth_code", "test_code_verifier")
        )

        # Next should be token exchange
        token_request = await auth_flow.asend(oauth_metadata_response)
        assert str(token_request.url) == "https://mcp.linear.app/token"

        # Send successful token response
        token_response = httpx.Response(
            200,
            content=b'{"access_token": "linear_token", "token_type": "Bearer", "expires_in": 3600}',
            request=token_request,
        )

        # Final request with auth header
        final_request = await auth_flow.asend(token_response)
        assert final_request.headers["Authorization"] == "Bearer linear_token"
        assert str(final_request.url) == "https://mcp.linear.app/sse"

        # Complete flow
        final_response = httpx.Response(200, request=final_request)
        try:
            await auth_flow.asend(final_response)
        except StopAsyncIteration:
            pass

    @pytest.mark.anyio
    async def test_legacy_server_with_different_prm_and_root_urls(
        self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
    ):
        """Test PRM fallback with different WWW-Authenticate and root URLs."""

        async def redirect_handler(url: str) -> None:
            pass  # pragma: no cover

        async def callback_handler() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
        )

        provider.context.current_tokens = None
        provider.context.token_expiry_time = None
        provider._initialized = True

        provider.context.client_info = OAuthClientInformationFull(
            client_id="existing_client",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        )

        test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")
        auth_flow = provider.async_auth_flow(test_request)

        await auth_flow.__anext__()

        # 401 with custom WWW-Authenticate PRM URL
        response = httpx.Response(
            401,
            headers={
                "WWW-Authenticate": 'Bearer resource_metadata="https://custom.prm.com/.well-known/oauth-protected-resource"'
            },
            request=test_request,
        )

        # Try custom PRM URL first
        prm_request_1 = await auth_flow.asend(response)
        assert str(prm_request_1.url) == "https://custom.prm.com/.well-known/oauth-protected-resource"

        # Not served there
        prm_response_1 = httpx.Response(404, request=prm_request_1)

        # Try path-based fallback
        prm_request_2 = await auth_flow.asend(prm_response_1)
        assert str(prm_request_2.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"

        # Returns 404
        prm_response_2 = httpx.Response(404, request=prm_request_2)

        # Try root fallback
        prm_request_3 = await auth_flow.asend(prm_response_2)
        assert str(prm_request_3.url) == "https://api.example.com/.well-known/oauth-protected-resource"

        # Also returns 404 - all PRM URLs failed
        prm_response_3 = httpx.Response(404, request=prm_request_3)

        # Should fall back to root OAuth discovery
        oauth_metadata_request = await auth_flow.asend(prm_response_3)
        assert str(oauth_metadata_request.url) == "https://api.example.com/.well-known/oauth-authorization-server"

        # Complete the flow
        oauth_metadata_response = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://api.example.com", '
                b'"authorization_endpoint": "https://api.example.com/authorize", '
                b'"token_endpoint": "https://api.example.com/token"}'
            ),
            request=oauth_metadata_request,
        )

        provider._perform_authorization_code_grant = mock.AsyncMock(
            return_value=("test_auth_code", "test_code_verifier")
        )

        token_request = await auth_flow.asend(oauth_metadata_response)
        assert str(token_request.url) == "https://api.example.com/token"

        token_response = httpx.Response(
            200,
            content=b'{"access_token": "test_token", "token_type": "Bearer", "expires_in": 3600}',
            request=token_request,
        )

        final_request = await auth_flow.asend(token_response)
        assert final_request.headers["Authorization"] == "Bearer test_token"

        final_response = httpx.Response(200, request=final_request)
        try:
            await auth_flow.asend(final_response)
        except StopAsyncIteration:
            pass


class TestSEP985Discovery:
    """Test SEP-985 protected resource metadata discovery with fallback."""

    @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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
        )

        # Test with 401 response without WWW-Authenticate header
        init_response = httpx.Response(
            status_code=401, headers={}, request=httpx.Request("GET", "https://api.example.com/v1/mcp")
        )

        # Build discovery URLs
        discovery_urls = 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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
        )

        # Ensure no tokens are stored
        provider.context.current_tokens = None
        provider.context.token_expiry_time = None
        provider._initialized = True

        # Mock client info to skip DCR
        provider.context.client_info = OAuthClientInformationFull(
            client_id="existing_client",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        )

        # Create a test request
        test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")

        # Mock the auth flow
        auth_flow = provider.async_auth_flow(test_request)

        # First request should be the original request without auth header
        request = await auth_flow.__anext__()
        assert "Authorization" not in request.headers

        # Send a 401 response without WWW-Authenticate header
        response = httpx.Response(401, headers={}, request=test_request)

        # Next request should be to discover protected resource metadata (path-based)
        discovery_request_1 = await auth_flow.asend(response)
        assert str(discovery_request_1.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
        assert discovery_request_1.method == "GET"

        # Send 404 response for path-based discovery
        discovery_response_1 = httpx.Response(404, request=discovery_request_1)

        # Next request should be to root-based well-known URI
        discovery_request_2 = await auth_flow.asend(discovery_response_1)
        assert str(discovery_request_2.url) == "https://api.example.com/.well-known/oauth-protected-resource"
        assert discovery_request_2.method == "GET"

        # Send successful discovery response
        discovery_response_2 = httpx.Response(
            200,
            content=(
                b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
            ),
            request=discovery_request_2,
        )

        # Mock the rest of the OAuth flow
        provider._perform_authorization = mock.AsyncMock(return_value=("test_auth_code", "test_code_verifier"))

        # Next should be OAuth metadata discovery
        oauth_metadata_request = await auth_flow.asend(discovery_response_2)
        assert oauth_metadata_request.method == "GET"

        # Complete the flow
        oauth_metadata_response = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://auth.example.com", '
                b'"authorization_endpoint": "https://auth.example.com/authorize", '
                b'"token_endpoint": "https://auth.example.com/token"}'
            ),
            request=oauth_metadata_request,
        )

        token_request = await auth_flow.asend(oauth_metadata_response)
        token_response = httpx.Response(
            200,
            content=(
                b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600, '
                b'"refresh_token": "new_refresh_token"}'
            ),
            request=token_request,
        )

        final_request = await auth_flow.asend(token_response)
        final_response = httpx.Response(200, request=final_request)
        try:
            await auth_flow.asend(final_response)
        except StopAsyncIteration:  # pragma: no cover
            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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
        )

        # Test with 401 response with WWW-Authenticate header
        init_response = httpx.Response(
            status_code=401,
            headers={
                "WWW-Authenticate": 'Bearer resource_metadata="https://custom.example.com/.well-known/oauth-protected-resource"'
            },
            request=httpx.Request("GET", "https://api.example.com/v1/mcp"),
        )

        # Build discovery URLs
        discovery_urls = 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 = httpx.Response(
            status_code=401,
            headers={"WWW-Authenticate": www_auth_header},
            request=httpx.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 = httpx.Response(
            status_code=401, headers=headers, request=httpx.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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
            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() -> tuple[str, str | None]:
            return "test_auth_code", "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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
            client_metadata_url="https://example.com/client",
        )

        provider.context.current_tokens = None
        provider.context.token_expiry_time = None
        provider._initialized = True

        test_request = httpx.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 = httpx.Response(401, headers={}, request=test_request)

        # PRM discovery
        prm_request = await auth_flow.asend(response)
        prm_response = httpx.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 = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://auth.example.com", '
                b'"authorization_endpoint": "https://auth.example.com/authorize", '
                b'"token_endpoint": "https://auth.example.com/token", '
                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 = httpx.Response(
            200,
            content=b'{"access_token": "test_token", "token_type": "Bearer", "expires_in": 3600}',
            request=token_request,
        )

        final_request = await auth_flow.asend(token_response)
        assert final_request.headers["Authorization"] == "Bearer test_token"

        final_response = httpx.Response(200, request=final_request)
        try:
            await auth_flow.asend(final_response)
        except StopAsyncIteration:
            pass

    @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() -> tuple[str, str | None]:
            return "test_auth_code", "test_state"  # pragma: no cover

        provider = OAuthClientProvider(
            server_url="https://api.example.com/v1/mcp",
            client_metadata=client_metadata,
            storage=mock_storage,
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
            client_metadata_url="https://example.com/client",
        )

        provider.context.current_tokens = None
        provider.context.token_expiry_time = None
        provider._initialized = True

        test_request = httpx.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 = httpx.Response(401, headers={}, request=test_request)

        # PRM discovery
        prm_request = await auth_flow.asend(response)
        prm_response = httpx.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 = httpx.Response(
            200,
            content=(
                b'{"issuer": "https://auth.example.com", '
                b'"authorization_endpoint": "https://auth.example.com/authorize", '
                b'"token_endpoint": "https://auth.example.com/token", '
                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 = httpx.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 = httpx.Response(
            200,
            content=b'{"access_token": "test_token", "token_type": "Bearer", "expires_in": 3600}',
            request=token_request,
        )

        final_request = await auth_flow.asend(token_response)
        final_response = httpx.Response(200, request=final_request)
        try:
            await auth_flow.asend(final_response)
        except StopAsyncIteration:
            pass


@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_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 - both already have trailing slashes
    await provider._validate_resource_match(prm)


@pytest.mark.anyio
async def test_get_resource_url_falls_back_when_prm_mismatches(
    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
) -> None:
    """get_resource_url returns 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() == "https://api.example.com/v1/mcp"


def _prepare_full_flow(provider: OAuthClientProvider, client_info: OAuthClientInformationFull | None) -> list[str]:
    """Reset `provider` for a full flow with `client_info` as the stored registration, and wire a
    redirect/callback pair that echoes the `state` of the last authorization URL it was sent to.
    Returns the list the redirect handler appends authorization URLs to."""
    provider.context.current_tokens = None
    provider.context.token_expiry_time = None
    provider._initialized = True
    provider.context.client_info = client_info
    redirects: list[str] = []

    async def record_redirect(url: str) -> None:
        redirects.append(url)

    async def echo_callback() -> tuple[str, str | None]:
        return "auth_code", parse_qs(urlparse(redirects[-1]).query)["state"][0]

    provider.context.redirect_handler = record_redirect
    provider.context.callback_handler = echo_callback
    return redirects


def _asm(issuer: str, *, token_origin: str | None = None, registration: bool = False) -> bytes:
    metadata = {
        "issuer": issuer,
        "authorization_endpoint": f"{issuer}/authorize",
        "token_endpoint": f"{token_origin or issuer}/token",
    }
    if registration:
        metadata["registration_endpoint"] = f"{issuer}/register"
    return json.dumps(metadata).encode()


@pytest.mark.anyio
async def test_metadata_issuer_must_match_the_advertised_authorization_server(oauth_provider: OAuthClientProvider):
    """RFC 8414 section 3.3: metadata fetched for the PRM-advertised authorization server must
    name that server as its issuer; metadata naming another issuer is refused before
    registration, authorization or token requests are built from it."""
    _prepare_full_flow(oauth_provider, None)
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    prm_req = await auth_flow.asend(httpx.Response(401, request=request))
    prm_response = httpx.Response(
        200,
        content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
        request=prm_req,
    )
    asm_req = await auth_flow.asend(prm_response)
    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"

    asm_response = httpx.Response(200, content=_asm("https://other-as.example.com", registration=True), request=asm_req)
    with pytest.raises(OAuthFlowError) as exc_info:
        await auth_flow.asend(asm_response)

    assert str(exc_info.value) == snapshot(
        "Authorization server metadata issuer mismatch: https://other-as.example.com/ != https://auth.example.com/"
    )


@pytest.mark.anyio
async def test_legacy_fallback_metadata_naming_a_different_issuer_is_refused(oauth_provider: OAuthClientProvider):
    """RFC 8414 section 3.3 on the legacy no-PRM path: metadata served from the resource server's
    own well-known must name that origin as its issuer.

    Metadata naming a different authorization server is refused before any authorization or
    token request is built, so a stored confidential client is never presented to the endpoints
    that metadata lists.
    """
    _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="confidential-client",
            client_secret="stored-secret",
            token_endpoint_auth_method="client_secret_post",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        ),
    )
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    # 401 without WWW-Authenticate; both PRM well-knowns 404; legacy root ASM discovery.
    prm_req = await auth_flow.asend(httpx.Response(401, request=request))
    prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"

    # The resource origin's well-known names another server as issuer while listing its own
    # token endpoint.
    asm_response = httpx.Response(
        200, content=_asm("https://other-as.example.com", token_origin="https://api.example.com"), request=asm_req
    )
    with pytest.raises(OAuthFlowError) as exc_info:
        await auth_flow.asend(asm_response)

    assert str(exc_info.value) == snapshot(
        "Authorization server metadata issuer mismatch: https://other-as.example.com/ != https://api.example.com/"
    )


_ISSUER = "https://as.example.com/tenant"


def _issuer_metadata(issuer: str = _ISSUER) -> OAuthMetadata:
    return OAuthMetadata.model_validate(
        {"issuer": issuer, "authorization_endpoint": f"{issuer}/authorize", "token_endpoint": f"{issuer}/token"}
    )


def test_validate_metadata_issuer_accepts_match():
    validate_metadata_issuer(_issuer_metadata(_ISSUER), _ISSUER)


def test_validate_metadata_issuer_rejects_mismatch():
    with pytest.raises(OAuthFlowError, match="issuer mismatch"):
        validate_metadata_issuer(_issuer_metadata("https://other-as.example.com/tenant"), _ISSUER)


@pytest.mark.parametrize(
    ("issuer", "expected"),
    [
        pytest.param("https://as.example.com/", "https://as.example.com", id="metadata-has-root-slash"),
        pytest.param("https://as.example.com/", "https://as.example.com/", id="both-have-root-slash"),
    ],
)
def test_validate_metadata_issuer_treats_empty_path_and_root_slash_as_the_same_issuer(issuer: str, expected: str):
    """SDK-defined tolerance: an origin with an empty path and the same origin with a lone `/`
    identify the same server (RFC 3986 section 6.2.3). A root issuer always parses to the `/`
    form here, while the legacy discovery URL is built from the bare origin, so the two must
    compare equal."""
    validate_metadata_issuer(_issuer_metadata(issuer), expected)


@pytest.mark.parametrize(
    ("issuer", "expected"),
    [
        pytest.param("https://as.example.com/tenant/", "https://as.example.com/tenant", id="non-root-trailing-slash"),
        pytest.param("https://as.example.com/tenant", "https://as.example.com", id="different-path"),
        pytest.param("http://as.example.com/", "https://as.example.com", id="different-scheme"),
        pytest.param("https://as.example.com:8443/", "https://as.example.com", id="different-port"),
        pytest.param("https://as.example.com//", "https://as.example.com", id="double-slash"),
    ],
)
def test_validate_metadata_issuer_root_slash_tolerance_does_not_extend_further(issuer: str, expected: str):
    """The empty-path tolerance is exactly that: any other difference is still a mismatch."""
    with pytest.raises(OAuthFlowError, match="metadata issuer mismatch"):
        validate_metadata_issuer(_issuer_metadata(issuer), 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_root_slash_is_the_same_issuer():
    """A binding written as the bare origin matches the `/` form a root URL parses to, and back."""
    info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as")
    assert credentials_match_issuer(info, "https://as/", None) is True
    info.issuer = "https://as/"
    assert credentials_match_issuer(info, "https://as", None) is True


def test_credentials_match_issuer_root_slash_tolerance_does_not_extend_to_other_paths():
    """SDK-defined: a trailing slash on a non-root path is a different issuer."""
    info = OAuthClientInformationFull(
        client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as.example.com/tenant"
    )
    assert credentials_match_issuer(info, "https://as.example.com/tenant/", None) is False


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
@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", "redirect_uris": ["http://localhost:3030/callback"], "issuer": echoed_issuer}
    ).encode()

    client_info = await handle_registration_response(httpx.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\xe9"}'.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."""
    with pytest.raises(OAuthRegistrationError):
        await handle_registration_response(httpx.Response(201, content=content))


@pytest.mark.anyio
async def test_stored_credentials_are_not_presented_to_a_different_authorization_server(
    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
):
    """SEP-2352: stored credentials are bound to the authorization server that registered them.

    Steps:
    1. Storage holds a confidential client bound to `https://auth.example.com/`.
    2. PRM now advertises `https://other-as.example.com` -> the stored client and its tokens are
       discarded before that server's metadata is fetched.
    3. Metadata for the new server is discovered -> the flow registers there and the token
       request carries the new client, not the discarded `client_id`/`client_secret`.
    4. The new registration is recorded as bound to the new server.
    """
    redirects = _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="confidential-client",
            client_secret="stored-secret",
            token_endpoint_auth_method="client_secret_post",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
            issuer="https://auth.example.com/",
        ),
    )
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    prm_req = await auth_flow.asend(httpx.Response(401, request=request))
    prm_response = httpx.Response(
        200,
        content=(
            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://other-as.example.com"]}'
        ),
        request=prm_req,
    )
    asm_req = await auth_flow.asend(prm_response)
    assert str(asm_req.url) == "https://other-as.example.com/.well-known/oauth-authorization-server"
    assert oauth_provider.context.client_info is None

    register_req = await auth_flow.asend(
        httpx.Response(200, content=_asm("https://other-as.example.com", registration=True), request=asm_req)
    )
    assert register_req.method == "POST"
    assert str(register_req.url) == "https://other-as.example.com/register"
    register_response = httpx.Response(
        201,
        json={"client_id": "new-client", "redirect_uris": ["http://localhost:3030/callback"]},
        request=register_req,
    )

    token_req = await auth_flow.asend(register_response)
    assert str(token_req.url) == "https://other-as.example.com/token"
    assert redirects[-1].startswith("https://other-as.example.com/authorize?")
    token_form = parse_qs(token_req.content.decode())
    assert token_form["client_id"] == ["new-client"]
    assert "client_secret" not in token_form

    stored = await mock_storage.get_client_info()
    assert stored is not None
    assert stored.client_id == "new-client"
    assert stored.issuer == "https://other-as.example.com/"
    await auth_flow.aclose()


@pytest.mark.anyio
async def test_stored_credentials_bound_to_the_advertised_authorization_server_are_kept(
    oauth_provider: OAuthClientProvider,
):
    """SEP-2352 positive control: a stored client bound to the server PRM advertises (written
    with or without the root slash) is reused - no registration request is made."""
    _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="bound-client",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
            issuer="https://auth.example.com",
        ),
    )
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    prm_req = await auth_flow.asend(httpx.Response(401, request=request))
    prm_response = httpx.Response(
        200,
        content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
        request=prm_req,
    )
    asm_req = await auth_flow.asend(prm_response)
    assert oauth_provider.context.client_info is not None

    token_req = await auth_flow.asend(
        httpx.Response(200, content=_asm("https://auth.example.com", registration=True), request=asm_req)
    )
    assert str(token_req.url) == "https://auth.example.com/token"
    assert parse_qs(token_req.content.decode())["client_id"] == ["bound-client"]
    await auth_flow.aclose()


@pytest.mark.anyio
async def test_issuer_binding_evaluated_against_the_server_origin_when_prm_discovery_failed(
    oauth_provider: OAuthClientProvider,
):
    """SEP-2352: on the legacy no-PRM path the binding check uses the resource server's origin.

    PRM discovery fails (404) so `auth_server_url` stays `None`; the legacy well-known URL is
    built from the resource server's origin, which is therefore the issuer any metadata found
    there must carry (RFC 8414 section 3.3). Stored credentials bound to a different issuer are
    discarded before that metadata is fetched, and the flow re-registers.
    """
    _prepare_full_flow(
        oauth_provider,
        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(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    # PRM discovery: path-based then root, both 404.
    prm_req = await auth_flow.asend(httpx.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(httpx.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): the stale credentials are already
    # gone when the request is issued.
    asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
    assert oauth_provider.context.client_info is None

    # 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(
        httpx.Response(200, content=_asm("https://api.example.com", registration=True), request=asm_req)
    )
    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(
    "first_response",
    [(401, {}), (403, {"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin"'})],
    ids=["401", "403-insufficient-scope"],
)
async def test_legacy_fallback_without_metadata_re_registers_instead_of_presenting_credentials_bound_elsewhere(
    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken, first_response: tuple[int, dict[str, str]]
):
    """SEP-2352 on the legacy no-PRM path when no metadata is served at all, whether the flow starts
    from a 401 or from a 403 scope challenge with no metadata held.

    Steps:
    1. Storage holds a token and a confidential client bound to a different authorization server.
    2. Both PRM well-knowns 404 -> the expected issuer is the resource server's origin, so the
       stored client is discarded before ASM discovery.
    3. The origin's ASM well-known 404s too -> the flow registers a fresh client at the
       origin's default `/register` and authorizes with it; the token request to the origin's
       default `/token` carries the new client and none of the discarded credentials.
    """
    redirects = _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="confidential-client",
            client_secret="stored-secret",
            token_endpoint_auth_method="client_secret_post",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
            issuer="https://other-as.example.com",
        ),
    )
    oauth_provider.context.current_tokens = valid_tokens
    oauth_provider.context.token_expiry_time = time.time() + 1800
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    status, headers = first_response
    prm_req = await auth_flow.asend(httpx.Response(status, headers=headers, request=request))
    prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
    assert oauth_provider.context.client_info is None

    # No metadata at the origin either: register at the origin's default endpoint.
    register_req = await auth_flow.asend(httpx.Response(404, request=asm_req))
    assert register_req.method == "POST"
    assert str(register_req.url) == "https://api.example.com/register"
    register_response = httpx.Response(
        201,
        json={
            "client_id": "origin-client",
            "redirect_uris": ["http://localhost:3030/callback"],
            "token_endpoint_auth_method": "none",
        },
        request=register_req,
    )

    token_req = await auth_flow.asend(register_response)
    assert str(token_req.url) == "https://api.example.com/token"
    assert redirects[-1].startswith("https://api.example.com/authorize?")
    token_form = parse_qs(token_req.content.decode())
    assert token_form["client_id"] == ["origin-client"]
    assert "client_secret" not in token_form
    await auth_flow.aclose()


@pytest.mark.anyio
async def test_scope_step_up_discovers_the_authorization_server_before_reauthorizing(
    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
):
    """SDK-defined: a 403 scope challenge runs discovery first when no metadata is held, so
    re-authorization targets the advertised server.

    Steps:
    1. A restarted client holds a token and a registration but no authorization server metadata.
    2. The first response is 403 insufficient_scope -> the next requests are PRM (at the challenge's
       `resource_metadata` URL) then ASM discovery.
    3. The authorization redirect and the token request use the discovered server's endpoints and
       ask for the challenged scope.
    """
    redirects = _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="registered-client", redirect_uris=[AnyUrl("http://localhost:3030/callback")]
        ),
    )
    oauth_provider.context.current_tokens = valid_tokens
    oauth_provider.context.token_expiry_time = time.time() + 1800
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()
    response_403 = httpx.Response(
        403,
        headers={
            "WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin",'
            ' resource_metadata="https://api.example.com/v1/mcp/resource-metadata"'
        },
        request=request,
    )

    prm_request = await auth_flow.asend(response_403)
    assert (prm_request.method, str(prm_request.url)) == ("GET", "https://api.example.com/v1/mcp/resource-metadata")
    prm = b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
    asm_request = await auth_flow.asend(httpx.Response(200, content=prm, request=prm_request))
    assert str(asm_request.url) == "https://auth.example.com/.well-known/oauth-authorization-server"

    token_request = await auth_flow.asend(
        httpx.Response(200, content=_asm("https://auth.example.com"), request=asm_request)
    )

    assert redirects[-1].startswith("https://auth.example.com/authorize?")
    assert parse_qs(urlparse(redirects[-1]).query)["scope"] == ["admin"]
    assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token")
    await auth_flow.aclose()


@pytest.mark.anyio
async def test_scope_step_up_reuses_metadata_discovered_earlier_in_the_process(
    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
):
    """SDK-defined: once metadata has been discovered in this process, a step-up re-authorizes with it
    directly (no discovery requests) and asks for the challenged scope."""
    redirects = _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="registered-client", redirect_uris=[AnyUrl("http://localhost:3030/callback")]
        ),
    )
    oauth_provider.context.current_tokens = valid_tokens
    oauth_provider.context.token_expiry_time = time.time() + 1800
    oauth_provider.context.auth_server_url = "https://auth.example.com/"
    oauth_provider.context.oauth_metadata = OAuthMetadata.model_validate_json(_asm("https://auth.example.com"))
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()
    response_403 = httpx.Response(
        403, headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin"'}, request=request
    )

    token_request = await auth_flow.asend(response_403)

    assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token")
    assert redirects[-1].startswith("https://auth.example.com/authorize?")
    assert parse_qs(urlparse(redirects[-1]).query)["scope"] == ["admin"]
    await auth_flow.aclose()


@pytest.mark.anyio
async def test_403_without_a_scope_challenge_is_returned_to_the_caller(
    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
):
    """SDK-defined: a 403 that is not an insufficient_scope challenge ends the flow; the request is
    not retried."""
    _prepare_full_flow(oauth_provider, None)
    oauth_provider.context.current_tokens = valid_tokens
    oauth_provider.context.token_expiry_time = time.time() + 1800
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()

    with pytest.raises(StopAsyncIteration):
        await auth_flow.asend(
            httpx.Response(403, headers={"WWW-Authenticate": 'Bearer error="access_denied"'}, request=request)
        )


@pytest.mark.anyio
@pytest.mark.parametrize("status", [500, 503, 429])
async def test_a_failing_resource_metadata_request_stops_the_flow_and_keeps_stored_credentials(
    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken, status: int
):
    """SDK-defined: a server error (or 429) on a protected resource metadata request says nothing about
    whether the server publishes that metadata. The remaining well-known locations are still tried, but
    when none answers the flow stops instead of taking the legacy path, and a registration bound to the
    advertised authorization server and its tokens stay as they were."""
    bound = OAuthClientInformationFull(
        client_id="registered-client",
        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        issuer="https://auth.example.com/",
    )
    _prepare_full_flow(oauth_provider, bound)
    oauth_provider.context.current_tokens = valid_tokens
    oauth_provider.context.token_expiry_time = time.time() + 1800
    flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await flow.__anext__()
    prm_request = await flow.asend(httpx.Response(401, request=request))
    root_prm_request = await flow.asend(httpx.Response(status, request=prm_request))
    assert str(root_prm_request.url) == "https://api.example.com/.well-known/oauth-protected-resource"

    with pytest.raises(OAuthFlowError) as exc_info:
        await flow.asend(httpx.Response(404, request=root_prm_request))

    assert str(exc_info.value) == f"Protected resource metadata request failed: HTTP {status}"
    assert oauth_provider.context.client_info == bound
    assert oauth_provider.context.current_tokens == valid_tokens


@pytest.mark.anyio
async def test_a_failing_resource_metadata_location_does_not_matter_when_another_one_answers(
    oauth_provider: OAuthClientProvider,
):
    """SDK-defined: the well-known locations are tried in order; an error at one of them is forgotten
    once a later one returns the metadata."""
    _prepare_full_flow(oauth_provider, None)
    flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await flow.__anext__()
    prm_request = await flow.asend(httpx.Response(401, request=request))
    root_prm_request = await flow.asend(httpx.Response(503, request=prm_request))
    prm = b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'

    asm_request = await flow.asend(httpx.Response(200, content=prm, request=root_prm_request))

    assert str(asm_request.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
    await flow.aclose()


@pytest.mark.anyio
@pytest.mark.parametrize(
    "served_issuer", ["https://api.example.com", "https://api.example.com/"], ids=["bare", "root-slash"]
)
async def test_legacy_fallback_accepts_the_origin_issuer_for_a_server_url_in_any_spelling(
    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage, served_issuer: str
):
    """SDK-defined: on the legacy no-PRM path the expected issuer is the resource server's origin; a
    `server_url` written with an upper-case host and an explicit default port still matches metadata
    naming that origin, with or without its trailing slash, and the flow proceeds to registration."""

    async def redirect_handler(url: str) -> None:
        raise NotImplementedError

    async def callback_handler() -> tuple[str, str | None]:
        raise NotImplementedError

    provider = OAuthClientProvider(
        server_url="https://API.Example.com:443/v1/mcp",
        client_metadata=client_metadata,
        storage=mock_storage,
        redirect_handler=redirect_handler,
        callback_handler=callback_handler,
    )
    auth_flow = provider.async_auth_flow(httpx.Request("GET", "https://API.Example.com:443/v1/mcp"))
    request = await auth_flow.__anext__()
    prm_req = await auth_flow.asend(httpx.Response(401, request=request))
    prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
    asm = {
        "issuer": served_issuer,
        "authorization_endpoint": "https://api.example.com/authorize",
        "token_endpoint": "https://api.example.com/token",
        "registration_endpoint": "https://api.example.com/register",
    }

    register_req = await auth_flow.asend(httpx.Response(200, json=asm, request=asm_req))

    assert (register_req.method, str(register_req.url)) == ("POST", "https://api.example.com/register")
    await auth_flow.aclose()


@pytest.mark.anyio
@pytest.mark.parametrize(
    "asm_responses",
    [
        pytest.param([httpx.Response(404), httpx.Response(404)], id="asm-discovery-failed"),
        pytest.param(
            [httpx.Response(200, content=_asm("https://new-as.example.com"))],
            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[httpx.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.
    """
    _prepare_full_flow(
        oauth_provider,
        OAuthClientInformationFull(
            client_id="stale-client",
            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
            issuer="https://api.example.com/",
        ),
    )
    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
    request = await auth_flow.__anext__()
    response_401 = httpx.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 = httpx.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 = httpx.Response(
        201,
        json={"client_id": "fallback-client", "redirect_uris": ["http://localhost:3030/callback"]},
        request=dcr_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
    await auth_flow.aclose()


@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.
    """
    _prepare_full_flow(oauth_provider, None)
    auth_flow = oauth_provider.async_auth_flow(httpx.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(httpx.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(httpx.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(httpx.Response(404, request=prm_req))
    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"

    # DCR falls back to the resource origin's /register - the issuer's own host.
    dcr_req = await auth_flow.asend(httpx.Response(200, content=_asm("https://api.example.com"), request=asm_req))
    assert dcr_req.method == "POST"
    assert str(dcr_req.url) == "https://api.example.com/register"
    dcr_response = httpx.Response(
        201,
        json={"client_id": "embedded-client", "redirect_uris": ["http://localhost:3030/callback"]},
        request=dcr_req,
    )
    await auth_flow.asend(dcr_response)

    stored = await mock_storage.get_client_info()
    assert stored is not None
    assert stored.client_id == "embedded-client"
    assert stored.issuer == "https://api.example.com/"
    await auth_flow.aclose()