diff --git a/BROKER.md b/BROKER.md index c35cd15..1b4312e 100644 --- a/BROKER.md +++ b/BROKER.md @@ -111,6 +111,17 @@ When built with `WOLFMQTT_STATIC_MEMORY`, the broker uses fixed-size arrays inst | `BROKER_TIMEOUT_MS` | 1000 | `select()` timeout | | `BROKER_LISTEN_BACKLOG` | 128 | Listen queue depth | +Retaining a message is best-effort. A `RETAIN=1` PUBLISH whose retained copy +cannot be stored (the retained table is full at `BROKER_MAX_RETAINED`, the +payload exceeds `BROKER_MAX_PAYLOAD_LEN`, or an allocation fails) is still +delivered to every current subscriber and acknowledged with success at any QoS +or protocol level. Only the retained copy is skipped, so a later subscriber will +not receive that message until the topic is published again with room to store +it. The skipped retained copy is logged broker-side. This matches how servers +such as Mosquitto treat a full retained store: live delivery is never sacrificed +to retention. Raise `BROKER_MAX_RETAINED` (or `BROKER_MAX_PAYLOAD_LEN`) if +retained-topic capacity matters for your deployment. + The static offline queue is broker-owned fixed storage. Its dominant RAM cost is approximately `sessions * messages * (topic length + data length)` bytes, plus queue metadata. A new persistent CONNECT is refused when all session slots diff --git a/ChangeLog.md b/ChangeLog.md index f1b1c6a..248c6f5 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -29,6 +29,20 @@ previous manual keep-alive loop under `WOLFMQTT_NO_TIME` (#501) * API / Behavior Changes + - The broker now treats retaining a message as best-effort. When a + `RETAIN=1` PUBLISH cannot be stored (retained table full, oversized + payload, or allocation failure), the message is still delivered to all + current subscribers and acknowledged with success at every QoS and + protocol level; only the retained copy is skipped, and the skip is logged. + This matches Mosquitto and replaces the previous inconsistent handling + that could drop delivery on a retained-store failure. See `BROKER.md`. + - The client rejects an inbound v5 PUBLISH that carries a Topic Alias and no + longer advertises a nonzero Topic Alias Maximum in `CONNECT`. Inbound + alias resolution is not implemented, so the client advertises Topic Alias + Maximum 0 and a server that sends an alias anyway is treated as a protocol + error. Outbound Topic Alias (client to server) is unchanged. An + application that supplies a nonzero `MQTT_PROP_TOPIC_ALIAS_MAX` now gets + `MQTT_CODE_ERROR_PROPERTY` from `MqttClient_Connect`. - A v5 `CONNECT` now advertises `Receive Maximum` set to `MQTT_MAX_RECV_QOS2` (16 by default) unless the application supplied its own `MQTT_PROP_RECEIVE_MAX`. This bounds the QoS 1 and QoS 2 PUBLISH diff --git a/README.md b/README.md index 055ad64..77d0fec 100644 --- a/README.md +++ b/README.md @@ -276,7 +276,7 @@ The following v5.0 specification features are supported by the wolfMQTT client: * Maximum packet size * Server assigned client identifier * Subscription ID -* Topic Alias +* Topic Alias (outbound PUBLISH only; inbound aliases are rejected because the client advertises a Topic Alias Maximum of 0) The v5 enabled wolfMQTT client was tested with the following MQTT v5 brokers: * Mosquitto diff --git a/examples/pub-sub/mqtt-pub.c b/examples/pub-sub/mqtt-pub.c index abf3130..86f730e 100644 --- a/examples/pub-sub/mqtt-pub.c +++ b/examples/pub-sub/mqtt-pub.c @@ -297,10 +297,11 @@ int pub_client(MQTTCtx *mqttCtx) prop->data_int = (word32)mqttCtx->max_packet_size; } { - /* Topic Alias Maximum */ + /* Topic Alias Maximum. Advertise 0: the client does not resolve inbound + * Topic Aliases, so a conforming server must not send any. */ MqttProp* prop = MqttClient_PropsAdd(&mqttCtx->connect.props); prop->type = MQTT_PROP_TOPIC_ALIAS_MAX; - prop->data_short = mqttCtx->topic_alias_max; + prop->data_short = 0; } if (mqttCtx->clean_session == 0) { /* Session expiry interval */ diff --git a/examples/pub-sub/mqtt-sub.c b/examples/pub-sub/mqtt-sub.c index c4e7497..a8109c3 100644 --- a/examples/pub-sub/mqtt-sub.c +++ b/examples/pub-sub/mqtt-sub.c @@ -364,7 +364,7 @@ int sub_client(MQTTCtx *mqttCtx) /* Topic Alias Maximum */ MqttProp* prop = MqttClient_PropsAdd(&mqttCtx->connect.props); prop->type = MQTT_PROP_TOPIC_ALIAS_MAX; - prop->data_short = mqttCtx->topic_alias_max; + prop->data_short = 0; } if (mqttCtx->clean_session == 0) { /* Session expiry interval */ diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index cf44495..b065993 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -7600,9 +7600,6 @@ static int BrokerHandle_Publish(BrokerClient* bc, int rx_len, byte* payload = NULL; char* topic = NULL; MqttQoS eff_qos; -#if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_RETAINED) - int retain_rc = MQTT_CODE_SUCCESS; -#endif #if WOLFMQTT_MAX_QOS >= 2 int qos2_duplicate = 0; #endif @@ -7801,37 +7798,20 @@ static int BrokerHandle_Publish(BrokerClient* bc, int rx_len, int ret_rc = BrokerRetained_Store(broker, topic, payload, pub.total_len, pub.qos, expiry); if (ret_rc != MQTT_CODE_SUCCESS) { + /* Retaining is best-effort: a store failure (table full, + * oversized payload, transient alloc) does not reject the + * PUBLISH. The message is still delivered live and + * acknowledged; only the retained copy is not kept. */ WBLOG_ERR(broker, "Retained store failed: %s", MqttClient_ReturnCodeToString(ret_rc)); } -#ifdef WOLFMQTT_V5 - retain_rc = ret_rc; -#endif } } } #endif /* WOLFMQTT_BROKER_RETAINED */ -#if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_RETAINED) && \ - WOLFMQTT_MAX_QOS >= 2 - /* [MQTT-4.3.3] Only a v5 failure PUBREC releases the QoS 2 packet id for - * reuse, so on a rejected retained store drop the dedup entry added above - * for v5 clients only; a v3 client completes the handshake normally and a - * leftover entry would misread its next reuse as a duplicate. */ - if (bc->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5 && - retain_rc != MQTT_CODE_SUCCESS && pub.qos == MQTT_QOS_2) { - BrokerInboundQos2_Remove(bc, pub.packet_id); - } -#endif - - /* Skip fan-out for a QoS 2 duplicate (already delivered [MQTT-4.3.3]) and - * for a v5 QoS 1/2 publish whose retained store was rejected (negatively - * acknowledged below). A v3 client cannot be told of the failure and - * QoS 0 has no ack path, so both are still forwarded best-effort. */ -#if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_RETAINED) - if (bc->protocol_level < MQTT_CONNECT_PROTOCOL_LEVEL_5 || - retain_rc == MQTT_CODE_SUCCESS || pub.qos == MQTT_QOS_0) -#endif + /* Fan-out is skipped for QoS 2 duplicates: subscribers already received the + * application message from the original PUBLISH ([MQTT-4.3.3]). */ if ( #if WOLFMQTT_MAX_QOS >= 2 !qos2_duplicate && @@ -8147,13 +8127,6 @@ static int BrokerHandle_Publish(BrokerClient* bc, int rx_len, #ifdef WOLFMQTT_V5 resp.protocol_level = bc->protocol_level; resp.reason_code = MQTT_REASON_SUCCESS; - /* A retained-store failure must not be ACKed as success: tell the - * publisher the quota was exceeded [MQTT-3.4.2]. */ -#ifdef WOLFMQTT_BROKER_RETAINED - if (retain_rc != MQTT_CODE_SUCCESS) { - resp.reason_code = MQTT_REASON_QUOTA_EXCEEDED; - } -#endif resp.props = NULL; #endif rc = MqttEncode_PublishResp(bc->tx_buf, BROKER_CLIENT_TX_SZ(bc), diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 3e4c922..1ce10bb 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -26,13 +26,6 @@ #include "wolfmqtt/mqtt_client.h" -/* Mirror the shared property-pool cap from mqtt_packet.c; a build overriding it - * via CFLAGS defines it for this file too. Bounds caller-property traversal - * here the same way MqttEncode_Props does. */ -#ifndef MQTT_MAX_PROPS -#define MQTT_MAX_PROPS 30 -#endif - /* Secure memory zeroing - uses volatile pointer to prevent the compiler * from optimizing away the stores (dead-store elimination). * Declared WOLFMQTT_LOCAL in mqtt_client.h so the MQTT-SN client can reuse it @@ -1422,7 +1415,8 @@ static int MqttClient_DecodePacket(MqttClient* client, byte* rx_buf, if (prop->type == MQTT_PROP_TOPIC_ALIAS) { MqttProps_Free(p_publish->props); p_publish->props = NULL; - rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA); + rc = MQTT_TRACE_ERROR( + MQTT_CODE_ERROR_MALFORMED_DATA); break; } } @@ -3132,6 +3126,27 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); } +#ifdef WOLFMQTT_V5 + /* The client does not resolve inbound Topic Aliases, so it must not + * advertise the capability. Reject a nonzero caller-supplied Topic Alias + * Maximum before sending CONNECT rather than advertising support it cannot + * honor and then rejecting the server's aliased PUBLISH. The scan is + * bounded like MqttEncode_Props so a cyclic list cannot spin. */ + if (client->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5) { + ta_count = 0; + for (ta_prop = mc_connect->props; ta_prop != NULL; + ta_prop = ta_prop->next) { + if (++ta_count > MQTT_MAX_PROPS) { + break; + } + if (ta_prop->type == MQTT_PROP_TOPIC_ALIAS_MAX && + ta_prop->data_short != 0) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PROPERTY); + } + } + } +#endif + #ifndef WOLFMQTT_NO_SESSION_REPLAY if (mc_connect->stat.write == MQTT_MSG_PAYLOAD) { /* MQTT_MSG_PAYLOAD is not part of the CONNECT write sequence @@ -3264,22 +3279,6 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) client->server_recv_max = 65535; client->server_recv_max_negotiated = 65535; client->topic_alias_max = 0; - - /* The client does not resolve inbound Topic Aliases, so it must not - * advertise the capability: clamp any caller-supplied Topic Alias - * Maximum to 0 so a conforming server never sends one. The traversal is - * bounded like MqttEncode_Props so a cyclic list cannot spin here; a - * list longer than the cap is rejected by MqttEncode_Connect below. */ - ta_count = 0; - for (ta_prop = mc_connect->props; ta_prop != NULL; - ta_prop = ta_prop->next) { - if (++ta_count > MQTT_MAX_PROPS) { - break; - } - if (ta_prop->type == MQTT_PROP_TOPIC_ALIAS_MAX) { - ta_prop->data_short = 0; - } - } #endif #if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 2 diff --git a/src/mqtt_packet.c b/src/mqtt_packet.c index 80599aa..d3edade 100644 --- a/src/mqtt_packet.c +++ b/src/mqtt_packet.c @@ -127,10 +127,6 @@ static const struct MqttPropMatrix gPropMatrix[] = { { MQTT_PROP_TYPE_MAX, MQTT_DATA_TYPE_NONE, 0 } }; -/* Maximum number of active properties - overridable */ -#ifndef MQTT_MAX_PROPS -#define MQTT_MAX_PROPS 30 -#endif /* WOLFMQTT_DYN_PROP allows property allocation using malloc */ #ifndef WOLFMQTT_DYN_PROP @@ -1171,11 +1167,11 @@ int MqttDecode_Props(MqttPacketType packet, MqttProp** props, byte* pbuf, tmp++; total++; prop_len--; - /* [MQTT-3.1.2-28/29] Request Response/Problem Information - * MUST be 0 or 1; any other value is a Protocol Error. */ - if ((cur_prop->type == MQTT_PROP_REQ_RESP_INFO || - cur_prop->type == MQTT_PROP_REQ_PROB_INFO) && - cur_prop->data_byte > 1) { + /* Every MQTT 5 Byte property is Boolean-valued (0 or 1), and + * Maximum QoS shares the same {0,1} domain; any other value is + * a Protocol Error. Mirrors MqttEncode_Props so a decoded + * property always re-encodes. */ + if (cur_prop->data_byte > 1) { rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PROPERTY); } break; diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 11b38ff..1c06c3b 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -1442,12 +1442,6 @@ static size_t build_v5_connect_emptyid_userprops(byte* out, size_t out_sz, return pos; } -/* Mirror the library's shared property-pool cap (mqtt_packet.c); a build that - * overrides it via CFLAGS defines it for this file too. */ -#ifndef MQTT_MAX_PROPS - #define MQTT_MAX_PROPS 30 -#endif - /* A v5 CONNECT with an empty ClientId that also fills the shared property pool * with User Properties must still receive its mandatory Assigned Client * Identifier in CONNACK: the decoded CONNECT properties are released before the @@ -4243,22 +4237,62 @@ TEST(broker_retained_clock_rollback_not_expired) MqttBroker_Free(&broker); } +#if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 1 +/* Return the Reason Code of the first PUBACK/PUBREC in a captured stream. A v5 + * response that omits the reason byte (remain <= 2) means Success; -1 means no + * such packet was found. */ +static int first_publish_resp_reason(const byte* buf, size_t len) +{ + size_t pos = 0; + while (pos < len) { + byte type = (byte)((buf[pos] >> 4) & 0x0F); + size_t remain = 0; + size_t mult = 1; + size_t hdr_len = 1; + int vbi_complete = 0; + while (pos + hdr_len < len && hdr_len <= 5) { + byte b = buf[pos + hdr_len]; + remain += (size_t)(b & 0x7F) * mult; + hdr_len++; + if ((b & 0x80) == 0) { vbi_complete = 1; break; } + mult *= 128; + } + if (!vbi_complete) { + break; + } + if (type == MQTT_PACKET_TYPE_PUBLISH_ACK || + type == MQTT_PACKET_TYPE_PUBLISH_REC) { + if (remain <= 2) { + return MQTT_REASON_SUCCESS; + } + if (pos + hdr_len + 2 < len) { + return buf[pos + hdr_len + 2]; + } + return -1; + } + if (remain > len - pos - hdr_len) { + break; + } + pos += hdr_len + remain; + } + return -1; +} +#endif /* WOLFMQTT_V5 && WOLFMQTT_MAX_QOS >= 1 */ + #if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 2 -/* A v5 QoS 2 retained PUBLISH the broker cannot store is answered with a - * failure PUBREC. The rejected message is not delivered (its topic subscriber - * sees nothing) and, per [MQTT-4.3.3], its packet id is released: a later - * reuse of that id is a fresh delivery, not a silently discarded duplicate. */ -TEST(qos2_retained_store_failure_releases_packet_id) +/* Best-effort retain: a v5 QoS 2 PUBLISH whose retained store is full is still + * delivered live to subscribers and answered with a success PUBREC. Only the + * retained copy is skipped; the store stays at capacity. */ +TEST(qos2_retained_store_full_delivers_live) { MqttBroker broker; MqttBrokerNet net; int i; byte fill[8]; - int fail_topic_pubs; - int reuse_topic_pubs; + int sub_pubs; int pub_pubrecs; - /* v3.1.1 subscriber "A" on "z", the rejected publish's topic. */ + /* v3.1.1 subscriber "A" on "z". */ static const byte connect_sub_z[] = { 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'A' @@ -4266,14 +4300,6 @@ TEST(qos2_retained_store_failure_releases_packet_id) static const byte subscribe_z[] = { 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'z', 0x02 }; - /* v3.1.1 subscriber "C" on "x", the reused-id publish's topic. */ - static const byte connect_sub_x[] = { - 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', - 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'C' - }; - static const byte subscribe_x[] = { - 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x02 - }; /* v5 publisher "B". */ static const byte connect_pub[] = { 0x10, 0x0E, 0x00, 0x04, 'M', 'Q', 'T', 'T', @@ -4290,27 +4316,17 @@ TEST(qos2_retained_store_failure_releases_packet_id) 0x35, 0x0B, 0x00, 0x01, 'z', 0x00, 0x07, 0x00, 'f', 'i', 'r', 's', 't' }; - /* v5 QoS 2 PUBLISH reusing packet_id=7, topic "x", payload "second". - * remain = topic(3) + id(2) + props(1) + payload(6) = 12 */ - static const byte publish_reuse[] = { - 0x34, 0x0C, 0x00, 0x01, 'x', 0x00, 0x07, 0x00, - 's', 'e', 'c', 'o', 'n', 'd' - }; 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)); - reset_mock_clients(4); + reset_mock_clients(3); mock_client_input_append(0, connect_sub_z, sizeof(connect_sub_z)); mock_client_input_append(0, subscribe_z, sizeof(subscribe_z)); mock_client_input_append(1, connect_pub, sizeof(connect_pub)); mock_client_input_append(2, connect_fill, sizeof(connect_fill)); - mock_client_input_append(3, connect_sub_x, sizeof(connect_sub_x)); - mock_client_input_append(3, subscribe_x, sizeof(subscribe_x)); - /* Fill the retained store to its cap with distinct QoS 0 topics so the - * next retained store is rejected. */ for (i = 0; i < BROKER_MAX_RETAINED; i++) { fill[0] = 0x31; /* PUBLISH, retain=1, QoS 0 */ fill[1] = 0x06; /* remain = 6 */ @@ -4326,35 +4342,24 @@ TEST(qos2_retained_store_failure_releases_packet_id) } ASSERT_EQ(BROKER_MAX_RETAINED, broker.retained_count); - /* The publisher's QoS 2 retained PUBLISH cannot be stored, so it is - * rejected with a failure PUBREC, not forwarded, and its packet id is - * released. */ + /* Table is full, so the retained copy is skipped, but the message is still + * delivered live and answered with a success PUBREC. */ mock_client_input_append(1, publish_retain, sizeof(publish_retain)); for (i = 0; i < 16; i++) { MqttBroker_Step(&broker); } - /* The rejected topic was not stored. */ ASSERT_EQ(BROKER_MAX_RETAINED, broker.retained_count); - /* Reusing the released packet id is a fresh delivery, not a duplicate. */ - mock_client_input_append(1, publish_reuse, sizeof(publish_reuse)); - for (i = 0; i < 16; i++) { - MqttBroker_Step(&broker); - } - - fail_topic_pubs = count_packets_of_type(g_clients[0].out_buf, + sub_pubs = count_packets_of_type(g_clients[0].out_buf, g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH); - reuse_topic_pubs = count_packets_of_type(g_clients[3].out_buf, - g_clients[3].out_len, MQTT_PACKET_TYPE_PUBLISH); pub_pubrecs = count_packets_of_type(g_clients[1].out_buf, g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH_REC); - /* The rejected publish is not forwarded, while the reused-id publish is; - * pre-fix these were reversed (rejected delivered, reuse dropped as a - * duplicate). Each inbound QoS 2 PUBLISH still got one PUBREC. */ - ASSERT_EQ(0, fail_topic_pubs); - ASSERT_EQ(1, reuse_topic_pubs); - ASSERT_EQ(2, pub_pubrecs); + ASSERT_EQ(1, sub_pubs); /* delivered live despite full table */ + ASSERT_EQ(1, pub_pubrecs); + ASSERT_EQ(MQTT_REASON_SUCCESS, + first_publish_resp_reason(g_clients[1].out_buf, + g_clients[1].out_len)); MqttBroker_Stop(&broker); MqttBroker_Free(&broker); @@ -4362,10 +4367,10 @@ TEST(qos2_retained_store_failure_releases_packet_id) #endif /* WOLFMQTT_V5 && WOLFMQTT_MAX_QOS >= 2 */ #if WOLFMQTT_MAX_QOS >= 2 -/* A v3.1.1 publisher cannot be told a retained store was rejected: PUBACK and - * PUBREC carry no reason code, so the broker must keep delivering the live - * message and complete the QoS 2 handshake rather than silently dropping it. */ -TEST(qos2_retained_store_failure_v311_still_delivers) +/* Best-effort retain: a v3.1.1 QoS 2 PUBLISH whose retained store is full is + * still delivered live to subscribers, and its handshake completes normally. + * Only the retained copy is skipped. */ +TEST(qos2_retained_store_full_v311_delivers_live) { MqttBroker broker; MqttBrokerNet net; @@ -4428,8 +4433,8 @@ TEST(qos2_retained_store_failure_v311_still_delivers) } ASSERT_EQ(BROKER_MAX_RETAINED, broker.retained_count); - /* Retained store is full, so this PUBLISH cannot be retained, but the v3 - * publisher must still see it fanned out and the handshake complete. */ + /* Table is full, so the retained copy is skipped, but the message is still + * delivered live and the v3.1.1 handshake completes normally. */ mock_client_input_append(1, publish_retain, sizeof(publish_retain)); mock_client_input_append(1, pubrel, sizeof(pubrel)); for (i = 0; i < 16; i++) { @@ -4443,7 +4448,7 @@ TEST(qos2_retained_store_failure_v311_still_delivers) pub_pubcomps = count_packets_of_type(g_clients[1].out_buf, g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH_COMP); - ASSERT_EQ(1, sub_pubs); /* live message still delivered */ + ASSERT_EQ(1, sub_pubs); /* delivered live despite full table */ ASSERT_EQ(1, pub_pubrecs); ASSERT_EQ(1, pub_pubcomps); /* handshake completes normally */ @@ -4451,6 +4456,163 @@ TEST(qos2_retained_store_failure_v311_still_delivers) MqttBroker_Free(&broker); } #endif /* WOLFMQTT_MAX_QOS >= 2 */ + +#if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 1 +/* Best-effort retain on the QoS 1 path most applications hit: a v5 QoS 1 + * PUBLISH whose retained store is full is still delivered live and answered + * with a success PUBACK. Only the retained copy is skipped. */ +TEST(qos1_retained_store_full_delivers_live) +{ + MqttBroker broker; + MqttBrokerNet net; + int i; + byte fill[8]; + int sub_pubs; + int pub_pubacks; + + /* v3.1.1 subscriber "A" on "x". */ + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'A' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + /* v5 publisher "B". */ + static const byte connect_pub[] = { + 0x10, 0x0E, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x05, 0x02, 0x00, 0x3C, 0x00, 0x00, 0x01, 'B' + }; + /* v3.1.1 filler "F" used to exhaust the retained store. */ + static const byte connect_fill[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'F' + }; + /* v5 QoS 1 retained PUBLISH, packet_id=7, topic "x", payload "live". + * remain = topic(3) + id(2) + props(1) + payload(4) = 10 */ + static const byte publish_retain[] = { + 0x33, 0x0A, 0x00, 0x01, 'x', 0x00, 0x07, 0x00, + 'l', 'i', 'v', 'e' + }; + + 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)); + + reset_mock_clients(3); + 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)); + mock_client_input_append(2, connect_fill, sizeof(connect_fill)); + for (i = 0; i < BROKER_MAX_RETAINED; i++) { + fill[0] = 0x31; /* PUBLISH, retain=1, QoS 0 */ + fill[1] = 0x06; /* remain = 6 */ + fill[2] = 0x00; fill[3] = 0x03; /* topic len 3 */ + fill[4] = 'r'; + fill[5] = (byte)('0' + (i / 10)); + fill[6] = (byte)('0' + (i % 10)); + fill[7] = 'x'; /* payload */ + mock_client_input_append(2, fill, sizeof(fill)); + } + for (i = 0; i < BROKER_MAX_RETAINED + 32; i++) { + MqttBroker_Step(&broker); + } + ASSERT_EQ(BROKER_MAX_RETAINED, broker.retained_count); + + mock_client_input_append(1, publish_retain, sizeof(publish_retain)); + for (i = 0; i < 16; i++) { + MqttBroker_Step(&broker); + } + ASSERT_EQ(BROKER_MAX_RETAINED, broker.retained_count); + + sub_pubs = count_packets_of_type(g_clients[0].out_buf, + g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH); + pub_pubacks = count_packets_of_type(g_clients[1].out_buf, + g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH_ACK); + + ASSERT_EQ(1, sub_pubs); /* delivered live despite full table */ + ASSERT_EQ(1, pub_pubacks); + ASSERT_EQ(MQTT_REASON_SUCCESS, + first_publish_resp_reason(g_clients[1].out_buf, + g_clients[1].out_len)); + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} + +#ifndef WOLFMQTT_STATIC_MEMORY +/* Best-effort retain also covers a transient store failure: an allocation + * failure in the retained store still delivers the message live and + * acknowledges success, and stores nothing. */ +TEST(qos1_retained_store_alloc_failure_delivers_live) +{ + MqttBroker broker; + MqttBrokerNet net; + int i; + int sub_pubs; + int pub_pubacks; + + /* v3.1.1 subscriber "A" on "x". */ + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'A' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + /* v5 publisher "B". */ + static const byte connect_pub[] = { + 0x10, 0x0E, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x05, 0x02, 0x00, 0x3C, 0x00, 0x00, 0x01, 'B' + }; + /* v5 QoS 1 retained PUBLISH, packet_id=7, topic "x", payload "live". + * remain = topic(3) + id(2) + props(1) + payload(4) = 10 */ + static const byte publish_retain[] = { + 0x33, 0x0A, 0x00, 0x01, 'x', 0x00, 0x07, 0x00, + 'l', 'i', 'v', 'e' + }; + + 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)); + + 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++) { + MqttBroker_Step(&broker); + } + + /* The retained table is far below capacity, so the store fails only because + * the injected allocation fails. The topic copy allocates first and + * succeeds; the next allocation is the retained store's, which fails. */ + broker_test_fail_alloc_after(1); + mock_client_input_append(1, publish_retain, sizeof(publish_retain)); + for (i = 0; i < 16; i++) { + MqttBroker_Step(&broker); + } + broker_test_disable_alloc_failure(); + + sub_pubs = count_packets_of_type(g_clients[0].out_buf, + g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH); + pub_pubacks = count_packets_of_type(g_clients[1].out_buf, + g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH_ACK); + + ASSERT_EQ(1, g_alloc_failure_count); /* the retained store did fail */ + ASSERT_EQ(0, broker.retained_count); /* nothing was stored */ + ASSERT_EQ(1, sub_pubs); /* but still delivered live */ + ASSERT_EQ(1, pub_pubacks); + ASSERT_EQ(MQTT_REASON_SUCCESS, + first_publish_resp_reason(g_clients[1].out_buf, + g_clients[1].out_len)); /* acked success, not Quota */ + + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} +#endif /* !WOLFMQTT_STATIC_MEMORY */ +#endif /* WOLFMQTT_V5 && WOLFMQTT_MAX_QOS >= 1 */ #endif /* WOLFMQTT_BROKER_RETAINED && !WOLFMQTT_STATIC_MEMORY */ #ifndef WOLFMQTT_STATIC_MEMORY @@ -9638,10 +9800,14 @@ int main(int argc, char** argv) RUN_TEST(broker_retained_list_capped); RUN_TEST(broker_retained_clock_rollback_not_expired); #if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 2 - RUN_TEST(qos2_retained_store_failure_releases_packet_id); + RUN_TEST(qos2_retained_store_full_delivers_live); #endif #if WOLFMQTT_MAX_QOS >= 2 - RUN_TEST(qos2_retained_store_failure_v311_still_delivers); + RUN_TEST(qos2_retained_store_full_v311_delivers_live); +#endif +#if defined(WOLFMQTT_V5) && WOLFMQTT_MAX_QOS >= 1 + RUN_TEST(qos1_retained_store_full_delivers_live); + RUN_TEST(qos1_retained_store_alloc_failure_delivers_live); #endif RUN_TEST(broker_retained_scrub_after_completed_write); #ifdef WOLFMQTT_NONBLOCK diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index c1f69e6..1cea250 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -537,15 +537,13 @@ TEST(connect_clears_tx_buf_credentials) #ifdef WOLFMQTT_V5 /* The client cannot resolve inbound Topic Aliases, so MqttClient_Connect must * not advertise the capability: a caller-supplied nonzero Topic Alias Maximum - * is clamped to 0 in the CONNECT it sends. */ -TEST(connect_clamps_inbound_topic_alias_max) + * is rejected before any CONNECT is sent, rather than advertised and then + * contradicted when the server's aliased PUBLISH is refused. */ +TEST(connect_rejects_nonzero_inbound_topic_alias_max) { int rc; MqttConnect connect; MqttProp ta_max_prop; - /* CONNECT property wire bytes: TOPIC_ALIAS_MAX(0x22) + 2-byte value. */ - static const byte alias_max_five[] = { 0x22, 0x00, 0x05 }; - static const byte alias_max_zero[] = { 0x22, 0x00, 0x00 }; rc = test_init_client(); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); @@ -566,17 +564,16 @@ TEST(connect_clamps_inbound_topic_alias_max) connect.props = &ta_max_prop; rc = MqttClient_Connect(&test_client, &connect); - /* The read mock cannot deliver a CONNECT_ACK, so the call returns an error - * after the CONNECT is written and captured. */ - ASSERT_NE(MQTT_CODE_SUCCESS, rc); - ASSERT_TRUE(connect_mock_xfer > 0); + ASSERT_EQ(MQTT_CODE_ERROR_PROPERTY, rc); + /* Rejected before the transport was touched. */ + ASSERT_EQ(0, connect_mock_xfer); - /* The advertised maximum was clamped: the nonzero value never reached the - * wire, and a zero value did. */ - ASSERT_FALSE(buf_contains(connect_mock_sent, connect_mock_xfer, - (const char*)alias_max_five, 3)); - ASSERT_TRUE(buf_contains(connect_mock_sent, connect_mock_xfer, - (const char*)alias_max_zero, 3)); + /* A zero (or absent) Topic Alias Maximum is accepted and does reach the + * write path. */ + ta_max_prop.data_short = 0; + rc = MqttClient_Connect(&test_client, &connect); + ASSERT_NE(MQTT_CODE_ERROR_PROPERTY, rc); + ASSERT_TRUE(connect_mock_xfer > 0); } #endif /* WOLFMQTT_V5 */ @@ -2220,7 +2217,9 @@ TEST(publish_v5_within_max_packet_size_allowed) /* MQTT 5.0 section 3.2.2.3.4: Maximum QoS can only be 0 or 1. Feed an * independently constructed CONNACK containing 2 and require the client to - * reject the connection instead of normalizing the invalid wire value. */ + * reject the connection instead of normalizing the invalid wire value. The + * out-of-range Byte value is now caught in MqttDecode_Props at the wire + * boundary (MQTT_CODE_ERROR_PROPERTY). */ TEST(connect_accepted_connack_rejects_illegal_max_qos) { int rc; @@ -2253,14 +2252,14 @@ TEST(connect_accepted_connack_rejects_illegal_max_qos) rc = MqttClient_Connect(&test_client, &connect); } - ASSERT_EQ(MQTT_CODE_ERROR_SERVER_PROP, rc); - ASSERT_EQ(MQTT_CONNECT_ACK_CODE_ACCEPTED, connect.ack.return_code); + ASSERT_EQ(MQTT_CODE_ERROR_PROPERTY, rc); ASSERT_EQ(WOLFMQTT_MAX_QOS, test_client.max_qos); } /* MQTT 5.0 section 3.2.2.3.5: Retain Available can only be 0 or 1. Feed an * independently constructed CONNACK containing 2 and require a protocol - * failure rather than accepting it as Retain Available=1. */ + * failure rather than accepting it as Retain Available=1. The out-of-range + * Byte value is now caught in MqttDecode_Props (MQTT_CODE_ERROR_PROPERTY). */ TEST(connect_accepted_connack_rejects_illegal_retain_available) { int rc; @@ -2294,8 +2293,7 @@ TEST(connect_accepted_connack_rejects_illegal_retain_available) rc = MqttClient_Connect(&test_client, &connect); } - ASSERT_EQ(MQTT_CODE_ERROR_SERVER_PROP, rc); - ASSERT_EQ(MQTT_CONNECT_ACK_CODE_ACCEPTED, connect.ack.return_code); + ASSERT_EQ(MQTT_CODE_ERROR_PROPERTY, rc); ASSERT_EQ(1, test_client.retain_avail); } #endif /* WOLFMQTT_V5 */ @@ -7582,7 +7580,7 @@ void run_mqtt_client_tests(void) RUN_TEST(second_connect_on_same_network_connection_rejected); RUN_TEST(connect_clears_tx_buf_credentials); #ifdef WOLFMQTT_V5 - RUN_TEST(connect_clamps_inbound_topic_alias_max); + RUN_TEST(connect_rejects_nonzero_inbound_topic_alias_max); #endif RUN_TEST(connect_accepted_connack_returns_success); RUN_TEST(connect_clean_session_present_mismatch_refused); diff --git a/tests/test_mqtt_packet.c b/tests/test_mqtt_packet.c index 20d09f3..b27cea9 100644 --- a/tests/test_mqtt_packet.c +++ b/tests/test_mqtt_packet.c @@ -1477,6 +1477,23 @@ TEST(decode_publish_v5_topic_alias_zero_rejected) ASSERT_NULL(pub.props); } +/* A Byte property value other than 0 or 1 is a Protocol Error and must be + * rejected at decode, symmetric with MqttEncode_Props, so a decoded property + * always re-encodes. Wire: PUBLISH QoS 0, topic "t", props_len=2, + * PAYLOAD_FORMAT_IND(1)=2. */ +TEST(decode_publish_v5_byte_property_out_of_range_rejected) +{ + byte buf[] = { 0x30, 0x07, 0x00, 0x01, 't', 0x02, 0x01, 0x02, 'x' }; + MqttPublish pub; + int rc; + + XMEMSET(&pub, 0, sizeof(pub)); + pub.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + rc = MqttDecode_Publish(buf, (int)sizeof(buf), &pub); + ASSERT_EQ(MQTT_CODE_ERROR_PROPERTY, rc); + ASSERT_NULL(pub.props); +} + /* [MQTT-3.3.2-14] A Response Topic is a Topic Name and MUST NOT contain * wildcards. Wire: PUBLISH QoS 0, topic "t", props_len=6, RESP_TOPIC(8)="a/#". */ TEST(decode_publish_v5_response_topic_wildcard_rejected) @@ -6966,6 +6983,7 @@ void run_mqtt_packet_tests(void) RUN_TEST(decode_publish_v5_empty_topic_no_alias_rejected); RUN_TEST(decode_publish_v5_subscription_id_zero_rejected); RUN_TEST(decode_publish_v5_topic_alias_zero_rejected); + RUN_TEST(decode_publish_v5_byte_property_out_of_range_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); diff --git a/wolfmqtt/mqtt_packet.h b/wolfmqtt/mqtt_packet.h index a86475f..25efb2b 100644 --- a/wolfmqtt/mqtt_packet.h +++ b/wolfmqtt/mqtt_packet.h @@ -48,6 +48,16 @@ #define MAX_MQTT_TOPICS 12 #endif +/* Maximum number of MQTT v5 properties in one packet, and the size of the + * shared property pool. Also bounds property-list traversal outside the pool + * allocator. Override in user_settings.h to trade memory for a larger set. */ +#ifndef MQTT_MAX_PROPS +#define MQTT_MAX_PROPS 30 +#endif +#if (MQTT_MAX_PROPS < 1) || (MQTT_MAX_PROPS > 65535) + #error "MQTT_MAX_PROPS must be between 1 and 65535" +#endif + /* WOLFMQTT_NO_UTF8_VALIDATION * Define to disable RFC 3629 UTF-8 well-formedness validation on the * encode side (MqttEncode_Utf8Ok). Decode-side validation in