From 0c824f09d545f82715da4d022015aa5645eedb00 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 12 Aug 2026 13:17:45 -0700 Subject: [PATCH] feat(gateway): observe paired node desktops (#122724) * feat(gateway): stream paired node desktops * chore(protocol): refresh desktop observe model * fix(gateway): preserve desktop stream boundaries * fix(gateway): keep desktop streams lifetime-bound * fix(gateway): harden node desktop stream lifecycle * fix(gateway): stabilize node desktop lifecycle setup * chore(plugin-sdk): refresh API baselines --- .../OpenClawProtocol/GatewayModels.swift | 6 +- .../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/gateway/configuration-reference.md | 66 +++- docs/nodes/index.md | 6 +- .../src/schema/desktop.test.ts | 19 +- .../gateway-protocol/src/schema/desktop.ts | 8 + src/gateway/desktop/attachment.test.ts | 104 ++++- src/gateway/desktop/attachment.ts | 8 + src/gateway/desktop/host-source-errors.ts | 23 +- .../desktop/node-observe.integration.test.ts | 266 +++++++++++++ src/gateway/desktop/node-source-context.ts | 11 + src/gateway/desktop/node-source.ts | 288 ++++++++++++++ .../desktop/node-stream-broker.test.ts | 273 +++++++++++++ src/gateway/desktop/node-stream-broker.ts | 301 +++++++++++++++ src/gateway/desktop/observe-bridge.test.ts | 1 + src/gateway/desktop/observe-bridge.ts | 23 +- src/gateway/desktop/session-registry.ts | 182 ++++++++- src/gateway/node-command-policy.test.ts | 54 +++ src/gateway/node-command-policy.ts | 4 +- src/gateway/server-http.ts | 20 + src/gateway/server-kernel-request-runtime.ts | 2 + src/gateway/server-lifecycle.ts | 20 + .../environments.desktop.test.ts | 30 +- .../server-methods/environments.test.ts | 47 +++ src/gateway/server-methods/environments.ts | 81 +++- .../server-node-session-runtime.test.ts | 19 + src/gateway/server-node-session-runtime.ts | 2 + src/gateway/server-request-context.ts | 5 + src/gateway/server-runtime-state-prepare.ts | 25 +- src/gateway/server-runtime-state.ts | 4 + .../worker-environments/desktop-tunnel.ts | 4 +- src/node-host/desktop-stream-command.test.ts | 111 ++++++ src/node-host/desktop-stream-command.ts | 361 ++++++++++++++++++ src/node-host/gateway-candidate-connection.ts | 4 +- .../invoke-agent-cli-claude-handler.ts | 5 + src/node-host/invoke.ts | 23 ++ src/node-host/runner.test.ts | 1 + src/node-host/runner.ts | 4 +- src/node-host/runtime.test.ts | 41 +- src/node-host/runtime.ts | 23 +- src/shared/node-desktop-stream.ts | 2 + ui/src/components/desktop/desktop-panel.ts | 30 +- ui/src/e2e/desktop-panel.e2e.test.ts | 67 ++++ 59 files changed, 2526 insertions(+), 82 deletions(-) create mode 100644 src/gateway/desktop/node-observe.integration.test.ts create mode 100644 src/gateway/desktop/node-source-context.ts create mode 100644 src/gateway/desktop/node-source.ts create mode 100644 src/gateway/desktop/node-stream-broker.test.ts create mode 100644 src/gateway/desktop/node-stream-broker.ts create mode 100644 src/node-host/desktop-stream-command.test.ts create mode 100644 src/node-host/desktop-stream-command.ts create mode 100644 src/shared/node-desktop-stream.ts diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index a6f19816c7df..036abff5b50f 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -2304,6 +2304,7 @@ public struct DesktopObserveResult: Codable, Sendable { public let control: Bool public let vncpassword: String? public let auth: String? + public let preauthenticated: Bool? public init( transport: String, @@ -2311,7 +2312,8 @@ public struct DesktopObserveResult: Codable, Sendable { expiresatms: Int, control: Bool, vncpassword: String? = nil, - auth: String? = nil) + auth: String? = nil, + preauthenticated: Bool? = nil) { self.transport = transport self.wspath = wspath @@ -2319,6 +2321,7 @@ public struct DesktopObserveResult: Codable, Sendable { self.control = control self.vncpassword = vncpassword self.auth = auth + self.preauthenticated = preauthenticated } private enum CodingKeys: String, CodingKey { @@ -2328,6 +2331,7 @@ public struct DesktopObserveResult: Codable, Sendable { case control case vncpassword = "vncPassword" case auth + case preauthenticated } } 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 dcd2ac08e887..54f7192104cb 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":"0afe8b5e500dce4237ca4b5023ab225cfd31a9897e0d4bfdb2341886df035952","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} +{"contentHash":"4cbdbf71af85d9f128c7508ff8fc731e26d6534a005cf198ee65a29900c77340","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 1ca9cba851fc..b1526d0dae97 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json @@ -1 +1 @@ -{"contentHash":"4580a38b9e871c7d448c90cd247673929caaa7a298877d96d8e3f7e7c970b27a","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} +{"contentHash":"23d191b0194005162c3004d7094de2c2ab3130519cc1dd8e93b9fd3b7de1887a","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 f0ede275a3c8..5b27720c7966 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-core.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-core.json @@ -1 +1 @@ -{"contentHash":"b84b53ca4792a410eb488ae4fb303e362c3c84d8eb72e37cf9c922183792a7cd","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} +{"contentHash":"1a06845da584661ae0964b92fb78a6e30b5b49eb76e8607d7d27e22b9cea06e5","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 b7fb7fb60212..20bf4473f5de 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":"cd582aff7d7278ddc71a3c99bd4f8b46e2083411f02096e8649e74b39ab001ea","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} +{"contentHash":"3a12bd6e0a49b35124afb94210e13e08d009c3879f28e5d397f7c3155155ef8d","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 31ba5e415d4b..6c80082d21eb 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-message.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-message.json @@ -1 +1 @@ -{"contentHash":"abb22832153a23cd5187f92f2cdee6d81889d61697fae04c8647eb3370a1aa93","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} +{"contentHash":"986d193f0ee65375bba05e796533e619ef5d5861d3cb4b76d3233ff94f60ae26","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 aa67a89ff3e7..28cec6e0111c 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json @@ -1 +1 @@ -{"contentHash":"a82dcfdcc661b2567ba2ba631d589887d5ca29cfe1d8600696fe0db859f4969f","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} +{"contentHash":"f34d5a25428beca66e891f24c358befc2ffa900ae5efe66770039e95b6d554f8","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 a6e29608d23f..a56078e9807e 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":"034d00199482a5490cb8202aa2ff6c5acd438486e8ab616b5f76674af7fb78c4","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} +{"contentHash":"c16c7681de92566c0c886aae20c229422185d65a6a1be627e97fe4152b9a0cd9","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 310ebd0e5de0..aa248db3fbf4 100644 --- a/docs/.generated/plugin-sdk-api-baseline/core.json +++ b/docs/.generated/plugin-sdk-api-baseline/core.json @@ -1 +1 @@ -{"contentHash":"2efff507ffe87196b554e29043eadc0ba17a32864f3ccd1864d4200add9d8f87","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} +{"contentHash":"c8683be7c78ece135932acd07c4b5c8e840177d1fe16f9da298202287dd4b305","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 e0a1e4af68ef..7173408ca6d8 100644 --- a/docs/.generated/plugin-sdk-api-baseline/discord.json +++ b/docs/.generated/plugin-sdk-api-baseline/discord.json @@ -1 +1 @@ -{"contentHash":"31bfbc294becf37f8ee037a0f8f50c69e4d04e097ff9bea16089115ee93b80df","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} +{"contentHash":"cf626c5629096ad218de506f7e35d22806efcbd4eacccc3d84df13aa726f7f33","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 d231bce30e8e..3577dc570ffe 100644 --- a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json @@ -1 +1 @@ -{"contentHash":"bb371c8a354a150acf264357e3cbcb755efc6f42d9a2143e5a9254c3bc3c2aa8","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} +{"contentHash":"6b238d77a12894d47525c52b8e564656292178054b6f7dfcba4c5f4ba721c124","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 5788f56f29af..edb32ab690d9 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":"07dbe53008bb2487dd63f28db4d81424cfb67498e4508aae6d86667e78fe1855","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} +{"contentHash":"449257831591673a7f5bb4a2664cf5931cdc7ba9baa380f4b25513fb11a66c13","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 c9e854aa8c44..b6b7558fca13 100644 --- a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json @@ -1 +1 @@ -{"contentHash":"5ccd913160d86827e36459bf6c6e88a169306dd24e913b85e288cbbf2ea12157","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} +{"contentHash":"63740457d088b93b3f42f5f5991ca566c03c7024a471bfcee9f4bae3dd1f609e","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 8006d43c8b2c..5feb08347e3f 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json @@ -1 +1 @@ -{"contentHash":"d9ae989750b9f69072b5767cdf8995bde3f431da315e031cb80f35d2052a3917","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} +{"contentHash":"73b7833128355b975c6084eeb195b2d2278982fb42e00253ad1d26249cb6bd51","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 a2c1e7fd652d..dfb33d6f5d21 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json @@ -1 +1 @@ -{"contentHash":"dce833138906cea4ce741068911c2e4818ebc37693edb2d2883e54eb7b25778e","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} +{"contentHash":"0b4f0f3996add880e30ae03f78f1ee8220d67eece0b85c89204e9e384f242d8f","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 d3a4f4a7148f..38b450d99fea 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":"0010661448f741237afcb47b75bdc2178d962e4e5751dfc4d26fdb1c4c487997","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} +{"contentHash":"1cd3f740a75bcc58e73253165e13bef74f7e4472d59d2cb0c5f3a14a5944a4e6","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 a189078481cd..eb9f2e3c4bba 100644 --- a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json +++ b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json @@ -1 +1 @@ -{"contentHash":"a392d8bbb2ababa55a16ac115beff0ed0b790723c8e6c69c0a46d44c6984ed4e","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} +{"contentHash":"9f785c65c9d357a29d1449fe6ac3c4bbafe9cd638298485f209ba3547bf118ee","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 1fc735ceeae6..ca3eb7d4c1a2 100644 --- a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json +++ b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json @@ -1 +1 @@ -{"contentHash":"a9f8b45ec3684fdb607c50fc47c41f62255c9ac0eae282b42441df58400fce31","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} +{"contentHash":"d99313fa91889a8eb69c5331f28dac834d133c7dadd2b6fafae69975afdeb56e","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md index c2f3670707e2..687bcff1241c 100644 --- a/docs/gateway/configuration-reference.md +++ b/docs/gateway/configuration-reference.md @@ -619,6 +619,70 @@ or logs. `openclaw doctor` can offer an explicitly confirmed `sudo launchctl` repair when Screen Sharing is off; enabling the macOS system service may expose it on other network interfaces according to macOS Sharing settings. +### Paired node desktops + +A paired macOS, Windows, or Linux node can expose its own desktop in the same +Control UI Desktop panel. This path is intentionally off by default and always +uses an existing node-local RFB server on `127.0.0.1`; the Gateway never asks a +node to connect to a caller-selected host or port. + +On the node machine, enable the desktop source and configure attach mode: + +```json5 +{ + desktop: { + host: { + enabled: true, + port: 5900, + // passwordFile: "/path/to/vnc-password.txt", + }, + }, +} +``` + +Restart the node host after changing this config. `managed: true` is a Gateway +host feature and does not start a managed desktop inside a node host; paired +nodes must already have a loopback RFB server. + +On the Gateway, explicitly arm the dangerous command and restart: + +```json5 +{ + gateway: { + nodes: { + commands: { + allow: ["desktop.stream"], + // deny: ["desktop.stream"], // deny always wins + }, + }, + }, +} +``` + +The node reconnect advertises `desktop.stream` as a pairing-surface upgrade. +Inspect `openclaw nodes pending`, then approve the new request with +`openclaw nodes approve `. The node appears in the Desktop picker +only while it is connected and the effective approved command remains allowed. + +For VncAuth, `desktop.host.passwordFile` stays on the node and is delivered only +to the Gateway's authenticated relay. Without a password file, the Control UI +prompts for the VNC password. macOS ARD credentials are always prompted per +observation. The Gateway completes ARD or VNC authentication before exposing a +no-auth RFB handshake to the browser, so credentials are not returned in URLs, +logs, or RPC results. + +Desktop bytes use a dedicated outbound binary WebSocket from the node. The +normal node invoke remains only as the cancellable lifecycle handle and never +carries framebuffer data. Reconnecting or changing the node's pairing +generation closes active relays. To disarm the feature, remove +`desktop.stream` from `commands.allow` or add it to `commands.deny`, restart the +Gateway, and reconnect the node. + +If the node is missing from the picker, verify all four gates: the node-local +desktop config, the loopback RFB listener, the approved pairing update, and the +Gateway allow/deny policy. After changing any of them, restart the affected +Gateway or node host and check `openclaw nodes pending` again. + --- ## Gateway @@ -781,7 +845,7 @@ it on other network interfaces according to macOS Sharing settings. - `gateway.nodes.pairing.autoApproveLocal`: silently approves pairing, role upgrades, and scope upgrades from trusted local connections (default: `true`). Set `false` to require explicit approval for every device; metadata-only reconnect refreshes remain automatic. - `gateway.nodes.pairing.autoApproveCidrs`: optional CIDR/IP allowlist for auto-approving first-time node device pairing with no requested scopes. It is disabled when unset. This does not auto-approve operator/browser/Control UI/WebChat pairing, and it does not auto-approve role, scope, metadata, or public-key upgrades. - `gateway.nodes.pairing.sshVerify`: SSH-verified auto-approval for first-time node device pairing (default: enabled). The gateway SSHes back to the pairing host (BatchMode, strict host keys) and approves only on an exact `openclaw node identity` device-key match. Same eligibility floor as `autoApproveCidrs`; probes are limited to private/CGNAT source addresses unless `cidrs` overrides them. Set `false` to disable, or `{ user, identity, timeoutMs, cidrs }` to tune. See [Node pairing](/gateway/pairing#ssh-verified-device-auto-approval-default). -- `gateway.nodes.commands.allow` / `gateway.nodes.commands.deny`: global allow/deny shaping for declared node commands after pairing and platform allowlist evaluation. `commands.allow` is the one-time persistent enable for classified commands such as `camera.snap`, `camera.clip`, `screen.record`, `health.summary`, `sms.search`, and `sms.send`; `commands.deny` removes a command even if a platform default or explicit allow would otherwise include it. Computer and mobile UI control instead rely on default-off node-local enablement plus pairing. iOS Health permission, Android SMS permission, and Gateway command authorization are independent. After a node changes its declared command list, reject and re-approve that device pairing so the gateway stores the updated command snapshot. +- `gateway.nodes.commands.allow` / `gateway.nodes.commands.deny`: global allow/deny shaping for declared node commands after pairing and platform allowlist evaluation. `commands.allow` is the one-time persistent enable for classified commands such as `camera.snap`, `camera.clip`, `desktop.stream`, `screen.record`, `health.summary`, `sms.search`, and `sms.send`; `commands.deny` removes a command even if a platform default or explicit allow would otherwise include it. Computer and mobile UI control instead rely on default-off node-local enablement plus pairing. iOS Health permission, Android SMS permission, and Gateway command authorization are independent. After a node changes its declared command list, reject and re-approve that device pairing so the gateway stores the updated command snapshot. - `gateway.tools.deny`: extra tool names blocked for HTTP `POST /tools/invoke` (extends default deny list). - `gateway.tools.allow`: remove tool names from the default HTTP deny list for owner/admin callers. This does not upgrade identity-bearing `operator.write` diff --git a/docs/nodes/index.md b/docs/nodes/index.md index cfc879195d0b..6802062b9b4c 100644 --- a/docs/nodes/index.md +++ b/docs/nodes/index.md @@ -484,7 +484,7 @@ These rows describe the Gateway policy ceiling, not the commands implemented by Desktop host commands (`system.run`, `system.run.prepare`, `system.which`, `browser.proxy`, `browser.proxy.upload.v1`, `mcp.tools.call.v1`, and `screen.snapshot` on macOS/Windows/Linux) are not part of the static platform-default table above. They become available once the operator approves a pairing request that declares them, after which the node's approved command set carries them forward on reconnect. -Dangerous or privacy-heavy commands require a one-time persistent opt-in with `gateway.nodes.commands.allow`, even if a node declares them: `camera.snap`, `camera.clip`, `camera.ptz.control`, `screen.record`, `contacts.add`, `calendar.add`, `reminders.add`, `health.summary`, `sms.send`, `sms.search`. `gateway.nodes.commands.deny` always wins over defaults and extra allowlist entries. See [HealthKit summaries](/platforms/ios-healthkit) for the iPhone consent gate and [Computer use](/nodes/computer-use) for the local enablement, pairing, capability, and tool-policy gates around desktop input. +Dangerous or privacy-heavy commands require a one-time persistent opt-in with `gateway.nodes.commands.allow`, even if a node declares them: `camera.snap`, `camera.clip`, `camera.ptz.control`, `desktop.stream`, `screen.record`, `contacts.add`, `calendar.add`, `reminders.add`, `health.summary`, `sms.send`, `sms.search`. `gateway.nodes.commands.deny` always wins over defaults and extra allowlist entries. See [Paired node desktops](/gateway/configuration-reference#paired-node-desktops), [HealthKit summaries](/platforms/ios-healthkit), and [Computer use](/nodes/computer-use) for the local enablement, pairing, capability, and tool-policy gates around desktop access. Plugin-owned node commands can add a Gateway node-invoke policy. That policy runs after the allowlist check and before forwarding to the node, so raw `node.invoke`, CLI helpers, and dedicated agent tools share the same plugin permission boundary. Dangerous plugin node commands still require explicit `gateway.nodes.commands.allow` opt-in. @@ -511,9 +511,9 @@ Node-related settings live under `gateway.nodes` and `tools.exec`: pluginTools: { enabled: true, }, - // Persistently enable dangerous/privacy-heavy node commands (camera.snap, etc.). + // Persistently enable dangerous/privacy-heavy node commands. commands: { - allow: ["camera.snap", "screen.record"], + allow: ["camera.snap", "desktop.stream", "screen.record"], // Block exact command names even if defaults or commands.allow include them. deny: ["camera.clip"], }, diff --git a/packages/gateway-protocol/src/schema/desktop.test.ts b/packages/gateway-protocol/src/schema/desktop.test.ts index cc4d7de55c84..827c127056f3 100644 --- a/packages/gateway-protocol/src/schema/desktop.test.ts +++ b/packages/gateway-protocol/src/schema/desktop.test.ts @@ -7,7 +7,7 @@ import { } from "../index.js"; describe("desktop protocol schemas", () => { - it("accepts host and environment observe sources while rejecting unknown source kinds", () => { + it("accepts host, environment, and node observe sources", () => { expect(validateDesktopObserveParams({ source: { kind: "host" }, control: true })).toBe(true); expect( validateDesktopObserveParams({ @@ -20,7 +20,21 @@ describe("desktop protocol schemas", () => { source: { kind: "environment", environmentId: "worker:one" }, }), ).toBe(true); - expect(validateDesktopObserveParams({ source: { kind: "node", nodeId: "one" } })).toBe(false); + expect(validateDesktopObserveParams({ source: { kind: "node", nodeId: "one" } })).toBe(true); + expect( + validateDesktopObserveParams({ + source: { kind: "node", nodeId: "one" }, + credentials: { username: "operator", password: "secret" }, + }), + ).toBe(true); + expect( + validateDesktopObserveParams({ + source: { kind: "node", nodeId: "one" }, + credentials: { password: "secret" }, + }), + ).toBe(true); + expect(validateDesktopObserveParams({ source: { kind: "node", nodeId: "" } })).toBe(false); + expect(validateDesktopObserveParams({ source: { kind: "future" } })).toBe(false); expect( validateDesktopObserveParams({ source: { kind: "environment", environmentId: "worker:one" }, @@ -55,6 +69,7 @@ describe("desktop protocol schemas", () => { expiresAtMs: 1, control: false, auth: "ard-account", + preauthenticated: true, }), ).toBe(true); expect( diff --git a/packages/gateway-protocol/src/schema/desktop.ts b/packages/gateway-protocol/src/schema/desktop.ts index 593b85715d4c..e4882fe2629f 100644 --- a/packages/gateway-protocol/src/schema/desktop.ts +++ b/packages/gateway-protocol/src/schema/desktop.ts @@ -8,6 +8,7 @@ import { NonEmptyString } from "./primitives.js"; export const DesktopSourceSchema = Type.Union([ closedObject({ kind: Type.Literal("host") }), closedObject({ kind: Type.Literal("environment"), environmentId: NonEmptyString }), + closedObject({ kind: Type.Literal("node"), nodeId: NonEmptyString }), ]); const DesktopObserveCredentialsSchema = closedObject({ @@ -26,6 +27,11 @@ export const DesktopObserveParamsSchema = Type.Union([ source: closedObject({ kind: Type.Literal("environment"), environmentId: NonEmptyString }), control: Type.Optional(Type.Boolean()), }), + closedObject({ + source: closedObject({ kind: Type.Literal("node"), nodeId: NonEmptyString }), + control: Type.Optional(Type.Boolean()), + credentials: Type.Optional(DesktopObserveCredentialsSchema), + }), ]); export const DesktopObserveResultSchema = closedObject({ @@ -36,6 +42,8 @@ export const DesktopObserveResultSchema = closedObject({ vncPassword: Type.Optional(NonEmptyString), // Auth drives credential prompting without coupling clients to RFB security numbers. auth: Type.Optional(Type.String({ enum: ["none", "vnc-password", "ard-account"] })), + // Gateway-side pre-auth keeps credentials out of the browser RFB client. + preauthenticated: Type.Optional(Type.Boolean()), }); export const DesktopLaunchParamsSchema = closedObject({ diff --git a/src/gateway/desktop/attachment.test.ts b/src/gateway/desktop/attachment.test.ts index 3bf7e639b8ee..d91b32f59c11 100644 --- a/src/gateway/desktop/attachment.test.ts +++ b/src/gateway/desktop/attachment.test.ts @@ -1,11 +1,14 @@ import net from "node:net"; -import { afterEach, describe, expect, it } from "vitest"; +import { PassThrough } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { connectRfbAttachment } from "./attachment.js"; +import { createDesktopSessionRegistry } from "./session-registry.js"; const servers: net.Server[] = []; const sockets: net.Socket[] = []; afterEach(async () => { + vi.useRealTimers(); for (const socket of sockets.splice(0)) { socket.destroy(); } @@ -45,4 +48,103 @@ describe("RFB attachments", () => { await expect(accepted).resolves.toBeUndefined(); }); + + it("does not claim a stream that closed before observer redemption", async () => { + const registry = createDesktopSessionRegistry(); + await registry.acquire({ + sourceKey: "node:one", + ownerEpoch: 1, + start: async () => ({ + attachment: { kind: "tcp", host: "127.0.0.1", port: 5900 }, + }), + }); + const stream = new PassThrough(); + const reservation = registry.reserveObserver("node:one", 1); + if (!reservation) { + throw new Error("expected observer reservation"); + } + const attachment = registry.publishStream({ + sourceKey: "node:one", + ownerEpoch: 1, + stream, + reservation, + }); + if (!attachment) { + throw new Error("expected stream attachment"); + } + const closed = new Promise((resolve) => { + stream.once("close", () => resolve()); + }); + stream.destroy(); + await closed; + + expect(registry.claimStream(attachment)).toBeUndefined(); + await registry.stopAll(); + }); + + it("refreshes the cleanup deadline when an idle stream session is reactivated", async () => { + vi.useFakeTimers(); + const teardown = vi.fn(async () => undefined); + const registry = createDesktopSessionRegistry({ lingerMs: 25 }); + await registry.activate({ sourceKey: "node:one", ownerEpoch: 1, teardown }); + await vi.advanceTimersByTimeAsync(20); + await registry.activate({ sourceKey: "node:one", ownerEpoch: 1 }); + await vi.advanceTimersByTimeAsync(20); + expect(teardown).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(5); + expect(teardown).toHaveBeenCalled(); + }); + + it("bounds pending observer reservations before streams are started", async () => { + const registry = createDesktopSessionRegistry(); + await registry.activate({ sourceKey: "node:one", ownerEpoch: 1 }); + const reservations = Array.from({ length: 8 }, () => registry.reserveObserver("node:one", 1)); + expect(reservations.every(Boolean)).toBe(true); + expect(registry.reserveObserver("node:one", 1)).toBeUndefined(); + + reservations[0]?.release(); + expect(registry.reserveObserver("node:one", 1)).toBeDefined(); + await registry.stopAll(); + }); + + it("keeps a reserved observer session alive and rearms cleanup on release", async () => { + vi.useFakeTimers(); + const teardown = vi.fn(async () => undefined); + const registry = createDesktopSessionRegistry({ lingerMs: 25 }); + await registry.activate({ sourceKey: "node:one", ownerEpoch: 1, teardown }); + const reservation = registry.reserveObserver("node:one", 1); + if (!reservation) { + throw new Error("expected observer reservation"); + } + await vi.advanceTimersByTimeAsync(100); + expect(teardown).not.toHaveBeenCalled(); + + reservation.release(); + await vi.advanceTimersByTimeAsync(25); + expect(teardown).toHaveBeenCalled(); + }); + + it("does not linger-stop a reservation when another observer disconnects", async () => { + vi.useFakeTimers(); + const teardown = vi.fn(async () => undefined); + const registry = createDesktopSessionRegistry({ lingerMs: 25 }); + await registry.activate({ sourceKey: "node:one", ownerEpoch: 1, teardown }); + const observer = registry.attachObserver("node:one", { + ownerEpoch: 1, + control: false, + close: () => {}, + }); + const reservation = registry.reserveObserver("node:one", 1); + if (!observer || !reservation) { + throw new Error("expected observer and reservation"); + } + observer.release(); + await vi.advanceTimersByTimeAsync(100); + expect(teardown).not.toHaveBeenCalled(); + + reservation.release(); + await vi.advanceTimersByTimeAsync(25); + expect(teardown).toHaveBeenCalled(); + }); }); diff --git a/src/gateway/desktop/attachment.ts b/src/gateway/desktop/attachment.ts index bde7cdcb7623..664e09c46bfe 100644 --- a/src/gateway/desktop/attachment.ts +++ b/src/gateway/desktop/attachment.ts @@ -1,11 +1,19 @@ import net from "node:net"; +import type { Duplex } from "node:stream"; export type RfbAttachment = | { kind: "unix-socket"; socketPath: string } | { kind: "tcp"; host: "127.0.0.1"; port: number }; +/** One already-connected RFB transport published by a remote desktop source. */ +type RfbStreamAttachment = { kind: "stream"; streamId: string }; + +export type DesktopRfbAttachment = RfbAttachment | RfbStreamAttachment; + export function connectRfbAttachment(attachment: RfbAttachment): net.Socket { return attachment.kind === "unix-socket" ? net.connect(attachment.socketPath) : net.connect(attachment.port, attachment.host); } + +export type ConnectedRfbStream = Duplex; diff --git a/src/gateway/desktop/host-source-errors.ts b/src/gateway/desktop/host-source-errors.ts index 72cb6830a345..9fb1db79387d 100644 --- a/src/gateway/desktop/host-source-errors.ts +++ b/src/gateway/desktop/host-source-errors.ts @@ -1,15 +1,26 @@ -export class HostDesktopCredentialsRequiredError extends Error { - readonly auth = "ard-account" as const; +export class DesktopCredentialsRequiredError extends Error { readonly detailCode = "DESKTOP_CREDENTIALS_REQUIRED" as const; + constructor( + readonly auth: "vnc-password" | "ard-account", + message: string, + ) { + super(message); + this.name = "DesktopCredentialsRequiredError"; + } +} + +export class HostDesktopCredentialsRequiredError extends DesktopCredentialsRequiredError { + declare readonly auth: "ard-account"; + constructor() { - super("macOS account credentials are required to observe Screen Sharing"); + super("ard-account", "macOS account credentials are required to observe Screen Sharing"); this.name = "HostDesktopCredentialsRequiredError"; } } -export function isHostDesktopCredentialsRequiredError( +export function isDesktopCredentialsRequiredError( error: unknown, -): error is HostDesktopCredentialsRequiredError { - return error instanceof HostDesktopCredentialsRequiredError; +): error is DesktopCredentialsRequiredError { + return error instanceof DesktopCredentialsRequiredError; } diff --git a/src/gateway/desktop/node-observe.integration.test.ts b/src/gateway/desktop/node-observe.integration.test.ts new file mode 100644 index 000000000000..8f6093368989 --- /dev/null +++ b/src/gateway/desktop/node-observe.integration.test.ts @@ -0,0 +1,266 @@ +import http from "node:http"; +import net from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { WebSocket, type RawData } from "ws"; +import { invokeNodeDesktopStream } from "../../node-host/desktop-stream-command.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../../shared/node-desktop-stream.js"; +import type { NodeRegistry } from "../node-registry.js"; +import { createNodeDesktopService } from "./node-source.js"; +import { createNodeDesktopStreamBroker } from "./node-stream-broker.js"; +import { handleDesktopObserveUpgrade } from "./observe-bridge.js"; +import { createDesktopSessionRegistry } from "./session-registry.js"; + +const VERSION = Buffer.from("RFB 003.008\n", "ascii"); +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); +}); + +class SocketReader { + private buffered = Buffer.alloc(0); + private readonly waiters = new Set<() => void>(); + + constructor(socket: net.Socket) { + socket.on("data", (chunk) => { + this.buffered = Buffer.concat([ + this.buffered, + Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk), + ]); + for (const waiter of this.waiters) { + waiter(); + } + this.waiters.clear(); + }); + } + + async readExactly(length: number): Promise { + while (this.buffered.length < length) { + await new Promise((resolve) => { + this.waiters.add(resolve); + }); + } + const value = this.buffered.subarray(0, length); + this.buffered = this.buffered.subarray(length); + return value; + } +} + +class WebSocketReader { + private readonly chunks: Buffer[] = []; + private readonly waiters: Array<(chunk: Buffer) => void> = []; + + constructor(ws: WebSocket) { + ws.on("message", (data: RawData) => { + const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer); + const waiter = this.waiters.shift(); + if (waiter) { + waiter(chunk); + } else { + this.chunks.push(chunk); + } + }); + } + + async next(): Promise { + return ( + this.chunks.shift() ?? + (await new Promise((resolve) => { + this.waiters.push(resolve); + })) + ); + } +} + +describe("paired node desktop observe integration", () => { + it("relays pixels through the node attach socket and drops view-only input", async () => { + const rfbPeers = new Set(); + let connectionCount = 0; + let completedStreams = 0; + let resolveRfbScript!: () => void; + let rejectRfbScript!: (error: Error) => void; + const rfbScript = new Promise((resolve, reject) => { + resolveRfbScript = resolve; + rejectRfbScript = reject; + }); + const rfbServer = net.createServer((socket) => { + rfbPeers.add(socket); + socket.once("close", () => rfbPeers.delete(socket)); + connectionCount += 1; + const connectionIndex = connectionCount; + const reader = new SocketReader(socket); + void (async () => { + try { + socket.write(VERSION); + expect(await reader.readExactly(VERSION.length)).toEqual(VERSION); + socket.write(Buffer.from([1, 2])); + if (connectionIndex % 2 === 1) { + return; + } + expect(await reader.readExactly(1)).toEqual(Buffer.from([2])); + socket.write(Buffer.alloc(16, 7)); + expect(await reader.readExactly(16)).toHaveLength(16); + socket.write(Buffer.alloc(4)); + + expect(await reader.readExactly(1)).toEqual(Buffer.from([1])); + socket.write(Buffer.from("pixel-update", "ascii")); + const framebufferRequest = Buffer.from([3, 1, 0, 0, 0, 0, 0, 64, 0, 64]); + expect(await reader.readExactly(framebufferRequest.length)).toEqual(framebufferRequest); + completedStreams += 1; + if (completedStreams === 2) { + resolveRfbScript(); + } + } catch (error) { + rejectRfbScript(error instanceof Error ? error : new Error(String(error))); + } + })(); + }); + await new Promise((resolve) => { + rfbServer.listen(0, "127.0.0.1", resolve); + }); + const rfbAddress = rfbServer.address(); + if (!rfbAddress || typeof rfbAddress === "string") { + throw new Error("expected RFB address"); + } + cleanups.push( + async () => + await new Promise((resolve) => { + for (const peer of rfbPeers) { + peer.destroy(); + } + rfbServer.close(() => resolve()); + }), + ); + + const desktopRegistry = createDesktopSessionRegistry({ lingerMs: 10 }); + const streamBroker = createNodeDesktopStreamBroker(); + cleanups.push(async () => desktopRegistry.stopAll()); + const httpServer = http.createServer(); + let gatewayUrl = ""; + const nodeSession = { + nodeId: "node-1", + connId: "conn-1", + pairingGeneration: "generation-1", + platform: "linux", + deviceFamily: "Linux", + commands: [NODE_DESKTOP_STREAM_COMMAND], + }; + const nodeRegistry = { + get: () => nodeSession, + getForPairingGeneration: (_nodeId: string, generation: string) => + generation === nodeSession.pairingGeneration ? nodeSession : undefined, + isConnectionCurrentPairingState: async (connId: string) => connId === nodeSession.connId, + invoke: async (request: { + params?: unknown; + signal?: AbortSignal; + onProgress?: (chunk: string) => void; + }) => { + try { + await invokeNodeDesktopStream({ + paramsJSON: JSON.stringify({ + ...(request.params as { ticket: string; attachPath: string }), + }), + gatewayUrl, + config: { enabled: true, port: rfbAddress.port }, + signal: request.signal ?? new AbortController().signal, + emitStatus: async (status) => request.onProgress?.(status), + }); + return { ok: true }; + } catch (error) { + return { + ok: false, + error: { message: error instanceof Error ? error.message : String(error) }, + }; + } + }, + } as unknown as NodeRegistry; + httpServer.on("upgrade", (req, socket, head) => { + void (async () => { + if (await streamBroker.handleUpgrade(req, socket, head, nodeRegistry)) { + return; + } + handleDesktopObserveUpgrade(req, socket, head, { registry: desktopRegistry }); + })(); + }); + await new Promise((resolve) => { + httpServer.listen(0, "127.0.0.1", resolve); + }); + const httpAddress = httpServer.address(); + if (!httpAddress || typeof httpAddress === "string") { + throw new Error("expected Gateway address"); + } + gatewayUrl = `ws://127.0.0.1:${httpAddress.port}`; + cleanups.push( + async () => + await new Promise((resolve) => { + httpServer.close(() => resolve()); + }), + ); + + const service = createNodeDesktopService({ + getConfig: () => ({ + gateway: { nodes: { commands: { allow: [NODE_DESKTOP_STREAM_COMMAND] } } }, + }), + nodeRegistry, + desktopRegistry, + streamBroker, + }); + const observed = await service.observe({ + nodeId: nodeSession.nodeId, + control: false, + credentials: { password: "memory-only-password" }, + }); + expect(observed.auth).toBe("vnc-password"); + + const ws = new WebSocket(`${gatewayUrl}${observed.wsPath}`); + const browser = new WebSocketReader(ws); + cleanups.push(async () => ws.terminate()); + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); + expect(await browser.next()).toEqual(VERSION); + ws.send(Buffer.concat([VERSION, Buffer.from([1, 0])])); + expect(await browser.next()).toEqual(Buffer.from([1, 1])); + expect(await browser.next()).toEqual(Buffer.alloc(4)); + expect(await browser.next()).toEqual(Buffer.from("pixel-update", "ascii")); + + const keyEvent = Buffer.from([4, 1, 0, 0, 0, 0, 0, 65]); + const framebufferRequest = Buffer.from([3, 1, 0, 0, 0, 0, 0, 64, 0, 64]); + ws.send(Buffer.concat([keyEvent, framebufferRequest])); + + const secondObserved = await service.observe({ + nodeId: nodeSession.nodeId, + control: false, + credentials: { password: "memory-only-password" }, + }); + const secondWs = new WebSocket(`${gatewayUrl}${secondObserved.wsPath}`); + const secondBrowser = new WebSocketReader(secondWs); + cleanups.push(async () => secondWs.terminate()); + await new Promise((resolve, reject) => { + secondWs.once("open", resolve); + secondWs.once("error", reject); + }); + expect(await secondBrowser.next()).toEqual(VERSION); + secondWs.send(Buffer.concat([VERSION, Buffer.from([1, 0])])); + expect(await secondBrowser.next()).toEqual(Buffer.from([1, 1])); + expect(await secondBrowser.next()).toEqual(Buffer.alloc(4)); + expect(await secondBrowser.next()).toEqual(Buffer.from("pixel-update", "ascii")); + expect(ws.readyState).toBe(WebSocket.OPEN); + secondWs.send(Buffer.concat([keyEvent, framebufferRequest])); + + await expect(rfbScript).resolves.toBeUndefined(); + await vi.waitFor(() => expect(connectionCount).toBe(4)); + const firstClosed = new Promise((resolve) => { + ws.once("close", () => resolve()); + }); + const secondClosed = new Promise((resolve) => { + secondWs.once("close", () => resolve()); + }); + + await service.stopNode(nodeSession.nodeId); + + await Promise.all([firstClosed, secondClosed]); + await vi.waitFor(() => expect(rfbPeers.size).toBe(0)); + }); +}); diff --git a/src/gateway/desktop/node-source-context.ts b/src/gateway/desktop/node-source-context.ts new file mode 100644 index 000000000000..a3a47e99cb82 --- /dev/null +++ b/src/gateway/desktop/node-source-context.ts @@ -0,0 +1,11 @@ +import type { NodeDesktopService } from "./node-source.js"; + +export const NODE_DESKTOP_SERVICE_CONTEXT = Symbol("openclaw.nodeDesktopService"); + +type NodeDesktopServiceContext = { + [NODE_DESKTOP_SERVICE_CONTEXT]?: NodeDesktopService; +}; + +export function getNodeDesktopService(context: object): NodeDesktopService | undefined { + return (context as NodeDesktopServiceContext)[NODE_DESKTOP_SERVICE_CONTEXT]; +} diff --git a/src/gateway/desktop/node-source.ts b/src/gateway/desktop/node-source.ts new file mode 100644 index 000000000000..084de700f51c --- /dev/null +++ b/src/gateway/desktop/node-source.ts @@ -0,0 +1,288 @@ +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { registerSecretValueForRedaction } from "../../logging/secret-redaction-registry.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../../shared/node-desktop-stream.js"; +import { isNodeCommandAllowed, resolveNodeCommandAllowlist } from "../node-command-policy.js"; +import type { NodeRegistry } from "../node-registry.js"; +import { DesktopCredentialsRequiredError } from "./host-source-errors.js"; +import type { NodeDesktopStreamBroker } from "./node-stream-broker.js"; +import { mintDesktopObserverToken } from "./observe-bridge.js"; +import type { DesktopSessionRegistry } from "./session-registry.js"; + +type NodeDesktopObserveResult = { + transport: "rfb"; + wsPath: string; + expiresAtMs: number; + control: boolean; + auth: "vnc-password" | "ard-account"; + preauthenticated: true; +}; + +function invocationError(result: Awaited>): Error { + const message = result.error?.message?.trim(); + return new Error(message || "node desktop stream closed before attachment"); +} + +type ActiveNodeDesktopStream = { + controller: AbortController; + ticket?: ReturnType; + stream?: import("node:stream").Duplex; + invocation?: ReturnType; + reservation?: ReturnType; + reservationTransferred: boolean; + unclaimedTimer?: ReturnType; + stopped: boolean; +}; + +type NodeDesktopSession = { + connId: string; + pairingGeneration: string; + ownerEpoch: number; + active: Set; +}; + +async function stopActiveStream(active: ActiveNodeDesktopStream): Promise { + if (active.stopped) { + return; + } + retireActiveStream(active); + await active.invocation?.catch(() => undefined); +} + +function retireActiveStream(active: ActiveNodeDesktopStream): void { + if (active.stopped) { + return; + } + active.stopped = true; + clearTimeout(active.unclaimedTimer); + active.ticket?.cancel(); + active.controller.abort(); + if (!active.reservationTransferred) { + active.reservation?.release(); + } + active.stream?.destroy(); +} + +/** Combines node command policy, ticket redemption, and desktop session ownership. */ +export function createNodeDesktopService(params: { + getConfig: () => OpenClawConfig; + nodeRegistry: NodeRegistry; + desktopRegistry: DesktopSessionRegistry; + streamBroker: NodeDesktopStreamBroker; +}) { + const ownerEpochs = new Map(); + const sessions = new Map(); + + const ensureSession = async (request: { + sourceKey: string; + connId: string; + pairingGeneration: string; + }): Promise => { + const current = sessions.get(request.sourceKey); + if ( + current?.connId === request.connId && + current.pairingGeneration === request.pairingGeneration + ) { + await params.desktopRegistry.activate({ + sourceKey: request.sourceKey, + ownerEpoch: current.ownerEpoch, + }); + return current; + } + + const ownerEpoch = (ownerEpochs.get(request.sourceKey) ?? 0) + 1; + ownerEpochs.set(request.sourceKey, ownerEpoch); + const session: NodeDesktopSession = { + connId: request.connId, + pairingGeneration: request.pairingGeneration, + ownerEpoch, + active: new Set(), + }; + sessions.set(request.sourceKey, session); + try { + await params.desktopRegistry.activate({ + sourceKey: request.sourceKey, + ownerEpoch, + teardown: async () => { + if (sessions.get(request.sourceKey) === session) { + sessions.delete(request.sourceKey); + } + await Promise.all([...session.active].map(stopActiveStream)); + session.active.clear(); + }, + }); + return session; + } catch (error) { + if (sessions.get(request.sourceKey) === session) { + sessions.delete(request.sourceKey); + } + throw error; + } + }; + + return { + async stopNode(nodeId: string): Promise { + const sourceKey = `node:${nodeId}`; + const session = sessions.get(sourceKey); + if (session) { + await params.desktopRegistry.stop(sourceKey, session.ownerEpoch); + } + }, + async observe(request: { + nodeId: string; + control: boolean; + credentials?: { username?: string; password?: string }; + }): Promise { + const node = params.nodeRegistry.get(request.nodeId); + if (!node?.pairingGeneration) { + throw new Error("node desktop is unavailable; reconnect and approve the node capability"); + } + const pairingGeneration = node.pairingGeneration; + // NodeSession.commands is the generation-bound effective approval surface; + // declaredCommands remains the broader capability advertised at connect. + const allowlist = resolveNodeCommandAllowlist(params.getConfig(), node); + const commandPolicy = isNodeCommandAllowed({ + command: NODE_DESKTOP_STREAM_COMMAND, + declaredCommands: node.commands, + allowlist, + }); + if (!commandPolicy.ok) { + throw new Error( + "node desktop is not enabled; explicitly allow and approve desktop.stream for this node", + ); + } + + const sourceKey = `node:${request.nodeId}`; + const session = await ensureSession({ + sourceKey, + connId: node.connId, + pairingGeneration, + }); + const active: ActiveNodeDesktopStream = { + controller: new AbortController(), + reservation: params.desktopRegistry.reserveObserver(sourceKey, session.ownerEpoch), + reservationTransferred: false, + stopped: false, + }; + if (!active.reservation) { + throw new Error("node desktop observer limit reached"); + } + session.active.add(active); + active.ticket = params.streamBroker.mint({ + nodeId: request.nodeId, + connId: node.connId, + pairingGeneration, + }); + active.invocation = params.nodeRegistry.invoke({ + nodeId: request.nodeId, + expectedConnId: node.connId, + expectedPairingGeneration: pairingGeneration, + command: NODE_DESKTOP_STREAM_COMMAND, + params: { ticket: active.ticket.ticket, attachPath: active.ticket.attachPath }, + timeoutMs: 0, + onProgress: () => {}, + signal: active.controller.signal, + }); + const invocationFinished = active.invocation.then((result) => { + throw invocationError(result); + }); + void invocationFinished.catch(() => undefined); + + let attached: Awaited; + try { + attached = await Promise.race([active.ticket.attached, invocationFinished]); + } catch (error) { + await stopActiveStream(active); + session.active.delete(active); + throw error; + } + active.stream = attached.stream; + + let password: string | undefined; + try { + if (attached.auth === "vnc-password") { + password = attached.vncPassword ?? request.credentials?.password; + if (!password) { + throw new DesktopCredentialsRequiredError( + "vnc-password", + "VNC password is required to observe this node", + ); + } + registerSecretValueForRedaction(password); + } else { + const username = request.credentials?.username?.trim() ?? ""; + const ardPassword = request.credentials?.password ?? ""; + if (!username || !ardPassword) { + throw new DesktopCredentialsRequiredError( + "ard-account", + "macOS account credentials are required to observe this node", + ); + } + registerSecretValueForRedaction(ardPassword); + } + } catch (error) { + await stopActiveStream(active); + session.active.delete(active); + throw error; + } + + const attachment = params.desktopRegistry.publishStream({ + sourceKey, + ownerEpoch: session.ownerEpoch, + stream: attached.stream, + reservation: active.reservation, + }); + if (!attachment) { + await stopActiveStream(active); + session.active.delete(active); + throw new Error("node desktop session was superseded before publication"); + } + active.reservationTransferred = true; + const credentials = request.credentials; + const preauth = + attached.auth === "ard-account" + ? { + auth: attached.auth, + credentials: { + username: credentials?.username?.trim() ?? "", + password: credentials?.password ?? "", + }, + } + : { + auth: attached.auth, + credentials: { password: password ?? credentials?.password ?? "" }, + }; + const minted = mintDesktopObserverToken({ + sourceKey, + ownerEpoch: session.ownerEpoch, + control: request.control, + attachment, + preauth, + }); + active.unclaimedTimer = setTimeout( + () => { + if (params.desktopRegistry.hasPendingStream(attachment)) { + void stopActiveStream(active).then(() => session.active.delete(active)); + } + }, + Math.max(0, minted.expiresAtMs - Date.now()), + ); + active.unclaimedTimer.unref?.(); + void active.invocation + .finally(() => { + retireActiveStream(active); + session.active.delete(active); + }) + .catch(() => undefined); + return { + transport: "rfb", + wsPath: `/desktop/observe?token=${minted.token}`, + expiresAtMs: minted.expiresAtMs, + control: request.control, + auth: attached.auth, + preauthenticated: true, + }; + }, + }; +} + +export type NodeDesktopService = ReturnType; diff --git a/src/gateway/desktop/node-stream-broker.test.ts b/src/gateway/desktop/node-stream-broker.test.ts new file mode 100644 index 000000000000..5df71eb38ecb --- /dev/null +++ b/src/gateway/desktop/node-stream-broker.test.ts @@ -0,0 +1,273 @@ +import http from "node:http"; +import net from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { WebSocket } from "ws"; +import { createNodeDesktopStreamBroker } from "./node-stream-broker.js"; + +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); +}); + +async function startBrokerServer(params: { + broker: ReturnType; + session: { connId: string; pairingGeneration: string }; + pairingCurrent?: () => boolean | Promise; +}) { + const registry = { + getForPairingGeneration: (_nodeId: string, pairingGeneration: string) => + pairingGeneration === params.session.pairingGeneration + ? { connId: params.session.connId } + : undefined, + isConnectionCurrentPairingState: async (connId: string) => + connId === params.session.connId && (await (params.pairingCurrent?.() ?? true)), + }; + const server = http.createServer(); + server.on("upgrade", (req, socket, head) => { + void params.broker.handleUpgrade(req, socket, head, registry as never); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("expected broker test address"); + } + cleanups.push( + async () => + await new Promise((resolve) => { + server.close(() => resolve()); + }), + ); + return `ws://127.0.0.1:${address.port}`; +} + +async function connectAndSend(url: string, metadata: object): Promise { + const ws = new WebSocket(url); + cleanups.push(async () => ws.terminate()); + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); + ws.send(Buffer.from(JSON.stringify(metadata)), { binary: true }); + return ws; +} + +async function expectUnauthorized(url: string): Promise { + const ws = new WebSocket(url); + cleanups.push(async () => ws.terminate()); + await expect( + new Promise((resolve, reject) => { + ws.once("unexpected-response", (_request, response) => resolve(response.statusCode ?? 0)); + ws.once("open", () => reject(new Error("unexpected node desktop attachment"))); + ws.once("error", () => undefined); + }), + ).resolves.toBe(401); +} + +describe("node desktop stream tickets", () => { + it("is single-use and resolves one ticket-bound binary stream", async () => { + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + + await connectAndSend(`${baseUrl}${minted.attachPath}`, { auth: "vnc-password" }); + const attached = await minted.attached; + expect(attached.auth).toBe("vnc-password"); + attached.stream.destroy(); + + await expectUnauthorized(`${baseUrl}${minted.attachPath}`); + }); + + it("buffers early RFB bytes while the pairing binding is rechecked", async () => { + let pairingChecks = 0; + let releaseRecheck!: () => void; + const recheck = new Promise((resolve) => { + releaseRecheck = resolve; + }); + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ + broker, + session, + pairingCurrent: async () => { + pairingChecks += 1; + if (pairingChecks > 1) { + await recheck; + } + return true; + }, + }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const ws = await connectAndSend(`${baseUrl}${minted.attachPath}`, { + auth: "vnc-password", + }); + const earlyBanner = Buffer.from("RFB 003.008\n", "ascii"); + ws.send(earlyBanner, { binary: true }); + await vi.waitFor(() => expect(pairingChecks).toBe(2)); + releaseRecheck(); + + const attached = await minted.attached; + await expect( + new Promise((resolve) => { + attached.stream.once("data", resolve); + }), + ).resolves.toEqual(earlyBanner); + attached.stream.destroy(); + }); + + it("rejects a stream error during the asynchronous pairing handoff", async () => { + let pairingChecks = 0; + let releaseRecheck!: () => void; + const recheck = new Promise((resolve) => { + releaseRecheck = resolve; + }); + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ + broker, + session, + pairingCurrent: async () => { + pairingChecks += 1; + if (pairingChecks > 1) { + await recheck; + } + return true; + }, + }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const ws = await connectAndSend(`${baseUrl}${minted.attachPath}`, { + auth: "vnc-password", + }); + await vi.waitFor(() => expect(pairingChecks).toBe(2)); + ws.send(Buffer.alloc(65 * 1024), { binary: true }); + + await expect(minted.attached).rejects.toThrow(); + releaseRecheck(); + }); + + it("rejects invalid metadata without exposing later WebSocket errors", async () => { + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const ws = await connectAndSend(`${baseUrl}${minted.attachPath}`, { auth: "none" }); + ws.send(Buffer.alloc(65 * 1024), { binary: true }); + + await expect(minted.attached).rejects.toThrow(); + }); + + it("rejects an expired ticket before upgrading", async () => { + let now = 1_000; + const broker = createNodeDesktopStreamBroker({ ttlMs: 60_000, now: () => now }); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + now = minted.expiresAtMs; + + await expectUnauthorized(`${baseUrl}${minted.attachPath}`); + await expect(minted.attached).rejects.toThrow("expired"); + }); + + it("rejects a ticket after connection or pairing generation replacement", async () => { + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const oldConnection = broker.mint({ nodeId: "node-1", ...session }); + session.connId = "conn-2"; + await expectUnauthorized(`${baseUrl}${oldConnection.attachPath}`); + await expect(oldConnection.attached).rejects.toThrow("stale"); + + const oldGeneration = broker.mint({ nodeId: "node-1", ...session }); + session.pairingGeneration = "generation-2"; + await expectUnauthorized(`${baseUrl}${oldGeneration.attachPath}`); + await expect(oldGeneration.attached).rejects.toThrow("stale"); + }); + + it("rechecks the pairing generation after a delayed metadata frame", async () => { + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const ws = new WebSocket(`${baseUrl}${minted.attachPath}`); + cleanups.push(async () => ws.terminate()); + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); + session.pairingGeneration = "generation-2"; + ws.send(Buffer.from(JSON.stringify({ auth: "vnc-password" })), { binary: true }); + + await expect(minted.attached).rejects.toThrow("stale"); + }); + + it("keeps a redeemed ticket cancellable while metadata is pending", async () => { + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ broker, session }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const ws = new WebSocket(`${baseUrl}${minted.attachPath}`); + cleanups.push(async () => ws.terminate()); + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); + const closed = new Promise((resolve) => { + ws.once("close", () => resolve()); + }); + + minted.cancel(); + + await expect(minted.attached).rejects.toThrow("cancelled"); + await expect(closed).resolves.toBeUndefined(); + }); + + it("rejects when the raw upgrade socket closes during pairing authorization", async () => { + let pairingChecks = 0; + let releaseCheck!: () => void; + const check = new Promise((resolve) => { + releaseCheck = resolve; + }); + const broker = createNodeDesktopStreamBroker(); + const session = { connId: "conn-1", pairingGeneration: "generation-1" }; + const baseUrl = await startBrokerServer({ + broker, + session, + pairingCurrent: async () => { + pairingChecks += 1; + await check; + return true; + }, + }); + const minted = broker.mint({ nodeId: "node-1", ...session }); + const url = new URL(baseUrl); + const socket = net.createConnection(Number(url.port), url.hostname); + cleanups.push(async () => { + socket.destroy(); + }); + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + socket.write( + [ + `GET ${minted.attachPath} HTTP/1.1`, + `Host: ${url.host}`, + "Connection: Upgrade", + "Upgrade: websocket", + "Sec-WebSocket-Version: 13", + "Sec-WebSocket-Key: dGVzdC1ub25jZS0xMjM0NQ==", + "", + "", + ].join("\r\n"), + ); + await vi.waitFor(() => expect(pairingChecks).toBe(1)); + + socket.destroy(); + + await expect(minted.attached).rejects.toThrow("authorization"); + releaseCheck(); + }); +}); diff --git a/src/gateway/desktop/node-stream-broker.ts b/src/gateway/desktop/node-stream-broker.ts new file mode 100644 index 000000000000..abf57fdedef1 --- /dev/null +++ b/src/gateway/desktop/node-stream-broker.ts @@ -0,0 +1,301 @@ +import crypto from "node:crypto"; +import type { IncomingMessage } from "node:http"; +import type { Duplex } from "node:stream"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { createWebSocketStream, WebSocket, WebSocketServer, type RawData } from "ws"; +import { registerSecretValueForRedaction } from "../../logging/secret-redaction-registry.js"; +import { NODE_DESKTOP_ATTACH_PATH } from "../../shared/node-desktop-stream.js"; +import type { NodeRegistry } from "../node-registry.js"; + +const DEFAULT_TICKET_TTL_MS = 60_000; +const TICKET_PATTERN = /^[a-f0-9]{48}$/u; +const MAX_ATTACH_FRAME_BYTES = 64 * 1024; + +type NodeDesktopStreamMetadata = { + auth: "vnc-password" | "ard-account"; + vncPassword?: string; +}; + +type AttachedNodeDesktopStream = NodeDesktopStreamMetadata & { stream: Duplex }; + +type NodeDesktopStreamBinding = { + nodeId: string; + connId: string; + pairingGeneration: string; +}; + +type TicketEntry = { + binding: NodeDesktopStreamBinding; + expiresAtMs: number; + resolve: (attached: AttachedNodeDesktopStream) => void; + reject: (error: Error) => void; + timer: ReturnType; + redeemed: boolean; + settled: boolean; + socket?: Duplex; + ws?: WebSocket; +}; + +type TicketNodeRegistry = Pick< + NodeRegistry, + "getForPairingGeneration" | "isConnectionCurrentPairingState" +>; + +function rawDataBuffer(data: RawData): Buffer { + if (Buffer.isBuffer(data)) { + return data; + } + if (Array.isArray(data)) { + return Buffer.concat(data); + } + return Buffer.from(data); +} + +function parseStreamMetadata(data: RawData, isBinary: boolean): NodeDesktopStreamMetadata { + const buffer = rawDataBuffer(data); + if (!isBinary || buffer.length === 0 || buffer.length > MAX_ATTACH_FRAME_BYTES) { + throw new Error("invalid node desktop attach metadata"); + } + let value: unknown; + try { + value = JSON.parse(buffer.toString("utf8")); + } catch { + throw new Error("invalid node desktop attach metadata"); + } + if (!isRecord(value) || (value.auth !== "vnc-password" && value.auth !== "ard-account")) { + throw new Error("invalid node desktop attach metadata"); + } + const keys = Object.keys(value); + if (keys.some((key) => key !== "auth" && key !== "vncPassword")) { + throw new Error("invalid node desktop attach metadata"); + } + if (value.vncPassword !== undefined && typeof value.vncPassword !== "string") { + throw new Error("invalid node desktop attach metadata"); + } + if (value.auth === "ard-account" && value.vncPassword !== undefined) { + throw new Error("invalid node desktop attach metadata"); + } + const vncPassword = typeof value.vncPassword === "string" ? value.vncPassword : undefined; + if (vncPassword) { + registerSecretValueForRedaction(vncPassword); + } + return { auth: value.auth, ...(vncPassword ? { vncPassword } : {}) }; +} + +function writeUnauthorized(socket: Duplex): void { + socket.write("HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n"); + socket.destroy(); +} + +function readAttachedStream( + ws: WebSocket, + onStreamError: (error: Error) => void, +): Promise<{ metadata: NodeDesktopStreamMetadata; stream: Duplex }> { + return new Promise((resolve, reject) => { + const cleanup = () => { + ws.off("message", onMessage); + ws.off("close", onClose); + }; + const onMessage = (data: RawData, isBinary: boolean) => { + cleanup(); + try { + const metadata = parseStreamMetadata(data, isBinary); + // Install the stream listener in the same message turn. The pairing + // recheck may yield, but early RFB banner bytes must already be buffered. + const stream = createWebSocketStream(ws, { allowHalfOpen: false }); + // Retain a safety listener through the asynchronous registry handoff; + // the observer bridge adds its own lifecycle handler after claiming it. + stream.on("error", onStreamError); + resolve({ + metadata, + stream, + }); + } catch (error) { + reject(error instanceof Error ? error : new Error(String(error))); + } + }; + const onClose = () => { + cleanup(); + reject(new Error("node desktop stream closed before attach")); + }; + const onError = (error: Error) => { + cleanup(); + onStreamError(error); + reject(error); + }; + ws.once("message", onMessage); + ws.once("close", onClose); + // Keep this listener for the WebSocket lifetime. Invalid metadata never + // creates a Duplex, so it remains the terminal protocol-error guard. + ws.on("error", onError); + }); +} + +/** Owns one-time node stream tickets and turns redeemed WebSockets into RFB duplexes. */ +export function createNodeDesktopStreamBroker(deps: { ttlMs?: number; now?: () => number } = {}) { + const ttlMs = deps.ttlMs ?? DEFAULT_TICKET_TTL_MS; + const now = deps.now ?? Date.now; + const tickets = new Map(); + const wss = new WebSocketServer({ noServer: true, maxPayload: MAX_ATTACH_FRAME_BYTES }); + + const remove = (ticket: string): TicketEntry | undefined => { + const entry = tickets.get(ticket); + if (!entry) { + return undefined; + } + tickets.delete(ticket); + clearTimeout(entry.timer); + return entry; + }; + + const rejectTicket = (ticket: string, error: Error): void => { + const entry = remove(ticket); + if (!entry || entry.settled) { + return; + } + entry.settled = true; + entry.reject(error); + entry.ws?.close(1008, "node desktop attach rejected"); + entry.socket?.destroy(); + }; + + const resolveTicket = (ticket: string, attached: AttachedNodeDesktopStream): void => { + const entry = remove(ticket); + if (!entry || entry.settled) { + attached.stream.destroy(); + return; + } + entry.settled = true; + entry.resolve(attached); + }; + + function mint(binding: NodeDesktopStreamBinding) { + const ticket = crypto.randomBytes(24).toString("hex"); + const expiresAtMs = now() + ttlMs; + let resolve!: (attached: AttachedNodeDesktopStream) => void; + let reject!: (error: Error) => void; + const attached = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + void attached.catch(() => undefined); + const timer = setTimeout(() => { + rejectTicket(ticket, new Error("node desktop stream ticket expired")); + }, ttlMs); + timer.unref?.(); + tickets.set(ticket, { + binding, + expiresAtMs, + resolve, + reject, + timer, + redeemed: false, + settled: false, + }); + return { + ticket, + attachPath: `${NODE_DESKTOP_ATTACH_PATH}?ticket=${ticket}`, + expiresAtMs, + attached, + cancel() { + rejectTicket(ticket, new Error("node desktop stream ticket cancelled")); + }, + }; + } + + const bindingIsCurrent = async ( + registry: TicketNodeRegistry, + binding: NodeDesktopStreamBinding, + ): Promise => { + const current = registry.getForPairingGeneration(binding.nodeId, binding.pairingGeneration); + if (!current || current.connId !== binding.connId) { + return false; + } + if (!(await registry.isConnectionCurrentPairingState(binding.connId))) { + return false; + } + const rechecked = registry.getForPairingGeneration(binding.nodeId, binding.pairingGeneration); + return rechecked?.connId === binding.connId; + }; + + async function handleUpgrade( + req: IncomingMessage, + socket: Duplex, + head: Buffer, + registry: TicketNodeRegistry, + ): Promise { + const resource = new URL(req.url ?? "/", "http://127.0.0.1"); + if (resource.pathname !== NODE_DESKTOP_ATTACH_PATH) { + return false; + } + const ticket = (resource.searchParams.get("ticket") ?? "").trim(); + if (!TICKET_PATTERN.test(ticket)) { + writeUnauthorized(socket); + return true; + } + const entry = tickets.get(ticket); + if (!entry || entry.redeemed || entry.expiresAtMs <= now()) { + writeUnauthorized(socket); + if (entry && !entry.redeemed) { + rejectTicket(ticket, new Error("node desktop stream ticket expired")); + } + return true; + } + entry.redeemed = true; + entry.socket = socket; + const onSocketError = (error: Error) => rejectTicket(ticket, error); + const onSocketClose = () => + rejectTicket(ticket, new Error("node desktop attach closed during authorization")); + socket.once("error", onSocketError); + socket.once("end", onSocketClose); + socket.once("close", onSocketClose); + let current: boolean; + try { + current = await bindingIsCurrent(registry, entry.binding); + } catch { + current = false; + } + if (entry.settled) { + return true; + } + if (!current) { + socket.off("error", onSocketError); + socket.off("end", onSocketClose); + socket.off("close", onSocketClose); + writeUnauthorized(socket); + rejectTicket(ticket, new Error("node desktop stream ticket binding is stale")); + return true; + } + socket.off("error", onSocketError); + socket.off("end", onSocketClose); + socket.off("close", onSocketClose); + try { + wss.handleUpgrade(req, socket, head, (ws) => { + entry.socket = undefined; + entry.ws = ws; + const attached = readAttachedStream(ws, (error) => rejectTicket(ticket, error)); + void (async () => { + try { + const resolved = await attached; + if (!(await bindingIsCurrent(registry, entry.binding))) { + throw new Error("node desktop stream ticket binding is stale"); + } + resolveTicket(ticket, { + ...resolved.metadata, + stream: resolved.stream, + }); + } catch (error) { + rejectTicket(ticket, error instanceof Error ? error : new Error(String(error))); + } + })(); + }); + } catch (error) { + rejectTicket(ticket, error instanceof Error ? error : new Error(String(error))); + } + return true; + } + + return { mint, handleUpgrade }; +} + +export type NodeDesktopStreamBroker = ReturnType; diff --git a/src/gateway/desktop/observe-bridge.test.ts b/src/gateway/desktop/observe-bridge.test.ts index af5157896d36..b571cf92e1ae 100644 --- a/src/gateway/desktop/observe-bridge.test.ts +++ b/src/gateway/desktop/observe-bridge.test.ts @@ -62,6 +62,7 @@ async function createProxyHarness( httpServer.on("upgrade", (req, socket, head) => { handleDesktopObserveUpgrade(req, socket, head, { registry: { + claimStream: () => undefined, attachObserver: (_environmentId, observer) => { closeObserver.mockImplementation((code: number, reason: string) => { observer.close(code, reason); diff --git a/src/gateway/desktop/observe-bridge.ts b/src/gateway/desktop/observe-bridge.ts index 45fa6fbbe475..a2381cd32830 100644 --- a/src/gateway/desktop/observe-bridge.ts +++ b/src/gateway/desktop/observe-bridge.ts @@ -2,7 +2,7 @@ import crypto from "node:crypto"; import type { IncomingMessage } from "node:http"; import type { Duplex } from "node:stream"; import { WebSocket, WebSocketServer, type RawData } from "ws"; -import { connectRfbAttachment, type RfbAttachment } from "./attachment.js"; +import { connectRfbAttachment, type DesktopRfbAttachment } from "./attachment.js"; import { preauthenticateRfb, RfbPreauthBuffer, @@ -24,7 +24,7 @@ type DesktopObserverTokenEntry = { sourceKey: string; ownerEpoch: number; control: boolean; - attachment: RfbAttachment; + attachment: DesktopRfbAttachment; preauth?: RfbPreauthDescriptor; expiresAt: number; }; @@ -54,7 +54,7 @@ export function mintDesktopObserverToken(params: { sourceKey: string; ownerEpoch: number; control: boolean; - attachment: RfbAttachment; + attachment: DesktopRfbAttachment; preauth?: RfbPreauthDescriptor; nowMs?: number; }): { token: string; expiresAtMs: number } { @@ -178,7 +178,7 @@ export function handleDesktopObserveUpgrade( socket: Duplex, head: Buffer, deps: { - registry: Pick; + registry: Pick; getBufferedAmount?: (ws: WebSocket) => number; }, ): boolean { @@ -193,6 +193,12 @@ export function handleDesktopObserveUpgrade( return true; } desktopObserverWss.handleUpgrade(req, socket, head, (ws) => { + const claimedStream = + entry.attachment.kind === "stream" ? deps.registry.claimStream(entry.attachment) : undefined; + if (entry.attachment.kind === "stream" && !claimedStream) { + ws.close(1013, "desktop stream unavailable"); + return; + } // View-only is enforced here at the RFB message boundary; the UI setting is only UX. const observer = deps.registry.attachObserver(entry.sourceKey, { control: entry.control, @@ -200,10 +206,17 @@ export function handleDesktopObserveUpgrade( close: (code, reason) => ws.close(code, reason), }); if (!observer) { + claimedStream?.destroy(); ws.close(1013, "desktop observer limit"); return; } - const desktopSocket = connectRfbAttachment(entry.attachment); + const desktopSocket = + entry.attachment.kind === "stream" ? claimedStream : connectRfbAttachment(entry.attachment); + if (!desktopSocket) { + observer.release(); + ws.close(1013, "desktop stream unavailable"); + return; + } let closed = false; let negotiating = Boolean(entry.preauth); let resumeTimer: ReturnType | undefined; diff --git a/src/gateway/desktop/session-registry.ts b/src/gateway/desktop/session-registry.ts index 65230f66a95b..95de7811043e 100644 --- a/src/gateway/desktop/session-registry.ts +++ b/src/gateway/desktop/session-registry.ts @@ -1,4 +1,5 @@ -import type { RfbAttachment } from "./attachment.js"; +import { randomUUID } from "node:crypto"; +import type { ConnectedRfbStream, DesktopRfbAttachment } from "./attachment.js"; const DEFAULT_LINGER_MS = 60_000; const MAX_OBSERVERS = 8; @@ -25,7 +26,7 @@ type DesktopSessionObserver = { }; type DesktopSessionAcquireResult = { - attachment: RfbAttachment; + attachment: DesktopRfbAttachment; auth?: "vnc-password" | "ard-account"; vncPassword?: string; }; @@ -37,22 +38,27 @@ type DesktopSessionAcquireRequest = { teardown?: () => Promise; }; +type DesktopSessionActivateRequest = Omit; +type DesktopSessionStartResult = DesktopSessionAcquireResult | undefined; + type ObserverEntry = DesktopSessionObserver & { released: boolean }; type DesktopSessionEntry = { sourceKey: string; ownerEpoch: number; initialization?: Promise; stopPromise?: Promise; - ready: Promise; - resolveReady: (result: DesktopSessionAcquireResult) => void; + ready: Promise; + resolveReady: (result: DesktopSessionStartResult) => void; rejectReady: (error: Error) => void; readySettled: boolean; observers: Set; + observerReservations: Set; controller?: ObserverEntry; lingerTimer?: ReturnType; stopped: boolean; - start: DesktopSessionAcquireRequest["start"]; + start: (isCurrent: () => boolean) => Promise; teardown?: DesktopSessionAcquireRequest["teardown"]; + pendingStreams: Map; }; /** Owns per-source desktop sessions and their connected observer lifetimes. */ @@ -105,6 +111,12 @@ export function createDesktopSessionRegistry( } entry.observers.clear(); entry.controller = undefined; + for (const pending of entry.pendingStreams.values()) { + pending.reservation.release(); + pending.stream.destroy(); + } + entry.pendingStreams.clear(); + entry.observerReservations.clear(); if (!entry.readySettled) { entry.readySettled = true; entry.rejectReady(new DesktopSessionStoppedError()); @@ -118,9 +130,17 @@ export function createDesktopSessionRegistry( return entry.stopPromise; }; - async function acquire( - request: DesktopSessionAcquireRequest, - ): Promise { + const scheduleLinger = (entry: DesktopSessionEntry): void => { + clearTimeout(entry.lingerTimer); + entry.lingerTimer = setTimeout(() => void stopEntry(entry), lingerMs); + entry.lingerTimer.unref?.(); + }; + + async function startSession( + request: + | DesktopSessionAcquireRequest + | (DesktopSessionActivateRequest & { start: () => Promise }), + ): Promise { claimOwnerEpoch(request.sourceKey, request.ownerEpoch); const current = entries.get(request.sourceKey); if (current) { @@ -132,9 +152,9 @@ export function createDesktopSessionRegistry( } } - let resolveReady!: (result: DesktopSessionAcquireResult) => void; + let resolveReady!: (result: DesktopSessionStartResult) => void; let rejectReady!: (error: Error) => void; - const ready = new Promise((resolve, reject) => { + const ready = new Promise((resolve, reject) => { resolveReady = resolve; rejectReady = reject; }); @@ -147,6 +167,8 @@ export function createDesktopSessionRegistry( rejectReady, readySettled: false, observers: new Set(), + observerReservations: new Set(), + pendingStreams: new Map(), stopped: false, start: request.start, ...(request.teardown ? { teardown: request.teardown } : {}), @@ -176,9 +198,36 @@ export function createDesktopSessionRegistry( return await ready; } + async function acquire( + request: DesktopSessionAcquireRequest, + ): Promise { + const result = await startSession(request); + if (!result) { + throw new Error("Desktop session attachment is unavailable"); + } + return result; + } + + async function activate(request: DesktopSessionActivateRequest): Promise { + await startSession({ ...request, start: async () => undefined }); + const entry = entries.get(request.sourceKey); + if ( + entry?.ownerEpoch === request.ownerEpoch && + entry.observers.size === 0 && + entry.observerReservations.size === 0 + ) { + scheduleLinger(entry); + } + } + function attachObserver(sourceKey: string, observer: DesktopSessionObserver) { const entry = entries.get(sourceKey); - if (!entry || !entry.readySettled || entry.stopped || entry.observers.size >= MAX_OBSERVERS) { + if ( + !entry || + !entry.readySettled || + entry.stopped || + entry.observers.size + entry.observerReservations.size >= MAX_OBSERVERS + ) { return undefined; } // A token minted against a replaced entry must not reach this one; otherwise a stale @@ -210,14 +259,114 @@ export function createDesktopSessionRegistry( if (entry.controller === attached) { entry.controller = undefined; } - if (entry.observers.size === 0 && isCurrent(entry)) { - entry.lingerTimer = setTimeout(() => void stopEntry(entry), lingerMs); - entry.lingerTimer.unref?.(); + if ( + entry.observers.size === 0 && + entry.observerReservations.size === 0 && + isCurrent(entry) + ) { + scheduleLinger(entry); } }, }; } + function reserveObserver(sourceKey: string, ownerEpoch: number) { + const entry = entries.get(sourceKey); + if ( + !entry || + entry.stopped || + entry.ownerEpoch !== ownerEpoch || + entry.observers.size + entry.observerReservations.size >= MAX_OBSERVERS + ) { + return undefined; + } + const reservationId = Symbol("desktop-observer"); + entry.observerReservations.add(reservationId); + clearTimeout(entry.lingerTimer); + entry.lingerTimer = undefined; + let released = false; + return { + sourceKey, + ownerEpoch, + release() { + if (released) { + return; + } + released = true; + entry.observerReservations.delete(reservationId); + if ( + entry.observers.size === 0 && + entry.observerReservations.size === 0 && + isCurrent(entry) + ) { + scheduleLinger(entry); + } + }, + }; + } + + function publishStream(params: { + sourceKey: string; + ownerEpoch: number; + stream: ConnectedRfbStream; + reservation: NonNullable>; + }) { + const entry = entries.get(params.sourceKey); + if ( + !entry || + entry.stopped || + entry.ownerEpoch !== params.ownerEpoch || + params.reservation.sourceKey !== params.sourceKey || + params.reservation.ownerEpoch !== params.ownerEpoch + ) { + params.reservation.release(); + params.stream.destroy(); + return undefined; + } + if (params.stream.destroyed || params.stream.readableEnded || params.stream.writableEnded) { + params.reservation.release(); + params.stream.destroy(); + return undefined; + } + const streamId = randomUUID(); + const pending = { stream: params.stream, reservation: params.reservation }; + entry.pendingStreams.set(streamId, pending); + params.stream.once("close", () => { + if (entry.pendingStreams.get(streamId) === pending) { + entry.pendingStreams.delete(streamId); + params.reservation.release(); + } + }); + return { kind: "stream", streamId } as const; + } + + function claimStream(attachment: { kind: "stream"; streamId: string }) { + for (const entry of entries.values()) { + const pending = entry.pendingStreams.get(attachment.streamId); + if (!pending) { + continue; + } + entry.pendingStreams.delete(attachment.streamId); + pending.reservation.release(); + const stream = pending.stream; + if (stream.destroyed || stream.readableEnded || stream.writableEnded) { + stream.destroy(); + return undefined; + } + return stream; + } + return undefined; + } + + function hasPendingStream(attachment: { kind: "stream"; streamId: string }): boolean { + for (const entry of entries.values()) { + if (entry.pendingStreams.has(attachment.streamId)) { + return true; + } + } + return false; + } + async function stop(sourceKey: string, ownerEpoch?: number): Promise { const entry = entries.get(sourceKey); if (entry && (ownerEpoch === undefined || ownerEpoch === entry.ownerEpoch)) { @@ -242,7 +391,12 @@ export function createDesktopSessionRegistry( return { acquire, + activate, attachObserver, + publishStream, + claimStream, + hasPendingStream, + reserveObserver, claimOwnerEpoch, isOwnerEpochCurrent: (sourceKey: string, ownerEpoch: number) => claimedOwnerEpochs.get(sourceKey) === ownerEpoch, diff --git a/src/gateway/node-command-policy.test.ts b/src/gateway/node-command-policy.test.ts index 4dda1d978196..ce82b5b851d4 100644 --- a/src/gateway/node-command-policy.test.ts +++ b/src/gateway/node-command-policy.test.ts @@ -9,6 +9,7 @@ import { import type { OpenClawConfig } from "../config/types.openclaw.js"; import { createEmptyPluginRegistry } from "../plugins/registry-empty.js"; import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../plugins/runtime.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import { isForegroundRestrictedPluginNodeCommand, isNodeCommandAllowed, @@ -45,6 +46,59 @@ describe("gateway/node-command-policy", () => { return registry; } + it("keeps desktop streaming dangerous, advertised, explicitly allowed, and deny-wins", () => { + const node = { + platform: "linux", + deviceFamily: "Linux", + commands: [NODE_DESKTOP_STREAM_COMMAND], + approvedCommands: [NODE_DESKTOP_STREAM_COMMAND], + }; + expect( + resolveNodeCommandAllowlist({} as OpenClawConfig, node).has(NODE_DESKTOP_STREAM_COMMAND), + ).toBe(false); + expect( + resolveNodePairingCommandAllowlist({} as OpenClawConfig, { + platform: node.platform, + deviceFamily: node.deviceFamily, + commands: node.commands, + }).has(NODE_DESKTOP_STREAM_COMMAND), + ).toBe(false); + + const allowedConfig = { + gateway: { nodes: { commands: { allow: [NODE_DESKTOP_STREAM_COMMAND] } } }, + } as OpenClawConfig; + const allowed = resolveNodeCommandAllowlist(allowedConfig, node); + expect( + isNodeCommandAllowed({ + command: NODE_DESKTOP_STREAM_COMMAND, + declaredCommands: node.commands, + allowlist: allowed, + }), + ).toEqual({ ok: true }); + expect( + isNodeCommandAllowed({ + command: NODE_DESKTOP_STREAM_COMMAND, + declaredCommands: [], + allowlist: allowed, + }), + ).toEqual({ ok: false, reason: "node did not declare commands" }); + + const denied = resolveNodeCommandAllowlist( + { + gateway: { + nodes: { + commands: { + allow: [NODE_DESKTOP_STREAM_COMMAND], + deny: [NODE_DESKTOP_STREAM_COMMAND], + }, + }, + }, + } as OpenClawConfig, + node, + ); + expect(denied.has(NODE_DESKTOP_STREAM_COMMAND)).toBe(false); + }); + it("normalizes declared node commands against the allowlist", () => { const allowlist = new Set(["canvas.snapshot", "system.run"]); expect( diff --git a/src/gateway/node-command-policy.ts b/src/gateway/node-command-policy.ts index b4047261fce0..e85e76921f85 100644 --- a/src/gateway/node-command-policy.ts +++ b/src/gateway/node-command-policy.ts @@ -15,6 +15,7 @@ import { NODE_SYSTEM_RUN_COMMANDS, } from "../infra/node-commands.js"; import { getActivePluginGatewayNodePolicyRegistry } from "../plugins/runtime.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import { normalizeDeviceMetadataForPolicy } from "./device-metadata-normalization.js"; import { MOBILE_NODE_COMMANDS } from "./node-command-policy-mobile.js"; import type { NodeSession } from "./node-registry.js"; @@ -25,7 +26,7 @@ const MAC_CAMERA_COMMANDS = ["camera.ptz.status"]; const CAMERA_DANGEROUS_COMMANDS = ["camera.snap", "camera.clip", "camera.ptz.control"]; const SCREEN_COMMANDS = ["screen.snapshot"]; -const SCREEN_DANGEROUS_COMMANDS = ["screen.record"]; +const SCREEN_DANGEROUS_COMMANDS = ["screen.record", NODE_DESKTOP_STREAM_COMMAND]; // Desktop computer use is advertised only while the node-local control is // enabled. Pairing approval of that advertised surface is the durable grant. @@ -92,6 +93,7 @@ const DESKTOP_HOST_COMMANDS = new Set([ NODE_MCP_TOOLS_CALL_COMMAND, NODE_AGENT_CLI_CLAUDE_RUN_COMMAND, ...SCREEN_COMMANDS, + NODE_DESKTOP_STREAM_COMMAND, ]); const UNKNOWN_PLATFORM_COMMANDS = [ ...CAMERA_COMMANDS, diff --git a/src/gateway/server-http.ts b/src/gateway/server-http.ts index 2024d7e325a9..8476bb99ae37 100644 --- a/src/gateway/server-http.ts +++ b/src/gateway/server-http.ts @@ -25,6 +25,7 @@ import { isGatewayWorkAdmissionClosed, } from "../process/gateway-work-admission.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; +import { NODE_DESKTOP_ATTACH_PATH } from "../shared/node-desktop-stream.js"; import { resolveRuntimeServiceVersion } from "../version.js"; import { resolveAssistantIdentity } from "./assistant-identity.js"; import type { AuthRateLimiter } from "./auth-rate-limit.js"; @@ -46,6 +47,7 @@ import { isControlUiPluginManagerRequest, } from "./control-ui-routing.js"; import type { ControlUiRootState } from "./control-ui.js"; +import type { NodeDesktopStreamBroker } from "./desktop/node-stream-broker.js"; import type { DesktopSessionRegistry } from "./desktop/session-registry.js"; import { classifyGatewayProbePath, @@ -63,6 +65,7 @@ import { normalizePluginNodeCapabilityScopedUrl, type PluginNodeCapabilitySurface, } from "./plugin-node-capability.js"; +import type { GatewayRequestContext } from "./server-methods/types.js"; import type { HooksRequestHandler } from "./server/hooks-request-handler.js"; import { runWithGatewayHttpWorkAdmission, @@ -940,6 +943,8 @@ export function attachGatewayUpgradeHandler(opts: { /** Optional logger for error diagnostics. */ log?: { warn: (msg: string) => void }; desktopSessionRegistry?: DesktopSessionRegistry; + nodeDesktopStreamBroker?: NodeDesktopStreamBroker; + getGatewayRequestContext?: () => GatewayRequestContext | undefined; }) { const { httpServer, @@ -1119,6 +1124,21 @@ export function attachGatewayUpgradeHandler(opts: { }); return; } + if (requestPath === NODE_DESKTOP_ATTACH_PATH) { + const context = opts.getGatewayRequestContext?.(); + if (!opts.nodeDesktopStreamBroker || !context) { + writeGatewayUpgradeServiceUnavailable(socket, "node desktop attach unavailable"); + socket.destroy(); + return; + } + if (isGatewayWorkAdmissionClosed()) { + writeGatewayUpgradeServiceUnavailable(socket, "Gateway websocket admission closed"); + socket.destroy(); + return; + } + await opts.nodeDesktopStreamBroker.handleUpgrade(req, socket, head, context.nodeRegistry); + return; + } // Plugin-owned upgrade routes have already had the opportunity to claim the socket. // Core Gateway control connections remain reachable while suspension is prepared. try { diff --git a/src/gateway/server-kernel-request-runtime.ts b/src/gateway/server-kernel-request-runtime.ts index 7b47f29d0e34..587d2257b120 100644 --- a/src/gateway/server-kernel-request-runtime.ts +++ b/src/gateway/server-kernel-request-runtime.ts @@ -63,6 +63,7 @@ export async function prepareGatewayKernelRequestRuntime(params: { claimControlUiDeviceAuthMigration, releaseControlUiDeviceAuthMigrationClaim, nodeRegistry, + nodeDesktopService, workerEnvironmentService, hostDesktopService, workerEnvironmentStartup, @@ -163,6 +164,7 @@ export async function prepareGatewayKernelRequestRuntime(params: { releaseControlUiDeviceAuthMigrationClaim: (deviceId: string) => releaseControlUiDeviceAuthMigrationClaim(deviceId, { env: process.env }), nodeRegistry, + ...(nodeDesktopService ? { nodeDesktopService } : {}), ...(workerEnvironmentService ? { workerEnvironmentService } : {}), ...(hostDesktopService ? { hostDesktopService } : {}), ...(workerEnvironmentStartup diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index b2a2fb68033f..bf9ad3a09690 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -100,6 +100,8 @@ export async function prepareGatewayLifecycle(params: { defaultWorkspaceDir, activeTaskCount, residentRegistry, + desktopSessionRegistry, + nodeDesktopStreamBroker, } = runtime; const completeControlUiDeviceAuthMigrationForEffectiveOperator = ( device: EffectiveOperatorDeviceIdentity, @@ -155,6 +157,9 @@ export async function prepareGatewayLifecycle(params: { const unsubscribeSessionMessageEvents: GatewayRequestContext["unsubscribeSessionMessageEvents"] = (connId, sessionKey) => sessionMessageSubscribers.unsubscribe(connId, sessionKey); const restartRecoveryCandidates = new Map(); + const nodeDesktopServiceRef: { + current?: import("./desktop/node-source.js").NodeDesktopService; + } = {}; const { createGatewayNodeSessionRuntime } = await import("./server-node-session-runtime.js"); const { nodeRegistry, @@ -175,11 +180,25 @@ export async function prepareGatewayLifecycle(params: { nodePluginToolsEnabled: cfgAtStart.gateway?.nodes?.pluginTools?.enabled !== false, nodeSkillsEnabled: cfgAtStart.gateway?.nodes?.allowSkills !== false, onPairingInvalidated: ({ nodeId, connId }) => { + void nodeDesktopServiceRef.current?.stopNode(nodeId); upsertPresence(nodeId, { reason: "disconnect" }); broadcastPresenceSnapshot({ broadcast, incrementPresenceVersion, getHealthVersion }); removeRemoteNodeInfoForConnection(nodeId, connId); }, + onPairingGenerationChanged: ({ nodeId }) => { + void nodeDesktopServiceRef.current?.stopNode(nodeId); + }, }); + const nodeDesktopService = + desktopSessionRegistry && nodeDesktopStreamBroker + ? (await import("./desktop/node-source.js")).createNodeDesktopService({ + getConfig: getRuntimeConfig, + nodeRegistry, + desktopRegistry: desktopSessionRegistry, + streamBroker: nodeDesktopStreamBroker, + }) + : undefined; + nodeDesktopServiceRef.current = nodeDesktopService; const { createWatchNodeHttpRuntime } = await import("./watch-node-http.js"); const watchNodeHttpRuntime = createWatchNodeHttpRuntime({ nodeRegistry, @@ -633,6 +652,7 @@ export async function prepareGatewayLifecycle(params: { unsubscribeSessionMessageEvents, restartRecoveryCandidates, nodeRegistry, + nodeDesktopService, nodePresenceTimers, nodeSendToSession, nodeSendToAllSubscribed, diff --git a/src/gateway/server-methods/environments.desktop.test.ts b/src/gateway/server-methods/environments.desktop.test.ts index f595b934cb59..7ae5d28d4fc5 100644 --- a/src/gateway/server-methods/environments.desktop.test.ts +++ b/src/gateway/server-methods/environments.desktop.test.ts @@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { ErrorCodes } from "../../../packages/gateway-protocol/src/index.js"; import { HostDesktopCredentialsRequiredError } from "../desktop/host-source-errors.js"; import { createHostDesktopService } from "../desktop/host-source.js"; +import { NODE_DESKTOP_SERVICE_CONTEXT } from "../desktop/node-source-context.js"; import { createDesktopSessionRegistry } from "../desktop/session-registry.js"; import { environmentsHandlers } from "./environments.js"; @@ -160,12 +161,31 @@ describe("desktop gateway methods", () => { }); it("rejects unknown desktop source kinds before dispatch", async () => { - const [ok, , error] = await invoke( - "desktop.observe", - { source: { kind: "node", nodeId: "one" } }, - {}, - ); + const [ok, , error] = await invoke("desktop.observe", { source: { kind: "future" } }, {}); expect(ok).toBe(false); expect(error.code).toBe(ErrorCodes.INVALID_REQUEST); }); + + it("forwards node credentials only to the paired-node desktop service", async () => { + const observe = vi.fn(async () => ({ + transport: "rfb" as const, + wsPath: "/desktop/observe?token=node", + expiresAtMs: 42, + control: false, + auth: "vnc-password" as const, + })); + const credentials = { password: "memory-only-node-password" }; + const [ok, result] = await invoke( + "desktop.observe", + { source: { kind: "node", nodeId: "node-1" }, credentials }, + { [NODE_DESKTOP_SERVICE_CONTEXT]: { observe } }, + ); + expect(ok).toBe(true); + expect(result).not.toHaveProperty("vncPassword"); + expect(observe).toHaveBeenCalledWith({ + nodeId: "node-1", + control: false, + credentials, + }); + }); }); diff --git a/src/gateway/server-methods/environments.test.ts b/src/gateway/server-methods/environments.test.ts index 20e34ceaad34..c3c2ec949339 100644 --- a/src/gateway/server-methods/environments.test.ts +++ b/src/gateway/server-methods/environments.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ErrorCodes } from "../../../packages/gateway-protocol/src/index.js"; import { listNodePairing } from "../../infra/device-pairing-node.js"; import { listDevicePairing } from "../../infra/device-pairing.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../../shared/node-desktop-stream.js"; import type { WorkerEnvironmentServiceRecord } from "../worker-environments/service-contract.js"; import type { WorkerEnvironmentRecord } from "../worker-environments/store.js"; import { environmentsHandlers, summarizeWorkerEnvironment } from "./environments.js"; @@ -237,6 +238,52 @@ describe("environment gateway methods", () => { }); }); + it("marks only connected, advertised, and explicitly allowed nodes as desktop sources", async () => { + const context = mockContext(); + context.getRuntimeConfig = () => + ({ + gateway: { nodes: { commands: { allow: [NODE_DESKTOP_STREAM_COMMAND] } } }, + }) as never; + context.nodeRegistry.listConnectedForPairingStates = () => + [ + { + nodeId: "node-desktop", + connId: "conn-desktop", + displayName: "Desktop Node", + platform: "linux", + deviceFamily: "Linux", + caps: [], + commands: [NODE_DESKTOP_STREAM_COMMAND], + connectedAtMs: 123, + }, + { + nodeId: "node-without-command", + connId: "conn-plain", + displayName: "Plain Node", + platform: "linux", + deviceFamily: "Linux", + caps: [], + commands: [], + connectedAtMs: 123, + }, + ] as never; + const respond = vi.fn(); + await environmentsHandlers["environments.list"]?.({ + params: {}, + respond, + context, + } as never); + const environments = respond.mock.calls[0]?.[1].environments as Array<{ + id: string; + desktop?: boolean; + }>; + expect(environments.find((entry) => entry.id === "node:node-desktop")?.desktop).toBe(true); + expect( + environments.find((entry) => entry.id === "node:node-without-command")?.desktop, + ).toBeUndefined(); + expect(environments.find((entry) => entry.id === "node:node-offline")?.desktop).toBeUndefined(); + }); + it("appends worker metadata with stable sessions and elapsed times", async () => { const service = workerService({ list: vi.fn(() => [ diff --git a/src/gateway/server-methods/environments.ts b/src/gateway/server-methods/environments.ts index 47cbc8d681d4..35e5cbd1ca5c 100644 --- a/src/gateway/server-methods/environments.ts +++ b/src/gateway/server-methods/environments.ts @@ -15,9 +15,12 @@ import { } from "../../../packages/gateway-protocol/src/index.js"; import { listNodePairing } from "../../infra/device-pairing-node.js"; import { listDevicePairing, resolveNodePairingState } from "../../infra/device-pairing.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../../shared/node-desktop-stream.js"; import type { NodeListNode } from "../../shared/node-list-types.js"; -import { isHostDesktopCredentialsRequiredError } from "../desktop/host-source-errors.js"; +import { isDesktopCredentialsRequiredError } from "../desktop/host-source-errors.js"; +import { getNodeDesktopService } from "../desktop/node-source-context.js"; import { createKnownNodeCatalog, listKnownNodes } from "../node-catalog.js"; +import { isNodeCommandAllowed, resolveNodeCommandAllowlist } from "../node-command-policy.js"; import type { WorkerEnvironmentServiceRecord } from "../worker-environments/service-contract.js"; import type { WorkerEnvironmentState } from "../worker-environments/state.js"; import { formatForLog } from "../ws-log.js"; @@ -57,11 +60,26 @@ function rejectInvalid( ) { return respondInvalidParams({ respond, method, validator }); } -function summarizeNodeEnvironment(node: NodeListNode): EnvironmentSummary { +function summarizeNodeEnvironment( + node: NodeListNode, + config: Parameters[0], +): EnvironmentSummary { // Expose both declared capabilities and command names so older node // runtimes still advertise useful execution surfaces in one stable list. const capabilities = uniqueSortedStrings(node.caps, node.commands); const platform = node.platform?.trim(); + const desktop = + node.connected === true && + isNodeCommandAllowed({ + command: NODE_DESKTOP_STREAM_COMMAND, + declaredCommands: node.commands, + allowlist: resolveNodeCommandAllowlist(config, { + platform: node.platform, + deviceFamily: node.deviceFamily, + commands: node.commands, + approvedCommands: node.commands, + }), + }).ok; return { id: `node:${node.nodeId}`, type: "node", @@ -70,6 +88,7 @@ function summarizeNodeEnvironment(node: NodeListNode): EnvironmentSummary { ...(platform ? { platform } : {}), sessionHost: false, trust: "persistent", + ...(desktop ? { desktop: true } : {}), ...(capabilities.length > 0 ? { capabilities } : {}), }; } @@ -121,11 +140,15 @@ async function listEnvironments(context: GatewayRequestContext): Promise summarizeNodeEnvironment(node, config)), + ]; } function listWorkerEnvironments(context: GatewayRequestContext): WorkerEnvironmentServiceRecord[] { try { @@ -207,7 +230,7 @@ async function respondDesktopObserve(params: { undefined, ); } catch (error) { - if (isHostDesktopCredentialsRequiredError(error)) { + if (isDesktopCredentialsRequiredError(error)) { params.respond( false, undefined, @@ -234,6 +257,54 @@ async function respondDesktopObserve(params: { return; } + if (params.request.source.kind === "node") { + const service = getNodeDesktopService(params.context); + if (!service) { + params.respond( + false, + undefined, + errorShape( + ErrorCodes.INVALID_REQUEST, + "node desktop is disabled; explicitly allow desktop.stream, then restart the gateway", + ), + ); + return; + } + try { + params.respond( + true, + await service.observe({ + nodeId: params.request.source.nodeId, + control: params.request.control ?? false, + ...("credentials" in params.request && params.request.credentials + ? { credentials: params.request.credentials } + : {}), + }), + undefined, + ); + } catch (error) { + if (isDesktopCredentialsRequiredError(error)) { + params.respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, error.message, { + details: { code: error.detailCode, auth: error.auth }, + }), + ); + return; + } + params.respond( + false, + undefined, + errorShape( + ErrorCodes.UNAVAILABLE, + error instanceof Error ? error.message : "node desktop observe unavailable", + ), + ); + } + return; + } + const service = params.context.workerEnvironmentService; if (!service) { params.respond( diff --git a/src/gateway/server-node-session-runtime.test.ts b/src/gateway/server-node-session-runtime.test.ts index fa51ff35b5e5..a679eb187f96 100644 --- a/src/gateway/server-node-session-runtime.test.ts +++ b/src/gateway/server-node-session-runtime.test.ts @@ -72,6 +72,25 @@ function registerNode( } describe("gateway node session runtime", () => { + test("publishes pairing-generation transitions to lifecycle consumers", () => { + const onPairingGenerationChanged = vi.fn(); + const runtime = createGatewayNodeSessionRuntime({ + broadcast: vi.fn(), + onPairingGenerationChanged, + sessionEventSubscribers: createSessionEventSubscriberRegistry(), + sessionMessageSubscribers: createSessionMessageSubscriberRegistry(), + }); + registerNode(runtime, "conn-original", "generation-a", []); + registerNode(runtime, "conn-replacement", "generation-b", []); + + expect(onPairingGenerationChanged).toHaveBeenCalledWith({ + nodeId: "node-a", + previousPairingGeneration: "generation-a", + nextPairingGeneration: "generation-b", + preserveSessionState: false, + }); + }); + test("forwards subscribed payload json without parsing it again", async () => { const frames: string[] = []; const runtime = createRuntime(async () => "generation-a"); diff --git a/src/gateway/server-node-session-runtime.ts b/src/gateway/server-node-session-runtime.ts index 97895e22633f..246e750322b2 100644 --- a/src/gateway/server-node-session-runtime.ts +++ b/src/gateway/server-node-session-runtime.ts @@ -29,6 +29,7 @@ export function createGatewayNodeSessionRuntime(params: { resolveCurrentPairingState?: NodeRegistryOptions["resolveCurrentPairingState"]; isPairingStateCurrent?: NodeRegistryOptions["isPairingStateCurrent"]; onPairingInvalidated?: NodeRegistryOptions["onPairingInvalidated"]; + onPairingGenerationChanged?: NodeRegistryOptions["onPairingGenerationChanged"]; sessionEventSubscribers: SessionEventSubscriberRegistry; sessionMessageSubscribers: SessionMessageSubscriberRegistry; }) { @@ -46,6 +47,7 @@ export function createGatewayNodeSessionRuntime(params: { ...change, preserveSubscriptions: change.preserveSessionState, }); + params.onPairingGenerationChanged?.(change); }, }); const nodePresenceTimers = new Map>(); diff --git a/src/gateway/server-request-context.ts b/src/gateway/server-request-context.ts index 77d130806fdb..32434eb890c7 100644 --- a/src/gateway/server-request-context.ts +++ b/src/gateway/server-request-context.ts @@ -10,6 +10,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { upsertPresence } from "../infra/system-presence.js"; import { resolveUserProfileId } from "../state/user-profiles.js"; import { buildAuthenticatedPresenceUser } from "./authenticated-presence-user.js"; +import { NODE_DESKTOP_SERVICE_CONTEXT } from "./desktop/node-source-context.js"; import type { GatewayServerLiveState } from "./server-live-state.js"; import type { GatewayClient, GatewayRequestContext } from "./server-methods/types.js"; import { disconnectAllSharedGatewayAuthClients } from "./server-shared-auth-generation.js"; @@ -79,6 +80,7 @@ type GatewayRequestContextParams = { scopes: string[]; }) => void; nodeRegistry: GatewayRequestContext["nodeRegistry"]; + nodeDesktopService?: import("./desktop/node-source.js").NodeDesktopService; workerEnvironmentService?: GatewayRequestContext["workerEnvironmentService"]; hostDesktopService?: GatewayRequestContext["hostDesktopService"]; workerSessionPlacementService?: GatewayRequestContext["workerSessionPlacementService"]; @@ -365,6 +367,9 @@ export function createGatewayRequestContext( releaseControlUiDeviceAuthMigrationClaim: params.releaseControlUiDeviceAuthMigrationClaim, completeControlUiDeviceAuthMigration: params.completeControlUiDeviceAuthMigration, nodeRegistry: params.nodeRegistry, + ...(params.nodeDesktopService + ? { [NODE_DESKTOP_SERVICE_CONTEXT]: params.nodeDesktopService } + : {}), ...(params.workerEnvironmentService ? { workerEnvironmentService: params.workerEnvironmentService } : {}), diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index 231641ad98be..5d6283c1d463 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -12,6 +12,7 @@ import { runtimeForLogger } from "../logging/subsystem.js"; import { isGatewayDraining } from "../process/command-queue.js"; import type { RuntimeEnv } from "../runtime.js"; import { getActiveSecretsRuntimeConfigSnapshot } from "../secrets/runtime-state.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import { createAuthRateLimiter, type AuthRateLimiter } from "./auth-rate-limit.js"; import { resolveGatewayAuth } from "./auth.js"; import { createDesktopSessionRegistry } from "./desktop/session-registry.js"; @@ -117,13 +118,29 @@ export async function prepareGatewayKernelState(params: { Boolean(workerEnvironmentStartup?.hasNonlocalPlacementRecords); const hostDesktopConfig = gatewayPluginConfigAtStart.desktop?.host; const hostDesktopEnabled = hostDesktopConfig?.enabled === true; + const nodeCommandConfig = gatewayPluginConfigAtStart.gateway?.nodes?.commands; + const nodeDesktopObserveAvailable = + (nodeCommandConfig?.allow ?? []).some( + (command) => command.trim() === NODE_DESKTOP_STREAM_COMMAND, + ) && + !(nodeCommandConfig?.deny ?? []).some( + (command) => command.trim() === NODE_DESKTOP_STREAM_COMMAND, + ); const workerGatewayEndpoint = { resolve: (() => undefined) as () => { host: "127.0.0.1" | "::1"; port: number } | undefined, }; const desktopSessionRegistry = - shouldStartWorkerEnvironmentService || hostDesktopEnabled + shouldStartWorkerEnvironmentService || hostDesktopEnabled || nodeDesktopObserveAvailable ? createDesktopSessionRegistry() : undefined; + const nodeDesktopStreamBroker = nodeDesktopObserveAvailable + ? ( + await startupTrace.measure( + "node-desktop.runtime-import", + () => import("./desktop/node-stream-broker.js"), + ) + ).createNodeDesktopStreamBroker() + : undefined; const hostDesktopService = hostDesktopConfig && hostDesktopEnabled && desktopSessionRegistry ? ( @@ -181,7 +198,8 @@ export async function prepareGatewayKernelState(params: { : undefined; const workerDesktopObserveAvailable = Boolean(workerEnvironmentService) && gatewayPluginConfigAtStart.cloudWorkers?.desktop === true; - const desktopObserveAvailable = workerDesktopObserveAvailable || Boolean(hostDesktopService); + const desktopObserveAvailable = + workerDesktopObserveAvailable || nodeDesktopObserveAvailable || Boolean(hostDesktopService); const channelLogs = Object.fromEntries( listGatewayStartupChannelPlugins().map((plugin) => [plugin.id, logChannels.child(plugin.id)]), ) as Record>; @@ -440,6 +458,7 @@ export async function prepareGatewayKernelState(params: { (await watchNodeRequestHandler.current?.(req, res)) ?? false, workerIngressEnabled: Boolean(workerEnvironmentService), desktopSessionRegistry, + nodeDesktopStreamBroker, clients: connectionState.clients, }); const { @@ -473,6 +492,8 @@ export async function prepareGatewayKernelState(params: { workerDesktopObserveAvailable, desktopObserveAvailable, desktopSessionRegistry, + nodeDesktopObserveAvailable, + nodeDesktopStreamBroker, hostDesktopService, channelLogs, channelRuntimeEnvs, diff --git a/src/gateway/server-runtime-state.ts b/src/gateway/server-runtime-state.ts index 04bf0f55e367..4244f7ebc4d6 100644 --- a/src/gateway/server-runtime-state.ts +++ b/src/gateway/server-runtime-state.ts @@ -19,6 +19,7 @@ import type { PluginRegistry } from "../plugins/registry.js"; import type { AuthRateLimiter } from "./auth-rate-limit.js"; import type { ResolvedGatewayAuth } from "./auth.js"; import type { ControlUiRootState } from "./control-ui.js"; +import type { NodeDesktopStreamBroker } from "./desktop/node-stream-broker.js"; import type { DesktopSessionRegistry } from "./desktop/session-registry.js"; import type { HooksConfigResolved } from "./hooks.js"; import type { AuthorizedGatewayHttpRequest } from "./http-auth-utils.js"; @@ -120,6 +121,7 @@ export async function createGatewayHttpTransport(params: { handleWatchNodeRequest?: (req: IncomingMessage, res: ServerResponse) => Promise; workerIngressEnabled?: boolean; desktopSessionRegistry?: DesktopSessionRegistry; + nodeDesktopStreamBroker?: NodeDesktopStreamBroker; clients: Set; }): Promise<{ httpServer: HttpServer; @@ -307,6 +309,8 @@ export async function createGatewayHttpTransport(params: { workerIngressEnabled: params.workerIngressEnabled, log: params.log, desktopSessionRegistry: params.desktopSessionRegistry, + nodeDesktopStreamBroker: params.nodeDesktopStreamBroker, + getGatewayRequestContext: params.getGatewayRequestContext, }); gatewayHttpServers.push(httpServer); httpServers.push(httpServer); diff --git a/src/gateway/worker-environments/desktop-tunnel.ts b/src/gateway/worker-environments/desktop-tunnel.ts index 60940d4b9990..9229508e712c 100644 --- a/src/gateway/worker-environments/desktop-tunnel.ts +++ b/src/gateway/worker-environments/desktop-tunnel.ts @@ -6,7 +6,7 @@ import type { WorkerDesktopEndpoint, WorkerSshEndpoint, } from "../../plugins/types.js"; -import type { RfbAttachment } from "../desktop/attachment.js"; +import type { DesktopRfbAttachment } from "../desktop/attachment.js"; import { createDesktopSessionRegistry, DesktopSessionStaleOwnerError, @@ -45,7 +45,7 @@ type DesktopAcquireRequest = { resolveIdentity: WorkerSshIdentityResolver; }; -type DesktopAcquireResult = { attachment: RfbAttachment; vncPassword?: string }; +type DesktopAcquireResult = { attachment: DesktopRfbAttachment; vncPassword?: string }; type DesktopAppLaunchEntry = { environmentId: string; diff --git a/src/node-host/desktop-stream-command.test.ts b/src/node-host/desktop-stream-command.test.ts new file mode 100644 index 000000000000..916a346ab643 --- /dev/null +++ b/src/node-host/desktop-stream-command.test.ts @@ -0,0 +1,111 @@ +import http from "node:http"; +import net from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { WebSocketServer } from "ws"; +import { invokeNodeDesktopStream } from "./desktop-stream-command.js"; + +const TICKET = "a".repeat(48); +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); +}); + +describe("node desktop stream command", () => { + it("refuses a caller-selected RFB target before dialing", async () => { + await expect( + invokeNodeDesktopStream({ + paramsJSON: JSON.stringify({ + ticket: TICKET, + attachPath: `/node-desktop/attach?ticket=${TICKET}`, + target: { host: "192.0.2.10", port: 5900 }, + }), + gatewayUrl: "ws://127.0.0.1:1", + config: { enabled: true }, + signal: new AbortController().signal, + }), + ).rejects.toThrow("unsupported fields"); + }); + + it("refuses an attach path that changes the connected gateway origin", async () => { + await expect( + invokeNodeDesktopStream({ + paramsJSON: JSON.stringify({ + ticket: TICKET, + attachPath: `//attacker.example/node-desktop/attach?ticket=${TICKET}`, + }), + gatewayUrl: "ws://127.0.0.1:1", + config: { enabled: true }, + signal: new AbortController().signal, + }), + ).rejects.toThrow("ticket and attachPath required"); + }); + + it("tears down both splice sockets when the invoke is cancelled", async () => { + const rfbPeers = new Set(); + const rfbServer = net.createServer((socket) => { + rfbPeers.add(socket); + socket.once("close", () => rfbPeers.delete(socket)); + socket.write(Buffer.from("RFB 003.008\n", "ascii")); + socket.once("data", () => socket.write(Buffer.from([1, 2]))); + }); + await new Promise((resolve) => { + rfbServer.listen(0, "127.0.0.1", resolve); + }); + const rfbAddress = rfbServer.address(); + if (!rfbAddress || typeof rfbAddress === "string") { + throw new Error("expected RFB test address"); + } + cleanups.push( + async () => + await new Promise((resolve) => { + for (const peer of rfbPeers) { + peer.destroy(); + } + rfbServer.close(() => resolve()); + }), + ); + + const httpServer = http.createServer(); + const wss = new WebSocketServer({ server: httpServer }); + let streamClosed = false; + wss.on("connection", (ws) => { + ws.once("close", () => { + streamClosed = true; + }); + }); + await new Promise((resolve) => { + httpServer.listen(0, "127.0.0.1", resolve); + }); + const gatewayAddress = httpServer.address(); + if (!gatewayAddress || typeof gatewayAddress === "string") { + throw new Error("expected Gateway test address"); + } + cleanups.push( + async () => + await new Promise((resolve) => { + wss.close(() => httpServer.close(() => resolve())); + }), + ); + + const controller = new AbortController(); + const emitStatus = vi.fn(async () => undefined); + const running = invokeNodeDesktopStream({ + paramsJSON: JSON.stringify({ + ticket: TICKET, + attachPath: `/node-desktop/attach?ticket=${TICKET}`, + }), + gatewayUrl: `ws://127.0.0.1:${gatewayAddress.port}`, + config: { enabled: true, port: rfbAddress.port }, + signal: controller.signal, + emitStatus, + }); + await vi.waitFor(() => expect(emitStatus).toHaveBeenCalledWith("desktop stream attached\n")); + + controller.abort(); + + await expect(running).resolves.toBeUndefined(); + await vi.waitFor(() => expect(streamClosed).toBe(true)); + await vi.waitFor(() => expect(rfbPeers.size).toBe(0)); + }); +}); diff --git a/src/node-host/desktop-stream-command.ts b/src/node-host/desktop-stream-command.ts new file mode 100644 index 000000000000..d1ad161a51c7 --- /dev/null +++ b/src/node-host/desktop-stream-command.ts @@ -0,0 +1,361 @@ +import fs from "node:fs/promises"; +import net from "node:net"; +import type { TLSSocket } from "node:tls"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { WebSocket, type ClientOptions, type RawData } from "ws"; +import type { DesktopHostConfig } from "../config/types.desktop.js"; +import { classifyRfbSecurity, probeRfbServer } from "../gateway/desktop/rfb-probe.js"; +import { normalizeFingerprint } from "../infra/tls/fingerprint.js"; +import { registerSecretValueForRedaction } from "../logging/secret-redaction-registry.js"; +import { NODE_DESKTOP_ATTACH_PATH } from "../shared/node-desktop-stream.js"; + +const DEFAULT_DESKTOP_PORT = 5900; +const PROBE_TIMEOUT_MS = 1_500; +const MAX_PAYLOAD_BYTES = 1024 * 1024; +const PAUSE_BUFFERED_BYTES = 4 * 1024 * 1024; +const RESUME_CHECK_MS = 25; +const TICKET_PATTERN = /^[a-f0-9]{48}$/u; + +type NodeDesktopStreamCommandParams = { + ticket: string; + attachPath: string; +}; + +type NodeDesktopStreamTarget = { + host: string; + port: number; +}; + +function decodeDesktopStreamParams(raw?: string | null): NodeDesktopStreamCommandParams { + let value: unknown; + try { + value = raw ? JSON.parse(raw) : undefined; + } catch { + throw new Error("INVALID_REQUEST: desktop stream params malformed JSON"); + } + if (!isRecord(value)) { + throw new Error("INVALID_REQUEST: desktop stream params required"); + } + const ticket = typeof value.ticket === "string" ? value.ticket.trim() : ""; + const attachPath = typeof value.attachPath === "string" ? value.attachPath.trim() : ""; + if ( + !TICKET_PATTERN.test(ticket) || + attachPath !== `${NODE_DESKTOP_ATTACH_PATH}?ticket=${ticket}` + ) { + throw new Error("INVALID_REQUEST: desktop stream ticket and attachPath required"); + } + const attachUrl = new URL(attachPath, "http://127.0.0.1"); + if (attachUrl.searchParams.get("ticket") !== ticket) { + throw new Error("INVALID_REQUEST: desktop stream ticket does not match attachPath"); + } + if (Object.keys(value).some((key) => key !== "ticket" && key !== "attachPath")) { + throw new Error("INVALID_REQUEST: desktop stream params contain unsupported fields"); + } + return { ticket, attachPath }; +} + +function websocketDataBuffer(data: RawData): Buffer { + if (Buffer.isBuffer(data)) { + return data; + } + if (Array.isArray(data)) { + return Buffer.concat(data); + } + return Buffer.from(data); +} + +function attachWebSocketUrl(gatewayUrl: string, attachPath: string): string { + const gateway = new URL(gatewayUrl); + const url = new URL(attachPath, gateway); + if (url.protocol !== "ws:" && url.protocol !== "wss:") { + throw new Error("desktop stream gateway URL must use WebSocket transport"); + } + if (url.origin !== gateway.origin || url.pathname !== NODE_DESKTOP_ATTACH_PATH) { + throw new Error("desktop stream attachPath must stay on the connected gateway"); + } + return url.toString(); +} + +function assertTlsSocketFingerprint(socket: TLSSocket, expectedRaw: string): void { + const expected = normalizeFingerprint(expectedRaw); + const actual = normalizeFingerprint(socket.getPeerCertificate().fingerprint256 ?? ""); + if (!expected || !actual || actual !== expected) { + throw new Error("gateway TLS fingerprint mismatch"); + } +} + +function createPinnedRequestFinisher( + expected: string, +): NonNullable { + return (request) => { + request.once("socket", (socket) => { + const tlsSocket = socket as TLSSocket; + tlsSocket.once("secureConnect", () => { + try { + assertTlsSocketFingerprint(tlsSocket, expected); + request.end(); + } catch (error) { + request.destroy(error instanceof Error ? error : new Error(String(error))); + } + }); + }); + }; +} + +function websocketOptions(url: string, tlsFingerprint?: string): ClientOptions { + if (!url.startsWith("wss:") || !tlsFingerprint?.trim()) { + return { maxPayload: MAX_PAYLOAD_BYTES }; + } + return { + maxPayload: MAX_PAYLOAD_BYTES, + rejectUnauthorized: false, + finishRequest: createPinnedRequestFinisher(tlsFingerprint), + }; +} + +function assertGatewayTlsFingerprint(ws: WebSocket, expectedRaw?: string): void { + if (!expectedRaw?.trim()) { + return; + } + const expected = normalizeFingerprint(expectedRaw); + const socket = ( + ws as WebSocket & { + _socket?: { getPeerCertificate?: () => { fingerprint256?: string } }; + } + )["_socket"]; + const actual = normalizeFingerprint(socket?.getPeerCertificate?.().fingerprint256 ?? ""); + if (!expected || !actual || actual !== expected) { + throw new Error("gateway TLS fingerprint mismatch"); + } +} + +async function readVncPassword(passwordFile?: string): Promise { + if (!passwordFile) { + return undefined; + } + const password = (await fs.readFile(passwordFile, "utf8")).replace(/[\r\n]+$/u, ""); + if (!password) { + throw new Error("desktop.host.passwordFile is empty"); + } + registerSecretValueForRedaction(password); + return password; +} + +async function waitForSocketConnect(socket: net.Socket): Promise { + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); +} + +async function waitForWebSocketOpen(ws: WebSocket): Promise { + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); +} + +async function sendAttachMetadata( + ws: WebSocket, + metadata: { auth: "vnc-password" | "ard-account"; vncPassword?: string }, +): Promise { + const buffer = Buffer.from(JSON.stringify(metadata), "utf8"); + try { + await new Promise((resolve, reject) => { + ws.send(buffer, { binary: true }, (error) => (error ? reject(error) : resolve())); + }); + } finally { + buffer.fill(0); + } +} + +function createDesktopStreamSplice(params: { rfbSocket: net.Socket; ws: WebSocket }) { + let resumeTimer: ReturnType | undefined; + let settled = false; + let finish!: (error?: Error) => void; + const done = new Promise((resolve, reject) => { + finish = (error?: Error) => { + if (settled) { + return; + } + settled = true; + clearInterval(resumeTimer); + if (error) { + reject(error); + } else { + resolve(); + } + }; + params.ws.on("message", (data, isBinary) => { + if (!isBinary) { + finish(new Error("gateway sent non-binary desktop stream data")); + return; + } + if (!params.rfbSocket.write(websocketDataBuffer(data))) { + params.ws.pause(); + params.rfbSocket.once("drain", () => params.ws.resume()); + } + }); + params.rfbSocket.on("data", (chunk) => { + if (params.ws.readyState !== WebSocket.OPEN) { + return; + } + params.ws.send(chunk, { binary: true }, (error) => error && finish(error)); + if (params.ws.bufferedAmount <= PAUSE_BUFFERED_BYTES || resumeTimer) { + return; + } + params.rfbSocket.pause(); + resumeTimer = setInterval(() => { + if (params.ws.bufferedAmount <= PAUSE_BUFFERED_BYTES) { + clearInterval(resumeTimer); + resumeTimer = undefined; + params.rfbSocket.resume(); + } + }, RESUME_CHECK_MS); + resumeTimer.unref?.(); + }); + params.ws.once("close", () => finish()); + params.ws.once("error", (error) => finish(error)); + params.rfbSocket.once("close", () => finish()); + params.rfbSocket.once("error", (error) => finish(error)); + }); + void done.catch(() => undefined); + return { + done, + start() { + if (params.rfbSocket.destroyed || params.ws.readyState !== WebSocket.OPEN) { + finish(); + return; + } + params.rfbSocket.resume(); + params.ws.resume(); + }, + }; +} + +/** Splices a node-local loopback RFB socket to a ticket-authenticated Gateway WebSocket. */ +async function runNodeDesktopStreamCommand(params: { + command: NodeDesktopStreamCommandParams; + gatewayUrl: string; + gatewayTlsFingerprint?: string; + target: NodeDesktopStreamTarget; + passwordFile?: string; + signal: AbortSignal; + emitStatus?: (status: string) => Promise; +}): Promise { + if (params.target.host !== "127.0.0.1") { + throw new Error("desktop stream target must be loopback"); + } + if ( + !Number.isInteger(params.target.port) || + params.target.port < 1 || + params.target.port > 65535 + ) { + throw new Error("desktop stream target port is invalid"); + } + void params.emitStatus?.("probing local RFB server\n").catch(() => undefined); + const probe = await probeRfbServer({ + host: "127.0.0.1", + port: params.target.port, + timeoutMs: PROBE_TIMEOUT_MS, + }); + if (probe.kind !== "rfb") { + throw new Error( + probe.kind === "not-rfb" + ? "desktop stream target is not an RFB server" + : "desktop stream loopback RFB server is unavailable", + ); + } + const auth = classifyRfbSecurity(probe.securityTypes); + if (auth === "none") { + throw new Error("refusing unauthenticated loopback RFB server"); + } + if (auth === "unsupported") { + throw new Error("loopback RFB server security is unsupported"); + } + const vncPassword = + auth === "vnc-password" ? await readVncPassword(params.passwordFile) : undefined; + if (params.signal.aborted) { + return; + } + + const rfbSocket = net.createConnection(params.target.port, "127.0.0.1"); + // The RFB server can send its banner as soon as TCP connects. Pause until the + // attach metadata is accepted so no stateful handshake bytes are lost. + rfbSocket.pause(); + const wsUrl = attachWebSocketUrl(params.gatewayUrl, params.command.attachPath); + const ws = new WebSocket(wsUrl, websocketOptions(wsUrl, params.gatewayTlsFingerprint)); + let aborted: boolean = params.signal.aborted; + let resolveAbort!: () => void; + const abort = new Promise((resolve) => { + resolveAbort = resolve; + }); + const onAbort = () => { + aborted = true; + rfbSocket.destroy(); + ws.terminate(); + resolveAbort(); + }; + params.signal.addEventListener("abort", onAbort, { once: true }); + if (aborted) { + onAbort(); + } + try { + await Promise.race([ + Promise.all([waitForSocketConnect(rfbSocket), waitForWebSocketOpen(ws)]), + abort, + ]); + if (aborted) { + return; + } + assertGatewayTlsFingerprint(ws, params.gatewayTlsFingerprint); + ws.pause(); + const splice = createDesktopStreamSplice({ rfbSocket, ws }); + await sendAttachMetadata(ws, { auth, ...(vncPassword ? { vncPassword } : {}) }); + void params.emitStatus?.("desktop stream attached\n").catch(() => undefined); + splice.start(); + await splice.done; + } catch (error) { + if (!aborted) { + throw error; + } + } finally { + params.signal.removeEventListener("abort", onAbort); + rfbSocket.destroy(); + if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) { + ws.close(); + } + } +} + +/** Runs the built-in command against the node-local desktop configuration. */ +export async function invokeNodeDesktopStream(params: { + paramsJSON?: string | null; + gatewayUrl?: string; + gatewayTlsFingerprint?: string; + config?: DesktopHostConfig; + signal?: AbortSignal; + emitStatus?: (status: string) => Promise; +}): Promise { + if (!params.gatewayUrl || !params.signal) { + throw new Error("desktop stream gateway connection is unavailable"); + } + if (params.config?.enabled !== true) { + throw new Error("desktop host streaming is disabled on this node"); + } + const command = decodeDesktopStreamParams(params.paramsJSON); + await runNodeDesktopStreamCommand({ + command, + gatewayUrl: params.gatewayUrl, + ...(params.gatewayTlsFingerprint + ? { gatewayTlsFingerprint: params.gatewayTlsFingerprint } + : {}), + target: { + host: "127.0.0.1", + port: params.config.port ?? DEFAULT_DESKTOP_PORT, + }, + ...(params.config.passwordFile ? { passwordFile: params.config.passwordFile } : {}), + signal: params.signal, + ...(params.emitStatus ? { emitStatus: params.emitStatus } : {}), + }); +} diff --git a/src/node-host/gateway-candidate-connection.ts b/src/node-host/gateway-candidate-connection.ts index df34108a29e9..7485261a2f7d 100644 --- a/src/node-host/gateway-candidate-connection.ts +++ b/src/node-host/gateway-candidate-connection.ts @@ -25,7 +25,7 @@ type GatewayCandidateConnectionParams = { candidates: readonly NodeHostGatewayConfig[]; clientOptions: CandidateConnectionOptions; onEvent: (event: GatewayCandidateEvent) => void; - onHelloOk: (hello: GatewayCandidateHello, url: string) => void; + onHelloOk: (hello: GatewayCandidateHello, url: string, tlsFingerprint?: string) => void; onConnectError: (error: Error) => void; onReconnectPaused: (info: GatewayReconnectPausedInfo) => void; onClose: (code: number, reason: string, info?: GatewayClientCloseInfo) => void; @@ -84,7 +84,7 @@ export function createNodeHostGatewayCandidateConnection(params: GatewayCandidat winnerSelected = true; params.onWinningCandidate(candidate); } - params.onHelloOk(hello, url); + params.onHelloOk(hello, url, candidate.tlsFingerprint); }, onConnectError: (error) => { if (currentCandidateIndex === candidateIndex) { diff --git a/src/node-host/invoke-agent-cli-claude-handler.ts b/src/node-host/invoke-agent-cli-claude-handler.ts index e51a71ad7301..d8368c2a92c7 100644 --- a/src/node-host/invoke-agent-cli-claude-handler.ts +++ b/src/node-host/invoke-agent-cli-claude-handler.ts @@ -1,3 +1,4 @@ +import type { DesktopHostConfig } from "../config/types.desktop.js"; import { createExecApprovalPolicySnapshot } from "../infra/exec-approvals.js"; import type { scanInstalledApps } from "../infra/installed-apps.js"; import type { OpenClawPluginNodeHostCommandIo } from "../plugins/types.js"; @@ -25,6 +26,10 @@ export type NodeHostInvokeRuntime = { installedAppsSharingEnabled?: boolean; installedAppsPlatform?: NodeJS.Platform; scanInstalledApps?: typeof scanInstalledApps; + gatewayUrl?: string; + gatewayTlsFingerprint?: string; + desktopHostConfig?: DesktopHostConfig; + emitProgress?: (text: string) => Promise; }; type ClaudeCliNodeInvokeDeps = Pick< diff --git a/src/node-host/invoke.ts b/src/node-host/invoke.ts index 7b897dcd0624..4e126edfe8d6 100644 --- a/src/node-host/invoke.ts +++ b/src/node-host/invoke.ts @@ -43,8 +43,10 @@ import { } from "../infra/node-commands.js"; import { logWarn } from "../logger.js"; import { runCommandWithTimeout } from "../process/exec.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import { truncateUtf8Prefix } from "../utils/utf8-truncate.js"; import type { NodeHostClient } from "./client.js"; +import { invokeNodeDesktopStream } from "./desktop-stream-command.js"; import { handleClaudeCliNodeInvoke, type NodeHostInvokeRuntime, @@ -617,6 +619,27 @@ async function dispatchInvoke( } return; } + if (command === NODE_DESKTOP_STREAM_COMMAND) { + try { + await invokeNodeDesktopStream({ + paramsJSON: frame.paramsJSON, + gatewayUrl: runtime.gatewayUrl, + gatewayTlsFingerprint: runtime.gatewayTlsFingerprint, + config: runtime.desktopHostConfig, + signal: runtime.signal, + emitStatus: runtime.emitProgress, + }); + await sendJsonPayloadResult(client, frame, { status: "closed" }); + } catch (error) { + await sendErrorResult( + client, + frame, + "UNAVAILABLE", + error instanceof Error ? error.message : "desktop stream unavailable", + ); + } + return; + } if (command === "system.execApprovals.get") { try { const snapshot = await ensureExecApprovalsSnapshot(); diff --git a/src/node-host/runner.test.ts b/src/node-host/runner.test.ts index 9f43597206fe..7ecb43b28b8e 100644 --- a/src/node-host/runner.test.ts +++ b/src/node-host/runner.test.ts @@ -61,6 +61,7 @@ const mocks = vi.hoisted(() => ({ handleInput: vi.fn(), cancel: vi.fn(), cancelAll: vi.fn(), + updateGatewayConnection: vi.fn(), close: vi.fn(async () => {}), }, })); diff --git a/src/node-host/runner.ts b/src/node-host/runner.ts index e538f2041134..66dc31ac1909 100644 --- a/src/node-host/runner.ts +++ b/src/node-host/runner.ts @@ -504,8 +504,9 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { void activeRuntime.invoke(payload); } }, - onHelloOk: (hello, url) => { + onHelloOk: (hello, url, tlsFingerprint) => { writeStderrLine(`node host gateway connected: ${url}`); + activeRuntime.updateGatewayConnection({ url, ...(tlsFingerprint ? { tlsFingerprint } : {}) }); gatewayConnectionGeneration += 1; gatewayHelloReceived = true; connectedGatewayProtocol = hello.protocol; @@ -533,6 +534,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { }, onClose: (code, reason) => { retireGatewayConnection(); + activeRuntime.updateGatewayConnection(); activeRuntime.cancelAll(); writeStderrLine(`node host gateway closed (${code}): ${reason}`); }, diff --git a/src/node-host/runtime.test.ts b/src/node-host/runtime.test.ts index 301ab2501d68..bf3b1e82dce1 100644 --- a/src/node-host/runtime.test.ts +++ b/src/node-host/runtime.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { NODE_DEVICE_APPS_COMMAND } from "../infra/node-commands.js"; import type { OpenClawPluginNodeHostCommandIo } from "../plugins/types.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import type { NodeHostClient } from "./client.js"; import { listRegisteredNodeHostCapsAndCommands } from "./plugin-node-host.js"; import { prepareNodeHostRuntime } from "./runtime.js"; @@ -8,6 +9,8 @@ import { prepareNodeHostRuntime } from "./runtime.js"; const mocks = vi.hoisted(() => ({ closeMcp: vi.fn(async () => undefined), handleInvoke: vi.fn(async () => undefined), + progressStartHeartbeats: vi.fn(), + progressWrite: vi.fn(async () => undefined), })); vi.mock("../infra/path-env.js", () => ({ @@ -29,8 +32,8 @@ vi.mock("./mcp.js", () => ({ vi.mock("./node-invoke-progress.js", () => ({ createNodeInvokeProgressWriter: vi.fn(() => ({ - startHeartbeats: vi.fn(), - write: vi.fn(async () => undefined), + startHeartbeats: mocks.progressStartHeartbeats, + write: mocks.progressWrite, stop: vi.fn(), flush: vi.fn(async () => undefined), })), @@ -180,6 +183,40 @@ describe("node-host invocation cancellation", () => { }); }); +describe("node-host desktop manifest", () => { + it("advertises desktop.stream only when the node-local desktop is enabled", async () => { + const disabled = await prepareNodeHostRuntime({ + config: {}, + env: { PATH: "/usr/bin" }, + platform: "linux", + }); + expect(disabled.manifest.commands).not.toContain(NODE_DESKTOP_STREAM_COMMAND); + + const enabled = await prepareNodeHostRuntime({ + config: { desktop: { host: { enabled: true } } }, + env: { PATH: "/usr/bin" }, + platform: "linux", + }); + expect(enabled.manifest.commands).toContain(NODE_DESKTOP_STREAM_COMMAND); + }); + + it("emits desktop statuses without control-channel heartbeats", async () => { + const runtime = await startRuntime(); + await runtime.invoke({ ...frame, command: NODE_DESKTOP_STREAM_COMMAND }); + + expect(mocks.progressStartHeartbeats).not.toHaveBeenCalled(); + const lastCall = mocks.handleInvoke.mock.calls.at(-1) as unknown[] | undefined; + const invokeRuntime = lastCall?.[4] as + | { + emitProgress?: (text: string) => Promise; + } + | undefined; + await invokeRuntime?.emitProgress?.("attached\n"); + expect(mocks.progressWrite).toHaveBeenCalledWith("attached\n"); + await runtime.close(); + }); +}); + describe("node-host invoke input dispatch", () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/src/node-host/runtime.ts b/src/node-host/runtime.ts index 8f3af5314847..e36e86967a28 100644 --- a/src/node-host/runtime.ts +++ b/src/node-host/runtime.ts @@ -20,6 +20,7 @@ import { logDebug } from "../logger.js"; import type { OpenClawPluginNodeHostCommandIo } from "../plugins/types.js"; import type { OpenClawPluginNodeHostCommandContext } from "../plugins/types.node-host.js"; import { BoundedBuffer } from "../shared/bounded-buffer.js"; +import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import type { NodeHostClient } from "./client.js"; import { handleInvoke, type NodeInvokeRequestPayload, type SkillBinsProvider } from "./invoke.js"; import { startNodeHostMcpManager, type NodeHostMcpManager } from "./mcp.js"; @@ -61,6 +62,7 @@ type ActiveNodeHostRuntime = { handleInput(invokeId: string, seq: number, payloadJSON: string): void; cancel(invokeId: string): void; cancelAll(): void; + updateGatewayConnection(connection?: { url: string; tlsFingerprint?: string }): void; close(): Promise; }; @@ -252,6 +254,9 @@ export async function prepareNodeHostRuntime(params?: { const platform = params?.platform ?? process.platform; const installedAppsSharingEnabled = platform === "darwin" && params?.installedAppsSharingEnabled === true; + const desktopStreamingEnabled = + (platform === "darwin" || platform === "linux" || platform === "win32") && + config.desktop?.host?.enabled === true; const availabilityContext = { config, env }; const resolvePluginNodeHost = () => listRegisteredNodeHostCapsAndCommands(availabilityContext, { @@ -281,6 +286,7 @@ export async function prepareNodeHostRuntime(params?: { NODE_FS_LIST_DIR_COMMAND, NODE_TERMINAL_UPLOAD_COMMAND, NODE_MCP_TOOLS_CALL_COMMAND, + ...(desktopStreamingEnabled ? [NODE_DESKTOP_STREAM_COMMAND] : []), ...(installedAppsSharingEnabled ? [NODE_DEVICE_APPS_COMMAND] : []), ...(claudePath ? [NODE_AGENT_CLI_CLAUDE_RUN_COMMAND] : []), ...pluginManifest.commands, @@ -307,6 +313,7 @@ export async function prepareNodeHostRuntime(params?: { }; let currentPluginNodeHost = pluginNodeHost; let currentManifest = manifest; + let gatewayConnection: { url: string; tlsFingerprint?: string } | undefined; let manager: NodeHostMcpManager | undefined; const startup = startNodeHostMcpManager(config.nodeHost?.mcp?.servers, { signal: mcpAbort.signal, @@ -348,6 +355,7 @@ export async function prepareNodeHostRuntime(params?: { return { async invoke(frame) { const duplexCommand = duplexEnabled && isRegisteredNodeHostCommandDuplex(frame.command); + const progressEnabled = duplexCommand || frame.command === NODE_DESKTOP_STREAM_COMMAND; const controller = new AbortController(); // Every command must remain cancellable after dispatch; only duplex // commands own ordered input and its pre-spawn buffer. @@ -373,7 +381,7 @@ export async function prepareNodeHostRuntime(params?: { // let its cleanup unregister the replacement invocation. activeInvokes.get(frame.id)?.controller.abort(); activeInvokes.set(frame.id, active); - const progress = duplexCommand + const progress = progressEnabled ? createNodeInvokeProgressWriter({ client, frame, @@ -381,7 +389,9 @@ export async function prepareNodeHostRuntime(params?: { onError: () => controller.abort(), }) : undefined; - progress?.startHeartbeats(); + if (duplexCommand) { + progress?.startHeartbeats(); + } const pluginCommandIo: OpenClawPluginNodeHostCommandIo | undefined = input && progress ? { @@ -399,6 +409,12 @@ export async function prepareNodeHostRuntime(params?: { ...(claudePath ? { claudePath } : {}), signal: controller.signal, ...(pluginCommandIo ? { pluginCommandIo } : {}), + ...(gatewayConnection?.url ? { gatewayUrl: gatewayConnection.url } : {}), + ...(gatewayConnection?.tlsFingerprint + ? { gatewayTlsFingerprint: gatewayConnection.tlsFingerprint } + : {}), + ...(config.desktop?.host ? { desktopHostConfig: config.desktop.host } : {}), + ...(progress ? { emitProgress: (text) => progress.write(text) } : {}), installedAppsSharingEnabled, installedAppsPlatform: platform, pluginCommandContext, @@ -426,6 +442,9 @@ export async function prepareNodeHostRuntime(params?: { } activeInvokes.clear(); }, + updateGatewayConnection(connection) { + gatewayConnection = connection; + }, async close() { this.cancelAll(); stopAvailabilityWatch(); diff --git a/src/shared/node-desktop-stream.ts b/src/shared/node-desktop-stream.ts new file mode 100644 index 000000000000..328d66867cde --- /dev/null +++ b/src/shared/node-desktop-stream.ts @@ -0,0 +1,2 @@ +export const NODE_DESKTOP_STREAM_COMMAND = "desktop.stream"; +export const NODE_DESKTOP_ATTACH_PATH = "/node-desktop/attach"; diff --git a/ui/src/components/desktop/desktop-panel.ts b/ui/src/components/desktop/desktop-panel.ts index 7d8e738bf703..4af4619d0468 100644 --- a/ui/src/components/desktop/desktop-panel.ts +++ b/ui/src/components/desktop/desktop-panel.ts @@ -50,9 +50,13 @@ type PendingDesktopConnection = { type ObservedDesktopConnection = PendingDesktopConnection & { observed: DesktopObserveResult }; function desktopSourceForEnvironment(environment: Pick): DesktopSource { - return environment.id === "gateway" - ? { kind: "host" } - : { kind: "environment", environmentId: environment.id }; + if (environment.id === "gateway") { + return { kind: "host" }; + } + if (environment.id.startsWith("node:") && environment.id.length > "node:".length) { + return { kind: "node", nodeId: environment.id.slice("node:".length) }; + } + return { kind: "environment", environmentId: environment.id }; } /** `` — dockable RFB access to Gateway desktop sources. */ @@ -248,7 +252,7 @@ class OpenClawDesktopPanel extends OpenClawLitElement { this.controlTakeoverRecoveryUsed = options.takeoverRecovery === true; try { const observeCredentials = - source.kind === "host" && + source.kind !== "environment" && this.credentials?.password && (this.credentialAuth === "vnc-password" || (this.credentialAuth === "ard-account" && this.credentials.username)) @@ -262,12 +266,18 @@ class OpenClawDesktopPanel extends OpenClawLitElement { if (operationId !== this.operationId) { return; } - const credentials = observed.vncPassword - ? { password: observed.vncPassword } - : observed.auth === "vnc-password" - ? this.credentials - : undefined; - if (observed.auth === "vnc-password" && !credentials?.password) { + const credentials = observed.preauthenticated + ? undefined + : observed.vncPassword + ? { password: observed.vncPassword } + : observed.auth === "vnc-password" + ? this.credentials + : undefined; + if ( + observed.auth === "vnc-password" && + observed.preauthenticated !== true && + !credentials?.password + ) { this.credentialAuth = "vnc-password"; this.pendingConnection = { environmentId, control, observed, operationId }; this.state = "credentials"; diff --git a/ui/src/e2e/desktop-panel.e2e.test.ts b/ui/src/e2e/desktop-panel.e2e.test.ts index 794664217408..b80047503ae6 100644 --- a/ui/src/e2e/desktop-panel.e2e.test.ts +++ b/ui/src/e2e/desktop-panel.e2e.test.ts @@ -270,6 +270,73 @@ suite.define(() => { }); }); + it("lists an observable node and connects with the node source arm", async () => { + await suite.withPage({ serviceWorkers: "block" }, async ({ page }) => { + const gateway = await installMockGateway(page, { + featureMethods: ["desktop.observe", "environments.list"], + methodResponses: { + "sessions.list": sessionsList("local"), + "environments.list": { + environments: [ + { + id: "node:paired-node", + type: "node", + status: "available", + desktop: true, + capabilities: ["desktop.stream"], + }, + { + id: "node:plain-node", + type: "node", + status: "available", + capabilities: ["screen.snapshot"], + }, + ], + }, + "desktop.observe": { + sequence: [ + { + __mockError: { + code: "INVALID_REQUEST", + message: "VNC password is required to observe this node", + details: { + code: "DESKTOP_CREDENTIALS_REQUIRED", + auth: "vnc-password", + }, + }, + }, + { + transport: "rfb", + wsPath: "/desktop/observe?token=node", + expiresAtMs: 60_000, + control: false, + auth: "vnc-password", + preauthenticated: true, + }, + ], + }, + }, + }); + const panel = await openDesktopPanel(page); + await gateway.waitForRequest("environments.list"); + await panel.getByText("node:paired-node", { exact: true }).waitFor(); + expect(await panel.getByText("node:plain-node", { exact: true }).count()).toBe(0); + await installDesktopClientFake(panel); + + await panel.getByRole("button", { name: "Connect", exact: true }).click(); + await panel.getByLabel("VNC password", { exact: true }).fill("node-password"); + await panel.getByRole("button", { name: "Connect", exact: true }).click(); + await expect.poll(async () => await panel.getAttribute("data-connect-count")).toBe("1"); + expect(await panel.getAttribute("data-used-credentials")).toBe("false"); + const observeRequests = await gateway.getRequests("desktop.observe"); + expect(observeRequests.at(-1)?.params).toEqual({ + source: { kind: "node", nodeId: "paired-node" }, + control: false, + credentials: { password: "node-password" }, + }); + }); + }); + it("launches advertised desktop apps and keeps observe controls working", async () => { await suite.withPage({ serviceWorkers: "block" }, async ({ page }) => { const gateway = await installMockGateway(page, {