mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-28 02:10:38 +02:00
Normalize resource URLs before comparison to support RFC 8707 query parameters (#2967)
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
Abhijeeth Padarthi
GitHub
parent
c1f50e49be
commit
cee99d1210
2 changed files with 283 additions and 3 deletions
|
|
@ -3,9 +3,63 @@
|
|||
import pytest
|
||||
from mcp.server.auth.provider import AuthorizationParams, AuthorizeError
|
||||
from mcp.shared.auth import OAuthClientInformationFull
|
||||
from pydantic import AnyUrl
|
||||
from pydantic import AnyHttpUrl, AnyUrl
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
||||
from fastmcp.server.auth.oauth_proxy.proxy import (
|
||||
_normalize_resource_url,
|
||||
_server_url_has_query,
|
||||
)
|
||||
|
||||
|
||||
class TestNormalizeResourceUrl:
|
||||
"""Unit tests for the _normalize_resource_url helper function."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url,expected",
|
||||
[
|
||||
# Basic URL unchanged
|
||||
("https://example.com/mcp", "https://example.com/mcp"),
|
||||
# Query parameters stripped
|
||||
("https://example.com/mcp?foo=bar", "https://example.com/mcp"),
|
||||
("https://example.com/mcp?a=1&b=2", "https://example.com/mcp"),
|
||||
# Fragments stripped
|
||||
("https://example.com/mcp#section", "https://example.com/mcp"),
|
||||
# Both query and fragment stripped
|
||||
("https://example.com/mcp?foo=bar#section", "https://example.com/mcp"),
|
||||
# Trailing slash stripped
|
||||
("https://example.com/mcp/", "https://example.com/mcp"),
|
||||
# Trailing slash with query params
|
||||
("https://example.com/mcp/?foo=bar", "https://example.com/mcp"),
|
||||
# Preserves path structure
|
||||
(
|
||||
"https://example.com/api/v2/mcp?kb_name=test",
|
||||
"https://example.com/api/v2/mcp",
|
||||
),
|
||||
# Preserves port
|
||||
("https://example.com:8080/mcp?foo=bar", "https://example.com:8080/mcp"),
|
||||
# Root path
|
||||
("https://example.com/?foo=bar", "https://example.com"),
|
||||
("https://example.com/", "https://example.com"),
|
||||
],
|
||||
)
|
||||
def test_normalizes_urls_correctly(self, url: str, expected: str):
|
||||
"""Test that URLs are normalized by stripping query params, fragments, and trailing slashes."""
|
||||
assert _normalize_resource_url(url) == expected
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url,has_query",
|
||||
[
|
||||
("https://example.com/mcp", False),
|
||||
("https://example.com/mcp?foo=bar", True),
|
||||
("https://example.com/mcp?", False), # Empty query string
|
||||
("https://example.com/mcp#fragment", False),
|
||||
("https://example.com/mcp?a=1&b=2", True),
|
||||
],
|
||||
)
|
||||
def test_server_url_has_query(self, url: str, has_query: bool):
|
||||
"""Test detection of query parameters in server URLs."""
|
||||
assert _server_url_has_query(url) == has_query
|
||||
|
||||
|
||||
class TestResourceURLValidation:
|
||||
|
|
@ -131,6 +185,185 @@ class TestResourceURLValidation:
|
|||
redirect_url = await proxy_with_resource_url.authorize(client, params)
|
||||
assert "/consent" in redirect_url
|
||||
|
||||
async def test_authorize_accepts_resource_with_query_params(
|
||||
self, proxy_with_resource_url
|
||||
):
|
||||
"""Test that authorization accepts resource URLs with query parameters.
|
||||
|
||||
Per RFC 8707, clients may include query parameters in resource URLs.
|
||||
ChatGPT sends resource URLs like ?kb_name=X, which should match the
|
||||
server's resource URL that doesn't include query params.
|
||||
"""
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
|
||||
await proxy_with_resource_url.register_client(client)
|
||||
|
||||
# Client requests resource with query params (like ChatGPT does)
|
||||
params = AuthorizationParams(
|
||||
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
||||
redirect_uri_provided_explicitly=True,
|
||||
state="client-state",
|
||||
code_challenge="challenge",
|
||||
scopes=["read"],
|
||||
resource="https://proxy.example.com/api/v2/mcp?kb_name=test",
|
||||
)
|
||||
|
||||
# Should succeed - base URL matches, query params are normalized away
|
||||
redirect_url = await proxy_with_resource_url.authorize(client, params)
|
||||
assert "/consent" in redirect_url
|
||||
|
||||
async def test_authorize_rejects_different_path_with_query_params(
|
||||
self, proxy_with_resource_url
|
||||
):
|
||||
"""Test that query param normalization doesn't bypass path validation."""
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
|
||||
await proxy_with_resource_url.register_client(client)
|
||||
|
||||
# Client requests wrong path but with query params
|
||||
params = AuthorizationParams(
|
||||
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
||||
redirect_uri_provided_explicitly=True,
|
||||
state="client-state",
|
||||
code_challenge="challenge",
|
||||
scopes=["read"],
|
||||
resource="https://proxy.example.com/wrong/path?kb_name=test",
|
||||
)
|
||||
|
||||
# Should fail - path doesn't match even after normalizing query params
|
||||
with pytest.raises(AuthorizeError) as exc_info:
|
||||
await proxy_with_resource_url.authorize(client, params)
|
||||
|
||||
assert exc_info.value.error == "invalid_target"
|
||||
|
||||
async def test_authorize_requires_exact_match_when_server_has_query_params(
|
||||
self, jwt_verifier
|
||||
):
|
||||
"""Test that when server URL has query params, exact match is required.
|
||||
|
||||
If a server configures its resource URL with query params (e.g., for
|
||||
multi-tenant or per-KB scoping), clients must provide the exact same
|
||||
query params. This prevents bypassing tenant isolation.
|
||||
"""
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=jwt_verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
proxy.set_mcp_path("/mcp")
|
||||
# Simulate server configured with query params for tenant scoping
|
||||
proxy._resource_url = AnyHttpUrl("https://proxy.example.com/mcp?tenant=acme")
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client)
|
||||
|
||||
# Client requests with DIFFERENT query params - should fail
|
||||
params = AuthorizationParams(
|
||||
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
||||
redirect_uri_provided_explicitly=True,
|
||||
state="client-state",
|
||||
code_challenge="challenge",
|
||||
scopes=["read"],
|
||||
resource="https://proxy.example.com/mcp?tenant=other", # Wrong tenant!
|
||||
)
|
||||
|
||||
with pytest.raises(AuthorizeError) as exc_info:
|
||||
await proxy.authorize(client, params)
|
||||
|
||||
assert exc_info.value.error == "invalid_target"
|
||||
|
||||
async def test_authorize_accepts_exact_match_when_server_has_query_params(
|
||||
self, jwt_verifier
|
||||
):
|
||||
"""Test that exact query param match succeeds when server has query params."""
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=jwt_verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
proxy.set_mcp_path("/mcp")
|
||||
# Simulate server configured with query params for tenant scoping
|
||||
proxy._resource_url = AnyHttpUrl("https://proxy.example.com/mcp?tenant=acme")
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client)
|
||||
|
||||
# Client requests with SAME query params - should succeed
|
||||
params = AuthorizationParams(
|
||||
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
||||
redirect_uri_provided_explicitly=True,
|
||||
state="client-state",
|
||||
code_challenge="challenge",
|
||||
scopes=["read"],
|
||||
resource="https://proxy.example.com/mcp?tenant=acme", # Exact match
|
||||
)
|
||||
|
||||
redirect_url = await proxy.authorize(client, params)
|
||||
assert "/consent" in redirect_url
|
||||
|
||||
async def test_authorize_rejects_no_query_when_server_has_query_params(
|
||||
self, jwt_verifier
|
||||
):
|
||||
"""Test that missing query params are rejected when server requires them."""
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=jwt_verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
proxy.set_mcp_path("/mcp")
|
||||
# Simulate server configured with query params for tenant scoping
|
||||
proxy._resource_url = AnyHttpUrl("https://proxy.example.com/mcp?tenant=acme")
|
||||
|
||||
client = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client)
|
||||
|
||||
# Client requests WITHOUT query params - should fail
|
||||
params = AuthorizationParams(
|
||||
redirect_uri=AnyUrl("http://localhost:12345/callback"),
|
||||
redirect_uri_provided_explicitly=True,
|
||||
state="client-state",
|
||||
code_challenge="challenge",
|
||||
scopes=["read"],
|
||||
resource="https://proxy.example.com/mcp", # Missing tenant param!
|
||||
)
|
||||
|
||||
with pytest.raises(AuthorizeError) as exc_info:
|
||||
await proxy.authorize(client, params)
|
||||
|
||||
assert exc_info.value.error == "invalid_target"
|
||||
|
||||
def test_set_mcp_path_creates_jwt_issuer_with_correct_audience(self, jwt_verifier):
|
||||
"""Test that set_mcp_path creates JWTIssuer with correct audience."""
|
||||
proxy = OAuthProxy(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue