Add rate limits to socket messages

Author: skadefro

OpenFlow/src/Config.ts

@@ -43,6 +43,8 @@ export class Config {
         Config.api_credential_cache_seconds = parseInt(Config.getEnv("api_credential_cache_seconds", "60000"));
         Config.api_rate_limit_points = parseInt(Config.getEnv("api_rate_limit_points", "40"));
         Config.api_rate_limit_duration = parseInt(Config.getEnv("api_rate_limit_duration", "5"));
+        Config.socket_rate_limit_points = parseInt(Config.getEnv("socket_rate_limit_points", "10"));
+        Config.socket_rate_limit_duration = parseInt(Config.getEnv("socket_rate_limit_duration", "1"));
 
         Config.client_heartbeat_timeout = parseInt(Config.getEnv("client_heartbeat_timeout", "60"));
 
@@ -119,6 +121,8 @@ export class Config {
     public static api_credential_cache_seconds: number = parseInt(Config.getEnv("api_credential_cache_seconds", "60000"));
     public static api_rate_limit_points: number = parseInt(Config.getEnv("api_rate_limit_points", "40"));
     public static api_rate_limit_duration: number = parseInt(Config.getEnv("api_rate_limit_duration", "5"));
+    public static socket_rate_limit_points: number = parseInt(Config.getEnv("socket_rate_limit_points", "10"));
+    public static socket_rate_limit_duration: number = parseInt(Config.getEnv("socket_rate_limit_duration", "1"));
 
     public static client_heartbeat_timeout: number = parseInt(Config.getEnv("client_heartbeat_timeout", "60"));
 

OpenFlow/src/LoginProvider.ts

@@ -20,7 +20,7 @@ import { Audit } from "./Audit";
 import * as saml from "saml20";
 const multer = require('multer');
 const GridFsStorage = require('multer-gridfs-storage');
-const { RateLimiterMemory, RateLimiterRes } = require('rate-limiter-flexible')
+const { RateLimiterMemory } = require('rate-limiter-flexible')
 import { GridFSBucket, ObjectID, Db, Cursor, Binary } from "mongodb";
 import { Base, User, NoderedUtil, TokenUser, WellknownIds, Rights, Role } from "openflow-api";
 import { DBHelper } from "./DBHelper";
@@ -35,11 +35,11 @@ const rateLimiter = (req: express.Request, res: express.Response, next: express.
     BaseRateLimiter
         .consume(req.ip)
         .then((e) => {
-            // console.log("NO_RATE_LIMIT consumedPoints: " + e.consumedPoints + " remainingPoints: " + e.remainingPoints);
+            // console.log("API_O_RATE_LIMIT consumedPoints: " + e.consumedPoints + " remainingPoints: " + e.remainingPoints);
             next();
         })
         .catch((e) => {
-            console.log("RATE_LIMIT consumedPoints: " + e.consumedPoints + " remainingPoints: " + e.remainingPoints + " msBeforeNext: " + e.msBeforeNext);
+            console.log("API_RATE_LIMIT consumedPoints: " + e.consumedPoints + " remainingPoints: " + e.remainingPoints + " msBeforeNext: " + e.msBeforeNext);
             res.status(429).json({ response: 'RATE_LIMIT' });
         });
 };

OpenFlow/src/WebSocketServerClient.ts

@@ -6,6 +6,12 @@ import { Config } from "./Config";
 import { amqpwrapper, QueueMessageOptions, amqpqueue } from "./amqpwrapper";
 import { NoderedUtil, Base, InsertOneMessage, QueueMessage, MapReduceMessage, QueryMessage, UpdateOneMessage, UpdateManyMessage, DeleteOneMessage, User, mapFunc, reduceFunc, finalizeFunc, QueuedMessage, QueuedMessageCallback, WatchEventMessage } from "openflow-api";
 import { ChangeStream } from "mongodb";
+const { RateLimiterMemory } = require('rate-limiter-flexible')
+
+const BaseRateLimiter = new RateLimiterMemory({
+    points: Config.socket_rate_limit_points,
+    duration: Config.socket_rate_limit_duration,
+});
 
 interface IHashTable<T> {
     [key: string]: T;
@@ -70,7 +76,7 @@ export class WebSocketServerClient {
         }
         logger.info("new client " + this.id + " from " + this.remoteip);
         socketObject.on("open", (e: Event): void => this.open(e));
-        socketObject.on("message", (e: string): void => this.message(e)); // e: MessageEvent
+        socketObject.on("message", (e: string): void => (this.message(e) as any)); // e: MessageEvent
         socketObject.on("error", (e: Event): void => this.error(e));
         socketObject.on("close", (e: CloseEvent): void => this.close(e));
     }
@@ -129,15 +135,30 @@ export class WebSocketServerClient {
             }
         }
     }
-    private message(message: string): void { // e: MessageEvent
+    private _message(message: string): void {
+        //this._logger.silly("WebSocket message received " + message);
+        let msg: SocketMessage = SocketMessage.fromjson(message);
+        this._logger.silly("WebSocket message received id: " + msg.id + " index: " + msg.index + " count: " + msg.count);
+        this._receiveQueue.push(msg);
+        if ((msg.index + 1) >= msg.count) this.ProcessQueue();
+    }
+    private async message(message: string): Promise<void> {
+        let username: string = "Unknown";
         try {
-            //this._logger.silly("WebSocket message received " + message);
-            let msg: SocketMessage = SocketMessage.fromjson(message);
-            this._logger.silly("WebSocket message received id: " + msg.id + " index: " + msg.index + " count: " + msg.count);
-            this._receiveQueue.push(msg);
-            this.ProcessQueue();
+            try {
+
+                if (!NoderedUtil.IsNullUndefinded(this.user)) { username = this.user.username; }
+
+                // await this.consume("me");
+                var res = await BaseRateLimiter.consume("me");
+                // console.log("SOCKET_NO_RATE_LIMIT consumedPoints: " + res.consumedPoints + " remainingPoints: " + res.remainingPoints);
+                this._message(message);
+            } catch (error) {
+                this._logger.debug("[" + username + "/" + this.clientagent + "/" + this.id + "] SOCKET_RATE_LIMIT consumedPoints: " + error.consumedPoints + " remainingPoints: " + error.remainingPoints + " msBeforeNext: " + error.msBeforeNext);
+                setTimeout(() => { this.message(message); }, Math.floor(Math.random() * 10) * 100);
+            }
         } catch (error) {
-            this._logger.error("WebSocket error encountered " + (error.message ? error.message : error));
+            this._logger.error("[" + username + "/" + this.clientagent + "/" + this.id + "] WebSocket error encountered " + (error.message ? error.message : error));
             const errormessage: Message = new Message(); errormessage.command = "error"; errormessage.data = (error.message ? error.message : error);
             this._socketObject.send(JSON.stringify(errormessage));
         }
@@ -163,7 +184,7 @@ export class WebSocketServerClient {
         if (this._socketObject != null) {
             try {
                 this._socketObject.removeListener("open", (e: Event): void => this.open(e));
-                this._socketObject.removeListener("message", (e: string): void => this.message(e)); // e: MessageEvent
+                this._socketObject.removeListener("message", (e: string): void => (this.message(e) as any)); // e: MessageEvent
                 this._socketObject.removeListener("error", (e: Event): void => this.error(e));
                 this._socketObject.removeListener("close", (e: CloseEvent): void => this.close(e));
             } catch (error) {

OpenFlowNodeRED/package.json

@@ -1,6 +1,6 @@
 {
   "name": "openflow-nodered",
-  "version": "1.1.60",
+  "version": "1.1.62",
   "description": "Simple wrapper around NodeRed, RabbitMQ and MongoDB to support a more scaleable NodeRed implementation.\r Also the \"backend\" for [OpenRPA](https://github.com/skadefro/OpenRPA)",
   "main": "index.js",
   "scripts": {

OpenFlowNodeRED/src/nodered/nodes/api_nodes.ts

@@ -351,7 +351,7 @@ export class api_update {
                     }
                     Promises.push(NoderedUtil.UpdateOne(this.config.collection, null, element, this.config.writeconcern, this.config.journal, msg.jwt));
                 }
-                this.node.status({ fill: "blue", shape: "dot", text: "processing " + y + " to " + (y + 49) });
+                this.node.status({ fill: "blue", shape: "dot", text: y + " to " + (y + 49) + " of " + data.length });
                 const tempresults = await Promise.all(Promises.map(p => p.catch(e => e)));
                 results = results.concat(tempresults);
                 Promises = [];
@@ -430,7 +430,7 @@ export class api_addorupdate {
                     }
                     Promises.push(NoderedUtil.InsertOrUpdateOne(this.config.collection, element, this.config.uniqeness, this.config.writeconcern, this.config.journal, msg.jwt));
                 }
-                this.node.status({ fill: "blue", shape: "dot", text: "processing " + y + " to " + (y + 49) });
+                this.node.status({ fill: "blue", shape: "dot", text: y + " to " + (y + 49) + " of " + data.length });
                 const tempresults = await Promise.all(Promises.map(p => p.catch(e => e)));
                 results = results.concat(tempresults);
                 Promises = [];
@@ -496,7 +496,7 @@ export class api_delete {
                     if (NoderedUtil.isObject(element)) { id = element._id; }
                     Promises.push(NoderedUtil.DeleteOne(this.config.collection, id, msg.jwt));
                 }
-                this.node.status({ fill: "blue", shape: "dot", text: "processing " + y + " to " + (y + 49) });
+                this.node.status({ fill: "blue", shape: "dot", text: y + " to " + (y + 49) + " of " + data.length });
                 const tempresults = await Promise.all(Promises.map(p => p.catch(e => e)));
                 results = results.concat(tempresults);
                 Promises = [];

VERSION

@@ -1 +1 @@
-1.1.60
+1.1.62

docker-compose-toolbox.yml

@@ -82,7 +82,7 @@ services:
       - "traefik.http.routers.nodered.rule=Host(`nodered1.toolbox.openrpa.dk`)"
       - "traefik.http.routers.nodered.entrypoints=web"
       - "traefik.http.services.nodered.loadbalancer.server.port=1880"
-    image: "cloudhack/openflownodered:1.1.60"
+    image: "cloudhack/openflownodered:1.1.62"
     container_name: "nodered"
     environment:
       #      - nodered_id=1

docker-compose-traefik-letsencrypt.yml

@@ -107,7 +107,7 @@ services:
       - "traefik.http.routers.nodered.entrypoints=web,websecure"
       - "traefik.http.services.nodered.loadbalancer.server.port=1880"
       - "traefik.http.routers.nodered.tls.certresolver=myresolver"
-    image: "cloudhack/openflownodered:1.1.60"
+    image: "cloudhack/openflownodered:1.1.62"
     container_name: "nodered"
     environment:
       #      - nodered_id=1

docker-compose-traefik.yml

@@ -82,7 +82,7 @@ services:
       - "traefik.http.routers.nodered.rule=Host(`nodered1.localhost.openrpa.dk`)"
       - "traefik.http.routers.nodered.entrypoints=web"
       - "traefik.http.services.nodered.loadbalancer.server.port=1880"
-    image: "cloudhack/openflownodered:1.1.60"
+    image: "cloudhack/openflownodered:1.1.62"
     container_name: "nodered"
     environment:
       #      - nodered_id=1

docker-compose.yml

@@ -52,7 +52,7 @@ services:
       - "80:80"
       - "5858:5858"
   nodered:
-    image: "cloudhack/openflownodered:1.1.60"
+    image: "cloudhack/openflownodered:1.1.62"
     environment:
       #      - nodered_id=1
       - nodered_sa=nodered1