From 11eaf8d28eaf8babfb2ba5d07b7358cabbbcdb2b Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 13 Jul 2026 00:20:15 -0400 Subject: [PATCH] feat(runtime): add relay control and data clients Co-authored-by: Orca --- .../relay/relay-control-client.test.ts | 248 +++++++++++++++++ .../runtime/relay/relay-control-client.ts | 252 ++++++++++++++++++ .../runtime/relay/relay-control-protocol.ts | 105 ++++++++ .../runtime/relay/relay-control-requests.ts | 112 ++++++++ src/main/runtime/relay/relay-host-proof.ts | 166 ++++++++++++ .../runtime/rpc/mobile-socket-wiring.test.ts | 3 +- src/main/runtime/rpc/mobile-socket-wiring.ts | 1 + src/main/runtime/rpc/relay-transport.test.ts | 102 +++++++ src/main/runtime/rpc/relay-transport.ts | 199 ++++++++++++++ 9 files changed, 1187 insertions(+), 1 deletion(-) create mode 100644 src/main/runtime/relay/relay-control-client.test.ts create mode 100644 src/main/runtime/relay/relay-control-client.ts create mode 100644 src/main/runtime/relay/relay-control-protocol.ts create mode 100644 src/main/runtime/relay/relay-control-requests.ts create mode 100644 src/main/runtime/relay/relay-host-proof.ts create mode 100644 src/main/runtime/rpc/relay-transport.test.ts create mode 100644 src/main/runtime/rpc/relay-transport.ts diff --git a/src/main/runtime/relay/relay-control-client.test.ts b/src/main/runtime/relay/relay-control-client.test.ts new file mode 100644 index 00000000000..275945768de --- /dev/null +++ b/src/main/runtime/relay/relay-control-client.test.ts @@ -0,0 +1,248 @@ +import { createHash, createHmac, randomBytes } from 'node:crypto' +import { afterEach, describe, expect, it, vi } from 'vitest' +import nacl from 'tweetnacl' +import { WebSocketServer, type WebSocket } from 'ws' +import type { E2EEKeypair } from '../e2ee-keypair' +import { RelayControlClient } from './relay-control-client' + +const encoder = new TextEncoder() +const HOST_PROOF_DOMAIN = 'orca-relay-host-proof/v1' +const CHALLENGE_DOMAIN = 'orca-relay-host-challenge/v1' + +function concat(parts: readonly Uint8Array[]): Uint8Array { + const output = new Uint8Array(parts.reduce((total, part) => total + part.byteLength, 0)) + let offset = 0 + for (const part of parts) { + output.set(part, offset) + offset += part.byteLength + } + return output +} + +function uint32(value: number): Uint8Array { + const bytes = new Uint8Array(4) + new DataView(bytes.buffer).setUint32(0, value, false) + return bytes +} + +function uint64(value: number): Uint8Array { + const bytes = new Uint8Array(8) + new DataView(bytes.buffer).setBigUint64(0, BigInt(value), false) + return bytes +} + +function field(name: string, value: Uint8Array): Uint8Array { + const encodedName = encoder.encode(name) + return concat([uint32(encodedName.byteLength), encodedName, uint32(value.byteLength), value]) +} + +function text(value: string): Uint8Array { + return encoder.encode(value) +} + +function buildTranscript(input: { + origin: string + relayKey: Uint8Array + nonce: Uint8Array + challengeId: string + issuedAt: number + expiresAt: number + relayHostId: string + hostKey: Uint8Array +}): Uint8Array { + return concat([ + field('protocol', text(HOST_PROOF_DOMAIN)), + field('version', new Uint8Array([1])), + field('relayOrigin', text(input.origin)), + field('relayEphemeralPublicKey', input.relayKey), + field('challengeNonce', input.nonce), + field('challengeId', text(input.challengeId)), + field('issuedAt', uint64(input.issuedAt)), + field('expiresAt', uint64(input.expiresAt)), + field('userId', text('user-1')), + field('profileId', text('profile-1')), + field('organizationId', text('org-1')), + field('relayHostId', text(input.relayHostId)), + field('hostPublicKey', input.hostKey), + field('assignmentEpoch', uint64(3)), + field('previousGeneration', new Uint8Array()), + field('resumeRequested', new Uint8Array([0])) + ]) +} + +function nextJson(ws: WebSocket): Promise> { + return new Promise((resolve) => { + ws.once('message', (raw) => resolve(JSON.parse(raw.toString()) as Record)) + }) +} + +describe('RelayControlClient', () => { + const servers: WebSocketServer[] = [] + const clients: RelayControlClient[] = [] + + afterEach(async () => { + for (const client of clients.splice(0)) { + client.closeNow() + } + await Promise.all( + servers.splice(0).map( + (server) => + new Promise((resolve) => { + for (const socket of server.clients) { + socket.terminate() + } + server.close(() => resolve()) + }) + ) + ) + }) + + it('proves the host key and drives control/data commands without URL credentials', async () => { + const server = new WebSocketServer({ port: 0, perMessageDeflate: false }) + servers.push(server) + await new Promise((resolve) => server.once('listening', resolve)) + const address = server.address() + if (!address || typeof address === 'string') { + throw new Error('expected TCP relay test server') + } + const origin = `http://127.0.0.1:${address.port}` + const hostKeys = nacl.box.keyPair() + const keypair: E2EEKeypair = { + publicKey: hostKeys.publicKey, + secretKey: hostKeys.secretKey, + publicKeyB64: Buffer.from(hostKeys.publicKey).toString('base64') + } + const relayHostId = createHash('sha256') + .update(hostKeys.publicKey) + .digest('base64url') + .slice(0, 16) + const accepted = new Promise<{ socket: WebSocket; authorization: string; path: string }>( + (resolve) => { + server.once('connection', (socket, request) => + resolve({ + socket, + authorization: String(request.headers.authorization), + path: request.url ?? '' + }) + ) + } + ) + const onConnectionOpen = vi.fn() + const onDrain = vi.fn() + const onClose = vi.fn() + const client = new RelayControlClient({ + cellUrl: origin, + relayJwt: 'scoped-token', + relayHostId, + assignmentEpoch: 3, + identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' }, + keypair, + appVersion: '1.2.3', + onConnectionOpen, + onDrain, + onClose + }) + clients.push(client) + const connecting = client.connect() + const { socket, authorization, path } = await accepted + expect(authorization).toBe('Bearer scoped-token') + expect(path).toBe('/v1/host/control') + const hello = await nextJson(socket) + expect(hello).toMatchObject({ + type: 'host-hello', + relayHostId, + assignmentEpoch: 3, + hostPublicKeyB64: keypair.publicKeyB64 + }) + + const relayKeys = nacl.box.keyPair() + const nonce = randomBytes(24) + const secret = randomBytes(32) + const issuedAt = Date.now() + const expiresAt = issuedAt + 10_000 + const transcript = buildTranscript({ + origin, + relayKey: relayKeys.publicKey, + nonce, + challengeId: 'challenge-1', + issuedAt, + expiresAt, + relayHostId, + hostKey: hostKeys.publicKey + }) + const plaintext = concat([ + text(`${CHALLENGE_DOMAIN}\0`), + uint32(transcript.byteLength), + transcript, + secret + ]) + const proofMessage = nextJson(socket) + socket.send( + JSON.stringify({ + type: 'host-challenge', + challengeId: 'challenge-1', + relayEphemeralPublicKeyB64: Buffer.from(relayKeys.publicKey).toString('base64'), + nonceB64: nonce.toString('base64'), + ciphertextB64: Buffer.from( + nacl.box(plaintext, nonce, hostKeys.publicKey, relayKeys.secretKey) + ).toString('base64'), + expiresAt + }) + ) + const proof = await proofMessage + expect(proof).toMatchObject({ type: 'host-challenge-ack', challengeId: 'challenge-1' }) + const expectedProof = createHmac('sha256', secret) + .update(text(`${HOST_PROOF_DOMAIN}\0ack\0`)) + .update(transcript) + .digest('base64') + expect(proof.proofB64).toBe(expectedProof) + + socket.send( + JSON.stringify({ + type: 'host-hello-ack', + v: 1, + generation: 4, + controlResumeSecret: randomBytes(32).toString('base64url'), + leaseExpiresAt: Date.now() + 60_000, + activeConnIds: [], + pendingConns: [] + }) + ) + await expect(connecting).resolves.toMatchObject({ generation: 4 }) + + socket.send(JSON.stringify({ type: 'ping', t: Date.now() })) + await expect(nextJson(socket)).resolves.toMatchObject({ type: 'pong' }) + socket.send( + JSON.stringify({ + type: 'conn-open', + connId: 'conn-1', + connTicket: randomBytes(32).toString('base64url'), + kind: 'invite', + relayDeviceId: 'device-1', + attachDeadlineMs: 10_000 + }) + ) + await vi.waitFor(() => expect(onConnectionOpen).toHaveBeenCalledOnce()) + + const inviteRequest = nextJson(socket) + const invitePromise = client.createInvite('device-1', 'invite-req') + await expect(inviteRequest).resolves.toEqual({ + type: 'invite-create', + reqId: 'invite-req', + relayDeviceId: 'device-1' + }) + socket.send( + JSON.stringify({ + type: 'invite-created', + reqId: 'invite-req', + inviteToken: randomBytes(32).toString('base64url'), + expiresAt: Date.now() + 60_000, + maxAttempts: 3 + }) + ) + await expect(invitePromise).resolves.toMatchObject({ reqId: 'invite-req' }) + + socket.send(JSON.stringify({ type: 'drain', graceMs: 5_000, recovery: 'resolve-director' })) + await vi.waitFor(() => expect(onDrain).toHaveBeenCalledOnce()) + }) +}) diff --git a/src/main/runtime/relay/relay-control-client.ts b/src/main/runtime/relay/relay-control-client.ts new file mode 100644 index 00000000000..7d1f3c5f0da --- /dev/null +++ b/src/main/runtime/relay/relay-control-client.ts @@ -0,0 +1,252 @@ +import { randomUUID } from 'node:crypto' +import WebSocket, { type RawData } from 'ws' +import { MOBILE_RELAY_CLOSE_CODE } from '../../../shared/mobile-relay-close-codes' +import type { E2EEKeypair } from '../e2ee-keypair' +import { + RelayConnectionOpenMessageSchema, + RelayDrainMessageSchema, + RelayHostChallengeMessageSchema, + RelayHostHelloAckMessageSchema, + RelayPingMessageSchema, + parseRelayControlMessage, + type RelayConnectionOpenMessage, + type RelayDrainMessage, + type RelayHostHelloAckMessage, + type RelayInviteCreatedMessage +} from './relay-control-protocol' +import { RelayControlRequests } from './relay-control-requests' +import { answerRelayHostChallenge } from './relay-host-proof' + +type RelayControlState = 'idle' | 'opening' | 'proving' | 'active' | 'draining' | 'closed' + +type RelayControlClientOptions = { + cellUrl: string + relayJwt: string + relayHostId: string + assignmentEpoch: number + identity: { userId: string; profileId: string; organizationId: string } + keypair: E2EEKeypair + appVersion: string + previousGeneration?: number + controlResumeSecret?: string + onConnectionOpen: (message: RelayConnectionOpenMessage) => void + onDrain: (message: RelayDrainMessage) => void + onClose: (code: number) => void + createSocket?: (url: string, relayJwt: string) => WebSocket +} + +function controlWebSocketUrl(cellUrl: string): { origin: string; url: string } { + const parsed = new URL(cellUrl) + if (parsed.pathname !== '/' || parsed.search || parsed.hash) { + throw new Error('relay_cell_url_must_be_an_origin') + } + const origin = parsed.origin + if (parsed.protocol === 'https:') { + parsed.protocol = 'wss:' + } else if (parsed.protocol === 'http:') { + parsed.protocol = 'ws:' + } else { + throw new Error('relay_cell_url_must_use_http') + } + return { origin, url: `${parsed.origin}/v1/host/control` } +} + +export class RelayControlClient { + private readonly options: RelayControlClientOptions + private readonly relayOrigin: string + private readonly controlUrl: string + private readonly createSocket: NonNullable + private readonly requests = new RelayControlRequests() + private socket: WebSocket | null = null + private state: RelayControlState = 'idle' + private connectResolve: ((ack: RelayHostHelloAckMessage) => void) | null = null + private connectReject: ((error: Error) => void) | null = null + + constructor(options: RelayControlClientOptions) { + this.options = options + const endpoint = controlWebSocketUrl(options.cellUrl) + this.relayOrigin = endpoint.origin + this.controlUrl = endpoint.url + this.createSocket = + options.createSocket ?? + ((url, token) => + new WebSocket(url, { + headers: { authorization: `Bearer ${token}` }, + perMessageDeflate: false, + maxPayload: 64 * 1024 + })) + } + + connect(): Promise { + if (this.state !== 'idle') { + return Promise.reject(new Error('relay_control_already_started')) + } + this.state = 'opening' + const socket = this.createSocket(this.controlUrl, this.options.relayJwt) + this.socket = socket + socket.once('open', () => this.sendHostHello()) + socket.on('message', (raw, isBinary) => { + if (isBinary) { + this.failProtocol('binary control message') + return + } + this.handleMessage(raw) + }) + socket.once('error', (error) => { + if (this.state === 'opening' || this.state === 'proving') { + this.connectReject?.(error) + this.clearConnectPromise() + } + }) + socket.once('close', (code) => this.handleClose(code)) + return new Promise((resolve, reject) => { + this.connectResolve = resolve + this.connectReject = reject + }) + } + + refreshAuthorization(relayJwt: string): void { + this.sendActive({ type: 'auth-refresh', relayJwt }) + } + + createInvite( + relayDeviceId: string, + reqId: string = randomUUID() + ): Promise { + return this.requests.createInvite(reqId, relayDeviceId, (payload) => this.sendActive(payload)) + } + + revokeDevice(relayDeviceId: string, reqId: string = randomUUID()): Promise { + return this.requests.revokeDevice(reqId, relayDeviceId, (payload) => this.sendActive(payload)) + } + + closeNow(): void { + this.state = 'closed' + this.requests.rejectAll(new Error('relay_control_closed')) + this.socket?.terminate() + this.socket = null + } + + private sendHostHello(): void { + if (!this.socket || this.state !== 'opening') { + return + } + this.state = 'proving' + this.socket.send( + JSON.stringify({ + type: 'host-hello', + v: 1, + relayHostId: this.options.relayHostId, + assignmentEpoch: this.options.assignmentEpoch, + hostPublicKeyB64: this.options.keypair.publicKeyB64, + appVersion: this.options.appVersion, + ...(this.options.previousGeneration === undefined + ? {} + : { previousGeneration: this.options.previousGeneration }), + ...(this.options.controlResumeSecret + ? { controlResumeSecret: this.options.controlResumeSecret } + : {}) + }) + ) + } + + private handleMessage(raw: RawData): void { + const message = parseRelayControlMessage(raw) + if (!message) { + this.failProtocol('invalid control JSON') + return + } + if (this.state === 'proving') { + this.handleProofMessage(message) + return + } + if (this.state !== 'active' && this.state !== 'draining') { + this.failProtocol('control message before activation') + return + } + if (RelayPingMessageSchema.safeParse(message).success) { + this.socket?.send(JSON.stringify({ type: 'pong', t: message.t })) + return + } + const connection = RelayConnectionOpenMessageSchema.safeParse(message) + if (connection.success && this.state === 'active') { + this.options.onConnectionOpen(connection.data) + return + } + const drain = RelayDrainMessageSchema.safeParse(message) + if (drain.success) { + this.state = 'draining' + this.options.onDrain(drain.data) + return + } + if (this.requests.resolveMessage(message)) { + return + } + this.failProtocol('unknown control message') + } + + private handleProofMessage(message: Record): void { + const challenge = RelayHostChallengeMessageSchema.safeParse(message) + if (challenge.success) { + const proofB64 = answerRelayHostChallenge(challenge.data, { + relayOrigin: this.relayOrigin, + ...this.options.identity, + relayHostId: this.options.relayHostId, + hostPublicKey: this.options.keypair.publicKey, + hostSecretKey: this.options.keypair.secretKey, + assignmentEpoch: this.options.assignmentEpoch, + previousGeneration: this.options.previousGeneration, + resumeRequested: Boolean(this.options.controlResumeSecret) + }) + if (!proofB64) { + this.failProtocol('invalid host challenge') + return + } + this.socket?.send( + JSON.stringify({ + type: 'host-challenge-ack', + challengeId: challenge.data.challengeId, + proofB64 + }) + ) + return + } + const ack = RelayHostHelloAckMessageSchema.safeParse(message) + if (!ack.success) { + this.failProtocol('invalid host proof message') + return + } + this.state = 'active' + this.connectResolve?.(ack.data) + this.clearConnectPromise() + } + + private sendActive(payload: object): void { + if (!this.socket || (this.state !== 'active' && this.state !== 'draining')) { + throw new Error('relay_control_not_active') + } + this.socket.send(JSON.stringify(payload)) + } + + private failProtocol(reason: string): void { + this.connectReject?.(new Error(reason)) + this.clearConnectPromise() + this.socket?.close(MOBILE_RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, reason) + } + + private handleClose(code: number): void { + const wasConnecting = this.state === 'opening' || this.state === 'proving' + this.state = 'closed' + if (wasConnecting) { + this.connectReject?.(new Error(`relay_control_closed_${code}`)) + this.clearConnectPromise() + } + this.requests.rejectAll(new Error(`relay_control_closed_${code}`)) + this.options.onClose(code) + } + + private clearConnectPromise(): void { + this.connectResolve = null + this.connectReject = null + } +} diff --git a/src/main/runtime/relay/relay-control-protocol.ts b/src/main/runtime/relay/relay-control-protocol.ts new file mode 100644 index 00000000000..24f3d7c7851 --- /dev/null +++ b/src/main/runtime/relay/relay-control-protocol.ts @@ -0,0 +1,105 @@ +import { z } from 'zod' +import type { RawData } from 'ws' + +const OpaqueIdSchema = z.string().min(1).max(128) +const Base64Url32ByteSchema = z.string().regex(/^[A-Za-z0-9_-]{43}$/) +const Base6432ByteSchema = z.string().regex(/^[A-Za-z0-9+/]{43}=$/) +const Base64Raw24ByteSchema = z.string().regex(/^[A-Za-z0-9+/]{32}$/) +const EpochMsSchema = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER) +const GenerationSchema = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER) + +export const RelayHostChallengeMessageSchema = z + .object({ + type: z.literal('host-challenge'), + challengeId: OpaqueIdSchema, + relayEphemeralPublicKeyB64: Base6432ByteSchema, + nonceB64: Base64Raw24ByteSchema, + ciphertextB64: z + .string() + .min(1) + .max(16 * 1024), + expiresAt: EpochMsSchema + }) + .strict() + +const PendingConnectionSchema = z + .object({ connId: OpaqueIdSchema, connTicket: Base64Url32ByteSchema }) + .strict() + +export const RelayHostHelloAckMessageSchema = z + .object({ + type: z.literal('host-hello-ack'), + v: z.literal(1), + generation: GenerationSchema, + controlResumeSecret: Base64Url32ByteSchema, + leaseExpiresAt: EpochMsSchema, + activeConnIds: z.array(OpaqueIdSchema).max(8), + pendingConns: z.array(PendingConnectionSchema).max(8) + }) + .strict() + +export const RelayConnectionOpenMessageSchema = z + .object({ + type: z.literal('conn-open'), + connId: OpaqueIdSchema, + connTicket: Base64Url32ByteSchema, + kind: z.enum(['invite', 'resume']), + relayDeviceId: OpaqueIdSchema, + attachDeadlineMs: z.number().int().positive().max(60_000) + }) + .strict() + +export const RelayDrainMessageSchema = z + .object({ + type: z.literal('drain'), + graceMs: z + .number() + .int() + .nonnegative() + .max(60 * 60 * 1000), + recovery: z.literal('resolve-director') + }) + .strict() + +export const RelayPingMessageSchema = z + .object({ type: z.literal('ping'), t: EpochMsSchema }) + .strict() + +export const RelayInviteCreatedMessageSchema = z + .object({ + type: z.literal('invite-created'), + reqId: OpaqueIdSchema, + inviteToken: Base64Url32ByteSchema, + expiresAt: EpochMsSchema, + maxAttempts: z.number().int().positive().max(16) + }) + .strict() + +export const RelayDeviceRevokedMessageSchema = z + .object({ type: z.literal('device-revoked'), reqId: OpaqueIdSchema }) + .strict() + +export const RelayControlErrorMessageSchema = z + .object({ + type: z.literal('control-error'), + reqId: OpaqueIdSchema.optional(), + code: z.string().min(1).max(128) + }) + .strict() + +export type RelayHostChallengeMessage = z.infer +export type RelayHostHelloAckMessage = z.infer +export type RelayConnectionOpenMessage = z.infer +export type RelayDrainMessage = z.infer +export type RelayInviteCreatedMessage = z.infer + +export function parseRelayControlMessage(raw: RawData): Record | null { + try { + const parsed = JSON.parse(raw.toString()) as unknown + return typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed) + ? (parsed as Record) + : null + } catch { + return null + } +} diff --git a/src/main/runtime/relay/relay-control-requests.ts b/src/main/runtime/relay/relay-control-requests.ts new file mode 100644 index 00000000000..8a56c7151d1 --- /dev/null +++ b/src/main/runtime/relay/relay-control-requests.ts @@ -0,0 +1,112 @@ +import { + RelayControlErrorMessageSchema, + RelayDeviceRevokedMessageSchema, + RelayInviteCreatedMessageSchema, + type RelayInviteCreatedMessage +} from './relay-control-protocol' + +type PendingRequest = { + kind: 'invite' | 'revoke' + resolve: (value: unknown) => void + reject: (error: Error) => void + timer: ReturnType +} + +export class RelayControlRequests { + private readonly pending = new Map() + + createInvite( + reqId: string, + relayDeviceId: string, + send: (payload: object) => void + ): Promise { + return this.request( + reqId, + 'invite', + { type: 'invite-create', reqId, relayDeviceId }, + send + ) as Promise + } + + revokeDevice( + reqId: string, + relayDeviceId: string, + send: (payload: object) => void + ): Promise { + return this.request( + reqId, + 'revoke', + { type: 'device-revoke', reqId, relayDeviceId }, + send + ) as Promise + } + + resolveMessage(message: Record): boolean { + const reqId = typeof message.reqId === 'string' ? message.reqId : null + const pending = reqId ? this.pending.get(reqId) : null + if (!pending || !reqId) { + return false + } + const error = RelayControlErrorMessageSchema.safeParse(message) + if (error.success) { + this.finish(reqId) + pending.reject(new Error(error.data.code)) + return true + } + if (pending.kind === 'invite') { + const invite = RelayInviteCreatedMessageSchema.safeParse(message) + if (!invite.success) { + return false + } + this.finish(reqId) + pending.resolve(invite.data) + return true + } + const revoked = RelayDeviceRevokedMessageSchema.safeParse(message) + if (!revoked.success) { + return false + } + this.finish(reqId) + pending.resolve(undefined) + return true + } + + rejectAll(error: Error): void { + for (const [reqId, pending] of this.pending) { + this.finish(reqId) + pending.reject(error) + } + } + + private request( + reqId: string, + kind: PendingRequest['kind'], + payload: object, + send: (payload: object) => void + ): Promise { + if (this.pending.has(reqId)) { + return Promise.reject(new Error('duplicate_relay_request_id')) + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(reqId) + reject(new Error('relay_control_request_timeout')) + }, 10_000) + this.pending.set(reqId, { kind, resolve, reject, timer }) + try { + send(payload) + } catch (error) { + this.finish(reqId) + reject(error) + } + }) + } + + private finish(reqId: string): void { + const pending = this.pending.get(reqId) + if (pending) { + clearTimeout(pending.timer) + this.pending.delete(reqId) + } + } +} diff --git a/src/main/runtime/relay/relay-host-proof.ts b/src/main/runtime/relay/relay-host-proof.ts new file mode 100644 index 00000000000..94db5b2f1b2 --- /dev/null +++ b/src/main/runtime/relay/relay-host-proof.ts @@ -0,0 +1,166 @@ +import { createHmac, timingSafeEqual } from 'node:crypto' +import nacl from 'tweetnacl' + +const HOST_PROOF_TRANSCRIPT_DOMAIN = 'orca-relay-host-proof/v1' +const HOST_CHALLENGE_PLAINTEXT_DOMAIN = 'orca-relay-host-challenge/v1' +const textEncoder = new TextEncoder() +const textDecoder = new TextDecoder() + +export type RelayHostChallenge = { + challengeId: string + relayEphemeralPublicKeyB64: string + nonceB64: string + ciphertextB64: string + expiresAt: number +} + +export type RelayHostProofContext = { + relayOrigin: string + userId: string + profileId: string + organizationId: string + relayHostId: string + hostPublicKey: Uint8Array + hostSecretKey: Uint8Array + assignmentEpoch: number + previousGeneration?: number + resumeRequested: boolean + now?: () => number +} + +function decodeCanonicalBase64(value: string, expectedBytes: number): Uint8Array | null { + if (!/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/.test(value)) { + return null + } + const decoded = Buffer.from(value, 'base64') + return decoded.byteLength === expectedBytes && decoded.toString('base64') === value + ? decoded + : null +} + +function uint64(value: number): Uint8Array { + const bytes = new Uint8Array(8) + new DataView(bytes.buffer).setBigUint64(0, BigInt(value), false) + return bytes +} + +function equal(left: Uint8Array | undefined, right: Uint8Array): boolean { + return Boolean(left && left.byteLength === right.byteLength && timingSafeEqual(left, right)) +} + +function parseTranscript(transcript: Uint8Array): Map | null { + const fields = new Map() + const view = new DataView(transcript.buffer, transcript.byteOffset, transcript.byteLength) + let offset = 0 + try { + while (offset < transcript.byteLength) { + const nameLength = view.getUint32(offset, false) + offset += 4 + const name = textDecoder.decode(transcript.slice(offset, offset + nameLength)) + offset += nameLength + const valueLength = view.getUint32(offset, false) + offset += 4 + if (fields.has(name) || offset + valueLength > transcript.byteLength) { + return null + } + fields.set(name, transcript.slice(offset, offset + valueLength)) + offset += valueLength + } + } catch { + return null + } + return offset === transcript.byteLength ? fields : null +} + +function readUint64(value: Uint8Array | undefined): number | null { + if (!value || value.byteLength !== 8) { + return null + } + const parsed = new DataView(value.buffer, value.byteOffset, value.byteLength).getBigUint64( + 0, + false + ) + return parsed <= BigInt(Number.MAX_SAFE_INTEGER) ? Number(parsed) : null +} + +function validateTranscript( + transcript: Uint8Array, + challenge: RelayHostChallenge, + context: RelayHostProofContext, + relayKey: Uint8Array, + nonce: Uint8Array +): boolean { + const fields = parseTranscript(transcript) + if (!fields || fields.size !== 16) { + return false + } + const now = (context.now ?? Date.now)() + const issuedAt = readUint64(fields.get('issuedAt')) + const expiresAt = readUint64(fields.get('expiresAt')) + const previousGeneration = fields.get('previousGeneration') + const expectedPrevious = + context.previousGeneration === undefined ? new Uint8Array() : uint64(context.previousGeneration) + return ( + issuedAt !== null && + issuedAt <= now && + now <= challenge.expiresAt && + challenge.expiresAt - issuedAt <= 10_000 && + expiresAt === challenge.expiresAt && + equal(fields.get('protocol'), textEncoder.encode(HOST_PROOF_TRANSCRIPT_DOMAIN)) && + equal(fields.get('version'), new Uint8Array([1])) && + equal(fields.get('relayOrigin'), textEncoder.encode(context.relayOrigin)) && + equal(fields.get('relayEphemeralPublicKey'), relayKey) && + equal(fields.get('challengeNonce'), nonce) && + equal(fields.get('challengeId'), textEncoder.encode(challenge.challengeId)) && + equal(fields.get('userId'), textEncoder.encode(context.userId)) && + equal(fields.get('profileId'), textEncoder.encode(context.profileId)) && + equal(fields.get('organizationId'), textEncoder.encode(context.organizationId)) && + equal(fields.get('relayHostId'), textEncoder.encode(context.relayHostId)) && + equal(fields.get('hostPublicKey'), context.hostPublicKey) && + equal(fields.get('assignmentEpoch'), uint64(context.assignmentEpoch)) && + equal(previousGeneration, expectedPrevious) && + equal(fields.get('resumeRequested'), new Uint8Array([context.resumeRequested ? 1 : 0])) + ) +} + +export function answerRelayHostChallenge( + challenge: RelayHostChallenge, + context: RelayHostProofContext +): string | null { + const relayKey = decodeCanonicalBase64(challenge.relayEphemeralPublicKeyB64, 32) + const nonce = decodeCanonicalBase64(challenge.nonceB64, 24) + const ciphertext = Buffer.from(challenge.ciphertextB64, 'base64') + if (!relayKey || !nonce || ciphertext.toString('base64') !== challenge.ciphertextB64) { + return null + } + const plaintext = nacl.box.open(ciphertext, nonce, relayKey, context.hostSecretKey) + if (!plaintext) { + return null + } + const domain = textEncoder.encode(`${HOST_CHALLENGE_PLAINTEXT_DOMAIN}\0`) + if ( + !equal(plaintext.slice(0, domain.byteLength), domain) || + plaintext.byteLength < domain.byteLength + 36 + ) { + return null + } + const transcriptLength = new DataView( + plaintext.buffer, + plaintext.byteOffset + domain.byteLength, + 4 + ).getUint32(0, false) + const transcriptStart = domain.byteLength + 4 + const secretStart = transcriptStart + transcriptLength + if (secretStart + 32 !== plaintext.byteLength) { + return null + } + const transcript = plaintext.slice(transcriptStart, secretStart) + if (!validateTranscript(transcript, challenge, context, relayKey, nonce)) { + return null + } + const secret = plaintext.slice(secretStart) + return createHmac('sha256', secret) + .update(textEncoder.encode(`${HOST_PROOF_TRANSCRIPT_DOMAIN}\0ack\0`)) + .update(transcript) + .digest('base64') +} diff --git a/src/main/runtime/rpc/mobile-socket-wiring.test.ts b/src/main/runtime/rpc/mobile-socket-wiring.test.ts index 4011e728597..c51e79b86ed 100644 --- a/src/main/runtime/rpc/mobile-socket-wiring.test.ts +++ b/src/main/runtime/rpc/mobile-socket-wiring.test.ts @@ -123,7 +123,8 @@ describe('MobileSocketWiring', () => { transport: 'relay', relayHostId: 'AbCdEf0123_-xyZ9', relayDeviceId: 'outer-device', - basisConnId: 'connection-1' + basisConnId: 'connection-1', + credentialKind: 'invite' } const wiring = new MobileSocketWiring({ deviceRegistry: registryFor('e2ee-device', 'valid-token'), diff --git a/src/main/runtime/rpc/mobile-socket-wiring.ts b/src/main/runtime/rpc/mobile-socket-wiring.ts index 9b8b75cc4b4..bb7ece7be51 100644 --- a/src/main/runtime/rpc/mobile-socket-wiring.ts +++ b/src/main/runtime/rpc/mobile-socket-wiring.ts @@ -13,6 +13,7 @@ export type MobileSocketTransportMetadata = relayHostId: string relayDeviceId: string basisConnId: string + credentialKind: 'invite' | 'resume' } export type MobileSocketTransport = { diff --git a/src/main/runtime/rpc/relay-transport.test.ts b/src/main/runtime/rpc/relay-transport.test.ts new file mode 100644 index 00000000000..38f221b49bc --- /dev/null +++ b/src/main/runtime/rpc/relay-transport.test.ts @@ -0,0 +1,102 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import WebSocketClient, { WebSocketServer, type WebSocket } from 'ws' +import { CloudRelayTransport } from './relay-transport' + +function nextMessage(ws: WebSocket): Promise<{ data: Buffer; isBinary: boolean }> { + return new Promise((resolve) => { + ws.once('message', (data, isBinary) => resolve({ data: Buffer.from(data as Buffer), isBinary })) + }) +} + +describe('CloudRelayTransport', () => { + const servers: WebSocketServer[] = [] + const transports: CloudRelayTransport[] = [] + + afterEach(async () => { + await Promise.all(transports.splice(0).map((transport) => transport.stop())) + await Promise.all( + servers.splice(0).map( + (server) => + new Promise((resolve) => { + for (const client of server.clients) { + client.terminate() + } + server.close(() => resolve()) + }) + ) + ) + }) + + it('authenticates one query-free host-data socket and forwards messages verbatim', async () => { + const server = new WebSocketServer({ port: 0, perMessageDeflate: false }) + servers.push(server) + await new Promise((resolve) => server.once('listening', resolve)) + const address = server.address() + if (typeof address === 'string' || address === null) { + throw new Error('expected TCP relay test server') + } + const accepted = new Promise<{ socket: WebSocket; path: string }>((resolve) => { + server.once('connection', (socket, request) => resolve({ socket, path: request.url ?? '' })) + }) + let clientSocket: WebSocketClient | null = null + const transport = new CloudRelayTransport({ + cellUrl: `http://127.0.0.1:${address.port}`, + relayHostId: 'AbCdEf0123_-xyZ9', + generation: 7, + createSocket: (url) => { + clientSocket = new WebSocketClient(url, { perMessageDeflate: false }) + return clientSocket + } + }) + transports.push(transport) + const received: (string | Uint8Array)[] = [] + transport.onMessage((message) => received.push(message)) + transport.onConnectionClose(vi.fn()) + await transport.start() + const opening = transport.openConnection({ + connId: 'conn/with spaces', + connTicket: 'ticket-1', + kind: 'resume', + relayDeviceId: 'device-1', + attachDeadlineMs: 1_000 + }) + const { socket, path } = await accepted + const auth = await nextMessage(socket) + await opening + + expect(path).toBe('/v1/host/data/conn%2Fwith%20spaces') + expect(auth.isBinary).toBe(false) + expect(JSON.parse(auth.data.toString())).toEqual({ + type: 'host-data-auth', + v: 1, + connTicket: 'ticket-1', + generation: 7 + }) + socket.send('e2ee-hello') + socket.send(Buffer.from([1, 2, 3]), { binary: true }) + await vi.waitFor(() => expect(received).toHaveLength(2)) + expect(received[0]).toBe('e2ee-hello') + expect(received[1]).toEqual(new Uint8Array([1, 2, 3])) + + expect(clientSocket).not.toBeNull() + const metadata = transport.metadataFor(clientSocket!) + expect(metadata).toEqual({ + transport: 'relay', + relayHostId: 'AbCdEf0123_-xyZ9', + relayDeviceId: 'device-1', + basisConnId: 'conn/with spaces', + credentialKind: 'resume' + }) + }) + + it('rejects non-origin cell URLs before opening a socket', () => { + expect( + () => + new CloudRelayTransport({ + cellUrl: 'https://relay.example/path?credential=forbidden', + relayHostId: 'AbCdEf0123_-xyZ9', + generation: 1 + }) + ).toThrow('relay_cell_url_must_be_an_origin') + }) +}) diff --git a/src/main/runtime/rpc/relay-transport.ts b/src/main/runtime/rpc/relay-transport.ts new file mode 100644 index 00000000000..8ffee907c37 --- /dev/null +++ b/src/main/runtime/rpc/relay-transport.ts @@ -0,0 +1,199 @@ +import WebSocket from 'ws' +import type { RpcTransport } from './transport' +import type { MobileSocketTransport, MobileSocketTransportMetadata } from './mobile-socket-wiring' + +const MAX_RELAY_MESSAGE_BYTES = 1024 * 1024 + +type RelayMessagePayload = string | Uint8Array + +export type RelayConnectionOpen = { + connId: string + connTicket: string + kind: 'invite' | 'resume' + relayDeviceId: string + attachDeadlineMs: number +} + +type CloudRelayTransportOptions = { + cellUrl: string + relayHostId: string + generation: number + createSocket?: (url: string) => WebSocket +} + +function relayWebSocketOrigin(cellUrl: string): string { + const url = new URL(cellUrl) + if (url.pathname !== '/' || url.search || url.hash) { + throw new Error('relay_cell_url_must_be_an_origin') + } + if (url.protocol === 'https:') { + url.protocol = 'wss:' + } else if (url.protocol === 'http:') { + url.protocol = 'ws:' + } else { + throw new Error('relay_cell_url_must_use_http') + } + return url.origin +} + +export class CloudRelayTransport implements RpcTransport, MobileSocketTransport { + private readonly cellWebSocketOrigin: string + private readonly relayHostId: string + private readonly generation: number + private readonly createSocket: (url: string) => WebSocket + private readonly socketsByConnectionId = new Map() + private readonly metadataBySocket = new Map() + private readonly clientIds = new Map() + private messageHandler: Parameters[0] | null = null + private closeHandler: Parameters[0] | null = null + private stopped = false + + constructor(options: CloudRelayTransportOptions) { + this.cellWebSocketOrigin = relayWebSocketOrigin(options.cellUrl) + this.relayHostId = options.relayHostId + this.generation = options.generation + this.createSocket = + options.createSocket ?? + ((url) => + new WebSocket(url, { perMessageDeflate: false, maxPayload: MAX_RELAY_MESSAGE_BYTES })) + } + + onMessage(handler: Parameters[0]): void { + this.messageHandler = handler + } + + onConnectionClose(handler: Parameters[0]): void { + this.closeHandler = handler + } + + metadataFor(ws: WebSocket): MobileSocketTransportMetadata { + const metadata = this.metadataBySocket.get(ws) + if (!metadata) { + throw new Error('unknown_relay_socket') + } + return metadata + } + + setClientId(ws: WebSocket, clientId: string): void { + if (this.metadataBySocket.has(ws)) { + this.clientIds.set(ws, clientId) + } + } + + terminateClientConnections(clientId: string): number { + const sockets = Array.from(this.clientIds.entries()) + .filter(([, candidate]) => candidate === clientId) + .map(([socket]) => socket) + for (const socket of sockets) { + socket.terminate() + } + return sockets.length + } + + async start(): Promise { + this.stopped = false + } + + async stop(): Promise { + this.stopped = true + const sockets = [...this.metadataBySocket.keys()] + for (const socket of sockets) { + socket.terminate() + } + await Promise.all(sockets.map((socket) => this.waitForClose(socket))) + } + + async openConnection(connection: RelayConnectionOpen): Promise { + if (this.stopped) { + throw new Error('relay_transport_stopped') + } + if (this.socketsByConnectionId.has(connection.connId)) { + return + } + const url = `${this.cellWebSocketOrigin}/v1/host/data/${encodeURIComponent(connection.connId)}` + const socket = this.createSocket(url) + const metadata: MobileSocketTransportMetadata = { + transport: 'relay', + relayHostId: this.relayHostId, + relayDeviceId: connection.relayDeviceId, + basisConnId: connection.connId, + credentialKind: connection.kind + } + this.socketsByConnectionId.set(connection.connId, socket) + this.metadataBySocket.set(socket, metadata) + + await new Promise((resolve, reject) => { + let opened = false + let attached = false + let finalized = false + const deadline = setTimeout(() => { + socket.terminate() + if (!opened) { + reject(new Error('relay_host_data_attach_timeout')) + } + }, connection.attachDeadlineMs) + const finalize = (): void => { + if (finalized) { + return + } + finalized = true + clearTimeout(deadline) + this.socketsByConnectionId.delete(connection.connId) + this.metadataBySocket.delete(socket) + const clientId = this.clientIds.get(socket) ?? null + this.clientIds.delete(socket) + const hasOtherConnections = + clientId !== null && [...this.clientIds.values()].includes(clientId) + this.closeHandler?.(clientId, socket, hasOtherConnections) + } + socket.on('message', (raw, isBinary) => { + if (!attached) { + attached = true + clearTimeout(deadline) + } + const message: RelayMessagePayload = isBinary + ? new Uint8Array(raw as Buffer) + : raw.toString() + this.messageHandler?.( + message, + (response) => { + if (socket.readyState === socket.OPEN) { + socket.send(response) + } + }, + socket + ) + }) + socket.once('open', () => { + opened = true + const networkSocket = ( + socket as unknown as { _socket?: { setNoDelay(value: boolean): void } } + )._socket + networkSocket?.setNoDelay(true) + socket.send( + JSON.stringify({ + type: 'host-data-auth', + v: 1, + connTicket: connection.connTicket, + generation: this.generation + }) + ) + resolve() + }) + socket.once('error', (error) => { + if (!opened) { + finalize() + reject(error) + } + }) + socket.once('close', finalize) + }) + } + + private waitForClose(socket: WebSocket): Promise { + if (socket.readyState === socket.CLOSED) { + return Promise.resolve() + } + return new Promise((resolve) => socket.once('close', resolve)) + } +}