Added checking of valid topicFilter format in unsubscribe

ChangeLog.txt

@@ -3,6 +3,7 @@ Version 0.8:
     Fixed bug, doesn't close the channel when connect doesn't contains a valid proto name, issue #65
     Fixed bug, does not check for zero length topic (filters) in PUBLISH, SUBSCRIBE, and UNSUBSCRIBE, issue #67
     Fixed issue #69
+    Fixed bug #68, UNSUBSCRIBE has to fail on bad topicFilter names
 
 Version 0.7:
     Fix issue #38, added aligned to MQTT 3.1.1

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -19,6 +19,7 @@ import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.dsl.Disruptor;
 import java.nio.ByteBuffer;
+import java.text.ParseException;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -596,6 +597,14 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
 
         for (String topic : topics) {
+            boolean validTopic = SubscriptionsStore.validate(topic);
+            if (!validTopic) {
+                //close the connection, not valid topicFilter is a protocol violation
+                session.close(true);
+                LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID);
+                return;
+            }
+
             subscriptions.removeSubscription(topic, clientID);
             m_sessionsStore.removeSubscription(topic, clientID);
             m_interceptor.notifyTopicUnsubscribed(topic);
@@ -634,7 +643,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         LOG.info("<{}> subscribed to topic <{}> with QoS {}", 
                 newSubscription.getClientId(), topic, 
                 AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
-        boolean validTopic = SubscriptionsStore.validate(newSubscription);
+        boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
         if (!validTopic) {
             //send SUBACK with 0x80 for this topic filter
             return false;

broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java

@@ -33,12 +33,12 @@ public class SubscriptionsStore {
     /**
      * Check if the topic filter of the subscription is well formed
      * */
-    public static boolean validate(Subscription newSubscription) {
+    public static boolean validate(String topicFilter) {
         try {
-            parseTopic(newSubscription.topicFilter);
+            parseTopic(topicFilter);
             return true;
         } catch (ParseException pex) {
-            LOG.info("Bad matching topic filter <{}>", newSubscription.topicFilter);
+            LOG.info("Bad matching topic filter <{}>", topicFilter);
             return false;
         }
     }

broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java

@@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit;
 
 import io.netty.util.AttributeKey;
 import org.eclipse.moquette.interception.InterceptHandler;
+import org.eclipse.moquette.proto.messages.*;
 import org.eclipse.moquette.server.netty.NettyChannel;
 import org.eclipse.moquette.spi.IMatchingCondition;
 import org.eclipse.moquette.spi.IMessagesStore;
@@ -31,14 +32,8 @@ import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator;
 import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
 import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
 import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1_1;
-import org.eclipse.moquette.proto.messages.AbstractMessage;
+
 import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
-import org.eclipse.moquette.proto.messages.ConnAckMessage;
-import org.eclipse.moquette.proto.messages.ConnectMessage;
-import org.eclipse.moquette.proto.messages.DisconnectMessage;
-import org.eclipse.moquette.proto.messages.PublishMessage;
-import org.eclipse.moquette.proto.messages.SubAckMessage;
-import org.eclipse.moquette.proto.messages.SubscribeMessage;
 import org.eclipse.moquette.server.ServerChannel;
 import static org.junit.Assert.*;
 import org.junit.Before;
@@ -505,7 +500,7 @@ public class ProtocolProcessorTest {
     @Test
     public void testSubscribeWithBadFormattedTopic() {
         SubscribeMessage msg = new SubscribeMessage();
-        msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), BAD_FORMATTED_TOPIC));
+        msg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), BAD_FORMATTED_TOPIC));
         m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID);
         m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false);
         subscriptions.clearAllSubscriptions();
@@ -522,6 +517,24 @@ public class ProtocolProcessorTest {
         assertEquals(QOSType.FAILURE, qosSubAcked.get(0));
     }
 
+    /*
+     * Check topicFilter is a valid MQTT topic filter (issue 68)
+     * */
+    @Test
+    public void testUnsubscribeWithBadFormattedTopic() {
+        UnsubscribeMessage msg = new UnsubscribeMessage();
+        msg.setMessageID(1);
+        msg.addTopicFilter(BAD_FORMATTED_TOPIC);
+        m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID);
+        m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false);
+
+        //Exercise
+        m_processor.processUnsubscribe(m_session, msg);
+
+        //Verify
+        assertTrue(m_session.isClosed());
+    }
+
     
     @Test
     public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {

netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java

@@ -40,7 +40,7 @@ class UnsubscribeDecoder extends DemuxDecoder {
         
         //check qos level
         if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
-            throw new CorruptedFrameException("Found an Usubscribe message with qos other than LEAST_ONE, was: " + message.getQos());
+            throw new CorruptedFrameException("Found an Unsubscribe message with qos other than LEAST_ONE, was: " + message.getQos());
         }
             
         int start = in.readerIndex();