diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py
index 50dee0462..2d7dd6472 100644
--- a/src/fastmcp/server/auth/oauth_proxy.py
+++ b/src/fastmcp/server/auth/oauth_proxy.py
@@ -375,6 +375,96 @@ def create_consent_html(
)
+def create_error_html(
+ error_title: str,
+ error_message: str,
+ error_details: dict[str, str] | None = None,
+ server_name: str | None = None,
+ server_icon_url: str | None = None,
+) -> str:
+ """Create a styled HTML error page for OAuth errors.
+
+ Args:
+ error_title: The error title (e.g., "OAuth Error", "Authorization Failed")
+ error_message: The main error message to display
+ error_details: Optional dictionary of error details to show (e.g., {"Error Code": "invalid_client"})
+ server_name: Optional server name to display
+ server_icon_url: Optional URL to server icon/logo
+
+ Returns:
+ Complete HTML page as a string
+ """
+ import html as html_module
+
+ error_message_escaped = html_module.escape(error_message)
+
+ # Build error message box
+ error_box = f"""
+
+
{error_message_escaped}
+
+
{html_module.escape(label)}:
+
{html_module.escape(value)}
+
+ Error Details
+
+ {detail_rows_html}
+
+
+ """
+
+ # Build the page content
+ content = f"""
+
+ {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
+
{html_module.escape(error_title)}
+ {error_box}
+ {details_section}
+
+ """
+
+ # Additional styles needed for this page
+ # Override .info-box.error to use normal text color instead of red
+ additional_styles = (
+ INFO_BOX_STYLES
+ + DETAILS_STYLES
+ + DETAIL_BOX_STYLES
+ + """
+ .info-box.error {
+ color: #111827;
+ }
+ """
+ )
+
+ # Simple CSP policy for error pages (no forms needed)
+ csp_policy = (
+ "default-src 'none'; style-src 'unsafe-inline'; img-src https:; base-uri 'none'"
+ )
+
+ return create_page(
+ content=content,
+ title=error_title,
+ additional_styles=additional_styles,
+ csp_policy=csp_policy,
+ )
+
+
# -------------------------------------------------------------------------
# Handler Classes
# -------------------------------------------------------------------------
@@ -1569,7 +1659,9 @@ class OAuthProxy(OAuthProvider):
# IdP Callback Forwarding
# -------------------------------------------------------------------------
- async def _handle_idp_callback(self, request: Request) -> RedirectResponse:
+ async def _handle_idp_callback(
+ self, request: Request
+ ) -> HTMLResponse | RedirectResponse:
"""Handle callback from upstream IdP and forward to client.
This implements the DCR-compliant callback forwarding:
@@ -1584,32 +1676,37 @@ class OAuthProxy(OAuthProvider):
error = request.query_params.get("error")
if error:
+ error_description = request.query_params.get("error_description")
logger.error(
"IdP callback error: %s - %s",
error,
- request.query_params.get("error_description"),
+ error_description,
)
- # TODO: Forward error to client callback
- return RedirectResponse(
- url=f"data:text/html,OAuth Error
{error}: {request.query_params.get('error_description', 'Unknown error')}
",
- status_code=302,
+ # Show error page to user
+ html_content = create_error_html(
+ error_title="OAuth Error",
+ error_message=f"Authentication failed: {error_description or 'Unknown error'}",
+ error_details={"Error Code": error} if error else None,
)
+ return HTMLResponse(content=html_content, status_code=400)
if not idp_code or not txn_id:
logger.error("IdP callback missing code or transaction ID")
- return RedirectResponse(
- url="data:text/html,OAuth Error
Missing authorization code or transaction ID
",
- status_code=302,
+ html_content = create_error_html(
+ error_title="OAuth Error",
+ error_message="Missing authorization code or transaction ID from the identity provider.",
)
+ return HTMLResponse(content=html_content, status_code=400)
# Look up transaction data
transaction_model = await self._transaction_store.get(key=txn_id)
if not transaction_model:
logger.error("IdP callback with invalid transaction ID: %s", txn_id)
- return RedirectResponse(
- url="data:text/html,OAuth Error
Invalid or expired transaction
",
- status_code=302,
+ html_content = create_error_html(
+ error_title="OAuth Error",
+ error_message="Invalid or expired authorization transaction. Please try authenticating again.",
)
+ return HTMLResponse(content=html_content, status_code=400)
transaction = transaction_model.model_dump()
# Exchange IdP code for tokens (server-side)
@@ -1663,11 +1760,11 @@ class OAuthProxy(OAuthProvider):
except Exception as e:
logger.error("IdP token exchange failed: %s", e)
- # TODO: Forward error to client callback
- return RedirectResponse(
- url=f"data:text/html,OAuth Error
Token exchange failed: {e}
",
- status_code=302,
+ html_content = create_error_html(
+ error_title="OAuth Error",
+ error_message=f"Token exchange with identity provider failed: {e}",
)
+ return HTMLResponse(content=html_content, status_code=500)
# Generate our own authorization code for the client
client_code = secrets.token_urlsafe(32)
@@ -1714,10 +1811,11 @@ class OAuthProxy(OAuthProvider):
except Exception as e:
logger.error("Error in IdP callback handler: %s", e, exc_info=True)
- return RedirectResponse(
- url="data:text/html,OAuth Error
Internal server error during IdP callback
",
- status_code=302,
+ html_content = create_error_html(
+ error_title="OAuth Error",
+ error_message="Internal server error during OAuth callback processing. Please try again.",
)
+ return HTMLResponse(content=html_content, status_code=500)
# -------------------------------------------------------------------------
# Consent Interstitial
diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py
index e3bc09d2d..c4c7140e1 100644
--- a/tests/server/auth/test_oauth_proxy.py
+++ b/tests/server/auth/test_oauth_proxy.py
@@ -1309,3 +1309,103 @@ class TestTokenHandlerErrorTransformation:
# Should pass through unchanged
assert response.status_code == 400
assert b'"error":"invalid_grant"' in response.body
+
+
+class TestErrorPageRendering:
+ """Test error page rendering for OAuth callback errors."""
+
+ def test_create_error_html_basic(self):
+ """Test basic error page generation."""
+ from fastmcp.server.auth.oauth_proxy import create_error_html
+
+ html = create_error_html(
+ error_title="Test Error",
+ error_message="This is a test error message",
+ )
+
+ # Verify it's valid HTML
+ assert "" in html
+ assert "Test Error" in html
+ assert "This is a test error message" in html
+ assert 'class="info-box error"' in html
+
+ def test_create_error_html_with_details(self):
+ """Test error page with error details."""
+ from fastmcp.server.auth.oauth_proxy import create_error_html
+
+ html = create_error_html(
+ error_title="OAuth Error",
+ error_message="Authentication failed",
+ error_details={
+ "Error Code": "invalid_scope",
+ "Description": "Requested scope does not exist",
+ },
+ )
+
+ # Verify error details are included
+ assert "Error Details" in html
+ assert "Error Code" in html
+ assert "invalid_scope" in html
+ assert "Description" in html
+ assert "Requested scope does not exist" in html
+
+ def test_create_error_html_escapes_user_input(self):
+ """Test that error page properly escapes HTML in user input."""
+ from fastmcp.server.auth.oauth_proxy import create_error_html
+
+ html = create_error_html(
+ error_title="Error ",
+ error_message="Message with HTML tags",
+ error_details={"Key" not in html
+ assert "<script>" in html
+ assert "HTML" not in html
+ assert "<b>HTML</b>" in html
+
+ async def test_callback_error_returns_html_page(self):
+ """Test that OAuth callback errors return styled HTML instead of data: URLs."""
+ from unittest.mock import Mock
+
+ from starlette.requests import Request
+ from starlette.responses import HTMLResponse
+
+ from fastmcp.server.auth.oauth_proxy import OAuthProxy
+ from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+ # Create a minimal OAuth proxy
+ provider = OAuthProxy(
+ upstream_authorization_endpoint="https://idp.example.com/authorize",
+ upstream_token_endpoint="https://idp.example.com/token",
+ upstream_client_id="test-client",
+ upstream_client_secret="test-secret",
+ token_verifier=JWTVerifier(
+ jwks_uri="https://idp.example.com/.well-known/jwks.json",
+ issuer="https://idp.example.com",
+ audience="test-client",
+ ),
+ base_url="http://localhost:8000",
+ jwt_signing_key="test-signing-key",
+ )
+
+ # Mock a request with an error from the IdP
+ mock_request = Mock(spec=Request)
+ mock_request.query_params = {
+ "error": "invalid_scope",
+ "error_description": "The application asked for scope 'read' that doesn't exist",
+ "state": "test-state",
+ }
+
+ # Call the callback handler
+ response = await provider._handle_idp_callback(mock_request)
+
+ # Verify we get an HTMLResponse, not a RedirectResponse
+ assert isinstance(response, HTMLResponse)
+ assert response.status_code == 400
+
+ # Verify the response contains the error message
+ assert b"invalid_scope" in response.body
+ assert b"doesn't exist" in response.body # HTML-escaped apostrophe
+ assert b"OAuth Error" in response.body