Changed DemuxDecoder.decodeCommonHeader to also validate the fixed part flags (if needed)

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

@@ -17,6 +17,7 @@ package org.dna.mqtt.moquette.parser.netty;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.CorruptedFrameException;
 import java.util.List;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 
@@ -27,13 +28,41 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 abstract class DemuxDecoder {
     abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
     
+    /**
+     * Decodes the first 2 bytes of the MQTT packet.
+     * The first byte contain the packet operation code and the flags,
+     * the second byte and more contains the overall packet length.
+     */
     protected boolean decodeCommonHeader(AbstractMessage message, ByteBuf in) {
+        return genericDecodeCommonHeader(message, null, in);
+    }
+    
+    /**
+     * Do the same as the @see#decodeCommonHeader but having a strong validation on the flags values
+     */
+    protected boolean decodeCommonHeader(AbstractMessage message, int expectedFlags, ByteBuf in) {
+        return genericDecodeCommonHeader(message, expectedFlags, in);
+    }
+    
+    
+    private boolean genericDecodeCommonHeader(AbstractMessage message, Integer expectedFlagsOpt, ByteBuf in) {
         //Common decoding part
         if (in.readableBytes() < 2) {
             return false;
         }
         byte h1 = in.readByte();
         byte messageType = (byte) ((h1 & 0x00F0) >> 4);
+        
+        byte flags = (byte) (h1 & 0x0F);
+        if (expectedFlagsOpt != null) {
+            int expectedFlags = expectedFlagsOpt;
+            if ((byte) expectedFlags != flags) {
+                String hexExpected = Integer.toHexString(expectedFlags);
+                String hexReceived = Integer.toHexString(flags);
+                throw new CorruptedFrameException(String.format("Received a message with fixed header flags (%s) != expected (%s)", hexReceived, hexExpected));
+            }
+        }
+        
         boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1);
         byte qosLevel = (byte) ((h1 & 0x0006) >> 1);
         boolean retainFlag = ((byte) (h1 & 0x0001) == 1);
@@ -49,5 +78,4 @@ abstract class DemuxDecoder {
         message.setRemainingLength(remainingLength);
         return true;
     }
-    
 }

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

@@ -21,7 +21,6 @@ import io.netty.handler.codec.CorruptedFrameException;
 import io.netty.util.AttributeMap;
 import java.io.UnsupportedEncodingException;
 import java.util.List;
-import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
 import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
 
@@ -41,16 +40,11 @@ class PubRelDecoder extends DemuxDecoder {
         in.resetReaderIndex();
         //Common decoding part
         MessageIDMessage message = new PubRelMessage();
-        if (!decodeCommonHeader(message, in)) {
+        if (!decodeCommonHeader(message, 0x02, in)) {
             in.resetReaderIndex();
             return;
         }
         
-        //if 3.1.1, check the flags (dup, retain and qos == 0)
-        if (message.isDupFlag() || message.isRetainFlag() || message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
-            throw new CorruptedFrameException("Received a PURREL with fixed header flags != b0010");
-        }
-        
         //read  messageIDs
         message.setMessageID(in.readUnsignedShort());
         out.add(message);