mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 08:03:13 +00:00
fix(relay): bound stuck desktop setup
This commit is contained in:
@@ -117,6 +117,72 @@ describe('RelayControlClient', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('rejects a control handshake that never receives a proof response', async () => {
|
||||
const server = new WebSocketServer({ port: 0, perMessageDeflate: false })
|
||||
servers.push(server)
|
||||
await new Promise<void>((resolve) => server.once('listening', resolve))
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string') {
|
||||
throw new Error('expected TCP relay test server')
|
||||
}
|
||||
const keypair = nacl.box.keyPair()
|
||||
const client = new RelayControlClient({
|
||||
cellUrl: `http://127.0.0.1:${address.port}`,
|
||||
relayJwt: 'scoped-token',
|
||||
relayHostId: createHash('sha256').update(keypair.publicKey).digest('base64url').slice(0, 16),
|
||||
assignmentEpoch: 1,
|
||||
identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' },
|
||||
keypair: {
|
||||
...keypair,
|
||||
publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
|
||||
},
|
||||
appVersion: '1.2.3',
|
||||
onConnectionOpen: vi.fn(),
|
||||
onDrain: vi.fn(),
|
||||
onClose: vi.fn(),
|
||||
connectDeadlineMs: 20
|
||||
})
|
||||
clients.push(client)
|
||||
|
||||
await expect(client.connect()).rejects.toThrow('relay_control_connect_timeout')
|
||||
})
|
||||
|
||||
it('settles an opening control immediately when ownership closes', async () => {
|
||||
const server = new WebSocketServer({ port: 0, perMessageDeflate: false })
|
||||
servers.push(server)
|
||||
await new Promise<void>((resolve) => server.once('listening', resolve))
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string') {
|
||||
throw new Error('expected TCP relay test server')
|
||||
}
|
||||
const keypair = nacl.box.keyPair()
|
||||
const client = new RelayControlClient({
|
||||
cellUrl: `http://127.0.0.1:${address.port}`,
|
||||
relayJwt: 'scoped-token',
|
||||
relayHostId: createHash('sha256').update(keypair.publicKey).digest('base64url').slice(0, 16),
|
||||
assignmentEpoch: 1,
|
||||
identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' },
|
||||
keypair: {
|
||||
...keypair,
|
||||
publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
|
||||
},
|
||||
appVersion: '1.2.3',
|
||||
onConnectionOpen: vi.fn(),
|
||||
onDrain: vi.fn(),
|
||||
onClose: vi.fn()
|
||||
})
|
||||
clients.push(client)
|
||||
const accepted = new Promise<void>((resolve) => {
|
||||
server.once('connection', () => resolve())
|
||||
})
|
||||
const connecting = client.connect()
|
||||
|
||||
await accepted
|
||||
client.closeNow()
|
||||
|
||||
await expect(connecting).rejects.toThrow('relay_control_closed')
|
||||
})
|
||||
|
||||
it('proves the host key and drives control/data commands without URL credentials', async () => {
|
||||
const server = new WebSocketServer({ port: 0, perMessageDeflate: false })
|
||||
servers.push(server)
|
||||
|
||||
@@ -36,8 +36,11 @@ type RelayControlClientOptions = {
|
||||
onDrain: (message: RelayDrainMessage) => void
|
||||
onClose: (code: number) => void
|
||||
createSocket?: (url: string, relayJwt: string) => WebSocket
|
||||
connectDeadlineMs?: number
|
||||
}
|
||||
|
||||
const RELAY_CONTROL_CONNECT_DEADLINE_MS = 15_000
|
||||
|
||||
function controlWebSocketUrl(cellUrl: string): { origin: string; url: string } {
|
||||
const parsed = new URL(cellUrl)
|
||||
if (parsed.pathname !== '/' || parsed.search || parsed.hash) {
|
||||
@@ -64,6 +67,7 @@ export class RelayControlClient {
|
||||
private state: RelayControlState = 'idle'
|
||||
private connectResolve: ((ack: RelayHostHelloAckMessage) => void) | null = null
|
||||
private connectReject: ((error: Error) => void) | null = null
|
||||
private connectTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
constructor(options: RelayControlClientOptions) {
|
||||
this.options = options
|
||||
@@ -102,6 +106,12 @@ export class RelayControlClient {
|
||||
}
|
||||
})
|
||||
socket.once('close', (code) => this.handleClose(code))
|
||||
// Recovery cannot advance while an upgrade/proof promise remains pending forever.
|
||||
this.connectTimer = setTimeout(
|
||||
() => this.expireConnect(),
|
||||
this.options.connectDeadlineMs ?? RELAY_CONTROL_CONNECT_DEADLINE_MS
|
||||
)
|
||||
this.connectTimer.unref()
|
||||
return new Promise((resolve, reject) => {
|
||||
this.connectResolve = resolve
|
||||
this.connectReject = reject
|
||||
@@ -155,7 +165,12 @@ export class RelayControlClient {
|
||||
}
|
||||
|
||||
closeNow(): void {
|
||||
const wasConnecting = this.state === 'opening' || this.state === 'proving'
|
||||
this.state = 'closed'
|
||||
if (wasConnecting) {
|
||||
this.connectReject?.(new Error('relay_control_closed'))
|
||||
this.clearConnectPromise()
|
||||
}
|
||||
this.requests.rejectAll(new Error('relay_control_closed'))
|
||||
this.socket?.terminate()
|
||||
this.socket = null
|
||||
@@ -279,7 +294,20 @@ export class RelayControlClient {
|
||||
this.options.onClose(code)
|
||||
}
|
||||
|
||||
private expireConnect(): void {
|
||||
if (this.state !== 'opening' && this.state !== 'proving') {
|
||||
return
|
||||
}
|
||||
this.connectReject?.(new Error('relay_control_connect_timeout'))
|
||||
this.clearConnectPromise()
|
||||
this.socket?.terminate()
|
||||
}
|
||||
|
||||
private clearConnectPromise(): void {
|
||||
if (this.connectTimer) {
|
||||
clearTimeout(this.connectTimer)
|
||||
this.connectTimer = null
|
||||
}
|
||||
this.connectResolve = null
|
||||
this.connectReject = null
|
||||
}
|
||||
|
||||
@@ -57,6 +57,49 @@ describe('relay HTTP client', () => {
|
||||
expect(fetch.mock.calls[0]?.[1]?.headers).toMatchObject({
|
||||
authorization: 'Bearer scoped-token'
|
||||
})
|
||||
expect(fetch.mock.calls[0]?.[1]?.signal).toBeInstanceOf(AbortSignal)
|
||||
})
|
||||
|
||||
it('aborts a blackholed assignment request so recovery can retry', async () => {
|
||||
const fetch = vi.fn<typeof globalThis.fetch>(
|
||||
async (_url, init) =>
|
||||
await new Promise<Response>((_resolve, reject) => {
|
||||
init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true })
|
||||
})
|
||||
)
|
||||
|
||||
await expect(
|
||||
requestRelayAssignment({
|
||||
directorUrl: 'https://relay.example',
|
||||
relayToken: 'scoped-token',
|
||||
relayHostId: 'AbCdEf0123_-xyZ9',
|
||||
requestDeadlineMs: 5,
|
||||
fetch
|
||||
})
|
||||
).rejects.toMatchObject({ name: 'TimeoutError' })
|
||||
})
|
||||
|
||||
it('aborts a blackholed token exchange so recovery can retry', async () => {
|
||||
const keypair = nacl.box.keyPair()
|
||||
const fetch = vi.fn<typeof globalThis.fetch>(
|
||||
async (_url, init) =>
|
||||
await new Promise<Response>((_resolve, reject) => {
|
||||
init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true })
|
||||
})
|
||||
)
|
||||
|
||||
await expect(
|
||||
exchangeRelayAuthorization({
|
||||
endpoint: 'https://auth.example/v1/desktop/auth/relay-token',
|
||||
accessToken: 'ordinary-access-token',
|
||||
keypair: {
|
||||
...keypair,
|
||||
publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
|
||||
},
|
||||
requestDeadlineMs: 5,
|
||||
fetch
|
||||
})
|
||||
).rejects.toMatchObject({ name: 'TimeoutError' })
|
||||
})
|
||||
|
||||
it('rejects data-plane supplied non-origin URLs', async () => {
|
||||
|
||||
@@ -4,6 +4,8 @@ import type { E2EEKeypair } from '../e2ee-keypair'
|
||||
import { cancelUnreadResponseBody } from '../../lib/unread-response-body'
|
||||
import { readFetchResponseJsonWithinLimit } from '../../lib/fetch-response-body'
|
||||
|
||||
const RELAY_HTTP_REQUEST_DEADLINE_MS = 15_000
|
||||
|
||||
const RelayTokenResponseSchema = z
|
||||
.object({
|
||||
relayToken: z
|
||||
@@ -72,6 +74,7 @@ export async function exchangeRelayAuthorization(input: {
|
||||
accessToken: string
|
||||
keypair: E2EEKeypair
|
||||
fetch?: typeof globalThis.fetch
|
||||
requestDeadlineMs?: number
|
||||
}): Promise<RelayAuthorization> {
|
||||
const relayHostId = deriveRelayHostId(input.keypair.publicKey)
|
||||
const response = await (input.fetch ?? globalThis.fetch)(input.endpoint, {
|
||||
@@ -80,6 +83,8 @@ export async function exchangeRelayAuthorization(input: {
|
||||
authorization: `Bearer ${input.accessToken}`,
|
||||
'content-type': 'application/json'
|
||||
},
|
||||
// A blackholed request must settle so the coordinator can advance its bounded retry state.
|
||||
signal: AbortSignal.timeout(input.requestDeadlineMs ?? RELAY_HTTP_REQUEST_DEADLINE_MS),
|
||||
body: JSON.stringify({ relayHostId, hostPublicKeyB64: input.keypair.publicKeyB64 })
|
||||
})
|
||||
if (!response.ok) {
|
||||
@@ -100,6 +105,7 @@ export async function requestRelayAssignment(input: {
|
||||
relayToken: string
|
||||
relayHostId: string
|
||||
fetch?: typeof globalThis.fetch
|
||||
requestDeadlineMs?: number
|
||||
}): Promise<RelayAssignment> {
|
||||
if (!isAllowedRelayOrigin(input.directorUrl)) {
|
||||
throw new RelayHttpError('assignment', 400)
|
||||
@@ -110,6 +116,7 @@ export async function requestRelayAssignment(input: {
|
||||
authorization: `Bearer ${input.relayToken}`,
|
||||
'content-type': 'application/json'
|
||||
},
|
||||
signal: AbortSignal.timeout(input.requestDeadlineMs ?? RELAY_HTTP_REQUEST_DEADLINE_MS),
|
||||
body: JSON.stringify({ v: 1, relayHostId: input.relayHostId })
|
||||
})
|
||||
if (!response.ok) {
|
||||
|
||||
Reference in New Issue
Block a user