diff --git a/.github/workflows/broker-check.yml b/.github/workflows/broker-check.yml index b2e8ee1..c288766 100644 --- a/.github/workflows/broker-check.yml +++ b/.github/workflows/broker-check.yml @@ -37,13 +37,23 @@ jobs: - name: "Broker TLS-only (no insecure)" cflags: "" wolfmqtt_opts: "--enable-broker --enable-tls --disable-broker-insecure" + - name: "Broker with WebSocket" + cflags: "" + wolfmqtt_opts: "--enable-broker --enable-websocket" + extra_deps: "libwebsockets-dev" + wolfssl_opts: "--enable-opensslcoexist" + - name: "Broker with WebSocket + TLS" + cflags: "" + wolfmqtt_opts: "--enable-broker --enable-tls --enable-websocket" + extra_deps: "libwebsockets-dev" + wolfssl_opts: "--enable-opensslcoexist --enable-enckeys" steps: - name: Install dependencies run: | export DEBIAN_FRONTEND=noninteractive sudo apt-get update - sudo apt-get install -y mosquitto-clients + sudo apt-get install -y mosquitto-clients ${{ matrix.extra_deps }} - uses: actions/checkout@master with: @@ -54,7 +64,7 @@ jobs: run: ./autogen.sh - name: wolfssl configure working-directory: ./wolfssl - run: ./configure --enable-enckeys + run: ./configure ${{ matrix.wolfssl_opts || '--enable-enckeys' }} - name: wolfssl make working-directory: ./wolfssl run: make diff --git a/examples/websocket/net_libwebsockets.c b/examples/websocket/net_libwebsockets.c index 2f8739c..db8da06 100644 --- a/examples/websocket/net_libwebsockets.c +++ b/examples/websocket/net_libwebsockets.c @@ -33,6 +33,11 @@ #include +/* Compatibility for older libwebsockets versions (pre-4.1) */ +#ifndef LWS_PROTOCOL_LIST_TERM + #define LWS_PROTOCOL_LIST_TERM { NULL, NULL, 0, 0, 0, NULL, 0 } +#endif + /* Network context for libwebsockets */ typedef struct _LibwebsockContext { struct lws_context *context; @@ -74,37 +79,14 @@ static int callback_mqtt(struct lws *wsi, enum lws_callback_reasons reason, XMEMCPY(net->rx_buffer + net->rx_len, in, len); net->rx_len += len; } else { - /* Buffer overflow - handle error */ - lwsl_err("WebSocket receive buffer overflow" - "- dropping oldest data\n"); - - /* Simple approach: If new data is larger than buffer, - * just keep newest data */ - if (len >= sizeof(net->rx_buffer)) { - /* New data is larger than entire buffer, - * keep only what fits */ - /* Cast to byte pointer to allow pointer arithmetic */ - const byte* in_bytes = (const byte*)in; - XMEMCPY(net->rx_buffer, - &in_bytes[len - sizeof(net->rx_buffer)], - sizeof(net->rx_buffer)); - net->rx_len = sizeof(net->rx_buffer); - } else { - /* Keep as much new data as possible */ - size_t keep_bytes = sizeof(net->rx_buffer) - len; - - /* Move the portion of old data we want to - * keep to the beginning */ - if (keep_bytes > 0 && net->rx_len > 0) { - XMEMMOVE(net->rx_buffer, - net->rx_buffer + (net->rx_len - keep_bytes), - keep_bytes); - } - - /* Append all new data */ - XMEMCPY(net->rx_buffer + keep_bytes, in, len); - net->rx_len = keep_bytes + len; - } + /* Dropping bytes would desynchronize MQTT packet framing, + * so treat overflow as a fatal protocol error. */ + lwsl_err("WebSocket receive buffer overflow " + "(have=%d, need=%d, max=%d)\n", + (int)net->rx_len, (int)len, + (int)sizeof(net->rx_buffer)); + net->status = -1; + return -1; /* close connection */ } } } diff --git a/scripts/broker.test b/scripts/broker.test index 4c8510a..dc1d541 100755 --- a/scripts/broker.test +++ b/scripts/broker.test @@ -887,7 +887,7 @@ if [ "$has_websocket" = "yes" ] && [ -x ./$ws_client_bin ]; then broker_pid=$no_pid fi ./$broker_bin -p $tcp_port -w $ws_port \ - >"${TMP_DIR}/t17_broker.log" 2>&1 & + >"${TMP_DIR}/t22_broker.log" 2>&1 & broker_pid=$! check_broker $tcp_port check_broker $ws_port @@ -899,29 +899,29 @@ if [ "$has_websocket" = "yes" ] && [ -x ./$ws_client_bin ]; then # Use stdbuf to force line-buffered stdout (otherwise printf output # is fully buffered when redirected to a file and lost on kill). timeout 15 stdbuf -oL ./$ws_client_bin -h 127.0.0.1 -p $ws_port \ - >"${TMP_DIR}/t17.log" 2>&1 & - T17_WS_PID=$! - TEST_PIDS+=($T17_WS_PID) + >"${TMP_DIR}/t22.log" 2>&1 & + T22_WS_PID=$! + TEST_PIDS+=($T22_WS_PID) sleep 3 # Verify WS client connected and subscribed before publishing. # Check both client log and broker log (broker always prints via # PRINTF which is unbuffered). ws_connected=no - if grep -q "MQTT Connected" "${TMP_DIR}/t17.log" 2>/dev/null; then + if grep -q "MQTT Connected" "${TMP_DIR}/t22.log" 2>/dev/null; then ws_connected=yes - elif grep -q "SUBACK sock=-1" "${TMP_DIR}/t17_broker.log" 2>/dev/null; then + elif grep -q "SUBACK sock=-1" "${TMP_DIR}/t22_broker.log" 2>/dev/null; then ws_connected=yes fi if [ "$ws_connected" = "yes" ]; then # Publish from TCP client to the WS client's subscribed topic ./$pub_bin -T -h 127.0.0.1 -p $tcp_port -n "test/topic" -m "ws_hello" \ - >"${TMP_DIR}/t17_pub.log" 2>&1 + >"${TMP_DIR}/t22_pub.log" 2>&1 sleep 2 fi - kill $T17_WS_PID 2>/dev/null - wait $T17_WS_PID 2>/dev/null || true + kill $T22_WS_PID 2>/dev/null + wait $T22_WS_PID 2>/dev/null || true TEST_PIDS=() - if grep -q "ws_hello" "${TMP_DIR}/t17.log" 2>/dev/null; then + if grep -q "ws_hello" "${TMP_DIR}/t22.log" 2>/dev/null; then echo "PASS: WebSocket connect and subscribe (message received)" elif [ "$ws_connected" = "yes" ]; then echo "PASS: WebSocket connect and subscribe (connected ok)" diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 0106180..3c958e2 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -465,6 +465,11 @@ static void BrokerTls_Free(MqttBroker* broker) #include +/* Compatibility for older libwebsockets versions (pre-4.1) */ +#ifndef LWS_PROTOCOL_LIST_TERM + #define LWS_PROTOCOL_LIST_TERM { NULL, NULL, 0, 0, 0, NULL, 0 } +#endif + /* Forward declaration for the no-op connect callback (defined after WS section) */ static int BrokerNetConnect(void* context, const char* host, word16 port, int timeout_ms); @@ -622,25 +627,13 @@ static int callback_broker_mqtt(struct lws *wsi, ws->rx_len += len; } else { - WBLOG_ERR(broker, "broker: ws rx buffer overflow (wsi=%p)", - (void*)wsi); - /* Drop oldest data to make room */ - if (len >= sizeof(ws->rx_buffer)) { - const byte* in_bytes = (const byte*)in; - XMEMCPY(ws->rx_buffer, - &in_bytes[len - sizeof(ws->rx_buffer)], - sizeof(ws->rx_buffer)); - ws->rx_len = sizeof(ws->rx_buffer); - } - else { - size_t keep = sizeof(ws->rx_buffer) - len; - if (keep > 0 && ws->rx_len > 0) { - XMEMMOVE(ws->rx_buffer, - ws->rx_buffer + (ws->rx_len - keep), keep); - } - XMEMCPY(ws->rx_buffer + keep, in, len); - ws->rx_len = keep + len; - } + /* Dropping bytes would desynchronize MQTT packet framing, + * so treat overflow as a fatal protocol error. */ + WBLOG_ERR(broker, "broker: ws rx buffer overflow " + "(wsi=%p, have=%d, need=%d, max=%d)", + (void*)wsi, (int)ws->rx_len, (int)len, + (int)sizeof(ws->rx_buffer)); + return -1; /* close connection */ } } else if (reason == LWS_CALLBACK_SERVER_WRITEABLE) { @@ -653,14 +646,17 @@ static int callback_broker_mqtt(struct lws *wsi, if (ws->tx_pending != NULL && ws->tx_len > 0) { int n = lws_write(wsi, ws->tx_pending + LWS_PRE, ws->tx_len, LWS_WRITE_BINARY); + if (n < (int)ws->tx_len) { + WBLOG_ERR(broker, "broker: ws write failed (wsi=%p, " + "n=%d, len=%d)", (void*)wsi, n, (int)ws->tx_len); + WOLFMQTT_FREE(ws->tx_pending); + ws->tx_pending = NULL; + ws->tx_len = 0; + return -1; + } WOLFMQTT_FREE(ws->tx_pending); ws->tx_pending = NULL; ws->tx_len = 0; - if (n < 0) { - WBLOG_ERR(broker, "broker: ws write failed (wsi=%p)", - (void*)wsi); - return -1; - } } } else if (reason == LWS_CALLBACK_CLOSED) { @@ -804,7 +800,15 @@ static int BrokerWsNetDisconnect(void* context) } if (ws->wsi != NULL && ws->status > 0) { + BrokerClient **bc_ptr; WBLOG_INFO(bc->broker, "broker: ws disconnect (wsi=%p)", (void*)ws->wsi); + /* Clear lws per-session user data so that any later callbacks + * (e.g. LWS_CALLBACK_CLOSED) see NULL and skip processing. + * Without this, the callback would dereference the freed bc. */ + bc_ptr = (BrokerClient**)lws_wsi_user(ws->wsi); + if (bc_ptr != NULL) { + *bc_ptr = NULL; + } lws_close_reason(ws->wsi, LWS_CLOSE_STATUS_NORMAL, NULL, 0); ws->wsi = NULL; } diff --git a/wolfmqtt/mqtt_broker.h b/wolfmqtt/mqtt_broker.h index 218b321..24a28e6 100644 --- a/wolfmqtt/mqtt_broker.h +++ b/wolfmqtt/mqtt_broker.h @@ -167,6 +167,11 @@ typedef struct MqttBrokerNet { /* WebSocket per-client context */ /* -------------------------------------------------------------------------- */ #ifdef ENABLE_MQTT_WEBSOCKET +#ifdef WOLFMQTT_STATIC_MEMORY + #error "WebSocket support (ENABLE_MQTT_WEBSOCKET) is incompatible with " \ + "static memory mode (WOLFMQTT_STATIC_MEMORY). libwebsockets " \ + "requires dynamic allocation internally." +#endif #ifndef BROKER_WS_RX_BUF_SZ #define BROKER_WS_RX_BUF_SZ BROKER_RX_BUF_SZ #endif