feat: implement end-to-end encryption for relay connections

This commit is contained in:
Mohamed Boudra
2026-02-03 13:40:44 +07:00
parent a7a39ffd01
commit 6ed70e24c7
23 changed files with 1004 additions and 115 deletions
+1
View File
@@ -184,6 +184,7 @@ export default async function globalSetup() {
process.env.E2E_DAEMON_PORT = String(port);
process.env.E2E_RELAY_PORT = String(relayPort);
process.env.E2E_RELAY_SESSION_ID = offer.sessionId;
process.env.E2E_RELAY_DAEMON_PUBLIC_KEY = offer.daemonPublicKeyB64;
process.env.E2E_METRO_PORT = String(metroPort);
console.log(`[e2e] Test daemon started on port ${port}, Metro on port ${metroPort}, home: ${paseoHome}`);
@@ -3,8 +3,11 @@ import { test, expect } from './fixtures';
test('connects via relay when direct endpoints fail', async ({ page }) => {
const relayPort = process.env.E2E_RELAY_PORT;
const sessionId = process.env.E2E_RELAY_SESSION_ID;
if (!relayPort || !sessionId) {
throw new Error('E2E_RELAY_PORT or E2E_RELAY_SESSION_ID is not set (expected from globalSetup).');
const daemonPublicKeyB64 = process.env.E2E_RELAY_DAEMON_PUBLIC_KEY;
if (!relayPort || !sessionId || !daemonPublicKeyB64) {
throw new Error(
'E2E_RELAY_PORT, E2E_RELAY_SESSION_ID, or E2E_RELAY_DAEMON_PUBLIC_KEY is not set (expected from globalSetup).'
);
}
const nowIso = new Date().toISOString();
@@ -14,6 +17,7 @@ test('connects via relay when direct endpoints fail', async ({ page }) => {
id: 'relay-only-daemon',
label: 'relay-daemon',
endpoints: [relayEndpoint],
daemonPublicKeyB64,
relay: { endpoint: relayEndpoint, sessionId },
metadata: null,
createdAt: nowIso,
+2 -1
View File
@@ -5,6 +5,7 @@ const path = require("path");
const projectRoot = __dirname;
const serverSrcRoot = path.resolve(projectRoot, "../server/src");
const relaySrcRoot = path.resolve(projectRoot, "../relay/src");
const config = getDefaultConfig(projectRoot);
const defaultResolveRequest = config.resolver.resolveRequest ?? resolve;
@@ -13,7 +14,7 @@ config.resolver.resolveRequest = (context, moduleName, platform) => {
const origin = context.originModulePath;
if (
origin &&
origin.startsWith(serverSrcRoot) &&
(origin.startsWith(serverSrcRoot) || origin.startsWith(relaySrcRoot)) &&
moduleName.endsWith(".js")
) {
const tsModuleName = moduleName.replace(/\.js$/, ".ts");
+3
View File
@@ -41,8 +41,10 @@
"@react-navigation/bottom-tabs": "^7.4.0",
"@react-navigation/elements": "^2.6.3",
"@react-navigation/native": "^7.1.8",
"@sphereon/isomorphic-webcrypto": "^2.5.0-rn-crypto.2",
"@tanstack/react-query": "^5.90.11",
"@tauri-apps/api": "^2.9.1",
"base64-js": "^1.5.1",
"buffer": "^6.0.3",
"expo": "^54.0.18",
"expo-audio": "~1.0.13",
@@ -85,6 +87,7 @@
"react-native-reanimated": "~4.1.1",
"react-native-safe-area-context": "~5.6.0",
"react-native-screens": "~4.16.0",
"react-native-securerandom": "^1.0.1",
"react-native-svg": "^15.14.0",
"react-native-unistyles": "^3.0.15",
"react-native-web": "~0.21.0",
@@ -10,6 +10,7 @@ import { useEffect, useMemo, useRef, useState } from "react";
function buildCandidateUrls(daemon: HostProfile): string[] {
const endpoints = daemon.endpoints ?? [];
const sessionId = daemon.relay?.sessionId ?? null;
const relayEndpoint = daemon.relay?.endpoint ?? null;
const lastKnownGood =
typeof daemon.metadata?.lastKnownGoodEndpoint === "string"
? (daemon.metadata.lastKnownGoodEndpoint as string)
@@ -20,21 +21,24 @@ function buildCandidateUrls(daemon: HostProfile): string[] {
if (!out.includes(url)) out.push(url);
};
if (lastKnownGood) {
const isLastKnownRelay = !!relayEndpoint && lastKnownGood === relayEndpoint;
const directEndpoints = relayEndpoint
? endpoints.filter((endpoint) => endpoint !== relayEndpoint)
: endpoints;
if (lastKnownGood && !isLastKnownRelay) {
push(buildDaemonWebSocketUrl(lastKnownGood));
}
for (const endpoint of endpoints) {
for (const endpoint of directEndpoints) {
push(buildDaemonWebSocketUrl(endpoint));
}
if (sessionId) {
if (lastKnownGood) {
push(buildRelayWebSocketUrl({ endpoint: lastKnownGood, sessionId }));
}
for (const endpoint of endpoints) {
push(buildRelayWebSocketUrl({ endpoint, sessionId }));
if (sessionId && relayEndpoint) {
if (isLastKnownRelay) {
push(buildRelayWebSocketUrl({ endpoint: relayEndpoint, sessionId }));
}
push(buildRelayWebSocketUrl({ endpoint: relayEndpoint, sessionId }));
}
return out;
@@ -123,7 +127,12 @@ function ManagedDaemonSession({ daemon }: { daemon: HostProfile }) {
]);
return (
<SessionProvider key={`${daemon.id}:${activeUrl}`} serverUrl={activeUrl} serverId={daemon.id}>
<SessionProvider
key={`${daemon.id}:${activeUrl}`}
serverUrl={activeUrl}
serverId={daemon.id}
daemonPublicKeyB64={daemon.daemonPublicKeyB64}
>
{null}
</SessionProvider>
);
@@ -326,6 +326,7 @@ interface SessionProviderProps {
children: ReactNode;
serverUrl: string;
serverId: string;
daemonPublicKeyB64?: string;
}
// SessionProvider: Daemon client message handler that updates Zustand store
@@ -333,8 +334,9 @@ export function SessionProvider({
children,
serverUrl,
serverId,
daemonPublicKeyB64,
}: SessionProviderProps) {
const client = useDaemonClient(serverUrl);
const client = useDaemonClient(serverUrl, { daemonPublicKeyB64 });
const [connectionSnapshot, setConnectionSnapshot] =
useState<DaemonConnectionSnapshot>(() =>
mapConnectionState(client.getConnectionState(), client.lastError)
+27 -2
View File
@@ -9,19 +9,44 @@ function runDaemonRequest(label: string, promise: Promise<unknown>): void {
});
}
export function useDaemonClient(url: string): DaemonClientV2 {
type DaemonClientOptions = {
daemonPublicKeyB64?: string;
};
function isRelayWebSocketUrl(url: string): boolean {
try {
const parsed = new URL(url);
return parsed.searchParams.get("role") === "client" && parsed.searchParams.has("session");
} catch {
return false;
}
}
export function useDaemonClient(
url: string,
options: DaemonClientOptions = {}
): DaemonClientV2 {
const client = useMemo(
() => {
const tauriTransportFactory = createTauriWebSocketTransportFactory();
const relayConnection = isRelayWebSocketUrl(url);
return new DaemonClientV2({
url,
suppressSendErrors: true,
...(tauriTransportFactory
? { transportFactory: tauriTransportFactory }
: {}),
...(relayConnection
? {
e2ee: {
enabled: true,
daemonPublicKeyB64: options.daemonPublicKeyB64,
},
}
: {}),
});
},
[url]
[options.daemonPublicKeyB64, url]
);
useEffect(() => {
+32 -4
View File
@@ -7,16 +7,44 @@ declare global {
}
export function polyfillCrypto(): void {
if (typeof globalThis.crypto === "undefined") {
(globalThis as any).crypto = {};
let webcrypto: Crypto | null = null;
try {
// Prefer the React Native entrypoint to avoid the browser shim.
// eslint-disable-next-line @typescript-eslint/no-var-requires
const rnModule = require("@sphereon/isomorphic-webcrypto/src/react-native");
webcrypto = (rnModule?.default ?? rnModule) as Crypto;
} catch {
// eslint-disable-next-line @typescript-eslint/no-var-requires
const fallbackModule = require("@sphereon/isomorphic-webcrypto");
webcrypto = (fallbackModule?.default ?? fallbackModule) as Crypto;
}
if (typeof globalThis.crypto.randomUUID !== "function") {
const existing = (globalThis as any).crypto as Crypto | null | undefined;
let target = existing;
if (!target || typeof (target as Crypto).subtle === "undefined") {
target = (webcrypto && typeof webcrypto === "object" ? (webcrypto as Crypto) : undefined) ?? ({} as Crypto);
(globalThis as any).crypto = target;
}
const ensureSecure = (globalThis as any).crypto?.ensureSecure as
| (() => Promise<void>)
| undefined;
if (typeof ensureSecure === "function") {
void ensureSecure();
}
if (typeof (globalThis as any).crypto?.randomUUID !== "function") {
if (!globalThis.crypto) {
(globalThis as any).crypto = {} as Crypto;
}
globalThis.crypto.randomUUID = () =>
ExpoCrypto.randomUUID() as `${string}-${string}-${string}-${string}-${string}`;
}
if (typeof globalThis.crypto.getRandomValues !== "function") {
if (typeof (globalThis as any).crypto?.getRandomValues !== "function") {
if (!globalThis.crypto) {
(globalThis as any).crypto = {} as Crypto;
}
globalThis.crypto.getRandomValues = <T extends ArrayBufferView>(array: T): T => {
return ExpoCrypto.getRandomValues(array as any) as T;
};
+2
View File
@@ -14,6 +14,8 @@
"test:watch": "vitest"
},
"dependencies": {
"@sphereon/isomorphic-webcrypto": "^2.5.0-rn-crypto.2",
"base64-js": "^1.5.1",
"ws": "^8.14.2"
},
"devDependencies": {
+12
View File
@@ -0,0 +1,12 @@
import { fromByteArray, toByteArray } from "base64-js";
export function arrayBufferToBase64(buffer: ArrayBuffer): string {
return fromByteArray(new Uint8Array(buffer));
}
export function base64ToArrayBuffer(base64: string): ArrayBuffer {
const bytes = toByteArray(base64);
const out = new Uint8Array(bytes.byteLength);
out.set(bytes);
return out.buffer;
}
+3 -20
View File
@@ -1,3 +1,4 @@
/// <reference lib="dom" />
/**
* E2EE crypto primitives using WebCrypto API.
*
@@ -6,6 +7,8 @@
* - AES-256-GCM for authenticated encryption
*/
import { arrayBufferToBase64, base64ToArrayBuffer } from "./base64.js";
const ECDH_ALGORITHM = { name: "ECDH", namedCurve: "P-256" };
const AES_ALGORITHM = { name: "AES-GCM", length: 256 };
const IV_LENGTH = 12;
@@ -127,23 +130,3 @@ export async function decrypt(
return plaintext;
}
}
// --- Helpers ---
function arrayBufferToBase64(buffer: ArrayBuffer): string {
const bytes = new Uint8Array(buffer);
let binary = "";
for (let i = 0; i < bytes.byteLength; i++) {
binary += String.fromCharCode(bytes[i]);
}
return btoa(binary);
}
function base64ToArrayBuffer(base64: string): ArrayBuffer {
const binary = atob(base64);
const bytes = new Uint8Array(binary.length);
for (let i = 0; i < binary.length; i++) {
bytes[i] = binary.charCodeAt(i);
}
return bytes.buffer;
}
+2 -20
View File
@@ -1,3 +1,4 @@
/// <reference lib="dom" />
/**
* Encrypted channel that wraps a WebSocket-like transport.
*
@@ -13,6 +14,7 @@ import {
encrypt,
decrypt,
} from "./crypto.js";
import { arrayBufferToBase64, base64ToArrayBuffer } from "./base64.js";
export interface Transport {
send(data: string | ArrayBuffer): void;
@@ -193,23 +195,3 @@ export class EncryptedChannel {
return this.state === "open";
}
}
// --- Helpers ---
function arrayBufferToBase64(buffer: ArrayBuffer): string {
const bytes = new Uint8Array(buffer);
let binary = "";
for (let i = 0; i < bytes.byteLength; i++) {
binary += String.fromCharCode(bytes[i]);
}
return btoa(binary);
}
function base64ToArrayBuffer(base64: string): ArrayBuffer {
const binary = atob(base64);
const bytes = new Uint8Array(binary.length);
for (let i = 0; i < binary.length; i++) {
bytes[i] = binary.charCodeAt(i);
}
return bytes.buffer;
}
+218 -1
View File
@@ -53,6 +53,11 @@ import type {
AgentSessionConfig,
} from "../server/agent/agent-sdk-types.js";
import { getAgentProviderDefinition } from "../server/agent/provider-manifest.js";
import {
createClientChannel,
type EncryptedChannel,
type Transport as RelayTransport,
} from "@paseo/relay";
export interface Logger {
debug(obj: object, msg?: string): void;
@@ -146,6 +151,10 @@ export type DaemonClientV2Config = {
transportFactory?: DaemonTransportFactory;
webSocketFactory?: WebSocketFactory;
logger?: Logger;
e2ee?: {
enabled?: boolean;
daemonPublicKeyB64?: string;
};
reconnect?: {
enabled?: boolean;
baseDelayMs?: number;
@@ -305,11 +314,19 @@ export class DaemonClientV2 {
try {
this.cleanupTransport();
const transportFactory =
const baseTransportFactory =
this.config.transportFactory ??
createWebSocketTransportFactory(
this.config.webSocketFactory ?? defaultWebSocketFactory
);
const transportFactory =
this.config.e2ee?.enabled === true
? createEncryptedTransportFactory(
baseTransportFactory,
this.config.e2ee?.daemonPublicKeyB64,
this.logger
)
: baseTransportFactory;
const transport = transportFactory({ url: this.config.url, headers });
this.transport = transport;
@@ -2325,6 +2342,206 @@ function createWebSocketTransportFactory(
};
}
function createEncryptedTransportFactory(
baseFactory: DaemonTransportFactory,
daemonPublicKeyB64: string | undefined,
logger: Logger
): DaemonTransportFactory {
if (!daemonPublicKeyB64) {
throw new Error("daemonPublicKeyB64 is required for relay E2EE");
}
return ({ url, headers }) => {
const base = baseFactory({ url, headers });
return createEncryptedTransport(base, daemonPublicKeyB64, logger);
};
}
function createEncryptedTransport(
base: DaemonTransport,
daemonPublicKeyB64: string,
logger: Logger
): DaemonTransport {
let channel: EncryptedChannel | null = null;
let opened = false;
let closed = false;
const openHandlers = new Set<() => void>();
const closeHandlers = new Set<(event?: unknown) => void>();
const errorHandlers = new Set<(event?: unknown) => void>();
const messageHandlers = new Set<(data: unknown) => void>();
const emitOpen = () => {
if (opened || closed) return;
opened = true;
for (const handler of openHandlers) {
try {
handler();
} catch {
// no-op
}
}
};
const emitClose = (event?: unknown) => {
if (closed) return;
closed = true;
for (const handler of closeHandlers) {
try {
handler(event);
} catch {
// no-op
}
}
};
const emitError = (event?: unknown) => {
if (closed) return;
for (const handler of errorHandlers) {
try {
handler(event);
} catch {
// no-op
}
}
};
const emitMessage = (data: unknown) => {
if (closed) return;
for (const handler of messageHandlers) {
try {
handler(data);
} catch {
// no-op
}
}
};
const relayTransport: RelayTransport = {
send: (data) => {
if (typeof data === "string") {
base.send(data);
return;
}
if (data instanceof ArrayBuffer) {
if (typeof TextDecoder !== "undefined") {
base.send(new TextDecoder().decode(data));
return;
}
if (typeof Buffer !== "undefined") {
base.send(Buffer.from(data).toString("utf8"));
return;
}
base.send(String(data));
return;
}
base.send(String(data));
},
close: (code?: number, reason?: string) => base.close(code, reason),
onmessage: null,
onclose: null,
onerror: null,
};
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),
});
} catch (error) {
logger.warn({ err: error }, "relay_e2ee_handshake_failed");
emitError(error);
base.close(1011, "E2EE handshake failed");
}
};
base.onOpen(() => {
void startHandshake();
});
base.onMessage((event) => {
relayTransport.onmessage?.(extractRelayMessageData(event));
});
base.onClose((event) => {
const record = event as { code?: number; reason?: string } | undefined;
relayTransport.onclose?.(record?.code ?? 0, record?.reason ?? "");
emitClose(event);
});
base.onError((event) => {
relayTransport.onerror?.(
event instanceof Error ? event : new Error(String(event))
);
emitError(event);
});
return {
send: (data) => {
if (!channel) {
throw new Error("Encrypted channel not ready");
}
void channel.send(data).catch((error) => {
emitError(error);
});
},
close: (code?: number, reason?: string) => {
if (channel) {
channel.close(code, reason);
} else {
base.close(code, reason);
}
emitClose({ code, reason });
},
onMessage: (handler) => {
messageHandlers.add(handler);
return () => messageHandlers.delete(handler);
},
onOpen: (handler) => {
openHandlers.add(handler);
if (opened) {
try {
handler();
} catch {
// no-op
}
}
return () => openHandlers.delete(handler);
},
onClose: (handler) => {
closeHandlers.add(handler);
if (closed) {
try {
handler();
} catch {
// no-op
}
}
return () => closeHandlers.delete(handler);
},
onError: (handler) => {
errorHandlers.add(handler);
return () => errorHandlers.delete(handler);
},
};
}
function extractRelayMessageData(event: unknown): string | ArrayBuffer {
const raw =
event && typeof event === "object" && "data" in event
? (event as { data: unknown }).data
: event;
if (typeof raw === "string") return raw;
if (raw instanceof ArrayBuffer) return raw;
if (ArrayBuffer.isView(raw)) {
const view = new Uint8Array(raw.buffer, raw.byteOffset, raw.byteLength);
const out = new Uint8Array(view.byteLength);
out.set(view);
return out.buffer;
}
return String(raw ?? "");
}
function bindWsHandler(
ws: WebSocketLike,
event: "open" | "close" | "error" | "message",
+4
View File
@@ -53,6 +53,7 @@ import {
createConnectionOfferV1,
encodeOfferToFragmentUrl,
} from "./connection-offer.js";
import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js";
import { printPairingQrIfEnabled } from "./pairing-qr.js";
import { startRelayTransport, type RelayTransportController } from "./relay-transport.js";
import type {
@@ -102,6 +103,7 @@ export async function createPaseoDaemon(
): Promise<PaseoDaemon> {
const logger = rootLogger.child({ module: "bootstrap" });
const connectionSessionId = randomUUID();
const daemonKeyPair = await loadOrCreateDaemonKeyPair(config.paseoHome, logger);
let relayTransport: RelayTransportController | null = null;
const agentMcpRoute = config.agentMcpRoute;
@@ -442,6 +444,7 @@ export async function createPaseoDaemon(
const offer = await createConnectionOfferV1({
sessionId: connectionSessionId,
endpoints,
daemonPublicKeyB64: daemonKeyPair.publicKeyB64,
});
const url = encodeOfferToFragmentUrl({ offer, appBaseUrl });
@@ -455,6 +458,7 @@ export async function createPaseoDaemon(
attachSocket: (ws) => wsServer.attachExternalSocket(ws),
relayEndpoint,
sessionId: connectionSessionId,
daemonKeyPair: daemonKeyPair.keyPair,
});
}
} else {
+2 -15
View File
@@ -1,5 +1,4 @@
import os from "node:os";
import { webcrypto } from "node:crypto";
import { z } from "zod";
export const ConnectionOfferV1Schema = z.object({
@@ -52,14 +51,13 @@ export function buildOfferEndpoints({
export async function createConnectionOfferV1(args: {
sessionId: string;
endpoints: string[];
daemonPublicKeyB64: string;
}): Promise<ConnectionOfferV1> {
const daemonPublicKeyB64 = await generateDaemonPublicKeyB64();
return ConnectionOfferV1Schema.parse({
v: 1,
sessionId: args.sessionId,
endpoints: args.endpoints,
daemonPublicKeyB64,
daemonPublicKeyB64: args.daemonPublicKeyB64,
});
}
@@ -90,17 +88,6 @@ function getPrimaryLanIp(): string | null {
return null;
}
async function generateDaemonPublicKeyB64(): Promise<string> {
const keyPair = await webcrypto.subtle.generateKey(
{ name: "ECDH", namedCurve: "P-256" },
true,
["deriveBits"]
);
const raw = await webcrypto.subtle.exportKey("raw", keyPair.publicKey);
return Buffer.from(new Uint8Array(raw)).toString("base64");
}
function dedupePreserveOrder(values: string[]): string[] {
const out: string[] = [];
const seen = new Set<string>();
@@ -5,6 +5,7 @@ import { Writable } from "node:stream";
import net from "node:net";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { createClientChannel, type Transport } from "@paseo/relay";
import { createRelayServer } from "@paseo/relay/node";
function createCapturingLogger() {
@@ -34,7 +35,10 @@ function parseOfferUrlFromLogs(lines: string[]): string {
throw new Error(`pairing_offer log not found. saw ${lines.length} lines`);
}
function decodeOfferFromFragmentUrl(url: string): { sessionId: string } {
function decodeOfferFromFragmentUrl(url: string): {
sessionId: string;
daemonPublicKeyB64: string;
} {
const marker = "#offer=";
const idx = url.indexOf(marker);
if (idx === -1) {
@@ -42,9 +46,10 @@ function decodeOfferFromFragmentUrl(url: string): { sessionId: string } {
}
const encoded = url.slice(idx + marker.length);
const json = Buffer.from(encoded, "base64url").toString("utf8");
const offer = JSON.parse(json) as { sessionId?: string };
const offer = JSON.parse(json) as { sessionId?: string; daemonPublicKeyB64?: string };
if (!offer.sessionId) throw new Error("offer.sessionId missing");
return { sessionId: offer.sessionId };
if (!offer.daemonPublicKeyB64) throw new Error("offer.daemonPublicKeyB64 missing");
return { sessionId: offer.sessionId, daemonPublicKeyB64: offer.daemonPublicKeyB64 };
}
async function getAvailablePort(): Promise<number> {
@@ -62,7 +67,7 @@ async function getAvailablePort(): Promise<number> {
});
}
describe("Relay transport (plaintext) - daemon E2E", () => {
describe("Relay transport (E2EE) - daemon E2E", () => {
test(
"daemon connects to relay and client ping/pong works through relay",
async () => {
@@ -82,7 +87,7 @@ describe("Relay transport (plaintext) - daemon E2E", () => {
try {
const offerUrl = parseOfferUrlFromLogs(lines);
const { sessionId } = decodeOfferFromFragmentUrl(offerUrl);
const { sessionId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
const ws = new WebSocket(
`ws://127.0.0.1:${relayPort}/ws?session=${encodeURIComponent(
@@ -96,24 +101,53 @@ describe("Relay transport (plaintext) - daemon E2E", () => {
reject(new Error("timed out waiting for pong"));
}, 20000);
ws.on("open", () => {
ws.send(JSON.stringify({ type: "ping" }));
});
const transport: Transport = {
send: (data) => ws.send(data),
close: (code?: number, reason?: string) => ws.close(code, reason),
onmessage: null,
onclose: null,
onerror: null,
};
ws.on("message", (data) => {
clearTimeout(timeout);
try {
resolve(JSON.parse(data.toString()));
} catch (err) {
reject(err);
} finally {
ws.close();
}
transport.onmessage?.(typeof data === "string" ? data : data.toString());
});
ws.on("close", (code, reason) => {
transport.onclose?.(code, reason.toString());
});
ws.on("error", (err) => {
transport.onerror?.(err);
});
ws.on("error", (err) => {
clearTimeout(timeout);
reject(err);
ws.on("open", async () => {
try {
const channel = await createClientChannel(
transport,
daemonPublicKeyB64,
{
onmessage: (data) => {
clearTimeout(timeout);
try {
const payload =
typeof data === "string" ? JSON.parse(data) : data;
resolve(payload);
} catch (err) {
reject(err);
} finally {
ws.close();
}
},
onerror: (err) => {
clearTimeout(timeout);
reject(err);
},
}
);
await channel.send(JSON.stringify({ type: "ping" }));
} catch (err) {
clearTimeout(timeout);
reject(err);
}
});
});
@@ -0,0 +1,78 @@
/// <reference lib="dom" />
import { webcrypto } from "node:crypto";
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import path from "node:path";
import { z } from "zod";
const KeyPairSchema = z.object({
v: z.literal(1),
publicKeyJwk: z.record(z.any()),
privateKeyJwk: z.record(z.any()),
});
type StoredKeyPair = z.infer<typeof KeyPairSchema>;
type LoggerLike = {
child(bindings: Record<string, unknown>): LoggerLike;
info(...args: any[]): void;
warn(...args: any[]): void;
};
const KEYPAIR_FILENAME = "daemon-keypair.json";
const ECDH_ALGORITHM = { name: "ECDH", namedCurve: "P-256" };
export type DaemonKeyPairBundle = {
keyPair: CryptoKeyPair;
publicKeyB64: string;
};
export async function loadOrCreateDaemonKeyPair(
paseoHome: string,
logger?: LoggerLike
): Promise<DaemonKeyPairBundle> {
const log = logger?.child({ module: "daemon-keypair" });
const filePath = path.join(paseoHome, KEYPAIR_FILENAME);
if (existsSync(filePath)) {
try {
const raw = readFileSync(filePath, "utf8");
const parsed = KeyPairSchema.parse(JSON.parse(raw)) as StoredKeyPair;
const [publicKey, privateKey] = await Promise.all([
webcrypto.subtle.importKey("jwk", parsed.publicKeyJwk, ECDH_ALGORITHM, true, []),
webcrypto.subtle.importKey("jwk", parsed.privateKeyJwk, ECDH_ALGORITHM, true, [
"deriveBits",
]),
]);
const publicKeyB64 = await exportPublicKeyB64(publicKey);
log?.info({ filePath }, "Loaded daemon keypair");
return { keyPair: { publicKey, privateKey }, publicKeyB64 };
} catch (error) {
log?.warn({ err: error, filePath }, "Failed to load daemon keypair, regenerating");
}
}
const keyPair = (await webcrypto.subtle.generateKey(ECDH_ALGORITHM, true, [
"deriveBits",
])) as CryptoKeyPair;
const [publicKeyJwk, privateKeyJwk, publicKeyB64] = await Promise.all([
webcrypto.subtle.exportKey("jwk", keyPair.publicKey),
webcrypto.subtle.exportKey("jwk", keyPair.privateKey),
exportPublicKeyB64(keyPair.publicKey),
]);
const payload: StoredKeyPair = {
v: 1,
publicKeyJwk: publicKeyJwk as Record<string, unknown>,
privateKeyJwk: privateKeyJwk as Record<string, unknown>,
};
writeFileSync(filePath, JSON.stringify(payload, null, 2) + "\n");
log?.info({ filePath }, "Saved daemon keypair");
return { keyPair, publicKeyB64 };
}
async function exportPublicKeyB64(publicKey: CryptoKey): Promise<string> {
const raw = await webcrypto.subtle.exportKey("raw", publicKey);
return Buffer.from(new Uint8Array(raw)).toString("base64");
}
+130 -2
View File
@@ -1,22 +1,39 @@
/// <reference lib="dom" />
import { EventEmitter } from "node:events";
import WebSocket from "ws";
import type pino from "pino";
import {
createDaemonChannel,
type EncryptedChannel,
type Transport as RelayTransport,
} from "@paseo/relay";
type RelayTransportOptions = {
logger: pino.Logger;
attachSocket: (ws: WebSocket) => Promise<void>;
attachSocket: (ws: RelaySocketLike) => Promise<void>;
relayEndpoint: string; // "host:port"
sessionId: string;
daemonKeyPair?: CryptoKeyPair;
};
export type RelayTransportController = {
stop: () => Promise<void>;
};
type RelaySocketLike = {
readyState: number;
send: (data: string) => void;
close: (code?: number, reason?: string) => void;
on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void;
once: (event: "close" | "error", listener: (...args: any[]) => void) => void;
};
export function startRelayTransport({
logger,
attachSocket,
relayEndpoint,
sessionId,
daemonKeyPair,
}: RelayTransportOptions): RelayTransportController {
const relayLogger = logger.child({ module: "relay-transport" });
@@ -56,7 +73,11 @@ export function startRelayTransport({
if (attached) return;
attached = true;
void attachSocket(socket);
if (daemonKeyPair) {
void attachEncryptedSocket(socket, daemonKeyPair, relayLogger, attachSocket);
} else {
void attachSocket(socket);
}
});
socket.on("close", (code, reason) => {
@@ -90,6 +111,113 @@ export function startRelayTransport({
return { stop };
}
async function attachEncryptedSocket(
socket: WebSocket,
daemonKeyPair: CryptoKeyPair,
logger: pino.Logger,
attachSocket: (ws: RelaySocketLike) => Promise<void>
): Promise<void> {
try {
const relayTransport = createRelayTransportAdapter(socket);
const emitter = new EventEmitter();
const channel = await createDaemonChannel(relayTransport, daemonKeyPair, {
onmessage: (data) => emitter.emit("message", data),
onclose: (code, reason) => emitter.emit("close", code, reason),
onerror: (error) => {
logger.warn({ err: error }, "relay_e2ee_error");
emitter.emit("error", error);
},
});
const encryptedSocket = createEncryptedSocket(channel, emitter);
await attachSocket(encryptedSocket);
} catch (error) {
logger.warn({ err: error }, "relay_e2ee_handshake_failed");
try {
socket.close(1011, "E2EE handshake failed");
} catch {
// ignore
}
}
}
function createRelayTransportAdapter(socket: WebSocket): RelayTransport {
const relayTransport: RelayTransport = {
send: (data) => socket.send(data),
close: (code?: number, reason?: string) => socket.close(code, reason),
onmessage: null,
onclose: null,
onerror: null,
};
socket.on("message", (data) => {
relayTransport.onmessage?.(normalizeMessageData(data));
});
socket.on("close", (code, reason) => {
relayTransport.onclose?.(code, reason.toString());
});
socket.on("error", (err) => {
relayTransport.onerror?.(err instanceof Error ? err : new Error(String(err)));
});
return relayTransport;
}
function createEncryptedSocket(
channel: EncryptedChannel,
emitter: EventEmitter
): RelaySocketLike {
let readyState = 1;
channel.setState("open");
const close = (code?: number, reason?: string) => {
if (readyState === 3) return;
readyState = 3;
channel.close(code, reason);
};
emitter.on("close", () => {
if (readyState === 3) return;
readyState = 3;
});
return {
get readyState() {
return readyState;
},
send: (data) => {
void channel.send(data).catch((error) => {
emitter.emit("error", error);
});
},
close,
on: (event, listener) => {
emitter.on(event, listener);
},
once: (event, listener) => {
emitter.once(event, listener);
},
};
}
function normalizeMessageData(data: unknown): string | ArrayBuffer {
if (typeof data === "string") return data;
if (data instanceof ArrayBuffer) return data;
if (ArrayBuffer.isView(data)) {
const view = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
const out = new Uint8Array(view.byteLength);
out.set(view);
return out.buffer;
}
if (Buffer.isBuffer(data)) {
const view = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
const out = new Uint8Array(view.byteLength);
out.set(view);
return out.buffer;
}
return String(data);
}
function buildRelayWebSocketUrl(
relayEndpoint: string,
sessionId: string,
@@ -86,7 +86,9 @@ export class VoiceAssistantWebSocketServer {
this.bridge.broadcast(message);
}
public async attachExternalSocket(ws: Parameters<WebSocketSessionBridge["attach"]>[0]): Promise<void> {
public async attachExternalSocket(
ws: Parameters<WebSocketSessionBridge["attach"]>[0]
): Promise<void> {
const fakeRequest = {
url: "/ws",
headers: {},
@@ -1,5 +1,4 @@
import type { IncomingMessage } from "http";
import type { WebSocket } from "ws";
import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
import { join } from "path";
import {
@@ -22,9 +21,17 @@ import type pino from "pino";
export type AgentMcpTransportFactory = () => Promise<Transport>;
type WebSocketLike = {
readyState: number;
send: (data: string) => void;
close: (code?: number, reason?: string) => void;
on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void;
once: (event: "close" | "error", listener: (...args: any[]) => void) => void;
};
export class WebSocketSessionBridge {
private readonly logger: pino.Logger;
private readonly sessions: Map<WebSocket, Session> = new Map();
private readonly sessions: Map<WebSocketLike, Session> = new Map();
private clientIdCounter = 0;
private readonly agentManager: AgentManager;
private readonly agentStorage: AgentStorage;
@@ -86,7 +93,7 @@ export class WebSocketSessionBridge {
return this.sessions.size;
}
public async attach(ws: WebSocket, _request: IncomingMessage): Promise<void> {
public async attach(ws: WebSocketLike, _request: IncomingMessage): Promise<void> {
const clientId = `client-${++this.clientIdCounter}`;
const connectionLogger = this.logger.child({ clientId });
@@ -131,7 +138,7 @@ export class WebSocketSessionBridge {
});
}
private async detach(ws: WebSocket, connectionLogger: pino.Logger, clientId: string): Promise<void> {
private async detach(ws: WebSocketLike, connectionLogger: pino.Logger, clientId: string): Promise<void> {
const session = this.sessions.get(ws);
if (!session) return;
@@ -144,7 +151,10 @@ export class WebSocketSessionBridge {
this.sessions.delete(ws);
}
private async handleRawMessage(ws: WebSocket, data: Buffer | ArrayBuffer | Buffer[]): Promise<void> {
private async handleRawMessage(
ws: WebSocketLike,
data: Buffer | ArrayBuffer | Buffer[] | string
): Promise<void> {
try {
const parsed = JSON.parse(data.toString());
const message = WSInboundMessageSchema.parse(parsed);
@@ -237,7 +247,7 @@ export class WebSocketSessionBridge {
}
}
private sendToClient(ws: WebSocket, message: WSOutboundMessage): void {
private sendToClient(ws: WebSocketLike, message: WSOutboundMessage): void {
// WebSocket.OPEN = 1
if (ws.readyState === 1) {
ws.send(JSON.stringify(message));
@@ -366,7 +376,7 @@ export class WebSocketSessionBridge {
reason: "finished" | "error" | "permission";
}): void {
const clientEntries: Array<{
ws: WebSocket;
ws: WebSocketLike;
state: {
deviceType: "web" | "mobile" | null;
focusedAgentId: string | null;
+1 -1
View File
@@ -3,7 +3,7 @@
"target": "ES2020",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2020"],
"lib": ["ES2020", "DOM"],
"baseUrl": "./src",
"paths": {
"@server/*": ["./*"]