add abuse checks and abstract class for logic

app/realtime.php

@@ -10,6 +10,7 @@ use Appwrite\Database\Database;
 use Appwrite\Database\Document;
 use Appwrite\Database\Validator\Authorization;
 use Appwrite\Extend\PDO;
+use Appwrite\Realtime\Realtime;
 use Swoole\WebSocket\Server;
 use Swoole\Http\Request;
 use Swoole\Process;
@@ -20,6 +21,8 @@ use Utopia\Config\Config;
 use Utopia\Registry\Registry;
 use Utopia\Swoole\Request as SwooleRequest;
 use PDO as PDONative;
+use Utopia\Abuse\Abuse;
+use Utopia\Abuse\Adapters\TimeLimit;
 
 /**
  * TODO List
@@ -127,28 +130,7 @@ $server->on("workerStart", function ($server, $workerId) use (&$subscriptions, &
                  */
                 $event = json_decode($payload, true);
 
-                $receivers = [];
-
-                foreach ($connections as $fd => $connection) {
-                    if ($connection['projectId'] !== $event['project']) {
-                        continue;
-                    }
-
-                    foreach ($connection['roles'] as $role) {
-                        if (\array_key_exists($role, $subscriptions[$event['project']])) {
-                            foreach ($event['data']['channels'] as $channel) {
-                                if (\array_key_exists($channel, $subscriptions[$event['project']][$role]) && \in_array($role, $event['permissions'])) {
-                                    foreach (array_keys($subscriptions[$event['project']][$role][$channel]) as $ids) {
-                                        $receivers[] = $ids;
-                                    }
-                                    break;
-                                }
-                            }
-                        }
-                    }
-                }
-
-                $receivers = array_keys(array_flip($receivers));
+                $receivers = Realtime::identifyReceivers($event, $connections, $subscriptions);
 
                 foreach ($receivers as $receiver) {
                     if ($server->exist($receiver) && $server->isEstablished($receiver)) {
@@ -260,28 +242,47 @@ $server->on('open', function (Server $server, Request $request) use (&$connectio
 
     $channels = $request->getQuery('channels', []);
 
+    /*
+     * Abuse Check
+     */
+    $timeLimit = new TimeLimit('url:{url},ip:{ip}', 60, 60, function () use ($register) {
+        return $register->get('db');
+    });
+    $timeLimit->setNamespace('app_' . $project->getId());
+    $timeLimit
+        ->setParam('{ip}', $request->getIP())
+        ->setParam('{url}', $request->getURI());
+
+    $abuse = new Abuse($timeLimit);
+
+    if ($abuse->check() && App::getEnv('_APP_OPTIONS_ABUSE', 'enabled') === 'enabled') {
+        $server->push($connection, 'Too many requests');
+        $server->close($connection);
+    }
+
+    /*
+     *  Project Check
+     */
     if (empty($project->getId())) {
         $server->push($connection, 'Missing or unknown project ID');
         $server->close($connection);
     }
 
-    if (empty($request->getQuery('channels', []))) {
-        $server->push($connection, 'Missing or unknown channels');
-        $server->close($connection);
-    }
+    Realtime::setUser($user);
 
     $roles = ['*', 'user:' . $user->getId(), 'role:' . (($user->isEmpty()) ? Auth::USER_ROLE_GUEST : Auth::USER_ROLE_MEMBER)];
     $channels = array_flip($channels);
 
-    \array_map(function ($node) use (&$roles) {
-        if (isset($node['teamId']) && isset($node['roles'])) {
-            $roles[] = 'team:' . $node['teamId'];
+    Realtime::parseChannels($channels);
+    Realtime::parseRoles($roles);
 
-            foreach ($node['roles'] as $nodeRole) { // Set all team roles
-                $roles[] = 'team:' . $node['teamId'] . '/' . $nodeRole;
-            }
-        }
-    }, $user->getAttribute('memberships', []));
+    /**
+     * Channels Check
+     */
+    if (empty($request->getQuery('channels', []))) {
+        $server->push($connection, 'Missing channels');
+        $server->close($connection);
+    }
 
     /**
      * Build Subscriptions Tree
@@ -315,6 +316,8 @@ $server->on('open', function (Server $server, Request $request) use (&$connectio
         'projectId' => $project->getId(),
         'roles' => $roles,
     ];
+
+    $server->push($connection, json_encode($channels));
 });
 
 $server->on('message', function (Server $server, Frame $frame) {

src/Appwrite/Event/Realtime.php

@@ -22,6 +22,11 @@ class Realtime
      */
     protected $channels = [];
 
+    /**
+     * @var array
+     */
+    protected $permissions = [];
+
     /**
      * @var Document
      */
@@ -106,22 +111,26 @@ class Realtime
         switch (true) {
             case strpos($this->event, 'account.') === 0:
                 $this->channels[] = 'account.' . $this->payload->getId();
+                $this->permissions = ['user:' . $this->payload->getId()];
 
                 break;
             case strpos($this->event, 'database.collections.') === 0:
                 $this->channels[] = 'collections';
                 $this->channels[] = 'collections.' . $this->payload->getId();
+                $this->permissions = $this->payload->getAttribute('$permissions.read');
 
                 break;
             case strpos($this->event, 'database.documents.') === 0:
                 $this->channels[] = 'documents';
                 $this->channels[] = 'collections.' . $this->payload->getAttribute('$collection') . '.documents';
                 $this->channels[] = 'documents.' . $this->payload->getId();
+                $this->permissions = $this->payload->getAttribute('$permissions.read');
 
                 break;
             case strpos($this->event, 'storage.') === 0:
                 $this->channels[] = 'files';
                 $this->channels[] = 'files.' . $this->payload->getId();
+                $this->permissions = $this->payload->getAttribute('$permissions.read');
 
                 break;
         }
@@ -141,7 +150,7 @@ class Realtime
         $redis->connect(App::getEnv('_APP_REDIS_HOST', ''), App::getEnv('_APP_REDIS_PORT', ''));
         $redis->publish('realtime', json_encode([
             'project' => $this->project,
-            'permissions' => $this->payload->getAttribute('$permissions.read'),
+            'permissions' => $this->permissions,
             'data' => [
                 'event' => $this->event,
                 'channels' => $this->channels,

src/Appwrite/Realtime/Realtime.php

@@ -0,0 +1,91 @@
+<?php
+
+namespace Appwrite\Realtime;
+
+use Appwrite\Database\Document;
+
+class Realtime
+{
+    /**
+     * @var Document $user
+     */
+    static $user;
+
+    /**
+     * @param Document $user
+     */
+    static function setUser(Document $user)
+    {
+        self::$user = $user;
+    }
+
+    /**
+     * @param array $channels
+     */
+    static function parseChannels(array &$channels)
+    {
+        foreach ($channels as $key => $value) {
+            if (strpos($key, 'account.') === 0) {
+                unset($channels[$key]);
+            } elseif ($key === 'account') {
+                if (!empty(self::$user->getId())) {
+                    $channels['account.' . self::$user->getId()] = $value;
+                }
+                unset($channels['account']);
+            }
+        }
+
+        if (\array_key_exists('account', $channels)) {
+            if (self::$user->getId()) {
+                $channels['account.' . self::$user->getId()] = $channels['account'];
+            }
+            unset($channels['account']);
+        }
+    }
+
+    /**
+     * @param array $roles
+     */
+    static function parseRoles(array &$roles)
+    {
+        \array_map(function ($node) use (&$roles) {
+            if (isset($node['teamId']) && isset($node['roles'])) {
+                $roles[] = 'team:' . $node['teamId'];
+
+                foreach ($node['roles'] as $nodeRole) { // Set all team roles
+                    $roles[] = 'team:' . $node['teamId'] . '/' . $nodeRole;
+                }
+            }
+        }, self::$user->getAttribute('memberships', []));
+    }
+
+    /**
+     * @param array $event
+     * @param array $connections
+     * @param array $subscriptions
+     */
+    static function identifyReceivers(array &$event, array &$connections, array &$subscriptions)
+    {
+        $receivers = [];
+        foreach ($connections as $fd => $connection) {
+            if ($connection['projectId'] !== $event['project']) {
+                continue;
+            }
+
+            foreach ($connection['roles'] as $role) {
+                if (\array_key_exists($role, $subscriptions[$event['project']])) {
+                    foreach ($event['data']['channels'] as $channel) {
+                        if (\array_key_exists($channel, $subscriptions[$event['project']][$role]) && \in_array($role, $event['permissions'])) {
+                            foreach (array_keys($subscriptions[$event['project']][$role][$channel]) as $ids) {
+                                $receivers[] = $ids;
+                            }
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        return array_keys(array_flip($receivers));
+    }
+}