diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java index a009647d..f4eba67c 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java +++ b/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 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; } - } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubRelDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubRelDecoder.java index 6e2031ac..10b53341 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubRelDecoder.java +++ b/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);