feat: US-191 - Add WebSocket project-matrix fixture using ws package

Author: NathanFlurry

packages/secure-exec-core/isolate-runtime/src/common/runtime-globals.d.ts

@@ -46,6 +46,9 @@ import type {
 	ChildProcessSpawnSyncBridgeRef,
 	ChildProcessStdinCloseBridgeRef,
 	ChildProcessStdinWriteBridgeRef,
+	UpgradeSocketWriteRawBridgeRef,
+	UpgradeSocketEndRawBridgeRef,
+	UpgradeSocketDestroyRawBridgeRef,
 } from "../../../src/shared/bridge-contract.js";
 
 type RuntimeGlobalExposer = (name: string, value: unknown) => void;
@@ -94,6 +97,9 @@ declare global {
 	var _networkHttpRequestRaw: NetworkHttpRequestRawBridgeRef;
 	var _networkHttpServerListenRaw: NetworkHttpServerListenRawBridgeRef;
 	var _networkHttpServerCloseRaw: NetworkHttpServerCloseRawBridgeRef;
+	var _upgradeSocketWriteRaw: UpgradeSocketWriteRawBridgeRef;
+	var _upgradeSocketEndRaw: UpgradeSocketEndRawBridgeRef;
+	var _upgradeSocketDestroyRaw: UpgradeSocketDestroyRawBridgeRef;
 	var _childProcessSpawnStart: ChildProcessSpawnStartBridgeRef;
 	var _childProcessStdinWrite: ChildProcessStdinWriteBridgeRef;
 	var _childProcessStdinClose: ChildProcessStdinCloseBridgeRef;

packages/secure-exec-core/src/bridge/network.ts

@@ -15,6 +15,9 @@ import type {
 	NetworkHttpServerListenRawBridgeRef,
 	RegisterHandleBridgeFn,
 	UnregisterHandleBridgeFn,
+	UpgradeSocketWriteRawBridgeRef,
+	UpgradeSocketEndRawBridgeRef,
+	UpgradeSocketDestroyRawBridgeRef,
 } from "../shared/bridge-contract.js";
 
 // Declare host bridge References
@@ -32,6 +35,18 @@ declare const _networkHttpServerCloseRaw:
   | NetworkHttpServerCloseRawBridgeRef
   | undefined;
 
+declare const _upgradeSocketWriteRaw:
+  | UpgradeSocketWriteRawBridgeRef
+  | undefined;
+
+declare const _upgradeSocketEndRaw:
+  | UpgradeSocketEndRawBridgeRef
+  | undefined;
+
+declare const _upgradeSocketDestroyRaw:
+  | UpgradeSocketDestroyRawBridgeRef
+  | undefined;
+
 declare const _registerHandle:
   | RegisterHandleBridgeFn
   | undefined;
@@ -745,15 +760,27 @@ export class ClientRequest {
         statusText?: string;
         body?: string;
         trailers?: Record<string, string>;
+        upgradeSocketId?: number;
       };
 
       this.finished = true;
 
       // 101 Switching Protocols → fire 'upgrade' event
       if (response.status === 101) {
         const res = new IncomingMessage(response);
-        const head = typeof Buffer !== "undefined" ? Buffer.alloc(0) : new Uint8Array(0);
-        this._emit("upgrade", res, this.socket, head);
+        // Use UpgradeSocket for bidirectional data relay when socketId is available
+        let socket: FakeSocket | UpgradeSocket = this.socket;
+        if (response.upgradeSocketId != null) {
+          socket = new UpgradeSocket(response.upgradeSocketId, {
+            host: this._options.hostname as string,
+            port: Number(this._options.port) || 80,
+          });
+          upgradeSocketInstances.set(response.upgradeSocketId, socket);
+        }
+        const head = typeof Buffer !== "undefined"
+          ? (response.body ? Buffer.from(response.body, "base64") : Buffer.alloc(0))
+          : new Uint8Array(0);
+        this._emit("upgrade", res, socket, head);
         return;
       }
 
@@ -1020,6 +1047,8 @@ const serverRequestListeners = new Map<
   number,
   (incoming: ServerIncomingMessage, outgoing: ServerResponseBridge) => unknown
 >();
+// Server instances indexed by serverId — used by upgrade dispatch to emit 'upgrade' events
+const serverInstances = new Map<number, Server>();
 
 class ServerIncomingMessage {
   headers: Record<string, string>;
@@ -1344,9 +1373,11 @@ class Server {
     } else {
       serverRequestListeners.set(this._serverId, () => undefined);
     }
+    serverInstances.set(this._serverId, this);
   }
 
-  private _emit(event: string, ...args: unknown[]): void {
+  /** @internal Emit an event — used by upgrade dispatch to fire 'upgrade' events. */
+  _emit(event: string, ...args: unknown[]): void {
     const listeners = this._listeners[event];
     if (!listeners || listeners.length === 0) return;
     listeners.slice().forEach((listener) => listener(...args));
@@ -1415,6 +1446,7 @@ class Server {
         }
         this.listening = false;
         this._address = null;
+        serverInstances.delete(this._serverId);
         if (this._handleId && typeof _unregisterHandle === "function") {
           _unregisterHandle(this._handleId);
         }
@@ -1534,6 +1566,203 @@ async function dispatchServerRequest(
   return JSON.stringify(outgoing.serialize());
 }
 
+// Upgrade socket for bidirectional data relay through the host bridge
+const upgradeSocketInstances = new Map<number, UpgradeSocket>();
+
+class UpgradeSocket {
+  remoteAddress: string;
+  remotePort: number;
+  localAddress = "127.0.0.1";
+  localPort = 0;
+  connecting = false;
+  destroyed = false;
+  writable = true;
+  readable = true;
+  readyState = "open";
+  bytesWritten = 0;
+  private _listeners: Record<string, EventListener[]> = {};
+  private _socketId: number;
+
+  // Readable stream state stub for ws compatibility (socketOnClose checks _readableState.endEmitted)
+  _readableState = { endEmitted: false };
+  _writableState = { finished: false, errorEmitted: false };
+
+  constructor(socketId: number, options?: { host?: string; port?: number }) {
+    this._socketId = socketId;
+    this.remoteAddress = options?.host || "127.0.0.1";
+    this.remotePort = options?.port || 80;
+  }
+
+  setTimeout(_ms: number, _cb?: () => void): this { return this; }
+  setNoDelay(_noDelay?: boolean): this { return this; }
+  setKeepAlive(_enable?: boolean, _delay?: number): this { return this; }
+  ref(): this { return this; }
+  unref(): this { return this; }
+  cork(): void {}
+  uncork(): void {}
+  pause(): this { return this; }
+  resume(): this { return this; }
+  address(): { address: string; family: string; port: number } {
+    return { address: this.localAddress, family: "IPv4", port: this.localPort };
+  }
+
+  on(event: string, listener: EventListener): this {
+    if (!this._listeners[event]) this._listeners[event] = [];
+    this._listeners[event].push(listener);
+    return this;
+  }
+
+  addListener(event: string, listener: EventListener): this {
+    return this.on(event, listener);
+  }
+
+  once(event: string, listener: EventListener): this {
+    const wrapper = (...args: unknown[]): void => {
+      this.off(event, wrapper);
+      listener(...args);
+    };
+    return this.on(event, wrapper);
+  }
+
+  off(event: string, listener: EventListener): this {
+    if (this._listeners[event]) {
+      const idx = this._listeners[event].indexOf(listener);
+      if (idx !== -1) this._listeners[event].splice(idx, 1);
+    }
+    return this;
+  }
+
+  removeListener(event: string, listener: EventListener): this {
+    return this.off(event, listener);
+  }
+
+  removeAllListeners(event?: string): this {
+    if (event) {
+      delete this._listeners[event];
+    } else {
+      this._listeners = {};
+    }
+    return this;
+  }
+
+  emit(event: string, ...args: unknown[]): boolean {
+    const handlers = this._listeners[event];
+    if (handlers) handlers.slice().forEach((fn) => fn.call(this, ...args));
+    return handlers !== undefined && handlers.length > 0;
+  }
+
+  listenerCount(event: string): number {
+    return this._listeners[event]?.length || 0;
+  }
+
+  // Allow arbitrary property assignment (used by ws for Symbol properties)
+  [key: string | symbol]: unknown;
+
+  write(data: unknown, encodingOrCb?: string | (() => void), cb?: (() => void)): boolean {
+    if (this.destroyed) return false;
+    const callback = typeof encodingOrCb === "function" ? encodingOrCb : cb;
+    if (typeof _upgradeSocketWriteRaw !== "undefined") {
+      let base64: string;
+      if (typeof Buffer !== "undefined" && Buffer.isBuffer(data)) {
+        base64 = data.toString("base64");
+      } else if (typeof data === "string") {
+        base64 = typeof Buffer !== "undefined" ? Buffer.from(data).toString("base64") : btoa(data);
+      } else if (data instanceof Uint8Array) {
+        base64 = typeof Buffer !== "undefined" ? Buffer.from(data).toString("base64") : btoa(String.fromCharCode(...data));
+      } else {
+        base64 = typeof Buffer !== "undefined" ? Buffer.from(String(data)).toString("base64") : btoa(String(data));
+      }
+      this.bytesWritten += base64.length;
+      _upgradeSocketWriteRaw.applySync(undefined, [this._socketId, base64]);
+    }
+    if (callback) callback();
+    return true;
+  }
+
+  end(data?: unknown): this {
+    if (data) this.write(data);
+    if (typeof _upgradeSocketEndRaw !== "undefined" && !this.destroyed) {
+      _upgradeSocketEndRaw.applySync(undefined, [this._socketId]);
+    }
+    this.writable = false;
+    this.emit("finish");
+    return this;
+  }
+
+  destroy(err?: Error): this {
+    if (this.destroyed) return this;
+    this.destroyed = true;
+    this.writable = false;
+    this.readable = false;
+    this._readableState.endEmitted = true;
+    this._writableState.finished = true;
+    if (typeof _upgradeSocketDestroyRaw !== "undefined") {
+      _upgradeSocketDestroyRaw.applySync(undefined, [this._socketId]);
+    }
+    upgradeSocketInstances.delete(this._socketId);
+    if (err) this.emit("error", err);
+    this.emit("close", false);
+    return this;
+  }
+
+  // Push data received from the host into this socket
+  _pushData(data: Buffer | Uint8Array): void {
+    this.emit("data", data);
+  }
+
+  // Signal end-of-stream from the host
+  _pushEnd(): void {
+    this.readable = false;
+    this._readableState.endEmitted = true;
+    this._writableState.finished = true;
+    this.emit("end");
+    this.emit("close", false);
+    upgradeSocketInstances.delete(this._socketId);
+  }
+}
+
+/** Route an incoming HTTP upgrade to the server's 'upgrade' event listeners. */
+function dispatchUpgradeRequest(
+  serverId: number,
+  requestJson: string,
+  headBase64: string,
+  socketId: number
+): void {
+  const server = serverInstances.get(serverId);
+  if (!server) {
+    throw new Error(`Unknown HTTP server for upgrade: ${serverId}`);
+  }
+
+  const request = JSON.parse(requestJson) as SerializedServerRequest;
+  const incoming = new ServerIncomingMessage(request);
+  const head = typeof Buffer !== "undefined" ? Buffer.from(headBase64, "base64") : new Uint8Array(0);
+
+  const socket = new UpgradeSocket(socketId, {
+    host: incoming.headers["host"]?.split(":")[0] || "127.0.0.1",
+  });
+  upgradeSocketInstances.set(socketId, socket);
+
+  // Emit 'upgrade' on the server — ws.WebSocketServer listens for this
+  server._emit("upgrade", incoming, socket, head);
+}
+
+/** Push data from host to an upgrade socket. */
+function onUpgradeSocketData(socketId: number, dataBase64: string): void {
+  const socket = upgradeSocketInstances.get(socketId);
+  if (socket) {
+    const data = typeof Buffer !== "undefined" ? Buffer.from(dataBase64, "base64") : new Uint8Array(0);
+    socket._pushData(data);
+  }
+}
+
+/** Signal end-of-stream from host to an upgrade socket. */
+function onUpgradeSocketEnd(socketId: number): void {
+  const socket = upgradeSocketInstances.get(socketId);
+  if (socket) {
+    socket._pushEnd();
+  }
+}
+
 // Function-based ServerResponse constructor — allows .call() inheritance
 // used by light-my-request (Fastify's inject), which does
 // http.ServerResponse.call(this, req) + util.inherits(Response, http.ServerResponse)
@@ -1693,6 +1922,9 @@ exposeCustomGlobal("_httpsModule", https);
 exposeCustomGlobal("_http2Module", http2);
 exposeCustomGlobal("_dnsModule", dns);
 exposeCustomGlobal("_httpServerDispatch", dispatchServerRequest);
+exposeCustomGlobal("_httpServerUpgradeDispatch", dispatchUpgradeRequest);
+exposeCustomGlobal("_upgradeSocketData", onUpgradeSocketData);
+exposeCustomGlobal("_upgradeSocketEnd", onUpgradeSocketEnd);
 
 // Harden fetch API globals (non-writable, non-configurable)
 exposeCustomGlobal("fetch", fetch);

packages/secure-exec-core/src/index.ts

@@ -148,6 +148,9 @@ export type {
 	NetworkHttpRequestRawBridgeRef,
 	NetworkHttpServerCloseRawBridgeRef,
 	NetworkHttpServerListenRawBridgeRef,
+	UpgradeSocketWriteRawBridgeRef,
+	UpgradeSocketEndRawBridgeRef,
+	UpgradeSocketDestroyRawBridgeRef,
 	ProcessErrorBridgeRef,
 	ProcessLogBridgeRef,
 	RegisterHandleBridgeFn,

packages/secure-exec-core/src/shared/bridge-contract.ts

@@ -63,6 +63,9 @@ export const HOST_BRIDGE_GLOBAL_KEYS = {
 	networkHttpRequestRaw: "_networkHttpRequestRaw",
 	networkHttpServerListenRaw: "_networkHttpServerListenRaw",
 	networkHttpServerCloseRaw: "_networkHttpServerCloseRaw",
+	upgradeSocketWriteRaw: "_upgradeSocketWriteRaw",
+	upgradeSocketEndRaw: "_upgradeSocketEndRaw",
+	upgradeSocketDestroyRaw: "_upgradeSocketDestroyRaw",
 	ptySetRawMode: "_ptySetRawMode",
 	processConfig: "_processConfig",
 	osConfig: "_osConfig",
@@ -85,6 +88,9 @@ export const RUNTIME_BRIDGE_GLOBAL_KEYS = {
 	http2Module: "_http2Module",
 	dnsModule: "_dnsModule",
 	httpServerDispatch: "_httpServerDispatch",
+	httpServerUpgradeDispatch: "_httpServerUpgradeDispatch",
+	upgradeSocketData: "_upgradeSocketData",
+	upgradeSocketEnd: "_upgradeSocketEnd",
 	fsFacade: "_fs",
 	requireFrom: "_requireFrom",
 	moduleCache: "_moduleCache",
@@ -246,6 +252,9 @@ export type NetworkDnsLookupRawBridgeRef = BridgeApplyRef<[string], string>;
 export type NetworkHttpRequestRawBridgeRef = BridgeApplyRef<[string, string], string>;
 export type NetworkHttpServerListenRawBridgeRef = BridgeApplyRef<[string], string>;
 export type NetworkHttpServerCloseRawBridgeRef = BridgeApplyRef<[number], void>;
+export type UpgradeSocketWriteRawBridgeRef = BridgeApplySyncRef<[number, string], void>;
+export type UpgradeSocketEndRawBridgeRef = BridgeApplySyncRef<[number], void>;
+export type UpgradeSocketDestroyRawBridgeRef = BridgeApplySyncRef<[number], void>;
 
 // PTY boundary contracts.
 export type PtySetRawModeBridgeRef = BridgeApplySyncRef<[boolean], void>;