diff --git a/broker/pom.xml b/broker/pom.xml
index 8162d928..977ef61b 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -118,7 +118,7 @@
mapdb
-
+
1.0.7
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 edf9357b..d41738dc 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
@@ -602,38 +602,38 @@ class ProtocolProcessor implements EventHandler {
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
- for (SubscribeMessage.Couple req : msg.subscriptions()) {
- AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
- Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
- subscribeSingleTopic(newSubscription, req.getTopicFilter());
- }
-
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
- //reply with requested qos
- for(SubscribeMessage.Couple req : msg.subscriptions()) {
+ for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
- ackMessage.addType(qos);
+ Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
+ boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
+ ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE);
}
-
+
LOG.debug("SUBACK for packetID {}", msg.getMessageID());
session.write(ackMessage);
}
- private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
+ private boolean subscribeSingleTopic(Subscription newSubscription, final String topic) {
LOG.info("<{}> subscribed to topic <{}> with QoS {}",
newSubscription.getClientId(), topic,
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
String clientID = newSubscription.getClientId();
+ boolean validTopic = SubscriptionsStore.validate(newSubscription);
+ if (!validTopic) {
+ //send SUBACK with 0x80 for this topic filter
+ return false;
+ }
m_sessionsStore.addNewSubscription(newSubscription, clientID);
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
- return SubscriptionsStore.matchTopics(key, topic);
+ return SubscriptionsStore.matchTopics(key, topic);
}
});
@@ -645,6 +645,7 @@ class ProtocolProcessor implements EventHandler {
m_messagesStore.nextPacketID(newSubscription.getClientId());
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
+ return true;
}
private void disruptorPublish(OutputMessagingEvent msgEvent) {
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 1440c387..918f6de2 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
@@ -30,6 +30,19 @@ import org.slf4j.LoggerFactory;
*/
public class SubscriptionsStore {
+ /**
+ * Check if the topic filter of the subscription is well formed
+ * */
+ public static boolean validate(Subscription newSubscription) {
+ try {
+ parseTopic(newSubscription.topicFilter);
+ return true;
+ } catch (ParseException pex) {
+ LOG.info("Bad matching topic filter <{}>", newSubscription.topicFilter);
+ return false;
+ }
+ }
+
public static interface IVisitor {
void visit(TreeNode node);
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 2db13398..390bba9f 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
@@ -54,6 +54,7 @@ public class ProtocolProcessorTest {
final static String FAKE_CLIENT_ID2 = "FAKE_456";
final static String FAKE_PUBLISHER_ID = "Publisher";
final static String FAKE_TOPIC = "/news";
+ final static String BAD_FORMATTED_TOPIC = "#MQTTClient";
final static String TEST_USER = "fakeuser";
final static String TEST_PWD = "fakepwd";
@@ -456,6 +457,27 @@ public class ProtocolProcessorTest {
assertTrue(subscriptions.contains(expectedSubscription));
}
+
+ @Test
+ public void testSubscribeWithBadFormattedTopic() {
+ SubscribeMessage msg = new SubscribeMessage();
+ 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();
+ assertEquals(0, subscriptions.size());
+
+ //Exercise
+ m_processor.processSubscribe(m_session, msg);
+
+ //Verify
+ assertEquals(0, subscriptions.size());
+ assertTrue(m_receivedMessage instanceof SubAckMessage);
+ List qosSubAcked = ((SubAckMessage) m_receivedMessage).types();
+ assertEquals(1, qosSubAcked.size());
+ assertEquals(QOSType.FAILURE, qosSubAcked.get(0));
+ }
+
@Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java
index 212667ee..7835f0da 100644
--- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java
+++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java
@@ -18,6 +18,7 @@ package org.eclipse.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.eclipse.moquette.proto.messages.AbstractMessage;
+import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
import org.eclipse.moquette.proto.messages.SubAckMessage;
/**
@@ -38,8 +39,9 @@ class SubAckEncoder extends DemuxEncoder {
buff.writeByte(AbstractMessage.SUBACK << 4 );
buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
buff.writeShort(message.getMessageID());
- for (AbstractMessage.QOSType c : message.types()) {
- buff.writeByte(c.ordinal());
+ for (QOSType c : message.types()) {
+ int qosValue = (c == QOSType.FAILURE) ? qosValue = 0x80 : c.ordinal();
+ buff.writeByte(qosValue);
}
out.writeBytes(buff);
diff --git a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java
index 659ca330..e0732790 100644
--- a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java
+++ b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java
@@ -38,7 +38,7 @@ public abstract class AbstractMessage {
public static final byte DISCONNECT = 14; //Client is Disconnecting
public static enum QOSType {
- MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED;
+ MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED, FAILURE;
public static String formatQoS(QOSType qos) {
return String.format("%d - %s", qos.ordinal(), qos.name());
diff --git a/tools_scripts/integration/issue43_badFormattedTopicSub.groovy b/tools_scripts/integration/issue43_badFormattedTopicSub.groovy
new file mode 100644
index 00000000..46aa1ee2
--- /dev/null
+++ b/tools_scripts/integration/issue43_badFormattedTopicSub.groovy
@@ -0,0 +1,28 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.Message
+import org.fusesource.mqtt.client.QoS
+import org.fusesource.mqtt.client.Topic
+
+String host = args.size() == 1 ? args[0] : "localhost"
+
+MQTT mqttSub = new MQTT()
+mqttSub.setHost(host, 1883)
+mqttSub.setClientId("Subscriber(A)")
+mqttSub.cleanSession = false
+//AT_MOST_ONCE,
+//AT_LEAST_ONCE,
+//EXACTLY_ONCE
+BlockingConnection clientSub = mqttSub.blockingConnection()
+clientSub.connect()
+println "Sub connected"
+Topic[] topics = [new Topic("#MQTTClient", QoS.AT_MOST_ONCE)]
+byte[] qoses = clientSub.subscribe(topics)
+println "Sub subscribed"
+clientSub.disconnect()
+clientSub.kill()
+println "Sub disconnected"
+
+println "Finished!!\n\n"