From 3cc4841416ae6ba8e6e4d6893da129c2050bc5ff Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 11 Nov 2026 11:17:29 +0000 Subject: [PATCH] fix(gateway): reassemble WebChat frames before JSON parse libwebsockets spills a frame into WS_RX_SPILL chunks and reports the final fragment only when the packet is complete. Parsing each chunk dropped a WS_TEXT_MAX chat message once quotes or backslashes expanded the JSON past that spill. Co-authored-by: esadrianno --- CHANGELOG.md | 3 + Makefile | 24 ++++-- src/gateway/http.c | 4 +- src/gateway/http_lws.c | 28 +++--- src/gateway/http_lws.h | 6 +- src/gateway/ws.c | 24 ++++++ src/gateway/ws.h | 59 +++++++++++++ src/gateway/ws_rx.c | 140 ++++++++++++++++++++++++++++++ tests/test_ws_rx.c | 189 +++++++++++++++++++++++++++++++++++++++++ 9 files changed, 450 insertions(+), 27 deletions(-) create mode 100644 src/gateway/ws_rx.c create mode 100644 tests/test_ws_rx.c diff --git a/CHANGELOG.md b/CHANGELOG.md index 0252119..3f7990a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ All notable changes to ShellClaw are documented here. Format follows [Keep a Cha ## [Unreleased] +### Fixed +- WebChat reassembles libwebsockets receive spills before parsing, so a chat message whose JSON escapes grow past the 32 KiB + 64 byte callback buffer reaches the agent instead of being dropped. + ## [1.0.1] - 2026-10-18 ### Fixed diff --git a/Makefile b/Makefile index 7280bcb..a808e67 100644 --- a/Makefile +++ b/Makefile @@ -100,6 +100,7 @@ HTTP_LWS_O := $(if $(filter 1,$(GATEWAY)),src/gateway/http_lws.o,) ROUTES_O := $(if $(filter 1,$(GATEWAY)),src/gateway/routes.o,) ROUTES_HARDWARE_O := $(if $(filter 1,$(GATEWAY)),src/gateway/routes_hardware.o,) WS_O := $(if $(filter 1,$(GATEWAY)),src/gateway/ws.o,) +WS_RX_O := $(if $(filter 1,$(GATEWAY)),src/gateway/ws_rx.o,) # Tools (Task 7) SHELL_O := src/tools/shell.o WEBSEARCH_O := src/tools/web_search.o @@ -172,7 +173,7 @@ OBJS := $(CORE_OBJS) $(VENDOR_OBJS) PROVIDER_OBJS := $(PROVIDER_COMMON_O) $(STUB_O) $(ROUTER_O) $(ANTHROPIC_O) $(OPENAI_COMPAT_O) $(OPENAI_O) $(LOCAL_O) CHANNEL_OBJS := $(CHANNEL_COMMON_O) $(CHANNEL_CLI_O) $(CHANNEL_TG_O) $(CHANNEL_DISCORD_O) $(DISCORD_HELPERS_O) $(CHANNEL_HEARTBEAT_O) $(CHANNEL_WEBCHAT_O) ASAP_HTTP_BODY_O := $(if $(filter 1,$(GATEWAY)),src/gateway/asap_http_body.o,) -GATEWAY_OBJS := $(AUTH_O) $(STATIC_O) $(HTTP_O) $(HTTP_LWS_O) $(ASAP_HTTP_BODY_O) $(ROUTES_O) $(ROUTES_HARDWARE_O) $(WS_O) $(RATE_LIMIT_O) +GATEWAY_OBJS := $(AUTH_O) $(STATIC_O) $(HTTP_O) $(HTTP_LWS_O) $(ASAP_HTTP_BODY_O) $(ROUTES_O) $(ROUTES_HARDWARE_O) $(WS_O) $(WS_RX_O) $(RATE_LIMIT_O) TOOL_OBJS := $(SHELL_O) $(WEBSEARCH_O) $(FILE_O) $(REGISTRY_O) $(CONTEXT_O) $(CONTEXT_CACHE_O) $(CONTEXT_HTTP_O) $(CONTEXT_GEO_O) $(CRYPTO_LINK) $(HARDWARE_STUB_O) $(HARDWARE_INIT_O) $(HARDWARE_GPIO_SNAPSHOT_O) $(HARDWARE_TEGRASTATS_O) $(HARDWARE_TOOLS_O) $(BOARD_DETECT_O) $(HARDWARE_I2C_O) $(HARDWARE_CAMERA_O) $(HARDWARE_LIBGPIOD_O) $(CRON_O) $(ASAP_INVOKE_O) ASAP_OBJS := $(MANIFEST_O) $(MANIFEST_PROFILES_O) $(MANIFEST_BUILD_O) $(MANIFEST_SIGN_O) $(MANIFEST_KEYS_O) $(JCS_O) $(ENVELOPE_O) $(ULID_O) $(CLIENT_O) $(ASAP_REGISTRY_O) $(SERVER_O) $(ASAP_LOG_O) SANDBOX_OBJS := $(SANDBOX_O) $(ALLOWLIST_O) @@ -386,6 +387,9 @@ $(ROUTES_HARDWARE_O): src/gateway/routes_hardware.c src/gateway/routes_hardware. $(WS_O): src/gateway/ws.c src/gateway/ws.h $(CC) $(CFLAGS) $(INC) $(GATEWAY_CFLAGS) -c -o $@ src/gateway/ws.c +src/gateway/ws_rx.o: src/gateway/ws_rx.c src/gateway/ws.h vendor/cJSON/cJSON.h + $(CC) $(CFLAGS) $(INC) -c -o $@ src/gateway/ws_rx.c + $(SHELL_O): src/tools/shell.c src/tools/tool.h src/tools/shell.h src/core/config.h \ src/sandbox/sandbox.h src/sandbox/allowlist.h $(CC) $(CFLAGS) $(INC) -c -o $@ src/tools/shell.c @@ -647,9 +651,14 @@ test_cron: tests/test_cron.c $(CRON_O) $(CRYPTO_LINK) $(MEMORY_O) $(SQLITE3_O) $ $(CC) $(CFLAGS) $(LDFLAGS) $(INC) -o $(BINDIR)/$@ tests/test_cron.c $(CRON_O) $(CRYPTO_LINK) $(MEMORY_O) $(SQLITE3_O) $(CHANNEL_COMMON_O) $(CJSON_O) $(AGENT_O) $(SKILL_O) $(CONFIG_O) $(TOML_O) $(PROVIDER_COMMON_O) $(LDLIBS) -pthread $(DSYM_SCRIPT) -test_ws: tests/test_ws.c $(WS_TEST_O) +test_ws: tests/test_ws.c $(WS_TEST_O) src/gateway/ws_rx.o $(CJSON_O) + @mkdir -p $(BINDIR) + $(CC) $(CFLAGS) $(LDFLAGS) $(INC) -o $(BINDIR)/$@ tests/test_ws.c $(WS_TEST_O) src/gateway/ws_rx.o $(CJSON_O) $(LDLIBS) -pthread + $(DSYM_SCRIPT) + +test_ws_rx: tests/test_ws_rx.c $(WS_TEST_O) src/gateway/ws_rx.o $(CJSON_O) @mkdir -p $(BINDIR) - $(CC) $(CFLAGS) $(LDFLAGS) $(INC) -o $(BINDIR)/$@ tests/test_ws.c $(WS_TEST_O) $(LDLIBS) -pthread + $(CC) $(CFLAGS) $(LDFLAGS) $(INC) -o $(BINDIR)/$@ tests/test_ws_rx.c $(WS_TEST_O) src/gateway/ws_rx.o $(CJSON_O) $(LDLIBS) -pthread $(DSYM_SCRIPT) test_context: tests/test_context.c $(CONTEXT_TEST_OBJS) $(CONFIG_O) $(TOML_O) $(CJSON_O) @@ -865,7 +874,7 @@ static: --suppress=variableScope:src/vendor/tweetnacl/tweetnacl.c \ -q src/ -test: test_config test_config_patch test_memory test_skill test_provider test_anthropic test_openai test_local_provider test_router test_heartbeat test_agent test_reload test_channel test_cli test_shell test_file test_telegram test_discord_helpers test_web_search test_cron test_context test_dispatch test_crypto test_hardware_stub test_board_detect test_hardware_libgpiod test_hardware_i2c test_hardware_camera test_pin_tables test_hardware_init test_hardware_gpio_snapshot test_hardware_tegrastats test_hardware_tools test_registry test_ws test_manifest $(ASAP_UNIT_TESTS) test_sandbox test_allowlist test_rate_limit test_daemon_smoke test_bootstrap_keys test_update_script test_install_script test_download_model test_web_dashboard test_routes_hardware +test: test_config test_config_patch test_memory test_skill test_provider test_anthropic test_openai test_local_provider test_router test_heartbeat test_agent test_reload test_channel test_cli test_shell test_file test_telegram test_discord_helpers test_web_search test_cron test_context test_dispatch test_crypto test_hardware_stub test_board_detect test_hardware_libgpiod test_hardware_i2c test_hardware_camera test_pin_tables test_hardware_init test_hardware_gpio_snapshot test_hardware_tegrastats test_hardware_tools test_registry test_ws test_ws_rx test_manifest $(ASAP_UNIT_TESTS) test_sandbox test_allowlist test_rate_limit test_daemon_smoke test_bootstrap_keys test_update_script test_install_script test_download_model test_web_dashboard test_routes_hardware $(BINDIR)/test_config $(BINDIR)/test_config_patch $(BINDIR)/test_memory @@ -901,6 +910,7 @@ test: test_config test_config_patch test_memory test_skill test_provider test_an $(BINDIR)/test_hardware_tools $(BINDIR)/test_registry $(BINDIR)/test_ws + $(BINDIR)/test_ws_rx @for t in $(ASAP_UNIT_TESTS); do $(BINDIR)/$$t || exit 1; done $(BINDIR)/test_sandbox $(BINDIR)/test_allowlist @@ -922,7 +932,7 @@ COVERAGE_DIR := build/coverage COVERAGE_MIN := 80 coverage: clean - $(MAKE) BUILD=coverage GATEWAY=0 test_config test_config_patch test_memory test_skill test_provider test_anthropic test_openai test_local_provider test_router test_heartbeat test_agent test_reload test_channel test_cli test_shell test_file test_telegram test_discord_helpers test_web_search test_cron test_context test_dispatch test_crypto test_hardware_stub test_board_detect test_hardware_libgpiod test_hardware_i2c test_hardware_camera test_pin_tables test_hardware_init test_hardware_gpio_snapshot test_hardware_tegrastats test_hardware_tools test_registry test_ws test_manifest_build test_manifest_keys test_jcs $(ASAP_UNIT_TESTS) test_sandbox test_allowlist test_rate_limit test_auth + $(MAKE) BUILD=coverage GATEWAY=0 test_config test_config_patch test_memory test_skill test_provider test_anthropic test_openai test_local_provider test_router test_heartbeat test_agent test_reload test_channel test_cli test_shell test_file test_telegram test_discord_helpers test_web_search test_cron test_context test_dispatch test_crypto test_hardware_stub test_board_detect test_hardware_libgpiod test_hardware_i2c test_hardware_camera test_pin_tables test_hardware_init test_hardware_gpio_snapshot test_hardware_tegrastats test_hardware_tools test_registry test_ws test_ws_rx test_manifest_build test_manifest_keys test_jcs $(ASAP_UNIT_TESTS) test_sandbox test_allowlist test_rate_limit test_auth @if [ "$(GATEWAY)" = "1" ]; then $(MAKE) BUILD=coverage GATEWAY=1 shellclaw test_gateway_http test_static; fi @chmod +x scripts/coverage.sh @@ -934,9 +944,9 @@ clean-root-dsym: @rm -f shellclaw test_agent test_anthropic test_channel test_cli test_config test_file test_memory test_local_provider test_openai test_provider test_router test_shell test_skill test_telegram test_web_search test_ws clean: clean-root-dsym - rm -f $(OBJS) $(PROVIDER_COMMON_O) $(STUB_O) $(ANTHROPIC_O) $(OPENAI_COMPAT_O) $(OPENAI_O) $(LOCAL_O) $(ROUTER_O) $(CJSON_O) $(TWEETNACL_O) $(ANTHROPIC_TEST_O) $(OPENAI_TEST_O) $(LOCAL_TEST_O) $(CONTEXT_TEST_OBJS) $(HEARTBEAT_TEST_O) $(CHANNEL_TG_TEST_O) $(CHANNEL_COMMON_O) $(CHANNEL_STUB_O) $(CHANNEL_CLI_O) $(CHANNEL_TG_O) $(CHANNEL_DISCORD_O) $(DISCORD_HELPERS_O) $(CHANNEL_HEARTBEAT_O) $(CHANNEL_WEBCHAT_O) $(AUTH_O) $(STATIC_O) $(HTTP_O) $(HTTP_LWS_O) $(ASAP_HTTP_BODY_O) $(ROUTES_O) $(ROUTES_HARDWARE_O) $(WS_O) $(MANIFEST_O) $(MANIFEST_PROFILES_O) $(MANIFEST_BUILD_O) $(MANIFEST_SIGN_O) $(MANIFEST_KEYS_O) $(ENVELOPE_O) $(ULID_O) $(CLIENT_O) $(ASAP_REGISTRY_O) $(SERVER_O) $(ASAP_LOG_O) $(RATE_LIMIT_O) $(SHELL_O) $(WEBSEARCH_O) $(FILE_O) $(REGISTRY_O) $(CONTEXT_O) $(CONTEXT_CACHE_O) $(CONTEXT_HTTP_O) $(CONTEXT_GEO_O) $(CRYPTO_O) $(JCS_O) $(HARDWARE_STUB_O) $(HARDWARE_INIT_O) $(HARDWARE_GPIO_SNAPSHOT_O) $(HARDWARE_TEGRASTATS_O) $(HARDWARE_TOOLS_O) $(BOARD_DETECT_O) src/hardware/hardware_libgpiod.o $(HARDWARE_I2C_O) $(HARDWARE_CAMERA_O) $(CRON_O) $(ASAP_INVOKE_O) $(SANDBOX_CORE_O) $(SANDBOX_LANDLOCK_O) $(SANDBOX_ENV_O) $(ALLOWLIST_SCAN_O) $(ALLOWLIST_PATH_O) + rm -f $(OBJS) $(PROVIDER_COMMON_O) $(STUB_O) $(ANTHROPIC_O) $(OPENAI_COMPAT_O) $(OPENAI_O) $(LOCAL_O) $(ROUTER_O) $(CJSON_O) $(TWEETNACL_O) $(ANTHROPIC_TEST_O) $(OPENAI_TEST_O) $(LOCAL_TEST_O) $(CONTEXT_TEST_OBJS) $(HEARTBEAT_TEST_O) $(CHANNEL_TG_TEST_O) $(CHANNEL_COMMON_O) $(CHANNEL_STUB_O) $(CHANNEL_CLI_O) $(CHANNEL_TG_O) $(CHANNEL_DISCORD_O) $(DISCORD_HELPERS_O) $(CHANNEL_HEARTBEAT_O) $(CHANNEL_WEBCHAT_O) $(AUTH_O) $(STATIC_O) $(HTTP_O) $(HTTP_LWS_O) $(ASAP_HTTP_BODY_O) $(ROUTES_O) $(ROUTES_HARDWARE_O) $(WS_O) src/gateway/ws_rx.o $(MANIFEST_O) $(MANIFEST_PROFILES_O) $(MANIFEST_BUILD_O) $(MANIFEST_SIGN_O) $(MANIFEST_KEYS_O) $(ENVELOPE_O) $(ULID_O) $(CLIENT_O) $(ASAP_REGISTRY_O) $(SERVER_O) $(ASAP_LOG_O) $(RATE_LIMIT_O) $(SHELL_O) $(WEBSEARCH_O) $(FILE_O) $(REGISTRY_O) $(CONTEXT_O) $(CONTEXT_CACHE_O) $(CONTEXT_HTTP_O) $(CONTEXT_GEO_O) $(CRYPTO_O) $(JCS_O) $(HARDWARE_STUB_O) $(HARDWARE_INIT_O) $(HARDWARE_GPIO_SNAPSHOT_O) $(HARDWARE_TEGRASTATS_O) $(HARDWARE_TOOLS_O) $(BOARD_DETECT_O) src/hardware/hardware_libgpiod.o $(HARDWARE_I2C_O) $(HARDWARE_CAMERA_O) $(CRON_O) $(ASAP_INVOKE_O) $(SANDBOX_CORE_O) $(SANDBOX_LANDLOCK_O) $(SANDBOX_ENV_O) $(ALLOWLIST_SCAN_O) $(ALLOWLIST_PATH_O) rm -f src/gateway/ui_assets.h find . -name '*.gcno' -o -name '*.gcda' -o -name '*.gcov' | xargs rm -f 2>/dev/null || true - rm -f $(WS_TEST_O) $(SANDBOX_TEST_O) $(BINDIR)/asap_registry_test.o $(BINDIR)/asap_invoke_test.o $(CONTEXT_TEST_OBJS) $(HEARTBEAT_TEST_O) $(BINDIR)/shellclaw $(BINDIR)/test_tweetnacl_smoke $(BINDIR)/test_config $(BINDIR)/test_config_patch $(BINDIR)/test_memory $(BINDIR)/test_skill $(BINDIR)/test_provider $(BINDIR)/test_anthropic $(BINDIR)/test_openai $(BINDIR)/test_local_provider $(BINDIR)/test_router $(BINDIR)/test_heartbeat $(BINDIR)/test_crypto $(BINDIR)/test_hardware_stub $(BINDIR)/test_board_detect $(BINDIR)/test_hardware_libgpiod $(BINDIR)/test_hardware_i2c $(BINDIR)/test_hardware_camera $(BINDIR)/test_pin_tables $(BINDIR)/test_hardware_init $(BINDIR)/test_hardware_tools $(BINDIR)/test_registry $(BINDIR)/test_ws $(BINDIR)/test_agent $(BINDIR)/test_channel $(BINDIR)/test_cli $(BINDIR)/test_shell $(BINDIR)/test_file $(BINDIR)/test_telegram $(BINDIR)/test_discord_helpers $(BINDIR)/test_web_search $(BINDIR)/test_cron $(BINDIR)/test_context $(BINDIR)/test_manifest_build $(BINDIR)/test_manifest_keys $(BINDIR)/test_jcs $(BINDIR)/test_asap_envelope $(BINDIR)/test_asap_ulid $(BINDIR)/test_asap_client $(BINDIR)/test_asap_registry $(BINDIR)/test_asap_server $(BINDIR)/test_asap_invoke $(BINDIR)/test_asap_log $(BINDIR)/test_auth $(BINDIR)/test_gateway_http $(BINDIR)/test_gateway_db_lock $(BINDIR)/test_static $(BINDIR)/test_sandbox $(BINDIR)/test_allowlist $(BINDIR)/test_rate_limit + rm -f $(WS_TEST_O) $(SANDBOX_TEST_O) $(BINDIR)/asap_registry_test.o $(BINDIR)/asap_invoke_test.o $(CONTEXT_TEST_OBJS) $(HEARTBEAT_TEST_O) $(BINDIR)/shellclaw $(BINDIR)/test_tweetnacl_smoke $(BINDIR)/test_config $(BINDIR)/test_config_patch $(BINDIR)/test_memory $(BINDIR)/test_skill $(BINDIR)/test_provider $(BINDIR)/test_anthropic $(BINDIR)/test_openai $(BINDIR)/test_local_provider $(BINDIR)/test_router $(BINDIR)/test_heartbeat $(BINDIR)/test_crypto $(BINDIR)/test_hardware_stub $(BINDIR)/test_board_detect $(BINDIR)/test_hardware_libgpiod $(BINDIR)/test_hardware_i2c $(BINDIR)/test_hardware_camera $(BINDIR)/test_pin_tables $(BINDIR)/test_hardware_init $(BINDIR)/test_hardware_tools $(BINDIR)/test_registry $(BINDIR)/test_ws $(BINDIR)/test_ws_rx $(BINDIR)/test_agent $(BINDIR)/test_channel $(BINDIR)/test_cli $(BINDIR)/test_shell $(BINDIR)/test_file $(BINDIR)/test_telegram $(BINDIR)/test_discord_helpers $(BINDIR)/test_web_search $(BINDIR)/test_cron $(BINDIR)/test_context $(BINDIR)/test_manifest_build $(BINDIR)/test_manifest_keys $(BINDIR)/test_jcs $(BINDIR)/test_asap_envelope $(BINDIR)/test_asap_ulid $(BINDIR)/test_asap_client $(BINDIR)/test_asap_registry $(BINDIR)/test_asap_server $(BINDIR)/test_asap_invoke $(BINDIR)/test_asap_log $(BINDIR)/test_auth $(BINDIR)/test_gateway_http $(BINDIR)/test_gateway_db_lock $(BINDIR)/test_static $(BINDIR)/test_sandbox $(BINDIR)/test_allowlist $(BINDIR)/test_rate_limit rm -rf $(BINDIR)/*.dSYM $(DSYMDIR) rm -f $(BOOTSTRAP_DISPATCH_STUB_O) $(TOOL_RELOAD_STUB_O) $(RELOAD_CHANNEL_STUB_O) $(HTTP_RELOAD_STUB_O) diff --git a/src/gateway/http.c b/src/gateway/http.c index 976a30c..26aaf41 100644 --- a/src/gateway/http.c +++ b/src/gateway/http.c @@ -42,8 +42,10 @@ static const struct lws_protocols protocols[] = { { .name = NULL }, }; +_Static_assert(WS_RX_BUFFER_SIZE == WS_RX_SPILL, + "WebChat rx_buffer_size must match the reassembly spill"); _Static_assert(WS_RX_BUFFER_SIZE >= (size_t)WS_TEXT_MAX + 32, - "WebChat rx_buffer_size must fit JSON envelope around WS_TEXT_MAX"); + "WebChat rx_buffer_size must fit an unescaped JSON envelope"); static const struct lws_http_mount mount_http = { .mountpoint = "/", diff --git a/src/gateway/http_lws.c b/src/gateway/http_lws.c index 9794007..2b099f7 100644 --- a/src/gateway/http_lws.c +++ b/src/gateway/http_lws.c @@ -11,7 +11,6 @@ #include "gateway/uri_match.h" #include "gateway/static.h" #include "gateway/ws.h" -#include "cJSON.h" #include #include #include @@ -463,23 +462,16 @@ int ws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, } case LWS_CALLBACK_RECEIVE: { int conn_id = (int)(intptr_t)lws_wsi_user(wsi); - if (conn_id <= 0) break; - if (len > 0 && in) { - char *buf = malloc(len + 1); - if (buf) { - memcpy(buf, in, len); - buf[len] = '\0'; - cJSON *root = cJSON_Parse(buf); - free(buf); - if (root) { - cJSON *type = cJSON_GetObjectItem(root, "type"); - cJSON *text = cJSON_GetObjectItem(root, "text"); - if (cJSON_IsString(type) && strcmp(type->valuestring, "message") == 0 && - cJSON_IsString(text) && text->valuestring) - ws_push_incoming(conn_id, text->valuestring); - cJSON_Delete(root); - } - } + char *text = NULL; + int fed; + /* user_space holds the conn id (lws_set_wsi_user), so the + * partial frame lives on the connection table instead. */ + if (conn_id <= 0) + break; + fed = ws_conn_rx_feed(conn_id, in, len, lws_is_final_fragment(wsi), &text); + if (fed == 1 && text) { + ws_push_incoming(conn_id, text); + free(text); } break; } diff --git a/src/gateway/http_lws.h b/src/gateway/http_lws.h index 45954b3..9b7b900 100644 --- a/src/gateway/http_lws.h +++ b/src/gateway/http_lws.h @@ -25,7 +25,11 @@ extern "C" { #define CONFIG_BODY_MAX 65536 #define BODY_BUF_SIZE CONFIG_BODY_MAX #define ASAP_BODY_MAX (1024 * 1024) -/** Per-callback WebChat RX. Must fit `{"type":"message","text":...}` at WS_TEXT_MAX. */ +/** + * Per-callback WebChat RX spill (WS_RX_SPILL). + * A legal WS_TEXT_MAX body of quotes or backslashes is larger than this + * on the wire. ws_conn_rx_feed() reassembles until the final fragment. + */ #define WS_RX_BUFFER_SIZE (WS_TEXT_MAX + 64) enum http_method { diff --git a/src/gateway/ws.c b/src/gateway/ws.c index 11b68a0..3a7a16b 100644 --- a/src/gateway/ws.c +++ b/src/gateway/ws.c @@ -66,6 +66,7 @@ typedef struct ws_msg { typedef struct { ws_conn_t wsi; int conn_id; + ws_rx_acc_t rx; } ws_conn_entry_t; static ws_conn_entry_t g_conns[MAX_CONNECTIONS]; @@ -138,6 +139,7 @@ int ws_register_conn(int conn_id, ws_conn_t wsi) pthread_mutex_lock(&g_mutex); for (int i = 0; i < MAX_CONNECTIONS; i++) { if (g_conns[i].wsi == NULL) { + ws_rx_acc_free(&g_conns[i].rx); g_conns[i].wsi = wsi; g_conns[i].conn_id = conn_id; registered = 1; @@ -227,6 +229,7 @@ void ws_unregister_conn(int conn_id) pthread_mutex_lock(&g_mutex); for (i = 0; i < MAX_CONNECTIONS; i++) { if (g_conns[i].wsi && g_conns[i].conn_id == conn_id) { + ws_rx_acc_free(&g_conns[i].rx); g_conns[i].wsi = NULL; g_conns[i].conn_id = 0; break; @@ -442,6 +445,27 @@ void ws_cleanup(void) g_outgoing_head = NULL; g_outgoing_tail = NULL; g_outgoing_count = 0; + for (int i = 0; i < MAX_CONNECTIONS; i++) + ws_rx_acc_free(&g_conns[i].rx); memset(g_conns, 0, sizeof(g_conns)); pthread_mutex_unlock(&g_mutex); } + +int ws_conn_rx_feed(int conn_id, const void *in, size_t len, int is_final, + char **text_out) +{ + int rc = 0; + + if (!text_out) + return -1; + *text_out = NULL; + pthread_mutex_lock(&g_mutex); + for (int i = 0; i < MAX_CONNECTIONS; i++) { + if (!g_conns[i].wsi || g_conns[i].conn_id != conn_id) + continue; + rc = ws_rx_acc_feed(&g_conns[i].rx, in, len, is_final, text_out); + break; + } + pthread_mutex_unlock(&g_mutex); + return rc; +} diff --git a/src/gateway/ws.h b/src/gateway/ws.h index 0d0fcba..021f912 100644 --- a/src/gateway/ws.h +++ b/src/gateway/ws.h @@ -16,6 +16,19 @@ extern "C" { #define WS_TEXT_MAX (32 * 1024) /** C-string dest bytes for a WS_TEXT_MAX payload, including the NUL. */ #define WS_TEXT_BUF_SIZE (WS_TEXT_MAX + 1) +/** + * One libwebsockets receive spill for the ws protocol. + * Not a whole-message cap: JSON escaping can outgrow this. + */ +#define WS_RX_SPILL ((size_t)WS_TEXT_MAX + 64u) +/** + * Largest inbound WebChat JSON object to reassemble. + * JSON.stringify expands one input byte to at most six (`\u00XX`), + * and the `{"type":"message","text":...}` envelope fits in the slack. + * + * Example: a WS_TEXT_MAX string of backslashes is accepted; one byte more is not. + */ +#define WS_RX_JSON_MAX ((size_t)WS_TEXT_MAX * 6u + 64u) /** * True when @p len_out bytes fit in a WS_TEXT_MAX on-wire text frame. @@ -113,6 +126,52 @@ int ws_has_pending_outgoing(int conn_id); */ void ws_on_service_cancel(void); +/** + * In-progress WebChat frame. libwebsockets spills one websocket message + * across several LWS_CALLBACK_RECEIVE calls. The bytes live here until + * lws_is_final_fragment() is true. Zero the struct before the first feed + * and call ws_rx_acc_free() when the connection goes away. + */ +typedef struct ws_rx_acc { + char *buf; + size_t len; + size_t cap; + int skip; +} ws_rx_acc_t; + +/** Drop any partial frame held in @p acc. Safe on a zeroed struct. */ +void ws_rx_acc_free(ws_rx_acc_t *acc); + +/** + * Append one RX chunk. On the final chunk, a complete `type:message` + * whose text fits in WS_TEXT_MAX is returned in @p text_out (malloc'd). + * + * @param acc Accumulator for this connection. + * @param in Chunk bytes. May be NULL only when @p len is 0. + * @param len Chunk length. + * @param is_final Non-zero when this chunk ends the websocket message. + * @param text_out Receives the message text, or NULL when more chunks + * are required or the message was dropped. + * @return 1 when @p text_out is set, 0 when nothing is ready, -1 on OOM. + * + * Example: feed each LWS_CALLBACK_RECEIVE until the return value is 1. + */ +int ws_rx_acc_feed(ws_rx_acc_t *acc, const void *in, size_t len, int is_final, + char **text_out); + +/** + * Same as ws_rx_acc_feed() for a registered `conn_id`. + * The accumulator is tracked with the connection, because lws_set_wsi_user() + * stores the conn id in the only per-wsi slot libwebsockets gives us. + * + * @return 1 when @p text_out is set, 0 when nothing is ready or @p conn_id + * is not registered, -1 on OOM. + * + * Example: `ws_conn_rx_feed(conn_id, in, len, lws_is_final_fragment(wsi), &text);` + */ +int ws_conn_rx_feed(int conn_id, const void *in, size_t len, int is_final, + char **text_out); + /** * Get next connection ID for new connections. * diff --git a/src/gateway/ws_rx.c b/src/gateway/ws_rx.c new file mode 100644 index 0000000..9afe79e --- /dev/null +++ b/src/gateway/ws_rx.c @@ -0,0 +1,140 @@ +/** + * @file ws_rx.c + * @brief Reassemble one WebChat websocket message before JSON parse. + * + * libwebsockets spills a frame into rx_buffer_size chunks (WS_RX_SPILL) + * and only then reports the final fragment. Parsing each chunk drops a + * message whose escaped JSON is larger than one spill. + */ +#include "gateway/ws.h" +#include "cJSON.h" +#include +#include +#include + +void ws_rx_acc_free(ws_rx_acc_t *acc) +{ + if (!acc) + return; + free(acc->buf); + acc->buf = NULL; + acc->len = 0; + acc->cap = 0; + acc->skip = 0; +} + +static int ws_rx_too_big(const ws_rx_acc_t *acc, size_t chunk_len) +{ + if (chunk_len == 0) + return 0; + if (acc->len > WS_RX_JSON_MAX) + return 1; + return chunk_len > WS_RX_JSON_MAX - acc->len; +} + +static int ws_rx_append(ws_rx_acc_t *acc, const void *in, size_t chunk_len) +{ + size_t need; + char *grown; + size_t ncap; + + if (chunk_len == 0) + return 0; + if (!in) + return -1; + need = acc->len + chunk_len + 1; + if (acc->cap < need) { + ncap = acc->cap ? acc->cap * 2 : 4096; + while (ncap < need) + ncap *= 2; + if (ncap > WS_RX_JSON_MAX + 1) + ncap = WS_RX_JSON_MAX + 1; + if (ncap < need) + return -1; + grown = realloc(acc->buf, ncap); + if (!grown) + return -1; + acc->buf = grown; + acc->cap = ncap; + } + memcpy(acc->buf + acc->len, in, chunk_len); + acc->len += chunk_len; + acc->buf[acc->len] = '\0'; + return 0; +} + +static int ws_rx_copy_message_text(const cJSON *root, char **text_out) +{ + const cJSON *type_item; + const cJSON *text_item; + size_t n; + char *copy; + + type_item = cJSON_GetObjectItem(root, "type"); + text_item = cJSON_GetObjectItem(root, "text"); + if (!cJSON_IsString(type_item) || !type_item->valuestring || + strcmp(type_item->valuestring, "message") != 0 || + !cJSON_IsString(text_item) || !text_item->valuestring) + return 0; + n = strlen(text_item->valuestring); + if (n == 0 || n > (size_t)WS_TEXT_MAX) + return 0; + copy = malloc(n + 1); + if (!copy) + return -1; + memcpy(copy, text_item->valuestring, n + 1); + *text_out = copy; + return 1; +} + +static int ws_rx_take_text(const ws_rx_acc_t *acc, char **text_out) +{ + cJSON *root; + int rc; + + if (!acc->buf || acc->len == 0) + return 0; + root = cJSON_Parse(acc->buf); + if (!root) + return 0; + rc = ws_rx_copy_message_text(root, text_out); + cJSON_Delete(root); + return rc; +} + +static void ws_rx_drop_oversize(ws_rx_acc_t *acc, int is_final) +{ + ws_rx_acc_free(acc); + acc->skip = is_final ? 0 : 1; + fprintf(stderr, "shellclaw: ws: drop inbound message over %zu bytes\n", + WS_RX_JSON_MAX); +} + +int ws_rx_acc_feed(ws_rx_acc_t *acc, const void *in, size_t len, int is_final, + char **text_out) +{ + int rc; + + if (!acc || !text_out) + return -1; + *text_out = NULL; + if (acc->skip) { + if (is_final) + acc->skip = 0; + return 0; + } + if (ws_rx_too_big(acc, len)) { + ws_rx_drop_oversize(acc, is_final); + return 0; + } + if (ws_rx_append(acc, in, len) != 0) { + ws_rx_acc_free(acc); + acc->skip = is_final ? 0 : 1; + return -1; + } + if (!is_final) + return 0; + rc = ws_rx_take_text(acc, text_out); + ws_rx_acc_free(acc); + return rc; +} diff --git a/tests/test_ws_rx.c b/tests/test_ws_rx.c new file mode 100644 index 0000000..00c7bed --- /dev/null +++ b/tests/test_ws_rx.c @@ -0,0 +1,189 @@ +/** + * @file test_ws_rx.c + * @brief WebChat RX reassembly across libwebsockets spills (no libwebsockets). + */ +#define _POSIX_C_SOURCE 200809L + +#include "gateway/ws.h" +#include +#include +#include +#include + +#define ASSERT(c) \ + do { \ + if (!(c)) { \ + fprintf(stderr, "FAIL: %s:%d %s\n", __FILE__, __LINE__, #c); \ + return 1; \ + } \ + } while (0) +#define RUN(t) \ + do { \ + int _r = (t); \ + if (_r) \ + return _r; \ + } while (0) + +static char *json_message(const char *text, size_t *wire_len) +{ + const char *prefix = "{\"type\":\"message\",\"text\":\""; + size_t n = strlen(text); + size_t i; + size_t j; + char *wire = malloc(n * 2 + 64); + + if (!wire) + return NULL; + j = strlen(prefix); + memcpy(wire, prefix, j); + for (i = 0; i < n; i++) { + if (text[i] == '\\' || text[i] == '"') + wire[j++] = '\\'; + wire[j++] = text[i]; + } + wire[j++] = '"'; + wire[j++] = '}'; + wire[j] = '\0'; + *wire_len = j; + return wire; +} + +static int feed_spills(ws_rx_acc_t *acc, const char *wire, size_t wire_len, char **got) +{ + size_t off = 0; + int rc = 0; + + *got = NULL; + while (off < wire_len) { + size_t n = wire_len - off; + int is_final; + char *part = NULL; + + if (n > WS_RX_SPILL) + n = WS_RX_SPILL; + is_final = off + n == wire_len; + rc = ws_rx_acc_feed(acc, wire + off, n, is_final, &part); + off += n; + if (!is_final) { + if (rc != 0 || part != NULL) + return -1; + continue; + } + *got = part; + } + return rc; +} + +/** A max-length body of backslashes expands past one RX spill. */ +static int test_escaped_max_rejoins_across_spills(void) +{ + ws_rx_acc_t acc; + char *text; + char *wire; + char *got = NULL; + size_t wire_len = 0; + int rc; + + text = malloc((size_t)WS_TEXT_MAX + 1); + ASSERT(text != NULL); + memset(text, '\\', (size_t)WS_TEXT_MAX); + text[WS_TEXT_MAX] = '\0'; + wire = json_message(text, &wire_len); + ASSERT(wire != NULL); + ASSERT(wire_len > WS_RX_SPILL); + memset(&acc, 0, sizeof(acc)); + rc = feed_spills(&acc, wire, wire_len, &got); + ASSERT(rc == 1); + ASSERT(got != NULL); + ASSERT(strcmp(got, text) == 0); + ASSERT(acc.buf == NULL); + free(got); + free(wire); + free(text); + ws_rx_acc_free(&acc); + return 0; +} + +static int test_oversize_frame_does_not_poison_next(void) +{ + ws_rx_acc_t acc; + char *huge; + char *got = NULL; + const char *ok = "{\"type\":\"message\",\"text\":\"hi\"}"; + + memset(&acc, 0, sizeof(acc)); + huge = malloc(WS_RX_JSON_MAX + 1); + ASSERT(huge != NULL); + memset(huge, '{', WS_RX_JSON_MAX + 1); + ASSERT(ws_rx_acc_feed(&acc, huge, WS_RX_JSON_MAX + 1, 0, &got) == 0); + ASSERT(got == NULL); + ASSERT(ws_rx_acc_feed(&acc, "rest", 4, 1, &got) == 0); + ASSERT(got == NULL); + ASSERT(ws_rx_acc_feed(&acc, ok, strlen(ok), 1, &got) == 1); + ASSERT(got != NULL && strcmp(got, "hi") == 0); + free(got); + free(huge); + ws_rx_acc_free(&acc); + return 0; +} + +static int test_text_past_ws_text_max_is_dropped(void) +{ + ws_rx_acc_t acc; + char *text; + char *wire; + char *got = NULL; + size_t wire_len = 0; + + text = malloc((size_t)WS_TEXT_MAX + 2); + ASSERT(text != NULL); + memset(text, 'a', (size_t)WS_TEXT_MAX + 1); + text[WS_TEXT_MAX + 1] = '\0'; + wire = json_message(text, &wire_len); + ASSERT(wire != NULL); + memset(&acc, 0, sizeof(acc)); + ASSERT(feed_spills(&acc, wire, wire_len, &got) == 0); + ASSERT(got == NULL); + ASSERT(acc.buf == NULL); + free(wire); + free(text); + ws_rx_acc_free(&acc); + return 0; +} + +static int test_conn_feed_joins_and_unregister_drops_partial(void) +{ + char *got = NULL; + const char *head = "{\"type\":\"message\",\"te"; + const char *tail = "xt\":\"ping\"}"; + const char *next = "{\"type\":\"message\",\"text\":\"next\"}"; + + ws_cleanup(); + ASSERT(ws_register_conn(4, (ws_conn_t)(intptr_t)4) == 0); + ASSERT(ws_conn_rx_feed(4, head, strlen(head), 0, &got) == 0); + ASSERT(got == NULL); + ASSERT(ws_conn_rx_feed(4, tail, strlen(tail), 1, &got) == 1); + ASSERT(got != NULL && strcmp(got, "ping") == 0); + free(got); + got = NULL; + ASSERT(ws_conn_rx_feed(4, head, strlen(head), 0, &got) == 0); + ws_unregister_conn(4); + ASSERT(ws_conn_rx_feed(4, tail, strlen(tail), 1, &got) == 0); + ASSERT(got == NULL); + ASSERT(ws_register_conn(4, (ws_conn_t)(intptr_t)4) == 0); + ASSERT(ws_conn_rx_feed(4, next, strlen(next), 1, &got) == 1); + ASSERT(got != NULL && strcmp(got, "next") == 0); + free(got); + ws_cleanup(); + return 0; +} + +int main(void) +{ + RUN(test_escaped_max_rejoins_across_spills()); + RUN(test_oversize_frame_does_not_poison_next()); + RUN(test_text_past_ws_text_max_is_dropped()); + RUN(test_conn_feed_joins_and_unregister_drops_partial()); + printf("test_ws_rx: all tests passed\n"); + return 0; +}