Repository: moquette-io/moquette (java)

Issue:
Support topic that starts with `$` character
### MQTT specification

**The Server MUST NOT match Topic Filters starting with a wildcard character (# or +) with Topic Names beginning with a $ character** [MQTT-4.7.2-1].
The Server SHOULD prevent Clients from using such Topic Names to exchange messages with other Clients. Server implementations MAY use Topic Names that start with a leading $ character for other purposes.

#### Non-normative comment
- `$SYS/` has been widely adopted as a prefix to topics that contain Server-specific information or control APIs
- Applications cannot use a topic with a leading $ character for their own purposes

#### Non-normative comment
- A subscription to “#” **will not receive** any messages published to a topic beginning with a $
- A subscription to `+/monitor/Clients` **will not receive** any messages published to `$SYS/monitor/Clients`
- A subscription to `$SYS/#` **will receive** messages published to topics beginning with `$SYS/`
- A subscription to `$SYS/monitor/+` **will receive** messages published to `$SYS/monitor/Clients`
- For a Client to receive messages from topics that begin with `$SYS/` and from topics that don’t begin with a $, it **has to subscribe to both** `#` and `$SYS/#`


### Implementation

- [x] update the topic matching algorithm so that TopicFilters  (the paths on subscriptions tree) that start with `#`or `+` doesn't match TopicNames with $ character
- [x] avoid a client to publish messages on topics that start with `$`
- [x] create integration tests to verify the conditions and also unit tests

Pull request description:
Implements reserved topic matching prescriptions
## Release notes
Implements requirements on reserved topics (starts with $). Implements the matching rules and avoid to proceed with processing on client's publishes on those topics.


## What does this PR do?

<!-- Mandatory
Explain here the changes you made on the PR. Please explain the WHAT: patterns used, algorithms implemented, design architecture, message processing, etc.

Example:
  Expose 'bla bla' in the docker environment variables and update logstash.yml to surface this config option.
  
  This commit exposes the 'bla bla' variable in the docker by adding it in env2yaml.go, which translates from
  being an environment variable to a proper yaml config.
  
  Additionally, this PR exposes this setting for bla bla to the moquette.conf file.
-->
Updates the mathcing algorithm inside `CTrie` to negatively match topic names in the reserved form (start with `$`) against wildcards position as first token of a topic filter.
Refactored the tests to reuse integration fixture also outside the MQTT5 test suite, because this change regards all protocols starting from MQTT3.1.1.

## Why is it important/What is the impact to the user?

Avoid the clients to use improperly reserved topics to exchange payloads. In particular avoid wildcard matching to listen inadvertently and receive unexpected topics, the clients has to explicitly subscribe to the reserved topics. 

## Checklist

<!-- Mandatory
Add a checklist of things that are required to be reviewed in order to have the PR approved

List here all the items you have verified BEFORE sending this PR. Please DO NOT remove any item, striking through those that do not apply. (Just in case, strikethrough uses two tildes. ~~Scratch this.~~)
-->

- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- ~~[ ] I have made corresponding changes to the documentation~~
- ~~[ ] I have made corresponding change to the default configuration files (and/or docker env variables)~~
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] I have updated the Changelog if it's a feature or a fix that has to be reported

## Author's Checklist

<!-- Recommended
Add a checklist of things that are required to be reviewed in order to have the PR approved
-->
- [ ]

## How to test this PR locally

<!-- Recommended
Explain here how this PR will be tested by the reviewer: commands, dependencies, steps, etc.
-->

## Related issues

<!-- Recommended
Link related issues below. Insert the issue link or reference after the word "Closes" if merging this should automatically close it.

- Closes #123
- Relates #123
- Requires #123
- Superseeds #123
-->
- Closes #792

## Use cases

<!-- Recommended
Explain here the different behaviors that this PR introduces or modifies in this project, user roles, environment configuration, etc.

If you are familiar with Gherkin test scenarios, we recommend its usage: https://cucumber.io/docs/gherkin/reference/
-->


Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index f7d18c20..19e0b30c 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
 Version 0.18-SNAPSHOT:
+   [fix] Implements requirements on reserved topics (starts with $). Implements the matching rules and avoid to proceed with processing on client's publishes on those topics (#793)
    [feature] Handle will delay interval and MQTT5's Will optional properties (#770)
    [fix] Handle empty collector batches in PostOffice (#777)
    [feature] Introduce a new option "PEER_CERTIFICATE_AS_USERNAME". When enabled, the client certificate is provided as the username to the authenticator.
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index 8ae88e56..7aa8b6bb 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -608,6 +608,11 @@ final class MQTTConnection {
             dropConnection();
         }
 
+        if (!topic.isEmpty() && topic.headToken().isReserved()) {
+            LOG.warn("Avoid to publish on topic which contains reserved topic (starts with $)");
+            return PostOffice.RouteResult.failed(clientId);
+        }
+
         // retain else msg is cleaned by the NewNettyMQTTHandler and is not available
         // in execution by SessionEventLoop
         msg.retain();
@@ -615,8 +620,9 @@ final class MQTTConnection {
             case AT_MOST_ONCE:
                 return postOffice.routeCommand(clientId, "PUB QoS0", () -> {
                     checkMatchSessionLoop(clientId);
-                    if (!isBoundToSession())
+                    if (!isBoundToSession()) {
                         return null;
+                    }
                     postOffice.receivedPublishQos0(topic, username, clientId, msg);
                     return null;
                 }).ifFailed(msg::release);
diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java b/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java
index 70ba7d30..39351400 100644
--- a/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java
+++ b/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java
@@ -50,55 +50,68 @@ public class CTrie {
         MATCH, GODEEP, STOP
     }
 
-    private NavigationAction evaluate(Topic topic, CNode cnode) {
+    private NavigationAction evaluate(Topic topicName, CNode cnode, int depth) {
+        // depth 0 is the root node of all the topics, so for topic filter
+        // monitor/sensor we have <root> -> monitor -> sensor
+        final boolean isFirstLevel = depth == 1;
         if (Token.MULTI.equals(cnode.getToken())) {
+            Token token = topicName.headToken();
+            if (token != null && token.isReserved() && isFirstLevel) {
+                // [MQTT-4.7.2-1] single wildcard can't match reserved topics
+                // if reserved token is the first of the topicName
+                return NavigationAction.STOP;
+            }
             return NavigationAction.MATCH;
         }
-        if (topic.isEmpty()) {
+        if (topicName.isEmpty()) {
             return NavigationAction.STOP;
         }
-        final Token token = topic.headToken();
-        if (!(Token.SINGLE.equals(cnode.getToken()) || cnode.getToken().equals(token) || ROOT.equals(cnode.getToken()))) {
-            return NavigationAction.STOP;
+        final Token token = topicName.headToken();
+        if (Token.SINGLE.equals(cnode.getToken()) || cnode.getToken().equals(token) || ROOT.equals(cnode.getToken())) {
+            if (Token.SINGLE.equals(cnode.getToken()) && token.isReserved() && isFirstLevel) {
+                // [MQTT-4.7.2-1] single wildcard can't match reserved topics
+                return NavigationAction.STOP;
+            }
+            return NavigationAction.GODEEP;
         }
-        return NavigationAction.GODEEP;
+        return NavigationAction.STOP;
     }
 
-    public List<Subscription> recursiveMatch(Topic topic) {
-        return recursiveMatch(topic, this.root);
+    public List<Subscription> recursiveMatch(Topic topicName) {
+        return recursiveMatch(topicName, this.root, 0);
     }
 
-    private List<Subscription> recursiveMatch(Topic topic, INode inode) {
+    private List<Subscription> recursiveMatch(Topic topicName, INode inode, int depth) {
         CNode cnode = inode.mainNode();
         if (cnode instanceof TNode) {
             return Collections.emptyList();
         }
-        NavigationAction action = evaluate(topic, cnode);
+        NavigationAction action = evaluate(topicName, cnode, depth);
         if (action == NavigationAction.MATCH) {
             return cnode.subscriptions;
         }
         if (action == NavigationAction.STOP) {
             return Collections.emptyList();
         }
-        Topic remainingTopic = (ROOT.equals(cnode.getToken())) ? topic : topic.exceptHeadToken();
+        Topic remainingTopic = (ROOT.equals(cnode.getToken())) ? topicName : topicName.exceptHeadToken();
         List<Subscription> subscriptions = new ArrayList<>();
 
         // We should only consider the maximum three children children of
         // type #, + or exact match
         Optional<INode> subInode = cnode.childOf(Token.MULTI);
         if (subInode.isPresent()) {
-            subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get()));
+            subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get(), depth + 1));
         }
         subInode = cnode.childOf(Token.SINGLE);
         if (subInode.isPresent()) {
-            subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get()));
+            subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get(), depth + 1));
         }
         if (remainingTopic.isEmpty()) {
             subscriptions.addAll(cnode.subscriptions);
         } else {
             subInode = cnode.childOf(remainingTopic.headToken());
             if (subInode.isPresent()) {
-                subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get()));
+                subscriptions.addAll(recursiveMatch(remainingTopic, subInode.get(), depth + 1));
             }
         }
         return subscriptions;
diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java
index 5ce0f32e..5b381a6b 100644
--- a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java
+++ b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java
@@ -71,18 +71,18 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
      * contain character # and + because they are reserved to listeners subscriptions, and not topic
      * publishing.
      *
-     * @param topic
-     *            to use fo searching matching subscriptions.
+     * @param topicName
+     *            to use for search matching subscriptions.
      * @return the list of matching subscriptions, or empty if not matching.
      */
     @Override
-    public List<Subscription> matchWithoutQosSharpening(Topic topic) {
-        return ctrie.recursiveMatch(topic);
+    public List<Subscription> matchWithoutQosSharpening(Topic topicName) {
+        return ctrie.recursiveMatch(topicName);
     }
 
     @Override
-    public List<Subscription> matchQosSharpening(Topic topic) {
-        final List<Subscription> subscriptions = matchWithoutQosSharpening(topic);
+    public List<Subscription> matchQosSharpening(Topic topicName) {
+        final List<Subscription> subscriptions = matchWithoutQosSharpening(topicName);
 
         Map<String, Subscription> subsGroupedByClient = new HashMap<>();
         for (Subscription sub : subscriptions) {
diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java b/broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java
index a524e550..28ce7290 100644
--- a/broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java
+++ b/broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java
@@ -20,6 +20,9 @@ import io.moquette.broker.ISubscriptionsRepository;
 import java.util.List;
 import java.util.Set;
 
+/**
+ * Contains all topic filters that are used to  match against topic names.
+ * */
 public interface ISubscriptionsDirectory {
 
     void init(ISubscriptionsRepository sessionsRepository);
diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/Token.java b/broker/src/main/java/io/moquette/broker/subscriptions/Token.java
index b8ccb00e..ca3544ed 100644
--- a/broker/src/main/java/io/moquette/broker/subscriptions/Token.java
+++ b/broker/src/main/java/io/moquette/broker/subscriptions/Token.java
@@ -34,18 +34,6 @@ public class Token implements Comparable<Token> {
         return name;
     }
 
-    protected boolean match(Token t) {
-        if (MULTI.equals(t) || SINGLE.equals(t)) {
-            return false;
-        }
-
-        if (MULTI.equals(this) || SINGLE.equals(this)) {
-            return true;
-        }
-
-        return equals(t);
-    }
-
     @Override
     public int hashCode() {
         int hash = 7;
@@ -86,4 +74,11 @@ public class Token implements Comparable<Token> {
         }
         return name.compareTo(other.name);
     }
+
+    /**
+     * Token which starts with $ is reserved
+     * */
+    public boolean isReserved() {
+        return name.startsWith("$");
+    }
 }


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java
index 6deddc6c..7baba221 100644
--- a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java
+++ b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java
@@ -171,26 +171,39 @@ public class CTrieSubscriptionDirectoryMatchingTest {
         assertMatch("foo/bar/+", "foo/bar/");
     }
 
-    private void assertMatch(String s, String t) {
+    @Test
+    public void givenTopicFilterStartingWithSingleWildcardDoesntMatchSpecialTopicNames() {
+        assertNotMatch("+/monitor/clients", "$SYS/monitor/clients");
+        assertMatch("outer/+/inner", "outer/$something/inner");
+        assertMatch("$SYS/monitor/+", "$SYS/monitor/clients");
+    }
+
+    @Test
+    public void givenTopicFilterStartingWithMultiWildcardDoesntMatchSpecialTopicNames() {
+        assertNotMatch("#", "$SYS/monitor/clients");
+        assertMatch("$SYS/#", "$SYS");
+    }
+
+    private void assertMatch(String topicFilter, String topicName) {
         sut = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
         sut.init(sessionsRepository);
 
-        Subscription sub = clientSubOnTopic("AnySensor1", s);
+        Subscription sub = clientSubOnTopic("AnySensor1", topicFilter);
         sut.add(sub);
 
-        assertThat(sut.matchWithoutQosSharpening(asTopic(t))).isNotEmpty();
+        assertThat(sut.matchWithoutQosSharpening(asTopic(topicName))).isNotEmpty();
     }
 
-    private void assertNotMatch(String subscription, String topic) {
+    private void assertNotMatch(String topicFilter, String topicName) {
         sut = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
         sut.init(sessionsRepository);
 
-        Subscription sub = clientSubOnTopic("AnySensor1", subscription);
+        Subscription sub = clientSubOnTopic("AnySensor1", topicFilter);
         sut.add(sub);
 
-        assertThat(sut.matchWithoutQosSharpening(asTopic(topic))).isEmpty();
+        assertThat(sut.matchWithoutQosSharpening(asTopic(topicName))).isEmpty();
     }
 
     @Test
diff --git a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java
index 88aa6394..7e7ac4b5 100644
--- a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java
+++ b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java
@@ -114,8 +114,8 @@ public class CTrieTest {
         assertTrue(subscriptions.contains(happinessSensor));
     }
 
-    static Subscription clientSubOnTopic(String clientID, String topicName) {
-        return new Subscription(clientID, asTopic(topicName), null);
+    static Subscription clientSubOnTopic(String clientID, String topicFilter) {
+        return new Subscription(clientID, asTopic(topicFilter), null);
     }
 
     @Test
diff --git a/broker/src/test/java/io/moquette/integration/PublishTest.java b/broker/src/test/java/io/moquette/integration/PublishTest.java
new file mode 100644
index 00000000..b7265689
--- /dev/null
+++ b/broker/src/test/java/io/moquette/integration/PublishTest.java
@@ -0,0 +1,85 @@
+package io.moquette.integration;
+
+import com.hivemq.client.mqtt.MqttClient;
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt3.Mqtt3BlockingClient;
+import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
+import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAckReturnCode;
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import io.moquette.integration.mqtt5.AbstractServerIntegrationTest;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class PublishTest extends AbstractServerIntegrationTest {
+    @Override
+    public String clientName() {
+        return "Publisher";
+    }
+
+    @Test
+    public void givenAConnectedClientWhenItPublishToReservedTopicNoPublishIsForwarded() throws InterruptedException {
+        // given a connected client
+        Mqtt3BlockingClient publisher = createHiveMqV3Client("publisher");
+
+        Mqtt3BlockingClient subscriber = createHiveMqV3Client("subscriber");
+
+        subscriber.subscribeWith()
+            .topicFilter("$sys/monitor")
+            .qos(MqttQos.AT_MOST_ONCE)
+            .send();
+
+        // when send a publish to reserved topic
+        publisher.publishWith()
+            .topic("$sys/monitor")
+            .payload("Something new".getBytes(StandardCharsets.UTF_8))
+            .send();
+
+        // then no message is published
+        verifyNoMessageIsReceived(subscriber, Duration.ofSeconds(2));
+    }
+
+    @NotNull
+    private static Mqtt3BlockingClient createHiveMqV3Client(String clientId) {
+        Mqtt3BlockingClient publisher = MqttClient.builder()
+            .useMqttVersion3()
+            .identifier(clientId)
+            .serverHost("localhost")
+            .serverPort(1883)
+            .buildBlocking();
+        Mqtt3ConnAck connectAck = publisher.connect();
+        assertEquals(Mqtt3ConnAckReturnCode.SUCCESS, connectAck.getReturnCode(), "Accept plain connection");
+        return publisher;
+    }
+
+    private static void verifyMessageIsReceived(String expectedPayload, Mqtt3BlockingClient client, Duration timeout) throws InterruptedException {
+        try (Mqtt3BlockingClient.Mqtt3Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
+            Optional<Mqtt3Publish> publishMessage = publishes.receive(timeout.getSeconds(), TimeUnit.SECONDS);
+            final String payload = publishMessage.map(Mqtt3Publish::getPayloadAsBytes)
+                .map(b -> new String(b, StandardCharsets.UTF_8))
+                .orElse("Failed to load payload");
+            assertEquals(expectedPayload, payload, "Something new has to be received");
+        }
+    }
+
+    private static void verifyNoMessageIsReceived(Mqtt3BlockingClient testamentSubscriber, Duration timeout) throws InterruptedException {
+        try (Mqtt3BlockingClient.Mqtt3Publishes publishes = testamentSubscriber.publishes(MqttGlobalPublishFilter.ALL)) {
+            Optional<Mqtt3Publish> publishedWill = publishes.receive(timeout.getSeconds(), TimeUnit.SECONDS);
+
+            // verify no published will in 10 seconds
+            assertFalse(publishedWill.isPresent(), "No message should be published");
+        }
+    }
+
+
+}
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
index a98cec90..88d31d1b 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/AbstractServerIntegrationTest.java
@@ -16,7 +16,7 @@ import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Properties;
 
-abstract class AbstractServerIntegrationTest {
+public abstract class AbstractServerIntegrationTest {
     Server broker;
     IConfig config;
 
@@ -26,7 +26,7 @@ abstract class AbstractServerIntegrationTest {
 
     Client lowLevelClient;
 
-    abstract String clientName();
+    public abstract String clientName();
 
     protected void startServer(String dbPath) throws IOException {
         broker = new Server();
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
index cec94054..0a71897d 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectAckTest.java
@@ -21,7 +21,7 @@ class ConnectAckTest extends  AbstractServerIntegrationTest {
     private MqttConnAckMessage connAck;
 
     @Override
-    String clientName() {
+    public String clientName() {
         return "client";
     }
 
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
index 1aa2bb5e..83a7e178 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
@@ -54,7 +54,7 @@ class ConnectTest extends AbstractServerIntegrationTest {
     }
 
     @Override
-    String clientName() {
+    public String clientName() {
         return "subscriber";
     }