fix: gracefully decode v3.1.1 packets in v5 sessions (mixed broker compat) (#40)

* fix: gracefully decode v3.1.1 packets in v5 sessions (mixed broker compat)

Some brokers (e.g. mqtt.meshtastic.pt) accept an MQTT 5.0 CONNECT but
respond with 3.1.1-style packets (no properties section in CONNACK,
SUBACK, UNSUBACK, PUBLISH, PUBACK, etc.).  This caused a crash in the
properties decoder: 'Not enough bytes for boolean at offset 4' or
'Unknown property ID: 0x85', killing the read loop and triggering an
infinite reconnect cycle.

Fix: introduce tryDecodePropertiesSection() — a try/catch wrapper around
the strict decoder that returns null on IllegalArgumentException.  Applied
in decodePublish, decodePubAckLike (size >= 4), decodeSubAck, and
decodeUnsubAck so that when v5 properties parsing fails, the decoder
falls back to treating the bytes as 3.1.1 (no properties, bare reason
codes / raw payload).

This mirrors HiveMQ's approach for CONNACK (tryDecodeMqtt3) but extends
it to all response packet types for full compatibility with misconfigured
brokers that advertise v5 but speak v3.1.1.

Also adds bounds checks and hex-dump diagnostics to the properties
decoder for easier debugging of future protocol issues.

Tested successfully against mqtt.meshtastic.pt:1883 — 10/10 publishes
succeed, incoming mesh traffic decoded correctly, zero reconnects.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* chore: bump version to 0.3.6

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jamesarich

gradle.properties

@@ -16,7 +16,7 @@ android.nonTransitiveRClass=true
 #Picked up automatically by com.vanniktech.maven.publish. Override in CI with -PVERSION_NAME=...
 GROUP=org.meshtastic
 POM_ARTIFACT_ID=mqtt-client
-VERSION_NAME=0.3.5
+VERSION_NAME=0.3.6
 #Compose Desktop — allow Homebrew/OpenJDK vendors for local `packageReleaseDmg` runs.
 #CI uses Temurin via setup-java so this is a no-op there.
 compose.desktop.packaging.checkJdkVendor=false

library/src/commonMain/kotlin/org/meshtastic/mqtt/MqttConnection.kt

@@ -58,6 +58,7 @@ import org.meshtastic.mqtt.packet.UnsubAck
 import org.meshtastic.mqtt.packet.Unsubscribe
 import org.meshtastic.mqtt.packet.decodePacket
 import org.meshtastic.mqtt.packet.encode
+import org.meshtastic.mqtt.packet.hexDump
 import kotlin.concurrent.Volatile
 import kotlin.time.TimeMark
 import kotlin.time.TimeSource
@@ -830,7 +831,11 @@ internal class MqttConnection(
                             try {
                                 decodePacket(bytes, version)
                             } catch (e: IllegalArgumentException) {
-                                log.error(TAG) { "Failed to decode packet: ${e.message}" }
+                                log.error(TAG) {
+                                    "Failed to decode packet (${bytes.size} bytes, " +
+                                        "version=$version): ${e.message}. " +
+                                        "Raw hex: ${bytes.hexDump()}"
+                                }
                                 throw e
                             }
                         log.debug(TAG) { "Received ${packet.packetType}" }

library/src/commonMain/kotlin/org/meshtastic/mqtt/packet/MqttDecoder.kt

@@ -250,9 +250,16 @@ private fun decodePublish(
 
     val properties: MqttProperties
     if (version.supportsProperties) {
-        val (props, propsConsumed) = decodePropertiesSection(body, pos)
-        properties = props
-        pos += propsConsumed
+        // Some brokers accept a v5 CONNECT but relay messages from 3.1.1 clients
+        // without a properties section. Try v5 decode; fall back to treating
+        // remaining bytes as raw payload (3.1.1 style).
+        val propertiesResult = tryDecodePropertiesSection(body, pos)
+        if (propertiesResult != null) {
+            properties = propertiesResult.first
+            pos += propertiesResult.second
+        } else {
+            properties = MqttProperties.EMPTY
+        }
     } else {
         properties = MqttProperties.EMPTY
     }
@@ -297,7 +304,7 @@ private fun <T : MqttPacket> decodePubAckLike(
         return factory(packetId, reasonCode, MqttProperties.EMPTY)
     }
 
-    val (properties, _) = decodePropertiesSection(body, 3)
+    val properties = tryDecodePropertiesSection(body, 3)?.first ?: MqttProperties.EMPTY
     return factory(packetId, reasonCode, properties)
 }
 
@@ -376,9 +383,16 @@ private fun decodeSubAck(
 
     val properties: MqttProperties
     if (version.supportsProperties) {
-        val (props, propsConsumed) = decodePropertiesSection(body, pos)
-        properties = props
-        pos += propsConsumed
+        // Some brokers accept a v5 CONNECT but respond with 3.1.1-style SUBACKs
+        // (no properties section). Detect this by trying v5 decode first; if properties
+        // parsing fails, fall back to treating remaining bytes as bare return codes.
+        val propertiesResult = tryDecodePropertiesSection(body, pos)
+        if (propertiesResult != null) {
+            properties = propertiesResult.first
+            pos += propertiesResult.second
+        } else {
+            properties = MqttProperties.EMPTY
+        }
     } else {
         properties = MqttProperties.EMPTY
     }
@@ -461,8 +475,16 @@ private fun decodeUnsubAck(
         )
     }
 
-    val (properties, propsConsumed) = decodePropertiesSection(body, pos)
-    pos += propsConsumed
+    // Some brokers accept a v5 CONNECT but respond with 3.1.1-style packets.
+    // Try v5 properties decode; fall back to bare reason codes on failure.
+    val propertiesResult = tryDecodePropertiesSection(body, pos)
+    val properties: MqttProperties
+    if (propertiesResult != null) {
+        properties = propertiesResult.first
+        pos += propertiesResult.second
+    } else {
+        properties = MqttProperties.EMPTY
+    }
 
     val reasonCodes = mutableListOf<ReasonCode>()
     while (pos < body.size) {
@@ -548,7 +570,25 @@ private fun decodePropertiesSection(
     return props to (lengthResult.bytesConsumed + propsLength)
 }
 
-private fun ByteArray.toHexDump(): String {
+/**
+ * Try to decode a properties section. Returns null if the bytes don't form a valid
+ * properties section — this handles brokers that accept a v5 CONNECT but respond with
+ * 3.1.1-style packets (no properties section in SUBACK/UNSUBACK/etc.).
+ */
+@Suppress("SwallowedException")
+private fun tryDecodePropertiesSection(
+    bytes: ByteArray,
+    offset: Int,
+): Pair<MqttProperties, Int>? =
+    try {
+        decodePropertiesSection(bytes, offset)
+    } catch (_: IllegalArgumentException) {
+        null
+    }
+
+private fun ByteArray.toHexDump(): String = hexDump()
+
+internal fun ByteArray.hexDump(): String {
     fun Byte.hex(): String {
         val i = toInt() and 0xFF
         val hi = "0123456789abcdef"[i ushr 4]

library/src/commonMain/kotlin/org/meshtastic/mqtt/packet/MqttProperties.kt

@@ -242,10 +242,20 @@ internal fun decodeProperties(
     var sharedSubscriptionAvailable: Boolean? = null
 
     while (pos < end) {
+        require(pos < bytes.size) {
+            "Property section overflows buffer: pos=$pos, end=$end, " +
+                "bodySize=${bytes.size}. Hex: ${bytes.hexDump()}"
+        }
         val idResult = VariableByteInt.decode(bytes, pos)
         pos += idResult.bytesConsumed
         val propertyId = idResult.value
 
+        // Bounds-check: the property value must start within the declared section
+        require(pos <= end && pos <= bytes.size) {
+            "Property 0x${propertyId.toString(16)} value overflows section: " +
+                "pos=$pos, end=$end, bodySize=${bytes.size}. Hex: ${bytes.hexDump()}"
+        }
+
         // Multi-occurrence properties: User Property and Subscription Identifier
         val isMultiOccurrence =
             propertyId == PropertyId.USER_PROPERTY ||
@@ -445,9 +455,15 @@ private fun decodeBooleanByte(
     bytes: ByteArray,
     offset: Int,
 ): Boolean {
-    require(offset < bytes.size) { "Not enough bytes for boolean at offset $offset" }
+    require(offset < bytes.size) {
+        "Not enough bytes for boolean at offset $offset " +
+            "(bodySize=${bytes.size}). Hex: ${bytes.hexDump()}"
+    }
     val b = bytes[offset].toInt() and 0xFF
-    require(b == 0 || b == 1) { "Malformed boolean value 0x${b.toString(16)} at offset $offset (§1.5.1.4)" }
+    require(b == 0 || b == 1) {
+        "Malformed boolean value 0x${b.toString(16)} at offset $offset (§1.5.1.4). " +
+            "Hex: ${bytes.hexDump()}"
+    }
     return b == 1
 }
 

library/src/commonTest/kotlin/org/meshtastic/mqtt/MqttConnectionTest.kt

@@ -1145,7 +1145,7 @@ class MqttConnectionTest {
         }
 
     @Test
-    fun v5ConnectionDisconnectsOnMalformedPacket() =
+    fun v5ConnectionDecodesV311PublishGracefully() =
         runTest {
             val transport = FakeTransport()
             val connection = MqttConnection(transport, defaultConfig(), this)
@@ -1154,8 +1154,9 @@ class MqttConnectionTest {
             connection.connect(endpoint)
             advanceUntilIdle()
 
-            // Send a v3.1.1-encoded packet to a v5 connection — should be treated as
-            // a decode error and disconnect, not silently downgrade protocol version.
+            // Some brokers accept a v5 CONNECT but relay PUBLISH messages from
+            // v3.1.1 clients without a properties section.  The decoder should
+            // fall back gracefully (empty properties) instead of disconnecting.
             val v311Publish =
                 Publish(
                     topicName = "test/topic",
@@ -1166,10 +1167,13 @@ class MqttConnectionTest {
             advanceUntilIdle()
 
             val state = connection.connectionState.value
-            assertIs<ConnectionState.Disconnected>(
+            assertIs<ConnectionState.Connected>(
                 state,
-                "Connection should disconnect on malformed packet, not downgrade protocol",
+                "Connection should stay connected when receiving a v3.1.1 PUBLISH in a v5 session",
             )
+
+            connection.disconnect()
+            advanceUntilIdle()
         }
 
     // --- Session Present validation (§3.2.2.1.1) ---