diff --git a/ChangeLog.txt b/ChangeLog.txt index ba5badf6..595d841d 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,6 +1,7 @@ Version 0.8: Introduced authenticator and authorizator implementations. - Fixed bug, doesnt' close the channel when connect doesn't contains a valid proto name, issue #65 + Fixed bug, doesn't close the channel when connect doesn't contains a valid proto name, issue #65 + Fixed bug, does not check for zero length topic (filters) in PUBLISH, SUBSCRIBE, and UNSUBSCRIBE, issue #67 Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/PublishDecoder.java index e80b7ba3..d7e5b36b 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/PublishDecoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/PublishDecoder.java @@ -67,7 +67,11 @@ class PublishDecoder extends DemuxDecoder { return; } if (topic.contains("+") || topic.contains("#")) { - throw new CorruptedFrameException("Received a PUBLISH with topic containting wild card chars, topic: " + topic); + throw new CorruptedFrameException("Received a PUBLISH with topic containing wild card chars, topic: " + topic); + } + //check topic is at least one char [MQTT-4.7.3-1] + if (topic.length() == 0) { + throw new CorruptedFrameException("Received a PUBLISH with topic without any character"); } message.setTopicName(topic); diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubscribeDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubscribeDecoder.java index 1522f16b..9fcbc45c 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubscribeDecoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubscribeDecoder.java @@ -41,16 +41,16 @@ class SubscribeDecoder extends DemuxDecoder { //check qos level if (message.getQos() != QOSType.LEAST_ONE) { - throw new CorruptedFrameException("Received Subscribe message with QoS other than LEAST_ONE, was: " + message.getQos()); + throw new CorruptedFrameException("Received SUBSCRIBE message with QoS other than LEAST_ONE, was: " + message.getQos()); } int start = in.readerIndex(); //read messageIDs message.setMessageID(in.readUnsignedShort()); - int readed = in.readerIndex() - start; - while (readed < message.getRemainingLength()) { + int read = in.readerIndex() - start; + while (read < message.getRemainingLength()) { decodeSubscription(in, message); - readed = in.readerIndex()- start; + read = in.readerIndex() - start; } if (message.subscriptions().isEmpty()) { @@ -65,6 +65,10 @@ class SubscribeDecoder extends DemuxDecoder { */ private void decodeSubscription(ByteBuf in, SubscribeMessage message) throws UnsupportedEncodingException { String topic = Utils.decodeString(in); + //check topic is at least one char [MQTT-4.7.3-1] + if (topic.length() == 0) { + throw new CorruptedFrameException("Received a SUBSCRIBE with empty topic filter"); + } 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)); diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java index 083499c2..6ffe3ae5 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoder.java @@ -46,10 +46,15 @@ class UnsubscribeDecoder extends DemuxDecoder { int start = in.readerIndex(); //read messageIDs message.setMessageID(in.readUnsignedShort()); - int readed = in.readerIndex()- start; - while (readed < message.getRemainingLength()) { - message.addTopicFilter(Utils.decodeString(in)); - readed = in.readerIndex()- start; + int read = in.readerIndex() - start; + while (read < message.getRemainingLength()) { + String topicFilter = Utils.decodeString(in); + //check topic is at least one char [MQTT-4.7.3-1] + if (topicFilter.length() == 0) { + throw new CorruptedFrameException("Received an UNSUBSCRIBE with empty topic filter"); + } + message.addTopicFilter(topicFilter); + read = in.readerIndex() - start; } if (message.topicFilters().isEmpty()) { throw new CorruptedFrameException("unsubscribe MUST have got at least 1 topic"); diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/PublishDecoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/PublishDecoderTest.java index 7edde60b..7ecf1e12 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/PublishDecoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/PublishDecoderTest.java @@ -48,7 +48,7 @@ public class PublishDecoderTest { @Before public void setUp() { m_msgdec = new PublishDecoder(); - m_results = new ArrayList(); + m_results = new ArrayList<>(); m_attrMap = new DefaultAttributeMap(); } @@ -57,7 +57,7 @@ public class PublishDecoderTest { m_buff = Unpooled.buffer(14); initHeader(m_buff); - //Excercise + //Exercise m_msgdec.decode(m_attrMap, m_buff, m_results); assertFalse(m_results.isEmpty()); @@ -105,7 +105,7 @@ public class PublishDecoderTest { @Test(expected = CorruptedFrameException.class) public void testDup0WithQoS0_3_1_1() throws Exception { - m_buff = m_buff = preparePubclishWithQosFlags((byte) 0x08); + m_buff = preparePubclishWithQosFlags((byte) 0x08); //Exercise m_msgdec.decode(m_attrMap, m_buff, m_results); @@ -121,7 +121,7 @@ public class PublishDecoderTest { @Test(expected = CorruptedFrameException.class) public void testTopicWithWildCards() throws Exception { - byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte length 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x2B /*+*/, 0x23 /*#*/, //[/top+#] string 2 len + 6 content 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; @@ -134,7 +134,7 @@ public class PublishDecoderTest { @Test public void testBugOnRealCase() throws Exception { - byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte length 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; @@ -198,6 +198,17 @@ public class PublishDecoderTest { Buffer expectedPayload = ByteBuffer.allocate(15).put("Test my payload".getBytes()).flip(); assertEquals(expectedPayload, message.getPayload()); } + + /* + * Check topic is at least one char [MQTT-4.7.3-1] + * */ + @Test(expected = CorruptedFrameException.class) + public void testMinimumTopicLength() throws Exception { + ByteBuf packet = packetWithTopic(""); + + //Exercise + m_msgdec.decode(m_attrMap, packet, m_results); + } private void initHeader(ByteBuf buff) throws IllegalAccessException { ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); @@ -205,6 +216,15 @@ public class PublishDecoderTest { //topic name buff.writeBytes(tmp); } + + private ByteBuf packetWithTopic(String topicName) throws IllegalAccessException { + ByteBuf buff = Unpooled.buffer(14); + ByteBuf topicBuff = Unpooled.buffer(4).writeBytes(Utils.encodeString(topicName)); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4).writeBytes(Utils.encodeRemainingLength(topicBuff.readableBytes())); + //topic name + buff.writeBytes(topicBuff); + return buff; + } private void initHeaderWithMessageID(ByteBuf buff, int messageID) throws IllegalAccessException { ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubscribeDecoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubscribeDecoderTest.java index 693de911..f242ceeb 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubscribeDecoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubscribeDecoderTest.java @@ -38,7 +38,7 @@ public class SubscribeDecoderTest { @Before public void setUp() { m_msgdec = new SubscribeDecoder(); - m_results = new ArrayList(); + m_results = new ArrayList<>(); } @Test(expected = CorruptedFrameException.class) @@ -46,7 +46,7 @@ public class SubscribeDecoderTest { m_buff = Unpooled.buffer(2); initHeaderBadQos(m_buff); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); } @@ -57,7 +57,7 @@ public class SubscribeDecoderTest { SubscribeMessage.Couple c2 = new SubscribeMessage.Couple((byte)1, "c/d/e"); initMultiTopic(m_buff, 123, c1, c2); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); //Verify @@ -72,7 +72,7 @@ public class SubscribeDecoderTest { m_buff = Unpooled.buffer(4); initMultiTopic(m_buff, 123); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); //Verify @@ -88,7 +88,7 @@ public class SubscribeDecoderTest { SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)2, "a/b"); initPollutedTopic(m_buff, 123, c1); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); //Verify @@ -97,6 +97,19 @@ public class SubscribeDecoderTest { assertEquals(2, message.subscriptions().size()); assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType()); } + + /* + * Check topic is at least one char [MQTT-4.7.3-1] + * */ + @Test(expected = CorruptedFrameException.class) + public void testMinimumTopicLength() throws Exception { + m_buff = Unpooled.buffer(4); + SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)2, ""); + initMultiTopic(m_buff, 123, c1); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + } private void initHeaderBadQos(ByteBuf buff) { buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4).writeByte(0); diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoderTest.java index 8b528ae0..98b9ece7 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/UnsubscribeDecoderTest.java @@ -39,7 +39,7 @@ public class UnsubscribeDecoderTest { @Before public void setUp() { m_msgdec = new UnsubscribeDecoder(); - m_results = new ArrayList(); + m_results = new ArrayList<>(); } @@ -48,7 +48,7 @@ public class UnsubscribeDecoderTest { m_buff = Unpooled.buffer(2); initHeaderBadQos(m_buff); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); } @@ -59,7 +59,7 @@ public class UnsubscribeDecoderTest { String topic2 = "c/d/e"; initMultiTopic(m_buff, 123, topic1, topic2); - //Excercise + //Exercise m_msgdec.decode(null, m_buff, m_results); //Verify @@ -74,9 +74,21 @@ public class UnsubscribeDecoderTest { @Test(expected = CorruptedFrameException.class) public void testFailOnEmptyTopic() throws Exception { m_buff = Unpooled.buffer(4); - initMultiTopic(m_buff, 123); + initMultiTopic(m_buff, 123, ""); - //Excercise + //Exercise + m_msgdec.decode(null, m_buff, m_results); + } + + /* + * Check topic is at least one char [MQTT-4.7.3-1] + * */ + @Test(expected = CorruptedFrameException.class) + public void testMinimumTopicLength() throws Exception { + m_buff = Unpooled.buffer(4); + initMultiTopic(m_buff, 123); + + //Exercise m_msgdec.decode(null, m_buff, m_results); } @@ -94,9 +106,9 @@ public class UnsubscribeDecoderTest { assertFalse(m_results.isEmpty()); UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0); assertNotNull(message); - assertEquals(0x01, (int)message.getMessageID()); + assertEquals(0x01, (int) message.getMessageID()); } - + private void initHeaderBadQos(ByteBuf buff) { buff.clear().writeByte(AbstractMessage.UNSUBSCRIBE << 4).writeByte(0);