feat: add support for Express middlewares

This commit implements middlewares at the Engine.IO level, because
Socket.IO middlewares are meant for namespace authorization and are not
executed during a classic HTTP request/response cycle.

A workaround was possible by using the allowRequest option and the
"headers" event, but this feels way cleaner and works with upgrade
requests too.

Syntax:

```js
engine.use((req, res, next) => {
  // do something

  next();
});

// with express-session
import session from "express-session";

engine.use(session({
  secret: "keyboard cat",
  resave: false,
  saveUninitialized: true,
  cookie: { secure: true }
});

// with helmet
import helmet from "helmet";

engine.use(helmet());
```

Related:

- #668
- #651
- socketio/socket.io#4609
- socketio/socket.io#3933
- a lot of other issues asking for compatibility with express-session

Author: darrachequesne

lib/server.ts

@@ -7,12 +7,19 @@ import { Socket } from "./socket";
 import debugModule from "debug";
 import { serialize } from "cookie";
 import { Server as DEFAULT_WS_ENGINE } from "ws";
-import { IncomingMessage, Server as HttpServer } from "http";
-import { CookieSerializeOptions } from "cookie";
-import { CorsOptions, CorsOptionsDelegate } from "cors";
+import type {
+  IncomingMessage,
+  Server as HttpServer,
+  ServerResponse,
+} from "http";
+import type { CookieSerializeOptions } from "cookie";
+import type { CorsOptions, CorsOptionsDelegate } from "cors";
+import type { Duplex } from "stream";
 
 const debug = debugModule("engine");
 
+const kResponseHeaders = Symbol("responseHeaders");
+
 type Transport = "polling" | "websocket";
 
 export interface AttachOptions {
@@ -119,12 +126,26 @@ export interface ServerOptions {
   allowEIO3?: boolean;
 }
 
+/**
+ * An Express-compatible middleware.
+ *
+ * Middleware functions are functions that have access to the request object (req), the response object (res), and the
+ * next middleware function in the application’s request-response cycle.
+ *
+ * @see https://expressjs.com/en/guide/using-middleware.html
+ */
+type Middleware = (
+  req: IncomingMessage,
+  res: ServerResponse,
+  next: () => void
+) => void;
+
 export abstract class BaseServer extends EventEmitter {
   public opts: ServerOptions;
 
   protected clients: any;
   private clientsCount: number;
-  protected corsMiddleware: Function;
+  protected middlewares: Middleware[] = [];
 
   /**
    * Server constructor.
@@ -170,7 +191,7 @@ export abstract class BaseServer extends EventEmitter {
     }
 
     if (this.opts.cors) {
-      this.corsMiddleware = require("cors")(this.opts.cors);
+      this.use(require("cors")(this.opts.cors));
     }
 
     if (opts.perMessageDeflate) {
@@ -289,6 +310,52 @@ export abstract class BaseServer extends EventEmitter {
     fn();
   }
 
+  /**
+   * Adds a new middleware.
+   *
+   * @example
+   * import helmet from "helmet";
+   *
+   * engine.use(helmet());
+   *
+   * @param fn
+   */
+  public use(fn: Middleware) {
+    this.middlewares.push(fn);
+  }
+
+  /**
+   * Apply the middlewares to the request.
+   *
+   * @param req
+   * @param res
+   * @param callback
+   * @protected
+   */
+  protected _applyMiddlewares(
+    req: IncomingMessage,
+    res: ServerResponse,
+    callback: () => void
+  ) {
+    if (this.middlewares.length === 0) {
+      debug("no middleware to apply, skipping");
+      return callback();
+    }
+
+    const apply = (i) => {
+      debug("applying middleware n°%d", i + 1);
+      this.middlewares[i](req, res, () => {
+        if (i + 1 < this.middlewares.length) {
+          apply(i + 1);
+        } else {
+          callback();
+        }
+      });
+    };
+
+    apply(0);
+  }
+
   /**
    * Closes all clients.
    *
@@ -449,6 +516,40 @@ export abstract class BaseServer extends EventEmitter {
   };
 }
 
+/**
+ * Exposes a subset of the http.ServerResponse interface, in order to be able to apply the middlewares to an upgrade
+ * request.
+ *
+ * @see https://nodejs.org/api/http.html#class-httpserverresponse
+ */
+class WebSocketResponse {
+  constructor(readonly req, readonly socket: Duplex) {
+    // temporarily store the response headers on the req object (see the "headers" event)
+    req[kResponseHeaders] = {};
+  }
+
+  public setHeader(name: string, value: any) {
+    this.req[kResponseHeaders][name] = value;
+  }
+
+  public getHeader(name: string) {
+    return this.req[kResponseHeaders][name];
+  }
+
+  public removeHeader(name: string) {
+    delete this.req[kResponseHeaders][name];
+  }
+
+  public write() {}
+
+  public writeHead() {}
+
+  public end() {
+    // we could return a proper error code, but the WebSocket client will emit an "error" event anyway.
+    this.socket.destroy();
+  }
+}
+
 export class Server extends BaseServer {
   public httpServer?: HttpServer;
   private ws: any;
@@ -474,7 +575,8 @@ export class Server extends BaseServer {
       this.ws.on("headers", (headersArray, req) => {
         // note: 'ws' uses an array of headers, while Engine.IO uses an object (response.writeHead() accepts both formats)
         // we could also try to parse the array and then sync the values, but that will be error-prone
-        const additionalHeaders = {};
+        const additionalHeaders = req[kResponseHeaders] || {};
+        delete req[kResponseHeaders];
 
         const isInitialRequest = !req._query.sid;
         if (isInitialRequest) {
@@ -483,6 +585,7 @@ export class Server extends BaseServer {
 
         this.emit("headers", additionalHeaders, req);
 
+        debug("writing headers: %j", additionalHeaders);
         Object.keys(additionalHeaders).forEach((key) => {
           headersArray.push(`${key}: ${additionalHeaders[key]}`);
         });
@@ -517,13 +620,14 @@ export class Server extends BaseServer {
   /**
    * Handles an Engine.IO HTTP request.
    *
-   * @param {http.IncomingMessage} request
-   * @param {http.ServerResponse|http.OutgoingMessage} response
+   * @param {IncomingMessage} req
+   * @param {ServerResponse} res
    * @api public
    */
-  public handleRequest(req, res) {
+  public handleRequest(req: IncomingMessage, res: ServerResponse) {
     debug('handling "%s" http request "%s"', req.method, req.url);
     this.prepare(req);
+    // @ts-ignore
     req.res = res;
 
     const callback = (errorCode, errorContext) => {
@@ -538,51 +642,62 @@ export class Server extends BaseServer {
         return;
       }
 
+      // @ts-ignore
       if (req._query.sid) {
         debug("setting new request for existing client");
+        // @ts-ignore
         this.clients[req._query.sid].transport.onRequest(req);
       } else {
         const closeConnection = (errorCode, errorContext) =>
           abortRequest(res, errorCode, errorContext);
+        // @ts-ignore
         this.handshake(req._query.transport, req, closeConnection);
       }
     };
 
-    if (this.corsMiddleware) {
-      this.corsMiddleware.call(null, req, res, () => {
-        this.verify(req, false, callback);
-      });
-    } else {
+    this._applyMiddlewares(req, res, () => {
       this.verify(req, false, callback);
-    }
+    });
   }
 
   /**
    * Handles an Engine.IO HTTP Upgrade.
    *
    * @api public
    */
-  public handleUpgrade(req, socket, upgradeHead) {
+  public handleUpgrade(
+    req: IncomingMessage,
+    socket: Duplex,
+    upgradeHead: Buffer
+  ) {
     this.prepare(req);
 
-    this.verify(req, true, (errorCode, errorContext) => {
-      if (errorCode) {
-        this.emit("connection_error", {
-          req,
-          code: errorCode,
-          message: Server.errorMessages[errorCode],
-          context: errorContext,
-        });
-        abortUpgrade(socket, errorCode, errorContext);
-        return;
-      }
+    const res = new WebSocketResponse(req, socket);
 
-      const head = Buffer.from(upgradeHead);
-      upgradeHead = null;
+    this._applyMiddlewares(req, res as unknown as ServerResponse, () => {
+      this.verify(req, true, (errorCode, errorContext) => {
+        if (errorCode) {
+          this.emit("connection_error", {
+            req,
+            code: errorCode,
+            message: Server.errorMessages[errorCode],
+            context: errorContext,
+          });
+          abortUpgrade(socket, errorCode, errorContext);
+          return;
+        }
 
-      // delegate to ws
-      this.ws.handleUpgrade(req, socket, head, (websocket) => {
-        this.onWebSocket(req, socket, websocket);
+        const head = Buffer.from(upgradeHead);
+        upgradeHead = null;
+
+        // some middlewares (like express-session) wait for the writeHead() call to flush their headers
+        // see https://github.com/expressjs/session/blob/1010fadc2f071ddf2add94235d72224cf65159c6/index.js#L220-L244
+        res.writeHead();
+
+        // delegate to ws
+        this.ws.handleUpgrade(req, socket, head, (websocket) => {
+          this.onWebSocket(req, socket, websocket);
+        });
       });
     });
   }

lib/userver.ts

@@ -34,6 +34,7 @@ export class uServer extends BaseServer {
    */
   private prepare(req, res: HttpResponse) {
     req.method = req.getMethod().toUpperCase();
+    req.url = req.getUrl();
 
     const params = new URLSearchParams(req.getQuery());
     req._query = Object.fromEntries(params.entries());
@@ -91,6 +92,23 @@ export class uServer extends BaseServer {
       });
   }
 
+  override _applyMiddlewares(req: any, res: any, callback: () => void): void {
+    if (this.middlewares.length === 0) {
+      return callback();
+    }
+
+    // needed to buffer headers until the status is computed
+    req.res = new ResponseWrapper(res);
+
+    super._applyMiddlewares(req, req.res, () => {
+      // some middlewares (like express-session) wait for the writeHead() call to flush their headers
+      // see https://github.com/expressjs/session/blob/1010fadc2f071ddf2add94235d72224cf65159c6/index.js#L220-L244
+      req.res.writeHead();
+
+      callback();
+    });
+  }
+
   private handleRequest(
     res: HttpResponse,
     req: HttpRequest & { res: any; _query: any }
@@ -100,104 +118,99 @@ export class uServer extends BaseServer {
 
     req.res = res;
 
-    const callback = (errorCode, errorContext) => {
-      if (errorCode !== undefined) {
-        this.emit("connection_error", {
-          req,
-          code: errorCode,
-          message: Server.errorMessages[errorCode],
-          context: errorContext,
-        });
-        this.abortRequest(req.res, errorCode, errorContext);
-        return;
-      }
-
-      if (req._query.sid) {
-        debug("setting new request for existing client");
-        this.clients[req._query.sid].transport.onRequest(req);
-      } else {
-        const closeConnection = (errorCode, errorContext) =>
-          this.abortRequest(res, errorCode, errorContext);
-        this.handshake(req._query.transport, req, closeConnection);
-      }
-    };
-
-    if (this.corsMiddleware) {
-      // needed to buffer headers until the status is computed
-      req.res = new ResponseWrapper(res);
+    this._applyMiddlewares(req, res, () => {
+      this.verify(req, false, (errorCode, errorContext) => {
+        if (errorCode !== undefined) {
+          this.emit("connection_error", {
+            req,
+            code: errorCode,
+            message: Server.errorMessages[errorCode],
+            context: errorContext,
+          });
+          this.abortRequest(req.res, errorCode, errorContext);
+          return;
+        }
 
-      this.corsMiddleware.call(null, req, req.res, () => {
-        this.verify(req, false, callback);
+        if (req._query.sid) {
+          debug("setting new request for existing client");
+          this.clients[req._query.sid].transport.onRequest(req);
+        } else {
+          const closeConnection = (errorCode, errorContext) =>
+            this.abortRequest(res, errorCode, errorContext);
+          this.handshake(req._query.transport, req, closeConnection);
+        }
       });
-    } else {
-      this.verify(req, false, callback);
-    }
+    });
   }
 
   private handleUpgrade(
     res: HttpResponse,
-    req: HttpRequest & { _query: any },
+    req: HttpRequest & { res: any; _query: any },
     context
   ) {
     debug("on upgrade");
 
     this.prepare(req, res);
 
-    // @ts-ignore
     req.res = res;
 
-    this.verify(req, true, async (errorCode, errorContext) => {
-      if (errorCode) {
-        this.emit("connection_error", {
-          req,
-          code: errorCode,
-          message: Server.errorMessages[errorCode],
-          context: errorContext,
-        });
-        this.abortRequest(res, errorCode, errorContext);
-        return;
-      }
-
-      const id = req._query.sid;
-      let transport;
-
-      if (id) {
-        const client = this.clients[id];
-        if (!client) {
-          debug("upgrade attempt for closed client");
-          res.close();
-        } else if (client.upgrading) {
-          debug("transport has already been trying to upgrade");
-          res.close();
-        } else if (client.upgraded) {
-          debug("transport had already been upgraded");
-          res.close();
-        } else {
-          debug("upgrading existing transport");
-          transport = this.createTransport(req._query.transport, req);
-          client.maybeUpgrade(transport);
-        }
-      } else {
-        transport = await this.handshake(
-          req._query.transport,
-          req,
-          (errorCode, errorContext) =>
-            this.abortRequest(res, errorCode, errorContext)
-        );
-        if (!transport) {
+    this._applyMiddlewares(req, res, () => {
+      this.verify(req, true, async (errorCode, errorContext) => {
+        if (errorCode) {
+          this.emit("connection_error", {
+            req,
+            code: errorCode,
+            message: Server.errorMessages[errorCode],
+            context: errorContext,
+          });
+          this.abortRequest(res, errorCode, errorContext);
           return;
         }
-      }
 
-      res.upgrade(
-        {
-          transport,
-        },
-        req.getHeader("sec-websocket-key"),
-        req.getHeader("sec-websocket-protocol"),
-        req.getHeader("sec-websocket-extensions"),
-        context
-      );
+        const id = req._query.sid;
+        let transport;
+
+        if (id) {
+          const client = this.clients[id];
+          if (!client) {
+            debug("upgrade attempt for closed client");
+            res.close();
+          } else if (client.upgrading) {
+            debug("transport has already been trying to upgrade");
+            res.close();
+          } else if (client.upgraded) {
+            debug("transport had already been upgraded");
+            res.close();
+          } else {
+            debug("upgrading existing transport");
+            transport = this.createTransport(req._query.transport, req);
+            client.maybeUpgrade(transport);
+          }
+        } else {
+          transport = await this.handshake(
+            req._query.transport,
+            req,
+            (errorCode, errorContext) =>
+              this.abortRequest(res, errorCode, errorContext)
+          );
+          if (!transport) {
+            return;
+          }
+        }
+
+        // calling writeStatus() triggers the flushing of any header added in a middleware
+        req.res.writeStatus("101 Switching Protocols");
+
+        res.upgrade(
+          {
+            transport,
+          },
+          req.getHeader("sec-websocket-key"),
+          req.getHeader("sec-websocket-protocol"),
+          req.getHeader("sec-websocket-extensions"),
+          context
+        );
+      });
     });
   }
 
@@ -233,11 +246,29 @@ class ResponseWrapper {
   constructor(readonly res: HttpResponse) {}
 
   public set statusCode(status: number) {
+    if (!status) {
+      return;
+    }
+    // FIXME: handle all status codes?
     this.writeStatus(status === 200 ? "200 OK" : "204 No Content");
   }
 
+  public writeHead(status: number) {
+    this.statusCode = status;
+  }
+
   public setHeader(key, value) {
-    this.writeHeader(key, value);
+    if (Array.isArray(value)) {
+      value.forEach((val) => {
+        this.writeHeader(key, val);
+      });
+    } else {
+      this.writeHeader(key, value);
+    }
+  }
+
+  public removeHeader() {
+    // FIXME: not implemented
   }
 
   // needed by vary: https://github.com/jshttp/vary/blob/5d725d059b3871025cf753e9dfa08924d0bcfa8f/index.js#L134

package-lock.json

@@ -48,6 +48,8 @@
     "engine.io-client": "6.3.0",
     "engine.io-client-v3": "npm:engine.io-client@3.5.2",
     "expect.js": "^0.3.1",
+    "express-session": "^1.17.3",
+    "helmet": "^6.0.1",
     "mocha": "^9.1.3",
     "prettier": "^2.8.2",
     "rimraf": "^3.0.2",

package.json

@@ -0,0 +1,250 @@
+const listen = require("./common").listen;
+const expect = require("expect.js");
+const request = require("superagent");
+const { WebSocket } = require("ws");
+const helmet = require("helmet");
+const session = require("express-session");
+
+describe("middlewares", () => {
+  it("should apply middleware (polling)", (done) => {
+    const engine = listen((port) => {
+      engine.use((req, res, next) => {
+        res.setHeader("foo", "bar");
+        next();
+      });
+
+      request
+        .get(`http://localhost:${port}/engine.io/`)
+        .query({ EIO: 4, transport: "polling" })
+        .end((err, res) => {
+          expect(err).to.be(null);
+          expect(res.status).to.eql(200);
+          expect(res.headers["foo"]).to.eql("bar");
+
+          if (engine.httpServer) {
+            engine.httpServer.close();
+          }
+          done();
+        });
+    });
+  });
+
+  it("should apply middleware (websocket)", (done) => {
+    const engine = listen((port) => {
+      engine.use((req, res, next) => {
+        res.setHeader("foo", "bar");
+        next();
+      });
+
+      const socket = new WebSocket(
+        `ws://localhost:${port}/engine.io/?EIO=4&transport=websocket`
+      );
+
+      socket.on("upgrade", (res) => {
+        expect(res.headers["foo"]).to.eql("bar");
+
+        if (engine.httpServer) {
+          engine.httpServer.close();
+        }
+        done();
+      });
+
+      socket.on("open", () => {
+        socket.close();
+      });
+    });
+  });
+
+  it("should apply all middlewares in order", (done) => {
+    const engine = listen((port) => {
+      let count = 0;
+
+      engine.use((req, res, next) => {
+        expect(++count).to.eql(1);
+        next();
+      });
+
+      engine.use((req, res, next) => {
+        expect(++count).to.eql(2);
+        next();
+      });
+
+      engine.use((req, res, next) => {
+        expect(++count).to.eql(3);
+        next();
+      });
+
+      request
+        .get(`http://localhost:${port}/engine.io/`)
+        .query({ EIO: 4, transport: "polling" })
+        .end((err, res) => {
+          expect(err).to.be(null);
+          expect(res.status).to.eql(200);
+
+          if (engine.httpServer) {
+            engine.httpServer.close();
+          }
+          done();
+        });
+    });
+  });
+
+  it("should end the request (polling)", function (done) {
+    if (process.env.EIO_WS_ENGINE === "uws") {
+      return this.skip();
+    }
+    const engine = listen((port) => {
+      engine.use((req, res, _next) => {
+        res.writeHead(503);
+        res.end();
+      });
+
+      engine.on("connection", () => {
+        done(new Error("should not happen"));
+      });
+
+      request
+        .get(`http://localhost:${port}/engine.io/`)
+        .query({ EIO: 4, transport: "polling" })
+        .end((err, res) => {
+          expect(err).to.be.an(Error);
+          expect(res.status).to.eql(503);
+
+          if (engine.httpServer) {
+            engine.httpServer.close();
+          }
+          done();
+        });
+    });
+  });
+
+  it("should end the request (websocket)", (done) => {
+    const engine = listen((port) => {
+      engine.use((req, res, _next) => {
+        res.writeHead(503);
+        res.end();
+      });
+
+      engine.on("connection", () => {
+        done(new Error("should not happen"));
+      });
+
+      const socket = new WebSocket(
+        `ws://localhost:${port}/engine.io/?EIO=4&transport=websocket`
+      );
+
+      socket.addEventListener("error", () => {
+        if (engine.httpServer) {
+          engine.httpServer.close();
+        }
+        done();
+      });
+    });
+  });
+
+  it("should work with helmet (polling)", (done) => {
+    const engine = listen((port) => {
+      engine.use(helmet());
+
+      request
+        .get(`http://localhost:${port}/engine.io/`)
+        .query({ EIO: 4, transport: "polling" })
+        .end((err, res) => {
+          expect(err).to.be(null);
+          expect(res.status).to.eql(200);
+          expect(res.headers["x-download-options"]).to.eql("noopen");
+          expect(res.headers["x-content-type-options"]).to.eql("nosniff");
+
+          if (engine.httpServer) {
+            engine.httpServer.close();
+          }
+          done();
+        });
+    });
+  });
+
+  it("should work with helmet (websocket)", (done) => {
+    const engine = listen((port) => {
+      engine.use(helmet());
+
+      const socket = new WebSocket(
+        `ws://localhost:${port}/engine.io/?EIO=4&transport=websocket`
+      );
+
+      socket.on("upgrade", (res) => {
+        expect(res.headers["x-download-options"]).to.eql("noopen");
+        expect(res.headers["x-content-type-options"]).to.eql("nosniff");
+
+        if (engine.httpServer) {
+          engine.httpServer.close();
+        }
+        done();
+      });
+
+      socket.on("open", () => {
+        socket.close();
+      });
+    });
+  });
+
+  it("should work with express-session (polling)", (done) => {
+    const engine = listen((port) => {
+      engine.use(
+        session({
+          secret: "keyboard cat",
+          resave: false,
+          saveUninitialized: true,
+          cookie: {},
+        })
+      );
+
+      request
+        .get(`http://localhost:${port}/engine.io/`)
+        .query({ EIO: 4, transport: "polling" })
+        .end((err, res) => {
+          expect(err).to.be(null);
+          // expect(res.status).to.eql(200);
+          expect(res.headers["set-cookie"][0].startsWith("connect.sid=")).to.be(
+            true
+          );
+
+          if (engine.httpServer) {
+            engine.httpServer.close();
+          }
+          done();
+        });
+    });
+  });
+
+  it("should work with express-session (websocket)", (done) => {
+    const engine = listen((port) => {
+      engine.use(
+        session({
+          secret: "keyboard cat",
+          resave: false,
+          saveUninitialized: true,
+          cookie: {},
+        })
+      );
+
+      const socket = new WebSocket(
+        `ws://localhost:${port}/engine.io/?EIO=4&transport=websocket`
+      );
+
+      socket.on("upgrade", (res) => {
+        expect(res.headers["set-cookie"][0].startsWith("connect.sid=")).to.be(
+          true
+        );
+
+        if (engine.httpServer) {
+          engine.httpServer.close();
+        }
+        done();
+      });
+
+      socket.on("open", () => {
+        socket.close();
+      });
+    });
+  });
+});