WolfSSL
/
wolfMQTT
mirror of https://github.com/wolfSSL/wolfMQTT.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Make broker retained store best-effort and tighten property validation

pull/618/head
Aidan Garske 2026-09-17 10:19:13 -07:00
parent 2fc5d43dd7
commit 375d19af5b
12 changed files with 339 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
BROKER.md
View File

@ -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

14
ChangeLog.md
View File

@ -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

2
README.md
View File

@ -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

5
examples/pub-sub/mqtt-pub.c
View File

@ -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 */

2
examples/pub-sub/mqtt-sub.c
View File

@ -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 */

39
src/mqtt_broker.c
View File

@ -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),

47
src/mqtt_client.c
View File

@ -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

14
src/mqtt_packet.c
View File

@ -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;

288
tests/test_broker_connect.c
View File

@ -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

42
tests/test_mqtt_client.c
View File

@ -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);

18
tests/test_mqtt_packet.c
View File

@ -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);

10
wolfmqtt/mqtt_packet.h
View File

@ -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