This is an automated email from the ASF dual-hosted git repository.

cshannon pushed a commit to branch activemq-5.19.x
in repository https://gitbox.apache.org/repos/asf/activemq.git


The following commit(s) were added to refs/heads/activemq-5.19.x by this push:
     new cfa37f9975 fix: MQTTWireFormat rejecting valid 4-byte remaining 
lengths (#2517) (#2522)
cfa37f9975 is described below

commit cfa37f9975b1094ad6f30d53cf0e8fe27048468f
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Fri Sep 4 10:47:15 2026 -0400

    fix: MQTTWireFormat rejecting valid 4-byte remaining lengths (#2517) (#2522)
    
    Fix MQTTWireFormat rejecting valid 4-byte remaining lengths
    
    (cherry picked from commit 15a182dca5eee5b9370f74a324e6220256530cb4)
    
    Co-authored-by: Hailey <[email protected]>
---
 .../activemq/transport/mqtt/MQTTWireFormat.java    |  4 +-
 .../activemq/transport/mqtt/MQTTCodecTest.java     | 53 ++++++++++++++++++++++
 2 files changed, 55 insertions(+), 2 deletions(-)

diff --git 
a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTWireFormat.java
 
b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTWireFormat.java
index 9e79fd25f4..ff00c6c923 100644
--- 
a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTWireFormat.java
+++ 
b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTWireFormat.java
@@ -92,12 +92,12 @@ public class MQTTWireFormat implements WireFormat {
         int length = 0;
         do {
             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");
             }
+            length += (digit & 0x7F) * multiplier;
+            multiplier <<= 7;
         }
         while ((digit & 0x80) != 0);
 
diff --git 
a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java
 
b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java
index 994bff4e67..6bc1314733 100644
--- 
a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java
+++ 
b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java
@@ -362,4 +362,57 @@ public class MQTTCodecTest {
         }
     }
 
+    @Test
+    public void testCodecAcceptsFourByteRemainingLength() throws Exception {
+        // 2,097,152 is the smallest value that requires a 4-byte Remaining 
Length.
+        // Payload size = target RL - variable header overhead (2 topic length 
+ topic + 2 packet id)
+        final String topic = "TOPIC";
+        final int overhead = 2 + topic.length() + 2;  // topic length field + 
topic + message id (QoS 1)
+        final int payloadSize = 2_097_152 - overhead;
+
+        PUBLISH publish = new PUBLISH();
+        publish.qos(QoS.AT_LEAST_ONCE);
+        publish.messageId((short) 1);
+        publish.topicName(new UTF8Buffer(topic));
+        publish.payload(new Buffer(new byte[payloadSize]));
+
+        Buffer marshalled;
+        try (DataByteArrayOutputStream output = new 
DataByteArrayOutputStream()) {
+            wireFormat.marshal(publish.encode(), output);
+            marshalled = output.toBuffer();
+        }
+
+        try (DataByteArrayInputStream input = new 
DataByteArrayInputStream(marshalled)) {
+            codec.parse(input, marshalled.length());
+        }
+
+        assertEquals("Expected one frame from a valid 4-byte Remaining 
Length", 1, frames.size());
+        PUBLISH decoded = new PUBLISH().decode(frames.get(0));
+        assertEquals(payloadSize, decoded.payload().length());
+    }
+
+    @Test
+    public void testUnmarshalAcceptsFourByteRemainingLength() throws Exception 
{
+        final String topic = "TOPIC";
+        final int overhead = 2 + topic.length() + 2;
+        final int payloadSize = 2_097_152 - overhead;
+
+        PUBLISH publish = new PUBLISH();
+        publish.qos(QoS.AT_LEAST_ONCE);
+        publish.messageId((short) 1);
+        publish.topicName(new UTF8Buffer(topic));
+        publish.payload(new Buffer(new byte[payloadSize]));
+
+        Buffer marshalled;
+        try (DataByteArrayOutputStream output = new 
DataByteArrayOutputStream()) {
+            wireFormat.marshal(publish.encode(), output);
+            marshalled = output.toBuffer();
+        }
+
+        MQTTFrame frame = (MQTTFrame) wireFormat.unmarshal(
+                new ByteSequence(marshalled.data, marshalled.offset, 
marshalled.length));
+        PUBLISH decoded = new PUBLISH().decode(frame);
+        assertEquals(payloadSize, decoded.payload().length());
+    }
+
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact


Reply via email to