First draft: Connection limiter

app/controllers/api/rmq_utils_controller.rb

@@ -5,6 +5,42 @@ module Api
   # Any other response results in denial.
   # Results are cached for 10 minutes to prevent too many requests to the API.
   class RmqUtilsController < Api::AbstractController
+    class BrokerConnectionLimiter
+      class RateLimit < StandardError; end
+
+      attr_reader :cache
+
+      CACHE_KEY_TPL = "mqtt_limiter:%s"
+      TTL = 60 * 10 # Ten Minutes
+      PER_DEVICE_MAX = 20
+      MAX_GUEST_COUNT = 512
+
+      def self.current(cache = Rails.cache.redis)
+        self.new(cache)
+      end
+
+      def initialize(cache)
+        @cache = cache
+      end
+
+      def maybe_continue(username)
+        max = (username == FARMBOT_DEMO_USER) ?
+          MAX_GUEST_COUNT : PER_DEVICE_MAX
+        key = CACHE_KEY_TPL % username
+        total = (cache.get(key) || "0").to_i
+        needs_ttl = cache.ttl(key) < 1
+
+        if total < max
+          cache.incr(key)
+          cache.expire(key, TTL) if needs_ttl
+          yield
+        else
+          Rollbar.error(WARNING % username)
+          raise RateLimit, username
+        end
+      end
+    end
+
     # List of AMQP/MQTT topics we support in the following format:
     # "bot.device_123.<MAIN TOPIC HERE>"
     BOT_CHANNELS = %w(
@@ -40,6 +76,7 @@ module Api
     class PasswordFailure < Exception; end
 
     rescue_from PasswordFailure, with: :report_suspicious_behavior
+    rescue_from BrokerConnectionLimiter::RateLimit, with: :deny
 
     skip_before_action :check_fbos_version, except: []
     skip_before_action :authenticate_user!, except: []
@@ -47,21 +84,26 @@ module Api
     before_action :always_allow_admin, except: [:user_action]
 
     def user_action # Session entrypoint - Part I
-      # Example JSON:
-      #   "username"  => "foo@bar.com",
-      #   "password"  => "******",
-      #   "vhost"     => "/",
-      #   "client_id" => "MQTT_FX_Client",
-      case username_param
-      # NOTE: "guest" is not the same as
-      #       "farmbot_demo". We intentionally
-      #       differentiate to avoid accidental
-      #       security issues. -RC
-      when "guest" then deny
-      when FARMBOT_DEMO_USER then allow
-      when "admin" then authenticate_admin
-      else
-        device_id_in_username == current_device.id ? allow : deny
+      BrokerConnectionLimiter
+        .current
+        .maybe_continue(username_param) do
+        # Example JSON:
+        #   "username"  => "foo@bar.com",
+        #   "password"  => "******",
+        #   "vhost"     => "/",
+        #   "client_id" => "MQTT_FX_Client",
+        case username_param
+        # NOTE: "guest" is not the same as
+        #       "farmbot_demo". We intentionally
+        #       differentiate to avoid accidental
+        #       security issues. -RC
+        when "guest" then deny
+        when FARMBOT_DEMO_USER then allow
+        when "admin" then authenticate_admin
+        else
+          is_ok = device_id_in_username == current_device.id
+          is_ok ? allow : deny
+        end
       end
     end
 

app/lib/broker_connecction_limiter.rb

@@ -1,45 +0,0 @@
-class BrokerConnectionLimiter
-  class RateLimit < StandardError; end
-
-  attr_reader :cache
-
-  CACHE_KEY_TPL = "mqtt_limiter:%s"
-  TTL = 60 * 10 # Ten Minutes
-  LIMIT = 10
-
-  def self.current(cache = Rails.cache.redis)
-    self.new(cache)
-  end
-
-  def initialize(cache)
-    @cache = cache
-  end
-
-  def maybe_continue(username)
-    if allow?(username)
-      yield
-    else
-      raise RateLimit, username
-    end
-  end
-
-  def count_for(username)
-    client.get(key(username)) || 0
-  end
-
-  def increment(username)
-    client.pipelined do
-      k = key(username)
-      client.incr(k)
-      client.expire(k, TTL)
-    end
-  end
-
-  def allow?(username)
-    count_for(username) < LIMIT
-  end
-
-  def key(username)
-    CACHE_KEY_TPL % username
-  end
-end

app/lib/log_service.rb

@@ -29,7 +29,7 @@ class LogService < AbstractServiceRunner
     dev.maybe_unthrottle
     Log.deliver(dev, Logs::Create.run!(log, device: dev))
   rescue => x
-    Rollbar.error(x, "CC: CONNOR")
+    Rollbar.error(x)
   end
 
   def warn_user(data, violation)