.github/workflows/ci.yml

@@ -17,7 +17,6 @@ jobs:
     strategy:
       matrix:
         node-version:
-          - 10
           - 18
 
     steps:

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 2023
 
+- [6.5.0](#650-2023-06-16) (Jun 2023)
 - [6.4.2](#642-2023-05-02) (May 2023)
 - [6.4.1](#641-2023-02-20) (Feb 2023)
 - [6.4.0](#640-2023-02-06) (Feb 2023)
@@ -47,6 +48,91 @@
 
 # Release notes
 
+## [6.5.0](https://github.com/socketio/engine.io/compare/6.4.2...6.5.0) (2023-06-16)
+
+
+### Bug Fixes
+
+* **uws:** discard any write to an aborted uWS response ([#682](https://github.com/socketio/engine.io/issues/682)) ([3144d27](https://github.com/socketio/engine.io/commit/3144d274584ae3b96cca4e609c66c56d534f1715))
+
+
+### Features
+
+#### Support for WebTransport
+
+The Engine.IO server can now use WebTransport as the underlying transport.
+
+WebTransport is a web API that uses the HTTP/3 protocol as a bidirectional transport. It's intended for two-way communications between a web client and an HTTP/3 server.
+
+References:
+
+- https://w3c.github.io/webtransport/
+- https://developer.mozilla.org/en-US/docs/Web/API/WebTransport
+- https://developer.chrome.com/articles/webtransport/
+
+Until WebTransport support lands [in Node.js](https://github.com/nodejs/node/issues/38478), you can use the `@fails-components/webtransport` package:
+
+```js
+import { readFileSync } from "fs";
+import { createServer } from "https";
+import { Server } from "engine.io";
+import { Http3Server } from "@fails-components/webtransport";
+
+// WARNING: the total length of the validity period MUST NOT exceed two weeks (https://w3c.github.io/webtransport/#custom-certificate-requirements)
+const cert = readFileSync("/path/to/my/cert.pem");
+const key = readFileSync("/path/to/my/key.pem");
+
+const httpsServer = createServer({
+  key,
+  cert
+});
+
+httpsServer.listen(3000);
+
+const engine = new Server({
+  transports: ["polling", "websocket", "webtransport"] // WebTransport is not enabled by default
+});
+
+engine.attach(httpsServer);
+
+const h3Server = new Http3Server({
+  port: 3000,
+  host: "0.0.0.0",
+  secret: "changeit",
+  cert,
+  privKey: key,
+});
+
+(async () => {
+  const stream = await h3Server.sessionStream("/engine.io/");
+  const sessionReader = stream.getReader();
+
+  while (true) {
+    const { done, value } = await sessionReader.read();
+    if (done) {
+      break;
+    }
+    engine.onWebTransportSession(value);
+  }
+})();
+
+h3Server.startServer();
+```
+
+Added in [123b68c](https://github.com/socketio/engine.io/commit/123b68c04f9e971f59b526e0f967a488ee6b0116).
+
+
+### Credits
+
+Huge thanks to [@OxleyS](https://github.com/OxleyS) for helping!
+
+
+### Dependencies
+
+- [`ws@~8.11.0`](https://github.com/websockets/ws/releases/tag/8.11.0) (no change)
+
+
+
 ## [6.4.2](https://github.com/socketio/engine.io/compare/6.4.1...6.4.2) (2023-05-02)
 
 :warning: This release contains an important security fix :warning:

SECURITY.md

@@ -19,5 +19,7 @@ We will get back to you as soon as possible and publish a fix if necessary.
 
 ## History
 
-- Jan 2022: [Uncaught exception in engine.io](https://github.com/socketio/engine.io/security/advisories/GHSA-273r-mgr4-v34f) (CVE-2022-21676)
-- Nov 2022: [Uncaught exception in engine.io](https://github.com/socketio/engine.io/security/advisories/GHSA-r7qp-cfhv-p84w) (CVE-2022-41940)
+- Feb 2020: [Resource exhaustion in engine.io](https://github.com/advisories/GHSA-j4f2-536g-r55m) (CVE-2020-36048)
+- Jan 2022: [Uncaught exception in engine.io](https://github.com/advisories/GHSA-273r-mgr4-v34f) (CVE-2022-21676)
+- Nov 2022: [Uncaught exception in engine.io](https://github.com/advisories/GHSA-r7qp-cfhv-p84w) (CVE-2022-41940)
+- May 2023: [Uncaught exception in engine.io](https://github.com/advisories/GHSA-q9mw-68c2-j6m5) (CVE-2023-31125)

lib/server.ts

@@ -15,10 +15,12 @@ import type {
 import type { CookieSerializeOptions } from "cookie";
 import type { CorsOptions, CorsOptionsDelegate } from "cors";
 import type { Duplex } from "stream";
+import { WebTransport } from "./transports/webtransport";
 
 const debug = debugModule("engine");
 
 const kResponseHeaders = Symbol("responseHeaders");
+const TEXT_DECODER = new TextDecoder();
 
 type Transport = "polling" | "websocket";
 
@@ -78,7 +80,13 @@ export interface ServerOptions {
     fn: (err: string | null | undefined, success: boolean) => void
   ) => void;
   /**
-   * the low-level transports that are enabled
+   * The low-level transports that are enabled. WebTransport is disabled by default and must be manually enabled:
+   *
+   * @example
+   * new Server({
+   *   transports: ["polling", "websocket", "webtransport"]
+   * });
+   *
    * @default ["polling", "websocket"]
    */
   transports?: Transport[];
@@ -140,6 +148,17 @@ type Middleware = (
   next: (err?: any) => void
 ) => void;
 
+function parseSessionId(handshake: string) {
+  if (handshake.startsWith("0{")) {
+    try {
+      const parsed = JSON.parse(handshake.substring(1));
+      if (typeof parsed.sid === "string") {
+        return parsed.sid;
+      }
+    } catch (e) {}
+  }
+}
+
 export abstract class BaseServer extends EventEmitter {
   public opts: ServerOptions;
 
@@ -166,7 +185,7 @@ export abstract class BaseServer extends EventEmitter {
         pingInterval: 25000,
         upgradeTimeout: 10000,
         maxHttpBufferSize: 1e6,
-        transports: Object.keys(transports),
+        transports: ["polling", "websocket"], // WebTransport is disabled by default
         allowUpgrades: true,
         httpCompression: {
           threshold: 1024,
@@ -245,7 +264,11 @@ export abstract class BaseServer extends EventEmitter {
   protected verify(req, upgrade, fn) {
     // transport check
     const transport = req._query.transport;
-    if (!~this.opts.transports.indexOf(transport)) {
+    // WebTransport does not go through the verify() method, see the onWebTransportSession() method
+    if (
+      !~this.opts.transports.indexOf(transport) ||
+      transport === "webtransport"
+    ) {
       debug('unknown transport "%s"', transport);
       return fn(Server.errors.UNKNOWN_TRANSPORT, { transport });
     }
@@ -495,6 +518,85 @@ export abstract class BaseServer extends EventEmitter {
     return transport;
   }
 
+  public async onWebTransportSession(session: any) {
+    const timeout = setTimeout(() => {
+      debug(
+        "the client failed to establish a bidirectional stream in the given period"
+      );
+      session.close();
+    }, this.opts.upgradeTimeout);
+
+    const streamReader = session.incomingBidirectionalStreams.getReader();
+    const result = await streamReader.read();
+
+    if (result.done) {
+      debug("session is closed");
+      return;
+    }
+
+    const stream = result.value;
+    const reader = stream.readable.getReader();
+
+    // reading the first packet of the stream
+    const { value, done } = await reader.read();
+    if (done) {
+      debug("stream is closed");
+      return;
+    }
+
+    clearTimeout(timeout);
+    const handshake = TEXT_DECODER.decode(value);
+
+    // handshake is either
+    // "0" => new session
+    // '0{"sid":"xxxx"}' => upgrade
+    if (handshake === "0") {
+      const transport = new WebTransport(session, stream, reader);
+
+      // note: we cannot use "this.generateId()", because there is no "req" argument
+      const id = base64id.generateId();
+      debug('handshaking client "%s" (WebTransport)', id);
+
+      const socket = new Socket(id, this, transport, null, 4);
+
+      this.clients[id] = socket;
+      this.clientsCount++;
+
+      socket.once("close", () => {
+        delete this.clients[id];
+        this.clientsCount--;
+      });
+
+      this.emit("connection", socket);
+      return;
+    }
+
+    const sid = parseSessionId(handshake);
+
+    if (!sid) {
+      debug("invalid WebTransport handshake");
+      return session.close();
+    }
+
+    const client = this.clients[sid];
+
+    if (!client) {
+      debug("upgrade attempt for closed client");
+      session.close();
+    } else if (client.upgrading) {
+      debug("transport has already been trying to upgrade");
+      session.close();
+    } else if (client.upgraded) {
+      debug("transport had already been upgraded");
+      session.close();
+    } else {
+      debug("upgrading existing transport");
+
+      const transport = new WebTransport(session, stream, reader);
+      client.maybeUpgrade(transport);
+    }
+  }
+
   protected abstract createTransport(transportName, req);
 
   /**

lib/socket.ts

@@ -70,10 +70,15 @@ export class Socket extends EventEmitter {
     this.protocol = protocol;
 
     // Cache IP since it might not be in the req later
-    if (req.websocket && req.websocket._socket) {
-      this.remoteAddress = req.websocket._socket.remoteAddress;
+    if (req) {
+      if (req.websocket && req.websocket._socket) {
+        this.remoteAddress = req.websocket._socket.remoteAddress;
+      } else {
+        this.remoteAddress = req.connection.remoteAddress;
+      }
     } else {
-      this.remoteAddress = req.connection.remoteAddress;
+      // TODO there is currently no way to get the IP address of the client when it connects with WebTransport
+      //  see https://github.com/fails-components/webtransport/issues/114
     }
 
     this.checkIntervalTimer = null;

lib/transport.ts

@@ -136,8 +136,26 @@ export abstract class Transport extends EventEmitter {
     this.emit("close");
   }
 
+  /**
+   * Advertise framing support.
+   */
   abstract get supportsFraming();
+
+  /**
+   * The name of the transport.
+   */
   abstract get name();
+
+  /**
+   * Sends an array of packets.
+   *
+   * @param {Array} packets
+   * @package
+   */
   abstract send(packets);
+
+  /**
+   * Closes the transport.
+   */
   abstract doClose(fn?);
 }

lib/transports-uws/polling.ts

@@ -162,7 +162,9 @@ export class Polling extends Transport {
     const onEnd = (buffer) => {
       this.onData(buffer.toString());
       this.onDataRequestCleanup();
-      res.end("ok");
+      res.cork(() => {
+        res.end("ok");
+      });
     };
 
     res.onAborted(() => {
@@ -312,10 +314,12 @@ export class Polling extends Transport {
 
     const respond = (data) => {
       this.headers(this.req, headers);
-      Object.keys(headers).forEach((key) => {
-        this.res.writeHeader(key, String(headers[key]));
+      this.res.cork(() => {
+        Object.keys(headers).forEach((key) => {
+          this.res.writeHeader(key, String(headers[key]));
+        });
+        this.res.end(data);
       });
-      this.res.end(data);
       callback();
     };
 

lib/transports/index.ts

@@ -1,10 +1,12 @@
 import { Polling as XHR } from "./polling";
 import { JSONP } from "./polling-jsonp";
 import { WebSocket } from "./websocket";
+import { WebTransport } from "./webtransport";
 
 export default {
   polling: polling,
   websocket: WebSocket,
+  webtransport: WebTransport,
 };
 
 /**
@@ -21,4 +23,4 @@ function polling(req) {
   }
 }
 
-polling.upgradesTo = ["websocket"];
+polling.upgradesTo = ["websocket", "webtransport"];