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 a3724d319d69cf949f365196595e31c605ac6169..e23850783619a85f600fb73a2be01773455aa6c8 100644 (file)
--- 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 5ff4a695a1c10a0903228be406d27d80909cd2e7..b47587781682a91aac12c08ec4ffadd465692c21 100644 (file)
--- 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 ccdc7657b187c13b93cb0c1581d4beaba8cf733c..a5b9ba148e08bcc2aab38c1b387ece7cdb923101 100644 (file)
--- 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 ff9fed98afc5ea97469a898156c70e7206c8e08c..abcc9d25a967a7d2602ed28bfa4a5be4dee9cd26 100644 (file)
--- 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 5d2101ee881a6e6a0c0c04240bf75d8115d8b187..3bc9c3aceba06e93ae4491d4a7da930b115ef2d3 100644 (file)
--- 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 85f453eb59fca47a197249277a3ad59223ec4314..f67459bf55baf4783c04e96230d671824e076307 100644 (file)
--- 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 27f30552147740279e1dbd9632470e94bce63bc3..a5aad668f5e0fc3928133fbf5b02c075538c7d63 100644 (file)
--- 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 7f31b246622b4a82efe852ee26ecfec0f6407298..9863c5b0dcaa71d6ce50f348f95c4a49324b989d 100644 (file)
--- 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 cdaec3ba935647c78d0175fc68482f5a3383fa63..46f4505a6a11fd2436cbfa2127f17eb07b115e77 100644 (file)
--- 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 b559383c937aa81ee70778fc8e1902889977ce31..70c00797194a92198731f0b056ab71625a3ed646 100644 (file)
--- 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 42d6e9dc2995dfb709f9bf3f2dea3a05a01bcd3c..488fb5a9cfc8d78dd2cce13cf0dc072bbc0b02f4 100755 (executable)
--- 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)