Tunnel WebSockets end to end

.github/workflows/ci.yml

@@ -23,6 +23,8 @@ jobs:
       - run: pnpm install
       - run: pnpm build
       - run: pnpm test
+      # Example tests need the pinned Bun/Deno versions above; local runs skip them.
+      - run: pnpm test:examples
       - run: pnpm lint
       - name: arethetypeswrong
         run: |

README.md

@@ -65,6 +65,37 @@ await new Promise(() => {}); // stay alive until killed
 
 That's all you need! No local ports, just a fetch function.
 
+### WebSockets
+
+Tunnels forward WebSockets too: `npx captun 3000` exposes any local WebSocket
+server (socket.io, `ws`, Bun, Deno, ...) with handshake headers, subprotocols,
+binary messages, and close codes passing through. In code, a fetch handler
+accepts WebSockets Workers-style on any runtime:
+
+```ts
+import {
+  createCaptunTunnel,
+  createWebSocketResponse,
+  isWebSocketUpgradeRequest,
+  WebSocketPair,
+} from "captun";
+
+await createCaptunTunnel({
+  fetch(request) {
+    if (!isWebSocketUpgradeRequest(request)) return new Response("hello");
+    const pair = new WebSocketPair();
+    pair[1].accept();
+    pair[1].addEventListener("message", (event) => pair[1].send(`echo:${event.data}`));
+    return createWebSocketResponse(pair[0]);
+  },
+});
+```
+
+Connections are relayed message by message over the tunnel, so ping/pong and
+compression are per-hop, close codes outside 1000/3000–4999 degrade to a plain
+close, and messages are capped at 16MiB so one oversized frame can't take down
+the tunnel.
+
 ### Vite plugin
 
 `captun/vite` serves your Vite dev server (and `vite preview`) through a public tunnel URL — handy for receiving webhooks against local code, sharing work in progress, or pointing remote devices and agents at your dev server.

package.json

@@ -87,7 +87,8 @@
     "deploy:worker": "pnpm run build:hosted-browser-module && wrangler deploy",
     "deploy:hosted": "pnpm run build:hosted-browser-module && wrangler deploy --config wrangler.hosted.jsonc",
     "dev": "wrangler dev",
-    "test": "vitest run",
+    "test": "vitest run test/",
+    "test:examples": "vitest run examples/",
     "test:unit": "vitest run test/worker.test.ts",
     "cli": "tsx src/cli/bin.ts",
     "smoke": "./scripts/smoke-test.sh",
@@ -126,5 +127,8 @@
       "optional": true
     }
   },
+  "engines": {
+    "node": ">=22.4"
+  },
   "packageManager": "pnpm@10.11.1"
 }

src/cli/bin.ts

@@ -17,7 +17,14 @@ import {
   CaptunTunnelConnectError,
   createCaptunTunnel,
   HOSTED_CAPTUN_GATEWAY,
+  isWebSocketUpgradeRequest,
+  pipeWebSocketToHandle,
   randomConnectToken,
+  webSocketHandleFromSocket,
+  type Fetcher,
+  type WebSocketConnectResult,
+  type WebSocketFetcher,
+  type WebSocketHandle,
 } from "../index.js";
 import { assertLocalTargetAcceptingConnections } from "./local-target.js";
 import { withSpinner } from "./spinner.js";
@@ -522,7 +529,8 @@ async function connectTunnelWithRetry(
           gateway: tunnel.gateway,
           name: tunnel.name,
           token: tunnel.token,
-          fetch: fetcher,
+          fetch: fetcher.fetch,
+          connectWebSocket: fetcher.connectWebSocket,
         }),
       );
     } catch (error) {
@@ -536,42 +544,144 @@ async function connectTunnelWithRetry(
   throw new Error("unreachable");
 }
 
-function makeTunnelFetcher(tunnel: ResolvedTunnel) {
-  return async (request: Request) => {
-    if (isCaptunHealthRequest(request)) return captunHealthResponse();
+function makeTunnelFetcher(tunnel: ResolvedTunnel): Fetcher & WebSocketFetcher {
+  return {
+    fetch: async (request: Request) => {
+      if (isWebSocketUpgradeRequest(request)) {
+        return new Response("Use connectWebSocket for WebSocket tunnel requests\n", {
+          status: 400,
+        });
+      }
 
-    const url = new URL(request.url);
-    const requestStartedAt = performance.now();
-    const rayId = request.headers.get("cf-ray") || "-";
-    try {
-      const response = await fetch(
-        new Request(`${tunnel.target}${url.pathname}${url.search}`, request),
-      );
-      logRequest(tunnel.requestLogs, {
-        method: request.method,
-        path: `${url.pathname}${url.search}`,
-        rayId,
-        status: response.status,
-        startedAt: requestStartedAt,
-      });
-      return response;
-    } catch {
-      const response = new Response(
-        `Request reached the captun cli, but ${tunnel.target} is not accepting connections\n`,
+      if (isCaptunHealthRequest(request)) return captunHealthResponse();
+
+      const url = new URL(request.url);
+      const requestStartedAt = performance.now();
+      const rayId = request.headers.get("cf-ray") || "-";
+      try {
+        const response = await fetch(
+          new Request(`${tunnel.target}${url.pathname}${url.search}`, request),
+        );
+        logRequest(tunnel.requestLogs, {
+          method: request.method,
+          path: `${url.pathname}${url.search}`,
+          rayId,
+          status: response.status,
+          startedAt: requestStartedAt,
+        });
+        return response;
+      } catch {
+        const response = new Response(
+          `Request reached the captun cli, but ${tunnel.target} is not accepting connections\n`,
+          { status: 502 },
+        );
+        logRequest(tunnel.requestLogs, {
+          method: request.method,
+          path: `${url.pathname}${url.search}`,
+          rayId,
+          status: response.status,
+          startedAt: requestStartedAt,
+        });
+        return response;
+      }
+    },
+
+    connectWebSocket: (request, remote) => connectTargetWebSocket(tunnel, request, remote),
+  };
+}
+
+async function connectTargetWebSocket(
+  tunnel: ResolvedTunnel,
+  request: Request,
+  remote: WebSocketHandle,
+): Promise<WebSocketConnectResult> {
+  const url = new URL(request.url);
+  const requestStartedAt = performance.now();
+  const log = (status: number) =>
+    logRequest(tunnel.requestLogs, {
+      method: request.method,
+      path: `${url.pathname}${url.search}`,
+      rayId: request.headers.get("cf-ray") || "-",
+      status,
+      startedAt: requestStartedAt,
+    });
+
+  const targetUrl = new URL(`${tunnel.target}${url.pathname}${url.search}`);
+  targetUrl.protocol = targetUrl.protocol === "https:" ? "wss:" : "ws:";
+  const targetSocket = new WebSocket(targetUrl, {
+    protocols: request.headers
+      .get("sec-websocket-protocol")
+      ?.split(",")
+      .map((protocol) => protocol.trim())
+      .filter(Boolean),
+    headers: forwardedHandshakeHeaders(request.headers),
+    // Node's WebSocket (undici) accepts { protocols, headers }; the DOM type doesn't.
+  } as unknown as string[]);
+
+  try {
+    await waitForWebSocketOpen(targetSocket);
+    // The local server may have closed right after the handshake; reject the
+    // public upgrade cleanly instead of accepting an already-dead socket.
+    if (targetSocket.readyState !== WebSocket.OPEN) throw new Error("WebSocket closed after open");
+  } catch {
+    targetSocket.close();
+    log(502);
+    return {
+      accepted: false,
+      response: new Response(
+        `Request reached the captun cli, but ${targetUrl.origin} did not accept the WebSocket\n`,
         { status: 502 },
-      );
-      logRequest(tunnel.requestLogs, {
-        method: request.method,
-        path: `${url.pathname}${url.search}`,
-        rayId,
-        status: response.status,
-        startedAt: requestStartedAt,
-      });
-      return response;
-    }
+      ),
+    };
+  }
+
+  log(101);
+  pipeWebSocketToHandle(targetSocket, remote);
+  return {
+    accepted: true,
+    protocol: targetSocket.protocol || undefined,
+    socket: webSocketHandleFromSocket(targetSocket),
   };
 }
 
+/**
+ * Handshake headers to forward to the local WebSocket server, so cookie or
+ * token auth behaves the same as tunneled HTTP requests. Hop-by-hop headers
+ * and the Sec-WebSocket-* family stay out: the local WebSocket client
+ * negotiates its own handshake (subprotocols are passed separately).
+ */
+function forwardedHandshakeHeaders(headers: Headers) {
+  const skip = new Set(["connection", "host", "keep-alive", "te", "trailer", "upgrade"]);
+  const forwarded: Record<string, string> = {};
+  for (const [name, value] of headers) {
+    if (skip.has(name) || name.startsWith("sec-websocket-") || name.startsWith("proxy-")) continue;
+    forwarded[name] = value;
+  }
+  return forwarded;
+}
+
+async function waitForWebSocketOpen(socket: WebSocket) {
+  if (socket.readyState === WebSocket.OPEN) return;
+  if (socket.readyState !== WebSocket.CONNECTING) throw new Error("WebSocket closed before open");
+
+  const listeners = new AbortController();
+  await new Promise<void>((resolveOpen, rejectOpen) => {
+    const settle = (callback: () => void) => {
+      listeners.abort();
+      callback();
+    };
+    socket.addEventListener("open", () => settle(resolveOpen), { signal: listeners.signal });
+    socket.addEventListener("error", () => settle(() => rejectOpen(new Error("WebSocket error"))), {
+      signal: listeners.signal,
+    });
+    socket.addEventListener(
+      "close",
+      () => settle(() => rejectOpen(new Error("WebSocket closed before open"))),
+      { signal: listeners.signal },
+    );
+  });
+}
+
 function tunnelConnectError(tunnel: ResolvedTunnel, cause: unknown) {
   const hostname = new URL(tunnel.gateway).hostname;
   const message = cause instanceof Error ? cause.message : String(cause);

src/hosted/worker.ts

@@ -9,6 +9,7 @@ import {
 import {
   connectTokenFromRequest,
   GATEWAY_CONNECT_QUERY_PARAM,
+  isWebSocketUpgradeRequest,
   TUNNEL_CONNECT_DIAGNOSTIC_HEADER,
   TUNNEL_NAME_QUERY_PARAM,
 } from "../index.js";
@@ -102,17 +103,20 @@ export default {
       customHostname,
       tunnelName,
     });
-    const response = await shard.forward(
+    const tunnelRequest = createTunnelForwardRequest(forwarded, {
       tunnelName,
-      createTunnelForwardRequest(forwarded, tunnelUrl),
-    );
+      tunnelUrl,
+    });
+    const response = isWebSocketUpgradeRequest(request)
+      ? await shard.fetch(tunnelRequest)
+      : await shard.forward(tunnelName, tunnelRequest);
     return stripSetCookieHeadersOutsideTunnel(response, new URL(tunnelUrl).hostname);
   },
 } satisfies ExportedHandler<HostedCaptunEnv>;
 
 async function connectTunnel(request: Request, env: HostedCaptunEnv) {
   const diagnostic = isConnectDiagnostic(request);
-  if (!diagnostic && request.headers.get("upgrade") !== "websocket") {
+  if (!diagnostic && !isWebSocketUpgradeRequest(request)) {
     return new Response("Expected WebSocket upgrade\n", { status: 400 });
   }
 
@@ -158,7 +162,7 @@ function isGatewayConnectRequest(request: Request) {
 }
 
 function isConnectDiagnostic(request: Request) {
-  if (request.headers.get("upgrade") === "websocket") return false;
+  if (isWebSocketUpgradeRequest(request)) return false;
   return request.headers.get(TUNNEL_CONNECT_DIAGNOSTIC_HEADER) === "1";
 }
 
@@ -177,6 +181,9 @@ function createForwardedRequest(request: Request, customHostname: string | undef
 }
 
 function stripSetCookieHeadersOutsideTunnel(response: Response, tunnelHostname: string) {
+  // An upgraded response carries the webSocket and cannot be reconstructed.
+  if (response.status === 101) return response;
+
   const setCookies = setCookieHeaders(response.headers);
   if (setCookies.length === 0) return response;