mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(mobile): relay a paired phone's server-workspace requests through the desktop (hidden) (#26435)
* feat(runtime): let a paired desktop register its phones as delegated mobile devices A desktop paired to this runtime can now call pairing.delegatedMobileDevice.sync to make each of its phones an ordinary mobile-scope device it owns, so the phone mirrors that host through the desktop. Children are never pending, inherit the parent's reach, and die with the parent's grant. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * refactor(runtime): move the mobile device revoke primitive below dispatch Dispatch binds the caller-bound delegated-device context, so the revoke path it needs now lives in a layer under it as a pure move; this drops the abstract method seam. No behaviour change. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(runtime): keep a runtime grant until its delegated phones are revoked A failed child cleanup now leaves the parent in place so a retry can still find the children, and revokeRuntimeAccess lives beside the revoke it cascades to. Adds restart, rename and bounds coverage, shares credential minting, and drops the timing-dependent census. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(runtime): fail a delegated sync whose dropped phone cannot be revoked Matches the parent revoke: a child whose cleanup could not be saved is kept and the sync errors before minting anything, so the desktop retries instead of believing the server matches its list. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * feat(mobile): relay a paired phone's server-workspace requests through the desktop A phone mirrors its desktop, including workspaces on the desktop's configured servers. The desktop now forwards a request that names `executionHost: runtime:<env>` to that server over one socket per (phone, server), signed in as the phone's delegated device; replies pass through untouched except stream ids. Hidden until the phone client ships: the relay capability is not advertised yet. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * docs(mobile): state the desktop relay contract in its entry module main dropped docs/reference, so the relay's contract lives beside the code. Also pins that a server whose delegated sync fails (e.g. delegated_device_revoke_failed) is unavailable until a later request syncs it again. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * test(mobile): pin the relay's configured-server adapter The production hosts adapter was only typechecked; this pins that it resolves configured servers by exact id only (never a name or a removed server), fences by pairing revision and runtime identity, and stops reporting retirements once unsubscribed. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * refactor(mobile): tighten the desktop relay after elegance review Reuses the shared E2EE socket opener, folds relay state into fewer fields, never sends a request whose link ended while opening, passes failure codes through, relays only this desktop's own phones, and drops the unused host-state reader until S2 needs one. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(mobile): bound relay frames and stop sticky relay verdicts Restores the 8 MiB frame cap on every remote runtime socket, re-checks an update-needed server on each new link so an in-place upgrade relays, answers a phone the server just declined without syncing again, and ends a link whose mirrored capability update cannot be queued. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(mobile): run targeted desktop-owned calls on the desktop The phone tags every call on a workspace screen with its execution host, so a paired-desktop method carrying `runtime:<env>` now runs locally instead of failing 'forbidden'. The Mac stays the only reader of the routing table. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(ci): run the remote-runtime WebSocket opener test on Node Under the bun vitest project, `ws` resolves to Bun's built-in client, which ignores maxPayload, so the frame-cap test hung. The opener only ships on Node/Electron, so run its test in the Node runtime project like its siblings. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * refactor(mobile-relay): derive the phone allowlist from the routing table Adding a phone method now takes one edit (its route). A server-targeted call on a legacy non-E2EE socket is refused instead of running locally, and only execution-host methods enter the relay. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
@@ -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',
|
||||
|
||||
@@ -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'])
|
||||
})
|
||||
})
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, RetainedRemoteRuntimeSubscription>()
|
||||
@@ -73,6 +74,7 @@ export function invalidateRuntimeEnvironmentTransport(environmentId: string): Pr
|
||||
advanceRuntimeEnvironmentTransportGeneration(environmentId)
|
||||
closeRemoteRuntimeRequestConnection(environmentId)
|
||||
closeSubscriptionsForEnvironment(environmentId)
|
||||
retireMobileDesktopRelayEnvironment(environmentId)
|
||||
reactivateRuntimeEnvironmentStatus(environmentId)
|
||||
return retirePairedRuntimeBrowserClientHostEnvironment(
|
||||
environmentId,
|
||||
|
||||
@@ -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<string, DelegatedPhoneGrant>
|
||||
// Every phone this sync asked for; one asked for but not granted waits for the next sync.
|
||||
asked: ReadonlySet<string>
|
||||
// Phones whose host token was refused (revoked on the host); not retried until the next sync.
|
||||
refused: Set<string>
|
||||
}
|
||||
|
||||
export async function syncDelegatedPhoneGrants(
|
||||
hosts: MobileDesktopRelayHosts,
|
||||
host: MobileDesktopRelayHost,
|
||||
phones: DelegatedMobileDeviceSyncParams['phones']
|
||||
): Promise<DelegatedPhoneGrants> {
|
||||
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()
|
||||
}
|
||||
}
|
||||
@@ -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<MobileDesktopRelayHost | null>
|
||||
/** A call as the desktop itself, over its own connection to the server. */
|
||||
call: (
|
||||
host: MobileDesktopRelayHost,
|
||||
method: string,
|
||||
params: unknown
|
||||
) => Promise<RuntimeRpcResponse<unknown>>
|
||||
/** Fires when a server is removed, re-paired or disconnected; returns an unsubscribe. */
|
||||
onEnvironmentRetired: (listener: (environmentId: string) => void) => () => void
|
||||
}
|
||||
@@ -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<typeof PassthroughSocketModule>()
|
||||
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<string, unknown>
|
||||
|
||||
describe('mobile desktop relay: phone -> desktop -> server', () => {
|
||||
const cleanups: (() => Promise<void> | 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<void>((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<ReturnType<typeof connectPhone>>
|
||||
|
||||
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)
|
||||
})
|
||||
})
|
||||
@@ -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<unknown> | Promise<RuntimeRpcResponse<unknown>> = () =>
|
||||
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' }
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -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<ArrayBufferLike>) => boolean | void
|
||||
}
|
||||
|
||||
type GrantsEntry = {
|
||||
promise: Promise<DelegatedPhoneGrants>
|
||||
settled: boolean
|
||||
result?: DelegatedPhoneGrants
|
||||
}
|
||||
|
||||
type OpenedRelayLink = { socket: RemoteRuntimePassthroughSocket; hostToken: string }
|
||||
|
||||
class RelayLink {
|
||||
readonly replies: RelayedPhoneReplies
|
||||
readonly opened: Promise<OpenedRelayLink>
|
||||
|
||||
constructor(
|
||||
readonly environmentId: string,
|
||||
readonly phone: RelayedPhone,
|
||||
allocateStreamId: () => number,
|
||||
open: (link: RelayLink) => Promise<OpenedRelayLink>
|
||||
) {
|
||||
this.replies = new RelayedPhoneReplies(allocateStreamId)
|
||||
this.opened = open(this)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Relays a phone request carrying `executionHost: runtime:<env>` 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<string, RelayLink>()
|
||||
private readonly grants = new Map<string, GrantsEntry>()
|
||||
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<void> {
|
||||
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<OpenedRelayLink> {
|
||||
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<DelegatedPhoneGrants> {
|
||||
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<DelegatedPhoneGrants> {
|
||||
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 } })
|
||||
}
|
||||
@@ -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'])
|
||||
})
|
||||
})
|
||||
@@ -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<string, { hostStreamIds: number[]; swallow: boolean }>()
|
||||
private readonly desktopStreamIds = new Map<number, number>()
|
||||
|
||||
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<ArrayBufferLike>): Uint8Array<ArrayBufferLike> | 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<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
@@ -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([])
|
||||
})
|
||||
})
|
||||
@@ -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<string, RelayedStreamCarrier> = 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']
|
||||
])
|
||||
@@ -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:<env>`), validated where it is read.
|
||||
executionHost?: unknown
|
||||
}
|
||||
|
||||
export type LegacyCoordinatorAuthorityProof = Readonly<{
|
||||
|
||||
@@ -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<string, unknown>[] = []
|
||||
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<string, unknown> = 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'
|
||||
}
|
||||
}
|
||||
])
|
||||
})
|
||||
})
|
||||
@@ -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)
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -10,7 +10,11 @@ export class RuntimeRpcMobileDevices extends RuntimeRpcRequestAdmission {
|
||||
private onPushUnregisterQueued?: () => void
|
||||
|
||||
async revokeMobileDevice(deviceId: string): Promise<boolean> {
|
||||
return this.revokeMobileDeviceNow(deviceId)
|
||||
const revoked = this.revokeMobileDeviceNow(deviceId)
|
||||
if (revoked) {
|
||||
this.mobileDesktopRelay?.phonesChanged()
|
||||
}
|
||||
return revoked
|
||||
}
|
||||
|
||||
private revokeMobileDeviceNow(deviceId: string): boolean {
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -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<string, MobileRpcMethodRoute> = new Map([
|
||||
...PAIRED_DESKTOP_METHODS.map((method) => [method, 'paired-desktop'] as const),
|
||||
...EXECUTION_HOST_METHODS.map((method) => [method, 'execution-host'] as const)
|
||||
])
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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<unknown>) | null = null
|
||||
protected mobileRelayPairingOfferQueue: Promise<void> = Promise.resolve()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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<ArrayBufferLike>) => 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<RemoteRuntimePassthroughSocket> {
|
||||
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)
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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<void>((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))
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<ArrayBufferLike>) => 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) {
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user