puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-22 01:02:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(hosts): share daemon links across isolated routed sessions

Browse Source 
Multiplex logical sessions over the monitored daemon connection and wrap original binary bytes in routing headers. Admit channels under their authenticated parent and current grants; preserve scoped delivery and physical send completion.
This commit is contained in:
Mohamed Boudra
2026-10-21 10:33:12 +02:00
parent 34185def78
commit 09c2a24846
23 changed files with 1384 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/connected-hosts.md
+23 -13
View File
@@ -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:<B principal>:<A serverId>:<A producer>`, 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
packages/client/src/daemon-client.ts
+65 -6
View File
@@ -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<string> {
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 } : {}),
},
{
packages/client/src/hosts/index.ts
+20 -23
View File
@@ -14,23 +14,13 @@ export interface PaseoHostActions {
client(serverId: string): Promise<PaseoApi>;
}
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<string>;
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,
packages/client/src/node/buffering.ts
+1 -1
View File
@@ -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;
}
packages/client/src/node/connector.test.ts
+21
View File
@@ -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<void>((resolve) => transport.onOpen(resolve));
const received = new Promise<Uint8Array>((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);
});
packages/client/src/node/index.ts
+10 -11
View File
@@ -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<DaemonClientConfig, "url" | "transportFactory">,
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),
});
}
packages/protocol/src/binary-frames/demux.test.ts
+38
View File
@@ -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");
});
});
packages/protocol/src/binary-frames/demux.ts
+6
View File
@@ -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:
packages/protocol/src/binary-frames/host-route.ts
+33
View File
@@ -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;
}
}
packages/protocol/src/binary-frames/index.ts
+1
View File
@@ -1,3 +1,4 @@
export * from "./demux.js";
export * from "./file-transfer.js";
export * from "./terminal.js";
export * from "./host-route.js";
packages/protocol/src/hosts.ts
+14 -3
View File
@@ -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() }),
});
packages/protocol/src/messages.ts
+5
View File
@@ -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,
packages/server/src/server/authorization/operation-permissions.ts
+2
View File
@@ -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,
packages/server/src/server/connected-hosts/hosts.e2e.test.ts
+546 -31
View File
@@ -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<void>((resolve) => wss.close(() => resolve()));
await new Promise<void>((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<void>((resolve) => wss.close(() => resolve()));
await new Promise<void>((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<void>((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<void>((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();
});
packages/server/src/server/connected-hosts/index.ts
+80 -71
View File
@@ -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<typeof setTimeout> | 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");
}
packages/server/src/server/connected-hosts/link.ts
+210
View File
@@ -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<void>,
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<void> {
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<string, { scopeId: string; socket: VirtualHostSocket }>();
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<void>;
},
) {}
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<void> {
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();
}
}
packages/server/src/server/connected-hosts/session.ts
+26 -8
View File
@@ -1,3 +1,4 @@
import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
import type {
SessionInboundMessage,
SessionOutboundMessage,
@@ -13,7 +14,7 @@ type HostRequest<T extends SessionInboundMessage["type"]> = Extract<
interface SessionRoute {
scopeId: string;
source: object;
send(data: string, binary: boolean): void;
send(data: string | Uint8Array): void;
release(): Promise<void>;
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,
});
packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts
+63 -15
View File
@@ -281,6 +281,7 @@ async function waitForCapturedLog(
let remote: PaseoApi | undefined;
let sibling: PaseoApi | undefined;
let terminalRoute: Awaited<ReturnType<typeof openConnectedHostTerminal>> | undefined;
let wire: Awaited<ReturnType<typeof relayWire>> | 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<net.Socket>();
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<void>((resolve) => server.close(() => resolve()));
},
};
}
packages/server/src/server/session.test.ts
+39
View File
@@ -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<void>((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 });
}
});
packages/server/src/server/session.ts
+29 -9
View File
@@ -445,7 +445,7 @@ export interface SessionOptions {
appVersion?: string | null;
clientCapabilities?: Record<string, unknown> | null;
onMessage: (msg: SessionOutboundMessage) => void;
onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void;
onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void | Promise<void>;
onBinaryMessage?: (frame: Uint8Array) => void;
onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
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<void>)
| 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<void> {
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<void>): Promise<void> {
let pending: Promise<void>;
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);
packages/server/src/server/session/owned-subscriptions/index.test.ts
+37
View File
@@ -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();
});
packages/server/src/server/session/owned-subscriptions/index.ts
+23 -4
View File
@@ -51,6 +51,8 @@ export interface OwnedSubscription {
readonly signal: AbortSignal;
emit(message: SessionOutboundMessage): void;
emitBinary(frame: Uint8Array): void;
emitAndWait(message: SessionOutboundMessage): Promise<void>;
emitBinaryAndWait(frame: Uint8Array): Promise<void>;
release(): Promise<void>;
}
@@ -67,8 +69,14 @@ export class SessionDelivery {
private closing?: Promise<void>;
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<void>,
private readonly sendBinary: (
source: object,
frame: Uint8Array,
) => void | Promise<void> = () => {},
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<void> {
packages/server/src/server/websocket-server.ts
+92 -9
View File
@@ -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<void>;
onBinaryMessage?: (frame: Uint8Array) => void;
onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
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<WebSocketLike, DaemonLinkEndpoint>();
private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
private readonly socketIdentities: Map<WebSocketLike, WebSocketConnectionIdentity> = 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<WebSocketLike>,
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<void> {
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<void> {
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<WSInboundMessage, { type: "session" }>,
): Promise<void> {
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) {