fastmcp/tests/server/auth/test_enhanced_error_responses.py

"""Tests for enhanced OAuth error responses.

This test suite covers:
1. Enhanced authorization handler (HTML and JSON error pages)
2. Enhanced middleware (better error messages)
3. Content negotiation
4. Server branding in error pages
"""

import asyncio
from urllib.parse import parse_qs, quote, urlparse

import pytest
from key_value.aio.stores.memory import MemoryStore
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyHttpUrl, AnyUrl
from starlette.applications import Starlette
from starlette.testclient import TestClient

from fastmcp import FastMCP
from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.oauth_proxy import OAuthProxy
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
from fastmcp.server.http import create_streamable_http_app


class _UnderScopedTokenVerifier(TokenVerifier):
    def __init__(self, required_scopes: list[str]):
        super().__init__(required_scopes=required_scopes)

    async def verify_token(self, token: str) -> AccessToken:
        return AccessToken(token=token, client_id="test-client", scopes=["other"])


class _UnderScopedOAuthProxy(OAuthProxy):
    async def verify_token(self, token: str) -> AccessToken:
        return AccessToken(token=token, client_id="test-client", scopes=["other"])


class _DirectClientRedirectOAuthProxy(OAuthProxy):
    """Proxy whose `authorize()` bypasses consent/upstream entirely and
    redirects straight back to the client with a `code` — the pattern used
    by providers (or tests) that short-circuit the standard
    consent -> upstream IdP -> callback flow. OAuthProxy's own `authorize()`
    never does this itself, but a subclass legitimately can, and
    `AuthorizationHandler.handle()` must still attach `iss` to whatever
    redirect comes back.

    Appends its own `code`/`state` with `&` rather than an unconditional
    `?` so this still produces a well-formed URL when `redirect_uri` is a
    registered redirect that already carries its own query string (e.g. a
    client-supplied `iss`)."""

    async def authorize(self, client, params):  # type: ignore[override]
        separator = "&" if "?" in str(params.redirect_uri) else "?"
        return (
            f"{params.redirect_uri}{separator}code=test-auth-code&state={params.state}"
        )


class _DirectClientRedirectWithIssOAuthProxy(OAuthProxy):
    """Like `_DirectClientRedirectOAuthProxy`, but the provider's
    `authorize()` override already put its own `iss` on the redirect —
    simulating a provider that is itself RFC 9207-aware (or, when
    `redirect_iss` doesn't match this server's issuer, a provider bug).
    `response_kind` selects whether the redirect looks like a success
    (`code`) or error (`error`) response; `AuthorizationHandler.handle()`
    must not duplicate `iss` on either."""

    def __init__(
        self,
        *args,
        redirect_iss: str,
        response_kind: str = "code",
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self._redirect_iss = redirect_iss
        self._response_kind = response_kind

    async def authorize(self, client, params):  # type: ignore[override]
        payload = (
            f"code=test-auth-code&state={params.state}"
            if self._response_kind == "code"
            else f"error=access_denied&state={params.state}"
        )
        iss = quote(self._redirect_iss, safe="")
        return f"{params.redirect_uri}?{payload}&iss={iss}"


class TestEnhancedAuthorizationHandler:
    """Tests for enhanced authorization handler error responses."""

    @pytest.fixture
    def rsa_key_pair(self) -> RSAKeyPair:
        """Generate RSA key pair for testing."""
        return RSAKeyPair.generate()

    @pytest.fixture
    def oauth_proxy(self, rsa_key_pair):
        """Create OAuth proxy for testing."""
        return OAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )

    def test_unregistered_client_returns_html_for_browser(self, oauth_proxy):
        """Test that unregistered client returns styled HTML for browser requests."""
        app = Starlette(routes=oauth_proxy.get_routes())

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                headers={"Accept": "text/html"},
            )

            # Should return 400 with HTML content
            assert response.status_code == 400
            assert "text/html" in response.headers["content-type"]

            # HTML should contain error message
            html = response.text
            assert "Client Not Registered" in html
            assert "unregistered-client-id" in html
            assert "To fix this" in html
            assert "Close this browser window" in html
            assert "Clear authentication tokens" in html

            # Should have Link header for registration endpoint
            assert "Link" in response.headers
            assert "/register" in response.headers["Link"]

    def test_unregistered_client_returns_json_for_api(self, oauth_proxy):
        """Test that unregistered client returns enhanced JSON for API clients."""
        app = Starlette(routes=oauth_proxy.get_routes())

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                headers={"Accept": "application/json"},
            )

            # Should return 400 with JSON content
            assert response.status_code == 400
            assert "application/json" in response.headers["content-type"]

            # JSON should have enhanced error response
            data = response.json()
            assert data["error"] == "invalid_request"
            assert "unregistered-client-id" in data["error_description"]
            assert data["state"] == "test-state"

            # Should include registration endpoint hints
            assert "registration_endpoint" in data
            assert data["registration_endpoint"] == "https://myserver.com/register"
            assert "authorization_server_metadata" in data

            # Should have Link header
            assert "Link" in response.headers
            assert "/register" in response.headers["Link"]

    def test_successful_authorization_not_enhanced(self, oauth_proxy):
        """Test that successful authorizations are not modified by enhancement."""
        app = Starlette(routes=oauth_proxy.get_routes())

        # Register a valid client first
        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
        )

        # Need to register synchronously
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                headers={"Accept": "text/html"},
                follow_redirects=False,
            )

            # Should redirect to consent page (302), not return error
            assert response.status_code == 302
            assert "/consent" in response.headers["location"]

    def test_redirect_error_includes_proxy_issuer(self, oauth_proxy):
        """Authorization error redirects should include RFC 9207 issuer."""
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )

        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                    "scope": "admin",
                },
                headers={"Accept": "text/html"},
                follow_redirects=False,
            )

            assert response.status_code == 302
            query_params = parse_qs(urlparse(response.headers["location"]).query)
            assert query_params["error"] == ["invalid_scope"]
            assert query_params["state"] == ["test-state"]
            assert query_params["iss"] == ["https://myserver.com/"]

    def test_redirect_error_matches_path_base_url_metadata_issuer(self, rsa_key_pair):
        """Authorization error redirects should match the metadata issuer exactly."""
        oauth_proxy = OAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://proxy.example.com/oauth",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )

        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata_response = client.get("/.well-known/oauth-authorization-server")
            metadata = metadata_response.json()

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                    "scope": "admin",
                },
                headers={"Accept": "text/html"},
                follow_redirects=False,
            )

            assert metadata["issuer"] == "https://proxy.example.com/oauth"
            assert response.status_code == 302
            query_params = parse_qs(urlparse(response.headers["location"]).query)
            assert query_params["error"] == ["invalid_scope"]
            assert query_params["state"] == ["test-state"]
            assert query_params["iss"] == [metadata["issuer"]]

    def test_success_redirect_from_authorize_override_includes_issuer(
        self, rsa_key_pair
    ):
        """RFC 9207 regression: a `code` redirect returned directly by
        `authorize()` (bypassing consent/upstream) must carry `iss` too, not
        just `error` redirects.

        `AuthorizationHandler.handle()` previously only attached `iss` when
        the SDK's redirect contained an `error` parameter. The base
        `OAuthProxy.authorize()` never redirects straight to the client, so
        this gap was invisible until a provider override (or a test mock,
        like the GitHub provider integration test) returned the client
        redirect directly — at which point the server was advertising
        `authorization_response_iss_parameter_supported: true` while
        silently breaking RFC 9207-aware clients on this path.
        """
        oauth_proxy = _DirectClientRedirectOAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata = client.get("/.well-known/oauth-authorization-server").json()
            assert metadata["authorization_response_iss_parameter_supported"] is True

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                follow_redirects=False,
            )

        assert response.status_code == 302
        query_params = parse_qs(urlparse(response.headers["location"]).query)
        assert query_params["code"] == ["test-auth-code"]
        assert query_params["state"] == ["test-state"]
        assert query_params["iss"] == [metadata["issuer"]]

    def test_success_redirect_does_not_duplicate_iss_already_in_redirect_uri(
        self, rsa_key_pair
    ):
        """RFC 9207 P2 regression: a registered redirect_uri may already
        carry its own `iss` query parameter — distinct from the provider
        adding one itself (covered by
        `test_success_redirect_with_matching_iss_not_duplicated` below).
        `AuthorizationHandler.handle()` must still land on exactly one
        `iss` (the canonical value), with every other query byte on the
        registered URI preserved untouched.
        """
        oauth_proxy = _DirectClientRedirectOAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_redirect_uri = (
            "http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
        )
        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl(client_redirect_uri)],
            scope="read",
        )
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata = client.get("/.well-known/oauth-authorization-server").json()

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": client_redirect_uri,
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                follow_redirects=False,
            )

        assert response.status_code == 302
        location = response.headers["location"]
        query = urlparse(location).query
        query_params = parse_qs(query)
        assert query_params["code"] == ["test-auth-code"]
        assert query_params["state"] == ["test-state"]
        # Exactly one `iss`, corrected to the canonical value -- a
        # duplicate would make this list have length 2.
        assert query_params["iss"] == [metadata["issuer"]]
        # Other query bytes from the registered redirect_uri survive
        # byte-for-byte.
        assert "flag" in query
        assert "sig=%FF%FE" in query

    def test_success_redirect_with_matching_iss_not_duplicated(self, rsa_key_pair):
        """If a provider's `authorize()` override already stamped the
        correct `iss` on its redirect, `handle()` must not append a second
        one — RFC 6749 §3.1 forbids a response parameter appearing twice.
        """
        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
            redirect_iss="https://myserver.com/",
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata = client.get("/.well-known/oauth-authorization-server").json()

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                follow_redirects=False,
            )

        assert response.status_code == 302
        query_params = parse_qs(urlparse(response.headers["location"]).query)
        # Exactly one `iss` (a duplicate would make this list have length 2).
        assert query_params["iss"] == [metadata["issuer"]]

    def test_success_redirect_with_mismatched_iss_is_corrected(self, rsa_key_pair):
        """A provider's `authorize()` override can put an `iss` on its
        redirect that doesn't match what this server advertises in its own
        discovery document (`self._issuer`). An RFC 9207 client validates
        `iss` against that document, so the mismatched value is already
        unusable to a spec-compliant client. `handle()` corrects it to the
        canonical value rather than leaving the broken value in place or
        appending a second `iss` (which RFC 6749 §3.1 forbids outright).

        This is a deliberate policy choice, not the only defensible one —
        see the comment in `AuthorizationHandler.handle()` for the
        reasoning, and update this test if that policy changes.
        """
        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
            redirect_iss="https://wrong-issuer.example.com/",
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata = client.get("/.well-known/oauth-authorization-server").json()

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                follow_redirects=False,
            )

        assert response.status_code == 302
        query_params = parse_qs(urlparse(response.headers["location"]).query)
        # Exactly one `iss`, corrected to the canonical value rather than
        # left mismatched or duplicated.
        assert query_params["iss"] == [metadata["issuer"]]
        assert query_params["iss"] != ["https://wrong-issuer.example.com/"]

    def test_error_redirect_with_existing_iss_not_duplicated(self, rsa_key_pair):
        """The duplication guard applies to error redirects too, not just
        success ones — a provider override can construct an `error`
        redirect that already carries `iss`.
        """
        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=rsa_key_pair.public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
            redirect_iss="https://myserver.com/",
            response_kind="error",
        )
        app = Starlette(routes=oauth_proxy.get_routes())

        client_info = OAuthClientInformationFull(
            client_id="valid-client",
            client_secret="valid-secret",
            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
            scope="read",
        )
        asyncio.run(oauth_proxy.register_client(client_info))

        with TestClient(app) as client:
            metadata = client.get("/.well-known/oauth-authorization-server").json()

            response = client.get(
                "/authorize",
                params={
                    "client_id": "valid-client",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                    "state": "test-state",
                },
                follow_redirects=False,
            )

        assert response.status_code == 302
        query_params = parse_qs(urlparse(response.headers["location"]).query)
        assert query_params["error"] == ["access_denied"]
        assert query_params["iss"] == [metadata["issuer"]]

    def test_html_error_includes_server_branding(self, oauth_proxy):
        """Test that HTML error page includes server branding from FastMCP instance."""
        from mcp_types import Icon

        # Create FastMCP server with custom branding
        mcp = FastMCP(
            "My Custom Server",
            icons=[Icon(src="https://example.com/icon.png", mime_type="image/png")],
        )

        # Create app with OAuth routes
        app = Starlette(routes=oauth_proxy.get_routes())
        # Attach FastMCP instance to app state (same as done in http.py)
        app.state.fastmcp_server = mcp

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                },
                headers={"Accept": "text/html"},
            )

            assert response.status_code == 400
            html = response.text

            # Should include custom server icon
            assert "https://example.com/icon.png" in html


class TestEnhancedRequireAuthMiddleware:
    """Tests for enhanced authentication middleware error messages."""

    @staticmethod
    def create_scoped_app(
        required_scopes: list[str],
        scopes_supported: list[str],
        challenge_scopes: list[str] | None = None,
    ) -> Starlette:
        auth = RemoteAuthProvider(
            token_verifier=_UnderScopedTokenVerifier(required_scopes),
            authorization_servers=[AnyHttpUrl("https://auth.example.com")],
            base_url="http://localhost:8000",
            scopes_supported=scopes_supported,
            challenge_scopes=challenge_scopes,
        )
        return FastMCP("Test Server", auth=auth).http_app()

    @staticmethod
    def create_oauth_app() -> Starlette:
        from key_value.aio.stores.memory import MemoryStore

        auth = _UnderScopedOAuthProxy(
            upstream_authorization_endpoint="https://auth.example.com/authorize",
            upstream_token_endpoint="https://auth.example.com/token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=_UnderScopedTokenVerifier(["openid"]),
            base_url="http://localhost:8000",
            valid_scopes=["openid", "email", "calendar"],
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )
        return FastMCP("Test Server", auth=auth).http_app()

    @pytest.fixture
    def rsa_key_pair(self) -> RSAKeyPair:
        """Generate RSA key pair for testing."""
        return RSAKeyPair.generate()

    @pytest.fixture
    def jwt_verifier(self, rsa_key_pair):
        """Create JWT verifier for testing."""
        return JWTVerifier(
            public_key=rsa_key_pair.public_key,
            issuer="https://test.com",
            audience="https://test.com",
            base_url="https://test.com",
        )

    def test_missing_auth_no_error_attribute(self, jwt_verifier):
        """Test that missing auth returns 401 without error attribute (RFC 6750 §3.1)."""
        server = FastMCP("Test Server")

        @server.tool
        def test_tool() -> str:
            return "test"

        app = create_streamable_http_app(
            server=server,
            streamable_http_path="/mcp",
            auth=jwt_verifier,
        )

        with TestClient(app) as client:
            # Request without Authorization header
            response = client.post("/mcp")

            assert response.status_code == 401
            assert "www-authenticate" in response.headers

            # Per RFC 6750 §3.1: no error attribute when auth is missing
            www_auth = response.headers["www-authenticate"]
            assert "error=" not in www_auth
            assert response.content == b""

    def test_missing_auth_challenge_includes_supported_scopes(self):
        app = self.create_scoped_app(
            required_scopes=["read"],
            scopes_supported=["api://client-id/read"],
            challenge_scopes=["api://client-id/read"],
        )

        with TestClient(app) as client:
            response = client.post("/mcp")

        assert response.status_code == 401
        assert response.headers["www-authenticate"] == (
            'Bearer scope="api://client-id/read", '
            'resource_metadata="http://localhost:8000/'
            '.well-known/oauth-protected-resource/mcp"'
        )

    def test_insufficient_scope_challenge_includes_supported_scopes(self):
        app = self.create_scoped_app(
            required_scopes=["read"],
            scopes_supported=["api://client-id/read"],
            challenge_scopes=["api://client-id/read"],
        )

        with TestClient(app) as client:
            response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})

        assert response.status_code == 403
        assert response.headers["www-authenticate"] == (
            'Bearer error="insufficient_scope", '
            'error_description="Required scope: read", '
            'scope="api://client-id/read", '
            'resource_metadata="http://localhost:8000/'
            '.well-known/oauth-protected-resource/mcp"'
        )

    def test_missing_auth_challenge_uses_required_scope_with_empty_catalog(self):
        app = self.create_scoped_app(required_scopes=["read"], scopes_supported=[])

        with TestClient(app) as client:
            response = client.post("/mcp")

        assert response.status_code == 401
        assert response.headers["www-authenticate"] == (
            'Bearer scope="read", resource_metadata="http://localhost:8000/'
            '.well-known/oauth-protected-resource/mcp"'
        )

    def test_remote_missing_auth_challenge_excludes_optional_catalog_scopes(self):
        app = self.create_scoped_app(
            required_scopes=["read"],
            scopes_supported=["read", "admin"],
        )

        with TestClient(app) as client:
            response = client.post("/mcp")
            metadata = client.get("/.well-known/oauth-protected-resource/mcp").json()

        assert response.status_code == 401
        assert 'scope="read"' in response.headers["www-authenticate"]
        assert "admin" not in response.headers["www-authenticate"]
        assert metadata["scopes_supported"] == ["read", "admin"]

    def test_remote_insufficient_scope_challenge_excludes_optional_catalog_scopes(
        self,
    ):
        app = self.create_scoped_app(
            required_scopes=["read"],
            scopes_supported=["read", "admin"],
        )

        with TestClient(app) as client:
            response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})

        assert response.status_code == 403
        assert 'scope="read"' in response.headers["www-authenticate"]
        assert "admin" not in response.headers["www-authenticate"]

    def test_oauth_missing_auth_challenge_excludes_optional_scopes(self):
        app = self.create_oauth_app()

        with TestClient(app) as client:
            response = client.post("/mcp")
            metadata = client.get("/.well-known/oauth-protected-resource/mcp").json()

        assert response.status_code == 401
        assert 'scope="openid"' in response.headers["www-authenticate"]
        assert "email" not in response.headers["www-authenticate"]
        assert "calendar" not in response.headers["www-authenticate"]
        assert metadata["scopes_supported"] == ["openid", "email", "calendar"]

    def test_oauth_insufficient_scope_challenge_excludes_optional_scopes(self):
        app = self.create_oauth_app()

        with TestClient(app) as client:
            response = client.post("/mcp", headers={"Authorization": "Bearer narrow"})

        assert response.status_code == 403
        assert 'scope="openid"' in response.headers["www-authenticate"]
        assert "email" not in response.headers["www-authenticate"]
        assert "calendar" not in response.headers["www-authenticate"]

    def test_invalid_token_enhanced_error_message(self, jwt_verifier):
        """Test that invalid_token errors have enhanced error messages."""
        server = FastMCP("Test Server")

        @server.tool
        def test_tool() -> str:
            return "test"

        app = create_streamable_http_app(
            server=server,
            streamable_http_path="/mcp",
            auth=jwt_verifier,
        )

        with TestClient(app) as client:
            # Request WITH an invalid Authorization header
            response = client.post(
                "/mcp", headers={"Authorization": "Bearer invalid-token"}
            )

            assert response.status_code == 401
            assert "www-authenticate" in response.headers

            # Check enhanced error message
            data = response.json()
            assert data["error"] == "invalid_token"
            # Should have enhanced description with resolution steps
            assert "clear authentication tokens" in data["error_description"]
            assert "automatically re-register" in data["error_description"]

    def test_invalid_token_www_authenticate_header_format(self, jwt_verifier):
        """Test that invalid token WWW-Authenticate header includes error attribute."""
        server = FastMCP("Test Server")
        app = create_streamable_http_app(
            server=server,
            streamable_http_path="/mcp",
            auth=jwt_verifier,
        )

        with TestClient(app) as client:
            # Request WITH an invalid token
            response = client.post(
                "/mcp", headers={"Authorization": "Bearer invalid-token"}
            )

            assert response.status_code == 401
            www_auth = response.headers["www-authenticate"]

            # Should follow Bearer challenge format with error
            assert www_auth.startswith("Bearer ")
            assert 'error="invalid_token"' in www_auth
            assert "error_description=" in www_auth

    def test_insufficient_scope_not_enhanced(self, rsa_key_pair):
        """Test that insufficient_scope errors are not modified."""
        # Create a valid token with wrong scopes
        jwt_verifier = JWTVerifier(
            public_key=rsa_key_pair.public_key,
            issuer="https://test.com",
            audience="https://test.com",
            base_url="https://test.com",
        )

        server = FastMCP("Test Server")

        @server.tool
        def test_tool() -> str:
            return "test"

        app = create_streamable_http_app(
            server=server,
            streamable_http_path="/mcp",
            auth=jwt_verifier,
        )

        # Note: Testing insufficient_scope would require mocking the verifier
        # to return a token with wrong scopes. For now, we verify the middleware
        # is properly in place by checking it rejects unauthenticated requests.
        with TestClient(app) as client:
            response = client.post("/mcp")
            # Without a valid token, we get invalid_token
            assert response.status_code == 401


class TestContentNegotiation:
    """Tests for content negotiation in error responses."""

    @pytest.fixture
    def oauth_proxy(self):
        """Create OAuth proxy for testing."""
        return OAuthProxy(
            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
            upstream_token_endpoint="https://github.com/login/oauth/access_token",
            upstream_client_id="test-client-id",
            upstream_client_secret="test-client-secret",
            token_verifier=JWTVerifier(
                public_key=RSAKeyPair.generate().public_key,
                issuer="https://test.com",
                audience="https://test.com",
                base_url="https://test.com",
            ),
            base_url="https://myserver.com",
            jwt_signing_key="test-secret",
            client_storage=MemoryStore(),
        )

    def test_html_preferred_when_both_accepted(self, oauth_proxy):
        """Test that HTML is preferred when both text/html and application/json are accepted."""
        app = Starlette(routes=oauth_proxy.get_routes())

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                },
                headers={"Accept": "text/html,application/json"},
            )

            # Should prefer HTML
            assert response.status_code == 400
            assert "text/html" in response.headers["content-type"]

    def test_json_when_only_json_accepted(self, oauth_proxy):
        """Test that JSON is returned when only application/json is accepted."""
        app = Starlette(routes=oauth_proxy.get_routes())

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                },
                headers={"Accept": "application/json"},
            )

            assert response.status_code == 400
            assert "application/json" in response.headers["content-type"]

    def test_json_when_no_accept_header(self, oauth_proxy):
        """Test that JSON is returned when no Accept header is provided."""
        app = Starlette(routes=oauth_proxy.get_routes())

        with TestClient(app) as client:
            response = client.get(
                "/authorize",
                params={
                    "client_id": "unregistered-client-id",
                    "redirect_uri": "http://localhost:12345/callback",
                    "response_type": "code",
                    "code_challenge": "test-challenge",
                },
            )

            # Without Accept header, should return JSON (API default)
            assert response.status_code == 400
            assert "application/json" in response.headers["content-type"]