From 50cde731372b0aa3a2742fbee86fbd69b19775a8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 21 Oct 2026 13:33:56 +0200 Subject: [PATCH] Extract relay endpoint security boundary from transport and admission Move the daemon-side handshake, record framing, and handshake timeout out of relay-transport.ts into websocket/secure-relay-ingress.ts, and give both endpoints one secure-channel entrypoint in @getpaseo/relay/e2ee. Relay transport now attaches plaintext sockets; bootstrap secures them before admission. The client selects endpoint security from configuration instead of inferring it from the relay URL. Behavior-preserving: the legacy construction is unchanged and no new security guarantee is claimed. --- .../src/daemon-client-relay-e2ee-transport.ts | 26 +-- .../src/daemon-client-transport.test.ts | 8 +- .../client/src/daemon-client-transport.ts | 1 - packages/client/src/daemon-client.ts | 11 +- packages/relay/src/e2ee.ts | 9 + packages/relay/src/secure-channel.ts | 84 ++++++++ packages/server/src/server/bootstrap.ts | 20 +- packages/server/src/server/relay-runtime.ts | 3 - .../server/src/server/relay-transport.test.ts | 107 ---------- packages/server/src/server/relay-transport.ts | 170 +-------------- .../server/src/server/websocket-server.ts | 6 + .../websocket/encrypted-relay-socket.test.ts | 4 - .../websocket/encrypted-relay-socket.ts | 3 - .../websocket/secure-relay-ingress.test.ts | 188 +++++++++++++++++ .../server/websocket/secure-relay-ingress.ts | 194 ++++++++++++++++++ 15 files changed, 517 insertions(+), 317 deletions(-) create mode 100644 packages/relay/src/secure-channel.ts create mode 100644 packages/server/src/server/websocket/secure-relay-ingress.test.ts create mode 100644 packages/server/src/server/websocket/secure-relay-ingress.ts diff --git a/packages/client/src/daemon-client-relay-e2ee-transport.ts b/packages/client/src/daemon-client-relay-e2ee-transport.ts index 237699e600..343bdbd116 100644 --- a/packages/client/src/daemon-client-relay-e2ee-transport.ts +++ b/packages/client/src/daemon-client-relay-e2ee-transport.ts @@ -1,7 +1,7 @@ import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { - createClientChannel, - type EncryptedChannel, + openClientSecureChannel, + type SecureChannel, type Transport as RelayTransport, } from "@getpaseo/relay/e2ee"; import type { @@ -37,20 +37,13 @@ export function createRelayE2eeTransportFactory(args: { }; } -export function createRelayTransportFactory( - baseFactory: DaemonTransportFactory, -): DaemonTransportFactory { - // The relay upgrade precedes E2EE. Only the encrypted hello may carry daemon credentials. - return ({ url }) => baseFactory({ url }); -} - export function createEncryptedTransport( base: DaemonTransport, daemonPublicKeyB64: string, logger: TransportLogger, assertOutboundCapacity: (wireBytes: number) => void = () => {}, ): DaemonTransport { - let channel: EncryptedChannel | null = null; + let channel: SecureChannel | null = null; let opened = false; let closed = false; @@ -113,11 +106,14 @@ export function createEncryptedTransport( const startHandshake = async () => { try { - channel = await createClientChannel(relayTransport, daemonPublicKeyB64, { - onopen: emitOpen, - onmessage: (data) => emitMessage(data), - onclose: (code, reason) => emitClose({ code, reason }), - onerror: (error) => emitError(error), + channel = await openClientSecureChannel(relayTransport, { + daemonPublicKeyB64, + events: { + onopen: emitOpen, + onmessage: (data) => emitMessage(data), + onclose: (code, reason) => emitClose({ code, reason }), + onerror: (error) => emitError(error), + }, }); } catch (error) { logger.warn({ err: normalizeTransportError(error) }, "relay_e2ee_handshake_failed"); diff --git a/packages/client/src/daemon-client-transport.test.ts b/packages/client/src/daemon-client-transport.test.ts index 6f80717dd1..27f5caf513 100644 --- a/packages/client/src/daemon-client-transport.test.ts +++ b/packages/client/src/daemon-client-transport.test.ts @@ -10,16 +10,16 @@ import { extractRelayMessage, } from "./daemon-client-transport.js"; -const createClientChannelMock = vi.hoisted(() => vi.fn()); +const openClientSecureChannelMock = vi.hoisted(() => vi.fn()); vi.mock("@getpaseo/relay/e2ee", () => ({ - createClientChannel: createClientChannelMock, + openClientSecureChannel: openClientSecureChannelMock, })); describe("daemon-client transport helpers", () => { test("createEncryptedTransport closes handshake failures with browser-safe code", async () => { - createClientChannelMock.mockReset(); - createClientChannelMock.mockRejectedValueOnce(new Error("handshake failed")); + openClientSecureChannelMock.mockReset(); + openClientSecureChannelMock.mockRejectedValueOnce(new Error("handshake failed")); const connection: { open: (() => void) | null } = { open: null }; const close = vi.fn(); diff --git a/packages/client/src/daemon-client-transport.ts b/packages/client/src/daemon-client-transport.ts index 01933fd1ec..05bd08bbc6 100644 --- a/packages/client/src/daemon-client-transport.ts +++ b/packages/client/src/daemon-client-transport.ts @@ -17,7 +17,6 @@ export { export { createEncryptedTransport, createRelayE2eeTransportFactory, - createRelayTransportFactory, } from "./daemon-client-relay-e2ee-transport.js"; export { bindWsHandler, diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 0741b08147..563d1ee151 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -162,7 +162,6 @@ import { } from "@getpaseo/protocol/binary-frames/index"; import { createRelayE2eeTransportFactory, - createRelayTransportFactory, createWebSocketTransportFactory, decodeMessageData, defaultWebSocketFactory, @@ -1448,14 +1447,12 @@ export class DaemonClient { this.config.transportFactory ?? createWebSocketTransportFactory(this.config.webSocketFactory ?? defaultWebSocketFactory); const isRelayTransport = isRelayClientWebSocketUrl(this.config.url); - const shouldUseRelayE2ee = this.config.e2ee?.enabled === true && isRelayTransport; this.assertEncryptedRelayAuth(selected.helloAuth, isRelayTransport); - let transportFactory = isRelayTransport - ? createRelayTransportFactory(baseTransportFactory) - : baseTransportFactory; - if (shouldUseRelayE2ee) { - const daemonPublicKeyB64 = this.config.e2ee?.daemonPublicKeyB64; + // Endpoint security is selected by configuration, never inferred from the URL. + let transportFactory = baseTransportFactory; + if (this.config.e2ee?.enabled === true) { + const daemonPublicKeyB64 = this.config.e2ee.daemonPublicKeyB64; if (!daemonPublicKeyB64) { throw new Error("daemonPublicKeyB64 is required for relay E2EE"); } diff --git a/packages/relay/src/e2ee.ts b/packages/relay/src/e2ee.ts index c3a3b852fb..af59369701 100644 --- a/packages/relay/src/e2ee.ts +++ b/packages/relay/src/e2ee.ts @@ -1,3 +1,12 @@ +export { openClientSecureChannel, openDaemonSecureChannel } from "./secure-channel.js"; +export type { + ClientSecureChannelOptions, + DaemonSecureChannelOptions, + SecureChannel, + SecureChannelEvents, + SecurityMode, +} from "./secure-channel.js"; + export { createClientChannel, createDaemonChannel, EncryptedChannel } from "./encrypted-channel.js"; export type { Transport, TransportMessage, EncryptedChannelEvents } from "./encrypted-channel.js"; diff --git a/packages/relay/src/secure-channel.ts b/packages/relay/src/secure-channel.ts new file mode 100644 index 0000000000..e6017c032b --- /dev/null +++ b/packages/relay/src/secure-channel.ts @@ -0,0 +1,84 @@ +/// +/** + * Endpoint security for relay-carried daemon sessions. + * + * This is the one entrypoint both endpoints use to secure a transport. It owns + * the handshake, record protection, retries, termination, and the negotiated + * security mode. Relay adapters own routing and physical I/O; daemon admission + * owns credentials and authorization. + * + * "legacy" is the pre-existing ephemeral-static box construction in + * `encrypted-channel.ts`. It has no replay protection. + */ + +import { + createClientChannel, + createDaemonChannel, + type EncryptedChannelEvents, + type Transport, +} from "./encrypted-channel.js"; +import type { KeyPair } from "./crypto.js"; + +export type SecurityMode = "legacy"; + +export type SecureChannelEvents = EncryptedChannelEvents; + +export interface SecureChannel { + send(data: string | ArrayBuffer): Promise; + close(code?: number, reason?: string): void; + isOpen(): boolean; + securityMode(): SecurityMode; + /** Bytes the transport will carry for `data`, including protection overhead. */ + outboundWireByteLength(data: string | ArrayBuffer): number; + onTransitionToOpen(cb: () => void): void; + onClose(cb: () => void): void; +} + +export interface ClientSecureChannelOptions { + daemonPublicKeyB64: string; + events?: SecureChannelEvents; +} + +export interface DaemonSecureChannelOptions { + daemonKeyPair: KeyPair; + events?: SecureChannelEvents; +} + +/** + * Secures a transport as the initiator. Resolves once the handshake has been + * started; `events.onopen` fires when the channel is usable. + */ +export async function openClientSecureChannel( + transport: Transport, + options: ClientSecureChannelOptions, +): Promise { + const channel = await createClientChannel(transport, options.daemonPublicKeyB64, options.events); + return withSecurityMode(channel, "legacy"); +} + +/** + * Secures a transport as the responder. Resolves once the peer's handshake has + * been accepted and the channel is open. + */ +export async function openDaemonSecureChannel( + transport: Transport, + options: DaemonSecureChannelOptions, +): Promise { + const channel = await createDaemonChannel(transport, options.daemonKeyPair, options.events); + return withSecurityMode(channel, "legacy"); +} + +function withSecurityMode( + channel: Omit, + mode: SecurityMode, +): SecureChannel { + return { + send: (data) => channel.send(data), + close: (code, reason) => channel.close(code, reason), + isOpen: () => channel.isOpen(), + securityMode: () => mode, + outboundWireByteLength: (data) => channel.outboundWireByteLength(data), + onTransitionToOpen: (cb) => channel.onTransitionToOpen(cb), + onClose: (cb) => channel.onClose(cb), + }; +} diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 5125a0b923..d24cd997f0 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -172,6 +172,7 @@ import { createConfiguredTerminalManager } from "../terminal/terminal-manager-fa import { applyTerminalAgentHookSetting } from "../terminal/agent-hooks/terminal-agent-hook-setting.js"; import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js"; import { createRelayRuntime, type RelayRuntime } from "./relay-runtime.js"; +import { admitSecuredRelaySocket } from "./websocket/secure-relay-ingress.js"; import type { PushNotificationSender } from "./push/index.js"; import { getOrCreateServerId } from "./server-id.js"; import { resolveDaemonVersion } from "./daemon-version.js"; @@ -1761,6 +1762,7 @@ export async function createPaseoDaemon( await pluginRuntime.start(); providerSnapshotManager.settlePluginProviders(); wsServer.beginAcceptingConnections(); + const relaySecurityLogger = logger.child({ module: "relay-security" }); relayRuntime = createRelayRuntime({ config: { enabled: relayEnabled, @@ -1770,12 +1772,20 @@ export async function createPaseoDaemon( publicUseTls: relayPublicUseTls, }, logger, - attachSocket: async (ws, metadata) => { - if (!wsServer) throw new Error("WebSocket server is not ready"); - await wsServer.attachExternalSocket(ws, metadata); - }, + attachSocket: (ws, metadata) => + admitSecuredRelaySocket( + ws, + { daemonKeyPair: daemonKeyPair.keyPair, logger: relaySecurityLogger }, + async (secured) => { + if (!wsServer) throw new Error("WebSocket server is not ready"); + await wsServer.attachExternalSocket(secured.socket, { + ...metadata, + transport: "relay", + securityMode: secured.securityMode, + }); + }, + ), serverId, - daemonKeyPair: daemonKeyPair.keyPair, }); daemonConfigStore.onFieldChange("relay.enabled", (value) => { relayRuntime?.setEnabled(value === true); diff --git a/packages/server/src/server/relay-runtime.ts b/packages/server/src/server/relay-runtime.ts index e9f2ccf158..885aaa9b6e 100644 --- a/packages/server/src/server/relay-runtime.ts +++ b/packages/server/src/server/relay-runtime.ts @@ -1,5 +1,4 @@ import type pino from "pino"; -import type { KeyPair } from "@getpaseo/relay/e2ee"; import type { ExternalSocketMetadata } from "./websocket-server.js"; import { startRelayTransport, @@ -20,7 +19,6 @@ interface RelayRuntimeOptions { logger: pino.Logger; attachSocket(ws: RelaySocketLike, metadata?: ExternalSocketMetadata): Promise; serverId: string; - daemonKeyPair: KeyPair; startTransport?: typeof startRelayTransport; } @@ -43,7 +41,6 @@ export function createRelayRuntime(options: RelayRuntimeOptions): RelayRuntime { relayEndpoint: config.endpoint, relayUseTls: config.useTls, serverId: options.serverId, - daemonKeyPair: options.daemonKeyPair, }); } diff --git a/packages/server/src/server/relay-transport.test.ts b/packages/server/src/server/relay-transport.test.ts index 2dcf29b948..26868e7112 100644 --- a/packages/server/src/server/relay-transport.test.ts +++ b/packages/server/src/server/relay-transport.test.ts @@ -1,7 +1,5 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import pino from "pino"; -import { createClientChannel, type Transport } from "@getpaseo/relay/e2ee"; -import { exportPublicKey, generateKeyPair } from "@getpaseo/relay"; import { startRelayTransport } from "./relay-transport"; function createMockLogger() { @@ -146,37 +144,6 @@ describe("relay-transport control lifecycle", () => { vi.useRealTimers(); }); - test("releases a stalled encrypted data connection without restarting control", async () => { - vi.useFakeTimers(); - let attachments = 0; - controllers.push( - startRelayTransport({ - logger: pino({ level: "silent" }), - attachSocket: async () => { - attachments += 1; - }, - relayEndpoint: "relay.paseo.sh:443", - relayUseTls: true, - serverId: "srv_test", - daemonKeyPair: generateKeyPair(), - createWebSocket: relay.createWebSocket, - }), - ); - const control = relay.sockets[0]; - control.open(); - const connected = JSON.stringify({ type: "connected", connectionId: "stalled" }); - control.message(connected); - const stalled = relay.sockets[1]; - stalled.open(); - await vi.advanceTimersByTimeAsync(15_000); - expect(stalled.terminateCalls).toBe(1); - expect(attachments).toBe(0); - expect(control.terminateCalls).toBe(0); - control.message(connected); - expect(relay.sockets).toHaveLength(3); - expect(relay.sockets[2].url).toBe(stalled.url); - }); - test("logs relay_control_connected only after first valid control message", () => { const logger = createMockLogger(); const controller = startRelayTransport({ @@ -286,80 +253,6 @@ describe("relay-transport control lifecycle", () => { ]); }); - test("encrypted sends wait for the physical data socket callback", async () => { - const logger = createMockLogger(); - const daemonKeyPair = generateKeyPair(); - let resolveAttached: ((socket: unknown) => void) | undefined; - const attached = new Promise((resolve) => { - resolveAttached = resolve; - }); - const controller = startRelayTransport({ - logger: logger as unknown as pino.Logger, - attachSocket: async (socket) => resolveAttached?.(socket), - relayEndpoint: "relay.paseo.sh:443", - relayUseTls: true, - serverId: "srv_test", - daemonKeyPair, - createWebSocket: relay.createWebSocket, - }); - controllers.push(controller); - - const control = relay.sockets[0]; - control.open(); - control.message(JSON.stringify({ type: "sync", connectionIds: [] }), false); - control.message(JSON.stringify({ type: "connected", connectionId: "clt_test" }), false); - - const dataSocket = relay.sockets[1]; - dataSocket.deferSendCompletion = true; - dataSocket.open(); - let clientTransport: Transport; - clientTransport = { - send: (data) => dataSocket.message(data, data instanceof ArrayBuffer), - close: () => undefined, - onmessage: null, - onclose: null, - onerror: null, - }; - dataSocket.onSend = (data) => { - clientTransport.onmessage?.({ - data: data instanceof Uint8Array ? data.slice().buffer : data, - isBinary: data instanceof ArrayBuffer || data instanceof Uint8Array, - }); - }; - let resolveClientOpen: (() => void) | undefined; - const clientOpen = new Promise((resolve) => { - resolveClientOpen = resolve; - }); - await createClientChannel(clientTransport, exportPublicKey(daemonKeyPair.publicKey), { - onopen: () => resolveClientOpen?.(), - }); - - let attachedCompleted = false; - void attached.then(() => { - attachedCompleted = true; - return undefined; - }); - await clientOpen; - await Promise.resolve(); - expect(attachedCompleted).toBe(false); - dataSocket.completeNextSend(); - const encryptedSocket = (await attached) as { - send: (data: Uint8Array) => void | Promise; - }; - let completed = false; - - const sending = Promise.resolve(encryptedSocket.send(new Uint8Array([1, 2, 3]))).then(() => { - completed = true; - return undefined; - }); - await Promise.resolve(); - expect(completed).toBe(false); - - dataSocket.completeNextSend(); - await sending; - expect(completed).toBe(true); - }); - test("uses relayUseTls for control and data socket URLs", () => { const logger = createMockLogger(); const controller = startRelayTransport({ diff --git a/packages/server/src/server/relay-transport.ts b/packages/server/src/server/relay-transport.ts index 113cce605c..08fcd826d5 100644 --- a/packages/server/src/server/relay-transport.ts +++ b/packages/server/src/server/relay-transport.ts @@ -1,15 +1,8 @@ /// -import { EventEmitter } from "node:events"; import { WebSocket } from "ws"; import type pino from "pino"; -import { - createDaemonChannel, - type Transport as RelayTransport, - type KeyPair, -} from "@getpaseo/relay/e2ee"; import { buildRelayWebSocketUrl } from "@getpaseo/protocol/daemon-endpoints"; import type { ExternalSocketMetadata } from "./websocket-server.js"; -import { createEncryptedRelaySocket } from "./websocket/encrypted-relay-socket.js"; export interface RelayTransportOptions { logger: pino.Logger; @@ -17,7 +10,6 @@ export interface RelayTransportOptions { relayEndpoint: string; // "host:port" relayUseTls: boolean; serverId: string; - daemonKeyPair?: KeyPair; createWebSocket?: RelayWebSocketFactory; } @@ -30,13 +22,12 @@ export interface RelaySocketLike { bufferedAmount?: number; send: (data: string | Uint8Array | ArrayBuffer, callback?: (error?: Error) => void) => void; close: (code?: number, reason?: string) => void; - terminate?: () => void; + terminate: () => void; on: (event: "message" | "close" | "error", listener: (...args: unknown[]) => void) => void; once: (event: "close" | "error", listener: (...args: unknown[]) => void) => void; } interface RelayWebSocketLike extends RelaySocketLike { - terminate: () => void; ping: () => void; on: ( event: "open" | "message" | "close" | "error" | "pong", @@ -118,7 +109,6 @@ export function startRelayTransport({ relayEndpoint, relayUseTls, serverId, - daemonKeyPair, createWebSocket = createDefaultRelayWebSocket, }: RelayTransportOptions): RelayTransportController { const relayLogger = logger.child({ module: "relay-transport" }); @@ -385,17 +375,7 @@ export function startRelayTransport({ externalSessionKey: `session:${connectionId}`, relayConnectionId: connectionId, }; - if (daemonKeyPair) { - void attachEncryptedSocket( - socket, - daemonKeyPair, - relayLogger.child({ connectionId }), - attachSocket, - externalMetadata, - ); - } else { - void attachSocket(socket, externalMetadata); - } + void attachSocket(socket, externalMetadata); }); socket.on("close", (code, reason) => { @@ -418,149 +398,3 @@ export function startRelayTransport({ return { stop }; } - -async function attachEncryptedSocket( - socket: RelayWebSocketLike, - daemonKeyPair: KeyPair, - logger: pino.Logger, - attachSocket: (ws: RelaySocketLike, metadata?: ExternalSocketMetadata) => Promise, - metadata?: ExternalSocketMetadata, -): Promise { - const handshakeTimeout = setTimeout(() => { - logger.warn("relay_e2ee_handshake_timeout_terminating"); - socket.terminate(); - }, 15_000); - try { - const relayTransport = createRelayTransportAdapter(socket, logger); - const emitter = new EventEmitter(); - const pendingMessages: Array = []; - let attached = false; - const emitMessage = (data: string | ArrayBuffer) => { - if (attached) { - emitter.emit("message", data); - return; - } - pendingMessages.push(data); - }; - const channel = await createDaemonChannel(relayTransport, daemonKeyPair, { - onmessage: emitMessage, - onclose: (code, reason) => emitter.emit("close", code, reason), - onerror: (error) => { - logger.warn({ err: error }, "relay_e2ee_error"); - emitter.emit("error", error); - }, - }); - clearTimeout(handshakeTimeout); - const encryptedSocket = createEncryptedRelaySocket({ - channel, - emitter, - getTransportBufferedAmount: () => socket.bufferedAmount, - terminateTransport: () => socket.terminate(), - }); - await attachSocket(encryptedSocket, metadata); - attached = true; - for (const message of pendingMessages) { - emitter.emit("message", message); - } - pendingMessages.length = 0; - } catch (error) { - logger.warn({ err: error }, "relay_e2ee_handshake_failed"); - try { - socket.close(1011, "E2EE handshake failed"); - } catch { - // ignore - } - } finally { - clearTimeout(handshakeTimeout); - } -} - -function createRelayTransportAdapter( - socket: RelayWebSocketLike, - logger: pino.Logger, -): RelayTransport { - const relayTransport: RelayTransport = { - send: (data) => - new Promise((resolve, reject) => { - try { - socket.send(data, (error) => { - if (!error) { - resolve(); - return; - } - logger.warn({ err: error }, "relay_socket_send_failed"); - reject(error); - }); - } catch (error) { - const err = error instanceof Error ? error : new Error(String(error)); - logger.warn({ err }, "relay_socket_send_failed"); - reject(err); - } - }), - close: (code?: number, reason?: string) => socket.close(code, reason), - onmessage: null, - onclose: null, - onerror: null, - }; - - socket.on("message", (data, isBinary) => { - const binary = isBinary === true; - relayTransport.onmessage?.({ data: normalizeMessageData(data, binary), isBinary: binary }); - }); - socket.on("close", (code, reason) => { - const closeCode = typeof code === "number" ? code : 1006; - relayTransport.onclose?.(closeCode, String(reason ?? "")); - }); - socket.on("error", (err) => { - relayTransport.onerror?.(err instanceof Error ? err : new Error(String(err))); - }); - - return relayTransport; -} - -function normalizeMessageData(data: unknown, isBinary: boolean): string | ArrayBuffer { - if (!isBinary) { - if (typeof data === "string") return data; - const buffer = bufferFromWsData(data); - if (buffer) return buffer.toString("utf8"); - return String(data); - } - - if (data instanceof ArrayBuffer) return data; - - const buffer = bufferFromWsData(data); - if (buffer) { - const view = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); - const out = new Uint8Array(view.byteLength); - out.set(view); - return out.buffer; - } - - return String(data); -} - -function bufferFromWsData(data: unknown): Buffer | null { - if (Buffer.isBuffer(data)) return data; - if (Array.isArray(data)) { - const buffers: Buffer[] = []; - for (const part of data) { - if (Buffer.isBuffer(part)) { - buffers.push(part); - } else if (part instanceof ArrayBuffer) { - buffers.push(Buffer.from(part)); - } else if (ArrayBuffer.isView(part)) { - buffers.push(Buffer.from(part.buffer, part.byteOffset, part.byteLength)); - } else if (typeof part === "string") { - buffers.push(Buffer.from(part, "utf8")); - } else { - return null; - } - } - return Buffer.concat(buffers); - } - if (data instanceof ArrayBuffer) return Buffer.from(data); - if (ArrayBuffer.isView(data)) { - return Buffer.from(data.buffer, data.byteOffset, data.byteLength); - } - return null; -} diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 8f36e1a2b8..0200fc4f87 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -103,6 +103,7 @@ import { } from "./lifecycle-reasons.js"; import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; +import type { SecurityMode } from "@getpaseo/relay/e2ee"; import type { BrowserToolsBroker } from "./browser-tools/broker.js"; import type { DaemonRuntimeConfig } from "./session/daemon/daemon-session.js"; import { DirectorySyncService } from "./directory-sync/index.js"; @@ -124,6 +125,8 @@ export interface ExternalSocketMetadata { transport: "relay" | "hub"; externalSessionKey?: string; relayConnectionId?: string; + /** Negotiated endpoint security for relay-carried sockets. */ + securityMode?: SecurityMode; hubDaemonId?: string; } @@ -151,6 +154,7 @@ interface WebSocketConnectionIdentity { userAgent?: string; remoteAddress?: string; relayConnectionId?: string; + securityMode?: SecurityMode; hubDaemonId?: string; clientId?: string; sessionId?: string; @@ -2944,6 +2948,7 @@ function createWebSocketConnectionIdentity( ...(requestMetadata.userAgent ? { userAgent: requestMetadata.userAgent } : {}), ...(requestMetadata.remoteAddress ? { remoteAddress: requestMetadata.remoteAddress } : {}), ...(metadata?.relayConnectionId ? { relayConnectionId: metadata.relayConnectionId } : {}), + ...(metadata?.securityMode ? { securityMode: metadata.securityMode } : {}), ...(metadata?.hubDaemonId ? { hubDaemonId: metadata.hubDaemonId } : {}), }; } @@ -2962,6 +2967,7 @@ function toConnectionLogFields(identity: WebSocketConnectionIdentity): Record = []; private resolveSend: (() => void) | null = null; - setState(state: "open"): void { - expect(state).toBe("open"); - } - send(data: string | ArrayBuffer): Promise { this.sent.push(data); return new Promise((resolve) => { diff --git a/packages/server/src/server/websocket/encrypted-relay-socket.ts b/packages/server/src/server/websocket/encrypted-relay-socket.ts index 3b76a49711..f11162761c 100644 --- a/packages/server/src/server/websocket/encrypted-relay-socket.ts +++ b/packages/server/src/server/websocket/encrypted-relay-socket.ts @@ -2,7 +2,6 @@ import { EventEmitter } from "node:events"; import { MAX_PHYSICAL_SOCKET_BUFFERED_BYTES } from "./physical-socket.js"; export interface EncryptedRelayChannel { - setState: (state: "open") => void; send: (data: string | ArrayBuffer) => Promise; outboundWireByteLength: (data: string | ArrayBuffer) => number; close: (code?: number, reason?: string) => void; @@ -27,8 +26,6 @@ export function createEncryptedRelaySocket(params: { const { channel, emitter, getTransportBufferedAmount, terminateTransport } = params; let readyState = 1; - channel.setState("open"); - const terminate = () => { if (readyState === 3) return; readyState = 3; diff --git a/packages/server/src/server/websocket/secure-relay-ingress.test.ts b/packages/server/src/server/websocket/secure-relay-ingress.test.ts new file mode 100644 index 0000000000..f0df25b202 --- /dev/null +++ b/packages/server/src/server/websocket/secure-relay-ingress.test.ts @@ -0,0 +1,188 @@ +import { afterEach, expect, test, vi } from "vitest"; +import pino from "pino"; +import { exportPublicKey, generateKeyPair, openClientSecureChannel } from "@getpaseo/relay/e2ee"; +import type { Transport } from "@getpaseo/relay/e2ee"; +import { admitSecuredRelaySocket, type SecuredRelaySocket } from "./secure-relay-ingress.js"; + +type Listener = (...args: unknown[]) => void; + +class FakeIngressSocket { + readyState = 1; + bufferedAmount = 0; + terminateCalls = 0; + closes: Array<{ code?: number; reason?: string }> = []; + deferSendCompletion = false; + onSend: ((data: string | Uint8Array | ArrayBuffer) => void) | null = null; + private pendingSendCallbacks: Array<(error?: Error) => void> = []; + private listeners = new Map(); + + on(event: "message" | "close" | "error", listener: Listener) { + this.listeners.set(event, [...(this.listeners.get(event) ?? []), listener]); + } + + send(data: string | Uint8Array | ArrayBuffer, callback?: (error?: Error) => void) { + this.onSend?.(data); + if (!callback) return; + if (this.deferSendCompletion) { + this.pendingSendCallbacks.push(callback); + return; + } + callback(); + } + + completeNextSend() { + this.pendingSendCallbacks.shift()?.(); + } + + close(code?: number, reason?: string) { + this.closes.push({ code, reason }); + this.readyState = 3; + } + + terminate() { + this.terminateCalls += 1; + this.readyState = 3; + this.emit("close", 1006, ""); + } + + message(data: string | ArrayBuffer) { + this.emit("message", data, data instanceof ArrayBuffer); + } + + private emit(event: string, ...args: unknown[]) { + for (const listener of this.listeners.get(event) ?? []) listener(...args); + } +} + +const logger = pino({ level: "silent" }); + +/** Wires a client secure channel to the fake socket so both ends talk directly. */ +async function openClientAgainst( + socket: FakeIngressSocket, + daemonPublicKeyB64: string, + events: { onopen?: () => void; onmessage?: (data: string | ArrayBuffer) => void } = {}, +) { + const clientTransport: Transport = { + send: (data) => socket.message(data), + close: () => undefined, + onmessage: null, + onclose: null, + onerror: null, + }; + socket.onSend = (data) => { + clientTransport.onmessage?.({ + data: data instanceof Uint8Array ? data.slice().buffer : data, + isBinary: data instanceof ArrayBuffer || data instanceof Uint8Array, + }); + }; + return openClientSecureChannel(clientTransport, { daemonPublicKeyB64, events }); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +test("terminates a relay socket whose peer never completes the handshake", async () => { + vi.useFakeTimers(); + const socket = new FakeIngressSocket(); + let attachments = 0; + const admitted = admitSecuredRelaySocket( + socket, + { daemonKeyPair: generateKeyPair(), logger }, + async () => { + attachments += 1; + }, + ); + await vi.advanceTimersByTimeAsync(15_000); + await admitted; + expect(socket.terminateCalls).toBe(1); + expect(attachments).toBe(0); +}); + +test("admits the secured socket with its negotiated security mode", async () => { + const socket = new FakeIngressSocket(); + const daemonKeyPair = generateKeyPair(); + let secured: SecuredRelaySocket | null = null; + const admitted = admitSecuredRelaySocket(socket, { daemonKeyPair, logger }, async (next) => { + secured = next; + }); + await openClientAgainst(socket, exportPublicKey(daemonKeyPair.publicKey)); + await admitted; + expect(secured).not.toBeNull(); + expect(secured!.securityMode).toBe("legacy"); + expect(secured!.socket.readyState).toBe(1); +}); + +test("delivers records received before admission completes, in order, after attach", async () => { + const socket = new FakeIngressSocket(); + const daemonKeyPair = generateKeyPair(); + let releaseAttach: (() => void) | undefined; + const attachGate = new Promise((resolve) => { + releaseAttach = resolve; + }); + const received: unknown[] = []; + let secured: SecuredRelaySocket | null = null; + const admitted = admitSecuredRelaySocket(socket, { daemonKeyPair, logger }, async (next) => { + secured = next; + next.socket.on("message", (data) => received.push(data)); + await attachGate; + }); + let clientOpen: () => void = () => undefined; + const opened = new Promise((resolve) => { + clientOpen = resolve; + }); + const client = await openClientAgainst(socket, exportPublicKey(daemonKeyPair.publicKey), { + onopen: () => clientOpen(), + }); + await opened; + await vi.waitFor(() => expect(secured).not.toBeNull()); + await client.send("first"); + await client.send("second"); + expect(received).toEqual([]); + releaseAttach?.(); + await admitted; + expect(received).toEqual(["first", "second"]); +}); + +test("encrypted sends wait for the physical socket send callback", async () => { + const socket = new FakeIngressSocket(); + const daemonKeyPair = generateKeyPair(); + socket.deferSendCompletion = true; + let resolveAttached: ((secured: SecuredRelaySocket) => void) | undefined; + const attached = new Promise((resolve) => { + resolveAttached = resolve; + }); + void admitSecuredRelaySocket(socket, { daemonKeyPair, logger }, async (next) => + resolveAttached?.(next), + ); + let resolveClientOpen: (() => void) | undefined; + const clientOpen = new Promise((resolve) => { + resolveClientOpen = resolve; + }); + await openClientAgainst(socket, exportPublicKey(daemonKeyPair.publicKey), { + onopen: () => resolveClientOpen?.(), + }); + + let attachedCompleted = false; + void attached.then(() => { + attachedCompleted = true; + return undefined; + }); + await clientOpen; + await Promise.resolve(); + expect(attachedCompleted).toBe(false); + socket.completeNextSend(); + const { socket: encryptedSocket } = await attached; + + let completed = false; + const sending = Promise.resolve(encryptedSocket.send(new Uint8Array([1, 2, 3]))).then(() => { + completed = true; + return undefined; + }); + await Promise.resolve(); + expect(completed).toBe(false); + + socket.completeNextSend(); + await sending; + expect(completed).toBe(true); +}); diff --git a/packages/server/src/server/websocket/secure-relay-ingress.ts b/packages/server/src/server/websocket/secure-relay-ingress.ts new file mode 100644 index 0000000000..fce633c469 --- /dev/null +++ b/packages/server/src/server/websocket/secure-relay-ingress.ts @@ -0,0 +1,194 @@ +/// +/** + * Endpoint security boundary for relay-carried daemon sessions. + * + * A relay data socket arrives here as plaintext bytes. This module runs the + * shared secure channel over it and hands the secured socket to session + * admission, together with the negotiated security mode. Relay routing lives in + * `relay-transport.ts`; credentials and authorization live in admission. + */ +import { EventEmitter } from "node:events"; +import type pino from "pino"; +import { + openDaemonSecureChannel, + type KeyPair, + type SecurityMode, + type Transport as RelayTransport, +} from "@getpaseo/relay/e2ee"; +import { createEncryptedRelaySocket, type EncryptedRelaySocket } from "./encrypted-relay-socket.js"; + +const HANDSHAKE_TIMEOUT_MS = 15_000; + +export interface RelayIngressSocket { + readyState: number; + bufferedAmount?: number; + send: (data: string | Uint8Array | ArrayBuffer, callback?: (error?: Error) => void) => void; + close: (code?: number, reason?: string) => void; + terminate: () => void; + on: (event: "message" | "close" | "error", listener: (...args: unknown[]) => void) => void; +} + +export interface SecuredRelaySocket { + socket: EncryptedRelaySocket; + securityMode: SecurityMode; +} + +export interface SecureRelayIngressOptions { + daemonKeyPair: KeyPair; + logger: pino.Logger; +} + +/** + * Secures `socket` and admits it through `attach`. Messages the peer sends + * before `attach` resolves are delivered once it has. + * + * A peer that never completes the handshake is terminated after + * {@link HANDSHAKE_TIMEOUT_MS}; a failed handshake closes the socket. + */ +export async function admitSecuredRelaySocket( + socket: RelayIngressSocket, + options: SecureRelayIngressOptions, + attach: (secured: SecuredRelaySocket) => Promise, +): Promise { + const { logger } = options; + const handshakeTimeout = setTimeout(() => { + logger.warn("relay_e2ee_handshake_timeout_terminating"); + socket.terminate(); + }, HANDSHAKE_TIMEOUT_MS); + try { + const relayTransport = createRelayTransportAdapter(socket, logger); + const emitter = new EventEmitter(); + const pendingMessages: Array = []; + let attached = false; + const emitMessage = (data: string | ArrayBuffer) => { + if (attached) { + emitter.emit("message", data); + return; + } + pendingMessages.push(data); + }; + const channel = await openDaemonSecureChannel(relayTransport, { + daemonKeyPair: options.daemonKeyPair, + events: { + onmessage: emitMessage, + onclose: (code, reason) => emitter.emit("close", code, reason), + onerror: (error) => { + logger.warn({ err: error }, "relay_e2ee_error"); + emitter.emit("error", error); + }, + }, + }); + clearTimeout(handshakeTimeout); + const encryptedSocket = createEncryptedRelaySocket({ + channel, + emitter, + getTransportBufferedAmount: () => socket.bufferedAmount, + terminateTransport: () => socket.terminate(), + }); + await attach({ socket: encryptedSocket, securityMode: channel.securityMode() }); + attached = true; + for (const message of pendingMessages) { + emitter.emit("message", message); + } + pendingMessages.length = 0; + } catch (error) { + logger.warn({ err: error }, "relay_e2ee_handshake_failed"); + try { + socket.close(1011, "E2EE handshake failed"); + } catch { + // ignore + } + } finally { + clearTimeout(handshakeTimeout); + } +} + +function createRelayTransportAdapter( + socket: RelayIngressSocket, + logger: pino.Logger, +): RelayTransport { + const relayTransport: RelayTransport = { + send: (data) => + new Promise((resolve, reject) => { + try { + socket.send(data, (error) => { + if (!error) { + resolve(); + return; + } + logger.warn({ err: error }, "relay_socket_send_failed"); + reject(error); + }); + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + logger.warn({ err }, "relay_socket_send_failed"); + reject(err); + } + }), + close: (code?: number, reason?: string) => socket.close(code, reason), + onmessage: null, + onclose: null, + onerror: null, + }; + + socket.on("message", (data, isBinary) => { + const binary = isBinary === true; + relayTransport.onmessage?.({ data: normalizeMessageData(data, binary), isBinary: binary }); + }); + socket.on("close", (code, reason) => { + const closeCode = typeof code === "number" ? code : 1006; + relayTransport.onclose?.(closeCode, String(reason ?? "")); + }); + socket.on("error", (err) => { + relayTransport.onerror?.(err instanceof Error ? err : new Error(String(err))); + }); + + return relayTransport; +} + +function normalizeMessageData(data: unknown, isBinary: boolean): string | ArrayBuffer { + if (!isBinary) { + if (typeof data === "string") return data; + const buffer = bufferFromWsData(data); + if (buffer) return buffer.toString("utf8"); + return String(data); + } + + if (data instanceof ArrayBuffer) return data; + + const buffer = bufferFromWsData(data); + if (buffer) { + const view = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); + const out = new Uint8Array(view.byteLength); + out.set(view); + return out.buffer; + } + + return String(data); +} + +function bufferFromWsData(data: unknown): Buffer | null { + if (Buffer.isBuffer(data)) return data; + if (Array.isArray(data)) { + const buffers: Buffer[] = []; + for (const part of data) { + if (Buffer.isBuffer(part)) { + buffers.push(part); + } else if (part instanceof ArrayBuffer) { + buffers.push(Buffer.from(part)); + } else if (ArrayBuffer.isView(part)) { + buffers.push(Buffer.from(part.buffer, part.byteOffset, part.byteLength)); + } else if (typeof part === "string") { + buffers.push(Buffer.from(part, "utf8")); + } else { + return null; + } + } + return Buffer.concat(buffers); + } + if (data instanceof ArrayBuffer) return Buffer.from(data); + if (ArrayBuffer.isView(data)) { + return Buffer.from(data.buffer, data.byteOffset, data.byteLength); + } + return null; +}