From: Stefan Eissing Date: Tue, 28 Jul 2026 12:27:13 +0000 (+0200) Subject: websocket: pause writing and meta data fix X-Git-Url: http://git.ipfire.org/gitweb/index.cgi?a=commitdiff_plain;h=3974491c97cbff6ca39dd249c02b24e30171f265;p=thirdparty%2Fcurl.git websocket: pause writing and meta data fix When writing a decoded chunk of websocket data, always flush the writer chain so that buffered data gets delivered before the ws meta data gets updated. Add client writer flags CURL_CW_FLAG_BLOWUP for writer types that may significantly enlarge write sizes. This flag causes the pause writer to be added and shrinks the write chunk sizes. We do not want that for content decoders like WS that do not change the size. Add test_20_13 to check that large frames are paused/unpaused correctly with the matching meta data. Fixes #22413 Reported-by: Hendrik Hübner Closes #22416 --- diff --git a/lib/content_encoding.c b/lib/content_encoding.c index a3724d319d..e238507836 100644 --- a/lib/content_encoding.c +++ b/lib/content_encoding.c @@ -297,6 +297,7 @@ static void deflate_do_close(struct Curl_easy *data, static const struct Curl_cwtype deflate_encoding = { "deflate", NULL, + CURL_CW_FLAG_BLOWUP, deflate_do_init, deflate_do_write, Curl_cwriter_def_flush, @@ -359,6 +360,7 @@ static void gzip_do_close(struct Curl_easy *data, static const struct Curl_cwtype gzip_encoding = { "gzip", "x-gzip", + CURL_CW_FLAG_BLOWUP, gzip_do_init, gzip_do_write, Curl_cwriter_def_flush, @@ -493,6 +495,7 @@ static void brotli_do_close(struct Curl_easy *data, static const struct Curl_cwtype brotli_encoding = { "br", NULL, + CURL_CW_FLAG_BLOWUP, brotli_do_init, brotli_do_write, Curl_cwriter_def_flush, @@ -607,6 +610,7 @@ static void zstd_do_close(struct Curl_easy *data, static const struct Curl_cwtype zstd_encoding = { "zstd", NULL, + CURL_CW_FLAG_BLOWUP, zstd_do_init, zstd_do_write, Curl_cwriter_def_flush, @@ -619,6 +623,7 @@ static const struct Curl_cwtype zstd_encoding = { static const struct Curl_cwtype identity_encoding = { "identity", "none", + 0, Curl_cwriter_def_init, Curl_cwriter_def_write, Curl_cwriter_def_flush, @@ -707,6 +712,7 @@ static void error_do_close(struct Curl_easy *data, static const struct Curl_cwtype error_writer = { "ce-error", NULL, + 0, error_do_init, error_do_write, Curl_cwriter_def_flush, diff --git a/lib/cw-out.c b/lib/cw-out.c index 5ff4a695a1..b475877816 100644 --- a/lib/cw-out.c +++ b/lib/cw-out.c @@ -190,9 +190,9 @@ static CURLcode cw_out_cb_write(struct Curl_easy *data, nwritten = wcb((char *)CURL_UNCONST(buf), 1, blen, wcb_data); CURL_CBAPI_END(&guard); } - CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu", + CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes, type=%x -> %zu", blen, (otype == CW_OUT_HDS) ? "header" : "body", - nwritten); + (unsigned int)otype, nwritten); if(nwritten == CURL_WRITEFUNC_PAUSE) { if(data->conn->scheme->flags & PROTOPT_NONETWORK) { /* Protocols that work without network cannot be paused. This is @@ -429,8 +429,6 @@ static CURLcode cw_out_write(struct Curl_easy *data, CURLcode result; bool flush_all = !!(type & CLIENTWRITE_EOS); - CURL_TRC_WRITE(data, "[OUT] write %zu bytes, type=%x", - blen, (unsigned int)type); if((type & CLIENTWRITE_BODY) || ((type & CLIENTWRITE_HEADER) && data->set.include_header)) { cw_out_type otype = (!blen && (type & CLIENTWRITE_0LEN)) ? @@ -454,32 +452,34 @@ static CURLcode cw_out_do_flush(struct Curl_easy *data, bool flush_all) { struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out; - CURLcode result = CURLE_OK; if(ctx->errored) return CURLE_WRITE_ERROR; - if(data->req.writer.paused) - return CURLE_OK; /* not doing it */ - result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all); - if(result) { - ctx->errored = TRUE; - cw_out_bufs_free(ctx); - return result; + if(!data->req.writer.paused && ctx->buf) { + CURLcode result; + + CURL_TRC_WRITE(data, "[OUT] flush"); + result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all); + if(result) { + ctx->errored = TRUE; + cw_out_bufs_free(ctx); + return result; + } } - return result; + return CURLE_OK; } static CURLcode cw_out_flush(struct Curl_easy *data, struct Curl_cwriter *writer) { - CURL_TRC_WRITE(data, "[OUT] flush"); return cw_out_do_flush(data, writer, FALSE); } const struct Curl_cwtype Curl_cwt_out = { "cw-out", NULL, + 0, cw_out_init, cw_out_write, cw_out_flush, diff --git a/lib/cw-pause.c b/lib/cw-pause.c index ccdc7657b1..a5b9ba148e 100644 --- a/lib/cw-pause.c +++ b/lib/cw-pause.c @@ -207,6 +207,7 @@ static CURLcode cw_pause_write(struct Curl_easy *data, const struct Curl_cwtype Curl_cwt_pause = { "cw-pause", NULL, + 0, cw_pause_init, cw_pause_write, cw_pause_flush, diff --git a/lib/ftp.c b/lib/ftp.c index ff9fed98af..abcc9d25a9 100644 --- a/lib/ftp.c +++ b/lib/ftp.c @@ -441,6 +441,7 @@ static CURLcode ftp_cw_lc_write(struct Curl_easy *data, static const struct Curl_cwtype ftp_cw_lc = { "ftp-lineconv", NULL, + 0, Curl_cwriter_def_init, ftp_cw_lc_write, Curl_cwriter_def_flush, diff --git a/lib/headers.c b/lib/headers.c index 5d2101ee88..3bc9c3aceb 100644 --- a/lib/headers.c +++ b/lib/headers.c @@ -315,6 +315,7 @@ static CURLcode hds_cw_collect_write(struct Curl_easy *data, static const struct Curl_cwtype hds_cw_collect = { "hds-collect", NULL, + 0, Curl_cwriter_def_init, hds_cw_collect_write, Curl_cwriter_def_flush, diff --git a/lib/http_chunks.c b/lib/http_chunks.c index 85f453eb59..f67459bf55 100644 --- a/lib/http_chunks.c +++ b/lib/http_chunks.c @@ -465,6 +465,7 @@ static CURLcode cw_chunked_write(struct Curl_easy *data, const struct Curl_cwtype Curl_httpchunk_unencoder = { "chunked", NULL, + 0, cw_chunked_init, cw_chunked_write, Curl_cwriter_def_flush, diff --git a/lib/sendf.c b/lib/sendf.c index 27f3055214..a5aad668f5 100644 --- a/lib/sendf.c +++ b/lib/sendf.c @@ -293,6 +293,7 @@ static CURLcode cw_download_write(struct Curl_easy *data, static const struct Curl_cwtype cw_download = { "protocol", NULL, + 0, Curl_cwriter_def_init, cw_download_write, Curl_cwriter_def_flush, @@ -315,6 +316,7 @@ static CURLcode cw_raw_write(struct Curl_easy *data, static const struct Curl_cwtype cw_raw = { "raw", NULL, + 0, Curl_cwriter_def_init, cw_raw_write, Curl_cwriter_def_flush, @@ -484,9 +486,9 @@ CURLcode Curl_cwriter_add(struct Curl_easy *data, return result; } - if(writer->phase == CURL_CW_CONTENT_DECODE) { - /* On adding a content decoder, add the pause writer. Do this - * BEFORE the given writer as any failure will make the + if(writer->cwt->flags & CURL_CW_FLAG_BLOWUP) { + /* On adding a writer that may blow up write sizes, e.g. zip bombs, + * add the pause writer. Do this first as any failure will make the * caller destroy the writer again. */ result = cwriter_ensure_pause_writer(data); if(result) diff --git a/lib/sendf.h b/lib/sendf.h index 7f31b24662..9863c5b0dc 100644 --- a/lib/sendf.h +++ b/lib/sendf.h @@ -110,10 +110,14 @@ typedef enum { CURL_CW_CLIENT /* data written to client */ } Curl_cwriter_phase; +/* writer may blow up size of write data, e.g. zip bombs */ +#define CURL_CW_FLAG_BLOWUP (1U << 0) + /* Client Writer Type, provides the implementation */ struct Curl_cwtype { const char *name; /* writer name. */ const char *alias; /* writer name alias, maybe NULL. */ + uint8_t flags; /* flags for writer behaviour */ CURLcode (*do_init)(struct Curl_easy *data, struct Curl_cwriter *writer); CURLcode (*do_write)(struct Curl_easy *data, diff --git a/lib/ws.c b/lib/ws.c index cdaec3ba93..46f4505a6a 100644 --- a/lib/ws.c +++ b/lib/ws.c @@ -695,6 +695,7 @@ static CURLcode ws_cw_dec_next(const uint8_t *buf, size_t buflen, update_meta(ws, frame_age, frame_flags, payload_offset, payload_len, buflen); + CURL_TRC_WRITE(data, "[WS] pass %zu decoded bytes", buflen); result = Curl_cwriter_write(data, ctx->next_writer, (ctx->cw_type | CLIENTWRITE_0LEN), (const char *)buf, buflen); @@ -713,7 +714,7 @@ static CURLcode ws_cw_write(struct Curl_easy *data, struct websocket *ws; CURLcode result = CURLE_OK; - CURL_TRC_WRITE(data, "ws_cw_write(len=%zu, type=%d)", nbytes, type); + CURL_TRC_WRITE(data, "[WS] write(len=%zu, type=%d)", nbytes, type); if(!(type & CLIENTWRITE_BODY) || data->set.ws_raw_mode) return Curl_cwriter_write(data, writer->next, type, buf, nbytes); @@ -733,6 +734,10 @@ static CURLcode ws_cw_write(struct Curl_easy *data, } } + result = Curl_cwriter_flush(data, writer->next); + if(result) + goto out; + while(!Curl_bufq_is_empty(&ctx->buf) && !Curl_cwriter_is_paused(data)) { struct ws_cw_dec_ctx pass_ctx; pass_ctx.data = data; @@ -822,6 +827,7 @@ out: static const struct Curl_cwtype ws_cw_decode = { "ws-decode", NULL, + 0, ws_cw_init, ws_cw_write, ws_cw_flush, diff --git a/tests/http/test_20_websockets.py b/tests/http/test_20_websockets.py index b559383c93..70c0079719 100644 --- a/tests/http/test_20_websockets.py +++ b/tests/http/test_20_websockets.py @@ -325,12 +325,22 @@ class TestWebsockets: rss2 = r.profile.stats['rss'] / (1024 * 1024) assert (rss1 * 1.1) >= rss2, 'bad memory increase' - # test frame delivery when pausing - def test_20_12_pause_frames(self, env: Env, ws_4frames): + # test small frames delivery when pausing + def test_20_12_pause_frames_small(self, env: Env, ws_4frames): payload = 127 * "x" client = LocalClient(env=env, name='cli_ws_pause') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{ws_4frames.port}/' + url = f'ws://localhost:{ws_4frames.port}/small' + r = client.run(args=[url, payload]) + r.check_exit_code(0) + + # test small frames delivery when pausing + def test_20_13_pause_frames_large(self, env: Env, ws_4frames): + payload = 127 * "x" + client = LocalClient(env=env, name='cli_ws_pause') + if not client.exists(): + pytest.skip(f'example client not built: {client.name}') + url = f'ws://localhost:{ws_4frames.port}/large' r = client.run(args=[url, payload]) r.check_exit_code(0) diff --git a/tests/http/testenv/ws_4frames_server.py b/tests/http/testenv/ws_4frames_server.py index 42d6e9dc29..488fb5a9cf 100755 --- a/tests/http/testenv/ws_4frames_server.py +++ b/tests/http/testenv/ws_4frames_server.py @@ -29,21 +29,30 @@ import logging import websockets -MESSAGES = [ +MESSAGES_SMALL = [ "Hello 1", "Hello 2", "Hello 3", "Hello 4", ] +MESSAGES_LARGE = [ + b"x" * 65536, + b"x" * 65536, + b"x" * 65536, + b"x" * 65536, +] + async def handler(websocket): peer = websocket.remote_address print(f"client from {peer[0]}:{peer[1]}", flush=True) print("handshake complete", flush=True) + msgs = MESSAGES_LARGE if websocket.request.path == '/large' else MESSAGES_SMALL + await asyncio.sleep(0.1) - for index, payload in enumerate(MESSAGES, start=1): + for index, payload in enumerate(msgs, start=1): await websocket.send(payload) print(f"sent frame {index}: {payload!r}", flush=True) # await asyncio.sleep(0.2)