From 055593f2592e1d7f5f48587c999ca7068a3be90d Mon Sep 17 00:00:00 2001 From: Antoine Bluchet Date: Mon, 27 Jul 2026 11:15:43 +0200 Subject: [PATCH] Merge commit from fork processSSEStream() appended every 4 KiB chunk to $sseBuffer and only flushed on the "\n\n" SSE event delimiter. A server streaming data without ever sending the delimiter grew the buffer without bound, exhausting client memory (GHSA-7m52-jw36-44r3). Cap the buffer (default 8 MiB, configurable via the constructor). When appending the next chunk would exceed the cap before a complete event is seen, abort the stream and resolve the in-flight request with an error immediately, so the caller fails fast instead of spinning until its request timeout. Reimplements tonghuaroot's advisory fix: makes the cap configurable rather than a hardcoded constant, fails the pending request fast instead of relying on the timeout, and adds regression tests. Co-authored-by: tonghuaroot <23011166+tonghuaroot@users.noreply.github.com> --- src/Client/Transport/HttpTransport.php | 65 +++++++++- .../Client/Transport/HttpTransportTest.php | 117 ++++++++++++++++++ 2 files changed, 177 insertions(+), 5 deletions(-) create mode 100644 tests/Unit/Client/Transport/HttpTransportTest.php diff --git a/src/Client/Transport/HttpTransport.php b/src/Client/Transport/HttpTransport.php index 33869f7..dd3255f 100644 --- a/src/Client/Transport/HttpTransport.php +++ b/src/Client/Transport/HttpTransport.php @@ -14,6 +14,7 @@ namespace Mcp\Client\Transport; use Http\Discovery\Psr17FactoryDiscovery; use Http\Discovery\Psr18ClientDiscovery; use Mcp\Exception\ConnectionException; +use Mcp\Exception\InvalidArgumentException; use Mcp\Schema\JsonRpc\Error; use Mcp\Schema\JsonRpc\Response; use Psr\Http\Client\ClientInterface; @@ -52,11 +53,24 @@ class HttpTransport extends BaseTransport private string $sseBuffer = ''; /** - * @param string $endpoint The MCP server endpoint URL - * @param array $headers Additional headers to send - * @param ClientInterface|null $httpClient PSR-18 HTTP client (auto-discovered if null) - * @param RequestFactoryInterface|null $requestFactory PSR-17 request factory (auto-discovered if null) - * @param StreamFactoryInterface|null $streamFactory PSR-17 stream factory (auto-discovered if null) + * Default cap on the bytes buffered while waiting for a complete SSE event. + */ + public const DEFAULT_MAX_SSE_BUFFER_BYTES = 8 * 1024 * 1024; + + private readonly int $maxSseBufferBytes; + + /** + * @param string $endpoint The MCP server endpoint URL + * @param array $headers Additional headers to send + * @param ClientInterface|null $httpClient PSR-18 HTTP client (auto-discovered if null) + * @param RequestFactoryInterface|null $requestFactory PSR-17 request factory (auto-discovered if null) + * @param StreamFactoryInterface|null $streamFactory PSR-17 stream factory (auto-discovered if null) + * @param int $maxSseBufferBytes Maximum bytes buffered while waiting for a complete + * SSE event. A server that never sends the "\n\n" event + * delimiter would otherwise grow the buffer without bound + * and exhaust client memory; reaching the cap aborts the + * stream instead. Raise it for servers that legitimately + * emit single events larger than the default. */ public function __construct( private readonly string $endpoint, @@ -65,9 +79,15 @@ class HttpTransport extends BaseTransport ?RequestFactoryInterface $requestFactory = null, ?StreamFactoryInterface $streamFactory = null, ?LoggerInterface $logger = null, + int $maxSseBufferBytes = self::DEFAULT_MAX_SSE_BUFFER_BYTES, ) { parent::__construct($logger); + if ($maxSseBufferBytes < 1) { + throw new InvalidArgumentException(\sprintf('The maximum SSE buffer size must be a positive number of bytes, got %d.', $maxSseBufferBytes)); + } + + $this->maxSseBufferBytes = $maxSseBufferBytes; $this->httpClient = $httpClient ?? Psr18ClientDiscovery::find(); $this->requestFactory = $requestFactory ?? Psr17FactoryDiscovery::findRequestFactory(); $this->streamFactory = $streamFactory ?? Psr17FactoryDiscovery::findStreamFactory(); @@ -200,6 +220,12 @@ class HttpTransport extends BaseTransport if (!$this->activeStream->eof()) { $chunk = $this->activeStream->read(4096); if ('' !== $chunk) { + if (\strlen($this->sseBuffer) + \strlen($chunk) > $this->maxSseBufferBytes) { + $this->abortSseStream(\sprintf('buffered %d bytes without a complete event, exceeding the %d byte limit', \strlen($this->sseBuffer) + \strlen($chunk), $this->maxSseBufferBytes)); + + return; + } + $this->sseBuffer .= $chunk; } } @@ -218,6 +244,35 @@ class HttpTransport extends BaseTransport } } + /** + * Tear down the active SSE stream and fail any in-flight request. + * + * The waiting fiber is resolved with an error immediately so the caller + * fails fast, rather than spinning until the request timeout elapses. + */ + private function abortSseStream(string $reason): void + { + $bufferedBytes = \strlen($this->sseBuffer); + $this->sseBuffer = ''; + $this->activeStream = null; + + $this->logger->warning('Aborting SSE stream: '.$reason, [ + 'session_id' => $this->sessionId, + 'buffered_bytes' => $bufferedBytes, + 'max_sse_buffer_bytes' => $this->maxSseBufferBytes, + ]); + + if (null === $this->state) { + return; + } + + foreach ($this->state->getPendingRequests() as $pending) { + $requestId = $pending['request_id']; + $error = Error::forInternalError('SSE stream aborted: '.$reason, $requestId); + $this->state->storeResponse($requestId, $error->jsonSerialize()); + } + } + /** * Parse a single SSE event and handle the message. */ diff --git a/tests/Unit/Client/Transport/HttpTransportTest.php b/tests/Unit/Client/Transport/HttpTransportTest.php new file mode 100644 index 0000000..100107d --- /dev/null +++ b/tests/Unit/Client/Transport/HttpTransportTest.php @@ -0,0 +1,117 @@ +factory = new Psr17Factory(); + } + + #[TestDox('SSE stream is aborted before the buffer can exceed the configured cap')] + public function testSseBufferIsBoundedByConfiguredCap(): void + { + $transport = $this->createTransport(maxSseBufferBytes: 64); + $state = new ClientState(); + $state->addPendingRequest(1, 30); + $transport->setState($state); + + // A server that streams data without ever sending the "\n\n" delimiter. + $this->setActiveStream($transport, $this->factory->createStream(str_repeat('a', 4096))); + + $this->invokeProcessSseStream($transport); + + self::assertSame('', $this->readPrivate($transport, 'sseBuffer'), 'buffer must be cleared on abort'); + self::assertNull($this->readPrivate($transport, 'activeStream'), 'stream must be released on abort'); + } + + #[TestDox('aborting the SSE stream fails the in-flight request immediately instead of waiting for its timeout')] + public function testAbortFailsPendingRequestFast(): void + { + $transport = $this->createTransport(maxSseBufferBytes: 64); + $state = new ClientState(); + $state->addPendingRequest(1, 30); + $transport->setState($state); + + $this->setActiveStream($transport, $this->factory->createStream(str_repeat('a', 4096))); + + $this->invokeProcessSseStream($transport); + + $response = $state->consumeResponse(1); + self::assertInstanceOf(Error::class, $response); + self::assertSame(Error::INTERNAL_ERROR, $response->code); + self::assertSame(1, $response->id); + } + + #[TestDox('well-formed delimited events are parsed and dispatched')] + public function testWellFormedEventsStillParse(): void + { + $transport = $this->createTransport(); + $messages = []; + $transport->onMessage(static function (string $message) use (&$messages): void { + $messages[] = $message; + }); + + $this->setActiveStream($transport, $this->factory->createStream("data: hello\n\ndata: world\n\n")); + + $this->invokeProcessSseStream($transport); + + self::assertSame(['hello', 'world'], $messages); + } + + #[TestDox('the buffer cap must be a positive number of bytes')] + public function testRejectsNonPositiveCap(): void + { + $this->expectException(InvalidArgumentException::class); + + $this->createTransport(maxSseBufferBytes: 0); + } + + private function createTransport(int $maxSseBufferBytes = 8 * 1024 * 1024): HttpTransport + { + return new HttpTransport( + endpoint: 'https://example.test/mcp', + httpClient: $this->createMock(ClientInterface::class), + requestFactory: $this->factory, + streamFactory: $this->factory, + maxSseBufferBytes: $maxSseBufferBytes, + ); + } + + private function setActiveStream(HttpTransport $transport, StreamInterface $stream): void + { + (new \ReflectionProperty($transport, 'activeStream'))->setValue($transport, $stream); + } + + private function invokeProcessSseStream(HttpTransport $transport): void + { + (new \ReflectionMethod($transport, 'processSSEStream'))->invoke($transport); + } + + private function readPrivate(HttpTransport $transport, string $property): mixed + { + return (new \ReflectionProperty($transport, $property))->getValue($transport); + } +}