CVE-2024-24750: Uncontrolled Resource Consumption (CWE-400)
value: Undici is an HTTP/1.1 client, written from scratch for Node.js. In affected versions calling `fetch(url)` and not consuming the incoming body ((or consuming it very slowing) will lead to a memory leak. This issue has been addressed in version 6.6.1. Users are advised to upgrade. Users unable to upgrade should make sure to always consume the incoming body.
CWE description: The product does not properly control the allocation and maintenance of a limited resource, thereby enabling an actor to influence the amount of resources consumed, eventually leading to the exhaustion of available resources.
Published: 2024-02-16T22:15Z
Repository: https://github.com/nodejs/undici
Language: JavaScript
Files changed: lib/fetch/index.js, test/fetch/pull-dont-push.js
Commit date: 2024-02-05 09:35:45 +0100
Commit message: Merge pull request from GHSA-9f24-jqhm-jfcw

* fetch: pull don't push

Signed-off-by: Matteo Collina <hello@matteocollina.com>

* added tests

Signed-off-by: Matteo Collina <hello@matteocollina.com>

---------

Signed-off-by: Matteo Collina <hello@matteocollina.com>
Diff: diff --git a/lib/fetch/index.js b/lib/fetch/index.js
index a3d02271d..6fe684477 100644
--- a/lib/fetch/index.js
+++ b/lib/fetch/index.js
@@ -1097,24 +1097,25 @@ function fetchFinale (fetchParams, response) {
     // 4. Set internalResponse’s body’s stream to the result of internalResponse’s body’s stream piped through transformStream.
     internalResponse.body.stream.pipeThrough(transformStream)
 
     const byteStream = new ReadableStream({
       readableStream: transformStream.readable,
-      async start (controller) {
+      async pull (controller) {
         const reader = this.readableStream.getReader()
 
-        while (true) {
+        while (controller.desiredSize >= 0) {
           const { done, value } = await reader.read()
 
           if (done) {
             queueMicrotask(() => readableStreamClose(controller))
             break
           }
 
           controller.enqueue(value)
         }
       },
+      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 }),
       type: 'bytes'
     })
 
     internalResponse.body.stream = byteStream
   }
@@ -1925,20 +1926,22 @@ async function httpNetworkFetch (
   // 15. Let stream be a new ReadableStream.
   // 16. Set up stream with byte reading support with pullAlgorithm set to pullAlgorithm,
   //     cancelAlgorithm set to cancelAlgorithm.
   const stream = new ReadableStream(
     {
+      highWaterMark: 16384,
       async start (controller) {
         fetchParams.controller.controller = controller
       },
       async pull (controller) {
         await pullAlgorithm(controller)
       },
       async cancel (reason) {
         await cancelAlgorithm(reason)
       },
-      type: 'bytes'
+      type: 'bytes',
+      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 })
     }
   )
 
   // 17. Run these steps, but abort when the ongoing fetch is terminated:
 
diff --git a/test/fetch/pull-dont-push.js b/test/fetch/pull-dont-push.js
new file mode 100644
index 000000000..24bad7c39
--- /dev/null
+++ b/test/fetch/pull-dont-push.js
@@ -0,0 +1,53 @@
+'use strict'
+
+const { test } = require('node:test')
+const assert = require('node:assert')
+const { fetch } = require('../..')
+const { createServer } = require('http')
+const { once } = require('events')
+const { Readable, pipeline } = require('stream')
+const { setTimeout: sleep } = require('timers/promises')
+
+const { closeServerAsPromise } = require('../utils/node-http')
+
+test('Allow the usage of custom implementation of AbortController', async (t) => {
+  let count = 0
+  let socket
+  const server = createServer((req, res) => {
+    res.statusCode = 200
+    socket = res.socket
+
+    // infinite stream
+    const stream = new Readable({
+      read () {
+        this.push('a')
+        if (count++ > 1000000) {
+          this.push(null)
+        }
+      }
+    })
+
+    pipeline(stream, res, () => {})
+  })
+
+  t.after(closeServerAsPromise(server))
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  t.diagnostic('server listening on port %d', server.address().port)
+  const res = await fetch(`http://localhost:${server.address().port}`)
+  t.diagnostic('fetched')
+
+  // Some time is needed to fill the buffer
+  await sleep(1000)
+
+  assert.strictEqual(socket.bytesWritten < 1024 * 1024, true) // 1 MB
+  socket.destroy()
+
+  // consume the  stream
+  try {
+    /* eslint-disable-next-line no-empty, no-unused-vars */
+    for await (const chunk of res.body) {}
+  } catch {}
+})

Vulnerable code:       async start (controller) {
        while (true) {
      type: 'bytes'
Fixed code:       async pull (controller) {
        while (controller.desiredSize >= 0) {
      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 }),
      highWaterMark: 16384,
      type: 'bytes',
      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 })
'use strict'

const { test } = require('node:test')
const assert = require('node:assert')
const { fetch } = require('../..')
const { createServer } = require('http')
const { once } = require('events')
const { Readable, pipeline } = require('stream')
const { setTimeout: sleep } = require('timers/promises')

const { closeServerAsPromise } = require('../utils/node-http')

test('Allow the usage of custom implementation of AbortController', async (t) => {
  let count = 0
  let socket
  const server = createServer((req, res) => {
    res.statusCode = 200
    socket = res.socket

    // infinite stream
    const stream = new Readable({
      read () {
        this.push('a')
        if (count++ > 1000000) {
          this.push(null)
        }
      }
    })

    pipeline(stream, res, () => {})
  })

  t.after(closeServerAsPromise(server))

  server.listen(0)
  await once(server, 'listening')

  t.diagnostic('server listening on port %d', server.address().port)
  const res = await fetch(`http://localhost:${server.address().port}`)
  t.diagnostic('fetched')

  // Some time is needed to fill the buffer
  await sleep(1000)

  assert.strictEqual(socket.bytesWritten < 1024 * 1024, true) // 1 MB
  socket.destroy()

  // consume the  stream
  try {
    /* eslint-disable-next-line no-empty, no-unused-vars */
    for await (const chunk of res.body) {}
  } catch {}
})