diff --git a/docs/connected-hosts.md b/docs/connected-hosts.md index 28383be16b..ded6313a2d 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -17,10 +17,12 @@ access can be changed by users with `tunnel.manage`; connection and metadata cha producer or `access.manage`. Removing a producer-managed entry allows that producer to add it again. Disabling retains the opt-out. -Each routed API owns a separate session. A derives its identity from the admitted origin session, -producer, and logical SDK scope. A replaces forwarded identity and grants, and B intersects those -grants with the authenticated link's authority. Reconnect reads current authority. Changing caller -permissions closes existing routes. A routed API cannot select a second hop. +Each enabled registration owns one physical daemon link, shared by status monitoring and every +routed API scope. Each scope owns an isolated virtual session on B. A derives its identity from the +admitted origin session, producer, and logical SDK scope. B admits virtual sessions under the actual +physical link principal, then intersects forwarded caller grants with that link's current authority. +Changing or resuming a parent's grants closes its existing channels; reconnect re-admits demand with +current authority. A routed API cannot select a second hop, and saving a host grants no reverse access. A producer that registers a host through a routed API is qualified on B as `route:::`, with each component percent-encoded. This namespace belongs to the authenticated link @@ -31,12 +33,13 @@ Trusted server plugins retain the authority of their runtime; they are not sandb ## Connections and recovery The shared Node connector owns the target grammar and each connection attempt, including a fresh -SSH process. SSH reaches an already running daemon. The daemon keeps unreachable registrations and -retries them. Identity is pinned when discovered. Self and duplicate registrations remain inspectable; -a changed destination requires explicit identity replacement. +SSH process. SSH reaches an already running daemon. The shared client connection lifecycle owns +physical reconnect and keepalive; the registry retains configuration, identity acceptance, and recovery +status. Unreachable registrations remain saved. Identity is pinned when discovered. Self and duplicate +registrations remain inspectable; a changed destination requires explicit identity replacement. -Both daemons must advertise `connectedHosts`. A checks B's capability and identity before forwarding -session operations. There is no local fallback. An older daemon continues to serve existing local +Both daemons advertise `connectedHosts`; the shared link additionally requires `hostMultiplexing`. +A completes capability and pinned-identity checks on each physical recovery before admitting channels. There is no local fallback. An older daemon continues to serve existing local operations; requesting this feature gives update guidance. Disposing a routed API releases its observations and session. It does not remove the saved host or @@ -45,10 +48,17 @@ through the usual owned-subscription contract; requests made while offline fail. returns `outcome_unknown` with the destination ID. Inspect the destination before retrying a mutation. No mutation is automatically replayed. -Host links cap saved registrations, simultaneous routes, frames, per-socket output, and shared queued -bytes. Slow consumers lose their route with a resource-limit error instead of retaining unbounded -output. Routed sessions release server resources on disconnect; subscription demand is rebuilt by -its owning client on reconnect. +Text session envelopes carry a channel identifier. Binary routing adds an opcode, a bounded UTF-8 +channel identifier, and the original terminal or file-transfer bytes. It never converts those bytes to +base64 JSON. The existing direct binary opcodes and 100 MiB message limit remain unchanged; routed +payloads reserve space for the routing header inside that limit. Relay encryption retains its own +existing frame limit. + +Host links cap saved registrations, simultaneous channels, handshake lifetime, frames, physical +output, and aggregate queued bytes. Capacity checks use the actual escaped outer JSON or binary +frame size. Awaitable output completes when the physical send completes. An overflowing channel +closes without closing siblings or the saved link. Routed sessions release resources on disconnect; +subscription demand is rebuilt by its owning client on reconnect. See the [SDK reference](../public-docs/sdk/reference.md#clienthosts), [CLI reference](../public-docs/cli.md#connected-hosts), and diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index ae88f1aace..5434622ef1 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1,3 +1,9 @@ +import { + HOST_ROUTE_OPCODE, + encodeHostRouteFrame, + decodeHostRouteFrame, + type HostRouteFrame, +} from "@getpaseo/protocol/binary-frames/index"; import { messageTooLargeError } from "./daemon-client-transport-utils.js"; import type { HostAdd, HostUpdate, HostDirectory } from "@getpaseo/protocol/hosts"; import { legacyUsageIcon } from "./legacy-usage-icons.js"; @@ -334,7 +340,7 @@ export type ConnectionState = | { status: "idle" } | { status: "connecting"; attempt: number } | { status: "connected" } - | { status: "disconnected"; reason?: string } + | { status: "disconnected"; reason?: string; nextRetryAt?: string; failureCode?: string } | { status: "disposed" }; export type DaemonEvent = @@ -1753,15 +1759,43 @@ export class DaemonClient { }); } } - sendHostFrame(routeId: string, data: string, binary: boolean): void { - this.sendSessionMessageStrict({ type: "hosts.route.frame.request", routeId, data, binary }); + async openHostLink(channelId: string): Promise { + this.requireConnectedHosts(); + if (this.getLastServerInfoMessage()?.features?.hostMultiplexing !== true) + throw Object.assign(new Error("Update the destination daemon"), { + code: "unsupported_capability", + }); + return ( + await this.sendCorrelatedSessionRequest({ + message: { type: "hosts.link.open.request", channelId }, + responseType: "hosts.link.open.response", + }) + ).routeId; + } + sendHostFrame(routeId: string, data: string | Uint8Array): void { + if (typeof data === "string") { + this.sendSessionMessageStrict({ + type: "hosts.route.frame.request", + routeId, + data, + binary: false, + }); + } else { + if (!this.isConnected || !this.transport) throw this.unavailableError(); + this.sendTransportFrame(encodeHostRouteFrame(routeId, data)); + } + } + private readonly hostFrameListeners = new Set<(frame: HostRouteFrame) => void>(); + subscribeHostFrames(handler: (frame: HostRouteFrame) => void): () => void { + this.hostFrameListeners.add(handler); + return () => this.hostFrameListeners.delete(handler); } cancelHostRoute(scopeId: string): void { - if (this.isConnected) + if (this.isConnected && this.transport) this.sendSessionMessageStrict({ type: "hosts.route.close.request", scopeId }); } closeHostRoute(routeId: string): void { - if (this.isConnected) + if (this.isConnected && this.transport) this.sendSessionMessageStrict({ type: "hosts.route.close.request", routeId }); } @@ -6669,6 +6703,16 @@ export class DaemonClient { } private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean { + const hostFrame = decodeHostRouteFrame(rawBytes); + if (rawBytes[0] === HOST_ROUTE_OPCODE && !hostFrame) { + this.disposeTransport(1008, "Invalid host routing frame"); + this.scheduleReconnect({ reason: "Invalid host routing frame", event: "INVALID_HOST_FRAME" }); + return true; + } + if (hostFrame) { + for (const handler of this.hostFrameListeners) handler(hostFrame); + return true; + } const fileFrame = decodeFileTransferFrame(rawBytes); if (fileFrame) { this.traceInstant("paseo.ws.message.inbound", { @@ -6903,11 +6947,26 @@ export class DaemonClient { private emitDisconnectedStateForReconnect( reason: string | undefined, - input: { reason?: string; event?: string; reasonCode?: string } | undefined, + input: { reason?: string; event?: string; reasonCode?: string; failure?: Error } | undefined, ): void { this.updateConnectionState( { status: "disconnected", + ...(input?.failure && "code" in input.failure && typeof input.failure.code === "string" + ? { failureCode: input.failure.code } + : {}), + ...(this.shouldReconnect && this.config.reconnect?.enabled !== false + ? { + nextRetryAt: new Date( + Date.now() + + Math.min( + (this.config.reconnect?.baseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS) * + 2 ** this.reconnectAttempt, + this.config.reconnect?.maxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS, + ), + ).toISOString(), + } + : {}), ...(reason ? { reason } : {}), }, { diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index cb2419e156..45b057fb07 100644 --- a/packages/client/src/hosts/index.ts +++ b/packages/client/src/hosts/index.ts @@ -14,23 +14,13 @@ export interface PaseoHostActions { client(serverId: string): Promise; } -function encode(data: Uint8Array | ArrayBuffer): string { - const bytes = data instanceof Uint8Array ? data : new Uint8Array(data); - let result = ""; - for (let start = 0; start < bytes.length; start += 8192) - result += String.fromCharCode(...bytes.subarray(start, start + 8192)); - return globalThis.btoa(result); -} -function decode(data: string): Uint8Array { - return Uint8Array.from(globalThis.atob(data), (char) => char.charCodeAt(0)); -} - -function routeTransport( +export function createHostChannelTransport( origin: DaemonClient, serverId: string, scopeId: string, - initialRouteId: string, + initialRouteId: string | null, state: { failure: Error | null }, + link = false, ): DaemonTransportFactory { let first: string | null = initialRouteId; return () => { @@ -46,18 +36,19 @@ function routeTransport( closed = true; if (error instanceof Error) state.failure = Object.assign(error, { serverId }); for (const cleanup of cleanups) cleanup(); - if (routeId) origin.closeHostRoute(routeId); + try { + if (routeId) origin.closeHostRoute(routeId); + else origin.cancelHostRoute(scopeId); + } catch { + /* Transport already closed. */ + } if (error) for (const handler of errors) handler(error); for (const handler of closes) handler(error); }; cleanups.push( origin.subscribeRawMessages((message) => { if (message.type === "hosts.route.frame" && message.payload.subscriptionId === routeId) { - for (const handler of messages) - handler( - message.payload.binary ? decode(message.payload.data) : message.payload.data, - message.payload.binary, - ); + for (const handler of messages) handler(message.payload.data, false); } if (message.type === "hosts.route.closed" && message.payload.subscriptionId === routeId) close( @@ -68,12 +59,19 @@ function routeTransport( ); }), ); + cleanups.push( + origin.subscribeHostFrames((frame) => { + if (frame.routeId === routeId) for (const handler of messages) handler(frame.payload, true); + }), + ); cleanups.push( origin.subscribeConnectionStatus((connectionState) => { if (connectionState.status !== "connected") close(); }), ); - const opening = first ? Promise.resolve(first) : origin.openHostRoute(serverId, scopeId); + let opening: Promise; + if (first) opening = Promise.resolve(first); + else opening = link ? origin.openHostLink(scopeId) : origin.openHostRoute(serverId, scopeId); first = null; void opening .then((id) => { @@ -96,8 +94,7 @@ function routeTransport( }); origin.sendHostFrame( routeId, - typeof data === "string" ? data : encode(data), - typeof data !== "string", + typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data), ); }, close: () => close(), @@ -175,7 +172,7 @@ export function createHostActions( const client = new DaemonClient({ url: "ws://routed-host/ws", clientId: scopeId, - transportFactory: routeTransport(origin, serverId, scopeId, routeId, state), + transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId, state), routedClient: true, routedServerId: serverId, routedFailure: () => state.failure, diff --git a/packages/client/src/node/buffering.ts b/packages/client/src/node/buffering.ts index 1a06a973a0..06bed49bf0 100644 --- a/packages/client/src/node/buffering.ts +++ b/packages/client/src/node/buffering.ts @@ -1,7 +1,7 @@ /** Counts bytes until the socket's send callback, shared by all links owned by a daemon. */ export class HostTransportBudget { private used = 0; - constructor(readonly limit = 64 * 1024 * 1024) {} + constructor(readonly limit = 256 * 1024 * 1024) {} get bufferedBytes(): number { return this.used; } diff --git a/packages/client/src/node/connector.test.ts b/packages/client/src/node/connector.test.ts index 5a0ffb68cf..3de65c2dcb 100644 --- a/packages/client/src/node/connector.test.ts +++ b/packages/client/src/node/connector.test.ts @@ -51,3 +51,24 @@ test("a slow consumer cannot exceed the shared byte budget and close releases re transport.close(); await expect.poll(() => budget.bufferedBytes).toBe(0); }); +test("direct connector preserves 25 MiB binary frames in both directions", async () => { + const server = await endpoint(); + const connected = once(server.wss, "connection"); + const transport = createHostTransportFactory(server.url)({ url: server.url }); + cleanup.push(async () => transport.close()); + const ready = new Promise((resolve) => transport.onOpen(resolve)); + const received = new Promise((resolve, reject) => { + transport.onMessage((bytes, binary) => binary && resolve(bytes as Uint8Array)); + transport.onError(reject); + }); + const [socket] = (await connected) as [WebSocket]; + await ready; + const bytes = new Uint8Array(25 * 1024 * 1024).fill(173); + const echoed = once(socket, "message"); + transport.send(bytes); + const [sent, binary] = await echoed; + expect(binary).toBe(true); + expect(Buffer.from(sent as Uint8Array).equals(bytes)).toBe(true); + socket.send(bytes); + expect(Buffer.from(await received).equals(bytes)).toBe(true); +}); diff --git a/packages/client/src/node/index.ts b/packages/client/src/node/index.ts index 41372b738b..e015f53cd4 100644 --- a/packages/client/src/node/index.ts +++ b/packages/client/src/node/index.ts @@ -1,3 +1,4 @@ +export { createHostChannelTransport } from "../hosts/index.js"; import { HostTransportBudget } from "./buffering.js"; export { HostTransportBudget } from "./buffering.js"; import { homedir } from "node:os"; @@ -32,6 +33,7 @@ export function createHostTransportFactory( let closeEvent: unknown; let transport: DaemonTransport | undefined; let teardown = () => {}; + let disposeSocket = () => {}; let sshTunnel: SshTunnel | null = null; const close = (code?: number, reason?: string) => { if (closed) return; @@ -40,6 +42,7 @@ export function createHostTransportFactory( transport?.close(code, reason); } finally { try { + disposeSocket(); teardown(); } finally { for (const handler of closes) handler(closeEvent); @@ -66,9 +69,10 @@ export function createHostTransportFactory( { headers: options?.headers, ...(socketPath ? { socketPath } : {}), - maxPayload: 24 * 1024 * 1024, + maxPayload: 100 * 1024 * 1024, }, ); + disposeSocket = () => socket.terminate(); const pending = new Set<() => void>(); const send = socket.send.bind(socket); socket.send = (data, ...args: unknown[]) => { @@ -76,19 +80,12 @@ export function createHostTransportFactory( typeof data === "string" ? Buffer.byteLength(data) : Buffer.byteLength(data as Uint8Array); - if (socket.bufferedAmount + bytes > 24 * 1024 * 1024) { - socket.terminate(); + if (socket.bufferedAmount + bytes > 128 * 1024 * 1024) { throw Object.assign(new Error("Host socket buffer limit reached"), { code: "resource_limit", }); } - let release: () => void; - try { - release = budget.reserve(bytes); - } catch (error) { - socket.terminate(); - throw error; - } + const release = budget.reserve(bytes); pending.add(release); const callback = args.find( (value): value is (error?: Error) => void => typeof value === "function", @@ -129,6 +126,7 @@ export function createHostTransportFactory( close(); }); transport.onError((error) => { + if (closed) return; const detail = sshTunnel?.failureDetail() ?? ""; const failure = /permission denied|authentication failed|host key verification failed|REMOTE HOST IDENTIFICATION HAS CHANGED/i.test( @@ -179,12 +177,13 @@ export function createHostTransportFactory( export function createHostClient( connection: string, config: Omit, + budget?: HostTransportBudget, ): DaemonClient { const target = parseHostTarget(connection); return new DaemonClient({ ...config, url: "ws://host-connector/ws", ...(target.password ? { password: target.password } : {}), - transportFactory: createHostTransportFactory(connection), + transportFactory: createHostTransportFactory(connection, budget), }); } diff --git a/packages/protocol/src/binary-frames/demux.test.ts b/packages/protocol/src/binary-frames/demux.test.ts index 718a7c6763..c56e417528 100644 --- a/packages/protocol/src/binary-frames/demux.test.ts +++ b/packages/protocol/src/binary-frames/demux.test.ts @@ -2,6 +2,9 @@ import { describe, expect, it } from "vitest"; import { decodeBinaryFrame, + encodeHostRouteFrame, + decodeHostRouteFrame, + MAX_HOST_ROUTE_ID_BYTES, encodeFileTransferFrame, encodeTerminalStreamFrame, FileTransferOpcode, @@ -51,3 +54,38 @@ describe("binary frame demux", () => { expect(decodeBinaryFrame(new Uint8Array([0xff, 0]))).toBeNull(); }); }); + +describe("host route binary framing", () => { + it("wraps terminal and file frames unchanged and leaves their direct interpretation intact", () => { + const frames = [ + encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot: 7, payload: "echo" }), + encodeFileTransferFrame({ + opcode: FileTransferOpcode.FileChunk, + requestId: "upload", + payload: new Uint8Array([0, 255, 128, 1]), + }), + ]; + for (const payload of frames) { + const wire = encodeHostRouteFrame("channel", payload); + const decoded = decodeBinaryFrame(wire); + expect(decoded).toEqual({ kind: "host_route", frame: { routeId: "channel", payload } }); + expect(wire.subarray(10)).toEqual(payload); + expect(decodeBinaryFrame(decodeHostRouteFrame(wire)!.payload)).toEqual( + decodeBinaryFrame(payload), + ); + } + }); + it("rejects truncated, oversized identifiers and invalid UTF-8 without throwing", () => { + for (const bytes of [ + new Uint8Array([0x30]), + new Uint8Array([0x30, 0, 4, 65]), + new Uint8Array([0x30, 0, 0]), + new Uint8Array([0x30, 1, 1]), + new Uint8Array([0x30, 0, 1, 255]), + ]) + expect(decodeBinaryFrame(bytes)).toBeNull(); + expect(() => + encodeHostRouteFrame("a".repeat(MAX_HOST_ROUTE_ID_BYTES + 1), new Uint8Array()), + ).toThrow("Invalid host route ID"); + }); +}); diff --git a/packages/protocol/src/binary-frames/demux.ts b/packages/protocol/src/binary-frames/demux.ts index 31aca68cc6..a8503b898f 100644 --- a/packages/protocol/src/binary-frames/demux.ts +++ b/packages/protocol/src/binary-frames/demux.ts @@ -1,3 +1,4 @@ +import { HOST_ROUTE_OPCODE, decodeHostRouteFrame, type HostRouteFrame } from "./host-route.js"; import { decodeFileTransferFrame, FileTransferOpcode, @@ -10,11 +11,16 @@ import { } from "./terminal.js"; export type BinaryFrame = + | { kind: "host_route"; frame: HostRouteFrame } | { kind: "terminal"; frame: TerminalStreamFrame } | { kind: "file_transfer"; frame: FileTransferFrame }; export function decodeBinaryFrame(bytes: Uint8Array): BinaryFrame | null { switch (bytes[0]) { + case HOST_ROUTE_OPCODE: { + const frame = decodeHostRouteFrame(bytes); + return frame ? { kind: "host_route", frame } : null; + } case TerminalStreamOpcode.Output: case TerminalStreamOpcode.Input: case TerminalStreamOpcode.Resize: diff --git a/packages/protocol/src/binary-frames/host-route.ts b/packages/protocol/src/binary-frames/host-route.ts new file mode 100644 index 0000000000..f45cd714f9 --- /dev/null +++ b/packages/protocol/src/binary-frames/host-route.ts @@ -0,0 +1,33 @@ +import { MAX_WEBSOCKET_MESSAGE_BYTES } from "../websocket-message-limit.js"; +/** Routing adds a bounded header; the enclosed terminal/file frame is never re-encoded. */ +export const HOST_ROUTE_OPCODE = 0x30; +export const MAX_HOST_ROUTE_ID_BYTES = 256; +export const MAX_HOST_ROUTE_HEADER_BYTES = 3 + MAX_HOST_ROUTE_ID_BYTES; +export const MAX_ROUTED_PAYLOAD_BYTES = MAX_WEBSOCKET_MESSAGE_BYTES - MAX_HOST_ROUTE_HEADER_BYTES; +export interface HostRouteFrame { + routeId: string; + payload: Uint8Array; +} +export function encodeHostRouteFrame(routeId: string, payload: Uint8Array): Uint8Array { + const id = new TextEncoder().encode(routeId); + if (!id.length || id.length > MAX_HOST_ROUTE_ID_BYTES) throw new Error("Invalid host route ID"); + if (payload.byteLength > MAX_ROUTED_PAYLOAD_BYTES) throw new Error("Host frame is too large"); + const frame = new Uint8Array(3 + id.length + payload.byteLength); + frame[0] = HOST_ROUTE_OPCODE; + new DataView(frame.buffer).setUint16(1, id.length); + frame.set(id, 3); + frame.set(payload, 3 + id.length); + return frame; +} +export function decodeHostRouteFrame(bytes: Uint8Array): HostRouteFrame | null { + if (bytes[0] !== HOST_ROUTE_OPCODE || bytes.length < 3) return null; + const length = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getUint16(1); + if (!length || length > MAX_HOST_ROUTE_ID_BYTES || bytes.length < 3 + length) return null; + if (bytes.length - 3 - length > MAX_ROUTED_PAYLOAD_BYTES) return null; + try { + const routeId = new TextDecoder("utf-8", { fatal: true }).decode(bytes.subarray(3, 3 + length)); + return { routeId, payload: bytes.subarray(3 + length) }; + } catch { + return null; + } +} diff --git a/packages/protocol/src/binary-frames/index.ts b/packages/protocol/src/binary-frames/index.ts index 533b8fe453..cb24172960 100644 --- a/packages/protocol/src/binary-frames/index.ts +++ b/packages/protocol/src/binary-frames/index.ts @@ -1,3 +1,4 @@ export * from "./demux.js"; export * from "./file-transfer.js"; export * from "./terminal.js"; +export * from "./host-route.js"; diff --git a/packages/protocol/src/hosts.ts b/packages/protocol/src/hosts.ts index 7f4edaa0dd..7d36a50236 100644 --- a/packages/protocol/src/hosts.ts +++ b/packages/protocol/src/hosts.ts @@ -66,8 +66,8 @@ export const HostsRouteOpenRequestSchema = z.object({ export const HostsRouteFrameRequestSchema = z.object({ type: z.literal("hosts.route.frame.request"), routeId: z.string(), - data: z.string().max(24 * 1024 * 1024), - binary: z.boolean(), + data: z.string().max(100 * 1024 * 1024), + binary: z.literal(false), }); export const HostsRouteCloseRequestSchema = z.object({ type: z.literal("hosts.route.close.request"), @@ -104,9 +104,20 @@ export const HostsRouteOpenResponseSchema = z.object({ }); export const HostsRouteFrameSchema = z.object({ type: z.literal("hosts.route.frame"), - payload: z.object({ subscriptionId: z.string(), data: z.string(), binary: z.boolean() }), + payload: z.object({ subscriptionId: z.string(), data: z.string(), binary: z.literal(false) }), }); export const HostsRouteClosedSchema = z.object({ type: z.literal("hosts.route.closed"), payload: z.object({ subscriptionId: z.string(), code: z.string(), message: z.string() }), }); + +// Logical B sessions are admitted under the authenticated physical link, never hello auth. +export const HostsLinkOpenRequestSchema = z.object({ + type: z.literal("hosts.link.open.request"), + requestId: z.string(), + channelId: z.string().min(1).max(256), +}); +export const HostsLinkOpenResponseSchema = z.object({ + type: z.literal("hosts.link.open.response"), + payload: z.object({ requestId: z.string(), routeId: z.string() }), +}); diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 4d3df09c8c..4be3229b63 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -4,6 +4,7 @@ import { HostsAddRequestSchema, HostsUpdateRequestSchema, HostsRemoveRequestSchema, + HostsLinkOpenRequestSchema, HostsRouteOpenRequestSchema, HostsRouteFrameRequestSchema, HostsRouteCloseRequestSchema, @@ -12,6 +13,7 @@ import { HostsAddResponseSchema, HostsUpdateResponseSchema, HostsRemoveResponseSchema, + HostsLinkOpenResponseSchema, HostsRouteOpenResponseSchema, HostsRouteFrameSchema, HostsRouteClosedSchema, @@ -3229,6 +3231,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ HostsAddRequestSchema, HostsUpdateRequestSchema, HostsRemoveRequestSchema, + HostsLinkOpenRequestSchema, HostsRouteOpenRequestSchema, HostsRouteFrameRequestSchema, HostsRouteCloseRequestSchema, @@ -3626,6 +3629,7 @@ export const ServerInfoStatusPayloadSchema = z backgroundWorkspaces: z.boolean().optional(), providersSnapshot: z.boolean().optional(), connectedHosts: z.boolean().optional(), + hostMultiplexing: z.boolean().optional(), usageSources: z.boolean().optional(), // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10. providersSnapshotCwd: z.boolean().optional(), @@ -6912,6 +6916,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ HostsAddResponseSchema, HostsUpdateResponseSchema, HostsRemoveResponseSchema, + HostsLinkOpenResponseSchema, HostsRouteOpenResponseSchema, HostsRouteFrameSchema, HostsRouteClosedSchema, diff --git a/packages/server/src/server/authorization/operation-permissions.ts b/packages/server/src/server/authorization/operation-permissions.ts index aee93b5cbe..80599a74a3 100644 --- a/packages/server/src/server/authorization/operation-permissions.ts +++ b/packages/server/src/server/authorization/operation-permissions.ts @@ -10,6 +10,7 @@ const INBOUND_PERMISSION = { "hosts.add.request": "tunnel.manage", "hosts.update.request": "tunnel.manage", "hosts.remove.request": "tunnel.manage", + "hosts.link.open.request": null, "hosts.route.open.request": null, "hosts.route.frame.request": null, "hosts.route.close.request": null, @@ -228,6 +229,7 @@ const OUTBOUND_PERMISSION = { "hosts.add.response": "tunnel.manage", "hosts.update.response": "tunnel.manage", "hosts.remove.response": "tunnel.manage", + "hosts.link.open.response": null, "hosts.route.open.response": null, "hosts.route.frame": null, "hosts.route.closed": null, diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index ef4189d2e6..3fa0064781 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -1,6 +1,6 @@ import { afterEach, expect, test } from "vitest"; import { createPaseoApi } from "@getpaseo/client"; -import { createHostClient } from "@getpaseo/client/node"; +import { createHostClient, createHostChannelTransport } from "@getpaseo/client/node"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import type { HostDirectory } from "@getpaseo/protocol/hosts"; @@ -91,18 +91,22 @@ test("SDK saves unreachable hosts, rejects syntax and preserves source identity test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origin disposal", async () => { const b = await daemon(); + const wire = await hostWire(b); const a = await daemon(); const { paseo } = await api(a); const observation = paseo.hosts.observe(); expect((await observation.ready).hosts).toEqual([]); - const saved = await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "B" }); + const saved = await paseo.hosts.add({ connection: wire.connection, label: "B" }); const directory = await online(paseo); const serverId = directory.hosts[0]!.serverId!; const first = await paseo.hosts.client(serverId); const second = await paseo.hosts.client(serverId); + expect(wire.connections()).toBe(1); + expect(wire.active()).toBe(1); const workspaces = await second.workspaces.list({ subscribe: {} }); expect(workspaces.entries).toEqual([]); await first.dispose(); + expect(wire.active()).toBe(1); expect((await second.workspaces.list()).entries).toEqual([]); await expect(second.hosts.client(directory.self.serverId)).rejects.toMatchObject({ code: "unsupported_capability", @@ -123,7 +127,10 @@ test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origi { timeout: 20000 }, ) .toBe(0); + expect(wire.connections()).toBe(2); + expect(wire.active()).toBe(1); await second.dispose(); + expect(wire.active()).toBe(1); await observation.release(); await paseo.dispose(); }); @@ -160,9 +167,10 @@ test("routed authority is bounded by the caller, with isolated simultaneous sess test("distinct origin sessions cannot collide through a caller-selected route scope, including terminal binary output", async () => { const b = await daemon(); + const wire = await hostWire(b); const a = await daemon(); const admin = await api(a); - await admin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + await admin.paseo.hosts.add({ connection: wire.connection }); const serverId = (await online(admin.paseo)).hosts[0]!.serverId!; const remote = await admin.paseo.hosts.client(serverId); const workspace = await remote.workspaces.create({ @@ -184,12 +192,45 @@ test("distinct origin sessions cannot collide through a caller-selected route sc const second = await RoutePeer.open(secondOrigin.client, serverId); await first.subscribeTerminal(one.id); await second.subscribeTerminal(two.id); + expect(wire.connections()).toBe(1); first.input("FIRST_ONLY\n"); second.input("SECOND_ONLY\n"); await expect.poll(() => first.output()).toContain("FIRST_ONLY"); await expect.poll(() => second.output()).toContain("SECOND_ONLY"); expect(first.output()).not.toContain("SECOND_ONLY"); expect(second.output()).not.toContain("FIRST_ONLY"); + const firstBytes = new Uint8Array([0, 255, 128, 173]); + const secondBytes = new Uint8Array([173, 128, 255, 0]); + const paths = await Promise.all([ + first.upload("first.bin", firstBytes), + second.upload("second.bin", secondBytes), + ]); + expect((await readFile(paths[0]!)).equals(firstBytes)).toBe(true); + expect((await readFile(paths[1]!)).equals(secondBytes)).toBe(true); + + const binary = wire.frames.filter((frame) => frame.binary); + expect(binary.length).toBeGreaterThan(0); + expect(binary.every((frame) => decodeHostRouteFrame(frame.bytes) !== null)).toBe(true); + expect( + binary.some((frame) => { + const payload = decodeHostRouteFrame(frame.bytes)!.payload; + return Buffer.from(payload).equals( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + slot: 0, + payload: "FIRST_ONLY\n", + }), + ); + }), + ).toBe(true); + const json = wire.frames + .filter((frame) => !frame.binary) + .map((frame) => JSON.parse(frame.bytes.toString())); + expect(json.filter((frame) => frame.type === "hello")).toHaveLength(1); + expect(json.some((frame) => frame.message?.type === "hosts.link.open.request")).toBe(true); + expect( + json.some((frame) => frame.message?.binary === true || frame.message?.payload?.binary === true), + ).toBe(false); await first.close(); second.input("STILL_SECOND\n"); await expect.poll(() => second.output()).toContain("STILL_SECOND"); @@ -200,6 +241,9 @@ test("distinct origin sessions cannot collide through a caller-selected route sc }); import { + encodeFileTransferFrame, + FileTransferOpcode, + decodeHostRouteFrame, decodeTerminalStreamFrame, encodeTerminalStreamFrame, TerminalStreamOpcode, @@ -209,7 +253,7 @@ import { type SessionOutboundMessage, type SessionInboundMessage, } from "@getpaseo/protocol/messages"; -import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { DaemonClient } from "@getpaseo/client/internal/daemon-client"; class RoutePeer { readonly messages: SessionOutboundMessage[] = []; readonly bytes: Uint8Array[] = []; @@ -219,15 +263,21 @@ class RoutePeer { private readonly origin: DaemonClient, private readonly routeId: string, ) { - this.detach = origin.subscribeRawMessages((message) => { + const binary = origin.subscribeHostFrames((frame) => { + if (frame.routeId === routeId) this.bytes.push(frame.payload); + }); + const text = origin.subscribeRawMessages((message) => { if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId) return; - if (message.payload.binary) this.bytes.push(Buffer.from(message.payload.data, "base64")); - else { + { const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data)); if (frame.type === "session") this.messages.push(frame.message); } }); + this.detach = () => { + binary(); + text(); + }; } static async open(origin: DaemonClient, serverId: string) { const peer = new RoutePeer( @@ -243,7 +293,6 @@ class RoutePeer { protocolVersion: 1, capabilities: { owned_subscriptions: true }, }), - false, ); await expect .poll( @@ -264,7 +313,7 @@ class RoutePeer { return peer; } send(message: SessionInboundMessage) { - this.origin.sendHostFrame(this.routeId, JSON.stringify({ type: "session", message }), false); + this.origin.sendHostFrame(this.routeId, JSON.stringify({ type: "session", message })); } async subscribeTerminal(terminalId: string) { this.send({ type: "subscribe_terminal_request", terminalId, requestId: "same-request" }); @@ -284,16 +333,48 @@ class RoutePeer { input(text: string) { this.origin.sendHostFrame( this.routeId, - Buffer.from( - encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Input, - slot: this.slot, - payload: text, - }), - ).toString("base64"), - true, + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + slot: this.slot, + payload: text, + }), ); } + async upload(fileName: string, bytes: Uint8Array) { + const requestId = "same-upload-request"; + const modifiedAt = new Date().toISOString(); + this.send({ + type: "file.upload.request", + requestId, + fileName, + mimeType: "application/octet-stream", + size: bytes.length, + modifiedAt, + }); + for (const frame of [ + encodeFileTransferFrame({ + opcode: FileTransferOpcode.FileBegin, + requestId, + metadata: { + mime: "application/octet-stream", + size: bytes.length, + encoding: "binary", + modifiedAt, + fileName, + }, + }), + encodeFileTransferFrame({ opcode: FileTransferOpcode.FileChunk, requestId, payload: bytes }), + encodeFileTransferFrame({ opcode: FileTransferOpcode.FileEnd, requestId }), + ]) + this.origin.sendHostFrame(this.routeId, frame); + await expect + .poll(() => this.messages.some((message) => message.type === "file.upload.response")) + .toBe(true); + const response = this.messages.find((message) => message.type === "file.upload.response"); + if (response?.type !== "file.upload.response" || !response.payload.file) + throw new Error("Upload failed"); + return response.payload.file.path; + } output() { return this.bytes .flatMap((bytes) => { @@ -310,9 +391,9 @@ class RoutePeer { } } -import { mkdir, writeFile, stat, mkdtemp, rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; +import { mkdir, writeFile, readFile, chmod, stat, mkdtemp, rm } from "node:fs/promises"; +import { join, dirname, basename } from "node:path"; +import { tmpdir, userInfo } from "node:os"; import { resolveDaemonVersion } from "../daemon-version.js"; test("identity discovery failures remain inspectable and changing a pinned destination requires explicit replacement", async () => { @@ -433,7 +514,7 @@ export default function contribute(server) { expect((await direct.paseo.workspaces.list()).entries).toHaveLength(1); }); -import { execFile } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; import { promisify } from "node:util"; import { fileURLToPath } from "node:url"; @@ -543,7 +624,7 @@ test("resuming an A session with narrower grants terminates existing routes and await remote.dispose(); }); -import { WebSocketServer } from "ws"; +import { WebSocketServer, WebSocket } from "ws"; import { createServer } from "node:http"; import { once } from "node:events"; import { hashSync } from "bcryptjs"; @@ -669,14 +750,10 @@ test("disposing an origin during routed handshake settles acquisition and closes let routeHello = false; let routeClosed = false; ws.on("connection", (socket) => { - const monitor = ++connections === 1; - socket.on("close", () => { - if (!monitor) routeClosed = true; - }); + connections++; socket.on("message", (data) => { const frame = JSON.parse(data.toString()); - if (frame.type !== "hello") return; - if (monitor) + if (frame.type === "hello") socket.send( JSON.stringify({ type: "session", @@ -686,13 +763,28 @@ test("disposing an origin during routed handshake settles acquisition and closes status: "server_info", serverId: "srv_pending", hostname: "pending", - version: "0.11.0", - features: { connectedHosts: true, ownedSubscriptions: true }, + version: "0.12.0", + features: { + connectedHosts: true, + hostMultiplexing: true, + ownedSubscriptions: true, + }, }, }, }), ); - else routeHello = true; + if (frame.message?.type === "hosts.link.open.request") + socket.send( + JSON.stringify({ + type: "session", + message: { + type: "hosts.link.open.response", + payload: { requestId: frame.message.requestId, routeId: "pending-channel" }, + }, + }), + ); + if (frame.message?.type === "hosts.route.frame.request") routeHello = true; + if (frame.message?.type === "hosts.route.close.request") routeClosed = true; }); }); http.listen(0, "127.0.0.1"); @@ -715,4 +807,427 @@ test("disposing an origin during routed handshake settles acquisition and closes expect(await result).toBeInstanceOf(Error); await expect.poll(() => routeClosed).toBe(true); expect(ws.clients.size).toBe(1); + expect(connections).toBe(1); +}); + +test("source-key registration remains idempotent at directory capacity", async () => { + const { ConnectedHosts } = await import("./index.js"); + const home = await mkdtemp(join(tmpdir(), "hosts-capacity-")); + cleanup.push(() => rm(home, { recursive: true, force: true })); + const rows = Array.from({ length: 128 }, (_, index) => ({ + registrationId: `saved-${index}`, + serverId: null, + label: null, + source: { id: "plugin:capacity", key: `key-${index}` }, + metadata: {}, + enabled: false, + status: "disabled", + targetDescription: "127.0.0.1:1", + lastOnlineAt: null, + nextRetryAt: null, + error: null, + connection: "127.0.0.1:1", + })); + await writeFile(join(home, "hosts.json"), JSON.stringify(rows)); + const registry = new ConnectedHosts(home, { serverId: "self", label: "self" }); + cleanup.push(() => registry.close()); + const actor = { sourceId: "plugin:capacity", manageSources: false }; + await expect( + registry.add({ connection: "127.0.0.1:1", sourceKey: "key-0" }, actor), + ).resolves.toMatchObject({ registrationId: "saved-0", enabled: false }); + await expect( + registry.add({ connection: "127.0.0.1:1", sourceKey: "new" }, actor), + ).rejects.toMatchObject({ code: "resource_limit" }); +}); + +/** Real WebSocket proxy records the A–B physical wire without replacing either daemon. */ +async function hostWire(b: TestPaseoDaemon) { + const http = createServer(); + const wss = new WebSocketServer({ server: http }); + const frames: { binary: boolean; bytes: Buffer; direction: "toB" | "fromB" }[] = []; + let connections = 0; + wss.on("connection", (downstream) => { + connections++; + const upstream = new WebSocket(`ws://127.0.0.1:${b.port}/ws`); + const pending: (() => void)[] = []; + upstream.on("open", () => { + for (const send of pending.splice(0)) send(); + }); + downstream.on("message", (data, binary) => { + const bytes = Buffer.from(data as Buffer); + frames.push({ binary, bytes, direction: "toB" }); + const send = () => { + if (upstream.readyState === 1) upstream.send(bytes, { binary }); + }; + if (upstream.readyState === 0) pending.push(send); + else send(); + }); + upstream.on("message", (data, binary) => { + const bytes = Buffer.from(data as Buffer); + frames.push({ binary, bytes, direction: "fromB" }); + if (downstream.readyState === 1) downstream.send(bytes, { binary }); + }); + upstream.on("error", () => downstream.close()); + upstream.on("close", () => downstream.close()); + downstream.on("close", () => upstream.close()); + }); + http.listen(0, "127.0.0.1"); + await once(http, "listening"); + cleanup.push(async () => { + for (const socket of wss.clients) socket.terminate(); + await new Promise((resolve) => wss.close(() => resolve())); + await new Promise((resolve) => http.close(() => resolve())); + }); + const address = http.address(); + if (!address || typeof address === "string") throw new Error("Missing wire port"); + return { + connection: `127.0.0.1:${address.port}`, + frames, + connections: () => connections, + active: () => wss.clients.size, + }; +} + +test("a resumed physical link re-admits every logical session under its current grant ceiling", async () => { + const b = await daemon(); + const parent = createHostClient(`127.0.0.1:${b.port}`, { clientId: "physical-grants" }); + cleanup.push(() => parent.close()); + await parent.connect(); + const serverId = parent.getLastServerInfoMessage()!.serverId; + const logical = new DaemonClient({ + url: "ws://logical/ws", + clientId: "logical", + transportFactory: createHostChannelTransport( + parent, + serverId, + "logical", + null, + { failure: null }, + true, + ), + routedClient: true, + routedServerId: serverId, + }); + cleanup.push(() => logical.close()); + await logical.connect(); + const scoped = createPaseoApi(logical); + cleanup.push(() => scoped.dispose()); + const bytes = Uint8Array.from({ length: 256 * 1024 + 123 }, (_, index) => index % 251); + const upload = await logical.uploadFile({ + fileName: "routed.bin", + mimeType: "application/octet-stream", + bytes, + }); + expect(upload.error).toBeNull(); + const downloaded = await logical.readFile( + dirname(upload.file!.path), + basename(upload.file!.path), + ); + expect(Buffer.from(downloaded.bytes).equals(bytes)).toBe(true); + + const observed = await scoped.workspaces.list({ subscribe: {} }); + const snapshots: unknown[] = []; + observed.subscription.subscribe({ snapshot: (value) => snapshots.push(value), update() {} }); + const narrowed = createHostClient(`127.0.0.1:${b.port}`, { + clientId: "physical-grants", + sessionPermissions: ["daemon.read", "workspace.read"], + reconnect: { enabled: false }, + }); + cleanup.push(() => narrowed.close()); + await narrowed.connect(); + await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(2); + expect(logical.getLastServerInfoMessage()?.permissions).toEqual([ + "daemon.read", + "workspace.read", + ]); + await expect( + scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } }), + ).rejects.toMatchObject({ code: "permission_denied" }); + const restored = createHostClient(`127.0.0.1:${b.port}`, { + clientId: "physical-grants", + reconnect: { enabled: false }, + }); + cleanup.push(() => restored.close()); + await restored.connect(); + await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(3); + expect(logical.getLastServerInfoMessage()?.permissions).toContain("workspace.write"); + await scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } }); +}); + +import { DaemonLinkEndpoint, VirtualHostSocket } from "./index.js"; +import { SessionDelivery } from "../session/owned-subscriptions/index.js"; +import { sendBoundedPhysicalFrameAndWait } from "../websocket/physical-socket.js"; + +test("channel output accounts for escaped text, awaits physical drain, and preserves its sibling", async () => { + const http = createServer(); + const wss = new WebSocketServer({ server: http }); + http.listen(0, "127.0.0.1"); + await once(http, "listening"); + const address = http.address(); + if (!address || typeof address === "string") throw new Error("Missing port"); + const accepted = once(wss, "connection"); + const receiver = new WebSocket(`ws://127.0.0.1:${address.port}`); + const received: { binary: boolean; bytes: Buffer }[] = []; + receiver.on("message", (bytes, binary) => + received.push({ bytes: Buffer.from(bytes as Buffer), binary }), + ); + const [physical] = (await accepted) as [WebSocket]; + await once(receiver, "open"); + cleanup.push(async () => { + receiver.terminate(); + physical.terminate(); + await new Promise((resolve) => wss.close(() => resolve())); + await new Promise((resolve) => http.close(() => resolve())); + }); + const delivery = new SessionDelivery( + async (source, message) => { + expect(delivery.permits(source, message)).toBe(true); + await sendBoundedPhysicalFrameAndWait({ + socket: physical, + frame: JSON.stringify({ type: "session", message }), + onHighWater: () => { + throw new Error("Shared link overflowed"); + }, + }); + }, + async (source, frame) => { + expect(delivery.permitsBinary(source, frame)).toBe(true); + await sendBoundedPhysicalFrameAndWait({ + socket: physical, + frame, + onHighWater: () => { + throw new Error("Shared link overflowed"); + }, + }); + }, + ); + delivery.attach(physical, true); + const channels: VirtualHostSocket[] = []; + const endpoint = new DaemonLinkEndpoint({ + identity: "authenticated", + permissions: () => ["daemon.read"], + routed: () => false, + bufferedAmount: () => physical.bufferedAmount, + delivery, + admit: async (socket) => { + channels.push(socket); + }, + }); + cleanup.push(async () => { + endpoint.close(); + await delivery.close(); + }); + for (const channelId of ["slow", "sibling"]) { + const request = { type: "hosts.link.open.request" as const, requestId: channelId, channelId }; + await delivery.request(physical, request, () => endpoint.dispatch(request)); + } + await expect(channels[0]!.send('"'.repeat(34 * 1024 * 1024))).rejects.toThrow( + "Host channel output limit reached", + ); + expect(channels[0]!.readyState).toBe(3); + expect(physical.readyState).toBe(1); + await channels[1]!.send("sibling-before"); + await expect + .poll(() => received.some((frame) => !frame.binary && frame.bytes.includes("sibling-before"))) + .toBe(true); + receiver.pause(); + let completed = false; + const bytes = new Uint8Array(8 * 1024 * 1024).fill(173); + const pending = channels[1]!.send(bytes).then(() => { + completed = true; + return undefined; + }); + await expect.poll(() => physical.bufferedAmount).toBeGreaterThan(0); + expect(completed).toBe(false); + expect(physical.bufferedAmount).toBeLessThan(64 * 1024 * 1024); + receiver.resume(); + await pending; + await channels[1]!.send("sibling-after"); + await expect + .poll(() => received.some((frame) => !frame.binary && frame.bytes.includes("sibling-after"))) + .toBe(true); + const binary = received.find((frame) => frame.binary)!; + expect(Buffer.from(decodeHostRouteFrame(binary.bytes)!.payload).equals(bytes)).toBe(true); + expect(wss.clients.size).toBe(1); +}); + +import { existsSync } from "node:fs"; +test.skipIf(process.platform !== "linux" || !existsSync("/usr/sbin/sshd"))( + "real SSH shares one tunnel across scopes and owns fresh attempts after recovery", + async () => { + const root = await mkdtemp(join(tmpdir(), "hosts-ssh-")); + cleanup.push(() => rm(root, { recursive: true, force: true })); + const run = promisify(execFile); + for (const name of ["host", "client"]) + await run("ssh-keygen", ["-t", "ed25519", "-N", "", "-f", join(root, name)]); + const authorized = await readFile(join(root, "client.pub")); + await writeFile(join(root, "authorized_keys"), authorized); + const portServer = createServer(); + portServer.listen(0, "127.0.0.1"); + await once(portServer, "listening"); + const address = portServer.address(); + if (!address || typeof address === "string") throw new Error("Missing SSH port"); + await new Promise((resolve) => portServer.close(() => resolve())); + await writeFile( + join(root, "sshd_config"), + `Port ${address.port}\nListenAddress 127.0.0.1\nHostKey ${root}/host\nPidFile ${root}/pid\nAuthorizedKeysFile ${root}/authorized_keys\nStrictModes no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nUsePAM no\nAllowTcpForwarding yes\n`, + ); + await writeFile( + join(root, "ssh"), + `#!/bin/sh\nprintf 'attempt\\n' >> '${root}/attempts'\nexec /usr/bin/ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile='${root}/known_hosts' -i '${root}/client' "$@"\n`, + ); + await chmod(join(root, "ssh"), 0o700); + const originalPath = process.env.PATH; + process.env.PATH = `${root}:${originalPath}`; + cleanup.push(async () => { + process.env.PATH = originalPath; + }); + const b = await createTestPaseoDaemon({ auth: { password: hashSync("ssh-password", 4) } }); + cleanup.push(() => b.close()); + const a = await daemon(); + const { paseo, client } = await api(a); + const host = await paseo.hosts.add({ + connection: `ssh://${userInfo().username}@127.0.0.1:${address.port}?daemonPort=${b.port}&password=ssh-password`, + }); + await expect.poll(async () => (await paseo.hosts.list()).hosts[0]?.status).toBe("offline"); + const sshd = spawn("/usr/sbin/sshd", [ + "-D", + "-f", + join(root, "sshd_config"), + "-E", + join(root, "sshd.log"), + ]); + cleanup.push(async () => { + sshd.kill(); + }); + await online(paseo); + const attempts = async () => + (await readFile(join(root, "attempts"), "utf8")).trim().split("\n").length; + const before = await attempts(); + const first = await paseo.hosts.client(b.daemon.getServerId()); + const sibling = await paseo.hosts.client(b.daemon.getServerId()); + expect((await sibling.workspaces.list()).entries).toEqual([]); + expect(await attempts()).toBe(before); + await first.dispose(); + expect((await sibling.workspaces.list()).entries).toEqual([]); + expect(await attempts()).toBe(before); + const workspace = await sibling.workspaces.create({ + source: { kind: "directory", path: b.staticDir }, + }); + const terminal = await workspace.terminals.create({ + command: process.execPath, + args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"], + }); + const stream = await RoutePeer.open(client, b.daemon.getServerId()); + await stream.subscribeTerminal(terminal.id); + stream.input("SSH_BINARY_BEFORE\n"); + await expect.poll(() => stream.output()).toContain("SSH_BINARY_BEFORE"); + await stream.close(); + expect(await attempts()).toBe(before); + + await paseo.hosts.update(host.registrationId, { enabled: false }); + await paseo.hosts.update(host.registrationId, { enabled: true }); + await online(paseo); + await expect + .poll( + async () => { + try { + return (await sibling.workspaces.list()).entries.length; + } catch { + return -1; + } + }, + { timeout: 15000 }, + ) + .toBe(1); + expect(await attempts()).toBe(before + 1); + const resumed = await RoutePeer.open(client, b.daemon.getServerId()); + await resumed.subscribeTerminal(terminal.id); + resumed.input("SSH_BINARY_AFTER\n"); + await expect.poll(() => resumed.output()).toContain("SSH_BINARY_AFTER"); + await resumed.close(); + await terminal.kill(); + expect(await attempts()).toBe(before + 1); + await sibling.dispose(); + await paseo.hosts.update(host.registrationId, { enabled: false }); + await writeFile(join(root, "authorized_keys"), ""); + await paseo.hosts.update(host.registrationId, { enabled: true }); + await expect + .poll(async () => (await paseo.hosts.list()).hosts[0]?.error?.code, { timeout: 15000 }) + .toBe("authentication_failed"); + expect((await paseo.hosts.list()).hosts[0]?.nextRetryAt).toBeNull(); + await writeFile(join(root, "authorized_keys"), authorized); + await paseo.hosts.update(host.registrationId, { enabled: true }); + await online(paseo); + }, + 60000, +); + +test("malformed routing headers terminate their physical source and leave the daemon usable", async () => { + const b = await daemon(); + const socket = new WebSocket(`ws://127.0.0.1:${b.port}/ws`); + cleanup.push(async () => socket.terminate()); + await once(socket, "open"); + const info = once(socket, "message"); + socket.send( + JSON.stringify({ + type: "hello", + clientId: "malformed-route", + clientType: "cli", + protocolVersion: 1, + capabilities: { owned_subscriptions: true }, + }), + ); + await info; + const opened = once(socket, "message"); + socket.send( + JSON.stringify({ + type: "session", + message: { type: "hosts.link.open.request", requestId: "channel", channelId: "channel" }, + }), + ); + const [response] = await opened; + expect(JSON.parse(String(response)).message.type).toBe("hosts.link.open.response"); + const closed = once(socket, "close"); + socket.send(new Uint8Array([0x30, 0, 3, 97])); + expect((await closed)[0]).toBe(1008); + expect((await (await api(b)).paseo.workspaces.list()).entries).toEqual([]); +}); + +test("cancelling a logical scope during admission settles it and releases its owned channel", async () => { + const source = {}; + const messages: SessionOutboundMessage[] = []; + const delivery = new SessionDelivery((_source, message) => { + messages.push(message); + }); + delivery.attach(source, true); + let pendingSocket: VirtualHostSocket | undefined; + const endpoint = new DaemonLinkEndpoint({ + identity: "physical", + permissions: () => ["daemon.read"], + routed: () => false, + bufferedAmount: () => 0, + delivery, + admit: (socket) => + new Promise((resolve) => { + pendingSocket = socket; + socket.once("close", resolve); + }), + }); + const open = { + type: "hosts.link.open.request" as const, + requestId: "pending", + channelId: "logical-scope", + }; + const acquiring = delivery.request(source, open, () => endpoint.dispatch(open)); + expect(pendingSocket).toBeDefined(); + const cancel = { type: "hosts.route.close.request" as const, scopeId: "logical-scope" }; + expect(endpoint.handles(cancel)).toBe(true); + await delivery.request(source, cancel, () => endpoint.dispatch(cancel)); + await acquiring; + expect(pendingSocket!.readyState).toBe(3); + expect(messages.some((message) => message.type === "hosts.link.open.response")).toBe(false); + expect(messages.filter((message) => message.type === "hosts.route.closed")).toHaveLength(1); + endpoint.close(); + await delivery.close(); }); diff --git a/packages/server/src/server/connected-hosts/index.ts b/packages/server/src/server/connected-hosts/index.ts index 83a160119b..83f5173fab 100644 --- a/packages/server/src/server/connected-hosts/index.ts +++ b/packages/server/src/server/connected-hosts/index.ts @@ -1,3 +1,4 @@ +export { DaemonLinkEndpoint, VirtualHostSocket } from "./link.js"; export { ConnectedHostSession } from "./session.js"; import { randomUUID, createHash } from "node:crypto"; import { mkdir, readFile, rename, writeFile, rm } from "node:fs/promises"; @@ -5,7 +6,7 @@ import { join } from "node:path"; import { z } from "zod"; import { createHostClient, - createHostTransportFactory, + createHostChannelTransport, parseHostTarget, HostTransportBudget, } from "@getpaseo/client/node"; @@ -37,6 +38,8 @@ export interface HostActor { manageSources: boolean; } interface Live { + client: DaemonClient; + accepted: boolean; stop(): void; } interface Route { @@ -118,8 +121,6 @@ export class ConnectedHosts { return this.mutate(async () => { input = HostAddSchema.parse(input); const target = parseHostTarget(input.connection); - if (this.rows.length >= 128) - throw new HostError("resource_limit", "Remove a host before adding another (limit 128)"); const source = { id: actor.sourceId, key: input.sourceKey ?? randomUUID() }; const existing = this.rows.find( (row) => row.source.id === source.id && row.source.key === source.key, @@ -137,6 +138,8 @@ export class ConnectedHosts { "This source key already exists. Update its registration explicitly.", ); } + if (this.rows.length >= 128) + throw new HostError("resource_limit", "Remove a host before adding another (limit 128)"); const row: Saved = { registrationId: `hst_${randomUUID()}`, serverId: null, @@ -233,55 +236,26 @@ export class ConnectedHosts { } private start(row: Saved): void { if (!row.enabled || this.closed) return; - let client: DaemonClient | null = null; - let timer: ReturnType | undefined; let stopped = false; - let connecting = false; + const client = createHostClient( + row.connection, + { + clientId: `host-link-${this.self.serverId}-${row.registrationId}`, + connectTimeoutMs: 10000, + }, + this.budget, + ); const live: Live = { + client, + accepted: false, stop: () => { stopped = true; - clearTimeout(timer); - void client?.close(); + void client.close(); }, }; this.live.set(row.registrationId, live); - const retry = () => { - if (stopped || connecting || timer || row.status === "error") return; - row.status = "offline"; - row.error = { - code: "host_offline", - message: "Cannot reach host. Check the connection; retrying automatically.", - }; - row.nextRetryAt = new Date(Date.now() + 5000).toISOString(); - this.changed(); - timer = setTimeout(() => { - timer = undefined; - void attempt(); - }, 5000); - timer.unref(); - }; - const attempt = async () => { - if (stopped) return; - await client?.close(); - client = createHostClient(row.connection, { - clientId: `host-monitor-${this.self.serverId}-${row.registrationId}`, - reconnect: { enabled: false }, - connectTimeoutMs: 10000, - }); - client.subscribeConnectionStatus((state) => { - if (!stopped && state.status === "disconnected") { - for (const route of this.routes) - if (route.registrationId === row.registrationId) route.close(); - retry(); - } - }); - connecting = true; - row.status = "connecting"; - row.nextRetryAt = null; - this.changed(); + const accept = async () => { try { - await client.connect(); - if (stopped) return; const info = client.getLastServerInfoMessage(); if (!info) throw new HostError("host_offline", "No daemon identity received"); if (row.serverId && info.serverId !== row.serverId) @@ -306,7 +280,7 @@ export class ConnectedHosts { "duplicate_host", "Another registration already owns this daemon. Remove the duplicate registration.", ); - if (info.features?.connectedHosts !== true) + if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true) throw new HostError( "unsupported_capability", "Update the destination daemon to use connected hosts.", @@ -325,25 +299,55 @@ export class ConnectedHosts { nextRetryAt: null, }; await this.persist(this.rows.map((entry) => (entry === row ? next : entry))); + if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return; Object.assign(row, next); + live.accepted = true; this.changed(); }); } catch (error) { if (stopped) return; - const failure = connectionFailure(error, client.authFailureReason !== null); - if (failure) { - row.status = "error"; - row.error = failure; - row.nextRetryAt = null; - this.changed(); - } + row.status = "error"; + row.error = connectionFailure(error, client.authFailureReason !== null) ?? { + code: "host_offline", + message: "Host connection failed", + }; + row.nextRetryAt = null; + this.changed(); await client.close(); - } finally { - connecting = false; - if (!stopped && !client.isConnected) retry(); } }; - void attempt(); + client.subscribeConnectionStatus((state) => { + if (stopped || row.status === "error") return; + live.accepted = false; + if (state.status === "connected") { + void accept(); + return; + } + for (const route of this.routes) + if (route.registrationId === row.registrationId) route.close(); + row.status = state.status === "connecting" ? "connecting" : "offline"; + row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null; + row.error = + state.status === "disconnected" + ? (connectionFailure( + Object.assign(new Error("Check the SSH configuration and daemon credentials"), { + code: state.failureCode, + }), + client.authFailureReason !== null, + ) ?? { + code: "host_offline", + message: "Cannot reach host. Check the connection; retrying automatically.", + }) + : null; + if (row.error?.code === "authentication_failed") { + row.status = "error"; + row.nextRetryAt = null; + client.setReconnectEnabled(false); + void client.close(); + } + this.changed(); + }); + void client.connect().catch(() => {}); } async openRoute(input: { serverId: string; @@ -352,9 +356,9 @@ export class ConnectedHosts { originSessionId: string; permissions: readonly DaemonPermission[]; signal: AbortSignal; - frame(data: string, binary: boolean): void; + frame(data: string | Uint8Array): void; ended(code: string, message: string): void; - }): Promise<{ send(data: string, binary: boolean): void; close(): void }> { + }): Promise<{ send(data: string | Uint8Array): void; close(): void }> { await this.ready; const row = this.rows.find((entry) => entry.serverId === input.serverId); if (!row) throw new HostError("host_unknown", `Unknown host ${input.serverId}`, input.serverId); @@ -371,10 +375,18 @@ export class ConnectedHosts { input.serverId, ); if (this.routes.size >= 128) throw new HostError("resource_limit", "Too many routed sessions"); - const transport = createHostTransportFactory( - row.connection, - this.budget, - )({ url: "ws://host-route/ws" }); + const live = this.live.get(row.registrationId); + const client = live?.client; + if (!live?.accepted || !client?.isConnected) + throw new HostError("host_offline", "Host is offline", input.serverId); + const transport = createHostChannelTransport( + client, + input.serverId, + randomUUID(), + null, + { failure: null }, + true, + )({ url: "ws://host-link/ws" }); let closed = false; let verified = false; let helloSent = false; @@ -460,22 +472,19 @@ export class ConnectedHosts { return; } } - input.frame( - binary ? Buffer.from(data as Uint8Array).toString("base64") : String(data), - binary, - ); + input.frame(binary ? (data as Uint8Array) : String(data)); }); await ready; if (closed) throw new HostError("host_offline", "Host closed during connection"); return { close: () => route.close(), - send: (data, binary) => { + send: (data) => { + const binary = typeof data !== "string"; if (closed) throw new HostError("host_offline", "Host is offline"); if (!helloSent) { if (binary) throw new HostError("invalid_message", "Expected hello"); - const hello = JSON.parse(data) as WSHelloMessage; + const hello = JSON.parse(data as string) as WSHelloMessage; if (hello.type !== "hello") throw new HostError("invalid_message", "Expected hello"); - const target = parseHostTarget(row.connection); const clientId = `route-${createHash("sha256") .update( JSON.stringify([ @@ -493,7 +502,7 @@ export class ConnectedHosts { routed: true, producer: { serverId: this.self.serverId, sourceId: input.sourceId }, sessionPermissions: input.permissions, - auth: target.password ? { kind: "password", password: target.password } : undefined, + auth: undefined, }), ); helloSent = true; @@ -501,7 +510,7 @@ export class ConnectedHosts { } if (!verified) throw new HostError("host_offline", "Host identity has not been verified"); try { - transport.send(binary ? Buffer.from(data, "base64") : data); + transport.send(data); } catch { route.close("resource_limit", "Host buffer limit reached or transport unavailable"); } diff --git a/packages/server/src/server/connected-hosts/link.ts b/packages/server/src/server/connected-hosts/link.ts new file mode 100644 index 0000000000..85fe06ed4e --- /dev/null +++ b/packages/server/src/server/connected-hosts/link.ts @@ -0,0 +1,210 @@ +import type { SessionDelivery } from "../session/owned-subscriptions/index.js"; +import { EventEmitter } from "node:events"; +import { createHash } from "node:crypto"; +import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index"; +import type { + DaemonPermission, + SessionInboundMessage, + SessionOutboundMessage, +} from "@getpaseo/protocol/messages"; + +/** A logical socket has no network resources or authority of its own. */ +export class VirtualHostSocket extends EventEmitter { + readyState = 1; + constructor( + private readonly output: (data: string | Uint8Array) => Promise, + private readonly ended: (code: number, reason: string) => void, + private readonly buffered: () => number, + private readonly transformHello: (data: string) => string, + ) { + super(); + } + get bufferedAmount(): number { + return this.buffered(); + } + private helloPending = true; + feed(data: string | Uint8Array): void { + if (this.readyState !== 1) return; + if (this.helloPending) { + if (typeof data !== "string") { + this.close(1008, "Expected hello"); + return; + } + try { + data = this.transformHello(data); + } catch { + this.close(1008, "Invalid hello"); + return; + } + this.helloPending = false; + } + this.emit("message", typeof data === "string" ? data : Buffer.from(data)); + } + async send(data: string | Uint8Array | ArrayBuffer): Promise { + if (this.readyState !== 1) throw new Error("Host channel closed"); + await this.output( + typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data), + ); + } + close(code = 1000, reason = "Host channel closed"): void { + if (this.readyState !== 1) return; + this.readyState = 3; + this.ended(code, reason); + this.emit("close", 1000, "Host channel closed"); + } +} + +/** One authenticated physical source owns all its channels and their admission ceiling. */ +export class DaemonLinkEndpoint { + private readonly channels = new Map(); + private closed = false; + constructor( + private readonly parent: { + identity: string; + permissions(): readonly DaemonPermission[]; + routed(): boolean; + bufferedAmount(): number | null; + delivery: SessionDelivery; + admit(socket: VirtualHostSocket, permissions: readonly DaemonPermission[]): Promise; + }, + ) {} + handles(message: SessionInboundMessage): boolean { + return ( + message.type === "hosts.link.open.request" || + (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) || + (message.type === "hosts.route.close.request" && + ((!!message.routeId && this.channels.has(message.routeId)) || + [...this.channels.values()].some((channel) => channel.scopeId === message.scopeId))) + ); + } + async dispatch(message: SessionInboundMessage): Promise { + if (message.type === "hosts.link.open.request") { + const error = (code: string, text: string) => + this.parent.delivery.reply({ + type: "rpc_error", + payload: { + requestId: message.requestId, + requestType: message.type, + code, + error: text, + }, + }); + if (this.closed || this.parent.routed()) { + error("unsupported_capability", "Connected hosts support one hop"); + return; + } + if (this.channels.size >= 128) { + error("resource_limit", "Too many host channels"); + return; + } + let socket: VirtualHostSocket; + const owner = this.parent.delivery.begin("host-link", undefined, () => { + this.channels.delete(owner.id); + socket?.close(); + }); + const id = owner.id; + // Capture actual authenticated authority. The caller hello can only reduce it. + const ceiling = [...this.parent.permissions()]; + socket = new VirtualHostSocket( + async (data) => { + const output: SessionOutboundMessage | null = + typeof data === "string" + ? { + type: "hosts.route.frame", + payload: { subscriptionId: id, data, binary: false }, + } + : null; + const frame = typeof data === "string" ? null : encodeHostRouteFrame(id, data); + const bytes = output + ? Buffer.byteLength(JSON.stringify({ type: "session", message: output })) + : frame!.byteLength; + const buffered = this.parent.bufferedAmount(); + // Account for the exact escaped outer frame, retaining control-frame headroom. + if (buffered === null || buffered + bytes > 64 * 1024 * 1024 - 65536) { + socket.close(1009, "Host channel output limit reached"); + throw new Error("Host channel output limit reached"); + } + try { + if (output) await owner.emitAndWait(output); + else await owner.emitBinaryAndWait(frame!); + } catch (failure) { + socket.close(); + throw failure; + } + }, + (code, reason) => { + this.channels.delete(id); + let failure = "host_offline"; + if (code === 1009) failure = "resource_limit"; + if (code === 1008) failure = "permission_denied"; + owner.emit({ + type: "hosts.route.closed", + payload: { + subscriptionId: id, + code: failure, + message: reason, + }, + }); + void owner.release(); + }, + () => this.parent.bufferedAmount() ?? 0, + (data) => { + const hello = JSON.parse(data); + if (hello.type !== "hello" || typeof hello.clientId !== "string") + throw new Error("Expected hello"); + return JSON.stringify({ + ...hello, + auth: undefined, + routed: true, + clientId: `link-${createHash("sha256") + .update(JSON.stringify([this.parent.identity, hello.clientId])) + .digest("hex")}`, + }); + }, + ); + this.channels.set(id, { scopeId: message.channelId, socket }); + try { + await this.parent.admit(socket, ceiling); + if (this.closed || socket.readyState !== 1) { + socket.close(); + return; + } + this.parent.delivery.reply({ + type: "hosts.link.open.response", + payload: { requestId: message.requestId, routeId: id }, + }); + } catch { + socket.close(); + error("host_offline", "Could not admit host channel"); + } + return; + } + if (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) { + this.channels.get(message.routeId)!.socket.feed(message.data); + return; + } + if (message.type === "hosts.route.close.request") { + for (const [id, channel] of this.channels) { + if (message.routeId === id || (message.scopeId && message.scopeId === channel.scopeId)) + channel.socket.close(); + } + } + return; + } + binary(frame: HostRouteFrame): boolean { + const socket = this.channels.get(frame.routeId)?.socket; + if (!socket) return false; + socket.feed(frame.payload); + return true; + } + close(reason: "host_offline" | "permission_denied" = "host_offline"): void { + if (this.closed) return; + this.closed = true; + for (const { socket } of this.channels.values()) + socket.close( + reason === "permission_denied" ? 1008 : 1000, + "Physical link authority or connection changed", + ); + this.channels.clear(); + } +} diff --git a/packages/server/src/server/connected-hosts/session.ts b/packages/server/src/server/connected-hosts/session.ts index 1c8fd5036d..811daaa723 100644 --- a/packages/server/src/server/connected-hosts/session.ts +++ b/packages/server/src/server/connected-hosts/session.ts @@ -1,3 +1,4 @@ +import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index"; import type { SessionInboundMessage, SessionOutboundMessage, @@ -13,7 +14,7 @@ type HostRequest = Extract< interface SessionRoute { scopeId: string; source: object; - send(data: string, binary: boolean): void; + send(data: string | Uint8Array): void; release(): Promise; fail(code: string, message: string): void; } @@ -31,6 +32,15 @@ export class ConnectedHostSession { private readonly originSessionId: string, private readonly bufferedAmount: (source: object) => number | null, ) {} + binary(frame: HostRouteFrame, source: object): void { + const route = this.routes.get(frame.routeId); + if (!route || route.source !== source) return; + try { + route.send(frame.payload); + } catch { + route.fail("resource_limit", "Host input buffer limit reached"); + } + } setSource(sourceId: string): void { this.actor.sourceId = sourceId; } @@ -116,7 +126,7 @@ export class ConnectedHostSession { const route = this.routes.get(message.routeId); if (!route || route.source !== this.delivery.currentSource) throw new HostError("host_unknown", "Unknown routed session"); - route.send(message.data, message.binary); + route.send(message.data); return; } case "hosts.route.close.request": { @@ -196,17 +206,25 @@ export class ConnectedHostSession { originSessionId: this.originSessionId, permissions: this.permissions(), signal: owner.signal, - frame: (data, binary) => { - const size = Buffer.byteLength(data) + 512; + frame: (data) => { + const output: SessionOutboundMessage | null = + typeof data === "string" + ? { + type: "hosts.route.frame", + payload: { subscriptionId: owner.id, data, binary: false }, + } + : null; + const binary = typeof data === "string" ? null : encodeHostRouteFrame(owner.id, data); + const size = output + ? Buffer.byteLength(JSON.stringify({ type: "session", message: output })) + : binary!.byteLength; const buffered = this.bufferedAmount(owner.source); if (buffered === null || buffered + size > 32 * 1024 * 1024) { fail("resource_limit", "The receiving client cannot keep up with host output"); return; } - owner.emit({ - type: "hosts.route.frame", - payload: { subscriptionId: owner.id, data, binary }, - }); + if (output) owner.emit(output); + else owner.emitBinary(binary!); }, ended: fail, }); diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts index 4a21810ddf..85030a4151 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -281,6 +281,7 @@ async function waitForCapturedLog( let remote: PaseoApi | undefined; let sibling: PaseoApi | undefined; let terminalRoute: Awaited> | undefined; + let wire: Awaited> | undefined; try { await startRelay({ useLocalRelay: true }); const port = relayPort; @@ -301,7 +302,8 @@ async function waitForCapturedLog( relayEndpoint: b.config.relayEndpoint, }); const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl); - const connection = new URL(`relay://127.0.0.1:${port}/${serverId}`); + wire = await relayWire(port); + const connection = new URL(`relay://127.0.0.1:${wire.port}/${serverId}`); connection.searchParams.set("key", daemonPublicKeyB64); connection.searchParams.set("password", "wrong-password"); const saved = await api.hosts.add({ connection: connection.toString() }); @@ -314,8 +316,12 @@ async function waitForCapturedLog( await expect .poll(async () => (await api!.hosts.list()).hosts[0]?.status, { timeout: 20000 }) .toBe("online"); + const admittedConnections = wire.connections(); + expect(wire.active()).toBe(1); remote = await api.hosts.client(serverId); sibling = await api.hosts.client(serverId); + expect(wire.connections()).toBe(admittedConnections); + expect(wire.active()).toBe(1); expect((await remote.workspaces.list()).entries).toEqual([]); const observed = await remote.workspaces.list({ subscribe: {} }); let snapshots = 0; @@ -333,6 +339,7 @@ async function waitForCapturedLog( args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"], }); terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id); + expect(wire.connections()).toBe(admittedConnections); terminalRoute.write("ENCRYPTED_BINARY_BEFORE\n"); await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_BEFORE"); terminalRoute.close(); @@ -366,6 +373,7 @@ async function waitForCapturedLog( ) .toBe(1); await expect.poll(() => snapshots, { timeout: 15000 }).toBeGreaterThan(before); + expect(wire.active()).toBe(1); terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id); terminalRoute.write("ENCRYPTED_BINARY_AFTER\n"); await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_AFTER"); @@ -398,6 +406,7 @@ async function waitForCapturedLog( await origin?.close(); await a?.close(); await b?.close(); + await wire?.close(); await stopRelay(); } }, 120000); @@ -1080,19 +1089,20 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t const routeId = await origin.openHostRoute(serverId, crypto.randomUUID()); const messages: SessionOutboundMessage[] = []; let output = ""; + const detachBinary = origin.subscribeHostFrames((routed) => { + if (routed.routeId !== routeId) return; + const frame = decodeTerminalStreamFrame(routed.payload); + if (frame?.opcode === TerminalStreamOpcode.Output) + output += new TextDecoder().decode(frame.payload); + }); const detach = origin.subscribeRawMessages((message) => { if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId) return; - if (message.payload.binary) { - const frame = decodeTerminalStreamFrame(Buffer.from(message.payload.data, "base64")); - if (frame?.opcode === TerminalStreamOpcode.Output) - output += new TextDecoder().decode(frame.payload); - } else { - const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data)); - if (frame.type === "session") messages.push(frame.message); - } + const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data)); + if (frame.type === "session") messages.push(frame.message); }); const close = () => { detach(); + detachBinary(); origin.closeHostRoute(routeId); }; try { @@ -1104,7 +1114,6 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t clientType: "cli", protocolVersion: 1, }), - false, ); await expect .poll(() => @@ -1119,7 +1128,6 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t type: "session", message: { type: "subscribe_terminal_request", terminalId, requestId: "binary" }, }), - false, ); await expect .poll(() => messages.some((message) => message.type === "subscribe_terminal_response")) @@ -1134,10 +1142,7 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t write: (payload: string) => origin.sendHostFrame( routeId, - Buffer.from( - encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }), - ).toString("base64"), - true, + encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }), ), }; } catch (error) { @@ -1145,3 +1150,46 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t throw error; } } + +async function relayWire(relayPort: number) { + const sockets = new Set(); + let connections = 0; + let active = 0; + const server = net.createServer((source) => { + const target = net.connect(relayPort, "127.0.0.1"); + sockets.add(source); + sockets.add(target); + let counted = false; + target.on("connect", () => { + counted = true; + connections++; + active++; + }); + target.on("error", () => source.destroy()); + source.on("error", () => target.destroy()); + target.on("close", () => { + if (counted) active--; + sockets.delete(target); + source.destroy(); + }); + source.on("close", () => { + sockets.delete(source); + target.destroy(); + }); + source.pipe(target); + target.pipe(source); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing relay wire port"); + return { + port: address.port, + connections: () => connections, + active: () => active, + close: async () => { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 5b85d12a2b..b837502f0c 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -288,6 +288,7 @@ vi.mock("./worktree-bootstrap.js", async (importOriginal) => { }); interface SessionForTestOptions { + sendToSource?: SessionOptions["onMessageToSource"]; clientId?: string; permissions?: readonly DaemonPermission[]; agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown }; @@ -441,6 +442,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { daemonRuntimeConfig: options.daemonRuntimeConfig, permissions: options.permissions ?? OWNER_PERMISSIONS, }; + if (options.sendToSource) sessionOptions.onMessageToSource = options.sendToSource; return new Session(sessionOptions); } @@ -5813,3 +5815,40 @@ test("provider snapshots preserve versionless visibility while capabilities upda ]); expect(references.compactSnapshot!.entries[0]!.modes![0]!.icon).toBe("ShieldCheck"); }); + +test("a failing direct source observes fire-and-forget file replies without unhandled rejections", async () => { + const root = mkdtempSync(join(tmpdir(), "session-send-failure-")); + writeFileSync(join(root, "readme.txt"), "source reply"); + let attempts = 0; + const unhandled: unknown[] = []; + const collect = (error: unknown) => unhandled.push(error); + process.on("unhandledRejection", collect); + const session = createSessionForTest({ + sendToSource: async () => { + attempts++; + throw new Error("Physical source closed during send"); + }, + }); + try { + const source = {}; + session.updateClientCapabilities(null, source); + await session.handleMessage( + { + type: "file_explorer_request", + requestId: "direct-file", + cwd: root, + path: "readme.txt", + mode: "file", + acceptBinary: false, + }, + source, + ); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(attempts).toBeGreaterThan(0); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", collect); + await session.cleanup(); + rmSync(root, { recursive: true, force: true }); + } +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index ba517436f4..69f2c53b45 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -445,7 +445,7 @@ export interface SessionOptions { appVersion?: string | null; clientCapabilities?: Record | null; onMessage: (msg: SessionOutboundMessage) => void; - onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void; + onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void | Promise; onBinaryMessage?: (frame: Uint8Array) => void; onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise; getTransportBufferedAmount?: (source?: object) => number | null; @@ -691,14 +691,12 @@ export class Session { readonly delivery = new SessionDelivery( (source, message) => { if (!this.authorization.allowsOutbound(message)) return; - if (this.onMessageToSource) this.onMessageToSource(source, message); - else this.onMessage(message); + if (this.onMessageToSource) return this.onMessageToSource(source, message); + this.onMessage(message); }, (source, frame) => { if (this.onBinaryMessageToSource) { - void this.onBinaryMessageToSource(source, frame).catch((error) => - this.sessionLogger.warn({ err: error }, "Failed to emit binary frame"), - ); + return this.onBinaryMessageToSource(source, frame); } else this.emitBinary(frame); }, (source, message) => this.workspaceSetupMessageForClient(message, source), @@ -720,7 +718,7 @@ export class Session { private readonly sessionId: string; private readonly onMessage: (msg: SessionOutboundMessage) => void; private readonly onMessageToSource: - | ((source: object, msg: SessionOutboundMessage) => void) + | ((source: object, msg: SessionOutboundMessage) => void | Promise) | null; private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null; private readonly onBinaryMessageToSource: @@ -885,9 +883,13 @@ export class Session { this.hostSourceId = hostSource(options); this.connectedHostSession = createHostSession(options, this); this.onMessage = onMessage; - this.onMessageToSource = onMessageToSource ?? null; + this.onMessageToSource = onMessageToSource + ? (source, message) => this.observeSourceSend(() => onMessageToSource(source, message)) + : null; this.onBinaryMessage = onBinaryMessage ?? null; - this.onBinaryMessageToSource = onBinaryMessageToSource ?? null; + this.onBinaryMessageToSource = onBinaryMessageToSource + ? (source, frame) => this.observeSourceSend(() => onBinaryMessageToSource(source, frame)) + : null; this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0); this.onLifecycleIntent = onLifecycleIntent ?? null; this.onWorkspaceRecovered = onWorkspaceRecovered ?? null; @@ -3205,6 +3207,10 @@ export class Session { } public async handleBinaryFrame(binaryFrame: BinaryFrame, source: object): Promise { + if (binaryFrame.kind === "host_route") { + this.connectedHostSession?.binary(binaryFrame.frame, source); + return; + } if (!this.authorization.allowsPermission("workspace.write")) { return; } @@ -8579,6 +8585,20 @@ export class Session { this.emitBinary(frame); } + private observeSourceSend(send: () => void | Promise): Promise { + let pending: Promise; + try { + pending = Promise.resolve(send()); + } catch (error) { + pending = Promise.reject(error); + } + // Legacy producers fire and forget; owned streams await the same failure. + void pending.catch((error) => + this.sessionLogger.warn({ err: error }, "Failed to emit source frame"), + ); + return pending; + } + private emitForSource(msg: SessionOutboundMessage, source?: object): void { if (source && this.onMessageToSource) { msg = this.delivery.authorizeReply(msg, source); diff --git a/packages/server/src/server/session/owned-subscriptions/index.test.ts b/packages/server/src/server/session/owned-subscriptions/index.test.ts index 8e8e33a4ce..4f4ab58382 100644 --- a/packages/server/src/server/session/owned-subscriptions/index.test.ts +++ b/packages/server/src/server/session/owned-subscriptions/index.test.ts @@ -169,3 +169,40 @@ test("failed teardown still closes the delivery and rejects late work", async () delivery.request(source, { type: "ping", requestId: "late" }, async () => {}), ).rejects.toThrow("Session delivery is closed"); }); + +test("owned send completion retains source proofs and reports physical failures to awaiting producers", async () => { + const source = {}; + const failure = new Error("Physical send failed"); + const delivery = new SessionDelivery( + async (socket, message) => { + expect(socket).toBe(source); + expect(delivery.permits(socket, message)).toBe(true); + throw failure; + }, + async (socket, frame) => { + expect(socket).toBe(source); + expect(delivery.permitsBinary(socket, frame)).toBe(true); + throw failure; + }, + ); + delivery.attach(source, true); + await delivery.request( + source, + { type: "hosts.link.open.request", requestId: "open", channelId: "scope" }, + async () => { + const owner = delivery.begin("host-link", undefined, () => {}); + await expect( + owner.emitAndWait({ + type: "hosts.route.frame", + payload: { subscriptionId: owner.id, data: "text", binary: false }, + }), + ).rejects.toBe(failure); + await expect(owner.emitBinaryAndWait(new Uint8Array([0, 255]))).rejects.toBe(failure); + await owner.release(); + await expect(owner.emitBinaryAndWait(new Uint8Array([0]))).rejects.toThrow( + "Subscription source is closed", + ); + }, + ); + await delivery.close(); +}); diff --git a/packages/server/src/server/session/owned-subscriptions/index.ts b/packages/server/src/server/session/owned-subscriptions/index.ts index 6ee5c91fa5..c17e5829be 100644 --- a/packages/server/src/server/session/owned-subscriptions/index.ts +++ b/packages/server/src/server/session/owned-subscriptions/index.ts @@ -51,6 +51,8 @@ export interface OwnedSubscription { readonly signal: AbortSignal; emit(message: SessionOutboundMessage): void; emitBinary(frame: Uint8Array): void; + emitAndWait(message: SessionOutboundMessage): Promise; + emitBinaryAndWait(frame: Uint8Array): Promise; release(): Promise; } @@ -67,8 +69,14 @@ export class SessionDelivery { private closing?: Promise; constructor( - private readonly send: (source: object, message: SessionOutboundMessage) => void, - private readonly sendBinary: (source: object, frame: Uint8Array) => void = () => {}, + private readonly send: ( + source: object, + message: SessionOutboundMessage, + ) => void | Promise, + private readonly sendBinary: ( + source: object, + frame: Uint8Array, + ) => void | Promise = () => {}, private readonly project: ( source: object, message: SessionOutboundMessage, @@ -319,7 +327,18 @@ export class SessionDelivery { emitBinary: (frame) => { if (!owner.active || !source.active) return; this.proofs.set(frame, owner); - this.sendBinary(source.socket, frame); + void Promise.resolve(this.sendBinary(source.socket, frame)).catch(() => {}); + }, + emitAndWait: async (message) => { + if (!owner.active || !source.active) throw new Error("Subscription source is closed"); + const projected = this.project(source.socket, withSubscriptionId(message, responseId)); + this.proofs.set(projected, owner); + await this.send(source.socket, projected); + }, + emitBinaryAndWait: async (frame) => { + if (!owner.active || !source.active) throw new Error("Subscription source is closed"); + this.proofs.set(frame, owner); + await this.sendBinary(source.socket, frame); }, release: () => this.releaseOwner(owner), }; @@ -414,7 +433,7 @@ export class SessionDelivery { private sendOwned(owner: DeliveryOwner, message: SessionOutboundMessage): void { const projected = this.project(owner.source.socket, message); this.proofs.set(projected, owner); - this.send(owner.source.socket, projected); + void Promise.resolve(this.send(owner.source.socket, projected)).catch(() => {}); } private releaseOwner(owner: RetainedOwner): Promise { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index d158815d3b..8d14e27e9f 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,3 +1,5 @@ +import { HOST_ROUTE_OPCODE } from "@getpaseo/protocol/binary-frames/index"; +import { DaemonLinkEndpoint, VirtualHostSocket } from "./connected-hosts/index.js"; import { ConnectedHosts } from "./connected-hosts/index.js"; import { stat } from "node:fs/promises"; import type { CreationSnapshot } from "@getpaseo/protocol/messages"; @@ -438,6 +440,7 @@ export interface WebSocketLike { } interface SessionConnectionBase { + hubExecutionAgents?: HubExecutionAgents; permissionCeiling?: readonly DaemonPermission[]; session: Session; principalId: string; @@ -469,7 +472,7 @@ interface SocketSessionOptions { permissions: readonly DaemonPermission[]; connectionLogger: pino.Logger; onMessage: (message: SessionOutboundMessage) => void; - onMessageToSource?: (source: object, message: SessionOutboundMessage) => void; + onMessageToSource?: (source: object, message: SessionOutboundMessage) => void | Promise; onBinaryMessage?: (frame: Uint8Array) => void; onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise; getTransportBufferedAmount?: (source?: object) => number | null; @@ -530,6 +533,7 @@ function requireWebSocketServices(params: { export class VoiceAssistantWebSocketServer { private readonly logger: pino.Logger; private readonly wss: WebSocketServer; + private readonly hostLinks = new Map(); private readonly pendingConnections: Map = new Map(); private readonly sessions: Map = new Map(); private readonly socketIdentities: Map = new Map(); @@ -1052,6 +1056,12 @@ export class VoiceAssistantWebSocketServer { principalId: string, permissions: readonly DaemonPermission[], ): void { + for (const [socket, link] of this.hostLinks) { + if (this.sessions.get(socket)?.principalId === principalId) { + link.close("permission_denied"); + this.hostLinks.delete(socket); + } + } for (const pending of this.pendingConnections.values()) { if (pending.admission?.principalId === principalId) { pending.admission = { ...pending.admission, permissions }; @@ -1169,6 +1179,20 @@ export class VoiceAssistantWebSocketServer { message: WSOutboundMessage, reportRejection = false, ): void { + const prepared = this.prepareJsonDelivery(sockets, message, reportRejection); + if (!prepared) return; + for (const ws of prepared.sockets) { + this.sendFrameToClient(ws, prepared.payload, prepared.bytes, () => { + this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount); + }); + } + } + + private prepareJsonDelivery( + sockets: Iterable, + message: WSOutboundMessage, + reportRejection: boolean, + ): { sockets: WebSocketLike[]; payload: string; bytes: number } | null { const writableSockets = [...sockets].filter((ws) => { if ( message.type === "session" && @@ -1190,7 +1214,7 @@ export class VoiceAssistantWebSocketServer { return this.ensureOutboundCapacity(ws, 0); }); if (writableSockets.length === 0) { - return; + return null; } let payload: string; @@ -1198,14 +1222,30 @@ export class VoiceAssistantWebSocketServer { payload = JSON.stringify(message); } catch (err) { this.logger.warn({ err }, "ws_serialize_failed"); - return; + return null; } - const payloadBytes = outboundFrameByteLength(payload); - for (const ws of writableSockets) { - this.sendFrameToClient(ws, payload, payloadBytes, () => { - this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount); + return { sockets: writableSockets, payload, bytes: outboundFrameByteLength(payload) }; + } + + private async sendMessageToClientAndWait( + ws: WebSocketLike, + message: WSOutboundMessage, + ): Promise { + const prepared = this.prepareJsonDelivery([ws], message, true); + if (!prepared) throw new Error("Source message delivery unavailable"); + try { + const sent = await sendBoundedPhysicalFrameAndWait({ + socket: ws, + frame: prepared.payload, + frameBytes: prepared.bytes, + onHighWater: () => this.closeAtOutboundHighWater(ws), }); + if (!sent) throw new Error("Source is closed"); + this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount); + } catch (err) { + this.logger.warn({ err }, "ws_send_failed"); + throw err; } } @@ -1415,11 +1455,11 @@ export class VoiceAssistantWebSocketServer { } this.sendToConnection(connection, wrapSessionMessage(msg)); }, - onMessageToSource: (source, msg) => { + onMessageToSource: async (source, msg) => { if (!connection || !connection.sockets.has(source as WebSocketLike)) { return; } - this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg)); + await this.sendMessageToClientAndWait(source as WebSocketLike, wrapSessionMessage(msg)); }, onBinaryMessage: (frame) => { if (!connection) { @@ -1460,6 +1500,7 @@ export class VoiceAssistantWebSocketServer { const base: SessionConnectionBase = { session, principalId: admission.principalId, + hubExecutionAgents: admission.hubExecutionAgents, sessionKey: sessionConnectionKey(admission.principalId, clientId), clientId, appVersion, @@ -1481,6 +1522,7 @@ export class VoiceAssistantWebSocketServer { getConnectedHostBufferedAmount: () => { let total = 0; for (const socket of this.sessions.keys()) { + if (socket instanceof VirtualHostSocket) continue; if (typeof socket.bufferedAmount !== "number") return null; total += socket.bufferedAmount; } @@ -1790,6 +1832,10 @@ export class VoiceAssistantWebSocketServer { clearTimeout(existing.externalDisconnectCleanupTimeout); existing.externalDisconnectCleanupTimeout = null; } + for (const socket of existing.sockets) { + this.hostLinks.get(socket)?.close("permission_denied"); + this.hostLinks.delete(socket); + } existing.permissionCeiling = message.sessionPermissions; existing.session.setPermissions(pending.admission!.permissions); existing.session.setRouted(message.routed === true); @@ -1853,6 +1899,7 @@ export class VoiceAssistantWebSocketServer { ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}), features: { connectedHosts: true, + hostMultiplexing: true, usageSources: true, ownedSubscriptions: true, agentRequestReceipts: true, @@ -2096,6 +2143,9 @@ export class VoiceAssistantWebSocketServer { error?: Error; }, ): Promise { + this.hostLinks.get(ws)?.close(); + this.hostLinks.delete(ws); + this.applicationSocketLease.release(ws); const identity = this.socketIdentities.get(ws); const identityFields = identity ? toConnectionLogFields(identity) : {}; @@ -2290,6 +2340,12 @@ export class VoiceAssistantWebSocketServer { return false; } const decodedFrame = decodeBinaryFrame(asBytes); + if (asBytes[0] === HOST_ROUTE_OPCODE && !decodedFrame) { + ws.close(1008, "Invalid host routing frame"); + return true; + } + if (decodedFrame?.kind === "host_route" && this.hostLinks.get(ws)?.binary(decodedFrame.frame)) + return true; if (!decodedFrame) { return false; } @@ -2444,11 +2500,38 @@ export class VoiceAssistantWebSocketServer { } } + private hostLink(ws: WebSocketLike, connection: SessionConnection): DaemonLinkEndpoint { + let link = this.hostLinks.get(ws); + if (!link) { + link = new DaemonLinkEndpoint({ + identity: JSON.stringify([connection.principalId, connection.clientId]), + permissions: () => connection.session.getPermissions(), + routed: () => connection.session.isRouted(), + bufferedAmount: () => ws.bufferedAmount ?? null, + delivery: connection.session.delivery, + admit: (socket, permissions) => + this.attachSocket(socket, undefined, undefined, false, { + principalId: connection.principalId, + permissions, + }), + }); + this.hostLinks.set(ws, link); + } + return link; + } + private async dispatchSessionMessage( ws: WebSocketLike, activeConnection: SessionConnection, message: Extract, ): Promise { + const link = this.hostLink(ws, activeConnection); + if (link.handles(message.message)) { + await activeConnection.session.delivery.request(ws, message.message, () => + link.dispatch(message.message), + ); + return; + } this.recordInboundSessionRequestType(message.message.type); const controlRpc = getControlRpcLogInfo(message.message); if (controlRpc) {