Fixed issue #412 adding a check on Topic's string length in processPublish

broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java

@@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
 import java.util.Optional;
 
 import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
+import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
 import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
 
@@ -58,6 +59,12 @@ class Qos1PublishHandler extends QosPublishHandler {
     void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
         // verify if topic can be write
         final Topic topic = new Topic(msg.variableHeader().topicName());
+        topic.getTokens();
+        if (!topic.isValid()) {
+            LOG.warn("Invalid topic format, force close the connection");
+            channel.close().addListener(CLOSE_ON_FAILURE);
+            return;
+        }
         String clientID = NettyUtils.clientID(channel);
         String username = NettyUtils.userName(channel);
         if (!m_authorizator.canWrite(topic, username, clientID)) {

broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java

@@ -72,6 +72,10 @@ public class Topic implements Serializable {
     }
 
     private List<Token> parseTopic(String topic) throws ParseException {
+        if (topic.length() == 0) {
+            throw new ParseException("Bad format of topic, topic MUST be at least 1 character [MQTT-4.7.3-1] and " +
+                                     "this was empty", 0);
+        }
         List<Token> res = new ArrayList<>();
         String[] splitted = topic.split("/");
 

broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java

@@ -85,7 +85,11 @@ abstract class AbstractProtocolProcessorCommonUtils {
     }
 
     void verifyNoPublishIsReceived() {
-        final Object messageReceived = m_channel.readOutbound();
+        verifyNoPublishIsReceived(this.m_channel);
+    }
+
+    void verifyNoPublishIsReceived(EmbeddedChannel channel) {
+        final Object messageReceived = channel.readOutbound();
         assertNull("Received an out message from processor while not expected", messageReceived);
     }
 

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java

@@ -341,31 +341,22 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
 
         assertFalse("First 'subscriber' channel MUST be closed by the broker", clientXA.isOpen());
         verifyPublishIsReceived(clientYA, "Hello 2", AT_MOST_ONCE);
+    }
 
-//        LOG.info("*** testForceClientDisconnection_issue118 ***");
-//        MessageCollector cbSubscriber1 = new MessageCollector();
-//        MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
-//        LOG.info("Connected 'subscriber' first time");
-//        clientXA.subscribe("topic", 0);
-//        LOG.info("Subscribed 'topic' from 'subscriber' first time");
-//
-//        MqttClient clientXB = createClient("publisher", "X");
-//        LOG.info("Connected 'publisher' first time");
-//        clientXB.publish("topic", "Hello".getBytes(), 2, false);
-//        LOG.info("Published on 'topic' from 'publisher' first time");
-//
-//        LOG.info("Creating second new 'subscriber'");
-//        MessageCollector cbSubscriber2 = new MessageCollector();
-//        MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
-//        LOG.info("Connected 'subscriber' second time");
-//        clientYA.subscribe("topic", 0);
-//
-//        MqttClient clientYB = createClient("publisher", "Y");
-//        clientYB.publish("topic", "Hello 2".getBytes(), 2, true);
-//
-//        // Verify that the second subscriber client get notified and not the first.
-//        assertTrue(cbSubscriber1.connectionLost());
-//        assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload()));
+    @Test
+    public void testPublishToEmptyTopic() {
+        EmbeddedChannel subscriber = new EmbeddedChannel();
+        connectAsClient(subscriber, "Subscriber");
+
+        subscribe(subscriber, "#", AT_LEAST_ONCE);
+
+        // Exercise
+        connect();
+        publishToAs(FAKE_CLIENT_ID, "", AT_LEAST_ONCE, 1, false);
+
+        // Verify
+        verifyNoPublishIsReceived(subscriber);
+        assertFalse("On empty topic subscription the channel MUST be closed", m_channel.isOpen());
     }
 
 }