Fixed issue #43, not clean handlng of bad formatted topic filters during subscrion

2015-06-20 18:27:43 +02:00
parent c4ebd75008
commit 867d78b650
7 changed files with 82 additions and 16 deletions
@@ -602,38 +602,38 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
        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<IMessagesStore.StoredMessage> 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<ValueEvent> {
                    m_messagesStore.nextPacketID(newSubscription.getClientId());
            sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
        }
+        return true;
    }
    
    private void disruptorPublish(OutputMessagingEvent msgEvent) {
@@ -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<T> {
        void visit(TreeNode node);