Improve OAuth protected resource metadata URL construction per RFC 9728 (#1407)
This commit is contained in:
parent
814c9c024a
commit
df3e428ee8
4 changed files with 192 additions and 32 deletions
|
|
@ -32,7 +32,7 @@ class ResourceServerSettings(BaseSettings):
|
|||
# Server settings
|
||||
host: str = "localhost"
|
||||
port: int = 8001
|
||||
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:8001")
|
||||
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:8001/mcp")
|
||||
|
||||
# Authorization Server settings
|
||||
auth_server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:9000")
|
||||
|
|
@ -137,7 +137,7 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http
|
|||
|
||||
# Create settings
|
||||
host = "localhost"
|
||||
server_url = f"http://{host}:{port}"
|
||||
server_url = f"http://{host}:{port}/mcp"
|
||||
settings = ResourceServerSettings(
|
||||
host=host,
|
||||
port=port,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pydantic import AnyHttpUrl
|
||||
from starlette.middleware.cors import CORSMiddleware
|
||||
|
|
@ -186,6 +187,25 @@ def build_metadata(
|
|||
return metadata
|
||||
|
||||
|
||||
def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
|
||||
"""
|
||||
Build RFC 9728 compliant protected resource metadata URL.
|
||||
|
||||
Inserts /.well-known/oauth-protected-resource between host and resource path
|
||||
as specified in RFC 9728 §3.1.
|
||||
|
||||
Args:
|
||||
resource_server_url: The resource server URL (e.g., https://example.com/mcp)
|
||||
|
||||
Returns:
|
||||
The metadata URL (e.g., https://example.com/.well-known/oauth-protected-resource/mcp)
|
||||
"""
|
||||
parsed = urlparse(str(resource_server_url))
|
||||
# Handle trailing slash: if path is just "/", treat as empty
|
||||
resource_path = parsed.path if parsed.path != "/" else ""
|
||||
return AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-protected-resource{resource_path}")
|
||||
|
||||
|
||||
def create_protected_resource_routes(
|
||||
resource_url: AnyHttpUrl,
|
||||
authorization_servers: list[AnyHttpUrl],
|
||||
|
|
@ -218,9 +238,15 @@ def create_protected_resource_routes(
|
|||
|
||||
handler = ProtectedResourceMetadataHandler(metadata)
|
||||
|
||||
# RFC 9728 §3.1: Register route at /.well-known/oauth-protected-resource + resource path
|
||||
metadata_url = build_resource_metadata_url(resource_url)
|
||||
# Extract just the path part for route registration
|
||||
parsed = urlparse(str(metadata_url))
|
||||
well_known_path = parsed.path
|
||||
|
||||
return [
|
||||
Route(
|
||||
"/.well-known/oauth-protected-resource",
|
||||
well_known_path,
|
||||
endpoint=cors_middleware(handler.handle, ["GET", "OPTIONS"]),
|
||||
methods=["GET", "OPTIONS"],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -825,11 +825,10 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
# Determine resource metadata URL
|
||||
resource_metadata_url = None
|
||||
if self.settings.auth and self.settings.auth.resource_server_url:
|
||||
from pydantic import AnyHttpUrl
|
||||
from mcp.server.auth.routes import build_resource_metadata_url
|
||||
|
||||
resource_metadata_url = AnyHttpUrl(
|
||||
str(self.settings.auth.resource_server_url).rstrip("/") + "/.well-known/oauth-protected-resource"
|
||||
)
|
||||
# Build compliant metadata URL for WWW-Authenticate header
|
||||
resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)
|
||||
|
||||
# Auth is enabled, wrap the endpoints with RequireAuthMiddleware
|
||||
routes.append(
|
||||
|
|
@ -938,11 +937,10 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
# Determine resource metadata URL
|
||||
resource_metadata_url = None
|
||||
if self.settings.auth and self.settings.auth.resource_server_url:
|
||||
from pydantic import AnyHttpUrl
|
||||
from mcp.server.auth.routes import build_resource_metadata_url
|
||||
|
||||
resource_metadata_url = AnyHttpUrl(
|
||||
str(self.settings.auth.resource_server_url).rstrip("/") + "/.well-known/oauth-protected-resource"
|
||||
)
|
||||
# Build compliant metadata URL for WWW-Authenticate header
|
||||
resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)
|
||||
|
||||
routes.append(
|
||||
Route(
|
||||
|
|
@ -961,23 +959,13 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
|
||||
# Add protected resource metadata endpoint if configured as RS
|
||||
if self.settings.auth and self.settings.auth.resource_server_url:
|
||||
from mcp.server.auth.handlers.metadata import ProtectedResourceMetadataHandler
|
||||
from mcp.server.auth.routes import cors_middleware
|
||||
from mcp.shared.auth import ProtectedResourceMetadata
|
||||
from mcp.server.auth.routes import create_protected_resource_routes
|
||||
|
||||
protected_resource_metadata = ProtectedResourceMetadata(
|
||||
resource=self.settings.auth.resource_server_url,
|
||||
authorization_servers=[self.settings.auth.issuer_url],
|
||||
scopes_supported=self.settings.auth.required_scopes,
|
||||
)
|
||||
routes.append(
|
||||
Route(
|
||||
"/.well-known/oauth-protected-resource",
|
||||
endpoint=cors_middleware(
|
||||
ProtectedResourceMetadataHandler(protected_resource_metadata).handle,
|
||||
["GET", "OPTIONS"],
|
||||
),
|
||||
methods=["GET", "OPTIONS"],
|
||||
routes.extend(
|
||||
create_protected_resource_routes(
|
||||
resource_url=self.settings.auth.resource_server_url,
|
||||
authorization_servers=[self.settings.auth.issuer_url],
|
||||
scopes_supported=self.settings.auth.required_scopes,
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ from inline_snapshot import snapshot
|
|||
from pydantic import AnyHttpUrl
|
||||
from starlette.applications import Starlette
|
||||
|
||||
from mcp.server.auth.routes import create_protected_resource_routes
|
||||
from mcp.server.auth.routes import build_resource_metadata_url, create_protected_resource_routes
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -36,10 +36,11 @@ async def test_client(test_app: Starlette):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_endpoint(test_client: httpx.AsyncClient):
|
||||
"""Test the OAuth 2.0 Protected Resource metadata endpoint."""
|
||||
async def test_metadata_endpoint_with_path(test_client: httpx.AsyncClient):
|
||||
"""Test the OAuth 2.0 Protected Resource metadata endpoint for path-based resource."""
|
||||
|
||||
response = await test_client.get("/.well-known/oauth-protected-resource")
|
||||
# For resource with path "/resource", metadata should be accessible at the path-aware location
|
||||
response = await test_client.get("/.well-known/oauth-protected-resource/resource")
|
||||
assert response.json() == snapshot(
|
||||
{
|
||||
"resource": "https://example.com/resource",
|
||||
|
|
@ -50,3 +51,148 @@ async def test_metadata_endpoint(test_client: httpx.AsyncClient):
|
|||
"bearer_methods_supported": ["header"],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_endpoint_root_path_returns_404(test_client: httpx.AsyncClient):
|
||||
"""Test that root path returns 404 for path-based resource."""
|
||||
|
||||
# Root path should return 404 for path-based resources
|
||||
response = await test_client.get("/.well-known/oauth-protected-resource")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def root_resource_app():
|
||||
"""Fixture to create protected resource routes for root-level resource."""
|
||||
|
||||
# Create routes for a resource without path component
|
||||
protected_resource_routes = create_protected_resource_routes(
|
||||
resource_url=AnyHttpUrl("https://example.com"),
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
scopes_supported=["read"],
|
||||
resource_name="Root Resource",
|
||||
)
|
||||
|
||||
app = Starlette(routes=protected_resource_routes)
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def root_resource_client(root_resource_app: Starlette):
|
||||
"""Fixture to create an HTTP client for the root resource app."""
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.ASGITransport(app=root_resource_app), base_url="https://mcptest.com"
|
||||
) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_endpoint_without_path(root_resource_client: httpx.AsyncClient):
|
||||
"""Test metadata endpoint for root-level resource."""
|
||||
|
||||
# For root resource, metadata should be at standard location
|
||||
response = await root_resource_client.get("/.well-known/oauth-protected-resource")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == snapshot(
|
||||
{
|
||||
"resource": "https://example.com/",
|
||||
"authorization_servers": ["https://auth.example.com/"],
|
||||
"scopes_supported": ["read"],
|
||||
"resource_name": "Root Resource",
|
||||
"bearer_methods_supported": ["header"],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class TestMetadataUrlConstruction:
|
||||
"""Test URL construction utility function."""
|
||||
|
||||
def test_url_without_path(self):
|
||||
"""Test URL construction for resource without path component."""
|
||||
resource_url = AnyHttpUrl("https://example.com")
|
||||
result = build_resource_metadata_url(resource_url)
|
||||
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
def test_url_with_path_component(self):
|
||||
"""Test URL construction for resource with path component."""
|
||||
resource_url = AnyHttpUrl("https://example.com/mcp")
|
||||
result = build_resource_metadata_url(resource_url)
|
||||
assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"
|
||||
|
||||
def test_url_with_trailing_slash_only(self):
|
||||
"""Test URL construction for resource with trailing slash only."""
|
||||
resource_url = AnyHttpUrl("https://example.com/")
|
||||
result = build_resource_metadata_url(resource_url)
|
||||
# Trailing slash should be treated as empty path
|
||||
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"resource_url,expected_url",
|
||||
[
|
||||
("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
|
||||
("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
|
||||
("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
|
||||
("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
|
||||
],
|
||||
)
|
||||
def test_various_resource_configurations(self, resource_url: str, expected_url: str):
|
||||
"""Test URL construction with various resource configurations."""
|
||||
result = build_resource_metadata_url(AnyHttpUrl(resource_url))
|
||||
assert str(result) == expected_url
|
||||
|
||||
|
||||
class TestRouteConsistency:
|
||||
"""Test consistency between URL generation and route registration."""
|
||||
|
||||
def test_route_path_matches_metadata_url(self):
|
||||
"""Test that route path matches the generated metadata URL."""
|
||||
resource_url = AnyHttpUrl("https://example.com/mcp")
|
||||
|
||||
# Generate metadata URL
|
||||
metadata_url = build_resource_metadata_url(resource_url)
|
||||
|
||||
# Create routes
|
||||
routes = create_protected_resource_routes(
|
||||
resource_url=resource_url,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
)
|
||||
|
||||
# Extract path from metadata URL
|
||||
from urllib.parse import urlparse
|
||||
|
||||
metadata_path = urlparse(str(metadata_url)).path
|
||||
|
||||
# Verify consistency
|
||||
assert len(routes) == 1
|
||||
assert routes[0].path == metadata_path
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"resource_url,expected_path",
|
||||
[
|
||||
("https://example.com", "/.well-known/oauth-protected-resource"),
|
||||
("https://example.com/", "/.well-known/oauth-protected-resource"),
|
||||
("https://example.com/mcp", "/.well-known/oauth-protected-resource/mcp"),
|
||||
],
|
||||
)
|
||||
def test_consistent_paths_for_various_resources(self, resource_url: str, expected_path: str):
|
||||
"""Test that URL generation and route creation are consistent."""
|
||||
resource_url_obj = AnyHttpUrl(resource_url)
|
||||
|
||||
# Test URL generation
|
||||
metadata_url = build_resource_metadata_url(resource_url_obj)
|
||||
from urllib.parse import urlparse
|
||||
|
||||
url_path = urlparse(str(metadata_url)).path
|
||||
|
||||
# Test route creation
|
||||
routes = create_protected_resource_routes(
|
||||
resource_url=resource_url_obj,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
)
|
||||
route_path = routes[0].path
|
||||
|
||||
# Both should match expected path
|
||||
assert url_path == expected_path
|
||||
assert route_path == expected_path
|
||||
assert url_path == route_path
|
||||
|
|
|
|||
Loading…
Reference in a new issue