From 8dac0e97544c44497af6aa397636465ff5076bc1 Mon Sep 17 00:00:00 2001 From: Navneet <155615649+nkcbuilds@users.noreply.github.com> Date: Mon, 19 Oct 2026 07:08:08 +0530 Subject: [PATCH 1/3] fix(server): refresh managed Codex credentials during active turns --- apps/server/src/bin.ts | 5 +- .../src/provider/CodexChatGptAuth.test.ts | 19 + apps/server/src/provider/CodexChatGptAuth.ts | 174 ++++--- .../src/provider/CodexManagedRuntime.test.ts | 476 +++++++++++++++++- .../src/provider/CodexManagedRuntime.ts | 168 ++++--- .../provider/CodexManagedTokenSource.test.ts | 461 +++++++++++++++++ .../src/provider/CodexManagedTokenSource.ts | 252 ++++++++++ .../src/provider/Drivers/CodexDriver.test.ts | 7 +- .../provider/Drivers/CodexManagedProvider.ts | 40 +- .../provider/codexManagedTokenCommand.test.ts | 125 +++++ .../src/provider/codexManagedTokenCommand.ts | 67 +++ 11 files changed, 1624 insertions(+), 170 deletions(-) create mode 100644 apps/server/src/provider/CodexManagedTokenSource.test.ts create mode 100644 apps/server/src/provider/CodexManagedTokenSource.ts create mode 100644 apps/server/src/provider/codexManagedTokenCommand.test.ts create mode 100644 apps/server/src/provider/codexManagedTokenCommand.ts diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index 4e01009c77c4a..459fb7b9f2f1d 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -14,7 +14,10 @@ if ( }) ) { const command = process.argv[2]; - if (command === "acp-mcp-bridge" || command === "acp-mcp-call") { + if (command === "__managed-chatgpt-token") { + const { runCodexManagedTokenCommand } = await import("./provider/codexManagedTokenCommand.ts"); + await runCodexManagedTokenCommand(process.argv.slice(3)); + } else if (command === "acp-mcp-bridge" || command === "acp-mcp-call") { const { runAcpMcpCliFastPath } = await import("./mcp/AcpMcpStdioBridge.ts"); await runAcpMcpCliFastPath(command, process.argv.slice(3)); } else { diff --git a/apps/server/src/provider/CodexChatGptAuth.test.ts b/apps/server/src/provider/CodexChatGptAuth.test.ts index 859537b25cb6d..33b786ffa87c9 100644 --- a/apps/server/src/provider/CodexChatGptAuth.test.ts +++ b/apps/server/src/provider/CodexChatGptAuth.test.ts @@ -1753,6 +1753,25 @@ it.effect("primary completes OAuth and destination imports and owns the refresh ), ); +it.effect("profile import stops credential users before taking the auth access lock", () => + provision( + Effect.gen(function* () { + const h = yield* makeHarness; + yield* h.signIn; + yield* h.phase("succeeded"); + const profile = yield* h.auth.exportProfile; + const destination = yield* h.destination; + yield* destination.auth.controller.importProfile!(profile, Effect.void); + + const imported = yield* destination.auth.controller.importProfile!( + profile, + destination.auth.access.pipe(Effect.asVoid), + ); + assert.strictEqual(imported.phase, "succeeded"); + }), + ), +); + it.effect("transferred profiles reject mismatched identities without overwriting credentials", () => provision( Effect.gen(function* () { diff --git a/apps/server/src/provider/CodexChatGptAuth.ts b/apps/server/src/provider/CodexChatGptAuth.ts index 6a393ace4c605..d7b738456fbad 100644 --- a/apps/server/src/provider/CodexChatGptAuth.ts +++ b/apps/server/src/provider/CodexChatGptAuth.ts @@ -54,6 +54,19 @@ const Record = Schema.Struct({ issuer: Schema.optionalKey(Schema.String), }); export type CodexChatGptCredentials = typeof Record.Type; + +export const codexChatGptProfileKey = (record: CodexChatGptCredentials) => + NodeCrypto.createHash("sha256") + .update( + JSON.stringify([ + record.clientId, + record.subject, + record.idToken ?? null, + record.issuer ?? null, + ]), + ) + .digest("hex"); + const Sessions = Schema.Struct({ activeClientId: Schema.NullOr(Schema.String), sessions: Schema.Array(Record), @@ -713,73 +726,86 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* ), ); }); - const access = lock.withPermit( - withSessionLock( - Effect.uninterruptible( - Effect.gen(function* () { - const saved = yield* read; - if (Option.isNone(saved)) - return yield* failure("access", "Sign in with ChatGPT to use managed Codex."); - let record = saved.value; - if (!record.scopes.includes(REQUIRED_SCOPE)) - return yield* failure( - "sharing", - "Token sharing is disabled. Sign in with ChatGPT and enable token sharing.", - ); - const now = yield* Clock.currentTimeMillis; - if (record.expiresAt - now > 60_000) return record; - if (record.earliestRefreshAt !== null && record.earliestRefreshAt > now) { - if (record.expiresAt > now) return record; - return yield* failure( - "refresh", - "ChatGPT cannot renew this connection yet. Try again shortly.", - ); - } - if (!record.refreshToken) { - yield* remove; - return yield* failure("refresh", "Your ChatGPT connection expired. Sign in again."); - } - const tokens = yield* exchange( - new URLSearchParams({ - grant_type: "refresh_token", - client_id: record.clientId, - refresh_token: record.refreshToken, - resource, - }), - ).pipe( - Effect.interruptible, - Effect.timeout("20 seconds"), - Effect.mapError((error) => - error._tag === "ProviderSetupError" && error.operation === "client" - ? error - : failure( - "refresh", - "Could not renew the ChatGPT connection. Retry, or sign in again.", - ), - ), - ); - if (!tokens.refresh_token || tokens.token_type.toLowerCase() !== "bearer") { - return yield* failure("refresh", "ChatGPT returned an invalid renewal. Sign in again."); - } - record = { - ...record, - accessToken: tokens.access_token, - refreshToken: tokens.refresh_token, - expiresAt: (yield* Clock.currentTimeMillis) + tokens.expires_in * 1000, - earliestRefreshAt: earliest(tokens.earliest_refresh_at), - scopes: tokens.scope.split(/\s+/).filter(Boolean), - }; - yield* save(record); - if (!record.scopes.includes(REQUIRED_SCOPE)) - return yield* failure( - "sharing", - "ChatGPT token sharing is no longer enabled. Sign in again.", + const accessForProfile = (expectedProfileKey?: string) => + lock.withPermit( + withSessionLock( + Effect.uninterruptible( + Effect.gen(function* () { + const saved = yield* read; + if (Option.isNone(saved)) + return yield* failure("access", "Sign in with ChatGPT to use managed Codex."); + let record = saved.value; + if ( + expectedProfileKey !== undefined && + codexChatGptProfileKey(record) !== expectedProfileKey + ) + return yield* failure( + "access", + "The ChatGPT connection changed. Start a new session.", + ); + if (!record.scopes.includes(REQUIRED_SCOPE)) + return yield* failure( + "sharing", + "Token sharing is disabled. Sign in with ChatGPT and enable token sharing.", + ); + const now = yield* Clock.currentTimeMillis; + if (record.expiresAt - now > 60_000) return record; + if (record.earliestRefreshAt !== null && record.earliestRefreshAt > now) { + if (record.expiresAt > now) return record; + return yield* failure( + "refresh", + "ChatGPT cannot renew this connection yet. Try again shortly.", + ); + } + if (!record.refreshToken) { + yield* remove; + return yield* failure("refresh", "Your ChatGPT connection expired. Sign in again."); + } + const tokens = yield* exchange( + new URLSearchParams({ + grant_type: "refresh_token", + client_id: record.clientId, + refresh_token: record.refreshToken, + resource, + }), + ).pipe( + Effect.interruptible, + Effect.timeout("20 seconds"), + Effect.mapError((error) => + error._tag === "ProviderSetupError" && error.operation === "client" + ? error + : failure( + "refresh", + "Could not renew the ChatGPT connection. Retry, or sign in again.", + ), + ), ); - return record; - }), + if (!tokens.refresh_token || tokens.token_type.toLowerCase() !== "bearer") { + return yield* failure( + "refresh", + "ChatGPT returned an invalid renewal. Sign in again.", + ); + } + record = { + ...record, + accessToken: tokens.access_token, + refreshToken: tokens.refresh_token, + expiresAt: (yield* Clock.currentTimeMillis) + tokens.expires_in * 1000, + earliestRefreshAt: earliest(tokens.earliest_refresh_at), + scopes: tokens.scope.split(/\s+/).filter(Boolean), + }; + yield* save(record); + if (!record.scopes.includes(REQUIRED_SCOPE)) + return yield* failure( + "sharing", + "ChatGPT token sharing is no longer enabled. Sign in again.", + ); + return record; + }), + ), ), - ), - ); + ); + const access = accessForProfile(); const logout = Effect.gen(function* () { const credentials = Option.getOrUndefined(yield* read); let confirmed = !credentials?.refreshToken; @@ -964,12 +990,17 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* yield* saveRegistration(registration); yield* save(profile.credentials, true); }); - return lock.withPermit( - track( - "transfer", - { flow: "primary_handoff" }, - validate.pipe( - Effect.andThen(controller.adoptCredentials!(withSessionLock(saveImported), stopSessions)), + // Stop session scopes before taking the auth/session locks used by access. + // A closing managed Codex scope can be draining an admitted token refresh. + return track( + "transfer", + { flow: "primary_handoff" }, + validate.pipe( + Effect.andThen( + controller.adoptCredentials!( + lock.withPermit(withSessionLock(saveImported)), + stopSessions, + ), ), ), ); @@ -978,6 +1009,7 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* controller: { ...controller, reconnectProfile, importProfile }, read, access, + accessForProfile, exportProfile, revoke: lock.withPermit(withSessionLock(remove)), }; diff --git a/apps/server/src/provider/CodexManagedRuntime.test.ts b/apps/server/src/provider/CodexManagedRuntime.test.ts index 69d27f1377b6b..2a998cbb30475 100644 --- a/apps/server/src/provider/CodexManagedRuntime.test.ts +++ b/apps/server/src/provider/CodexManagedRuntime.test.ts @@ -1,10 +1,13 @@ // @effect-diagnostics nodeBuiltinImport:off - checks the host's default Codex directory without writing to it. +import * as NodeNet from "node:net"; import * as NodeOS from "node:os"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { CodexSettings, EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Layer from "effect/Layer"; @@ -15,6 +18,7 @@ import * as ServerConfig from "../config.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; +import { codexChatGptProfileKey } from "./CodexChatGptAuth.ts"; import { makeCodexManagedRuntime } from "./CodexManagedRuntime.ts"; import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; import { codexAppServerArgs } from "./codexLaunchArgs.ts"; @@ -22,6 +26,16 @@ import { resolveManagedCodexHomeLayout } from "./CodexManagedHome.ts"; const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); const decodeSettings = Schema.decodeSync(CodexSettings); + +const readToken = (endpoint: string) => + new Promise((resolve, reject) => { + const socket = NodeNet.createConnection(endpoint); + const chunks: Buffer[] = []; + socket.once("connect", () => socket.write("token\n")); + socket.on("data", (chunk: Buffer) => chunks.push(chunk)); + socket.once("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + socket.once("error", reject); + }); it.effect("managed home defaults to the global Codex home and honors configured home paths", () => Effect.gen(function* () { const path = yield* Path.Path; @@ -44,17 +58,383 @@ it.effect("managed home defaults to the global Codex home and honors configured ProviderInstanceId.make("codex-work"), decodeSettings({ homePath: "/custom/shared", shadowHomePath: "/custom/shadow" }), ); - assert.equal(configured.sharedHomePath, "/custom/shared"); - assert.equal(configured.effectiveHomePath, "/custom/shadow"); + assert.equal(configured.sharedHomePath, path.resolve("/custom/shared")); + assert.equal(configured.effectiveHomePath, path.resolve("/custom/shadow")); }).pipe(Effect.provide(NodeServices.layer)), ); +it.effect("prepares the local runtime before consuming refreshable credentials", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const instanceId = ProviderInstanceId.make("codex"); + const homePath = path.join(config.baseDir, "blocked-codex-home"); + yield* fs.writeFileString(homePath, "not-a-directory"); + + const data = new Map(); + let leases = 0; + const secrets = ServerSecretStore.ServerSecretStore.of({ + get: (key) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))), + set: (key, value) => + Effect.sync(() => { + data.set(key, value); + }), + remove: (key) => + Effect.sync(() => { + data.delete(key); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + let authRequests = 0; + + yield* Effect.gen(function* () { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + yield* store.set( + new TextEncoder().encode( + JSON.stringify({ + clientId: "oaiapp_test", + accessToken: "expired-access", + refreshToken: "refresh-must-not-be-consumed", + expiresAt: 0, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + subject: "test-user", + email: null, + }), + ), + ); + const runtime = yield* makeCodexManagedRuntime({ + instanceId, + enabled: true, + config: decodeSettings({ setupMode: "managed", homePath }), + environment: {}, + }); + assert.equal((yield* runtime.resolve.pipe(Effect.scoped, Effect.result))._tag, "Failure"); + assert.equal(authRequests, 0); + assert.equal(leases, 0); + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provide( + Layer.mock(CodexInstallation.CodexInstallation)({ + managedDirectory: "/isolated/tools/codex", + acquire: () => + Effect.gen(function* () { + leases++; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + leases--; + }), + ); + return { + executablePath: "/isolated/tools/codex/bin/codex", + managedVersionDirectory: "/isolated/tools/codex", + source: "managed" as const, + version: "0.156.1", + }; + }), + }), + ), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + authRequests++; + return HttpClientResponse.fromWeb(request, new Response(null, { status: 500 })); + }), + ), + ), + ); + }).pipe( + Effect.scoped, + Effect.provide( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-managed-runtime-order-", + }).pipe(Layer.provideMerge(NodeServices.layer)), + ), + ), +); + +it.effect("profile-bound runtime resolution rejects a replacement before refreshing it", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const instanceId = ProviderInstanceId.make("codex"); + const homePath = path.join(config.baseDir, "codex-home"); + yield* fs.makeDirectory(homePath, { recursive: true }); + + const data = new Map(); + const secrets = ServerSecretStore.ServerSecretStore.of({ + get: (key) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))), + set: (key, value) => + Effect.sync(() => { + data.set(key, value); + }), + remove: (key) => + Effect.sync(() => { + data.delete(key); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + let authRequests = 0; + + yield* Effect.gen(function* () { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + const original = { + clientId: "oaiapp_a", + accessToken: "access-a", + refreshToken: "refresh-a", + expiresAt: Number.MAX_SAFE_INTEGER, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + subject: "user-a", + email: null, + idToken: "identity-a", + issuer: "https://auth.openai.com", + }; + yield* store.set(new TextEncoder().encode(JSON.stringify(original))); + const runtime = yield* makeCodexManagedRuntime({ + instanceId, + enabled: true, + config: decodeSettings({ setupMode: "managed", homePath }), + environment: {}, + }); + yield* store.set( + new TextEncoder().encode( + JSON.stringify({ + ...original, + clientId: "oaiapp_b", + accessToken: "expired-b", + refreshToken: "refresh-b-must-not-be-consumed", + expiresAt: 0, + subject: "user-b", + idToken: "identity-b", + }), + ), + ); + + const result = yield* runtime + .resolveForProfile(codexChatGptProfileKey(original)) + .pipe(Effect.scoped, Effect.result); + assert.equal(result._tag, "Failure"); + assert.equal(authRequests, 0); + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provide( + Layer.mock(CodexInstallation.CodexInstallation)({ + managedDirectory: "/isolated/tools/codex", + acquire: () => + Effect.succeed({ + executablePath: "/isolated/tools/codex/bin/codex", + managedVersionDirectory: "/isolated/tools/codex", + source: "managed", + version: "0.156.1", + }), + }), + ), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + authRequests++; + return HttpClientResponse.fromWeb(request, new Response(null, { status: 500 })); + }), + ), + ), + ); + }).pipe( + Effect.scoped, + Effect.provide( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-managed-runtime-profile-order-", + }).pipe(Layer.provideMerge(NodeServices.layer)), + ), + ), +); + +it.effect("logout drains an admitted refresh before revoking and removing credentials", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const instanceId = ProviderInstanceId.make("codex"); + const homePath = path.join(config.baseDir, "codex-home"); + yield* fs.makeDirectory(homePath, { recursive: true }); + + const data = new Map(); + const events: string[] = []; + const refreshEntered = yield* Deferred.make(); + const releaseRefresh = yield* Deferred.make(); + let credentialKey: string | undefined; + const secrets = ServerSecretStore.ServerSecretStore.of({ + get: (key) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))), + set: (key, value) => + Effect.sync(() => { + data.set(key, value); + if (key === credentialKey && new TextDecoder().decode(value).includes("rotated-refresh")) + events.push("refresh-saved"); + }), + remove: (key) => + Effect.sync(() => { + data.delete(key); + if (key === credentialKey) events.push("credentials-removed"); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + + yield* Effect.gen(function* () { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + credentialKey = store.binding.key; + const original = { + clientId: "oaiapp_test", + accessToken: "access-a", + refreshToken: "refresh-a", + expiresAt: Number.MAX_SAFE_INTEGER, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + subject: "test-user", + email: null, + idToken: "identity-a", + issuer: "https://auth.openai.com", + }; + yield* store.set(new TextEncoder().encode(JSON.stringify(original))); + const runtime = yield* makeCodexManagedRuntime({ + instanceId, + enabled: true, + config: decodeSettings({ setupMode: "managed", homePath }), + environment: {}, + }); + const effective = yield* runtime.auth.controller.withAccess!(runtime.resolve); + const providerConfig = codexAppServerArgs(effective.config.launchArgs).find((arg) => + arg.startsWith("model_providers.openai_token_sharing_"), + ); + const encodedAuthArgs = /args=(\[.*\]),timeout_ms=/u.exec(providerConfig!)?.[1]; + const authArgs = JSON.parse(encodedAuthArgs!) as string[]; + const endpoint = authArgs.at(-1)!; + + yield* store.set(new TextEncoder().encode(JSON.stringify({ ...original, expiresAt: 0 }))); + const helper = yield* Effect.promise(() => + readToken(endpoint).then( + () => undefined, + () => undefined, + ), + ).pipe(Effect.forkChild); + yield* Deferred.await(refreshEntered); + + const logout = yield* runtime.auth.controller.logout(Effect.void).pipe(Effect.forkChild); + yield* Effect.yieldNow; + assert.isUndefined(logout.pollUnsafe()); + + yield* Deferred.succeed(releaseRefresh, undefined); + yield* Fiber.join(logout); + yield* Fiber.join(helper); + + assert.deepEqual(events, [ + "refresh-entered", + "refresh-response", + "refresh-saved", + "lease-released", + "revoke", + "credentials-removed", + ]); + assert.isTrue(Option.isNone(yield* store.get)); + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provide( + Layer.mock(CodexInstallation.CodexInstallation)({ + managedDirectory: "/isolated/tools/codex", + acquire: () => + Effect.acquireRelease( + Effect.succeed({ + executablePath: "/isolated/tools/codex/bin/codex", + managedVersionDirectory: "/isolated/tools/codex", + source: "managed" as const, + version: "0.156.1", + }), + () => Effect.sync(() => events.push("lease-released")), + ), + }), + ), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => { + if (request.url.endsWith("/.well-known/openid-configuration")) + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + Response.json({ + issuer: "https://auth.openai.com", + authorization_endpoint: "https://auth.openai.com/authorize", + token_endpoint: "https://auth.openai.com/token", + jwks_uri: "https://auth.openai.com/jwks", + revocation_endpoint: "https://auth.openai.com/revoke", + }), + ), + ); + if (request.url === "https://auth.openai.com/token") + return Effect.gen(function* () { + events.push("refresh-entered"); + yield* Deferred.succeed(refreshEntered, undefined); + yield* Deferred.await(releaseRefresh); + events.push("refresh-response"); + return HttpClientResponse.fromWeb( + request, + Response.json({ + access_token: "rotated-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + token_type: "Bearer", + scope: "chatgpt.tokens.use.direct", + }), + ); + }); + if (request.url === "https://auth.openai.com/revoke") { + events.push("revoke"); + return Effect.succeed( + HttpClientResponse.fromWeb(request, new Response(null, { status: 200 })), + ); + } + return Effect.die(`Unexpected request: ${request.url}`); + }), + ), + ); + }).pipe( + Effect.scoped, + Effect.provide( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-managed-runtime-logout-order-", + }).pipe(Layer.provideMerge(NodeServices.layer)), + ), + ), +); + it.effect.each( (["managed", "local"] as const).flatMap((source) => - (["primary", "additional"] as const).map((account) => ({ source, account })), + (["primary", "additional"] as const).flatMap((account) => + (["node", "electron"] as const).map((runtimeHost) => ({ source, account, runtimeHost })), + ), ), )( - "$source Codex $account account shares home state without routing owned tokens through ambient CLI overrides", - ({ source, account }) => + "$source Codex $account account ($runtimeHost host) shares home state without routing owned tokens through ambient CLI overrides", + ({ source, account, runtimeHost }) => Effect.gen(function* () { const instanceId = ProviderInstanceId.make( account === "primary" ? "codex" : "codex-personal", @@ -111,15 +491,19 @@ it.effect.each( scopes: ["chatgpt.tokens.use.direct"], subject: "test-user", email: null, + idToken: "dummy-id-token", + issuer: "https://auth.openai.com", }); yield* store.set(new TextEncoder().encode(json)); - const ambient = { + const ambient: NodeJS.ProcessEnv = { + ...(runtimeHost === "electron" ? { ELECTRON_RUN_AS_NODE: "1" } : {}), CODEX_HOME: "/user/.codex", OPENAI_API_KEY: "dummy-global-key", OPENAI_BASE_URL: "https://user-proxy.test", T3CODE_CODEX_LAUNCH_ARGS: "--config model_provider=global-proxy", PATH: "/usr/bin", }; + const ambientBefore = JSON.stringify(ambient); const runtime = yield* makeCodexManagedRuntime({ instanceId, enabled: true, @@ -127,7 +511,7 @@ it.effect.each( environment: ambient, }); yield* Effect.gen(function* () { - const effective = yield* runtime.resolve; + const effective = yield* runtime.auth.controller.withAccess!(runtime.resolve); assert.strictEqual(leases, 1); assert.strictEqual(effective.config.binaryPath, executable.executablePath); assert.notStrictEqual(effective.config.homePath, ambient.CODEX_HOME); @@ -137,7 +521,7 @@ it.effect.each( assert.equal(runtime.homeLayout.mode, "direct"); } else { assert.include(effective.config.homePath, instanceId); - assert.include(effective.config.homePath, "userdata/providers/codex"); + assert.include(effective.config.homePath, path.join("userdata", "providers", "codex")); assert.equal(runtime.homeLayout.mode, "authOverlay"); assert.equal( yield* fs.readLink(path.join(effective.config.homePath, "sessions")), @@ -149,35 +533,91 @@ it.effect.each( ); assert.isFalse(yield* fs.exists(path.join(effective.config.homePath, "auth.json"))); } - assert.strictEqual(effective.environment.ACCESS_TOKEN, "dummy-owned-access"); + assert.isUndefined(effective.environment.ACCESS_TOKEN); + assert.strictEqual( + effective.environment.ELECTRON_RUN_AS_NODE, + ambient.ELECTRON_RUN_AS_NODE, + ); + assert.isUndefined(effective.environment.T3CODE_MANAGED_CODEX_AUTH_URL); + assert.isUndefined(effective.environment.T3CODE_MANAGED_CODEX_AUTH_SECRET); assert.isUndefined(effective.environment.OPENAI_API_KEY); assert.isUndefined(effective.environment.OPENAI_BASE_URL); assert.isUndefined(effective.environment.T3CODE_CODEX_LAUNCH_ARGS); const args = codexAppServerArgs(effective.config.launchArgs); - assert.include( - args, - 'model_providers.openai_token_sharing.base_url="https://api.openai.com/v1"', + assert.isTrue( + args.some( + (arg) => arg.startsWith("model_provider=") && arg.includes("openai_token_sharing_"), + ), ); - assert.include( - args, - 'model_providers.openai_token_sharing.model_catalog_url="https://api.openai.com/v1/models"', + const providerConfig = args.find((arg) => + arg.startsWith("model_providers.openai_token_sharing_"), ); + assert.isDefined(providerConfig); + assert.include(providerConfig!, 'base_url="https://api.openai.com/v1"'); + assert.include(providerConfig!, 'model_catalog_url="https://api.openai.com/v1/models"'); assert.include(args, "features.api_key_model_discovery=true"); assert.include(args, "features.multi_agent=false"); assert.include(args, "features.multi_agent_v2=false"); + assert.notInclude(effective.config.launchArgs, "shell_environment_policy"); + assert.include(providerConfig!, "auth={"); + assert.include(providerConfig!, "command="); + assert.include(providerConfig!, "args="); + assert.include(providerConfig!, "timeout_ms=80000"); + assert.include(providerConfig!, "refresh_interval_ms=0"); + const encodedAuthArgs = /args=(\[.*\]),timeout_ms=/u.exec(providerConfig!)?.[1]; + assert.isDefined(encodedAuthArgs); + const authArgs = JSON.parse(encodedAuthArgs!) as string[]; + assert.include(authArgs, "__managed-chatgpt-token"); + const endpoint = authArgs.at(-1); + assert.isDefined(endpoint); + const token = yield* Effect.promise(() => readToken(endpoint!)); + assert.equal(token, "dummy-owned-access"); + assert.notInclude(providerConfig!, 'env_key="ACCESS_TOKEN"'); assert.notInclude(effective.config.launchArgs, "model_catalog_json"); assert.notInclude(effective.config.launchArgs, "x-openai-chatpass-test"); - assert.include(args, "model_providers.openai_token_sharing.supports_websockets=false"); - assert.include(args, "model_providers.openai_token_sharing.requires_openai_auth=false"); - assert.notInclude(effective.config.launchArgs, "dummy-owned-access"); + assert.include(providerConfig!, "supports_websockets=false"); + assert.include(providerConfig!, "requires_openai_auth=false"); + for (const sensitive of [ + "dummy-owned-access", + "dummy-refresh", + "dummy-id-token", + "test-user", + ]) { + assert.notInclude(effective.config.launchArgs, sensitive); + assert.notInclude(JSON.stringify(effective.environment), sensitive); + } assert.strictEqual(ambient.CODEX_HOME, "/user/.codex"); + if (source === "managed" && account === "primary") { + yield* runtime.auth.controller.logout(Effect.void); + const closed = yield* Effect.promise(() => + readToken(endpoint!).then( + () => false, + () => true, + ), + ); + assert.isTrue(closed); + } else if (source === "local" && account === "primary" && runtimeHost === "node") { + yield* runtime.auth.controller.invalidate!; + const closed = yield* Effect.promise(() => + readToken(endpoint!).then( + () => false, + () => true, + ), + ); + assert.isTrue(closed); + } }).pipe(Effect.scoped); assert.strictEqual(leases, 0); + assert.equal(JSON.stringify(ambient), ambientBefore); yield* runtime.auth.controller.logout(Effect.void); assert.equal( yield* fs.readFileString(path.join(sharedHome, "auth.json")), "native-auth-unchanged", ); + assert.equal( + yield* fs.readFileString(path.join(sharedHome, "config.toml")), + "# shared config\n", + ); }).pipe( Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { diff --git a/apps/server/src/provider/CodexManagedRuntime.ts b/apps/server/src/provider/CodexManagedRuntime.ts index 826d777f1bd70..d9115850f313d 100644 --- a/apps/server/src/provider/CodexManagedRuntime.ts +++ b/apps/server/src/provider/CodexManagedRuntime.ts @@ -4,9 +4,12 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import * as ServerConfig from "../config.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; -import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import { codexChatGptProfileKey, makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import { makeCodexManagedTokenSource } from "./CodexManagedTokenSource.ts"; +import { codexManagedTokenCommandName } from "./codexManagedTokenCommand.ts"; import { materializeCodexShadowHome } from "./Drivers/CodexHomeLayout.ts"; export interface CodexEffectiveRuntime { @@ -15,21 +18,24 @@ export interface CodexEffectiveRuntime { readonly revision: string; } const decodeSettings = Schema.decodeSync(CodexSettings); -const managedCodexLaunchArgs = [ - 'model_provider="openai_token_sharing"', - 'model_providers.openai_token_sharing.name="OpenAI Token Sharing"', - 'model_providers.openai_token_sharing.base_url="https://api.openai.com/v1"', - 'model_providers.openai_token_sharing.model_catalog_url="https://api.openai.com/v1/models"', - "features.api_key_model_discovery=true", - "features.multi_agent=false", - "features.multi_agent_v2=false", - 'model_providers.openai_token_sharing.env_key="ACCESS_TOKEN"', - 'model_providers.openai_token_sharing.wire_api="responses"', - "model_providers.openai_token_sharing.requires_openai_auth=false", - "model_providers.openai_token_sharing.supports_websockets=false", -] - .map((value) => `-c '${value}'`) - .join(" "); +const managedCodexLaunchArgs = ( + profileKey: string, + command: string, + args: ReadonlyArray, +) => { + const providerKey = `openai_token_sharing_${profileKey}`; + return [ + `model_provider=${JSON.stringify(providerKey)}`, + // Use a profile-specific inline table so recursive config merging cannot + // inherit env_key/auth settings from a user's openai_token_sharing entry. + `model_providers.${providerKey}={name="OpenAI Token Sharing",base_url="https://api.openai.com/v1",model_catalog_url="https://api.openai.com/v1/models",wire_api="responses",requires_openai_auth=false,supports_websockets=false,auth={command=${JSON.stringify(command)},args=${JSON.stringify(args)},timeout_ms=80000,refresh_interval_ms=0}}`, + "features.api_key_model_discovery=true", + "features.multi_agent=false", + "features.multi_agent_v2=false", + ] + .map((value) => `-c ${JSON.stringify(value)}`) + .join(" "); +}; export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(function* (options: { readonly instanceId: ProviderInstanceId; @@ -41,6 +47,7 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func const config = yield* ServerConfig.ServerConfig; const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; + const invocation = yield* resolveSelfInvocation(); const auth = yield* makeCodexChatGptAuth({ instanceId: options.instanceId, defaultReturnUrl: new URL( @@ -54,60 +61,77 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func options.config, ); const homePath = homeLayout.effectiveHomePath ?? homeLayout.sharedHomePath; - const resolve = Effect.gen(function* () { - const executable = yield* installation.acquire().pipe( - Effect.mapError( - () => - new ProviderSetupError({ - instanceId: options.instanceId, - operation: "install", - detail: "Set up managed Codex before starting a session.", - }), - ), - ); - const credentials = yield* auth.access; - yield* materializeCodexShadowHome(homeLayout).pipe( - Effect.provideService(FileSystem.FileSystem, fs), - Effect.provideService(Path.Path, path), - Effect.mapError( - (cause) => - new ProviderSetupError({ - instanceId: options.instanceId, - operation: "runtime", - detail: cause.message, - }), - ), - ); - yield* fs.makeDirectory(homePath, { recursive: true }).pipe( - Effect.mapError( - () => - new ProviderSetupError({ - instanceId: options.instanceId, - operation: "runtime", - detail: "Could not prepare the managed Codex runtime.", - }), - ), - ); - // Ambient CLI overrides cannot redirect a T3-owned token to a different provider. - const environment: NodeJS.ProcessEnv = { - ...options.environment, - ACCESS_TOKEN: credentials.accessToken, - CODEX_HOME: homePath, - }; - delete environment.T3CODE_CODEX_LAUNCH_ARGS; - delete environment.OPENAI_API_KEY; - delete environment.OPENAI_BASE_URL; - return { - config: decodeSettings({ - enabled: options.enabled, - setupMode: "managed", - binaryPath: executable.executablePath, - homePath, - launchArgs: managedCodexLaunchArgs, - }), - environment, - revision: credentials.accessToken, - } satisfies CodexEffectiveRuntime; - }); - return { auth, resolve, installation, homePath, homeLayout }; + const resolveForProfile = (expectedProfileKey?: string) => + Effect.gen(function* () { + const executable = yield* installation.acquire().pipe( + Effect.mapError( + () => + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "install", + detail: "Set up managed Codex before starting a session.", + }), + ), + ); + yield* materializeCodexShadowHome(homeLayout).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + Effect.mapError( + (cause) => + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "runtime", + detail: cause.message, + }), + ), + ); + yield* fs.makeDirectory(homePath, { recursive: true }).pipe( + Effect.mapError( + () => + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "runtime", + detail: "Could not prepare the managed Codex runtime.", + }), + ), + ); + // Finish local runtime preparation before touching refreshable OAuth state. + const credentials = + expectedProfileKey === undefined + ? yield* auth.access + : yield* auth.accessForProfile(expectedProfileKey); + const profileKey = codexChatGptProfileKey(credentials); + const source = yield* makeCodexManagedTokenSource( + options.instanceId, + credentials, + auth.accessForProfile(profileKey), + ); + const authArgs = selfInvocationArgs(invocation, [ + codexManagedTokenCommandName, + source.endpoint, + ]); + // Ambient CLI overrides cannot redirect a T3-owned token to a different provider. + // The auth command receives only a private local IPC endpoint, never a token or capability. + const environment: NodeJS.ProcessEnv = { + ...options.environment, + CODEX_HOME: homePath, + }; + delete environment.ACCESS_TOKEN; + delete environment.T3CODE_CODEX_LAUNCH_ARGS; + delete environment.OPENAI_API_KEY; + delete environment.OPENAI_BASE_URL; + return { + config: decodeSettings({ + enabled: options.enabled, + setupMode: "managed", + binaryPath: executable.executablePath, + homePath, + launchArgs: managedCodexLaunchArgs(source.profileKey, invocation.command, authArgs), + }), + environment, + revision: source.profileKey, + } satisfies CodexEffectiveRuntime; + }); + const resolve = resolveForProfile(); + return { auth, resolve, resolveForProfile, installation, homePath, homeLayout }; }); diff --git a/apps/server/src/provider/CodexManagedTokenSource.test.ts b/apps/server/src/provider/CodexManagedTokenSource.test.ts new file mode 100644 index 0000000000000..bf973c6845763 --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenSource.test.ts @@ -0,0 +1,461 @@ +// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - Exercises native socket timing at the private IPC boundary. +import * as NodeFSP from "node:fs/promises"; +import * as NodeNet from "node:net"; +import * as NodePath from "node:path"; +import * as NodeTimers from "node:timers"; +import * as NodeTimersPromises from "node:timers/promises"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Scope from "effect/Scope"; +import { HttpClient, HttpClientResponse } from "effect/http"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { + codexChatGptProfileKey, + makeCodexChatGptAuth, + type CodexChatGptCredentials, +} from "./CodexChatGptAuth.ts"; +import { makeCodexManagedTokenSource } from "./CodexManagedTokenSource.ts"; +import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; + +const instanceId = ProviderInstanceId.make("codex-managed-token-source-test"); +const windowsHost = HostProcess.Platform.defaultValue() === "win32"; +const credentials = { + clientId: "oaiapp_test", + accessToken: "access-a", + refreshToken: "refresh-a", + expiresAt: Number.MAX_SAFE_INTEGER, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + subject: "subject-a", + email: null, + idToken: "identity-a", + issuer: "https://auth.openai.com", +} satisfies CodexChatGptCredentials; + +const request = (source: { readonly endpoint: string }, frame = "token\n") => + new Promise((resolve) => { + const socket = NodeNet.createConnection(source.endpoint); + const chunks: Buffer[] = []; + let settled = false; + const finish = () => { + if (settled) return; + settled = true; + resolve(Buffer.concat(chunks).toString("utf8")); + }; + socket.once("connect", () => socket.write(frame)); + socket.on("data", (chunk: Buffer) => chunks.push(chunk)); + socket.once("end", finish); + socket.once("close", finish); + socket.once("error", finish); + }); + +it.effect("serves only the launch profile through a private scoped IPC endpoint", () => + Effect.gen(function* () { + const current = yield* Ref.make(credentials); + const source = yield* makeCodexManagedTokenSource(instanceId, credentials, Ref.get(current)); + + assert.equal(yield* Effect.promise(() => request(source, "wrong\n")), ""); + assert.equal(yield* Effect.promise(() => request(source)), "access-a"); + + yield* Ref.set(current, { ...credentials, accessToken: "access-b" }); + assert.equal(yield* Effect.promise(() => request(source)), "access-b"); + + yield* Ref.set(current, { + ...credentials, + idToken: "replacement-identity", + accessToken: "must-not-leak", + }); + assert.equal(yield* Effect.promise(() => request(source)), ""); + + if (windowsHost) { + assert.match(source.endpoint, /^\\\\\.\\pipe\\t3-code-managed-codex-/u); + } else { + const socket = yield* Effect.promise(() => NodeFSP.lstat(source.endpoint)); + const directory = yield* Effect.promise(() => + NodeFSP.lstat(NodePath.dirname(source.endpoint)), + ); + assert.isTrue(socket.isSocket()); + assert.equal(socket.mode & 0o777, 0o600); + assert.equal(directory.mode & 0o777, 0o700); + } + }).pipe(Effect.scoped), +); + +it.effect("concurrent IPC reads rotate an expired token once and persist the new refresh", () => + Effect.gen(function* () { + const data = new Map(); + const secrets = ServerSecretStore.ServerSecretStore.of({ + get: (key) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))), + set: (key, value) => + Effect.sync(() => { + data.set(key, value); + }), + remove: (key) => + Effect.sync(() => { + data.delete(key); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + let exchanges = 0; + yield* Effect.gen(function* () { + const expired = { + ...credentials, + accessToken: "expired-access", + refreshToken: "once-only-refresh", + expiresAt: 0, + }; + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + yield* store.set(new TextEncoder().encode(JSON.stringify(expired))); + const auth = yield* makeCodexChatGptAuth({ instanceId }); + const profileKey = codexChatGptProfileKey(expired); + const source = yield* makeCodexManagedTokenSource( + instanceId, + expired, + auth.accessForProfile(profileKey), + ); + const responses = yield* Effect.all( + [Effect.promise(() => request(source)), Effect.promise(() => request(source))], + { concurrency: "unbounded" }, + ); + assert.deepEqual(responses, ["rotated-access", "rotated-access"]); + assert.equal(exchanges, 1); + const saved = Option.getOrThrow(yield* auth.read); + assert.equal(saved.accessToken, "rotated-access"); + assert.equal(saved.refreshToken, "rotated-refresh"); + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + if (request.url.endsWith("/.well-known/openid-configuration")) + return HttpClientResponse.fromWeb( + request, + Response.json({ + issuer: "https://auth.openai.com", + authorization_endpoint: "https://auth.openai.com/authorize", + token_endpoint: "https://auth.openai.com/token", + jwks_uri: "https://auth.openai.com/jwks", + }), + ); + assert.equal(request.url, "https://auth.openai.com/token"); + exchanges++; + assert.equal(exchanges, 1, "a rotating refresh token must not be replayed"); + return HttpClientResponse.fromWeb( + request, + Response.json({ + access_token: "rotated-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + token_type: "Bearer", + scope: "chatgpt.tokens.use.direct", + }), + ); + }), + ), + ), + Effect.provide(NodeServices.layer), + ); + }).pipe(Effect.scoped), +); + +it.effect("shares one in-flight credential read across concurrent helper requests", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + let reads = 0; + const access = Effect.gen(function* () { + reads++; + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + return credentials; + }); + const source = yield* makeCodexManagedTokenSource(instanceId, credentials, access); + const requests = yield* Effect.all( + Array.from({ length: 8 }, () => Effect.promise(() => request(source))), + { concurrency: "unbounded" }, + ).pipe(Effect.forkChild); + + yield* Deferred.await(entered); + for (let turn = 0; turn < 5; turn++) + yield* Effect.promise( + () => + new Promise((resolve) => { + setImmediate(resolve); + }), + ); + assert.equal(reads, 1); + + yield* Deferred.succeed(release, undefined); + assert.deepEqual(yield* Fiber.join(requests), Array(8).fill("access-a")); + }).pipe(Effect.scoped), +); + +it.effect("enforces an absolute handshake deadline even while a peer trickles bytes", () => + Effect.gen(function* () { + const source = yield* makeCodexManagedTokenSource( + instanceId, + credentials, + Effect.succeed(credentials), + ); + const closedBeforeWatchdog = yield* Effect.promise( + () => + new Promise((resolve) => { + const socket = NodeNet.createConnection(source.endpoint); + let settled = false; + let interval: NodeJS.Timeout | undefined; + let watchdog: NodeJS.Timeout | undefined; + const finish = (closed: boolean) => { + if (settled) return; + settled = true; + if (interval !== undefined) NodeTimers.clearInterval(interval); + if (watchdog !== undefined) NodeTimers.clearTimeout(watchdog); + socket.destroy(); + resolve(closed); + }; + socket.once("connect", () => { + interval = NodeTimers.setInterval(() => { + if (!socket.destroyed) socket.write("t"); + }, 500); + watchdog = NodeTimers.setTimeout(() => finish(false), 7_000); + }); + socket.once("close", () => finish(true)); + socket.once("error", () => finish(true)); + }), + ); + assert.isTrue(closedBeforeWatchdog); + }).pipe(Effect.scoped), +); + +it.effect("profile-bound access rejects replacement before consuming its refresh token", () => + Effect.gen(function* () { + const data = new Map(); + const secrets = ServerSecretStore.ServerSecretStore.of({ + get: (key) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))), + set: (key, value) => + Effect.sync(() => { + data.set(key, value); + }), + remove: (key) => + Effect.sync(() => { + data.delete(key); + }), + create: () => Effect.die("unused"), + getOrCreateRandom: () => Effect.die("unused"), + }); + let exchanges = 0; + yield* Effect.gen(function* () { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + yield* store.set(new TextEncoder().encode(JSON.stringify(credentials))); + const auth = yield* makeCodexChatGptAuth({ instanceId }); + const source = yield* makeCodexManagedTokenSource( + instanceId, + credentials, + auth.accessForProfile(codexChatGptProfileKey(credentials)), + ); + yield* store.set( + new TextEncoder().encode( + JSON.stringify({ + ...credentials, + idToken: "replacement-identity", + accessToken: "replacement-access", + refreshToken: "replacement-refresh", + expiresAt: 0, + }), + ), + ); + assert.equal(yield* Effect.promise(() => request(source)), ""); + assert.equal(exchanges, 0); + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + if (request.url.endsWith("/.well-known/openid-configuration")) + return HttpClientResponse.fromWeb( + request, + Response.json({ + issuer: "https://auth.openai.com", + authorization_endpoint: "https://auth.openai.com/authorize", + token_endpoint: "https://auth.openai.com/token", + jwks_uri: "https://auth.openai.com/jwks", + }), + ); + exchanges++; + return HttpClientResponse.fromWeb( + request, + Response.json({ + access_token: "must-not-be-used", + refresh_token: "must-not-be-used", + expires_in: 3600, + token_type: "Bearer", + scope: "chatgpt.tokens.use.direct", + }), + ); + }), + ), + ), + Effect.provide(NodeServices.layer), + ); + }).pipe(Effect.scoped), +); + +it.effect("finishes an admitted refresh after the helper disconnects", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const completed = yield* Deferred.make(); + const access = Effect.gen(function* () { + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + yield* Deferred.succeed(completed, undefined); + return { ...credentials, accessToken: "rotated-after-disconnect" }; + }); + const source = yield* makeCodexManagedTokenSource(instanceId, credentials, access); + const pending = NodeNet.createConnection(source.endpoint); + pending.on("error", () => {}); + pending.once("connect", () => pending.write("token\n")); + + yield* Deferred.await(entered); + pending.destroy(); + yield* Deferred.succeed(release, undefined); + yield* Deferred.await(completed); + }).pipe(Effect.scoped), +); + +it.effect("scope shutdown drains an admitted credential refresh", () => + Effect.gen(function* () { + let enter!: () => void; + const entered = new Promise((resolve) => { + enter = resolve; + }); + let release!: () => void; + let completed = false; + const access = Effect.promise( + () => + new Promise((resolve) => { + enter(); + release = () => { + completed = true; + resolve({ ...credentials, accessToken: "rotated-before-close" }); + }; + }), + ); + const scope = yield* Scope.make(); + const source = yield* makeCodexManagedTokenSource(instanceId, credentials, access).pipe( + Scope.provide(scope), + ); + const pending = NodeNet.createConnection(source.endpoint); + pending.on("error", () => {}); + pending.once("connect", () => pending.write("token\n")); + yield* Effect.promise(() => entered); + + void NodeTimersPromises.setTimeout(50).then(release); + yield* Scope.close(scope, Exit.void); + assert.isTrue(completed); + pending.destroy(); + }), +); + +it.effect("does not admit new credential reads after scope shutdown begins", () => + Effect.gen(function* () { + let reads = 0; + const scope = yield* Scope.make(); + const source = yield* makeCodexManagedTokenSource( + instanceId, + credentials, + Effect.sync(() => { + reads++; + return credentials; + }), + ).pipe(Scope.provide(scope)); + const socket = NodeNet.createConnection(source.endpoint); + socket.on("error", () => {}); + yield* Effect.promise( + () => + new Promise((resolve) => { + socket.once("connect", resolve); + }), + ); + const closed = new Promise((resolve) => socket.once("close", () => resolve())); + const shutdown = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild); + yield* Effect.yieldNow; + socket.write("token\n"); + yield* Fiber.join(shutdown); + yield* Effect.promise(() => closed); + assert.equal(reads, 0); + }), +); + +it.effect("scope shutdown drops requestless peers and removes the IPC endpoint", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const source = yield* makeCodexManagedTokenSource( + instanceId, + credentials, + Effect.succeed(credentials), + ).pipe(Scope.provide(scope)); + const socket = NodeNet.createConnection(source.endpoint); + let unsolicitedBytes = 0; + socket.on("error", () => {}); + socket.on("data", (chunk: Buffer) => { + unsolicitedBytes += chunk.length; + }); + yield* Effect.promise( + () => + new Promise((resolve) => { + socket.once("connect", resolve); + }), + ); + const closed = new Promise((resolve) => socket.once("close", () => resolve())); + yield* Scope.close(scope, Exit.void); + yield* Effect.promise(() => closed); + assert.equal(unsolicitedBytes, 0); + + if (!windowsHost) { + const remains = yield* Effect.promise(() => + NodeFSP.lstat(NodePath.dirname(source.endpoint)).then( + () => true, + () => false, + ), + ); + assert.isFalse(remains); + } + }), +); + +it.effect("closes the IPC listener with its owning scope", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const source = yield* makeCodexManagedTokenSource( + instanceId, + credentials, + Effect.succeed(credentials), + ).pipe(Scope.provide(scope)); + + assert.equal(yield* Effect.promise(() => request(source)), "access-a"); + yield* Scope.close(scope, Exit.void); + assert.equal(yield* Effect.promise(() => request(source)), ""); + }), +); diff --git a/apps/server/src/provider/CodexManagedTokenSource.ts b/apps/server/src/provider/CodexManagedTokenSource.ts new file mode 100644 index 0000000000000..c09f165c2b1f3 --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenSource.ts @@ -0,0 +1,252 @@ +// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - Native socket callbacks own absolute handshake deadlines. +import * as NodeCrypto from "node:crypto"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeNet from "node:net"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeTimers from "node:timers"; +import { ProviderSetupError, type ProviderInstanceId } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Effect from "effect/Effect"; +import { codexChatGptProfileKey, type CodexChatGptCredentials } from "./CodexChatGptAuth.ts"; + +const REQUEST = Buffer.from("token\n"); +const REQUEST_BYTES = 64; +const RESPONSE_BYTES = 64 * 1024; +const HANDSHAKE_TIMEOUT_MS = 5_000; +const UNIX_SOCKET_PATH_BYTES = 100; + +interface OwnedEndpoint { + readonly endpoint: string; + readonly directory?: string; +} + +const privateEndpoint = async ( + platform: NodeJS.Platform, + uid: number | undefined, +): Promise => { + if (platform === "win32") { + // Keep Node/libuv's default named-pipe DACL; do not broaden it with + // readableAll/writableAll. The random name is routing, not a same-user capability. + return { + endpoint: `\\\\.\\pipe\\t3-code-managed-codex-${NodeCrypto.randomUUID()}`, + }; + } + + if (uid === undefined) throw new Error("Missing local user identity"); + + const candidates = [NodeOS.tmpdir(), "/tmp"]; + let lastError: unknown; + for (const candidate of candidates) { + try { + const parent = await NodeFSP.realpath(candidate); + const stat = await NodeFSP.lstat(parent); + const trusted = + stat.isDirectory() && + !stat.isSymbolicLink() && + ((stat.uid === uid && (stat.mode & 0o022) === 0) || + (stat.uid === 0 && ((stat.mode & 0o022) === 0 || (stat.mode & 0o1000) !== 0))); + if (!trusted) throw new Error("Untrusted IPC parent"); + + const directory = await NodeFSP.mkdtemp(NodePath.join(parent, "t3-code-token-")); + try { + await NodeFSP.chmod(directory, 0o700); + const created = await NodeFSP.lstat(directory); + if ( + !created.isDirectory() || + created.isSymbolicLink() || + created.uid !== uid || + (created.mode & 0o777) !== 0o700 + ) { + throw new Error("Unprotected IPC directory"); + } + const endpoint = NodePath.join(directory, "s"); + if (Buffer.byteLength(endpoint) > UNIX_SOCKET_PATH_BYTES) + throw new Error("IPC path is too long"); + return { endpoint, directory }; + } catch (cause) { + await NodeFSP.rmdir(directory).catch(() => undefined); + throw cause; + } + } catch (cause) { + lastError = cause; + } + } + throw lastError ?? new Error("Could not allocate private IPC endpoint"); +}; + +export const makeCodexManagedTokenSource = Effect.fn("makeCodexManagedTokenSource")(function* ( + instanceId: ProviderInstanceId, + launchCredentials: CodexChatGptCredentials, + access: Effect.Effect, +) { + const platform = yield* HostProcess.Platform; + const uid = yield* HostProcess.UserId; + const context = yield* Effect.context(); + const runPromise = Effect.runPromiseWith(context); + const launchProfileKey = codexChatGptProfileKey(launchCredentials); + const inFlight = new Set>(); + const sockets = new Set(); + let credentialRead: Promise | undefined; + let closing = false; + + // Concurrent helpers must share one credential read so a failing refresh cannot + // queue repeated serialized OAuth attempts behind the auth/session locks. + const readCredentials = () => { + if (credentialRead !== undefined) return credentialRead; + const pending = runPromise(access); + credentialRead = pending; + const clear = () => { + if (credentialRead === pending) credentialRead = undefined; + }; + void pending.then(clear, clear); + return pending; + }; + + const owned = yield* Effect.acquireRelease( + Effect.tryPromise({ + try: () => privateEndpoint(platform, uid), + catch: (cause) => + new ProviderSetupError({ + instanceId, + operation: "runtime", + detail: "Could not prepare the private managed Codex credential channel.", + cause, + }), + }), + ({ endpoint, directory }) => + Effect.promise(async () => { + if (platform !== "win32") await NodeFSP.unlink(endpoint).catch(() => undefined); + if (directory) await NodeFSP.rmdir(directory).catch(() => undefined); + }), + ); + + // Acquire the listener after its endpoint owner. Scope finalizers run LIFO, + // so shutdown closes/drains the server before unlinking/removing the endpoint. + const server = yield* Effect.acquireRelease( + Effect.tryPromise({ + try: () => + new Promise((resolve, reject) => { + const server = NodeNet.createServer((socket) => { + if (closing || sockets.size >= 32) { + socket.destroy(); + return; + } + sockets.add(socket); + const handshakeTimer = NodeTimers.setTimeout( + () => socket.destroy(), + HANDSHAKE_TIMEOUT_MS, + ); + handshakeTimer.unref(); + socket.once("close", () => { + NodeTimers.clearTimeout(handshakeTimer); + sockets.delete(socket); + }); + socket.once("error", () => {}); + + let request = Buffer.alloc(0); + const onData = (chunk: Buffer) => { + if (request.length + chunk.length > REQUEST_BYTES) { + socket.destroy(); + return; + } + request = Buffer.concat([request, chunk]); + const newline = request.indexOf(10); + if (newline < 0) return; + socket.off("data", onData); + if (closing || newline !== request.length - 1 || !request.equals(REQUEST)) { + socket.destroy(); + return; + } + + NodeTimers.clearTimeout(handshakeTimer); + let finish!: () => void; + const tracked = new Promise((resolve) => { + finish = resolve; + }); + inFlight.add(tracked); + void readCredentials() + .then((credentials) => { + if ( + socket.destroyed || + codexChatGptProfileKey(credentials) !== launchProfileKey || + Buffer.byteLength(credentials.accessToken) > RESPONSE_BYTES + ) { + socket.destroy(); + return; + } + socket.end(credentials.accessToken); + }) + .catch(() => socket.destroy()) + .finally(() => { + inFlight.delete(tracked); + finish(); + }); + }; + socket.on("data", onData); + }); + server.maxConnections = 32; + const onStartupError = (cause: Error) => reject(cause); + server.once("error", onStartupError); + server.listen(owned.endpoint, () => { + server.off("error", onStartupError); + // A post-start listener error must fail this credential channel, + // not crash the whole T3 server process or admit more credential work. + server.on("error", () => { + closing = true; + for (const socket of sockets) socket.destroy(); + if (server.listening) server.close(); + }); + resolve(server); + }); + }), + catch: (cause) => + new ProviderSetupError({ + instanceId, + operation: "runtime", + detail: "Could not start the private managed Codex credential channel.", + cause, + }), + }), + (server) => + Effect.promise(async () => { + // This is the admission cutoff: callbacks that run after this point + // must not start a fresh credential read. + closing = true; + for (const socket of sockets) socket.destroy(); + const closed = new Promise((resolve) => { + if (!server.listening) { + resolve(); + return; + } + server.close(() => resolve()); + }); + await Promise.allSettled(inFlight); + await closed; + }), + ); + if (platform !== "win32") + yield* Effect.tryPromise({ + try: () => NodeFSP.chmod(owned.endpoint, 0o600), + catch: (cause) => + new ProviderSetupError({ + instanceId, + operation: "runtime", + detail: "Could not secure the private managed Codex credential channel.", + cause, + }), + }); + + const address = server.address(); + if (address === null) + return yield* new ProviderSetupError({ + instanceId, + operation: "runtime", + detail: "Could not resolve the private managed Codex credential channel.", + }); + + return { + endpoint: owned.endpoint, + profileKey: launchProfileKey, + }; +}); diff --git a/apps/server/src/provider/Drivers/CodexDriver.test.ts b/apps/server/src/provider/Drivers/CodexDriver.test.ts index 51301ef43c50f..747144e7ed43e 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.test.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.test.ts @@ -205,7 +205,7 @@ it.layer(layerTest)("CodexDriver", (it) => { expect(restored.auth.email).toBe("account@example.test"); expect(restored.runtimePaths?.homePath).toBe(sharedHome); expect(restored.runtimePaths?.shadowHomePath).toContain( - `providers/codex/${instanceId}/shadow`, + NodePath.join("providers", "codex", instanceId, "shadow"), ); yield* Deferred.await(observedAccount); @@ -226,7 +226,10 @@ it.layer(layerTest)("CodexDriver", (it) => { const launch = launches[0]!; expect(launch.settings.binaryPath).toBe(executable.executablePath); expect(launch.settings.launchArgs).toContain("openai_token_sharing"); - expect(launch.environment.ACCESS_TOKEN).toBe("dummy-owned-access"); + expect(launch.settings.launchArgs).toContain("__managed-chatgpt-token"); + expect(launch.environment.ACCESS_TOKEN).toBeUndefined(); + expect(launch.environment.T3CODE_MANAGED_CODEX_AUTH_URL).toBeUndefined(); + expect(launch.environment.T3CODE_MANAGED_CODEX_AUTH_SECRET).toBeUndefined(); expect(launch.environment.OPENAI_API_KEY).toBeUndefined(); expect(launch.environment.CODEX_HOME).toBe(launch.settings.homePath); const before = yield* instance.auth!.subscribe("test-owner").pipe(Stream.runHead); diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index eb44345840232..298ba023ed566 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -1,4 +1,9 @@ -import { ProviderDriverKind, TextGenerationError, type CodexSettings } from "@t3tools/contracts"; +import { + ProviderDriverKind, + ProviderSetupError, + TextGenerationError, + type CodexSettings, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; @@ -9,6 +14,7 @@ import { makeCodexTextGeneration } from "../../textGeneration/CodexTextGeneratio import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import { chatGptModels } from "../CodexChatGptModels.ts"; import { makeCodexManagedRuntime } from "../CodexManagedRuntime.ts"; +import { codexChatGptProfileKey } from "../CodexChatGptAuth.ts"; import { ProviderDriverError } from "../Errors.ts"; import { createCodexAdapterV2 } from "../../orchestration-v2/Adapters/CodexAdapterV2.ts"; import { @@ -119,7 +125,10 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu label: "ChatGPT", ...(saved.value.email?.trim() ? { email: saved.value.email.trim() } : {}), }; - return yield* runtime.auth.controller.withAccess!(runtime.resolve).pipe( + const savedProfileKey = codexChatGptProfileKey(saved.value); + return yield* runtime.auth.controller.withAccess!( + runtime.resolveForProfile(savedProfileKey), + ).pipe( Effect.flatMap((effective) => Effect.gen(function* () { const draft = yield* checkCodexProviderStatus( @@ -128,10 +137,29 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu effective.environment, managedAuth, ); - const models = yield* chatGptModels( - effective.environment.ACCESS_TOKEN!, - draft.models, - ).pipe(Effect.provideService(HttpClient.HttpClient, http)); + const credentials = yield* runtime.auth.accessForProfile(effective.revision); + if ( + codexChatGptProfileKey(credentials) !== effective.revision || + savedProfileKey !== effective.revision + ) + return yield* new ProviderSetupError({ + instanceId, + operation: "models", + detail: "The ChatGPT connection changed while checking Codex. Retry.", + }); + const models = yield* chatGptModels(credentials.accessToken, draft.models).pipe( + Effect.provideService(HttpClient.HttpClient, http), + ); + const current = yield* runtime.auth.read; + if ( + Option.isNone(current) || + codexChatGptProfileKey(current.value) !== effective.revision + ) + return yield* new ProviderSetupError({ + instanceId, + operation: "models", + detail: "The ChatGPT connection changed while checking Codex. Retry.", + }); return { ...draft, models }; }), ), diff --git a/apps/server/src/provider/codexManagedTokenCommand.test.ts b/apps/server/src/provider/codexManagedTokenCommand.test.ts new file mode 100644 index 0000000000000..0d698ae4a4b4b --- /dev/null +++ b/apps/server/src/provider/codexManagedTokenCommand.test.ts @@ -0,0 +1,125 @@ +// @effect-diagnostics nodeBuiltinImport:off - launches the actual fast-path helper process. +import * as NodeChildProcess from "node:child_process"; +import * as NodeCrypto from "node:crypto"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeNet from "node:net"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeURL from "node:url"; +import { assert, it } from "@effect/vitest"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Effect from "effect/Effect"; + +const bin = NodeURL.fileURLToPath(new URL("../bin.ts", import.meta.url)); +const windowsHost = HostProcess.Platform.defaultValue() === "win32"; +const endpoint = () => + windowsHost + ? `\\\\.\\pipe\\t3-code-token-test-${NodeCrypto.randomUUID()}` + : NodePath.join(NodeOS.tmpdir(), `t3-code-token-test-${NodeCrypto.randomUUID()}.sock`); + +const run = (args: ReadonlyArray) => + Effect.promise( + () => + new Promise<{ code: number | null; stdout: string; stderr: string }>((resolve, reject) => { + const child = NodeChildProcess.spawn( + process.execPath, + [bin, "__managed-chatgpt-token", ...args], + { + env: process.env, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk)); + child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk)); + child.once("error", reject); + child.once("close", (code) => resolve({ code, stdout, stderr })); + }), + ); + +const serve = (body: string, onRequest?: (request: string) => void) => + Effect.acquireRelease( + Effect.promise( + () => + new Promise<{ server: NodeNet.Server; endpoint: string }>((resolve, reject) => { + const address = endpoint(); + const server = NodeNet.createServer((socket) => { + let request = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + request += chunk; + if (!request.includes("\n")) return; + onRequest?.(request); + socket.end(body); + }); + }); + server.once("error", reject); + server.listen(address, () => resolve({ server, endpoint: address })); + }), + ), + ({ server, endpoint }) => + Effect.promise( + () => + new Promise((resolve) => { + server.close(() => { + if (!windowsHost) + void NodeFSP.unlink(endpoint) + .catch(() => undefined) + .finally(resolve); + else resolve(); + }); + }), + ), + ); + +it.effect("fast-path helper prints only the token supplied by its scoped IPC channel", () => + Effect.scoped( + Effect.gen(function* () { + let request = ""; + const source = yield* serve("token-from-bridge", (value) => { + request = value; + }); + const result = yield* run([source.endpoint]); + assert.equal(request, "token\n"); + assert.equal(result.code, 0); + assert.equal(result.stdout, "token-from-bridge"); + assert.equal(result.stderr, ""); + }), + ), +); + +it.effect.each([ + { label: "empty", body: "" }, + { label: "oversized", body: "x".repeat(64 * 1024 + 1) }, +])("helper rejects $label token responses", ({ body }) => + Effect.scoped( + Effect.gen(function* () { + const source = yield* serve(body); + const result = yield* run([source.endpoint]); + assert.equal(result.code, 1); + assert.equal(result.stdout, ""); + assert.equal(result.stderr, "Could not renew managed Codex credentials.\n"); + }), + ), +); + +it.effect("helper fails closed for an unavailable IPC endpoint", () => + Effect.gen(function* () { + const result = yield* run([endpoint()]); + assert.equal(result.code, 1); + assert.equal(result.stdout, ""); + assert.equal(result.stderr, "Could not renew managed Codex credentials.\n"); + }), +); + +it.effect.each([[], ["one", "two"]] as const)( + "helper rejects malformed command arguments: %j", + (args) => + Effect.gen(function* () { + const result = yield* run(args); + assert.equal(result.code, 1); + assert.equal(result.stdout, ""); + assert.equal(result.stderr, "Could not renew managed Codex credentials.\n"); + }), +); diff --git a/apps/server/src/provider/codexManagedTokenCommand.ts b/apps/server/src/provider/codexManagedTokenCommand.ts new file mode 100644 index 0000000000000..ec9aa4c7d0e6f --- /dev/null +++ b/apps/server/src/provider/codexManagedTokenCommand.ts @@ -0,0 +1,67 @@ +// @effect-diagnostics nodeBuiltinImport:off - lean self-invoked private IPC client. +import * as NodeNet from "node:net"; + +export const codexManagedTokenCommandName = "__managed-chatgpt-token"; + +const REQUEST = "token\n"; +const MAX_ENDPOINT_BYTES = 2048; +const MAX_TOKEN_BYTES = 64 * 1024; +const REQUEST_TIMEOUT_MS = 80_000; + +const readToken = (endpoint: string) => + new Promise((resolve, reject) => { + let settled = false; + const fail = () => { + if (settled) return; + settled = true; + reject(new Error("managed Codex credential transport failed")); + }; + if ( + endpoint.length === 0 || + Buffer.byteLength(endpoint) > MAX_ENDPOINT_BYTES || + endpoint.includes("\0") + ) { + fail(); + return; + } + + const socket = NodeNet.createConnection(endpoint); + const chunks: Buffer[] = []; + let bytes = 0; + socket.setTimeout(REQUEST_TIMEOUT_MS, () => { + socket.destroy(); + fail(); + }); + socket.once("connect", () => socket.write(REQUEST)); + socket.on("data", (chunk: Buffer) => { + bytes += chunk.length; + if (bytes > MAX_TOKEN_BYTES) { + socket.destroy(); + fail(); + return; + } + chunks.push(chunk); + }); + socket.once("end", () => { + if (settled) return; + const token = Buffer.concat(chunks).toString("utf8").trim(); + if (token.length === 0) { + fail(); + return; + } + settled = true; + resolve(token); + }); + socket.once("error", fail); + }); + +export async function runCodexManagedTokenCommand(args: ReadonlyArray): Promise { + try { + if (args.length !== 1) throw new Error("invalid managed Codex credential arguments"); + const token = await readToken(args[0]!); + process.stdout.write(token); + } catch { + process.stderr.write("Could not renew managed Codex credentials.\n"); + process.exitCode = 1; + } +} From 1d4d0fbd4d80bc2412b0649ca6554b2b087a2b37 Mon Sep 17 00:00:00 2001 From: Navneet <155615649+nkcbuilds@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:03:30 +0530 Subject: [PATCH 2/3] fix(server): serialize transferred credential validation --- .../src/provider/CodexChatGptAuth.test.ts | 6 ++- apps/server/src/provider/CodexChatGptAuth.ts | 11 ++-- packages/provider-core/src/server/auth.ts | 1 + .../src/server/providerAuthFlow.test.ts | 51 +++++++++++++++++++ .../src/server/providerAuthFlow.ts | 3 +- 5 files changed, 63 insertions(+), 9 deletions(-) diff --git a/apps/server/src/provider/CodexChatGptAuth.test.ts b/apps/server/src/provider/CodexChatGptAuth.test.ts index 33b786ffa87c9..561cce9750bef 100644 --- a/apps/server/src/provider/CodexChatGptAuth.test.ts +++ b/apps/server/src/provider/CodexChatGptAuth.test.ts @@ -1789,11 +1789,15 @@ it.effect("transferred profiles reject mismatched identities without overwriting { ...profile.credentials, scopes: ["openid"] }, { ...profile.credentials, expiresAt: 0 }, ]) { + let stopped = false; const result = yield* destination.auth.controller.importProfile!( { ...profile, credentials }, - Effect.void, + Effect.sync(() => { + stopped = true; + }), ).pipe(Effect.result); assert.strictEqual(result._tag, "Failure"); + assert.isFalse(stopped); assert.deepEqual(Option.getOrThrow(yield* destination.auth.read), original); } }), diff --git a/apps/server/src/provider/CodexChatGptAuth.ts b/apps/server/src/provider/CodexChatGptAuth.ts index d7b738456fbad..cd703154cd302 100644 --- a/apps/server/src/provider/CodexChatGptAuth.ts +++ b/apps/server/src/provider/CodexChatGptAuth.ts @@ -995,13 +995,10 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* return track( "transfer", { flow: "primary_handoff" }, - validate.pipe( - Effect.andThen( - controller.adoptCredentials!( - lock.withPermit(withSessionLock(saveImported)), - stopSessions, - ), - ), + controller.adoptCredentials!( + lock.withPermit(withSessionLock(saveImported)), + stopSessions, + lock.withPermit(validate), ), ); }; diff --git a/packages/provider-core/src/server/auth.ts b/packages/provider-core/src/server/auth.ts index 25f30b8fd0c43..df39ebeeeca52 100644 --- a/packages/provider-core/src/server/auth.ts +++ b/packages/provider-core/src/server/auth.ts @@ -21,6 +21,7 @@ export interface ProviderAuthController { readonly adoptCredentials?: ( update: Effect.Effect, stopSessions: Effect.Effect, + prepare?: Effect.Effect, ) => Effect.Effect; readonly isChangingCredentials?: Effect.Effect; readonly invalidate?: Effect.Effect; diff --git a/packages/provider-core/src/server/providerAuthFlow.test.ts b/packages/provider-core/src/server/providerAuthFlow.test.ts index 2351243f1e0ef..1f19855b1110a 100644 --- a/packages/provider-core/src/server/providerAuthFlow.test.ts +++ b/packages/provider-core/src/server/providerAuthFlow.test.ts @@ -557,3 +557,54 @@ it.effect("profile import stops owned sessions and gates access until credential ); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + +it.effect("profile import claims credential changes before pre-stop validation", () => + Effect.gen(function* () { + const validationStarted = yield* Deferred.make(); + const releaseValidation = yield* Deferred.make(); + let stopped = false; + let loggedOut = false; + const controller = yield* ProviderAuthFlow.make({ + instanceId, + credentialBinding: { owner: "t3", key: "handoff-validation" }, + methods: Effect.succeed([method]), + authenticate: () => Effect.void, + logout: Effect.sync(() => { + loggedOut = true; + }), + }); + const imported = yield* controller.adoptCredentials!( + Effect.void, + Effect.sync(() => { + stopped = true; + }), + Deferred.succeed(validationStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseValidation)), + Effect.andThen( + Effect.fail( + new ProviderSetupError({ + instanceId, + operation: "import", + detail: "invalid transferred credentials", + }), + ), + ), + ), + ).pipe(Effect.result, Effect.forkChild); + + yield* Deferred.await(validationStarted); + assert.isTrue(yield* controller.isChangingCredentials!); + const concurrentLogout = yield* controller.logout(Effect.void).pipe(Effect.result); + assert.strictEqual(concurrentLogout._tag, "Failure"); + assert.isFalse(loggedOut); + assert.isFalse(stopped); + + yield* Deferred.succeed(releaseValidation, undefined); + assert.strictEqual((yield* Fiber.join(imported))._tag, "Failure"); + assert.isFalse(stopped); + + yield* controller.logout(Effect.void); + assert.isTrue(loggedOut); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + diff --git a/packages/provider-core/src/server/providerAuthFlow.ts b/packages/provider-core/src/server/providerAuthFlow.ts index c596f8ebc3e5f..144cd55c0d028 100644 --- a/packages/provider-core/src/server/providerAuthFlow.ts +++ b/packages/provider-core/src/server/providerAuthFlow.ts @@ -140,7 +140,7 @@ export const make = Effect.fn("ProviderAuthFlow.make")(function* (options: { const controller: ProviderAuth.ProviderAuthController = { credentialBinding: options.credentialBinding, - adoptCredentials: (update, stopSessions) => + adoptCredentials: (update, stopSessions, prepare = Effect.void) => Effect.gen(function* () { yield* lock.withPermit( Effect.gen(function* () { @@ -154,6 +154,7 @@ export const make = Effect.fn("ProviderAuthFlow.make")(function* (options: { }), ); const result = yield* Effect.gen(function* () { + yield* prepare; yield* stopOwnedSessions; yield* stopSessions; yield* update; From 9bdbbcb4263c2a490260c21c268e88144b8fabc4 Mon Sep 17 00:00:00 2001 From: Navneet Chaudhary Date: Mon, 19 Oct 2026 18:11:11 +0530 Subject: [PATCH 3/3] chore(server): format credential race test --- packages/provider-core/src/server/providerAuthFlow.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/provider-core/src/server/providerAuthFlow.test.ts b/packages/provider-core/src/server/providerAuthFlow.test.ts index 1f19855b1110a..86cfe3369ab17 100644 --- a/packages/provider-core/src/server/providerAuthFlow.test.ts +++ b/packages/provider-core/src/server/providerAuthFlow.test.ts @@ -607,4 +607,3 @@ it.effect("profile import claims credential changes before pre-stop validation", assert.isTrue(loggedOut); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); -