Merge pull request #1547 from cshannon/AMQ-9810-2

AMQ-9810 - Add additional validation for MQTT wireformat

Author: tabish121

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTCodec.java

@@ -25,7 +25,7 @@
 
 public class MQTTCodec {
 
-    private static final int MAX_MULTIPLIER = (int) Math.pow(2, 21);
+    static final int MAX_MULTIPLIER = (int) Math.pow(2, 21);
 
     private final MQTTFrameSink frameSink;
     private final MQTTWireFormat wireFormat;

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTWireFormat.java

@@ -29,6 +29,8 @@
 import org.fusesource.hawtbuf.Buffer;
 import org.fusesource.mqtt.codec.MQTTFrame;
 
+import static org.apache.activemq.transport.mqtt.MQTTCodec.MAX_MULTIPLIER;
+
 /**
  * Implements marshalling and unmarsalling the <a
  * href="http://mqtt.org/">MQTT</a> protocol.
@@ -92,6 +94,10 @@ public Object unmarshal(DataInput dataIn) throws IOException {
             digit = dataIn.readByte();
             length += (digit & 0x7F) * multiplier;
             multiplier <<= 7;
+            // MQTT protocol limits Remaining Length to 4 bytes
+            if (multiplier == MAX_MULTIPLIER && (digit & 128) != 0) {
+                throw new IOException("Remaining length exceeds 4 bytes");
+            }
         }
         while ((digit & 0x80) != 0);
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.activemq.util.ByteSequence;
 import org.fusesource.hawtbuf.Buffer;
 import org.fusesource.hawtbuf.DataByteArrayInputStream;
 import org.fusesource.hawtbuf.DataByteArrayOutputStream;
@@ -316,7 +317,6 @@ public void testMessageDecodingPerformance() throws Exception {
         LOG.info("Total time to process: {}", TimeUnit.MILLISECONDS.toSeconds(duration));
     }
 
-
     @Test
     public void testParseInvalidRemainingLengthField() throws Exception {
         try {
@@ -330,6 +330,7 @@ public void testParseInvalidRemainingLengthField() throws Exception {
             fail("Parsing should have failed invalid remaining length field");
         } catch (IOException e) {
             // expected
+            assertEquals("Remaining length exceeds 4 bytes", e.getMessage());
         }
     }
 
@@ -344,6 +345,20 @@ public void testPartialReadInvalidRemainingLengthField() throws Exception {
             fail("Parsing should have failed invalid remaining length field");
         } catch (IOException e) {
             // expected
+            assertEquals("Remaining length exceeds 4 bytes", e.getMessage());
+        }
+    }
+
+    @Test
+    public void testUnmarshalInvalidRemainingLengthField() {
+        try {
+            // Test Invalid remaining field checking using the marshaller
+            wireFormat.unmarshal(new ByteSequence(new byte[]{CONNECT.TYPE, (byte) 0x81, (byte) 0x81,
+                    (byte) 0x81, (byte) 0x81}));
+            fail("Parsing should have failed invalid remaining length field");
+        } catch (IOException e) {
+            // expected
+            assertEquals("Remaining length exceeds 4 bytes", e.getMessage());
         }
     }