From 6083bfe4881d23ff8391d452e7d7957554ca4c21 Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Wed, 2 Sep 2026 11:29:58 -0700 Subject: [PATCH 01/10] F-12444 - Refuse untrackable QoS 2 PUBLISH when dedup table full --- src/mqtt_client.c | 57 ++++++++++-- tests/test_mqtt_client.c | 184 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+), 5 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 6be1c8fe5..1e434cff7 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -565,6 +565,21 @@ static int MqttClient_RecvQos2_Contains(const MqttClient* client, return 0; } +static int MqttClient_RecvQos2_HasFreeSlot(const MqttClient* client) +{ + int i; + + if (client == NULL) { + return 0; + } + for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { + if (client->recv_qos2_pending[i] == 0) { + return 1; + } + } + return 0; +} + static void MqttClient_RecvQos2_Add(MqttClient* client, word16 packet_id) { int i; @@ -573,15 +588,14 @@ static void MqttClient_RecvQos2_Add(MqttClient* client, word16 packet_id) MqttClient_RecvQos2_Contains(client, packet_id)) { return; } + /* A new id is only delivered after MqttClient_Publish_ReadPayload confirmed + * a free slot, so this loop always finds one. */ for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { if (client->recv_qos2_pending[i] == 0) { client->recv_qos2_pending[i] = packet_id; return; } } - /* Table full: this id stays untracked, so a later retransmit of it may - * still reach the application. Delivery correctness is preserved; only the - * duplicate suppression is best effort under a flood of unacked QoS 2. */ } static void MqttClient_RecvQos2_Remove(MqttClient* client, word16 packet_id) @@ -1697,8 +1711,11 @@ static int MqttClient_HandlePacket(MqttClient* client, } #ifdef WOLFMQTT_V5 - /* Copy response code in case changed by callback */ + /* Copy response code in case changed by callback, then clear it on + * the (possibly caller-owned) publish object so a later reuse does + * not inherit this ack's reason, e.g. a quota rejection. */ resp->reason_code = publish->resp.reason_code; + publish->resp.reason_code = MQTT_REASON_SUCCESS; #endif /* Populate information needed for ack */ resp->packet_type = (packet_qos == MQTT_QOS_1) ? @@ -3590,8 +3607,16 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, * keep the stream in sync, but not delivered to the application again * [MQTT-4.3.3-10]. Re-derived from the packet id so it survives non-blocking * re-entry into this function. */ - int suppress_cb = (publish->qos == MQTT_QOS_2 && + int is_dup = (publish->qos == MQTT_QOS_2 && MqttClient_RecvQos2_Contains(client, publish->packet_id)); + /* A new QoS 2 id that cannot be recorded because the de-duplication table is + * full must not be delivered: its slot would go untracked and a later + * retransmit would reach the application a second time [MQTT-4.3.3-10]. The + * payload is still drained to keep the stream in sync, then the exchange is + * refused without a PUBREC so the sender retries once a slot is free. */ + int untrackable = (publish->qos == MQTT_QOS_2 && !is_dup && + !MqttClient_RecvQos2_HasFreeSlot(client)); + int suppress_cb = (is_dup || untrackable); #endif /* Handle packet callback and read remaining payload */ @@ -3674,6 +3699,28 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, } } while (!msg_done); +#if WOLFMQTT_MAX_QOS >= 2 + /* The new QoS 2 id could not be tracked (dedup table full) and the drained + * payload was not delivered. Complete the exchange so the connection is not + * left livelocked on a PUBLISH that is retransmitted forever. */ + if (rc == MQTT_CODE_SUCCESS && untrackable) { + #ifdef WOLFMQTT_V5 + if (client->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5) { + /* Reject on the PUBREC with Quota Exceeded so the peer ends the + * exchange without a PUBREL and may retry once a slot frees. The + * 0x80 error bit makes MqttClient_HandlePacket skip tracking. */ + publish->resp.reason_code = MQTT_REASON_QUOTA_EXCEEDED; + } + else + #endif + { + /* MQTT 3.1.1 PUBREC carries no reason code, so the message cannot be + * refused in-band. Fail fatally to drop the connection. */ + rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PACKET_ID); + } + } +#endif + /* No message callback registered to deliver this incoming PUBLISH. The * payload was drained above to keep the stream in sync, but the application * never saw it. Return a distinct error instead of success so the caller is diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index a6627302a..be10d7526 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -402,6 +402,19 @@ static int g_ack_id_count; * test can confirm the client echoed the id of the PUBLISH it is acknowledging. */ static int g_last_ack_id; +/* Counts msg_cb invocations so a test can assert an incoming PUBLISH was (or was + * not) delivered to the application. */ +#if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) +static int g_msg_cb_calls; +static int mock_msg_cb(MqttClient* client, MqttMessage* message, byte msg_new, + byte msg_done) +{ + (void)client; (void)message; (void)msg_new; (void)msg_done; + g_msg_cb_calls++; + return MQTT_CODE_SUCCESS; +} +#endif + static int mock_net_write_accept(void *context, const byte* buf, int buf_len, int timeout_ms) { @@ -6067,6 +6080,170 @@ TEST(wait_message_qos2_null_msg_cb_errors_no_ack) ASSERT_FALSE(g_pubresp_written); } +#if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) +/* [MQTT-4.3.3-10] The inbound QoS 2 dedup table records each delivered packet id + * until its PUBREL, so a retransmit is acknowledged again without a second + * delivery. When the table is full a new packet id cannot be recorded, and + * delivering it anyway would let a later retransmit reach the application twice. + * The client must instead complete the exchange without delivering: MQTT 3.1.1 + * has no way to refuse a PUBLISH in-band, so the connection is dropped rather + * than left livelocked on an id that can never be tracked. */ +TEST(wait_message_qos2_full_dedup_table_v311_disconnects) +{ + int rc; + int i; + word16 new_id = (word16)(MQTT_MAX_RECV_QOS2 + 1); + /* v3.1.1 QoS2 PUBLISH: type|qos2=0x34, remain=7, topic "a" (0x0001 'a'), + * packet_id filled below, payload "hi". new_id is one past the ids used to + * fill the table, so it is a genuinely new, untrackable exchange. */ + byte publish_qos2[] = { 0x34, 0x07, 0x00, 0x01, 'a', 0x00, 0x00, 'h', 'i' }; + publish_qos2[5] = (byte)(new_id >> 8); + publish_qos2[6] = (byte)(new_id & 0xFF); + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); + test_client.msg_cb = mock_msg_cb; + g_msg_cb_calls = 0; + + /* Occupy every dedup slot with distinct in-flight ids awaiting PUBREL. */ + for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { + test_client.recv_qos2_pending[i] = (word16)(i + 1); + } + + g_pubresp_written = 0; + g_frames_written = 0; + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, publish_qos2, sizeof(publish_qos2)); + g_canned_len = (int)sizeof(publish_qos2); + g_canned_pos = 0; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_WaitMessage(&test_client, TEST_CMD_TIMEOUT_MS); + } + + /* Pre-fix the message was delivered and a PUBREC sent while the id went + * untracked. Now it is not delivered, no PUBREC is sent, and the connection + * is torn down. */ + ASSERT_EQ(MQTT_CODE_ERROR_PACKET_ID, rc); + ASSERT_EQ(0, g_msg_cb_calls); + ASSERT_FALSE(g_pubresp_written); + ASSERT_EQ(0, g_frames_written); + ASSERT_EQ(0, (int)(MqttClient_Flags(&test_client, 0, 0) & + MQTT_CLIENT_FLAG_IS_CONNECTED)); +} + +#ifdef WOLFMQTT_V5 +/* The MQTT 5 counterpart keeps the connection: an untrackable new QoS 2 id is + * refused on the PUBREC with Quota Exceeded (0x97), so the peer ends the + * exchange and may retry once a tracking slot frees. The message is still not + * delivered to the application. */ +TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec) +{ + int rc; + int i; + word16 new_id = (word16)(MQTT_MAX_RECV_QOS2 + 1); + /* v5 QoS2 PUBLISH: type|qos2=0x34, remain=8, topic "a" (0x0001 'a'), + * packet_id filled below, prop_len=0, payload "hi". */ + byte publish_qos2[] = { 0x34, 0x08, 0x00, 0x01, 'a', 0x00, 0x00, 0x00, + 'h', 'i' }; + publish_qos2[5] = (byte)(new_id >> 8); + publish_qos2[6] = (byte)(new_id & 0xFF); + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); + test_client.msg_cb = mock_msg_cb; + g_msg_cb_calls = 0; + + for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { + test_client.recv_qos2_pending[i] = (word16)(i + 1); + } + + g_pubresp_written = 0; + g_last_ack_written = MQTT_PACKET_TYPE_RESERVED; + connect_mock_xfer = 0; + XMEMSET(connect_mock_sent, 0, sizeof(connect_mock_sent)); + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, publish_qos2, sizeof(publish_qos2)); + g_canned_len = (int)sizeof(publish_qos2); + g_canned_pos = 0; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_WaitMessage(&test_client, TEST_CMD_TIMEOUT_MS); + } + + /* Not delivered, but a PUBREC carrying Quota Exceeded is sent and the + * connection stays up. connect_mock_sent[4] is the PUBREC reason byte + * (0x50, remlen, id_hi, id_lo, reason). */ + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0, g_msg_cb_calls); + ASSERT_TRUE(g_pubresp_written); + ASSERT_EQ(MQTT_PACKET_TYPE_PUBLISH_REC, g_last_ack_written); + ASSERT_EQ(MQTT_REASON_QUOTA_EXCEEDED, connect_mock_sent[4]); + ASSERT_NE(0, (int)(MqttClient_Flags(&test_client, 0, 0) & + MQTT_CLIENT_FLAG_IS_CONNECTED)); +} + +/* The quota-rejection reason is written into the publish object to reach the + * outgoing PUBREC. On a caller-owned MqttObject (the public WaitMessage_ex path) + * it must be cleared afterwards, or a later PUBLISH reusing the object would + * inherit the rejection and send a spurious negative PUBREC. */ +TEST(wait_message_ex_qos2_quota_reason_not_retained) +{ + int rc; + int i; + MqttObject obj; + word16 new_id = (word16)(MQTT_MAX_RECV_QOS2 + 1); + /* v5 QoS2 PUBLISH for a new, untrackable id: type|qos2=0x34, remain=8, + * topic "a", packet_id, prop_len=0, "hi". */ + byte publish_qos2[] = { 0x34, 0x08, 0x00, 0x01, 'a', 0x00, 0x00, 0x00, + 'h', 'i' }; + publish_qos2[5] = (byte)(new_id >> 8); + publish_qos2[6] = (byte)(new_id & 0xFF); + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); + test_client.msg_cb = mock_msg_cb; + g_msg_cb_calls = 0; + + for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { + test_client.recv_qos2_pending[i] = (word16)(i + 1); + } + + XMEMSET(&obj, 0, sizeof(obj)); + connect_mock_xfer = 0; + XMEMSET(connect_mock_sent, 0, sizeof(connect_mock_sent)); + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, publish_qos2, sizeof(publish_qos2)); + g_canned_len = (int)sizeof(publish_qos2); + g_canned_pos = 0; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_WaitMessage_ex(&test_client, &obj, TEST_CMD_TIMEOUT_MS); + } + + /* The PUBREC carried Quota Exceeded, but the caller object must not retain + * it (pre-fix it stayed 0x97 and leaked into the next reuse). */ + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(MQTT_REASON_QUOTA_EXCEEDED, connect_mock_sent[4]); + ASSERT_EQ(MQTT_REASON_SUCCESS, obj.publish.resp.reason_code); +} +#endif /* WOLFMQTT_V5 */ +#endif /* WOLFMQTT_MAX_QOS >= 2 && MQTT_MAX_RECV_QOS2 < 65535 */ + #ifdef WOLFMQTT_V5 /* #6217 property-leak regression: a v5 incoming PUBLISH carrying properties must * not leak its retained property list when there is no msg_cb. The decoder keeps @@ -7480,6 +7657,13 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_qos1_null_msg_cb_errors_no_ack); RUN_TEST(wait_message_qos0_null_msg_cb_errors); RUN_TEST(wait_message_qos2_null_msg_cb_errors_no_ack); +#if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) + RUN_TEST(wait_message_qos2_full_dedup_table_v311_disconnects); +#ifdef WOLFMQTT_V5 + RUN_TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec); + RUN_TEST(wait_message_ex_qos2_quota_reason_not_retained); +#endif +#endif #ifdef WOLFMQTT_V5 RUN_TEST(wait_message_v5_props_null_msg_cb_frees_props); RUN_TEST(wait_message_v5_empty_topic_with_alias_delivered); From 91d99079d027991005a4d2df75d896b3b519e9fb Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Wed, 2 Sep 2026 13:09:47 -0700 Subject: [PATCH 02/10] F-12445 - Keep durable outbound queue on orphan session reclaim --- src/mqtt_broker.c | 12 ++--- tests/test_broker_connect.c | 100 ++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 8 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index aebf5fd17..ed90ce3f9 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -3804,14 +3804,10 @@ static int BrokerOrphan_Reclaim(MqttBroker* broker, BrokerClient* new_bc) WBLOG_INFO(broker, "broker: orphan reclaimed client_id=%s queued=%d", BrokerLog_Sanitize(new_bc->client_id), new_bc->out_q_count); -#ifdef WOLFMQTT_BROKER_PERSIST - /* The reclaimed queue is now in a LIVE BrokerClient. Persisted - * records for this client_id are no longer authoritative - the - * subscriber will receive these via the upcoming drain and ack - * them. Wipe the on-disk copies so a subsequent crash doesn't - * re-deliver them. */ - (void)BrokerPersist_DelOutQueue(broker, new_bc->client_id); -#endif + /* The durable OUTQ records are kept: each is removed only by its terminal + * acknowledgement (BrokerClient_OnPubAck / OnPubComp). Wiping them here + * would lose every unacknowledged message if the broker stopped after + * reclaim but before the drain re-sent and the peer acknowledged. */ BrokerOrphan_Remove(broker, o); return 1; } diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 40851934c..4b376d60f 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -7696,6 +7696,19 @@ static BrokerOutPub* persist_order_make_out(const char* topic, MqttQoS qos, return out; } +/* Counts deletions of durable outbound-queue records so a test can assert an + * orphan reclaim does not wipe them. */ +static int g_reclaim_outq_del; +static int persist_reclaim_kv_del(void* ctx, byte ns, const byte* key, + word16 key_len) +{ + (void)ctx; (void)key; (void)key_len; + if (ns == BROKER_PERSIST_NS_OUTQ) { + g_reclaim_outq_del++; + } + return MQTT_CODE_SUCCESS; +} + #ifdef WOLFMQTT_NONBLOCK /* A partially transmitted QoS 1 PUBLISH is a retransmission after a broker * restart, so its replay must carry DUP=1 [MQTT-3.3.1-1]. */ @@ -7906,6 +7919,92 @@ TEST(persist_mixed_qos_queue_preserves_fifo) MqttBroker_Free(&restored); } +/* An orphan reclaim must keep the durable outbound-queue records: each is + * removed only by its own terminal acknowledgement. Wiping them at reclaim + * loses every unacknowledged QoS message if the broker restarts after the + * reclaim but before the resumed session acknowledges. */ +TEST(orphan_reclaim_keeps_persisted_outq) +{ + MqttBroker broker; + MqttBrokerNet net; + MqttBrokerPersistHooks hooks; + PersistOrderStore store; + BrokerClient* sub_bc; + int i; + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + static const byte publish_x[] = { + 0x32, 0x08, 0x00, 0x01, 'x', 0x00, 0x07, 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + XMEMSET(&hooks, 0, sizeof(hooks)); + XMEMSET(&store, 0, sizeof(store)); + hooks.kv_put = persist_order_put; + hooks.kv_get = persist_order_get; + hooks.kv_del = persist_reclaim_kv_del; + hooks.kv_iter = persist_order_iter; + hooks.ctx = &store; +#ifdef WOLFMQTT_BROKER_PERSIST_ENCRYPT + hooks.derive_key = persist_order_derive_key; +#endif + + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_SetPersistHooks(&broker, &hooks)); + ASSERT_EQ(MQTT_CODE_SUCCESS, BrokerPersist_Restore(&broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + /* Persistent subscriber "S" subscribes; publisher "P" sends one QoS 1 + * message. The subscriber never PUBACKs, so it stays persisted and + * unacknowledged. */ + reset_mock_clients(2); + mock_client_input_append(0, connect_sub, sizeof(connect_sub)); + mock_client_input_append(0, subscribe_x, sizeof(subscribe_x)); + mock_client_input_append(1, connect_pub, sizeof(connect_pub)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&broker); + } + sub_bc = find_broker_client(&broker, "S"); + ASSERT_NOT_NULL(sub_bc); + mock_client_input_append(1, publish_x, sizeof(publish_x)); + for (i = 0; i < 12; i++) { + (void)MqttBroker_Step(&broker); + } + ASSERT_NOT_NULL(sub_bc->out_q_head); + + /* Drop the subscriber so its session is orphaned; BrokerOrphan_Take + * shadow-writes the unacknowledged QoS 1 message to durable storage. */ + sub_bc->connected = 0; + g_clients[0].read_err = 1; + (void)MqttBroker_Step(&broker); + ASSERT_EQ(1, broker.orphan_session_count); + ASSERT_EQ(1, store.outq_count); + + /* Reconnect the same persistent client id, which reclaims the orphan. The + * reclaim must not delete the durable OUTQ records. */ + g_reclaim_outq_del = 0; + reset_mock_clients(1); + mock_client_input_append(0, connect_sub, sizeof(connect_sub)); + for (i = 0; i < 12; i++) { + (void)MqttBroker_Step(&broker); + } + ASSERT_EQ(0, broker.orphan_session_count); + ASSERT_EQ(0, g_reclaim_outq_del); + + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} + TEST(persist_restore_packet_id_wrap_preserves_fifo) { MqttBroker source; @@ -9312,6 +9411,7 @@ int main(int argc, char** argv) RUN_TEST(persist_partial_publish_restart_keeps_dup); #endif RUN_TEST(persist_mixed_qos_queue_preserves_fifo); + RUN_TEST(orphan_reclaim_keeps_persisted_outq); #endif RUN_TEST(persist_stream_only_hooks_rejected); RUN_TEST(persist_mixed_kv_and_stream_hooks_accepted); From aa66d426613724156d656407334958a002cd144a Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Wed, 2 Sep 2026 14:56:52 -0700 Subject: [PATCH 03/10] F-12478 - Require re-authentication to reuse the CONNECT auth method --- src/mqtt_client.c | 71 +++++++++++++++++++++++++++++++++------- tests/test_mqtt_client.c | 44 +++++++++++++++++++++++++ wolfmqtt/mqtt_client.h | 20 +++++++++++ 3 files changed, 123 insertions(+), 12 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 1e434cff7..e4b45a861 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -2815,19 +2815,31 @@ int MqttClient_SetPropertyCallback(MqttClient *client, MqttPropertyCb propCb, #ifdef WOLFMQTT_V5 /* Return 1 if the CONNECT carries an Authentication Method property, i.e. the * connection is negotiating enhanced authentication [MQTT-4.12]. */ -static int MqttConnect_HasAuthMethod(const MqttConnect* mc_connect) +/* Record whether the CONNECT carried an Authentication Method and, if so, its + * value, so a later client-initiated AUTH can be required to reuse the same + * method [MQTT-4.12.0-1]. A value longer than MQTT_AUTH_METHOD_MAX is noted by + * auth_method_len but not stored, and MqttClient_Auth then refuses re-auth + * because it cannot verify the match. */ +static void MqttClient_StoreAuthMethod(MqttClient* client, + const MqttConnect* mc_connect) { const MqttProp* prop; - if (mc_connect == NULL) { - return 0; - } - for (prop = mc_connect->props; prop != NULL; prop = prop->next) { + client->auth_method_set = 0; + client->auth_method_len = 0; + for (prop = (mc_connect != NULL) ? mc_connect->props : NULL; + prop != NULL; prop = prop->next) { if (prop->type == MQTT_PROP_AUTH_METHOD) { - return 1; + client->auth_method_set = 1; + client->auth_method_len = prop->data_str.len; + if (prop->data_str.len <= MQTT_AUTH_METHOD_MAX && + prop->data_str.str != NULL) { + XMEMCPY(client->auth_method, prop->data_str.str, + prop->data_str.len); + } + return; } } - return 0; } #if WOLFMQTT_MAX_QOS >= 2 @@ -3172,12 +3184,12 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) } XMEMSET(&mc_connect->ack, 0, sizeof(mc_connect->ack)); - /* Record whether this connection negotiates enhanced authentication so - * a later MqttClient_Auth can be refused when no Authentication Method - * was sent [MQTT-4.12.0-1]. Recomputed each connect, so it also resets + /* Record whether this connection negotiates enhanced authentication and + * retain the Authentication Method value so a later MqttClient_Auth can + * be refused when no method was sent and required to reuse the same + * method [MQTT-4.12.0-1]. Recomputed each connect, so it also resets * across a reconnect on the same client. */ - client->auth_method_set = - (byte)MqttConnect_HasAuthMethod(mc_connect); + MqttClient_StoreAuthMethod(client, mc_connect); #endif /* Warn if credentials are being sent without TLS */ #ifdef WOLFMQTT_DEBUG_CLIENT @@ -5057,6 +5069,35 @@ static int MqttClient_AuthEx(MqttClient *client, MqttAuth* auth, return rc; } +/* Return 1 only if the AUTH packet carries the same Authentication Method value + * that CONNECT negotiated [MQTT-4.12.0-1]. A missing method, a length or byte + * mismatch, or a CONNECT method too long to have been stored all fail. */ +static int MqttClient_AuthMethodMatches(const MqttClient* client, + const MqttAuth* auth) +{ + const MqttProp* prop; + + if (client->auth_method_len > MQTT_AUTH_METHOD_MAX) { + return 0; + } + for (prop = auth->props; prop != NULL; prop = prop->next) { + if (prop->type == MQTT_PROP_AUTH_METHOD) { + if (prop->data_str.len != client->auth_method_len) { + return 0; + } + if (client->auth_method_len == 0) { + return 1; + } + if (prop->data_str.str == NULL) { + return 0; + } + return (XMEMCMP(prop->data_str.str, client->auth_method, + client->auth_method_len) == 0); + } + } + return 0; +} + int MqttClient_Auth(MqttClient *client, MqttAuth* auth) { if (client == NULL || auth == NULL) { @@ -5070,6 +5111,12 @@ int MqttClient_Auth(MqttClient *client, MqttAuth* auth) client->auth_method_set == 0) { return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); } + /* [MQTT-4.12.0-1] Re-authentication must reuse the negotiated method: refuse + * an AUTH whose Authentication Method differs from or is missing relative to + * the one CONNECT carried, so the mechanism cannot be switched mid-session. */ + if (!MqttClient_AuthMethodMatches(client, auth)) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); + } return MqttClient_AuthEx(client, auth, &auth->stat, #ifdef WOLFMQTT_MULTITHREAD &auth->pendResp, diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index be10d7526..4225deb01 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -1371,6 +1371,49 @@ TEST(auth_v311_session_rejected_before_write) MqttClient_PropsFree(auth.props); } +/* [MQTT-4.12.0-1] Re-authentication must reuse the CONNECT Authentication + * Method. After a CONNECT that negotiated "SCRAM-SHA-256", an AUTH selecting a + * different method or omitting it must be refused before it reaches the wire. */ +TEST(auth_mismatched_method_rejected) +{ + int rc; + MqttAuth auth; + MqttProp* prop; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + + rc = run_connect_v5_with_auth_method(1); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(1, (int)test_client.auth_method_set); + + XMEMSET(&auth, 0, sizeof(auth)); + auth.reason_code = MQTT_REASON_CONT_AUTH; + prop = MqttClient_PropsAdd(&auth.props); + ASSERT_NOT_NULL(prop); + prop->type = MQTT_PROP_AUTH_METHOD; + prop->data_str.str = (char*)"PLAIN"; + prop->data_str.len = (word16)XSTRLEN("PLAIN"); + + g_frames_written = 0; + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read; + + rc = MqttClient_Auth(&test_client, &auth); + ASSERT_EQ(MQTT_CODE_ERROR_BAD_ARG, rc); + ASSERT_EQ(0, g_frames_written); + MqttClient_PropsFree(auth.props); + + XMEMSET(&auth, 0, sizeof(auth)); + auth.reason_code = MQTT_REASON_CONT_AUTH; + g_frames_written = 0; + + rc = MqttClient_Auth(&test_client, &auth); + ASSERT_EQ(MQTT_CODE_ERROR_BAD_ARG, rc); + ASSERT_EQ(0, g_frames_written); +} + /* MQTT v5: a refused CONNACK (non-zero return code) must NOT mutate long-lived * client state even when it carries server properties, otherwise a rejected or * malicious broker could shrink the client's packet-size cap or lower its QoS @@ -7477,6 +7520,7 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_v311_rejects_disconnect_packet_type); RUN_TEST(wait_message_v5_accepts_disconnect_packet_type); RUN_TEST(auth_v311_session_rejected_before_write); + RUN_TEST(auth_mismatched_method_rejected); RUN_TEST(connect_refused_connack_preserves_v5_defaults); RUN_TEST(connect_accepted_connack_rejects_illegal_max_qos); RUN_TEST(connect_accepted_connack_rejects_illegal_retain_available); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index e90e61002..b2eea3eaa 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -367,6 +367,16 @@ typedef struct _MqttSendId { byte ack_type; } MqttSendId; +#ifdef WOLFMQTT_V5 + /* Bytes of the CONNECT Authentication Method retained to enforce that a + * later client-initiated AUTH reuses the same method [MQTT-4.12.0-1]. A + * CONNECT method longer than this cannot be verified and re-authentication + * is refused; override in user_settings.h if longer names are needed. */ + #ifndef MQTT_AUTH_METHOD_MAX + #define MQTT_AUTH_METHOD_MAX 32 + #endif +#endif + /* Client structure */ typedef struct _MqttClient { word32 flags; /* MqttClientFlags */ @@ -492,6 +502,16 @@ typedef struct _MqttClient { * off. */ int replayIdx; #endif + +#ifdef WOLFMQTT_V5 + /* CONNECT Authentication Method value, retained so a later client-initiated + * AUTH must reuse the same method [MQTT-4.12.0-1]. auth_method_len is the + * full value length; when it exceeds MQTT_AUTH_METHOD_MAX the value could + * not be stored and re-auth is refused. Placed at the end of the struct so + * existing member offsets are unchanged. */ + word16 auth_method_len; + byte auth_method[MQTT_AUTH_METHOD_MAX]; +#endif } MqttClient; #ifdef WOLFMQTT_SN From ad0fbd0c9bd37e41aad772f47ae02b1ac8022cfa Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Wed, 2 Sep 2026 15:21:20 -0700 Subject: [PATCH 04/10] F-12479 - Require QoS 2 before advancing a PUBLISH on PUBREC --- src/mqtt_broker.c | 18 ++++++++- tests/test_broker_connect.c | 81 +++++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index ed90ce3f9..dd184098d 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2381,6 +2381,15 @@ static int BrokerClient_OnPubRec(BrokerClient* bc, word16 packet_id) (int)bc->sock, (unsigned)packet_id); return 0; } + /* A PUBREC belongs only to a QoS 2 exchange. One arriving for a QoS 1 + * PUBLISH is a peer protocol error: leave the entry awaiting its PUBACK + * and tell the caller not to answer with a PUBREL the flow cannot back. */ + if (e->qos != MQTT_QOS_2) { + WBLOG_DBG(bc->broker, + "broker: PUBREC for QoS %d PUBLISH sock=%d packet_id=%u", + (int)e->qos, (int)bc->sock, (unsigned)packet_id); + return -1; + } e->state = BROKER_OUTQ_PUBREL_SENT; /* Inflight stays counted - the delivery is still outstanding until * PUBCOMP returns. */ @@ -8231,8 +8240,13 @@ static int BrokerHandle_PublishRec(BrokerClient* bc, int rx_len) * PUBREL we send below is correlated to this entry; PUBCOMP from the * subscriber will then close it out. A spurious PUBREC (no matching * entry) still gets a PUBREL response for idempotency, just no - * queue state change. */ - (void)BrokerClient_OnPubRec(bc, resp.packet_id); + * queue state change. A negative return means the PUBREC targeted a + * non-QoS 2 PUBLISH, a Protocol Error [MQTT-4.13.1-1]: that entry stays + * awaiting its PUBACK, no PUBREL is sent, and the fatal code closes the + * peer. */ + if (BrokerClient_OnPubRec(bc, resp.packet_id) < 0) { + return MQTT_CODE_ERROR_PACKET_TYPE; + } #endif #ifdef WOLFMQTT_STATIC_MEMORY tracked = BrokerStaticOrphan_OnPubRec(bc->broker, bc, resp.packet_id); diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 4b376d60f..d3a48e96b 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -7536,6 +7536,84 @@ TEST(will_qos1_routes_through_outq) } #endif /* WOLFMQTT_BROKER_WILL && !WOLFMQTT_STATIC_MEMORY */ +#if !defined(WOLFMQTT_STATIC_MEMORY) && (WOLFMQTT_MAX_QOS >= 2) +/* A PUBREC belongs only to a QoS 2 exchange. One arriving for an outbound QoS 1 + * PUBLISH must not advance that entry to PUBREL_SENT nor trigger a PUBREL: the + * message stays unacknowledged until its PUBACK. */ +TEST(pubrec_for_qos1_publish_not_advanced) +{ + MqttBroker broker; + MqttBrokerNet net; + BrokerClient* sub_bc; + word16 packet_id; + int i; + byte pubrec[] = { 0x50, 0x02, 0x00, 0x00 }; + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + static const byte publish_x[] = { + 0x32, 0x08, 0x00, 0x01, 'x', 0x00, 0x07, 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + /* Subscriber "S" gets one QoS 1 PUBLISH in flight, awaiting PUBACK. */ + reset_mock_clients(2); + mock_client_input_append(0, connect_sub, sizeof(connect_sub)); + mock_client_input_append(0, subscribe_x, sizeof(subscribe_x)); + mock_client_input_append(1, connect_pub, sizeof(connect_pub)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&broker); + } + sub_bc = find_broker_client(&broker, "S"); + ASSERT_NOT_NULL(sub_bc); + mock_client_input_append(1, publish_x, sizeof(publish_x)); + for (i = 0; i < 12; i++) { + (void)MqttBroker_Step(&broker); + } + ASSERT_NOT_NULL(sub_bc->out_q_head); + ASSERT_EQ(MQTT_QOS_1, (int)sub_bc->out_q_head->qos); + ASSERT_EQ(BROKER_OUTQ_PUBLISH_SENT, sub_bc->out_q_head->state); + packet_id = sub_bc->out_q_head->packet_id; + + /* The subscriber wrongly answers the QoS 1 PUBLISH with a PUBREC. */ + pubrec[2] = (byte)(packet_id >> 8); + pubrec[3] = (byte)(packet_id & 0xFF); + mock_client_input_append(0, pubrec, sizeof(pubrec)); + for (i = 0; i < 8; i++) { + (void)MqttBroker_Step(&broker); + } + + /* Pre-fix the entry advanced to PUBREL_SENT and a PUBREL was sent while + * the connection stayed open. A PUBREC on a QoS 1 PUBLISH is a Protocol + * Error [MQTT-4.13.1-1]: no PUBREL may be sent and the peer is closed. The + * persistent session is orphaned with the entry still awaiting its PUBACK. + * sub_bc is freed by the close, so the entry is inspected via the orphan. */ + ASSERT_EQ(0, count_packets_of_type(g_clients[0].out_buf, + g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH_REL)); + ASSERT_TRUE(g_clients[0].closed); + ASSERT_EQ(1, broker.orphan_session_count); + ASSERT_NOT_NULL(broker.orphan_sessions); + ASSERT_NOT_NULL(broker.orphan_sessions->out_q_head); + ASSERT_EQ(BROKER_OUTQ_PUBLISH_SENT, + broker.orphan_sessions->out_q_head->state); + + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} +#endif /* !WOLFMQTT_STATIC_MEMORY && WOLFMQTT_MAX_QOS >= 2 */ + #ifdef WOLFMQTT_BROKER_PERSIST static int persist_test_write_file(const char* path, const byte* data, word32 data_len) @@ -9402,6 +9480,9 @@ int main(int argc, char** argv) RUN_TEST(connect_v5_max_packet_size_zero_protocol_error); #endif #endif +#if !defined(WOLFMQTT_STATIC_MEMORY) && (WOLFMQTT_MAX_QOS >= 2) + RUN_TEST(pubrec_for_qos1_publish_not_advanced); +#endif #ifdef WOLFMQTT_BROKER_PERSIST RUN_TEST(persist_parent_component_rejected_as_bad_argument); RUN_TEST(persist_iter_skips_fifo_without_blocking); From ad73c4d80fd2bc5606020eb0b29f7bfb29522815 Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Thu, 3 Sep 2026 09:32:33 -0700 Subject: [PATCH 05/10] F-12443 - Retain MQTT-SN write ownership across partial writes --- src/mqtt_sn_client.c | 627 +++++++++++++++++++++--------------- tests/test_mqtt_sn_client.c | 141 +++++++- wolfmqtt/mqtt_client.h | 4 + 3 files changed, 517 insertions(+), 255 deletions(-) diff --git a/src/mqtt_sn_client.c b/src/mqtt_sn_client.c index 922de060f..4e58fc8df 100644 --- a/src/mqtt_sn_client.c +++ b/src/mqtt_sn_client.c @@ -948,6 +948,63 @@ int SN_Client_SetRegisterCallback(MqttClient *client, return rc; } +#ifdef WOLFMQTT_MULTITHREAD +/* Unlink a pending response registered for a send that will not be resumed. */ +static void SN_Client_UnlinkPendResp(MqttClient* client, + MqttPendResp* pendResp) +{ + if (pendResp != NULL && wm_SemLock(&client->lockClient) == 0) { + MqttClient_RespList_Remove(client, pendResp); + wm_SemUnlock(&client->lockClient); + } +} +#endif + +/* Drive the packet already encoded in tx_buf to the transport under the write + * ownership taken by MqttWriteStart. Returns MQTT_CODE_CONTINUE with ownership + * and the pending response kept, so the caller resumes here on its next call; + * except that when nothing reached the transport and + * WOLFMQTT_ALLOW_NODATA_UNLOCK is set, the writer is released, the pending + * response unlinked, and *stat reset to MQTT_MSG_BEGIN so the next call + * re-encodes. Returns MQTT_CODE_SUCCESS on completion with the writer released. + * On a short or failed write the writer is released, the pending response + * unlinked, *stat reset, and the write result returned. */ +static int SN_Client_WriteOwned(MqttClient* client, MqttMsgStat* stat +#ifdef WOLFMQTT_MULTITHREAD + , MqttPendResp* pendResp +#endif + ) +{ + int xfer; + int rc; + + /* Snapshot under ownership: once the writer is released another sender may + * re-encode and change client->write.len before the comparison below. */ + xfer = client->write.len; + rc = MqttPacket_Write(client, client->tx_buf, xfer); + if (rc == MQTT_CODE_CONTINUE) { + #ifdef WOLFMQTT_ALLOW_NODATA_UNLOCK + if (client->write.total == 0) { + MqttWriteStop(client, stat); + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, pendResp); + #endif + stat->write = MQTT_MSG_BEGIN; + } + #endif + return rc; + } + MqttWriteStop(client, stat); + if (rc != xfer) { + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, pendResp); + #endif + stat->write = MQTT_MSG_BEGIN; + return rc; + } + return MQTT_CODE_SUCCESS; +} + int SN_Client_SearchGW(MqttClient *client, SN_SearchGw *search) { int rc; @@ -958,13 +1015,12 @@ int SN_Client_SearchGW(MqttClient *client, SN_SearchGw *search) } if (search->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &search->stat); if (rc != 0) { return rc; } - #endif /* Encode the search packet */ rc = SN_Encode_SearchGW(client->tx_buf, client->tx_buf_len, @@ -975,9 +1031,7 @@ int SN_Client_SearchGW(MqttClient *client, SN_SearchGw *search) SN_MSG_TYPE_SEARCHGW); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &search->stat); return rc; } client->write.len = rc; @@ -992,26 +1046,23 @@ int SN_Client_SearchGW(MqttClient *client, SN_SearchGw *search) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &search->stat); return rc; /* Error locking client */ } #endif + search->stat.write = MQTT_MSG_HEADER; + } + if (search->stat.write == MQTT_MSG_HEADER) { /* Send search for gateway packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &search->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &search->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &search->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif search->stat.write = MQTT_MSG_WAIT; } @@ -1103,13 +1154,12 @@ static int SN_WillTopic(MqttClient *client, SN_Will *will) case MQTT_MSG_HEADER: { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread + * is reported as MQTT_CODE_CONTINUE instead of deadlocking. */ + rc = MqttWriteStart(client, &will->stat); if (rc != 0) { return rc; } - #endif /* Encode Will Topic */ rc = SN_Encode_WillTopic(client->tx_buf, client->tx_buf_len, @@ -1120,23 +1170,35 @@ static int SN_WillTopic(MqttClient *client, SN_Will *will) SN_MSG_TYPE_WILLTOPIC); #endif if (rc > 0) { - /* Send Will Topic packet */ client->write.len = rc; - rc = MqttPacket_Write(client, client->tx_buf, - client->write.len); - if (rc == client->write.len) { - rc = 0; - } + will->stat.write = MQTT_MSG_PAYLOAD; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + else { + MqttWriteStop(client, &will->stat); + will->stat.write = MQTT_MSG_BEGIN; + break; + } + } + FALL_THROUGH; - #ifdef WOLFMQTT_NONBLOCK + case MQTT_MSG_PAYLOAD: + { + /* Send Will Topic packet; on MQTT_CODE_CONTINUE ownership is kept + * and this state resumes the write rather than re-encoding. */ + rc = SN_Client_WriteOwned(client, &will->stat + #ifdef WOLFMQTT_MULTITHREAD + , NULL + #endif + ); if (rc == MQTT_CODE_CONTINUE) { - return rc; /* resume send on next call */ + /* A zero-progress release under WOLFMQTT_ALLOW_NODATA_UNLOCK + * resets to BEGIN, but the WILLTOPICREQ was already consumed: + * retry from the send phase, not the request wait. */ + if (will->stat.write == MQTT_MSG_BEGIN) { + will->stat.write = MQTT_MSG_HEADER; + } + return rc; } - #endif /* reset state */ will->stat.write = MQTT_MSG_BEGIN; @@ -1144,7 +1206,6 @@ static int SN_WillTopic(MqttClient *client, SN_Will *will) } case MQTT_MSG_AUTH: - case MQTT_MSG_PAYLOAD: case MQTT_MSG_PAYLOAD2: case MQTT_MSG_ACK: default: @@ -1224,13 +1285,12 @@ static int SN_WillMessage(MqttClient *client, SN_Will *will) case MQTT_MSG_HEADER: { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread + * is reported as MQTT_CODE_CONTINUE instead of deadlocking. */ + rc = MqttWriteStart(client, &will->stat); if (rc != 0) { return rc; } - #endif /* Encode Will Message */ rc = SN_Encode_WillMsg(client->tx_buf, client->tx_buf_len, will); @@ -1240,34 +1300,52 @@ static int SN_WillMessage(MqttClient *client, SN_Will *will) SN_MSG_TYPE_WILLMSG); #endif if (rc > 0) { - /* Send Will Message packet */ client->write.len = rc; - rc = MqttPacket_Write(client, client->tx_buf, - client->write.len); - if (rc == client->write.len) { - rc = 0; - } + will->stat.write = MQTT_MSG_PAYLOAD; + } + else { + CLIENT_FORCE_ZERO(client->tx_buf, client->write.len); + MqttWriteStop(client, &will->stat); + will->stat.write = MQTT_MSG_BEGIN; + break; } + } + FALL_THROUGH; - #ifdef WOLFMQTT_NONBLOCK + case MQTT_MSG_PAYLOAD: + { + int xfer; + + /* Send Will Message packet. The length is snapshotted under + * ownership so the completion check cannot be skewed by another + * sender once the writer is released. */ + xfer = client->write.len; + rc = MqttPacket_Write(client, client->tx_buf, xfer); if (rc == MQTT_CODE_CONTINUE) { - /* Send not complete: tx_buf still holds the will payload and is - * needed to resume, so do not scrub it yet. */ - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); + #ifdef WOLFMQTT_ALLOW_NODATA_UNLOCK + if (client->write.total == 0) { + /* Nothing reached the transport: scrub and release the + * writer so other operations proceed; re-encode next. */ + CLIENT_FORCE_ZERO(client->tx_buf, xfer); + MqttWriteStop(client, &will->stat); + will->stat.write = MQTT_MSG_HEADER; + } #endif - return rc; /* resume send on next call */ + /* Unfinished write: tx_buf still holds the will payload and is + * needed to resume, so keep ownership, do not scrub, and resume + * here on the next call rather than re-encoding. */ + return rc; } - #endif /* The encoded WILLMSG contains the will payload (potentially - * sensitive). Scrub tx_buf before releasing lockSend so another + * sensitive). Scrub tx_buf before releasing the writer so another * thread cannot observe residual plaintext (mirrors the mitigation * in MqttClient_Connect). */ - CLIENT_FORCE_ZERO(client->tx_buf, client->write.len); - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + CLIENT_FORCE_ZERO(client->tx_buf, xfer); + MqttWriteStop(client, &will->stat); + if (rc == xfer) { + rc = 0; + } /* reset state */ will->stat.write = MQTT_MSG_BEGIN; @@ -1275,7 +1353,6 @@ static int SN_WillMessage(MqttClient *client, SN_Will *will) } case MQTT_MSG_AUTH: - case MQTT_MSG_PAYLOAD: case MQTT_MSG_PAYLOAD2: case MQTT_MSG_ACK: default: @@ -1314,13 +1391,12 @@ int SN_Client_Connect(MqttClient *client, SN_Connect *mc_connect) mc_connect->will_done = SN_WILL_DONE_NONE; - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &mc_connect->stat); if (rc != 0) { return rc; } - #endif /* Encode the connect packet */ rc = SN_Encode_Connect(client->tx_buf, client->tx_buf_len, mc_connect); @@ -1330,9 +1406,7 @@ int SN_Client_Connect(MqttClient *client, SN_Connect *mc_connect) SN_MSG_TYPE_CONNECT, 0, 0); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &mc_connect->stat); return rc; } client->write.len = rc; @@ -1347,26 +1421,23 @@ int SN_Client_Connect(MqttClient *client, SN_Connect *mc_connect) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &mc_connect->stat); return rc; /* Error locking client */ } #endif + mc_connect->stat.write = MQTT_MSG_HEADER; + } + if (mc_connect->stat.write == MQTT_MSG_HEADER) { /* Send connect packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &mc_connect->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &mc_connect->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &mc_connect->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif mc_connect->stat.write = MQTT_MSG_WAIT; } @@ -1432,13 +1503,12 @@ int SN_Client_WillTopicUpdate(MqttClient *client, SN_Will *will) } if (will->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &will->stat); if (rc != 0) { return rc; } - #endif /* Encode Will Topic Update */ rc = SN_Encode_WillTopicUpdate(client->tx_buf, @@ -1449,9 +1519,7 @@ int SN_Client_WillTopicUpdate(MqttClient *client, SN_Will *will) SN_MSG_TYPE_WILLTOPICUPD); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &will->stat); return rc; } client->write.len = rc; @@ -1466,26 +1534,23 @@ int SN_Client_WillTopicUpdate(MqttClient *client, SN_Will *will) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &will->stat); return rc; /* Error locking client */ } #endif + will->stat.write = MQTT_MSG_HEADER; + } + if (will->stat.write == MQTT_MSG_HEADER) { /* Send Will Topic Update packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &will->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &will->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &will->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif will->stat.write = MQTT_MSG_WAIT; } @@ -1520,14 +1585,12 @@ int SN_Client_WillMsgUpdate(MqttClient *client, SN_Will *will) } if (will->stat.write == MQTT_MSG_BEGIN) { - int xfer = 0; - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &will->stat); if (rc != 0) { return rc; } - #endif /* Encode Will Message Update */ rc = SN_Encode_WillMsgUpdate(client->tx_buf, client->tx_buf_len, will); @@ -1537,13 +1600,10 @@ int SN_Client_WillMsgUpdate(MqttClient *client, SN_Will *will) SN_MSG_TYPE_WILLMSGUPD); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &will->stat); return rc; } client->write.len = rc; - xfer = client->write.len; #ifdef WOLFMQTT_MULTITHREAD rc = wm_SemLock(&client->lockClient); @@ -1555,48 +1615,68 @@ int SN_Client_WillMsgUpdate(MqttClient *client, SN_Will *will) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - CLIENT_FORCE_ZERO(client->tx_buf, xfer); - wm_SemUnlock(&client->lockSend); + CLIENT_FORCE_ZERO(client->tx_buf, client->write.len); + MqttWriteStop(client, &will->stat); return rc; /* Error locking client */ } #endif - /* Send Will Message Update packet. Save write.len into xfer first: the - * encoded WILLMSGUPD holds will->willMsg (a possibly rotated/secret will - * payload) in tx_buf, which must be scrubbed before lockSend is released - * on every return path below. - * - * This differs intentionally from MqttClient_Connect and SN_WillMessage: - * those keep a resume state (MQTT_MSG_HEADER) and so must NOT scrub on - * MQTT_CODE_CONTINUE, because tx_buf is still needed to finish a partial - * non-blocking send. This function has no such resume state - it re-runs - * this whole MQTT_MSG_BEGIN block (re-encoding tx_buf) on every call. A - * non-blocking partial write returns MQTT_CODE_CONTINUE, which is != xfer - * and therefore lands in the error branch below; scrubbing there is safe - * because the next call re-encodes the identical bytes before the write - * resumes. */ + will->stat.write = MQTT_MSG_HEADER; + } + if (will->stat.write == MQTT_MSG_HEADER) { + int xfer; + + /* Send Will Message Update packet. The encoded WILLMSGUPD holds + * will->willMsg (a possibly rotated/secret will payload), which must + * never stay in the shared tx_buf across an API return. So the packet + * is re-encoded on every pass: a partial send scrubs tx_buf before + * returning, the identical bytes are regenerated here, and + * MqttPacket_Write resumes from the preserved write offset. The length + * is snapshotted under ownership so the completion check cannot be + * skewed by another sender. */ + rc = SN_Encode_WillMsgUpdate(client->tx_buf, client->tx_buf_len, will); + if (rc <= 0) { + MqttWriteStop(client, &will->stat); + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, &will->pendResp); + #endif + will->stat.write = MQTT_MSG_BEGIN; + return rc; + } + client->write.len = rc; + xfer = rc; rc = MqttPacket_Write(client, client->tx_buf, xfer); - if (rc != xfer) { - /* Send failed (or returned MQTT_CODE_CONTINUE): scrub the will - * payload from tx_buf before releasing lockSend so another thread - - * or a later memory/core-dump inspection - cannot recover residual - * plaintext. */ + if (rc == MQTT_CODE_CONTINUE) { + /* Never leave the will payload in tx_buf between calls; the resume + * pass re-encodes it. */ CLIENT_FORCE_ZERO(client->tx_buf, xfer); - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &will->pendResp); - wm_SemUnlock(&client->lockClient); + #ifdef WOLFMQTT_ALLOW_NODATA_UNLOCK + if (client->write.total == 0) { + /* Nothing reached the transport: release the writer and the + * pending response so other operations proceed. */ + MqttWriteStop(client, &will->stat); + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, &will->pendResp); + #endif + will->stat.write = MQTT_MSG_BEGIN; } #endif + /* Unfinished write: keep ownership and the pending response and + * resume here on the next call. */ return rc; } - /* WILLMSGUPD sent: scrub the will payload from tx_buf before releasing - * lockSend, for the same reason as the error path above. */ + /* Scrub the will payload from tx_buf before releasing the writer so + * another thread, or a later memory/core-dump inspection, cannot + * recover residual plaintext. */ CLIENT_FORCE_ZERO(client->tx_buf, xfer); - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &will->stat); + if (rc != xfer) { + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, &will->pendResp); + #endif + will->stat.write = MQTT_MSG_BEGIN; + return rc; + } will->stat.write = MQTT_MSG_WAIT; } @@ -1632,13 +1712,12 @@ int SN_Client_Subscribe(MqttClient *client, SN_Subscribe *subscribe) } if (subscribe->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &subscribe->stat); if (rc != 0) { return rc; } - #endif /* Encode the subscribe packet */ rc = SN_Encode_Subscribe(client->tx_buf, client->tx_buf_len, @@ -1649,9 +1728,7 @@ int SN_Client_Subscribe(MqttClient *client, SN_Subscribe *subscribe) SN_MSG_TYPE_SUBSCRIBE, subscribe->qos); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &subscribe->stat); return rc; } client->write.len = rc; @@ -1666,26 +1743,23 @@ int SN_Client_Subscribe(MqttClient *client, SN_Subscribe *subscribe) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &subscribe->stat); return rc; /* Error locking client */ } #endif + subscribe->stat.write = MQTT_MSG_HEADER; + } + if (subscribe->stat.write == MQTT_MSG_HEADER) { /* Send subscribe packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &subscribe->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &subscribe->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &subscribe->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif subscribe->stat.write = MQTT_MSG_WAIT; } @@ -1737,13 +1811,12 @@ int SN_Client_Publish(MqttClient *client, SN_Publish *publish) { case MQTT_MSG_BEGIN: { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread + * is reported as MQTT_CODE_CONTINUE instead of deadlocking. */ + rc = MqttWriteStart(client, &publish->stat); if (rc != 0) { return rc; } - #endif /* Encode the publish packet */ rc = SN_Encode_Publish(client->tx_buf, client->tx_buf_len, @@ -1756,9 +1829,7 @@ int SN_Client_Publish(MqttClient *client, SN_Publish *publish) publish->qos); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &publish->stat); return rc; } @@ -1781,7 +1852,7 @@ int SN_Client_Publish(MqttClient *client, SN_Publish *publish) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &publish->stat); return rc; /* Error locking client */ } } @@ -1795,27 +1866,42 @@ int SN_Client_Publish(MqttClient *client, SN_Publish *publish) case MQTT_MSG_PAYLOAD: case MQTT_MSG_PAYLOAD2: { - /* Send packet and payload */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - #ifdef WOLFMQTT_NONBLOCK - if (rc == MQTT_CODE_CONTINUE) + int xfer; + + /* Send packet and payload. The length is snapshotted under + * ownership so the completion check cannot be skewed by another + * sender once the writer is released. */ + xfer = client->write.len; + rc = MqttPacket_Write(client, client->tx_buf, xfer); + if (rc == MQTT_CODE_CONTINUE) { + #ifdef WOLFMQTT_ALLOW_NODATA_UNLOCK + if (client->write.total == 0) { + /* Nothing reached the transport: release the writer so + * other operations proceed; re-encode on the next call. */ + MqttWriteStop(client, &publish->stat); + #ifdef WOLFMQTT_MULTITHREAD + SN_Client_UnlinkPendResp(client, &publish->pendResp); + #endif + publish->stat.write = MQTT_MSG_BEGIN; + } + #endif + /* Unfinished write: keep ownership and the pending response + * and resume here on the next call. */ return rc; - #endif - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + } + MqttWriteStop(client, &publish->stat); if (rc < 0) { #ifdef WOLFMQTT_MULTITHREAD - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &publish->pendResp); - wm_SemUnlock(&client->lockClient); - } + SN_Client_UnlinkPendResp(client, &publish->pendResp); #endif + /* The writer was released and its state cleared, so the object + * must re-encode on its next use rather than resume. */ + publish->stat.write = MQTT_MSG_BEGIN; return rc; } - if (rc == client->write.len) { + if (rc == xfer) { rc = MQTT_CODE_SUCCESS; } else { @@ -1896,13 +1982,12 @@ int SN_Client_Unsubscribe(MqttClient *client, SN_Unsubscribe *unsubscribe) } if (unsubscribe->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &unsubscribe->stat); if (rc != 0) { return rc; } - #endif /* Encode the subscribe packet */ rc = SN_Encode_Unsubscribe(client->tx_buf, client->tx_buf_len, @@ -1913,9 +1998,7 @@ int SN_Client_Unsubscribe(MqttClient *client, SN_Unsubscribe *unsubscribe) SN_MSG_TYPE_UNSUBSCRIBE); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &unsubscribe->stat); return rc; } client->write.len = rc; @@ -1931,26 +2014,23 @@ int SN_Client_Unsubscribe(MqttClient *client, SN_Unsubscribe *unsubscribe) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &unsubscribe->stat); return rc; /* Error locking client */ } #endif + unsubscribe->stat.write = MQTT_MSG_HEADER; + } + if (unsubscribe->stat.write == MQTT_MSG_HEADER) { /* Send unsubscribe packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &unsubscribe->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &unsubscribe->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &unsubscribe->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif unsubscribe->stat.write = MQTT_MSG_WAIT; } @@ -1986,13 +2066,12 @@ int SN_Client_Register(MqttClient *client, SN_Register *regist) } if (regist->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, ®ist->stat); if (rc != 0) { return rc; } - #endif /* Encode the register packet */ rc = SN_Encode_Register(client->tx_buf, client->tx_buf_len, regist); @@ -2002,9 +2081,7 @@ int SN_Client_Register(MqttClient *client, SN_Register *regist) SN_MSG_TYPE_REGISTER); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, ®ist->stat); return rc; } client->write.len = rc; @@ -2019,26 +2096,23 @@ int SN_Client_Register(MqttClient *client, SN_Register *regist) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, ®ist->stat); return rc; /* Error locking client */ } #endif + regist->stat.write = MQTT_MSG_HEADER; + } + if (regist->stat.write == MQTT_MSG_HEADER) { /* Send register packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, ®ist->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, ®ist->pendResp); - wm_SemUnlock(&client->lockClient); - } + , ®ist->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif regist->stat.write = MQTT_MSG_WAIT; } @@ -2125,13 +2199,12 @@ int SN_Client_Ping(MqttClient *client, SN_PingReq *ping) } if (ping->stat.write == MQTT_MSG_BEGIN) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &ping->stat); if (rc != 0) { goto ping_done; } - #endif /* Encode the ping packet as a request */ rc = SN_Encode_Ping(client->tx_buf, client->tx_buf_len, ping, @@ -2142,9 +2215,7 @@ int SN_Client_Ping(MqttClient *client, SN_PingReq *ping) SN_MSG_TYPE_PING_REQ); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &ping->stat); goto ping_done; } client->write.len = rc; @@ -2159,26 +2230,23 @@ int SN_Client_Ping(MqttClient *client, SN_PingReq *ping) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &ping->stat); goto ping_done; /* Error locking client */ } #endif + ping->stat.write = MQTT_MSG_HEADER; + } + if (ping->stat.write == MQTT_MSG_HEADER) { /* Send ping req packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &ping->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &ping->pendResp); - wm_SemUnlock(&client->lockClient); - } + , &ping->pendResp #endif + ); + if (rc != MQTT_CODE_SUCCESS) { goto ping_done; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif ping->stat.write = MQTT_MSG_WAIT; } @@ -2211,9 +2279,68 @@ int SN_Client_Ping(MqttClient *client, SN_PingReq *ping) return rc; } +/* Per-call claim on the client-owned disconnectSN, mirroring the NULL ping's + * pingSN_busy: it rejects re-entry and concurrent callers for the duration of + * one call, while write ownership serializes the resumable send itself. */ +static int SN_Client_DisconnectInternalClaim(MqttClient *client) +{ +#ifdef WOLFMQTT_MULTITHREAD + int rc; + + rc = wm_SemLock(&client->lockClient); + if (rc == 0) { + if (client->disconnectSN_busy) { + rc = MQTT_CODE_CONTINUE; + } + else { + client->disconnectSN_busy = 1; + } + wm_SemUnlock(&client->lockClient); + } + return rc; +#else + if (client->disconnectSN_busy) { + return MQTT_CODE_CONTINUE; + } + client->disconnectSN_busy = 1; + return MQTT_CODE_SUCCESS; +#endif +} + +static void SN_Client_DisconnectInternalRelease(MqttClient *client) +{ +#ifdef WOLFMQTT_MULTITHREAD + if (wm_SemLock(&client->lockClient) == 0) { + client->disconnectSN_busy = 0; + wm_SemUnlock(&client->lockClient); + } +#else + client->disconnectSN_busy = 0; +#endif +} + int SN_Client_Disconnect(MqttClient *client) { - return SN_Client_Disconnect_ex(client, NULL); + int rc; + + if (client == NULL) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); + } + rc = SN_Client_DisconnectInternalClaim(client); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } + /* A plain disconnect uses the client-owned object so a partial write keeps + * its resume position across MQTT_CODE_CONTINUE. write == BEGIN means no + * exchange is active, so stale terminal state is cleared before reuse. */ + if (client->disconnectSN.stat.write == MQTT_MSG_BEGIN) { + XMEMSET(&client->disconnectSN, 0, sizeof(client->disconnectSN)); + } + rc = SN_Client_Disconnect_ex(client, &client->disconnectSN); + /* Release the per-call claim on every result, including CONTINUE, so the + * next invocation can resume while concurrent calls are still rejected. */ + SN_Client_DisconnectInternalRelease(client); + return rc; } int SN_Client_Disconnect_ex(MqttClient *client, SN_Disconnect *disconnect) @@ -2224,15 +2351,19 @@ int SN_Client_Disconnect_ex(MqttClient *client, SN_Disconnect *disconnect) if (client == NULL) { return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); } + if (disconnect == NULL) { + /* A plain disconnect goes through the guarded wrapper, which claims the + * client-owned disconnectSN for this call and passes it back here. */ + return SN_Client_Disconnect(client); + } - if ((disconnect == NULL) || (disconnect->stat.write == MQTT_MSG_BEGIN)) { - #ifdef WOLFMQTT_MULTITHREAD - /* Lock send socket mutex */ - rc = wm_SemLock(&client->lockSend); + if (disconnect->stat.write == MQTT_MSG_BEGIN) { + /* Take write ownership; a write already in progress on this thread is + * reported as MQTT_CODE_CONTINUE instead of deadlocking on lockSend. */ + rc = MqttWriteStart(client, &disconnect->stat); if (rc != 0) { return rc; } - #endif /* Encode the disconnect packet */ rc = SN_Encode_Disconnect(client->tx_buf, client->tx_buf_len, @@ -2243,15 +2374,13 @@ int SN_Client_Disconnect_ex(MqttClient *client, SN_Disconnect *disconnect) SN_MSG_TYPE_DISCONNECT); #endif if (rc <= 0) { - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif + MqttWriteStop(client, &disconnect->stat); return rc; } client->write.len = rc; #ifdef WOLFMQTT_MULTITHREAD - if ((disconnect != NULL) && (disconnect->sleepTmr != 0)) { + if (disconnect->sleepTmr != 0) { rc = wm_SemLock(&client->lockClient); if (rc == 0) { /* inform other threads of expected response */ @@ -2261,32 +2390,28 @@ int SN_Client_Disconnect_ex(MqttClient *client, SN_Disconnect *disconnect) wm_SemUnlock(&client->lockClient); } if (rc != 0) { - wm_SemUnlock(&client->lockSend); + MqttWriteStop(client, &disconnect->stat); return rc; /* Error locking client */ } } #endif + disconnect->stat.write = MQTT_MSG_HEADER; + } + if (disconnect->stat.write == MQTT_MSG_HEADER) { /* Send disconnect packet */ - rc = MqttPacket_Write(client, client->tx_buf, client->write.len); - if (rc != client->write.len) { + rc = SN_Client_WriteOwned(client, &disconnect->stat #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - if ((disconnect != NULL) && (disconnect->sleepTmr != 0)) { - if (wm_SemLock(&client->lockClient) == 0) { - MqttClient_RespList_Remove(client, &disconnect->pendResp); - wm_SemUnlock(&client->lockClient); - } - } + , (disconnect->sleepTmr != 0) ? &disconnect->pendResp : NULL #endif + ); + if (rc != MQTT_CODE_SUCCESS) { return rc; } - #ifdef WOLFMQTT_MULTITHREAD - wm_SemUnlock(&client->lockSend); - #endif /* Only a sleep request is acknowledged by the gateway */ - if ((disconnect == NULL) || (disconnect->sleepTmr == 0)) { + if (disconnect->sleepTmr == 0) { + disconnect->stat.write = MQTT_MSG_BEGIN; return MQTT_CODE_SUCCESS; } diff --git a/tests/test_mqtt_sn_client.c b/tests/test_mqtt_sn_client.c index 340defd50..41542ef05 100644 --- a/tests/test_mqtt_sn_client.c +++ b/tests/test_mqtt_sn_client.c @@ -1968,6 +1968,130 @@ TEST(sn_register_auto_reply_resumes_partial_write) } #endif +#ifdef WOLFMQTT_NONBLOCK +/* A plain SN_Client_Disconnect (NULL object) must keep its resume position + * across MQTT_CODE_CONTINUE through the client-owned state: a transport that + * accepts one byte per write still completes exactly one DISCONNECT frame, + * never restarting at byte zero and duplicating the prefix. */ +TEST(sn_disconnect_chunked_write_completes_exact_frame) +{ + static const byte disconnect_frame[] = { 0x02, SN_MSG_TYPE_DISCONNECT }; + int rc; + int i; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + g_mock.write_chunk = 1; + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = SN_Client_Disconnect(&g_client); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ((int)sizeof(disconnect_frame), g_mock.out_len); + ASSERT_MEM_EQ(disconnect_frame, g_mock.out, sizeof(disconnect_frame)); +} +#endif + +/* A publish object must be reusable after a terminal write error: the failed + * send resets its state so the next attempt re-encodes instead of resuming a + * cleared, zero-length write. */ +TEST(sn_publish_reusable_after_write_error) +{ + SN_Publish publish; + word16 topic_id = SN_TEST_PUB_TOPIC_ID; + int rc; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + sn_publish_setup(&publish, &topic_id, MQTT_QOS_0); + g_mock.write_fail_rc = MQTT_CODE_ERROR_NETWORK; + rc = SN_Client_Publish(&g_client, &publish); + ASSERT_TRUE(rc < 0); + + g_mock.write_fail_rc = 0; + rc = SN_Client_Publish(&g_client, &publish); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_TRUE(g_mock.out_len > 0); +} + +#if defined(WOLFMQTT_NONBLOCK) && defined(WOLFMQTT_MULTITHREAD) && \ + !defined(WOLFMQTT_ALLOW_NODATA_UNLOCK) +/* A request whose write returns MQTT_CODE_CONTINUE must keep write ownership + * and its pending response: tx_buf and client->write still describe the + * unfinished packet, so releasing the writer would let another sender overwrite + * them and resume from a stale offset. The next call must resume that same + * write and put exactly one packet on the wire. */ +TEST(sn_register_partial_write_retains_send_state) +{ + SN_Register regist; + int rc; + int i; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + XMEMSET(®ist, 0, sizeof(regist)); + regist.topicName = "wolf/reg"; + regist.packet_id = 7; + g_mock.write_continue_count = 1; + + rc = SN_Client_Register(&g_client, ®ist); + ASSERT_EQ(MQTT_CODE_CONTINUE, rc); + /* The write is unfinished, so the request still owns the send path: + * pre-fix the state stayed at BEGIN and the pending response was + * dropped. */ + ASSERT_EQ(MQTT_MSG_HEADER, (int)regist.stat.write); + ASSERT_NOT_NULL(g_client.firstPendResp); + + /* Resume: the same packet completes, then the REGACK is consumed. */ + mock_net_push(&g_mock, SN_REGACK_ACCEPTED_FRAME, + (int)sizeof(SN_REGACK_ACCEPTED_FRAME)); + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = SN_Client_Register(&g_client, ®ist); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + /* Exactly one REGISTER reached the wire: an SN packet's first byte is its + * length, so a re-encoded or duplicated send would exceed it. */ + ASSERT_TRUE(g_mock.out_len > 1); + ASSERT_EQ((int)g_mock.out[0], g_mock.out_len); + ASSERT_EQ(SN_MSG_TYPE_REGISTER, g_mock.out[1]); + ASSERT_NULL(g_client.firstPendResp); +} + +/* Retaining write ownership across MQTT_CODE_CONTINUE must not deadlock a + * second send from the same thread: MqttWriteStart reports the in-progress + * write as MQTT_CODE_CONTINUE instead of blocking on the non-recursive + * lockSend, and the original request still resumes and completes. */ +TEST(sn_register_partial_write_second_send_not_deadlocked) +{ + SN_Register regist; + SN_PingReq ping; + int rc; + int i; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + XMEMSET(®ist, 0, sizeof(regist)); + regist.topicName = "wolf/reg"; + regist.packet_id = 7; + g_mock.write_continue_count = 1; + rc = SN_Client_Register(&g_client, ®ist); + ASSERT_EQ(MQTT_CODE_CONTINUE, rc); + + /* A second send while the Register write is unfinished must be turned + * away as busy, not block forever on the lock this thread holds. */ + XMEMSET(&ping, 0, sizeof(ping)); + rc = SN_Client_Ping(&g_client, &ping); + ASSERT_EQ(MQTT_CODE_CONTINUE, rc); + + /* The original request still resumes and completes. */ + mock_net_push(&g_mock, SN_REGACK_ACCEPTED_FRAME, + (int)sizeof(SN_REGACK_ACCEPTED_FRAME)); + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = SN_Client_Register(&g_client, ®ist); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_NULL(g_client.firstPendResp); +} +#endif + /* REGISTER has the same callback-before-response ordering requirement: the * callback runs once, then REGACK waits for the busy writer to be released. */ #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_NONBLOCK) && \ @@ -2105,12 +2229,15 @@ TEST(sn_pingreq_zero_progress_releases_writer) ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); - /* A completed direct send leaves cumulative write.total populated because - * the MQTT-SN send path owns lockSend directly rather than using - * MqttWriteStop. The following response still starts with zero progress. */ + /* A completed direct send now releases the writer through MqttWriteStop, + * which clears the cumulative write.total. Validate that invariant, then + * plant a stale nonzero total explicitly so the following zero-progress + * response must judge its own progress rather than inherit this value. */ sn_publish_setup(&publish, &topic_id, MQTT_QOS_0); ASSERT_EQ(MQTT_CODE_SUCCESS, SN_Client_Publish(&g_client, &publish)); - ASSERT_TRUE(g_client.write.total > 0); + ASSERT_EQ(0, (int)g_client.write.total); + ASSERT_TRUE(g_mock.out_len > 0); + g_client.write.total = g_mock.out_len; g_mock.out_len = 0; g_mock.write_zero_count = 1; @@ -2845,6 +2972,7 @@ int main(int argc, char** argv) RUN_TEST(sn_publish_qos1_no_continue); RUN_TEST(sn_publish_qos2_no_continue); RUN_TEST(sn_publish_qos0_no_pendresp); + RUN_TEST(sn_publish_reusable_after_write_error); RUN_TEST(sn_unsubscribe_no_continue); RUN_TEST(sn_publish_incoming_null_msg_cb_errors_no_ack); RUN_TEST(sn_ping_no_continue); @@ -2871,6 +2999,11 @@ int main(int argc, char** argv) #ifdef WOLFMQTT_NONBLOCK RUN_TEST(sn_publish_auto_reply_resumes_partial_write); RUN_TEST(sn_register_auto_reply_resumes_partial_write); + RUN_TEST(sn_disconnect_chunked_write_completes_exact_frame); +#if defined(WOLFMQTT_MULTITHREAD) && !defined(WOLFMQTT_ALLOW_NODATA_UNLOCK) + RUN_TEST(sn_register_partial_write_retains_send_state); + RUN_TEST(sn_register_partial_write_second_send_not_deadlocked); +#endif RUN_TEST(sn_pubrec_auto_reply_resumes_partial_write); RUN_TEST(sn_pingreq_auto_reply_resumes_zero_progress_write); #endif diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index b2eea3eaa..5863c17f1 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -406,6 +406,10 @@ typedef struct _MqttClient { SN_Object msgSN; SN_PingReq pingSN; /* persistent state for a NULL SN ping request */ byte pingSN_busy; /* one caller at a time owns pingSN */ + /* Persistent state for a NULL SN disconnect, so a partial write keeps its + * resume position across MQTT_CODE_CONTINUE like every other send. */ + SN_Disconnect disconnectSN; + byte disconnectSN_busy; /* one caller at a time owns disconnectSN */ SN_MsgType sn_wait_packet_type; word16 sn_wait_packet_id; SN_ClientRegisterCb reg_cb; From 50dd4639804915d64151330b2f3587db79e71ea0 Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Thu, 3 Sep 2026 09:48:52 -0700 Subject: [PATCH 06/10] Shorten inline comments to three lines or fewer --- src/mqtt_broker.c | 13 +++++-------- src/mqtt_client.c | 16 ++++++---------- src/mqtt_sn_client.c | 31 ++++++++++--------------------- wolfmqtt/mqtt_client.h | 15 ++++++--------- 4 files changed, 27 insertions(+), 48 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index dd184098d..ae9c6d1b0 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -3813,10 +3813,9 @@ static int BrokerOrphan_Reclaim(MqttBroker* broker, BrokerClient* new_bc) WBLOG_INFO(broker, "broker: orphan reclaimed client_id=%s queued=%d", BrokerLog_Sanitize(new_bc->client_id), new_bc->out_q_count); - /* The durable OUTQ records are kept: each is removed only by its terminal - * acknowledgement (BrokerClient_OnPubAck / OnPubComp). Wiping them here - * would lose every unacknowledged message if the broker stopped after - * reclaim but before the drain re-sent and the peer acknowledged. */ + /* Keep the durable OUTQ records: each is removed only by its terminal ack + * (BrokerClient_OnPubAck / OnPubComp). Wiping them here would lose every + * unacknowledged message if the broker stopped before those acks. */ BrokerOrphan_Remove(broker, o); return 1; } @@ -8240,10 +8239,8 @@ static int BrokerHandle_PublishRec(BrokerClient* bc, int rx_len) * PUBREL we send below is correlated to this entry; PUBCOMP from the * subscriber will then close it out. A spurious PUBREC (no matching * entry) still gets a PUBREL response for idempotency, just no - * queue state change. A negative return means the PUBREC targeted a - * non-QoS 2 PUBLISH, a Protocol Error [MQTT-4.13.1-1]: that entry stays - * awaiting its PUBACK, no PUBREL is sent, and the fatal code closes the - * peer. */ + * queue state change. A negative return means a QoS 1 target, a Protocol + * Error [MQTT-4.13.1-1]: no PUBREL, and the fatal code closes the peer. */ if (BrokerClient_OnPubRec(bc, resp.packet_id) < 0) { return MQTT_CODE_ERROR_PACKET_TYPE; } diff --git a/src/mqtt_client.c b/src/mqtt_client.c index e4b45a861..4962204ec 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -2815,11 +2815,9 @@ int MqttClient_SetPropertyCallback(MqttClient *client, MqttPropertyCb propCb, #ifdef WOLFMQTT_V5 /* Return 1 if the CONNECT carries an Authentication Method property, i.e. the * connection is negotiating enhanced authentication [MQTT-4.12]. */ -/* Record whether the CONNECT carried an Authentication Method and, if so, its - * value, so a later client-initiated AUTH can be required to reuse the same - * method [MQTT-4.12.0-1]. A value longer than MQTT_AUTH_METHOD_MAX is noted by - * auth_method_len but not stored, and MqttClient_Auth then refuses re-auth - * because it cannot verify the match. */ +/* Record the CONNECT Authentication Method (presence and value) so a later AUTH + * must reuse it [MQTT-4.12.0-1]. A value longer than MQTT_AUTH_METHOD_MAX sets + * auth_method_len but is not stored, so MqttClient_Auth refuses re-auth. */ static void MqttClient_StoreAuthMethod(MqttClient* client, const MqttConnect* mc_connect) { @@ -3621,11 +3619,9 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, * re-entry into this function. */ int is_dup = (publish->qos == MQTT_QOS_2 && MqttClient_RecvQos2_Contains(client, publish->packet_id)); - /* A new QoS 2 id that cannot be recorded because the de-duplication table is - * full must not be delivered: its slot would go untracked and a later - * retransmit would reach the application a second time [MQTT-4.3.3-10]. The - * payload is still drained to keep the stream in sync, then the exchange is - * refused without a PUBREC so the sender retries once a slot is free. */ + /* A new QoS 2 id that will not fit the dedup table must not be delivered: + * untracked, its retransmit would reach the application a second time + * [MQTT-4.3.3-10]. Drained to stay in sync, then the exchange is refused. */ int untrackable = (publish->qos == MQTT_QOS_2 && !is_dup && !MqttClient_RecvQos2_HasFreeSlot(client)); int suppress_cb = (is_dup || untrackable); diff --git a/src/mqtt_sn_client.c b/src/mqtt_sn_client.c index 4e58fc8df..b650c8b7e 100644 --- a/src/mqtt_sn_client.c +++ b/src/mqtt_sn_client.c @@ -960,15 +960,10 @@ static void SN_Client_UnlinkPendResp(MqttClient* client, } #endif -/* Drive the packet already encoded in tx_buf to the transport under the write - * ownership taken by MqttWriteStart. Returns MQTT_CODE_CONTINUE with ownership - * and the pending response kept, so the caller resumes here on its next call; - * except that when nothing reached the transport and - * WOLFMQTT_ALLOW_NODATA_UNLOCK is set, the writer is released, the pending - * response unlinked, and *stat reset to MQTT_MSG_BEGIN so the next call - * re-encodes. Returns MQTT_CODE_SUCCESS on completion with the writer released. - * On a short or failed write the writer is released, the pending response - * unlinked, *stat reset, and the write result returned. */ +/* Send the tx_buf packet under the ownership taken by MqttWriteStart. On + * MQTT_CODE_CONTINUE ownership and the pending response are kept for resume + * (released and reset to BEGIN only when a zero-progress ALLOW_NODATA_UNLOCK + * write occurred); on success/short/failed writes the writer is released. */ static int SN_Client_WriteOwned(MqttClient* client, MqttMsgStat* stat #ifdef WOLFMQTT_MULTITHREAD , MqttPendResp* pendResp @@ -1337,10 +1332,8 @@ static int SN_WillMessage(MqttClient *client, SN_Will *will) return rc; } - /* The encoded WILLMSG contains the will payload (potentially - * sensitive). Scrub tx_buf before releasing the writer so another - * thread cannot observe residual plaintext (mirrors the mitigation - * in MqttClient_Connect). */ + /* Scrub the will payload from tx_buf before releasing the writer so + * another thread cannot observe residual plaintext. */ CLIENT_FORCE_ZERO(client->tx_buf, xfer); MqttWriteStop(client, &will->stat); if (rc == xfer) { @@ -1626,14 +1619,10 @@ int SN_Client_WillMsgUpdate(MqttClient *client, SN_Will *will) if (will->stat.write == MQTT_MSG_HEADER) { int xfer; - /* Send Will Message Update packet. The encoded WILLMSGUPD holds - * will->willMsg (a possibly rotated/secret will payload), which must - * never stay in the shared tx_buf across an API return. So the packet - * is re-encoded on every pass: a partial send scrubs tx_buf before - * returning, the identical bytes are regenerated here, and - * MqttPacket_Write resumes from the preserved write offset. The length - * is snapshotted under ownership so the completion check cannot be - * skewed by another sender. */ + /* The encoded WILLMSGUPD holds a possibly secret will payload that must + * not stay in tx_buf across a return, so it is re-encoded each pass: + * a partial send scrubs, the next pass regenerates the identical bytes, + * and MqttPacket_Write resumes from the preserved write offset. */ rc = SN_Encode_WillMsgUpdate(client->tx_buf, client->tx_buf_len, will); if (rc <= 0) { MqttWriteStop(client, &will->stat); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index 5863c17f1..cad3004c8 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -368,10 +368,9 @@ typedef struct _MqttSendId { } MqttSendId; #ifdef WOLFMQTT_V5 - /* Bytes of the CONNECT Authentication Method retained to enforce that a - * later client-initiated AUTH reuses the same method [MQTT-4.12.0-1]. A - * CONNECT method longer than this cannot be verified and re-authentication - * is refused; override in user_settings.h if longer names are needed. */ + /* Max CONNECT Authentication Method bytes retained to enforce reuse by a + * later AUTH [MQTT-4.12.0-1]; a longer method cannot be verified so re-auth + * is refused. Override in user_settings.h for longer names. */ #ifndef MQTT_AUTH_METHOD_MAX #define MQTT_AUTH_METHOD_MAX 32 #endif @@ -508,11 +507,9 @@ typedef struct _MqttClient { #endif #ifdef WOLFMQTT_V5 - /* CONNECT Authentication Method value, retained so a later client-initiated - * AUTH must reuse the same method [MQTT-4.12.0-1]. auth_method_len is the - * full value length; when it exceeds MQTT_AUTH_METHOD_MAX the value could - * not be stored and re-auth is refused. Placed at the end of the struct so - * existing member offsets are unchanged. */ + /* CONNECT Authentication Method, retained so a later AUTH must reuse it + * [MQTT-4.12.0-1]. auth_method_len is the full length; when it exceeds + * MQTT_AUTH_METHOD_MAX the value is not stored and re-auth is refused. */ word16 auth_method_len; byte auth_method[MQTT_AUTH_METHOD_MAX]; #endif From 06accd38f1fd50599b08c330d78336dfb6e403d8 Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Thu, 3 Sep 2026 10:08:06 -0700 Subject: [PATCH 07/10] Fix test build errors from a duplicate counter and an SN test guard --- tests/test_mqtt_client.c | 14 +++++++------- tests/test_mqtt_sn_client.c | 2 ++ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 4225deb01..0c1aec10c 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -405,12 +405,12 @@ static int g_last_ack_id; /* Counts msg_cb invocations so a test can assert an incoming PUBLISH was (or was * not) delivered to the application. */ #if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) -static int g_msg_cb_calls; +static int g_dedup_msg_cb_calls; static int mock_msg_cb(MqttClient* client, MqttMessage* message, byte msg_new, byte msg_done) { (void)client; (void)message; (void)msg_new; (void)msg_done; - g_msg_cb_calls++; + g_dedup_msg_cb_calls++; return MQTT_CODE_SUCCESS; } #endif @@ -6150,7 +6150,7 @@ TEST(wait_message_qos2_full_dedup_table_v311_disconnects) #endif (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); test_client.msg_cb = mock_msg_cb; - g_msg_cb_calls = 0; + g_dedup_msg_cb_calls = 0; /* Occupy every dedup slot with distinct in-flight ids awaiting PUBREL. */ for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { @@ -6174,7 +6174,7 @@ TEST(wait_message_qos2_full_dedup_table_v311_disconnects) * untracked. Now it is not delivered, no PUBREC is sent, and the connection * is torn down. */ ASSERT_EQ(MQTT_CODE_ERROR_PACKET_ID, rc); - ASSERT_EQ(0, g_msg_cb_calls); + ASSERT_EQ(0, g_dedup_msg_cb_calls); ASSERT_FALSE(g_pubresp_written); ASSERT_EQ(0, g_frames_written); ASSERT_EQ(0, (int)(MqttClient_Flags(&test_client, 0, 0) & @@ -6203,7 +6203,7 @@ TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec) test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); test_client.msg_cb = mock_msg_cb; - g_msg_cb_calls = 0; + g_dedup_msg_cb_calls = 0; for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { test_client.recv_qos2_pending[i] = (word16)(i + 1); @@ -6228,7 +6228,7 @@ TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec) * connection stays up. connect_mock_sent[4] is the PUBREC reason byte * (0x50, remlen, id_hi, id_lo, reason). */ ASSERT_EQ(MQTT_CODE_SUCCESS, rc); - ASSERT_EQ(0, g_msg_cb_calls); + ASSERT_EQ(0, g_dedup_msg_cb_calls); ASSERT_TRUE(g_pubresp_written); ASSERT_EQ(MQTT_PACKET_TYPE_PUBLISH_REC, g_last_ack_written); ASSERT_EQ(MQTT_REASON_QUOTA_EXCEEDED, connect_mock_sent[4]); @@ -6258,7 +6258,7 @@ TEST(wait_message_ex_qos2_quota_reason_not_retained) test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_IS_CONNECTED); test_client.msg_cb = mock_msg_cb; - g_msg_cb_calls = 0; + g_dedup_msg_cb_calls = 0; for (i = 0; i < MQTT_MAX_RECV_QOS2; i++) { test_client.recv_qos2_pending[i] = (word16)(i + 1); diff --git a/tests/test_mqtt_sn_client.c b/tests/test_mqtt_sn_client.c index 41542ef05..917b4a82e 100644 --- a/tests/test_mqtt_sn_client.c +++ b/tests/test_mqtt_sn_client.c @@ -2972,7 +2972,9 @@ int main(int argc, char** argv) RUN_TEST(sn_publish_qos1_no_continue); RUN_TEST(sn_publish_qos2_no_continue); RUN_TEST(sn_publish_qos0_no_pendresp); +#if defined(WOLFMQTT_NONBLOCK) || defined(WOLFMQTT_MULTITHREAD) RUN_TEST(sn_publish_reusable_after_write_error); +#endif RUN_TEST(sn_unsubscribe_no_continue); RUN_TEST(sn_publish_incoming_null_msg_cb_errors_no_ack); RUN_TEST(sn_ping_no_continue); From 8cd9066c73c433b4302db4088e4ef07e36667373 Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Thu, 3 Sep 2026 11:06:57 -0700 Subject: [PATCH 08/10] Report a short MQTT-SN write as a network error not success --- src/mqtt_sn_client.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/mqtt_sn_client.c b/src/mqtt_sn_client.c index b650c8b7e..20589f56a 100644 --- a/src/mqtt_sn_client.c +++ b/src/mqtt_sn_client.c @@ -995,7 +995,10 @@ static int SN_Client_WriteOwned(MqttClient* client, MqttMsgStat* stat SN_Client_UnlinkPendResp(client, pendResp); #endif stat->write = MQTT_MSG_BEGIN; - return rc; + /* A non-negative short write (e.g. 0 bytes) is not success: report a + * network error so callers do not advance to awaiting a reply for a + * packet that never fully reached the transport. */ + return (rc >= 0) ? MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK) : rc; } return MQTT_CODE_SUCCESS; } From 6b2564f7d79541b39466dee570e17fe84974846d Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Thu, 3 Sep 2026 11:06:57 -0700 Subject: [PATCH 09/10] Keep MQTT 3.1.1 connected when the QoS 2 dedup table is full --- src/mqtt_client.c | 16 ++++++---------- tests/test_mqtt_client.c | 29 +++++++++++++---------------- 2 files changed, 19 insertions(+), 26 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 4962204ec..41f2097cd 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -3709,23 +3709,19 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, #if WOLFMQTT_MAX_QOS >= 2 /* The new QoS 2 id could not be tracked (dedup table full) and the drained - * payload was not delivered. Complete the exchange so the connection is not - * left livelocked on a PUBLISH that is retransmitted forever. */ + * payload was not delivered. Acknowledge so the connection is not livelocked + * on an endlessly retransmitted PUBLISH; the id stays suppressed on every + * retransmit, so the message is dropped rather than delivered twice. */ if (rc == MQTT_CODE_SUCCESS && untrackable) { #ifdef WOLFMQTT_V5 if (client->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5) { /* Reject on the PUBREC with Quota Exceeded so the peer ends the - * exchange without a PUBREL and may retry once a slot frees. The - * 0x80 error bit makes MqttClient_HandlePacket skip tracking. */ + * exchange without a PUBREL and may retry once a slot frees. */ publish->resp.reason_code = MQTT_REASON_QUOTA_EXCEEDED; } - else + /* MQTT 3.1.1 has no in-band rejection, so it sends a normal PUBREC and + * keeps the connection open. */ #endif - { - /* MQTT 3.1.1 PUBREC carries no reason code, so the message cannot be - * refused in-band. Fail fatally to drop the connection. */ - rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PACKET_ID); - } } #endif diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 0c1aec10c..33014314a 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -6124,14 +6124,13 @@ TEST(wait_message_qos2_null_msg_cb_errors_no_ack) } #if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) -/* [MQTT-4.3.3-10] The inbound QoS 2 dedup table records each delivered packet id - * until its PUBREL, so a retransmit is acknowledged again without a second - * delivery. When the table is full a new packet id cannot be recorded, and - * delivering it anyway would let a later retransmit reach the application twice. - * The client must instead complete the exchange without delivering: MQTT 3.1.1 - * has no way to refuse a PUBLISH in-band, so the connection is dropped rather - * than left livelocked on an id that can never be tracked. */ -TEST(wait_message_qos2_full_dedup_table_v311_disconnects) +/* [MQTT-4.3.3-10] When the inbound QoS 2 dedup table is full a new packet id + * cannot be recorded, and delivering it would let a later retransmit reach the + * application twice. The client instead drops the message but MQTT 3.1.1 still + * answers with a normal PUBREC and keeps the connection: the id stays suppressed + * on every retransmit, so it is never delivered nor duplicated, and the session + * is not torn down. */ +TEST(wait_message_qos2_full_dedup_table_v311_pubrec_kept_open) { int rc; int i; @@ -6170,14 +6169,12 @@ TEST(wait_message_qos2_full_dedup_table_v311_disconnects) rc = MqttClient_WaitMessage(&test_client, TEST_CMD_TIMEOUT_MS); } - /* Pre-fix the message was delivered and a PUBREC sent while the id went - * untracked. Now it is not delivered, no PUBREC is sent, and the connection - * is torn down. */ - ASSERT_EQ(MQTT_CODE_ERROR_PACKET_ID, rc); + /* Pre-fix the untracked message was delivered. Now it is not delivered, a + * PUBREC is still sent, and the connection stays open. */ + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); ASSERT_EQ(0, g_dedup_msg_cb_calls); - ASSERT_FALSE(g_pubresp_written); - ASSERT_EQ(0, g_frames_written); - ASSERT_EQ(0, (int)(MqttClient_Flags(&test_client, 0, 0) & + ASSERT_TRUE(g_pubresp_written); + ASSERT_NE(0, (int)(MqttClient_Flags(&test_client, 0, 0) & MQTT_CLIENT_FLAG_IS_CONNECTED)); } @@ -7702,7 +7699,7 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_qos0_null_msg_cb_errors); RUN_TEST(wait_message_qos2_null_msg_cb_errors_no_ack); #if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) - RUN_TEST(wait_message_qos2_full_dedup_table_v311_disconnects); + RUN_TEST(wait_message_qos2_full_dedup_table_v311_pubrec_kept_open); #ifdef WOLFMQTT_V5 RUN_TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec); RUN_TEST(wait_message_ex_qos2_quota_reason_not_retained); From 276d86368a17ea7c586fce86bebd3ab6a1b0441b Mon Sep 17 00:00:00 2001 From: Aidan Garske Date: Mon, 14 Sep 2026 14:59:06 -0700 Subject: [PATCH 10/10] Fix MQTT-SN short writes and QoS 2 delivery errors --- src/mqtt_client.c | 11 ++++--- src/mqtt_sn_client.c | 18 +++++------ tests/test_mqtt_client.c | 22 ++++++------- tests/test_mqtt_sn_client.c | 62 ++++++++++++++++++++++++++++++++++++- 4 files changed, 85 insertions(+), 28 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 41f2097cd..1f927155e 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -3709,9 +3709,8 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, #if WOLFMQTT_MAX_QOS >= 2 /* The new QoS 2 id could not be tracked (dedup table full) and the drained - * payload was not delivered. Acknowledge so the connection is not livelocked - * on an endlessly retransmitted PUBLISH; the id stays suppressed on every - * retransmit, so the message is dropped rather than delivered twice. */ + * payload was not delivered. MQTT 3.1.1 cannot reject the PUBLISH in-band, + * so fail without sending a successful PUBREC. */ if (rc == MQTT_CODE_SUCCESS && untrackable) { #ifdef WOLFMQTT_V5 if (client->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5) { @@ -3719,9 +3718,11 @@ static int MqttClient_Publish_ReadPayload(MqttClient* client, * exchange without a PUBREL and may retry once a slot frees. */ publish->resp.reason_code = MQTT_REASON_QUOTA_EXCEEDED; } - /* MQTT 3.1.1 has no in-band rejection, so it sends a normal PUBREC and - * keeps the connection open. */ + else #endif + { + rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PACKET_ID); + } } #endif diff --git a/src/mqtt_sn_client.c b/src/mqtt_sn_client.c index 20589f56a..403ec07a4 100644 --- a/src/mqtt_sn_client.c +++ b/src/mqtt_sn_client.c @@ -1342,6 +1342,9 @@ static int SN_WillMessage(MqttClient *client, SN_Will *will) if (rc == xfer) { rc = 0; } + else if (rc >= 0) { + rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK); + } /* reset state */ will->stat.write = MQTT_MSG_BEGIN; @@ -1667,7 +1670,8 @@ int SN_Client_WillMsgUpdate(MqttClient *client, SN_Will *will) SN_Client_UnlinkPendResp(client, &will->pendResp); #endif will->stat.write = MQTT_MSG_BEGIN; - return rc; + return (rc >= 0) ? + MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK) : rc; } will->stat.write = MQTT_MSG_WAIT; @@ -1883,22 +1887,18 @@ int SN_Client_Publish(MqttClient *client, SN_Publish *publish) } MqttWriteStop(client, &publish->stat); - if (rc < 0) { + if (rc != xfer) { #ifdef WOLFMQTT_MULTITHREAD SN_Client_UnlinkPendResp(client, &publish->pendResp); #endif /* The writer was released and its state cleared, so the object * must re-encode on its next use rather than resume. */ publish->stat.write = MQTT_MSG_BEGIN; - return rc; + return (rc >= 0) ? + MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK) : rc; } - if (rc == xfer) { - rc = MQTT_CODE_SUCCESS; - } - else { - rc = -1; - } + rc = MQTT_CODE_SUCCESS; /* if not expecting a reply, the reset state and exit */ if ((publish->qos == MQTT_QOS_0) || diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 33014314a..d984be3a1 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -6124,13 +6124,10 @@ TEST(wait_message_qos2_null_msg_cb_errors_no_ack) } #if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) -/* [MQTT-4.3.3-10] When the inbound QoS 2 dedup table is full a new packet id - * cannot be recorded, and delivering it would let a later retransmit reach the - * application twice. The client instead drops the message but MQTT 3.1.1 still - * answers with a normal PUBREC and keeps the connection: the id stays suppressed - * on every retransmit, so it is never delivered nor duplicated, and the session - * is not torn down. */ -TEST(wait_message_qos2_full_dedup_table_v311_pubrec_kept_open) +/* [MQTT-4.3.3-10] A new QoS 2 packet id cannot be delivered when the full + * dedup table cannot record it. MQTT 3.1.1 cannot reject the PUBLISH in-band, + * so report an error without sending a successful PUBREC. */ +TEST(wait_message_qos2_full_dedup_table_v311_errors_no_ack) { int rc; int i; @@ -6169,12 +6166,11 @@ TEST(wait_message_qos2_full_dedup_table_v311_pubrec_kept_open) rc = MqttClient_WaitMessage(&test_client, TEST_CMD_TIMEOUT_MS); } - /* Pre-fix the untracked message was delivered. Now it is not delivered, a - * PUBREC is still sent, and the connection stays open. */ - ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(MQTT_CODE_ERROR_PACKET_ID, rc); ASSERT_EQ(0, g_dedup_msg_cb_calls); - ASSERT_TRUE(g_pubresp_written); - ASSERT_NE(0, (int)(MqttClient_Flags(&test_client, 0, 0) & + ASSERT_FALSE(g_pubresp_written); + ASSERT_EQ(0, g_frames_written); + ASSERT_EQ(0, (int)(MqttClient_Flags(&test_client, 0, 0) & MQTT_CLIENT_FLAG_IS_CONNECTED)); } @@ -7699,7 +7695,7 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_qos0_null_msg_cb_errors); RUN_TEST(wait_message_qos2_null_msg_cb_errors_no_ack); #if (WOLFMQTT_MAX_QOS >= 2) && (MQTT_MAX_RECV_QOS2 < 65535) - RUN_TEST(wait_message_qos2_full_dedup_table_v311_pubrec_kept_open); + RUN_TEST(wait_message_qos2_full_dedup_table_v311_errors_no_ack); #ifdef WOLFMQTT_V5 RUN_TEST(wait_message_qos2_full_dedup_table_v5_rejects_with_pubrec); RUN_TEST(wait_message_ex_qos2_quota_reason_not_retained); diff --git a/tests/test_mqtt_sn_client.c b/tests/test_mqtt_sn_client.c index 917b4a82e..bc06614ed 100644 --- a/tests/test_mqtt_sn_client.c +++ b/tests/test_mqtt_sn_client.c @@ -103,6 +103,7 @@ typedef struct MockNet { * write) */ int write_chunk; /* maximum bytes accepted per write */ int write_zero_count; /* zero-progress writes before accepting */ + int write_zero_type; /* zero only for this MQTT-SN packet type */ int write_continue_count; /* async continuations before accepting */ int read_calls; @@ -236,7 +237,8 @@ static int mock_write(void *ctx, const byte* buf, int buf_len, int timeout_ms) net->write_continue_count--; return MQTT_CODE_CONTINUE; } - if (net->write_zero_count > 0) { + if (net->write_zero_count > 0 && (net->write_zero_type == 0 || + (buf_len > 1 && buf[1] == net->write_zero_type))) { net->write_zero_count--; return 0; } @@ -2638,6 +2640,59 @@ TEST(sn_unsubscribe_crossthread_unsuback_routing) #endif /* WOLFMQTT_NONBLOCK || WOLFMQTT_MULTITHREAD */ +#ifndef WOLFMQTT_NONBLOCK +TEST(sn_connect_willmsg_zero_write_returns_network_error) +{ + SN_Connect mc; + int rc; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + mock_net_push(&g_mock, WILLTOPICREQ_FRAME, (int)sizeof(WILLTOPICREQ_FRAME)); + mock_net_push(&g_mock, WILLMSGREQ_FRAME, (int)sizeof(WILLMSGREQ_FRAME)); + sn_will_setup_connect(&mc); + g_mock.write_zero_count = 1; + g_mock.write_zero_type = SN_MSG_TYPE_WILLMSG; + + rc = sn_connect_pump(&mc, NULL); + + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_NO_PENDRESP(); +} + +TEST(sn_willmsgupd_zero_write_returns_network_error) +{ + SN_Will will; + int rc; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + XMEMSET(&will, 0, sizeof(will)); + will.willMsg = (byte*)"offline"; + will.willMsgLen = 7; + g_mock.write_zero_count = 1; + + rc = SN_Client_WillMsgUpdate(&g_client, &will); + + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_NO_PENDRESP(); +} + +TEST(sn_publish_qos1_zero_write_returns_network_error) +{ + SN_Publish publish; + word16 topic_id = SN_TEST_PUB_TOPIC_ID; + int rc; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0)); + sn_publish_setup(&publish, &topic_id, MQTT_QOS_1); + g_mock.write_zero_count = 1; + + rc = SN_Client_Publish(&g_client, &publish); + + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_NO_PENDRESP(); +} +#endif /* !WOLFMQTT_NONBLOCK */ + /* ============================================================================ * SN ping pending-response lifecycle tests (use-after-scope regression, #3132) * @@ -2974,6 +3029,11 @@ int main(int argc, char** argv) RUN_TEST(sn_publish_qos0_no_pendresp); #if defined(WOLFMQTT_NONBLOCK) || defined(WOLFMQTT_MULTITHREAD) RUN_TEST(sn_publish_reusable_after_write_error); +#endif +#ifndef WOLFMQTT_NONBLOCK + RUN_TEST(sn_connect_willmsg_zero_write_returns_network_error); + RUN_TEST(sn_willmsgupd_zero_write_returns_network_error); + RUN_TEST(sn_publish_qos1_zero_write_returns_network_error); #endif RUN_TEST(sn_unsubscribe_no_continue); RUN_TEST(sn_publish_incoming_null_msg_cb_errors_no_ack);