diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java index 98caa8da..19f4c77f 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java @@ -18,8 +18,14 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; +import io.netty.util.AttributeMap; import java.io.UnsupportedEncodingException; import java.util.List; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1_1; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** @@ -27,9 +33,16 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage; * @author andrea */ public class ConnectDecoder extends DemuxDecoder { - + + static final AttributeKey CONNECT_STATUS = new AttributeKey("connected"); + @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws UnsupportedEncodingException { + decode((AttributeMap)ctx, in, out); + } + +// @Override + void decode(AttributeMap ctx, ByteBuf in, List out) throws UnsupportedEncodingException { in.resetReaderIndex(); //Common decoding part ConnectMessage message = new ConnectMessage(); @@ -40,24 +53,70 @@ public class ConnectDecoder extends DemuxDecoder { int remainingLength = message.getRemainingLength(); int start = in.readerIndex(); - //Connect specific decoding part - //ProtocolName 8 bytes - if (in.readableBytes() < 12) { - in.resetReaderIndex(); - return; + int protocolNameLen = in.readUnsignedShort(); + byte[] encProtoName; + String protoName; + Attribute versionAttr = ctx.attr(MQTTDecoder.PROTOCOL_VERSION); + switch (protocolNameLen) { + case 6: + //MQTT version 3.1 "MQIsdp" + //ProtocolName 8 bytes or 6 bytes + if (in.readableBytes() < 10) { + in.resetReaderIndex(); + return; + } + + encProtoName = new byte[6]; + in.readBytes(encProtoName); + protoName = new String(encProtoName, "UTF-8"); + if (!"MQIsdp".equals(protoName)) { + in.resetReaderIndex(); + throw new CorruptedFrameException("Invalid protoName: " + protoName); + } + message.setProtocolName(protoName); + + versionAttr.set((int) VERSION_3_1); + break; + case 4: + //MQTT version 3.1.1 "MQTT" + //ProtocolName 6 bytes + if (in.readableBytes() < 8) { + in.resetReaderIndex(); + return; + } + encProtoName = new byte[4]; + in.readBytes(encProtoName); + protoName = new String(encProtoName, "UTF-8"); + if (!"MQTT".equals(protoName)) { + in.resetReaderIndex(); + throw new CorruptedFrameException("Invalid protoName: " + protoName); + } + message.setProtocolName(protoName); + versionAttr.set((int) VERSION_3_1_1); + break; + default: + //protocol broken + throw new CorruptedFrameException("Invalid protoName size: " + protocolNameLen); } - byte[] encProtoName = new byte[6]; - in.skipBytes(2); //size, is 0x06 - in.readBytes(encProtoName); - String protoName = new String(encProtoName, "UTF-8"); - if (!"MQIsdp".equals(protoName)) { - in.resetReaderIndex(); - throw new CorruptedFrameException("Invalid protoName: " + protoName); - } - message.setProtocolName(protoName); - - //ProtocolVersion 1 byte (value 0x03) + + //ProtocolVersion 1 byte (value 0x03 for 3.1, 0x04 for 3.1.1) message.setProcotolVersion(in.readByte()); + if (message.getProcotolVersion() == VERSION_3_1_1) { + //if 3.1.1, check the flags (dup, retain and qos == 0) + if (message.isDupFlag() || message.isRetainFlag() || message.getQos() != AbstractMessage.QOSType.MOST_ONE) { + throw new CorruptedFrameException("Received a CONNECT with fixed header flags != 0"); + } + + //check if this is another connect from the same client on the same session + Attribute connectAttr = ctx.attr(ConnectDecoder.CONNECT_STATUS); + Boolean alreadyConnected = connectAttr.get(); + if (alreadyConnected == null) { + //never set + connectAttr.set(true); + } else if (alreadyConnected) { + throw new CorruptedFrameException("Received a second CONNECT on the same network connection"); + } + } //Connection flag byte connFlags = in.readByte(); @@ -88,7 +147,8 @@ public class ConnectDecoder extends DemuxDecoder { int keepAlive = in.readUnsignedShort(); message.setKeepAlive(keepAlive); - if (remainingLength == 12) { + if ((remainingLength == 12 && message.getProcotolVersion() == 3) || + (remainingLength == 10 && message.getProcotolVersion() == 4)) { out.add(message); return; } @@ -121,7 +181,7 @@ public class ConnectDecoder extends DemuxDecoder { message.setWillMessage(willMessage); } - //Compatibility check wieth v3.0, remaining length has precedence over + //Compatibility check with v3.0, remaining length has precedence over //the user and password flags int readed = in.readerIndex() - start; if (readed == remainingLength) { diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java index 8eb52677..0bfbdf87 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java @@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.AttributeKey; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -30,7 +31,10 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage; */ public class MQTTDecoder extends ByteToMessageDecoder { - private Map m_decoderMap = new HashMap(); + //3 = 3.1, 4 = 3.1.1 + static final AttributeKey PROTOCOL_VERSION = new AttributeKey("version"); + + private final Map m_decoderMap = new HashMap(); public MQTTDecoder() { m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder()); diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java index ffdc4b13..ae34484b 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java @@ -30,6 +30,9 @@ public class Utils { public static final int MAX_LENGTH_LIMIT = 268435455; + public static final byte VERSION_3_1 = 3; + public static final byte VERSION_3_1_1 = 4; + static byte readMessageType(ByteBuf in) { byte h1 = in.readByte(); byte messageType = (byte) ((h1 & 0x00F0) >> 4); diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java index 7c46671c..624d1074 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java @@ -18,9 +18,14 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.Attribute; +import io.netty.util.AttributeMap; +import io.netty.util.DefaultAttributeMap; import java.io.UnsupportedEncodingException; import java.util.ArrayList; import java.util.List; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1_1; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import static org.junit.Assert.assertEquals; @@ -35,10 +40,12 @@ public class ConnectDecoderTest { ByteBuf m_buff; ConnectDecoder m_msgdec; + AttributeMap attrMap; @Before public void setUp() { m_msgdec = new ConnectDecoder(); + attrMap = new DefaultAttributeMap(); } @Test @@ -48,13 +55,59 @@ public class ConnectDecoderTest { List results = new ArrayList(); //Excercise - m_msgdec.decode(null, m_buff, results); + m_msgdec.decode(this.attrMap, m_buff, results); //Verify assertFalse(results.isEmpty()); verifyBaseHeader((ConnectMessage)results.get(0)); } + @Test + public void testBaseHeader_311() throws UnsupportedEncodingException { + m_buff = Unpooled.buffer(12); + initBaseHeader311(m_buff); + List results = new ArrayList(); + + //Excercise + m_msgdec.decode(this.attrMap, m_buff, results); + + //Verify + assertFalse(results.isEmpty()); + verifyBaseHeader311((ConnectMessage)results.get(0)); + Attribute attr = this.attrMap.attr(MQTTDecoder.PROTOCOL_VERSION); + assertEquals(VERSION_3_1_1, attr.get().intValue()); + } + + @Test(expected = CorruptedFrameException.class) + public void testBaseHeader_311_withFlagsTouched() throws UnsupportedEncodingException { + m_buff = Unpooled.buffer(12); + initBaseHeader311_withFixedFlags(m_buff, (byte) 0x01); + List results = new ArrayList(); + + //Excercise + m_msgdec.decode(this.attrMap, m_buff, results); + + //Verify + assertFalse(results.isEmpty()); + verifyBaseHeader311((ConnectMessage)results.get(0)); + Attribute attr = this.attrMap.attr(MQTTDecoder.PROTOCOL_VERSION); + assertEquals(VERSION_3_1_1, attr.get().intValue()); + } + + @Test(expected = CorruptedFrameException.class) + public void testDoubleConnectInTheSameSession() throws Exception { + //setup the connection session as already connected + Attribute connectAttr = this.attrMap.attr(ConnectDecoder.CONNECT_STATUS); + connectAttr.set(true); + + m_buff = Unpooled.buffer(12); + initBaseHeader311(m_buff); + List results = new ArrayList(); + + //Excercise + m_msgdec.decode(this.attrMap, m_buff, results); + } + @Test public void testBaseHeader_ClientID() throws UnsupportedEncodingException, Exception { m_buff = Unpooled.buffer(40); @@ -65,7 +118,7 @@ public class ConnectDecoderTest { List results = new ArrayList(); //Excercise - m_msgdec.decode(null, m_buff, results); + m_msgdec.decode(this.attrMap, m_buff, results); //Verify assertFalse(results.isEmpty()); @@ -89,7 +142,7 @@ public class ConnectDecoderTest { List results = new ArrayList(); //Excercise - m_msgdec.decode(null, m_buff, results); + m_msgdec.decode(this.attrMap, m_buff, results); //Verify assertFalse(results.isEmpty()); @@ -104,21 +157,21 @@ public class ConnectDecoderTest { } @Test(expected = CorruptedFrameException.class) - public void testBadFlagUserPwd() throws UnsupportedEncodingException, Exception { + public void testBadFlagUserPwd() throws Exception { m_buff = Unpooled.buffer(14); m_buff.writeByte((AbstractMessage.CONNECT << 4)).writeByte(12); //Proto name encodeString(m_buff, "MQIsdp"); //version - m_buff.writeByte(3); + m_buff.writeByte(VERSION_3_1); //conn flags m_buff.writeByte(0x4E); //sets user to false and password to true //keepAlive m_buff.writeByte(0).writeByte(0x0A); - List results = new ArrayList(); + List results = new ArrayList(); //Excercise - m_msgdec.decode(null, m_buff, results); + m_msgdec.decode(this.attrMap, m_buff, results); } @@ -138,6 +191,22 @@ public class ConnectDecoderTest { buff.writeBytes(new byte[]{(byte)0, (byte) 0x0A}); } + private void initBaseHeader311(ByteBuf buff) throws UnsupportedEncodingException { + initBaseHeader311_withFixedFlags(buff, (byte) 0); + } + + private void initBaseHeader311_withFixedFlags(ByteBuf buff, byte fixedFlags) throws UnsupportedEncodingException { + buff.clear().writeByte((byte)(AbstractMessage.CONNECT << 4) | fixedFlags).writeByte((byte)0x0A); + //Proto name + encodeString(buff, "MQTT"); + //version + buff.writeByte(VERSION_3_1_1); + //conn flags + buff.writeByte(0xCE); + //keepAlive + buff.writeBytes(new byte[]{(byte)0, (byte) 0x0A}); + } + /** * Encode and insert the given string into the given buff */ @@ -149,7 +218,20 @@ public class ConnectDecoderTest { private void verifyBaseHeader(ConnectMessage connMessage) { assertNotNull(connMessage); assertEquals("MQIsdp", connMessage.getProtocolName()); - assertEquals(3, connMessage.getProcotolVersion()); + assertEquals(VERSION_3_1, connMessage.getProcotolVersion()); + assertTrue(connMessage.isUserFlag()); + assertTrue(connMessage.isPasswordFlag()); + assertTrue(connMessage.isCleanSession()); + assertEquals(10, connMessage.getKeepAlive()); + assertTrue(connMessage.isWillFlag()); + assertFalse(connMessage.isWillRetain()); + assertEquals(1, connMessage.getWillQos()); + } + + private void verifyBaseHeader311(ConnectMessage connMessage) { + assertNotNull(connMessage); + assertEquals("MQTT", connMessage.getProtocolName()); + assertEquals(VERSION_3_1_1, connMessage.getProcotolVersion()); assertTrue(connMessage.isUserFlag()); assertTrue(connMessage.isPasswordFlag()); assertTrue(connMessage.isCleanSession());