http: fix handling of HTTP upgrades with bodies

Previously, we ignored all indicators of the body (content-length or
transfer-encoding headers) and treated any information following
the headers as part of the upgraded stream. We now fully process the
requests bodies separately instead, allowing us to automatically handle
correct parsing of the body like any other HTTP request.

This is a breaking change if you are currently accepting HTTP Upgrade
requests with request bodies. In this case, before now you will have
received the request body and then the upgraded data on the same stream
without any distinction or HTTP parsing applies.

Now, you will need to separately read the request body from the request
(the 1st argument) and the upgraded data from the upgrade stream (the
2nd argument). If you're not interested in request bodies, you can
continue to just read from the upgrade stream directly.

This change tries to maximize backward compatibility by ensuring the 2nd
argument behaves exactly like the underlying socket even in this case.
Nonetheless, it is best not to treat the 2nd argument as a socket:
instead, use this argument for access to the upgrade stream data, and
use request.socket if you need direct access to the underlying socket
itself.

Author: pimterry

doc/api/http.md

@@ -745,7 +745,7 @@ added: v0.1.94
 -->
 
 * `response` {http.IncomingMessage}
-* `socket` {stream.Duplex}
+* `stream` {stream.Duplex}
 * `head` {Buffer}
 
 Emitted each time a server responds to a request with an upgrade. If this
@@ -768,13 +768,13 @@ const server = http.createServer((req, res) => {
   res.writeHead(200, { 'Content-Type': 'text/plain' });
   res.end('okay');
 });
-server.on('upgrade', (req, socket, head) => {
-  socket.write('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+server.on('upgrade', (req, stream, head) => {
+  stream.write('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
                'Upgrade: WebSocket\r\n' +
                'Connection: Upgrade\r\n' +
                '\r\n');
 
-  socket.pipe(socket); // echo back
+  stream.pipe(stream); // echo back
 });
 
 // Now that server is running
@@ -793,9 +793,9 @@ server.listen(1337, '127.0.0.1', () => {
   const req = http.request(options);
   req.end();
 
-  req.on('upgrade', (res, socket, upgradeHead) => {
+  req.on('upgrade', (res, stream, upgradeHead) => {
     console.log('got upgraded!');
-    socket.end();
+    stream.end();
     process.exit(0);
   });
 });
@@ -809,13 +809,13 @@ const server = http.createServer((req, res) => {
   res.writeHead(200, { 'Content-Type': 'text/plain' });
   res.end('okay');
 });
-server.on('upgrade', (req, socket, head) => {
-  socket.write('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+server.on('upgrade', (req, stream, head) => {
+  stream.write('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
                'Upgrade: WebSocket\r\n' +
                'Connection: Upgrade\r\n' +
                '\r\n');
 
-  socket.pipe(socket); // echo back
+  stream.pipe(stream); // echo back
 });
 
 // Now that server is running
@@ -834,9 +834,9 @@ server.listen(1337, '127.0.0.1', () => {
   const req = http.request(options);
   req.end();
 
-  req.on('upgrade', (res, socket, upgradeHead) => {
+  req.on('upgrade', (res, stream, upgradeHead) => {
     console.log('got upgraded!');
-    socket.end();
+    stream.end();
     process.exit(0);
   });
 });
@@ -1674,6 +1674,14 @@ per connection (in the case of HTTP Keep-Alive connections).
 <!-- YAML
 added: v0.1.94
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/60016
+    description: Request bodies are no longer exposed raw (unparsed) on the
+                 socket argument. Instead, if a body is received, the stream
+                 argument will be a duplex that emits socket content only
+                 after the request body, while the parsed request body data
+                 will be emitted from the request, just as in normal server
+                 `'request'` events.
   - version:
      - v24.9.0
      - v22.21.0
@@ -1689,31 +1697,38 @@ changes:
 
 * `request` {http.IncomingMessage} Arguments for the HTTP request, as it is in
   the [`'request'`][] event
-* `socket` {stream.Duplex} Network socket between the server and client
+* `stream` {stream.Duplex} The upgraded stream between the server and client
 * `head` {Buffer} The first packet of the upgraded stream (may be empty)
 
 Emitted each time a client's HTTP upgrade request is accepted. By default
 all HTTP upgrade requests are ignored (i.e. only regular `'request'` events
 are emitted, sticking with the normal HTTP request/response flow) unless you
 listen to this event, in which case they are all accepted (i.e. the `'upgrade'`
 event is emitted instead, and future communication must handled directly
-through the raw socket). You can control this more precisely by using the
+through the raw stream). You can control this more precisely by using the
 server `shouldUpgradeCallback` option.
 
 Listening to this event is optional and clients cannot insist on a protocol
 change.
 
-After this event is emitted, the request's socket will not have a `'data'`
-event listener, meaning it will need to be bound in order to handle data
-sent to the server on that socket.
-
 If an upgrade is accepted by `shouldUpgradeCallback` but no event handler
-is registered then the socket is destroyed, resulting in an immediate
+is registered then the socket will be destroyed, resulting in an immediate
 connection closure for the client.
 
-This event is guaranteed to be passed an instance of the {net.Socket} class,
-a subclass of {stream.Duplex}, unless the user specifies a socket
-type other than {net.Socket}.
+In the uncommon case that the incoming request has a body, this body will be
+parsed as normal, separate to the upgrade stream, and the raw stream data will
+only begin after it has completed. To ensure that reading from the stream isn't
+blocked by waiting for the request body to be read, any reads on the stream
+will start the request body flowing automatically. If you want to read the
+request body, ensure that you do so (i.e. you attach `'data'` listeners)
+before starting to read from the upgraded stream.
+
+The stream argument will typically be the {net.Socket} instance used by the
+request, but in some cases (such as with a request body) it may be a duplex
+stream (though socket properties are exposed for backward compatibility). If
+required, you can access the raw connection underlying the request via
+[`request.socket`][], which is guaranteed to be an instance of {net.Socket}
+unless the user specified another socket type.
 
 ### `server.close([callback])`
 

lib/_http_server.js

@@ -26,14 +26,19 @@ const {
   Error,
   MathMin,
   NumberIsFinite,
+  ObjectGetPrototypeOf,
   ObjectKeys,
   ObjectSetPrototypeOf,
+  Proxy,
   ReflectApply,
+  ReflectGet,
+  ReflectSet,
   Symbol,
   SymbolAsyncDispose,
   SymbolFor,
 } = primordials;
 
+const { Duplex } = require('stream');
 const net = require('net');
 const EE = require('events');
 const assert = require('internal/assert');
@@ -43,6 +48,7 @@ const {
   continueExpression,
   chunkExpression,
   kIncomingMessage,
+  kSocket,
   HTTPParser,
   isLenient,
   _checkInvalidHeaderChar: checkInvalidHeaderChar,
@@ -106,6 +112,7 @@ const onResponseFinishChannel = dc.channel('http.server.response.finish');
 
 const kServerResponse = Symbol('ServerResponse');
 const kServerResponseStatistics = Symbol('ServerResponseStatistics');
+const kUpgradeStream = Symbol('UpgradeStream');
 
 const kOptimizeEmptyRequests = Symbol('OptimizeEmptyRequestsOption');
 
@@ -953,6 +960,106 @@ function socketOnError(e) {
   }
 }
 
+class UpgradeStream extends Duplex {
+  constructor(socket, req) {
+    super({
+      allowHalfOpen: socket.allowHalfOpen,
+    });
+
+    this[kSocket] = socket;
+    this[kIncomingMessage] = req;
+
+    // Proxy error, end & closure events immediately.
+    socket.on('error', (err) => this.destroy(err));
+
+    socket.on('close', () => this.destroy());
+    this.on('close', () => socket.destroy());
+
+    socket.on('end', () => {
+      this.push(null);
+
+      // Match the socket behaviour, where 'end' will fire despite no 'data'
+      // listeners if a socket with no pending data ends:
+      if (this.readableLength === 0) {
+        this.resume();
+      }
+    });
+
+    // Other events (most notably, reading) all only
+    // activate after requestBodyCompleted is called.
+
+    // This stream wraps itself in a proxy which forwards all non-stream
+    // property lookups back to the underlying socket, for backward
+    // compatibility.
+    // eslint-disable-next-line no-constructor-return
+    return new Proxy(this, {
+      __proto__: null,
+      has(target, prop) {
+        return prop in target || prop in socket;
+      },
+      get(target, prop, receiver) {
+        if (prop in target) {
+          return ReflectGet(target, prop, receiver);
+        }
+
+        return socket[prop];
+      },
+      set(target, prop, value, receiver) {
+        if (prop in target) {
+          return ReflectSet(target, prop, value, receiver);
+        }
+
+        socket[prop] = value;
+        return true;
+      },
+      getPrototypeOf(target) {
+        return ObjectGetPrototypeOf(socket);
+      },
+    });
+  }
+
+  requestBodyCompleted(upgradeHead) {
+    this[kIncomingMessage] = null;
+
+    // When the request body is completed, we begin streaming all the
+    // post-body data for the upgraded protocol:
+    if (upgradeHead?.length > 0) {
+      if (!this.push(upgradeHead)) {
+        this[kSocket].pause();
+      }
+    }
+
+    this[kSocket].on('data', (data) => {
+      if (!this.push(data)) {
+        this[kSocket].pause();
+      }
+    });
+  }
+
+  _read(size) {
+    // Reading the upgrade stream starts the request stream flowing. It's
+    // important that this happens, even if there are no listeners, or it
+    // would be impossible to read this without explicitly reading all the
+    // request body first, which is backward incompatible & awkward.
+    this[kIncomingMessage]?.resume();
+
+    this[kSocket].resume();
+  }
+
+  _final(callback) {
+    this[kSocket].end(callback);
+  }
+
+  _write(chunk, encoding, callback) {
+    this[kSocket].write(chunk, encoding, callback);
+  }
+
+  _destroy(err, callback) {
+    this[kSocket].destroy(err);
+    callback(err);
+  }
+}
+
 function onParserExecuteCommon(server, socket, parser, state, ret, d) {
   if (ret instanceof Error) {
     prepareError(ret, parser, d);
@@ -962,28 +1069,56 @@ function onParserExecuteCommon(server, socket, parser, state, ret, d) {
     // Upgrade or CONNECT
     const req = parser.incoming;
     debug('SERVER upgrade or connect', req.method);
+    const eventName = req.method === 'CONNECT' ? 'connect' : 'upgrade';
+
+    let upgradeStream;
+    if (req.complete) {
+      d ||= parser.getCurrentBuffer();
+
+      socket.removeListener('data', state.onData);
+      socket.removeListener('end', state.onEnd);
+      socket.removeListener('close', state.onClose);
+      socket.removeListener('drain', state.onDrain);
+      socket.removeListener('error', socketOnError);
+      socket.removeListener('timeout', socketOnTimeout);
+
+      unconsume(parser, socket);
+      parser.finish();
+      freeParser(parser, req, socket);
+      parser = null;
+
+      // If the request is complete (no body, or all body read upfront) then
+      // we just emit the socket directly as the upgrade stream.
+      upgradeStream = socket;
+    } else {
+      // If the body hasn't been fully parsed yet, we emit immediately but
+      // we add a wrapper around the socket to not expose incoming data
+      // until the request body has finished.
+
+      if (socket[kUpgradeStream]) {
+        // We've already emitted the incomplete upgrade - nothing do to
+        // until actual body parsing completion.
+        return;
+      }
 
-    d ||= parser.getCurrentBuffer();
+      d ||= Buffer.alloc(0);
 
-    socket.removeListener('data', state.onData);
-    socket.removeListener('end', state.onEnd);
-    socket.removeListener('close', state.onClose);
-    socket.removeListener('drain', state.onDrain);
-    socket.removeListener('error', socketOnError);
-    socket.removeListener('timeout', socketOnTimeout);
-    unconsume(parser, socket);
-    parser.finish();
-    freeParser(parser, req, socket);
-    parser = null;
+      upgradeStream = new UpgradeStream(socket, req);
+      socket[kUpgradeStream] = upgradeStream;
+    }
 
-    const eventName = req.method === 'CONNECT' ? 'connect' : 'upgrade';
     if (server.listenerCount(eventName) > 0) {
       debug('SERVER have listener for %s', eventName);
-      const bodyHead = d.slice(ret, d.length);
 
-      socket.readableFlowing = null;
+      const bodyHead = d.slice(ret, d.length);
 
-      server.emit(eventName, req, socket, bodyHead);
+      if (req.complete && socket[kUpgradeStream]) {
+        // Previously emitted, now completed - just activate the stream
+        socket[kUpgradeStream].requestBodyCompleted(bodyHead);
+      } else {
+        socket.readableFlowing = null;
+        server.emit(eventName, req, upgradeStream, bodyHead);
+      }
     } else {
       // Got upgrade or CONNECT method, but have no handler.
       socket.destroy();
@@ -1089,8 +1224,9 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
   if (req.upgrade) {
     req.upgrade = req.method === 'CONNECT' ||
                   !!server.shouldUpgradeCallback(req);
-    if (req.upgrade)
-      return 2;
+    if (req.upgrade) {
+      return 0;
+    }
   }
 
   state.incoming.push(req);