diff --git a/src/fastmcp/server/auth/oauth_proxy/proxy.py b/src/fastmcp/server/auth/oauth_proxy/proxy.py index f81f0884d..7914009a3 100644 --- a/src/fastmcp/server/auth/oauth_proxy/proxy.py +++ b/src/fastmcp/server/auth/oauth_proxy/proxy.py @@ -23,7 +23,7 @@ import secrets import time from base64 import urlsafe_b64encode from typing import Any -from urllib.parse import urlencode +from urllib.parse import urlencode, urlparse, urlunparse import httpx from authlib.common.security import generate_token @@ -78,6 +78,30 @@ from fastmcp.utilities.logging import get_logger logger = get_logger(__name__) +def _normalize_resource_url(url: str) -> str: + """Normalize a resource URL by removing query parameters and trailing slashes. + + RFC 8707 allows clients to include query parameters in resource URLs, but the + server's configured resource URL typically doesn't include them. This function + normalizes URLs for comparison by stripping query params and fragments. + + Args: + url: The URL to normalize + + Returns: + Normalized URL with scheme, host, and path only (no query/fragment) + """ + parsed = urlparse(str(url)) + return urlunparse( + (parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "", "") + ) + + +def _server_url_has_query(url: str) -> bool: + """Check if a URL has query parameters.""" + return bool(urlparse(str(url)).query) + + class OAuthProxy(OAuthProvider, ConsentMixin): """OAuth provider that presents a DCR-compliant interface while proxying to non-DCR IDPs. @@ -617,9 +641,32 @@ class OAuthProxy(OAuthProvider, ConsentMixin): """ # Security check: validate client's requested resource matches this server # This prevents tokens intended for one server from being used on another + # + # Per RFC 8707, clients may include query parameters in resource URLs (e.g., + # ChatGPT sends ?kb_name=X). We handle two cases: + # + # 1. Server URL has NO query params: normalize both URLs (strip query/fragment) + # to allow clients like ChatGPT that add query params to still match. + # + # 2. Server URL HAS query params (e.g., multi-tenant ?tenant=X): require exact + # match to prevent clients from bypassing tenant isolation by changing params. + # + # Claude doesn't send a resource parameter at all, so this check is skipped. client_resource = getattr(params, "resource", None) if client_resource and self._resource_url: - if str(client_resource) != str(self._resource_url): + server_url = str(self._resource_url) + client_url = str(client_resource) + + if _server_url_has_query(server_url): + # Server has query params - require exact match for security + urls_match = client_url.rstrip("/") == server_url.rstrip("/") + else: + # Server has no query params - normalize both for comparison + urls_match = _normalize_resource_url( + client_url + ) == _normalize_resource_url(server_url) + + if not urls_match: logger.warning( "Resource mismatch: client requested %s but server is %s", client_resource, diff --git a/tests/server/auth/oauth_proxy/test_config.py b/tests/server/auth/oauth_proxy/test_config.py index b69a1d80f..ae3abe002 100644 --- a/tests/server/auth/oauth_proxy/test_config.py +++ b/tests/server/auth/oauth_proxy/test_config.py @@ -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(