Sync high-value upstream fixes: response queue poisoning (GHSA-35p6-xmwp-9g52), multi-byte setEncoding, null queue guard

lib/api/readable.js

@@ -15,6 +15,7 @@ const kContentType = Symbol('kContentType')
 const kContentLength = Symbol('kContentLength')
 const kUsed = Symbol('kUsed')
 const kBytesRead = Symbol('kBytesRead')
+const kPreservedBuffer = Symbol('kPreservedBuffer')
 
 const noop = () => {}
 
@@ -295,7 +296,37 @@ class BodyReadable extends Readable {
    */
   setEncoding (encoding) {
     if (Buffer.isEncoding(encoding)) {
-      this._readableState.encoding = encoding
+      // Preserve raw Buffer chunks for the consume path (body.text(),
+      // body.json(), etc.) before super.setEncoding() replaces them
+      // with decoded strings. Without this, the consume path would
+      // lose access to the original bytes — some of which may be held
+      // by the decoder for incomplete multi-byte sequences, and the
+      // rest converted to strings that can't be safely concatenated
+      // byte-wise.
+      const state = this._readableState
+      const buffer = state.buffer
+      if (buffer && state.length > 0) {
+        const bufferIndex = state.bufferIndex ?? 0
+        const preserved = []
+        const source = typeof buffer.slice === 'function'
+          ? buffer.slice(bufferIndex)
+          : buffer
+        for (const data of source) {
+          if (Buffer.isBuffer(data)) {
+            preserved.push(data)
+          }
+        }
+        if (preserved.length > 0) {
+          this[kPreservedBuffer] = (this[kPreservedBuffer] || []).concat(preserved)
+        }
+      }
+
+      // Delegate to Node.js Readable.setEncoding() which initializes a
+      // StringDecoder and re-encodes already-buffered chunks. This properly
+      // handles multi-byte sequences split at chunk boundaries for the
+      // for-await / on('data') paths. Without this, Node.js uses
+      // buf.toString(encoding) on each chunk, producing U+FFFD for split chars.
+      super.setEncoding(encoding)
     }
     return this
   }
@@ -390,7 +421,17 @@ function consumeStart (consume) {
 
   const { _readableState: state } = consume.stream
 
-  if (state.bufferIndex) {
+  // If setEncoding() was called, state.buffer may contain decoded strings
+  // (which would break Buffer.concat in chunksDecode). Use the preserved
+  // raw Buffers (saved before super.setEncoding() in setEncoding()) for
+  // byte-level accurate consumption. Otherwise read from state.buffer.
+  const preserved = consume.stream[kPreservedBuffer]
+  if (preserved && preserved.length > 0) {
+    for (const chunk of preserved) {
+      consumePush(consume, chunk)
+    }
+    consume.stream[kPreservedBuffer] = null
+  } else if (state.bufferIndex) {
     const start = state.bufferIndex
     const end = state.buffer.length
     for (let n = start; n < end; n++) {
@@ -403,10 +444,12 @@ function consumeStart (consume) {
   }
 
   if (state.endEmitted) {
-    consumeEnd(this[kConsume], this._readableState.encoding)
+    // consumeStart is a plain function, so `this` is undefined here in strict
+    // mode — use the consume argument and captured state, not `this`.
+    consumeEnd(consume, state.encoding)
   } else {
     consume.stream.on('end', function () {
-      consumeEnd(this[kConsume], this._readableState.encoding)
+      consumeEnd(consume, state.encoding)
     })
   }
 

lib/dispatcher/client-h1.js

@@ -57,6 +57,9 @@ const constants = require('../llhttp/constants.js')
 const EMPTY_BUF = Buffer.alloc(0)
 const FastBuffer = Buffer[Symbol.species]
 const removeAllListeners = util.removeAllListeners
+const kIdleSocketValidation = Symbol('kIdleSocketValidation')
+const kIdleSocketValidationTimeout = Symbol('kIdleSocketValidationTimeout')
+const kSocketUsed = Symbol('kSocketUsed')
 
 async function lazyllhttp () {
   const llhttpWasmData = process.env.JEST_WORKER_ID ? require('../llhttp/llhttp-wasm.js') : undefined
@@ -473,6 +476,15 @@ class Parser {
       return -1
     }
 
+    // A response arriving while nothing is inflight means a previously idle
+    // keep-alive socket received an unsolicited/early response. Matching it
+    // against the next request would poison the response queue, so discard the
+    // socket instead (GHSA-35p6-xmwp-9g52).
+    if (client[kRunning] === 0) {
+      util.destroy(socket, new SocketError('bad response', util.getSocketInfo(socket)))
+      return -1
+    }
+
     const request = client[kQueue][client[kRunningIdx]]
     if (!request) {
       return -1
@@ -601,6 +613,13 @@ class Parser {
       return -1
     }
 
+    // See onMessageBegin: response headers without an inflight request mean a
+    // poisoned idle socket (GHSA-35p6-xmwp-9g52).
+    if (client[kRunning] === 0) {
+      util.destroy(socket, new SocketError('bad response', util.getSocketInfo(socket)))
+      return -1
+    }
+
     const request = client[kQueue][client[kRunningIdx]]
 
     /* istanbul ignore next: difficult to make a test case for */
@@ -783,6 +802,10 @@ class Parser {
     request.onComplete(headers)
 
     client[kQueue][client[kRunningIdx]++] = null
+    // Once the socket has served a response it is a reuse candidate; a
+    // subsequently idle socket must be revalidated before dispatching the next
+    // pending request (see resumeH1 / GHSA-35p6-xmwp-9g52).
+    socket[kSocketUsed] = client[kPending] === 0
 
     if (socket[kWriting]) {
       assert(client[kRunning] === 0)
@@ -864,6 +887,9 @@ async function connectH1 (client, socket) {
   socket[kWriting] = false
   socket[kReset] = false
   socket[kBlocking] = false
+  socket[kIdleSocketValidation] = 0
+  socket[kIdleSocketValidationTimeout] = null
+  socket[kSocketUsed] = false
   socket[kParser] = new Parser(client, socket, llhttpInstance)
 
   util.addListener(socket, 'error', onHttpSocketError)
@@ -906,7 +932,7 @@ async function connectH1 (client, socket) {
      * @returns {boolean}
      */
     busy (request) {
-      if (socket[kWriting] || socket[kReset] || socket[kBlocking]) {
+      if (socket[kWriting] || socket[kReset] || socket[kBlocking] || socket[kIdleSocketValidation] === 1) {
         return true
       }
 
@@ -986,6 +1012,8 @@ function onHttpSocketEnd () {
 function onHttpSocketClose () {
   const parser = this[kParser]
 
+  clearIdleSocketValidation(this)
+
   if (parser) {
     if (!this[kError] && parser.statusCode && !parser.shouldKeepAlive) {
       this[kError] = parser.finish() || this[kError]
@@ -1032,6 +1060,28 @@ function onSocketClose () {
   this[kClosed] = true
 }
 
+function clearIdleSocketValidation (socket) {
+  if (socket[kIdleSocketValidationTimeout]) {
+    clearTimeout(socket[kIdleSocketValidationTimeout])
+    socket[kIdleSocketValidationTimeout] = null
+  }
+
+  socket[kIdleSocketValidation] = 0
+}
+
+function scheduleIdleSocketValidation (client, socket) {
+  socket[kIdleSocketValidation] = 1
+  socket[kIdleSocketValidationTimeout] = setTimeout(() => {
+    socket[kIdleSocketValidationTimeout] = null
+    socket[kIdleSocketValidation] = 2
+
+    if (client[kSocket] === socket && !socket.destroyed) {
+      client[kResume]()
+    }
+  }, 0)
+  socket[kIdleSocketValidationTimeout].unref?.()
+}
+
 /**
  * @param {import('./client.js')} client
  */
@@ -1049,7 +1099,47 @@ function resumeH1 (client) {
       socket[kNoRef] = false
     }
 
+    // Before dispatching a pending request onto a previously-used idle socket,
+    // proactively read from it once to surface any unsolicited bytes a
+    // misbehaving/malicious peer may have injected while it was idle. The
+    // one-tick validation window lets a stray response (which onMessageBegin
+    // turns into a 'bad response' teardown) close the socket before we bind the
+    // next request to it, preventing response queue poisoning
+    // (GHSA-35p6-xmwp-9g52). busy() reports the socket as busy while
+    // validation (state 1) is pending so no request is dispatched meanwhile.
+    if (client[kRunning] === 0 && client[kPending] > 0 && socket[kSocketUsed]) {
+      if (socket[kIdleSocketValidation] === 0) {
+        scheduleIdleSocketValidation(client, socket)
+        socket[kParser].readMore()
+        if (socket.destroyed) {
+          return
+        }
+        return
+      }
+
+      if (socket[kIdleSocketValidation] === 1) {
+        socket[kParser].readMore()
+        if (socket.destroyed) {
+          return
+        }
+        return
+      }
+    }
+
+    if (client[kRunning] === 0) {
+      socket[kParser].readMore()
+      if (socket.destroyed) {
+        return
+      }
+    }
+
     if (client[kSize] === 0) {
+      // Socket is fully idle. If validation was scheduled/completed but the
+      // pending request that triggered it went away before being written (e.g.
+      // it was aborted), reset validation state here so the socket is
+      // revalidated on its next reuse instead of getting stuck in state 1/2 and
+      // skipping the poisoning check (GHSA-35p6-xmwp-9g52).
+      clearIdleSocketValidation(socket)
       if (socket[kParser].timeoutType !== TIMEOUT_KEEP_ALIVE) {
         socket[kParser].setTimeout(client[kKeepAliveTimeoutValue], TIMEOUT_KEEP_ALIVE)
       }
@@ -1141,6 +1231,10 @@ function writeH1 (client, request) {
 
   const socket = client[kSocket]
 
+  // The socket is being handed a request; cancel any in-flight idle validation
+  // and reset its state so the reuse guard in resumeH1 starts clean next time.
+  clearIdleSocketValidation(socket)
+
   /**
    * @param {Error} [err]
    * @returns {void}

lib/dispatcher/client.js

@@ -323,7 +323,9 @@ class Client extends DispatcherBase {
       const requests = this[kQueue].splice(this[kPendingIdx])
       for (let i = 0; i < requests.length; i++) {
         const request = requests[i]
-        util.errorRequest(this, request, err)
+        if (request != null) {
+          util.errorRequest(this, request, err)
+        }
       }
 
       const callback = () => {
@@ -362,7 +364,9 @@ function onError (client, err) {
 
     for (let i = 0; i < requests.length; i++) {
       const request = requests[i]
-      util.errorRequest(client, request, err)
+      if (request != null) {
+        util.errorRequest(client, request, err)
+      }
     }
     assert(client[kSize] === 0)
   }

test/readable-setencoding-multibyte.js

@@ -0,0 +1,58 @@
+'use strict'
+
+const { tspl } = require('@matteo.collina/tspl')
+const { test, after } = require('node:test')
+const { createServer } = require('node:http')
+const { Client } = require('..')
+
+test('setEncoding(\'utf8\') handles 3-byte UTF-8 characters split across chunks', async (t) => {
+  t = tspl(t, { plan: 2 })
+
+  // CJK character '傳' is 3 bytes: 0xe5, 0x82, 0xb3
+  // Build a payload where this character will be split at the chunk boundary
+  const cjkChar = '傳' // U+50B3, bytes: e5 82 b3
+  const prefix = 'a'.repeat(10) // 10 ASCII bytes
+  const text = prefix + cjkChar + 'end'
+  const buf = Buffer.from(text) // 10 + 3 + 3 = 16 bytes
+
+  // Split at byte 11, which is in the middle of the 3-byte CJK character
+  // prefix (10 bytes) + first byte of '傳' (0xe5) | remaining 2 bytes (0x82 0xb3) + 'end'
+  const chunk1 = buf.subarray(0, 11)
+  const chunk2 = buf.subarray(11)
+
+  const server = createServer({ joinDuplicateHeaders: true }, (req, res) => {
+    // Send raw buffers to ensure the split is exactly where we want it
+    res.writeHead(200, { 'content-type': 'text/plain; charset=utf-8' })
+    res.write(chunk1)
+    // Use setTimeout to force separate TCP packets / chunks
+    setTimeout(() => {
+      res.end(chunk2)
+    }, 50)
+  })
+  after(() => {
+    server.closeAllConnections?.()
+    server.close()
+  })
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(client.destroy.bind(client))
+
+    const { body } = await client.request({
+      path: '/',
+      method: 'GET'
+    })
+    body.setEncoding('utf8')
+
+    let result = ''
+    for await (const chunk of body) {
+      result += chunk
+    }
+
+    // Must not contain U+FFFD replacement characters
+    t.strictEqual(result.includes('\ufffd'), false, 'should not contain U+FFFD replacement characters')
+    t.strictEqual(result, text, 'decoded text should match original')
+  })
+
+  await t.completed
+})

test/response-queue-poisoning.js

@@ -0,0 +1,57 @@
+'use strict'
+
+const assert = require('node:assert')
+const { once } = require('node:events')
+const { createServer } = require('node:http')
+const { after, test } = require('node:test')
+const { Client } = require('..')
+
+function readBody (body) {
+  return new Promise((resolve, reject) => {
+    let data = ''
+    body.setEncoding('latin1')
+    body.on('data', chunk => { data += chunk })
+    body.on('end', () => resolve(data))
+    body.on('error', reject)
+  })
+}
+
+test('should not reuse an idle socket with buffered unsolicited response bytes', async () => {
+  let evilServerSocket
+
+  const server = createServer((req, res) => {
+    if (!evilServerSocket) {
+      evilServerSocket = req.socket
+    }
+
+    res.end(req.url)
+  })
+  after(() => server.close())
+
+  await new Promise(resolve => server.listen(0, resolve))
+
+  const client = new Client(`http://localhost:${server.address().port}`, {
+    keepAliveTimeout: 300e3
+  })
+  after(() => client.close())
+
+  const response1 = await client.request({ path: '/request1', method: 'GET' })
+  assert.strictEqual(await readBody(response1.body), '/request1')
+
+  const disconnected = once(client, 'disconnect')
+
+  evilServerSocket.write(
+    'HTTP/1.1 200 OK\r\n' +
+    'Poison-Free-Socket: true\r\n' +
+    'Connection: keep-alive\r\n' +
+    'Keep-Alive: timeout=300\r\n' +
+    'Content-Length: 0\r\n' +
+    '\r\n'
+  )
+
+  await disconnected
+
+  const response2 = await client.request({ path: '/request2', method: 'GET' })
+  assert.strictEqual(response2.headers['poison-free-socket'], undefined)
+  assert.strictEqual(await readBody(response2.body), '/request2')
+})