mirror of
https://github.com/stablyai/orca.git
synced 2026-09-28 00:02:41 +00:00
fix(relay): bound the control socket's connect phase at the transport
The relay control socket was constructed with no `handshakeTimeout`, the same gap fixed for the remote-runtime transports. It was not a live defect: the class-level `connectDeadlineMs` (15s) also covers a stalled connect, and that deadline does fire — its `unref()` is safe because the pending TCP connect is itself a ref'd libuv handle that holds the event loop open. Measured in a bare Node process: unref'd timer with an empty loop never fires (exit at 0ms), but the same timer alongside a black-holed connect fired at 2003ms. It was a defect waiting on a refactor. The two bounds cover different phases, and the class deadline covers the connect phase only incidentally. DO NOT REMOVE EITHER BOUND AS REDUNDANT. They are not. Proven by mutation: - Remove the transport bound -> a stalled *connect* falls through to the class deadline, rejecting with `relay_control_connect_timeout` after the full deadline instead of the transport error. - Remove the class deadline -> a stall during the *proving* phase (socket open, host proof never answered) is unbounded; the incumbent test hangs 30s. `handshakeTimeout` cannot see that phase at all. Reuses `remoteRuntimeConnectOptions` rather than forking a second helper, and moves the construction into `relay-control-socket-factory.ts` so a caller that needs a relay control socket gets the bound instead of re-deriving an unbounded one. `handshakeTimeoutMs` is settable apart from `connectDeadlineMs` so a test can stall the connect alone and assert which bound produced the rejection — error identity, not elapsed time. The connect-bound ratchet now covers the relay site and asserts the site still resolves, so an allowlist that silently stopped matching cannot pass vacuously.
This commit is contained in:
@@ -9,7 +9,6 @@ import {
|
||||
RelayHostChallengeMessageSchema,
|
||||
RelayHostHelloAckMessageSchema,
|
||||
RelayPingMessageSchema,
|
||||
RELAY_HOST_CAPABILITY_HEADERS,
|
||||
encodeRelayHostHello,
|
||||
parseRelayControlMessage,
|
||||
type RelayConnectionOpenMessage,
|
||||
@@ -25,6 +24,7 @@ import {
|
||||
RelayControlSilenceWatchdog
|
||||
} from './relay-control-silence-watchdog'
|
||||
import { closeRelayControlSocket } from './relay-control-socket-close'
|
||||
import { createRelayControlSocket } from './relay-control-socket-factory'
|
||||
import { controlWebSocketUrl } from './relay-control-url'
|
||||
|
||||
type RelayControlState = 'idle' | 'opening' | 'proving' | 'active' | 'draining' | 'closed'
|
||||
@@ -44,6 +44,9 @@ type RelayControlClientOptions = {
|
||||
onClose: (code: number) => void
|
||||
createSocket?: (url: string, relayJwt: string) => WebSocket
|
||||
connectDeadlineMs?: number
|
||||
// Why: settable apart from connectDeadlineMs so a test can stall the connect
|
||||
// phase alone and prove which of the two bounds fired.
|
||||
handshakeTimeoutMs?: number
|
||||
silenceLimitMs?: number
|
||||
}
|
||||
|
||||
@@ -74,11 +77,13 @@ export class RelayControlClient {
|
||||
this.createSocket =
|
||||
options.createSocket ??
|
||||
((url, token) =>
|
||||
new WebSocket(url, {
|
||||
headers: { authorization: `Bearer ${token}`, ...RELAY_HOST_CAPABILITY_HEADERS },
|
||||
perMessageDeflate: false,
|
||||
maxPayload: 64 * 1024
|
||||
}))
|
||||
createRelayControlSocket(
|
||||
url,
|
||||
token,
|
||||
options.handshakeTimeoutMs ??
|
||||
options.connectDeadlineMs ??
|
||||
RELAY_CONTROL_CONNECT_DEADLINE_MS
|
||||
))
|
||||
}
|
||||
|
||||
connect(): Promise<RelayHostHelloAckMessage> {
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { readFileSync } from 'node:fs'
|
||||
import { createServer, type Server, type Socket } from 'node:net'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import nacl from 'tweetnacl'
|
||||
import { isRemoteRuntimeConnectTimeout } from '../../../shared/remote-runtime-connect-bound'
|
||||
import { RelayControlClient } from './relay-control-client'
|
||||
|
||||
const servers = new Set<Server>()
|
||||
const sockets = new Set<Socket>()
|
||||
|
||||
afterEach(async () => {
|
||||
for (const socket of sockets) {
|
||||
socket.destroy()
|
||||
}
|
||||
sockets.clear()
|
||||
await Promise.all(
|
||||
[...servers].map(
|
||||
(server) =>
|
||||
new Promise<void>((resolve) => {
|
||||
server.close(() => resolve())
|
||||
})
|
||||
)
|
||||
)
|
||||
servers.clear()
|
||||
})
|
||||
|
||||
/**
|
||||
* Accepts TCP and never answers the HTTP upgrade, so the socket never opens.
|
||||
* The incumbent relay test stalls the *proving* phase instead, which the
|
||||
* transport bound cannot see — only this shape distinguishes the two bounds.
|
||||
*/
|
||||
async function listenSilentUpgradeServer(): Promise<string> {
|
||||
const server = createServer((socket) => {
|
||||
sockets.add(socket)
|
||||
socket.once('close', () => sockets.delete(socket))
|
||||
})
|
||||
servers.add(server)
|
||||
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') {
|
||||
throw new Error('expected a TCP address')
|
||||
}
|
||||
return `http://127.0.0.1:${address.port}`
|
||||
}
|
||||
|
||||
function buildClient(cellUrl: string, overrides: { handshakeTimeoutMs?: number }) {
|
||||
const keypair = nacl.box.keyPair()
|
||||
return new RelayControlClient({
|
||||
cellUrl,
|
||||
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(),
|
||||
// Why: 50x the transport bound, so whichever error arrives names the bound
|
||||
// that produced it rather than the one that merely exists.
|
||||
connectDeadlineMs: 5_000,
|
||||
...overrides
|
||||
})
|
||||
}
|
||||
|
||||
describe('relay control connect transport bound', () => {
|
||||
it('bounds a connect that never opens, and names the transport bound', async () => {
|
||||
const cellUrl = await listenSilentUpgradeServer()
|
||||
const client = buildClient(cellUrl, { handshakeTimeoutMs: 100 })
|
||||
|
||||
const error = await client.connect().then(
|
||||
() => null,
|
||||
(reason: unknown) => reason as Error
|
||||
)
|
||||
|
||||
expect(error).toBeInstanceOf(Error)
|
||||
// The transport bound fired, not the class deadline that also covers this.
|
||||
expect(isRemoteRuntimeConnectTimeout(error)).toBe(true)
|
||||
expect(error?.message).not.toContain('relay_control_connect_timeout')
|
||||
})
|
||||
|
||||
// Why: defence in depth only works if both bounds survive. Either one removed
|
||||
// as "redundant" leaves a phase uncovered.
|
||||
it('keeps both bounds, since each covers a phase the other cannot', () => {
|
||||
const client = readFileSync(join(__dirname, 'relay-control-client.ts'), 'utf8')
|
||||
const factory = readFileSync(join(__dirname, 'relay-control-socket-factory.ts'), 'utf8')
|
||||
expect(factory).toContain('remoteRuntimeConnectOptions(')
|
||||
expect(client).toContain('this.connectTimer = setTimeout(')
|
||||
expect(client).toContain("new Error('relay_control_connect_timeout')")
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,35 @@
|
||||
import WebSocket from 'ws'
|
||||
import { remoteRuntimeConnectOptions } from '../../../shared/remote-runtime-connect-bound'
|
||||
import { RELAY_HOST_CAPABILITY_HEADERS } from './relay-control-protocol'
|
||||
|
||||
/**
|
||||
* Builds the relay control socket with a transport-level connect bound.
|
||||
*
|
||||
* Why this exists separately from `connectDeadlineMs` in `RelayControlClient`:
|
||||
* that deadline bounds the whole handshake including the host proof, but it can
|
||||
* only run once the socket object exists, and it is the class's to arm. A
|
||||
* black-holed relay never opens and never errors, so the connect sub-phase
|
||||
* needs its own bound at the transport — the same one the remote-runtime
|
||||
* transports use. Both are kept deliberately: they cover different phases and
|
||||
* neither is redundant. See #18191.
|
||||
*
|
||||
* Keeping the construction here means a caller that reaches for the relay
|
||||
* control socket gets the bound, rather than re-deriving an unbounded one.
|
||||
*/
|
||||
export function createRelayControlSocket(
|
||||
url: string,
|
||||
relayJwt: string,
|
||||
connectBoundMs: number
|
||||
): WebSocket {
|
||||
return new WebSocket(
|
||||
url,
|
||||
remoteRuntimeConnectOptions(
|
||||
{
|
||||
headers: { authorization: `Bearer ${relayJwt}`, ...RELAY_HOST_CAPABILITY_HEADERS },
|
||||
perMessageDeflate: false,
|
||||
maxPayload: 64 * 1024
|
||||
},
|
||||
connectBoundMs
|
||||
)
|
||||
)
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import { readFileSync, readdirSync } from 'node:fs'
|
||||
import { existsSync, readFileSync, readdirSync } from 'node:fs'
|
||||
import { createServer, type Server, type Socket } from 'node:net'
|
||||
import { join } from 'node:path'
|
||||
import { basename, join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { generateKeyPair, publicKeyToBase64 } from './e2ee-crypto'
|
||||
import type { RemoteRuntimeClientError } from './remote-runtime-client-error'
|
||||
@@ -15,6 +15,36 @@ import { openRemoteRuntimeWebSocket } from './remote-runtime-request-websocket'
|
||||
const servers = new Set<Server>()
|
||||
const sockets = new Set<Socket>()
|
||||
|
||||
const RELAY_CONTROL_SOCKET_FACTORY = join('relay', 'relay-control-socket-factory.ts')
|
||||
|
||||
/**
|
||||
* Files whose WebSocket construction must carry the connect bound. The shared
|
||||
* remote-runtime transports are swept by prefix; sites outside this directory
|
||||
* are listed explicitly so adding one is a deliberate act rather than a glob
|
||||
* accident. Other WebSocket sites (relay data transport, emulator control)
|
||||
* carry their own bounds and are deliberately not covered here.
|
||||
*/
|
||||
function coveredSocketSources(): string[] {
|
||||
const shared = readdirSync(__dirname)
|
||||
.filter(
|
||||
(name) =>
|
||||
name.startsWith('remote-runtime-') && name.endsWith('.ts') && !name.includes('.test.')
|
||||
)
|
||||
.map((name) => join(__dirname, name))
|
||||
const relaySocketFactory = join(
|
||||
__dirname,
|
||||
'..',
|
||||
'main',
|
||||
'runtime',
|
||||
'relay',
|
||||
'relay-control-socket-factory.ts'
|
||||
)
|
||||
if (!existsSync(relaySocketFactory)) {
|
||||
throw new Error(`connect-bound ratchet lost its relay site: ${relaySocketFactory}`)
|
||||
}
|
||||
return [...shared, relaySocketFactory]
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
for (const socket of sockets) {
|
||||
socket.destroy()
|
||||
@@ -59,25 +89,26 @@ describe('remote runtime connect bound', () => {
|
||||
|
||||
// Why: the bound only helps if every Node-side remote-runtime socket carries
|
||||
// it; a new transport that calls `new WebSocket` directly reintroduces #18191.
|
||||
it('routes every shared remote-runtime WebSocket through the bounded options', () => {
|
||||
const dir = join(__dirname)
|
||||
it('routes every covered WebSocket construction through the bounded options', () => {
|
||||
const offenders: string[] = []
|
||||
let scannedConstructions = 0
|
||||
for (const name of readdirSync(dir)) {
|
||||
if (!name.startsWith('remote-runtime-') || !name.endsWith('.ts') || name.includes('.test.')) {
|
||||
continue
|
||||
}
|
||||
const source = readFileSync(join(dir, name), 'utf8')
|
||||
for (const path of coveredSocketSources()) {
|
||||
const source = readFileSync(path, 'utf8')
|
||||
const constructions = source.split('new WebSocket(').length - 1
|
||||
const bounded = source.split('remoteRuntimeConnectOptions(').length - 1
|
||||
scannedConstructions += constructions
|
||||
if (constructions > bounded) {
|
||||
offenders.push(`${name}: ${constructions} WebSocket(s), ${bounded} bounded`)
|
||||
offenders.push(`${basename(path)}: ${constructions} WebSocket(s), ${bounded} bounded`)
|
||||
}
|
||||
}
|
||||
expect(offenders).toEqual([])
|
||||
// Guards against the scan silently matching nothing and passing vacuously.
|
||||
expect(scannedConstructions).toBeGreaterThan(0)
|
||||
// Guards the relay site specifically: an allowlist that quietly stopped
|
||||
// resolving a path would still satisfy the count above.
|
||||
expect(
|
||||
coveredSocketSources().some((path) => path.endsWith(RELAY_CONTROL_SOCKET_FACTORY))
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('reports an unanswered host as unreachable rather than as an empty result', async () => {
|
||||
|
||||
Reference in New Issue
Block a user