Apply ACL controls to PUBLISH operations

amqtt/broker.py

@@ -620,21 +620,32 @@ async def client_connected(
                             % client_session.client_id
                         )
                         break
-                    await self.plugins_manager.fire_event(
-                        EVENT_BROKER_MESSAGE_RECEIVED,
-                        client_id=client_session.client_id,
-                        message=app_message,
-                    )
-                    await self._broadcast_message(
-                        client_session, app_message.topic, app_message.data
+
+                    # See if the user is allowed to publish to this topic.
+                    permitted = await self.topic_filtering(
+                            client_session, topic=app_message.topic, action='publish'
                     )
-                    if app_message.publish_packet.retain_flag:
-                        self.retain_message(
-                            client_session,
-                            app_message.topic,
-                            app_message.data,
-                            app_message.qos,
+                    if not permitted:
+                        self.logger.info(
+                                "%s forbidden TOPIC %s sent in PUBLISH message.",
+                                client_session.client_id, app_message.topic
+                        )
+                    else:
+                        await self.plugins_manager.fire_event(
+                            EVENT_BROKER_MESSAGE_RECEIVED,
+                            client_id=client_session.client_id,
+                            message=app_message,
+                        )
+                        await self._broadcast_message(
+                            client_session, app_message.topic, app_message.data
                         )
+                        if app_message.publish_packet.retain_flag:
+                            self.retain_message(
+                                client_session,
+                                app_message.topic,
+                                app_message.data,
+                                app_message.qos,
+                            )
                     wait_deliver = asyncio.Task(
                         handler.mqtt_deliver_next_message(), loop=self._loop
                     )
@@ -703,7 +714,7 @@ async def authenticate(self, session: Session, listener):
         # If all plugins returned True, authentication is success
         return auth_result
 
-    async def topic_filtering(self, session: Session, topic):
+    async def topic_filtering(self, session: Session, topic, action: str):
         """
         This method call the topic_filtering method on registered plugins to check that the subscription is allowed.
         User is considered allowed if all plugins called return True.
@@ -713,7 +724,8 @@ async def topic_filtering(self, session: Session, topic):
          - None if topic filtering can't be achieved (then plugin result is then ignored)
         :param session:
         :param listener:
-        :param topic: Topic in which the client wants to subscribe
+        :param topic: Topic in which the client wants to subscribe / publish
+        :param action: What is being done with the topic?  subscribe or publish
         :return:
         """
         topic_plugins = None
@@ -724,6 +736,7 @@ async def topic_filtering(self, session: Session, topic):
             "topic_filtering",
             session=session,
             topic=topic,
+            action=action,
             filter_plugins=topic_plugins,
         )
         topic_result = True
@@ -773,7 +786,7 @@ async def add_subscription(self, subscription, session):
                         # [MQTT-4.7.1-3] + wildcard character must occupy entire level
                         return 0x80
             # Check if the client is authorised to connect to the topic
-            permitted = await self.topic_filtering(session, topic=a_filter)
+            permitted = await self.topic_filtering(session, topic=a_filter, action='subscribe')
             if not permitted:
                 return 0x80
             qos = subscription[1]

amqtt/plugins/topic_checking.py

@@ -7,6 +7,7 @@ def __init__(self, context):
             self.context.logger.warning(
                 "'topic-check' section not found in context configuration"
             )
+            self.topic_config = None
 
     def topic_filtering(self, *args, **kwargs):
         if not self.topic_config:
@@ -66,11 +67,24 @@ async def topic_filtering(self, *args, **kwargs):
         if filter_result:
             session = kwargs.get("session", None)
             req_topic = kwargs.get("topic", None)
+            action = kwargs.get("action", None)
+
+            # hbmqtt and older amqtt do not support publish filtering
+            if (action == "publish") and ("publish-acl" not in self.topic_config):
+                # maintain backward compatibility, assume permitted
+                return True
+
             if req_topic:
                 username = session.username
                 if username is None:
                     username = "anonymous"
-                allowed_topics = self.topic_config["acl"].get(username, None)
+
+                if action == "publish":
+                    acl = self.topic_config["publish-acl"]
+                elif action == "subscribe":
+                    acl = self.topic_config["acl"]
+
+                allowed_topics = acl.get(username, None)
                 if allowed_topics:
                     for allowed_topic in allowed_topics:
                         if self.topic_ac(req_topic, allowed_topic):
@@ -80,3 +94,5 @@ async def topic_filtering(self, *args, **kwargs):
                     return False
             else:
                 return False
+        else:
+            return False