diff --git a/docs-site/src/content/docs/guides/remote-link.md b/docs-site/src/content/docs/guides/remote-link.md index 3add1b64d6e..d51f6beffd8 100644 --- a/docs-site/src/content/docs/guides/remote-link.md +++ b/docs-site/src/content/docs/guides/remote-link.md @@ -99,6 +99,16 @@ When a step fails, the dashboard shows the reason and, when SSH reported one, a - **The remote computer did not report an OpenCodex version**: `ocx --version` on that computer printed something else, for example the usage text of an unsupported Windows install. - **Requests fail with `home_unreachable` while the Home is down**: a Child never serves requests from its own providers. When the tunnel is down before a request is forwarded, the Child answers a retryable 503 and sends nothing to the Home. If the tunnel drops after forwarding started, the answer is also `home_unreachable`, but the Home may already have received the request. A failed link relay authentication answers `link_relay_auth_failed` instead: upgrade the Home and Child or re-link. Whether GOAT may move that device's work to its own sign-ins is the per-device local fallback permission on the Home's **Remote Link** page, or `ocx fallback local on --device `; see [Fallback off](/guides/fallback-off/). +### Retryable 503 diagnostics + +A relay-generated `503` includes a server-generated `x-opencodex-link-request-id`. Use it to correlate a failure with local evidence under the Child's OpenCodex configuration home: `link/diagnostics.jsonl` and one rotated backup, `link/diagnostics.jsonl.1`. Each file is capped at 256 KiB, with private permissions (directory 0700 and files 0600 on POSIX; private ACLs on Windows). Records contain only fixed event/outcome/state categories, generated correlation IDs, numeric timing/generation/attempt metadata and timestamps, never secrets, request bodies, headers, URLs, account identifiers or raw errors. Each event/outcome pair is limited to one record per second; the next record includes a count of suppressed events, so an absent record does not prove an absent failure. A backward wall-clock adjustment starts a new rate-limit window, including when the writer queue is full, and preserves the suppression count. An asynchronous writer keeps at most 56 pending records, counts overflow as suppression, and yields while Windows ACL work runs. Logging is best effort and cannot change whether a request is admitted. Graceful signal shutdown and standalone recycling wait up to one second for queued diagnostics before exiting; stalled storage, forced termination and power loss can still lose records. + +`owner_lookup` records refused ownership checks; routine successful checks are silent. `owner_lookup.elapsedMs` measures the owner lookup itself. For `auth_timeout`, `relay_retry` and `relay_failure` report the actual transport-attempt duration in `elapsedMs` (normally about five seconds). A `relay_recovered` event with outcome `recovered` is emitted only after a retry receives an upstream response with `upstream.ok` (HTTP 2xx); it does not prove the streamed body completed. Rate limiting makes request-ID correlation sampled evidence, not a guaranteed trace of every failed request. + +Before sending data, OpenCodex allows one shared preflight retry for either an owner-lookup deadline or a specifically identified five-second authentication deadline. The retry obtains fresh owner proof where required and a fresh physical socket, then authenticates that socket before sending the link credential or body. It does not reuse cached owner approval, bypass security checks, restart SSH or OpenCodex, or replay a dispatched request body. The two-second owner-lookup and five-second authentication deadlines are unchanged for each attempt; other failures do not qualify for this preflight retry. + +This can recover these transient deadline paths and helps diagnose future failures. It does not establish the historical cause of [#7297](https://github.com/lidge-jun/opencodex/issues/7297). + ## Security The Child uses the Home computer's providers and provider credentials through the link. The Home creates a separate link key for each Child; removing the link revokes that key. On the Child, the key stays inside OpenCodex: credentials that Codex or Claude Code send there are not forwarded to the Home, including Bearer, Azure `api-key`, Anthropic-compatible `x-api-key`, and Google `x-goog-api-key` forms. Any program on the Child that reaches `127.0.0.1:` uses the Home without a key, the same local trust a standalone install gives. Web pages from other sites are refused. Compare the host fingerprint before confirmation so a wrong machine or changed host key is not accepted by mistake. Dashboard sessions issued from a Tailscale identity cannot manage machine links. diff --git a/scripts/test-layout/layout-clients.json b/scripts/test-layout/layout-clients.json index 9224a33affb..1e66ac94f41 100644 --- a/scripts/test-layout/layout-clients.json +++ b/scripts/test-layout/layout-clients.json @@ -1,4 +1,5 @@ { + "link-diagnostics.test.ts": "clients", "gentle-shell-path.test.ts": "clients", "gentle-shell-client.test.ts": "clients", "droid-managed-reasoning-defaults.test.ts": "clients", diff --git a/src/client/link-relay-transport.ts b/src/client/link-relay-transport.ts index b43bf4da7b1..ff43ca659e2 100644 --- a/src/client/link-relay-transport.ts +++ b/src/client/link-relay-transport.ts @@ -27,6 +27,14 @@ export class LinkRelayAuthenticationError extends Error { } } +/** Only the credential-free preflight deadline expired; no data request has begun. */ +export class LinkRelayAuthTimeoutError extends Error { + constructor() { + super("link relay authentication timed out"); + this.name = "LinkRelayAuthTimeoutError"; + } +} + /** One physical connection, ever. A closed keep-alive socket cannot trigger a silent reconnect. */ class SingleConnectionAgent extends Agent { socket: Socket | undefined; @@ -153,8 +161,17 @@ async function responseOnConnection(incoming: IncomingMessage, agent: SingleConn let peerReport: Record | undefined; -/** Authenticate and stream one request on the SAME socket; never fall back to an ordinary fetch. */ -export async function fetchBoundLinkRelay(target: BoundLinkRelayTarget, destination: string, init: RequestInit): Promise { +/** + * Authenticate and stream one request on the SAME socket; never fall back to an ordinary fetch. + * LinkRelayAuthTimeoutError certifies that only the credential-free preflight expired before + * data-request construction. Caller aborts and failures after that boundary keep their own errors. + */ +export async function fetchBoundLinkRelay( + target: BoundLinkRelayTarget, destination: string, init: RequestInit, + options: { authTimeoutMs?: number } = {}, +): Promise { + const authTimeoutMs = options.authTimeoutMs ?? LINK_RELAY_AUTH_TIMEOUT_MS; + if (!Number.isFinite(authTimeoutMs) || authTimeoutMs <= 0) throw new RangeError("invalid link relay auth timeout"); if (!target.linkId || !target.apiKeyId) throw new LinkRelayAuthenticationError(); const url = new URL(destination); if (url.origin !== `http://127.0.0.1:${target.tunnelPort}`) throw new LinkRelayAuthenticationError(); @@ -163,7 +180,8 @@ export async function fetchBoundLinkRelay(target: BoundLinkRelayTarget, destinat probe.search = new URLSearchParams({ version: LINK_RELAY_AUTH_VERSION, key: target.apiKeyId, link: target.linkId, nonce: challenge.nonce, proof: challenge.caller }).toString(); const agent = new SingleConnectionAgent({ keepAlive: true, maxSockets: 1, maxFreeSockets: 1 }); - const deadline = clearableDeadline(LINK_RELAY_AUTH_TIMEOUT_MS, init.signal ?? undefined); + const deadline = clearableDeadline(authTimeoutMs, init.signal ?? undefined); + let dataRequestStarted = false; try { // The challenge also carries this Child's version and features for the Home's link status. peerReport ??= linkPeerReportHeaders(packageVersion("0.0.0"), buildPeerFeatures()); @@ -177,15 +195,20 @@ export async function fetchBoundLinkRelay(target: BoundLinkRelayTarget, destinat for await (const chunk of response) { if (chunk.length) throw new LinkRelayAuthenticationError(); } + deadline.signal.throwIfAborted(); deadline.clear(); init.signal?.throwIfAborted(); if (!socket || socket.destroyed || socket !== agent.socket) throw new LinkRelayAuthenticationError(); const headers = new Headers(init.headers); headers.set(LINK_RELAY_SESSION_HEADER, challenge.nonce); + dataRequestStarted = true; const incoming = await requestOnConnection(url, { ...init, headers }, agent, socket); return await responseOnConnection(incoming, agent, init.method ?? "GET"); } catch (error) { + const authDeadlineExpired = !dataRequestStarted && deadline.didExpire() + && !(error instanceof LinkRelayAuthenticationError); await agent.close(); + if (authDeadlineExpired && !init.signal?.aborted) throw new LinkRelayAuthTimeoutError(); throw error; } finally { deadline.clear(); } } diff --git a/src/client/link-relay.ts b/src/client/link-relay.ts index d46e406cdca..2ac599c50e1 100644 --- a/src/client/link-relay.ts +++ b/src/client/link-relay.ts @@ -1,4 +1,5 @@ -import { fetchBoundLinkRelay, LinkRelayAuthenticationError } from "./link-relay-transport"; +import { randomUUID } from "node:crypto"; +import { fetchBoundLinkRelay, LinkRelayAuthenticationError, LinkRelayAuthTimeoutError } from "./link-relay-transport"; import { homeUnreachableBody } from "./home-unreachable"; import { boundedRelayResponseStream, @@ -6,6 +7,7 @@ import { headersWithinLimit, validateHubRelayRequestHeaders, } from "./hub-relay"; +import type { LinkDiagnosticRecord, LinkDiagnosticSink } from "../link/diagnostics"; import { linkRouteAllowed } from "../link/routes"; import { isLinkPort } from "../link/ports"; import { resolveInboundBodyLimitBytes } from "../server/request-decompress"; @@ -26,14 +28,23 @@ export interface LinkRelayClock { clearTimeout: typeof clearTimeout; } +/** A fresh admission verdict plus optional diagnostic evidence; metadata never grants admission. */ +export interface LinkAdmissionResult { + kind: "match" | "mismatch" | "lookup_error" | "deadline" | "stale" | "identity_unknown" | "identity_mismatch" | "unavailable"; + elapsedMs?: number; + generation?: number; + state?: string; + checkId?: string; +} + /** * The Child's tunnel as the relay sees it (the client link supervisor implements it). A request * waits on it only while the tunnel is being (re)established. Every relayed fetch waits for a * fresh bounded asynchronous ownership proof, including after a reconnect or refused retry. */ export interface LinkTunnelGate { - /** True only while a keyed probe has established this live tunnel. */ - connected(): boolean | Promise; + /** Admission and its evidence are returned together; only match permits a fetch. */ + connected(context?: { requestId?: string }): LinkAdmissionResult | Promise; /** True while the tunnel is connecting or reconnecting. */ pending(): boolean; /** @@ -51,12 +62,13 @@ export interface LinkTunnelGate { * supervisor, and a relay with no gate at all still refuses every request. */ export const HOME_INITIATED_LINK_TUNNEL: LinkTunnelGate = { - connected: () => true, + connected: () => ({ kind: "match" }), pending: () => false, waitForConnected: async () => false, }; export interface LinkRelayDeps { + diagnostic?: LinkDiagnosticSink; /** Trusted complete-send test seam. Production always uses the connection-bound transport. */ fetchImpl?: typeof fetch; clock?: LinkRelayClock; @@ -136,6 +148,7 @@ function relayAuthenticationFailed(error: string): Response { }); } +/** Preserve path/query on a fixed IPv4 loopback origin; throw when the tunnel port is invalid. */ export function linkRelayDestination(url: URL, target: Pick): string { if (!isLinkPort(target.tunnelPort)) { throw new RangeError("invalid link tunnel port"); @@ -180,6 +193,7 @@ function linkRequestHeaders( return headers; } +/** Drop hop-by-hop and stale length/encoding headers before exposing the decoded response. */ export function sanitizeLinkResponseHeaders(source: Headers): Headers { const connectionNamed = new Set((source.get("connection") ?? "") .split(",").map(value => value.trim().toLowerCase()).filter(Boolean)); @@ -328,6 +342,11 @@ function idleBoundedStream( }); } +/** + * Admit an allowlisted route and stream it through the authenticated tunnel with bounded bodies. + * Owner/auth preflight deadlines share one recovery attempt before data dispatch; an ambiguous + * post-dispatch failure never earns that recovery. Diagnostic failures cannot change admission. + */ export async function relayLinkDataRequest( req: Request, target: LinkRelayTarget, @@ -352,6 +371,16 @@ export async function relayLinkDataRequest( return jsonError(431, "link relay request headers too large"); } + const requestId = randomUUID(); + let admission: LinkAdmissionResult = { kind: "unavailable" }; + let recoveryAttempts = 0; + const diagnostic = (event: LinkDiagnosticRecord["event"], outcome: LinkDiagnosticRecord["outcome"], elapsedMs?: number) => { + try { deps.diagnostic?.({ event, outcome, requestId, + checkId: admission.checkId, generation: admission.generation, + state: admission.state, elapsedMs: admission.elapsedMs, + ...(elapsedMs !== undefined ? { elapsedMs } : {}), attempt: recoveryAttempts }); } + catch { /* diagnostics never alter admission */ } + }; const relayAbort = new AbortController(); const clock = deps.clock ?? defaultClock; let headerTimer: ReturnType | undefined = clock.setTimeout(() => { @@ -390,20 +419,45 @@ export async function relayLinkDataRequest( const remaining = holdUntil - now; return remaining > 0 && await gate.waitForConnected(remaining, relayAbort.signal); }; - const tunnelUnavailable = async (): Promise => { + const tunnelUnavailable = async ( + reason: LinkDiagnosticRecord["outcome"] = admission.kind === "match" ? "unavailable" : admission.kind, + message = "link tunnel unavailable", + elapsedMs?: number, + ): Promise => { + diagnostic("relay_failure", relayAbort.signal.aborted + ? req.signal.aborted ? "client_abort" : "header_timeout" : reason, elapsedMs); cleanup(); try { await body?.cancel(); } catch { /* best effort */ } - return homeUnreachable("link tunnel unavailable"); + const response = reason === "auth_failed" ? relayAuthenticationFailed(message) : homeUnreachable(message); + response.headers.set("x-opencodex-link-request-id", requestId); + return response; + }; + const retryPreflight = (reason: "deadline" | "auth_timeout", elapsedMs?: number): boolean => { + if (recoveryAttempts >= 1 || relayAbort.signal.aborted || bodyOverflow || bodyCancelled || body?.locked) return false; + recoveryAttempts += 1; + diagnostic("relay_retry", reason, elapsedMs); + return true; + }; + const checkConnected = async (): Promise => { + admission = tunnel ? await tunnel.connected({ requestId }) : { kind: "unavailable" }; + return admission.kind === "match"; }; const readyForFetch = async (): Promise => { - if (!tunnel) return false; - if (await tunnel.connected()) return true; - return tunnel.pending() && await holdForTunnel(tunnel) && await tunnel.connected(); + if (!tunnel || relayAbort.signal.aborted) return false; + if (await checkConnected()) return !relayAbort.signal.aborted; + // No fetch or body read has occurred. A deadline proves neither a foreign owner nor a + // broken SSH process: give this request one fresh proof, without caching or restarting it. + if (admission.kind === "deadline" && retryPreflight("deadline")) { + if (await checkConnected()) return !relayAbort.signal.aborted; + } + return !relayAbort.signal.aborted && tunnel.pending() && await holdForTunnel(tunnel) + && await checkConnected() && !relayAbort.signal.aborted; }; let upstream: Response | undefined; while (!upstream) { if (!await readyForFetch()) return await tunnelUnavailable(); + const fetchStartedAt = performance.now(); try { const init: RequestInit & { duplex?: "half" } = { method: req.method, @@ -415,24 +469,28 @@ export async function relayLinkDataRequest( upstream = await (deps.fetchImpl ? deps.fetchImpl(destination, init) : fetchBoundLinkRelay(target, destination, init)); } catch (error) { if (error instanceof LinkRelayAuthenticationError) { - cleanup(); - try { await body?.cancel(); } catch { /* an upload might already be drained */ } - return relayAuthenticationFailed(error.message); + return await tunnelUnavailable("auth_failed", error.message); } if (bodyOverflow) { cleanup(); return jsonError(413, "link relay request body too large"); } + if (error instanceof LinkRelayAuthTimeoutError) { + // The transport certifies that only its credential-free challenge was sent. A new + // iteration obtains a fresh ownership proof and a new authenticated physical socket. + const elapsedMs = performance.now() - fetchStartedAt; + if (retryPreflight("auth_timeout", elapsedMs)) continue; + return await tunnelUnavailable("auth_timeout", "link tunnel unavailable", elapsedMs); + } // A refused connection sent nothing, and an untouched body can go again once the tunnel // is back inside the same hold window. Anything else is never replayed. const resendable = connectionRefused(error) && (body === null || (!body.locked && !bodyCancelled)); - if (!tunnel || !resendable || !tunnel.pending() || !await holdForTunnel(tunnel)) return await tunnelUnavailable(); + if (!tunnel || !resendable || !tunnel.pending() || !await holdForTunnel(tunnel)) return await tunnelUnavailable("transport_error"); } } if (relayAbort.signal.aborted) { - cleanup(); try { await upstream.body?.cancel(); } catch { /* best effort */ } - return homeUnreachable("link tunnel unavailable"); + return await tunnelUnavailable(); } const sse = isSse(upstream.headers); @@ -449,6 +507,7 @@ export async function relayLinkDataRequest( try { await upstream.body?.cancel(); } catch { /* best effort */ } return jsonError(502, "link relay response body too large"); } + if (recoveryAttempts > 0 && upstream.ok) diagnostic("relay_recovered", "recovered"); if (req.method === "HEAD" || !upstream.body) { cleanup(); try { await upstream.body?.cancel(); } catch { /* best effort */ } diff --git a/src/client/link-tunnel.ts b/src/client/link-tunnel.ts index 4174aeb398d..ba62615e181 100644 --- a/src/client/link-tunnel.ts +++ b/src/client/link-tunnel.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { chmodSync, mkdirSync, readFileSync, statSync, unlinkSync } from "node:fs"; import { execFile } from "node:child_process"; import { readFile as readFileAsync } from "node:fs/promises"; @@ -20,10 +21,11 @@ import { type TunnelState, } from "../link/tunnel-state"; import { isLinkPort } from "../link/ports"; -import { ownsIpv4LoopbackListener } from "../server/port-reclaim"; +import { inspectIpv4LoopbackListener, type LoopbackOwnerLookupResult } from "../server/port-reclaim"; +import type { LinkDiagnosticSink } from "../link/diagnostics"; import { isLinkConnection, readClientConnectionState } from "./state"; import { clientLinkStatePath, readClientLinkState, type ClientLinkState } from "./link-state"; -import type { LinkTunnelGate } from "./link-relay"; +import type { LinkTunnelGate, LinkAdmissionResult } from "./link-relay"; /** * The client-owned `ssh -N -L 127.0.0.1::127.0.0.1: ` @@ -134,7 +136,8 @@ export interface ClientLinkSupervisorDeps extends ClientLinkTunnelDeps, OrphanRe /** The link key for the keyed readiness probe. The runtime passes its cached key source. */ linkKey?: () => string | null; /** Async local IPv4 LISTEN ownership; an unknown or foreign owner never receives the link key. */ - ownsLoopbackListener?: (port: number, pid: number) => Promise; + inspectLoopbackListener?: (port: number, pid: number) => Promise; + diagnostic?: LinkDiagnosticSink; fetchImpl?: typeof fetch; now?: () => number; random?: () => number; @@ -563,6 +566,10 @@ async function probeTunnel(fetchImpl: typeof fetch, tunnelPort: number, key: str } } +/** + * Own the Child's SSH lifecycle and reconnect holds. Each connected() call checks fresh listener + * ownership and fences results against tunnel replacement or stop; diagnostics cannot admit it. + */ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): ClientLinkSupervisor { const sidecarPath = (): string => clientLinkStatePath(deps.configDir); const readSidecar = deps.readSidecar ?? readWhenFileChanges(sidecarPath, () => readClientLinkState(sidecarPath())); @@ -573,7 +580,7 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): const policy = CLIENT_TUNNEL_RETRY_POLICY; const isAlive = deps.isAlive ?? defaultIsAlive; const linkKey = deps.linkKey ?? (() => null); - const ownerLookup = deps.ownsLoopbackListener ?? ownsIpv4LoopbackListener; + const ownerLookup = deps.inspectLoopbackListener ?? inspectIpv4LoopbackListener; const fetchImpl = deps.fetchImpl ?? fetch; const setSupervisorTimer = deps.setTimer ?? ((callback: () => void, ms: number) => { const interval = setInterval(callback, ms); @@ -620,9 +627,16 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): if (pid === null || pid === undefined || (adopted !== null && !adoptedRecord)) return null; return { pid, generation: ownerGeneration }; }; - const listenerOwnedByTunnel = async (port: number): Promise => { + const listenerOwnedByTunnel = async (port: number, context?: { requestId?: string }): Promise => { const candidate = ownerCandidate(port); - if (!candidate) return false; + if (!candidate) return { kind: "unavailable" }; + const checkId = randomUUID(); + const report = (kind: LinkAdmissionResult["kind"], elapsedMs = 0): LinkAdmissionResult => { + const check = { kind, elapsedMs, checkId, generation: candidate.generation, state: state.kind }; + if (kind !== "match") try { deps.diagnostic?.({ event: "owner_lookup", outcome: kind, requestId: context?.requestId, + checkId, generation: candidate.generation, state: state.kind, elapsedMs }); } catch { /* best effort */ } + return check; + }; if (adopted !== null) { const record = adoptedRecord; const identity = record ? await inspectAdoptedIdentity(record, deps) : "unknown"; @@ -640,16 +654,28 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): invalidateOwnerProof(); setState(IDLE); } - return false; + return report(identity === "mismatch" ? "identity_mismatch" : "identity_unknown"); } } - const owned = await Promise.resolve().then(() => ownerLookup(port, candidate.pid)).catch(() => false); - return owned && !stopping && ownerGeneration === candidate.generation - && ownerCandidate(port)?.pid === candidate.pid; + const result = await Promise.resolve().then(() => ownerLookup(port, candidate.pid)) + .catch((): LoopbackOwnerLookupResult => ({ kind: "lookup_error", elapsedMs: 0 })); + if (stopping || failure !== undefined || ownerGeneration !== candidate.generation || ownerCandidate(port)?.pid !== candidate.pid) { + return report("stale", result.elapsedMs); + } + return report(result.kind, result.elapsedMs); + }; + const connected = async (context?: { requestId?: string }): Promise => { + if (!started || stopping || failure !== undefined || state.kind !== "connected" || linkId === null || tunnelPort === null) { + return { kind: "unavailable" }; + } + const result = await listenerOwnedByTunnel(tunnelPort, context); + // State may change during the asynchronous lookup or an optional diagnostic callback. + if (result.kind === "match" && (!started || stopping || failure !== undefined + || state.kind !== "connected" || result.generation !== ownerGeneration)) { + return { ...result, kind: "stale" }; + } + return result; }; - const connected = async (): Promise => started && !stopping && failure === undefined - && state.kind === "connected" && linkId !== null && tunnelPort !== null - && await listenerOwnedByTunnel(tunnelPort); const settleWaiters = (): void => { if (waiters.size === 0) return; @@ -800,7 +826,7 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): probeFlight = flight; const wasConnected = state.kind === "connected"; void (async () => { - if (!await listenerOwnedByTunnel(tunnelPort)) return; + if ((await listenerOwnedByTunnel(tunnelPort)).kind !== "match") return; if (probeFlight !== flight || stopping || flight.abort.signal.aborted || child !== flight.child || adopted !== flight.adopted) return; const key = linkKey(); @@ -808,7 +834,7 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): const result = await probeTunnel(fetchImpl, tunnelPort, key, flight.abort.signal); // An aborted probe, or one whose listener changed while it ran, proves nothing. if (probeFlight !== flight || stopping || child !== flight.child || adopted !== flight.adopted - || !await listenerOwnedByTunnel(tunnelPort)) return; + || (await listenerOwnedByTunnel(tunnelPort)).kind !== "match") return; applyProbe(result, wasConnected); })().catch(() => undefined).finally(() => { if (probeFlight === flight) probeFlight = undefined; @@ -971,7 +997,7 @@ export function createClientLinkSupervisor(deps: ClientLinkSupervisorDeps = {}): pending, connected, waitForConnected(timeoutMs: number, signal?: AbortSignal): Promise { - if (state.kind === "connected") return connected(); + if (state.kind === "connected") return connected().then(result => result.kind === "match"); if (!pending() || waiters.size >= CLIENT_LINK_MAX_HOLDS || signal?.aborted || !(timeoutMs > 0)) { return Promise.resolve(false); } diff --git a/src/client/machine-listener.ts b/src/client/machine-listener.ts index 5009507d641..6d107e62702 100644 --- a/src/client/machine-listener.ts +++ b/src/client/machine-listener.ts @@ -1,3 +1,4 @@ +import { recordLinkDiagnostic } from "../link/diagnostics"; import type { Server } from "bun"; import { loadConfig } from "../config"; import { browserSecurityHeaders, requestPolicyView } from "../server/auth-cors"; @@ -152,7 +153,7 @@ export function startMachineListener( apiKeyId: connection.apiKeyId, policy: requestPolicyView(config, "127.0.0.1"), linkKey: deps.linkKeySource ?? createLinkKeySource(connection.tokenFingerprint, deps.linkKey), - relay: { fetchImpl: deps.fetchImpl, bodyLimitBytes: inboundBodyLimit, tunnel: deps.linkTunnel, ...deps.linkRelay }, + relay: { diagnostic: recordLinkDiagnostic, fetchImpl: deps.fetchImpl, bodyLimitBytes: inboundBodyLimit, tunnel: deps.linkTunnel, ...deps.linkRelay }, } : null; return (deps.serve ?? (options => Bun.serve(options)))({ diff --git a/src/client/runtime.ts b/src/client/runtime.ts index cb12800eb14..a547811ec6f 100644 --- a/src/client/runtime.ts +++ b/src/client/runtime.ts @@ -1,3 +1,4 @@ +import { flushLinkDiagnostics, recordLinkDiagnostic } from "../link/diagnostics"; import { existsSync } from "node:fs"; import type { Server } from "bun"; import { siblingRuntimeField, withSiblingMarker } from "../codex/sibling-start"; @@ -164,6 +165,7 @@ export async function recycleStandalone( console.warn(`[client] listener stop failed: ${error instanceof Error ? error.message : String(error)}`); } cleanup(); + await flushLinkDiagnostics(); // Recycling back to standalone after `ocx disconnect` must actually bring a standalone // proxy back, under every launch shape. // @@ -318,6 +320,7 @@ export async function startClientRuntime( const serve = async (rollback?: { port: number; portWaitMs: number }): Promise => { const supervisor = supervised ? createClientLinkSupervisor({ + diagnostic: recordLinkDiagnostic, onLinkEnded: () => scheduleStandaloneRecycle(state.value.tokenFingerprint), linkKey, }) @@ -417,7 +420,10 @@ export async function startClientRuntime( await recycleStandalone(state.value.tokenFingerprint); } }, - exitProcess: code => { process.exit(code); }, + exitProcess: async code => { + await flushLinkDiagnostics(); + process.exit(code); + }, }); installCrashGuards(); @@ -444,6 +450,7 @@ export async function startClientRuntime( console.warn(`[client] listener stop failed: ${error instanceof Error ? error.message : String(error)}`); } finally { cleanup(); + await flushLinkDiagnostics(); // launchd relaunches only unsuccessful exits; an external signal must still count. process.exit(handledSignalExitCode(signal)); } diff --git a/src/link/diagnostics.ts b/src/link/diagnostics.ts new file mode 100644 index 00000000000..f2203f40974 --- /dev/null +++ b/src/link/diagnostics.ts @@ -0,0 +1,182 @@ +import { constants } from "node:fs"; +import { chmod, lstat, mkdir, open, rename, unlink } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; +import { hardenSecretDirAsync, hardenSecretPathAsync, reattributeHardenedSecretPath, windowsSecretAclApplies } from "../lib/windows-secret-acl"; +import { linkDir } from "./paths"; +import type { TunnelState } from "./tunnel-state"; + +export type LinkDiagnosticRecord = { + event: "owner_lookup" | "relay_retry" | "relay_failure" | "relay_recovered"; + outcome: "match" | "mismatch" | "lookup_error" | "deadline" | "stale" | "unavailable" + | "identity_unknown" | "identity_mismatch" | "auth_timeout" | "auth_failed" + | "transport_error" | "client_abort" | "header_timeout" | "recovered"; + requestId?: string; + checkId?: string; + generation?: number; + state?: string; + elapsedMs?: number; + attempt?: number; +}; +export type LinkDiagnosticSink = (record: LinkDiagnosticRecord) => void; + +const EVENTS = new Set(["owner_lookup", "relay_retry", "relay_failure", "relay_recovered"]); +const OUTCOMES = new Set([ + "match", "mismatch", "lookup_error", "deadline", "stale", "unavailable", "identity_unknown", + "identity_mismatch", "auth_timeout", "auth_failed", "transport_error", "client_abort", "header_timeout", "recovered", +]); +const STATES = new Set(["idle", "connecting", "connected", "reconnecting", "failed"]); +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +const MAX_BYTES = 256 * 1024; + +/** Copy only bounded enum, UUID and numeric fields; unknown fields never reach disk. */ +function metadata(record: LinkDiagnosticRecord): LinkDiagnosticRecord | null { + if (!record) return null; + const { event, outcome, state } = record; + if (!EVENTS.has(event) || !OUTCOMES.has(outcome)) return null; + const result: LinkDiagnosticRecord = { event, outcome }; + for (const key of ["requestId", "checkId"] as const) { + const value = record[key]; + if (typeof value === "string" && value.length === 36 && UUID.test(value)) result[key] = value; + } + if (typeof state === "string" && STATES.has(state as TunnelState["kind"])) result.state = state; + for (const key of ["generation", "elapsedMs", "attempt"] as const) { + const value = record[key]; + if (typeof value === "number" && Number.isFinite(value) && value >= 0 && value <= Number.MAX_SAFE_INTEGER + && (key === "elapsedMs" || Number.isSafeInteger(value))) result[key] = value; + } + return result; +} + +/** Append through a private no-follow handle, keeping at most one 256 KiB rotated backup. */ +async function append(dir: string, line: string): Promise { + assertNotRealHomeUnderTest(dirname(dir)); + await mkdir(dir, { recursive: true, mode: 0o700 }); + const directory = await lstat(dir); + if (!directory.isDirectory() || directory.isSymbolicLink()) return; + const windows = windowsSecretAclApplies(); + if (windows) await hardenSecretDirAsync(dir, { required: true }); + else await chmod(dir, 0o700); + const path = join(dir, "diagnostics.jsonl"); + const backup = `${path}.1`; + let file: Awaited> | undefined; + try { + // No read of log contents, and no following a file symlink or writing to a pipe. + const flags = constants.O_WRONLY | constants.O_APPEND | constants.O_CREAT | constants.O_NOFOLLOW | constants.O_NONBLOCK; + file = await open(path, flags, 0o600); + const current = await file.stat(); + if (!current.isFile() || current.nlink !== 1) return; + if (windows) await hardenSecretPathAsync(path, { required: true }); + else await file.chmod(0o600); + const previous = await lstat(backup).catch(error => { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + return null; + }); + if (previous) { + if (!previous.isFile() || previous.isSymbolicLink() || previous.nlink !== 1) return; + if (previous.size > MAX_BYTES) await unlink(backup); + else if (windows) await hardenSecretPathAsync(backup, { required: true }); + else await chmod(backup, 0o600); + } + if (current.size > MAX_BYTES) await file.truncate(0); + else if (current.size + Buffer.byteLength(line) > MAX_BYTES) { + await file.close(); + file = undefined; + await rename(path, backup); + file = await open(path, flags, 0o600); + const fresh = await file.stat(); + if (!fresh.isFile() || fresh.nlink !== 1 || fresh.size + Buffer.byteLength(line) > MAX_BYTES) return; + if (windows) await hardenSecretPathAsync(path, { required: true }); + else await file.chmod(0o600); + } + await file.write(line); + if (windows) { + const written = await file.stat({ bigint: true }); + const named = await lstat(path, { bigint: true }); + if (!named.isSymbolicLink() && named.dev === written.dev && named.ino === written.ino) { + // Windows publishes content-write ChangeTime on close; retain only the same object's ACL memo. + await file.close(); + file = undefined; + reattributeHardenedSecretPath(path); + } + } + } finally { + await file?.close(); + } +} + +/** + * Create a best-effort metadata sink with one-second event/outcome buckets and a bounded queue. + * File and ACL work starts on the first eligible record and never blocks admission. flush() waits + * at most one second; its timeout does not cancel the writer or guarantee durable storage. + */ +export function createLinkDiagnosticSink( + { configDir, now = Date.now }: { configDir?: string; now?: () => number } = {}, +): LinkDiagnosticSink & { flush: (timeoutMs?: number) => Promise } { + // At most 56 buckets; correlation IDs never become keys or retained memory. + const buckets = new Map(); + const queue: Array<{ dir: string; line: string }> = []; + let flight: Promise | undefined; + const drain = async () => { + while (queue.length) { + const record = queue.shift()!; + try { await append(record.dir, record.line); } catch { /* Best effort; never publish raw errors. */ } + } + }; + const startWriter = () => { + if (flight) return; + flight = drain().finally(() => { + flight = undefined; + if (queue.length) startWriter(); + }); + }; + const sink: LinkDiagnosticSink = record => { + try { + const safe = metadata(record); + if (!safe) return; + const at = now(); + if (!Number.isSafeInteger(at) || at < 0) return; + const key = `${safe.event}:${safe.outcome}`; + const bucket = buckets.get(key); + // Wall-clock rollback starts a new window, even when the writer is backed up. + if (bucket && at >= bucket.at && at - bucket.at < 1_000) { + bucket.suppressed = Math.min(Number.MAX_SAFE_INTEGER, bucket.suppressed + 1); + return; + } + if (queue.length >= 56) { + buckets.set(key, { at: Math.min(bucket?.at ?? at, at), + suppressed: Math.min(Number.MAX_SAFE_INTEGER, (bucket?.suppressed ?? 0) + 1) }); + return; + } + const suppressed = bucket?.suppressed ?? 0; + // Failed writes are rate-limited too; diagnostics cannot hammer an unavailable disk. + buckets.set(key, { at, suppressed: 0 }); + queue.push({ dir: linkDir(configDir), line: `${JSON.stringify({ ...safe, timestamp: at, ...(suppressed ? { suppressed } : {}) })}\n` }); + startWriter(); + } catch { /* Diagnostics must never change request admission or expose raw errors. */ } + }; + return Object.assign(sink, { flush: async (timeoutMs = 1_000) => { + if (!flight) return; + // Keep shutdown bounded even if an asynchronous filesystem or ACL operation stalls. + // Timeout leaves the existing writer alone; it may still finish before process exit. + let timer: ReturnType | undefined; + try { + await Promise.race([ + (async () => { while (flight) await flight; })(), + new Promise(resolve => { timer = setTimeout(resolve, Math.min(1_000, Math.max(0, timeoutMs))); }), + ]); + } catch { /* Best effort; never publish raw errors. */ } + finally { clearTimeout(timer); } + } }); +} + +let defaultSink: ReturnType | undefined; +/** Lazily enqueue sanitized evidence in the default sink without propagating diagnostic failures. */ +export function recordLinkDiagnostic(record: LinkDiagnosticRecord): void { + try { (defaultSink ??= createLinkDiagnosticSink())(record); } catch { /* Best effort. */ } +} + +/** Drain existing diagnostics for at most one second without activating an unused sink. */ +export async function flushLinkDiagnostics(): Promise { + await defaultSink?.flush(); +} diff --git a/src/server/port-reclaim.ts b/src/server/port-reclaim.ts index 73ddc7caefa..c3dcc0231bf 100644 --- a/src/server/port-reclaim.ts +++ b/src/server/port-reclaim.ts @@ -405,22 +405,37 @@ async function runOwnerLookup(file: string, args: string[], timeoutMs: number): return stdout; } +export type LoopbackOwnerLookupResult = { + kind: "match" | "mismatch" | "lookup_error" | "deadline"; + elapsedMs: number; +}; + export interface LoopbackOwnerLookupIo { platform?: NodeJS.Platform; + nowMs?: () => number; + scheduleDeadline?: (callback: () => void, timeoutMs: number) => () => void; readProc?: (path: string) => Promise; listFds?: (path: string) => Promise; readFdLink?: (path: string) => Promise; run?: (file: string, args: string[], timeoutMs: number) => Promise; } -/** Async, bounded proof that this PID owns the IPv4 loopback LISTEN socket used by the relay. */ -export async function ownsIpv4LoopbackListener( +/** + * Classify fresh IPv4 loopback LISTEN ownership within the bounded lookup deadline. + * Only match proves the expected PID; mismatch, inspection failure and timeout remain distinct. + * elapsedMs uses a monotonic clock by default, and no raw OS output is returned. + */ +export async function inspectIpv4LoopbackListener( port: number, expectedPid: number, io: LoopbackOwnerLookupIo = {}, -): Promise { +): Promise { + const now = io.nowMs ?? (() => performance.now()); + const startedAt = now(); + const result = (kind: LoopbackOwnerLookupResult["kind"]): LoopbackOwnerLookupResult => + ({ kind, elapsedMs: Math.max(0, now() - startedAt) }); if (!Number.isInteger(port) || port < 1 || port > 65535 - || !Number.isSafeInteger(expectedPid) || expectedPid < 1) return false; + || !Number.isSafeInteger(expectedPid) || expectedPid < 1) return result("mismatch"); const platform = io.platform ?? process.platform; const lookup = async (): Promise => { if (platform === "linux") { @@ -458,13 +473,40 @@ export async function ownsIpv4LoopbackListener( } return false; }; - let timer: ReturnType | undefined; - const deadline = new Promise(resolve => { - timer = setTimeout(() => resolve(false), OWNER_LOOKUP_TIMEOUT_MS); + const scheduleDeadline = io.scheduleDeadline ?? ((callback, timeoutMs) => { + const timer = setTimeout(callback, timeoutMs); timer.unref?.(); + return () => clearTimeout(timer); }); - try { return await Promise.race([lookup().catch(() => false), deadline]); } - finally { if (timer) clearTimeout(timer); } + let cancelDeadline: (() => void) | undefined; + const deadline = new Promise(resolve => { + cancelDeadline = scheduleDeadline(() => resolve("deadline"), OWNER_LOOKUP_TIMEOUT_MS); + }); + const inspected = lookup().then( + matches => matches ? "match" as const : "mismatch" as const, + error => { + const failure = error as { + killed?: boolean; signal?: string | null; code?: unknown; stdout?: unknown; stderr?: unknown; + } | null; + // lsof exits 1 with no output when no socket matches its selectors. + if (platform === "darwin" && failure?.code === 1 && !failure.killed + && failure.signal == null && failure.stdout === "" && failure.stderr === "") return "mismatch" as const; + // execFile's timeout kills with SIGTERM; other spawn/exit/buffer errors are lookup failures. + return failure?.killed === true && failure.signal === "SIGTERM" && failure.code === null + ? "deadline" as const : "lookup_error" as const; + }, + ); + try { return result(await Promise.race([inspected, deadline])); } + finally { cancelDeadline?.(); } +} + +/** Boolean compatibility API: only a proven match admits the listener. */ +export async function ownsIpv4LoopbackListener( + port: number, + expectedPid: number, + io: LoopbackOwnerLookupIo = {}, +): Promise { + return (await inspectIpv4LoopbackListener(port, expectedPid, io)).kind === "match"; } /** diff --git a/structure/remote-link.md b/structure/remote-link.md index 69daf34d44c..dc08dae2a45 100644 --- a/structure/remote-link.md +++ b/structure/remote-link.md @@ -60,7 +60,7 @@ Applying a link probes the host key into a temporary file, waits for the operato ## Client link transport -The Child relay requires a positive `connected()` verdict from `src/client/link-tunnel.ts` before every fetch. Before the first keyed readiness probe and after a tunnel restart, the supervisor asynchronously proves the exact `127.0.0.1:` LISTEN socket belongs to its SSH child; an adopted process also needs matching pidfile argv and start time. `src/server/port-reclaim.ts` uses Linux `/proc/net/tcp{,6}` plus the expected PID's `/proc//fd` socket symlinks without external tools, macOS `lsof` and Windows `netstat` with bounded asynchronous execution. Every key-bearing probe and relayed request makes a fresh bounded asynchronous owner lookup; an adopted process also has its current argv and start time rechecked on every admission. An unreadable or timed-out identity denies only that admission and is retried; a confirmed mismatch releases the adopted PID without signalling it and lets the next tick start a fresh tunnel. An unknown socket-owner lookup likewise denies the current admission. A missing supervisor, or a failed or stopped tunnel, returns a retryable 503 without sending the link key or body to the persisted loopback port; its body carries `code: "home_unreachable"` under the [fallback policy](fallback-policy.md#hub-down). A Home-initiated Child is the exception: the Home owns that link's `ssh -R` forward, so the Child has no sidecar, no supervisor and no SSH process whose socket it could prove. A join writes `/link/child-initiated.json` (the link id, mode 0600) beside the sidecar and removes it with the sidecar, so a Child-initiated link that lost its sidecar is recognised (`isChildInitiatedLink` in `src/client/link-state.ts`; an unreadable marker counts as present) and gets no gate. Only a link-mode runtime with neither the sidecar nor a marker for its link id is treated as Home-initiated: `src/client/runtime.ts` passes it `HOME_INITIATED_LINK_TUNNEL` (`src/client/link-relay.ts`), which omits local process ownership proof and reconnect holding but still uses the connection-bound relay authentication below. A refused connection answers 503. A held request rechecks the verdict after its wait and before each retry. +The Child relay requires a `connected()` result with `kind: "match"` from `src/client/link-tunnel.ts` before every fetch. The gate returns one typed `LinkAdmissionResult` containing the verdict and bounded timing/generation/check metadata; diagnostic callbacks do not supply admission or recovery decisions. The supervisor has one typed ownership dependency, `inspectLoopbackListener`. Before the first keyed readiness probe and after a tunnel restart, the supervisor asynchronously proves the exact `127.0.0.1:` LISTEN socket belongs to its SSH child; an adopted process also needs matching pidfile argv and start time. `src/server/port-reclaim.ts` uses Linux `/proc/net/tcp{,6}` plus the expected PID's `/proc//fd` socket symlinks without external tools, macOS `lsof` and Windows `netstat` with bounded asynchronous execution. Every key-bearing probe and relayed request makes a fresh bounded asynchronous owner lookup; an adopted process also has its current argv and start time rechecked on every admission. An unreadable or timed-out identity denies only that admission and is retried; a confirmed mismatch releases the adopted PID without signalling it and lets the next tick start a fresh tunnel. An unknown socket-owner lookup likewise denies the current admission. A missing supervisor, or a failed or stopped tunnel, returns a retryable 503 without sending the link key or body to the persisted loopback port; its body carries `code: "home_unreachable"` under the [fallback policy](fallback-policy.md#hub-down). A Home-initiated Child is the exception: the Home owns that link's `ssh -R` forward, so the Child has no sidecar, no supervisor and no SSH process whose socket it could prove. A join writes `/link/child-initiated.json` (the link id, mode 0600) beside the sidecar and removes it with the sidecar, so a Child-initiated link that lost its sidecar is recognised (`isChildInitiatedLink` in `src/client/link-state.ts`; an unreadable marker counts as present) and gets no gate. Only a link-mode runtime with neither the sidecar nor a marker for its link id is treated as Home-initiated: `src/client/runtime.ts` passes it `HOME_INITIATED_LINK_TUNNEL` (`src/client/link-relay.ts`), which omits local process ownership proof and reconnect holding but still uses the connection-bound relay authentication below. A refused connection answers 503. A held request rechecks the verdict after its wait and before each retry. The listener can change between an ownership check and the TCP connect. The data relay now authenticates that physical connection before sending the key or body (see [Connection-bound relay authentication](#connection-bound-relay-authentication)); this does not change the separate readiness/enrollment probes. A private Unix-socket forward could additionally remove competing TCP listeners from those probe paths. @@ -78,6 +78,18 @@ Regression coverage lives in `tests/clients/link-ssh-argv.test.ts`, `tests/clien Enrollment cancellation in `src/client/connect.ts` reaches actual network requests and every subsequent write boundary. `src/client/link-join.ts` awaits that transaction's commit or completed rollback instead of racing a separate failure against its terminal result. An exit observed before commit aborts enrollment, drains local rollback, and only then revokes the issued key; an exit queued after the synchronous commit retains the committed link for restart. Readiness polling also observes tunnel exit while sleeping, rather than waiting out its deadline. `tests/server/link-join-route.test.ts` and `tests/clients/client-link-connect.test.ts` pin those boundaries. +## Bounded diagnostics and preflight recovery + +On Windows, a successful append refreshes the existing ACL memo only after matching the path to the open file's identity and closing the handle. Repeated writes avoid redundant ACL subprocesses; replacing the file or changing its permissions still requires fresh hardening. Exits from a bound client restart drain the same bounded diagnostic queue as signal shutdown and standalone recycle. + +The `diagnostics.ts` module in `src/link/` owns best-effort local evidence in `/link/diagnostics.jsonl` and its single rotated backup, `diagnostics.jsonl.1`, each capped at 256 KiB. The link directory is private (0700 on POSIX), files are private (0600), and Windows uses the secret-path ACL helpers. Diagnostics accept only fixed event/outcome/state values, generated UUID request/check IDs, numeric generation, elapsed time and attempt metadata, and a timestamp; they exclude secrets, request bodies, headers, URLs, account identifiers and raw errors. Each event/outcome pair emits at most once per second; the next emitted record carries the intervening `suppressed` count. A backward wall-clock adjustment starts a new rate-limit window, including when the writer queue is full, and preserves the suppression count. An asynchronous writer keeps at most 56 pending records, counts overflow as suppression, and yields while Windows ACL work runs. Write failures are also rate-limited and never change admission. Graceful signal shutdown and standalone recycling wait up to one second for queued diagnostics before exiting; stalled storage, forced termination and power loss can still lose records. Relay-generated 503 responses carry a server-generated `x-opencodex-link-request-id` for correlation; incoming caller IDs are not trusted as this identifier. + +`owner_lookup` records only refused ownership checks; routine successful checks are silent. `owner_lookup.elapsedMs` measures the owner lookup itself. For `auth_timeout`, `relay_retry` and `relay_failure` report the actual transport-attempt duration in `elapsedMs` (normally about five seconds). A `relay_recovered` event with outcome `recovered` is emitted only after a retry receives an upstream response with `upstream.ok` (HTTP 2xx); it does not prove the streamed body completed. Rate limiting makes request-ID correlation sampled evidence, not a guaranteed trace of every failed request. + +`src/client/link-relay.ts` shares one preflight recovery attempt per request between an owner-lookup `deadline` and the typed `LinkRelayAuthTimeoutError` from `src/client/link-relay-transport.ts`. Only expiry of the credential-free authentication deadline before data dispatch produces that type; arbitrary aborts, authentication failures and transport errors do not qualify. Recovery obtains fresh owner proof where required and opens a fresh physical socket, authenticating it before sending any link credential or body. It never caches owner approval, bypasses security, restarts SSH or OpenCodex, or replays a dispatched body. The owner-lookup and authentication deadlines remain two and five seconds respectively, per attempt. Existing reconnect holding remains separate. + +This recovers those transient preflight deadline paths and records bounded evidence for future failures. It does not establish the historical cause of [#7297](https://github.com/lidge-jun/opencodex/issues/7297); timing clusters and synthetic reproductions alone do not classify past production failures. + ## Connection-bound relay authentication The production relay's complete-send boundary is `src/client/link-relay-transport.ts`. A dedicated HTTP agent opens exactly one TCP socket. Before any admission header or body is sent, a credential-free, five-second challenge proves the stored link/key identity using direction-separated HMACs from `src/link/relay-auth.ts`. The data request uses that same authenticated socket, with a one-use session nonce; a closed socket cannot be silently replaced. A peer lacking this protocol is refused with an upgrade/re-link error, never retried through an unauthenticated send. That refusal is a 503 with `code: "link_relay_auth_failed"` (`src/client/link-relay.ts`), never `home_unreachable`: the peer answered, so it is not a Home outage and must not open device-local fallback ([fallback policy](fallback-policy.md#hub-down)). Home-initiated reverse tunnels use the same check. `LinkRelayDeps.fetchImpl` is a trusted complete-send test seam, not a user setting; production supplies no substitute. Bun's Node HTTP compatibility is confined to this leaf so streamed request caps, backpressure, credential replacement, decompression, response status, caller cancellation and SSE limits remain owned by the existing relay. diff --git a/tests/clients/client-home-unreachable.test.ts b/tests/clients/client-home-unreachable.test.ts index b8c7103a37c..9ce57ed293f 100644 --- a/tests/clients/client-home-unreachable.test.ts +++ b/tests/clients/client-home-unreachable.test.ts @@ -68,7 +68,7 @@ describe("home_unreachable answers", () => { headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "m", input: "hello" }), }), { tunnelPort: 41001, admissionKey: "fixture-link-key", linkId: "lnk_0123456789abcdef", apiKeyId: "key_link" }, { - tunnel: { connected: () => true, pending: () => false, waitForConnected: async () => true }, + tunnel: { connected: () => ({ kind: "match" }), pending: () => false, waitForConnected: async () => true }, fetchImpl: (async () => { throw new LinkRelayAuthenticationError(unrecognized); }) as unknown as typeof fetch, }); expect(response.status).toBe(503); diff --git a/tests/clients/client-link-relay.test.ts b/tests/clients/client-link-relay.test.ts index 6842a69a2b4..4fd8d8bd41e 100644 --- a/tests/clients/client-link-relay.test.ts +++ b/tests/clients/client-link-relay.test.ts @@ -15,7 +15,11 @@ import { sanitizeLinkResponseHeaders, type LinkRelayClock, type LinkTunnelGate, + type LinkAdmissionResult, } from "../../src/client/link-relay"; +import { LinkRelayAuthenticationError, LinkRelayAuthTimeoutError } from "../../src/client/link-relay-transport"; +import type { LinkDiagnosticRecord } from "../../src/link/diagnostics"; +import { LINK_RELAY_AUTH_PATH } from "../../src/link/relay-auth"; import { HUB_RELAY_REQUEST_BODY_MAX_BYTES } from "../../src/client/hub-relay"; import { HOME_UNREACHABLE_MESSAGE } from "../../src/client/home-unreachable"; import { startMachineListener } from "../../src/client/machine-listener"; @@ -25,7 +29,7 @@ import { removeTreeWithRetry } from "../helpers/remove-tree"; const LINK_KEY = `ocx_data_${"d".repeat(40)}`; const target = { tunnelPort: 12000, admissionKey: LINK_KEY }; -const connectedTunnel: LinkTunnelGate = { connected: () => true, pending: () => false, waitForConnected: async () => true }; +const connectedTunnel: LinkTunnelGate = { connected: () => ({ kind: "match" }), pending: () => false, waitForConnected: async () => true }; // Payload and framing cases provide a proven tunnel; admission cases call the implementation directly. function relayLinkDataRequest( req: Request, @@ -474,7 +478,7 @@ function manualGate(initiallyPending = true) { const state = { pending: initiallyPending, connected: !initiallyPending }; let open: (connected: boolean) => void = () => {}; const tunnel: LinkTunnelGate = { - connected: () => state.connected, + connected: () => ({ kind: state.connected ? "match" : "unavailable" }), pending: () => state.pending, waitForConnected: timeoutMs => { waits.push(timeoutMs); @@ -497,7 +501,7 @@ describe("client link relay while the tunnel reconnects", () => { for (const state of ["missing", "failed", "stopped"] as const) { test(`does not send the committed key or fetch while supervision is ${state}`, async () => { const forwarded: string[] = []; - const unavailable = { pending: () => false, connected: () => false, waitForConnected: async () => false }; + const unavailable = { pending: () => false, connected: () => ({ kind: "unavailable" as const }), waitForConnected: async () => false }; const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "{}" }), target, { ...(state === "missing" ? {} : { tunnel: unavailable }), fetchImpl: (async (_input, init) => { @@ -535,7 +539,7 @@ describe("client link relay while the tunnel reconnects", () => { let calls = 0; const response = await relayLinkDataRequestImpl(new Request("http://127.0.0.1:10100/v1/models"), target, { tunnel: { - connected: () => connected, + connected: () => ({ kind: connected ? "match" : "unavailable" }), pending: () => pending, waitForConnected: async () => { pending = false; connected = false; return true; }, }, @@ -597,22 +601,25 @@ describe("client link relay while the tunnel reconnects", () => { expect(calls).toBe(0); }); - test("a connected tunnel adds no wait", async () => { + test("a connected tunnel adds no wait or routine diagnostic", async () => { let waited = 0; - const tunnel: LinkTunnelGate = { connected: () => true, pending: () => false, waitForConnected: async () => { waited += 1; return true; } }; + const records: LinkDiagnosticRecord[] = []; + const tunnel: LinkTunnelGate = { connected: () => ({ kind: "match" }), pending: () => false, waitForConnected: async () => { waited += 1; return true; } }; const response = await relayLinkDataRequest(relayRequest({ method: "POST", body: "{}" }), target, { tunnel, + diagnostic: row => { records.push(row); }, fetchImpl: (async () => Response.json({ ok: true })) as typeof fetch, }); expect(response.status).toBe(200); expect(waited).toBe(0); + expect(records).toEqual([]); }); test("never replays a request after a reset, even while the tunnel reconnects", async () => { let pending = false; let calls = 0; let waited = 0; - const tunnel: LinkTunnelGate = { connected: () => !pending, pending: () => pending, waitForConnected: async () => { waited += 1; return true; } }; + const tunnel: LinkTunnelGate = { connected: () => ({ kind: pending ? "unavailable" : "match" }), pending: () => pending, waitForConnected: async () => { waited += 1; return true; } }; const response = await relayLinkDataRequest(relayRequest({ method: "POST", body: '{"input":"once"}' }), target, { tunnel, fetchImpl: (async (_input, init) => { @@ -634,7 +641,7 @@ describe("client link relay while the tunnel reconnects", () => { let clock = 1_000; const waits: number[] = []; const tunnel: LinkTunnelGate = { - connected: () => !pending, + connected: () => ({ kind: pending ? "unavailable" : "match" }), pending: () => pending, waitForConnected: async timeoutMs => { waits.push(timeoutMs); @@ -669,7 +676,7 @@ describe("client link relay while the tunnel reconnects", () => { calls = 0; let spentConnected = false; const spent = await relayLinkDataRequest(relayRequest({ method: "POST", body: "{}" }), target, { - tunnel: { connected: () => spentConnected, pending: () => !spentConnected, + tunnel: { connected: () => ({ kind: spentConnected ? "match" : "unavailable" }), pending: () => !spentConnected, waitForConnected: async () => { clock += LINK_RELAY_HOLD_MS; spentConnected = true; return true; } }, now: () => clock, fetchImpl: (async () => { @@ -693,7 +700,7 @@ describe("client link relay while the tunnel reconnects", () => { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ input: "x".repeat(100_000) }), }), { tunnelPort: port, admissionKey: LINK_KEY }, { tunnel: { - connected: () => !pending, + connected: () => ({ kind: pending ? "unavailable" : "match" }), pending: () => pending, waitForConnected: async () => { home = Bun.serve({ hostname: "127.0.0.1", port, async fetch(req) { received.push(await req.text()); return Response.json({ ok: true }); } }); @@ -737,3 +744,227 @@ describe("client link relay while the tunnel reconnects", () => { expect(await response.json()).toEqual({ relayed: true }); }); }); + +test("a transient owner deadline retries before forwarding even when diagnostics throw", async () => { + const { gate, checks } = observedGate(["deadline", "match"]); + let forwarded = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "untouched" }), target, { + tunnel: gate, + diagnostic: () => { throw new Error("diagnostic sink unavailable"); }, + // This trusted fetch seam records only the final data dispatch. + fetchImpl: (async (_url, init) => { forwarded += 1; expect(await new Response(init?.body).text()).toBe("untouched"); return Response.json({ ok: true }); }) as typeof fetch, + }); + expect(response.status).toBe(200); + expect(checks()).toBe(2); + expect(forwarded).toBe(1); +}); + + +function observedGate(outcomes: Array, onCheck?: () => void) { + let checks = 0; + const gate: LinkTunnelGate = { + connected() { + const kind = outcomes[Math.min(checks++, outcomes.length - 1)]; + onCheck?.(); + return { kind, elapsedMs: kind === "deadline" ? 2000 : 1, generation: 2, state: "connected" }; + }, pending: () => false, waitForConnected: async () => false, + }; + return { gate, checks: () => checks }; +} + +test.each(["mismatch", "lookup_error", "stale", "identity_unknown", "identity_mismatch", "unavailable"] as const)( + "does not retry an owner refusal without a deadline: %s", async kind => { + const { gate, checks } = observedGate([kind]); + let forwards = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "private" }), target, { + tunnel: gate, fetchImpl: (async () => { forwards++; return new Response("unexpected"); }) as typeof fetch, + }); + expect(response.status).toBe(503); + expect(checks()).toBe(1); + expect(forwards).toBe(0); + }); + +test("repeated owner deadlines stop after one recovery and correlate the final refusal", async () => { + const { gate, checks } = observedGate(["deadline"]); + const records: LinkDiagnosticRecord[] = []; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST" }), target, { tunnel: gate, diagnostic: row => records.push(row) }); + expect(response.status).toBe(503); + expect(checks()).toBe(2); + expect(response.headers.get("retry-after")).toBe("1"); + const requestId = response.headers.get("x-opencodex-link-request-id"); + expect(requestId).toMatch(/^[0-9a-f-]{36}$/); + expect(records.map(row => [row.event, row.outcome, row.attempt, row.requestId])).toEqual([ + ["relay_retry", "deadline", 1, requestId], ["relay_failure", "deadline", 1, requestId], + ]); +}); + +test("an auth deadline retries fresh admission and delivers the original body once", async () => { + const { gate, checks } = observedGate(["match"]); + const records: LinkDiagnosticRecord[] = []; + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "original bytes" }), target, { + tunnel: gate, diagnostic: row => { records.push(row); throw new Error("logging unavailable"); }, + fetchImpl: (async (_url, init) => { + if (++attempts === 1) throw new LinkRelayAuthTimeoutError(); + expect(await new Response(init?.body).text()).toBe("original bytes"); + return new Response("ok"); + }) as typeof fetch, + }); + expect(await response.text()).toBe("ok"); + expect(checks()).toBe(2); + expect(attempts).toBe(2); + expect(records.map(row => [row.event, row.outcome])).toEqual([ + ["relay_retry", "auth_timeout"], ["relay_recovered", "recovered"], + ]); +}); + +test("an upstream 500 after a preflight retry is not reported as recovered", async () => { + const { gate, checks } = observedGate(["match"]); + const records: LinkDiagnosticRecord[] = []; + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST" }), target, { + tunnel: gate, diagnostic: row => records.push(row), + fetchImpl: (async () => { + if (++attempts === 1) throw new LinkRelayAuthTimeoutError(); + return new Response("upstream failed", { status: 500 }); + }) as typeof fetch, + }); + expect(response.status).toBe(500); + expect(await response.text()).toBe("upstream failed"); + expect(checks()).toBe(2); + expect(attempts).toBe(2); + expect(records.map(row => [row.event, row.outcome])).toEqual([["relay_retry", "auth_timeout"]]); +}); + +test.each([false, true])("owner and auth deadlines share a single recovery budget (auth first: %s)", async authFirst => { + const { gate, checks } = observedGate(authFirst ? ["match", "deadline"] : ["deadline", "match"]); + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "original" }), target, { + tunnel: gate, fetchImpl: (async () => { attempts++; throw new LinkRelayAuthTimeoutError(); }) as typeof fetch, + }); + expect(response.status).toBe(503); + expect(checks()).toBe(2); + expect(attempts).toBe(1); +}); + +test.each([new LinkRelayAuthenticationError(), new Error("socket reset after dispatch")])( + "never retries authentication refusal or ambiguous transport failure: %s", async failure => { + const { gate, checks } = observedGate(["match"]); + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST" }), target, { + tunnel: gate, fetchImpl: (async () => { attempts++; throw failure; }) as typeof fetch, + }); + expect(response.status).toBe(503); + expect(checks()).toBe(1); + expect(attempts).toBe(1); + }); + +test("caller cancellation during an owner deadline prevents both recovery and dispatch", async () => { + const abort = new AbortController(); + const { gate, checks } = observedGate(["deadline", "match"], () => abort.abort()); + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", signal: abort.signal }), target, { + tunnel: gate, fetchImpl: (async () => { attempts++; return new Response("unexpected"); }) as typeof fetch, + }); + expect(response.status).toBe(503); + expect(checks()).toBe(1); + expect(attempts).toBe(0); +}); + +test("a real five-second silent preflight recovers on a new socket and nonce without replaying data", async () => { + let challenges = 0, dispatches = 0; + const records: LinkDiagnosticRecord[] = []; + const nonces: string[] = []; + const ports: number[] = []; + const sessions = createLinkRelaySessions({ fingerprints: () => [serviceApiTokenFingerprint(LINK_KEY)] }); + const home = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch(req, server) { + const url = new URL(req.url); + if (url.pathname === LINK_RELAY_AUTH_PATH) { + challenges++; + nonces.push(url.searchParams.get("nonce")!); + ports.push(server.requestIP(req)!.port); + expect(req.headers.has("authorization")).toBe(false); + if (challenges === 1) return new Promise(() => {}); + } + return sessions.dispatch(req, server, async data => { + dispatches++; + expect(await data.text()).toBe("sent exactly once"); + return new Response("recovered"); + }); + } }); + servers.push(home); + const { gate, checks } = observedGate(["match"]); + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "sent exactly once" }), { + ...target, tunnelPort: home.port!, linkId: `lnk_${"a".repeat(16)}`, apiKeyId: "ocx_data_fixture", + }, { tunnel: gate, diagnostic: row => records.push(row) }); + expect(response.status).toBe(200); + expect(await response.text()).toBe("recovered"); + expect(challenges).toBe(2); + expect(new Set(nonces).size).toBe(2); + expect(new Set(ports).size).toBe(2); + expect(checks()).toBe(2); + expect(dispatches).toBe(1); + const retry = records.find(row => row.event === "relay_retry" && row.outcome === "auth_timeout"); + expect(retry?.elapsedMs).toBeGreaterThanOrEqual(4500); + expect(retry?.elapsedMs).toBeLessThan(10_000); +}, 12_000); + + +test("repeated auth deadlines stop after the second preflight without dispatching data", async () => { + const { gate, checks } = observedGate(["match"]); + const records: LinkDiagnosticRecord[] = []; + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: "untouched" }), target, { + tunnel: gate, diagnostic: row => records.push(row), fetchImpl: (async () => { + attempts++; + await Bun.sleep(20); + throw new LinkRelayAuthTimeoutError(); + }) as typeof fetch, + }); + expect(response.status).toBe(503); + expect(checks()).toBe(2); + expect(attempts).toBe(2); + expect(records.map(row => [row.event, row.outcome])).toEqual([ + ["relay_retry", "auth_timeout"], ["relay_failure", "auth_timeout"], + ]); + for (const record of records) expect(record.elapsedMs).toBeGreaterThanOrEqual(15); +}); + +test("concurrent requests keep their recovery budget and correlation separate", async () => { + const records: LinkDiagnosticRecord[] = []; + const first = observedGate(["deadline"]); + const second = observedGate(["deadline"]); + const responses = await Promise.all([first, second].map(({ gate }) => relayLinkDataRequestImpl( + relayRequest({ method: "POST" }), target, { tunnel: gate, diagnostic: row => records.push(row) }, + ))); + const ids = responses.map(response => response.headers.get("x-opencodex-link-request-id")); + expect(new Set(ids).size).toBe(2); + for (const id of ids) { + expect(records.filter(row => row.requestId === id).map(row => [row.event, row.attempt])).toEqual([ + ["relay_retry", 1], ["relay_failure", 1], + ]); + } + expect(first.checks()).toBe(2); + expect(second.checks()).toBe(2); +}); + + +test("an oversized streamed body remains a 413 when the credential-free auth deadline expires", async () => { + const records: LinkDiagnosticRecord[] = []; + let attempts = 0; + const response = await relayLinkDataRequestImpl(relayRequest({ method: "POST", body: new ReadableStream({ + start(controller) { controller.enqueue(new Uint8Array(4)); controller.close(); }, + }) }), target, { + tunnel: connectedTunnel, bodyLimitBytes: 3, diagnostic: row => records.push(row), + fetchImpl: (async () => { + attempts++; + // The cap may prefetch while authentication is pending, before data dispatch starts. + await new Promise(resolve => setTimeout(resolve, 0)); + throw new LinkRelayAuthTimeoutError(); + }) as typeof fetch, + }); + expect(response.status).toBe(413); + expect(await response.json()).toEqual({ error: "link relay request body too large" }); + expect(attempts).toBe(1); + expect(records.some(row => row.event === "relay_retry")).toBe(false); +}); diff --git a/tests/clients/client-link-runtime.test.ts b/tests/clients/client-link-runtime.test.ts index cbe3c033165..d4299b341e0 100644 --- a/tests/clients/client-link-runtime.test.ts +++ b/tests/clients/client-link-runtime.test.ts @@ -8,7 +8,7 @@ import { bindClientListener, scheduleStandaloneRecycle, startClientRuntime } fro import type { MachineListenerDeps } from "../../src/client/machine-listener"; import type { OcxClientConnectionConfig } from "../../src/types"; import { clientLinkStatePath, writeClientLinkState } from "../../src/client/link-state"; -import { fixturePath } from "../helpers/repo-root"; +import { fixturePath, repoPath } from "../helpers/repo-root"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { watchdogMs } from "../helpers/ci-watchdog"; @@ -84,6 +84,77 @@ test("link runtime refuses a busy configured port instead of selecting an epheme } }); +test.each(["SIGINT", "SIGTERM", "restart"])("%s shutdown persists queued diagnostics before process exit", async signal => { + const root = mkdtempSync(join(tmpdir(), "ocx-client-signal-diagnostics-")); + const source = ` + import { mock } from "bun:test"; + import { readFileSync } from "node:fs"; + import { join } from "node:path"; + import { getDefaultConfig, saveConfig } from ${JSON.stringify(repoPath("src/config.ts"))}; + import { startClientRuntime } from ${JSON.stringify(repoPath("src/client/runtime.ts"))}; + import { recordLinkDiagnostic } from ${JSON.stringify(repoPath("src/link/diagnostics.ts"))}; + import { createLocalAttestationChallenge } from ${JSON.stringify(repoPath("src/lib/local-management-attestation.ts"))}; + import { createSystemRestartCapability, SYSTEM_RESTART_CAPABILITY_HEADER, + SYSTEM_RESTART_EXPECTED_PID_HEADER, SYSTEM_RESTART_NONCE_HEADER, SYSTEM_RESTART_PATH + } from ${JSON.stringify(repoPath("src/lib/system-restart-contract.ts"))}; + const record = () => { + recordLinkDiagnostic({ event: "relay_failure", outcome: "auth_failed" }); + recordLinkDiagnostic({ event: "relay_retry", outcome: "deadline" }); + recordLinkDiagnostic({ event: "relay_recovered", outcome: "recovered" }); + }; + if (${JSON.stringify(signal)} === "restart") { + const replacement = await import(${JSON.stringify(repoPath("src/server/restart-replacement.ts"))}); + mock.module(${JSON.stringify(repoPath("src/server/restart-replacement.ts"))}, () => ({ + ...replacement, + spawnReplacementStart: async () => { record(); }, + })); + } + const config = getDefaultConfig(); + config.port = ${freePort()}; + config.runtimeRole = "client"; + config.client = ${JSON.stringify(linkClientState())}; + saveConfig(config); + await startClientRuntime({ block: false }, { restartSupervisor: async () => null }); + if (${JSON.stringify(signal)} === "restart") { + const { attestationSecret, port } = JSON.parse(readFileSync(join(process.env.OPENCODEX_HOME, "runtime-port.json"), "utf8")); + const nonce = createLocalAttestationChallenge(); + const response = await fetch('http://127.0.0.1:' + port + SYSTEM_RESTART_PATH, { + method: "POST", headers: { + Connection: "close", + [SYSTEM_RESTART_EXPECTED_PID_HEADER]: String(process.pid), + [SYSTEM_RESTART_NONCE_HEADER]: nonce, + [SYSTEM_RESTART_CAPABILITY_HEADER]: createSystemRestartCapability(attestationSecret, nonce, "POST", SYSTEM_RESTART_PATH, process.pid, port), + }, + }); + await response.arrayBuffer(); + if (response.status !== 202) throw new Error('restart refused: ' + response.status); + } else { + record(); + process.emit(${JSON.stringify(signal)}, ${JSON.stringify(signal)}); + } + `; + const child = Bun.spawn([process.execPath, "-e", source], { + cwd: root, + env: { ...process.env, HOME: root, USERPROFILE: root, CODEX_HOME: join(root, "codex"), + OPENCODEX_HOME: join(root, "ocx"), OCX_SERVICE_MANAGED: "0" }, + stdout: "ignore", stderr: "pipe", + }); + const errors = new Response(child.stderr).text(); + try { + const exit = await waitFor(() => child.exitCode, "signal shutdown"); + expect({ exit, stderr: await errors }).toEqual({ exit: 0, stderr: "" }); + const records = readFileSync(join(root, "ocx", "link", "diagnostics.jsonl"), "utf8") + .trim().split("\n").map(line => JSON.parse(line)); + expect(records.map(({ event }) => event)).toEqual(["relay_failure", "relay_retry", "relay_recovered"]); + expect(existsSync(join(root, "ocx", "runtime-port.json"))).toBe(false); + } finally { + if (child.exitCode === null) child.kill(); + await child.exited; + await errors; + removeTreeWithRetry(root); + } +}, watchdogMs(60_000)); + test("an ended link recycles a connected sibling after listener cleanup", async () => { const root = mkdtempSync(join(tmpdir(), "ocx-client-sibling-recycle-")); const home = join(root, "ocx"); diff --git a/tests/clients/client-link-tunnel.test.ts b/tests/clients/client-link-tunnel.test.ts index aee48f0ddc4..d87ce260ea1 100644 --- a/tests/clients/client-link-tunnel.test.ts +++ b/tests/clients/client-link-tunnel.test.ts @@ -455,7 +455,7 @@ function supervisorHarness(overrides: Partial = {}) { random: () => 0.5, readProcessStartTime: pid => `fixture-start-${pid}`, linkKey: () => LINK_KEY, - ownsLoopbackListener: async (_port, pid) => ssh.live().some(item => item.child.pid === pid), + inspectLoopbackListener: async (_port, pid) => ({ kind: ssh.live().some(item => item.child.pid === pid) ? "match" : "mismatch", elapsedMs: 1 }), fetchImpl: (async (input, init) => { probes.push({ url: String(input), keyed: new Headers(init?.headers).get("x-opencodex-api-key") === LINK_KEY, cache: init?.cache }); if (view.readyz === "refused") throw Object.assign(new Error("Unable to connect"), { code: "ConnectionRefused" }); @@ -503,7 +503,7 @@ test("a competing listener cannot receive the probe key or relayed traffic befor } }); const h = supervisorHarness({ fetchImpl: fetch, - ownsLoopbackListener: async () => false, + inspectLoopbackListener: async () => ({ kind: "mismatch", elapsedMs: 1 }), }); h.view.sidecar = { ...sidecar(), tunnelPort: competing.port! }; try { @@ -513,7 +513,7 @@ test("a competing listener cannot receive the probe key or relayed traffic befor method: "POST", headers: { "Content-Type": "application/json" }, body: '{"input":"private"}', }), { tunnelPort: competing.port!, admissionKey: LINK_KEY }, { tunnel: h.supervisor, holdMs: 10, fetchImpl: fetch }); expect(response.status).toBe(503); - expect(await h.supervisor.connected()).toBe(false); + expect(await h.supervisor.connected()).toMatchObject({ kind: "unavailable" }); expect(received).toEqual([]); } finally { await h.close(); @@ -824,7 +824,7 @@ test("an adopted tunnel is probed only when pidfile identity and socket owner ag platform: "darwin", isAlive: pid => pid === 42 || pid === process.pid, readProcessInfo: pid => pid === 42 ? { ppid: process.pid, args: argv.join(" "), startTime: "adopted-start" } : null, - ownsLoopbackListener: async (_port, pid) => pid === 42, + inspectLoopbackListener: async (_port, pid) => ({ kind: pid === 42 ? "match" : "mismatch", elapsedMs: 1 }), }); mkdirSync(join(h.configDir, "link"), { recursive: true }); writeFileSync(clientTunnelPidfilePath(h.configDir), JSON.stringify({ version: 1, @@ -833,7 +833,7 @@ test("an adopted tunnel is probed only when pidfile identity and socket owner ag await h.start(); expect(h.ssh.children).toHaveLength(0); await h.step(); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); expect(h.probes).toHaveLength(1); expect(h.probes[0]?.keyed).toBe(true); } finally { @@ -853,7 +853,7 @@ test("one unreadable adopted identity denies only this admission and recovers", if (unreadable) { unreadable = false; return null; } return { ppid: process.pid, args: argv.join(" "), startTime: "adopted-start" }; }, - ownsLoopbackListener: async (_port, pid) => pid === 42, + inspectLoopbackListener: async (_port, pid) => ({ kind: pid === 42 ? "match" : "mismatch", elapsedMs: 1 }), }); mkdirSync(join(h.configDir, "link"), { recursive: true }); writeFileSync(clientTunnelPidfilePath(h.configDir), JSON.stringify({ version: 1, @@ -868,7 +868,7 @@ test("one unreadable adopted identity denies only this admission and recovers", try { await h.start(); await h.step(); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); unreadable = true; expect((await relay()).status).toBe(503); expect(forwarded).toBe(0); @@ -893,7 +893,7 @@ test("an adopted PID with changed start identity cannot receive another probe or isAlive: pid => pid === 42 || pid === process.pid, readProcessInfo: pid => pid === 42 ? { ppid: process.pid, args: argv.join(" "), startTime: currentStart } : null, - ownsLoopbackListener: async (_port, pid) => pid === 42 || pid >= 31_000, + inspectLoopbackListener: async (_port, pid) => ({ kind: pid === 42 || pid >= 31_000 ? "match" : "mismatch", elapsedMs: 1 }), signal: (_pid, signal) => { signals.push(signal); }, }); mkdirSync(join(h.configDir, "link"), { recursive: true }); @@ -903,7 +903,7 @@ test("an adopted PID with changed start identity cannot receive another probe or try { await h.start(); await h.step(); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); expect(h.probes).toHaveLength(1); currentStart = "reused-start"; h.view.clock += 1_500; @@ -928,13 +928,13 @@ test("an adopted PID with changed start identity cannot receive another probe or test("a connected tunnel stops relaying when another PID takes its port", async () => { let owners: number[] = []; - const h = supervisorHarness({ ownsLoopbackListener: async (_port, pid) => owners.includes(pid) }); + const h = supervisorHarness({ inspectLoopbackListener: async (_port, pid) => ({ kind: owners.includes(pid) ? "match" : "mismatch", elapsedMs: 1 }) }); let forwarded = 0; try { await h.start(); owners = [h.ssh.children[0]!.child.pid]; await h.step(); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); owners = [process.pid]; const response = await relayLinkDataRequest(new Request("http://127.0.0.1:10100/v1/responses", { method: "POST", body: '{"input":"private"}', @@ -944,7 +944,7 @@ test("a connected tunnel stops relaying when another PID takes its port", async }); expect(response.status).toBe(503); expect(forwarded).toBe(0); - expect(await h.supervisor.connected()).toBe(false); + expect(await h.supervisor.connected()).toMatchObject({ kind: "mismatch" }); } finally { await h.close(); } @@ -953,7 +953,7 @@ test("a connected tunnel stops relaying when another PID takes its port", async test("each relayed request obtains a fresh asynchronous owner proof", async () => { let lookups = 0; const h = supervisorHarness({ - ownsLoopbackListener: async () => { lookups += 1; await Bun.sleep(1); return true; }, + inspectLoopbackListener: async () => { lookups += 1; await Bun.sleep(1); return { kind: "match", elapsedMs: 1 }; }, }); const relay = () => relayLinkDataRequest(new Request("http://127.0.0.1:10100/v1/models"), { tunnelPort: sidecar().tunnelPort, admissionKey: LINK_KEY }, { @@ -977,7 +977,7 @@ test("each relayed request obtains a fresh asynchronous owner proof", async () = test("a replacement SSH child needs a new ownership proof before its first keyed probe", async () => { let lookups = 0; - const h = supervisorHarness({ ownsLoopbackListener: async () => { lookups += 1; return true; } }); + const h = supervisorHarness({ inspectLoopbackListener: async () => { lookups += 1; return { kind: "match", elapsedMs: 1 }; } }); try { await h.start(); await h.step(); @@ -1040,19 +1040,19 @@ test("an unreadable read does not end the link; three in a row or an explicit di test("requests wait on a reconnect only while it lasts and are released when it connects or fails", async () => { const h = supervisorHarness(); try { - expect(await h.supervisor.connected()).toBe(false); + expect(await h.supervisor.connected()).toMatchObject({ kind: "unavailable" }); expect(h.supervisor.pending()).toBe(false); await h.start(); expect(h.supervisor.pending()).toBe(true); await h.step(); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); expect(h.supervisor.pending()).toBe(false); expect(await h.supervisor.waitForConnected(15_000)).toBe(true); h.view.readyz = "refused"; h.ssh.children[0]!.exit(255, "Connection reset by peer"); await h.settle(); - expect(await h.supervisor.connected()).toBe(false); + expect(await h.supervisor.connected()).toMatchObject({ kind: "unavailable" }); expect(h.supervisor.pending()).toBe(true); let released: boolean | undefined; const waiting = h.supervisor.waitForConnected(15_000).then(value => { released = value; return value; }); @@ -1062,7 +1062,7 @@ test("requests wait on a reconnect only while it lasts and are released when it h.view.readyz = 200; await h.step(); expect(await waiting).toBe(true); - expect(await h.supervisor.connected()).toBe(true); + expect(await h.supervisor.connected()).toMatchObject({ kind: "match" }); h.ssh.children[1]!.exit(255, "Connection reset by peer"); await h.settle(); @@ -1072,7 +1072,7 @@ test("requests wait on a reconnect only while it lasts and are released when it await h.settle(); expect(await failing).toBe(false); expect(h.supervisor.status()).toMatchObject({ state: { kind: "failed", reason: "hostkey" } }); - expect(await h.supervisor.connected()).toBe(false); + expect(await h.supervisor.connected()).toMatchObject({ kind: "unavailable" }); expect(h.supervisor.pending()).toBe(false); expect(await h.supervisor.waitForConnected(15_000)).toBe(false); } finally { @@ -1124,3 +1124,153 @@ test("the supervisor's own reads parse a file again only after it changed", () = rmSync(dir, { recursive: true, force: true }); } }); + +for (const kind of ["match", "mismatch", "deadline"] as const) { + test(`fresh ${kind} owner proof reports correlated timing and admits only a match`, async () => { + let result: Awaited>> = { kind: "match", elapsedMs: 1 }; + const checks: import("../../src/client/link-relay").LinkAdmissionResult[] = []; + const events: import("../../src/link/diagnostics").LinkDiagnosticRecord[] = []; + const h = supervisorHarness({ inspectLoopbackListener: async () => result, diagnostic: event => { events.push(event); } }); + try { + await h.start(); + await h.step(); + expect(h.state()).toBe("connected"); + events.length = 0; + result = { kind, elapsedMs: 2_003.25 }; + for (let request = 0; request < 2; request += 1) { + checks.push(await h.supervisor.connected({ requestId: `request-${request}` })); + } + expect(checks).toHaveLength(2); + expect(events).toHaveLength(kind === "match" ? 0 : 2); + for (const [index, check] of checks.entries()) { + expect(check).toMatchObject({ kind, elapsedMs: 2_003.25, state: "connected" }); + expect(check.checkId).toEqual(expect.any(String)); + expect(check.checkId!.length).toBeGreaterThan(0); + expect(Number.isSafeInteger(check.generation)).toBe(true); + expect(check.generation!).toBeGreaterThan(0); + if (kind !== "match") expect(events[index]).toEqual({ event: "owner_lookup", outcome: kind, requestId: `request-${index}`, + checkId: check.checkId, generation: check.generation, state: "connected", elapsedMs: 2_003.25 }); + } + expect(checks[1]!.checkId).not.toBe(checks[0]!.checkId); + expect(checks[1]!.generation).toBe(checks[0]!.generation); + } finally { + await h.close(); + } + }); +} + +for (const transition of ["stop", "replacement", "sidecar failure"] as const) { + for (const kind of ["match", "deadline"] as const) { + test(`a delayed ${kind} after ${transition} is stale and cannot admit or report a deadline`, async () => { + type OwnerResult = Awaited>>; + const delayed = deferred(); + const entered = deferred(); + let delayNext = false; + const events: import("../../src/link/diagnostics").LinkDiagnosticRecord[] = []; + const h = supervisorHarness({ + inspectLoopbackListener: async () => { + if (delayNext) { + delayNext = false; + entered.resolve(); + return delayed.promise; + } + return { kind: "match", elapsedMs: 1 }; + }, + diagnostic: event => { events.push(event); }, + }); + try { + await h.start(); + await h.step(); + expect(h.state()).toBe("connected"); + const baseline = await h.supervisor.connected(); + expect(baseline.kind).toBe("match"); + events.length = 0; + delayNext = true; + const admission = h.supervisor.connected({ requestId: "delayed-request" }); + await entered.promise; + if (transition === "stop") { + await h.supervisor.stop(); + expect(h.state()).toBe("idle"); + } else if (transition === "sidecar failure") { + h.view.sidecar = "unreadable"; + await h.step(); + await h.step(); + expect(h.state()).toBe("connected"); + await h.step(); + expect(h.supervisor.status()).toEqual({ kind: "failed", reason: "sidecar_invalid" }); + } else { + h.ssh.children[0]!.exit(255, "Connection reset by peer"); + await h.settle(); + for (let tick = 0; tick < 10 && h.ssh.children.length < 2; tick += 1) await h.step(); + expect(h.ssh.children).toHaveLength(2); + await h.step(); + expect(h.state()).toBe("connected"); + } + delayed.resolve({ kind, elapsedMs: 2_004.5 }); + const check = await admission; + expect(check).toMatchObject({ kind: "stale", elapsedMs: 2_004.5, + generation: baseline.generation, state: transition === "replacement" ? "connected" : "idle" }); + expect(check.checkId).not.toBe(baseline.checkId); + expect(events.filter(event => event.requestId === "delayed-request")).toEqual([ + { event: "owner_lookup", outcome: "stale", requestId: "delayed-request", checkId: check.checkId, + generation: baseline.generation, state: transition === "replacement" ? "connected" : "idle", elapsedMs: 2_004.5 }, + ]); + if (transition === "replacement") { + const fresh = await h.supervisor.connected(); + expect(fresh.kind).toBe("match"); + expect(fresh.generation!).toBeGreaterThan(check.generation!); + } + } finally { + delayed.resolve({ kind: "match", elapsedMs: 1 }); + await h.close(); + } + }); + } +} + +for (const kind of ["match", "mismatch", "deadline"] as const) { + test(`throwing diagnostic sinks preserve ${kind} admission`, async () => { + let result: "match" | "mismatch" | "deadline" = "match"; + let diagnostics = 0; + const h = supervisorHarness({ + inspectLoopbackListener: async () => ({ kind: result, elapsedMs: 7 }), + diagnostic: () => { diagnostics += 1; throw new Error("diagnostic sink unavailable"); }, + }); + try { + await h.start(); + await h.step(); + expect(h.state()).toBe("connected"); + diagnostics = 0; + result = kind; + expect(await h.supervisor.connected({ requestId: "throwing-sinks" })).toMatchObject({ kind, elapsedMs: 7 }); + expect(diagnostics).toBe(kind === "match" ? 0 : 1); + expect(h.state()).toBe("connected"); + } finally { + await h.close(); + } + }); +} + +test("a stop triggered by failure diagnostics leaves admission closed", async () => { + let result: "match" | "mismatch" = "match"; + let stopping: Promise | undefined; + const h = supervisorHarness({ + inspectLoopbackListener: async () => ({ kind: result, elapsedMs: 1 }), + diagnostic: () => { stopping = h.supervisor.stop(); }, + }); + try { + await h.start(); + await h.step(); + expect(h.state()).toBe("connected"); + expect(stopping).toBeUndefined(); + result = "mismatch"; + expect(await h.supervisor.connected()).toMatchObject({ kind: "mismatch" }); + expect(stopping).toBeDefined(); + await stopping; + expect(h.state()).toBe("idle"); + expect(await h.supervisor.connected()).toMatchObject({ kind: "unavailable" }); + } finally { + await stopping; + await h.close(); + } +}); diff --git a/tests/clients/client-machine-listener.test.ts b/tests/clients/client-machine-listener.test.ts index 2b4ab69e3d5..3ef6a52179c 100644 --- a/tests/clients/client-machine-listener.test.ts +++ b/tests/clients/client-machine-listener.test.ts @@ -363,7 +363,7 @@ describe("client machine listener in link mode", () => { const server = startMachineListener(0, { state: linkConnection(options.fingerprint), managementAuthState: authState(), - linkTunnel: { connected: () => true, pending: () => false, waitForConnected: async () => true }, + linkTunnel: { connected: () => ({ kind: "match" }), pending: () => false, waitForConnected: async () => true }, fetchImpl: (async (input, init) => { upstream.push(new Request(String(input), init)); return options.reply?.() ?? Response.json({ relayed: true }); diff --git a/tests/clients/client-runtime.test.ts b/tests/clients/client-runtime.test.ts index 10463ddc30b..45da00f7d8a 100644 --- a/tests/clients/client-runtime.test.ts +++ b/tests/clients/client-runtime.test.ts @@ -1,8 +1,10 @@ import { describe, expect, test } from "bun:test"; -import { mkdtempSync, writeFileSync } from "node:fs"; +import { mkdtempSync, readFileSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { serviceApiTokenFingerprint } from "../../src/lib/service-secrets"; +import { recordLinkDiagnostic } from "../../src/link/diagnostics"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; import { recycleStandalone, standaloneRecycleEnv } from "../../src/client/runtime"; describe("standalone recycle environment", () => { @@ -129,6 +131,36 @@ describe("standalone recycle", () => { }); }); + test("persists queued diagnostics before a supervised recycle exits", async () => { + const dir = mkdtempSync(join(tmpdir(), "ocx-recycle-diagnostics-")); + const previousHome = process.env.OPENCODEX_HOME; + process.env.OPENCODEX_HOME = dir; + try { + await withServiceMarker("1", async () => { + recordLinkDiagnostic({ event: "relay_failure", outcome: "auth_failed" }); + recordLinkDiagnostic({ event: "relay_retry", outcome: "deadline" }); + recordLinkDiagnostic({ event: "relay_recovered", outcome: "recovered" }); + let exited = false; + await recycleStandalone(serviceApiTokenFingerprint("disconnected-hub-token"), { + configuredPort: () => 10456, + isDesktopSupervised: () => false, + exitProcess: code => { + expect(code).toBe(1); + const records = readFileSync(join(dir, "link", "diagnostics.jsonl"), "utf8") + .trim().split("\n").map(line => JSON.parse(line)); + expect(records.map(({ event }) => event)).toEqual(["relay_failure", "relay_retry", "relay_recovered"]); + exited = true; + }, + }); + expect(exited).toBe(true); + }); + } finally { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(dir); + } + }); + test("a supervised client still exits 1 without spawning", async () => { await withServiceMarker("1", async () => { const calls: string[] = []; diff --git a/tests/clients/link-diagnostics.test.ts b/tests/clients/link-diagnostics.test.ts new file mode 100644 index 00000000000..53cea0da6cc --- /dev/null +++ b/tests/clients/link-diagnostics.test.ts @@ -0,0 +1,296 @@ +import { afterEach, expect, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, renameSync, statSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createLinkDiagnosticSink, type LinkDiagnosticRecord } from "../../src/link/diagnostics"; +import { resetHardenedStateForTests, setAsyncIcaclsRunnerForTests, setIcaclsRunnerForTests, + setPlatformForTests, setStatForTests } from "../../src/lib/windows-secret-acl"; +import { setAsyncWindowsPrincipalRunnerForTests, resetWindowsPrincipalForTests } from "../../src/lib/windows-user-principal"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) removeTreeWithRetry(root); }); +function fixture() { + const configDir = mkdtempSync(join(tmpdir(), "ocx-link-diagnostics-")); + roots.push(configDir); + const dir = join(configDir, "link"); + const path = join(dir, "diagnostics.jsonl"); + return { configDir, dir, path }; +} +function records(path: string): Record[] { + return readFileSync(path, "utf8").trim().split("\n").map(line => JSON.parse(line)); +} + +test("writes only bounded metadata and opaque UUID correlation without config ownership side effects", async () => { + const { configDir, path } = fixture(); + const sink = createLinkDiagnosticSink({ configDir, now: () => 100 }); + expect(readdirSync(configDir)).toEqual([]); + const requestId = randomUUID(); + const checkId = randomUUID(); + sink({ + event: "owner_lookup", outcome: "match", requestId, checkId, + generation: 3, state: "connected", elapsedMs: 1.5, attempt: 2, + error: "private-error", body: "private-body", key: "private-key", url: "https://private.test", + linkId: "lnk_0123456789abcdef", suppressed: "private", timestamp: "private", + } as LinkDiagnosticRecord); + sink({ event: "relay_failure", outcome: "auth_timeout", requestId: "lnk_0123456789abcdef", + checkId: `${checkId}\nprivate`, state: "private-state", elapsedMs: Infinity, generation: -1, attempt: 0.5 }); + sink({ event: "private-event", outcome: "match" } as unknown as LinkDiagnosticRecord); + sink({ event: "owner_lookup", outcome: "private-outcome" } as unknown as LinkDiagnosticRecord); + await sink.flush(); + expect(records(path)).toEqual([ + { event: "owner_lookup", outcome: "match", requestId, checkId, generation: 3, + state: "connected", elapsedMs: 1.5, attempt: 2, timestamp: 100 }, + { event: "relay_failure", outcome: "auth_timeout", timestamp: 100 }, + ]); + expect(readFileSync(path, "utf8")).not.toContain("private"); + expect(readdirSync(configDir)).toEqual(["link"]); +}); + +test("deduplicates across changing IDs and reports suppression on the next eligible write", async () => { + const { configDir, path } = fixture(); + let now = 0; + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + const requestId = randomUUID(); + const checkId = randomUUID(); + sink({ event: "owner_lookup", outcome: "match", requestId, checkId }); + now = 999; + for (let i = 0; i < 50; i++) sink({ event: "owner_lookup", outcome: "match", requestId: randomUUID(), generation: i }); + sink({ event: "owner_lookup", outcome: "deadline" }); + sink({ event: "relay_retry", outcome: "match" }); + await sink.flush(); + expect(records(path)).toHaveLength(3); + now = 1_000; + sink({ event: "owner_lookup", outcome: "match" }); + now = 2_000; + sink({ event: "owner_lookup", outcome: "match" }); + await sink.flush(); + const written = records(path); + expect(written[0]).toEqual({ event: "owner_lookup", outcome: "match", requestId, checkId, timestamp: 0 }); + expect(written[3]).toEqual({ event: "owner_lookup", outcome: "match", timestamp: 1_000, suppressed: 50 }); + expect(written[4]).toEqual({ event: "owner_lookup", outcome: "match", timestamp: 2_000 }); +}); + +test("a backward clock step starts a new diagnostic window and preserves suppression counts", async () => { + const { configDir, path } = fixture(); + let now = 10_000; + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + const record: LinkDiagnosticRecord = { event: "relay_failure", outcome: "auth_timeout" }; + sink(record); + now = 10_500; + sink(record); + now = 5_000; + sink(record); + now = 5_500; + sink(record); + now = 6_000; + sink(record); + await sink.flush(); + expect(records(path)).toEqual([ + { ...record, timestamp: 10_000 }, + { ...record, timestamp: 5_000, suppressed: 1 }, + { ...record, timestamp: 6_000, suppressed: 1 }, + ]); +}); + +test("keeps private modes and rotates at 256 KiB with only one bounded backup", async () => { + const { configDir, dir, path } = fixture(); + let now = 0; + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + sink({ event: "owner_lookup", outcome: "match" }); + await sink.flush(); + if (process.platform !== "win32") { + expect(statSync(dir).mode & 0o777).toBe(0o700); + expect(statSync(path).mode & 0o777).toBe(0o600); + chmodSync(dir, 0o755); + chmodSync(path, 0o644); + } + // Seed the boundary directly instead of writing megabytes through the sink. + const seed = `${" ".repeat(256 * 1024 - 1)}\n`; + writeFileSync(path, seed); + writeFileSync(`${path}.1`, "old backup\n"); + now = 1_000; + sink({ event: "owner_lookup", outcome: "match" }); + await sink.flush(); + expect(readFileSync(`${path}.1`, "utf8")).toBe(seed); + expect(records(path)).toEqual([{ event: "owner_lookup", outcome: "match", timestamp: now }]); + writeFileSync(path, seed); + now = 2_000; + sink({ event: "relay_failure", outcome: "transport_error" }); + await sink.flush(); + expect(readdirSync(dir).sort()).toEqual(["diagnostics.jsonl", "diagnostics.jsonl.1"]); + expect(statSync(path).size).toBeLessThanOrEqual(256 * 1024); + expect(statSync(`${path}.1`).size).toBeLessThanOrEqual(256 * 1024); + expect(statSync(path).size + statSync(`${path}.1`).size).toBeLessThanOrEqual(512 * 1024); + if (process.platform !== "win32") { + expect(statSync(dir).mode & 0o777).toBe(0o700); + expect(statSync(path).mode & 0o777).toBe(0o600); + expect(statSync(`${path}.1`).mode & 0o777).toBe(0o600); + } +}); + +test("filesystem and clock failures never escape the sink", async () => { + const { configDir, dir, path } = fixture(); + let now = 0; + writeFileSync(dir, "not a directory"); + const record: LinkDiagnosticRecord = { event: "relay_failure", outcome: "transport_error" }; + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + expect(() => sink(record)).not.toThrow(); + await sink.flush(); + expect(existsSync(path)).toBe(false); + expect(() => createLinkDiagnosticSink({ configDir, now: () => { throw new Error("private"); } })(record)).not.toThrow(); + expect(readFileSync(dir, "utf8")).toBe("not a directory"); + unlinkSync(dir); + now = 999; + sink(record); + await sink.flush(); + expect(existsSync(path)).toBe(false); + now = 1_000; + sink(record); + await sink.flush(); + expect(records(path)).toEqual([{ ...record, timestamp: 1_000, suppressed: 1 }]); +}); + +test.skipIf(process.platform === "win32")("refuses symlink logs without modifying their targets", async () => { + const { configDir, dir, path } = fixture(); + const target = join(configDir, "foreign"); + mkdirSync(dir); + writeFileSync(target, "preserve"); + symlinkSync(target, path); + const sink = createLinkDiagnosticSink({ configDir }); + expect(() => sink({ event: "owner_lookup", outcome: "match" })).not.toThrow(); + await sink.flush(); + expect(readFileSync(target, "utf8")).toBe("preserve"); +}); + + +test("Windows diagnostic appends reuse ACLs only until the file or its permissions change", async () => { + const { configDir, path } = fixture(); + let now = 0, grants = 0; + let permissionFreshness: bigint | undefined; + const observedStat = (target: string) => { + const { dev, ino, ctimeNs } = statSync(target, { bigint: true }); + return { dev, ino, ctimeNs: target === path ? permissionFreshness ?? ctimeNs : ctimeNs }; + }; + setPlatformForTests("win32"); + resetHardenedStateForTests(); + setStatForTests(observedStat); + setAsyncWindowsPrincipalRunnerForTests(async () => ({ success: true, exitCode: 0, timedOut: false, + stdout: "S-1-5-21-1-2-3-1001\nTEST\\user\n" })); + setAsyncIcaclsRunnerForTests(async args => { + if (args[0] === path && args.includes("/grant:r")) grants++; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + const append = async () => { + sink({ event: "relay_failure", outcome: "auth_timeout" }); + await sink.flush(); + now += 1_000; + }; + try { + await append(); + expect(grants).toBe(1); + await append(); + expect(records(path)).toHaveLength(2); + expect(grants).toBe(1); + + // Keep the old object alive so replacing the path cannot recycle its inode. + renameSync(path, `${path}.old`); + writeFileSync(path, ""); + await append(); + expect(records(path)).toHaveLength(1); + expect(grants).toBe(2); + await append(); + expect(grants).toBe(2); + + const beforePermissions = observedStat(path); + chmodSync(path, 0o400); + chmodSync(path, 0o600); + // Model an observable permission change regardless of the host filesystem's ctime granularity. + permissionFreshness = beforePermissions.ctimeNs + 1n; + const afterPermissions = observedStat(path); + expect(afterPermissions.dev).toBe(beforePermissions.dev); + expect(afterPermissions.ino).toBe(beforePermissions.ino); + expect(afterPermissions.ctimeNs).not.toBe(beforePermissions.ctimeNs); + await append(); + expect(records(path)).toHaveLength(3); + expect(grants).toBe(3); + } finally { + await sink.flush(); + setAsyncIcaclsRunnerForTests(null); + setPlatformForTests(null); + setStatForTests(null); + setAsyncWindowsPrincipalRunnerForTests(null); + resetWindowsPrincipalForTests(); + resetHardenedStateForTests(); + } +}); + +test.each([false, true])("slow Windows ACL work yields and bounds the queue (clock rolls back: %s)", async rollback => { + const { configDir, path } = fixture(); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const blocked = new Promise(resolve => { release = resolve; }); + let first = true, now = 0, syncCalls = 0; + setPlatformForTests("win32"); + resetHardenedStateForTests(); + setAsyncWindowsPrincipalRunnerForTests(async () => ({ success: true, exitCode: 0, timedOut: false, + stdout: "S-1-5-21-1-2-3-1001\nTEST\\user\n" })); + setIcaclsRunnerForTests(() => { syncCalls++; throw new Error("synchronous ACL work is forbidden"); }); + setAsyncIcaclsRunnerForTests(async () => { + if (first) { first = false; enter(); await blocked; } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + const sink = createLinkDiagnosticSink({ configDir, now: () => now }); + try { + sink({ event: "owner_lookup", outcome: "match", attempt: 0 }); + await entered; + for (let attempt = 1; attempt <= 100; attempt++) { + now += 1_000; + sink({ event: "owner_lookup", outcome: "match", attempt }); + } + // A deadline must still be able to fire while the diagnostics child is stalled. + await new Promise(resolve => setTimeout(resolve, 0)); + expect(existsSync(path)).toBe(false); + expect(syncCalls).toBe(0); + if (rollback) { + now = 5_000; + sink({ event: "owner_lookup", outcome: "match", attempt: 101 }); + } + // Shutdown must return even while the ACL runner is still blocked. + let watchdog: ReturnType | undefined; + try { + expect(await Promise.race([ + sink.flush(10).then(() => "drained"), + new Promise(resolve => { watchdog = setTimeout(() => resolve("blocked"), 500); }), + ])).toBe("drained"); + expect(existsSync(path)).toBe(false); + } finally { + clearTimeout(watchdog); + } + release(); + await sink.flush(); + expect(records(path)).toHaveLength(57); // one active record plus at most 56 queued + if (rollback) { + now += 500; + sink({ event: "owner_lookup", outcome: "match", attempt: 102 }); + await sink.flush(); + expect(records(path)).toHaveLength(57); + } + now += rollback ? 500 : 1_000; + const attempt = rollback ? 103 : 101; + sink({ event: "owner_lookup", outcome: "match", attempt }); + await sink.flush(); + expect(records(path).at(-1)).toMatchObject({ attempt, timestamp: now, suppressed: rollback ? 46 : 44 }); + } finally { + release(); + await sink.flush(); + setAsyncIcaclsRunnerForTests(null); + setIcaclsRunnerForTests(null); + setPlatformForTests(null); + setAsyncWindowsPrincipalRunnerForTests(null); + resetWindowsPrincipalForTests(); + resetHardenedStateForTests(); + } +}); diff --git a/tests/clients/link-relay-bound-transport.test.ts b/tests/clients/link-relay-bound-transport.test.ts index 780f5e9213e..27803e5d1ce 100644 --- a/tests/clients/link-relay-bound-transport.test.ts +++ b/tests/clients/link-relay-bound-transport.test.ts @@ -1,8 +1,8 @@ import { describe, expect, test } from "bun:test"; -import { Agent, request as httpRequest } from "node:http"; +import { Agent, createServer, request as httpRequest, type RequestListener } from "node:http"; import type { Server } from "bun"; import { gzipSync } from "node:zlib"; -import { fetchBoundLinkRelay, LinkRelayAuthenticationError } from "../../src/client/link-relay-transport"; +import { fetchBoundLinkRelay, LinkRelayAuthenticationError, LinkRelayAuthTimeoutError } from "../../src/client/link-relay-transport"; import { relayLinkDataRequest, HOME_INITIATED_LINK_TUNNEL } from "../../src/client/link-relay"; import { createLinkRelaySessions } from "../../src/server/index/link-relay-sessions"; import { createLinkListenerLifecycle } from "../../src/server/index/link-listener"; @@ -43,6 +43,19 @@ async function rawGet(url: URL, agent: Agent, headers?: Record) }); } +async function withHttpPeer(handler: RequestListener, run: (port: number) => Promise) { + const server = createServer(handler); + try { + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("missing peer port"); + await run(address.port); + } finally { + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + } +} + describe("connection-bound link relay", () => { test("direction and both identities are authenticated", () => { const nonce = "c".repeat(64); @@ -108,16 +121,93 @@ describe("connection-bound link relay", () => { test("an unrecognized or invalid peer gets no credentials or body, and missing identity refuses locally", async () => { const requests: Request[] = []; - const home = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: req => { requests.push(req); return new Response(null, { status: 404 }); } }); + const home = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: req => { + requests.push(req); + return new Response(null, requests.length === 1 ? { status: 404 } + : { status: 204, headers: { "x-opencodex-link-proof": "0".repeat(64) } }); + } }); try { await expect(fetchBoundLinkRelay(target(home.port!), `${home.url}v1/models`, {})).rejects.toBeInstanceOf(LinkRelayAuthenticationError); expect(requests).toHaveLength(1); expect(requests[0]!.headers.has("authorization")).toBe(false); expect(requests[0]!.method).toBe("GET"); expect(requests[0]!.body).toBeNull(); + await expect(fetchBoundLinkRelay(target(home.port!), `${home.url}v1/models`, {})).rejects.toBeInstanceOf(LinkRelayAuthenticationError); + expect(requests).toHaveLength(2); + expect(requests[1]!.headers.has("authorization")).toBe(false); expect(requests[1]!.body).toBeNull(); await expect(fetchBoundLinkRelay({ tunnelPort: home.port!, admissionKey: key }, `${home.url}v1/models`, {})).rejects.toThrow(); - expect(requests).toHaveLength(1); + expect(requests).toHaveLength(2); } finally { await home.stop(true); } }); + test("a silent actual peer expires only the credential-free auth deadline without touching the body", async () => { + const seen: Array<{ url: string; authorization: string | undefined; body: string }> = []; + let bodyRead = false, bodyCancelled = false; + const body = new ReadableStream({ + pull() { bodyRead = true; }, cancel() { bodyCancelled = true; }, + }, { highWaterMark: 0 }); + await withHttpPeer(req => { + const record = { url: req.url!, authorization: req.headers.authorization, body: "" }; + seen.push(record); + req.on("data", chunk => { record.body += chunk.toString(); }); + }, async port => { + await expect(fetchBoundLinkRelay(target(port), `http://127.0.0.1:${port}/v1/responses`, { + method: "POST", headers: { authorization: `Bearer ${key}` }, body, + }, { authTimeoutMs: 20 })).rejects.toBeInstanceOf(LinkRelayAuthTimeoutError); + expect(seen).toHaveLength(1); + expect(new URL(seen[0]!.url, "http://peer").pathname).toBe(LINK_RELAY_AUTH_PATH); + expect(seen[0]!.authorization).toBeUndefined(); expect(seen[0]!.body).toBe(""); + expect(bodyRead).toBe(false); expect(bodyCancelled).toBe(false); expect(body.locked).toBe(false); + }); + }); + + test("parent abort during a real auth probe is not an auth deadline", async () => { + const parent = new AbortController(); + let probes = 0; + await withHttpPeer(() => { probes++; parent.abort(new DOMException("caller stopped", "AbortError")); }, async port => { + const error = await fetchBoundLinkRelay(target(port), `http://127.0.0.1:${port}/v1/models`, { + signal: parent.signal, + }, { authTimeoutMs: 40 }).catch(error => error); + expect(error).toBeInstanceOf(Error); + expect(error).not.toBeInstanceOf(LinkRelayAuthTimeoutError); + expect(probes).toBe(1); + }); + }); + + for (const failure of ["header timeout", "reset"] as const) { + test(`post-proof data ${failure} is never a preflight auth timeout`, async () => { + const parent = new AbortController(); + let timer: ReturnType | undefined; + let probes = 0, dataRequests = 0; + await withHttpPeer((req, res) => { + const url = new URL(req.url!, "http://peer"); + if (url.pathname === LINK_RELAY_AUTH_PATH) { + probes++; + res.writeHead(204, { "x-opencodex-link-proof": linkRelayProof(fp, "listener", keyId, linkId, + url.searchParams.get("nonce")!)! }); + res.end(); return; + } + dataRequests++; + if (failure === "reset") req.socket.destroy(); + else timer = setTimeout(() => parent.abort(new DOMException("data headers timed out", "TimeoutError")), 40); + }, async port => { + try { + const error = await fetchBoundLinkRelay(target(port), `http://127.0.0.1:${port}/v1/models`, { + headers: { authorization: `Bearer ${key}` }, signal: parent.signal, + }, { authTimeoutMs: 20 }).catch(error => error); + expect(error).toBeInstanceOf(Error); + expect(error).not.toBeInstanceOf(LinkRelayAuthTimeoutError); + expect(probes).toBe(1); expect(dataRequests).toBe(1); + } finally { clearTimeout(timer); parent.abort(); } + }); + }); + } + + test("auth deadline overrides must be finite and positive", async () => { + for (const authTimeoutMs of [0, -1, NaN, Infinity]) { + await expect(fetchBoundLinkRelay(target(1), "http://127.0.0.1:1/v1/models", {}, { authTimeoutMs })) + .rejects.toBeInstanceOf(RangeError); + } + }); + test("a removed last link cannot release its port while an authenticated response is still streaming", async () => { let current = store(); const body = gate(), entered = gate(); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 20aa180eed2..d8b5c34a870 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1,4 +1,4 @@ -{ "combo-description.test.ts": "routing", "combo-description-catalog.test.ts": "codex-integration", "catalog-auto-review-override.test.ts": "codex-integration", "management-auto-review-settings.test.ts": "server", "management-auto-review-auth.test.ts": "server", "prefer-model-alias-ids.test.ts": "server", "responses-mimo-envelope.test.ts": "responses", "responses-call-id-linkage.test.ts": "responses", +{ "link-diagnostics.test.ts": "clients", "combo-description.test.ts": "routing", "combo-description-catalog.test.ts": "codex-integration", "catalog-auto-review-override.test.ts": "codex-integration", "management-auto-review-settings.test.ts": "server", "management-auto-review-auth.test.ts": "server", "prefer-model-alias-ids.test.ts": "server", "responses-mimo-envelope.test.ts": "responses", "responses-call-id-linkage.test.ts": "responses", "bridge-terminal-refusal.test.ts": "adapters", "loop-terminal-refusal.test.ts": "images", "web-search-terminal-refusal.test.ts": "web-search", diff --git a/tests/server/port-reclaim.test.ts b/tests/server/port-reclaim.test.ts index 8025e77317a..07eccd6a7a1 100644 --- a/tests/server/port-reclaim.test.ts +++ b/tests/server/port-reclaim.test.ts @@ -9,6 +9,8 @@ import { parseListenEntriesFromSs, scanListenPidsForAddress, ownsIpv4LoopbackListener, + inspectIpv4LoopbackListener, + type LoopbackOwnerLookupIo, parseIpv4LoopbackListenPidsFromNetstat, parseProcLoopbackListenInodes, reclaimListenPort, @@ -85,8 +87,16 @@ describe("exact IPv4 loopback listener ownership", () => { }; expect(await ownsIpv4LoopbackListener(25000, 42, io)).toBe(true); expect(paths).toEqual(["/proc/42/fd"]); + expect((await inspectIpv4LoopbackListener(25000, 42, io)).kind).toBe("match"); expect(await ownsIpv4LoopbackListener(25000, 43, { ...io, readFdLink: async () => "socket:[44002]" })).toBe(false); + expect((await inspectIpv4LoopbackListener(25000, 43, { ...io, + readFdLink: async () => "socket:[44002]" })).kind).toBe("mismatch"); + expect((await inspectIpv4LoopbackListener(25000, 42, { ...io, + readProc: async () => { throw new Error("private proc detail"); } })).kind).toBe("lookup_error"); + expect((await inspectIpv4LoopbackListener(25000, 42, { ...io, + readProc: async () => tcp + "1: 0100007F:61A8 00000000:0000 0A 0 0 0 501 0 44003 1\n", + })).kind).toBe("mismatch"); }); test("Windows netstat ignores a foreign IPv6 listener on the same numeric port", () => { @@ -97,6 +107,119 @@ describe("exact IPv4 loopback listener ownership", () => { expect(parseIpv4LoopbackListenPidsFromNetstat(rows, 25000)).toEqual([42]); }); + test.each(["darwin", "win32"] as const)("%s reports match, mismatch and ambiguous owners with boolean compatibility", async platform => { + const row = (pid: number) => platform === "darwin" ? `${pid}\n` + : `TCP 127.0.0.1:25000 0.0.0.0:0 LISTENING ${pid}\n`; + for (const [stdout, kind] of [[row(42), "match"], [row(43), "mismatch"], + [row(42) + row(43), "mismatch"], ["", "mismatch"]] as const) { + let now = 100; + const io: LoopbackOwnerLookupIo = { platform, nowMs: () => now, + run: async (_file, _args, timeoutMs) => { + expect(timeoutMs).toBe(2_000); + now += 7; + return stdout; + } }; + expect(await inspectIpv4LoopbackListener(25000, 42, io)).toEqual({ kind, elapsedMs: 7 }); + expect(await ownsIpv4LoopbackListener(25000, 42, io)).toBe(kind === "match"); + } + }); + + test.each([ + ["darwin", {}, "mismatch"], + ["darwin", { signal: undefined }, "mismatch"], + ["darwin", { killed: undefined }, "mismatch"], + ["win32", {}, "lookup_error"], + ["darwin", { killed: true }, "lookup_error"], + ["darwin", { signal: "SIGTERM" }, "lookup_error"], + ["darwin", { code: 2 }, "lookup_error"], + ["darwin", { code: "1" }, "lookup_error"], + ["darwin", { stdout: "42\n" }, "lookup_error"], + ["darwin", { stderr: "lsof: permission denied" }, "lookup_error"], + ["darwin", { stdout: "\n" }, "lookup_error"], + ["darwin", { stderr: "\n" }, "lookup_error"], + ["darwin", { stdout: undefined }, "lookup_error"], + ["darwin", { stderr: undefined }, "lookup_error"], + ] as const)("%s empty lsof exit 1 with overrides %j reports %s", async (platform, overrides, kind) => { + const failure = Object.assign(new Error("Command failed: lsof"), { + code: 1, killed: false, signal: null, stdout: "", stderr: "", ...overrides, + }); + const io: LoopbackOwnerLookupIo = { platform, nowMs: () => 100, + run: async () => { throw failure; } }; + expect(await inspectIpv4LoopbackListener(25000, 42, io)).toEqual({ kind, elapsedMs: 0 }); + expect(await ownsIpv4LoopbackListener(25000, 42, io)).toBe(false); + }); + + test.each([ + [Object.assign(new Error("private spawn detail"), { code: "ENOENT" }), "lookup_error"], + [{ killed: true, signal: "SIGTERM", code: null }, "deadline"], + [{ killed: true, signal: "SIGTERM", code: "ERR_CHILD_PROCESS_STDIO_MAXBUFFER" }, "lookup_error"], + [{ killed: false, signal: "SIGTERM", code: null }, "lookup_error"], + [null, "lookup_error"], + ] as const)("rejected lookup %j returns only bounded metadata", async (failure, kind) => { + const io: LoopbackOwnerLookupIo = { platform: "darwin", nowMs: () => 100, + run: async () => { throw failure; } }; + expect(await inspectIpv4LoopbackListener(25000, 42, io)).toEqual({ kind, elapsedMs: 0 }); + expect(await ownsIpv4LoopbackListener(25000, 42, io)).toBe(false); + }); + + test.each(["resolve", "reject"] as const)("deadline stays final after late lookup %s", async completion => { + let now = 100; + let expire!: () => void; + let finish!: (stdout: string) => void; + let fail!: (error: Error) => void; + let cancelled = 0; + const io: LoopbackOwnerLookupIo = { platform: "darwin", nowMs: () => now, + run: () => new Promise((resolve, reject) => { finish = resolve; fail = reject; }), + scheduleDeadline: (callback, timeoutMs) => { + expect(timeoutMs).toBe(2_000); + expire = callback; + return () => { cancelled += 1; }; + } }; + const pending = inspectIpv4LoopbackListener(25000, 42, io); + now += 2_000; + expire(); + const result = await pending; + expect(result).toEqual({ kind: "deadline", elapsedMs: 2_000 }); + expect(cancelled).toBe(1); + now += 100; + if (completion === "resolve") finish("42\n"); + else fail(new Error("late private error")); + await Promise.resolve(); + expect(await pending).toEqual({ kind: "deadline", elapsedMs: 2_000 }); + expect(result).toEqual({ kind: "deadline", elapsedMs: 2_000 }); + const booleanPending = ownsIpv4LoopbackListener(25000, 42, io); + expire(); + expect(await booleanPending).toBe(false); + finish("42\n"); + }); + + test("successful lookup cancels the deadline and retains its result", async () => { + let expire!: () => void; + let cancelled = false; + const pending = inspectIpv4LoopbackListener(25000, 42, { platform: "darwin", nowMs: () => 10, + run: async () => "42\n", scheduleDeadline: callback => { + expire = callback; + return () => { cancelled = true; }; + } }); + expect(await pending).toEqual({ kind: "match", elapsedMs: 0 }); + expect(cancelled).toBe(true); + expire(); + expect(await pending).toEqual({ kind: "match", elapsedMs: 0 }); + }); + + test.each([[0, 42], [65536, 42], [1.5, 42], [25000, 0], [25000, NaN]])( + "invalid port/PID %j/%j fails closed before lookup", async (port, pid) => { + expect(await inspectIpv4LoopbackListener(port, pid, { nowMs: () => 1, + run: async () => { throw new Error("must not run"); } })).toEqual({ kind: "mismatch", elapsedMs: 0 }); + }); + + test.skipIf(process.platform !== "darwin")("a closed IPv4 loopback port reports mismatch with real lsof", async () => { + const server = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("ok") }); + const port = server.port!; + await server.stop(true); + expect((await inspectIpv4LoopbackListener(port, process.pid)).kind).toBe("mismatch"); + }); + test.skipIf(process.platform !== "darwin")("a foreign ::1 listener does not veto the owned IPv4 socket", async () => { const ipv4 = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("v4") }); const port = ipv4.port!;