mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 08:03:14 +00:00
feat(runtime): add relay control and data clients
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo-H
Orca
@@ -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<Record<string, unknown>> {
|
||||
return new Promise((resolve) => {
|
||||
ws.once('message', (raw) => resolve(JSON.parse(raw.toString()) as Record<string, unknown>))
|
||||
})
|
||||
}
|
||||
|
||||
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<void>((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<void>((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())
|
||||
})
|
||||
})
|
||||
@@ -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<RelayControlClientOptions['createSocket']>
|
||||
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<RelayHostHelloAckMessage> {
|
||||
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<RelayInviteCreatedMessage> {
|
||||
return this.requests.createInvite(reqId, relayDeviceId, (payload) => this.sendActive(payload))
|
||||
}
|
||||
|
||||
revokeDevice(relayDeviceId: string, reqId: string = randomUUID()): Promise<void> {
|
||||
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<string, unknown>): 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
|
||||
}
|
||||
}
|
||||
@@ -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<typeof RelayHostChallengeMessageSchema>
|
||||
export type RelayHostHelloAckMessage = z.infer<typeof RelayHostHelloAckMessageSchema>
|
||||
export type RelayConnectionOpenMessage = z.infer<typeof RelayConnectionOpenMessageSchema>
|
||||
export type RelayDrainMessage = z.infer<typeof RelayDrainMessageSchema>
|
||||
export type RelayInviteCreatedMessage = z.infer<typeof RelayInviteCreatedMessageSchema>
|
||||
|
||||
export function parseRelayControlMessage(raw: RawData): Record<string, unknown> | null {
|
||||
try {
|
||||
const parsed = JSON.parse(raw.toString()) as unknown
|
||||
return typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed)
|
||||
? (parsed as Record<string, unknown>)
|
||||
: null
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
@@ -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<typeof setTimeout>
|
||||
}
|
||||
|
||||
export class RelayControlRequests {
|
||||
private readonly pending = new Map<string, PendingRequest>()
|
||||
|
||||
createInvite(
|
||||
reqId: string,
|
||||
relayDeviceId: string,
|
||||
send: (payload: object) => void
|
||||
): Promise<RelayInviteCreatedMessage> {
|
||||
return this.request(
|
||||
reqId,
|
||||
'invite',
|
||||
{ type: 'invite-create', reqId, relayDeviceId },
|
||||
send
|
||||
) as Promise<RelayInviteCreatedMessage>
|
||||
}
|
||||
|
||||
revokeDevice(
|
||||
reqId: string,
|
||||
relayDeviceId: string,
|
||||
send: (payload: object) => void
|
||||
): Promise<void> {
|
||||
return this.request(
|
||||
reqId,
|
||||
'revoke',
|
||||
{ type: 'device-revoke', reqId, relayDeviceId },
|
||||
send
|
||||
) as Promise<void>
|
||||
}
|
||||
|
||||
resolveMessage(message: Record<string, unknown>): 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<unknown> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, Uint8Array> | null {
|
||||
const fields = new Map<string, Uint8Array>()
|
||||
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')
|
||||
}
|
||||
@@ -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'),
|
||||
|
||||
@@ -13,6 +13,7 @@ export type MobileSocketTransportMetadata =
|
||||
relayHostId: string
|
||||
relayDeviceId: string
|
||||
basisConnId: string
|
||||
credentialKind: 'invite' | 'resume'
|
||||
}
|
||||
|
||||
export type MobileSocketTransport = {
|
||||
|
||||
@@ -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<void>((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<void>((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<ArrayBufferLike>)[] = []
|
||||
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')
|
||||
})
|
||||
})
|
||||
@@ -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<ArrayBufferLike>
|
||||
|
||||
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<string, WebSocket>()
|
||||
private readonly metadataBySocket = new Map<WebSocket, MobileSocketTransportMetadata>()
|
||||
private readonly clientIds = new Map<WebSocket, string>()
|
||||
private messageHandler: Parameters<MobileSocketTransport['onMessage']>[0] | null = null
|
||||
private closeHandler: Parameters<MobileSocketTransport['onConnectionClose']>[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<MobileSocketTransport['onMessage']>[0]): void {
|
||||
this.messageHandler = handler
|
||||
}
|
||||
|
||||
onConnectionClose(handler: Parameters<MobileSocketTransport['onConnectionClose']>[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<void> {
|
||||
this.stopped = false
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
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<void> {
|
||||
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<void>((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<void> {
|
||||
if (socket.readyState === socket.CLOSED) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
return new Promise((resolve) => socket.once('close', resolve))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user