Merge branch 'backport_v5.5' into 'release/v5.5'

MR: Backport fixes to 5.5
See merge request espressif/esp-mqtt!315

lib/mqtt5_msg.c

@@ -6,6 +6,7 @@
 #include "esp_log.h"
 
 #define MQTT5_MAX_FIXED_HEADER_SIZE 5
+#define MQTT5_MAX_PROPERTY_STRING_LEN (16 * 1024)
 
 static const char *TAG = "mqtt5_msg";
 
@@ -51,17 +52,33 @@ static size_t get_variable_len(uint8_t *buffer, size_t offset, size_t buffer_len
 {
     *len_bytes = 0;
     size_t len = 0, i = 0;
-    for (i = offset; i < buffer_length; i ++) {
-        len += (buffer[i] & 0x7f) << (7 * (i - offset));
+
+    // MQTT Variable Byte Integer is max 4 bytes (MQTT v5 spec).
+    // Limit decoding to 4 bytes to avoid undefined shift behavior on malformed inputs.
+    for (i = offset; i < buffer_length && (i - offset) < 4; i ++) {
+        len += ((size_t)(buffer[i] & 0x7f)) << (7 * (i - offset));
+
         if ((buffer[i] & 0x80) == 0) {
             i ++;
             break;
         }
     }
+
+    // If the varint didn't terminate within 4 bytes, treat as invalid (0 bytes consumed).
+    if ((i - offset) == 4 && i <= buffer_length && (buffer[i - 1] & 0x80)) {
+        *len_bytes = 0;
+        return 0;
+    }
+
     *len_bytes = i - offset;
     return len;
 }
 
+static bool mqtt5_property_has_bytes(size_t property_offset, size_t needed, size_t property_len)
+{
+    return property_offset <= property_len && needed <= (property_len - property_offset);
+}
+
 static int update_property_len_value(mqtt_connection_t *connection, size_t property_len, int property_offset)
 {
     uint8_t encoded_lens[4] = {0}, len_bytes = 0;
@@ -225,19 +242,48 @@ static mqtt5_user_property_handle_t mqtt5_msg_get_user_property(uint8_t *buffer,
         uint8_t property_id = property[property_offset ++];
         switch (property_id) {
         case MQTT5_PROPERTY_REASON_STRING: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, buffer_length)) {
+                goto err;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, buffer_length)) {
+                goto err;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_REASON_STRING %.*s", len, &property[property_offset]);
             property_offset += len;
             continue;
         case MQTT5_PROPERTY_USER_PROPERTY: {
             uint8_t *key = NULL, *value = NULL;
             size_t key_len = 0, value_len = 0;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, buffer_length)) {
+                goto err;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, buffer_length)) {
+                goto err;
+            }
+
             key = &property[property_offset];
             key_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY key: %.*s", key_len, (char *)key);
             property_offset += len;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, buffer_length)) {
+                goto err;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, buffer_length)) {
+                goto err;
+            }
+
             value = &property[property_offset];
             value_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY value: %.*s", value_len, (char *)value);
@@ -334,19 +380,43 @@ char *mqtt5_get_publish_property_payload(uint8_t *buffer, size_t buffer_length,
         uint8_t property_id = property[property_offset ++];
         switch (property_id) {
         case MQTT5_PROPERTY_PAYLOAD_FORMAT_INDICATOR:
+            if (!mqtt5_property_has_bytes(property_offset, 1, *property_len)) {
+                return NULL;
+            }
+
             resp_property->payload_format_indicator = property[property_offset ++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_PAYLOAD_FORMAT_INDICATOR %d", resp_property->payload_format_indicator);
             continue;
         case MQTT5_PROPERTY_MESSAGE_EXPIRY_INTERVAL:
-            MQTT5_CONVERT_ONE_BYTE_TO_FOUR(resp_property->message_expiry_interval, property[property_offset ++], property[property_offset ++], property[property_offset ++], property[property_offset ++])
+            if (!mqtt5_property_has_bytes(property_offset, 4, *property_len)) {
+                return NULL;
+            }
+
+            MQTT5_CONVERT_ONE_BYTE_TO_FOUR(resp_property->message_expiry_interval, property[property_offset ++],
+                                           property[property_offset ++], property[property_offset ++], property[property_offset ++])
             ESP_LOGD(TAG, "MQTT5_PROPERTY_MESSAGE_EXPIRY_INTERVAL %"PRIu32, resp_property->message_expiry_interval);
             continue;
         case MQTT5_PROPERTY_TOPIC_ALIAS:
+            if (!mqtt5_property_has_bytes(property_offset, 2, *property_len)) {
+                return NULL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(resp_property->topic_alias, property[property_offset ++], property[property_offset ++])
             ESP_LOGD(TAG, "MQTT5_PROPERTY_TOPIC_ALIAS %d", resp_property->topic_alias);
             continue;
         case MQTT5_PROPERTY_RESPONSE_TOPIC:
-            MQTT5_CONVERT_ONE_BYTE_TO_TWO(resp_property->response_topic_len, property[property_offset ++], property[property_offset ++])
+            if (!mqtt5_property_has_bytes(property_offset, 2, *property_len)) {
+                return NULL;
+            }
+
+            MQTT5_CONVERT_ONE_BYTE_TO_TWO(resp_property->response_topic_len, property[property_offset ++],
+                                          property[property_offset ++])
+
+            if (resp_property->response_topic_len > MQTT5_MAX_PROPERTY_STRING_LEN ||
+                    !mqtt5_property_has_bytes(property_offset, resp_property->response_topic_len, *property_len)) {
+                return NULL;
+            }
+
             resp_property->response_topic = (char *)(property + property_offset);
             property_offset += resp_property->response_topic_len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_RESPONSE_TOPIC %.*s", resp_property->response_topic_len, resp_property->response_topic);
@@ -358,7 +428,12 @@ char *mqtt5_get_publish_property_payload(uint8_t *buffer, size_t buffer_length,
             ESP_LOGD(TAG, "MQTT5_PROPERTY_CORRELATION_DATA length %d", resp_property->correlation_data_len);
             continue;
         case MQTT5_PROPERTY_SUBSCRIBE_IDENTIFIER:
-            resp_property->subscribe_id = get_variable_len(property, property_offset, buffer_length, &len_bytes);
+            resp_property->subscribe_id = get_variable_len(property, property_offset, *property_len, &len_bytes);
+
+            if (!mqtt5_property_has_bytes(property_offset, len_bytes, *property_len)) {
+                return NULL;
+            }
+
             property_offset += len_bytes;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SUBSCRIBE_IDENTIFIER %d", resp_property->subscribe_id);
             continue;
@@ -371,12 +446,32 @@ char *mqtt5_get_publish_property_payload(uint8_t *buffer, size_t buffer_length,
         case MQTT5_PROPERTY_USER_PROPERTY: {
             uint8_t *key = NULL, *value = NULL;
             size_t key_len = 0, value_len = 0;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, *property_len)) {
+                return NULL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, *property_len)) {
+                return NULL;
+            }
+
             key = &property[property_offset];
             key_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY key: %.*s", key_len, (char *)key);
             property_offset += len;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, *property_len)) {
+                return NULL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, *property_len)) {
+                return NULL;
+            }
+
             value = &property[property_offset];
             value_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY value: %.*s", value_len, (char *)value);
@@ -390,7 +485,16 @@ char *mqtt5_get_publish_property_payload(uint8_t *buffer, size_t buffer_length,
             continue;
         }
         case MQTT5_PROPERTY_REASON_STRING: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, *property_len)) {
+                return NULL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, *property_len)) {
+                return NULL;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_REASON_STRING %.*s", len, &property[property_offset]);
             property_offset += len;
             continue;
@@ -424,6 +528,11 @@ char *mqtt5_get_suback_data(uint8_t *buffer, size_t *length, mqtt5_user_property
     if (offset < totlen) {
         size_t property_len = get_variable_len(buffer, offset, totlen, &len_bytes);
         offset += len_bytes;
+
+        if (property_len > (totlen - offset)) {
+            goto err;
+        }
+
         *user_property = mqtt5_msg_get_user_property(buffer + offset, property_len);
         offset += property_len;
         if (offset < totlen) {
@@ -453,6 +562,12 @@ char *mqtt5_get_puback_data(uint8_t *buffer, size_t *length, mqtt5_user_property
         if (offset < totlen) {
             size_t property_len = get_variable_len(buffer, offset, totlen, &len_bytes);
             offset += len_bytes;
+
+            if (property_len > (totlen - offset)) {
+                *length = 0;
+                return NULL;
+            }
+
             *user_property = mqtt5_msg_get_user_property(buffer + offset, property_len);
         }
         return data;
@@ -597,12 +712,23 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
     size_t property_len = get_variable_len(buffer, offset, buffer_len, &len_bytes);
     offset += len_bytes;
     uint16_t property_offset = 0, len = 0;
+
+    if (property_len > (buffer_len - offset)) {
+        ESP_LOGE(TAG, "Property length %d exceeds buffer bounds %d", property_len, buffer_len - offset);
+        return ESP_FAIL;
+    }
+
     uint8_t *property = (buffer + offset);
     while (property_offset < property_len) {
         uint8_t property_id = property[property_offset ++];
         switch (property_id) {
         case MQTT5_PROPERTY_SESSION_EXPIRY_INTERVAL:
-            MQTT5_CONVERT_ONE_BYTE_TO_FOUR(connection_property->session_expiry_interval, property[property_offset ++], property[property_offset ++], property[property_offset ++], property[property_offset ++])
+            if (!mqtt5_property_has_bytes(property_offset, 4, property_len)) {
+                return ESP_FAIL;
+            }
+
+            MQTT5_CONVERT_ONE_BYTE_TO_FOUR(connection_property->session_expiry_interval, property[property_offset ++],
+                                           property[property_offset ++], property[property_offset ++], property[property_offset ++])
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SESSION_EXPIRY_INTERVAL %"PRIu32, connection_property->session_expiry_interval);
             continue;
         case MQTT5_PROPERTY_RECEIVE_MAXIMUM:
@@ -610,10 +736,18 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             ESP_LOGD(TAG, "MQTT5_PROPERTY_RECEIVE_MAXIMUM %d", resp_property->receive_maximum);
             continue;
         case MQTT5_PROPERTY_MAXIMUM_QOS:
+            if (!mqtt5_property_has_bytes(property_offset, 1, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->max_qos = property[property_offset ++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_MAXIMUM_QOS %d", resp_property->max_qos);
             continue;
         case MQTT5_PROPERTY_RETAIN_AVAILABLE:
+            if (!mqtt5_property_has_bytes(property_offset, 1, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->retain_available = property[property_offset ++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_RETAIN_AVAILABLE %d", resp_property->retain_available);
             continue;
@@ -622,7 +756,17 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             ESP_LOGD(TAG, "MQTT5_PROPERTY_MAXIMUM_PACKET_SIZE %"PRIu32, resp_property->maximum_packet_size);
             continue;
         case MQTT5_PROPERTY_ASSIGNED_CLIENT_IDENTIFIER:
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                ESP_LOGE(TAG, "Sub-length %d exceeds property bounds", len);
+                return ESP_FAIL;
+            }
+
             if (connection_info->client_id) {
                 free(connection_info->client_id);
             }
@@ -641,19 +785,48 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             ESP_LOGD(TAG, "MQTT5_PROPERTY_TOPIC_ALIAS_MAXIMIM %d", resp_property->topic_alias_maximum);
             continue;
         case MQTT5_PROPERTY_REASON_STRING: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_REASON_STRING %.*s", len, &property[property_offset]);
             property_offset += len;
             continue;
         case MQTT5_PROPERTY_USER_PROPERTY: {
             uint8_t *key = NULL, *value = NULL;
             size_t key_len = 0, value_len = 0;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             key = &property[property_offset];
             key_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY key: %.*s", key_len, (char *)key);
             property_offset += len;
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             value = &property[property_offset];
             value_len = len;
             ESP_LOGD(TAG, "MQTT5_PROPERTY_USER_PROPERTY value: %.*s", value_len, (char *)value);
@@ -667,18 +840,34 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             continue;
         }
         case MQTT5_PROPERTY_WILDCARD_SUBSCR_AVAILABLE:
+            if (!mqtt5_property_has_bytes(property_offset, 1, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->wildcard_subscribe_available = property[property_offset++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_WILDCARD_SUBSCR_AVAILABLE %d", resp_property->wildcard_subscribe_available);
             continue;
         case MQTT5_PROPERTY_SUBSCR_IDENTIFIER_AVAILABLE:
+            if (!mqtt5_property_has_bytes(property_offset, 1, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->subscribe_identifiers_available = property[property_offset++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SUBSCR_IDENTIFIER_AVAILABLE %d", resp_property->subscribe_identifiers_available);
             continue;
         case MQTT5_PROPERTY_SHARED_SUBSCR_AVAILABLE:
+            if (!mqtt5_property_has_bytes(property_offset, 1, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->shared_subscribe_available = property[property_offset++];
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SHARED_SUBSCR_AVAILABLE %d", resp_property->shared_subscribe_available);
             continue;
         case MQTT5_PROPERTY_SERVER_KEEP_ALIVE:
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(connection_info->keepalive, property[property_offset ++], property[property_offset ++])
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SERVER_KEEP_ALIVE %lld", connection_info->keepalive);
             continue;
@@ -686,7 +875,17 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             if (resp_property->response_info) {
                 free(resp_property->response_info);
             }
+
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             resp_property->response_info = calloc(1, len + 1);
             if (!resp_property->response_info) {
                 ESP_LOGE(TAG, "Failed to calloc %d data", len);
@@ -698,17 +897,44 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
             ESP_LOGD(TAG, "MQTT5_PROPERTY_RESP_INFO %s", resp_property->response_info);
             continue;
         case MQTT5_PROPERTY_SERVER_REFERENCE: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_SERVER_REFERENCE %.*s", len, &property[property_offset]);
             property_offset += len;
             continue;
         case MQTT5_PROPERTY_AUTHENTICATION_METHOD: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_AUTHENTICATION_METHOD %.*s", len, &property[property_offset]);
             property_offset += len;
             continue;
         case MQTT5_PROPERTY_AUTHENTICATION_DATA: //only print now
+            if (!mqtt5_property_has_bytes(property_offset, 2, property_len)) {
+                return ESP_FAIL;
+            }
+
             MQTT5_CONVERT_ONE_BYTE_TO_TWO(len, property[property_offset ++], property[property_offset ++])
+
+            if (len > MQTT5_MAX_PROPERTY_STRING_LEN || !mqtt5_property_has_bytes(property_offset, len, property_len)) {
+                return ESP_FAIL;
+            }
+
             ESP_LOGD(TAG, "MQTT5_PROPERTY_AUTHENTICATION_DATA length %d", len);
             property_offset += len;
             continue;

lib/mqtt_msg.c

@@ -143,8 +143,9 @@ size_t mqtt_get_total_length(const uint8_t *buffer, size_t length, int *fixed_si
     int i;
     size_t totlen = 0;
 
-    for (i = 1; i < length; ++i) {
-        totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
+    for (i = 1; i < length && i <= 4; ++i) {
+        totlen += (size_t)(buffer[i] & 0x7f) << (7 * (i - 1));
+
         if ((buffer[i] & 0x80) == 0) {
             ++i;
             break;
@@ -223,7 +224,7 @@ char *mqtt_get_publish_data(uint8_t *buffer, size_t *length)
     int blength = *length;
     *length = 0;
 
-    for (i = 1; i < blength; ++i) {
+    for (i = 1; i < blength && i <= 4; ++i) {
         totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
         if ((buffer[i] & 0x80) == 0) {
             ++i;

mqtt5_client.c

@@ -574,13 +574,7 @@ esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client
             client->mqtt5_config->connect_property_info.session_expiry_interval = connect_property->session_expiry_interval;
         }
         if (connect_property->maximum_packet_size) {
-            if (connect_property->maximum_packet_size > client->mqtt_state.in_buffer_length) {
-                ESP_LOGW(TAG, "Connect maximum_packet_size property is over buffer_size(%d), Please first change it", client->mqtt_state.in_buffer_length);
-                MQTT_API_UNLOCK(client);
-                return ESP_FAIL;
-            } else {
-                client->mqtt5_config->connect_property_info.maximum_packet_size = connect_property->maximum_packet_size;
-            }
+            client->mqtt5_config->connect_property_info.maximum_packet_size = connect_property->maximum_packet_size;
         } else {
             client->mqtt5_config->connect_property_info.maximum_packet_size = client->mqtt_state.in_buffer_length;
         }

mqtt_client.c

@@ -71,6 +71,16 @@ static int esp_mqtt_handle_transport_read_error(int err, esp_mqtt_client_handle_
     return -2;
 }
 
+// Reset per-message pending state before composing a new outbound control packet
+static inline void mqtt_reset_pending_message(esp_mqtt_client_handle_t client)
+{
+    client->mqtt_state.pending_msg_id = 0;
+    client->mqtt_state.pending_msg_type = 0;
+    client->mqtt_state.pending_publish_qos = 0;
+    client->mqtt_state.connection.outbound_message.fragmented_msg_total_length = 0;
+    client->mqtt_state.connection.outbound_message.fragmented_msg_data_offset = 0;
+}
+
 #if MQTT_ENABLE_SSL
 enum esp_mqtt_ssl_cert_key_api {
     MQTT_SSL_DATA_API_CA_CERT,
@@ -741,6 +751,8 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
 {
     int read_len, connect_rsp_code = 0;
     client->wait_for_ping_resp = false;
+    mqtt_reset_pending_message(client);
+
     if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
 #ifdef MQTT_PROTOCOL_5
         mqtt5_msg_connect(&client->mqtt_state.connection,
@@ -1744,7 +1756,8 @@ static void esp_mqtt_task(void *pv)
                             ESP_LOGE(TAG, "Failed to remove queued qos0 message from the outbox");
                         }
                     }
-                    if (client->mqtt_state.pending_publish_qos > 0) {
+                    if (client->mqtt_state.pending_publish_qos > 0 &&
+                            mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
                         outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED);
 #ifdef MQTT_PROTOCOL_5
                         if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
@@ -1760,7 +1773,9 @@ static void esp_mqtt_task(void *pv)
                 if (item && (last_retransmit - msg_tick > client->config->message_retransmit_timeout))  {
                     if (mqtt_resend_queued(client, item) == ESP_OK) {
 #ifdef MQTT_PROTOCOL_5
-                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
+                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 &&
+                                client->mqtt_state.pending_publish_qos > 0 &&
+                                mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
                             esp_mqtt5_increment_packet_counter(client);
                         }
 #endif
@@ -1770,7 +1785,13 @@ static void esp_mqtt_task(void *pv)
                 if (item && (last_retransmit - msg_tick > client->config->message_retransmit_timeout))  {
                     if (mqtt_resend_pubrel(client, item) == ESP_OK) {
 #ifdef MQTT_PROTOCOL_5
-                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
+
+                        // Do not count PUBREL as a new inflight PUBLISH
+                        // Only PUBLISH QoS>0 contributes to inflight limitation
+                        // (outbound_message here is PUBREL, so this condition will be false)
+                        if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5 &&
+                                client->mqtt_state.pending_publish_qos > 0 &&
+                                mqtt_get_type(client->mqtt_state.connection.outbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
                             esp_mqtt5_increment_packet_counter(client);
                         }
 #endif
@@ -1931,10 +1952,7 @@ esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client)
 
         // Only send the disconnect message if the client is connected
         if (client->state == MQTT_STATE_CONNECTED) {
-            if (send_disconnect_msg(client) != ESP_OK) {
-                MQTT_API_UNLOCK(client);
-                return ESP_FAIL;
-            }
+            send_disconnect_msg(client);
         }
 
         client->run = false;
@@ -2029,8 +2047,10 @@ int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client,
     }
 
     ESP_LOGD(TAG, "Sent subscribe, first topic=%s, id: %d", topic_list[0].filter, client->mqtt_state.pending_msg_id);
+
+    int pending_msg_id = client->mqtt_state.pending_msg_id;
     MQTT_API_UNLOCK(client);
-    return client->mqtt_state.pending_msg_id;
+    return pending_msg_id;
 
 }
 int esp_mqtt_client_subscribe_single(esp_mqtt_client_handle_t client, const char *topic, int qos)
@@ -2050,6 +2070,9 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top
         return -1;
     }
     MQTT_API_LOCK(client);
+    // Reset pending state to avoid inheriting previous PUBLISH QoS or type
+    mqtt_reset_pending_message(client);
+
     if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
 #ifdef MQTT_PROTOCOL_5
         mqtt5_msg_unsubscribe(&client->mqtt_state.connection,
@@ -2085,8 +2108,10 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top
     }
 
     ESP_LOGD(TAG, "Sent Unsubscribe topic=%s, id: %d, successful", topic, client->mqtt_state.pending_msg_id);
+
+    int pending_msg_id = client->mqtt_state.pending_msg_id;
     MQTT_API_UNLOCK(client);
-    return client->mqtt_state.pending_msg_id;
+    return pending_msg_id;
 }
 
 static int make_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data,