From 07272c02599cba689757a03aebfa151ded031243 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Thu, 8 Oct 2026 00:14:50 -0400 Subject: [PATCH] feat(mobile): relay a paired phone's server-workspace requests through the desktop A phone mirrors its desktop, including workspaces on the desktop's configured servers. The desktop now forwards a request that names `executionHost: runtime:` to that server over one socket per (phone, server), signed in as the phone's delegated device; replies pass through untouched except stream ids. Hidden until the phone client ships: the relay capability is not advertised yet. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb --- .../runtime-environment-mobile-relay-hosts.ts | 40 ++ src/main/ipc/runtime-environments.ts | 2 + .../delegated-phone-grants.ts | 66 +++ .../mobile-desktop-relay-hosts.ts | 25 + .../mobile-desktop-relay.integration.test.ts | 445 ++++++++++++++++++ .../mobile-desktop-relay.test.ts | 140 ++++++ .../mobile-desktop-relay.ts | 287 +++++++++++ .../relayed-phone-frames.test.ts | 120 +++++ .../relayed-phone-frames.ts | 119 +++++ .../relayed-stream-carriers.test.ts | 24 + .../relayed-stream-carriers.ts | 19 + .../runtime-rpc/runtime-rpc-lifecycle.ts | 1 + .../runtime-rpc-mobile-method-routing.test.ts | 23 + .../runtime-rpc-mobile-method-routing.ts | 227 +++++++++ ...ntime-rpc-mobile-paired-desktop-methods.ts | 103 ++++ .../runtime-rpc/runtime-rpc-pairing.ts | 32 ++ .../runtime/runtime-rpc/runtime-rpc-state.ts | 3 + .../runtime-rpc-websocket-dispatch.ts | 61 +++ ...-process-runtime-launch-activation.test.ts | 4 + .../startup/main-process-runtime-launch.ts | 2 + .../delegated-mobile-device-contract.ts | 33 ++ src/shared/mobile-desktop-relay-contract.ts | 7 + .../remote-runtime-passthrough-socket.ts | 224 +++++++++ src/shared/terminal-stream-protocol.ts | 17 + 24 files changed, 2024 insertions(+) create mode 100644 src/main/ipc/runtime-environment-mobile-relay-hosts.ts create mode 100644 src/main/runtime/mobile-desktop-relay/delegated-phone-grants.ts create mode 100644 src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts create mode 100644 src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts create mode 100644 src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.test.ts create mode 100644 src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.ts create mode 100644 src/main/runtime/mobile-desktop-relay/relayed-phone-frames.test.ts create mode 100644 src/main/runtime/mobile-desktop-relay/relayed-phone-frames.ts create mode 100644 src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.test.ts create mode 100644 src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.ts create mode 100644 src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.test.ts create mode 100644 src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.ts create mode 100644 src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts create mode 100644 src/shared/delegated-mobile-device-contract.ts create mode 100644 src/shared/mobile-desktop-relay-contract.ts create mode 100644 src/shared/remote-runtime-passthrough-socket.ts diff --git a/src/main/ipc/runtime-environment-mobile-relay-hosts.ts b/src/main/ipc/runtime-environment-mobile-relay-hosts.ts new file mode 100644 index 00000000000..a18bfbfa37f --- /dev/null +++ b/src/main/ipc/runtime-environment-mobile-relay-hosts.ts @@ -0,0 +1,40 @@ +import { app } from 'electron' +import { listEnvironments } from '../../shared/runtime-environment-store' +import { getPreferredPairingOffer } from '../../shared/runtime-environments' +import type { MobileDesktopRelayHosts } from '../runtime/mobile-desktop-relay/mobile-desktop-relay-hosts' +import { resolveManagedRuntimeEnvironment } from './runtime-environment-managed-tunnel' +import { callRuntimeEnvironment } from './runtime-environment-transport-routing' + +const retirementListeners = new Set<(environmentId: string) => void>() + +/** Called wherever the desktop's own transport to a server is invalidated. */ +export function retireMobileDesktopRelayEnvironment(environmentId: string): void { + for (const listener of retirementListeners) { + listener(environmentId) + } +} + +/** The desktop's configured servers, addressed only by exact environment id. */ +export function createRuntimeEnvironmentMobileRelayHosts(): MobileDesktopRelayHosts { + // Why userData: configured servers live where the runtime-environment handlers keep them. + const userDataPath = app.getPath('userData') + return { + resolve: async (environmentId) => { + if (!listEnvironments(userDataPath).some((entry) => entry.id === environmentId)) { + return null + } + const environment = await resolveManagedRuntimeEnvironment(userDataPath, environmentId) + return { + environmentId, + fence: `${environment.pairingRevision ?? environment.createdAt}\0${environment.runtimeId}`, + pairing: getPreferredPairingOffer(environment) + } + }, + call: (host, method, params) => + callRuntimeEnvironment(userDataPath, host.environmentId, method, params), + onEnvironmentRetired: (listener) => { + retirementListeners.add(listener) + return () => retirementListeners.delete(listener) + } + } +} diff --git a/src/main/ipc/runtime-environments.ts b/src/main/ipc/runtime-environments.ts index 9e3506eb5fb..3cc62903e90 100644 --- a/src/main/ipc/runtime-environments.ts +++ b/src/main/ipc/runtime-environments.ts @@ -29,6 +29,7 @@ import { registerRuntimeSshAccessHandlers } from './runtime-ssh-access-handlers' import { retirePairedRuntimeBrowserClientHostEnvironment } from '../browser/paired-runtime-browser-client-host-runtime' import { registerRuntimeEnvironmentBrowserClientHostHandler } from './runtime-environment-browser-client-host-handler' import { advanceRuntimeEnvironmentCapabilityIncarnation } from './runtime-environment-capability-evidence' +import { retireMobileDesktopRelayEnvironment } from './runtime-environment-mobile-relay-hosts' import { watchRuntimeEnvironmentPreference } from './runtime-environment-preference' const remoteRuntimeSubscriptions = new Map() @@ -71,6 +72,7 @@ export function invalidateRuntimeEnvironmentTransport(environmentId: string): Pr advanceRuntimeEnvironmentTransportGeneration(environmentId) closeRemoteRuntimeRequestConnection(environmentId) closeSubscriptionsForEnvironment(environmentId) + retireMobileDesktopRelayEnvironment(environmentId) return retirePairedRuntimeBrowserClientHostEnvironment( environmentId, new Error('Runtime environment transport was invalidated') diff --git a/src/main/runtime/mobile-desktop-relay/delegated-phone-grants.ts b/src/main/runtime/mobile-desktop-relay/delegated-phone-grants.ts new file mode 100644 index 00000000000..3240c3af1f8 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/delegated-phone-grants.ts @@ -0,0 +1,66 @@ +import { z } from 'zod' +import { + DELEGATED_MOBILE_DEVICE_SYNC_METHOD, + DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES, + DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY, + type DelegatedMobileDeviceSyncParams +} from '../../../shared/delegated-mobile-device-contract' +import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error' +import type { MobileDesktopRelayHost, MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts' + +const SyncResultSchema = z.object({ + devices: z.array(z.object({ phoneKey: z.string(), deviceId: z.string(), token: z.string() })) +}) +const StatusCapabilitiesSchema = z.object({ capabilities: z.array(z.string()).optional() }) + +export type DelegatedPhoneGrant = { deviceId: string; token: string } + +/** One sync's outcome for a host; held in memory only, so the next sync is the only recovery. */ +export type DelegatedPhoneGrants = + | { kind: 'update-needed'; fence: string } + | { + kind: 'ready' + fence: string + grants: ReadonlyMap + // Phones whose host token was refused (revoked on the host); not retried until the next sync. + refused: Set + } + +export async function syncDelegatedPhoneGrants( + hosts: MobileDesktopRelayHosts, + host: MobileDesktopRelayHost, + phones: DelegatedMobileDeviceSyncParams['phones'] +): Promise { + const status = await hosts.call(host, 'status.get', undefined) + if (!status.ok) { + throw new RemoteRuntimeClientError('remote_runtime_unavailable', status.error.message) + } + const capabilities = StatusCapabilitiesSchema.safeParse(status.result).data?.capabilities ?? [] + if (!capabilities.includes(DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY)) { + return { kind: 'update-needed', fence: host.fence } + } + const response = await hosts.call(host, DELEGATED_MOBILE_DEVICE_SYNC_METHOD, { + phones: phones.slice(0, DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES) + }) + if (!response.ok) { + throw new RemoteRuntimeClientError('remote_runtime_unavailable', response.error.message) + } + const result = SyncResultSchema.safeParse(response.result) + if (!result.success) { + throw new RemoteRuntimeClientError( + 'invalid_runtime_response', + 'Remote Orca runtime returned an invalid delegated device sync.' + ) + } + return { + kind: 'ready', + fence: host.fence, + grants: new Map( + result.data.devices.map((device) => [ + device.phoneKey, + { deviceId: device.deviceId, token: device.token } + ]) + ), + refused: new Set() + } +} diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts new file mode 100644 index 00000000000..ff7727f8c6e --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts @@ -0,0 +1,25 @@ +import type { PairingOffer } from '../../../shared/pairing' +import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope' + +/** A configured server the desktop can relay to. */ +export type MobileDesktopRelayHost = { + environmentId: string + // Pairing revision and runtime identity; delegated grants from another fence are never reused. + fence: string + // The desktop's own runtime-scope pairing; the relay swaps in a phone's delegated token. + pairing: PairingOffer +} + +/** The desktop's configured servers, as the relay sees them. */ +export type MobileDesktopRelayHosts = { + /** Null when the id names no configured server; a phone-supplied endpoint is never used. */ + resolve: (environmentId: string) => Promise + /** A call as the desktop itself, over its own connection to the server. */ + call: ( + host: MobileDesktopRelayHost, + method: string, + params: unknown + ) => Promise> + /** Fires when a server is removed, re-paired or disconnected; returns an unsubscribe. */ + onEnvironmentRetired: (listener: (environmentId: string) => void) => () => void +} diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts new file mode 100644 index 00000000000..6e0210ed0bc --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts @@ -0,0 +1,445 @@ +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import WebSocket from 'ws' +import { z } from 'zod' +import { + decrypt, + decryptBytes, + deriveSharedKey, + encrypt, + generateKeyPair, + publicKeyFromBase64, + publicKeyToBase64 +} from '../../../shared/e2ee-crypto' +import { parsePairingCode, type PairingOffer } from '../../../shared/pairing' +import { + DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY, + DelegatedMobileDeviceSyncParamsSchema +} from '../../../shared/delegated-mobile-device-contract' +import { decodeTerminalStreamFrame } from '../../../shared/terminal-stream-protocol' +import type { DeviceEntry } from '../device-registry' +import { OrcaRuntimeService } from '../orca-runtime' +import { readRuntimeMetadata } from '../runtime-metadata' +import { OrcaRuntimeRpcServer } from '../runtime-rpc' +import { sendRequest } from '../runtime-rpc-test-harness' +import { makeStore } from '../runtime-rpc-worktree-store-fixtures' +import type { MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts' +import type * as PassthroughSocketModule from '../../../shared/remote-runtime-passthrough-socket' + +const passthroughOpens = vi.hoisted(() => { + const hostReplies: string[] = [] + return { count: 0, hostReplies } +}) +vi.mock('../../../shared/remote-runtime-passthrough-socket', async (importOriginal) => { + const original = await importOriginal() + return { + ...original, + openRemoteRuntimePassthroughSocket: ( + ...[pairing, capabilities, callbacks, options]: Parameters< + typeof original.openRemoteRuntimePassthroughSocket + > + ) => { + passthroughOpens.count += 1 + return original.openRemoteRuntimePassthroughSocket( + pairing, + capabilities, + { + ...callbacks, + onText: (plaintext) => { + passthroughOpens.hostReplies.push(plaintext) + callbacks.onText(plaintext) + } + }, + options + ) + } + } +}) + +vi.mock('../../git/worktree', () => { + const worktrees = [ + { + path: '/tmp/worktree-a', + head: 'abc', + branch: 'feature/foo', + isBare: false, + isMainWorktree: false + } + ] + return { + listWorktrees: vi.fn().mockResolvedValue(worktrees), + listWorktreesStrict: vi.fn().mockResolvedValue(worktrees) + } +}) + +/** Exposes the per-terminal phone seats the runtime keeps behind a protected map. */ +class SeatObservableRuntime extends OrcaRuntimeService { + seatClientIds(ptyId: string): string[] { + return [...(this.mobileSubscribers.get(ptyId)?.keys() ?? [])] + } +} + +const WORKTREE = 'id:repo-1::/tmp/worktree-a' +const TARGET = 'runtime:env-1' +type Frame = Record + +describe('mobile desktop relay: phone -> desktop -> server', () => { + const cleanups: (() => Promise | void)[] = [] + afterEach(async () => { + for (const cleanup of cleanups.splice(0).toReversed()) { + await cleanup() + } + }) + + async function startRuntime(spawnedOnThisHost: string[] = []) { + const writes: string[] = [] + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the fixture store carries the surface these RPCs read, as in the sibling streaming tests. + const runtime = new SeatObservableRuntime(makeStore() as never) + runtime.setPtyController({ + spawn: vi.fn().mockImplementation(async () => { + spawnedOnThisHost.push('pty-shared') + return { id: 'pty-shared' } + }), + write: (_ptyId, data) => { + writes.push(data) + return true + }, + kill: () => true, + getForegroundProcess: async () => null, + resize: () => true, + getSize: () => ({ cols: 80, rows: 24 }) + }) + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-relay-')) + const server = new OrcaRuntimeRpcServer({ + runtime, + userDataPath, + enableWebSocket: true, + wsPort: 0 + }) + await server.start() + cleanups.push(() => server.stop()) + return { server, runtime, writes, userDataPath } + } + + function pair( + server: OrcaRuntimeRpcServer, + scope: 'mobile' | 'runtime', + name: string + ): PairingOffer { + const offer = server.createPairingOffer({ address: '127.0.0.1', name, scope }) + if (!offer.available) { + throw new Error('pairing unavailable') + } + return parsePairingCode(offer.pairingUrl)! + } + + /** A server, and a desktop whose only configured server is it, as `env-1`. */ + async function startTopology() { + const host = await startRuntime() + const desktopSpawns: string[] = [] + const desktop = await startRuntime(desktopSpawns) + const desktopOnHost = pair(host.server, 'runtime', 'MacBook') + const registry = host.server.getDeviceRegistry()! + const children = new Map() + const syncedNames: string[][] = [] + const state = { capable: true, retire: (_id: string) => {} } + const ok = (result: unknown) => ({ + id: 'x', + ok: true as const, + result, + _meta: { runtimeId: 'h' } + }) + const hosts: MobileDesktopRelayHosts = { + resolve: async (environmentId) => + environmentId === 'env-1' + ? { environmentId, fence: 'pairing-1', pairing: desktopOnHost } + : null, + call: async (_host, method, params) => { + if (method === 'status.get') { + return ok({ + capabilities: state.capable ? [DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY] : [] + }) + } + // Stand-in for the server's sync (lane S1a): mint or return ordinary phone devices on the host. + const { phones } = DelegatedMobileDeviceSyncParamsSchema.parse(params) + syncedNames.push(phones.map((phone) => phone.name)) + for (const phone of phones) { + const existing = children.get(phone.phoneKey) + if (!existing || !registry.getDevice(existing.deviceId)) { + const device = registry.addDevice(phone.name, 'mobile') + registry.updateLastSeen(device.deviceId) + children.set(phone.phoneKey, device) + } + } + return ok({ + devices: phones.map((phone) => { + const device = children.get(phone.phoneKey)! + return { phoneKey: phone.phoneKey, deviceId: device.deviceId, token: device.token } + }) + }) + }, + onEnvironmentRetired: (listener) => { + state.retire = listener + return () => {} + } + } + desktop.server.setMobileDesktopRelayHosts(hosts) + const metadata = readRuntimeMetadata(host.userDataPath)! + const created = await sendRequest(metadata.transports[0]!.endpoint, { + id: 'create', + authToken: metadata.authToken, + method: 'terminal.create', + params: { + worktree: WORKTREE, + command: 'bash', + tabId: 'tab-1', + leafId: '11111111-1111-4111-8111-111111111111', + activate: true + } + }) + const handle = z.object({ terminal: z.object({ handle: z.string() }) }).parse(created.result) + .terminal.handle + return { host, desktop, desktopSpawns, children, syncedNames, state, handle } + } + + async function connectPhone(desktop: OrcaRuntimeRpcServer, name: string) { + const pairing = pair(desktop, 'mobile', name) + const keys = generateKeyPair() + const sharedKey = deriveSharedKey(keys.secretKey, publicKeyFromBase64(pairing.publicKeyB64)) + const ws = new WebSocket(pairing.endpoint) + const frames: Frame[] = [] + const binaries: Uint8Array[] = [] + let authenticated = false + const ready = new Promise((resolve) => { + ws.on('message', (data: Buffer, isBinary) => { + if (isBinary) { + binaries.push(decryptBytes(new Uint8Array(data), sharedKey)!) + return + } + const text = data.toString() + if (text.includes('e2ee_ready')) { + ws.send( + encrypt( + JSON.stringify({ type: 'e2ee_auth', deviceToken: pairing.deviceToken }), + sharedKey + ) + ) + return + } + const frame = z.record(z.string(), z.unknown()).parse(JSON.parse(decrypt(text, sharedKey)!)) + if (!authenticated) { + authenticated = frame.type === 'e2ee_authenticated' + resolve() + return + } + frames.push(frame) + }) + }) + ws.once('open', () => + ws.send( + JSON.stringify({ type: 'e2ee_hello', publicKeyB64: publicKeyToBase64(keys.publicKey) }) + ) + ) + await ready + cleanups.push(() => ws.close()) + const token = pairing.deviceToken + return { + token, + frames, + binaries, + close: () => ws.close(), + send(id: string, method: string, params: unknown, executionHost: string | null = TARGET) { + ws.send( + encrypt( + JSON.stringify({ + id, + deviceToken: token, + method, + params, + executionHost: executionHost ?? undefined + }), + sharedKey + ) + ) + }, + next: (id: string, predicate: (frame: Frame) => boolean = () => true) => + vi.waitFor(() => { + const frame = frames.find((candidate) => candidate.id === id && predicate(candidate)) + expect(frame).toBeDefined() + return frame! + }) + } + } + type Phone = Awaited> + + async function subscribe(phone: Phone, handle: string, id = 'sub') { + phone.send(id, 'terminal.subscribe', { + terminal: handle, + client: { id: phone.token, type: 'mobile' }, + viewport: { cols: 40, rows: 20 }, + capabilities: { terminalBinaryStream: 1 } + }) + const subscribed = await phone.next( + id, + (frame) => + z.object({ result: z.object({ type: z.literal('subscribed') }) }).safeParse(frame).success + ) + return z.object({ result: z.object({ streamId: z.number() }) }).parse(subscribed).result + .streamId + } + + const isOk = (frame: Frame) => frame.ok === true + const errorCode = (frame: Frame) => + z.object({ error: z.object({ code: z.string() }) }).parse(frame).error.code + const sendAccepted = (frame: Frame) => + z.object({ result: z.object({ send: z.object({ accepted: z.boolean() }) }) }).parse(frame) + .result.send.accepted + + it('relays a terminal subscribe, its binary stream, input and a query reply as the phone delegated device', async () => { + const { host, desktop, children, syncedNames, handle } = await startTopology() + const phone = await connectPhone(desktop.server, 'iPhone') + + const streamId = await subscribe(phone, handle) + // The snapshot arrives as binary frames under the id the phone was given. + await vi.waitFor(() => + expect( + phone.binaries.some((bytes) => decodeTerminalStreamFrame(bytes)?.streamId === streamId) + ).toBe(true) + ) + expect(syncedNames).toEqual([[`iPhone via ${desktop.runtime.readMachineName()}`]]) + const delegated = [...children.values()][0]! + // The host seats the phone's delegated device, never the desktop's token for it. + expect(host.runtime.seatClientIds('pty-shared')).toEqual([delegated.token]) + + phone.send('in', 'terminal.send', { + terminal: handle, + text: 'ls', + client: { id: phone.token, type: 'mobile' } + }) + expect(isOk(await phone.next('in'))).toBe(true) + expect(host.writes).toContain('ls') + // terminal.send cross-checks client.id against the socket's device: only a swapped token passes. + phone.send('query', 'terminal.send', { + terminal: handle, + text: '\x1b[0n', + enter: false, + inputKind: 'query-reply', + client: { id: phone.token, type: 'mobile' } + }) + expect(sendAccepted(await phone.next('query'))).toBe(true) + expect(host.writes).toContain('\x1b[0n') + }) + + it('gives two phones on one server terminal their own seats, and cleans up the one that disconnects', async () => { + const { host, desktop, children, handle } = await startTopology() + const phoneA = await connectPhone(desktop.server, 'Phone A') + const phoneB = await connectPhone(desktop.server, 'Phone B') + const streamA = await subscribe(phoneA, handle) + const streamB = await subscribe(phoneB, handle) + expect(streamA).not.toBe(streamB) + const tokens = [...children.values()].map((device) => device.token) + expect(new Set(tokens).size).toBe(2) + expect(host.runtime.seatClientIds('pty-shared').sort()).toEqual([...tokens].sort()) + + phoneA.close() + await vi.waitFor(() => expect(host.runtime.seatClientIds('pty-shared')).toHaveLength(1)) + expect(host.runtime.seatClientIds('pty-shared')).not.toContain(tokens[0]) + }) + + it('applies a phone capability update on the desktop and on its open server socket, answering once', async () => { + const { desktop, handle } = await startTopology() + const phone = await connectPhone(desktop.server, 'iPhone') + await subscribe(phone, handle) + phone.send( + 'caps', + 'runtime.clientCapabilities.update', + { clientCapabilities: ['relay.test.v1'] }, + null + ) + expect(await phone.next('caps')).toMatchObject({ + ok: true, + result: { clientCapabilities: ['relay.test.v1'] } + }) + // The server applied it to the relayed phone's socket; its answer stays with the relay. + await vi.waitFor(() => + expect( + passthroughOpens.hostReplies.some((frame) => + frame.includes('"clientCapabilities":["relay.test.v1"]') + ) + ).toBe(true) + ) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect( + phone.frames.filter((frame) => frame.id === 'caps' || String(frame.id).startsWith('relay-')) + ).toHaveLength(1) + }) + + it('stops relaying a phone the server revoked until the next sync', async () => { + const { host, desktop, children, state, handle } = await startTopology() + const phone = await connectPhone(desktop.server, 'iPhone') + await subscribe(phone, handle) + const revoked = [...children.values()][0]! + + await host.server.revokeMobileDevice(revoked.deviceId) + // The open stream ends as unavailable, never as an unpaired phone. + expect(errorCode(await phone.next('sub', (frame) => frame.ok === false))).toBe( + 'remote_runtime_unavailable' + ) + const opensBefore = passthroughOpens.count + phone.send('after-1', 'terminal.list', {}) + expect(errorCode(await phone.next('after-1'))).toBe('remote_runtime_unavailable') + phone.send('after-2', 'terminal.list', {}) + expect(errorCode(await phone.next('after-2'))).toBe('remote_runtime_unavailable') + // One refused attempt, then no retries of the dead token. + expect(passthroughOpens.count - opensBefore).toBe(1) + + state.retire('env-1') + phone.send('after-sync', 'terminal.list', {}) + expect(isOk(await phone.next('after-sync'))).toBe(true) + }) + + it('refuses desktop-owned methods and unknown servers without running anything on the desktop', async () => { + const { desktop, desktopSpawns } = await startTopology() + const phone = await connectPhone(desktop.server, 'iPhone') + + phone.send('settings', 'settings.get', {}) + expect(errorCode(await phone.next('settings'))).toBe('forbidden') + phone.send('multiplex', 'terminal.multiplex', {}) + expect(errorCode(await phone.next('multiplex'))).toBe('forbidden') + const create = { + worktree: WORKTREE, + command: 'bash', + tabId: 'tab-2', + leafId: '22222222-2222-4222-8222-222222222222' + } + phone.send('unknown', 'terminal.create', create, 'runtime:env-unknown') + expect(errorCode(await phone.next('unknown'))).toBe('remote_runtime_unavailable') + phone.send('garbled', 'terminal.create', create, 'runtime:') + expect(errorCode(await phone.next('garbled'))).toBe('invalid_argument') + expect(desktopSpawns).toEqual([]) + // No target is today's local call. + phone.send('local', 'terminal.list', {}, null) + expect(isOk(await phone.next('local'))).toBe(true) + }) + + it('reports a server without delegated devices as update-needed and relays nothing to it', async () => { + const { desktop, state } = await startTopology() + state.capable = false + const phone = await connectPhone(desktop.server, 'iPhone') + const opensBefore = passthroughOpens.count + phone.send('list', 'terminal.list', {}) + expect(errorCode(await phone.next('list'))).toBe('remote_runtime_unavailable') + await expect(desktop.server.getMobileDesktopRelayHostState('env-1')).resolves.toBe( + 'update-needed' + ) + await expect(desktop.server.getMobileDesktopRelayHostState('env-unknown')).resolves.toBe( + 'unavailable' + ) + expect(passthroughOpens.count).toBe(opensBefore) + state.capable = true + state.retire('env-1') + await expect(desktop.server.getMobileDesktopRelayHostState('env-1')).resolves.toBe('ready') + }) +}) diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.test.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.test.ts new file mode 100644 index 00000000000..550cf46a77d --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from 'vitest' +import { DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY } from '../../../shared/delegated-mobile-device-contract' +import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error' +import type { RemoteRuntimePassthroughCallbacks } from '../../../shared/remote-runtime-passthrough-socket' +import { MobileDesktopRelay, type RelayedPhone } from './mobile-desktop-relay' +import type { MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts' + +type FakeSocket = { + sent: string[] + capabilities: readonly string[] + callbacks: RemoteRuntimePassthroughCallbacks + closed: boolean +} +const sockets = vi.hoisted(() => { + const opened: FakeSocket[] = [] + return opened +}) +vi.mock('../../../shared/remote-runtime-passthrough-socket', () => ({ + openRemoteRuntimePassthroughSocket: async ( + _pairing: unknown, + capabilities: readonly string[], + callbacks: RemoteRuntimePassthroughCallbacks + ) => { + const socket: FakeSocket = { sent: [], capabilities, callbacks, closed: false } + sockets.push(socket) + return { + send: (frame: string) => { + socket.sent.push(frame) + return true + }, + close: () => { + socket.closed = true + } + } + } +})) + +const ok = (result: unknown) => ({ id: 'x', ok: true as const, result, _meta: { runtimeId: 'h' } }) + +function relayWithPhone() { + const hosts: MobileDesktopRelayHosts = { + resolve: async (environmentId) => ({ + environmentId, + fence: 'f', + pairing: { v: 2, endpoint: 'ws://127.0.0.1:1', deviceToken: 'desktop', publicKeyB64: 'k' } + }), + call: async (_host, method) => + method === 'status.get' + ? ok({ capabilities: [DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY] }) + : ok({ devices: [{ phoneKey: 'phone-1', deviceId: 'child', token: 'host-token' }] }), + onEnvironmentRetired: () => () => {} + } + const relay = new MobileDesktopRelay({ + hosts, + listPhones: () => [{ phoneKey: 'phone-1', name: 'iPhone via Mac' }], + allocateStreamId: () => 1 + }) + const replies: string[] = [] + let capabilities: readonly string[] = ['cap.a'] + const phone: RelayedPhone = { + connectionId: 'conn-1', + deviceId: 'phone-1', + deviceToken: 'desktop-token', + clientCapabilities: () => capabilities, + reply: (frame) => replies.push(frame), + sendBinary: () => true + } + return { relay, phone, replies, setCapabilities: (next: string[]) => (capabilities = next) } +} + +describe('MobileDesktopRelay', () => { + it('signs in with exactly the phone capabilities and mirrors its updates without a second answer', async () => { + sockets.length = 0 + const { relay, phone, replies, setCapabilities } = relayWithPhone() + relay.forward( + phone, + 'env-1', + { id: 'r1', method: 'terminal.list' }, + '{"id":"r1","method":"terminal.list"}' + ) + relay.forward( + phone, + 'env-2', + { id: 'r2', method: 'terminal.list' }, + '{"id":"r2","method":"terminal.list"}' + ) + await vi.waitFor(() => expect(sockets.map((socket) => socket.sent.length)).toEqual([1, 1])) + expect(sockets[0]!.capabilities).toEqual(['cap.a']) + + setCapabilities(['cap.b']) + const update = JSON.stringify({ + id: 'caps', + deviceToken: 'desktop-token', + method: 'runtime.clientCapabilities.update', + params: { clientCapabilities: ['cap.b'] } + }) + relay.forwardClientCapabilities('conn-1', update) + relay.forwardClientCapabilities('other-connection', update) + await vi.waitFor(() => expect(sockets.map((socket) => socket.sent.length)).toEqual([2, 2])) + for (const socket of sockets) { + const mirrored = JSON.parse(socket.sent[1]!) + expect(mirrored).toMatchObject({ + method: 'runtime.clientCapabilities.update', + params: { clientCapabilities: ['cap.b'] } + }) + expect(mirrored.id).not.toBe('caps') + expect(mirrored.deviceToken).toBeUndefined() + // The host's answer to the mirror is the relay's, not the phone's. + socket.callbacks.onText(JSON.stringify({ id: mirrored.id, ok: true, result: {}, _meta: {} })) + } + expect(replies).toEqual([]) + }) + + it('answers every request still open when the server socket drops, then closes on phone disconnect', async () => { + sockets.length = 0 + const { relay, phone, replies } = relayWithPhone() + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + relay.forward(phone, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}') + await vi.waitFor(() => expect(sockets[0]?.sent).toHaveLength(2)) + // One socket per (phone connection, server) carries both requests. + expect(sockets).toHaveLength(1) + sockets[0]!.callbacks.onText('{"id":"a","ok":true,"result":{},"_meta":{"runtimeId":"h"}}') + sockets[0]!.callbacks.onClose( + new RemoteRuntimeClientError('remote_runtime_unavailable', 'gone') + ) + expect(replies.map((frame) => JSON.parse(frame))).toEqual([ + { id: 'a', ok: true, result: {}, _meta: { runtimeId: 'h' } }, + expect.objectContaining({ + id: 'b', + ok: false, + error: expect.objectContaining({ code: 'remote_runtime_unavailable' }) + }) + ]) + + relay.forward(phone, 'env-1', { id: 'c', method: 'terminal.list' }, '{"id":"c"}') + await vi.waitFor(() => expect(sockets).toHaveLength(2)) + relay.closePhoneConnection('conn-1') + expect(sockets[1]!.closed).toBe(true) + }) +}) diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.ts new file mode 100644 index 00000000000..4926eb52ad9 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.ts @@ -0,0 +1,287 @@ +import { randomUUID } from 'node:crypto' +import type { MobileDesktopRelayHostState } from '../../../shared/mobile-desktop-relay-contract' +import type { DelegatedMobileDeviceSyncParams } from '../../../shared/delegated-mobile-device-contract' +import type { RuntimeCapability } from '../../../shared/protocol-version' +import { RemoteRuntimeClientError } from '../../../shared/remote-runtime-client-error' +import { + openRemoteRuntimePassthroughSocket, + type RemoteRuntimePassthroughSocket +} from '../../../shared/remote-runtime-passthrough-socket' +import { MOBILE_RPC_METHOD_ROUTES } from '../runtime-rpc/runtime-rpc-mobile-method-routing' +import { syncDelegatedPhoneGrants, type DelegatedPhoneGrants } from './delegated-phone-grants' +import type { MobileDesktopRelayHost, MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts' +import { RELAYED_STREAM_CARRIERS } from './relayed-stream-carriers' +import { RelayedPhoneReplies, rewriteRelayedPhoneRequest } from './relayed-phone-frames' + +const RELAY_CONNECT_TIMEOUT_MS = 15_000 + +/** The phone connection a relayed request arrived on. */ +export type RelayedPhone = { + connectionId: string + deviceId: string + deviceToken: string + clientCapabilities: () => readonly RuntimeCapability[] + reply: (frame: string) => void + sendBinary: (bytes: Uint8Array) => boolean | void +} + +type OpenedRelayLink = { socket: RemoteRuntimePassthroughSocket; hostToken: string } + +class RelayLink { + socket?: RemoteRuntimePassthroughSocket + readonly replies: RelayedPhoneReplies + readonly opened: Promise + + constructor( + readonly environmentId: string, + readonly phone: RelayedPhone, + allocateStreamId: () => number, + open: (link: RelayLink) => Promise + ) { + this.replies = new RelayedPhoneReplies(allocateStreamId) + this.opened = open(this) + } +} + +/** + * Relays a paired phone's requests for a server's workspace to that server, over one socket per + * (phone connection, server) signed in as the phone's delegated device. It never reads a reply + * beyond its id and stream ids; see docs/reference/mobile-desktop-relay.md. + */ +export class MobileDesktopRelay { + private readonly links = new Map() + private readonly grants = new Map>() + private readonly stopRetirementWatch: () => void + + constructor( + private readonly options: { + hosts: MobileDesktopRelayHosts + listPhones: () => DelegatedMobileDeviceSyncParams['phones'] + allocateStreamId: () => number + } + ) { + this.stopRetirementWatch = options.hosts.onEnvironmentRetired((environmentId) => + this.retireEnvironment(environmentId) + ) + } + + dispose(): void { + this.stopRetirementWatch() + for (const link of this.links.values()) { + this.endLink(link, unavailable('The paired desktop stopped relaying.')) + } + this.grants.clear() + } + + forward( + phone: RelayedPhone, + environmentId: string, + request: { id: string; method: string }, + frame: string + ): void { + if ( + MOBILE_RPC_METHOD_ROUTES.get(request.method) !== 'execution-host' || + RELAYED_STREAM_CARRIERS.get(request.method) === 'phone-binary-frames' + ) { + phone.reply( + failure(request.id, 'forbidden', `Method '${request.method}' runs on the paired desktop`) + ) + return + } + const link = this.linkFor(phone, environmentId) + link.replies.noteForwarded(request.id) + link.opened.then( + ({ socket, hostToken }) => { + if (!socket.send(rewriteRelayedPhoneRequest(frame, phone.deviceToken, hostToken))) { + this.endLink(link, unavailable('The server connection could not take the request.')) + } + }, + () => { + // The open failure already answered every request queued on this link. + } + ) + } + + /** Mirrors a phone's capability update to every server socket that phone has open. */ + forwardClientCapabilities(connectionId: string, frame: string): void { + for (const link of this.links.values()) { + if (link.phone.connectionId !== connectionId) { + continue + } + // Why a fresh id: the phone already has the desktop's answer to its own id. + const id = `relay-capabilities:${randomUUID()}` + link.replies.noteForwarded(id, { swallowReplies: true }) + link.opened.then( + ({ socket, hostToken }) => + socket.send(rewriteRelayedPhoneRequest(frame, link.phone.deviceToken, hostToken, id)), + () => {} + ) + } + } + + closePhoneConnection(connectionId: string): void { + for (const link of this.links.values()) { + if (link.phone.connectionId === connectionId) { + this.endLink(link, null) + } + } + } + + /** The server was removed, re-paired or disconnected: drop its sockets and grants. */ + retireEnvironment(environmentId: string): void { + this.grants.delete(environmentId) + for (const link of this.links.values()) { + if (link.environmentId === environmentId) { + this.endLink(link, unavailable('The server connection was reset.')) + } + } + } + + /** The desktop's paired phones changed: re-sync every server already holding grants. */ + phonesChanged(): void { + for (const environmentId of this.grants.keys()) { + this.grants.delete(environmentId) + void this.hostState(environmentId) + } + } + + async hostState(environmentId: string): Promise { + try { + const host = await this.options.hosts.resolve(environmentId) + return host ? (await this.grantsFor(host, null)).kind : 'unavailable' + } catch { + return 'unavailable' + } + } + + private linkFor(phone: RelayedPhone, environmentId: string): RelayLink { + const key = linkKey(phone.connectionId, environmentId) + const existing = this.links.get(key) + if (existing) { + return existing + } + const link = new RelayLink(environmentId, phone, this.options.allocateStreamId, (opening) => + this.open(opening) + ) + link.opened.catch((error: unknown) => { + this.endLink( + link, + error instanceof RemoteRuntimeClientError ? error : unavailable(String(error)) + ) + }) + this.links.set(key, link) + return link + } + + private async open(link: RelayLink): Promise { + const { phone } = link + const host = await this.options.hosts.resolve(link.environmentId) + if (!host) { + throw unavailable('The server is not configured on this computer.') + } + const grants = await this.grantsFor(host, phone.deviceId) + if (grants.kind === 'update-needed') { + throw unavailable('The server needs an Orca update to open its workspaces from a phone.') + } + const grant = grants.grants.get(phone.deviceId) + if (!grant || grants.refused.has(phone.deviceId)) { + throw unavailable('The server does not accept this phone.') + } + try { + const socket = await openRemoteRuntimePassthroughSocket( + { + ...host.pairing, + deviceToken: grant.token, + pairedDeviceId: grant.deviceId, + scope: 'mobile' + }, + phone.clientCapabilities(), + { + onText: (plaintext) => { + const frame = link.replies.text(plaintext) + if (frame !== null) { + phone.reply(frame) + } + }, + onBinary: (bytes) => { + const frame = link.replies.binary(bytes) + if (frame) { + phone.sendBinary(frame) + } + }, + onClose: (error) => this.endLink(link, error) + }, + { timeoutMs: RELAY_CONNECT_TIMEOUT_MS } + ) + if (this.links.get(linkKey(phone.connectionId, link.environmentId)) !== link) { + socket.close() + throw unavailable('The relayed request was cancelled.') + } + link.socket = socket + return { socket, hostToken: grant.token } + } catch (error) { + if (error instanceof RemoteRuntimeClientError && error.code === 'unauthorized') { + // Why: retrying a revoked token would only raise the host's "unpaired device" prompt. + grants.refused.add(phone.deviceId) + throw unavailable('The server revoked this phone.') + } + throw error + } + } + + private grantsFor( + host: MobileDesktopRelayHost, + phoneKey: string | null + ): Promise { + const cached = this.grants.get(host.environmentId) + const sync = (): Promise => { + const pending = syncDelegatedPhoneGrants(this.options.hosts, host, this.options.listPhones()) + this.grants.set(host.environmentId, pending) + pending.catch(() => { + // Why: an unreachable server is not an answer; the next request syncs again. + if (this.grants.get(host.environmentId) === pending) { + this.grants.delete(host.environmentId) + } + }) + return pending + } + if (!cached) { + return sync() + } + return cached.then((grants) => + grants.fence !== host.fence || + (phoneKey !== null && grants.kind === 'ready' && !grants.grants.has(phoneKey)) + ? this.grants.get(host.environmentId) === cached + ? sync() + : this.grantsFor(host, phoneKey) + : grants + ) + } + + private endLink(link: RelayLink, error: RemoteRuntimeClientError | null): void { + const key = linkKey(link.phone.connectionId, link.environmentId) + if (this.links.get(key) !== link) { + return + } + this.links.delete(key) + link.socket?.close() + const openRequestIds = link.replies.takeOpenRequestIds() + if (error) { + for (const id of openRequestIds) { + link.phone.reply(failure(id, 'remote_runtime_unavailable', error.message)) + } + } + } +} + +function linkKey(connectionId: string, environmentId: string): string { + return `${connectionId}\0${environmentId}` +} + +function unavailable(message: string): RemoteRuntimeClientError { + return new RemoteRuntimeClientError('remote_runtime_unavailable', message) +} + +function failure(id: string, code: string, message: string): string { + return JSON.stringify({ id, ok: false, error: { code, message }, _meta: { runtimeId: null } }) +} diff --git a/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.test.ts b/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.test.ts new file mode 100644 index 00000000000..d05b44a2439 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, it } from 'vitest' +import { + encodeTerminalStreamFrame, + decodeTerminalStreamFrame, + TerminalStreamOpcode +} from '../../../shared/terminal-stream-protocol' +import { + BrowserScreencastOpcode, + encodeBrowserScreencastFrame +} from '../../../shared/browser-screencast-protocol' +import { RelayedPhoneReplies, rewriteRelayedPhoneRequest } from './relayed-phone-frames' + +const DESKTOP_TOKEN = 'a'.repeat(48) +const HOST_TOKEN = 'b'.repeat(48) + +describe('rewriteRelayedPhoneRequest', () => { + it('swaps the desktop token everywhere and strips the envelope routing fields', () => { + const frame = JSON.stringify({ + id: 'r1', + deviceToken: DESKTOP_TOKEN, + executionHost: 'runtime:env-1', + method: 'terminal.unsubscribe', + params: { + subscriptionId: `terminal-1:${DESKTOP_TOKEN}`, + client: { id: DESKTOP_TOKEN, type: 'mobile' }, + nested: [{ mobileClient: { id: DESKTOP_TOKEN } }] + } + }) + const rewritten = rewriteRelayedPhoneRequest(frame, DESKTOP_TOKEN, HOST_TOKEN) + expect(frame).toContain(DESKTOP_TOKEN) + expect(rewritten).not.toContain(DESKTOP_TOKEN) + expect(JSON.parse(rewritten)).toEqual({ + id: 'r1', + method: 'terminal.unsubscribe', + params: { + subscriptionId: `terminal-1:${HOST_TOKEN}`, + client: { id: HOST_TOKEN, type: 'mobile' }, + nested: [{ mobileClient: { id: HOST_TOKEN } }] + } + }) + }) + + it('can re-id a frame the relay sends on its own behalf', () => { + const frame = JSON.stringify({ id: 'phone-id', method: 'm', params: {} }) + expect( + JSON.parse(rewriteRelayedPhoneRequest(frame, DESKTOP_TOKEN, HOST_TOKEN, 'relay-id')) + ).toEqual({ id: 'relay-id', method: 'm', params: {} }) + }) +}) + +describe('RelayedPhoneReplies', () => { + function replies(firstDesktopId = 100) { + let next = firstDesktopId + return new RelayedPhoneReplies(() => next++) + } + const subscribed = (id: string, streamId: number) => + JSON.stringify({ + id, + ok: true, + streaming: true, + result: { type: 'subscribed', streamId }, + _meta: { runtimeId: 'host' } + }) + const output = (streamId: number) => + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId, + seq: 7, + payload: new TextEncoder().encode('hi') + }) + + it('renumbers host stream ids in results and terminal frame headers into the desktop space', () => { + const relay = replies() + relay.noteForwarded('s1') + relay.noteForwarded('s2') + // Both hosts allocate from 1; the phone must see distinct ids. + expect(JSON.parse(relay.text(subscribed('s1', 1))!).result.streamId).toBe(100) + expect(JSON.parse(relay.text(subscribed('s2', 2))!).result.streamId).toBe(101) + const frame = decodeTerminalStreamFrame(relay.binary(output(2))!) + expect(frame).toMatchObject({ streamId: 101, seq: 7, opcode: TerminalStreamOpcode.Output }) + expect(new TextDecoder().decode(frame!.payload)).toBe('hi') + }) + + it('drops terminal frames for streams the phone was never told of and passes other binary untouched', () => { + const relay = replies() + expect(relay.binary(output(9))).toBeNull() + const screencast = encodeBrowserScreencastFrame({ + opcode: BrowserScreencastOpcode.Frame, + seq: 1, + format: 'jpeg', + metadata: { deviceWidth: 1 }, + image: new Uint8Array([1, 2, 3]) + }) + expect(relay.binary(screencast)).toBe(screencast) + }) + + it('passes non-stream replies through byte for byte and forgets a stream once it ends', () => { + const relay = replies() + relay.noteForwarded('s1') + relay.noteForwarded('q') + const plain = '{"id":"q","ok":true,"result":{"x":1},"_meta":{"runtimeId":"h"}}' + expect(relay.text(plain)).toBe(plain) + expect(relay.text('{"_keepalive":true}')).toBe('{"_keepalive":true}') + relay.text(subscribed('s1', 1)) + relay.text( + JSON.stringify({ id: 's1', ok: true, streaming: true, result: { type: 'end' }, _meta: {} }) + ) + expect(relay.binary(output(1))).toBeNull() + expect(relay.takeOpenRequestIds()).toEqual([]) + }) + + it('reports unfinished requests, and swallows replies to requests the relay sent itself', () => { + const relay = replies() + relay.noteForwarded('open') + relay.noteForwarded('mine', { swallowReplies: true }) + expect(relay.text('{"id":"mine","ok":true,"result":{},"_meta":{"runtimeId":"h"}}')).toBeNull() + relay.noteForwarded('mine-2', { swallowReplies: true }) + expect(relay.takeOpenRequestIds()).toEqual(['open']) + }) +}) diff --git a/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.ts b/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.ts new file mode 100644 index 00000000000..8da333afdfe --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.ts @@ -0,0 +1,119 @@ +import { + readTerminalStreamFrameId, + writeTerminalStreamFrameId +} from '../../../shared/terminal-stream-protocol' + +const STRIPPED_ENVELOPE_FIELDS = new Set(['deviceToken', 'executionHost']) + +// Why: the phone stamps its desktop token into params (client.id, mobileClient.id, `${terminal}:${token}` +// subscription ids). A field list would rot silently, so every occurrence in the text is swapped. +export function rewriteRelayedPhoneRequest( + frame: string, + phoneDesktopToken: string, + phoneHostToken: string, + requestId?: string +): string { + const parsed: unknown = JSON.parse(frame.split(phoneDesktopToken).join(phoneHostToken)) + if (!isRecord(parsed)) { + throw new Error('relayed_phone_frame_not_an_object') + } + // Why: the host binds identity to the authenticated socket; the routing target is the desktop's only. + const forwarded = Object.fromEntries( + Object.entries(parsed).filter(([field]) => !STRIPPED_ENVELOPE_FIELDS.has(field)) + ) + return JSON.stringify(requestId === undefined ? forwarded : { ...forwarded, id: requestId }) +} + +/** + * The host's replies on one relayed socket, passed through except for stream ids. Why renumber: + * every host allocates terminal stream ids from 1, and the phone routes binary frames by id + * across everything its desktop socket carries. + */ +export class RelayedPhoneReplies { + private readonly openRequests = new Map() + private readonly swallowedRequests = new Set() + private readonly desktopStreamIds = new Map() + + constructor(private readonly allocateStreamId: () => number) {} + + noteForwarded(requestId: string, options: { swallowReplies?: boolean } = {}): void { + this.openRequests.set(requestId, []) + if (options.swallowReplies) { + this.swallowedRequests.add(requestId) + } + } + + /** Request ids still waiting on the host; the caller fails them when the socket ends. */ + takeOpenRequestIds(): string[] { + const ids = [...this.openRequests.keys()].filter((id) => !this.swallowedRequests.has(id)) + this.openRequests.clear() + this.swallowedRequests.clear() + this.desktopStreamIds.clear() + return ids + } + + /** The frame to hand the phone, or null to drop it. */ + text(plaintext: string): string | null { + let parsed: unknown + try { + parsed = JSON.parse(plaintext) + } catch { + return plaintext + } + if (!isRecord(parsed) || typeof parsed.id !== 'string') { + return plaintext + } + const streamIds = this.openRequests.get(parsed.id) + const swallowed = this.swallowedRequests.has(parsed.id) + const result = parsed.result + let renumbered = false + if (streamIds && isRecord(result) && typeof result.streamId === 'number') { + result.streamId = this.desktopStreamId(result.streamId, streamIds) + renumbered = true + } + if (streamIds && (parsed.streaming !== true || (isRecord(result) && result.type === 'end'))) { + this.finish(parsed.id, streamIds) + } + if (swallowed) { + return null + } + return renumbered ? JSON.stringify(parsed) : plaintext + } + + /** The frame to hand the phone, or null to drop a terminal frame for a stream it was never told of. */ + binary(bytes: Uint8Array): Uint8Array | null { + const hostStreamId = readTerminalStreamFrameId(bytes) + if (hostStreamId === null) { + return bytes + } + const desktopStreamId = this.desktopStreamIds.get(hostStreamId) + if (desktopStreamId === undefined) { + return null + } + writeTerminalStreamFrameId(bytes, desktopStreamId) + return bytes + } + + private desktopStreamId(hostStreamId: number, requestStreamIds: number[]): number { + const known = this.desktopStreamIds.get(hostStreamId) + if (known !== undefined) { + return known + } + const allocated = this.allocateStreamId() + this.desktopStreamIds.set(hostStreamId, allocated) + requestStreamIds.push(hostStreamId) + return allocated + } + + private finish(requestId: string, hostStreamIds: number[]): void { + this.openRequests.delete(requestId) + this.swallowedRequests.delete(requestId) + for (const hostStreamId of hostStreamIds) { + this.desktopStreamIds.delete(hostStreamId) + } + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} diff --git a/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.test.ts b/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.test.ts new file mode 100644 index 00000000000..fffda55e8ab --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from 'vitest' +import { ALL_RPC_METHODS } from '../rpc/methods' +import { MOBILE_RPC_METHOD_ALLOWLIST } from '../runtime-rpc/runtime-rpc-mobile-method-allowlist' +import { MOBILE_RPC_METHOD_ROUTES } from '../runtime-rpc/runtime-rpc-mobile-method-routing' +import { RELAYED_STREAM_CARRIERS } from './relayed-stream-carriers' + +describe('relayed stream carrier census', () => { + const relayableStreams: string[] = ALL_RPC_METHODS.filter( + (method) => + 'stream' in method && + MOBILE_RPC_METHOD_ALLOWLIST.has(method.name) && + MOBILE_RPC_METHOD_ROUTES.get(method.name) === 'execution-host' + ).map((method) => method.name) + + it('classifies how every relayable stream names itself', () => { + expect(relayableStreams.filter((name) => !RELAYED_STREAM_CARRIERS.has(name))).toEqual([]) + }) + + it('classifies nothing that is not a relayable stream', () => { + expect( + [...RELAYED_STREAM_CARRIERS.keys()].filter((name) => !relayableStreams.includes(name)) + ).toEqual([]) + }) +}) diff --git a/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.ts b/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.ts new file mode 100644 index 00000000000..da3ef79f5ee --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/relayed-stream-carriers.ts @@ -0,0 +1,19 @@ +/** + * How each streaming method a phone can relay names its stream. `terminal-stream-id`: the host + * allocates ids in `result.streamId` and terminal binary headers, renumbered by RelayedPhoneReplies. + * `none`: replies carry no host-allocated id (the phone holds one screencast at a time). + * `phone-binary-frames`: the client drives the stream with binary frames, which the relay never + * forwards, so the call is refused rather than opened dead. The census test fails on a new stream. + */ +export type RelayedStreamCarrier = 'terminal-stream-id' | 'none' | 'phone-binary-frames' + +export const RELAYED_STREAM_CARRIERS: ReadonlyMap = new Map([ + ['terminal.subscribe', 'terminal-stream-id'], + ['terminal.multiplex', 'phone-binary-frames'], + ['browser.screencast', 'none'], + ['agentSession.subscribe', 'none'], + ['agentSession.subscribeStatus', 'none'], + ['nativeChat.subscribe', 'none'], + ['session.tabs.subscribe', 'none'], + ['session.tabs.subscribeAll', 'none'] +]) diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts index efa05957f40..236eaf75fc3 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts @@ -249,6 +249,7 @@ export class RuntimeRpcLifecycle extends RuntimeRpcWebSocketDispatch { return } this.abortWebSocketDispatches(socket.ws) + this.mobileDesktopRelay?.closePhoneConnection(socket.connectionId) // Why: subscriptions and binary streams are socket-scoped, but disconnect state is device-scoped across transports. this.runtime.cleanupSubscriptionsForConnection(socket.connectionId) this.runtime.cancelMobileDictationForConnection(socket.connectionId) diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.test.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.test.ts new file mode 100644 index 00000000000..7f36329b9b0 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from 'vitest' +import { MOBILE_RPC_METHOD_ALLOWLIST } from './runtime-rpc-mobile-method-allowlist' +import { + MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT, + MOBILE_RPC_METHOD_ROUTES +} from './runtime-rpc-mobile-method-routing' + +describe('mobile RPC method routing census', () => { + it('tags every allowlisted method exactly once', () => { + const untagged = [...MOBILE_RPC_METHOD_ALLOWLIST].filter( + (method) => !MOBILE_RPC_METHOD_ROUTES.has(method) + ) + expect(untagged).toEqual([]) + expect(MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT).toBe(MOBILE_RPC_METHOD_ROUTES.size) + }) + + it('tags nothing outside the allowlist', () => { + const stale = [...MOBILE_RPC_METHOD_ROUTES.keys()].filter( + (method) => !MOBILE_RPC_METHOD_ALLOWLIST.has(method) + ) + expect(stale).toEqual([]) + }) +}) diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.ts new file mode 100644 index 00000000000..2064193e087 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.ts @@ -0,0 +1,227 @@ +import { PAIRED_DESKTOP_METHODS } from './runtime-rpc-mobile-paired-desktop-methods' + +// Why: a relaying Mac sends `execution-host` calls to the host named by the call's target (itself when +// local) and keeps `paired-desktop` calls; every allowlisted method must pick one (census test). +export type MobileRpcMethodRoute = 'execution-host' | 'paired-desktop' + +const EXECUTION_HOST_METHODS = [ + 'agent.launch', + 'agent.launchReplay', + 'agentTeams.prepareLaunch', + 'agentTeams.tmuxCompat', + 'browser.back', + 'browser.dialogAccept', + 'browser.dialogDismiss', + 'browser.forward', + 'browser.goto', + 'browser.keyboardInsertText', + 'browser.keypress', + 'browser.mouseClick', + 'browser.mouseDown', + 'browser.mouseMove', + 'browser.mouseUp', + 'browser.mouseWheel', + 'browser.reload', + 'browser.screencast', + 'browser.screencast.unsubscribe', + 'browser.tabCreate', + 'browser.viewport', + // Pasted images must land on the host whose terminal reads the path. + 'clipboard.abortImageUpload', + 'clipboard.appendImageUploadChunk', + 'clipboard.commitImageUpload', + 'clipboard.saveImageAsTempFile', + 'clipboard.startImageUpload', + 'files.browseServerDir', + 'files.createFile', + 'files.list', + 'files.open', + 'files.openDiff', + 'files.read', + 'files.readChunk', + 'files.readDir', + 'files.readDocPreview', + 'files.readPreview', + 'files.readTerminalArtifact', + 'files.readTerminalArtifactPreview', + 'files.resolveTerminalPath', + 'files.searchPaths', + 'files.stat', + 'files.writeTerminalArtifact', + 'folderWorkspace.list', + 'git.abortMerge', + 'git.abortRebase', + 'git.branchCompare', + 'git.branchDiff', + 'git.bulkStage', + 'git.bulkUnstage', + 'git.cancelGenerateCommitMessage', + 'git.cancelGeneratePullRequestFields', + 'git.checkout', + 'git.commit', + 'git.commitCompare', + 'git.commitDiff', + 'git.diff', + 'git.discard', + 'git.discoverCommitMessageModels', + 'git.fastForward', + 'git.fetch', + 'git.forkSync', + 'git.generateCommitMessage', + 'git.generatePullRequestFields', + 'git.history', + 'git.localBranches', + 'git.pull', + 'git.push', + 'git.rebaseFromBase', + 'git.stage', + 'git.status', + 'git.unstage', + 'git.upstreamStatus', + // Repo-scoped provider calls run where the desktop runs them: on the repo's host, with its login. + 'github.addIssueComment', + 'github.addPRReviewComment', + 'github.addPRReviewCommentReply', + 'github.countWorkItems', + 'github.createIssue', + 'github.listAssignableUsers', + 'github.listLabels', + 'github.listWorkItems', + 'github.mergePR', + 'github.prCheckDetails', + 'github.prChecks', + 'github.prFileContents', + 'github.prForBranch', + 'github.removePRReviewers', + 'github.repoSlug', + 'github.requestPRReviewers', + 'github.rerunPRChecks', + 'github.resolveReviewThread', + 'github.setPRAutoMerge', + 'github.setPRFileViewed', + 'github.updateIssue', + 'github.updatePR', + 'github.updatePRState', + 'github.updatePRTitle', + 'github.workItem', + 'github.workItemByOwnerRepo', + 'github.workItemDetails', + 'gitlab.addIssueComment', + 'gitlab.addMRComment', + 'gitlab.createIssue', + 'gitlab.listWorkItems', + 'gitlab.mergeMR', + 'gitlab.resolveMRDiscussion', + 'gitlab.todos', + 'gitlab.updateIssue', + 'gitlab.updateMR', + 'gitlab.updateMRState', + 'gitlab.workItemByPath', + 'gitlab.workItemDetails', + // Shell and agent availability are facts about the machine that spawns the process. + 'host.gitBash.isAvailable', + 'host.platform', + 'host.pwsh.isAvailable', + 'host.wsl.isAvailable', + 'host.wsl.listDistros', + 'preflight.check', + 'preflight.detectAgents', + 'preflight.detectRemoteAgents', + 'hostedReview.create', + 'hostedReview.createStacked', + 'hostedReview.forBranch', + 'hostedReview.getCreationEligibility', + 'linear.resolveCurrentIssue', + 'markdown.readTab', + 'markdown.saveTab', + 'nativeChat.readSession', + 'nativeChat.subscribe', + 'nativeChat.unsubscribe', + 'orchestration.workerTerminalUserInput', + 'projectGroup.list', + 'repo.baseRefDefault', + 'repo.gitAvailable', + 'repo.hooks', + 'repo.list', + 'repo.saveSparsePreset', + 'repo.searchRefs', + 'repo.sparsePresets', + 'repo.update', + 'session.tabs.activate', + 'session.tabs.close', + 'session.tabs.closeLifecycle', + 'session.tabs.createTerminal', + 'session.tabs.list', + 'session.tabs.listAll', + 'session.tabs.move', + 'session.tabs.subscribe', + 'session.tabs.subscribeAll', + 'session.tabs.unsubscribe', + 'session.tabs.unsubscribeAll', + 'agentSession.cancel', + 'agentSession.close', + 'agentSession.commands', + 'agentSession.conversationCommand', + 'agentSession.create', + 'agentSession.createSupport', + 'agentSession.ensure', + 'agentSession.handoffStatus', + 'agentSession.history', + 'agentSession.hold', + 'agentSession.modelCatalog', + 'agentSession.options', + 'agentSession.queuedMessageDelete', + 'agentSession.queuedMessageSend', + 'agentSession.queuedMessagesResume', + 'agentSession.readVisual', + 'agentSession.release', + 'agentSession.respondToApproval', + 'agentSession.respondToQuestion', + 'agentSession.reveal', + 'agentSession.send', + 'agentSession.setOption', + 'agentSession.subscribe', + 'agentSession.subscribeStatus', + 'agentSession.unsubscribe', + 'terminal.adoptOrphans', + 'terminal.agentStatus', + 'terminal.clearBuffer', + 'terminal.close', + 'terminal.closeAll', + 'terminal.closeTab', + 'terminal.create', + 'terminal.createAgentSession', + 'terminal.ensureAgentSession', + 'terminal.focus', + 'terminal.isRunningAgent', + 'terminal.list', + 'terminal.multiplex', + 'terminal.read', + 'terminal.rename', + 'terminal.send', + 'terminal.setDisplayMode', + 'terminal.subscribe', + 'terminal.unsubscribe', + 'terminal.updateViewport', + 'terminal.wait', + 'worktree.activate', + 'worktree.create', + 'worktree.forceDeleteBranch', + 'worktree.listRetiredNames', + 'worktree.prefetchCreateBase', + 'worktree.ps', + 'worktree.resolveMrBase', + 'worktree.resolvePrBase', + 'worktree.rm', + 'worktree.set', + 'worktree.show', + 'worktree.sleep' +] as const + +export const MOBILE_RPC_METHOD_ROUTES: ReadonlyMap = new Map([ + ...PAIRED_DESKTOP_METHODS.map((method) => [method, 'paired-desktop'] as const), + ...EXECUTION_HOST_METHODS.map((method) => [method, 'execution-host'] as const) +]) + +export const MOBILE_RPC_METHOD_ROUTE_ENTRY_COUNT = + PAIRED_DESKTOP_METHODS.length + EXECUTION_HOST_METHODS.length diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts new file mode 100644 index 00000000000..c736f0e9577 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts @@ -0,0 +1,103 @@ +/** Allowlisted phone methods the relaying Mac always answers itself; never sent with a target. */ +export const PAIRED_DESKTOP_METHODS = [ + // Device and pairing concerns of the phone's own pairing. + 'mobileWeb.bundle.chunk', + 'mobileWeb.bundle.manifest', + 'mobileWeb.bundle.range', + 'notifications.getMissedSince', + 'notifications.registerPush', + 'notifications.subscribe', + 'notifications.testPush', + 'notifications.unregisterPush', + 'notifications.unsubscribe', + 'pairing.getEndpoints', + 'pairing.provisionRelay', + 'speech.dictation.cancel', + 'speech.dictation.chunk', + 'speech.dictation.finish', + 'speech.dictation.setup', + 'speech.dictation.start', + 'speech.models.delete', + 'speech.models.download', + 'speech.models.list', + // The Mac also forwards it to every open server connection of that phone. + 'runtime.clientCapabilities.update', + // The Mac republishes host changes it already observes. + 'runtime.clientEvents.subscribe', + 'runtime.clientEvents.unsubscribe', + 'status.get', + 'diagnostics.memory', + 'stats.summary', + 'ui.get', + 'ui.recordFeatureInteraction', + 'ui.set', + // The desktop's accounts pane, Linear and project pickers target its active runtime, not a workspace's host. + 'accounts.consumeCodexResetCredit', + 'accounts.list', + 'accounts.selectClaude', + 'accounts.selectCodex', + 'accounts.selectCodexForTarget', + 'accounts.subscribe', + 'accounts.unsubscribe', + 'linear.addIssueComment', + 'linear.agentSearchIssues', + 'linear.connect', + 'linear.createIssue', + 'linear.createProject', + 'linear.getCustomView', + 'linear.getIssue', + 'linear.getProject', + 'linear.issueComments', + 'linear.issueContext', + 'linear.listCustomViewIssues', + 'linear.listCustomViewProjects', + 'linear.listCustomViews', + 'linear.listIssues', + 'linear.listProjectIssues', + 'linear.listProjects', + 'linear.listTeams', + 'linear.mcpListIssues', + 'linear.searchIssues', + 'linear.selectWorkspace', + 'linear.status', + 'linear.teamLabels', + 'linear.teamMembers', + 'linear.teamStates', + 'linear.updateIssue', + 'github.project.addIssueCommentBySlug', + 'github.project.clearItemField', + 'github.project.deleteIssueCommentBySlug', + 'github.project.listAccessible', + 'github.project.listAssignableUsersBySlug', + 'github.project.listIssueTypesBySlug', + 'github.project.listLabelsBySlug', + 'github.project.listViews', + 'github.project.resolveRef', + 'github.project.updateIssueBySlug', + 'github.project.updateIssueCommentBySlug', + 'github.project.updateIssueTypeBySlug', + 'github.project.updateItemField', + 'github.project.updatePullRequestBySlug', + 'github.project.viewTable', + 'github.project.workItemDetailsBySlug', + // Desktop main federates session history across hosts itself (executionHostId / scope params). + 'aiVault.listSessions', + 'aiVault.prepareSessionResume', + 'aiVault.resolveSessionTitles', + 'aiVault.searchSessions', + 'aiVault.searchStatus', + // User preferences live on the Mac; the phone's settings screens mirror the desktop's. + 'settings.get', + 'settings.getTerminalQuickCommands', + 'settings.mutateNativeChatSessionOptions', + 'settings.update', + 'settings.updateTerminalQuickCommands', + 'terminal.getAutoRestoreFit', + 'terminal.setAutoRestoreFit', + // SSH targets are configured on the Mac. + 'ssh.connect', + 'ssh.getState', + 'ssh.listRemovedTargetLabels', + 'ssh.listTargetSummaries', + 'ssh.listTargets' +] as const diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts b/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts index 7d8bba9f958..cf2a05d0684 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts @@ -11,6 +11,10 @@ import { encodePairingOffer, PAIRING_OFFER_VERSION } from '../../../shared/pairi import type { RuntimePairingReach } from '../../../shared/runtime-pairing-reach' import { resolveAdvertisedPairingEndpoint } from '../pairing-endpoint' import { RuntimeRpcNetworkExposure } from './runtime-rpc-network-exposure' +import { MobileDesktopRelay } from '../mobile-desktop-relay/mobile-desktop-relay' +import type { MobileDesktopRelayHosts } from '../mobile-desktop-relay/mobile-desktop-relay-hosts' +import type { MobileDesktopRelayHostState } from '../../../shared/mobile-desktop-relay-contract' +import { allocateTerminalSubscriptionStreamId } from '../rpc/methods/terminal/terminal-subscription-stream-id' import { createWebClientUrl, DEVICE_REGISTRY_UNAVAILABLE_GUIDANCE, @@ -85,6 +89,33 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure { this.mobileRelayPairingProvider = provider } + setMobileDesktopRelayHosts(hosts: MobileDesktopRelayHosts | null): void { + this.mobileDesktopRelay?.dispose() + this.mobileDesktopRelay = hosts + ? new MobileDesktopRelay({ + hosts, + listPhones: () => this.listRelayedPhones(), + // Why: one allocator with local streams, so relayed ids never collide on the phone's socket. + allocateStreamId: allocateTerminalSubscriptionStreamId + }) + : null + } + + /** Whether a paired phone can open this server's workspaces through this desktop. */ + getMobileDesktopRelayHostState(environmentId: string): Promise { + return this.mobileDesktopRelay?.hostState(environmentId) ?? Promise.resolve('unavailable') + } + + private listRelayedPhones(): { phoneKey: string; name: string }[] { + const computerName = this.runtime.readMachineName() + return (this.deviceRegistry?.listDevices() ?? []) + .filter((device) => device.scope === 'mobile' && device.lastSeenAt > 0) + .map((device) => ({ + phoneKey: device.deviceId, + name: `${device.name} via ${computerName}`.slice(0, 128) + })) + } + async revokeMobileDevice(deviceId: string): Promise { const device = this.deviceRegistry?.getDevice(deviceId) if (device?.scope !== 'mobile') { @@ -104,6 +135,7 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure { this.mobileRelayPairingProvider?.onDemandStateChanged?.() this.runtime.forgetClientNavigationState(deviceId) this.mobileSocketWiring?.terminateDeviceConnections(device.token) + this.mobileDesktopRelay?.phonesChanged() return true } diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-state.ts b/src/main/runtime/runtime-rpc/runtime-rpc-state.ts index b6492a6683b..08af258f624 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-state.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-state.ts @@ -9,6 +9,7 @@ import type { DeviceRegistry } from '../device-registry' import type { E2EEKeypair } from '../e2ee-keypair' import type { UnpairedDeviceAuthThrottle } from '../rpc/unpaired-device-auth-throttle' import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring' +import type { MobileDesktopRelay } from '../mobile-desktop-relay/mobile-desktop-relay' import { RelayRevokeOutbox } from '../relay/relay-revoke-outbox' import { PushUnregisterOutbox } from '../push/push-unregister-outbox' import { RuntimeBinaryMessageRouter } from '../runtime-binary-message-router' @@ -70,6 +71,8 @@ export class RuntimeRpcState { // transports under the SAME wiring (see ensureMobileSocketWiring) instead of orphaning relay sockets. protected detachWebSocketWiring: (() => void) | null = null protected mobileRelayPairingProvider: MobileRelayPairingProvider | null = null + // Why: only a desktop with configured servers relays; headless serve leaves this unset. + protected mobileDesktopRelay: MobileDesktopRelay | null = null protected mobileRelayPairingOfferQueue: Promise = Promise.resolve() protected mobileRelayPairingOfferInFlight: { generation: number diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts b/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts index dd714b77528..46311b58a26 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts @@ -11,6 +11,7 @@ import type { DeviceScope } from '../device-registry' import { RuntimeRpcRequestAdmission } from './runtime-rpc-request-admission' import { classifyRuntimeLongPoll } from './runtime-rpc-long-poll' import { MOBILE_RPC_METHOD_ALLOWLIST } from './runtime-rpc-mobile-method-allowlist' +import { parseExecutionHostId } from '../../../shared/execution-host' // Why: status.get has no per-connection context in the dispatcher, so stamp the scope here at the transport boundary. function injectDeviceScope(response: string, scope: DeviceScope): string { @@ -27,6 +28,58 @@ function injectDeviceScope(response: string, scope: DeviceScope): string { } export class RuntimeRpcWebSocketDispatch extends RuntimeRpcRequestAdmission { + // Why: a phone names a server's workspace with `executionHost`; only configured servers are reached, + // and a target the desktop cannot reach is an error, never a silent local run. + private relayMobileRequest( + request: RpcRequest, + rawMessage: string, + reply: (response: string) => void, + sendBinary: (response: Uint8Array) => boolean | void, + socket: AuthenticatedMobileSocket + ): boolean { + if (request.method === 'runtime.clientCapabilities.update') { + this.mobileDesktopRelay?.forwardClientCapabilities(socket.connectionId, rawMessage) + } + const executionHost: unknown = Reflect.get(request, 'executionHost') + if (executionHost === undefined) { + return false + } + const target = parseExecutionHostId(typeof executionHost === 'string' ? executionHost : null) + if (target?.kind === 'local' || target?.kind === 'ssh') { + return false + } + if (!target) { + reply( + JSON.stringify(this.buildError(request.id, 'invalid_argument', 'Invalid executionHost')) + ) + } else if (!this.mobileDesktopRelay) { + reply( + JSON.stringify( + this.buildError( + request.id, + 'remote_runtime_unavailable', + 'This computer does not relay to servers' + ) + ) + ) + } else { + this.mobileDesktopRelay.forward( + { + connectionId: socket.connectionId, + deviceId: socket.device.deviceId, + deviceToken: socket.device.deviceToken, + clientCapabilities: () => socket.clientCapabilities, + reply, + sendBinary + }, + target.environmentId, + request, + rawMessage + ) + } + return true + } + // Why: WebSocket dispatch is streaming (multiple responses) and auths via per-device tokens, not the shared token. protected async handleWebSocketMessage( rawMessage: string, @@ -86,6 +139,14 @@ export class RuntimeRpcWebSocketDispatch extends RuntimeRpcRequestAdmission { return } + if ( + device.scope === 'mobile' && + authenticatedSocket && + this.relayMobileRequest(request, rawMessage, reply, sendBinary, authenticatedSocket) + ) { + return + } + // Why: bind deviceToken to this socket so ws.on('close') knows which mobile client disconnected. if (wsTransport && ws) { wsTransport.setClientId(ws, token) diff --git a/src/main/startup/main-process-runtime-launch-activation.test.ts b/src/main/startup/main-process-runtime-launch-activation.test.ts index 6fd2ec6ec57..d87cc4a806d 100644 --- a/src/main/startup/main-process-runtime-launch-activation.test.ts +++ b/src/main/startup/main-process-runtime-launch-activation.test.ts @@ -27,9 +27,13 @@ vi.mock('../runtime/runtime-rpc', () => ({ OrcaRuntimeRpcServer: class { start = vi.fn(async () => {}) setOnUnpairedDeviceAuthFailure = vi.fn() + setMobileDesktopRelayHosts = vi.fn() } })) vi.mock('../ipc/mobile', () => ({ registerMobileHandlers: vi.fn() })) +vi.mock('../ipc/runtime-environment-mobile-relay-hosts', () => ({ + createRuntimeEnvironmentMobileRelayHosts: vi.fn() +})) vi.mock('../ipc/pty', () => ({ getLocalPtyProvider: vi.fn(), registerHeadlessPtyRuntime: vi.fn() diff --git a/src/main/startup/main-process-runtime-launch.ts b/src/main/startup/main-process-runtime-launch.ts index f78a86f297a..9c439823041 100644 --- a/src/main/startup/main-process-runtime-launch.ts +++ b/src/main/startup/main-process-runtime-launch.ts @@ -8,6 +8,7 @@ import { } from '../persistence' import { OrcaRuntimeRpcServer } from '../runtime/runtime-rpc' import { registerMobileHandlers } from '../ipc/mobile' +import { createRuntimeEnvironmentMobileRelayHosts } from '../ipc/runtime-environment-mobile-relay-hosts' import { getLocalPtyProvider, registerHeadlessPtyRuntime } from '../ipc/pty' import { LocalPtyProvider } from '../providers/local-pty-provider' import { publishHeadlessRuntimeGraph } from '../runtime/headless-runtime-graph' @@ -95,6 +96,7 @@ function installRuntimeRpc( webClientRoot: getBundledWebClientRoot() }) state.runtimeRpc = runtimeRpc + runtimeRpc.setMobileDesktopRelayHosts(createRuntimeEnvironmentMobileRelayHosts()) registerMobileHandlers(runtimeRpc, { getRelayStatus: getDesktopRelayStatus, consumePendingUnpairedDeviceAuthFailure: (webContentsId) => { diff --git a/src/shared/delegated-mobile-device-contract.ts b/src/shared/delegated-mobile-device-contract.ts new file mode 100644 index 00000000000..55bf990b1a7 --- /dev/null +++ b/src/shared/delegated-mobile-device-contract.ts @@ -0,0 +1,33 @@ +import { z } from 'zod' + +// Why: a paired desktop relays each of its phones to this host as an ordinary phone device it owns. +export const DELEGATED_MOBILE_DEVICE_SYNC_METHOD = 'pairing.delegatedMobileDevice.sync' +export const DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES = 32 + +const DelegatedPhoneSchema = z + .object({ + phoneKey: z.string().min(1).max(128), + name: z.string().min(1).max(128) + }) + .strict() + +export const DelegatedMobileDeviceSyncParamsSchema = z + .object({ + phones: z + .array(DelegatedPhoneSchema) + .max(DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES) + .refine((phones) => new Set(phones.map((phone) => phone.phoneKey)).size === phones.length, { + message: 'duplicate phoneKey' + }) + }) + .strict() + +export type DelegatedMobileDeviceSyncParams = z.infer + +export type DelegatedMobileDeviceSyncResult = { + devices: { phoneKey: string; deviceId: string; token: string }[] +} + +// Why: the paired desktop relays a phone to this host only when the host can mint its delegated device. +export const DELEGATED_MOBILE_DEVICE_SYNC_RUNTIME_CAPABILITY = + 'pairing.delegated-mobile-device.v1' as const diff --git a/src/shared/mobile-desktop-relay-contract.ts b/src/shared/mobile-desktop-relay-contract.ts new file mode 100644 index 00000000000..1a3454fac4d --- /dev/null +++ b/src/shared/mobile-desktop-relay-contract.ts @@ -0,0 +1,7 @@ +// Why: a phone may put `executionHost` on a request only after its desktop advertises this; an older +// desktop ignores the field and would run the call on itself. Not in RUNTIME_CAPABILITIES yet: the +// phone's host-scoped client ships first (S2), and adding it there is the one line that turns relay on. +export const MOBILE_DESKTOP_RELAY_RUNTIME_CAPABILITY = 'mobile.desktop-relay.v1' as const + +/** A paired desktop's relay readiness for one of its servers. */ +export type MobileDesktopRelayHostState = 'ready' | 'update-needed' | 'unavailable' diff --git a/src/shared/remote-runtime-passthrough-socket.ts b/src/shared/remote-runtime-passthrough-socket.ts new file mode 100644 index 00000000000..a70e15fe4d2 --- /dev/null +++ b/src/shared/remote-runtime-passthrough-socket.ts @@ -0,0 +1,224 @@ +import WebSocket from 'ws' +import type { PairingOffer } from './pairing' +import type { RuntimeCapability } from './protocol-version' +import { + decrypt, + decryptBytes, + deriveSharedKey, + encrypt, + generateKeyPair, + publicKeyFromBase64, + publicKeyToBase64 +} from './e2ee-crypto' +import { + classifyRemoteRuntimeReadyFrame, + formatRemoteRuntimeCloseMessage, + parseRemoteRuntimeAuthenticatedFrame +} from './remote-runtime-client-handshake' +import { RemoteRuntimeClientError } from './remote-runtime-client-error' +import { + remoteRuntimeConnectFailureMessage, + remoteRuntimeConnectOptions +} from './remote-runtime-connect-bound' +import { + REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES, + serializeRemoteRuntimePayload +} from './remote-runtime-memory-limits' +import { closeRemoteRuntimeSocket } from './remote-runtime-socket-close' +import { + startRemoteRuntimeSocketLiveness, + type RemoteRuntimeSocketLivenessOptions +} from './remote-runtime-socket-liveness' +import { RemoteRuntimeSubscriptionOutbound } from './remote-runtime-subscription-outbound' + +export type RemoteRuntimePassthroughSocket = { + /** Queues one plaintext RPC frame; false when the socket can no longer carry it. */ + send: (plaintext: string) => boolean + close: () => void +} + +export type RemoteRuntimePassthroughCallbacks = { + onText: (plaintext: string) => void + onBinary: (bytes: Uint8Array) => void + /** Called once when an open socket ends for any reason other than `close()`. */ + onClose: (error: RemoteRuntimeClientError) => void +} + +/** + * An authenticated E2EE socket that carries any number of requests and streams and hands every + * decrypted frame back untouched. Why not the subscription transport: it owns one request id per + * socket and fails on any other, while a relay forwards whatever its client multiplexes. + */ +export function openRemoteRuntimePassthroughSocket( + pairing: PairingOffer, + clientCapabilities: readonly RuntimeCapability[], + callbacks: RemoteRuntimePassthroughCallbacks, + options: RemoteRuntimeSocketLivenessOptions & { timeoutMs: number } +): Promise { + return new Promise((resolve, reject) => { + const keyPair = generateKeyPair() + const sharedKey = deriveSharedKey(keyPair.secretKey, publicKeyFromBase64(pairing.publicKeyB64)) + let state: 'awaiting_ready' | 'awaiting_authenticated' | 'ready' | 'ended' = 'awaiting_ready' + let ws: WebSocket + const outbound = new RemoteRuntimeSubscriptionOutbound({ fail: (error) => end(error) }) + const handle: RemoteRuntimePassthroughSocket = { + send: (plaintext) => + state === 'ready' && ws.readyState === WebSocket.OPEN + ? outbound.enqueueRequest(ws, encrypt(plaintext, sharedKey)) + : false, + close: () => end(null) + } + const timeout = setTimeout( + () => + end( + new RemoteRuntimeClientError( + 'runtime_timeout', + 'Timed out connecting to the remote Orca runtime.' + ) + ), + options.timeoutMs + ) + const liveness = startRemoteRuntimeSocketLiveness({ + ping: () => { + if (ws.readyState === WebSocket.OPEN) { + ws.ping() + } + }, + onDead: () => + end( + new RemoteRuntimeClientError( + 'remote_runtime_unavailable', + 'Remote Orca runtime stopped responding; the connection was reset.' + ) + ), + options + }) + + function end(error: RemoteRuntimeClientError | null): void { + if (state === 'ended') { + return + } + const wasReady = state === 'ready' + state = 'ended' + clearTimeout(timeout) + liveness.stop() + outbound.releaseQueues() + outbound.retainSocketMemoryUntilClose(ws) + ws.off('message', onMessage) + closeRemoteRuntimeSocket(ws) + if (!wasReady) { + reject( + error ?? + new RemoteRuntimeClientError('remote_runtime_unavailable', 'Remote runtime closed.') + ) + } else if (error) { + callbacks.onClose(error) + } + } + + function protocolError(message: string): RemoteRuntimeClientError { + return new RemoteRuntimeClientError('invalid_runtime_response', message) + } + + function onMessage(data: WebSocket.RawData, isBinary: boolean): void { + liveness.noteActivity() + if (isBinary) { + // Why Buffer only: a socket left at the default binaryType delivers nothing else. + const bytes = + state === 'ready' && Buffer.isBuffer(data) + ? decryptBytes(new Uint8Array(data), sharedKey) + : null + if (!bytes) { + end(protocolError('Remote Orca runtime returned an unreadable binary frame.')) + return + } + callbacks.onBinary(bytes) + return + } + const frame = data.toString() + if (state === 'awaiting_ready') { + if (classifyRemoteRuntimeReadyFrame(frame) !== 'ready') { + end(protocolError('Remote Orca runtime returned an invalid E2EE handshake frame.')) + return + } + state = 'awaiting_authenticated' + ws.send( + encrypt( + serializeRemoteRuntimePayload({ + type: 'e2ee_auth', + deviceToken: pairing.deviceToken, + clientCapabilities + }), + sharedKey + ) + ) + return + } + const plaintext = decrypt(frame, sharedKey) + if (plaintext === null) { + end(protocolError('Remote Orca runtime returned an undecryptable frame.')) + return + } + if (state === 'ready') { + callbacks.onText(plaintext) + return + } + const authenticated = parseRemoteRuntimeAuthenticatedFrame(plaintext) + if (authenticated.kind !== 'authenticated') { + end( + authenticated.kind === 'rejected' && authenticated.unauthorized + ? new RemoteRuntimeClientError( + 'unauthorized', + 'Remote Orca runtime rejected the pairing token.' + ) + : protocolError('Remote Orca runtime returned an invalid E2EE auth frame.') + ) + return + } + state = 'ready' + clearTimeout(timeout) + resolve(handle) + } + + try { + ws = new WebSocket( + pairing.endpoint, + remoteRuntimeConnectOptions({ maxPayload: REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES }) + ) + } catch (error) { + clearTimeout(timeout) + liveness.stop() + reject( + new RemoteRuntimeClientError( + 'invalid_argument', + `Invalid remote endpoint: ${String(error)}` + ) + ) + return + } + ws.once('open', () => + ws.send( + JSON.stringify({ type: 'e2ee_hello', publicKeyB64: publicKeyToBase64(keyPair.publicKey) }) + ) + ) + ws.on('error', (error) => + end( + new RemoteRuntimeClientError( + 'remote_runtime_unavailable', + remoteRuntimeConnectFailureMessage(error, pairing.endpoint) + ) + ) + ) + ws.on('close', (code, reason) => + end( + new RemoteRuntimeClientError( + 'remote_runtime_unavailable', + formatRemoteRuntimeCloseMessage(code, reason) + ) + ) + ) + ws.on('message', onMessage) + ws.on('pong', () => liveness.noteActivity()) + ws.on('ping', () => liveness.noteActivity()) + }) +} diff --git a/src/shared/terminal-stream-protocol.ts b/src/shared/terminal-stream-protocol.ts index 4641fb101da..992fbb3d8df 100644 --- a/src/shared/terminal-stream-protocol.ts +++ b/src/shared/terminal-stream-protocol.ts @@ -79,6 +79,23 @@ export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFram } } +/** The stream id in a terminal frame's header, or null when the bytes are not a terminal frame. */ +export function readTerminalStreamFrameId(bytes: Uint8Array): number | null { + if ( + bytes.length < HEADER_BYTES || + bytes[0] !== TERMINAL_STREAM_KIND || + bytes[1] !== TERMINAL_STREAM_VERSION + ) { + return null + } + return new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getUint32(4, true) +} + +/** Overwrites the stream id of a frame `readTerminalStreamFrameId` accepted. */ +export function writeTerminalStreamFrameId(bytes: Uint8Array, streamId: number): void { + new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).setUint32(4, streamId, true) +} + export function encodeTerminalStreamJson(value: unknown): Uint8Array { return new TextEncoder().encode(JSON.stringify(value)) }