diff --git a/src/client/connect.ts b/src/client/connect.ts index 67ea8d0968b..7aa3e65e005 100644 --- a/src/client/connect.ts +++ b/src/client/connect.ts @@ -104,6 +104,8 @@ export interface LinkClientCredential { } export interface ClientConnectDeps { + /** Abort enrollment network work and refuse subsequent writes; rollback still drains. */ + signal?: AbortSignal; fetchImpl?: typeof fetch; now?: () => Date; lifecycleLockDeps?: ClientLifecycleLockDeps; @@ -539,6 +541,18 @@ export async function connectClient( options: ConnectOptions, deps: ClientConnectDeps = {}, ): Promise { + deps.signal?.throwIfAborted(); + const rawFetch = deps.fetchImpl ?? fetch; + const fetchImpl: typeof fetch = deps.signal ? Object.assign(async (...[input, init = {}]: Parameters) => { + deps.signal!.throwIfAborted(); + const signals = [deps.signal, init.signal, input instanceof Request ? input.signal : undefined] + .filter((signal): signal is AbortSignal => signal != null); + return rawFetch(input, { ...init, signal: AbortSignal.any(signals), redirect: "manual" }); + }, { preconnect: rawFetch.preconnect }) : rawFetch; + const assertActiveConnectingState = (fingerprint?: string) => { + deps.signal?.throwIfAborted(); + assertConnectingState(fingerprint); + }; let serverUrl = ""; let managementUrl = ""; let linkAdmissionToken: string | null = null; @@ -586,7 +600,7 @@ export async function connectClient( throw new Error("link mode requires a valid local config port"); } withClientLifecycleSync(() => withConfigMutationLockSync(() => { - assertConnectingState(); + assertActiveConnectingState(); const externalProvider = currentExternalCodexModelProvider(); if (externalProvider) throw new Error("connect refused: an external Codex provider owns config.toml"); if (linkMode) { @@ -603,7 +617,7 @@ export async function connectClient( } }), deps.lifecycleLockDeps); - const upstreamFetch = deps.fetchImpl ?? fetch; + const upstreamFetch = fetchImpl; const readinessFetch = linkMode ? (async (input, init = {}) => { const url = input instanceof Request ? input.url : String(input); @@ -622,18 +636,18 @@ export async function connectClient( managementUrl, localGuiOrigin(), options.credential.value, - { fetchImpl: deps.fetchImpl }, + { fetchImpl }, ); cleanupCredential = { kind: "gui-session", value: session }; } else if (!linkMode && options.credential.kind === "admin") { cleanupCredential = { kind: "admin", value: options.credential.value }; } if (!linkMode) { - issued = await issueClientKey(managementUrl, cleanupCredential!, clientKeyName(), { fetchImpl: deps.fetchImpl }); + issued = await issueClientKey(managementUrl, cleanupCredential!, clientKeyName(), { fetchImpl }); } const initialFiles = withClientLifecycleSync(() => withConfigMutationLockSync(() => { - assertConnectingState(linkMode ? pendingConnectFingerprint! : undefined); + assertActiveConnectingState(linkMode ? pendingConnectFingerprint! : undefined); const persisted = linkMode ? (() => { const current = readServiceApiTokenState(); @@ -656,7 +670,7 @@ export async function connectClient( if (!admissionToken) throw new Error("client admission credential unavailable"); const apiKeyId = linkMode ? (options.credential as LinkClientCredential).apiKeyId : issued!.id; const catalog = await downloadClientCatalog(serverUrl, admissionToken, { - fetchImpl: deps.fetchImpl, + fetchImpl, timeoutMs: options.catalogTimeoutMs, }); // Fail closed BEFORE the write (#4207). The hub being reachable and the credential working @@ -666,7 +680,7 @@ export async function connectClient( // than writing one and restoring it afterwards. assertClientCatalogCompatible(catalog.body, deps.catalogCompatibility); writtenCatalogFingerprint = withClientLifecycleSync(() => withConfigMutationLockSync(() => { - assertConnectingState(persisted.fingerprint); + assertActiveConnectingState(persisted.fingerprint); atomicWriteFile(DEFAULT_CATALOG_PATH, catalog.body); return sha256(catalog.body); }), deps.lifecycleLockDeps); @@ -679,7 +693,7 @@ export async function connectClient( routingTarget: target, catalogPath: DEFAULT_CATALOG_PATH, journalOwner: { kind: "client", apiKeyId }, - beforeClientWrite: () => assertConnectingState(persisted.fingerprint), + beforeClientWrite: () => assertActiveConnectingState(persisted.fingerprint), }); if (!preflight.success) throw new Error(preflight.message); @@ -688,7 +702,7 @@ export async function connectClient( routingTarget: target, catalogPath: DEFAULT_CATALOG_PATH, journalOwner: { kind: "client", apiKeyId }, - beforeClientWrite: () => assertConnectingState(persisted.fingerprint), + beforeClientWrite: () => assertActiveConnectingState(persisted.fingerprint), }); if (!injected.success || injected.status === "skipped") throw new Error(injected.message); injectionCommitted = true; @@ -715,7 +729,7 @@ export async function connectClient( ...(linkMode ? { transport: "link" as const, link: linkMetadata } : {}), }; withClientLifecycleSync(() => withConfigMutationLockSync(() => { - assertConnectingState(persisted.fingerprint); + assertActiveConnectingState(persisted.fingerprint); clearClientConnectPending(persisted.fingerprint); commitClientConnection(connection); committed = true; diff --git a/src/client/link-join.ts b/src/client/link-join.ts index ad2fe285de9..83c6163fc7a 100644 --- a/src/client/link-join.ts +++ b/src/client/link-join.ts @@ -287,7 +287,7 @@ async function waitForReady( } const remaining = deadline - now(); if (remaining <= 0) throw new ClientLinkJoinError("join_tunnel_failed"); - await sleep(Math.min(JOIN_TUNNEL_POLL_MS, remaining)); + await Promise.race([tunnelExited, sleep(Math.min(JOIN_TUNNEL_POLL_MS, remaining))]); } } @@ -360,29 +360,45 @@ export async function joinHome(deps: ClientLinkJoinDeps, input: { alias: string throw new ClientLinkJoinError(code); } + const enrollmentAbort = new AbortController(); + let enrollmentFinished = false; try { if (!tunnel) throw new ClientLinkJoinError("join_tunnel_failed"); const connect = deps.connect ?? connectClient; - // Keep watching the tunnel until the connection commits: an exited tunnel - // must not let the issued key ride out to whatever next holds the port. - await Promise.race([ - tunnel.exited.then(() => { throw new ClientLinkJoinError("join_tunnel_failed"); }), - connect({ - serverUrl: `http://127.0.0.1:${tunnelPort}`, - managementUrl: `http://127.0.0.1:${tunnelPort}`, - credential: { kind: "link", apiKeyId: issued.apiKeyId, key: issued.key }, - transport: "link", - link: { tunnelPort, linkId: issued.linkId }, - selectedClients: deps.selectedClients ?? ["codex", "claude"], - managementTransport: "direct", - }, { - fetchImpl: deps.fetchImpl, - ...deps.connectDeps, - }), - ]); + // A tunnel exit cancels work; it is NOT a competing terminal result. The connect + // transaction alone decides commit versus rollback, so an exit queued immediately + // after commit cannot revoke a key that a connected client has already retained. + void tunnel.exited.then(() => { + if (!enrollmentFinished) enrollmentAbort.abort(new ClientLinkJoinError("join_tunnel_failed")); + }); + const signal = deps.connectDeps?.signal + ? AbortSignal.any([enrollmentAbort.signal, deps.connectDeps.signal]) : enrollmentAbort.signal; + // Observe a tunnel that exited after readiness before starting any enrollment write. + await Promise.resolve(); + signal.throwIfAborted(); + await connect({ + serverUrl: `http://127.0.0.1:${tunnelPort}`, + managementUrl: `http://127.0.0.1:${tunnelPort}`, + credential: { kind: "link", apiKeyId: issued.apiKeyId, key: issued.key }, + transport: "link", + link: { tunnelPort, linkId: issued.linkId }, + selectedClients: deps.selectedClients ?? ["codex", "claude"], + managementTransport: "direct", + }, { + fetchImpl: deps.fetchImpl, + ...deps.connectDeps, + signal, + }); } catch (error) { + // connectClient has drained its local rollback before rejecting. Only then can + // the tunnel and the remote key be compensated without racing a late writer. + const tunnelAborted = enrollmentAbort.signal.aborted; + enrollmentFinished = true; await rollback(deps, issued.linkId, tunnel); - throw new ClientLinkJoinError(error instanceof ClientLinkJoinError ? error.code : "join_connect_failed"); + throw new ClientLinkJoinError(tunnelAborted ? "join_tunnel_failed" + : error instanceof ClientLinkJoinError ? error.code : "join_connect_failed"); + } finally { + enrollmentFinished = true; } await stopTunnel(tunnel); diff --git a/structure/remote-link.md b/structure/remote-link.md index cfad32f1cad..4ec1953c9aa 100644 --- a/structure/remote-link.md +++ b/structure/remote-link.md @@ -63,3 +63,5 @@ Codex keeps the standalone loopback routing: `routingTarget` in `src/client/conn > Decision record: [ADR-6032](decisions/ADR-6032-link-relay-credential-boundary.md) Regression coverage lives in `tests/clients/link-ssh-argv.test.ts`, `tests/clients/link-ssh-config.test.ts`, `tests/clients/link-tunnel-state.test.ts`, `tests/clients/link-store.test.ts`, `tests/clients/link-boundary.test.ts`, `tests/clients/link-routes.test.ts`, `tests/clients/client-link-connect.test.ts`, `tests/clients/client-link-relay.test.ts`, `tests/clients/client-machine-listener.test.ts`, `tests/clients/client-link-status.test.ts`, `tests/clients/client-link-runtime.test.ts`, `tests/codex-integration/injection-link-websocket.test.ts`, `tests/clients/link-supervisor.test.ts`, `tests/clients/link-status-projection.test.ts`, `tests/clients/link-admission-wait.test.ts`, `tests/clients/link-fingerprint.test.ts`, `tests/cli/cli-link.test.ts`, `tests/server/link-management-routes.test.ts`, `tests/server/link-join-route.test.ts`, `tests/server/port-reclaim.test.ts`, `tests/server/link-listener-lifecycle.test.ts`, `tests/clients/client-link-teardown.test.ts` and `gui/tests/remote-link.test.tsx`. + +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. diff --git a/tests/clients/client-link-connect.test.ts b/tests/clients/client-link-connect.test.ts index 9f323c9ead2..6b8130a9e33 100644 --- a/tests/clients/client-link-connect.test.ts +++ b/tests/clients/client-link-connect.test.ts @@ -7,6 +7,8 @@ import { getDefaultConfig, saveConfig } from "../../src/config"; import { clientConnectionSchema } from "../../src/config/schema/leaf-validators"; import { handleConnectCommand, handleDisconnectCommand } from "../../src/cli/connect"; import { readSecretBytes } from "../../src/cli/runtime-api"; +import { joinHome } from "../../src/client/link-join"; +import { quoteRemote, remoteOcxArgv } from "../../src/link/ssh-argv"; import { connectClient, routingTarget } from "../../src/client/connect"; import { readServiceApiTokenState } from "../../src/lib/service-secrets"; import { isLinkConnection, readClientConnectionState } from "../../src/client/state"; @@ -130,6 +132,81 @@ describe("client link connection contracts", () => { }); }); + test("cancellation during catalog download prevents late enrollment writes and drains token rollback", async () => { + await withLinkHome(async home => { + const prior = '{"models":[{"id":"prior"}]}\n'; + writeFileSync(DEFAULT_CATALOG_PATH, prior); + const abort = new AbortController(); + let cancelledFetch = false; + await expect(connectClient(linkOptions(), { + signal: abort.signal, + fetchImpl: async (input, init) => { + if (String(input).endsWith("/readyz")) return Response.json({ + service: "opencodex", version: "0.0.0", uptime: 1, pid: 1, port: 34567, + status: "ready", protocol: 1, minimumClientProtocol: 1, + managementUrl: "http://127.0.0.1:34567", + }); + abort.abort(new Error("fixture enrollment cancelled")); + cancelledFetch = init?.signal?.aborted === true; + // Even a fetch implementation returning after abort cannot authorize a write. + return Response.json({ models: [] }); + }, + lifecycleLockDeps: { lockPath: join(home, "lifecycle.sqlite") }, + })).rejects.toThrow("fixture enrollment cancelled"); + expect(cancelledFetch).toBe(true); + expect(readFileSync(DEFAULT_CATALOG_PATH, "utf8")).toBe(prior); + expect(readServiceApiTokenState()).toEqual({ kind: "absent" }); + expect(readClientConnectionState()).toEqual({ kind: "disconnected" }); + }); + }); + + test("a real enrollment commit survives a tunnel exit queued before join completion", async () => { + await withLinkHome(async home => { + let exit!: (code: number) => void; + const exited = new Promise(resolve => { exit = resolve; }); + let revoked = 0, restarted = 0; + const result = await joinHome({ + runner: { + run: async argv => { + if (argv.at(-1) === quoteRemote(remoteOcxArgv(["link", "revoke", "--link-id", linkId]))) { + revoked += 1; + return { code: 0, stdout: "", stderr: "" }; + } + return { code: 0, stdout: JSON.stringify({ linkId, apiKeyId: "key-1", key, listenerPort: 45678 }), stderr: "" }; + }, + spawnTunnel: () => { throw new Error("unexpected real tunnel"); }, + }, + knownHostsFile: join(home, "known-hosts"), + confirmedHost: { alias: "home", fingerprint: "SHA256:fixture", probedAt: 1 }, + now: () => 1, choosePort: async () => 34567, + writeState: () => {}, clearState: () => {}, readSidecar: () => null, + spawnTunnel: () => ({ pid: 123, exited, stop: async () => {} }), + scanListenPids: () => ({ ok: true, pids: [123] }), + selectedClients: ["claude"], + fetchImpl: async (input, init) => { + if (!String(input).endsWith("/readyz")) return Response.json({ models: [] }); + if (!new Headers(init?.headers).has("x-opencodex-api-key")) return new Response(null, { status: 401 }); + return Response.json({ service: "opencodex", version: "0.0.0", uptime: 1, pid: 123, + port: 34567, status: "ready", protocol: 1, minimumClientProtocol: 1, + managementUrl: "http://127.0.0.1:34567" }); + }, + connectDeps: { lifecycleLockDeps: { lockPath: join(home, "lifecycle.sqlite") }, + catalogCompatibility: { supportedEfforts: () => new Set() } }, + connect: async (options, deps) => { + const committed = await connectClient(options, deps); + exit(255); + return committed; + }, + scheduleRestart: () => { restarted += 1; }, + }, { alias: "home" }); + expect(result).toEqual({ linkId, apiKeyId: "key-1" }); + expect(readClientConnectionState()).toMatchObject({ kind: "connected", value: { link: { linkId } } }); + expect(readServiceApiTokenState()).toMatchObject({ kind: "present", token: key }); + expect(revoked).toBe(0); + expect(restarted).toBe(1); + }); + }); + test("catalog failure removes the pending link token and leaves config.client unset", async () => { await withLinkHome(async home => { await expect(connectClient(linkOptions(), { diff --git a/tests/server/link-join-route.test.ts b/tests/server/link-join-route.test.ts index 9ea2b8c26c1..4736a9607a6 100644 --- a/tests/server/link-join-route.test.ts +++ b/tests/server/link-join-route.test.ts @@ -471,25 +471,81 @@ describe("client initiated link join", () => { expect(revokeCalls(calls)).toHaveLength(1); }); + test("readiness polling stops after tunnel exit without waiting for its deadline", async () => { + const calls: string[][] = []; + let clock = 1, scans = 0, sleeps = 0, fetched = 0, connected = 0; + let releaseExit!: (code: number) => void; + const exited = new Promise(resolve => { releaseExit = resolve; }); + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + now: () => clock, + sleep: async ms => { + sleeps += 1; clock += ms; releaseExit(255); + await Promise.resolve(); + }, + writeState: () => {}, clearState: () => {}, + spawnTunnel: () => ({ pid: 123, exited, stop: async () => {} }), + scanListenPids: () => { scans += 1; return { ok: true, pids: [] }; }, + fetchImpl: async () => { fetched += 1; return new Response(null, { status: 401 }); }, + connect: (async () => { connected += 1; }) as typeof import("../../src/client/connect").connectClient, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(scans).toBe(1); + expect(sleeps).toBe(1); + expect(clock).toBe(101); + expect(fetched).toBe(0); + expect(connected).toBe(0); + expect(revokeCalls(calls)).toHaveLength(1); + }); + test("a tunnel that exits during connect cannot commit the connection", async () => { const calls: string[][] = []; let releaseExit!: (code: number) => void; const exited = new Promise(resolve => { releaseExit = resolve; }); let stopped = 0; let connectCommitted = false; + let connectDrained = false; await expect(joinHome(joinDeps({ runner: runnerFor(calls), writeState: () => {}, clearState: () => {}, spawnTunnel: () => ({ pid: 1, exited, stop: async () => { stopped += 1; } }), fetchImpl: challengedFetch(), - connect: (async () => { releaseExit(255); await new Promise(() => {}); connectCommitted = true; }) as typeof import("../../src/client/connect").connectClient, + connect: (async (_options, deps) => { + releaseExit(255); + try { + await new Promise(resolve => setTimeout(resolve, 10)); + deps?.signal?.throwIfAborted(); + connectCommitted = true; + } finally { connectDrained = true; } + }) as typeof import("../../src/client/connect").connectClient, }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(connectDrained).toBe(true); expect(connectCommitted).toBe(false); expect(stopped).toBe(1); expect(revokeCalls(calls)).toHaveLength(1); }); + test("an exit queued after enrollment commit does not revoke the committed link", async () => { + const calls: string[][] = [], order: string[] = []; + let releaseExit!: (code: number) => void; + const exited = new Promise(resolve => { releaseExit = resolve; }); + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + writeState: () => {}, clearState: () => { order.push("clear"); }, + spawnTunnel: () => ({ pid: 123, exited, stop: async () => { order.push("stop"); } }), + fetchImpl: challengedFetch(), + connect: (async (_options, deps) => { + deps?.signal?.throwIfAborted(); + // Mirrors connectClient's synchronous final commit: no await follows it. + order.push("commit"); + releaseExit(255); + }) as typeof import("../../src/client/connect").connectClient, + scheduleRestart: () => { order.push("restart"); }, + }), { alias: "home" })).resolves.toEqual({ linkId: LINK_ID, apiKeyId: API_KEY_ID }); + expect(order).toEqual(["commit", "stop", "restart"]); + expect(revokeCalls(calls)).toHaveLength(0); + }); + test("does not disclose the issued key when the tunnel exits during its spawn grace", async () => { const calls: string[][] = []; let fetches = 0; @@ -598,6 +654,28 @@ describe("client initiated link join", () => { expect(logs.mock.calls.flat().join(" ")).not.toContain(KEY); }); + for (const code of ["join_connect_failed", "admission_failed"] as const) { + test(`rollback tunnel exit preserves the original ${code} cause`, async () => { + const calls: string[][] = []; + let exit!: (code: number) => void; + const exited = new Promise(resolve => { exit = resolve; }); + let stopped = 0, cleared = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), writeState: () => {}, + clearState: () => { cleared += 1; }, + spawnTunnel: () => ({ pid: 123, exited, stop: async () => { stopped += 1; exit(0); } }), + fetchImpl: challengedFetch(), + connect: (async () => { + if (code === "admission_failed") throw new ClientLinkJoinError(code); + throw new Error("catalog validation failed"); + }) as typeof import("../../src/client/connect").connectClient, + }), { alias: "home" })).rejects.toMatchObject({ code }); + expect(stopped).toBe(1); + expect(cleared).toBe(1); + expect(revokeCalls(calls)).toHaveLength(1); + }); + } + test("keeps the sidecar and reports the link id when rollback revoke fails, then compensates before the next join", async () => { const sidecar = { linkId: LINK_ID,