Updated SUBSCRIBE to be 3.1.1 compliant

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoder.java

@@ -38,6 +38,7 @@ class SubscribeDecoder extends DemuxDecoder {
             in.resetReaderIndex();
             return;
         }
+        //TODO check reserved fixed flags = b0010
         
         //check qos level
         if (message.getQos() != QOSType.LEAST_ONE) {
@@ -53,6 +54,10 @@ class SubscribeDecoder extends DemuxDecoder {
             readed = in.readerIndex()- start;
         }
         
+        if (message.subscriptions().isEmpty()) {
+            throw new CorruptedFrameException("subscribe MUST have got at least 1 couple topic/QoS");
+        } 
+        
         out.add(message);
     }
     
@@ -61,7 +66,12 @@ class SubscribeDecoder extends DemuxDecoder {
      */
     private void decodeSubscription(ByteBuf in, SubscribeMessage message) throws UnsupportedEncodingException {
         String topic = Utils.decodeString(in);
-        byte qos = (byte)(in.readByte() & 0x03);
+        byte qosByte = in.readByte();
+        if ((qosByte & 0xFC) > 0) { //the first 6 bits is reserved => has to be 0
+            throw new CorruptedFrameException("subscribe MUST have QoS byte with reserved buts to 0, found " + Integer.toHexString(qosByte));
+        }
+        byte qos = (byte)(qosByte & 0x03);
+        //TODO check qos id 000000xx
         message.addSubscription(new SubscribeMessage.Couple(qos, topic));
     }
     

netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java

@@ -67,6 +67,37 @@ public class SubscribeDecoderTest {
         assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType());
     }
     
+    @Test(expected = CorruptedFrameException.class)
+    public void testFailOnEmptyTopic() throws Exception {
+        m_buff = Unpooled.buffer(4);
+        initMultiTopic(m_buff, 123);
+        
+        //Excercise
+        m_msgdec.decode(null, m_buff, m_results);
+
+        //Verify
+        assertFalse(m_results.isEmpty());
+        SubscribeMessage message = (SubscribeMessage)m_results.get(0); 
+        assertEquals(2, message.subscriptions().size());
+        assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType());
+    }
+    
+    @Test(expected = CorruptedFrameException.class)
+    public void testFailOnSubscriptionWithReservedQosPolluted() throws Exception {
+        m_buff = Unpooled.buffer(4);
+        SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)2, "a/b");
+        initPollutedTopic(m_buff, 123, c1);
+        
+        //Excercise
+        m_msgdec.decode(null, m_buff, m_results);
+
+        //Verify
+        assertFalse(m_results.isEmpty());
+        SubscribeMessage message = (SubscribeMessage)m_results.get(0); 
+        assertEquals(2, message.subscriptions().size());
+        assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType());
+    }
+    
     private void initHeaderBadQos(ByteBuf buff) {
         buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4).writeByte(0);
     }
@@ -83,4 +114,15 @@ public class SubscribeDecoderTest {
                 .writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes()));
         buff.writeBytes(topicBuffer);
     }
+    
+    private void initPollutedTopic(ByteBuf buff, int messageID, SubscribeMessage.Couple topic) throws IllegalAccessException {
+        ByteBuf topicBuffer = Unpooled.buffer(4);
+        topicBuffer.writeShort(messageID);
+        topicBuffer.writeBytes(Utils.encodeString(topic.getTopic()));
+        topicBuffer.writeByte(0xF0 | topic.getQos());
+        
+        buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02)
+                .writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes()));
+        buff.writeBytes(topicBuffer);
+    }
 }