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;
+}