[1.x] Ensures large requests are handled correctly (#87)

* add payload size tests

* wip

* correctly buffer messages

* add buffer test

config/reverb.php

@@ -35,6 +35,7 @@
             'options' => [
                 'tls' => [],
             ],
+            'max_request_size' => env('REVERB_MAX_REQUEST_SIZE', 10_000),
             'scaling' => [
                 'enabled' => env('REVERB_SCALING_ENABLED', false),
                 'channel' => env('REVERB_SCALING_CHANNEL', 'reverb'),

src/Servers/Reverb/Console/Commands/StartServer.php

@@ -56,6 +56,7 @@ public function handle(): void
             $host = $this->option('host') ?: $config['host'],
             $port = $this->option('port') ?: $config['port'],
             $hostname = $this->option('hostname') ?: $config['hostname'],
+            $config['max_request_size'] ?? 10_000,
             $config['options'] ?? [],
             loop: $loop
         );

src/Servers/Reverb/Factory.php

@@ -35,8 +35,15 @@ class Factory
     /**
      * Create a new WebSocket server instance.
      */
-    public static function make(string $host = '0.0.0.0', string $port = '8080', ?string $hostname = null, array $options = [], string $protocol = 'pusher', ?LoopInterface $loop = null): HttpServer
-    {
+    public static function make(
+        string $host = '0.0.0.0',
+        string $port = '8080',
+        ?string $hostname = null,
+        int $maxRequestSize = 10_000,
+        array $options = [],
+        string $protocol = 'pusher',
+        ?LoopInterface $loop = null
+    ): HttpServer {
         $loop = $loop ?: Loop::get();
 
         $router = match ($protocol) {
@@ -56,6 +63,7 @@ public static function make(string $host = '0.0.0.0', string $port = '8080', ?st
         return new HttpServer(
             new SocketServer($uri, $options, $loop),
             $router,
+            $maxRequestSize,
             $loop
         );
     }

src/Servers/Reverb/Http/Request.php

@@ -15,28 +15,31 @@ class Request
      */
     const EOM = "\r\n\r\n";
 
-    /**
-     * The maximum number of allowed bytes for the request.
-     *
-     * @var int
-     */
-    const MAX_SIZE = 10_000;
-
     /**
      * Turn the raw message into a Psr7 request.
      */
-    public static function from(string $message, Connection $connection): ?RequestInterface
+    public static function from(string $message, Connection $connection, int $maxRequestSize): ?RequestInterface
     {
         $connection->appendToBuffer($message);
 
-        if ($connection->bufferLength() > static::MAX_SIZE) {
-            throw new OverflowException('Maximum HTTP buffer size of '.static::MAX_SIZE.'exceeded.');
+        if ($connection->bufferLength() > $maxRequestSize) {
+            throw new OverflowException('Maximum HTTP buffer size of '.$maxRequestSize.'exceeded.');
         }
 
         if (static::isEndOfMessage($buffer = $connection->buffer())) {
+            $request = Message::parseRequest($buffer);
+
+            if (! $contentLength = $request->getHeader('Content-Length')) {
+                return $request;
+            }
+
+            if ($connection->bufferLength() < $contentLength[0] ?? 0) {
+                return null;
+            }
+
             $connection->clearBuffer();
 
-            return Message::parseRequest($buffer);
+            return $request;
         }
 
         return null;

src/Servers/Reverb/Http/Server.php

@@ -19,7 +19,7 @@ class Server
     /**
      * Create a new Http server instance.
      */
-    public function __construct(protected ServerInterface $socket, protected Router $router, protected ?LoopInterface $loop = null)
+    public function __construct(protected ServerInterface $socket, protected Router $router, protected int $maxRequestSize, protected ?LoopInterface $loop = null)
     {
         gc_disable();
 
@@ -67,7 +67,7 @@ protected function handleRequest(string $message, Connection $connection): void
     protected function createRequest(string $message, Connection $connection): ?RequestInterface
     {
         try {
-            $request = Request::from($message, $connection);
+            $request = Request::from($message, $connection, $this->maxRequestSize);
         } catch (OverflowException $e) {
             $this->close($connection, 413, 'Payload too large.');
         } catch (Throwable $e) {

tests/Feature/Protocols/Pusher/Reverb/EventsControllerTest.php

@@ -144,3 +144,24 @@
     expect($response->getStatusCode())->toBe(200);
     expect($response->getBody()->getContents())->toBe('{"channels":{"presence-test-channel-one":{},"test-channel-two":{"subscription_count":1}}}');
 });
+
+it('cannot trigger an event over the max message size', function () {
+    await($this->signedPostRequest('events', [
+        'name' => 'NewEvent',
+        'channel' => 'test-channel',
+        'data' => json_encode([str_repeat('a', 10_100)]),
+    ]));
+})->expectExceptionMessage('HTTP status code 413 (Request Entity Too Large)');
+
+it('can trigger an event within the max message size', function () {
+    $this->stopServer();
+    $this->startServer(maxRequestSize: 20_000);
+    $response = await($this->signedPostRequest('events', [
+        'name' => 'NewEvent',
+        'channel' => 'test-channel',
+        'data' => json_encode([str_repeat('a', 10_100)]),
+    ], appId: '654321'));
+
+    expect($response->getStatusCode())->toBe(200);
+    expect($response->getBody()->getContents())->toBe('{}');
+});

tests/Feature/Protocols/Pusher/Reverb/ServerTest.php

@@ -429,3 +429,44 @@
 
     expect($response)->toBe('{"event":"pusher:error","data":"{\"code\":4009,\"message\":\"Connection is unauthorized\"}"}');
 });
+
+it('rejects messages over the max allowed size', function () {
+    $connection = connect();
+
+    $response = send([
+        'event' => 'pusher:subscribe',
+        'data' => [
+            'channel' => 'my-channel',
+            'channel_data' => json_encode([str_repeat('a', 10_100)]),
+        ],
+    ], $connection);
+
+    expect($response)->toBe('Maximum message size exceeded');
+});
+
+it('allows message within the max allowed size', function () {
+    $connection = connect(key: 'reverb-key-2');
+
+    $response = send([
+        'event' => 'pusher:subscribe',
+        'data' => [
+            'channel' => 'my-channel',
+            'channel_data' => json_encode([str_repeat('a', 20_000)]),
+        ],
+    ], $connection);
+
+    expect($response)->toBe('{"event":"pusher_internal:subscription_succeeded","channel":"my-channel"}');
+});
+
+it('buffers large requests correctly', function () {
+    $this->stopServer();
+    $this->startServer(maxRequestSize: 200_000);
+    $response = await($this->signedPostRequest('events', [
+        'name' => 'NewEvent',
+        'channel' => 'test-channel',
+        'data' => json_encode([str_repeat('a', 150_000)]),
+    ], appId: '654321'));
+
+    expect($response->getStatusCode())->toBe(200);
+    expect($response->getBody()->getContents())->toBe('{}');
+});

tests/ReverbTestCase.php

@@ -81,10 +81,10 @@ public function usingRedis(): void
     /**
      * Start the WebSocket server.
      */
-    public function startServer(string $host = '0.0.0.0', string $port = '8080'): void
+    public function startServer(string $host = '0.0.0.0', string $port = '8080', int $maxRequestSize = 10_000): void
     {
         $this->resetFiber();
-        $this->server = Factory::make($host, $port, loop: $this->loop);
+        $this->server = Factory::make($host, $port, maxRequestSize: $maxRequestSize, loop: $this->loop);
     }
 
     /**