Skip to content

Commit 3bf10ac

Browse files
clueclue
committed
Handle parser errors by emitting error and closing connection
1 parent 8a05b9e commit 3bf10ac

File tree

4 files changed

+140
-36
lines changed

4 files changed

+140
-36
lines changed

‎src/Io/Buffer.php

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ public function prepend($str)
3737
*
3838
* @param int $len length in bytes, must be positive or zero
3939
* @return string
40-
* @throws \LogicException
40+
* @throws \UnderflowException
4141
*/
4242
public function read($len)
4343
{
@@ -53,7 +53,7 @@ public function read($len)
5353

5454
// ensure buffer size contains $len bytes by checking target buffer position
5555
if ($len < 0 || !isset($this->buffer[$this->bufferPos + $len - 1])) {
56-
throw new \LogicException('Not enough data in buffer to read ' . $len . ' bytes');
56+
throw new \UnderflowException('Not enough data in buffer to read ' . $len . ' bytes');
5757
}
5858
$buffer = \substr($this->buffer, $this->bufferPos, $len);
5959
$this->bufferPos += $len;
@@ -106,12 +106,12 @@ public function readBuffer($len)
106106
*
107107
* @param int $len length in bytes, must be positve and non-zero
108108
* @return void
109-
* @throws \LogicException
109+
* @throws \UnderflowException
110110
*/
111111
public function skip($len)
112112
{
113113
if ($len < 1 || !isset($this->buffer[$this->bufferPos + $len - 1])) {
114-
throw new \LogicException('Not enough data in buffer');
114+
throw new \UnderflowException('Not enough data in buffer');
115115
}
116116
$this->bufferPos += $len;
117117
}
@@ -220,13 +220,13 @@ public function readStringLen()
220220
* Reads string until NULL character
221221
*
222222
* @return string
223-
* @throws \LogicException
223+
* @throws \UnderflowException
224224
*/
225225
public function readStringNull()
226226
{
227227
$pos = \strpos($this->buffer, "\0", $this->bufferPos);
228228
if ($pos === false) {
229-
throw new \LogicException('Missing NULL character');
229+
throw new \UnderflowException('Missing NULL character');
230230
}
231231

232232
$ret = $this->read($pos - $this->bufferPos);

‎src/Io/Parser.php

Lines changed: 64 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
use React\MySQL\Commands\AuthenticateCommand;
66
use React\MySQL\Commands\QueryCommand;
77
use React\MySQL\Commands\QuitCommand;
8-
use React\MySQL\Exception;
8+
use React\MySQL\Exception as MysqlException;
99
use React\Stream\DuplexStreamInterface;
1010

1111
/**
@@ -25,6 +25,13 @@ class Parser
2525
const STATE_STANDBY = 0;
2626
const STATE_BODY = 1;
2727

28+
/**
29+
* The packet header always consists of 4 bytes, 3 bytes packet length + 1 byte sequence number
30+
*
31+
* @var integer
32+
*/
33+
const PACKET_SIZE_HEADER = 4;
34+
2835
/**
2936
* Keeps a reference to the command that is currently being processed.
3037
*
@@ -63,7 +70,20 @@ class Parser
6370
protected $serverStatus;
6471

6572
protected $rsState = 0;
66-
protected $pctSize = 0;
73+
74+
/**
75+
* Packet size expected in number of bytes
76+
*
77+
* Depending on `self::$state`, the Parser excepts either a packet header
78+
* (always 4 bytes) or the packet contents (n bytes determined by prior
79+
* packet header).
80+
*
81+
* @var int
82+
* @see self::$state
83+
* @see self::PACKET_SIZE_HEADER
84+
*/
85+
private $pctSize = self::PACKET_SIZE_HEADER;
86+
6787
protected $resultFields = [];
6888

6989
protected $insertId;
@@ -97,7 +117,7 @@ public function __construct(DuplexStreamInterface $stream, Executor $executor)
97117

98118
public function start()
99119
{
100-
$this->stream->on('data', [$this, 'parse']);
120+
$this->stream->on('data', [$this, 'handleData']);
101121
$this->stream->on('close', [$this, 'onClose']);
102122
}
103123

@@ -110,31 +130,53 @@ public function debug($message)
110130
}
111131
}
112132

113-
public function parse($data)
133+
/** @var string $data */
134+
public function handleData($data)
114135
{
115136
$this->buffer->append($data);
116-
packet:
117-
if ($this->state === self::STATE_STANDBY) {
118-
if ($this->buffer->length() < 4) {
137+
138+
if ($this->debug) {
139+
$this->debug('Received ' . strlen($data) . ' byte(s), buffer now has ' . ($len = $this->buffer->length()) . ' byte(s): ' . wordwrap(bin2hex($b = $this->buffer->read($len)), 2, ' ', true)); $this->buffer->append($b); // @codeCoverageIgnore
140+
}
141+
142+
while ($this->buffer->length() >= $this->pctSize) {
143+
if ($this->state === self::STATE_STANDBY) {
144+
$this->pctSize = $this->buffer->readInt3();
145+
//printf("packet size:%d\n", $this->pctSize);
146+
$this->state = self::STATE_BODY;
147+
$this->seq = $this->buffer->readInt1() + 1;
148+
}
149+
150+
$len = $this->buffer->length();
151+
if ($len < $this->pctSize) {
152+
$this->debug('Waiting for complete packet with ' . $len . '/' . $this->pctSize . ' bytes');
153+
119154
return;
120155
}
121156

122-
$this->pctSize = $this->buffer->readInt3();
123-
//printf("packet size:%d\n", $this->pctSize);
124-
$this->state = self::STATE_BODY;
125-
$this->seq = $this->buffer->readInt1() + 1;
126-
}
157+
$packet = $this->buffer->readBuffer($this->pctSize);
158+
$this->state = self::STATE_STANDBY;
159+
$this->pctSize = self::PACKET_SIZE_HEADER;
127160

128-
$len = $this->buffer->length();
129-
if ($len < $this->pctSize) {
130-
$this->debug('Waiting for complete packet with ' . $len . '/' . $this->pctSize . ' bytes');
161+
try {
162+
$this->parsePacket($packet);
163+
} catch (\UnderflowException $e) {
164+
$this->onError(new \UnexpectedValueException('Unexpected protocol error, received malformed packet: ' . $e->getMessage(), 0, $e));
165+
$this->stream->close();
166+
return;
167+
}
131168

132-
return;
169+
if ($packet->length() !== 0) {
170+
$this->onError(new \UnexpectedValueException('Unexpected protocol error, received malformed packet with ' . $packet->length() . ' unknown byte(s)'));
171+
$this->stream->close();
172+
return;
173+
}
133174
}
175+
}
134176

135-
$packet = $this->buffer->readBuffer($this->pctSize);
136-
$this->state = self::STATE_STANDBY;
137-
177+
/** @return void */
178+
private function parsePacket(Buffer $packet)
179+
{
138180
if ($this->debug) {
139181
$this->debug('Parse packet#' . $this->seq . ' with ' . ($len = $packet->length()) . ' bytes: ' . wordwrap(bin2hex($b = $packet->read($len)), 2, ' ', true)); $packet->append($b); // @codeCoverageIgnore
140182
}
@@ -146,7 +188,7 @@ public function parse($data)
146188
$this->phase = self::PHASE_AUTH_ERR;
147189

148190
$code = $packet->readInt2();
149-
$exception = new Exception($packet->read($packet->length()), $code);
191+
$exception = new MysqlException($packet->read($packet->length()), $code);
150192
$this->debug(sprintf("Error Packet:%d %s\n", $code, $exception->getMessage()));
151193

152194
// error during init phase also means we're not currently executing any command
@@ -186,7 +228,7 @@ public function parse($data)
186228
// error packet
187229
$code = $packet->readInt2();
188230
$packet->skip(6); // skip SQL state
189-
$exception = new Exception($packet->read($packet->length()), $code);
231+
$exception = new MysqlException($packet->read($packet->length()), $code);
190232
$this->debug(sprintf("Error Packet:%d %s\n", $code, $exception->getMessage()));
191233

192234
$this->onError($exception);
@@ -266,10 +308,6 @@ public function parse($data)
266308
}
267309
}
268310
}
269-
270-
// finished parsing packet, continue with next packet
271-
assert($packet->length() === 0);
272-
goto packet;
273311
}
274312

275313
private function onResultRow($row)
@@ -279,7 +317,7 @@ private function onResultRow($row)
279317
$command->emit('result', [$row]);
280318
}
281319

282-
private function onError(Exception $error)
320+
private function onError(\Exception $error)
283321
{
284322
$this->rsState = self::RS_STATE_HEADER;
285323
$this->resultFields = [];

‎tests/Io/BufferTest.php

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ public function testReadBeyondLimitThrows()
2222

2323
$buffer->append('hi');
2424

25-
$this->setExpectedException('LogicException');
25+
$this->setExpectedException('UnderflowException');
2626
$buffer->read(3);
2727
}
2828

@@ -71,7 +71,7 @@ public function testSkipZeroThrows()
7171

7272
$buffer->append('hi');
7373

74-
$this->setExpectedException('LogicException');
74+
$this->setExpectedException('UnderflowException');
7575
$buffer->skip(0);
7676
}
7777

@@ -81,7 +81,7 @@ public function testSkipBeyondLimitThrows()
8181

8282
$buffer->append('hi');
8383

84-
$this->setExpectedException('LogicException');
84+
$this->setExpectedException('UnderflowException');
8585
$buffer->skip(3);
8686
}
8787

@@ -214,7 +214,7 @@ public function testParseStringNullCharacterThrowsIfNullNotFound()
214214
$buffer = new Buffer();
215215
$buffer->append("hello");
216216

217-
$this->setExpectedException('LogicException');
217+
$this->setExpectedException('UnderflowException');
218218
$buffer->readStringNull();
219219
}
220220
}

‎tests/Io/ParserTest.php

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,72 @@ public function testReceivingErrorFrameForQueryAfterResultSetHeadersShouldEmitEr
149149
$this->assertEquals([], $ref->getValue($parser));
150150
}
151151

152+
public function testReceivingInvalidPacketWithMissingDataShouldEmitErrorAndCloseConnection()
153+
{
154+
$stream = new ThroughStream();
155+
156+
$command = new QueryCommand();
157+
$command->on('error', $this->expectCallableOnce());
158+
159+
$error = null;
160+
$command->on('error', function ($e) use (&$error) {
161+
$error = $e;
162+
});
163+
164+
$executor = new Executor();
165+
$executor->enqueue($command);
166+
167+
$parser = new Parser(new CompositeStream($stream, new ThroughStream()), $executor);
168+
$parser->start();
169+
170+
// hack to inject command as current command
171+
$ref = new \ReflectionProperty($parser, 'currCommand');
172+
$ref->setAccessible(true);
173+
$ref->setValue($parser, $command);
174+
175+
$stream->on('close', $this->expectCallableOnce());
176+
177+
$stream->write("\x32\0\0\0" . "\x0a" . "mysql\0" . str_repeat("\0", 43));
178+
179+
$this->assertTrue($error instanceof \UnexpectedValueException);
180+
$this->assertEquals('Unexpected protocol error, received malformed packet: Not enough data in buffer', $error->getMessage());
181+
$this->assertEquals(0, $error->getCode());
182+
$this->assertInstanceOf('UnderflowException', $error->getPrevious());
183+
}
184+
185+
public function testReceivingInvalidPacketWithExcessiveDataShouldEmitErrorAndCloseConnection()
186+
{
187+
$stream = new ThroughStream();
188+
189+
$command = new QueryCommand();
190+
$command->on('error', $this->expectCallableOnce());
191+
192+
$error = null;
193+
$command->on('error', function ($e) use (&$error) {
194+
$error = $e;
195+
});
196+
197+
$executor = new Executor();
198+
$executor->enqueue($command);
199+
200+
$parser = new Parser(new CompositeStream($stream, new ThroughStream()), $executor);
201+
$parser->start();
202+
203+
// hack to inject command as current command
204+
$ref = new \ReflectionProperty($parser, 'currCommand');
205+
$ref->setAccessible(true);
206+
$ref->setValue($parser, $command);
207+
208+
$stream->on('close', $this->expectCallableOnce());
209+
210+
$stream->write("\x34\0\0\0" . "\x0a" . "mysql\0" . str_repeat("\0", 45));
211+
212+
$this->assertTrue($error instanceof \UnexpectedValueException);
213+
$this->assertEquals('Unexpected protocol error, received malformed packet with 1 unknown byte(s)', $error->getMessage());
214+
$this->assertEquals(0, $error->getCode());
215+
$this->assertNull($error->getPrevious());
216+
}
217+
152218
public function testReceivingIncompleteErrorFrameDuringHandshakeShouldNotEmitError()
153219
{
154220
$stream = new ThroughStream();

0 commit comments

Comments
 (0)