ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

[codex] Add OAuthProxy RFC 9207 issuer responses (#4438)

Browse source 
* Add OAuthProxy issuer response parameter

* Cover OAuthProxy issuer error redirects

* Relax host origin guard defaults (#4439)

* Use exact issuer in authorize errors

* Restore HTTP host guard compatibility (#4472)

* Hugging Face Auth Integration (#4385)

Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>

* Docs: add v3.4.4 changelog entries (#4473)

* Explain unnormalized issuer; cover consent-denial path base_url

* Revert "Merge remote-tracking branch 'origin/release/3.x' into codex/oauth-proxy-rfc9207-issuer"

This reverts commit 9e34b1686c, reversing
changes made to 640dc60fe0.

* Preserve callback query bytes when appending iss/code/state params

add_query_params previously decoded the existing query with parse_qsl
and re-encoded it, mutating opaque or signed query strings (a valueless
?flag became ?flag=, non-UTF-8 percent-encoded bytes got replaced).
Append the newly-encoded params to the existing query string instead of
round-tripping it through parse/encode.

Also fixes a stray bare `httpx` reference in a test that should use
httpx2 following the SDK v2 migration.

* Attach RFC 9207 iss to authorize() success redirects too

AuthorizationHandler only added iss to error redirects from the SDK's
base handler, not to code redirects returned directly by authorize()
overrides that bypass consent/upstream (as GitHub's mocked test does).
Since metadata now unconditionally advertises
authorization_response_iss_parameter_supported, any client-facing
redirect missing iss hard-fails RFC 9207-aware clients.

Also fixes HeadlessOAuth, which parsed code/state from the redirect
but silently dropped iss, so the same regression would have masked
itself across every other provider integration test too.

* Carry RFC 9207 iss through the production OAuth callback path

OAuthProxy advertises authorization_response_iss_parameter_supported and
sends iss on every authorization redirect, but the client's production
callback chain (CallbackResponse -> OAuthCallbackResult -> OAuth.callback_handler)
had no iss field, so it was silently dropped and the SDK's
validate_authorization_response_iss rejected the callback. HeadlessOAuth
already carried iss through, which is why CI stayed green while real
clients failed.

Add iss to CallbackResponse and OAuthCallbackResult, thread it through
store_result_once for both success and error branches, and pass it into
AuthorizationCodeResult in OAuth.callback_handler.

* Don't duplicate iss when a provider redirect already carries one

* Consolidate RFC 9207 iss handling into a single redirect helper

Every client-facing authorization redirect must carry exactly one iss.
That invariant was being enforced by hand at five separate call sites,
each building its own params dict -- which is how the success-redirect
path shipped without iss in the first place, and how a registered
redirect_uri that already carries its own iss could end up duplicated.
Route all five sites through build_client_redirect(), which owns the
idempotent replace-or-append behavior so no caller can get it wrong.

---------

Co-authored-by: shaun smith <1936278+evalstate@users.noreply.github.com>
This commit is contained in:
Jeremiah Lowin 2026-07-19 09:52:43 -04:00 • committed by GitHub
commit 67e8448389
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1399 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

1
fastmcp_slim/fastmcp/client/auth/oauth.py
View file

@ -415,6 +415,7 @@ class OAuth(OAuthClientProvider):
return AuthorizationCodeResult(
code=result.code, # type: ignore[arg-type] # ty:ignore[invalid-argument-type]
state=result.state,
iss=result.iss,
)
except TimeoutError as e:
raise TimeoutError(

31
fastmcp_slim/fastmcp/client/oauth_callback.py
View file

@ -82,6 +82,11 @@ class CallbackResponse:
state: str | None = None
error: str | None = None
error_description: str | None = None
# RFC 9207: the authorization server's issuer identifier, sent on both
# success and error redirects once OAuthProxy advertises
# `authorization_response_iss_parameter_supported`. Must be captured
# here or `from_dict`'s annotation filter silently drops it.
iss: str | None = None
@classmethod
def from_dict(cls, data: dict[str, str]) -> CallbackResponse:
@ -98,6 +103,12 @@ class OAuthCallbackResult:
code: str | None = None
state: str | None = None
error: Exception | None = None
# RFC 9207 issuer identifier, captured on both success and error
# callbacks. The MCP SDK's `validate_authorization_response_iss` only
# consumes this on the success path (via `AuthorizationCodeResult.iss`),
# but it is stored unconditionally here so the error path never silently
# drops it either.
iss: str | None = None
def create_oauth_callback_server(
@ -127,6 +138,7 @@ def create_oauth_callback_server(
code: str | None = None,
state: str | None = None,
error: Exception | None = None,
iss: str | None = None,
) -> None:
"""Store the first callback result and ignore subsequent requests."""
if result_container is None or result_ready is None or result_ready.is_set():
@ -135,6 +147,7 @@ def create_oauth_callback_server(
result_container.code = code
result_container.state = state
result_container.error = error
result_container.iss = iss
result_ready.set()
async def callback_handler(request: Request):
@ -151,8 +164,13 @@ def create_oauth_callback_server(
else:
user_message = f"Authorization failed: {error_desc}"
# Store error and signal completion if result tracking provided
store_result_once(error=RuntimeError(user_message))
# Store error and signal completion if result tracking provided.
# RFC 9207: `iss` is captured here too, even though the callback
# ultimately raises instead of returning a result, so it isn't
# silently dropped for callers that want to inspect it.
store_result_once(
error=RuntimeError(user_message), iss=callback_response.iss
)
return create_secure_html_response(
create_callback_html(
@ -166,7 +184,9 @@ def create_oauth_callback_server(
user_message = "No authorization code was received from the server."
# Store error and signal completion if result tracking provided
store_result_once(error=RuntimeError(user_message))
store_result_once(
error=RuntimeError(user_message), iss=callback_response.iss
)
return create_secure_html_response(
create_callback_html(
@ -183,7 +203,9 @@ def create_oauth_callback_server(
)
# Store error and signal completion if result tracking provided
store_result_once(error=RuntimeError(user_message))
store_result_once(
error=RuntimeError(user_message), iss=callback_response.iss
)
return create_secure_html_response(
create_callback_html(
@ -196,6 +218,7 @@ def create_oauth_callback_server(
# Success case - store result and signal completion if result tracking provided
store_result_once(
code=callback_response.code,
iss=callback_response.iss,
state=callback_response.state,
)

31
fastmcp_slim/fastmcp/server/auth/handlers/authorize.py
View file

@ -15,6 +15,7 @@ from __future__ import annotations
import json
from typing import TYPE_CHECKING
from urllib.parse import parse_qs, urlparse
from mcp.server.auth.handlers.authorize import (
AuthorizationHandler as SDKAuthorizationHandler,
@ -23,6 +24,7 @@ from pydantic import AnyHttpUrl
from starlette.requests import Request
from starlette.responses import Response
from fastmcp.server.auth.redirect_validation import build_client_redirect
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.ui import (
INFO_BOX_STYLES,
@ -189,6 +191,10 @@ class AuthorizationHandler(SDKAuthorizationHandler):
server_icon_url: Optional server icon URL for branding
"""
super().__init__(provider)
# Unnormalized on purpose: this must match the discovery document's
# `issuer` field byte-for-byte per RFC 9207, and that field is built
# from the same unmodified base_url (see OAuthProxy.get_routes()).
self._issuer = str(base_url)
self._base_url = str(base_url).rstrip("/")
self._server_name = server_name
self._server_icon_url = server_icon_url
@ -209,6 +215,31 @@ class AuthorizationHandler(SDKAuthorizationHandler):
# Call the SDK handler
response = await super().handle(request)
if 300 <= response.status_code < 400 and "location" in response.headers:
redirect_url = response.headers["location"]
redirect_params = parse_qs(urlparse(redirect_url).query)
# RFC 9207: any client-facing authorization response — success
# (`code`) or error (`error`) — must carry `iss`. The base SDK
# handler's redirect target is normally `/consent` or the
# upstream IdP (neither carries `code`/`error`), but a provider
# can override `authorize()` to redirect straight back to the
# client (e.g. when consent/upstream is skipped entirely), so
# this must not be gated on "error" alone.
if "error" in redirect_params or "code" in redirect_params:
# `build_client_redirect` owns the "set `iss` idempotently"
# invariant: a provider's `authorize()` override (or the
# client's own registered redirect_uri) may already carry an
# `iss` — matching or not — and RFC 6749 §3.1 forbids a
# response parameter from appearing more than once. A
# mismatched existing value is already unusable to a
# spec-compliant client (it validates `iss` against the
# discovery document's `issuer`, i.e. `self._issuer`), so
# the helper corrects it to the canonical value rather than
# leaving it broken or appending a duplicate.
response.headers["location"] = build_client_redirect(
redirect_url, {}, iss=self._issuer
)
# Check if this is a client not found error
if response.status_code == 400:
# Try to extract client_id from request for enhanced error

17
fastmcp_slim/fastmcp/server/auth/oauth_proxy/consent.py
View file

@ -23,7 +23,10 @@ from starlette.responses import HTMLResponse, RedirectResponse
from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
from fastmcp.server.auth.oauth_proxy.ui import create_consent_html
from fastmcp.server.auth.redirect_validation import validate_redirect_uri
from fastmcp.server.auth.redirect_validation import (
build_client_redirect,
validate_redirect_uri,
)
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.ui import create_secure_html_response
@ -388,9 +391,12 @@ class ConsentMixin:
"error": "access_denied",
"state": txn.get("client_state") or "",
}
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
return RedirectResponse(
url=f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}",
url=build_client_redirect(
txn["client_redirect_uri"],
callback_params,
iss=str(self.base_url),
),
status_code=302,
)
else:
@ -563,9 +569,8 @@ class ConsentMixin:
"error": "access_denied",
"state": txn.get("client_state") or "",
}
sep = "&" if "?" in txn["client_redirect_uri"] else "?"
client_callback_url = (
f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}"
client_callback_url = build_client_redirect(
txn["client_redirect_uri"], callback_params, iss=str(self.base_url)
)
response = RedirectResponse(url=client_callback_url, status_code=302)

30
fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
View file

@ -26,7 +26,6 @@ from collections import OrderedDict
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, Literal
from urllib.parse import urlencode
import anyio
import httpx2
@ -103,7 +102,10 @@ from fastmcp.server.auth.oauth_proxy.models import (
)
from fastmcp.server.auth.oauth_proxy.ui import create_error_html
from fastmcp.server.auth.oauth_proxy.upstream import AsyncOAuth2Client
from fastmcp.server.auth.redirect_validation import validate_redirect_uri
from fastmcp.server.auth.redirect_validation import (
build_client_redirect,
validate_redirect_uri,
)
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
@ -2291,10 +2293,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
methods=["POST", "OPTIONS"],
)
)
elif (
(self._cimd_manager is not None or self._identity_assertion is not None)
and isinstance(route, Route)
and route.path.startswith("/.well-known/oauth-authorization-server")
elif isinstance(route, Route) and route.path.startswith(
"/.well-known/oauth-authorization-server"
):
client_registration_options = (
self.client_registration_options or ClientRegistrationOptions()
@ -2307,6 +2307,11 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
revocation_options,
supports_identity_assertion=self._identity_assertion is not None,
)
# RFC 9207: every authorization response we issue carries an
# `iss` matching this issuer byte-for-byte, so this route must
# always be overridden to advertise support — not just when
# CIMD or identity assertion is also enabled.
metadata.authorization_response_iss_parameter_supported = True
if self._cimd_manager is not None:
metadata.client_id_metadata_document_supported = True
existing = metadata.token_endpoint_auth_methods_supported or []
@ -2423,9 +2428,12 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
}
if error_description:
error_params["error_description"] = error_description
separator = "&" if "?" in client_redirect_uri else "?"
return RedirectResponse(
url=f"{client_redirect_uri}{separator}{urlencode(error_params)}",
url=build_client_redirect(
client_redirect_uri,
error_params,
iss=str(self.base_url),
),
status_code=302,
)
# No trusted redirect_uri available — show local error page
@ -2591,10 +2599,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
"state": client_state,
}
# Add query parameters to client redirect URI
separator = "&" if "?" in client_redirect_uri else "?"
client_callback_url = (
f"{client_redirect_uri}{separator}{urlencode(callback_params)}"
client_callback_url = build_client_redirect(
client_redirect_uri, callback_params, iss=str(self.base_url)
)
logger.debug(f"Forwarding to client callback for transaction {txn_id}")

95
fastmcp_slim/fastmcp/server/auth/redirect_validation.py
View file

@ -5,7 +5,7 @@ protecting against userinfo-based bypass attacks like http://localhost@evil.com.
"""
import fnmatch
from urllib.parse import unquote, urlparse
from urllib.parse import unquote, urlencode, urlparse, urlunparse
from pydantic import AnyUrl
@ -19,6 +19,99 @@ UNSAFE_REDIRECT_URI_SCHEMES = frozenset(
)
def add_query_params(url: str, params: dict[str, str]) -> str:
"""Append query parameters to a URL while preserving existing parameters.
The existing query string is appended to verbatim rather than decoded
and re-serialized, since registered redirect URIs may carry opaque or
signed query strings whose exact bytes matter to the receiving client
(for example, a valueless `?flag` must not become `?flag=`, and
non-UTF-8 percent-encoded sequences must not be replaced).
"""
parsed = urlparse(url)
new_query = urlencode(params)
query = f"{parsed.query}&{new_query}" if parsed.query else new_query
return urlunparse(parsed._replace(query=query))
def replace_query_param(url: str, key: str, value: str) -> str:
"""Replace the first occurrence of `key` in a URL's query string in place.
Like `add_query_params`, this does not round-trip the query through
`parse_qsl`/`urlencode`: every segment other than the matched one is
passed through byte-for-byte, so opaque or non-UTF-8 percent-encoded
values elsewhere in the query are left untouched. Only the matched
segment's encoding is replaced (with `key=value`, freshly
`urlencode`d). If `key` is not present, it is appended, matching
`add_query_params`'s behavior.
"""
parsed = urlparse(url)
segments = parsed.query.split("&") if parsed.query else []
new_segment = urlencode({key: value})
replaced = False
new_segments: list[str] = []
for segment in segments:
segment_key = segment.split("=", 1)[0]
if not replaced and unquote(segment_key) == key:
new_segments.append(new_segment)
replaced = True
else:
new_segments.append(segment)
if not replaced:
new_segments.append(new_segment)
return urlunparse(parsed._replace(query="&".join(new_segments)))
def build_client_redirect(url: str, params: dict[str, str], *, iss: str) -> str:
"""Build a client-facing authorization redirect that carries exactly one `iss`.
Every redirect this server sends back to an OAuth client from the
authorization endpoint -- success (carrying `code`) or error (carrying
`error`) -- must carry the proxy's RFC 9207 issuer exactly once (RFC
6749 §3.1 forbids a response parameter from appearing more than once).
A registered redirect_uri can legitimately carry its own `iss` query
parameter already (e.g. `https://client.example/callback?iss=tenant`),
so blindly appending the server's issuer on top of that would duplicate
it -- this is what every client-facing redirect site must get right,
and the reason this helper exists instead of five call sites each
reimplementing the same invariant by hand.
`params` is appended to `url` via `add_query_params` (verbatim, without
re-encoding the existing query -- see that function's docstring), and
`iss` is then set idempotently via `replace_query_param`: an existing
occurrence -- whether contributed by the registered redirect_uri or
already present in `url` -- is overwritten with the canonical value;
otherwise `iss` is appended.
`iss` is keyword-only and required so a caller cannot forget to pass
it. `params` must not itself contain an `"iss"` key -- pass it via the
`iss` keyword instead, so there is exactly one place the value can come
from.
Args:
url: The redirect target -- normally the client's registered
redirect_uri.
params: The response parameters to append (e.g. `code`/`state`, or
`error`/`error_description`). Must not include `"iss"`.
iss: The canonical RFC 9207 issuer, byte-for-byte equal to the
discovery document's `issuer` (`str(self.base_url)` /
`self._issuer` -- never the rstripped `self._base_url`).
Returns:
`url` with `params` appended and exactly one `iss` query parameter
set to `iss`.
"""
if "iss" in params:
raise ValueError(
"params must not include 'iss' -- pass it via the 'iss' keyword"
)
if params:
url = add_query_params(url, params)
return replace_query_param(url, "iss", iss)
def _parse_host_port(netloc: str) -> tuple[str | None, str | None]:
"""Parse host and port from netloc, handling wildcards.

3
fastmcp_slim/fastmcp/utilities/tests.py
View file

@ -270,6 +270,7 @@ class HeadlessOAuth(OAuth):
auth_code = query_params["code"][0]
state = query_params.get("state", [None])[0]
return AuthorizationCodeResult(code=auth_code, state=state)
iss = query_params.get("iss", [None])[0]
return AuthorizationCodeResult(code=auth_code, state=state, iss=iss)
else:
raise RuntimeError(f"Authorization failed: {response.status_code}")

39
tests/client/auth/test_oauth_client.py
View file

@ -3,6 +3,7 @@ import time
from unittest.mock import patch
from urllib.parse import urlparse
import anyio
import httpx2
import pytest
from key_value.aio.stores.memory import MemoryStore
@ -177,6 +178,44 @@ async def test_expired_dynamic_registration_is_retried():
assert provider.registration_count == 2
async def test_oauth_callback_handler_propagates_iss_to_authorization_code_result():
"""RFC 9207: `OAuth.callback_handler()` (the production, non-headless path)
must carry `iss` from the callback query string all the way into the
`AuthorizationCodeResult` handed back to the MCP SDK.
The MCP SDK's `validate_authorization_response_iss` raises when the
authorization server metadata advertises
`authorization_response_iss_parameter_supported` and the result it
receives has no `iss` -- so if this hop drops it, every production OAuth
login against an RFC 9207-compliant server (like OAuthProxy) fails, even
though the server sent `iss` correctly. `HeadlessOAuth` already carries
`iss` through for tests -- this test exercises the real `OAuth` class
that production clients actually use.
"""
oauth = OAuth(mcp_url="http://127.0.0.1:9999")
async def send_callback():
await anyio.sleep(0.1)
async with httpx2.AsyncClient() as client:
response = await client.get(
f"http://{oauth._callback_host}:{oauth.redirect_port}/callback",
params={
"code": "auth-code-123",
"state": "state-xyz",
"iss": "https://issuer.example.com",
},
)
assert response.status_code == 200
async with anyio.create_task_group() as tg:
tg.start_soon(send_callback)
result = await oauth.callback_handler()
assert result.code == "auth-code-123"
assert result.state == "state-xyz"
assert result.iss == "https://issuer.example.com"
class TestOAuthClientUrlHandling:
"""Tests for OAuth client URL handling (issue #2573)."""

83
tests/client/test_oauth_callback_race.py
View file

@ -48,3 +48,86 @@ def test_oauth_callback_server_uses_configured_host():
server = create_oauth_callback_server(port=find_available_port(), host="localhost")
assert server.config.host == "localhost"
async def test_oauth_callback_result_captures_iss():
"""RFC 9207: the `iss` query parameter must survive from the raw callback
request through to `OAuthCallbackResult`, the same as `code` and `state`.
OAuthProxy advertises `authorization_response_iss_parameter_supported` and
includes `iss` on every authorization redirect. If the callback server's
query-parsing chain (CallbackResponse.from_dict -> store_result_once ->
OAuthCallbackResult) drops it, the MCP SDK's `validate_authorization_response_iss`
rejects an otherwise-successful callback.
"""
port = find_available_port()
result = OAuthCallbackResult()
result_ready = anyio.Event()
server = create_oauth_callback_server(
port=port,
result_container=result,
result_ready=result_ready,
)
async with anyio.create_task_group() as tg:
tg.start_soon(server.serve)
await anyio.sleep(0.05)
async with httpx2.AsyncClient() as client:
response = await client.get(
f"http://127.0.0.1:{port}/callback",
params={
"code": "good",
"state": "s1",
"iss": "https://issuer.example.com",
},
)
assert response.status_code == 200
await result_ready.wait()
assert result.error is None
assert result.code == "good"
assert result.state == "s1"
assert result.iss == "https://issuer.example.com"
tg.cancel_scope.cancel()
async def test_oauth_callback_result_captures_iss_on_error():
"""RFC 9207 applies to error redirects too -- the server emits `iss` on
them, so the callback server must not silently drop it while building the
error result.
"""
port = find_available_port()
result = OAuthCallbackResult()
result_ready = anyio.Event()
server = create_oauth_callback_server(
port=port,
result_container=result,
result_ready=result_ready,
)
async with anyio.create_task_group() as tg:
tg.start_soon(server.serve)
await anyio.sleep(0.05)
async with httpx2.AsyncClient() as client:
response = await client.get(
f"http://127.0.0.1:{port}/callback",
params={
"error": "access_denied",
"state": "s1",
"iss": "https://issuer.example.com",
},
)
assert response.status_code == 400
await result_ready.wait()
assert result.error is not None
assert result.iss == "https://issuer.example.com"
tg.cancel_scope.cancel()

180
tests/server/auth/oauth_proxy/test_oauth_proxy.py
View file

@ -1,11 +1,14 @@
"""Tests for OAuth proxy initialization and configuration."""
import time
from unittest.mock import AsyncMock, patch
from urllib.parse import parse_qs, urlparse
import httpx2
import pytest
from key_value.aio.stores.memory import MemoryStore
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from starlette.applications import Starlette
from fastmcp.server.auth.oauth_proxy import OAuthProxy
@ -238,6 +241,34 @@ class TestOAuthProxyInitialization:
"none",
}
async def test_metadata_advertises_authorization_response_issuer_parameter(
self, jwt_verifier
):
"""OAuth metadata should advertise RFC 9207 authorization response issuers."""
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="client-123",
upstream_client_secret="secret-456",
token_verifier=jwt_verifier,
base_url="https://api.example.com",
jwt_signing_key="test-secret",
client_storage=MemoryStore(),
)
app = Starlette(routes=proxy.get_routes())
transport = httpx2.ASGITransport(app=app)
async with httpx2.AsyncClient(
transport=transport, base_url="https://api.example.com"
) as client:
response = await client.get("/.well-known/oauth-authorization-server")
assert response.status_code == 200
metadata = response.json()
assert metadata["issuer"] == "https://api.example.com/"
assert metadata["authorization_response_iss_parameter_supported"] is True
class TestOptionalClientSecret:
"""Tests for OAuthProxy without upstream_client_secret."""
@ -360,6 +391,61 @@ class TestIdpCallbackErrorForwarding:
assert params["error"] == ["access_denied"]
assert params["error_description"] == ["User denied access"]
assert params["state"] == [client_state]
assert params["iss"] == ["https://myserver.com/"]
async def test_error_redirect_does_not_duplicate_iss_already_in_redirect_uri(
self, oauth_proxy
):
"""RFC 9207 P2 regression: a registered redirect_uri may already
carry its own `iss` query parameter (e.g. a multi-tenant client
encoding its tenant in the callback URL). Forwarding an IdP error
must not append a second `iss` on top of it -- RFC 6749 §3.1
forbids a response parameter appearing more than once -- and every
other query byte on the registered URI (a valueless `flag` and a
non-UTF-8 percent-encoded `sig`) must survive untouched.
"""
txn_id = "test-txn-dup-iss"
client_redirect_uri = (
"http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
)
client_state = "client-state-abc"
transaction = OAuthTransaction(
txn_id=txn_id,
client_id="test-client",
client_redirect_uri=client_redirect_uri,
client_state=client_state,
code_challenge=None,
code_challenge_method="S256",
scopes=["read"],
created_at=time.time(),
)
await oauth_proxy._transaction_store.put(key=txn_id, value=transaction)
app = Starlette(routes=oauth_proxy.get_routes())
transport = httpx2.ASGITransport(app=app)
async with httpx2.AsyncClient(
transport=transport,
base_url="https://myserver.com",
follow_redirects=False,
) as client:
response = await client.get(
f"/auth/callback?error=access_denied&state={txn_id}"
)
assert response.status_code == 302
location = response.headers["location"]
query = urlparse(location).query
params = parse_qs(query)
# Exactly one `iss`, corrected to the canonical value -- a
# duplicate would make this list have length 2.
assert params["iss"] == ["https://myserver.com/"]
# Other query bytes from the registered redirect_uri survive
# byte-for-byte.
assert "flag" in query
assert "sig=%FF%FE" in query
async def test_error_with_unsafe_transaction_redirect_returns_html_error(
self, oauth_proxy
@ -412,3 +498,97 @@ class TestIdpCallbackErrorForwarding:
)
assert response.status_code == 400
class TestIdpCallbackSuccessForwarding:
"""Tests for the success (`code`) path in the IdP callback."""
async def test_success_redirect_does_not_duplicate_iss_already_in_redirect_uri(
self, jwt_verifier
):
"""RFC 9207 P2 regression at the success-redirect call site: a
registered redirect_uri already carrying `iss` must end up with
exactly one `iss` (the canonical value) after the proxy forwards
the exchanged authorization code, and every other query byte on the
registered URI must survive untouched.
"""
# Consent is disabled here because this test exercises callback
# forwarding, not the consent-binding-cookie check that the
# standard consent flow additionally requires.
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=jwt_verifier,
base_url="https://myserver.com",
redirect_path="/auth/callback",
jwt_signing_key="test-secret",
client_storage=MemoryStore(),
require_authorization_consent=False,
)
client_id = "success-dup-iss-client"
client_redirect_uri = (
"http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
)
client_info = OAuthClientInformationFull(
client_id=client_id,
client_secret="test-secret",
redirect_uris=[AnyUrl(client_redirect_uri)],
)
await oauth_proxy.register_client(client_info)
txn_id = "test-txn-success-dup-iss"
transaction = OAuthTransaction(
txn_id=txn_id,
client_id=client_id,
client_redirect_uri=client_redirect_uri,
client_state="client-state-success",
code_challenge=None,
code_challenge_method="S256",
scopes=["read"],
created_at=time.time(),
)
await oauth_proxy._transaction_store.put(key=txn_id, value=transaction)
app = Starlette(routes=oauth_proxy.get_routes())
transport = httpx2.ASGITransport(app=app)
with patch(
"fastmcp.server.auth.oauth_proxy.proxy.AsyncOAuth2Client"
) as MockClient:
mock_client = AsyncMock()
mock_client.fetch_token = AsyncMock(
return_value={
"access_token": "upstream-access-token",
"refresh_token": "upstream-refresh-token",
"expires_in": 3600,
"token_type": "Bearer",
}
)
MockClient.return_value = mock_client
async with httpx2.AsyncClient(
transport=transport,
base_url="https://myserver.com",
follow_redirects=False,
) as client:
response = await client.get(
f"/auth/callback?code=idp-authorization-code&state={txn_id}"
)
assert response.status_code == 302
location = response.headers["location"]
query = urlparse(location).query
params = parse_qs(query)
assert "code" in params
assert params["state"] == ["client-state-success"]
# Exactly one `iss`, corrected to the canonical value -- a
# duplicate would make this list have length 2.
assert params["iss"] == ["https://myserver.com/"]
# Other query bytes from the registered redirect_uri survive
# byte-for-byte.
assert "flag" in query
assert "sig=%FF%FE" in query

55
tests/server/auth/oauth_proxy/test_tokens.py
View file

@ -3,6 +3,7 @@
import logging
import time
from unittest.mock import AsyncMock, Mock, patch
from urllib.parse import parse_qs, urlparse
import pytest
from key_value.aio.stores.memory import MemoryStore
@ -249,6 +250,60 @@ class TestOAuthProxyTokenEndpointAuth:
mock_client.fetch_token.assert_awaited_once()
mock_client.aclose.assert_awaited_once()
async def test_callback_redirect_includes_proxy_issuer(self, jwt_verifier):
proxy = OAuthProxy(
upstream_authorization_endpoint="https://oauth.example.com/authorize",
upstream_token_endpoint="https://oauth.example.com/token",
upstream_client_id="client-id",
upstream_client_secret="client-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
require_authorization_consent=False,
jwt_signing_key="test-secret",
client_storage=MemoryStore(),
)
await proxy._transaction_store.put(
key="txn-id",
value=OAuthTransaction(
txn_id="txn-id",
client_id="test-client",
client_redirect_uri="http://localhost:12345/callback",
client_state="client-state",
code_challenge="",
code_challenge_method="S256",
scopes=["read"],
created_at=time.time(),
),
)
mock_request = Mock()
mock_request.query_params = {"code": "idp-code", "state": "txn-id"}
mock_request.cookies = {}
mock_client = AsyncMock()
mock_client.fetch_token = AsyncMock(
return_value={
"access_token": "upstream-access-token",
"refresh_token": "upstream-refresh-token",
"expires_in": 3600,
"token_type": "Bearer",
}
)
with patch.object(
proxy, "_create_upstream_oauth_client", return_value=mock_client
):
response = await proxy._handle_idp_callback(mock_request)
assert response.status_code == 302
location = response.headers["location"]
query_params = parse_qs(urlparse(location).query)
assert "code" in query_params
assert query_params["state"] == ["client-state"]
assert query_params["iss"] == ["https://proxy.example.com/"]
mock_client.aclose.assert_awaited_once()
async def test_callback_rejects_unsafe_transaction_redirect(self, jwt_verifier):
proxy = OAuthProxy(
upstream_authorization_endpoint="https://oauth.example.com/authorize",

452
tests/server/auth/test_enhanced_error_responses.py
View file

@ -7,7 +7,11 @@ This test suite covers:
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
@ -18,6 +22,7 @@ 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):
@ -33,6 +38,57 @@ class _UnderScopedOAuthProxy(OAuthProxy):
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."""
@ -44,8 +100,6 @@ class TestEnhancedAuthorizationHandler:
@pytest.fixture
def oauth_proxy(self, rsa_key_pair):
"""Create OAuth proxy for testing."""
from key_value.aio.stores.memory import MemoryStore
return OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
@ -143,8 +197,6 @@ class TestEnhancedAuthorizationHandler:
)
# Need to register synchronously
import asyncio
asyncio.run(oauth_proxy.register_client(client_info))
with TestClient(app) as client:
@ -165,6 +217,388 @@ class TestEnhancedAuthorizationHandler:
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
@ -251,8 +685,6 @@ class TestEnhancedRequireAuthMiddleware:
def test_missing_auth_no_error_attribute(self, jwt_verifier):
"""Test that missing auth returns 401 without error attribute (RFC 6750 §3.1)."""
from fastmcp.server.http import create_streamable_http_app
server = FastMCP("Test Server")
@server.tool
@ -381,8 +813,6 @@ class TestEnhancedRequireAuthMiddleware:
def test_invalid_token_enhanced_error_message(self, jwt_verifier):
"""Test that invalid_token errors have enhanced error messages."""
from fastmcp.server.http import create_streamable_http_app
server = FastMCP("Test Server")
@server.tool
@ -413,8 +843,6 @@ class TestEnhancedRequireAuthMiddleware:
def test_invalid_token_www_authenticate_header_format(self, jwt_verifier):
"""Test that invalid token WWW-Authenticate header includes error attribute."""
from fastmcp.server.http import create_streamable_http_app
server = FastMCP("Test Server")
app = create_streamable_http_app(
server=server,
@ -439,8 +867,6 @@ class TestEnhancedRequireAuthMiddleware:
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
from fastmcp.server.http import create_streamable_http_app
jwt_verifier = JWTVerifier(
public_key=rsa_key_pair.public_key,
issuer="https://test.com",
@ -475,8 +901,6 @@ class TestContentNegotiation:
@pytest.fixture
def oauth_proxy(self):
"""Create OAuth proxy for testing."""
from key_value.aio.stores.memory import MemoryStore
return OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",

189
tests/server/auth/test_oauth_consent_flow.py
View file

@ -111,6 +111,26 @@ def oauth_proxy_https_remember():
)
@pytest.fixture
def oauth_proxy_https_path():
"""OAuthProxy with a path component in base_url (no trailing slash).
Exercises the RFC 9207 issuer consistency across a base_url shape where
naive normalization (e.g. force-appending a trailing slash) would produce
an `iss` value that no longer matches the discovery document's `issuer`.
"""
return OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="client-id",
upstream_client_secret="client-secret",
token_verifier=_Verifier(),
base_url="https://myserver.example/oauth",
client_storage=MemoryStore(),
jwt_signing_key="test-secret",
)
async def _start_flow(
proxy: OAuthProxy, client_id: str, redirect: str
) -> tuple[str, str]:
@ -631,11 +651,180 @@ class TestConsentSecurity:
q = parse_qs(parsed.query)
assert q.get("error") == ["access_denied"]
assert q.get("state") == ["client-state-xyz"]
assert q.get("iss") == ["https://myserver.example/"]
# Signed denied cookie should be set
assert "MCP_DENIED_CLIENTS" in ";\n".join(
r.headers.get("set-cookie", "").splitlines()
)
async def test_deny_redirect_does_not_duplicate_iss_already_in_redirect_uri(
self, oauth_proxy_https_remember
):
"""RFC 9207 P2 regression: a registered redirect_uri may already
carry its own `iss` query parameter. Explicit consent denial must
not append a second `iss` on top of it -- RFC 6749 §3.1 forbids a
response parameter appearing more than once -- and every other
query byte on the registered URI (a valueless `flag` and a
non-UTF-8 percent-encoded `sig`) must survive untouched.
"""
client_redirect = "http://localhost:5009/callback?iss=tenant&flag&sig=%FF%FE"
txn_id, _ = await _start_flow(
oauth_proxy_https_remember, "client-dup-iss", client_redirect
)
app = Starlette(routes=oauth_proxy_https_remember.get_routes())
with TestClient(app) as c:
consent = c.get(f"/consent?txn_id={txn_id}")
csrf = _extract_csrf(consent.text)
assert csrf
for k, v in consent.cookies.items():
c.cookies.set(k, v)
r = c.post(
"/consent",
data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
follow_redirects=False,
)
assert r.status_code in (302, 303)
loc = r.headers.get("location", "")
query = urlparse(loc).query
q = parse_qs(query)
assert q.get("error") == ["access_denied"]
# Exactly one `iss`, corrected to the canonical value -- a
# duplicate would make this list have length 2.
assert q.get("iss") == ["https://myserver.example/"]
# Other query bytes from the registered redirect_uri survive
# byte-for-byte.
assert "flag" in query
assert "sig=%FF%FE" in query
async def test_deny_redirect_issuer_matches_path_base_url_metadata(
self, oauth_proxy_https_path
):
"""Consent-denial `iss` must match the discovery document exactly.
Regression test for a base_url with a path and no trailing slash
(`https://myserver.example/oauth`): the metadata `issuer` is the
unmodified base_url, so the denial redirect's `iss` must match it
byte-for-byte rather than force-appending a trailing slash.
"""
client_redirect = "http://localhost:5008/callback"
txn_id, _ = await _start_flow(
oauth_proxy_https_path, "client-path", client_redirect
)
app = Starlette(routes=oauth_proxy_https_path.get_routes())
with TestClient(app) as c:
metadata = c.get("/.well-known/oauth-authorization-server").json()
assert metadata["issuer"] == "https://myserver.example/oauth"
consent = c.get(f"/consent?txn_id={txn_id}")
csrf = _extract_csrf(consent.text)
assert csrf
for k, v in consent.cookies.items():
c.cookies.set(k, v)
r = c.post(
"/consent",
data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
follow_redirects=False,
)
assert r.status_code in (302, 303)
q = parse_qs(urlparse(r.headers.get("location", "")).query)
assert q.get("error") == ["access_denied"]
assert q.get("iss") == [metadata["issuer"]]
async def test_remembered_denial_redirects_with_issuer(
self, oauth_proxy_https_remember
):
"""Remembered consent denial redirects with RFC 9207 issuer."""
client_id = "client-denied"
redirect = "http://localhost:5007/callback"
txn_id, _ = await _start_flow(oauth_proxy_https_remember, client_id, redirect)
app = Starlette(routes=oauth_proxy_https_remember.get_routes())
with TestClient(app) as c:
consent = c.get(f"/consent?txn_id={txn_id}")
csrf = _extract_csrf(consent.text)
assert csrf
for k, v in consent.cookies.items():
c.cookies.set(k, v)
r = c.post(
"/consent",
data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
follow_redirects=False,
)
set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
m = re.search(r"__Host-MCP_DENIED_CLIENTS=([^;]+)", set_cookie)
assert m
denied_cookie = m.group(1)
new_txn, _ = await _start_flow(
oauth_proxy_https_remember, client_id, redirect
)
c.cookies.set("__Host-MCP_DENIED_CLIENTS", denied_cookie)
r2 = c.get(
f"/consent?txn_id={new_txn}",
headers={"Sec-Fetch-Site": "none"},
follow_redirects=False,
)
assert r2.status_code in (302, 303)
loc = r2.headers.get("location", "")
parsed = urlparse(loc)
assert parsed.scheme == "http" and parsed.netloc.startswith("localhost")
q = parse_qs(parsed.query)
assert q.get("error") == ["access_denied"]
assert q.get("state") == ["client-state-xyz"]
assert q.get("iss") == ["https://myserver.example/"]
async def test_remembered_denial_does_not_duplicate_iss_already_in_redirect_uri(
self, oauth_proxy_https_remember
):
"""RFC 9207 P2 regression: the *remembered/silent* denial path is a
separate call site from the explicit deny above, and must
independently avoid duplicating `iss` when the registered
redirect_uri already carries one.
"""
client_id = "client-denied-dup-iss"
redirect = "http://localhost:5010/callback?iss=tenant&flag&sig=%FF%FE"
txn_id, _ = await _start_flow(oauth_proxy_https_remember, client_id, redirect)
app = Starlette(routes=oauth_proxy_https_remember.get_routes())
with TestClient(app) as c:
consent = c.get(f"/consent?txn_id={txn_id}")
csrf = _extract_csrf(consent.text)
assert csrf
for k, v in consent.cookies.items():
c.cookies.set(k, v)
r = c.post(
"/consent",
data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
follow_redirects=False,
)
set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
m = re.search(r"__Host-MCP_DENIED_CLIENTS=([^;]+)", set_cookie)
assert m
denied_cookie = m.group(1)
new_txn, _ = await _start_flow(
oauth_proxy_https_remember, client_id, redirect
)
c.cookies.set("__Host-MCP_DENIED_CLIENTS", denied_cookie)
r2 = c.get(
f"/consent?txn_id={new_txn}",
headers={"Sec-Fetch-Site": "none"},
follow_redirects=False,
)
assert r2.status_code in (302, 303)
loc = r2.headers.get("location", "")
query = urlparse(loc).query
q = parse_qs(query)
assert q.get("error") == ["access_denied"]
assert q.get("state") == ["client-state-xyz"]
# Exactly one `iss`, corrected to the canonical value -- a
# duplicate would make this list have length 2.
assert q.get("iss") == ["https://myserver.example/"]
# Other query bytes from the registered redirect_uri survive
# byte-for-byte.
assert "flag" in query
assert "sig=%FF%FE" in query
async def test_approve_sets_cookie_and_redirects_to_upstream(
self, oauth_proxy_https_remember
):

231
tests/server/auth/test_redirect_validation.py
View file

@ -1,15 +1,246 @@
"""Tests for redirect URI validation in OAuth flows."""
import re
from pathlib import Path
from urllib.parse import parse_qs, parse_qsl, urlparse
import pytest
from pydantic import AnyUrl
import fastmcp.server.auth
from fastmcp.server.auth.redirect_validation import (
DEFAULT_LOCALHOST_PATTERNS,
add_query_params,
build_client_redirect,
matches_allowed_pattern,
replace_query_param,
validate_redirect_uri,
)
class TestAddQueryParams:
"""Test that add_query_params preserves the registered callback's exact query bytes.
A registered redirect URI may carry an opaque or signed query string.
Decoding it with parse_qsl and re-serializing with urlencode mutates it
(a valueless `?flag` becomes `?flag=`, and non-UTF-8 percent-encoded
bytes get replaced) which breaks clients that route on, or
cryptographically validate, the raw callback query.
"""
def test_preserves_valueless_param_and_non_utf8_bytes(self):
original_query = "flag&sig=%FF%FE"
url = f"https://client.example.com/callback?{original_query}"
result = add_query_params(
url,
{
"code": "abc123",
"state": "xyz state",
"iss": "https://issuer.example.com/",
},
)
result_query = urlparse(result).query
# The original query substring must survive byte-for-byte: the
# valueless `flag` must not become `flag=`, and the non-UTF-8
# percent-encoded `sig` value must not be decoded/replaced.
assert result_query.startswith(f"{original_query}&")
# New params are appended after a single `&`, correctly encoded.
appended = result_query[len(original_query) + 1 :]
assert dict(parse_qsl(appended)) == {
"code": "abc123",
"state": "xyz state",
"iss": "https://issuer.example.com/",
}
def test_empty_query_has_no_stray_ampersand(self):
url = "https://client.example.com/callback"
result = add_query_params(url, {"code": "abc123"})
assert result == "https://client.example.com/callback?code=abc123"
def test_appends_to_existing_ordinary_query(self):
url = "https://client.example.com/callback?foo=bar"
result = add_query_params(url, {"code": "abc123"})
assert result == "https://client.example.com/callback?foo=bar&code=abc123"
class TestReplaceQueryParam:
"""Direct tests for the idempotent replace-or-append primitive that
`build_client_redirect` relies on to guarantee exactly one `iss`.
"""
def test_replaces_existing_value_in_place_preserving_other_bytes(self):
url = "https://client.example.com/callback?iss=tenant&sig=%FF%FE"
result = replace_query_param(url, "iss", "https://issuer.example.com/")
assert urlparse(result).query == (
"iss=https%3A%2F%2Fissuer.example.com%2F&sig=%FF%FE"
)
def test_appends_when_key_absent(self):
url = "https://client.example.com/callback?sig=%FF%FE"
result = replace_query_param(url, "iss", "https://issuer.example.com/")
assert urlparse(result).query == (
"sig=%FF%FE&iss=https%3A%2F%2Fissuer.example.com%2F"
)
def test_only_first_occurrence_is_replaced(self):
"""A key appearing twice in the input is left with one replaced
occurrence and one untouched -- callers must not feed this function
an already-duplicated key and expect deduplication."""
url = "https://client.example.com/callback?iss=first&iss=second"
result = replace_query_param(url, "iss", "https://issuer.example.com/")
assert urlparse(result).query == (
"iss=https%3A%2F%2Fissuer.example.com%2F&iss=second"
)
class TestBuildClientRedirect:
"""Tests for the single helper that owns the client-facing-redirect
`iss` invariant: exactly one `iss`, set to the canonical value, with
every other query byte preserved verbatim.
This is the consolidation point for RFC 9207 support -- every redirect
the OAuth proxy sends back to a client (success or error, across all
five call sites that build one) must go through this function rather
than hand-building a params dict with its own `"iss"` key.
"""
def test_appends_params_and_iss_when_absent(self):
url = "https://client.example.com/callback"
result = build_client_redirect(
url,
{"code": "abc", "state": "xyz"},
iss="https://issuer.example.com/",
)
assert dict(parse_qsl(urlparse(result).query)) == {
"code": "abc",
"state": "xyz",
"iss": "https://issuer.example.com/",
}
def test_replaces_iss_already_present_in_registered_redirect_uri(self):
"""A registered redirect_uri may legitimately carry its own `iss`
query parameter (e.g. a multi-tenant client encoding its tenant in
the callback URL). Blindly appending the server's issuer on top of
that would yield two `iss` values -- RFC 6749 §3.1 forbids a
response parameter appearing more than once, so strict clients
reject the response or read the wrong value. This is the P2 defect
this helper exists to close off at every call site, not just one.
"""
url = "https://client.example.com/callback?iss=tenant&sig=%FF%FE"
result = build_client_redirect(
url,
{"code": "abc", "state": "xyz"},
iss="https://issuer.example.com/",
)
result_query = urlparse(result).query
iss_values = parse_qs(result_query)["iss"]
assert len(iss_values) == 1
assert iss_values == ["https://issuer.example.com/"]
def test_preserves_valueless_param_and_non_utf8_bytes_alongside_existing_iss(
self,
):
"""Exact end-to-end reproduction of the worked example from the P2
review comment: registered redirect_uri already has `iss`, a
valueless `flag`, and a non-UTF-8 percent-encoded `sig` -- all three
must survive the round trip through `add_query_params` +
`replace_query_param` untouched, with only `iss` rewritten in
place.
"""
url = "https://client.example.com/callback?iss=tenant&flag&sig=%FF%FE"
result = build_client_redirect(
url,
{"code": "abc", "state": "xyz state"},
iss="https://issuer.example.com/",
)
assert urlparse(result).query == (
"iss=https%3A%2F%2Fissuer.example.com%2F"
"&flag&sig=%FF%FE&code=abc&state=xyz+state"
)
def test_rejects_iss_hand_specified_in_params(self):
"""`iss` must come from the keyword-only `iss` argument, never from
the `params` dict -- this keeps exactly one place a caller can set
it, rather than two that could disagree."""
with pytest.raises(ValueError, match="iss"):
build_client_redirect(
"https://client.example.com/callback",
{"code": "abc", "iss": "sneaky"},
iss="https://issuer.example.com/",
)
def test_empty_params_does_not_add_stray_ampersand(self):
"""The authorize-handler call site passes no extra params (it only
needs to fix up `iss` on a URL the SDK already built) -- an empty
`params` dict must not introduce a trailing/stray `&`."""
url = "https://client.example.com/callback?code=abc&state=xyz"
result = build_client_redirect(url, {}, iss="https://issuer.example.com/")
assert urlparse(result).query == (
"code=abc&state=xyz&iss=https%3A%2F%2Fissuer.example.com%2F"
)
assert "&&" not in result
assert not result.endswith("&")
class TestNoHandSpecifiedIssOutsideHelper:
"""Guard against a future call site reintroducing the duplicate-`iss`
bug this PR consolidates away.
This is the sixth review round on the RFC 9207 `iss` work, and the last
two rounds were the same defect surfacing at different call sites: a
caller hand-building a params dict with its own `"iss"` key instead of
routing through `build_client_redirect`. Rather than trust that every
future redirect site remembers to do this, scan the directories that
build client-facing authorization redirects (`oauth_proxy/`,
`handlers/`) for a dict-literal `"iss"` key. `jwt_issuer.py` and the
JWT/Clerk providers legitimately use `"iss"` as a JWT claim name, but
those live outside these two directories, so this scan does not need to
special-case them.
"""
def test_no_dict_literal_iss_key_in_redirect_building_modules(self):
auth_root = Path(fastmcp.server.auth.__file__).parent
scan_dirs = [auth_root / "oauth_proxy", auth_root / "handlers"]
iss_dict_key = re.compile(r"""["']iss["']\s*:""")
offenders = [
str(path)
for scan_dir in scan_dirs
for path in scan_dir.rglob("*.py")
if iss_dict_key.search(path.read_text())
]
assert not offenders, (
"Found a hand-specified 'iss' dict key outside "
"build_client_redirect() in: "
f"{offenders}. Route this redirect through "
"fastmcp.server.auth.redirect_validation.build_client_redirect "
"instead so the duplicate-iss invariant stays centralized."
)
class TestMatchesAllowedPattern:
"""Test wildcard pattern matching for redirect URIs."""