From 04c3188a6c4843a88bef668a589bbd60345e3c32 Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Mon, 13 Jul 2026 11:15:55 -0500 Subject: [PATCH] Fenrir fixes --- src/mqtt_broker.c | 46 +++++--- src/mqtt_client.c | 19 +++- src/mqtt_packet.c | 10 ++ tests/test_mqtt_client.c | 214 ++++++++++++++++++++++++++++++++++++ tests/test_mqtt_packet.c | 31 ++++++ tests/test_mqtt_sn_client.c | 43 ++++++++ wolfmqtt/mqtt_packet.h | 2 + 7 files changed, 342 insertions(+), 23 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 898455b8..80e05e20 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2281,13 +2281,7 @@ static void BrokerOrphan_FreeContents(BrokerOrphanSession* o) cur = o->out_q_head; while (cur != NULL) { BrokerOutPub* next = cur->next; - if (cur->topic != NULL) { - WOLFMQTT_FREE(cur->topic); - } - if (cur->payload != NULL) { - WOLFMQTT_FREE(cur->payload); - } - WOLFMQTT_FREE(cur); + BrokerOutPub_Free(cur); cur = next; } o->out_q_head = NULL; @@ -2612,9 +2606,7 @@ static void BrokerOrphan_Enqueue(MqttBroker* broker, BrokerOrphanSession* o, head->packet_id); } #endif - if (head->topic) WOLFMQTT_FREE(head->topic); - if (head->payload) WOLFMQTT_FREE(head->payload); - WOLFMQTT_FREE(head); + BrokerOutPub_Free(head); } e = BrokerOutPub_Alloc(topic, payload, payload_len); @@ -3191,7 +3183,7 @@ static void BrokerRetained_ReapExpired(MqttBroker* broker, (now - rm->store_time) >= rm->expiry_sec) { WBLOG_DBG(broker, "broker: retained expired topic=%s", BrokerLog_Sanitize(rm->topic)); - XMEMSET(rm, 0, sizeof(BrokerRetainedMsg)); + BROKER_FORCE_ZERO(rm, sizeof(BrokerRetainedMsg)); } } #else @@ -3216,8 +3208,14 @@ static void BrokerRetained_ReapExpired(MqttBroker* broker, else { broker->retained = next; } - if (cur->topic != NULL) WOLFMQTT_FREE(cur->topic); - if (cur->payload != NULL) WOLFMQTT_FREE(cur->payload); + if (cur->topic != NULL) { + BROKER_FORCE_ZERO(cur->topic, XSTRLEN(cur->topic) + 1); + WOLFMQTT_FREE(cur->topic); + } + if (cur->payload != NULL) { + BROKER_FORCE_ZERO(cur->payload, cur->payload_len); + WOLFMQTT_FREE(cur->payload); + } WOLFMQTT_FREE(cur); if (broker->retained_count > 0) { broker->retained_count--; @@ -3855,7 +3853,7 @@ static void BrokerRetained_DeliverToClient(MqttBroker* broker, (now - rm->store_time) >= rm->expiry_sec) { WBLOG_DBG(broker, "broker: retained expired topic=%s", BrokerLog_Sanitize(rm->topic)); - XMEMSET(rm, 0, sizeof(BrokerRetainedMsg)); + BROKER_FORCE_ZERO(rm, sizeof(BrokerRetainedMsg)); continue; } if (BrokerTopicMatch(filter, rm->topic)) { @@ -3912,8 +3910,14 @@ static void BrokerRetained_DeliverToClient(MqttBroker* broker, else { broker->retained = rm_next; } - if (rm->topic) WOLFMQTT_FREE(rm->topic); - if (rm->payload) WOLFMQTT_FREE(rm->payload); + if (rm->topic) { + BROKER_FORCE_ZERO(rm->topic, XSTRLEN(rm->topic) + 1); + WOLFMQTT_FREE(rm->topic); + } + if (rm->payload) { + BROKER_FORCE_ZERO(rm->payload, rm->payload_len); + WOLFMQTT_FREE(rm->payload); + } WOLFMQTT_FREE(rm); if (broker->retained_count > 0) { broker->retained_count--; @@ -3972,8 +3976,14 @@ static void BrokerRetained_DeliverToClient(MqttBroker* broker, else { broker->retained = pnext; } - if (p->topic) WOLFMQTT_FREE(p->topic); - if (p->payload) WOLFMQTT_FREE(p->payload); + if (p->topic) { + BROKER_FORCE_ZERO(p->topic, XSTRLEN(p->topic) + 1); + WOLFMQTT_FREE(p->topic); + } + if (p->payload) { + BROKER_FORCE_ZERO(p->payload, p->payload_len); + WOLFMQTT_FREE(p->payload); + } WOLFMQTT_FREE(p); if (broker->retained_count > 0) { broker->retained_count--; diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 5ae00939..ce4484db 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -2108,7 +2108,7 @@ static int MqttClient_Publish_WritePayload(MqttClient *client, return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); if (pubCb) { /* use publish callback to get data */ - word32 tmp_len = publish->buffer_len; + word32 tmp_len; do { /* use the client->write.len to handle non-blocking re-entry when @@ -2121,15 +2121,24 @@ static int MqttClient_Publish_WritePayload(MqttClient *client, #endif return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_CALLBACK); } - } - if ((word32)client->write.len < publish->buffer_len) { - /* Last read */ - tmp_len = (word32)client->write.len; + /* Record how many valid bytes this callback produced (a short + * return marks the last read). Persisting it means a + * non-blocking resume does not mistake the in-progress chunk + * length for the fill length and drop the tail. */ + publish->intBuf_cb_len = + ((word32)client->write.len < publish->buffer_len) ? + (word32)client->write.len : publish->buffer_len; } + tmp_len = publish->intBuf_cb_len; + /* Send payload */ do { + /* Recompute the bytes remaining each pass so the final partial + * chunk copies only valid data and never reads past the end of + * the caller's payload buffer. */ + client->write.len = (int)(tmp_len - publish->intBuf_pos); if (client->write.len > client->tx_buf_len) { client->write.len = client->tx_buf_len; } diff --git a/src/mqtt_packet.c b/src/mqtt_packet.c index fc51f099..2f78344a 100644 --- a/src/mqtt_packet.c +++ b/src/mqtt_packet.c @@ -788,6 +788,16 @@ int MqttEncode_Props(MqttPacketType packet, MqttProp* props, byte* buf) } case MQTT_DATA_TYPE_STRING: { + /* [MQTT-3.3.2-14] A Response Topic is a Topic Name and MUST NOT + * contain wildcards. Reject here so the encoder never emits a + * property block its own decoder (or a peer) treats as + * malformed. */ + if (cur_prop->type == MQTT_PROP_RESP_TOPIC && + !MqttPacket_TopicNameValid(cur_prop->data_str.str, + cur_prop->data_str.len, + MQTT_CONNECT_PROTOCOL_LEVEL_5)) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA); + } tmp = MqttEncode_Data(buf, (const byte*)cur_prop->data_str.str, cur_prop->data_str.len); diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 406faaae..7dfa620c 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -963,6 +963,215 @@ TEST(publish_null_publish) ASSERT_EQ(MQTT_CODE_ERROR_BAD_ARG, rc); } +/* f-6805 streaming-publish chunk regression. The payload is larger than the + * client tx buffer and its length is not a multiple of it, so the callback loop + * must send a short final chunk. The backing buffer holds the valid payload + * followed by a guard region; the guard bytes must never reach the wire. */ +#define PUB_STREAM_VALID_LEN 300 /* > tx_buf (256), not a multiple of it */ +#define PUB_STREAM_GUARD_LEN 256 +#define PUB_STREAM_FILL 0xAA +#define PUB_STREAM_GUARD 0xEE + +#define PUB_MULTI_TOTAL_LEN 700 /* > buffer_len: 300 + 300 + 100 fills */ + +static byte pub_stream_backing[PUB_STREAM_VALID_LEN + PUB_STREAM_GUARD_LEN]; +static byte pub_stream_wire[1024]; +static int pub_stream_wire_len; +/* Length of the first write (the fixed header). Recorded so payload-byte counts + * can skip it and stay independent of the header's encoded byte values. */ +static int pub_stream_hdr_len; + +static int mock_net_write_accum(void *context, const byte* buf, int buf_len, + int timeout_ms) +{ + (void)context; (void)timeout_ms; + if (pub_stream_wire_len == 0) { + pub_stream_hdr_len = buf_len; /* first write is the fixed header */ + } + if (buf != NULL && buf_len > 0 && + pub_stream_wire_len + buf_len <= (int)sizeof(pub_stream_wire)) { + XMEMCPY(pub_stream_wire + pub_stream_wire_len, buf, (size_t)buf_len); + pub_stream_wire_len += buf_len; + } + return buf_len; /* accept the whole chunk */ +} + +/* Fills up to buffer_len bytes per call, tracking the total already sent via + * publish->buffer_pos, so a payload larger than the fill buffer drives multiple + * callback invocations (the outer streaming loop). */ +static int pub_multifill_cb(MqttPublish* publish) +{ + word32 remaining = publish->total_len - publish->buffer_pos; + word32 n = (remaining < publish->buffer_len) ? remaining : + publish->buffer_len; + XMEMSET(publish->buffer, PUB_STREAM_FILL, (size_t)n); + return (int)n; +} + +/* Fills only the valid region of the callback buffer, leaving the guard bytes + * that follow it untouched. */ +static int pub_stream_cb(MqttPublish* publish) +{ + XMEMSET(publish->buffer, PUB_STREAM_FILL, PUB_STREAM_VALID_LEN); + return PUB_STREAM_VALID_LEN; +} + +/* f-6805: the streaming-publish callback loop must copy only the bytes that + * remain on the final chunk. Before the fix the last copy reused the clamped + * tx_buf_len and read past publish->buffer, leaking adjacent memory onto the + * wire and over-sending the payload. Drive a 300-byte payload through a 256-byte + * tx buffer and assert none of the guard bytes past the payload reach the wire. */ +TEST(publish_stream_cb_final_chunk_no_overrun) +{ + int rc, i, guard_seen = 0; + MqttPublish publish; + + rc = test_init_client(); /* tx_buf_len = TEST_TX_BUF_SIZE (256) */ + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + pub_stream_wire_len = 0; + XMEMSET(pub_stream_wire, 0, sizeof(pub_stream_wire)); + XMEMSET(pub_stream_backing, PUB_STREAM_GUARD, sizeof(pub_stream_backing)); + test_net.write = mock_net_write_accum; + + XMEMSET(&publish, 0, sizeof(publish)); + publish.qos = MQTT_QOS_0; + publish.topic_name = "t"; + publish.buffer = pub_stream_backing; + publish.buffer_len = PUB_STREAM_VALID_LEN; + publish.total_len = PUB_STREAM_VALID_LEN; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Publish_ex(&test_client, &publish, pub_stream_cb); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* No guard byte (past the declared payload) may have reached the wire. */ + for (i = 0; i < pub_stream_wire_len; i++) { + if (pub_stream_wire[i] == PUB_STREAM_GUARD) { + guard_seen = 1; + break; + } + } + ASSERT_FALSE(guard_seen); + + /* Exactly the declared payload was transported (the header is a handful of + * bytes); the buggy path over-sent a full extra tx_buf chunk. */ + ASSERT_TRUE(pub_stream_wire_len <= PUB_STREAM_VALID_LEN + 16); +} + +/* f-6805 multi-fill coverage: when total_len > buffer_len the callback is + * invoked repeatedly, so the outer loop must advance buffer_pos by each fill and + * recompute intBuf_cb_len per fill. Drive a 700-byte payload through a 300-byte + * callback buffer and 256-byte tx buffer; assert the whole payload is + * transported exactly once, with no guard bytes. */ +TEST(publish_stream_cb_multifill_full_payload) +{ + int rc, i, fill_seen = 0, guard_seen = 0; + MqttPublish publish; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + pub_stream_wire_len = 0; + pub_stream_hdr_len = 0; + XMEMSET(pub_stream_wire, 0, sizeof(pub_stream_wire)); + XMEMSET(pub_stream_backing, PUB_STREAM_GUARD, sizeof(pub_stream_backing)); + test_net.write = mock_net_write_accum; + + XMEMSET(&publish, 0, sizeof(publish)); + publish.qos = MQTT_QOS_0; + publish.topic_name = "t"; + publish.buffer = pub_stream_backing; + publish.buffer_len = PUB_STREAM_VALID_LEN; /* 300 bytes per callback fill */ + publish.total_len = PUB_MULTI_TOTAL_LEN; /* 700 total -> 3 fills */ + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 40 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Publish_ex(&test_client, &publish, pub_multifill_cb); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* Skip the header bytes; count only the payload region. */ + for (i = pub_stream_hdr_len; i < pub_stream_wire_len; i++) { + if (pub_stream_wire[i] == PUB_STREAM_FILL) fill_seen++; + if (pub_stream_wire[i] == PUB_STREAM_GUARD) guard_seen++; + } + ASSERT_EQ(PUB_MULTI_TOTAL_LEN, fill_seen); + ASSERT_EQ(0, guard_seen); +} + +#ifdef WOLFMQTT_NONBLOCK +/* Forces exactly one partial (short) write on the first payload-sized chunk so + * MqttPacket_Write returns MQTT_CODE_CONTINUE and the publish re-enters + * mid-chunk. Each write's accepted bytes are appended to the wire capture. */ +static int pub_resume_did_partial; +static int mock_net_write_resume(void *context, const byte* buf, int buf_len, + int timeout_ms) +{ + int n = buf_len; + (void)context; (void)timeout_ms; + if (!pub_resume_did_partial && buf_len >= 64) { + n = 100; /* accept only part of the first chunk -> CONTINUE */ + pub_resume_did_partial = 1; + } + if (pub_stream_wire_len == 0) { + pub_stream_hdr_len = n; /* first write is the fixed header */ + } + if (buf != NULL && n > 0 && + pub_stream_wire_len + n <= (int)sizeof(pub_stream_wire)) { + XMEMCPY(pub_stream_wire + pub_stream_wire_len, buf, (size_t)n); + pub_stream_wire_len += n; + } + return n; +} + +/* f-6805 non-blocking resume: when the transport partially accepts a chunk the + * streaming publish re-enters mid-fill. The callback fill length must survive + * that re-entry (via publish->intBuf_cb_len); otherwise the resumed loop drops + * the payload tail and re-sends the head, over-sending and desyncing framing. + * Assert exactly the declared payload (all fill bytes, no guard bytes) reaches + * the wire. */ +TEST(publish_stream_cb_nonblock_resume_no_tail_drop) +{ + int rc, i, fill_seen = 0, guard_seen = 0; + MqttPublish publish; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + pub_stream_wire_len = 0; + pub_stream_hdr_len = 0; + pub_resume_did_partial = 0; + XMEMSET(pub_stream_wire, 0, sizeof(pub_stream_wire)); + XMEMSET(pub_stream_backing, PUB_STREAM_GUARD, sizeof(pub_stream_backing)); + test_net.write = mock_net_write_resume; + + XMEMSET(&publish, 0, sizeof(publish)); + publish.qos = MQTT_QOS_0; + publish.topic_name = "t"; + publish.buffer = pub_stream_backing; + publish.buffer_len = PUB_STREAM_VALID_LEN; + publish.total_len = PUB_STREAM_VALID_LEN; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 40 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Publish_ex(&test_client, &publish, pub_stream_cb); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* Skip the header bytes; count only the payload region. */ + for (i = pub_stream_hdr_len; i < pub_stream_wire_len; i++) { + if (pub_stream_wire[i] == PUB_STREAM_FILL) fill_seen++; + if (pub_stream_wire[i] == PUB_STREAM_GUARD) guard_seen++; + } + /* Exactly the declared payload, transported once, with no guard bytes. */ + ASSERT_EQ(PUB_STREAM_VALID_LEN, fill_seen); + ASSERT_EQ(0, guard_seen); +} +#endif /* WOLFMQTT_NONBLOCK */ + #ifdef WOLFMQTT_V5 /* Drives one QoS>0 publish to completion against the canned-response mock and * returns the final MqttClient_Publish result. The caller stages the broker's @@ -1928,6 +2137,11 @@ void run_mqtt_client_tests(void) /* MqttClient_Publish tests */ RUN_TEST(publish_null_client); RUN_TEST(publish_null_publish); + RUN_TEST(publish_stream_cb_final_chunk_no_overrun); + RUN_TEST(publish_stream_cb_multifill_full_payload); +#ifdef WOLFMQTT_NONBLOCK + RUN_TEST(publish_stream_cb_nonblock_resume_no_tail_drop); +#endif #ifdef WOLFMQTT_V5 RUN_TEST(publish_qos1_v5_broker_rejection_returns_publish_rejected); RUN_TEST(publish_qos1_v5_success_returns_success); diff --git a/tests/test_mqtt_packet.c b/tests/test_mqtt_packet.c index 3eea4888..9ab2b9e5 100644 --- a/tests/test_mqtt_packet.c +++ b/tests/test_mqtt_packet.c @@ -1331,6 +1331,36 @@ TEST(decode_publish_v5_response_topic_wildcard_rejected) ASSERT_NULL(pub.props); } +/* Encode-side counterpart of the test above. [MQTT-3.3.2-14] a Response Topic + * MUST NOT contain wildcards, so MqttEncode_Props must reject what + * MqttDecode_Props rejects instead of emitting a property block its own decoder + * (or a conformant peer) treats as malformed. MqttEncode_Publish maps any + * property-encode failure to the generic MQTT_CODE_ERROR_PROPERTY, so the wire + * packet is refused rather than serialized. */ +TEST(encode_publish_v5_response_topic_wildcard_rejected) +{ + byte tx_buf[64]; + MqttPublish pub; + MqttProp* prop; + int rc; + + XMEMSET(&pub, 0, sizeof(pub)); + pub.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + pub.topic_name = "t"; + pub.qos = MQTT_QOS_0; + + prop = MqttProps_Add(&pub.props); + ASSERT_TRUE(prop != NULL); + prop->type = MQTT_PROP_RESP_TOPIC; + prop->data_str.str = (char*)"a/#"; + prop->data_str.len = 3; + + rc = MqttEncode_Publish(tx_buf, (int)sizeof(tx_buf), &pub, 0); + ASSERT_EQ(MQTT_CODE_ERROR_PROPERTY, rc); + + MqttProps_Free(pub.props); +} + /* A single message may not carry more than the internal * MQTT_MAX_PROPS (default 30) properties; otherwise a peer can saturate the * shared property pool. 40 User Property entries exceeds that cap. The cap @@ -5074,6 +5104,7 @@ void run_mqtt_packet_tests(void) RUN_TEST(decode_publish_v5_subscription_id_zero_rejected); RUN_TEST(decode_publish_v5_topic_alias_zero_rejected); RUN_TEST(decode_publish_v5_response_topic_wildcard_rejected); + RUN_TEST(encode_publish_v5_response_topic_wildcard_rejected); RUN_TEST(decode_publish_v5_property_count_capped); RUN_TEST(decode_publish_v5_duplicate_singleton_prop_rejected); #endif diff --git a/tests/test_mqtt_sn_client.c b/tests/test_mqtt_sn_client.c index b8738c80..8a7a3b3c 100644 --- a/tests/test_mqtt_sn_client.c +++ b/tests/test_mqtt_sn_client.c @@ -210,6 +210,11 @@ static const byte WILLMSGREQ_FRAME[] = { 0x02, SN_MSG_TYPE_WILLMSGREQ }; static const byte CONNACK_FRAME[] = { 0x03, SN_MSG_TYPE_CONNACK, SN_RC_ACCEPTED }; +/* CONNACK where the gateway refuses the connection: total_len=3, type, + * return_code=SN_RC_CONGESTION (any non-accepted code). */ +static const byte CONNACK_REJECT_FRAME[] = { 0x03, SN_MSG_TYPE_CONNACK, + SN_RC_CONGESTION }; + /* Gateway response to a WILLMSGUPD: total_len=3, type, return_code. */ static const byte WILLMSGRESP_FRAME[] = { 0x03, SN_MSG_TYPE_WILLMSGRESP, SN_RC_ACCEPTED }; @@ -623,6 +628,43 @@ TEST(sn_connect_lwt_no_continue) ASSERT_NO_PENDRESP(); } +/* #6800: after SN_Client_WaitType decodes a CONNACK, the only thing that turns + * a gateway refusal into a caller-visible error is the return_code guard in + * SN_Client_Connect. Feed a CONNACK carrying a non-accepted return code and + * assert the refusal is reported as MQTT_CODE_ERROR_CONNECT_REFUSED, mirroring + * sn_subscribe_rejected. Without the guard this connect would report + * MQTT_CODE_SUCCESS over a link the gateway actually refused. Runs in every SN + * build. */ +TEST(sn_connect_refused) +{ + SN_Connect mc; + int rc; + + ASSERT_EQ(MQTT_CODE_SUCCESS, sn_client_init(0 /* no CONTINUE */)); + + mock_net_push(&g_mock, CONNACK_REJECT_FRAME, + (int)sizeof(CONNACK_REJECT_FRAME)); + + /* No LWT, so the gateway skips the WILLTOPICREQ/WILLMSGREQ exchange and the + * connect waits directly on the CONNACK. */ + XMEMSET(&mc, 0, sizeof(mc)); + mc.keep_alive_sec = 60; + mc.clean_session = 1; + mc.client_id = "wolfMQTT-sn-test"; + mc.protocol_level = SN_PROTOCOL_ID; + + rc = sn_connect_pump(&mc, NULL); + + /* Pre-fix (guard deleted) this returned MQTT_CODE_SUCCESS. */ + ASSERT_EQ(MQTT_CODE_ERROR_CONNECT_REFUSED, rc); + /* The raw gateway refusal code is still surfaced for diagnosis. */ + ASSERT_EQ(SN_RC_CONGESTION, mc.ack.return_code); + /* The client still sent the CONNECT before the refusal came back. */ + ASSERT_TRUE(g_mock.write_calls >= 1); + /* No pending response may be left dangling on the refusal path. */ + ASSERT_NO_PENDRESP(); +} + /* ============================================================================ * SN will-payload scrub test (#3137) * @@ -1438,6 +1480,7 @@ int main(int argc, char** argv) /* Happy path runs in every SN build (blocking and non-blocking). */ RUN_TEST(sn_connect_lwt_no_continue); + RUN_TEST(sn_connect_refused); RUN_TEST(sn_will_payload_scrubbed_after_send); RUN_TEST(sn_willmsgupd_payload_scrubbed_after_send); RUN_TEST(sn_willmsgupd_payload_scrubbed_on_write_error); diff --git a/wolfmqtt/mqtt_packet.h b/wolfmqtt/mqtt_packet.h index d10ffd7d..a8ae96b8 100644 --- a/wolfmqtt/mqtt_packet.h +++ b/wolfmqtt/mqtt_packet.h @@ -549,6 +549,8 @@ typedef struct _MqttMessage { MqttProp* props; byte protocol_level; #endif + word32 intBuf_cb_len; /* Valid bytes in the current publish-callback + fill; persisted across non-blocking re-entries */ } MqttMessage; typedef MqttMessage MqttPublish; /* Publish is message */