up

Author: pi0

docs/1.guide/7.proxy.md

@@ -9,7 +9,7 @@ icon: tabler:arrows-exchange
 crossws ships a small helper that returns a set of ready-made hooks which proxy every peer to an upstream WebSocket server. Use it to put crossws in front of an existing backend, split traffic across services, or bridge protocols between runtimes.
 
 > [!NOTE]
-> The proxy uses the global [`WebSocket`](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket) constructor to dial the upstream, which is available on Node.js ≥ 22, Bun, Deno, Cloudflare Workers, and browsers. On older Node versions, provide a `WebSocket` polyfill on `globalThis`.
+> The proxy uses the global [`WebSocket`](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket) constructor to dial the upstream, which is available on Node.js ≥ 22, Bun, Deno, Cloudflare Workers, and browsers. On older Node versions, pass a custom constructor via the [`WebSocket` option](#custom-websocket-constructor) or install a polyfill on `globalThis`.
 
 ## Usage
 
@@ -61,6 +61,20 @@ createWebSocketProxy({
 > [!WARNING]
 > The proxy commits to a subprotocol in the upgrade response before the upstream connection is established. If the upstream ultimately picks a different subprotocol (or rejects), the client will still see the one the proxy promised. Only keep `forwardProtocol` enabled when the upstream is known to accept the same subprotocols the client negotiates.
 
+## Custom WebSocket constructor
+
+Pass a `WebSocket` constructor via options to override the global — useful on Node.js < 22, to plug in a different client implementation, or to stub the upstream in tests.
+
+```ts
+import { WebSocket } from "ws";
+import { createWebSocketProxy } from "crossws";
+
+const hooks = createWebSocketProxy({
+  target: "wss://backend.example.com",
+  WebSocket: WebSocket as unknown as typeof globalThis.WebSocket,
+});
+```
+
 ## Combining with custom hooks
 
 `createWebSocketProxy()` returns a plain hooks object, so you can spread it and override individual hooks — for example, to authenticate the upgrade request before proxying:
@@ -91,5 +105,6 @@ Accepts either a target URL (`string` or `URL`), a resolver function, or an opti
 - **`forwardProtocol`** — `boolean` (default `true`). When enabled, the client's `sec-websocket-protocol` header is forwarded to the upstream and echoed back in the upgrade response.
 - **`maxBufferSize`** — `number` (default `1048576`, i.e. 1 MiB). Maximum number of bytes buffered per peer while the upstream is still connecting. When exceeded, the peer is closed with code `1009` (Message Too Big). Set to `0` to disable.
 - **`connectTimeout`** — `number` (default `10000`). Milliseconds to wait for the upstream WebSocket handshake to complete. If exceeded, the peer is closed with code `1011`. Set to `0` to disable.
+- **`WebSocket`** — `typeof WebSocket` (default `globalThis.WebSocket`). Custom `WebSocket` constructor used to dial the upstream. Falls back to the global when omitted; throws at setup time if neither is available.
 
 Returns a `Partial<Hooks>` object containing `upgrade`, `open`, `message`, `close`, and `error` hooks.

src/proxy.ts

@@ -41,6 +41,16 @@ export interface WebSocketProxyOptions {
    * @default 10000
    */
   connectTimeout?: number;
+
+  /**
+   * Custom `WebSocket` constructor used to dial the upstream. Useful when
+   * the runtime does not expose a global `WebSocket` (Node.js < 22) or
+   * when you want to use a different client implementation (e.g. `ws`,
+   * `undici`, a mock for tests).
+   *
+   * @default globalThis.WebSocket
+   */
+  WebSocket?: typeof WebSocket;
 }
 
 /**
@@ -64,6 +74,13 @@ export function createWebSocketProxy(
       ? { target }
       : target;
 
+  const WebSocketCtor = options.WebSocket ?? globalThis.WebSocket;
+  if (typeof WebSocketCtor !== "function") {
+    throw new TypeError(
+      "createWebSocketProxy requires a `WebSocket` constructor. Pass one via the `WebSocket` option, or use a runtime that provides a global `WebSocket` (Node.js >= 22, Bun, Deno, Cloudflare Workers, browsers).",
+    );
+  }
+
   const upstreams = new Map<string, UpstreamState>();
 
   return {
@@ -82,7 +99,7 @@ export function createWebSocketProxy(
       const url = _resolveTarget(options.target, peer);
       const protocols = _resolveProtocols(peer, options.forwardProtocol);
 
-      const ws = new WebSocket(url, protocols);
+      const ws = new WebSocketCtor(url, protocols);
       ws.binaryType = "arraybuffer";
 
       const state: UpstreamState = {
@@ -123,7 +140,7 @@ export function createWebSocketProxy(
         // closures to the client.
         if (upstreams.get(peer.id) !== state) return;
         _cleanupState(upstreams, peer.id, state);
-        _safeClose(peer, _remapCloseCode(event.code), event.reason);
+        _safeClose(peer, _remapIncomingCode(event.code), event.reason);
       });
 
       ws.addEventListener("error", () => {
@@ -136,22 +153,29 @@ export function createWebSocketProxy(
     message(peer, message) {
       const state = upstreams.get(peer.id);
       if (!state) return;
-      const data =
+      const raw =
         typeof message.rawData === "string"
           ? message.rawData
           : message.uint8Array();
       if (state.open) {
-        state.ws.send(data);
+        try {
+          state.ws.send(raw);
+        } catch {
+          // upstream may have transitioned to CLOSING between the check and send
+        }
         return;
       }
-      const size = typeof data === "string" ? data.length : data.byteLength;
+      const size = typeof raw === "string" ? raw.length : raw.byteLength;
       const limit = options.maxBufferSize ?? DEFAULT_MAX_BUFFER_SIZE;
       if (limit > 0 && state.bufferSize + size > limit) {
         _cleanupState(upstreams, peer.id, state);
         _safeClose(peer, 1009, "Proxy buffer limit exceeded");
         return;
       }
-      state.buffer.push(data);
+      // Copy binary views before buffering: the adapter may own the backing
+      // memory (e.g. Node's `ws` reuses Buffers in some paths) and the buffer
+      // may be flushed asynchronously once the upstream is open.
+      state.buffer.push(typeof raw === "string" ? raw : Uint8Array.from(raw));
       state.bufferSize += size;
     },
 
@@ -162,7 +186,7 @@ export function createWebSocketProxy(
       upstreams.delete(peer.id);
       try {
         state.ws.close(
-          _remapCloseCode(details.code),
+          _normalizeOutgoingCode(details.code),
           _truncateReason(details.reason),
         );
       } catch {
@@ -262,11 +286,25 @@ function _truncateReason(reason?: string): string | undefined {
   );
 }
 
-// Reserved codes must never appear in an outbound close frame.
-// 1005 (no status), 1006 (abnormal), 1015 (TLS failure) get remapped.
-function _remapCloseCode(code?: number): number | undefined {
+// Upstream close event → peer.close. Reserved pseudo-codes (1005/1006/1015)
+// must never appear on the wire, so they are rewritten. Everything else is
+// forwarded as-is; server-side peers can use the full 1000-4999 range.
+/** @internal exported for tests */
+export function _remapIncomingCode(code?: number): number | undefined {
   if (code === undefined) return undefined;
   if (code === 1005) return 1000;
   if (code === 1006 || code === 1015) return 1011;
   return code;
 }
+
+// Peer close → upstream `state.ws.close`. The upstream is a client-side
+// WebSocket, and WHATWG restricts close() to 1000 or 3000-4999 — anything
+// else (1001 going-away, 1008 policy, etc.) throws InvalidAccessError.
+// Normalize to 1000 so we don't silently fail to close the upstream.
+/** @internal exported for tests */
+export function _normalizeOutgoingCode(code?: number): number | undefined {
+  if (code === undefined) return undefined;
+  if (code === 1000) return 1000;
+  if (code >= 3000 && code <= 4999) return code;
+  return 1000;
+}

test/proxy.test.ts

@@ -3,6 +3,7 @@ import { createServer, Server } from "node:http";
 import { getRandomPort, waitForPort } from "get-port-please";
 import nodeAdapter from "../src/adapters/node.ts";
 import { createWebSocketProxy, defineHooks } from "../src/index.ts";
+import { _normalizeOutgoingCode, _remapIncomingCode } from "../src/proxy.ts";
 import { wsConnect } from "./_utils.ts";
 
 describe("createWebSocketProxy", () => {
@@ -18,7 +19,6 @@ describe("createWebSocketProxy", () => {
   let limitedProxyURL: string;
   let badProxyURL: string;
   let timeoutProxyURL: string;
-
   beforeAll(async () => {
     // Upstream echo server (crossws node adapter)
     const upstream = nodeAdapter({
@@ -244,3 +244,31 @@ describe("createWebSocketProxy", () => {
     expect(event.code).toBe(1009);
   });
 });
+
+describe("createWebSocketProxy internals", () => {
+  test("_normalizeOutgoingCode allows 1000 and 3000-4999 range", () => {
+    // state.ws is a client-side WebSocket; WHATWG forbids anything else.
+    expect(_normalizeOutgoingCode(undefined)).toBeUndefined();
+    expect(_normalizeOutgoingCode(1000)).toBe(1000);
+    expect(_normalizeOutgoingCode(3000)).toBe(3000);
+    expect(_normalizeOutgoingCode(4999)).toBe(4999);
+  });
+
+  test("_normalizeOutgoingCode rewrites reserved and disallowed codes to 1000", () => {
+    // Regression: state.ws.close(1005) would throw InvalidAccessError,
+    // leaking the upstream socket. 1001/1008 are valid server-side codes
+    // but still forbidden for client-side close(), so they also normalize.
+    for (const code of [1001, 1005, 1006, 1008, 1011, 1015, 2999, 5000]) {
+      expect(_normalizeOutgoingCode(code)).toBe(1000);
+    }
+  });
+
+  test("_remapIncomingCode rewrites reserved pseudo-codes before peer close", () => {
+    expect(_remapIncomingCode(undefined)).toBeUndefined();
+    expect(_remapIncomingCode(1000)).toBe(1000);
+    expect(_remapIncomingCode(1005)).toBe(1000);
+    expect(_remapIncomingCode(1006)).toBe(1011);
+    expect(_remapIncomingCode(1015)).toBe(1011);
+    expect(_remapIncomingCode(4321)).toBe(4321);
+  });
+});