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

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,

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,

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,

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,

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,

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,

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)

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,

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,

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)

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)