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

Fix OAuth Proxy resource parameter validation (#2764)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-12-26 21:23:31 -05:00 • committed by GitHub
commit 7cb00c9686
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 300 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

25
src/fastmcp/server/auth/auth.py
View file

@ -114,6 +114,8 @@ class AuthProvider(TokenVerifierProtocol):
base_url = AnyHttpUrl(base_url)
self.base_url = base_url
self.required_scopes = required_scopes or []
self._mcp_path: str | None = None
self._resource_url: AnyHttpUrl | None = None
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify a bearer token and return access info if valid.
@ -128,6 +130,20 @@ class AuthProvider(TokenVerifierProtocol):
"""
raise NotImplementedError("Subclasses must implement verify_token")
def set_mcp_path(self, mcp_path: str | None) -> None:
"""Set the MCP endpoint path and compute resource URL.
This method is called by get_routes() to configure the expected
resource URL before route creation. Subclasses can override to
perform additional initialization that depends on knowing the
MCP endpoint path.
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
"""
self._mcp_path = mcp_path
self._resource_url = self._get_resource_url(mcp_path)
def get_routes(
self,
mcp_path: str | None = None,
@ -407,6 +423,8 @@ class OAuthProvider(
Returns:
List of OAuth routes
"""
# Configure resource URL before creating routes
self.set_mcp_path(mcp_path)
# Create standard OAuth authorization server routes
# Pass base_url as issuer_url to ensure metadata declares endpoints where
@ -451,11 +469,8 @@ class OAuthProvider(
else:
oauth_routes.append(route)
# Get the resource URL based on the MCP path
resource_url = self._get_resource_url(mcp_path)
# Add protected resource routes if this server is also acting as a resource server
if resource_url:
if self._resource_url:
supported_scopes = (
self.client_registration_options.valid_scopes
if self.client_registration_options
@ -463,7 +478,7 @@ class OAuthProvider(
else self.required_scopes
)
protected_routes = create_protected_resource_routes(
resource_url=resource_url,
resource_url=self._resource_url,
authorization_servers=[cast(AnyHttpUrl, self.issuer_url)],
scopes_supported=supported_scopes,
)

84
src/fastmcp/server/auth/oauth_proxy.py
View file

@ -805,11 +805,10 @@ class OAuthProxy(OAuthProvider):
salt="fastmcp-jwt-signing-key",
)
self._jwt_issuer: JWTIssuer = JWTIssuer(
issuer=str(self.base_url),
audience=f"{str(self.base_url).rstrip('/')}/mcp",
signing_key=jwt_signing_key,
)
# Store JWT signing key for deferred JWTIssuer creation in set_mcp_path()
self._jwt_signing_key: bytes = jwt_signing_key
# JWTIssuer will be created in set_mcp_path() with correct audience
self._jwt_issuer: JWTIssuer | None = None
# If the user does not provide a store, we will provide an encrypted disk store
if client_storage is None:
@ -897,6 +896,47 @@ class OAuthProxy(OAuthProvider):
self._upstream_authorization_endpoint,
)
# -------------------------------------------------------------------------
# MCP Path Configuration
# -------------------------------------------------------------------------
def set_mcp_path(self, mcp_path: str | None) -> None:
"""Set the MCP endpoint path and create JWTIssuer with correct audience.
This method is called by get_routes() to configure the resource URL
and create the JWTIssuer. The JWT audience is set to the full resource
URL (e.g., http://localhost:8000/mcp) to ensure tokens are bound to
this specific MCP endpoint.
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
"""
super().set_mcp_path(mcp_path)
# Create JWT issuer with correct audience based on actual MCP path
# This ensures tokens are bound to the specific resource URL
self._jwt_issuer = JWTIssuer(
issuer=str(self.base_url),
audience=str(self._resource_url),
signing_key=self._jwt_signing_key,
)
logger.debug("Configured OAuth proxy for resource URL: %s", self._resource_url)
@property
def jwt_issuer(self) -> JWTIssuer:
"""Get the JWT issuer, ensuring it has been initialized.
The JWT issuer is created when set_mcp_path() is called (via get_routes()).
This property ensures a clear error if used before initialization.
"""
if self._jwt_issuer is None:
raise RuntimeError(
"JWT issuer not initialized. Ensure get_routes() is called "
"before token operations."
)
return self._jwt_issuer
# -------------------------------------------------------------------------
# PKCE Helper Methods
# -------------------------------------------------------------------------
@ -998,13 +1038,29 @@ class OAuthProxy(OAuthProvider):
"""Start OAuth transaction and route through consent interstitial.
Flow:
1. Store transaction with client details and PKCE (if forwarding)
2. Return local /consent URL; browser visits consent first
3. Consent handler redirects to upstream IdP if approved/already approved
1. Validate client's resource matches server's resource URL (security check)
2. Store transaction with client details and PKCE (if forwarding)
3. Return local /consent URL; browser visits consent first
4. Consent handler redirects to upstream IdP if approved/already approved
If consent is disabled (require_authorization_consent=False), skip the consent screen
and redirect directly to the upstream IdP.
"""
# Security check: validate client's requested resource matches this server
# This prevents tokens intended for one server from being used on another
client_resource = getattr(params, "resource", None)
if client_resource and self._resource_url:
if str(client_resource) != str(self._resource_url):
logger.warning(
"Resource mismatch: client requested %s but server is %s",
client_resource,
self._resource_url,
)
raise AuthorizeError(
error="invalid_target", # type: ignore[arg-type]
error_description="Resource does not match this server",
)
# Generate transaction ID for this authorization request
txn_id = secrets.token_urlsafe(32)
@ -1216,7 +1272,7 @@ class OAuthProxy(OAuthProvider):
# Issue minimal FastMCP access token (just a reference via JTI)
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
fastmcp_access_token = self._jwt_issuer.issue_access_token(
fastmcp_access_token = self.jwt_issuer.issue_access_token(
client_id=client.client_id,
scopes=authorization_code.scopes,
jti=access_jti,
@ -1227,7 +1283,7 @@ class OAuthProxy(OAuthProvider):
# Use upstream refresh token expiry to align lifetimes
fastmcp_refresh_token = None
if refresh_jti and refresh_expires_in:
fastmcp_refresh_token = self._jwt_issuer.issue_refresh_token(
fastmcp_refresh_token = self.jwt_issuer.issue_refresh_token(
client_id=client.client_id,
scopes=authorization_code.scopes,
jti=refresh_jti,
@ -1352,7 +1408,7 @@ class OAuthProxy(OAuthProvider):
"""
# Verify FastMCP refresh token
try:
refresh_payload = self._jwt_issuer.verify_token(refresh_token.token)
refresh_payload = self.jwt_issuer.verify_token(refresh_token.token)
refresh_jti = refresh_payload["jti"]
except Exception as e:
logger.debug("FastMCP refresh token validation failed: %s", e)
@ -1461,7 +1517,7 @@ class OAuthProxy(OAuthProvider):
if client.client_id is None:
raise TokenError("invalid_client", "Client ID is required")
new_access_jti = secrets.token_urlsafe(32)
new_fastmcp_access = self._jwt_issuer.issue_access_token(
new_fastmcp_access = self.jwt_issuer.issue_access_token(
client_id=client.client_id,
scopes=scopes,
jti=new_access_jti,
@ -1482,7 +1538,7 @@ class OAuthProxy(OAuthProvider):
# Issue NEW minimal FastMCP refresh token (rotation for security)
# Use upstream refresh token expiry to align lifetimes
new_refresh_jti = secrets.token_urlsafe(32)
new_fastmcp_refresh = self._jwt_issuer.issue_refresh_token(
new_fastmcp_refresh = self.jwt_issuer.issue_refresh_token(
client_id=client.client_id,
scopes=scopes,
jti=new_refresh_jti,
@ -1558,7 +1614,7 @@ class OAuthProxy(OAuthProvider):
"""
try:
# 1. Verify FastMCP JWT signature and claims
payload = self._jwt_issuer.verify_token(token)
payload = self.jwt_issuer.verify_token(token)
jti = payload["jti"]
# 2. Look up upstream token via JTI mapping

210
tests/server/auth/test_oauth_proxy.py
View file

@ -640,6 +640,9 @@ class TestOAuthProxyTokenEndpointAuth:
jwt_signing_key="test-secret",
)
# Initialize JWT issuer before token operations
proxy.set_mcp_path("/mcp")
# First, create a valid FastMCP token via full OAuth flow
client = OAuthClientInformationFull(
client_id="test-client",
@ -815,6 +818,9 @@ class TestOAuthProxyE2E:
jwt_signing_key="test-secret",
)
# Initialize JWT issuer before token operations
proxy.set_mcp_path("/mcp")
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
@ -1466,3 +1472,207 @@ class TestFallbackAccessTokenExpiry:
)
assert provider._fallback_access_token_expiry_seconds is None
class TestResourceURLValidation:
"""Tests for OAuth Proxy resource URL validation (GHSA-5h2m-4q8j-pqpj fix)."""
@pytest.fixture
def proxy_with_resource_url(self, jwt_verifier):
"""Create an OAuthProxy with set_mcp_path called."""
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",
)
# Use non-default path to prove fix isn't relying on old hardcoded /mcp
proxy.set_mcp_path("/api/v2/mcp")
return proxy
async def test_authorize_rejects_mismatched_resource(self, proxy_with_resource_url):
"""Test that authorization rejects requests with mismatched resource."""
from mcp.server.auth.provider import AuthorizeError
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 a different resource than the server's
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="challenge",
scopes=["read"],
resource="https://malicious-server.com/mcp", # Wrong resource
)
with pytest.raises(AuthorizeError) as exc_info:
await proxy_with_resource_url.authorize(client, params)
assert exc_info.value.error == "invalid_target"
assert "Resource does not match" in exc_info.value.error_description
async def test_authorize_accepts_matching_resource(self, proxy_with_resource_url):
"""Test that authorization accepts requests with matching resource."""
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 the correct resource (must match /api/v2/mcp path)
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", # Correct resource
)
# Should succeed (redirect to consent page)
redirect_url = await proxy_with_resource_url.authorize(client, params)
assert "/consent" in redirect_url
async def test_authorize_rejects_old_hardcoded_mcp_path(
self, proxy_with_resource_url
):
"""Test that old hardcoded /mcp path is rejected when server uses different path."""
from mcp.server.auth.provider import AuthorizeError
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 the old hardcoded /mcp path (would have worked before fix)
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", # Old hardcoded path
)
# Should fail because server is at /api/v2/mcp, not /mcp
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_accepts_no_resource(self, proxy_with_resource_url):
"""Test that authorization accepts requests without resource parameter."""
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 doesn't specify resource
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="challenge",
scopes=["read"],
# No resource parameter
)
# Should succeed (no resource check needed)
redirect_url = await proxy_with_resource_url.authorize(client, params)
assert "/consent" in redirect_url
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(
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",
)
# Before set_mcp_path, _jwt_issuer is None
assert proxy._jwt_issuer is None
# Call set_mcp_path with custom path
proxy.set_mcp_path("/custom/mcp")
# After set_mcp_path, _jwt_issuer should be created
assert proxy._jwt_issuer is not None
assert proxy.jwt_issuer.audience == "https://proxy.example.com/custom/mcp"
assert proxy.jwt_issuer.issuer == "https://proxy.example.com/"
def test_set_mcp_path_uses_base_url_if_no_path(self, jwt_verifier):
"""Test that set_mcp_path uses base_url as audience if no path provided."""
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(None)
assert proxy.jwt_issuer.audience == "https://proxy.example.com/"
def test_jwt_issuer_property_raises_if_not_initialized(self, jwt_verifier):
"""Test that jwt_issuer property raises if set_mcp_path not called."""
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",
)
with pytest.raises(RuntimeError) as exc_info:
_ = proxy.jwt_issuer
assert "JWT issuer not initialized" in str(exc_info.value)
def test_get_routes_calls_set_mcp_path(self, jwt_verifier):
"""Test that get_routes() calls set_mcp_path() to initialize JWT issuer."""
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",
)
# Before get_routes, _jwt_issuer is None
assert proxy._jwt_issuer is None
# get_routes should call set_mcp_path internally
proxy.get_routes("/api/mcp")
# After get_routes, _jwt_issuer should be created with correct audience
assert proxy._jwt_issuer is not None
assert proxy.jwt_issuer.audience == "https://proxy.example.com/api/mcp"