Add public_broadcast MQTT/AMQP channel

app/controllers/api/rmq_utils_controller.rb

@@ -24,7 +24,7 @@ class RmqUtilsController < Api::AbstractController
     # The only valid format for AMQP / MQTT topics.
     # Prevents a whole host of abuse / security issues.
     TOPIC_REGEX = Regexp.new("bot\\.device_\\d*\\.(#{BOT_CHANNELS})")
-    PUBLIC_CHANNELS = ["", ".\\*", ".\\#"].map { |x| "public_broadcast" + x }
+    PUBLIC_CHANNELS = ["", ".*", ".#"].map { |x| "public_broadcast" + x }
 
     MALFORMED_TOPIC = "malformed topic. Must match #{TOPIC_REGEX.inspect}"
     VHOST = ENV.fetch("MQTT_VHOST") { "/" }
@@ -78,10 +78,15 @@ def topic_action # Called during subscribe
       #   "vhost"       => "/",
       case routing_key_param
       when *PUBLIC_CHANNELS
-        permission_param == "read" ? allow : deny
+        if permission_param == "read"
+          allow
+        else
+          deny
+        end
       else
-        scrutinize_topic_string
-        device_id_in_topic == device_id_in_username ? allow : deny
+        if_topic_is_safe do
+          device_id_in_topic == device_id_in_username ? allow : deny
+        end
       end
     end
 
@@ -134,9 +139,12 @@ def permission_param
       @permission ||= params["permission"]
     end
 
-    def scrutinize_topic_string
-      is_ok = !!TOPIC_REGEX.match(routing_key_param)
-      render json: MALFORMED_TOPIC, status: 422 unless is_ok
+    def if_topic_is_safe
+      if !!TOPIC_REGEX.match(routing_key_param)
+        yield
+      else
+        render json: MALFORMED_TOPIC, status: 422
+      end
     end
 
     def device_id_in_topic

spec/controllers/api/rmq_utils/rmq_utils_controller_spec.rb

@@ -26,6 +26,24 @@
     end
   end
 
+  it "denies public_broadcast write access to non-admin users" do
+    routing_key = Api::RmqUtilsController::PUBLIC_CHANNELS.sample
+    permission  = ["write", "configure"].sample
+    p = credentials.merge(routing_key: routing_key, permission: permission)
+    post :topic_action, params: p
+    expect(response.body).to eq("deny")
+    expect(response.status).to eq(403)
+  end
+
+  it "allows public_broadcast read access to non-admin users" do
+    routing_key = Api::RmqUtilsController::PUBLIC_CHANNELS.sample
+    permission  = "read"
+    p = credentials.merge(routing_key: routing_key, permission: permission)
+    post :topic_action, params: p
+    expect(response.body).to eq("allow")
+    expect(response.status).to eq(200)
+  end
+
   it "allows access to ones own topic" do
     p = credentials.merge(routing_key: "bot.#{credentials[:username]}.logs")
     post :topic_action, params: p