diff --git a/config/scripts/vitest-node-runtime-files.mjs b/config/scripts/vitest-node-runtime-files.mjs index 2519b78ef03..77a972bb934 100644 --- a/config/scripts/vitest-node-runtime-files.mjs +++ b/config/scripts/vitest-node-runtime-files.mjs @@ -77,6 +77,8 @@ export const NODE_RUNTIME_INCLUDE = [ 'src/main/runtime/rpc/ws-empty-payload-frame-write.test.ts', 'src/shared/remote-runtime-client.test.ts', 'src/shared/remote-runtime-connect-bound.test.ts', + // Bun's built-in ws client ignores maxPayload; the product opener only runs on Node. + 'src/shared/remote-runtime-request-websocket.test.ts', 'src/shared/remote-runtime-subscription-connect-bound.test.ts', 'src/main/codex/codex-shared-server-probe.test.ts', 'src/main/opencode/opencode-launch-model-context-supervision.test.ts', diff --git a/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts b/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts new file mode 100644 index 00000000000..99ac27a765d --- /dev/null +++ b/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts @@ -0,0 +1,75 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => { + const environments: { id: string; name: string }[] = [] + return { environments, resolveManaged: vi.fn(), call: vi.fn() } +}) + +vi.mock('electron', () => ({ app: { getPath: () => '/user-data' } })) +vi.mock('../../shared/runtime-environment-store', () => ({ + listEnvironments: () => mocks.environments +})) +vi.mock('../../shared/runtime-environments', () => ({ + getPreferredPairingOffer: (environment: { endpoint: string }) => ({ + endpoint: environment.endpoint + }) +})) +vi.mock('./runtime-environment-managed-tunnel', () => ({ + resolveManagedRuntimeEnvironment: mocks.resolveManaged +})) +vi.mock('./runtime-environment-transport-routing', () => ({ + callRuntimeEnvironment: mocks.call +})) + +import { + createRuntimeEnvironmentMobileRelayHosts, + retireMobileDesktopRelayEnvironment +} from './runtime-environment-mobile-relay-hosts' + +describe('runtime environment mobile relay hosts', () => { + beforeEach(() => { + mocks.environments = [{ id: 'env-1', name: 'VM' }] + mocks.resolveManaged.mockReset().mockResolvedValue({ + id: 'env-1', + endpoint: 'ws://vm:6768', + pairingRevision: 7, + createdAt: 1, + runtimeId: 'runtime-a' + }) + mocks.call.mockReset() + }) + + it('resolves a configured server by exact id, fenced by pairing revision and runtime identity', async () => { + const hosts = createRuntimeEnvironmentMobileRelayHosts() + await expect(hosts.resolve('env-1')).resolves.toEqual({ + environmentId: 'env-1', + fence: '7\0runtime-a', + pairing: { endpoint: 'ws://vm:6768' } + }) + expect(mocks.resolveManaged).toHaveBeenCalledWith('/user-data', 'env-1') + }) + + it('answers null for a name, an unknown id or a removed server, without resolving it', async () => { + const hosts = createRuntimeEnvironmentMobileRelayHosts() + await expect(hosts.resolve('VM')).resolves.toBeNull() + await expect(hosts.resolve('env-2')).resolves.toBeNull() + mocks.environments = [] + await expect(hosts.resolve('env-1')).resolves.toBeNull() + expect(mocks.resolveManaged).not.toHaveBeenCalled() + }) + + it('calls the server as the desktop and reports retirements until unsubscribed', async () => { + const hosts = createRuntimeEnvironmentMobileRelayHosts() + mocks.call.mockResolvedValue({ id: 'x', ok: true, result: {}, _meta: { runtimeId: 'r' } }) + const host = (await hosts.resolve('env-1'))! + await hosts.call(host, 'status.get', undefined) + expect(mocks.call).toHaveBeenCalledWith('/user-data', 'env-1', 'status.get', undefined) + + const retired: string[] = [] + const stop = hosts.onEnvironmentRetired((id) => retired.push(id)) + retireMobileDesktopRelayEnvironment('env-1') + stop() + retireMobileDesktopRelayEnvironment('env-1') + expect(retired).toEqual(['env-1']) + }) +}) 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 20c53748643..418d558fd96 100644 --- a/src/main/ipc/runtime-environments.ts +++ b/src/main/ipc/runtime-environments.ts @@ -31,6 +31,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() @@ -73,6 +74,7 @@ export function invalidateRuntimeEnvironmentTransport(environmentId: string): Pr advanceRuntimeEnvironmentTransportGeneration(environmentId) closeRemoteRuntimeRequestConnection(environmentId) closeSubscriptionsForEnvironment(environmentId) + retireMobileDesktopRelayEnvironment(environmentId) reactivateRuntimeEnvironmentStatus(environmentId) return retirePairedRuntimeBrowserClientHostEnvironment( environmentId, 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..116f2cd5197 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/delegated-phone-grants.ts @@ -0,0 +1,69 @@ +import { z } from 'zod' +import { + DELEGATED_MOBILE_DEVICE_SYNC_METHOD, + DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES, + DELEGATED_MOBILE_DEVICES_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 + // Every phone this sync asked for; one asked for but not granted waits for the next sync. + asked: ReadonlySet + // 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_DEVICES_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 } + ]) + ), + asked: new Set(phones.map((phone) => phone.phoneKey)), + 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..e21d2b3b3d2 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts @@ -0,0 +1,488 @@ +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_METHOD, + DelegatedMobileDeviceSyncParamsSchema +} from '../../../shared/delegated-mobile-device-contract' +import { decodeTerminalStreamFrame } from '../../../shared/terminal-stream-protocol' +import { sendRemoteRuntimeRequest } from '../../../shared/remote-runtime-client' +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 = () => registry.listDelegatedMobileDevices(desktopOnHost.pairedDeviceId!) + const syncedNames: string[][] = [] + const state = { capable: true, retire: (_id: string) => {} } + const hosts: MobileDesktopRelayHosts = { + resolve: async (environmentId) => + environmentId === 'env-1' + ? { environmentId, fence: 'pairing-1', pairing: desktopOnHost } + : null, + call: async (_host, method, params) => { + if (method === DELEGATED_MOBILE_DEVICE_SYNC_METHOD) { + const { phones } = DelegatedMobileDeviceSyncParamsSchema.parse(params) + syncedNames.push(phones.map((phone) => phone.name)) + } + const response = await sendRemoteRuntimeRequest(desktopOnHost, method, params, 5_000) + // An older server: the same status without the delegated-devices capability. + return method === 'status.get' && !state.capable && response.ok + ? { + ...response, + result: { ...z.object({}).passthrough().parse(response.result), capabilities: [] } + } + : response + }, + 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, + deviceId: pairing.pairedDeviceId!, + 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() + // A phone another desktop relays to this one is that desktop's to relay, never re-relayed. + const registry = desktop.server.getDeviceRegistry()! + const otherDesktop = registry.getDevice( + pair(desktop.server, 'runtime', 'Other Mac').pairedDeviceId! + )! + registry.upsertDelegatedMobileDevices(otherDesktop, [ + { phoneKey: 'p', name: 'iPad via Other Mac' } + ]) + const phone = await connectPhone(desktop.server, 'iPhone') + + const streamId = await subscribe(phone, handle) + // Host and desktop share a process counter here, so compare against the id the host sent. + const hostStreamId = z + .object({ + id: z.literal('sub'), + result: z.object({ type: z.literal('subscribed'), streamId: z.number() }) + }) + .parse( + passthroughOpens.hostReplies + .map((frame) => JSON.parse(frame)) + .find((frame) => frame.id === 'sub' && frame.result?.type === 'subscribed') + ).result.streamId + expect(streamId).not.toBe(hostStreamId) + // The snapshot arrives as binary frames under the id the phone was given, never the host's. + await vi.waitFor(() => + expect( + phone.binaries.some((bytes) => decodeTerminalStreamFrame(bytes)?.streamId === streamId) + ).toBe(true) + ) + expect( + phone.binaries.some((bytes) => decodeTerminalStreamFrame(bytes)?.streamId === hostStreamId) + ).toBe(false) + expect(syncedNames).toEqual([[`iPhone via ${desktop.runtime.readMachineName()}`]]) + const delegated = children()[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().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('drops a phone from the server as soon as the desktop unpairs it', async () => { + const { host, desktop, children, handle } = await startTopology() + const kept = await connectPhone(desktop.server, 'Kept') + const unpaired = await connectPhone(desktop.server, 'Unpaired') + await subscribe(kept, handle) + expect(children().map((device) => device.phoneKey)).toEqual([kept.deviceId, unpaired.deviceId]) + + await desktop.server.revokeMobileDevice(unpaired.deviceId) + await vi.waitFor(() => + expect(children().map((device) => device.phoneKey)).toEqual([kept.deviceId]) + ) + expect(host.runtime.seatClientIds('pty-shared')).toEqual([children()[0]!.token]) + }) + + 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()[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('answers a targeted desktop-owned call on the desktop, exactly as the untargeted one', async () => { + const { desktop } = await startTopology() + const phone = await connectPhone(desktop.server, 'iPhone') + const opensBefore = passthroughOpens.count + + for (const method of ['settings.get', 'status.get']) { + phone.send(`${method}:targeted`, method, {}) + phone.send(`${method}:untargeted`, method, {}, null) + const targeted = await phone.next(`${method}:targeted`) + const untargeted = await phone.next(`${method}:untargeted`) + expect(isOk(targeted)).toBe(true) + expect(targeted.result).toEqual(untargeted.result) + } + expect(passthroughOpens.count).toBe(opensBefore) + // An execution-host method under the same target still relays. + phone.send('list', 'terminal.list', {}) + expect(isOk(await phone.next('list'))).toBe(true) + expect(passthroughOpens.count).toBe(opensBefore + 1) + }) + + it('refuses binary-frame 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('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, `local` and the desktop's own SSH hosts are today's local call. + const opensBefore = passthroughOpens.count + phone.send('absent', 'terminal.list', {}, null) + phone.send('local', 'terminal.list', {}, 'local') + phone.send('ssh', 'terminal.list', {}, 'ssh:devbox') + for (const id of ['absent', 'local', 'ssh']) { + expect(isOk(await phone.next(id))).toBe(true) + } + expect(passthroughOpens.count).toBe(opensBefore) + }) + + 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(await phone.next('list')).toMatchObject({ + error: { code: 'remote_runtime_unavailable', message: expect.stringContaining('update') } + }) + expect(passthroughOpens.count).toBe(opensBefore) + state.capable = true + state.retire('env-1') + phone.send('updated', 'terminal.list', {}) + expect(isOk(await phone.next('updated'))).toBe(true) + }) +}) 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..46e9a86c5f2 --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.test.ts @@ -0,0 +1,286 @@ +import { describe, expect, it, vi } from 'vitest' +import { DELEGATED_MOBILE_DEVICES_RUNTIME_CAPABILITY } from '../../../shared/delegated-mobile-device-contract' +import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope' +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 + accepts: boolean +} +const sockets = vi.hoisted(() => { + const opened: FakeSocket[] = [] + return opened +}) +// Holds every socket open until released, so a test can act while one is still connecting. +const openGate = vi.hoisted(() => ({ wait: Promise.resolve() })) +vi.mock('../../../shared/remote-runtime-passthrough-socket', () => ({ + openRemoteRuntimePassthroughSocket: async ( + _pairing: unknown, + capabilities: readonly string[], + callbacks: RemoteRuntimePassthroughCallbacks + ) => { + const socket: FakeSocket = { sent: [], capabilities, callbacks, closed: false, accepts: true } + sockets.push(socket) + await openGate.wait + return { + send: (frame: string) => { + socket.sent.push(frame) + return socket.accepts + }, + close: () => { + socket.closed = true + } + } + } +})) + +const ok = (result: unknown) => ({ id: 'x', ok: true as const, result, _meta: { runtimeId: 'h' } }) + +const SYNCED = ok({ + devices: [ + { phoneKey: 'phone-1', deviceId: 'child', token: 'host-token' }, + { phoneKey: 'phone-2', deviceId: 'child-2', token: 'host-token-2' } + ] +}) + +function relayWithPhone( + syncResponse: () => RuntimeRpcResponse | Promise> = () => + SYNCED +) { + const syncCalls: unknown[] = [] + const server = { capable: true } + 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, params) => { + if (method === 'status.get') { + return ok({ + capabilities: server.capable ? [DELEGATED_MOBILE_DEVICES_RUNTIME_CAPABILITY] : [] + }) + } + syncCalls.push(params) + return syncResponse() + }, + onEnvironmentRetired: () => () => {} + } + const relay = new MobileDesktopRelay({ + hosts, + listPhones: () => [ + { phoneKey: 'phone-1', name: 'iPhone via Mac' }, + { phoneKey: 'phone-2', name: 'iPad 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, + syncCalls, + server, + 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') + await vi.waitFor(() => expect(sockets[1]!.closed).toBe(true)) + }) + + it('treats a server whose sync failed as unavailable, and syncs again on the next request', async () => { + sockets.length = 0 + let failing = true + const { relay, phone, replies } = relayWithPhone(() => + failing + ? { + id: 'x', + ok: false, + error: { code: 'runtime_error', message: 'delegated_device_revoke_failed' } + } + : ok({ devices: [{ phoneKey: 'phone-1', deviceId: 'child', token: 'host-token' }] }) + ) + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(replies).toHaveLength(1)) + expect(JSON.parse(replies[0]!)).toMatchObject({ + id: 'a', + error: { code: 'remote_runtime_unavailable' } + }) + expect(sockets).toHaveLength(0) + + failing = false + relay.forward(phone, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}') + await vi.waitFor(() => expect(sockets[0]?.sent).toHaveLength(1)) + }) + + it('runs one sync when two phones race to the same server', async () => { + sockets.length = 0 + const { relay, phone, syncCalls } = relayWithPhone() + const tablet: RelayedPhone = { ...phone, connectionId: 'conn-2', deviceId: 'phone-2' } + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + relay.forward(tablet, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}') + await vi.waitFor(() => expect(sockets.map((socket) => socket.sent.length)).toEqual([1, 1])) + expect(syncCalls).toHaveLength(1) + }) + + it('closes a socket still opening when its server is removed, answering and never sending its requests', async () => { + sockets.length = 0 + let release = () => {} + openGate.wait = new Promise((resolve) => (release = resolve)) + const { relay, phone, replies } = relayWithPhone() + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(sockets).toHaveLength(1)) + relay.retireEnvironment('env-1') + expect(JSON.parse(replies[0]!)).toMatchObject({ + id: 'a', + error: { code: 'remote_runtime_unavailable' } + }) + release() + await vi.waitFor(() => expect(sockets[0]!.closed).toBe(true)) + expect(sockets[0]!.sent).toEqual([]) + openGate.wait = Promise.resolve() + }) + + it('opens nothing for a phone that disconnects while its server syncs, and answers nobody', async () => { + sockets.length = 0 + let finishSync = () => {} + const { relay, phone, replies, syncCalls } = relayWithPhone( + () => new Promise((resolve) => (finishSync = () => resolve(SYNCED))) + ) + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(syncCalls).toHaveLength(1)) + relay.closePhoneConnection('conn-1') + finishSync() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(sockets).toEqual([]) + expect(replies).toEqual([]) + }) + + it('relays to a server upgraded in place without waiting for a re-pair', async () => { + sockets.length = 0 + const { relay, phone, replies, server } = relayWithPhone() + server.capable = false + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(replies).toHaveLength(1)) + expect(JSON.parse(replies[0]!).error.message).toContain('update') + server.capable = true + relay.forward(phone, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}') + await vi.waitFor(() => expect(sockets[0]?.sent).toHaveLength(1)) + }) + + it('does not sync again for a phone the server just declined to grant', async () => { + sockets.length = 0 + const { relay, phone, replies, syncCalls } = relayWithPhone(() => + ok({ devices: [{ phoneKey: 'phone-2', deviceId: 'child-2', token: 'host-token-2' }] }) + ) + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(replies).toHaveLength(1)) + relay.forward(phone, 'env-1', { id: 'b', method: 'terminal.list' }, '{"id":"b"}') + await vi.waitFor(() => expect(replies).toHaveLength(2)) + expect(replies.map((frame) => JSON.parse(frame).error.code)).toEqual([ + 'remote_runtime_unavailable', + 'remote_runtime_unavailable' + ]) + expect(syncCalls).toHaveLength(1) + expect(sockets).toEqual([]) + }) + + it('ends the link when a mirrored capability update cannot be queued', async () => { + sockets.length = 0 + const { relay, phone, replies } = relayWithPhone() + relay.forward(phone, 'env-1', { id: 'a', method: 'terminal.list' }, '{"id":"a"}') + await vi.waitFor(() => expect(sockets[0]?.sent).toHaveLength(1)) + sockets[0]!.accepts = false + relay.forwardClientCapabilities( + 'conn-1', + '{"id":"caps","method":"runtime.clientCapabilities.update"}' + ) + await vi.waitFor(() => expect(sockets[0]!.closed).toBe(true)) + expect(JSON.parse(replies[0]!)).toMatchObject({ + id: 'a', + error: { code: 'remote_runtime_unavailable' } + }) + }) +}) 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..20386ee042c --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.ts @@ -0,0 +1,325 @@ +import { randomUUID } from 'node:crypto' +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 { 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 GrantsEntry = { + promise: Promise + settled: boolean + result?: DelegatedPhoneGrants +} + +type OpenedRelayLink = { socket: RemoteRuntimePassthroughSocket; hostToken: string } + +class RelayLink { + 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 phone request carrying `executionHost: runtime:` to that configured server, over one + * socket per (phone connection, server) signed in as the phone's delegated device (minted by + * `pairing.delegatedMobileDevice.sync`, tokens held in memory only). Contract: + * - The desktop never interprets replies; it renumbers only host stream ids (RELAYED_STREAM_CARRIERS). + * - Upstream, the phone's desktop token is swapped wherever it appears, not per field. + * - Only `execution-host` methods (MOBILE_RPC_METHOD_ROUTES) reach here; `paired-desktop` ones run locally. + * - A server without the delegated-devices capability is `update-needed`; nothing relays to it. + * - A server-side revoke of a phone is final until the next sync. + * - Hidden until S2: MOBILE_DESKTOP_RELAY_RUNTIME_CAPABILITY is not advertised yet. + */ +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 (RELAYED_STREAM_CARRIERS.get(request.method) === 'phone-binary-frames') { + phone.reply( + failure( + request.id, + 'forbidden', + `Method '${request.method}' is driven by binary frames, which the desktop does not relay` + ) + ) + return + } + const link = this.linkFor(phone, environmentId) + link.replies.noteForwarded(request.id) + link.opened.then( + ({ socket, hostToken }) => { + // Why: a link ended while opening already answered this request; never send it late. + if (!this.isCurrent(link)) { + return + } + 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 }) => { + if ( + this.isCurrent(link) && + !socket.send(rewriteRelayedPhoneRequest(frame, link.phone.deviceToken, hostToken, id)) + ) { + this.endLink(link, unavailable('The server connection could not take the request.')) + } + }, + () => {} + ) + } + } + + 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.resync(environmentId) + } + } + + private async resync(environmentId: string): Promise { + try { + const host = await this.options.hosts.resolve(environmentId) + if (host) { + await this.grantsFor(host, null) + } + } catch { + // An unreachable server syncs again on its next relayed request. + } + } + + 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.') + } + // Why: the phone left or the server was retired while syncing; open nothing for it. + if (!this.isCurrent(link)) { + throw unavailable('The relayed request was cancelled.') + } + 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 } + ) + 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) + if (cached && !cached.settled) { + // Why: phones racing to one server share its sync. + return cached.promise.then((grants) => + grants.fence === host.fence ? grants : this.grantsFor(host, phoneKey) + ) + } + const answered = + cached?.result?.kind === 'ready' && + cached.result.fence === host.fence && + (phoneKey === null || cached.result.asked.has(phoneKey)) + // Why update-needed is never reused: a server upgraded in place must relay without a re-pair. + return answered && cached.result ? Promise.resolve(cached.result) : this.sync(host) + } + + private sync(host: MobileDesktopRelayHost): Promise { + const entry: GrantsEntry = { + promise: syncDelegatedPhoneGrants(this.options.hosts, host, this.options.listPhones()), + settled: false + } + this.grants.set(host.environmentId, entry) + entry.promise.then( + (result) => { + entry.settled = true + entry.result = result + }, + () => { + // Why: an unreachable server is not an answer; the next request syncs again. + if (this.grants.get(host.environmentId) === entry) { + this.grants.delete(host.environmentId) + } + } + ) + return entry.promise + } + + private isCurrent(link: RelayLink): boolean { + return this.links.get(linkKey(link.phone.connectionId, link.environmentId)) === link + } + + private endLink(link: RelayLink, error: RemoteRuntimeClientError | null): void { + if (!this.isCurrent(link)) { + return + } + this.links.delete(linkKey(link.phone.connectionId, link.environmentId)) + link.opened.then( + ({ socket }) => socket.close(), + () => {} + ) + const openRequestIds = link.replies.takeOpenRequestIds() + if (error) { + for (const id of openRequestIds) { + link.phone.reply(failure(id, error.code, 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..e02301450fe --- /dev/null +++ b/src/main/runtime/mobile-desktop-relay/relayed-phone-frames.ts @@ -0,0 +1,117 @@ +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 { + // Why swallow: requests the relay sends on its own behalf have no phone waiting on them. + private readonly openRequests = new Map() + private readonly desktopStreamIds = new Map() + + constructor(private readonly allocateStreamId: () => number) {} + + noteForwarded(requestId: string, options: { swallowReplies?: boolean } = {}): void { + this.openRequests.set(requestId, { + hostStreamIds: [], + swallow: options.swallowReplies === true + }) + } + + /** Request ids still waiting on the host; the caller fails them when the socket ends. */ + takeOpenRequestIds(): string[] { + const ids = [...this.openRequests].filter(([, open]) => !open.swallow).map(([id]) => id) + this.openRequests.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 open = this.openRequests.get(parsed.id) + const result = parsed.result + let renumbered = false + if (open && isRecord(result) && typeof result.streamId === 'number') { + result.streamId = this.desktopStreamId(result.streamId, open.hostStreamIds) + renumbered = true + } + // Why bounded: a stream ended by its own unsubscribe sends no 'end'; it stays until the link closes. + if (open && (parsed.streaming !== true || (isRecord(result) && result.type === 'end'))) { + this.finish(parsed.id, open.hostStreamIds) + } + if (open?.swallow) { + 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) + 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/rpc/core.ts b/src/main/runtime/rpc/core.ts index 86226f83522..f58b10313bb 100644 --- a/src/main/runtime/rpc/core.ts +++ b/src/main/runtime/rpc/core.ts @@ -60,6 +60,8 @@ export type RpcRequest = { orchestrationCompatibilityEvidence?: OrchestrationCompatibilityEvidence // Why unknown: untrusted wire input; runtime-source-admission.ts validates it. expectedRuntimeSource?: unknown + // Why unknown: phone-supplied routing target (`runtime:`), validated where it is read. + executionHost?: unknown } export type LegacyCoordinatorAuthorityProof = Readonly<{ diff --git a/src/main/runtime/runtime-rpc-mobile-legacy-socket-relay-refusal.test.ts b/src/main/runtime/runtime-rpc-mobile-legacy-socket-relay-refusal.test.ts new file mode 100644 index 00000000000..331a06e628a --- /dev/null +++ b/src/main/runtime/runtime-rpc-mobile-legacy-socket-relay-refusal.test.ts @@ -0,0 +1,42 @@ +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' +import { OrcaRuntimeRpcServer } from './runtime-rpc' +import { DeviceRegistry } from './device-registry' +import { createMobileRpcSurfaceRuntime } from './runtime-rpc-mobile-method-allowlist-fixtures' + +describe('a server-targeted phone call on a legacy (non-E2EE) socket', () => { + it('is refused, never run on this desktop', async () => { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-legacy-relay-')) + const { runtime, mocks } = createMobileRpcSurfaceRuntime() + const server = new OrcaRuntimeRpcServer({ runtime, userDataPath, enableWebSocket: false }) + server['deviceRegistry'] = new DeviceRegistry(userDataPath) + const device = server['deviceRegistry']!.addDevice('phone', 'mobile') + const replies: Record[] = [] + await server['handleWebSocketMessage']( + JSON.stringify({ + id: 'req_1', + method: 'terminal.read', + executionHost: 'runtime:env-1', + deviceToken: device.token, + params: { terminal: 'terminal-1' } + }), + (response) => { + const parsed: Record = JSON.parse(response) + replies.push(parsed) + }, + () => {} + ) + expect(mocks.readTerminal).not.toHaveBeenCalled() + expect(replies).toMatchObject([ + { + ok: false, + error: { + code: 'remote_runtime_unavailable', + message: 'Servers are reachable only over an encrypted pairing' + } + } + ]) + }) +}) diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts index 88c711ebd70..3b01befecd5 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts @@ -261,6 +261,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-agent-session-methods.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts index 6ceb45937d8..f30ceee365a 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts @@ -1,4 +1,4 @@ -/** The structured agent-session methods a phone may call; spread into the mobile allowlist. */ +/** The structured agent-session methods a phone may call; spread into the execution-host routes. */ export const MOBILE_AGENT_SESSION_RPC_METHODS = [ 'agentSession.createSupport', 'agentSession.create', diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-devices.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-devices.ts index bf5eb52b4cc..73c5f243b7d 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-devices.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-devices.ts @@ -10,7 +10,11 @@ export class RuntimeRpcMobileDevices extends RuntimeRpcRequestAdmission { private onPushUnregisterQueued?: () => void async revokeMobileDevice(deviceId: string): Promise { - return this.revokeMobileDeviceNow(deviceId) + const revoked = this.revokeMobileDeviceNow(deviceId) + if (revoked) { + this.mobileDesktopRelay?.phonesChanged() + } + return revoked } private revokeMobileDeviceNow(deviceId: string): boolean { diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-allowlist.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-allowlist.ts index c6e43823a80..10a3b6a77ac 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-allowlist.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-allowlist.ts @@ -1,284 +1,4 @@ -import { MOBILE_AGENT_SESSION_RPC_METHODS } from './runtime-rpc-mobile-agent-session-methods' +import { MOBILE_RPC_METHOD_ROUTES } from './runtime-rpc-mobile-method-routing' -export const MOBILE_RPC_METHOD_ALLOWLIST = new Set([ - 'accounts.list', - 'accounts.consumeCodexResetCredit', - 'accounts.selectClaude', - 'accounts.selectCodex', - 'accounts.selectCodexForTarget', - 'accounts.subscribe', - 'accounts.unsubscribe', - 'agent.launch', - 'agent.launchReplay', - 'aiVault.listSessions', - 'aiVault.searchSessions', - 'aiVault.searchStatus', - 'aiVault.resolveSessionTitles', - 'aiVault.prepareSessionResume', - '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', - 'clipboard.abortImageUpload', - 'clipboard.appendImageUploadChunk', - 'clipboard.commitImageUpload', - 'clipboard.saveImageAsTempFile', - 'clipboard.startImageUpload', - 'diagnostics.memory', - 'files.browseServerDir', - 'files.createFile', - 'files.list', - 'files.open', - 'files.openDiff', - 'files.read', - 'files.readDocPreview', - 'files.readChunk', - 'files.readDir', - 'files.readPreview', - 'files.readTerminalArtifact', - 'files.readTerminalArtifactPreview', - 'files.resolveTerminalPath', - 'files.searchPaths', - 'files.stat', - 'files.writeTerminalArtifact', - 'folderWorkspace.list', - 'git.abortMerge', - 'git.abortRebase', - 'git.bulkStage', - 'git.bulkUnstage', - 'git.branchCompare', - 'git.branchDiff', - 'git.cancelGenerateCommitMessage', - 'git.cancelGeneratePullRequestFields', - 'git.checkout', - 'git.commit', - 'git.commitCompare', - 'git.commitDiff', - 'git.discard', - 'git.discoverCommitMessageModels', - 'git.diff', - 'git.fetch', - 'git.forkSync', - 'git.fastForward', - 'git.generateCommitMessage', - 'git.generatePullRequestFields', - 'git.history', - 'git.localBranches', - 'git.pull', - 'git.push', - 'git.rebaseFromBase', - '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.setPRAutoMerge', - 'github.requestPRReviewers', - 'github.removePRReviewers', - '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.prForBranch', - 'github.prFileContents', - 'github.prChecks', - 'github.prCheckDetails', - 'github.rerunPRChecks', - 'github.resolveReviewThread', - 'github.setPRFileViewed', - 'github.updateIssue', - 'github.updatePR', - 'github.updatePRTitle', - 'github.updatePRState', - 'github.repoSlug', - 'github.workItem', - // Cross-repo lookup: lets the mobile Smart picker resolve a pasted github.com URL for a different repo. - 'github.workItemByOwnerRepo', - 'github.workItemDetails', - 'gitlab.createIssue', - 'gitlab.addIssueComment', - 'gitlab.addMRComment', - 'gitlab.listWorkItems', - // Mobile Smart picker: resolve a pasted GitLab URL to an exact issue/MR (MR listing reuses gitlab.listWorkItems). - 'gitlab.workItemByPath', - 'gitlab.mergeMR', - 'gitlab.resolveMRDiscussion', - 'gitlab.todos', - 'gitlab.updateIssue', - 'gitlab.updateMR', - 'gitlab.updateMRState', - 'gitlab.workItemDetails', - 'host.gitBash.isAvailable', - 'host.platform', - 'host.pwsh.isAvailable', - 'host.wsl.isAvailable', - 'host.wsl.listDistros', - 'hostedReview.create', - 'hostedReview.createStacked', - 'hostedReview.forBranch', - 'hostedReview.getCreationEligibility', - 'linear.getCustomView', - 'linear.getIssue', - 'linear.getProject', - 'linear.agentSearchIssues', - 'linear.issueContext', - 'linear.resolveCurrentIssue', - 'linear.addIssueComment', - 'linear.connect', - 'linear.createIssue', - 'linear.createProject', - 'linear.issueComments', - 'linear.listCustomViewIssues', - 'linear.listCustomViewProjects', - 'linear.listCustomViews', - 'linear.listIssues', - 'linear.mcpListIssues', - 'linear.listProjectIssues', - 'linear.listProjects', - 'linear.teamLabels', - 'linear.teamMembers', - 'linear.listTeams', - 'linear.searchIssues', - 'linear.selectWorkspace', - 'linear.status', - 'linear.teamStates', - 'linear.updateIssue', - 'markdown.readTab', - 'markdown.saveTab', - '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', - 'preflight.check', - 'preflight.detectAgents', - 'preflight.detectRemoteAgents', - 'projectGroup.list', - 'repo.baseRefDefault', - 'repo.gitAvailable', - 'repo.hooks', - 'repo.list', - 'repo.saveSparsePreset', - 'repo.searchRefs', - 'repo.sparsePresets', - 'repo.update', - 'runtime.clientCapabilities.update', - 'runtime.clientEvents.subscribe', - 'runtime.clientEvents.unsubscribe', - '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', - ...MOBILE_AGENT_SESSION_RPC_METHODS, - 'nativeChat.readSession', - 'nativeChat.subscribe', - 'nativeChat.unsubscribe', - 'settings.get', - 'settings.getTerminalQuickCommands', - 'settings.mutateNativeChatSessionOptions', - 'settings.update', - 'settings.updateTerminalQuickCommands', - 'ssh.connect', - 'ssh.getState', - 'ssh.listRemovedTargetLabels', - 'ssh.listTargets', - 'ssh.listTargetSummaries', - 'speech.dictation.cancel', - 'speech.dictation.chunk', - 'speech.dictation.finish', - 'speech.dictation.setup', - 'speech.dictation.start', - 'speech.models.delete', - 'speech.models.download', - 'speech.models.list', - 'stats.summary', - 'status.get', - 'agentTeams.prepareLaunch', - 'agentTeams.tmuxCompat', - // Why: the phone reports a takeover out of band, the same signal the desktop renderer sends. - 'orchestration.workerTerminalUserInput', - 'terminal.clearBuffer', - 'terminal.close', - 'terminal.closeAll', - 'terminal.closeTab', - 'terminal.create', - 'terminal.createAgentSession', - 'terminal.ensureAgentSession', - 'terminal.focus', - 'terminal.agentStatus', - 'terminal.adoptOrphans', - 'terminal.getAutoRestoreFit', - 'terminal.isRunningAgent', - 'terminal.list', - 'terminal.multiplex', - 'terminal.read', - 'terminal.rename', - 'terminal.send', - 'terminal.setAutoRestoreFit', - 'terminal.setDisplayMode', - 'terminal.subscribe', - 'terminal.unsubscribe', - 'terminal.updateViewport', - 'terminal.wait', - 'ui.get', - 'ui.recordFeatureInteraction', - 'ui.set', - 'worktree.activate', - 'worktree.create', - 'worktree.forceDeleteBranch', - 'worktree.listRetiredNames', - 'worktree.prefetchCreateBase', - 'worktree.ps', - 'worktree.show', - 'worktree.resolveMrBase', - 'worktree.resolvePrBase', - 'worktree.rm', - 'worktree.set', - 'worktree.sleep' -]) +// Why: the routing table is the one list of phone methods, so allowlisting a method means routing it. +export const MOBILE_RPC_METHOD_ALLOWLIST = new Set(MOBILE_RPC_METHOD_ROUTES.keys()) 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..18a9de170a4 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, it } from 'vitest' +import { + EXECUTION_HOST_METHODS, + MOBILE_RPC_METHOD_ROUTES +} from './runtime-rpc-mobile-method-routing' +import { PAIRED_DESKTOP_METHODS } from './runtime-rpc-mobile-paired-desktop-methods' + +describe('mobile RPC method routing census', () => { + it('tags every routed method exactly once', () => { + expect(PAIRED_DESKTOP_METHODS.length + EXECUTION_HOST_METHODS.length).toBe( + MOBILE_RPC_METHOD_ROUTES.size + ) + }) +}) 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..0f6a506a925 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-method-routing.ts @@ -0,0 +1,204 @@ +import { MOBILE_AGENT_SESSION_RPC_METHODS } from './runtime-rpc-mobile-agent-session-methods' +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. A route is what allowlists a phone method. +export type MobileRpcMethodRoute = 'execution-host' | 'paired-desktop' + +export 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', + // Cross-repo lookup: lets the mobile Smart picker resolve a pasted github.com URL for a different repo. + 'github.workItemByOwnerRepo', + 'github.workItemDetails', + 'gitlab.addIssueComment', + 'gitlab.addMRComment', + 'gitlab.createIssue', + 'gitlab.listWorkItems', + 'gitlab.mergeMR', + 'gitlab.resolveMRDiscussion', + 'gitlab.todos', + 'gitlab.updateIssue', + 'gitlab.updateMR', + 'gitlab.updateMRState', + // Mobile Smart picker: resolve a pasted GitLab URL to an exact issue/MR (MR listing reuses gitlab.listWorkItems). + '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', + // Why: the phone reports a takeover out of band, the same signal the desktop renderer sends. + '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', + ...MOBILE_AGENT_SESSION_RPC_METHODS, + '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) +]) 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..14ac2c5dcf7 --- /dev/null +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-paired-desktop-methods.ts @@ -0,0 +1,103 @@ +/** Phone methods the relaying Mac always answers itself, even when the call names a server. */ +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 7c60b590eff..13538acc18d 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts @@ -8,6 +8,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 { DELEGATED_PHONE_NAME_MAX_CHARS } from '../../../shared/delegated-mobile-device-contract' +import { allocateTerminalSubscriptionStreamId } from '../rpc/methods/terminal/terminal-subscription-stream-id' import { createWebClientUrl, DEVICE_REGISTRY_UNAVAILABLE_GUIDANCE, @@ -84,6 +88,32 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure { this.mobileRelayPairingProviderInstaller = install } + 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 + } + + private listRelayedPhones(): { phoneKey: string; name: string }[] { + const computerName = this.runtime.readMachineName() + return (this.deviceRegistry?.listDevices() ?? []) + .filter( + // Why: phones another desktop relays here are that desktop's to relay, not ours. + (device) => + device.scope === 'mobile' && device.lastSeenAt > 0 && device.parentDeviceId === undefined + ) + .map((device) => ({ + phoneKey: device.deviceId, + name: `${device.name} via ${computerName}`.slice(0, DELEGATED_PHONE_NAME_MAX_CHARS) + })) + } + getWebSocketEndpoint(): string | null { const ws = this.transports.find((t) => t.kind === 'websocket') return ws?.endpoint ?? null diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-state.ts b/src/main/runtime/runtime-rpc/runtime-rpc-state.ts index 975085275ff..64b44e58733 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-state.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-state.ts @@ -10,6 +10,7 @@ import type { E2EEKeypair } from '../e2ee-keypair' import type { RuntimeHostDescriptor } from '../../../shared/runtime-host-descriptor' 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' @@ -72,6 +73,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 // Why: lets an automatic mint install a provider whose launch-time construction failed. protected mobileRelayPairingProviderInstaller: (() => Promise) | null = null protected mobileRelayPairingOfferQueue: Promise = Promise.resolve() 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 c3809ab190a..91236b64437 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts @@ -13,6 +13,9 @@ import { classifyRuntimeLongPoll } from './runtime-rpc-long-poll' import type { RpcBinarySender } from '../rpc/rpc-binary-sender' import type { RpcCallerScope } from '../rpc/rpc-caller-scope' import { limitRuntimeRpcReplySize } from './runtime-rpc-reply-size-limit' +import { MOBILE_RPC_METHOD_ALLOWLIST } from './runtime-rpc-mobile-method-allowlist' +import { MOBILE_RPC_METHOD_ROUTES } from './runtime-rpc-mobile-method-routing' +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 { @@ -29,6 +32,71 @@ function injectDeviceScope(response: string, scope: DeviceScope): string { } export class RuntimeRpcWebSocketDispatch extends RuntimeRpcMobileDevices { + // 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: RpcBinarySender, + socket: AuthenticatedMobileSocket | undefined + ): boolean { + const { executionHost } = request + if (executionHost === undefined) { + return false + } + const target = parseExecutionHostId(typeof executionHost === 'string' ? executionHost : null) + if (target?.kind === 'local' || target?.kind === 'ssh') { + return false + } + // Why: the phone tags every call on a workspace screen; only this Mac's routing table decides. + if (target && MOBILE_RPC_METHOD_ROUTES.get(request.method) !== 'execution-host') { + return false + } + if (!target) { + reply( + JSON.stringify(this.buildError(request.id, 'invalid_argument', 'Invalid executionHost')) + ) + } else if (!socket) { + reply( + JSON.stringify( + this.buildError( + request.id, + 'remote_runtime_unavailable', + 'Servers are reachable only over an encrypted pairing' + ) + ) + ) + } 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: (bytes) => { + sendBinary(bytes) + } + }, + 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, @@ -81,6 +149,20 @@ export class RuntimeRpcWebSocketDispatch extends RuntimeRpcMobileDevices { ? { kind: 'mobile', delegated: device.parentDeviceId !== undefined } : { kind: 'runtime-paired', grants: device.grants ?? [] } + // Why: relayed calls skip this dispatcher's scope check, so only allowlisted methods may relay. + if (device.scope === 'mobile' && MOBILE_RPC_METHOD_ALLOWLIST.has(request.method)) { + // Why: the phone's open server sockets must see the same capabilities the desktop does. + if (authenticatedSocket && request.method === 'runtime.clientCapabilities.update') { + this.mobileDesktopRelay?.forwardClientCapabilities( + authenticatedSocket.connectionId, + rawMessage + ) + } + if (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 cfc85d7b75e..9008145308e 100644 --- a/src/main/startup/main-process-runtime-launch-activation.test.ts +++ b/src/main/startup/main-process-runtime-launch-activation.test.ts @@ -30,6 +30,7 @@ vi.mock('../runtime/runtime-rpc', () => ({ launchOrder.push('rpc-start') }) setOnUnpairedDeviceAuthFailure = vi.fn() + setMobileDesktopRelayHosts = vi.fn() } })) vi.mock('../runtime/headless-runtime-graph', () => ({ publishHeadlessRuntimeGraph: vi.fn() })) @@ -39,6 +40,9 @@ vi.mock('./headless-serve-ssh-registration', () => ({ }) })) 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 80979d6ca54..b8050920f6c 100644 --- a/src/main/startup/main-process-runtime-launch.ts +++ b/src/main/startup/main-process-runtime-launch.ts @@ -6,6 +6,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' @@ -94,6 +95,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 index 9a0032eee27..ec248eda586 100644 --- a/src/shared/delegated-mobile-device-contract.ts +++ b/src/shared/delegated-mobile-device-contract.ts @@ -7,11 +7,12 @@ export const DELEGATED_MOBILE_DEVICE_SYNC_METHOD = 'pairing.delegatedMobileDevic export const DELEGATED_MOBILE_DEVICES_RUNTIME_CAPABILITY = 'pairing.delegated-mobile-devices.v1' satisfies (typeof RUNTIME_CAPABILITIES)[number] export const DELEGATED_MOBILE_DEVICE_SYNC_MAX_PHONES = 32 +export const DELEGATED_PHONE_NAME_MAX_CHARS = 128 const DelegatedPhoneSchema = z .object({ phoneKey: z.string().min(1).max(128), - name: z.string().min(1).max(128) + name: z.string().min(1).max(DELEGATED_PHONE_NAME_MAX_CHARS) }) .strict() diff --git a/src/shared/mobile-desktop-relay-contract.ts b/src/shared/mobile-desktop-relay-contract.ts new file mode 100644 index 00000000000..1ae407dce6d --- /dev/null +++ b/src/shared/mobile-desktop-relay-contract.ts @@ -0,0 +1,4 @@ +// 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 diff --git a/src/shared/remote-runtime-passthrough-socket.ts b/src/shared/remote-runtime-passthrough-socket.ts new file mode 100644 index 00000000000..50920fec667 --- /dev/null +++ b/src/shared/remote-runtime-passthrough-socket.ts @@ -0,0 +1,175 @@ +import WebSocket from 'ws' +import type { PairingOffer } from './pairing' +import type { RuntimeCapability } from './protocol-version' +import { decrypt, decryptBytes, encrypt } from './e2ee-crypto' +import { + classifyRemoteRuntimeReadyFrame, + formatRemoteRuntimeCloseMessage, + parseRemoteRuntimeAuthenticatedFrame +} from './remote-runtime-client-handshake' +import { RemoteRuntimeClientError } from './remote-runtime-client-error' +import { serializeRemoteRuntimePayload } from './remote-runtime-memory-limits' +import { openRemoteRuntimeWebSocket } from './remote-runtime-request-websocket' +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) => { + let state: 'awaiting_ready' | 'awaiting_authenticated' | 'ready' | 'ended' = 'awaiting_ready' + const opened = openRemoteRuntimeWebSocket(pairing, { + onTextFrame: (_ws, frame) => onTextFrame(frame), + onBinaryFrame: (_ws, frame) => { + liveness.noteActivity() + const bytes = state === 'ready' ? decryptBytes(frame, sharedKey) : null + if (!bytes) { + end(protocolError('Remote Orca runtime returned an unreadable binary frame.')) + return + } + callbacks.onBinary(bytes) + }, + onError: (_ws, error) => end(error), + onClose: (_ws, code, reason) => + end( + new RemoteRuntimeClientError( + 'remote_runtime_unavailable', + formatRemoteRuntimeCloseMessage(code, reason) + ) + ), + onPong: () => liveness.noteActivity(), + onPing: () => liveness.noteActivity() + }) + if (!opened.ok) { + reject(opened.error) + return + } + const { ws, sharedKey, cleanup } = opened.socket + const outbound = new RemoteRuntimeSubscriptionOutbound({ fail: (error) => end(error) }) + 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) + cleanup() + closeRemoteRuntimeSocket(ws) + if (!wasReady) { + reject(error ?? new RemoteRuntimeClientError('remote_runtime_unavailable', 'Closed.')) + } else if (error) { + callbacks.onClose(error) + } + } + + function protocolError(message: string): RemoteRuntimeClientError { + return new RemoteRuntimeClientError('invalid_runtime_response', message) + } + + function onTextFrame(frame: string): void { + liveness.noteActivity() + 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({ + send: (plaintext) => + state === 'ready' && ws.readyState === WebSocket.OPEN + ? outbound.enqueueRequest(ws, encrypt(plaintext, sharedKey)) + : false, + close: () => end(null) + }) + } + }) +} diff --git a/src/shared/remote-runtime-request-websocket.test.ts b/src/shared/remote-runtime-request-websocket.test.ts index 0458febd698..d025671f32f 100644 --- a/src/shared/remote-runtime-request-websocket.test.ts +++ b/src/shared/remote-runtime-request-websocket.test.ts @@ -2,6 +2,8 @@ import type { EventEmitter } from 'node:events' import { describe, expect, it, vi } from 'vitest' import { generateKeyPair, publicKeyToBase64 } from './e2ee-crypto' import { openRemoteRuntimeWebSocket } from './remote-runtime-request-websocket' +import { REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES } from './remote-runtime-memory-limits' +import { WebSocketServer } from 'ws' describe('openRemoteRuntimeWebSocket', () => { it('detaches Orca callback listeners when cleaned up', () => { @@ -39,4 +41,42 @@ describe('openRemoteRuntimeWebSocket', () => { expect(() => socketEvents.emit('error', new Error('late socket error'))).not.toThrow() opened.socket.ws.terminate() }) + + it('closes on a frame larger than any runtime sends', async () => { + const server = new WebSocketServer({ host: '127.0.0.1', port: 0 }) + server.on('connection', (ws) => + ws.send(Buffer.alloc(REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES + 1)) + ) + await new Promise((resolve) => server.once('listening', resolve)) + const address = server.address() + const port = typeof address === 'object' && address ? address.port : 0 + const onError = vi.fn() + const onBinaryFrame = vi.fn() + const closed = new Promise((resolve) => { + const opened = openRemoteRuntimeWebSocket( + { + v: 2, + endpoint: `ws://127.0.0.1:${port}`, + deviceToken: 'device-token', + publicKeyB64: publicKeyToBase64(generateKeyPair().publicKey) + }, + { + onClose: () => resolve(), + onError, + onTextFrame: vi.fn(), + onBinaryFrame + } + ) + if (!opened.ok) { + throw opened.error + } + }) + try { + await closed + expect(onError).toHaveBeenCalledOnce() + expect(onBinaryFrame).not.toHaveBeenCalled() + } finally { + await new Promise((resolve) => server.close(resolve)) + } + }) }) diff --git a/src/shared/remote-runtime-request-websocket.ts b/src/shared/remote-runtime-request-websocket.ts index ca2ddfec60d..03749478faf 100644 --- a/src/shared/remote-runtime-request-websocket.ts +++ b/src/shared/remote-runtime-request-websocket.ts @@ -11,6 +11,7 @@ import { remoteRuntimeConnectFailureMessage, remoteRuntimeConnectOptions } from './remote-runtime-connect-bound' +import { REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES } from './remote-runtime-memory-limits' import { invalidRemoteRuntimeResponseError, remoteRuntimeUnavailableError @@ -26,6 +27,8 @@ export type RemoteRuntimeWebSocketCallbacks = { onClose: (ws: WebSocket, code: number, reason: Buffer) => void onError: (ws: WebSocket, error: RemoteRuntimeClientError) => void onTextFrame: (ws: WebSocket, frame: string) => void + // Why optional: request sockets treat a binary frame as a protocol error; a relay forwards it. + onBinaryFrame?: (ws: WebSocket, frame: Uint8Array) => void // Why: protocol-level pongs (and server heartbeat pings) are the liveness // signal for detecting half-open tunnels that never deliver `close` (#7718). onPong?: (ws: WebSocket) => void @@ -64,6 +67,10 @@ export function openRemoteRuntimeWebSocket( } const onClose = (code: number, reason: Buffer): void => callbacks.onClose(ws, code, reason) const onMessage = (data: WebSocket.RawData, isBinary: boolean): void => { + if (isBinary && callbacks.onBinaryFrame && Buffer.isBuffer(data)) { + callbacks.onBinaryFrame(ws, new Uint8Array(data)) + return + } if (isBinary) { callbacks.onError( ws, @@ -129,7 +136,14 @@ function createSocket( try { return { ok: true, - ws: new WebSocket(pairing.endpoint, remoteRuntimeConnectOptions(undefined, connectTimeoutMs)), + // Why: no runtime frame exceeds this, so a larger one is a broken peer, not a payload to buffer. + ws: new WebSocket( + pairing.endpoint, + remoteRuntimeConnectOptions( + { maxPayload: REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES }, + connectTimeoutMs + ) + ), keyPair } } catch (error) { diff --git a/src/shared/terminal-stream-protocol.ts b/src/shared/terminal-stream-protocol.ts index 52b6c554784..0b74d7433bd 100644 --- a/src/shared/terminal-stream-protocol.ts +++ b/src/shared/terminal-stream-protocol.ts @@ -104,6 +104,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)) }