Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion src/Io/LineSeparatedReader.php
Original file line number Diff line number Diff line change
Expand Up @@ -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)
{
Expand Down Expand Up @@ -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()
Expand Down
6 changes: 6 additions & 0 deletions src/SshProcessConnector.php
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
6 changes: 6 additions & 0 deletions src/SshSocksConnector.php
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
16 changes: 16 additions & 0 deletions tests/IntegrationSshProcessConnectorTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down
16 changes: 16 additions & 0 deletions tests/IntegrationSshSocksConnectorTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
48 changes: 48 additions & 0 deletions tests/Io/LineSeparatedReaderTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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()
{
Expand Down