diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 7fffdee4..88919356 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/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)) { diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index c9379687..f3db6a8a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -72,6 +72,10 @@ public class Topic implements Serializable { } private List 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 res = new ArrayList<>(); String[] splitted = topic.split("/"); diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 63633e60..e13a2ce9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/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); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java index de92dde0..3b61ac2b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java +++ b/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()); } }