diff --git a/ChangeLog.txt b/ChangeLog.txt index 87192b39..94e1c014 100644 --- a/ChangeLog.txt +++ b/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 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 4d259a61..fc111731 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/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 { 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 { 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; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 29b78a39..fc4da271 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/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; } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 44361545..1b77b61b 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/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 { diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java index 5a2bd314..8a3b0aa6 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java +++ b/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();