diff --git a/src/fastmcp/server/auth/auth.py b/src/fastmcp/server/auth/auth.py index bc255b394..c3bbd1b17 100644 --- a/src/fastmcp/server/auth/auth.py +++ b/src/fastmcp/server/auth/auth.py @@ -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, ) diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py index 9968040e3..21b2d09cd 100644 --- a/src/fastmcp/server/auth/oauth_proxy.py +++ b/src/fastmcp/server/auth/oauth_proxy.py @@ -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 diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py index f389c5f86..223fb270d 100644 --- a/tests/server/auth/test_oauth_proxy.py +++ b/tests/server/auth/test_oauth_proxy.py @@ -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"