[1.x] WebSocket Rate Limiting Implementation

composer.json

@@ -22,6 +22,7 @@
         "php": "^8.2",
         "clue/redis-react": "^2.6",
         "guzzlehttp/psr7": "^2.6",
+        "illuminate/cache": "^10.47|^11.0|^12.0",
         "illuminate/console": "^10.47|^11.0|^12.0",
         "illuminate/contracts": "^10.47|^11.0|^12.0",
         "illuminate/http": "^10.47|^11.0|^12.0",

config/reverb.php

@@ -90,4 +90,19 @@
 
     ],
 
+    /*
+    |--------------------------------------------------------------------------
+    | WebSocket Rate Limiting
+    |--------------------------------------------------------------------------
+    |
+    | Configure rate limiting for WebSocket messages. This helps prevent
+    | abuse and ensures fair usage of the WebSocket server.
+    |
+    */
+    'rate_limiting' => [
+        'enabled' => env('REVERB_RATE_LIMITING_ENABLED', false),
+        'max_attempts' => env('REVERB_RATE_LIMITING_MAX_ATTEMPTS', 60),
+        'decay_seconds' => env('REVERB_RATE_LIMITING_DECAY_SECONDS', 10),
+        'terminate_on_limit' => env('REVERB_RATE_LIMITING_TERMINATE_ON_LIMIT', true),
+    ],
 ];

src/Protocols/Pusher/Exceptions/RateLimitExceededException.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace Laravel\Reverb\Protocols\Pusher\Exceptions;
+
+use Symfony\Component\HttpFoundation\Response as HttpFoundationResponse;
+
+/**
+ * Exception thrown when a client exceeds the rate limit for WebSocket connections.
+ */
+class RateLimitExceededException extends PusherException
+{
+    /**
+     * The error code associated with the exception.
+     *
+     * @var int
+     */
+    protected $code = HttpFoundationResponse::HTTP_TOO_MANY_REQUESTS;
+
+    /**
+     * The error message associated with the exception.
+     *
+     * @var string
+     */
+    protected $message = 'Too Many Requests';
+}

src/Protocols/Pusher/Server.php

@@ -14,13 +14,15 @@
 use Ratchet\RFC6455\Messaging\Frame;
 use Ratchet\RFC6455\Messaging\FrameInterface;
 use Throwable;
+use Laravel\Reverb\RateLimiting\RateLimitManager;
+use Illuminate\Cache\RateLimiter;
 
 class Server
 {
     /**
      * Create a new server instance.
      */
-    public function __construct(protected ChannelManager $channels, protected EventHandler $handler)
+    public function __construct(protected ChannelManager $channels, protected EventHandler $handler, protected RateLimitManager $rateLimitManager) 
     {
         //
     }
@@ -54,6 +56,8 @@ public function message(Connection $from, string $message): void
         $from->touch();
 
         try {
+            $this->rateLimiter($from);
+
             $event = json_decode($message, associative: true, flags: JSON_THROW_ON_ERROR);
 
             Validator::make($event, ['event' => ['required', 'string']])->validate();
@@ -75,6 +79,20 @@ public function message(Connection $from, string $message): void
         }
     }
 
+    /**
+     * Rate limit the incoming message.
+     * 
+     * @throws \Laravel\Reverb\Protocols\Pusher\Exceptions\RateLimitExceededException
+     */
+    protected function rateLimiter(Connection $from): void
+    {
+        if (!config('reverb.rate_limiting.enabled')) {
+            return;
+        }
+
+        $this->rateLimitManager->handle($from);
+    }
+
     /**
      * Handle a low-level WebSocket control frame.
      */

src/RateLimiting/RateLimitManager.php

@@ -0,0 +1,119 @@
+<?php
+
+namespace Laravel\Reverb\RateLimiting;
+
+use Illuminate\Cache\RateLimiter;
+use Laravel\Reverb\Contracts\Connection;
+use Laravel\Reverb\Protocols\Pusher\Exceptions\RateLimitExceededException;
+
+class RateLimitManager
+{
+    /**
+     * Create a new rate limit manager instance.
+     */
+    public function __construct(
+        protected RateLimiter $limiter, 
+        protected int $maxAttempts, 
+        protected int $decaySeconds,
+        protected bool $terminateOnLimit = true
+    ) {
+        //
+    }
+
+    /**
+     * Handle the incoming message and apply rate limiting.
+     * 
+     * @throws \Laravel\Reverb\Protocols\Pusher\Exceptions\RateLimitExceededException
+     */
+    public function handle(Connection $connection): void
+    {
+        $key = $this->resolveRequestSignature($connection);
+
+        if ($this->limiter->tooManyAttempts($key, $this->maxAttempts)) {
+            if ($this->terminateOnLimit) {
+                $connection->terminate();
+            }
+
+            throw new RateLimitExceededException();
+        }
+
+        $this->limiter->hit($key, $this->decaySeconds);
+    }
+
+    /**
+     * Check if a connection would exceed rate limits without incrementing the counter.
+     */
+    public function wouldExceedRateLimit(Connection $connection): bool
+    {
+        $key = $this->resolveRequestSignature($connection);
+
+        return $this->limiter->tooManyAttempts($key, $this->maxAttempts);
+    }
+
+    /**
+     * Get the number of remaining attempts for a connection.
+     */
+    public function remainingAttempts(Connection $connection): int
+    {
+        $key = $this->resolveRequestSignature($connection);
+
+        return $this->limiter->remaining($key, $this->maxAttempts);
+    }
+
+    /**
+     * Get the number of seconds until the rate limit is reset.
+     *
+     * @param  \Laravel\Reverb\Contracts\Connection  $connection
+     * @return int
+     */
+    public function availableIn(Connection $connection): int
+    {
+        $key = $this->resolveRequestSignature($connection);
+
+        return $this->limiter->availableIn($key);
+    }
+
+    /**
+     * Clear the rate limiting for the given connection.
+     *
+     * @param  \Laravel\Reverb\Contracts\Connection  $connection
+     * @return void
+     */
+    public function clear(Connection $connection): void
+    {
+        $key = $this->resolveRequestSignature($connection);
+
+        $this->limiter->clear($key);
+    }
+
+    /**
+     * Get the maximum number of attempts allowed.
+     *
+     * @return int
+     */
+    public function getMaxAttempts(): int
+    {
+        return $this->maxAttempts;
+    }
+
+    /**
+     * Get the decay seconds for the rate limiter.
+     *
+     * @return int
+     */
+    public function getDecaySeconds(): int
+    {
+        return $this->decaySeconds;
+    }
+
+    /**
+     * Resolve the request signature for the given connection.
+     */
+    protected function resolveRequestSignature(Connection $connection): string
+    {
+        return sha1(implode('|', [
+            $connection->id(),
+            $connection->app()->id(),
+        ]));
+    }
+} 

src/ReverbServiceProvider.php

@@ -2,10 +2,12 @@
 
 namespace Laravel\Reverb;
 
+use Illuminate\Cache\RateLimiter;
 use Illuminate\Support\ServiceProvider;
 use Laravel\Reverb\Console\Commands\InstallCommand;
 use Laravel\Reverb\Contracts\Logger;
 use Laravel\Reverb\Loggers\NullLogger;
+use Laravel\Reverb\RateLimiting\RateLimitManager;
 use Laravel\Reverb\Pulse\Livewire;
 use Livewire\LivewireManager;
 
@@ -25,6 +27,15 @@ public function register(): void
         $this->app->singleton(ServerProviderManager::class);
 
         $this->app->make(ServerProviderManager::class)->register();
+
+        $this->app->singleton(RateLimitManager::class, function ($app) {
+            return new RateLimitManager(
+                $app->make(RateLimiter::class),
+                config('reverb.rate_limiting.max_attempts', 10),
+                config('reverb.rate_limiting.decay_seconds', 10),
+                config('reverb.rate_limiting.terminate_on_limit', true)
+            );
+        });
     }
 
     /**

tests/FakeConnection.php

@@ -32,6 +32,13 @@ class FakeConnection extends BaseConnection
      */
     public $id;
 
+    /**
+     * Whether the connection has been terminated.
+     *
+     * @var bool
+     */
+    public $wasTerminated = false;
+
     /**
      * Create a new fake connection instance.
      */
@@ -108,7 +115,7 @@ public function control(string $type = Frame::OP_PING): void {}
      */
     public function terminate(): void
     {
-        //
+        $this->wasTerminated = true;
     }
 
     /**
@@ -142,4 +149,12 @@ public function assertHasBeenPinged(): void
     {
         Assert::assertTrue($this->hasBeenPinged);
     }
+
+    /**
+     * Assert the connection has been terminated.
+     */
+    public function assertHasBeenTerminated(): void
+    {
+        Assert::assertTrue($this->wasTerminated);
+    }
 }

tests/Feature/Protocols/Pusher/RateLimiter/RateLimitingTest.php

@@ -0,0 +1,85 @@
+<?php
+
+use Illuminate\Cache\RateLimiter;
+use Illuminate\Support\Facades\Config;
+use Laravel\Reverb\RateLimiting\RateLimitManager;
+use Laravel\Reverb\Protocols\Pusher\Exceptions\RateLimitExceededException;
+use Laravel\Reverb\Tests\FakeConnection;
+use Laravel\Reverb\Tests\ReverbTestCase;
+use Symfony\Component\HttpFoundation\Response as HttpFoundationResponse;
+
+uses(ReverbTestCase::class);
+
+beforeEach(function () {
+    Config::set('reverb.rate_limiting.enabled', true);
+    Config::set('reverb.rate_limiting.max_attempts', 10);
+    Config::set('reverb.rate_limiting.decay_seconds', 10);
+});
+
+it('allows messages when rate limiting is disabled', function () {
+    Config::set('reverb.rate_limiting.enabled', false);
+
+    $rateLimitManager = Mockery::mock(RateLimitManager::class);
+    $rateLimitManager->shouldNotReceive('handle');
+
+    $this->app->instance(RateLimitManager::class, $rateLimitManager);
+
+    $connection = new FakeConnection();
+    $this->app->make('Laravel\Reverb\Protocols\Pusher\Server')->message($connection, json_encode([
+        'event' => 'client-test-event',
+        'data' => ['message' => 'Hello World'],
+    ]));
+});
+
+it('rate limits messages when enabled', function () {
+    $rateLimitManager = Mockery::mock(RateLimitManager::class);
+    $rateLimitManager->shouldReceive('handle')
+        ->once()
+        ->andReturn(null);
+
+    $this->app->instance(RateLimitManager::class, $rateLimitManager);
+
+    $connection = new FakeConnection();
+    $this->app->make('Laravel\Reverb\Protocols\Pusher\Server')->message($connection, json_encode([
+        'event' => 'client-test-event',
+        'data' => ['message' => 'Hello World'],
+    ]));
+});
+
+it('blocks messages when over the rate limit', function () {
+    $rateLimitManager = Mockery::mock(RateLimitManager::class);
+    $rateLimitManager->shouldReceive('handle')
+        ->once()
+        ->andThrow(new RateLimitExceededException());
+
+    $this->app->instance(RateLimitManager::class, $rateLimitManager);
+
+    $connection = new FakeConnection();
+    $this->app->make('Laravel\Reverb\Protocols\Pusher\Server')->message($connection, json_encode([
+        'event' => 'client-test-event',
+        'data' => ['message' => 'Hello World'],
+    ]));
+
+    $connection->assertReceived([
+        'event' => 'pusher:error',
+        'data' => json_encode([
+            'code' => HttpFoundationResponse::HTTP_TOO_MANY_REQUESTS,
+            'message' => 'Too Many Requests',
+        ]),
+    ]);
+});
+
+it('uses correct configuration values', function () {
+    Config::set('reverb.rate_limiting.max_attempts', 5);
+    Config::set('reverb.rate_limiting.decay_seconds', 20);
+
+    $realRateLimiter = app(RateLimiter::class);
+    $realManager = new RateLimitManager(
+        $realRateLimiter,
+        Config::get('reverb.rate_limiting.max_attempts'),
+        Config::get('reverb.rate_limiting.decay_seconds')
+    );
+
+    expect($realManager->getMaxAttempts())->toBe(5);
+    expect($realManager->getDecaySeconds())->toBe(20);
+});