modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for DNS rebinding protections (#861)

Browse source
This commit is contained in:
David Dworken 2025-06-17 12:13:00 -07:00 • committed by GitHub
commit 86bb54cb28
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 799 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

13
tests/shared/test_sse.py
View file

@ -20,6 +20,7 @@ from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.server.transport_security import TransportSecuritySettings
from mcp.shared.exceptions import McpError
from mcp.types import (
EmptyResult,
@ -80,7 +81,11 @@ class ServerTest(Server):
# Test fixtures
def make_server_app() -> Starlette:
"""Create test Starlette app with SSE transport"""
sse = SseServerTransport("/messages/")
# Configure security with allowed hosts/origins for testing
security_settings = TransportSecuritySettings(
allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
)
sse = SseServerTransport("/messages/", security_settings=security_settings)
server = ServerTest()
async def handle_sse(request: Request) -> Response:
@ -339,7 +344,11 @@ class RequestContextServer(Server[object, Request]):
def run_context_server(server_port: int) -> None:
"""Run a server that captures request context"""
sse = SseServerTransport("/messages/")
# Configure security with allowed hosts/origins for testing
security_settings = TransportSecuritySettings(
allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
)
sse = SseServerTransport("/messages/", security_settings=security_settings)
context_server = RequestContextServer()
async def handle_sse(request: Request) -> Response:

10
tests/shared/test_streamable_http.py
View file

@ -37,6 +37,7 @@ from mcp.server.streamable_http import (
StreamId,
)
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.server.transport_security import TransportSecuritySettings
from mcp.shared.context import RequestContext
from mcp.shared.exceptions import McpError
from mcp.shared.message import (
@ -227,10 +228,14 @@ def create_app(is_json_response_enabled=False, event_store: EventStore | None =
server = ServerTest()
# Create the session manager
security_settings = TransportSecuritySettings(
allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
)
session_manager = StreamableHTTPSessionManager(
app=server,
event_store=event_store,
json_response=is_json_response_enabled,
security_settings=security_settings,
)
# Create an ASGI application that uses the session manager
@ -436,8 +441,9 @@ def test_content_type_validation(basic_server, basic_server_url):
},
data="This is not JSON",
)
assert response.status_code == 415
assert "Unsupported Media Type" in response.text
assert response.status_code == 400
assert "Invalid Content-Type" in response.text
def test_json_validation(basic_server, basic_server_url):