Add Testing Validation and Fixes for wolfMQTT

scripts/broker.test

@@ -880,6 +880,119 @@ else
 fi
 fi
 
+# --- Test 23: $-prefix wildcard guard [MQTT-4.7.2] ---
+echo ""
+echo "--- Test 23: \$-prefix wildcard guard [MQTT-4.7.2] ---"
+if [ "$skip_plain" = "yes" ]; then
+    echo "SKIP: \$-prefix wildcard guard (plain listener disabled)"
+elif [ "$has_wildcards" = "no" ]; then
+    echo "SKIP: \$-prefix wildcard guard (wildcards not built)"
+else
+start_broker
+# Subscribe to '#' (should NOT receive $SYS messages per MQTT-4.7.2)
+rm -f "${TMP_DIR}/t23_wild.ready" "${TMP_DIR}/t23_exact.ready"
+timeout 10 ./$sub_bin -T -h 127.0.0.1 -p $port -n "#" -i "sub_wild_dollar" \
+    -R "${TMP_DIR}/t23_wild.ready" >"${TMP_DIR}/t23_wild.log" 2>&1 &
+T23_WILD_PID=$!
+# Subscribe to exact '$SYS/test' (SHOULD receive the message)
+timeout 10 ./$sub_bin -T -h 127.0.0.1 -p $port -n '$SYS/test' -i "sub_exact_dollar" \
+    -R "${TMP_DIR}/t23_exact.ready" >"${TMP_DIR}/t23_exact.log" 2>&1 &
+T23_EXACT_PID=$!
+TEST_PIDS+=($T23_WILD_PID $T23_EXACT_PID)
+wait_for_file "${TMP_DIR}/t23_wild.ready" 5
+wait_for_file "${TMP_DIR}/t23_exact.ready" 5
+# Publish to $SYS/test
+./$pub_bin -T -h 127.0.0.1 -p $port -n '$SYS/test' -m "dollar_sys_msg" \
+    >"${TMP_DIR}/t23_pub.log" 2>&1
+sleep 0.3
+kill $T23_WILD_PID $T23_EXACT_PID 2>/dev/null
+wait $T23_WILD_PID 2>/dev/null || true
+wait $T23_EXACT_PID 2>/dev/null || true
+TEST_PIDS=()
+T23_WILD_GOT=no
+T23_EXACT_GOT=no
+grep -q "dollar_sys_msg" "${TMP_DIR}/t23_wild.log" 2>/dev/null && T23_WILD_GOT=yes
+grep -q "dollar_sys_msg" "${TMP_DIR}/t23_exact.log" 2>/dev/null && T23_EXACT_GOT=yes
+if [ "$T23_WILD_GOT" = "no" ] && [ "$T23_EXACT_GOT" = "yes" ]; then
+    echo "PASS: \$-prefix wildcard guard (# blocked, exact matched)"
+else
+    echo "FAIL: \$-prefix wildcard guard (wild_got=$T23_WILD_GOT, exact_got=$T23_EXACT_GOT)"
+    FAIL=1
+fi
+fi
+
+# --- Test 24: PUBLISH topic wildcard rejection [MQTT-3.3.2-2] ---
+echo ""
+echo "--- Test 24: PUBLISH topic wildcard rejection [MQTT-3.3.2-2] ---"
+if [ "$skip_plain" = "yes" ]; then
+    echo "SKIP: PUBLISH wildcard rejection (plain listener disabled)"
+elif [ "$has_wildcards" = "no" ]; then
+    echo "SKIP: PUBLISH wildcard rejection (wildcards not built)"
+else
+start_broker
+# Subscribe to a wildcard filter that would match if the broker allowed it
+rm -f "${TMP_DIR}/t24_sub.ready"
+timeout 10 ./$sub_bin -T -h 127.0.0.1 -p $port -n "test/wild/+" -i "sub_wild24" \
+    -R "${TMP_DIR}/t24_sub.ready" >"${TMP_DIR}/t24_sub.log" 2>&1 &
+T24_SUB_PID=$!
+TEST_PIDS+=($T24_SUB_PID)
+wait_for_file "${TMP_DIR}/t24_sub.ready" 5
+# Attempt to PUBLISH with '+' in the topic name (must be rejected by broker)
+./$pub_bin -T -h 127.0.0.1 -p $port -n "test/+/card" -m "bad_wildcard_plus" \
+    >"${TMP_DIR}/t24_pub_plus.log" 2>&1
+# Attempt to PUBLISH with '#' in the topic name (must be rejected by broker)
+./$pub_bin -T -h 127.0.0.1 -p $port -n "test/#" -m "bad_wildcard_hash" \
+    >"${TMP_DIR}/t24_pub_hash.log" 2>&1
+sleep 0.3
+kill $T24_SUB_PID 2>/dev/null
+wait $T24_SUB_PID 2>/dev/null || true
+TEST_PIDS=()
+# Verify subscriber did NOT receive either message
+T24_GOT_PLUS=no
+T24_GOT_HASH=no
+grep -q "bad_wildcard_plus" "${TMP_DIR}/t24_sub.log" 2>/dev/null && T24_GOT_PLUS=yes
+grep -q "bad_wildcard_hash" "${TMP_DIR}/t24_sub.log" 2>/dev/null && T24_GOT_HASH=yes
+if [ "$T24_GOT_PLUS" = "no" ] && [ "$T24_GOT_HASH" = "no" ]; then
+    echo "PASS: PUBLISH topic wildcard rejection (+ and # blocked)"
+else
+    echo "FAIL: PUBLISH wildcard rejection (got_plus=$T24_GOT_PLUS, got_hash=$T24_GOT_HASH)"
+    FAIL=1
+fi
+fi
+
+# --- Test 25: Multi-level wildcard matches parent [MQTT-4.7.1.2] ---
+echo ""
+echo "--- Test 25: Multi-level wildcard matches parent [MQTT-4.7.1.2] ---"
+if [ "$skip_plain" = "yes" ]; then
+    echo "SKIP: Multi-level wildcard parent match (plain listener disabled)"
+elif [ "$has_wildcards" = "no" ]; then
+    echo "SKIP: Multi-level wildcard parent match (wildcards not built)"
+else
+start_broker
+# Subscribe to 'sport/#' — per MQTT-4.7.1.2 this must also match 'sport'
+rm -f "${TMP_DIR}/t25_sub.ready"
+timeout 10 ./$sub_bin -T -h 127.0.0.1 -p $port -n "sport/#" -i "sub_parent25" \
+    -R "${TMP_DIR}/t25_sub.ready" >"${TMP_DIR}/t25_sub.log" 2>&1 &
+T25_SUB_PID=$!
+TEST_PIDS+=($T25_SUB_PID)
+wait_for_file "${TMP_DIR}/t25_sub.ready" 5
+# Publish to 'sport' (parent level, no trailing slash)
+./$pub_bin -T -h 127.0.0.1 -p $port -n "sport" -m "parent_match_msg" \
+    >"${TMP_DIR}/t25_pub.log" 2>&1
+sleep 0.3
+kill $T25_SUB_PID 2>/dev/null
+wait $T25_SUB_PID 2>/dev/null || true
+TEST_PIDS=()
+T25_GOT=no
+grep -q "parent_match_msg" "${TMP_DIR}/t25_sub.log" 2>/dev/null && T25_GOT=yes
+if [ "$T25_GOT" = "yes" ]; then
+    echo "PASS: Multi-level wildcard matches parent (sport/# matched sport)"
+else
+    echo "FAIL: Multi-level wildcard parent match (got=$T25_GOT)"
+    FAIL=1
+fi
+fi
+
 # --- WebSocket Tests ---
 ws_client_bin="examples/websocket/websocket_client"
 has_websocket=no

src/mqtt_broker.c

@@ -35,22 +35,7 @@
 
 #ifdef WOLFMQTT_BROKER
 
-/* Secure memory zeroing - prevents compiler dead-store elimination */
-#ifdef ENABLE_MQTT_TLS
-    #include <wolfssl/wolfcrypt/memory.h>
-    #define BROKER_FORCE_ZERO(mem, len) wc_ForceZero(mem, (word32)(len))
-#else
-    /* Local implementation matching wolfCrypt's ForceZero */
-    static void BrokerForceZero(void* mem, word32 len)
-    {
-        volatile byte* p = (volatile byte*)mem;
-        word32 i;
-        for (i = 0; i < len; i++) {
-            p[i] = 0;
-        }
-    }
-    #define BROKER_FORCE_ZERO(mem, len) BrokerForceZero(mem, (word32)(len))
-#endif
+#define BROKER_FORCE_ZERO(mem, len) Mqtt_ForceZero(mem, (word32)(len))
 
 /* -------------------------------------------------------------------------- */
 /* Platform includes                                                           */
@@ -887,6 +872,7 @@ static int callback_broker_mqtt(struct lws *wsi,
                 WOLFMQTT_FREE(ws->tx_pending);
                 ws->tx_pending = NULL;
                 ws->tx_len = 0;
+                ws->status = -1;
                 return -1;
             }
             WOLFMQTT_FREE(ws->tx_pending);
@@ -1037,6 +1023,12 @@ static int BrokerWsNetWrite(void* context, const byte* buf, int buf_len,
         return MQTT_CODE_ERROR_NETWORK;
     }
 
+    /* Check if the write callback reported an error (it frees tx_pending
+     * and sets status to -1 before returning -1 to lws) */
+    if (ws->status < 0) {
+        return MQTT_CODE_ERROR_NETWORK;
+    }
+
     return buf_len;
 }
 
@@ -2552,6 +2544,10 @@ static int BrokerTopicMatch(const char* filter, const char* topic)
             f++;
         }
         else if (*t == '/' || *f == '/') {
+            /* [MQTT-4.7.1.2] 'topic/#' must also match 'topic' itself */
+            if (*f == '/' && f[1] == '#' && f[2] == '\0' && *t == '\0') {
+                return 1;
+            }
             return 0;
         }
     }
@@ -3865,6 +3861,18 @@ int MqttBroker_Start(MqttBroker* broker)
     if (broker->auth_user || broker->auth_pass) {
         WBLOG_INFO(broker, "broker: auth enabled user=%s",
             broker->auth_user ? broker->auth_user : "(null)");
+    #ifdef ENABLE_MQTT_TLS
+    #ifndef WOLFMQTT_BROKER_NO_INSECURE
+        if (broker->use_tls &&
+            broker->port != broker->port_tls) {
+            WBLOG_ERR(broker,
+                "broker: WARNING: auth credentials exposed on plaintext "
+                "port %d. Rebuild with ./configure --disable-broker-insecure "
+                "for TLS-only",
+                broker->port);
+        }
+    #endif
+    #endif
     }
 #endif
 

src/mqtt_client.c

@@ -26,6 +26,8 @@
 
 #include "wolfmqtt/mqtt_client.h"
 
+#define CLIENT_FORCE_ZERO(mem, len) Mqtt_ForceZero(mem, (word32)(len))
+
 /* DOCUMENTED BUILD OPTIONS:
  *
  * WOLFMQTT_MULTITHREAD: Enables multi-thread support with mutex protection on
@@ -1664,6 +1666,14 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect)
     }
 
     if (mc_connect->stat.write == MQTT_MSG_BEGIN) {
+    #ifdef DEBUG_WOLFMQTT
+        /* Warn if credentials are being sent without TLS */
+        if ((mc_connect->username != NULL || mc_connect->password != NULL) &&
+            !(MqttClient_Flags(client, 0, 0) & MQTT_CLIENT_FLAG_IS_TLS)) {
+            PRINTF("Warning: MQTT credentials are being sent without TLS");
+        }
+    #endif
+
         /* Flag write active / lock mutex */
         if ((rc = MqttWriteStart(client, &mc_connect->stat)) != 0) {
             return rc;
@@ -1714,11 +1724,17 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect)
             && client->write.total > 0
         #endif
         ) {
-            /* keep send locked and return early */
+            /* keep send locked and return early.
+             * Note: tx_buf still contains credentials until write completes */
             return rc;
         }
     #endif
         MqttWriteStop(client, &mc_connect->stat);
+
+        /* Clear tx_buf to remove any plaintext credentials from memory.
+         * Use xfer (saved before MqttWriteStop zeroes client->write) */
+        CLIENT_FORCE_ZERO(client->tx_buf, xfer);
+
         if (rc != xfer) {
             MqttClient_CancelMessage(client, (MqttObject*)mc_connect);
             return rc;

src/mqtt_packet.c

@@ -244,6 +244,11 @@ int MqttDecode_Vbi(byte *buf, word32 *value, word32 buf_len)
         rc++;
     } while ((encodedByte & MQTT_PACKET_LEN_ENCODE_MASK) != 0);
 
+    /* [MQTT-1.5.5-1] Reject non-canonical overlong encodings */
+    if (rc > 1 && encodedByte == 0) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA);
+    }
+
     return (int)rc;
 }
 
@@ -1530,10 +1535,6 @@ int MqttEncode_PublishResp(byte* tx_buf, int tx_buf_len, byte type,
 #ifdef WOLFMQTT_V5
     if (publish_resp->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5)
     {
-        if (publish_resp->reason_code != MQTT_REASON_SUCCESS) {
-            /* Reason Code */
-            remain_len++;
-        }
         if (publish_resp->props != NULL) {
             /* Determine length of properties */
             props_len = MqttEncode_Props((MqttPacketType)type,
@@ -1550,6 +1551,11 @@ int MqttEncode_PublishResp(byte* tx_buf, int tx_buf_len, byte type,
             else
                 return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PROPERTY);
         }
+        if (publish_resp->reason_code != MQTT_REASON_SUCCESS ||
+            publish_resp->props != NULL) {
+            /* Reason Code */
+            remain_len++;
+        }
     }
 #endif
 
@@ -1574,7 +1580,8 @@ int MqttEncode_PublishResp(byte* tx_buf, int tx_buf_len, byte type,
 #ifdef WOLFMQTT_V5
     if (publish_resp->protocol_level >= MQTT_CONNECT_PROTOCOL_LEVEL_5)
     {
-        if (publish_resp->reason_code != MQTT_REASON_SUCCESS) {
+        if (publish_resp->reason_code != MQTT_REASON_SUCCESS ||
+            publish_resp->props != NULL) {
             /* Encode the Reason Code */
             *tx_payload++ = publish_resp->reason_code;
         }
@@ -1620,6 +1627,12 @@ int MqttDecode_PublishResp(byte* rx_buf, int rx_buf_len, byte type,
     if (header_len < 0) {
         return header_len;
     }
+
+    /* Validate remain_len (need at least packet_id) */
+    if (remain_len < MQTT_DATA_LEN_SIZE) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA);
+    }
+
     rx_payload = &rx_buf[header_len];
 
     /* Decode variable header */
@@ -1698,6 +1711,11 @@ int MqttEncode_Subscribe(byte *tx_buf, int tx_buf_len,
         return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG);
     }
 
+    /* [MQTT-2.3.1-1] SUBSCRIBE packets require a non-zero packet identifier */
+    if (subscribe->packet_id == 0) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PACKET_ID);
+    }
+
     /* Determine packet length */
     remain_len = MQTT_DATA_LEN_SIZE; /* For packet_id */
     for (i = 0; i < subscribe->topic_count; i++) {
@@ -1898,6 +1916,12 @@ int MqttDecode_SubscribeAck(byte* rx_buf, int rx_buf_len,
     if (header_len < 0) {
         return header_len;
     }
+
+    /* Validate remain_len (need at least packet_id) */
+    if (remain_len < MQTT_DATA_LEN_SIZE) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA);
+    }
+
     rx_payload = &rx_buf[header_len];
 
     /* Decode variable header */
@@ -1982,6 +2006,11 @@ int MqttEncode_Unsubscribe(byte *tx_buf, int tx_buf_len,
         return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG);
     }
 
+    /* [MQTT-2.3.1-1] UNSUBSCRIBE packets require a non-zero packet identifier */
+    if (unsubscribe->packet_id == 0) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_PACKET_ID);
+    }
+
     /* Determine packet length */
     remain_len = MQTT_DATA_LEN_SIZE; /* For packet_id */
     for (i = 0; i < unsubscribe->topic_count; i++) {
@@ -2170,6 +2199,12 @@ int MqttDecode_UnsubscribeAck(byte *rx_buf, int rx_buf_len,
     if (header_len < 0) {
         return header_len;
     }
+
+    /* Validate remain_len (need at least packet_id) */
+    if (remain_len < MQTT_DATA_LEN_SIZE) {
+        return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_MALFORMED_DATA);
+    }
+
     rx_payload = &rx_buf[header_len];
 
     /* Decode variable header */

tests/include.am

@@ -2,6 +2,13 @@
 # included from Top Level Makefile.am
 # All paths should be given relative to the root
 
+# Unit tests for packet encode/decode
+check_PROGRAMS += tests/unit_test
+tests_unit_test_SOURCES      = tests/unit_test.c
+tests_unit_test_LDADD        = src/libwolfmqtt.la
+tests_unit_test_DEPENDENCIES = src/libwolfmqtt.la
+tests_unit_test_CPPFLAGS     = $(AM_CPPFLAGS)
+
 if BUILD_FUZZ
 if BUILD_BROKER
 noinst_PROGRAMS += tests/fuzz/broker_fuzz