| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -4,27 +4,25 @@ | |
|
|
||
| import contextlib | ||
| import logging | ||
| from collections import deque | ||
| from collections.abc import AsyncIterator | ||
| from typing import TYPE_CHECKING, Any | ||
| from typing import TYPE_CHECKING, Any, Final | ||
| from uuid import uuid4 | ||
|
|
||
| import anyio | ||
| from anyio.abc import TaskStatus | ||
| from mcp_types import DEFAULT_NEGOTIATED_VERSION, INVALID_REQUEST, ErrorData, JSONRPCError | ||
| from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS | ||
| from starlette.datastructures import Headers | ||
| from starlette.requests import Request | ||
| from starlette.responses import Response | ||
| from starlette.types import Receive, Scope, Send | ||
| from starlette.types import ASGIApp, Message, Receive, Scope, Send | ||
|
|
||
| from mcp.server._streamable_http_modern import handle_modern_request | ||
| from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context | ||
| from mcp.server.connection import Connection | ||
| from mcp.server.runner import serve_connection, serve_loop | ||
| from mcp.server.streamable_http import ( | ||
| MCP_SESSION_ID_HEADER, | ||
| EventStore, | ||
| StreamableHTTPServerTransport, | ||
| ) | ||
| from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, EventStore, StreamableHTTPServerTransport | ||
| from mcp.server.transport_security import TransportSecuritySettings | ||
| from mcp.shared._compat import resync_tracer | ||
| from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER | ||
| Expand All | @@ -36,6 +34,9 @@ | |
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024 | ||
| """Default maximum Streamable HTTP request body size in bytes (4 MiB).""" | ||
|
|
||
|
|
||
| class StreamableHTTPSessionManager: | ||
| """Manages StreamableHTTP sessions with optional resumability via event store. | ||
| Expand Down Expand Up | @@ -69,6 +70,8 @@ | |
| retry_interval is also configured, ensure the idle timeout comfortably exceeds the retry interval to | ||
| avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800 | ||
| (30 minutes) is recommended for most deployments. | ||
| max_request_body_size: Maximum size in bytes for Streamable HTTP POST request bodies. Requests that | ||
| exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB. | ||
| """ | ||
|
|
||
| def __init__( | ||
| Expand All | @@ -80,11 +83,14 @@ | |
| security_settings: TransportSecuritySettings | None = None, | ||
| retry_interval: int | None = None, | ||
| session_idle_timeout: float | None = None, | ||
| max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE, | ||
| ): | ||
| if session_idle_timeout is not None and session_idle_timeout <= 0: | ||
| raise ValueError("session_idle_timeout must be a positive number of seconds") | ||
| if stateless and session_idle_timeout is not None: | ||
| raise RuntimeError("session_idle_timeout is not supported in stateless mode") | ||
| if max_request_body_size <= 0: | ||
| raise ValueError("max_request_body_size must be a positive number of bytes") | ||
|
|
||
| self.app = app | ||
| self.event_store = event_store | ||
| Expand All | @@ -93,6 +99,8 @@ | |
| self.security_settings = security_settings | ||
| self.retry_interval = retry_interval | ||
| self.session_idle_timeout = session_idle_timeout | ||
| self.max_request_body_size = max_request_body_size | ||
| self.asgi_app = RequestBodyLimitMiddleware(self._handle_request, max_request_body_size) | ||
|
|
||
| # Session tracking (only used if not stateless) | ||
| self._session_creation_lock = anyio.Lock() | ||
| Expand Down Expand Up | @@ -159,6 +167,9 @@ | |
|
|
||
| Dispatches to the appropriate handler based on stateless mode. | ||
| """ | ||
| await self.asgi_app(scope, receive, send) | ||
|
|
||
| async def _handle_request(self, scope: Scope, receive: Receive, send: Send) -> None: | ||
| if self._task_group is None: | ||
| raise RuntimeError("Task group is not initialized. Make sure to use run().") | ||
|
|
||
| Expand Down Expand Up | @@ -360,11 +371,71 @@ | |
| await response(scope, receive, send) | ||
|
|
||
|
|
||
| class RequestBodyLimitMiddleware: | ||
| """Reject oversized HTTP request bodies before invoking an ASGI application.""" | ||
|
Check notice on line 375 in src/mcp/server/streamable_http_manager.py
|
||
|
Comment thread
Comment on lines
+374
to
+375
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality🟣 Pre-existing issue, not introduced by this PR: the legacy SSE transport's POST endpoint (SseServerTransport.handle_post_message, src/mcp/server/sse.py:243) still buffers the entire request body unbounded via await request.body() — the same DoS vector this PR closes for Streamable HTTP. Since the new RequestBodyLimitMiddleware is a generic ASGI wrapper, applying it to the SSE message endpoint would be a natural one-line follow-up (or the docs could note explicitly that the limit covers Streamable HTTP only). Extended reasoning...What the issue isThis PR adds max_request_body_size enforcement (default 4 MiB, HTTP 413) for the Streamable HTTP transport via the new RequestBodyLimitMiddleware. The legacy SSE transport has the exact same unbounded-buffering exposure, and it remains open: SseServerTransport.handle_post_message (src/mcp/server/sse.py:205) reads the full POST body with body = await request.body() (src/mcp/server/sse.py:243) before any JSON validation, with no size check of any kind. The code pathThe SSE transport is still shipped and mountable — mcp.run(transport="sse") and sse_app() both wire sse.handle_post_message onto the message endpoint (see the Mount(message_path, app=sse.handle_post_message) routes in src/mcp/server/mcpserver/server.py). A request to that endpoint passes only header-level checks (transport security, session_id lookup) before line 243 buffers the entire body into memory. Starlette's Request.body() accumulates every chunk until the stream ends, so nothing bounds it. Step-by-step proof
Why this is pre-existing and non-blockingThe PR does not touch sse.py, and both documentation additions correctly scope the new limit: docs/migration.md says "Streamable HTTP request bodies are limited to 4 MiB" and the docs/run/index.md bullet sits under the Streamable HTTP section. So nothing in this PR is wrong or misleading — the gap predates it and is untouched by it. (For completeness: both earlier claude[bot] findings on this PR — handle_request bypassing the limit, and per-message dict memory amplification — are already fixed in the current revision: handle_request now delegates to asgi_app, and chunks coalesce into a single bytearray replayed as one message.) How to fix (follow-up)RequestBodyLimitMiddleware is deliberately transport-agnostic — it wraps any ASGI app and only inspects POST requests. Wrapping the SSE message endpoint, e.g. Mount(message_path, app=RequestBodyLimitMiddleware(sse.handle_post_message, max_request_body_size)) in sse_app(), would close the gap with essentially one line. Alternatively (or additionally), a one-line doc note stating that the 4 MiB limit applies to Streamable HTTP only would make the scope explicit for operators still serving SSE clients. Given SSE is legacy/discouraged (docs/run/index.md answers "When" with "You don't"), a follow-up issue is the right vehicle — this should not block the PR.
Sorry, something went wrong.
claude[bot] reacted with thumbs up emoji
claude[bot] reacted with thumbs down emoji
All reactions
|
||
|
|
||
| def __init__(self, app: ASGIApp, max_body_size: int) -> None: | ||
| self.app = app | ||
| self.max_body_size = max_body_size | ||
|
|
||
| async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: | ||
| if scope["type"] != "http" or scope["method"] != "POST": | ||
| await self.app(scope, receive, send) | ||
| return | ||
|
|
||
| headers = Headers(scope=scope) | ||
| content_length = headers.get("content-length") | ||
| if content_length is not None: | ||
| try: | ||
| declared_size = int(content_length) | ||
| except ValueError: | ||
| pass | ||
| else: | ||
| if declared_size > self.max_body_size: | ||
| response = Response("Request body too large", status_code=413) | ||
| return await response(scope, receive, send) | ||
|
|
||
| received_body = bytearray() | ||
| received_request = False | ||
| body_complete = False | ||
| trailing_message: Message | None = None | ||
| while True: | ||
| message = await receive() | ||
| if message["type"] != "http.request": | ||
| trailing_message = message | ||
| break | ||
|
|
||
| received_request = True | ||
| body = message.get("body", b"") | ||
| if len(received_body) + len(body) > self.max_body_size: | ||
| response = Response("Request body too large", status_code=413) | ||
| return await response(scope, receive, send) | ||
| received_body.extend(body) | ||
| if not message.get("more_body", False): | ||
| body_complete = True | ||
| break | ||
|
|
||
| cached_messages: deque[Message] = deque() | ||
| if received_request: | ||
| cached_messages.append( | ||
| {"type": "http.request", "body": bytes(received_body), "more_body": not body_complete} | ||
| ) | ||
| if trailing_message is not None: | ||
| cached_messages.append(trailing_message) | ||
|
|
||
| async def replay() -> Message: | ||
| if cached_messages: | ||
| return cached_messages.popleft() | ||
| return await receive() | ||
|
|
||
| await self.app(scope, replay, send) | ||
|
|
||
|
|
||
| class StreamableHTTPASGIApp: | ||
| """ASGI application for Streamable HTTP server transport.""" | ||
|
|
||
| def __init__(self, session_manager: StreamableHTTPSessionManager): | ||
| self.session_manager = session_manager | ||
|
|
||
| async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: | ||
| await self.session_manager.handle_request(scope, receive, send) | ||
| await self.session_manager.asgi_app(scope, receive, send) | ||
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.