Merge commit from fork
processInput() appended every 8 KiB stdout chunk to $inputBuffer and only drained it on a "\n". A spawned server that streams stdout without ever emitting a newline grew the buffer without bound, exhausting client memory (GHSA-6vrr-g4hw-f3hj). Cap the buffer (default 4 MiB, configurable via the constructor). When appending the next chunk would exceed the cap before a newline is seen, abort the read and resolve the in-flight request with an error immediately, so the caller fails fast instead of spinning until its request timeout. Based on localden's suggested fix in the advisory: makes the cap configurable, but resolves the pending request with an error (keeping runRequest's Response|Error contract) instead of throwing a TransportException out of the read loop, and leaves connection teardown to the caller. Adds regression tests. Co-authored-by: localden <53200638+localden@users.noreply.github.com>
This commit is contained in:
parent
055593f259
commit
529099f463
2 changed files with 161 additions and 4 deletions
|
|
@ -12,6 +12,7 @@
|
||||||
namespace Mcp\Client\Transport;
|
namespace Mcp\Client\Transport;
|
||||||
|
|
||||||
use Mcp\Exception\ConnectionException;
|
use Mcp\Exception\ConnectionException;
|
||||||
|
use Mcp\Exception\InvalidArgumentException;
|
||||||
use Mcp\Schema\JsonRpc\Error;
|
use Mcp\Schema\JsonRpc\Error;
|
||||||
use Mcp\Schema\JsonRpc\Response;
|
use Mcp\Schema\JsonRpc\Response;
|
||||||
use Psr\Log\LoggerInterface;
|
use Psr\Log\LoggerInterface;
|
||||||
|
|
@ -45,6 +46,11 @@ class StdioTransport extends BaseTransport
|
||||||
|
|
||||||
private string $inputBuffer = '';
|
private string $inputBuffer = '';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Default cap on the bytes buffered while waiting for a complete line.
|
||||||
|
*/
|
||||||
|
public const DEFAULT_MAX_BUFFER_SIZE = 4 * 1024 * 1024;
|
||||||
|
|
||||||
/** @var McpFiber|null */
|
/** @var McpFiber|null */
|
||||||
private ?\Fiber $activeFiber = null;
|
private ?\Fiber $activeFiber = null;
|
||||||
|
|
||||||
|
|
@ -52,10 +58,16 @@ class StdioTransport extends BaseTransport
|
||||||
private $activeProgressCallback;
|
private $activeProgressCallback;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param string $command The command to run
|
* @param string $command The command to run
|
||||||
* @param array<int, string> $args Command arguments
|
* @param array<int, string> $args Command arguments
|
||||||
* @param string|null $cwd Working directory
|
* @param string|null $cwd Working directory
|
||||||
* @param array<string, string>|null $env Environment variables
|
* @param array<string, string>|null $env Environment variables
|
||||||
|
* @param int $maxBufferSize Maximum bytes buffered while waiting for a complete line. The
|
||||||
|
* buffer is only drained on a "\n"; a spawned server that streams
|
||||||
|
* stdout without ever emitting a newline would otherwise grow it
|
||||||
|
* without bound and exhaust client memory. Reaching the cap aborts
|
||||||
|
* the read instead. Raise it for servers that emit single frames
|
||||||
|
* larger than the default.
|
||||||
*/
|
*/
|
||||||
public function __construct(
|
public function __construct(
|
||||||
private readonly string $command,
|
private readonly string $command,
|
||||||
|
|
@ -63,8 +75,13 @@ class StdioTransport extends BaseTransport
|
||||||
private readonly ?string $cwd = null,
|
private readonly ?string $cwd = null,
|
||||||
private readonly ?array $env = null,
|
private readonly ?array $env = null,
|
||||||
?LoggerInterface $logger = null,
|
?LoggerInterface $logger = null,
|
||||||
|
private readonly int $maxBufferSize = self::DEFAULT_MAX_BUFFER_SIZE,
|
||||||
) {
|
) {
|
||||||
parent::__construct($logger);
|
parent::__construct($logger);
|
||||||
|
|
||||||
|
if ($maxBufferSize < 1) {
|
||||||
|
throw new InvalidArgumentException(\sprintf('The maximum buffer size must be a positive number of bytes, got %d.', $maxBufferSize));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public function connect(): void
|
public function connect(): void
|
||||||
|
|
@ -223,6 +240,12 @@ class StdioTransport extends BaseTransport
|
||||||
|
|
||||||
$data = fread($this->stdout, 8192);
|
$data = fread($this->stdout, 8192);
|
||||||
if (false !== $data && '' !== $data) {
|
if (false !== $data && '' !== $data) {
|
||||||
|
if (\strlen($this->inputBuffer) + \strlen($data) > $this->maxBufferSize) {
|
||||||
|
$this->abortInput(\sprintf('buffered %d bytes without a newline, exceeding the %d byte limit', \strlen($this->inputBuffer) + \strlen($data), $this->maxBufferSize));
|
||||||
|
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
$this->inputBuffer .= $data;
|
$this->inputBuffer .= $data;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -237,6 +260,33 @@ class StdioTransport extends BaseTransport
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Discard the input buffer 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 abortInput(string $reason): void
|
||||||
|
{
|
||||||
|
$bufferedBytes = \strlen($this->inputBuffer);
|
||||||
|
$this->inputBuffer = '';
|
||||||
|
|
||||||
|
$this->logger->warning('Aborting stdio input: '.$reason, [
|
||||||
|
'buffered_bytes' => $bufferedBytes,
|
||||||
|
'max_buffer_size' => $this->maxBufferSize,
|
||||||
|
]);
|
||||||
|
|
||||||
|
if (null === $this->state) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach ($this->state->getPendingRequests() as $pending) {
|
||||||
|
$requestId = $pending['request_id'];
|
||||||
|
$error = Error::forInternalError('stdio input aborted: '.$reason, $requestId);
|
||||||
|
$this->state->storeResponse($requestId, $error->jsonSerialize());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private function processFiber(): void
|
private function processFiber(): void
|
||||||
{
|
{
|
||||||
if (null === $this->activeFiber || !$this->activeFiber->isSuspended()) {
|
if (null === $this->activeFiber || !$this->activeFiber->isSuspended()) {
|
||||||
|
|
|
||||||
107
tests/Unit/Client/Transport/StdioTransportTest.php
Normal file
107
tests/Unit/Client/Transport/StdioTransportTest.php
Normal file
|
|
@ -0,0 +1,107 @@
|
||||||
|
<?php
|
||||||
|
|
||||||
|
/*
|
||||||
|
* This file is part of the official PHP MCP SDK.
|
||||||
|
*
|
||||||
|
* A collaboration between Symfony and the PHP Foundation.
|
||||||
|
*
|
||||||
|
* For the full copyright and license information, please view the LICENSE
|
||||||
|
* file that was distributed with this source code.
|
||||||
|
*/
|
||||||
|
|
||||||
|
namespace Mcp\Tests\Unit\Client\Transport;
|
||||||
|
|
||||||
|
use Mcp\Client\State\ClientState;
|
||||||
|
use Mcp\Client\Transport\StdioTransport;
|
||||||
|
use Mcp\Exception\InvalidArgumentException;
|
||||||
|
use Mcp\Schema\JsonRpc\Error;
|
||||||
|
use PHPUnit\Framework\Attributes\TestDox;
|
||||||
|
use PHPUnit\Framework\TestCase;
|
||||||
|
|
||||||
|
final class StdioTransportTest extends TestCase
|
||||||
|
{
|
||||||
|
#[TestDox('input buffer is bounded: an over-length frame without a newline is aborted')]
|
||||||
|
public function testInputBufferIsBounded(): void
|
||||||
|
{
|
||||||
|
$transport = new StdioTransport(command: 'true', maxBufferSize: 64);
|
||||||
|
$state = new ClientState();
|
||||||
|
$state->addPendingRequest(1, 30);
|
||||||
|
$transport->setState($state);
|
||||||
|
|
||||||
|
// A server that floods stdout without ever emitting a newline.
|
||||||
|
$this->setStdout($transport, $this->stream(str_repeat('a', 8192)));
|
||||||
|
|
||||||
|
$this->invokeProcessInput($transport);
|
||||||
|
|
||||||
|
self::assertSame('', $this->readPrivate($transport, 'inputBuffer'), 'buffer must be cleared on abort');
|
||||||
|
}
|
||||||
|
|
||||||
|
#[TestDox('aborting the input fails the in-flight request immediately')]
|
||||||
|
public function testAbortFailsPendingRequestFast(): void
|
||||||
|
{
|
||||||
|
$transport = new StdioTransport(command: 'true', maxBufferSize: 64);
|
||||||
|
$state = new ClientState();
|
||||||
|
$state->addPendingRequest(1, 30);
|
||||||
|
$transport->setState($state);
|
||||||
|
|
||||||
|
$this->setStdout($transport, $this->stream(str_repeat('a', 8192)));
|
||||||
|
|
||||||
|
$this->invokeProcessInput($transport);
|
||||||
|
|
||||||
|
$response = $state->consumeResponse(1);
|
||||||
|
self::assertInstanceOf(Error::class, $response);
|
||||||
|
self::assertSame(Error::INTERNAL_ERROR, $response->code);
|
||||||
|
self::assertSame(1, $response->id);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[TestDox('newline-delimited frames within the cap are parsed and dispatched')]
|
||||||
|
public function testWellFormedFramesStillParse(): void
|
||||||
|
{
|
||||||
|
$transport = new StdioTransport(command: 'true');
|
||||||
|
$messages = [];
|
||||||
|
$transport->onMessage(static function (string $message) use (&$messages): void {
|
||||||
|
$messages[] = $message;
|
||||||
|
});
|
||||||
|
|
||||||
|
$this->setStdout($transport, $this->stream('{"a":1}'."\n".'{"b":2}'."\n"));
|
||||||
|
|
||||||
|
$this->invokeProcessInput($transport);
|
||||||
|
|
||||||
|
self::assertSame(['{"a":1}', '{"b":2}'], $messages);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[TestDox('the buffer cap must be a positive number of bytes')]
|
||||||
|
public function testRejectsNonPositiveCap(): void
|
||||||
|
{
|
||||||
|
$this->expectException(InvalidArgumentException::class);
|
||||||
|
|
||||||
|
new StdioTransport(command: 'true', maxBufferSize: 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @return resource
|
||||||
|
*/
|
||||||
|
private function stream(string $contents)
|
||||||
|
{
|
||||||
|
$stream = fopen('php://temp', 'r+');
|
||||||
|
fwrite($stream, $contents);
|
||||||
|
rewind($stream);
|
||||||
|
|
||||||
|
return $stream;
|
||||||
|
}
|
||||||
|
|
||||||
|
private function setStdout(StdioTransport $transport, mixed $stream): void
|
||||||
|
{
|
||||||
|
(new \ReflectionProperty($transport, 'stdout'))->setValue($transport, $stream);
|
||||||
|
}
|
||||||
|
|
||||||
|
private function invokeProcessInput(StdioTransport $transport): void
|
||||||
|
{
|
||||||
|
(new \ReflectionMethod($transport, 'processInput'))->invoke($transport);
|
||||||
|
}
|
||||||
|
|
||||||
|
private function readPrivate(StdioTransport $transport, string $property): mixed
|
||||||
|
{
|
||||||
|
return (new \ReflectionProperty($transport, $property))->getValue($transport);
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue