diff --git a/src/mcp/server/auth/__init__.py b/src/mcp/server/auth/__init__.py index 61b60e3487..34f2e3e1c9 100644 --- a/src/mcp/server/auth/__init__.py +++ b/src/mcp/server/auth/__init__.py @@ -1 +1,3 @@ """MCP OAuth server authorization components.""" + +from .url_validators import validate_issuer_url, validate_redirect_uri diff --git a/src/mcp/server/auth/handlers/register.py b/src/mcp/server/auth/handlers/register.py index 7fb14b2c43..ff41d95bd0 100644 --- a/src/mcp/server/auth/handlers/register.py +++ b/src/mcp/server/auth/handlers/register.py @@ -9,6 +9,7 @@ from starlette.responses import Response from mcp.server.auth.errors import stringify_pydantic_error +from mcp.server.auth.url_validators import validate_redirect_uri from mcp.server.auth.json_response import PydanticJSONResponse from mcp.server.auth.provider import OAuthAuthorizationServerProvider, RegistrationError, RegistrationErrorCode from mcp.server.auth.settings import ClientRegistrationOptions @@ -35,6 +36,20 @@ 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 + if client_metadata.redirect_uris: + for uri in 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..60b6f5d55e --- /dev/null +++ b/src/mcp/server/auth/url_validators.py @@ -0,0 +1,46 @@ +"""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 AnyHttpUrl + + +def validate_issuer_url(url: AnyHttpUrl): + """Validate that the issuer URL meets OAuth 2.0 requirements. + + Args: + url: The issuer URL to validate. + + Raises: + ValueError: If the issuer URL is invalid. + """ + if url.scheme != "https" and url.host not in ("localhost", "127.0.0.1", "[::1]"): + raise ValueError("Issuer URL must be HTTPS") + + if url.fragment: + raise ValueError("Issuer URL must not have a fragment") + if url.query: + raise ValueError("Issuer URL must not have a query string") + + +def validate_redirect_uri(url: AnyHttpUrl): + """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_routes.py b/tests/server/auth/test_routes.py index 58685c64c7..ad2935fdaf 100644 --- a/tests/server/auth/test_routes.py +++ b/tests/server/auth/test_routes.py @@ -3,6 +3,7 @@ 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 HTTPS"): + validate_redirect_uri(AnyHttpUrl("javascript:alert(1)")) + + +def test_validate_redirect_uri_file_scheme_rejected(): + with pytest.raises(ValueError, match="Redirect URI must use HTTPS"): + validate_redirect_uri(AnyHttpUrl("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#"))