From 522f06ffc69a581b02bd4e928013ec9868c03084 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 12 Aug 2026 09:59:29 -0700 Subject: [PATCH] feat(worker): support direct Gateway connections (#122683) * feat(worker): support direct gateway connections * docs(plan): link node worker provider PR --- .../agent-harness-runtime.json | 2 +- .../agent-harness.json | 2 +- .../plugin-sdk-api-baseline/channel-core.json | 2 +- .../channel-entry-contract.json | 2 +- .../channel-message.json | 2 +- .../channel-outbound.json | 2 +- .../channel-plugin-common.json | 2 +- .../plugin-sdk-api-baseline/core.json | 2 +- .../plugin-sdk-api-baseline/discord.json | 2 +- .../gateway-runtime.json | 2 +- .../inbound-reply-dispatch.json | 2 +- .../meeting-runtime.json | 2 +- .../plugin-sdk-api-baseline/plugin-entry.json | 2 +- .../plugin-runtime.json | 2 +- .../provider-catalog-runtime.json | 2 +- .../plugin-sdk-api-baseline/tool-plugin.json | 2 +- .../webhook-ingress.json | 2 +- docs/plan/runners.md | 4 +- packages/gateway-client/src/client.ts | 223 +++--------------- .../gateway-client/src/websocket-transport.ts | 177 ++++++++++++++ src/gateway/server.preauth-hardening.test.ts | 92 +++----- .../environment-access.test.ts | 6 +- .../placement-dispatch-test-harness.ts | 3 +- .../worker-environments/tunnel-contract.ts | 11 +- .../worker-environments/tunnel.test.ts | 16 ++ src/gateway/worker-environments/tunnel.ts | 28 ++- .../worker-turn-launcher.test.ts | 93 ++++---- .../worker-turn-launcher.ts | 12 +- .../worker-turn-payload.test.ts | 4 +- src/worker/launch-descriptor.test.ts | 46 +++- src/worker/launch-descriptor.ts | 21 +- src/worker/worker-connection-admission.ts | 31 +-- src/worker/worker-connection-contract.ts | 7 +- src/worker/worker-connection-endpoint.test.ts | 62 +++++ src/worker/worker-connection-endpoint.ts | 122 ++++++++++ src/worker/worker-connection.test.ts | 25 +- src/worker/worker-connection.ts | 5 + .../worker-fault-injection.test-support.ts | 6 +- src/worker/worker.runtime.test.ts | 18 +- src/worker/worker.runtime.ts | 2 +- 40 files changed, 672 insertions(+), 376 deletions(-) create mode 100644 packages/gateway-client/src/websocket-transport.ts create mode 100644 src/worker/worker-connection-endpoint.test.ts create mode 100644 src/worker/worker-connection-endpoint.ts diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json index 670f7cb4fe5a..5d3037193d13 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json @@ -1 +1 @@ -{"contentHash":"dc2dfe016cb61aa0c79c9647a53eff358ceeb1859859c0d0b6ca752160a879ee","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} +{"contentHash":"81edf86f9ac989d2c85a531a271396c8ca553bd40f80a3e93903b3cf34385146","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json index 56424b713962..230896ca2b18 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json @@ -1 +1 @@ -{"contentHash":"5375849cad6199c5faf3e16a8e67943be3c980b0587a6055d5ad3cd0f3b6d54b","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} +{"contentHash":"1bb99b68596361bc2ef9f1c8011f1d87fbbdc78d0bde58d9e89625a5e2bf2de7","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-core.json b/docs/.generated/plugin-sdk-api-baseline/channel-core.json index b71b5ac2d575..2754698ca1c0 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-core.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-core.json @@ -1 +1 @@ -{"contentHash":"983b9fd128e62f7bb399ba48879ab4817d399d5b28e935cd79468a96833015f5","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} +{"contentHash":"6c638b1fcf5a1cadce5a5363ae4cd8f29fe513feb90331272349ef5bf70bc63a","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json index 995b66585ad2..9d2d3ab6b776 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json @@ -1 +1 @@ -{"contentHash":"bff26d38a6bbe8e5f7c7bfce771c30a8898f445de4e37688dc89870ccd260c8d","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} +{"contentHash":"fa4140b50658ecfab11c323c24ba6f0fe0a5cb1608c8fa3e8eefa3e4e4192e77","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-message.json b/docs/.generated/plugin-sdk-api-baseline/channel-message.json index 2c4eba0d1336..d301e92a8907 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-message.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-message.json @@ -1 +1 @@ -{"contentHash":"a1a59370087b8f53d5fb24945ac2f0f3b516c7ca0c7d48f6293cd0b223c14f3d","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} +{"contentHash":"5c1b1998403a8527055fff05caa86f51ccd981c08ebd8ba8c7f9da19a8e16e33","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json index 57fe570b9d7d..60e4a3165445 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json @@ -1 +1 @@ -{"contentHash":"a1f6948c41d1a539819ae29e15b10a8d0db8a3a4f46dd24d5d0818da449ed727","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} +{"contentHash":"751d37b918948737d4324f6323e1f8f6151f991d55c3794e1908f79225745fc7","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json index 471a7e399671..7c34100884d7 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json @@ -1 +1 @@ -{"contentHash":"be6416bf69922f215e9a1f76a4fade27af0a74e3214d99b1ca1898b108c60490","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} +{"contentHash":"87d04d89c9a3b50fc63ea3da73e04bf4f6e5ce5738cfe1be11d4023aa680ec78","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} diff --git a/docs/.generated/plugin-sdk-api-baseline/core.json b/docs/.generated/plugin-sdk-api-baseline/core.json index f7f0e902f9a2..45d89448009f 100644 --- a/docs/.generated/plugin-sdk-api-baseline/core.json +++ b/docs/.generated/plugin-sdk-api-baseline/core.json @@ -1 +1 @@ -{"contentHash":"183298d9056641c8b595b8e35427138200bdd09c37d973c7ba5073126761189e","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} +{"contentHash":"2c95ec54b192b730e9a9c606f77e66a2d0dd9eb09dd66a3c6ad8fea9fca31585","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/discord.json b/docs/.generated/plugin-sdk-api-baseline/discord.json index 74fe7e6f199c..1e7953515e20 100644 --- a/docs/.generated/plugin-sdk-api-baseline/discord.json +++ b/docs/.generated/plugin-sdk-api-baseline/discord.json @@ -1 +1 @@ -{"contentHash":"662e95032c11179b8c3010e0792f23efb583dff086a3289c7351d75e5349d221","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} +{"contentHash":"d2051c433956de7d579b91d4e21e39d48334e63a632baf941bde0cef85ca8ac0","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} diff --git a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json index d647003dad6f..50ead4f63efc 100644 --- a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json @@ -1 +1 @@ -{"contentHash":"cfdf45c07d1df4fc251da2be882bbc57228148c88ffb8bd7fc3f6d53951bb89b","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} +{"contentHash":"adf5aa7e44a7bc6f6e0607fa35eae1fcb197749bb5a129c1848f85d61dc319a2","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json index a60fb9b5d4a1..d72a18256642 100644 --- a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json +++ b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json @@ -1 +1 @@ -{"contentHash":"1c90dcb9cd3cf2d53eca4a162201584cc7ed3cce67e6315258c2af40381a8ada","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} +{"contentHash":"a9e35fd79c8ba80ca1de7760b0b83a243a383bffbda1730818e5dec2a049ddce","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} diff --git a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json index e9508b029498..abac3ff08599 100644 --- a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json @@ -1 +1 @@ -{"contentHash":"7e36c02a156e70484c750a54d8771370ad97a6788d9c69ddd5fe7e768be4507d","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} +{"contentHash":"3ba67320fb902e12fdf2399c4d8e74403e78e74add8dfed8213154f4ab68a858","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json index a35b32b108c3..ff8ab001e754 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json @@ -1 +1 @@ -{"contentHash":"2e7ca18c70c9726f5f26496e015de059badfd83d3c36b43b4dcc640f286ecb74","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} +{"contentHash":"ca3136e446cf1fc294279e24078e3a4c6342a048b00f8012d519903c887ef3a7","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json index 5885b0f1d2c1..cb6c5737e5e5 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json @@ -1 +1 @@ -{"contentHash":"d2d5b75b1cf49439ef71a2b9e46db98fcd6431be792a63a6907e50d9247fe82b","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} +{"contentHash":"5f0b7be43fcb5e2240053e7ada316b35f28cbcdd0619b724463627961927dcb9","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json index 02ec719c3da8..7076b149adb2 100644 --- a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json @@ -1 +1 @@ -{"contentHash":"9d621db60ab93bb5b6cbaa6107030f780b4c863420ab7bf01e309f8c356bca5e","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} +{"contentHash":"5c9f0b8207f962f9c32228970abe15c23de32c08a00e3d42cbdf5fecbd4c1f1e","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json index 0e27008f7e63..9444a64e5a85 100644 --- a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json +++ b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json @@ -1 +1 @@ -{"contentHash":"0898758caccfb3cb789606e81cf692dc919e7f993a4d2f9c4487583d7e05f87a","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} +{"contentHash":"018adbdc87ee4a490fdea45b3ee249a55dca337d655cad8929e63986f4e633bb","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} diff --git a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json index 9f5caf27925b..8910c15ff1e5 100644 --- a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json +++ b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json @@ -1 +1 @@ -{"contentHash":"740dc824333ee4e1bfdfe410c8c9987914f0699f36e7a19d660e69d9bcd908d7","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} +{"contentHash":"1dac77955687176848405413e59d76ea0511e30722eaa32bf3a3cdd44c5a6395","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} diff --git a/docs/plan/runners.md b/docs/plan/runners.md index b947420fd465..d08f53f1639c 100644 --- a/docs/plan/runners.md +++ b/docs/plan/runners.md @@ -23,8 +23,8 @@ advances a milestone. | 3 | `openclaw connect` one-paste onboarding + `/j/` join route | in progress | #120768, #122499 | | 4 | Picker: grouping, placement, liveness, enrichment | in progress | #120804, #122531, #122635 | | F | Real-wire session boundary harness | landed | #121212 | -| 5 | Public worker ingress path | in progress | #122578, #122643 | -| 6 | Node worker provider (device runners) | not started | — | +| 5 | Public worker ingress path | landed | #122578, #122643 | +| 6 | Node worker provider (device runners) | in progress | #122683 | | 7 | Bundle push consent + runner updates | not started | — | | 8 | Stop-and-continue moves | not started | — | | 9 | Deletions (ssh sandbox, openshell, exec-host clones, …) | not started | — | diff --git a/packages/gateway-client/src/client.ts b/packages/gateway-client/src/client.ts index cd8a8a368b06..a9b956caf1e6 100644 --- a/packages/gateway-client/src/client.ts +++ b/packages/gateway-client/src/client.ts @@ -18,15 +18,11 @@ import { MIN_PROBE_PROTOCOL_VERSION, PROTOCOL_VERSION, } from "@openclaw/gateway-protocol/version"; -import { isLoopbackIpAddress, type ParsedIpAddress } from "@openclaw/net-policy/ip"; -import { isWssUrl } from "@openclaw/net-policy/url-protocol"; -import { WebSocket, type ClientOptions, type CertMeta } from "ws"; +import { WebSocket } from "ws"; import { isSensitiveUrlQueryParamName, normalizeFingerprint, normalizeGatewayErrorText, - parseGatewayIpAddress, - parseHostForAddressChecks, } from "./client-address-utils.js"; import { buildGatewayConnectAuth, @@ -56,6 +52,11 @@ import { resolveSafeTimeoutDelayMs, } from "./timeouts.js"; import { rawDataToString } from "./websocket-data.js"; +import { + GatewayWebSocketTransportConfigurationError, + isGatewayLoopbackHost, + resolveGatewayWebSocketTransport, +} from "./websocket-transport.js"; export type DeviceIdentity = { deviceId: string; @@ -127,91 +128,6 @@ function resolveHostDeps(overrides?: GatewayClientHostDeps): Required; } -const PRIVATE_OR_LOOPBACK_IPV4_RANGES = new Set([ - "loopback", - "private", - "linkLocal", - "carrierGradeNat", -]); - -const PRIVATE_OR_LOOPBACK_IPV6_RANGES = new Set([ - "loopback", - "linkLocal", - "uniqueLocal", - "deprecatedSiteLocal", -]); - -function isPrivateOrLoopbackIpAddress(address: ParsedIpAddress): boolean { - const ranges = - address.kind() === "ipv4" ? PRIVATE_OR_LOOPBACK_IPV4_RANGES : PRIVATE_OR_LOOPBACK_IPV6_RANGES; - return ranges.has(address.range()); -} - -function isLoopbackHost(host: string): boolean { - const parsed = parseHostForAddressChecks(host); - if (!parsed) { - return false; - } - if (parsed.isLocalhost) { - return true; - } - return isLoopbackIpAddress(parsed.unbracketedHost); -} - -function isPrivateOrLoopbackHost(host: string): boolean { - const parsed = parseHostForAddressChecks(host); - if (!parsed) { - return false; - } - if (parsed.isLocalhost) { - return true; - } - const address = parseGatewayIpAddress(parsed.unbracketedHost); - if (!address) { - return false; - } - return isPrivateOrLoopbackIpAddress(address); -} - -function isTrustedPlaintextWebSocketHost(hostname: string): boolean { - if (isPrivateOrLoopbackHost(hostname)) { - return true; - } - const normalized = hostname.toLowerCase().trim().replace(/\.+$/, ""); - // Plain ws:// is still useful for local discovery and Tailnet names. Public - // hostnames must use wss:// unless the caller opts into the private break-glass. - return normalized.endsWith(".local") || normalized.endsWith(".ts.net"); -} - -function isSecureWebSocketUrl(rawUrl: string, options?: { allowPrivateWs?: boolean }): boolean { - try { - const url = new URL(rawUrl); - const protocol = - url.protocol === "https:" ? "wss:" : url.protocol === "http:" ? "ws:" : url.protocol; - if (protocol === "wss:") { - return true; - } - if (protocol !== "ws:") { - return false; - } - if (isLoopbackHost(url.hostname) || isTrustedPlaintextWebSocketHost(url.hostname)) { - return true; - } - if (options?.allowPrivateWs === true) { - const hostForIpCheck = - url.hostname.startsWith("[") && url.hostname.endsWith("]") - ? url.hostname.slice(1, -1) - : url.hostname; - return ( - isPrivateOrLoopbackHost(url.hostname) || parseGatewayIpAddress(hostForIpCheck) === undefined - ); - } - return false; - } catch { - return false; - } -} - export type GatewayClientRequestOptions = GatewayProtocolRequestOptions; type AssembledConnect = { @@ -224,10 +140,6 @@ type AssembledConnect = { usingStoredDeviceToken: boolean | undefined; }; -type FingerprintCheckingClientOptions = Omit & { - checkServerIdentity?: (servername: string, cert: CertMeta) => Error | undefined; -}; - const DEFAULT_GATEWAY_CLIENT_URL = "ws://127.0.0.1:18789"; const DEFAULT_CLIENT_VERSION = "0.0.0"; const MAX_UPGRADE_ERROR_BODY_BYTES = 2 * 1024; @@ -299,9 +211,7 @@ export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTime } } -class GatewayClientSocketFactoryConfigurationError extends Error {} - -class GatewayClientTransportPolicyError extends GatewayClientSocketFactoryConfigurationError {} +class GatewayClientTransportPolicyError extends GatewayWebSocketTransportConfigurationError {} const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError"); @@ -531,7 +441,7 @@ export class GatewayClient { reconnect: { initialMs: 1_000, multiplier: 2, maxMs: 30_000 }, requestTimeoutMs: this.requestTimeoutMs, shouldRetrySocketFactoryError: (error) => - !(error instanceof GatewayClientSocketFactoryConfigurationError) && + !(error instanceof GatewayWebSocketTransportConfigurationError) && !(error instanceof SyntaxError) && !(error instanceof TypeError) && !(error instanceof RangeError), @@ -570,72 +480,26 @@ export class GatewayClient { private createSocket(handlers: GatewayProtocolSocketHandlers): GatewayProtocolSocket { const url = this.opts.url ?? DEFAULT_GATEWAY_CLIENT_URL; - const usesTls = isWssUrl(url); - if (this.opts.tlsFingerprint && !usesTls) { - throw new GatewayClientSocketFactoryConfigurationError( - "gateway tls fingerprint requires wss:// gateway url", - ); - } - - const allowPrivateWs = - (this.opts.env ?? process.env).OPENCLAW_ALLOW_INSECURE_PRIVATE_WS === "1"; // Block plaintext before device-token lookup. Credentials may be loaded from // host storage later in sendConnect(), and chat payloads are sensitive too. - if (!isSecureWebSocketUrl(url, { allowPrivateWs })) { - // Safe hostname extraction - avoid throwing on malformed URLs in error path - let displayHost = url; - try { - displayHost = new URL(url).hostname || url; - } catch { - // Use raw URL if parsing fails - } - throw new GatewayClientSocketFactoryConfigurationError( - `SECURITY ERROR: Cannot connect to "${displayHost}" over plaintext ws://. ` + - "Both credentials and chat data would be exposed to network interception. " + - "Use wss:// for remote URLs. Safe defaults: keep gateway.bind=loopback and connect via SSH tunnel " + - "(ssh -N -L 18789:127.0.0.1:18789 user@gateway-host), or use Tailscale Serve/Funnel. " + - (allowPrivateWs - ? "" - : "Break-glass (trusted private networks only): set OPENCLAW_ALLOW_INSECURE_PRIVATE_WS=1. ") + - "Run `openclaw doctor --fix` for guidance.", - ); - } - // Allow node screen snapshots and other large responses. - this.deps.beforeConnect(); - // Challenge timeout arms only after `open`. Bound the opening handshake so a - // peer that accepts TCP without upgrading cannot hang createSocket forever. const handshakeTimeoutMs = resolvePreauthHandshakeTimeoutMs({ env: this.opts.env, configuredTimeoutMs: this.opts.preauthHandshakeTimeoutMs, }); - const wsOptions: FingerprintCheckingClientOptions = { - maxPayload: 25 * 1024 * 1024, - handshakeTimeout: handshakeTimeoutMs, - ...(this.opts.origin ? { origin: this.opts.origin } : {}), - }; - if (usesTls && this.opts.tlsFingerprint) { - wsOptions.rejectUnauthorized = false; - wsOptions.checkServerIdentity = (_hostValue: string, cert: CertMeta) => { - const fingerprintValue = - typeof cert === "object" && cert && "fingerprint256" in cert - ? ((cert as { fingerprint256?: string }).fingerprint256 ?? "") - : ""; - const fingerprint = this.deps.normalizeTlsFingerprint( - typeof fingerprintValue === "string" ? fingerprintValue : "", - ); - const expected = this.deps.normalizeTlsFingerprint(this.opts.tlsFingerprint ?? ""); - if (!expected) { - return undefined; - } - if (!fingerprint) { - return new Error("Missing server TLS fingerprint"); - } - if (fingerprint !== expected) { - return new Error("Server TLS fingerprint mismatch"); - } - return undefined; - }; - } + const transport = resolveGatewayWebSocketTransport({ + url, + tlsFingerprint: this.opts.tlsFingerprint, + env: this.opts.env, + normalizeTlsFingerprint: this.deps.normalizeTlsFingerprint, + options: { + // Allow node screen snapshots and other large responses. The challenge + // timer starts after open, so separately bound the HTTP upgrade here. + maxPayload: 25 * 1024 * 1024, + handshakeTimeout: handshakeTimeoutMs, + ...(this.opts.origin ? { origin: this.opts.origin } : {}), + }, + }); + this.deps.beforeConnect(); let ws: WebSocket; // Managed proxies can intercept local traffic; the host owns the bypass // lifecycle and must remove it immediately after the socket is created. @@ -648,7 +512,7 @@ export class GatewayClient { ); } try { - ws = new WebSocket(url, wsOptions as ClientOptions); + ws = new WebSocket(url, transport.options); ws.binaryType = "nodebuffer"; } catch (error) { throw error instanceof Error ? error : new Error(String(error)); @@ -660,13 +524,11 @@ export class GatewayClient { let upgradeError: GatewayClientRequestError | undefined; ws.on("open", () => { handlers.open(); - if (usesTls && this.opts.tlsFingerprint) { - const tlsError = this.validateTlsFingerprint(); - if (tlsError) { - handlers.error(tlsError); - ws.close(1008, tlsError.message); - return; - } + const tlsError = transport.validateSocket(ws); + if (tlsError) { + handlers.error(tlsError); + ws.close(1008, tlsError.message); + return; } this.transportValidated = true; }); @@ -1327,7 +1189,7 @@ export class GatewayClient { : parsed.protocol === "http:" ? "ws:" : parsed.protocol; - if (isLoopbackHost(parsed.hostname)) { + if (isGatewayLoopbackHost(parsed.hostname)) { return true; } return protocol === "wss:" && Boolean(this.opts.tlsFingerprint?.trim()); @@ -1399,33 +1261,6 @@ export class GatewayClient { }, interval); } - private validateTlsFingerprint(): Error | null { - if (!this.opts.tlsFingerprint || !this.ws) { - return null; - } - const expected = this.deps.normalizeTlsFingerprint(this.opts.tlsFingerprint); - if (!expected) { - return new Error("gateway tls fingerprint missing"); - } - const socket = ( - this.ws as WebSocket & { - _socket?: { getPeerCertificate?: () => { fingerprint256?: string } }; - } - )["_socket"]; - if (!socket || typeof socket.getPeerCertificate !== "function") { - return new Error("gateway tls fingerprint unavailable"); - } - const cert = socket.getPeerCertificate(); - const fingerprint = this.deps.normalizeTlsFingerprint(cert?.fingerprint256 ?? ""); - if (!fingerprint) { - return new Error("gateway tls fingerprint unavailable"); - } - if (fingerprint !== expected) { - return new Error("gateway tls fingerprint mismatch"); - } - return null; - } - async request>( method: string, params?: unknown, diff --git a/packages/gateway-client/src/websocket-transport.ts b/packages/gateway-client/src/websocket-transport.ts new file mode 100644 index 000000000000..4f573e5408b2 --- /dev/null +++ b/packages/gateway-client/src/websocket-transport.ts @@ -0,0 +1,177 @@ +import { isLoopbackIpAddress, type ParsedIpAddress } from "@openclaw/net-policy/ip"; +import { isWssUrl } from "@openclaw/net-policy/url-protocol"; +import type { ClientOptions, CertMeta, WebSocket } from "ws"; +import { + normalizeFingerprint, + parseGatewayIpAddress, + parseHostForAddressChecks, +} from "./client-address-utils.js"; + +const PRIVATE_OR_LOOPBACK_IPV4_RANGES = new Set([ + "loopback", + "private", + "linkLocal", + "carrierGradeNat", +]); +const PRIVATE_OR_LOOPBACK_IPV6_RANGES = new Set([ + "loopback", + "linkLocal", + "uniqueLocal", + "deprecatedSiteLocal", +]); + +function isPrivateOrLoopbackIpAddress(address: ParsedIpAddress): boolean { + const ranges = + address.kind() === "ipv4" ? PRIVATE_OR_LOOPBACK_IPV4_RANGES : PRIVATE_OR_LOOPBACK_IPV6_RANGES; + return ranges.has(address.range()); +} + +export function isGatewayLoopbackHost(host: string): boolean { + const parsed = parseHostForAddressChecks(host); + return Boolean(parsed && (parsed.isLocalhost || isLoopbackIpAddress(parsed.unbracketedHost))); +} + +function isPrivateOrLoopbackHost(host: string): boolean { + const parsed = parseHostForAddressChecks(host); + if (!parsed) { + return false; + } + if (parsed.isLocalhost) { + return true; + } + const address = parseGatewayIpAddress(parsed.unbracketedHost); + return Boolean(address && isPrivateOrLoopbackIpAddress(address)); +} + +function isTrustedPlaintextWebSocketHost(hostname: string): boolean { + if (isPrivateOrLoopbackHost(hostname)) { + return true; + } + const normalized = hostname.toLowerCase().trim().replace(/\.+$/, ""); + return normalized.endsWith(".local") || normalized.endsWith(".ts.net"); +} + +function isSecureWebSocketUrl(rawUrl: string, options?: { allowPrivateWs?: boolean }): boolean { + try { + const url = new URL(rawUrl); + const protocol = + url.protocol === "https:" ? "wss:" : url.protocol === "http:" ? "ws:" : url.protocol; + if (protocol === "wss:") { + return true; + } + if (protocol !== "ws:") { + return false; + } + if (isGatewayLoopbackHost(url.hostname) || isTrustedPlaintextWebSocketHost(url.hostname)) { + return true; + } + if (options?.allowPrivateWs === true) { + const hostForIpCheck = + url.hostname.startsWith("[") && url.hostname.endsWith("]") + ? url.hostname.slice(1, -1) + : url.hostname; + return ( + isPrivateOrLoopbackHost(url.hostname) || parseGatewayIpAddress(hostForIpCheck) === undefined + ); + } + return false; + } catch { + return false; + } +} + +export class GatewayWebSocketTransportConfigurationError extends Error {} + +type FingerprintCheckingClientOptions = Omit & { + checkServerIdentity?: (servername: string, cert: CertMeta) => Error | undefined; +}; + +type GatewayWebSocketTransport = { + options: ClientOptions; + validateSocket(socket: WebSocket): Error | null; +}; + +export function resolveGatewayWebSocketTransport(params: { + url: string; + tlsFingerprint?: string; + env?: NodeJS.ProcessEnv; + options: Omit; + normalizeTlsFingerprint?: (fingerprint: string | undefined) => string; +}): GatewayWebSocketTransport { + const usesTls = isWssUrl(params.url); + if (params.tlsFingerprint && !usesTls) { + throw new GatewayWebSocketTransportConfigurationError( + "gateway tls fingerprint requires wss:// gateway url", + ); + } + const allowPrivateWs = (params.env ?? process.env).OPENCLAW_ALLOW_INSECURE_PRIVATE_WS === "1"; + if (!isSecureWebSocketUrl(params.url, { allowPrivateWs })) { + let displayHost = params.url; + try { + displayHost = new URL(params.url).hostname || params.url; + } catch { + // Use the raw URL when syntax is malformed. + } + throw new GatewayWebSocketTransportConfigurationError( + `SECURITY ERROR: Cannot connect to "${displayHost}" over plaintext ws://. ` + + "Both credentials and chat data would be exposed to network interception. " + + "Use wss:// for remote URLs. Safe defaults: keep gateway.bind=loopback and connect via SSH tunnel " + + "(ssh -N -L 18789:127.0.0.1:18789 user@gateway-host), or use Tailscale Serve/Funnel. " + + (allowPrivateWs + ? "" + : "Break-glass (trusted private networks only): set OPENCLAW_ALLOW_INSECURE_PRIVATE_WS=1. ") + + "Run `openclaw doctor --fix` for guidance.", + ); + } + + const normalize = params.normalizeTlsFingerprint ?? normalizeFingerprint; + const options: FingerprintCheckingClientOptions = { ...params.options }; + if (usesTls && params.tlsFingerprint) { + options.rejectUnauthorized = false; + options.checkServerIdentity = (_hostValue: string, cert: CertMeta) => { + const fingerprintValue = + typeof cert === "object" && cert && "fingerprint256" in cert + ? ((cert as { fingerprint256?: string }).fingerprint256 ?? "") + : ""; + const fingerprint = normalize(typeof fingerprintValue === "string" ? fingerprintValue : ""); + const expected = normalize(params.tlsFingerprint); + if (!expected) { + return undefined; + } + if (!fingerprint) { + return new Error("Missing server TLS fingerprint"); + } + if (fingerprint !== expected) { + return new Error("Server TLS fingerprint mismatch"); + } + return undefined; + }; + } + + return { + options: options as ClientOptions, + validateSocket: (socket) => { + if (!params.tlsFingerprint) { + return null; + } + const expected = normalize(params.tlsFingerprint); + if (!expected) { + return new Error("gateway tls fingerprint missing"); + } + const rawSocket = ( + socket as WebSocket & { + _socket?: { getPeerCertificate?: () => { fingerprint256?: string } }; + } + )["_socket"]; + if (!rawSocket || typeof rawSocket.getPeerCertificate !== "function") { + return new Error("gateway tls fingerprint unavailable"); + } + const cert = rawSocket.getPeerCertificate(); + const fingerprint = normalize(cert?.fingerprint256 ?? ""); + if (!fingerprint) { + return new Error("gateway tls fingerprint unavailable"); + } + return fingerprint === expected ? null : new Error("gateway tls fingerprint mismatch"); + }, + }; +} diff --git a/src/gateway/server.preauth-hardening.test.ts b/src/gateway/server.preauth-hardening.test.ts index 568c264d67c5..ff15c0fd03e4 100644 --- a/src/gateway/server.preauth-hardening.test.ts +++ b/src/gateway/server.preauth-hardening.test.ts @@ -2,7 +2,6 @@ * Gateway pre-auth hardening tests. */ import http from "node:http"; -import { rawDataToString } from "@openclaw/gateway-client/websocket-data"; import { afterEach, describe, expect, it, vi } from "vitest"; import { WebSocket, WebSocketServer } from "ws"; import { @@ -22,6 +21,7 @@ import { tryBeginGatewaySuspendAdmission, } from "../process/gateway-work-admission.js"; import { captureEnv, setTestEnvValue } from "../test-utils/env.js"; +import { createWorkerConnection } from "../worker/worker-connection.js"; import type { ResolvedGatewayAuth } from "./auth.js"; import { MAX_PREAUTH_PAYLOAD_BYTES } from "./server-constants.js"; import { @@ -232,7 +232,7 @@ describe("gateway pre-auth hardening", () => { } }); - it("admits a valid worker over the public path without a gateway challenge", async () => { + it("admits the production worker client over the public path without a gateway challenge", async () => { const clients = new Set(); const resolvedAuth: ResolvedGatewayAuth = { mode: "none", allowTailscale: false }; const httpServer = createGatewayHttpServer({ @@ -303,62 +303,48 @@ describe("gateway pre-auth hardening", () => { }); const address = httpServer.address(); const port = typeof address === "object" && address ? address.port : 0; - const client = new WebSocket(`ws://127.0.0.1:${port}${WORKER_PUBLIC_INGRESS_PATH}`); - const received: unknown[] = []; - client.on("message", (data) => received.push(JSON.parse(rawDataToString(data)))); + const client = createWorkerConnection({ + endpoint: { + kind: "websocket", + url: `ws://127.0.0.1:${port}${WORKER_PUBLIC_INGRESS_PATH}`, + }, + connectParams: { + minProtocol: PROTOCOL_VERSION, + maxProtocol: PROTOCOL_VERSION, + client: { + id: GATEWAY_CLIENT_IDS.WORKER, + version: "2026.8.12", + platform: "linux", + mode: GATEWAY_CLIENT_MODES.WORKER, + }, + role: "worker", + admission: { + environmentId: "worker-public", + credential: "public-worker-credential", + sessionId: null, + runId: null, + ownerEpoch: 1, + rpcSetVersion: 1, + handshake: { + bundleHash: "a".repeat(64), + openclawVersion: "2026.8.12", + protocolFeatures: [], + }, + }, + }, + reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, + admissionTimeoutMs: 2_000, + }); try { - await new Promise((resolve, reject) => { - client.once("open", resolve); - client.once("error", reject); + await client.start(); + expect(client.state).toMatchObject({ + kind: "ready", + hello: { type: "worker-hello-ok", environmentId: "worker-public" }, }); - client.send( - JSON.stringify({ - type: "req", - id: "connect-public-worker", - method: "connect", - params: { - minProtocol: PROTOCOL_VERSION, - maxProtocol: PROTOCOL_VERSION, - client: { - id: GATEWAY_CLIENT_IDS.WORKER, - version: "2026.8.12", - platform: "linux", - mode: GATEWAY_CLIENT_MODES.WORKER, - }, - role: "worker", - admission: { - environmentId: "worker-public", - credential: "public-worker-credential", - sessionId: null, - runId: null, - ownerEpoch: 1, - rpcSetVersion: 1, - handshake: { - bundleHash: "a".repeat(64), - openclawVersion: "2026.8.12", - protocolFeatures: [], - }, - }, - }, - }), - ); - await vi.waitFor(() => expect(received).toHaveLength(1)); - expect(received[0]).toMatchObject({ - type: "res", - id: "connect-public-worker", - ok: true, - payload: { type: "worker-hello-ok", environmentId: "worker-public" }, - }); - expect(received).not.toContainEqual( - expect.objectContaining({ type: "event", event: "connect.challenge" }), - ); expect(workerConnectionService.admitWorker).toHaveBeenCalledOnce(); } finally { - client.close(); - await new Promise((resolve) => { - client.once("close", () => resolve()); - }); + await client.stop(); await new Promise((resolve) => { wss.close(() => resolve()); }); diff --git a/src/gateway/worker-environments/environment-access.test.ts b/src/gateway/worker-environments/environment-access.test.ts index ad2ffedc0439..5c6a187f8611 100644 --- a/src/gateway/worker-environments/environment-access.test.ts +++ b/src/gateway/worker-environments/environment-access.test.ts @@ -23,7 +23,8 @@ describe("worker environment service", () => { return { environmentId: request.environmentId, ownerEpoch: request.ownerEpoch, - remoteSocketPath: "/tmp/worker/gateway.sock", + connectionEndpoint: { kind: "unix", socketPath: "/tmp/worker/gateway.sock" }, + launchTurn: vi.fn(), runWorkspaceCommand: vi.fn(), syncWorkspace: vi.fn(), stop: async () => {}, @@ -86,7 +87,8 @@ describe("worker environment service", () => { start: vi.fn(async (request: Parameters[0]) => ({ environmentId: request.environmentId, ownerEpoch: request.ownerEpoch, - remoteSocketPath: "/tmp/worker/gateway.sock", + connectionEndpoint: { kind: "unix", socketPath: "/tmp/worker/gateway.sock" }, + launchTurn: vi.fn(), runWorkspaceCommand: vi.fn(), syncWorkspace: vi.fn(), stop: async () => {}, diff --git a/src/gateway/worker-environments/placement-dispatch-test-harness.ts b/src/gateway/worker-environments/placement-dispatch-test-harness.ts index a2e3b31ab5d5..71092d2d4c98 100644 --- a/src/gateway/worker-environments/placement-dispatch-test-harness.ts +++ b/src/gateway/worker-environments/placement-dispatch-test-harness.ts @@ -175,7 +175,8 @@ export function createHarness( const tunnelHandle = (ownerEpoch: number): WorkerTunnelHandle => ({ environmentId: ready.environmentId, ownerEpoch, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix", socketPath: "/worker/gateway.sock" }, + launchTurn: vi.fn(), quiesceWorkspace: vi.fn(async () => { log.push("workspace:quiesce"); return { diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts index b054309c9798..b240a18d0f9f 100644 --- a/src/gateway/worker-environments/tunnel-contract.ts +++ b/src/gateway/worker-environments/tunnel-contract.ts @@ -1,4 +1,6 @@ import type { SpawnResult } from "../../process/exec.js"; +import type { WorkerLaunchDescriptor } from "../../worker/launch-descriptor.js"; +import type { WorkerConnectionEndpoint } from "../../worker/worker-connection-endpoint.js"; import type { WorkerWorkspaceApplyResult, WorkerWorkspaceReconciliationJournalAdapter, @@ -72,10 +74,17 @@ export type WorkerWorkspaceQuiescence = { resume(): Promise; }; +type WorkerTurnLaunchRequest = { + descriptor: WorkerLaunchDescriptor; + timeoutMs?: number; + signal?: AbortSignal; +}; + export type WorkerTunnelHandle = { environmentId: string; ownerEpoch: number; - remoteSocketPath: string; + connectionEndpoint: WorkerConnectionEndpoint; + launchTurn(request: WorkerTurnLaunchRequest): Promise; runWorkspaceCommand(command: WorkerWorkspaceCommand): Promise; quiesceWorkspace(remoteWorkspaceDir: string): Promise; syncWorkspace(request: WorkerWorkspaceSyncRequest): Promise; diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index 2d9093f5c422..4250a19c865e 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from "vitest"; +import type { WorkerLaunchDescriptor } from "../../worker/launch-descriptor.js"; import { createWorkerSshRunner } from "./tunnel-ssh-runner.js"; import { createWorkerTunnelManager } from "./tunnel.js"; import { @@ -68,6 +69,21 @@ describe("worker tunnel manager", () => { expect(workspace?.argv).toContain("ControlPath=none"); expect(workspace?.argv.at(-1)).toContain("pwd"); expect(fake.starts).toHaveLength(1); + expect(handle.connectionEndpoint).toMatchObject({ + kind: "unix", + socketPath: expect.stringMatching(/\/gateway\.sock$/u), + }); + const descriptor = { version: 3 } as unknown as WorkerLaunchDescriptor; + await expect(handle.launchTurn({ descriptor, timeoutMs: 123 })).resolves.toEqual(success()); + const launch = fake.runs.at(-1); + const remoteLaunchCommand = launch?.argv.at(-1) ?? ""; + expect(remoteLaunchCommand).toContain("'sh' '-c'"); + expect(remoteLaunchCommand).toContain( + 'exec node "$HOME/.openclaw-worker/$1/openclaw.mjs" worker', + ); + expect(remoteLaunchCommand).toContain(`'${BUNDLE_HASH}'`); + expect(launch?.options.input).toBe(JSON.stringify(descriptor)); + expect(launch?.options.timeoutMs).toBe(123); await handle.stop(); expect(tunnel?.process.stopCount).toBe(1); expect(manager.status("worker:one")).toBe("stopped"); diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts index 55ded6840b43..c02a3ef7881f 100644 --- a/src/gateway/worker-environments/tunnel.ts +++ b/src/gateway/worker-environments/tunnel.ts @@ -78,6 +78,7 @@ directory=$2 rm -f -- "$socket" rmdir -- "$directory" 2>/dev/null || true `; +const WORKER_LAUNCH_SCRIPT = 'exec node "$HOME/.openclaw-worker/$1/openclaw.mjs" worker'; type WorkerTunnelStartRequest = WorkerTunnelRequest & { bundleHash: string; @@ -230,11 +231,8 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = ).catch(() => undefined); }; - const createHandle = (entry: TunnelEntry): WorkerTunnelHandle => ({ - environmentId: entry.environmentId, - ownerEpoch: entry.ownerEpoch, - remoteSocketPath: entry.remoteSocketPath, - ...createWorkerWorkspaceActions({ + const createHandle = (entry: TunnelEntry): WorkerTunnelHandle => { + const workspace = createWorkerWorkspaceActions({ environmentId: entry.environmentId, sharedHost: entry.sharedHost, ownerSignal: entry.abortController.signal, @@ -243,9 +241,23 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = runner, tasks: entry.workspaceTasks, bundleHash: entry.bundleHash, - }), - stop: () => stop(entry.environmentId, entry.ownerEpoch), - }); + }); + return { + environmentId: entry.environmentId, + ownerEpoch: entry.ownerEpoch, + connectionEndpoint: { kind: "unix", socketPath: entry.remoteSocketPath }, + launchTurn: (request) => + workspace.runWorkspaceCommand({ + transportRetry: "never", + argv: ["sh", "-c", WORKER_LAUNCH_SCRIPT, "openclaw-worker", entry.bundleHash], + input: JSON.stringify(request.descriptor), + timeoutMs: request.timeoutMs, + signal: request.signal, + }), + ...workspace, + stop: () => stop(entry.environmentId, entry.ownerEpoch), + }; + }; const connect = async ( entry: TunnelEntry, diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index c2da8d17bebc..bf45913cf622 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -743,7 +743,7 @@ describe("worker turn launcher", () => { const tunnel: WorkerTunnelHandle = { environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => { @@ -754,14 +754,15 @@ describe("worker turn launcher", () => { expect(placements.listPendingWorkspaceResults()).toHaveLength(1); }), })), - runWorkspaceCommand: vi.fn(async (command): Promise => { + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async (request): Promise => { expect(placements.get(SESSION_ID)?.turnClaim).toMatchObject({ owner: "worker", runId: "run-worker-turn", ownerEpoch: OWNER_EPOCH, }); - descriptor = parseWorkerLaunchDescriptor(JSON.parse(command.input ?? "")); - expect(command.transportRetry).toBe("never"); + descriptor = parseWorkerLaunchDescriptor(structuredClone(request.descriptor)); + expect(request.timeoutMs).toBe(5_000); const activeRuntimeIdentity = await verifyAgentRuntimeIdentityToken( descriptor.assignment.agentRuntimeIdentityToken, ); @@ -770,14 +771,10 @@ describe("worker turn launcher", () => { activeRuntimeIdentity && createAgentRuntimeApprovalAuthorityValidator(placements)(activeRuntimeIdentity), ).toBe(true); - expect(command.argv).toEqual([ - "sh", - "-c", - 'exec node "$HOME/.openclaw-worker/$1/openclaw.mjs" worker', - "openclaw-worker", - BUNDLE_HASH, - ]); - expect(command.argv.join(" ")).not.toContain(credential().credential); + expect(descriptor.connectionEndpoint).toEqual({ + kind: "unix", + socketPath: "/worker/gateway.sock", + }); await Promise.resolve(); expect(acknowledgeCredentialDelivery).toHaveBeenCalledOnce(); const completed = openSessionManager(); @@ -885,7 +882,7 @@ describe("worker turn launcher", () => { expect(descriptor?.assignment.prompt).toBe("Inspect this workspace"); expect(descriptor?.assignment.suppressPromptTranscript).toBe(true); expect(descriptor?.assignment.agentId).toBe(sessionTarget.agentId); - expect(descriptor?.version).toBe(2); + expect(descriptor?.version).toBe(3); const verifiedRuntimeIdentity = await verifyAgentRuntimeIdentityToken( descriptor?.assignment.agentRuntimeIdentityToken, ); @@ -1007,13 +1004,14 @@ describe("worker turn launcher", () => { const tunnel: WorkerTunnelHandle = { environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand: vi.fn(async (command): Promise => { - descriptor = parseWorkerLaunchDescriptor(JSON.parse(command.input ?? "")); + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async (request): Promise => { + descriptor = parseWorkerLaunchDescriptor(structuredClone(request.descriptor)); const completed = openSessionManager(); const leafId = completed.appendMessage( makeAgentAssistantMessage({ @@ -1142,12 +1140,13 @@ describe("worker turn launcher", () => { const tunnel: WorkerTunnelHandle = { environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand: vi.fn(async (): Promise => { + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async (): Promise => { const completed = openSessionManager(); const leafId = completed.appendMessage( makeAgentAssistantMessage({ @@ -1227,12 +1226,13 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand: vi.fn(async (): Promise => { + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async (): Promise => { const completed = openSessionManager(); completed.appendMessage( makeAgentAssistantMessage({ @@ -1426,7 +1426,7 @@ describe("worker turn launcher", () => { isError: false, timestamp: 2, }); - const runWorkspaceCommand = vi.fn(async (): Promise => { + const launchTurn = vi.fn(async (): Promise => { throw new Error("unexpected worker handoff"); }); const acknowledgeCredentialDelivery = vi.fn(() => true); @@ -1434,9 +1434,10 @@ describe("worker turn launcher", () => { async (): Promise => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(), - runWorkspaceCommand, + runWorkspaceCommand: vi.fn(), + launchTurn, syncWorkspace: vi.fn(), reconcileWorkspace: vi.fn(), stop: vi.fn(async () => {}), @@ -1469,7 +1470,7 @@ describe("worker turn launcher", () => { ).rejects.toThrow(WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE); expect(startTunnel).toHaveBeenCalledOnce(); - expect(runWorkspaceCommand).not.toHaveBeenCalled(); + expect(launchTurn).not.toHaveBeenCalled(); expect(runLocal).not.toHaveBeenCalled(); expect(acknowledgeCredentialDelivery).not.toHaveBeenCalled(); expect(stopTunnel).not.toHaveBeenCalled(); @@ -1480,7 +1481,7 @@ describe("worker turn launcher", () => { it("preserves a terminal workspace result when the worker child later exits nonzero", async () => { seedActivePlacement(); const destroy = vi.fn(async () => attachedEnvironment()); - const runWorkspaceCommand = vi.fn(async (): Promise => { + const launchTurn = vi.fn(async (): Promise => { createWorkerSessionPlacementGate(placements).updateAckCursors({ sessionId: SESSION_ID, environmentId: ENVIRONMENT_ID, @@ -1504,9 +1505,10 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(), - runWorkspaceCommand, + runWorkspaceCommand: vi.fn(), + launchTurn, syncWorkspace: vi.fn(), reconcileWorkspace: vi.fn(), stop: vi.fn(async () => {}), @@ -1529,7 +1531,7 @@ describe("worker turn launcher", () => { ), ).rejects.toThrow("child cleanup failed"); - expect(runWorkspaceCommand).toHaveBeenCalledOnce(); + expect(launchTurn).toHaveBeenCalledOnce(); expect(destroy).not.toHaveBeenCalled(); expect(placements.listPendingWorkspaceResults()).toMatchObject([ { @@ -1656,12 +1658,13 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand: vi.fn(async () => { + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async () => { throw new Error("remote launch failed"); }), syncWorkspace: vi.fn(async () => { @@ -1732,8 +1735,9 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", - runWorkspaceCommand: vi.fn( + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn( async (): Promise => ({ stdout: "", stderr, @@ -1944,7 +1948,7 @@ describe("worker turn launcher", () => { killed: false; termination: "exit"; }>(); - const runWorkspaceCommand = vi.fn(() => { + const launchTurn = vi.fn(() => { commandStarted.resolve(); return commandFinished.promise; }); @@ -1955,12 +1959,13 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand, + runWorkspaceCommand: vi.fn(), + launchTurn, syncWorkspace: vi.fn(async () => { throw new Error("unexpected workspace sync"); }), @@ -1995,7 +2000,7 @@ describe("worker turn launcher", () => { meta: { durationMs: 1 }, })), ).rejects.toThrow("already has an active turn claim"); - expect(runWorkspaceCommand).toHaveBeenCalledOnce(); + expect(launchTurn).toHaveBeenCalledOnce(); const completed = openSessionManager(); const leafId = completed.appendMessage( @@ -2063,14 +2068,15 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand: vi.fn(async (command): Promise => { + runWorkspaceCommand: vi.fn(), + launchTurn: vi.fn(async (request): Promise => { launchCount += 1; - const descriptor = parseWorkerLaunchDescriptor(JSON.parse(command.input ?? "")); + const descriptor = parseWorkerLaunchDescriptor(structuredClone(request.descriptor)); turnIds.push(descriptor.assignment.turnId); if (launchCount === 1) { const completed = openSessionManager(); @@ -2219,7 +2225,7 @@ describe("worker turn launcher", () => { } return active; }; - const runWorkspaceCommand = vi.fn(async (): Promise => { + const launchTurn = vi.fn(async (): Promise => { workerStarted.resolve(); await resumeWorker.promise; expect(placements.get(SESSION_ID)).toMatchObject({ @@ -2261,12 +2267,13 @@ describe("worker turn launcher", () => { startTunnel: vi.fn(async () => ({ environmentId: ENVIRONMENT_ID, ownerEpoch: OWNER_EPOCH, - remoteSocketPath: "/worker/gateway.sock", + connectionEndpoint: { kind: "unix" as const, socketPath: "/worker/gateway.sock" }, quiesceWorkspace: vi.fn(async () => ({ assertActive: vi.fn(async () => {}), resume: vi.fn(async () => {}), })), - runWorkspaceCommand, + runWorkspaceCommand: vi.fn(), + launchTurn, syncWorkspace: vi.fn(async () => { throw new Error("unexpected workspace sync"); }), @@ -2349,7 +2356,7 @@ describe("worker turn launcher", () => { ); expect(result.payloads).toEqual([{ text: "Redispatched worker reply" }]); expect(redispatchCalls).toBe(1); - expect(runWorkspaceCommand).toHaveBeenCalledOnce(); + expect(launchTurn).toHaveBeenCalledOnce(); expect(runLocal).not.toHaveBeenCalled(); expect(placements.get(SESSION_ID)).toMatchObject({ state: "active", turnClaim: null }); }); diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index f02fe8a06696..68780760ce30 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -57,8 +57,6 @@ import { } from "./workspace-result-finalize.js"; import { workerWorkspaceResultRef } from "./workspace-result-staging.js"; -const WORKER_LAUNCH_SCRIPT = 'exec node "$HOME/.openclaw-worker/$1/openclaw.mjs" worker'; - type WorkerTurnEnvironmentService = Pick< WorkerEnvironmentService, | "acknowledgeCredentialDelivery" @@ -326,8 +324,8 @@ async function executeWorkerTurn(params: { messages: initialMessages, build: (agentRuntimeIdentityToken, windowedMessages) => parseWorkerLaunchDescriptor({ - version: 2, - socketPath: tunnel.remoteSocketPath, + version: 3, + connectionEndpoint: tunnel.connectionEndpoint, admission: { environmentId: placement.environmentId, credential: credential.credential, @@ -375,10 +373,8 @@ async function executeWorkerTurn(params: { turn.onExecutionPhase?.({ phase: "attempt_dispatch", backend: "cloud-worker" }); const handoffAbort = new AbortController(); params.onHandoff(); - const processPromise = tunnel.runWorkspaceCommand({ - transportRetry: "never", - argv: ["sh", "-c", WORKER_LAUNCH_SCRIPT, "openclaw-worker", placement.workerBundleHash], - input: JSON.stringify(descriptor), + const processPromise = tunnel.launchTurn({ + descriptor, timeoutMs: turn.timeoutMs, signal: turn.abortSignal ? AbortSignal.any([turn.abortSignal, handoffAbort.signal]) diff --git a/src/gateway/worker-environments/worker-turn-payload.test.ts b/src/gateway/worker-environments/worker-turn-payload.test.ts index 56e7ab9480ae..ceb86ca7919f 100644 --- a/src/gateway/worker-environments/worker-turn-payload.test.ts +++ b/src/gateway/worker-environments/worker-turn-payload.test.ts @@ -82,8 +82,8 @@ function buildDescriptor( operationalRunInstance: OperationalRunInstanceRef, ): WorkerLaunchDescriptor { return { - version: 2, - socketPath: "/tmp/worker.sock", + version: 3, + connectionEndpoint: { kind: "unix", socketPath: "/tmp/worker.sock" }, admission: { environmentId: "environment", credential: "worker-fixture-credential", diff --git a/src/worker/launch-descriptor.test.ts b/src/worker/launch-descriptor.test.ts index 6a134d19af9a..33f409b5a69b 100644 --- a/src/worker/launch-descriptor.test.ts +++ b/src/worker/launch-descriptor.test.ts @@ -11,8 +11,8 @@ import { buildWorkerConnectParams, parseWorkerLaunchDescriptor } from "./launch- function launchDescriptor(): WorkerLaunchDescriptor { return { - version: 2, - socketPath: "/tmp/openclaw-worker/gateway.sock", + version: 3, + connectionEndpoint: { kind: "unix", socketPath: "/tmp/openclaw-worker/gateway.sock" }, admission: { environmentId: "environment-1", credential: ["worker", "fixture", "value"].join("-"), @@ -62,6 +62,41 @@ describe("worker launch descriptor", () => { }); }); + it("accepts only closed Unix or public WebSocket connection endpoints", () => { + const descriptor = launchDescriptor(); + descriptor.connectionEndpoint = { + kind: "websocket", + url: "wss://gateway.example/tenant/__openclaw__/worker", + tlsFingerprint: "ab:".repeat(31) + "ab", + }; + expect(parseWorkerLaunchDescriptor(structuredClone(descriptor))).toEqual(descriptor); + + const invalidEndpoints: unknown[] = [ + { kind: "unix", socketPath: "gateway.sock" }, + { kind: "unix", socketPath: "/tmp/gateway:sock" }, + { kind: "websocket", url: "https://gateway.example/__openclaw__/worker" }, + { kind: "websocket", url: "ws://user@gateway.example/__openclaw__/worker" }, + { kind: "websocket", url: "wss://gateway.example/other" }, + { kind: "websocket", url: "wss://gateway.example/__openclaw__/worker?token=x" }, + { + kind: "websocket", + url: "ws://127.0.0.1/__openclaw__/worker", + tlsFingerprint: "ab".repeat(32), + }, + { + kind: "websocket", + url: "wss://gateway.example/__openclaw__/worker", + tlsFingerprint: "", + }, + { ...descriptor.connectionEndpoint, unexpected: true }, + ]; + for (const connectionEndpoint of invalidEndpoints) { + expect(() => parseWorkerLaunchDescriptor({ ...descriptor, connectionEndpoint })).toThrow( + "invalid worker launch descriptor", + ); + } + }); + it("rejects unknown fields at every launch-owned boundary", () => { const descriptor = launchDescriptor(); const cases: unknown[] = [ @@ -129,7 +164,7 @@ describe("worker launch descriptor", () => { const descriptor = launchDescriptor(); const { toolAuthority: _missing, ...assignmentWithoutAuthority } = descriptor.assignment; const cases: unknown[] = [ - { ...descriptor, version: 1 }, + { ...descriptor, version: 2 }, { ...descriptor, assignment: assignmentWithoutAuthority }, { ...descriptor, @@ -220,7 +255,10 @@ describe("worker launch descriptor", () => { it("rejects non-absolute paths, unattached sessions, and discontinuous event sequences", () => { const descriptor = launchDescriptor(); const cases: unknown[] = [ - { ...descriptor, socketPath: "gateway.sock" }, + { + ...descriptor, + connectionEndpoint: { kind: "unix", socketPath: "gateway.sock" }, + }, { ...descriptor, assignment: { ...descriptor.assignment, workspaceDir: "workspace" }, diff --git a/src/worker/launch-descriptor.ts b/src/worker/launch-descriptor.ts index 18da97c4b003..5df40886baaa 100644 --- a/src/worker/launch-descriptor.ts +++ b/src/worker/launch-descriptor.ts @@ -27,8 +27,12 @@ import { PROTOCOL_VERSION } from "../../packages/gateway-protocol/src/version.js import type { OperationalRunInstanceRef } from "../agents/admitted-run-context.js"; import { isWorkerToolName, type WorkerToolAuthority } from "./tool-authority.js"; import { isWorkerTranscriptMessageFrameSafe } from "./transcript-message.js"; +import { + parseWorkerConnectionEndpoint, + type WorkerConnectionEndpoint, +} from "./worker-connection-endpoint.js"; -const LAUNCH_VERSION = 2; +const LAUNCH_VERSION = 3; export type WorkerBrowserLaunchDescriptor = { cdpUrl: string; @@ -67,8 +71,8 @@ type WorkerLaunchAdmission = Omit & { }; export type WorkerLaunchDescriptor = { - version: 2; - socketPath: string; + version: 3; + connectionEndpoint: WorkerConnectionEndpoint; admission: WorkerLaunchAdmission; assignment: WorkerLaunchAssignment; }; @@ -261,20 +265,19 @@ export function buildWorkerConnectParams( export function parseWorkerLaunchDescriptor(value: unknown): WorkerLaunchDescriptor { if ( !isRecord(value) || - !hasExactKeys(value, ["version", "socketPath", "admission", "assignment"]) || - value.version !== LAUNCH_VERSION || - !isIdentifier(value.socketPath) || - !path.isAbsolute(value.socketPath) + !hasExactKeys(value, ["version", "connectionEndpoint", "admission", "assignment"]) || + value.version !== LAUNCH_VERSION ) { throw new Error("invalid worker launch descriptor"); } + const connectionEndpoint = parseWorkerConnectionEndpoint(value.connectionEndpoint); const assignment = parseAssignment(value.assignment); - if (!assignment || !isRecord(value.admission)) { + if (!connectionEndpoint || !assignment || !isRecord(value.admission)) { throw new Error("invalid worker launch descriptor"); } const candidate: WorkerLaunchDescriptor = { version: LAUNCH_VERSION, - socketPath: value.socketPath, + connectionEndpoint, admission: value.admission as WorkerLaunchAdmission, assignment, }; diff --git a/src/worker/worker-connection-admission.ts b/src/worker/worker-connection-admission.ts index 82fa9f89f72e..7f6701c89507 100644 --- a/src/worker/worker-connection-admission.ts +++ b/src/worker/worker-connection-admission.ts @@ -19,6 +19,10 @@ import { toWorkerConnectionError, type WorkerConnectionOptions, } from "./worker-connection-contract.js"; +import { + resolveWorkerConnectionTarget, + WorkerConnectionEndpointError, +} from "./worker-connection-endpoint.js"; import { closeInvalidWorkerFrame } from "./worker-connection-frames.js"; const RETRYABLE_CLOSE_REASONS = new Set([ @@ -68,25 +72,18 @@ export function isRetryableWorkerCloseReason(reason: WorkerProtocolCloseReason): return RETRYABLE_CLOSE_REASONS.has(reason); } -function workerSocketUrl(socketPath: string): string { - if (!socketPath.startsWith("/")) { - throw new Error("worker gateway socket path must be absolute"); - } - if (socketPath.includes(":")) { - throw new Error("worker gateway socket path must not contain a colon"); - } - return `ws+unix://${socketPath}:/`; -} - export function connectWorkerConnectionAttempt( options: WorkerConnectionAttemptOptions, ): Promise { const connectionOptions = options.connectionOptions; + const target = resolveWorkerConnectionTarget(connectionOptions.endpoint); + const socketOptions = { + ...target.options, + maxPayload: WORKER_PROTOCOL_MAX_INFERENCE_PAYLOAD_BYTES, + }; const socket = connectionOptions.createSocket - ? connectionOptions.createSocket(workerSocketUrl(connectionOptions.socketPath)) - : new WebSocket(workerSocketUrl(connectionOptions.socketPath), { - maxPayload: WORKER_PROTOCOL_MAX_INFERENCE_PAYLOAD_BYTES, - }); + ? connectionOptions.createSocket(target.url, socketOptions) + : new WebSocket(target.url, socketOptions); options.onSocket(socket); const admissionId = randomUUID(); let admitted = false; @@ -121,6 +118,12 @@ export function connectWorkerConnectionAttempt( socket.close(); return; } + const tlsError = target.validateSocket(socket); + if (tlsError) { + rejectAttempt(new WorkerConnectionEndpointError(tlsError.message)); + socket.close(1008, tlsError.message); + return; + } options.onAdmitting(); const frame: WorkerConnectRequestFrame = { type: "req", diff --git a/src/worker/worker-connection-contract.ts b/src/worker/worker-connection-contract.ts index 22f4e44eddbd..fdb126509b76 100644 --- a/src/worker/worker-connection-contract.ts +++ b/src/worker/worker-connection-contract.ts @@ -1,5 +1,5 @@ import { toStructuredErrorObject } from "@openclaw/normalization-core/error-coercion"; -import type { WebSocket } from "ws"; +import type { ClientOptions, WebSocket } from "ws"; import type { WorkerConnectParams, WorkerHeartbeatParams, @@ -7,6 +7,7 @@ import type { WorkerProtocolCloseReason, } from "../../packages/gateway-protocol/src/schema/worker-admission.js"; import type { BackoffPolicy } from "../infra/backoff.js"; +import type { WorkerConnectionEndpoint } from "./worker-connection-endpoint.js"; const FENCED_CLOSE_REASONS = new Set([ "credential-replaced", @@ -37,13 +38,13 @@ export type WorkerConnectionExit = | { kind: "stopped" }; export type WorkerConnectionOptions = { - socketPath: string; + endpoint: WorkerConnectionEndpoint; connectParams: WorkerConnectParams; reconnectBackoff?: BackoffPolicy; admissionTimeoutMs?: number; admissionDeadlineMs?: number; requestTimeoutMs?: number; - createSocket?: (url: string) => WebSocket; + createSocket?: (url: string, options: ClientOptions) => WebSocket; heartbeatStatus?: () => WorkerHeartbeatParams["status"]; }; diff --git a/src/worker/worker-connection-endpoint.test.ts b/src/worker/worker-connection-endpoint.test.ts new file mode 100644 index 000000000000..ec9ceed55221 --- /dev/null +++ b/src/worker/worker-connection-endpoint.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from "vitest"; +import type { CertMeta, WebSocket } from "ws"; +import { + parseWorkerConnectionEndpoint, + resolveWorkerConnectionTarget, +} from "./worker-connection-endpoint.js"; + +describe("worker connection endpoint", () => { + it("resolves Unix sockets through the existing ws+unix carrier", () => { + const endpoint = parseWorkerConnectionEndpoint({ + kind: "unix", + socketPath: "/tmp/openclaw-worker/gateway.sock", + }); + expect(endpoint).toBeDefined(); + + expect(resolveWorkerConnectionTarget(endpoint!)).toMatchObject({ + url: "ws+unix:///tmp/openclaw-worker/gateway.sock:/", + options: {}, + }); + }); + + it("applies the canonical TLS pin policy to public worker URLs", () => { + const fingerprint = "ab".repeat(32); + const endpoint = parseWorkerConnectionEndpoint({ + kind: "websocket", + url: "wss://gateway.example/tenant/__openclaw__/worker", + tlsFingerprint: fingerprint, + }); + expect(endpoint).toBeDefined(); + + const target = resolveWorkerConnectionTarget(endpoint!); + const checkServerIdentity = (hostname: string, cert: CertMeta) => + target.options.checkServerIdentity?.(hostname, cert); + expect(target.options.rejectUnauthorized).toBe(false); + expect( + checkServerIdentity("gateway.example", { + fingerprint256: fingerprint, + } as unknown as CertMeta), + ).toBeUndefined(); + expect( + checkServerIdentity("gateway.example", { + fingerprint256: "cd".repeat(32), + } as unknown as CertMeta), + ).toEqual(new Error("Server TLS fingerprint mismatch")); + + const socket = { + _socket: { getPeerCertificate: () => ({ fingerprint256: fingerprint }) }, + } as unknown as WebSocket; + expect(target.validateSocket(socket)).toBeNull(); + }); + + it("rejects public plaintext while retaining the private-network break-glass", () => { + const endpoint = { + kind: "websocket" as const, + url: "ws://gateway.example/__openclaw__/worker", + }; + expect(() => resolveWorkerConnectionTarget(endpoint, {})).toThrow("SECURITY ERROR"); + expect(() => + resolveWorkerConnectionTarget(endpoint, { OPENCLAW_ALLOW_INSECURE_PRIVATE_WS: "1" }), + ).not.toThrow(); + }); +}); diff --git a/src/worker/worker-connection-endpoint.ts b/src/worker/worker-connection-endpoint.ts new file mode 100644 index 000000000000..262da78c9647 --- /dev/null +++ b/src/worker/worker-connection-endpoint.ts @@ -0,0 +1,122 @@ +import path from "node:path"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import type { ClientOptions, WebSocket } from "ws"; +import { + GatewayWebSocketTransportConfigurationError, + resolveGatewayWebSocketTransport, +} from "../../packages/gateway-client/src/websocket-transport.js"; +import { WORKER_PUBLIC_INGRESS_PATH } from "../../packages/gateway-protocol/src/schema/worker-admission.js"; +import { WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH } from "../../packages/gateway-protocol/src/schema/worker-protocol-primitives.js"; + +export class WorkerConnectionEndpointError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkerConnectionEndpointError"; + } +} + +export type WorkerConnectionEndpoint = + | { kind: "unix"; socketPath: string } + | { kind: "websocket"; url: string; tlsFingerprint?: string }; + +function hasExactKeys(value: Record, required: string[], optional: string[] = []) { + const allowed = new Set([...required, ...optional]); + return ( + required.every((key) => key in value) && Object.keys(value).every((key) => allowed.has(key)) + ); +} + +function parseUnixEndpoint(value: Record): WorkerConnectionEndpoint | undefined { + if ( + !hasExactKeys(value, ["kind", "socketPath"]) || + value.kind !== "unix" || + typeof value.socketPath !== "string" || + value.socketPath.length > WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH || + !path.isAbsolute(value.socketPath) || + value.socketPath.includes(":") + ) { + return undefined; + } + return { kind: "unix", socketPath: value.socketPath }; +} + +function parseWebSocketEndpoint( + value: Record, +): WorkerConnectionEndpoint | undefined { + if ( + !hasExactKeys(value, ["kind", "url"], ["tlsFingerprint"]) || + value.kind !== "websocket" || + typeof value.url !== "string" || + value.url.length > 4_096 || + (value.tlsFingerprint !== undefined && + (typeof value.tlsFingerprint !== "string" || + value.tlsFingerprint.trim().length === 0 || + value.tlsFingerprint.length > 256)) + ) { + return undefined; + } + let url: URL; + try { + url = new URL(value.url); + } catch { + return undefined; + } + if ( + (url.protocol !== "ws:" && url.protocol !== "wss:") || + url.username !== "" || + url.password !== "" || + url.search !== "" || + url.hash !== "" || + !url.pathname.endsWith(WORKER_PUBLIC_INGRESS_PATH) || + (value.tlsFingerprint !== undefined && url.protocol !== "wss:") + ) { + return undefined; + } + return { + kind: "websocket", + url: value.url, + ...(value.tlsFingerprint === undefined ? {} : { tlsFingerprint: value.tlsFingerprint }), + }; +} + +export function parseWorkerConnectionEndpoint( + value: unknown, +): WorkerConnectionEndpoint | undefined { + if (!isRecord(value)) { + return undefined; + } + return parseUnixEndpoint(value) ?? parseWebSocketEndpoint(value); +} + +type WorkerConnectionTarget = { + url: string; + options: ClientOptions; + validateSocket(socket: WebSocket): Error | null; +}; + +export function resolveWorkerConnectionTarget( + endpoint: WorkerConnectionEndpoint, + env: NodeJS.ProcessEnv = process.env, +): WorkerConnectionTarget { + if (endpoint.kind === "unix") { + return { + url: `ws+unix://${endpoint.socketPath}:/`, + options: {}, + validateSocket: () => null, + }; + } + try { + const transport = resolveGatewayWebSocketTransport({ + url: endpoint.url, + tlsFingerprint: endpoint.tlsFingerprint, + env, + options: {}, + }); + return { url: endpoint.url, ...transport }; + } catch (error) { + if (error instanceof GatewayWebSocketTransportConfigurationError) { + throw new WorkerConnectionEndpointError(error.message); + } + throw error; + } +} diff --git a/src/worker/worker-connection.test.ts b/src/worker/worker-connection.test.ts index 8a7d6155667c..a787557cd925 100644 --- a/src/worker/worker-connection.test.ts +++ b/src/worker/worker-connection.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { WebSocket } from "ws"; import { GATEWAY_CLIENT_IDS, @@ -14,6 +14,7 @@ import type { WorkerInferenceTerminalFrame, } from "../../packages/gateway-protocol/src/schema/worker-inference.js"; import { toWorkerConnectionError } from "./worker-connection-contract.js"; +import { WorkerConnectionEndpointError } from "./worker-connection-endpoint.js"; import { WorkerConnectionFrameDispatcher } from "./worker-connection-frames.js"; import { createWorkerConnection, type WorkerConnectionState } from "./worker-connection.js"; @@ -44,7 +45,7 @@ const FRAME_CONNECT_PARAMS: WorkerConnectParams = { function createIdleConnection() { return createWorkerConnection({ - socketPath: "ws://127.0.0.1:1", + endpoint: { kind: "unix", socketPath: "/tmp/worker-listener-isolation.sock" }, connectParams: { minProtocol: 1, maxProtocol: 1, @@ -131,6 +132,26 @@ function installThrowingThenHealthyListeners(connection: ReturnType throwingCalls }; } +describe("worker connection endpoint failures", () => { + it("fails insecure public endpoints without entering reconnect backoff", async () => { + const createSocket = vi.fn(); + const connection = createWorkerConnection({ + endpoint: { + kind: "websocket", + url: "ws://gateway.example/__openclaw__/worker", + }, + connectParams: FRAME_CONNECT_PARAMS, + createSocket, + admissionDeadlineMs: 60_000, + reconnectBackoff: { initialMs: 30_000, maxMs: 30_000, factor: 1, jitter: 0 }, + }); + + await expect(connection.start()).rejects.toBeInstanceOf(WorkerConnectionEndpointError); + expect(connection.state).toMatchObject({ kind: "failed" }); + expect(createSocket).not.toHaveBeenCalled(); + }); +}); + describe("worker connection error coercion", () => { it("preserves structured non-Error causes", () => { const cause = { code: "ECONNRESET", status: 503 }; diff --git a/src/worker/worker-connection.ts b/src/worker/worker-connection.ts index d4245b329688..c04d509facda 100644 --- a/src/worker/worker-connection.ts +++ b/src/worker/worker-connection.ts @@ -42,6 +42,7 @@ import { type WorkerConnectionState, type WorkerFencedReason, } from "./worker-connection-contract.js"; +import { WorkerConnectionEndpointError } from "./worker-connection-endpoint.js"; import { WorkerConnectionFrameDispatcher } from "./worker-connection-frames.js"; export { @@ -287,6 +288,10 @@ export class WorkerConnection { this.handleAdmissionFailure(error); throw error; } + if (error instanceof WorkerConnectionEndpointError) { + this.finishFailed(error); + throw error; + } if (this.isTerminal()) { throw this.terminalError(); } diff --git a/src/worker/worker-fault-injection.test-support.ts b/src/worker/worker-fault-injection.test-support.ts index 2f683ba04041..f0b001d3ff63 100644 --- a/src/worker/worker-fault-injection.test-support.ts +++ b/src/worker/worker-fault-injection.test-support.ts @@ -316,8 +316,8 @@ export class ComposedGatewayHarness { const epoch = params.epoch ?? this.epoch; const credential = params.admissionProof ?? CREDENTIAL; return { - version: 2, - socketPath: this.socketPath, + version: 3, + connectionEndpoint: { kind: "unix", socketPath: this.socketPath }, admission: { environmentId: ENVIRONMENT_ID, credential, @@ -354,7 +354,7 @@ export class ComposedGatewayHarness { const descriptor = this.createDescriptor(params); const epoch = descriptor.admission.ownerEpoch; const connection = createWorkerConnection({ - socketPath: this.socketPath, + endpoint: { kind: "unix", socketPath: this.socketPath }, connectParams: buildWorkerConnectParams(descriptor), admissionTimeoutMs: 1_000, admissionDeadlineMs: 5_000, diff --git a/src/worker/worker.runtime.test.ts b/src/worker/worker.runtime.test.ts index e1fb96ac3911..249d736427a1 100644 --- a/src/worker/worker.runtime.test.ts +++ b/src/worker/worker.runtime.test.ts @@ -808,8 +808,8 @@ class FakeWorkerGateway { function descriptor(socketPath: string, workspaceDir: string): WorkerLaunchDescriptor { return { - version: 2, - socketPath, + version: 3, + connectionEndpoint: { kind: "unix", socketPath }, admission: { environmentId: "worker-environment", credential: CREDENTIAL, @@ -1026,7 +1026,7 @@ describe("worker runtime", () => { silenceSessionSpawnResponses: 2, }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), requestTimeoutMs: 25, reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, @@ -1451,7 +1451,7 @@ describe("worker reconnect clients", () => { it("isolates ready listener failures while admitting the worker and starting heartbeats", async () => { const { gateway, launch } = await setup({ heartbeatIntervalMs: 1 }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), }); let healthyReadyCalls = 0; @@ -1474,7 +1474,7 @@ describe("worker reconnect clients", () => { it("fails closed when the overall admission deadline expires", async () => { const { gateway, launch } = await setup({ admissionFailure: "gateway-unavailable" }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), admissionTimeoutMs: 25, admissionDeadlineMs: 250, @@ -1499,7 +1499,7 @@ describe("worker reconnect clients", () => { it("times out a silent admission attempt and admits on reconnect", async () => { const { gateway, launch } = await setup({ ignoreFirstAdmission: true }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), admissionTimeoutMs: 25, reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, @@ -1518,7 +1518,7 @@ describe("worker reconnect clients", () => { heartbeatIntervalMs: 1, }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), requestTimeoutMs: 25, reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, @@ -1538,7 +1538,7 @@ describe("worker reconnect clients", () => { silenceFirstInference: true, }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), requestTimeoutMs: 40, reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, @@ -1590,7 +1590,7 @@ describe("worker reconnect clients", () => { it("settles an in-flight commit and a later live emit after stop", async () => { const { gateway, launch } = await setup({ silenceFirstTranscript: true }); const connection = createWorkerConnection({ - socketPath: gateway.socketPath, + endpoint: { kind: "unix", socketPath: gateway.socketPath }, connectParams: buildWorkerConnectParams(launch), requestTimeoutMs: 5_000, reconnectBackoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, diff --git a/src/worker/worker.runtime.ts b/src/worker/worker.runtime.ts index 7fd5cab08eb0..2d8d352ac728 100644 --- a/src/worker/worker.runtime.ts +++ b/src/worker/worker.runtime.ts @@ -63,7 +63,7 @@ export async function runWorkerDescriptor( let resultFenceAcked = false; let forcedStopTimer: NodeJS.Timeout | undefined; const connection = createWorkerConnection({ - socketPath: descriptor.socketPath, + endpoint: descriptor.connectionEndpoint, connectParams: buildWorkerConnectParams(descriptor), }); const abortFromCaller = () => {