import httpx import pytest from pydantic import AnyHttpUrl from fastmcp import FastMCP from fastmcp.server.auth import RemoteAuthProvider from fastmcp.server.auth.providers.jwt import StaticTokenVerifier @pytest.fixture def test_tokens(): """Standard test tokens fixture for all auth tests.""" return { "test_token": { "client_id": "test-client", "scopes": ["read", "write"], } } class TestRemoteAuthProvider: """Test suite for RemoteAuthProvider.""" def test_init(self, test_tokens): """Test RemoteAuthProvider initialization.""" token_verifier = StaticTokenVerifier(tokens=test_tokens) auth_servers = [AnyHttpUrl("https://auth.example.com")] provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=auth_servers, base_url="https://api.example.com", ) assert provider.token_verifier is token_verifier assert provider.authorization_servers == auth_servers assert provider.base_url == AnyHttpUrl("https://api.example.com/") async def test_verify_token_delegates_to_verifier(self, test_tokens): """Test that verify_token delegates to the token verifier.""" # Use a different token for this specific test tokens = { "valid_token": { "client_id": "test-client", "scopes": [], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com", ) # Valid token result = await provider.verify_token("valid_token") assert result is not None assert result.token == "valid_token" assert result.client_id == "test-client" # Invalid token result = await provider.verify_token("invalid_token") assert result is None def test_get_routes_creates_protected_resource_routes(self, test_tokens): """Test that get_routes creates protected resource routes.""" token_verifier = StaticTokenVerifier(tokens=test_tokens) auth_servers = [AnyHttpUrl("https://auth.example.com")] provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=auth_servers, base_url="https://api.example.com", ) routes = provider.get_routes() assert len(routes) == 1 # Check that the route is the OAuth protected resource metadata endpoint # When called without mcp_path, it creates route at /.well-known/oauth-protected-resource route = routes[0] assert route.path == "/.well-known/oauth-protected-resource" assert route.methods is not None assert "GET" in route.methods def test_get_resource_url_with_well_known_path(self): """Test _get_resource_url returns correct URL for .well-known path.""" tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com", ) metadata_url = provider._get_resource_url( "/.well-known/oauth-protected-resource/mcp" ) assert metadata_url == AnyHttpUrl( "https://api.example.com/.well-known/oauth-protected-resource/mcp" ) def test_get_resource_url_with_nested_base_url(self): """Test _get_resource_url returns correct URL for .well-known path with nested base_url.""" tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com/v1/", ) metadata_url = provider._get_resource_url( "/.well-known/oauth-protected-resource/mcp" ) assert metadata_url == AnyHttpUrl( "https://api.example.com/v1/.well-known/oauth-protected-resource/mcp" ) def test_get_resource_url_handles_trailing_slash(self): """Test _get_resource_url handles trailing slash correctly.""" tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com/", ) metadata_url = provider._get_resource_url( "/.well-known/oauth-protected-resource/mcp" ) assert metadata_url == AnyHttpUrl( "https://api.example.com/.well-known/oauth-protected-resource/mcp" ) def test_get_resource_url_uses_resource_base_url_when_provided(self, test_tokens): """Test protected resource URLs are derived from resource_base_url when provided.""" token_verifier = StaticTokenVerifier(tokens=test_tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://auth.example.com/proxy", resource_base_url="https://api.example.com", ) assert provider._get_resource_url("/mcp") == AnyHttpUrl( "https://api.example.com/mcp" ) def test_init_preserves_legacy_positional_scopes_supported_slot(self, test_tokens): """Legacy positional scopes_supported should not bind to resource_base_url.""" token_verifier = StaticTokenVerifier(tokens=test_tokens) provider = RemoteAuthProvider( token_verifier, [AnyHttpUrl("https://auth.example.com")], "https://api.example.com", ["read"], ) assert provider._scopes_supported == ["read"] assert provider.resource_base_url is None assert provider._get_resource_url("/mcp") == AnyHttpUrl( "https://api.example.com/mcp" ) class TestRemoteAuthProviderIntegration: """Integration tests for RemoteAuthProvider with FastMCP server.""" @pytest.fixture def basic_auth_provider(self, test_tokens): """Basic RemoteAuthProvider fixture for testing.""" token_verifier = StaticTokenVerifier(tokens=test_tokens) return RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com", ) def _create_test_auth_provider( self, base_url="https://api.example.com", test_tokens=None, **kwargs ): """Helper to create a test RemoteAuthProvider with StaticTokenVerifier.""" tokens = kwargs.get( "tokens", test_tokens or { "test_token": { "client_id": "test-client", "scopes": ["read", "write"], } }, ) token_verifier = StaticTokenVerifier(tokens=tokens) return RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url=base_url, ) async def test_protected_resource_metadata_endpoint_status_code( self, basic_auth_provider ): """Test that the protected resource metadata endpoint returns 200.""" mcp = FastMCP("test-server", auth=basic_auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://api.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 async def test_protected_resource_metadata_endpoint_resource_field(self): """Test that the protected resource metadata endpoint returns correct resource field.""" auth_provider = self._create_test_auth_provider() mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://api.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") data = response.json() # This is the key test - ensure resource field contains the full MCP URL assert data["resource"] == "https://api.example.com/mcp" async def test_protected_resource_metadata_endpoint_authorization_servers_field( self, ): """Test that the protected resource metadata endpoint returns correct authorization_servers field.""" auth_provider = self._create_test_auth_provider() mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://api.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") data = response.json() assert data["authorization_servers"] == ["https://auth.example.com/"] @pytest.mark.parametrize( "base_url,expected_resource", [ ("https://api.example.com", "https://api.example.com/mcp"), ("https://api.example.com/", "https://api.example.com/mcp"), ("https://api.example.com/v1/", "https://api.example.com/v1/mcp"), ], ) async def test_base_url_configurations(self, base_url: str, expected_resource: str): """Test different base_url configurations.""" from urllib.parse import urlparse auth_provider = self._create_test_auth_provider(base_url=base_url) mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() # Extract the path from the expected resource to construct metadata URL resource_parsed = urlparse(expected_resource) # Remove leading slash if present to avoid double slashes resource_path = resource_parsed.path.lstrip("/") metadata_path = f"/.well-known/oauth-protected-resource/{resource_path}" async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://test.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get(metadata_path) assert response.status_code == 200 data = response.json() assert data["resource"] == expected_resource async def test_multiple_authorization_servers_resource_field(self): """Test resource field with multiple authorization servers.""" auth_servers = [ AnyHttpUrl("https://auth1.example.com"), AnyHttpUrl("https://auth2.example.com"), ] auth_provider = self._create_test_auth_provider() # Override the authorization servers auth_provider.authorization_servers = auth_servers mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://api.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") data = response.json() assert data["resource"] == "https://api.example.com/mcp" async def test_multiple_authorization_servers_list(self): """Test authorization_servers field with multiple authorization servers.""" auth_servers = [ AnyHttpUrl("https://auth1.example.com"), AnyHttpUrl("https://auth2.example.com"), ] auth_provider = self._create_test_auth_provider() # Override the authorization servers auth_provider.authorization_servers = auth_servers mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://api.example.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") data = response.json() assert set(data["authorization_servers"]) == { "https://auth1.example.com/", "https://auth2.example.com/", } async def test_token_verification_with_valid_auth_succeeds(self): """Test that requests with valid auth token succeed.""" # Note: This test focuses on HTTP-level authentication behavior # For the RemoteAuthProvider, the key test is that the OAuth discovery # endpoint correctly reports the resource server URL, which is tested above # This is primarily testing that the token verifier integration works tokens = { "valid_token": { "client_id": "test-client", "scopes": [], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com", ) # Test that the provider correctly delegates to the token verifier result = await provider.verify_token("valid_token") assert result is not None assert result.token == "valid_token" assert result.client_id == "test-client" result = await provider.verify_token("invalid_token") assert result is None async def test_token_verification_with_invalid_auth_fails(self): """Test that the provider correctly rejects invalid tokens.""" tokens = { "valid_token": { "client_id": "test-client", "scopes": [], } } token_verifier = StaticTokenVerifier(tokens=tokens) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://api.example.com", ) # Test that invalid tokens are rejected result = await provider.verify_token("invalid_token") assert result is None async def test_issue_1348_oauth_discovery_returns_correct_url(self): """Test that RemoteAuthProvider correctly returns the full MCP endpoint URL. This test confirms that RemoteAuthProvider works correctly and returns the resource URL with the MCP path appended to the base URL. """ tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) auth_provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://accounts.google.com")], base_url="https://my-server.com", ) mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://my-server.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 data = response.json() # The RemoteAuthProvider correctly returns the full MCP endpoint URL assert data["resource"] == "https://my-server.com/mcp" assert data["authorization_servers"] == ["https://accounts.google.com/"] async def test_resource_name_field(self): """Test that RemoteAuthProvider correctly returns the resource_name. This test confirms that RemoteAuthProvider works correctly and returns the exact resource_name specified. """ tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) auth_provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://accounts.google.com")], base_url="https://my-server.com", resource_name="My Test Resource", ) mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://my-server.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 data = response.json() # The RemoteAuthProvider correctly returns the resource_name assert data["resource_name"] == "My Test Resource" async def test_resource_documentation_field(self): """Test that RemoteAuthProvider correctly returns the resource_documentation. This test confirms that RemoteAuthProvider works correctly and returns the exact resource_documentation specified. """ tokens = { "test_token": { "client_id": "test-client", "scopes": ["read"], } } token_verifier = StaticTokenVerifier(tokens=tokens) auth_provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://accounts.google.com")], base_url="https://my-server.com", resource_documentation=AnyHttpUrl( "https://doc.my-server.com/resource-docs" ), ) mcp = FastMCP("test-server", auth=auth_provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://my-server.com", ) as client: # The metadata URL is path-aware per RFC 9728 response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 data = response.json() # The RemoteAuthProvider correctly returns the resource_documentation assert ( data["resource_documentation"] == "https://doc.my-server.com/resource-docs" ) async def test_scopes_supported_overrides_metadata(self): """Test that scopes_supported parameter overrides what's in metadata.""" token_verifier = StaticTokenVerifier( tokens={ "test": {"client_id": "c", "scopes": ["read"]}, }, required_scopes=["read"], ) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://my-server.com", scopes_supported=["api://my-api/read"], ) mcp = FastMCP("test-server", auth=provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://my-server.com", ) as client: response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 data = response.json() assert data["scopes_supported"] == ["api://my-api/read"] async def test_scopes_supported_defaults_to_verifier(self): """Test that metadata uses verifier scopes_supported when parameter not set.""" token_verifier = StaticTokenVerifier( tokens={ "test": {"client_id": "c", "scopes": ["read"]}, }, required_scopes=["read"], ) provider = RemoteAuthProvider( token_verifier=token_verifier, authorization_servers=[AnyHttpUrl("https://auth.example.com")], base_url="https://my-server.com", ) mcp = FastMCP("test-server", auth=provider) mcp_http_app = mcp.http_app() async with httpx.AsyncClient( transport=httpx.ASGITransport(app=mcp_http_app), base_url="https://my-server.com", ) as client: response = await client.get("/.well-known/oauth-protected-resource/mcp") assert response.status_code == 200 data = response.json() assert data["scopes_supported"] == ["read"]