mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 13:34:17 +02:00
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
parent
73b7f2e44d
commit
1cfd30840d
2 changed files with 56 additions and 3 deletions
|
|
@ -254,7 +254,11 @@ class HeadlessOAuth(OAuth):
|
|||
if response.status_code == 302:
|
||||
redirect_url = response.headers["location"]
|
||||
parsed = urlparse(redirect_url)
|
||||
query_params = parse_qs(parsed.query)
|
||||
# keep_blank_values=True so explicitly-empty params (e.g. ?state=)
|
||||
# survive parsing instead of being silently dropped. Real OAuth
|
||||
# callbacks can include empty `state` or `error_description`,
|
||||
# and downstream code distinguishes "" from missing.
|
||||
query_params = parse_qs(parsed.query, keep_blank_values=True)
|
||||
|
||||
if "error" in query_params:
|
||||
error = query_params["error"][0]
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
from unittest.mock import AsyncMock, patch
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
import fastmcp
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.utilities.tests import temporary_settings
|
||||
from fastmcp.utilities.tests import HeadlessOAuth, temporary_settings
|
||||
|
||||
|
||||
class TestTemporarySettings:
|
||||
|
|
@ -39,3 +41,50 @@ class TestTransportSetting:
|
|||
) as mock_stdio:
|
||||
await mcp.run_async(transport="stdio")
|
||||
mock_stdio.assert_called_once()
|
||||
|
||||
|
||||
class TestHeadlessOAuthCallbackHandler:
|
||||
"""Regression tests for #4056: blank query values must survive parse_qs.
|
||||
|
||||
The OAuth callback handler in HeadlessOAuth parses the redirect Location
|
||||
header. parse_qs without keep_blank_values=True silently drops keys whose
|
||||
value is empty (e.g. `?state=`), which mis-models real OAuth callbacks
|
||||
where an empty `state` is distinct from a missing one.
|
||||
"""
|
||||
|
||||
def _make_oauth_with_redirect(self, location: str) -> HeadlessOAuth:
|
||||
"""Build a HeadlessOAuth with a fake stored 302 response."""
|
||||
oauth = HeadlessOAuth.__new__(HeadlessOAuth)
|
||||
response = MagicMock()
|
||||
response.status_code = 302
|
||||
response.headers = {"location": location}
|
||||
oauth._stored_response = response
|
||||
return oauth
|
||||
|
||||
async def test_callback_preserves_blank_state(self):
|
||||
"""An explicitly-empty state must round-trip as "" rather than None."""
|
||||
oauth = self._make_oauth_with_redirect(
|
||||
"https://example.com/callback?code=abc&state="
|
||||
)
|
||||
auth_code, state = await oauth.callback_handler()
|
||||
assert auth_code == "abc"
|
||||
assert state == ""
|
||||
|
||||
async def test_callback_returns_none_when_state_missing(self):
|
||||
"""A truly missing state still returns None (default)."""
|
||||
oauth = self._make_oauth_with_redirect("https://example.com/callback?code=abc")
|
||||
auth_code, state = await oauth.callback_handler()
|
||||
assert auth_code == "abc"
|
||||
assert state is None
|
||||
|
||||
async def test_callback_uses_blank_error_description_verbatim(self):
|
||||
"""When the OAuth provider sends an empty error_description, surface
|
||||
it as "" rather than falling back to "Unknown error". The fallback is
|
||||
meant for the truly-absent case; with keep_blank_values=True the
|
||||
explicit empty value is preserved and used directly.
|
||||
"""
|
||||
oauth = self._make_oauth_with_redirect(
|
||||
"https://example.com/callback?error=invalid_request&error_description="
|
||||
)
|
||||
with pytest.raises(RuntimeError, match=r"invalid_request - $"):
|
||||
await oauth.callback_handler()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue