/* eslint-disable max-lines -- Why: this file is the single security boundary for the bundled CLI — transport setup, auth-token enforcement, admission control, keepalive framing, and orphan-socket sweeping all co-locate deliberately so a reviewer can audit the boundary in one sitting. Splitting this across files would scatter the invariants without reducing complexity. */ // Why: this is the single security boundary for the bundled CLI. It owns // auth-token enforcement, bootstrap-metadata publication, and transport // orchestration so a running runtime is always discoverable via exactly // one on-disk file. Method handling lives in `rpc/` and transport specifics // live in `rpc/unix-socket-transport.ts` and `rpc/ws-transport.ts`. import { randomBytes } from 'crypto' import { readdirSync, rmSync } from 'fs' import { join } from 'path' import type { RuntimeMetadata, RuntimeTransportMetadata } from '../../shared/runtime-bootstrap' import type { OrcaRuntimeService } from './orca-runtime' import { writeRuntimeMetadata } from './runtime-metadata' import { RpcDispatcher } from './rpc/dispatcher' import type { RpcRequest, RpcResponse } from './rpc/core' import { errorResponse } from './rpc/errors' import type { RpcMessageContext, RpcTransport } from './rpc/transport' import { UnixSocketTransport } from './rpc/unix-socket-transport' import { WebSocketTransport } from './rpc/ws-transport' import type { WebSocket } from 'ws' import { DeviceRegistry, type DeviceScope } from './device-registry' import { loadOrCreateE2EEKeypair, type E2EEKeypair } from './e2ee-keypair' import { E2EEChannel } from './rpc/e2ee-channel' import { encodePairingOffer, PAIRING_OFFER_VERSION } from '../../shared/pairing' import { decodeTerminalStreamFrame, type TerminalStreamFrame } from '../../shared/terminal-stream-protocol' const DEFAULT_WS_PORT = 6768 type OrcaRuntimeRpcServerOptions = { runtime: OrcaRuntimeService userDataPath: string pid?: number platform?: NodeJS.Platform enableWebSocket?: boolean wsPort?: number webClientRoot?: string // Why: test-only overrides for the two time-bound constants below. // Production callers must not pass these — defaults are set by the design // doc (§3.1) and changing them in production would weaken the admission // fence or flood the socket with keepalive frames. keepaliveIntervalMs?: number longPollCap?: number } // Why: after 10 s of a pending dispatch we emit a tiny `{"_keepalive":true}` // frame every 10 s until the handler resolves. Each write resets both the // server's own socket idle timer (30 s) and — once §3.1 ships on the client — // the client's idle timer, because any byte counts as socket activity. This // is the transport-layer fix for feedback #1: long-poll RPCs (i.e. // orchestration.check --wait) can now run past the 30 s/60 s idle caps // without either end tearing the socket down. See design doc §3.1. const KEEPALIVE_INTERVAL_MS = 10_000 // Why: long-poll slot cap. With keepalives a `check --wait --timeout-ms // 600000` can hold a connection for up to 10 minutes; unbounded that would // saturate MAX_RUNTIME_RPC_CONNECTIONS (32) with 32 waiting coordinators // and lock out normal short RPCs. Capping at half the connection budget // leaves the other half for short traffic. On overflow the server responds // immediately with `runtime_busy` (CLI exit 75) — fail fast, not silent // queuing. See design doc §3.1 + §7 risk #2. const LONG_POLL_CAP = 16 function resolvePairingEndpoint(rawEndpoint: string, address: string | null | undefined): string { const endpoint = new URL(rawEndpoint) const override = address?.trim() if (!override) { endpoint.hostname = '127.0.0.1' return formatWebSocketUrl(endpoint) } if (/^wss?:\/\//i.test(override)) { return formatWebSocketUrl(new URL(override)) } const parsed = parsePairingAddressOverride(override) endpoint.hostname = parsed.host.includes(':') ? `[${parsed.host.replace(/^\[|\]$/g, '')}]` : parsed.host if (parsed.port) { endpoint.port = parsed.port } return formatWebSocketUrl(endpoint) } function parsePairingAddressOverride(address: string): { host: string; port: string | null } { if (address.startsWith('[') || address.split(':').length === 2) { try { const parsed = new URL(`ws://${address}`) return { host: parsed.hostname.replace(/^\[|\]$/g, ''), port: parsed.port || null } } catch { return { host: address, port: null } } } return { host: address, port: null } } function formatWebSocketUrl(url: URL): string { const formatted = url.toString() return url.pathname === '/' && !url.search && !url.hash ? formatted.replace(/\/$/, '') : formatted } function createWebClientUrl(endpoint: string, pairingUrl: string): string { const url = new URL(endpoint) url.protocol = url.protocol === 'wss:' ? 'https:' : 'http:' url.pathname = webClientPathForEndpoint(url.pathname) url.search = '' // Why: pairing URLs include full runtime credentials. Keeping them in the // fragment avoids proxy logs and Referer headers while the web app loads. url.hash = `pairing=${encodeURIComponent(pairingUrl)}` return url.toString() } function webClientPathForEndpoint(pathname: string): string { if (!pathname || pathname === '/') { return '/web-index.html' } return `${pathname.replace(/\/$/, '')}/web-index.html` } const MOBILE_RPC_METHOD_ALLOWLIST = new Set([ 'accounts.list', 'accounts.selectClaude', 'accounts.selectCodex', 'accounts.subscribe', 'accounts.unsubscribe', 'browser.back', 'browser.dialogAccept', 'browser.dialogDismiss', 'browser.forward', 'browser.goto', 'browser.keyboardInsertText', 'browser.keypress', 'browser.mouseDown', 'browser.mouseClick', 'browser.mouseMove', 'browser.mouseUp', 'browser.mouseWheel', 'browser.reload', 'browser.screencast', 'browser.screencast.unsubscribe', 'browser.tabCreate', 'browser.viewport', 'files.list', 'files.open', 'files.openDiff', 'files.read', 'git.bulkStage', 'git.bulkUnstage', 'git.commit', 'git.discard', 'git.diff', 'git.fetch', 'git.pull', 'git.push', 'git.stage', 'git.status', 'git.unstage', 'git.upstreamStatus', 'github.createIssue', 'github.addIssueComment', 'github.addPRReviewComment', 'github.addPRReviewCommentReply', 'github.countWorkItems', 'github.listAssignableUsers', 'github.listLabels', 'github.listWorkItems', 'github.mergePR', 'github.requestPRReviewers', 'github.project.listAccessible', 'github.project.listAssignableUsersBySlug', 'github.project.listIssueTypesBySlug', 'github.project.listLabelsBySlug', 'github.project.listViews', 'github.project.resolveRef', 'github.project.addIssueCommentBySlug', 'github.project.updateIssueCommentBySlug', 'github.project.deleteIssueCommentBySlug', 'github.project.clearItemField', 'github.project.updateIssueBySlug', 'github.project.updateIssueTypeBySlug', 'github.project.updateItemField', 'github.project.updatePullRequestBySlug', 'github.project.viewTable', 'github.project.workItemDetailsBySlug', 'github.prFileContents', 'github.prChecks', 'github.rerunPRChecks', 'github.resolveReviewThread', 'github.setPRFileViewed', 'github.updateIssue', 'github.updatePR', 'github.updatePRTitle', 'github.updatePRState', 'github.repoSlug', 'github.workItem', 'github.workItemDetails', 'gitlab.createIssue', 'gitlab.addIssueComment', 'gitlab.addMRComment', 'gitlab.listWorkItems', 'gitlab.mergeMR', 'gitlab.todos', 'gitlab.updateIssue', 'gitlab.updateMR', 'gitlab.updateMRState', 'gitlab.workItemDetails', 'linear.getIssue', 'linear.addIssueComment', 'linear.connect', 'linear.createIssue', 'linear.issueComments', 'linear.listIssues', 'linear.listProjects', 'linear.teamLabels', 'linear.teamMembers', 'linear.listTeams', 'linear.searchIssues', 'linear.selectWorkspace', 'linear.status', 'linear.teamStates', 'linear.updateIssue', 'markdown.readTab', 'markdown.saveTab', 'notifications.subscribe', 'notifications.unsubscribe', 'preflight.check', 'preflight.detectAgents', 'preflight.detectRemoteAgents', 'repo.hooks', 'repo.list', 'repo.saveSparsePreset', 'repo.searchRefs', 'repo.sparsePresets', 'repo.update', 'session.tabs.activate', 'session.tabs.close', 'session.tabs.createTerminal', 'session.tabs.list', 'session.tabs.listAll', 'session.tabs.move', 'session.tabs.subscribe', 'session.tabs.subscribeAll', 'session.tabs.unsubscribe', 'settings.get', 'settings.update', 'ssh.connect', 'ssh.getState', 'speech.dictation.cancel', 'speech.dictation.chunk', 'speech.dictation.finish', 'speech.dictation.start', 'stats.summary', 'status.get', 'terminal.clearBuffer', 'terminal.close', 'terminal.create', 'terminal.focus', 'terminal.getAutoRestoreFit', 'terminal.list', 'terminal.multiplex', 'terminal.read', 'terminal.rename', 'terminal.send', 'terminal.setAutoRestoreFit', 'terminal.setDisplayMode', 'terminal.subscribe', 'terminal.unsubscribe', 'terminal.updateViewport', 'ui.get', 'ui.set', 'worktree.activate', 'worktree.create', 'worktree.ps', 'worktree.resolveMrBase', 'worktree.resolvePrBase', 'worktree.rm', 'worktree.set', 'worktree.sleep' ]) // Why: a long-poll request is one whose handler blocks waiting for an external // event. This function is the single place that classifies it — the long-poll // counter, abort wiring, keepalives, and runtime_busy admission check all // share this decision. See §3.1. function isLongPollRequest(request: RpcRequest): boolean { if (request.method === 'terminal.wait') { return true } if (request.method === 'orchestration.check') { const params = request.params as { wait?: unknown } | undefined return params?.wait === true } return false } export class OrcaRuntimeRpcServer { private readonly runtime: OrcaRuntimeService private readonly dispatcher: RpcDispatcher private readonly userDataPath: string private readonly pid: number private readonly platform: NodeJS.Platform private readonly enableWebSocket: boolean private readonly wsPort: number private readonly webClientRoot: string | undefined private readonly authToken = randomBytes(24).toString('hex') private readonly keepaliveIntervalMs: number private readonly longPollCap: number private deviceRegistry: DeviceRegistry | null = null private e2eeKeypair: E2EEKeypair | null = null private tlsFingerprint: string | null = null private wsTransport: WebSocketTransport | null = null private activeTransports: RpcTransport[] = [] private transports: RuntimeTransportMetadata[] = [] // Why: each WebSocket connection has its own E2EE channel that manages the // handshake and encrypt/decrypt lifecycle. Keyed by WebSocket instance. private e2eeChannels = new Map() // Why: stable per-WebSocket id used as the cleanup key for streaming // subscriptions, so the server can reap a closing socket's subscriptions // without affecting other live sockets that share the same deviceToken. private wsConnectionIds = new Map() private readonly binaryStreamHandlers = new Map< string, Map void> >() private readonly wsDispatchAbortStates = new Map< WebSocket, { controllers: Set; abortOnClose: () => void } >() // Why: separate from Node's server.maxConnections because we need to count // only long-running dispatches, not every in-flight short RPC. See §3.1 + // §7 risk #2. private activeLongPolls = 0 constructor({ runtime, userDataPath, pid = process.pid, platform = process.platform, enableWebSocket = false, wsPort = DEFAULT_WS_PORT, webClientRoot, keepaliveIntervalMs = KEEPALIVE_INTERVAL_MS, longPollCap = LONG_POLL_CAP }: OrcaRuntimeRpcServerOptions) { this.runtime = runtime this.dispatcher = new RpcDispatcher({ runtime }) this.userDataPath = userDataPath this.pid = pid this.platform = platform this.enableWebSocket = enableWebSocket this.wsPort = wsPort this.webClientRoot = webClientRoot this.keepaliveIntervalMs = keepaliveIntervalMs this.longPollCap = longPollCap } getDeviceRegistry(): DeviceRegistry | null { return this.deviceRegistry } getTlsFingerprint(): string | null { return this.tlsFingerprint } getE2EEPublicKey(): string | null { return this.e2eeKeypair?.publicKeyB64 ?? null } getE2EEKeypair(): E2EEKeypair | null { return this.e2eeKeypair } revokeMobileDevice(deviceId: string): boolean { const device = this.deviceRegistry?.getDevice(deviceId) if (device?.scope !== 'mobile' || !this.deviceRegistry?.removeDevice(deviceId)) { return false } this.wsTransport?.terminateClientConnections(device.token) return true } revokeRuntimeAccess(deviceId: string): boolean { const device = this.deviceRegistry?.getDevice(deviceId) if (device?.scope !== 'runtime' || !this.deviceRegistry?.removeDevice(deviceId)) { return false } this.wsTransport?.terminateClientConnections(device.token) return true } getWebSocketEndpoint(): string | null { const ws = this.transports.find((t) => t.kind === 'websocket') return ws?.endpoint ?? null } createPairingOffer(args: { address?: string | null name?: string rotate?: boolean scope?: DeviceScope }): | { available: false } | { available: true pairingUrl: string endpoint: string deviceId: string webClientUrl: string | null } { const rawEndpoint = this.getWebSocketEndpoint() const publicKeyB64 = this.getE2EEPublicKey() if (!rawEndpoint || !this.deviceRegistry || !publicKeyB64) { return { available: false } } const endpoint = resolvePairingEndpoint(rawEndpoint, args.address) const deviceName = args.name ?? `CLI ${new Date().toLocaleDateString()}` const scope = args.scope ?? 'runtime' const device = args.rotate ? this.deviceRegistry.rotatePendingDevice(deviceName, scope) : this.deviceRegistry.getOrCreatePendingDevice(deviceName, scope) const pairingUrl = encodePairingOffer({ v: PAIRING_OFFER_VERSION, endpoint, deviceToken: device.token, publicKeyB64 }) return { available: true, pairingUrl, endpoint, deviceId: device.deviceId, webClientUrl: this.webClientRoot && scope === 'runtime' ? createWebClientUrl(endpoint, pairingUrl) : null } } private registerBinaryStreamHandler( connectionId: string | undefined, streamId: number, handler: (frame: TerminalStreamFrame) => void ): () => void { if (!connectionId || !Number.isInteger(streamId) || streamId < 0) { return () => {} } let handlers = this.binaryStreamHandlers.get(connectionId) if (!handlers) { handlers = new Map() this.binaryStreamHandlers.set(connectionId, handlers) } handlers.set(streamId, handler) return () => { const current = this.binaryStreamHandlers.get(connectionId) if (!current || current.get(streamId) !== handler) { return } current.delete(streamId) if (current.size === 0) { this.binaryStreamHandlers.delete(connectionId) } } } private handleWebSocketBinaryMessage(bytes: Uint8Array, ws: WebSocket): void { const connectionId = this.wsConnectionIds.get(ws) if (!connectionId) { return } const frame = decodeTerminalStreamFrame(bytes) if (!frame) { return } this.binaryStreamHandlers.get(connectionId)?.get(frame.streamId)?.(frame) } private registerWebSocketDispatchAbort(ws: WebSocket): { signal: AbortSignal dispose: () => void } { const abortController = new AbortController() if (ws.readyState !== ws.OPEN) { abortController.abort() return { signal: abortController.signal, dispose: () => {} } } let state = this.wsDispatchAbortStates.get(ws) if (!state) { state = { controllers: new Set(), abortOnClose: () => this.abortWebSocketDispatches(ws) } this.wsDispatchAbortStates.set(ws, state) // Why: many streaming RPCs can share one WebSocket. A single socket-level // abort fan-out avoids MaxListenersExceededWarning while preserving cleanup. ws.on('close', state.abortOnClose) ws.on('error', state.abortOnClose) } state.controllers.add(abortController) return { signal: abortController.signal, dispose: () => { const current = this.wsDispatchAbortStates.get(ws) if (!current) { return } current.controllers.delete(abortController) if (current.controllers.size > 0) { return } this.wsDispatchAbortStates.delete(ws) ws.off('close', current.abortOnClose) ws.off('error', current.abortOnClose) } } } private abortWebSocketDispatches(ws: WebSocket): void { const state = this.wsDispatchAbortStates.get(ws) if (!state) { return } this.wsDispatchAbortStates.delete(ws) ws.off('close', state.abortOnClose) ws.off('error', state.abortOnClose) for (const controller of state.controllers) { controller.abort() } state.controllers.clear() } async start(): Promise { if (this.activeTransports.length > 0) { return } // Why: processes killed by SIGKILL / OOM-kill / forced-shutdown skip // stop() and leave behind `o--*.sock` files in userData. Sweeping // dead-pid sockets at startup keeps the directory from accumulating // orphans over the app's lifetime. Named-pipe transports on Windows do // not leave filesystem entries in userData, so the sweep is a no-op // there. if (this.platform !== 'win32') { sweepOrphanedRuntimeSockets(this.userDataPath, this.pid) } const transportMeta = createRuntimeTransportMetadata( this.userDataPath, this.pid, this.platform, this.runtime.getRuntimeId() ) const socketTransport = new UnixSocketTransport({ endpoint: transportMeta.endpoint, kind: transportMeta.kind as 'unix' | 'named-pipe', keepaliveIntervalMs: this.keepaliveIntervalMs }) // Why: Unix socket transport uses the shared runtime auth token. This is // the existing security model for CLI connections — the token lives in a // 0o600-permissioned file on disk. // Why: the `.catch` guarantees `reply()` always fires even if // `handleMessage` (or `JSON.stringify` on a pathological response) throws. // Without it, a throw would leave the client waiting for a terminal frame // that never arrives AND leak the dispatch's AbortController in the // transport's in-flight set until the 30 s socket idle timer closes the // connection. socketTransport.onMessage((msg, reply, context) => { void this.handleMessage(msg, context) .then((response) => { reply(JSON.stringify(response)) }) .catch((error) => { const message = error instanceof Error ? error.message : String(error) // Why: best-effort id recovery so the client can correlate the // error frame to its pending request. A malformed message would // have been caught by handleMessage and returned an envelope // instead of throwing, so in practice the id is always present. let id = 'unknown' try { const parsed = JSON.parse(msg) as { id?: unknown } if (typeof parsed.id === 'string' && parsed.id.length > 0) { id = parsed.id } } catch { // ignore — fall through with id='unknown' } reply(JSON.stringify(this.buildError(id, 'internal_error', message))) }) }) await socketTransport.start() const activeTransports: RpcTransport[] = [socketTransport] const transportsMeta: RuntimeTransportMetadata[] = [transportMeta] // Why: WebSocket transport is opt-in and starts alongside the Unix socket. // It uses per-device tokens and E2EE (application-layer encryption via // tweetnacl) rather than TLS, since React Native can't pin self-signed certs. if (this.enableWebSocket) { try { this.deviceRegistry = new DeviceRegistry(this.userDataPath) this.e2eeKeypair = loadOrCreateE2EEKeypair(this.userDataPath) const wsTransport = new WebSocketTransport({ host: '0.0.0.0', port: this.wsPort, staticRoot: this.webClientRoot }) this.wsTransport = wsTransport // Why: each WebSocket connection gets an E2EE channel that handles the // handshake before any RPC messages are processed. The channel decrypts // inbound messages and encrypts outbound replies transparently. wsTransport.onMessage((msg, _reply, ws) => { let channel = this.e2eeChannels.get(ws) if (!channel) { // Why: stable per-ws id used as the cleanup-index key for // streaming subscriptions, so the server can reap them exactly // when this socket closes (without affecting other live sockets // that share the same deviceToken). this.wsConnectionIds.set(ws, randomBytes(8).toString('hex')) channel = new E2EEChannel(ws, { serverSecretKey: this.e2eeKeypair!.secretKey, validateToken: (token) => this.deviceRegistry?.validateToken(token) != null, onReady: (ch) => { if (ch.deviceToken) { wsTransport.setClientId(ws, ch.deviceToken) // Why: mark the device as actually connected so it appears // in the "Paired Devices" list. Devices that were only // generated as QR codes but never scanned stay hidden. const device = this.deviceRegistry?.validateToken(ch.deviceToken) if (device) { this.deviceRegistry?.updateLastSeen(device.deviceId) } } }, onError: (code, reason) => { this.e2eeChannels.get(ws)?.destroy() this.e2eeChannels.delete(ws) ws.close(code, reason) } }) channel.onMessage((plaintext, encryptedReply, encryptedBinaryReply) => { const authenticatedDeviceToken = this.e2eeChannels.get(ws)?.deviceToken ?? null void this.handleWebSocketMessage( plaintext, encryptedReply, encryptedBinaryReply, wsTransport, ws, authenticatedDeviceToken ) }) channel.onBinaryMessage((bytes) => this.handleWebSocketBinaryMessage(bytes, ws)) this.e2eeChannels.set(ws, channel) } channel.handleRawMessage(msg) }) // Why: when a mobile client disconnects, the runtime must clean up // connection-scoped state like mobile-fit overrides and the E2EE // channel to prevent orphaned state. A single paired device can hold // multiple concurrent sockets (host screen + accounts screen, etc.), // so destroy the channel for THIS exact ws and skip the per-client // teardown when other sockets for the same token are still alive. wsTransport.onConnectionClose((clientId, ws, hasOtherConnections) => { this.abortWebSocketDispatches(ws) // Why: sweep streaming subscriptions for THIS ws regardless of // hasOtherConnections, so per-ws listeners (notifications, // accounts, terminal) don't leak across reconnects. This is // independent of the deviceToken-scoped onClientDisconnected. const connectionId = this.wsConnectionIds.get(ws) if (connectionId) { this.runtime.cleanupSubscriptionsForConnection(connectionId) this.runtime.cancelMobileDictationForConnection(connectionId) this.binaryStreamHandlers.delete(connectionId) this.wsConnectionIds.delete(ws) } const channel = this.e2eeChannels.get(ws) if (channel) { channel.destroy() this.e2eeChannels.delete(ws) } if (clientId && !hasOtherConnections) { this.runtime.onClientDisconnected(clientId) } }) await wsTransport.start() activeTransports.push(wsTransport) transportsMeta.push({ kind: 'websocket', endpoint: `ws://0.0.0.0:${wsTransport.resolvedPort}` }) } catch (error) { // Why: WebSocket transport is supplementary — the runtime must still // function if it fails to start (e.g., port in use). Log and continue // with Unix socket only. console.error('[runtime] Failed to start WebSocket transport:', error) this.wsTransport = null } } // Why: publish the transport into in-memory state before writing metadata // so the bootstrap file always contains the real endpoint/token pair. The // CLI only discovers the runtime through that file. this.activeTransports = activeTransports this.transports = transportsMeta try { this.writeMetadata() } catch (error) { // Why: a runtime that cannot publish bootstrap metadata is invisible to // the `orca` CLI. Close all transports immediately instead of leaving // behind a live but undiscoverable control plane. this.activeTransports = [] this.transports = [] await Promise.all(activeTransports.map((t) => t.stop().catch(() => {}))).catch(() => {}) throw error } } async stop(): Promise { const transports = this.activeTransports this.activeTransports = [] this.transports = [] this.wsTransport = null if (transports.length === 0) { return } await Promise.all(transports.map((t) => t.stop())) // Why: we intentionally leave the last metadata file behind instead of // deleting it on shutdown. Shared userData paths can briefly host multiple // Orca processes during restarts, updates, or development, and stale // metadata is safer than letting one process erase another live runtime's // bootstrap file. } // Why: Unix socket messages use one-shot dispatch (single response per // request) and the shared runtime auth token from the 0o600 metadata file. // The transport layer owns socket lifecycle, keepalive writes, and the // per-connection abort signal — this method just parses, auths, and // dispatches. See design doc §3.1. private async handleMessage( rawMessage: string, context?: RpcMessageContext ): Promise { // Why: empty messages are sent by the Unix socket transport layer when a // client exceeds the max message size. The transport closes the connection // after this response. if (!rawMessage) { return this.buildError('unknown', 'request_too_large', 'RPC request exceeds the maximum size') } const parsed = this.parseAndAuth(rawMessage) if ('error' in parsed) { return parsed.error } const request = parsed.request // Why: long-poll admission fence. Short RPCs bypass the counter entirely // — it only guards handlers that can block for minutes. See §7 risk #2. const longPoll = isLongPollRequest(request) if (longPoll && this.activeLongPolls >= this.longPollCap) { return this.buildError( request.id, 'runtime_busy', 'long-poll capacity reached; retry with backoff' ) } if (longPoll) { this.activeLongPolls += 1 // Why: arm the keepalive timer only for long-polls. Short RPCs never // touch it so the `setInterval` is never created. See §3.1. context?.startKeepalive() } try { return await this.dispatcher.dispatch(request, { signal: longPoll ? context?.signal : undefined }) } finally { if (longPoll) { this.activeLongPolls = Math.max(0, this.activeLongPolls - 1) } } } private parseAndAuth(rawMessage: string): { request: RpcRequest } | { error: RpcResponse } { let request: RpcRequest try { request = JSON.parse(rawMessage) as RpcRequest } catch { return { error: this.buildError('unknown', 'bad_request', 'Invalid JSON request') } } if (typeof request.id !== 'string' || request.id.length === 0) { return { error: this.buildError('unknown', 'bad_request', 'Missing request id') } } if (typeof request.method !== 'string' || request.method.length === 0) { return { error: this.buildError(request.id, 'bad_request', 'Missing RPC method') } } if (typeof request.authToken !== 'string' || request.authToken.length === 0) { return { error: this.buildError(request.id, 'unauthorized', 'Missing auth token') } } if (request.authToken !== this.authToken) { return { error: this.buildError(request.id, 'unauthorized', 'Invalid auth token') } } return { request } } // Why: WebSocket messages go through streaming dispatch which can emit // multiple responses. Auth uses per-device tokens from the device registry. private async handleWebSocketMessage( rawMessage: string, reply: (response: string) => void, sendBinary: (response: Uint8Array) => boolean | void, wsTransport?: WebSocketTransport, ws?: WebSocket, authenticatedDeviceToken?: string | null ): Promise { let request: RpcRequest try { request = JSON.parse(rawMessage) as RpcRequest } catch { reply(JSON.stringify(this.buildError('unknown', 'bad_request', 'Invalid JSON request'))) return } if (typeof request.id !== 'string' || request.id.length === 0) { reply(JSON.stringify(this.buildError('unknown', 'bad_request', 'Missing request id'))) return } if (typeof request.method !== 'string' || request.method.length === 0) { reply(JSON.stringify(this.buildError(request.id, 'bad_request', 'Missing RPC method'))) return } const requestToken = typeof (request as Record).deviceToken === 'string' ? ((request as Record).deviceToken as string) : null if (authenticatedDeviceToken && requestToken && requestToken !== authenticatedDeviceToken) { reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Device token mismatch'))) return } // Why: E2EE already authenticated the WebSocket channel. Use that bound // identity for authorization instead of trusting a repeated request field. const token = authenticatedDeviceToken ?? requestToken if (!token) { reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Missing device token'))) return } const device = this.deviceRegistry?.validateToken(token) if (!device) { reply(JSON.stringify(this.buildError(request.id, 'unauthorized', 'Invalid device token'))) return } if (device.scope === 'mobile' && !MOBILE_RPC_METHOD_ALLOWLIST.has(request.method)) { reply( JSON.stringify( this.buildError( request.id, 'forbidden', `Method '${request.method}' is not available to mobile clients` ) ) ) return } // Why: associate the deviceToken with this WebSocket so ws.on('close') // can notify the runtime which mobile client disconnected. if (wsTransport && ws) { wsTransport.setClientId(ws, token) } const longPoll = isLongPollRequest(request) if (longPoll && this.activeLongPolls >= this.longPollCap) { reply( JSON.stringify( this.buildError( request.id, 'runtime_busy', 'long-poll capacity reached; retry with backoff' ) ) ) return } const abortRegistration = ws ? this.registerWebSocketDispatchAbort(ws) : null if (longPoll) { this.activeLongPolls += 1 } const connectionId = ws ? this.wsConnectionIds.get(ws) : undefined try { await this.dispatcher.dispatchStreaming(request, reply, { connectionId, clientId: token, signal: abortRegistration?.signal, sendBinary, registerBinaryStreamHandler: (streamId, handler) => this.registerBinaryStreamHandler(connectionId, streamId, handler) }) } finally { abortRegistration?.dispose() if (longPoll) { this.activeLongPolls = Math.max(0, this.activeLongPolls - 1) } } } private buildError(id: string, code: string, message: string): RpcResponse { return errorResponse(id, { runtimeId: this.runtime.getRuntimeId() }, code, message) } private writeMetadata(): void { const metadata: RuntimeMetadata = { runtimeId: this.runtime.getRuntimeId(), pid: this.pid, transports: this.transports, authToken: this.authToken, startedAt: this.runtime.getStartedAt() } writeRuntimeMetadata(this.userDataPath, metadata) } } /** * Why: the regex MUST stay in lockstep with createRuntimeTransportMetadata() * below, which emits `o-${pid}-${endpointSuffix}.sock` where endpointSuffix * is `[A-Za-z0-9_-]{1,4}` (derived from a sanitised runtimeId prefix, or * `'rt'` as the fallback). The invariant is covered by a unit test so any * future change to the transport-name shape trips CI. */ export const RUNTIME_SOCKET_NAME_REGEX = /^o-(\d+)-[A-Za-z0-9_-]+\.sock$/ export function sweepOrphanedRuntimeSockets(userDataPath: string, ownPid: number): void { let entries: string[] try { entries = readdirSync(userDataPath) } catch { // Why: first-launch userData may not exist yet; the cold-start path // below will create it. Nothing to sweep in that case. return } for (const entry of entries) { const match = RUNTIME_SOCKET_NAME_REGEX.exec(entry) if (!match) { continue } const pid = Number(match[1]) if (!Number.isFinite(pid)) { continue } // Why: never touch the current process's socket. start() already // rmSync's it if it exists, but belt-and-braces — a bug in the own-pid // path here would rmSync a socket we're about to bind to. if (pid === ownPid) { continue } try { // Why: signal 0 is the POSIX liveness probe — it delivers no signal // but returns success iff the pid resolves AND the caller has // permission to signal it. ESRCH = no such process; EPERM = pid // exists but owned by another user, which is extremely unusual on a // desktop app's userData dir but we conservatively leave those // sockets alone. process.kill(pid, 0) } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ESRCH') { try { rmSync(join(userDataPath, entry), { force: true }) } catch { // Why: best-effort sweep — a permission error on unlink is fine // to ignore; the socket will be cleaned by a later start() or // by the OS on reboot. } } } } } export function createRuntimeTransportMetadata( userDataPath: string, pid: number, platform: NodeJS.Platform, runtimeId = 'runtime' ): RuntimeTransportMetadata { const endpointSuffix = runtimeId.replace(/[^a-zA-Z0-9_-]/g, '').slice(0, 4) || 'rt' if (platform === 'win32') { return { kind: 'named-pipe', // Why: Windows named pipes do not get the same chmod hardening path as // Unix sockets, so include a per-runtime suffix to avoid exposing a // stable, guessable control endpoint name across launches. endpoint: `\\\\.\\pipe\\orca-${pid}-${endpointSuffix}` } } return { kind: 'unix', endpoint: join(userDataPath, `o-${pid}-${endpointSuffix}.sock`) } }