diff --git a/src/Io/LineSeparatedReader.php b/src/Io/LineSeparatedReader.php index 8088867..9c18e37 100644 --- a/src/Io/LineSeparatedReader.php +++ b/src/Io/LineSeparatedReader.php @@ -13,6 +13,7 @@ class LineSeparatedReader extends EventEmitter implements ReadableStreamInterfac private $stream; private $buffer = ''; private $closed = false; + private $maxlength = 65536; public function __construct(ReadableStreamInterface $stream) { @@ -53,12 +54,16 @@ public function handleData($chunk) { $this->buffer .= $chunk; - while (($pos = strpos($this->buffer, "\n")) !== false) { + while (($pos = strpos($this->buffer, "\n")) !== false && $pos <= $this->maxlength) { $chunk = rtrim(substr($this->buffer, 0, $pos), "\r\n"); $this->buffer = (string)substr($this->buffer, $pos + 1); $this->emit('data', array($chunk)); } + + if (isset($this->buffer[$this->maxlength])) { + $this->handleError(new \OverflowException('Buffer size exceeded')); + } } public function handleEnd() diff --git a/src/SshProcessConnector.php b/src/SshProcessConnector.php index 372e892..3b6e691 100644 --- a/src/SshProcessConnector.php +++ b/src/SshProcessConnector.php @@ -138,6 +138,12 @@ public function connect($uri) $connected = true; }); + // terminate process if STDERR fails, e.g. because of excessive line length + $stderr->on('error', function (\Exception $e) use ($process, &$last) { + $last = $e->getMessage(); + $process->terminate(); + }); + // If STDERR closes before connection was established, explicitly close STDOUT stream. // The STDOUT stream starts in a paused state and as such will prevent the process exit // logic from triggering when it is not resumed. diff --git a/src/SshSocksConnector.php b/src/SshSocksConnector.php index ffff526..6a7197e 100644 --- a/src/SshSocksConnector.php +++ b/src/SshSocksConnector.php @@ -243,6 +243,12 @@ private function listen() } }); + // terminate process if STDERR fails, e.g. because of excessive line length + $stderr->on('error', function (\Exception $e) use ($process, &$last) { + $last = $e->getMessage(); + $process->terminate(); + }); + $process->on('exit', function ($code) use ($deferred, &$last) { $deferred->reject(new \RuntimeException( $last, diff --git a/tests/IntegrationSshProcessConnectorTest.php b/tests/IntegrationSshProcessConnectorTest.php index 25551d1..0ad2061 100644 --- a/tests/IntegrationSshProcessConnectorTest.php +++ b/tests/IntegrationSshProcessConnectorTest.php @@ -74,6 +74,22 @@ public function testConnectCancellationWillRejectWithException() Loop::run(); } + public function testConnectWillRejectWithExceptionAndTerminateProcessWhenProcessOutputsLineExceedingMaximumBufferSize() + { + $connector = new SshProcessConnector('host'); + + $ref = new \ReflectionProperty($connector, 'cmd'); + if (PHP_VERSION_ID < 80100) { + $ref->setAccessible(true); + } + $ref->setValue($connector, 'head -c 70000 /dev/zero | tr "\0" a >&2; exec sleep 10; #'); + + $promise = $connector->connect('example.com:80'); + + $this->setExpectedException('RuntimeException', 'Connection to example.com:80 failed because SSH client died (Buffer size exceeded)'); + \React\Async\await($promise); + } + public function testConnectWillResolveWithConnectionThatWillEmitImmediateDataFromProcessStdoutAfterChannelOpenConfirmMessage() { $connector = new SshProcessConnector('host'); diff --git a/tests/IntegrationSshSocksConnectorTest.php b/tests/IntegrationSshSocksConnectorTest.php index 87681c6..aeb16ea 100644 --- a/tests/IntegrationSshSocksConnectorTest.php +++ b/tests/IntegrationSshSocksConnectorTest.php @@ -117,4 +117,20 @@ public function testConnectWillRejectWithExceptionAndTerminateIdleProcessWhenSoc $this->assertInstanceOf('React\ChildProcess\Process', $process); $this->assertTrue($process->isTerminated()); } + + public function testConnectWillRejectWithExceptionAndTerminateProcessWhenProcessOutputsLineExceedingMaximumBufferSize() + { + $connector = new SshSocksConnector('host'); + + $ref = new \ReflectionProperty($connector, 'cmd'); + if (PHP_VERSION_ID < 80100) { + $ref->setAccessible(true); + } + $ref->setValue($connector, 'head -c 70000 /dev/zero | tr "\0" a >&2; exec sleep 10; #'); + + $promise = $connector->connect('example.com:80'); + + $this->setExpectedException('RuntimeException', 'Connection to example.com:80 failed because SSH client process died (Buffer size exceeded)'); + \React\Async\await($promise); + } } diff --git a/tests/Io/LineSeparatedReaderTest.php b/tests/Io/LineSeparatedReaderTest.php index 39ecaa1..cfb5ffc 100644 --- a/tests/Io/LineSeparatedReaderTest.php +++ b/tests/Io/LineSeparatedReaderTest.php @@ -36,6 +36,54 @@ public function testStreamDataWillEmitDataBehindNewlineWhenDataContainsDataBehin $input->write("hello!\nignored"); } + public function testStreamDataWillEmitDataWhenDataContainsNewlineAtMaximumBufferSize() + { + $input = new ThroughStream(); + $stream = new LineSeparatedReader($input); + + $stream->on('data', $this->expectCallableOnceWith(str_repeat('a', 65536))); + $stream->on('error', $this->expectCallableNever()); + + $input->write(str_repeat('a', 65536)); + $input->write("\n"); + } + + public function testStreamDataWillEmitErrorAndCloseWhenDataExceedsMaximumBufferSize() + { + $input = new ThroughStream(); + $stream = new LineSeparatedReader($input); + + $stream->on('data', $this->expectCallableNever()); + $stream->on('error', $this->expectCallableOnceWith($this->isInstanceOf('OverflowException'))); + $stream->on('close', $this->expectCallableOnce()); + + $input->write(str_repeat('a', 40000)); + $input->write(str_repeat('a', 40000)); + } + + public function testStreamDataWillEmitErrorAndCloseWhenDataContainsNewlineBehindMaximumBufferSize() + { + $input = new ThroughStream(); + $stream = new LineSeparatedReader($input); + + $stream->on('data', $this->expectCallableNever()); + $stream->on('error', $this->expectCallableOnceWith($this->isInstanceOf('OverflowException'))); + $stream->on('close', $this->expectCallableOnce()); + + $input->write(str_repeat('a', 65537) . "\n"); + } + + public function testStreamDataWillEmitDataAndErrorWhenDataBehindNewlineExceedsMaximumBufferSize() + { + $input = new ThroughStream(); + $stream = new LineSeparatedReader($input); + + $stream->on('data', $this->expectCallableOnceWith("hello!")); + $stream->on('error', $this->expectCallableOnceWith($this->isInstanceOf('OverflowException'))); + + $input->write("hello!\n" . str_repeat('a', 65537)); + } + public function testStreamEndWillEndWithoutDataWhenNoPreviousDataInBuffer() {