anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduced version, reserved flags untouchable and double connect 3.1.1 proto version

Browse Source
This commit is contained in:
selva.andre
2014-07-21 23:14:26 +02:00
parent d4e13e8543
commit 465d7123f6
4 changed files with 177 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java
+79 -19
View File
@@ -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<Boolean> CONNECT_STATUS = new AttributeKey<Boolean>("connected");
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws UnsupportedEncodingException {
decode((AttributeMap)ctx, in, out);
}
// @Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> 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<Integer> 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<Boolean> 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) {
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java
+5 -1
View File
@@ -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<Byte, DemuxDecoder> m_decoderMap = new HashMap<Byte, DemuxDecoder>();
//3 = 3.1, 4 = 3.1.1
static final AttributeKey<Integer> PROTOCOL_VERSION = new AttributeKey<Integer>("version");
private final Map<Byte, DemuxDecoder> m_decoderMap = new HashMap<Byte, DemuxDecoder>();
public MQTTDecoder() {
m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder());
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java
+3
View File
@@ -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);
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java
+90 -8
View File
@@ -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<Object> results = new ArrayList<Object >();
//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<Object> results = new ArrayList<Object>();
//Excercise
m_msgdec.decode(this.attrMap, m_buff, results);
//Verify
assertFalse(results.isEmpty());
verifyBaseHeader311((ConnectMessage)results.get(0));
Attribute<Integer> 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<Object> results = new ArrayList<Object>();
//Excercise
m_msgdec.decode(this.attrMap, m_buff, results);
//Verify
assertFalse(results.isEmpty());
verifyBaseHeader311((ConnectMessage)results.get(0));
Attribute<Integer> 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<Boolean> connectAttr = this.attrMap.attr(ConnectDecoder.CONNECT_STATUS);
connectAttr.set(true);
m_buff = Unpooled.buffer(12);
initBaseHeader311(m_buff);
List<Object> results = new ArrayList<Object>();
//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<Object> results = new ArrayList<Object >();
//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<Object> results = new ArrayList<Object >();
//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<Object> results = new ArrayList<Object >();
List<Object> results = new ArrayList<Object>();
//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());