diff --git a/CONTEXT.md b/CONTEXT.md index 9a9c6ac..f596da6 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -137,6 +137,7 @@ _Avoid_: Agent layer, agent product - The **Hosted Site** is part of the **Hosted Service** product surface, but it is not tunnel routing or **Tunnel Admission**. - **Hosted Site** code should not live in **Cloudflare Tunnel Gateway** core. A real browser package can wait until the demo surface needs it. - **Hosted Service** entrypoints may compose the **Cloudflare Tunnel Gateway**, but **Cloudflare Tunnel Gateway** core should remain understandable as a **Self-Hosted Deployment** with **Trusted Gateway Policy**. +- **Public Gateway Policy** implementation for `captun.sh` should live under `src/hosted/`; `src/worker.ts` should stay readable as the deployable **Cloudflare Tunnel Gateway** core. - The **Control Plane** governs future **Hosted Service** accounts, reservations, billing, and policy. - The **Agent Preview Use Case** uses the **Hosted Service** and may later use the **Control Plane**. - The **Agent Preview Use Case** should not shape the current gateway/core split until **Control Plane** support exists. diff --git a/src/cli/bin.ts b/src/cli/bin.ts index c231ef2..c29f790 100755 --- a/src/cli/bin.ts +++ b/src/cli/bin.ts @@ -13,10 +13,11 @@ import { createCli, yamlTableConsoleLogger } from "trpc-cli"; import { z } from "zod/v4"; import { color } from "./ansi.js"; import { CliFriendlyError } from "./cli-error.js"; -import { createCaptunTunnel } from "../index.js"; +import { CaptunTunnelConnectError, createCaptunTunnel } from "../index.js"; import { assertLocalTargetAcceptingConnections } from "./local-target.js"; import { withSpinner } from "./spinner.js"; -import { HOSTED_CAPTUN_GATEWAY } from "../routing.js"; +import { HOSTED_CAPTUN_GATEWAY, HOSTED_CAPTUN_HOSTNAME } from "../routing.js"; +import { randomConnectToken } from "../token.js"; import { captunHealthResponse, confirmTunnelHealth, @@ -57,8 +58,10 @@ export type TunnelReady = { export type CaptunCliRouterOptions = { readConfig?: () => Promise; writeConfig?: (config: Config) => Promise; + createTunnel?: typeof createCaptunTunnel; waitForShutdown?: () => Promise; onTunnelReady?: (ready: TunnelReady) => void | Promise; + tunnelRetries?: number; }; const adjectives = @@ -110,6 +113,8 @@ export function createCaptunCliRouter(options: CaptunCliRouterOptions = {}) { const tunnel = resolveTunnel(input, config); printTunnelOpening(tunnel); await runTunnelSession(tunnel, { + retries: options.tunnelRetries, + createTunnel: options.createTunnel, waitForShutdown: options.waitForShutdown, onReady: options.onTunnelReady, }); @@ -442,16 +447,22 @@ function resolveTunnel(input: TunnelCliInput, config?: Config): ResolvedTunnel { const name = input.name || randomName(); const target = normalizeTarget(input.target); + const token = input.token || config?.token || hostedGatewayToken(gateway); return { name, gateway, target, - token: input.token || config?.token, + token, requestLogs: input.requestLogs, }; } +function hostedGatewayToken(gateway: string) { + if (new URL(gateway).hostname !== HOSTED_CAPTUN_HOSTNAME) return undefined; + return randomConnectToken(); +} + function normalizeTarget(target: string) { const value = target.trim(); if (/^\d+$/.test(value)) return `http://127.0.0.1:${value}`; @@ -469,6 +480,7 @@ async function runTunnelSession( tunnel: ResolvedTunnel, opts: { retries?: number; + createTunnel?: typeof createCaptunTunnel; waitForShutdown?: () => Promise; onReady?: (ready: TunnelReady) => void | Promise; } = {}, @@ -476,7 +488,7 @@ async function runTunnelSession( const startedAt = performance.now(); await assertLocalTargetAcceptingConnections(tunnel.target); - const session = await connectTunnelWithRetry(tunnel, opts.retries || 0); + const session = await connectTunnelWithRetry(tunnel, opts.retries || 0, opts.createTunnel); try { await confirmTunnelHealth(session.url); console.log( @@ -492,7 +504,11 @@ async function runTunnelSession( } } -async function connectTunnelWithRetry(tunnel: ResolvedTunnel, retries: number) { +async function connectTunnelWithRetry( + tunnel: ResolvedTunnel, + retries: number, + createTunnel = createCaptunTunnel, +) { const fetcher = makeTunnelFetcher(tunnel); const maxAttempts = retries + 1; @@ -504,7 +520,7 @@ async function connectTunnelWithRetry(tunnel: ResolvedTunnel, retries: number) { : `Connecting to ${tunnel.gateway} (retry ${attempt - 1}/${retries})`; try { return await withSpinner(label, () => - createCaptunTunnel({ + createTunnel({ gateway: tunnel.gateway, name: tunnel.name, token: tunnel.token, @@ -562,7 +578,11 @@ function tunnelConnectError(tunnel: ResolvedTunnel, cause: unknown) { const hostname = new URL(tunnel.gateway).hostname; const message = cause instanceof Error ? cause.message : String(cause); const lines = [`Could not connect tunnel to ${color.cyan(tunnel.gateway)} (${message}).`]; - if (!hostname.endsWith(".workers.dev")) { + const connectRejection = cause instanceof CaptunTunnelConnectError ? cause.response : undefined; + const knownTunnelConflict = + connectRejection?.status === 409 && + connectRejection.body === "Tunnel name is already connected"; + if (!knownTunnelConflict && !hostname.endsWith(".workers.dev")) { // Dropping the leftmost label gives the zone-side wildcard parent — // `tunnel.mispwoso.com` -> `mispwoso.com`, `t.captun.example.com` -> `captun.example.com`. const wildcardParent = hostname.split(".").slice(1).join("."); diff --git a/src/hosted/rate-limit.ts b/src/hosted/rate-limit.ts new file mode 100644 index 0000000..57f6108 --- /dev/null +++ b/src/hosted/rate-limit.ts @@ -0,0 +1,212 @@ +import { DurableObject } from "cloudflare:workers"; + +export type HostedRateLimitEnv = { + HostedRateLimiter?: DurableObjectNamespace; + HOSTED_RATE_LIMIT_WINDOW_SECONDS?: string; + HOSTED_CONNECTS_PER_IP_PER_WINDOW?: string; + HOSTED_REQUESTS_PER_IP_PER_WINDOW?: string; + HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW?: string; + HOSTED_RATE_LIMIT_DISABLED?: string; +}; + +const DEFAULT_HOSTED_RATE_LIMIT_WINDOW_SECONDS = 60; +const DEFAULT_HOSTED_CONNECTS_PER_IP_PER_WINDOW = 30; +const DEFAULT_HOSTED_REQUESTS_PER_IP_PER_WINDOW = 600; +const DEFAULT_HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW = 1200; +const HOSTED_RATE_LIMIT_DIAGNOSTIC_WINDOW_MS = 2_000; + +type HostedRateLimitKind = "connect" | "request"; + +type HostedRateLimitInput = { limit: number; windowSeconds: number }; + +type HostedRateLimitResult = { ok: true } | { ok: false; limit: number; retryAfterSeconds: number }; + +type HostedRateLimitBucket = { + count: number; + resetAt: number; + lastRejectedAt?: number; +}; + +export class HostedRateLimiter extends DurableObject { + private bucket: HostedRateLimitBucket | undefined; + + check(input: HostedRateLimitInput): HostedRateLimitResult { + const now = Date.now(); + const bucket = this.activeBucket(now, now + input.windowSeconds * 1000); + if (bucket.count >= input.limit) { + bucket.lastRejectedAt = now; + return { + ok: false, + limit: input.limit, + retryAfterSeconds: Math.max(1, Math.ceil((bucket.resetAt - now) / 1000)), + }; + } + + bucket.count++; + return { ok: true }; + } + + diagnose(input: HostedRateLimitInput): HostedRateLimitResult { + const now = Date.now(); + const bucket = this.bucket; + if ( + bucket && + bucket.count >= input.limit && + bucket.resetAt > now && + bucket.lastRejectedAt && + now - bucket.lastRejectedAt <= HOSTED_RATE_LIMIT_DIAGNOSTIC_WINDOW_MS + ) { + return { + ok: false, + limit: input.limit, + retryAfterSeconds: Math.max(1, Math.ceil((bucket.resetAt - now) / 1000)), + }; + } + + return { ok: true }; + } + + private activeBucket(now: number, resetAt: number): HostedRateLimitBucket { + if (this.bucket && this.bucket.resetAt > now) return this.bucket; + const bucket: HostedRateLimitBucket = { count: 0, resetAt }; + this.bucket = bucket; + return bucket; + } +} + +export async function hostedRateLimitResponse(input: { + env: HostedRateLimitEnv; + request: Request; + tunnelName: string; + kind: HostedRateLimitKind; +}): Promise { + if (!input.env.HostedRateLimiter) { + return hostedRateLimiterMissingResponse(input.env); + } + + const config = hostedRateLimitConfig(input.env); + const checks = hostedRateLimitChecks({ + kind: input.kind, + clientKey: hostedClientKey(input.request), + tunnelName: input.tunnelName, + config, + }); + for (const check of checks) { + const limiter = input.env.HostedRateLimiter.getByName(hostedRateLimiterName(check.key)); + const result = await limiter.check({ + limit: check.limit, + windowSeconds: config.windowSeconds, + }); + if (!result.ok) return hostedRateLimitedResponse(result); + } + + return undefined; +} + +export async function hostedRateLimitDiagnosticResponse(input: { + env: HostedRateLimitEnv; + request: Request; + tunnelName: string; + kind: HostedRateLimitKind; +}): Promise { + if (!input.env.HostedRateLimiter) { + return hostedRateLimiterMissingResponse(input.env); + } + + const config = hostedRateLimitConfig(input.env); + const checks = hostedRateLimitChecks({ + kind: input.kind, + clientKey: hostedClientKey(input.request), + tunnelName: input.tunnelName, + config, + }); + for (const check of checks) { + const limiter = input.env.HostedRateLimiter.getByName(hostedRateLimiterName(check.key)); + const result = await limiter.diagnose({ + limit: check.limit, + windowSeconds: config.windowSeconds, + }); + if (!result.ok) return hostedRateLimitedResponse(result); + } + + return undefined; +} + +function hostedRateLimiterMissingResponse(env: HostedRateLimitEnv) { + if (env.HOSTED_RATE_LIMIT_DISABLED === "1") return undefined; + return new Response("Hosted rate limiter is not configured\n", { + status: 503, + headers: { + "content-type": "text/plain; charset=utf-8", + "cache-control": "no-store", + }, + }); +} + +function hostedRateLimitedResponse(result: Extract) { + return new Response(`Rate limit exceeded. Try again in ${result.retryAfterSeconds}s.\n`, { + status: 429, + headers: { + "content-type": "text/plain; charset=utf-8", + "cache-control": "no-store", + "retry-after": String(result.retryAfterSeconds), + "x-captun-rate-limit": String(result.limit), + }, + }); +} + +function hostedClientKey(request: Request) { + return request.headers.get("cf-connecting-ip") || "unknown"; +} + +function hostedRateLimitChecks(input: { + kind: HostedRateLimitKind; + clientKey: string; + tunnelName: string; + config: ReturnType; +}) { + if (input.kind === "connect") { + return [{ key: `connect:ip:${input.clientKey}`, limit: input.config.connectsPerIp }]; + } + + return [ + { key: `request:ip:${input.clientKey}`, limit: input.config.requestsPerIp }, + { key: `request:tunnel:${input.tunnelName}`, limit: input.config.requestsPerTunnel }, + ]; +} + +function hostedRateLimiterName(key: string) { + let hash = 2166136261; + for (let index = 0; index < key.length; index++) { + hash ^= key.charCodeAt(index); + hash = Math.imul(hash, 16777619); + } + return `bucket-${(hash >>> 0).toString(36)}`; +} + +function hostedRateLimitConfig(env: HostedRateLimitEnv) { + return { + windowSeconds: positiveInteger( + env.HOSTED_RATE_LIMIT_WINDOW_SECONDS, + DEFAULT_HOSTED_RATE_LIMIT_WINDOW_SECONDS, + ), + connectsPerIp: positiveInteger( + env.HOSTED_CONNECTS_PER_IP_PER_WINDOW, + DEFAULT_HOSTED_CONNECTS_PER_IP_PER_WINDOW, + ), + requestsPerIp: positiveInteger( + env.HOSTED_REQUESTS_PER_IP_PER_WINDOW, + DEFAULT_HOSTED_REQUESTS_PER_IP_PER_WINDOW, + ), + requestsPerTunnel: positiveInteger( + env.HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW, + DEFAULT_HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW, + ), + }; +} + +function positiveInteger(value: string | undefined, fallback: number) { + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 1) return fallback; + return parsed; +} diff --git a/src/hosted/site.ts b/src/hosted/site.ts index bfa354d..d9e2df7 100644 --- a/src/hosted/site.ts +++ b/src/hosted/site.ts @@ -265,7 +265,8 @@ const WWW_FAVICON = ` disposeSession(session), }; } @@ -296,11 +297,12 @@ class TunnelTargetFetcher extends RpcTarget { function gatewayConnectUrl(options) { const url = new URL(options.gateway || "https://captun.sh"); + const token = options.token || (url.hostname === "captun.sh" ? randomConnectToken() : undefined); url.protocol = url.protocol === "http:" ? "ws:" : "wss:"; url.searchParams.set("captun-connect", "1"); url.searchParams.set("captun-name", options.name || randomTunnelName()); - if (options.token) url.searchParams.set("captun-token", options.token); - return url; + if (token) url.searchParams.set("captun-token", token); + return { url, token }; } function waitUntilOpen(socket) { @@ -346,6 +348,12 @@ function randomTunnelName() { return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); } +function randomConnectToken() { + const bytes = new Uint8Array(16); + crypto.getRandomValues(bytes); + return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); +} + function disposeSession(session) { const disposeSymbol = Symbol.dispose; if (disposeSymbol && typeof session[disposeSymbol] === "function") session[disposeSymbol](); diff --git a/src/hosted/worker.ts b/src/hosted/worker.ts index 32c7c03..e32c3ea 100644 --- a/src/hosted/worker.ts +++ b/src/hosted/worker.ts @@ -1,21 +1,232 @@ -import cloudflareTunnelGateway, { CaptunServerShard, type CaptunEnv } from "../worker.js"; -import { GATEWAY_CONNECT_QUERY_PARAM } from "../routing.js"; +import { + captunServerShard, + createTunnelConnectRequest, + createTunnelForwardRequest, + CaptunServerShard as CloudflareTunnelGatewayShard, + type TunnelAdmission, + type TunnelAdmissionInput, +} from "../worker.js"; +import { + CONNECT_TOKEN_QUERY_PARAM, + GATEWAY_CONNECT_QUERY_PARAM, + getTunnelNameFromUrl, + getTunnelUrl, + HOSTED_CAPTUN_HOSTNAME, + isValidTunnelName, + RESERVED_TUNNEL_NAMES, + TUNNEL_CONNECT_DIAGNOSTIC_HEADER, + TUNNEL_NAME_QUERY_PARAM, +} from "../routing.js"; +import { + HostedRateLimiter, + hostedRateLimitDiagnosticResponse, + hostedRateLimitResponse, + type HostedRateLimitEnv, +} from "./rate-limit.js"; import { hostedCaptunResponse } from "./site.js"; -export { CaptunServerShard }; +export class CaptunServerShard extends CloudflareTunnelGatewayShard { + protected decideTunnelAdmission(input: TunnelAdmissionInput): TunnelAdmission { + const configuredToken = input.env.CAPTUN_TOKEN; + const token = connectToken(input.request) || undefined; + if (configuredToken) { + if (!token || !constantTimeEqual(token, configuredToken)) { + return { ok: false, response: reject("Unauthorized\n", 401) }; + } + return { ok: true, token }; + } + + if (!token) return { ok: false, response: reject("Missing tunnel token\n", 400) }; + if (!/^[a-zA-Z0-9._~-]{1,128}$/.test(token)) { + return { ok: false, response: reject("Invalid tunnel token\n", 400) }; + } + if (input.activeToken && input.activeToken !== token) { + return { ok: false, response: reject("Tunnel name is already connected\n", 409) }; + } + + return { ok: true, token }; + } +} + +export { HostedRateLimiter }; + +export type HostedCaptunEnv = HostedRateLimitEnv & { + CaptunServerShard: DurableObjectNamespace; + CAPTUN_TOKEN?: string; + CAPTUN_SECRET?: string; + SHARD_COUNT?: string; + CUSTOM_HOSTNAME?: string; +}; export default { - fetch(request: Request, env: CaptunEnv): Response | Promise { + async fetch(request: Request, env: HostedCaptunEnv): Promise { + if ("CAPTUN_SECRET" in env) throw new Error("CAPTUN_SECRET has been renamed to CAPTUN_TOKEN"); + if (env.CUSTOM_HOSTNAME !== HOSTED_CAPTUN_HOSTNAME) { + throw new Error("Hosted Captun Worker requires CUSTOM_HOSTNAME=captun.sh"); + } + if (isGatewayConnectRequest(request)) { - return cloudflareTunnelGateway.fetch(request, env); + return connectTunnel(request, env); } const hostedResponse = hostedCaptunResponse(request); if (hostedResponse) return hostedResponse; - return cloudflareTunnelGateway.fetch(request, env); + + const tunnelName = getTunnelNameFromUrl({ + customHostname: env.CUSTOM_HOSTNAME, + url: request.url, + }); + if (!tunnelName) return new Response("Missing tunnel name\n", { status: 404 }); + + if (RESERVED_TUNNEL_NAMES.includes(tunnelName)) { + return new Response("Reserved Captun tunnel name\n", { status: 404 }); + } + + const rateLimited = await hostedRateLimitResponse({ + env, + request, + tunnelName, + kind: "request", + }); + if (rateLimited) return rateLimited; + + const shard = captunServerShard(env, tunnelName); + const forwarded = new Request(request.url, request); + const tunnelUrl = getTunnelUrl({ + reqUrl: request.url, + customHostname: env.CUSTOM_HOSTNAME, + tunnelName, + }); + const response = await shard.forward( + tunnelName, + createTunnelForwardRequest(forwarded, tunnelUrl), + ); + return stripSetCookieHeadersOutsideTunnel(response, new URL(tunnelUrl).hostname); }, -} satisfies ExportedHandler; +} satisfies ExportedHandler; + +async function connectTunnel(request: Request, env: HostedCaptunEnv) { + const diagnostic = isConnectDiagnostic(request); + if (!diagnostic && request.headers.get("upgrade") !== "websocket") { + return new Response("Expected WebSocket upgrade\n", { status: 400 }); + } + + const url = new URL(request.url); + const tunnelName = url.searchParams.get(TUNNEL_NAME_QUERY_PARAM) || ""; + if (!isValidTunnelName(tunnelName) || RESERVED_TUNNEL_NAMES.includes(tunnelName)) { + return new Response("Missing tunnel name\n", { status: 404 }); + } + + const shard = captunServerShard(env, tunnelName); + const tunnelUrl = getTunnelUrl({ + reqUrl: request.url, + customHostname: env.CUSTOM_HOSTNAME, + tunnelName, + }); + const connectRequest = createTunnelConnectRequest({ request, tunnelName, tunnelUrl }); + + if (diagnostic) { + const rateLimited = await hostedRateLimitDiagnosticResponse({ + env, + request, + tunnelName, + kind: "connect", + }); + if (rateLimited) return rateLimited; + return shard.diagnoseConnect(tunnelName, connectRequest); + } + + const rateLimited = await hostedRateLimitResponse({ + env, + request, + tunnelName, + kind: "connect", + }); + if (rateLimited) return rateLimited; + + return shard.fetch(connectRequest); +} function isGatewayConnectRequest(request: Request) { return new URL(request.url).searchParams.get(GATEWAY_CONNECT_QUERY_PARAM) === "1"; } + +function isConnectDiagnostic(request: Request) { + if (request.headers.get("upgrade") === "websocket") return false; + return request.headers.get(TUNNEL_CONNECT_DIAGNOSTIC_HEADER) === "1"; +} + +function connectToken(request: Request) { + return new URL(request.url).searchParams.get(CONNECT_TOKEN_QUERY_PARAM); +} + +function stripSetCookieHeadersOutsideTunnel(response: Response, tunnelHostname: string) { + const setCookies = setCookieHeaders(response.headers); + if (setCookies.length === 0) return response; + + const safeCookies = setCookies.filter((cookie) => + setCookieIsScopedToTunnel(cookie, tunnelHostname), + ); + if (safeCookies.length === setCookies.length) return response; + + const headers = new Headers(response.headers); + headers.delete("set-cookie"); + for (const cookie of safeCookies) headers.append("set-cookie", cookie); + + return new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers, + }); +} + +function setCookieHeaders(headers: Headers) { + const getSetCookie = (headers as Headers & { getSetCookie?: () => string[] }).getSetCookie; + const cookies = getSetCookie ? getSetCookie.call(headers) : []; + if (cookies.length > 0) return cookies; + + const cookie = headers.get("set-cookie"); + return cookie ? [cookie] : []; +} + +function setCookieIsScopedToTunnel(cookie: string, tunnelHostname: string) { + const domains = setCookieDomains(cookie); + if (domains.length === 0) return true; + + // Add captun.sh to the public suffix list if/when people are using this. + return domains.every( + (domain) => domain === tunnelHostname || domain.endsWith(`.${tunnelHostname}`), + ); +} + +function setCookieDomains(cookie: string) { + const attributes = cookie.split(";").slice(1); + const domains: string[] = []; + for (const attribute of attributes) { + const [name, ...valueParts] = attribute.split("="); + if (name?.trim().toLowerCase() !== "domain") continue; + domains.push(valueParts.join("=").trim().replace(/^\./, "").toLowerCase()); + } + return domains; +} + +function reject(body: string, status: number) { + return new Response(body, { + status, + headers: { + "content-type": "text/plain; charset=utf-8", + "cache-control": "no-store", + }, + }); +} + +function constantTimeEqual(actual: string, expected: string) { + const actualBytes = new TextEncoder().encode(actual); + const expectedBytes = new TextEncoder().encode(expected); + if (actualBytes.length !== expectedBytes.length) return false; + let diff = 0; + for (let index = 0; index < actualBytes.length; index++) { + diff |= actualBytes[index]! ^ expectedBytes[index]!; + } + return diff === 0; +} diff --git a/src/index.ts b/src/index.ts index fbf4816..605415d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -3,10 +3,13 @@ import { CONNECT_TOKEN_QUERY_PARAM, GATEWAY_CONNECT_QUERY_PARAM, HOSTED_CAPTUN_GATEWAY, + HOSTED_CAPTUN_HOSTNAME, + TUNNEL_CONNECT_DIAGNOSTIC_HEADER, TUNNEL_NAME_QUERY_PARAM, } from "./routing.js"; import { acceptFetcherCapabilityFromSocket } from "./server-core.js"; import type { Fetcher, TunnelReady } from "./server-core.js"; +import { randomConnectToken } from "./token.js"; export type { Fetcher, FetcherStub } from "./server-core.js"; export { acceptFetcherCapabilityFromSocket } from "./server-core.js"; @@ -21,11 +24,25 @@ export type CaptunTunnel = Disposable & { token?: string; }; +export class CaptunTunnelConnectError extends Error { + response: { status: number; statusText: string; body: string } | undefined; + + constructor( + message: string, + response: { status: number; statusText: string; body: string } | undefined, + ) { + super(message); + this.name = "CaptunTunnelConnectError"; + this.response = response; + } +} + type TunnelClientCapability = Fetcher & { ready(tunnel: TunnelReady): void | Promise; }; const TUNNEL_READY_TIMEOUT_MS = 5_000; +const WEBSOCKET_REJECTION_PROBE_TIMEOUT_MS = 500; /** Creates a public tunnel by exposing a local fetch implementation to a Tunnel Gateway. */ export async function createCaptunTunnel( @@ -44,7 +61,7 @@ export async function createCaptunTunnel( }); const session = newWebSocketRpcSession(socket, fetcher); try { - await waitUntilOpen(socket); + await waitUntilOpen(socket, connect.url); const tunnel = await waitUntilReady(ready.promise); return { url: tunnel.url, @@ -60,10 +77,15 @@ export async function createCaptunTunnel( function gatewayConnectRequest(options: { gateway?: string | URL; name?: string; token?: string }) { const name = options.name || randomTunnelName(); const url = new URL(options.gateway || HOSTED_CAPTUN_GATEWAY); + const token = options.token || (isHostedCaptunGateway(url) ? randomConnectToken() : undefined); url.searchParams.set(GATEWAY_CONNECT_QUERY_PARAM, "1"); url.searchParams.set(TUNNEL_NAME_QUERY_PARAM, name); - if (options.token) url.searchParams.set(CONNECT_TOKEN_QUERY_PARAM, options.token); - return { url: url.toString(), name, token: options.token }; + if (token) url.searchParams.set(CONNECT_TOKEN_QUERY_PARAM, token); + return { url: url.toString(), name, token }; +} + +function isHostedCaptunGateway(url: URL) { + return url.hostname === HOSTED_CAPTUN_HOSTNAME; } function randomTunnelName() { @@ -97,7 +119,7 @@ function createWebSocket(url: string | URL) { return new WebSocket(connectUrl.href); } -async function waitUntilOpen(socket: WebSocket) { +async function waitUntilOpen(socket: WebSocket, connectUrl: string) { if (socket.readyState === WebSocket.OPEN) return; if (socket.readyState !== WebSocket.CONNECTING) { throw new Error("WebSocket closed before opening"); @@ -112,20 +134,59 @@ async function waitUntilOpen(socket: WebSocket) { socket.addEventListener("open", () => settle(resolve), { signal: listeners.signal }); socket.addEventListener( "error", - () => settle(() => reject(new Error("WebSocket connection failed"))), + () => + settle(() => { + void webSocketConnectionFailedError(connectUrl).then(reject); + }), { signal: listeners.signal }, ); socket.addEventListener( "close", (event) => { listeners.abort(); - reject(new Error(`WebSocket closed before opening: ${event.code} ${event.reason}`)); + void webSocketConnectionFailedError(connectUrl).then((error) => { + reject( + error.response + ? error + : new Error(`WebSocket closed before opening: ${event.code} ${event.reason}`), + ); + }); }, { signal: listeners.signal }, ); }); } +async function webSocketConnectionFailedError(connectUrl: string) { + const response = await readWebSocketRejection(connectUrl); + if (!response) return new CaptunTunnelConnectError("WebSocket connection failed", undefined); + return new CaptunTunnelConnectError( + `WebSocket connection failed: ${response.status} ${response.statusText}: ${response.body}`.trim(), + response, + ); +} + +async function readWebSocketRejection(connectUrl: string) { + const abort = new AbortController(); + const timeout = setTimeout(() => abort.abort(), WEBSOCKET_REJECTION_PROBE_TIMEOUT_MS); + try { + const response = await fetch(connectUrl, { + headers: { [TUNNEL_CONNECT_DIAGNOSTIC_HEADER]: "1" }, + signal: abort.signal, + }); + if (response.ok) return undefined; + return { + status: response.status, + statusText: response.statusText || "Rejected", + body: (await response.text()).trim(), + }; + } catch { + return undefined; + } finally { + clearTimeout(timeout); + } +} + async function waitUntilReady(promise: Promise) { let timeout: ReturnType | undefined; try { diff --git a/src/routing.ts b/src/routing.ts index bfb8b58..f03a8ab 100644 --- a/src/routing.ts +++ b/src/routing.ts @@ -3,6 +3,7 @@ export const HOSTED_CAPTUN_GATEWAY = "https://captun.sh"; export const GATEWAY_CONNECT_QUERY_PARAM = "captun-connect"; export const TUNNEL_NAME_QUERY_PARAM = "captun-name"; export const CONNECT_TOKEN_QUERY_PARAM = "captun-token"; +export const TUNNEL_CONNECT_DIAGNOSTIC_HEADER = "x-captun-connect-diagnostic"; export const RESERVED_TUNNEL_NAMES = [ "account", "accounts", diff --git a/src/token.ts b/src/token.ts new file mode 100644 index 0000000..af97b94 --- /dev/null +++ b/src/token.ts @@ -0,0 +1,5 @@ +export function randomConnectToken() { + const bytes = new Uint8Array(16); + crypto.getRandomValues(bytes); + return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); +} diff --git a/src/worker.ts b/src/worker.ts index 65c7bea..a49ee13 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -13,7 +13,7 @@ import { } from "./routing.js"; export type CaptunEnv = { - CaptunServerShard: DurableObjectNamespace; + CaptunServerShard: DurableObjectNamespace>; CAPTUN_TOKEN?: string; CAPTUN_SECRET?: string; SHARD_COUNT?: string; @@ -23,12 +23,27 @@ export type CaptunEnv = { /** Set by the top-level Worker on the WebSocket-upgrade request so the DO knows the tunnel. */ const TUNNEL_NAME_HEADER = "x-captun-tunnel-name"; +type CaptunShardBindingEnv = { + CaptunServerShard: DurableObjectNamespace>; + SHARD_COUNT?: string; +}; + type ActiveTunnel = { url: string; token?: string; fetcher: FetcherStub; }; +export type TunnelAdmission = + | { ok: true; token: string | undefined } + | { ok: false; response: Response }; + +export type TunnelAdmissionInput = { + request: Request; + env: Env; + activeToken: string | undefined; +}; + /** * A shard Durable Object owns many named tunnels. * @@ -37,8 +52,27 @@ type ActiveTunnel = { * more objects, which adds cold starts when new shards wake up but gives better * aggregate throughput for lots of concurrent large responses. */ -export class CaptunServerShard extends DurableObject { - private readonly tunnels = new Map(); +export class CaptunServerShard< + Env extends { CAPTUN_TOKEN?: string } = CaptunEnv, +> extends DurableObject { + private tunnels = new Map(); + + protected decideTunnelAdmission(input: TunnelAdmissionInput): TunnelAdmission { + const expected = input.env.CAPTUN_TOKEN; + if (expected) { + // Constant-time comparison to avoid leaking the gateway token via timing. + const actual = new TextEncoder().encode(connectToken(input.request) || ""); + const want = new TextEncoder().encode(expected); + if (!constantTimeEqual(actual, want)) { + return { ok: false, response: new Response("Unauthorized\n", { status: 401 }) }; + } + } + + return { + ok: true, + token: expected ? connectToken(input.request) || undefined : undefined, + }; + } // The DO's `fetch` only handles the WebSocket upgrade. The upgrade hand-off // is special-cased by the Workers runtime around `stub.fetch(...)` — a 101 @@ -53,24 +87,21 @@ export class CaptunServerShard extends DurableObject { const tunnelUrl = request.headers.get(TUNNEL_URL_HEADER); if (!tunnelUrl) return new Response("Missing tunnel URL\n", { status: 404 }); - const expected = this.env.CAPTUN_TOKEN; - if (expected) { - // Constant-time comparison to avoid leaking the gateway token via timing. - const actual = new TextEncoder().encode(connectToken(request) || ""); - const want = new TextEncoder().encode(expected); - if (!constantTimeEqual(actual, want)) { - return new Response("Unauthorized\n", { status: 401 }); - } - } + const activeTunnel = this.tunnels.get(tunnelName); + const admission = await this.decideTunnelAdmission({ + request, + env: this.env, + activeToken: activeTunnel?.token, + }); + if (!admission.ok) return admission.response; - const token = expected ? connectToken(request) || undefined : undefined; - this.tunnels.get(tunnelName)?.fetcher[Symbol.dispose](); + activeTunnel?.fetcher[Symbol.dispose](); const { response, fetcher } = acceptFetcherCapability({ onDisconnect: () => { if (this.tunnels.get(tunnelName)?.fetcher === fetcher) this.tunnels.delete(tunnelName); }, }); - const tunnel = { url: tunnelUrl, token, fetcher }; + const tunnel = { url: tunnelUrl, token: admission.token, fetcher }; this.tunnels.set(tunnelName, tunnel); queueMicrotask(() => { void fetcher.ready({ url: tunnel.url, token: tunnel.token }); @@ -78,6 +109,19 @@ export class CaptunServerShard extends DurableObject { return response; } + async diagnoseConnect(tunnelName: string, request: Request): Promise { + const admission = await this.decideTunnelAdmission({ + request, + env: this.env, + activeToken: this.tunnels.get(tunnelName)?.token, + }); + if (!admission.ok) return admission.response; + return new Response(null, { + status: 204, + headers: { "cache-control": "no-store" }, + }); + } + async forward(tunnelName: string, request: Request): Promise { const tunnel = this.tunnels.get(tunnelName)?.fetcher; if (!tunnel) return new Response("No tunnel client connected\n", { status: 503 }); @@ -116,9 +160,7 @@ export default { return new Response("Reserved Captun tunnel name\n", { status: 404 }); } - const shard = env.CaptunServerShard.getByName( - captunShardName(tunnelName, Number(env.SHARD_COUNT || 1)), - ); + const shard = captunServerShard(env, tunnelName); const forwarded = new Request(url, request); // Keep the canonical tunnel URL attached while crossing into the DO. @@ -127,9 +169,7 @@ export default { customHostname: env.CUSTOM_HOSTNAME, tunnelName, }); - const headers = new Headers(forwarded.headers); - headers.set(TUNNEL_URL_HEADER, tunnelUrl); - return shard.forward(tunnelName, new Request(forwarded, { headers })); + return shard.forward(tunnelName, createTunnelForwardRequest(forwarded, tunnelUrl)); }, } satisfies ExportedHandler; @@ -149,13 +189,8 @@ function connectTunnel(request: Request, env: CaptunEnv) { customHostname: env.CUSTOM_HOSTNAME, tunnelName, }); - const shard = env.CaptunServerShard.getByName( - captunShardName(tunnelName, Number(env.SHARD_COUNT || 1)), - ); - const headers = new Headers(request.headers); - headers.set(TUNNEL_NAME_HEADER, tunnelName); - headers.set(TUNNEL_URL_HEADER, tunnelUrl); - return shard.fetch(new Request(request, { headers })); + const shard = captunServerShard(env, tunnelName); + return shard.fetch(createTunnelConnectRequest({ request, tunnelName, tunnelUrl })); } function isGatewayConnectRequest(request: Request) { @@ -166,6 +201,30 @@ function connectToken(request: Request) { return new URL(request.url).searchParams.get(CONNECT_TOKEN_QUERY_PARAM); } +export function captunServerShard( + env: CaptunShardBindingEnv, + tunnelName: string, +): DurableObjectStub> { + return env.CaptunServerShard.getByName(captunShardName(tunnelName, Number(env.SHARD_COUNT || 1))); +} + +export function createTunnelConnectRequest(input: { + request: Request; + tunnelName: string; + tunnelUrl: string; +}): Request { + const headers = new Headers(input.request.headers); + headers.set(TUNNEL_NAME_HEADER, input.tunnelName); + headers.set(TUNNEL_URL_HEADER, input.tunnelUrl); + return new Request(input.request, { headers }); +} + +export function createTunnelForwardRequest(request: Request, tunnelUrl: string): Request { + const headers = new Headers(request.headers); + headers.set(TUNNEL_URL_HEADER, tunnelUrl); + return new Request(request, { headers }); +} + function constantTimeEqual(actual: Uint8Array, expected: Uint8Array) { if (actual.length !== expected.length) return false; let diff = 0; diff --git a/tasks/hosted-captun-sh.md b/tasks/hosted-captun-sh.md index aff3226..7748747 100644 --- a/tasks/hosted-captun-sh.md +++ b/tasks/hosted-captun-sh.md @@ -1,11 +1,11 @@ --- -status: initial-hosted-deployed +status: hosted-safety-in-review size: medium --- # Hosted captun.sh -Status summary: Initial hosted deployment is live on `captun.sh`. The CLI, library, and browser landing-page demo can create hosted random tunnels; the main missing work is a proper free/paid control plane, tunnel ownership, and throttling. +Status summary: Initial hosted deployment is live on `captun.sh`. The follow-up PR now adds anonymous tunnel tokens, same-token reconnects, different-token `409`s, hosted-only Durable Object rate-limit buckets, and hosted response cookie-domain stripping under `src/hosted/`. The main missing work is still a proper auth/payment control plane plus richer resource caps, observability, and a Public Suffix List submission after the service has enough active users. ## Initial public-hosted slice @@ -21,9 +21,10 @@ Status summary: Initial hosted deployment is live on `captun.sh`. The CLI, libra ## Safety and product follow-up - [ ] Use cryptographic random names for free hosted tunnels and keep friendly/custom subdomains behind auth or a paid reservation model. -- [ ] Add per-session tunnel ownership: first client claims a tunnel name, the same token can reconnect, and a different token gets `409` instead of evicting the active tunnel. -- [ ] Add Cloudflare Rate Limiting bindings for cheap edge throttles on connect attempts and forwarded requests. -- [ ] Add Durable Object backed global-ish limits for active tunnels, concurrent tunnels per IP/account, and suspicious reconnect churn. +- [x] Add per-session tunnel ownership: first client claims a tunnel name, the same token can reconnect, and a different token gets `409` instead of evicting the active tunnel. _Hosted `captun.sh` now requires a generated `captun-token`; `CaptunServerShard` keeps the active tunnel token and rejects mismatched reconnects._ +- [x] Add hosted throttles on connect attempts and forwarded requests. _Added a `HostedRateLimiter` Durable Object binding with per-IP connect/request buckets and per-tunnel request buckets in the hosted Worker entrypoint._ +- [x] Strip hosted tunnel response cookies scoped outside the tunnel's own subdomain. _`src/hosted/worker.ts` removes `Set-Cookie` headers whose `Domain` is broader than or outside the active tunnel hostname; add `captun.sh` to the Public Suffix List if/when people are using this._ +- [ ] Add broader Durable Object backed limits for active tunnels, concurrent tunnels per IP/account, and suspicious reconnect churn. - [ ] Add basic resource caps: max tunnel lifetime, idle timeout, in-flight request cap, request body size limit, and response streaming guardrails. - [ ] Add observability for rejected connects, `429`s, high-volume tunnel names, high-volume IPs, and top error classes. - [ ] Document an emergency shutdown path for disabling hosted `captun.sh` without affecting self-hosted deployments. @@ -34,3 +35,5 @@ Status summary: Initial hosted deployment is live on `captun.sh`. The CLI, libra - 2026-05-23: Public e2e passed against the live service with `CAPTUN_PUBLIC_E2E=1 pnpm vitest run test/public-hosted.test.ts`. - 2026-05-23: Reserved names and `www.captun.sh` verified against the live Worker. Apex redirect verified with `curl --resolve` against Cloudflare's authoritative A record while local resolver propagation was still uneven. - 2026-05-23: Browser demo deployed and manually verified with Playwriter. Clicking "create tunnel" produced a random `captun.sh` URL, and `curl` to that URL returned the browser-defined response. +- 2026-05-26: Hosted safety branch rebuilt on top of #16's gateway/token protocol. Public hosted clients now get generated tokens in the CLI, library, and browser module; self-hosted deployments without `CAPTUN_TOKEN` still permit trusted replacement. +- 2026-05-26: PSL submission deferred because current PSL guidance expects thousands of active users and more than two years of remaining domain registration. Hosted Worker now strips broad `Set-Cookie Domain` attributes as defense in depth before and after PSL inclusion. diff --git a/test/cli.test.ts b/test/cli.test.ts new file mode 100644 index 0000000..fdd4fca --- /dev/null +++ b/test/cli.test.ts @@ -0,0 +1,148 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; + +import { createRouterClient } from "@orpc/server"; +import { expect, test } from "vitest"; + +import { createCaptunCliRouter } from "../src/cli/bin.js"; +import { isCaptunHealthRequest } from "../src/cli/tunnel-health.js"; +import { CaptunTunnelConnectError } from "../src/index.js"; + +test("CLI tunnel connect errors do not blame DNS for active token conflicts", async () => { + await using target = await createTestServer(defaultTargetHandler); + + const router = createCaptunCliRouter({ + readConfig: async () => undefined, + createTunnel: async () => { + throw new CaptunTunnelConnectError( + "WebSocket connection failed: 409 Conflict: Tunnel name is already connected", + { status: 409, statusText: "Conflict", body: "Tunnel name is already connected" }, + ); + }, + }); + const client = createRouterClient(router); + + let caught: unknown; + try { + await client.tunnel({ + target: String(target.port), + gateway: "https://captun.sh", + name: "demo", + requestLogs: false, + }); + } catch (error) { + caught = error; + } + + expect(caught).toMatchObject({ + message: expect.stringContaining("Tunnel name is already connected"), + }); + expect(caught).not.toMatchObject({ + message: expect.stringContaining("DNS for"), + }); +}); + +test("CLI tunnel connect errors keep DNS guidance for unrelated 409 responses", async () => { + await using target = await createTestServer(defaultTargetHandler); + + const router = createCaptunCliRouter({ + readConfig: async () => undefined, + createTunnel: async () => { + throw new CaptunTunnelConnectError( + "WebSocket connection failed: 409 Conflict: Some other conflict", + { + status: 409, + statusText: "Conflict", + body: "Some other conflict", + }, + ); + }, + }); + const client = createRouterClient(router); + + let caught: unknown; + try { + await client.tunnel({ + target: String(target.port), + gateway: "https://custom.example.com", + name: "demo", + requestLogs: false, + }); + } catch (error) { + caught = error; + } + + expect(caught).toMatchObject({ + message: expect.stringContaining("Some other conflict"), + }); + expect(caught).toMatchObject({ + message: expect.stringContaining("DNS for"), + }); +}); + +test("CLI tunnel retries reuse the same generated hosted token", async () => { + await using target = await createTestServer(defaultTargetHandler); + const tokens: Array = []; + + const router = createCaptunCliRouter({ + readConfig: async () => undefined, + waitForShutdown: async () => {}, + tunnelRetries: 1, + createTunnel: async (options) => { + tokens.push(options.token); + if (tokens.length === 1) throw new Error("try again"); + return { + url: target.origin, + token: options.token, + [Symbol.dispose]() {}, + }; + }, + }); + const client = createRouterClient(router); + + await client.tunnel({ + target: String(target.port), + gateway: "https://captun.sh", + name: "demo", + requestLogs: false, + }); + + expect(tokens).toEqual([tokens[0], tokens[0]]); + expect(tokens[0]).toMatch(/^[a-f0-9]{32}$/); +}); + +function defaultTargetHandler(request: IncomingMessage, response: ServerResponse) { + const req = new Request(`http://127.0.0.1${request.url || "/"}`); + if (isCaptunHealthRequest(req)) { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: true })); + return; + } + + response.end("ok\n"); +} + +async function createTestServer( + handler: (req: IncomingMessage, res: ServerResponse) => void | Promise, +) { + const server = createServer((req, res) => { + void Promise.resolve(handler(req, res)).catch((error: unknown) => { + res.writeHead(500, { "content-type": "text/plain" }); + res.end(error instanceof Error ? error.stack : String(error)); + }); + }); + await new Promise((resolveListen, rejectListen) => { + server.once("error", rejectListen); + server.listen(0, "127.0.0.1", resolveListen); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Could not determine test-server port."); + } + return { + origin: `http://127.0.0.1:${address.port}`, + port: address.port, + async [Symbol.asyncDispose]() { + await new Promise((resolveClose) => server.close(() => resolveClose())); + }, + }; +} diff --git a/test/hosted-worker.test.ts b/test/hosted-worker.test.ts index 9f1a7aa..e13ddcc 100644 --- a/test/hosted-worker.test.ts +++ b/test/hosted-worker.test.ts @@ -1,5 +1,7 @@ +import { newWebSocketRpcSession, RpcTarget } from "capnweb"; import { expect, test } from "vitest"; -import { createHostedCaptunWorkerFixture } from "./miniflare.js"; +import { TUNNEL_CONNECT_DIAGNOSTIC_HEADER } from "../src/routing.js"; +import { createHostedCaptunWorkerFixture, createMiniflareWorkerFixture } from "./miniflare.js"; test("Hosted Captun redirects the apex hostname to www", async () => { await using fixture = await createHostedCaptunWorkerFixture(); @@ -16,7 +18,7 @@ test("Hosted Captun lets apex gateway connects reach the tunnel gateway", async await using fixture = await createHostedCaptunWorkerFixture(); const response = await fixture.worker.fetch( - "https://captun.sh/?captun-connect=1&captun-name=demo", + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", { headers: { upgrade: "websocket" }, redirect: "manual", @@ -79,6 +81,8 @@ test("Hosted Captun serves the browser demo module on www", async () => { const module = await response.text(); expect(module).toEqual(expect.stringContaining("createCaptunTunnel")); expect(module).toEqual(expect.stringContaining("captun-connect")); + expect(module).toEqual(expect.stringContaining("captun-token")); + expect(module).toEqual(expect.stringContaining("randomConnectToken")); expect(module).not.toEqual(expect.stringContaining("__captun-connect")); }); @@ -161,6 +165,376 @@ test("Hosted Captun landing page loads CodeMirror for the browser demo editor", ); }); +test("Hosted Captun rate limits tunnel connect attempts per client IP", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + CAPTUN_TOKEN: "secret", + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "1", + }); + const headers = { upgrade: "websocket", "cf-connecting-ip": "203.0.113.10" }; + + const first = await fixture.worker.fetch("https://captun.sh/?captun-connect=1&captun-name=one", { + headers, + }); + const second = await fixture.worker.fetch("https://captun.sh/?captun-connect=1&captun-name=two", { + headers, + }); + + expect(first).toMatchObject({ status: 401 }); + expect(second).toMatchObject({ status: 429 }); + expect(second.headers.get("retry-after")).toBe("60"); + expect(second.headers.get("cache-control")).toBe("no-store"); + expect(await second.text()).toBe("Rate limit exceeded. Try again in 60s.\n"); +}); + +test("Hosted Captun rate limits forwarded requests per client IP", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_REQUESTS_PER_IP_PER_WINDOW: "2", + HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW: "100", + }); + const headers = { "cf-connecting-ip": "203.0.113.20" }; + + const first = await fixture.worker.fetch("https://one.captun.sh/hello", { headers }); + const second = await fixture.worker.fetch("https://two.captun.sh/hello", { headers }); + const third = await fixture.worker.fetch("https://three.captun.sh/hello", { headers }); + + expect(first).toMatchObject({ status: 503 }); + expect(second).toMatchObject({ status: 503 }); + expect(third).toMatchObject({ status: 429 }); + expect(third.headers.get("x-captun-rate-limit")).toBe("2"); +}); + +test("Hosted Captun rate limits forwarded requests per tunnel name", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_REQUESTS_PER_IP_PER_WINDOW: "100", + HOSTED_REQUESTS_PER_TUNNEL_PER_WINDOW: "1", + }); + + const first = await fixture.worker.fetch("https://one.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.30" }, + }); + const second = await fixture.worker.fetch("https://one.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.31" }, + }); + const otherTunnel = await fixture.worker.fetch("https://two.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.31" }, + }); + + expect(first).toMatchObject({ status: 503 }); + expect(second).toMatchObject({ status: 429 }); + expect(otherTunnel).toMatchObject({ status: 503 }); +}); + +test("Hosted Captun fails closed when the rate limiter binding is missing", async () => { + await using fixture = await createMiniflareWorkerFixture({ + entryPoint: "src/hosted/worker.ts", + durableObjects: { + CaptunServerShard: { className: "CaptunServerShard" }, + }, + bindings: { CUSTOM_HOSTNAME: "captun.sh" }, + }); + + const response = await fixture.worker.fetch("https://one.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.50" }, + }); + + expect(response).toMatchObject({ status: 503 }); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(await response.text()).toBe("Hosted rate limiter is not configured\n"); +}); + +test("Hosted Captun only bypasses a missing rate limiter binding when explicitly disabled", async () => { + await using fixture = await createMiniflareWorkerFixture({ + entryPoint: "src/hosted/worker.ts", + durableObjects: { + CaptunServerShard: { className: "CaptunServerShard" }, + }, + bindings: { + CUSTOM_HOSTNAME: "captun.sh", + HOSTED_RATE_LIMIT_DISABLED: "1", + HOSTED_REQUESTS_PER_IP_PER_WINDOW: "1", + }, + }); + + const first = await fixture.worker.fetch("https://one.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.51" }, + }); + const second = await fixture.worker.fetch("https://two.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.51" }, + }); + + expect(first).toMatchObject({ status: 503 }); + expect(second).toMatchObject({ status: 503 }); +}); + +test("Hosted Captun does not trust spoofable forwarded IP headers for rate limiting", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_REQUESTS_PER_IP_PER_WINDOW: "1", + }); + + const first = await fixture.worker.fetch("https://one.captun.sh/hello", { + headers: { "x-forwarded-for": "203.0.113.60" }, + }); + const second = await fixture.worker.fetch("https://two.captun.sh/hello", { + headers: { "x-forwarded-for": "203.0.113.61" }, + }); + + expect(first).toMatchObject({ status: 503 }); + expect(second).toMatchObject({ status: 429 }); +}); + +test("Hosted Captun accepts the configured gateway token when token auth is enabled", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + CAPTUN_TOKEN: "secret", + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=secret", + response: "configured token\n", + clientIp: "203.0.113.65", + }); + + const response = await fixture.worker.fetch("https://demo.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.66" }, + }); + + expect(response).toMatchObject({ status: 200 }); + expect(await response.text()).toBe("configured token\n"); +}); + +test("Hosted Captun rejects invalid anonymous ownership tokens", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + + const response = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=no%20spaces", + { headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.67" } }, + ); + + expect(response).toMatchObject({ status: 400 }); + expect(await response.text()).toBe("Invalid tunnel token\n"); +}); + +test("Hosted Captun rejects a different token while a tunnel is active", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "token a\n", + clientIp: "203.0.113.70", + }); + + const conflict = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-b", + { headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.71" } }, + ); + expect(conflict).toMatchObject({ status: 409 }); + expect(await conflict.text()).toBe("Tunnel name is already connected\n"); + + const stillOwned = await fixture.worker.fetch("https://demo.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.72" }, + }); + + expect(stillOwned).toMatchObject({ status: 200 }); + expect(await stillOwned.text()).toBe("token a\n"); +}); + +test("Hosted Captun connect diagnostics do not replace active tunnels", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "token a\n", + clientIp: "203.0.113.74", + }); + + const diagnostic = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + { + headers: { + "cf-connecting-ip": "203.0.113.75", + [TUNNEL_CONNECT_DIAGNOSTIC_HEADER]: "1", + }, + }, + ); + const stillOwned = await fixture.worker.fetch("https://demo.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.76" }, + }); + + expect(diagnostic).toMatchObject({ status: 204 }); + expect(stillOwned).toMatchObject({ status: 200 }); + expect(await stillOwned.text()).toBe("token a\n"); +}); + +test("Hosted Captun connect diagnostics do not spend connect rate-limit slots", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "2", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "token a\n", + clientIp: "203.0.113.77", + }); + + const firstConflict = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-b", + { headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.78" } }, + ); + const diagnostic = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-b", + { + headers: { + "cf-connecting-ip": "203.0.113.78", + [TUNNEL_CONNECT_DIAGNOSTIC_HEADER]: "1", + }, + }, + ); + const secondConflict = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-c", + { headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.78" } }, + ); + + expect(firstConflict).toMatchObject({ status: 409 }); + expect(diagnostic).toMatchObject({ status: 409 }); + expect(await diagnostic.text()).toBe("Tunnel name is already connected\n"); + expect(secondConflict).toMatchObject({ status: 409 }); +}); + +test("Hosted Captun connect diagnostics surface recent connect rate limits", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "1", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "token a\n", + clientIp: "203.0.113.79", + }); + + const rateLimited = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-b", + { headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.79" } }, + ); + const diagnostic = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-b", + { + headers: { + "cf-connecting-ip": "203.0.113.79", + [TUNNEL_CONNECT_DIAGNOSTIC_HEADER]: "1", + }, + }, + ); + + expect(rateLimited).toMatchObject({ status: 429 }); + expect(diagnostic).toMatchObject({ status: 429 }); + expect(await diagnostic.text()).toMatch(/^Rate limit exceeded\. Try again in \d+s\.\n$/); +}); + +test("Hosted Captun connect diagnostics fail closed when rate limiter binding is missing", async () => { + await using fixture = await createMiniflareWorkerFixture({ + entryPoint: "src/hosted/worker.ts", + durableObjects: { CaptunServerShard: { className: "CaptunServerShard" } }, + bindings: { CUSTOM_HOSTNAME: "captun.sh" }, + }); + + const diagnostic = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + { + headers: { + "cf-connecting-ip": "203.0.113.84", + [TUNNEL_CONNECT_DIAGNOSTIC_HEADER]: "1", + }, + }, + ); + + expect(diagnostic).toMatchObject({ status: 503 }); + expect(await diagnostic.text()).toBe("Hosted rate limiter is not configured\n"); +}); + +test("Hosted Captun lets the same token replace its active tunnel", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + using _firstTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "first\n", + clientIp: "203.0.113.80", + }); + using _secondTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: "second\n", + clientIp: "203.0.113.81", + }); + + const response = await fixture.worker.fetch("https://demo.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.82" }, + }); + + expect(response).toMatchObject({ status: 200 }); + expect(await response.text()).toBe("second\n"); +}); + +test("Hosted Captun requires anonymous tokens for public hosted connections", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + + const response = await fixture.worker.fetch( + "https://captun.sh/?captun-connect=1&captun-name=demo", + { + headers: { upgrade: "websocket", "cf-connecting-ip": "203.0.113.90" }, + }, + ); + + expect(response).toMatchObject({ status: 400 }); + expect(await response.text()).toBe("Missing tunnel token\n"); +}); + +test("Hosted Captun strips response cookies scoped outside the tunnel hostname", async () => { + await using fixture = await createHostedCaptunWorkerFixture({ + HOSTED_CONNECTS_PER_IP_PER_WINDOW: "100", + }); + using _tokenTunnel = await createDirectWorkerTunnel({ + fixture, + url: "https://captun.sh/?captun-connect=1&captun-name=demo&captun-token=token-a", + response: () => { + const headers = new Headers(); + headers.append("set-cookie", "host=1; Path=/; Secure"); + headers.append("set-cookie", "tunnel=1; Domain=demo.captun.sh; Path=/; Secure"); + headers.append("set-cookie", "child=1; Domain=.child.demo.captun.sh; Path=/; Secure"); + headers.append("set-cookie", "root=1; Domain=captun.sh; Path=/; Secure"); + headers.append("set-cookie", "dotroot=1; Domain=.captun.sh; Path=/; Secure"); + headers.append("set-cookie", "sibling=1; Domain=other.captun.sh; Path=/; Secure"); + headers.append( + "set-cookie", + "duplicate=1; Domain=demo.captun.sh; Domain=captun.sh; Path=/; Secure", + ); + return new Response("cookies\n", { headers }); + }, + clientIp: "203.0.113.92", + }); + + const response = await fixture.worker.fetch("https://demo.captun.sh/hello", { + headers: { "cf-connecting-ip": "203.0.113.93" }, + }); + + expect(response).toMatchObject({ status: 200 }); + expect(response.headers.getSetCookie()).toEqual([ + "host=1; Path=/; Secure", + "tunnel=1; Domain=demo.captun.sh; Path=/; Secure", + "child=1; Domain=.child.demo.captun.sh; Path=/; Secure", + ]); + expect(await response.text()).toBe("cookies\n"); +}); + test.each([ "account", "accounts", @@ -192,6 +566,47 @@ test.each([ expect(await response.text()).toBe("Reserved Captun tunnel name\n"); }); +async function createDirectWorkerTunnel(options: { + fixture: any; + url: string; + response: string | (() => Response); + clientIp: string; +}) { + const response = await options.fixture.worker.fetch(options.url, { + headers: { + upgrade: "websocket", + "cf-connecting-ip": options.clientIp, + }, + }); + expect(response).toMatchObject({ status: 101 }); + + const socket = response.webSocket; + socket.accept(); + const session = newWebSocketRpcSession(socket, new TestTunnelFetcher(options.response)); + + return { + [Symbol.dispose]() { + session[Symbol.dispose](); + }, + }; +} + +class TestTunnelFetcher extends RpcTarget { + private response: string | (() => Response); + + constructor(response: string | (() => Response)) { + super(); + this.response = response; + } + + fetch() { + if (typeof this.response === "string") return new Response(this.response); + return this.response(); + } + + ready() {} +} + function textareaValue(html: string, id: string) { const match = html.match(new RegExp(``)); if (!match) throw new Error(`Missing textarea #${id}`); diff --git a/test/miniflare.ts b/test/miniflare.ts index 08c08e4..6b275fd 100644 --- a/test/miniflare.ts +++ b/test/miniflare.ts @@ -62,6 +62,7 @@ export function createHostedCaptunWorkerFixture(bindings: Record entryPoint: "src/hosted/worker.ts", durableObjects: { CaptunServerShard: { className: "CaptunServerShard" }, + HostedRateLimiter: { className: "HostedRateLimiter" }, }, bindings: { CUSTOM_HOSTNAME: "captun.sh", ...bindings }, }); diff --git a/test/worker.test.ts b/test/worker.test.ts index e8739d8..020cf2d 100644 --- a/test/worker.test.ts +++ b/test/worker.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; + import { expect, test } from "vitest"; import { createCaptunTunnel } from "../src/index.js"; import { captunHealthResponse, isCaptunHealthRequest } from "../src/cli/tunnel-health.js"; @@ -190,6 +192,19 @@ test("Captun Worker routes subdomain tunnel requests when CUSTOM_HOSTNAME is set expect(await response.text()).toBe("No tunnel client connected\n"); }); +test("Captun Worker ignores hosted rate-limit bindings in self-hosted folder routing", async () => { + await using fixture = await createCaptunWorkerFixture({ + HOSTED_REQUESTS_PER_IP_PER_WINDOW: "1", + }); + const headers = { "cf-connecting-ip": "203.0.113.40" }; + + const first = await fixture.worker.fetch(`${fixture.origin}/one/hello`, { headers }); + const second = await fixture.worker.fetch(`${fixture.origin}/two/hello`, { headers }); + + expect(first).toMatchObject({ status: 503 }); + expect(second).toMatchObject({ status: 503 }); +}); + test("Captun Worker rejects missing tunnel names before Durable Object dispatch", async () => { await using fixture = await createCaptunWorkerFixture({}); @@ -229,3 +244,88 @@ test("Captun Worker rejects the legacy CAPTUN_SECRET binding", async () => { "CAPTUN_SECRET has been renamed to CAPTUN_TOKEN", ); }); + +test("createCaptunTunnel surfaces rejected WebSocket upgrade response details", async () => { + await using rejection = await createRejectedWebSocketUpgradeServer({ + status: 409, + body: "Tunnel name is already connected\n", + }); + + await expect( + createCaptunTunnel({ + gateway: rejection.origin, + name: "demo", + fetch: () => new Response("unused\n"), + }), + ).rejects.toThrow(/409 Conflict: Tunnel name is already connected/); +}); + +test("createCaptunTunnel falls back when the rejected upgrade probe does not respond", async () => { + await using rejection = await createRejectedWebSocketUpgradeServer({ + status: 409, + body: "Tunnel name is already connected\n", + neverRespondToHttp: true, + }); + + let caught: unknown; + try { + await createCaptunTunnel({ + gateway: rejection.origin, + name: "demo", + fetch: () => new Response("unused\n"), + }); + } catch (error) { + caught = error; + } + + expect(caught).toMatchObject({ message: "WebSocket connection failed" }); +}); + +async function createRejectedWebSocketUpgradeServer(options: { + status: number; + body: string; + neverRespondToHttp?: boolean; +}) { + const sockets = new Set<{ destroy: () => void }>(); + const statusText = options.status === 409 ? "Conflict" : "Rejected"; + const server = createServer((_request, response) => { + if (options.neverRespondToHttp) return; + response.writeHead(options.status, { + "content-type": "text/plain; charset=utf-8", + "cache-control": "no-store", + }); + response.end(options.body); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + }); + server.on("upgrade", (_request, socket) => { + socket.write( + [ + `HTTP/1.1 ${options.status} ${statusText}`, + "Content-Type: text/plain; charset=utf-8", + "Cache-Control: no-store", + `Content-Length: ${Buffer.byteLength(options.body)}`, + "Connection: close", + "", + options.body, + ].join("\r\n"), + ); + socket.destroy(); + }); + await new Promise((resolveListen, rejectListen) => { + server.once("error", rejectListen); + server.listen(0, "127.0.0.1", resolveListen); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Could not start test server"); + + return { + origin: `http://127.0.0.1:${address.port}`, + async [Symbol.asyncDispose]() { + for (const socket of sockets) socket.destroy(); + await new Promise((resolveClose) => server.close(() => resolveClose())); + }, + }; +} diff --git a/wrangler.hosted.jsonc b/wrangler.hosted.jsonc index 78348cc..03e7301 100644 --- a/wrangler.hosted.jsonc +++ b/wrangler.hosted.jsonc @@ -10,9 +10,15 @@ "head_sampling_rate": 1, }, "durable_objects": { - "bindings": [{ "name": "CaptunServerShard", "class_name": "CaptunServerShard" }], + "bindings": [ + { "name": "CaptunServerShard", "class_name": "CaptunServerShard" }, + { "name": "HostedRateLimiter", "class_name": "HostedRateLimiter" }, + ], }, - "migrations": [{ "tag": "v1", "new_sqlite_classes": ["CaptunServerShard"] }], + "migrations": [ + { "tag": "v1", "new_sqlite_classes": ["CaptunServerShard"] }, + { "tag": "v2", "new_sqlite_classes": ["HostedRateLimiter"] }, + ], "vars": { "CUSTOM_HOSTNAME": "captun.sh", },