diff --git a/src/mcp/server/auth/handlers/register.py b/src/mcp/server/auth/handlers/register.py index 7fb14b2c43..1924598ca3 100644 --- a/src/mcp/server/auth/handlers/register.py +++ b/src/mcp/server/auth/handlers/register.py @@ -1,10 +1,10 @@ import secrets import time from dataclasses import dataclass -from typing import Any +from typing import Any, cast from uuid import uuid4 -from pydantic import BaseModel, ValidationError +from pydantic import AnyUrl, BaseModel, ValidationError from starlette.requests import Request from starlette.responses import Response @@ -12,6 +12,7 @@ from mcp.server.auth.json_response import PydanticJSONResponse from mcp.server.auth.provider import OAuthAuthorizationServerProvider, RegistrationError, RegistrationErrorCode from mcp.server.auth.settings import ClientRegistrationOptions +from mcp.server.auth.url_validators import validate_redirect_uri from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthClientInformationFull, OAuthClientMetadata # this alias is a no-op; it's just to separate out the types exposed to the @@ -35,6 +36,21 @@ async def handle(self, request: Request) -> Response: body = await request.body() client_metadata = OAuthClientMetadata.model_validate_json(body) + # Validate redirect_uris per RFC 7591 section 2. The metadata + # model requires a non-empty list (min_length=1), so no presence + # guard is needed; cast narrows the optional field for pyright. + for uri in cast(list[AnyUrl], client_metadata.redirect_uris): + try: + validate_redirect_uri(uri) + except ValueError as e: + return PydanticJSONResponse( + content=RegistrationErrorResponse( + error="invalid_redirect_uri", + error_description=str(e), + ), + status_code=400, + ) + # Scope validation is handled below except ValidationError as validation_error: return PydanticJSONResponse( diff --git a/src/mcp/server/auth/url_validators.py b/src/mcp/server/auth/url_validators.py new file mode 100644 index 0000000000..50944fc270 --- /dev/null +++ b/src/mcp/server/auth/url_validators.py @@ -0,0 +1,28 @@ +"""OAuth 2.0 URL validation helpers for MCP authorization servers. + +RFC 9700 4.1.1 and RFC 7591 2 require HTTPS for authorization endpoint URLs +and registered redirect_uris, with an HTTP loopback exception for local +development. +""" + +from pydantic import AnyUrl + + +def validate_redirect_uri(url: AnyUrl): + """Validate a registered redirect_uri for DCR. + + RFC 9700 section 4.1.1 and RFC 7591 section 2 require HTTPS for + redirect_uris, with an HTTP loopback exception for local development. + + Args: + url: The redirect URI to validate. + + Raises: + ValueError: If the redirect URI uses an unsafe scheme or contains + a fragment. + """ + if url.scheme not in ("http", "https"): + raise ValueError("Redirect URI must use an HTTP(S) scheme") + + if url.fragment is not None: + raise ValueError("Redirect URI must not contain a fragment") diff --git a/tests/server/auth/test_error_handling.py b/tests/server/auth/test_error_handling.py index cdd9caa16b..f765635b53 100644 --- a/tests/server/auth/test_error_handling.py +++ b/tests/server/auth/test_error_handling.py @@ -288,3 +288,39 @@ async def test_token_error_handling_refresh_token( data = refresh_response.json() assert data["error"] == "invalid_scope" assert data["error_description"] == "The requested scope is invalid" + + +@pytest.mark.anyio +async def test_registration_rejects_redirect_uri_with_fragment(client: httpx2.AsyncClient): + client_data = { + "redirect_uris": ["https://client.example.com/callback#frag"], + "token_endpoint_auth_method": "client_secret_post", + "grant_types": ["authorization_code", "refresh_token"], + "response_types": ["code"], + "client_name": "Test Client", + } + + response = await client.post("/register", json=client_data) + + assert response.status_code == 400, response.content + data = response.json() + assert data["error"] == "invalid_redirect_uri" + assert data["error_description"] == "Redirect URI must not contain a fragment" + + +@pytest.mark.anyio +async def test_registration_rejects_non_http_redirect_uri_scheme(client: httpx2.AsyncClient): + client_data = { + "redirect_uris": ["javascript:alert(1)"], + "token_endpoint_auth_method": "client_secret_post", + "grant_types": ["authorization_code", "refresh_token"], + "response_types": ["code"], + "client_name": "Test Client", + } + + response = await client.post("/register", json=client_data) + + assert response.status_code == 400, response.content + data = response.json() + assert data["error"] == "invalid_redirect_uri" + assert data["error_description"] == "Redirect URI must use an HTTP(S) scheme" diff --git a/tests/server/auth/test_routes.py b/tests/server/auth/test_routes.py index 58685c64c7..555556659e 100644 --- a/tests/server/auth/test_routes.py +++ b/tests/server/auth/test_routes.py @@ -1,8 +1,9 @@ import pytest -from pydantic import AnyHttpUrl +from pydantic import AnyHttpUrl, AnyUrl from mcp.server.auth.routes import build_metadata, validate_issuer_url from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions +from mcp.server.auth.url_validators import validate_redirect_uri def test_validate_issuer_url_https_allowed(): @@ -70,3 +71,43 @@ def test_build_metadata_serves_issuer_without_trailing_slash(): assert served["issuer"] == "https://as.example.com" assert served["authorization_endpoint"] == "https://as.example.com/authorize" assert served["token_endpoint"] == "https://as.example.com/token" + + +def test_validate_redirect_uri_https_allowed(): + validate_redirect_uri(AnyHttpUrl("https://example.com/cb")) + + +def test_validate_redirect_uri_http_localhost_allowed(): + validate_redirect_uri(AnyHttpUrl("http://localhost:3000/cb")) + + +def test_validate_redirect_uri_http_127_0_0_1_allowed(): + validate_redirect_uri(AnyHttpUrl("http://127.0.0.1:8080/cb")) + + +def test_validate_redirect_uri_http_ipv6_loopback_allowed(): + validate_redirect_uri(AnyHttpUrl("http://[::1]:9090/cb")) + + +def test_validate_redirect_uri_javascript_scheme_rejected(): + with pytest.raises(ValueError, match="Redirect URI must use an HTTP"): + validate_redirect_uri(AnyUrl("javascript:alert(1)")) + + +def test_validate_redirect_uri_file_scheme_rejected(): + with pytest.raises(ValueError, match="Redirect URI must use an HTTP"): + validate_redirect_uri(AnyUrl("file:///etc/passwd")) + + +def test_validate_redirect_uri_http_non_loopback_allowed(): + validate_redirect_uri(AnyHttpUrl("http://evil.com/cb")) + + +def test_validate_redirect_uri_fragment_rejected(): + with pytest.raises(ValueError, match="Redirect URI must not contain a fragment"): + validate_redirect_uri(AnyHttpUrl("https://example.com/cb#frag")) + + +def test_validate_redirect_uri_empty_fragment_rejected(): + with pytest.raises(ValueError, match="Redirect URI must not contain a fragment"): + validate_redirect_uri(AnyHttpUrl("https://example.com/cb#"))