Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions src/mcp/server/auth/handlers/register.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,18 @@
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

from mcp.server.auth.errors import stringify_pydantic_error
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
Expand All @@ -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):
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
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(
Expand Down
28 changes: 28 additions & 0 deletions src/mcp/server/auth/url_validators.py
Original file line number Diff line number Diff line change
@@ -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
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.


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")
36 changes: 36 additions & 0 deletions tests/server/auth/test_error_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
43 changes: 42 additions & 1 deletion tests/server/auth/test_routes.py
Original file line number Diff line number Diff line change
@@ -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():
Expand Down Expand Up @@ -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():

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: validate_redirect_uri accepts http:// for any host, not just loopback. The docstring promises RFC 9700 HTTPS enforcement with loopback exception, but only non-HTTP(S) schemes are rejected — http://evil.com/cb passes. This exposes authorization codes to interception over plain HTTP for non-loopback redirect URIs.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/server/auth/test_routes.py, line 102:

<comment>validate_redirect_uri accepts http:// for any host, not just loopback. The docstring promises RFC 9700 HTTPS enforcement with loopback exception, but only non-HTTP(S) schemes are rejected — http://evil.com/cb passes. This exposes authorization codes to interception over plain HTTP for non-loopback redirect URIs.</comment>

<file context>
@@ -99,9 +99,8 @@ def test_validate_redirect_uri_file_scheme_rejected():
-def test_validate_redirect_uri_http_non_loopback_rejected():
-    with pytest.raises(ValueError, match="Redirect URI must use HTTPS"):
-        validate_redirect_uri(AnyHttpUrl("http://evil.com/cb"))
+def test_validate_redirect_uri_http_non_loopback_allowed():
+    validate_redirect_uri(AnyHttpUrl("http://evil.com/cb"))
 
</file context>

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#"))
Loading