From d45d1cb80755f279a623c692884f077c1ecfb872 Mon Sep 17 00:00:00 2001 From: Tim Aronsson <70634109+Timpan4@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:59:54 +0200 Subject: [PATCH] fix(server): renew managed ChatGPT tokens in active Codex turns --- apps/server/src/bin.ts | 11 +- .../src/cli/managedChatGptToken.test.ts | 224 ++++++ apps/server/src/cli/managedChatGptToken.ts | 48 ++ .../AcpRegistryOrchestratorV2.live.test.ts | 4 + .../CursorOrchestratorV2.live.test.ts | 4 + .../GrokOrchestratorV2.live.test.ts | 4 + .../OpenCode2OrchestratorV2.live.test.ts | 4 + .../src/provider/CodexChatGptAuth.test.ts | 53 +- apps/server/src/provider/CodexChatGptAuth.ts | 22 +- .../src/provider/CodexManagedRuntime.test.ts | 50 +- .../src/provider/CodexManagedRuntime.ts | 62 +- .../provider/CodexManagedTokenBroker.test.ts | 671 ++++++++++++++++++ .../src/provider/CodexManagedTokenBroker.ts | 229 ++++++ .../CodexManagedTokenProtocol.test.ts | 105 +++ .../src/provider/CodexManagedTokenProtocol.ts | 72 ++ .../src/provider/Drivers/CodexDriver.test.ts | 102 ++- .../src/provider/Drivers/CodexDriver.ts | 4 +- .../provider/Drivers/CodexManagedProvider.ts | 28 +- .../provider/ProviderInstanceRegistry.test.ts | 11 + .../src/provider/ProviderRegistry.test.ts | 4 + apps/server/src/server.ts | 2 + 21 files changed, 1645 insertions(+), 69 deletions(-) create mode 100644 apps/server/src/cli/managedChatGptToken.test.ts create mode 100644 apps/server/src/cli/managedChatGptToken.ts create mode 100644 apps/server/src/provider/CodexManagedTokenBroker.test.ts create mode 100644 apps/server/src/provider/CodexManagedTokenBroker.ts create mode 100644 apps/server/src/provider/CodexManagedTokenProtocol.test.ts create mode 100644 apps/server/src/provider/CodexManagedTokenProtocol.ts diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index 063093874503..a5db6c1032de 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -3,9 +3,9 @@ * * Every ACP agent spawns `t3 acp-mcp-bridge` while opening its session, and * terminal-fallback agents run `t3 acp-mcp-call` per tool call, so their - * startup sits on first-message latency. Both dispatch here before the full - * CLI module graph (seconds of evaluation) loads; everything else defers to - * the real CLI in ./binCli.ts. + * startup sits on first-message latency. They and Codex's managed-token helper + * dispatch here before the full CLI module graph (seconds of evaluation) + * loads; everything else defers to the real CLI in ./binCli.ts. */ import { isEntrypoint } from "./entrypoint.ts"; @@ -17,7 +17,10 @@ if ( }) ) { const command = process.argv[2]; - if (command === "acp-mcp-bridge" || command === "acp-mcp-call") { + if (command === "__managed-chatgpt-token") { + const { runManagedChatGptTokenCli } = await import("./cli/managedChatGptToken.ts"); + await runManagedChatGptTokenCli(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/cli/managedChatGptToken.test.ts b/apps/server/src/cli/managedChatGptToken.test.ts new file mode 100644 index 000000000000..c63ab7ab3c7a --- /dev/null +++ b/apps/server/src/cli/managedChatGptToken.test.ts @@ -0,0 +1,224 @@ +// @effect-diagnostics nodeBuiltinImport:off - Runs the actual configured auth command in separate Node processes. +import * as NodeURL from "node:url"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { CodexSettings, EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Clock from "effect/Clock"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { HttpClient, HttpClientResponse } from "effect/http"; +import * as CodexManagedTokenBroker from "../provider/CodexManagedTokenBroker.ts"; +import * as ServerConfig from "../config.ts"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as CodexInstallation from "../provider/CodexInstallation.ts"; +import { makeCodexManagedRuntime } from "../provider/CodexManagedRuntime.ts"; +import * as ProviderCredentialStore from "../provider/ProviderCredentialStore.ts"; +import { codexAppServerArgs } from "../provider/codexLaunchArgs.ts"; + +const decodeSettings = Schema.decodeSync(CodexSettings); +const decodeCommand = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.Array(Schema.String)])), +); + +it.live( + "configured auth command reads the exact store, rotates once across processes, and rejects replaced or logged-out profiles", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const instanceId = ProviderInstanceId.make("codex-personal"); + const home = path.join(config.baseDir, "Codex home's space"); + const preload = path.join(config.baseDir, "oauth-mock.mjs"); + let renewals = 0; + yield* fs.makeDirectory(home); + // Helpers must remain transport-only and avoid the full CLI/auth/store graph. + yield* fs.writeFileString( + preload, + ` + import { registerHooks } from 'node:module'; + if (process.env.ELECTRON_RUN_AS_NODE !== undefined) throw new Error('Node-hosted helper must not gain an Electron flag'); + registerHooks({ resolve(specifier, context, next) { + if (['binCli', 'CodexChatGptAuth', 'ServerSecretStore', 'CodexManagedTokenBroker'].some(s => specifier.includes(s))) throw new Error('Server graph must not load'); + return next(specifier, context); + } }); + globalThis.fetch = () => { throw new Error('Transport helper cannot use network'); }; + `, + ); + yield* Effect.gen(function* () { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + const credentials = { + clientId: "oaiapp_test", + subject: "fake-subject", + idToken: "fake-id-token", + accessToken: "fake-initial-access", + refreshToken: "fake-refresh", + email: null, + expiresAt: (yield* Clock.currentTimeMillis) + 3_600_000, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + }; + const save = (record: typeof credentials) => + store.set(new TextEncoder().encode(JSON.stringify(record))); + yield* save(credentials); + const runtime = yield* makeCodexManagedRuntime({ + instanceId, + enabled: true, + config: decodeSettings({ setupMode: "managed", homePath: home }), + environment: { + PATH: process.env.PATH, + SystemRoot: process.env.SystemRoot, + T3CODE_HOME: path.join(config.baseDir, "wrong-home"), + ACCESS_TOKEN: "ambient-secret", + OPENAI_BASE_URL: "https://wrong-endpoint.invalid", + OPENAI_API_KEY: "ambient-key", + }, + }); + const effective = yield* runtime.resolve; + assert.isUndefined(effective.environment.ELECTRON_RUN_AS_NODE); + const table = codexAppServerArgs(effective.config.launchArgs).find((arg) => + arg.startsWith("model_providers.openai_token_sharing_"), + )!; + const match = /auth=\{command=(.*),args=(\[.*\]),timeout_ms=/u.exec(table)!; + const [command, args] = decodeCommand(`[${match[1]},${match[2]}]`); + const broker = yield* CodexManagedTokenBroker.CodexManagedTokenBroker; + assert.include(args, yield* broker.endpoint); + assert.notInclude(args, config.secretsDir); + for (const secret of [ + credentials.accessToken, + credentials.refreshToken, + credentials.idToken, + credentials.subject, + ]) { + assert.notInclude(table, secret); + assert.notInclude(JSON.stringify(effective.environment), secret); + } + const run = Effect.gen(function* () { + const child = yield* spawner.spawn( + ChildProcess.make(command, args, { + extendEnv: false, + env: { + ...effective.environment, + NODE_OPTIONS: `--import ${JSON.stringify(NodeURL.pathToFileURL(preload).href)}`, + }, + }), + ); + const [stdout, stderr, code] = yield* Effect.all( + [ + child.stdout.pipe(Stream.decodeText(), Stream.mkString), + child.stderr.pipe(Stream.decodeText(), Stream.mkString), + child.exitCode, + ], + { concurrency: "unbounded" }, + ); + return { stdout, stderr, code: Number(code) }; + }).pipe(Effect.scoped); + assert.deepEqual(yield* run, { stdout: credentials.accessToken, stderr: "", code: 0 }); + yield* save({ ...credentials, expiresAt: 0 }); + const results = yield* Effect.all([run, run], { concurrency: "unbounded" }); + for (const result of results) + assert.deepEqual(result, { stdout: "fake-renewed-access", stderr: "", code: 0 }); + assert.equal(renewals, 1); + assert.equal( + Option.getOrThrow(yield* runtime.auth.read).refreshToken, + "fake-rotated-refresh", + ); + assert.deepEqual( + (yield* fs.readDirectory(config.secretsDir)).filter( + (name) => name.endsWith(".lock") || name.endsWith(".tmp"), + ), + [], + ); + // Even the same registration and subject must not leak a replacement sign-in's token. + yield* save({ + ...credentials, + idToken: "replacement-id-token", + accessToken: "replacement-access", + }); + const replaced = yield* run; + assert.equal(replaced.code, 1); + assert.equal(replaced.stdout, ""); + assert.notInclude(replaced.stderr, "replacement-access"); + yield* store.remove; + const loggedOut = yield* run; + assert.equal(loggedOut.code, 1); + assert.equal(loggedOut.stdout, ""); + assert.isFalse(yield* fs.exists(path.join(config.baseDir, "wrong-home"))); + }).pipe( + Effect.provide( + CodexManagedTokenBroker.layer.pipe(Layer.provideMerge(ServerSecretStore.layer)), + ), + 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"); + renewals++; + assert.equal(renewals, 1, "consumed refresh must never be replayed"); + return HttpClientResponse.fromWeb( + request, + Response.json({ + access_token: "fake-renewed-access", + refresh_token: "fake-rotated-refresh", + token_type: "Bearer", + expires_in: 3600, + scope: "chatgpt.tokens.use.direct", + }), + ); + }), + ), + ), + ); + }).pipe( + Effect.scoped, + Effect.provideService(HostProcess.Arguments, [ + process.execPath, + NodeURL.fileURLToPath(new URL("../bin.ts", import.meta.url)), + ]), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed( + EnvironmentId.make("00000000-0000-4000-8000-000000000001"), + ), + }), + Effect.provide( + Layer.mock(CodexInstallation.CodexInstallation)({ + managedDirectory: "/test/managed-codex", + acquire: () => + Effect.succeed({ + executablePath: "/test/codex", + source: "local", + version: "0.162.1", + managedVersionDirectory: null, + }), + }), + ), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make(() => Effect.die("No parent network expected")), + ), + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-command-home's space-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ), + ), + ), +); diff --git a/apps/server/src/cli/managedChatGptToken.ts b/apps/server/src/cli/managedChatGptToken.ts new file mode 100644 index 000000000000..84ed18fb5228 --- /dev/null +++ b/apps/server/src/cli/managedChatGptToken.ts @@ -0,0 +1,48 @@ +// @effect-diagnostics nodeBuiltinImport:off - Lean self-invoked local IPC client; no auth/store graph. +import * as NodeNet from "node:net"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; +import * as Protocol from "../provider/CodexManagedTokenProtocol.ts"; + +const decodeArguments = Schema.decodeUnknownEffect( + Schema.Tuple([ + Schema.NonEmptyString, + Protocol.Request.fields.instanceId, + Protocol.Request.fields.environmentId, + Protocol.Request.fields.profileKey, + ]), +); +const encodeRequest = Schema.encodeSync(Schema.fromJsonString(Protocol.Request)); +const decodeResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(Protocol.Response), { + onExcessProperty: "error", +}); + +/** Codex's auth.command protocol is a bare token on stdout, never a CLI diagnostic. */ +export async function runManagedChatGptTokenCli(args: ReadonlyArray): Promise { + const program = Effect.gen(function* () { + const [endpoint, instanceId, environmentId, profileKey] = yield* decodeArguments(args); + const socket = yield* Effect.acquireRelease( + Effect.sync(() => NodeNet.createConnection(endpoint)), + (socket) => Effect.sync(() => socket.destroy()), + ); + socket.on("error", () => {}); + socket.once("connect", () => + socket.write(encodeRequest({ version: 1, instanceId, environmentId, profileKey }) + "\n"), + ); + return yield* Protocol.readFrame( + socket, + Protocol.RESPONSE_BYTES, + Protocol.COMMAND_TIMEOUT_MS, + ).pipe(Effect.flatMap(decodeResponse)); + }).pipe(Effect.scoped); + const result = await Effect.runPromiseExit(program); + if (Exit.isFailure(result)) { + process.stderr.write( + "Could not access the managed ChatGPT connection. Reconnect in T3 Code.\n", + ); + process.exitCode = 1; + return; + } + process.stdout.write(result.value.token); +} diff --git a/apps/server/src/orchestration-v2/AcpRegistryOrchestratorV2.live.test.ts b/apps/server/src/orchestration-v2/AcpRegistryOrchestratorV2.live.test.ts index da89b3bc6227..c158709306ff 100644 --- a/apps/server/src/orchestration-v2/AcpRegistryOrchestratorV2.live.test.ts +++ b/apps/server/src/orchestration-v2/AcpRegistryOrchestratorV2.live.test.ts @@ -26,6 +26,7 @@ import * as ServerConfig from "../config.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as AntigravityInstallation from "../provider/AntigravityInstallation.ts"; import * as CodexInstallation from "../provider/CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../provider/CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ModelManifest from "../provider/ModelManifest.ts"; import * as ProviderInstanceRegistryHydration from "../provider/ProviderInstanceRegistryHydration.ts"; @@ -121,6 +122,9 @@ const layerProviderInstanceRegistry = ProviderInstanceRegistryHydration.layer.pi Layer.provide(FetchHttpClient.layer), Layer.provide(layerPlatformTest), ), + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), // The Codex driver now resolves managed ChatGPT installs; these runs never launch Codex. Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", diff --git a/apps/server/src/orchestration-v2/CursorOrchestratorV2.live.test.ts b/apps/server/src/orchestration-v2/CursorOrchestratorV2.live.test.ts index c17888f5e053..5c0fe3501078 100644 --- a/apps/server/src/orchestration-v2/CursorOrchestratorV2.live.test.ts +++ b/apps/server/src/orchestration-v2/CursorOrchestratorV2.live.test.ts @@ -26,6 +26,7 @@ import * as ServerConfig from "../config.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as AntigravityInstallation from "../provider/AntigravityInstallation.ts"; import * as CodexInstallation from "../provider/CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../provider/CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ModelManifest from "../provider/ModelManifest.ts"; import * as ProviderInstanceRegistryHydration from "../provider/ProviderInstanceRegistryHydration.ts"; @@ -101,6 +102,9 @@ const layerProviderInstanceRegistry = ProviderInstanceRegistryHydration.layer.pi Layer.provide(FetchHttpClient.layer), Layer.provide(layerPlatformTest), ), + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), // The Codex driver now resolves managed ChatGPT installs; these runs never launch Codex. Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", diff --git a/apps/server/src/orchestration-v2/GrokOrchestratorV2.live.test.ts b/apps/server/src/orchestration-v2/GrokOrchestratorV2.live.test.ts index f3d3bb75ced6..424077771edf 100644 --- a/apps/server/src/orchestration-v2/GrokOrchestratorV2.live.test.ts +++ b/apps/server/src/orchestration-v2/GrokOrchestratorV2.live.test.ts @@ -26,6 +26,7 @@ import * as ServerConfig from "../config.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as AntigravityInstallation from "../provider/AntigravityInstallation.ts"; import * as CodexInstallation from "../provider/CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../provider/CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ModelManifest from "../provider/ModelManifest.ts"; import * as ProviderInstanceRegistryHydration from "../provider/ProviderInstanceRegistryHydration.ts"; @@ -100,6 +101,9 @@ const layerProviderInstanceRegistry = ProviderInstanceRegistryHydration.layer.pi Layer.provide(FetchHttpClient.layer), Layer.provide(layerPlatformTest), ), + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), // The Codex driver now resolves managed ChatGPT installs; these runs never launch Codex. Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", diff --git a/apps/server/src/orchestration-v2/OpenCode2OrchestratorV2.live.test.ts b/apps/server/src/orchestration-v2/OpenCode2OrchestratorV2.live.test.ts index 0a39c5cf374a..88233992e4f0 100644 --- a/apps/server/src/orchestration-v2/OpenCode2OrchestratorV2.live.test.ts +++ b/apps/server/src/orchestration-v2/OpenCode2OrchestratorV2.live.test.ts @@ -51,6 +51,7 @@ import * as ServerConfig from "../config.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as AntigravityInstallation from "../provider/AntigravityInstallation.ts"; import * as CodexInstallation from "../provider/CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../provider/CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ModelManifest from "../provider/ModelManifest.ts"; import * as ProviderInstanceRegistryHydration from "../provider/ProviderInstanceRegistryHydration.ts"; @@ -194,6 +195,9 @@ const layerProviderInstanceRegistry = ProviderInstanceRegistryHydration.layer.pi Layer.provide(FetchHttpClient.layer), Layer.provide(layerPlatformTest), ), + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), // The Codex driver now resolves managed ChatGPT installs; these runs never launch Codex. Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", diff --git a/apps/server/src/provider/CodexChatGptAuth.test.ts b/apps/server/src/provider/CodexChatGptAuth.test.ts index 96465947b651..caccd468fe26 100644 --- a/apps/server/src/provider/CodexChatGptAuth.test.ts +++ b/apps/server/src/provider/CodexChatGptAuth.test.ts @@ -276,6 +276,16 @@ const makeHarnessFor = Effect.fnUntraced(function* ( ); yield* store.set(new TextEncoder().encode(JSON.stringify({ ...record, expiresAt: 0 }))); }); + const recreateAuth = (expectedProfileKey?: string) => + makeCodexChatGptAuth({ + instanceId, + discoveryUrl: `${origin}/discovery`, + resource: `${origin}/v1`, + ...(expectedProfileKey !== undefined ? { expectedProfileKey } : {}), + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, environment), + ); return { auth, analyticsEvents, @@ -321,14 +331,8 @@ const makeHarnessFor = Effect.fnUntraced(function* ( ); return { auth: destination, bytes: destinationBytes }; }), - recreateAuth: makeCodexChatGptAuth({ - instanceId, - discoveryUrl: `${origin}/discovery`, - resource: `${origin}/v1`, - }).pipe( - Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), - Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, environment), - ), + recreateAuth: recreateAuth(), + bindProfile: recreateAuth, startRemote, signIn: signInWithMethod(), changeAccount: signInWithMethod("chatgpt-change-account"), @@ -770,6 +774,39 @@ it.effect( }), ), ); +it.effect( + "profile-bound access survives rotation but rejects account changes, logout, and reconnect", + () => + provision( + Effect.gen(function* () { + const h = yield* makeHarness; + yield* h.signIn; + yield* h.phase("succeeded"); + const key = yield* h.auth.profileKey(Option.getOrThrow(yield* h.auth.read)); + const bound = yield* h.bindProfile(key); + const replacement = yield* h.bindProfile(key); + yield* h.seedExpired; + const records = yield* Effect.all([bound.access, replacement.access], { + concurrency: "unbounded", + }); + assert.equal(h.refreshes(), 1); + assert.deepEqual(records[0], records[1]); + assert.equal(yield* h.auth.profileKey(records[0]), key); + h.setCallbackClientId("oaiapp_other_account"); + yield* h.changeAccount; + yield* h.phase("succeeded"); + assert.equal((yield* bound.access.pipe(Effect.flip)).operation, "access"); + yield* h.auth.controller.logout(Effect.void); + assert.equal((yield* bound.access.pipe(Effect.flip)).operation, "access"); + h.setCallbackClientId("oaiapp_test"); + yield* h.reconnectProfile("oaiapp_test"); + yield* h.phase("succeeded"); + assert.equal((yield* bound.access.pipe(Effect.flip)).operation, "access"); + assert.equal(h.refreshes(), 1); + }), + ), +); + it.effect("rejects invalid ID token nonce without saving credentials", () => provision( Effect.gen(function* () { diff --git a/apps/server/src/provider/CodexChatGptAuth.ts b/apps/server/src/provider/CodexChatGptAuth.ts index 2eda1e31533d..3e36a98f19cf 100644 --- a/apps/server/src/provider/CodexChatGptAuth.ts +++ b/apps/server/src/provider/CodexChatGptAuth.ts @@ -13,6 +13,7 @@ import * as Clock from "effect/Clock"; import * as Cause from "effect/Cause"; import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; import * as Crypto from "effect/Crypto"; +import * as Hex from "effect/encoding/Hex"; import * as Exit from "effect/Exit"; import { codexAuthCallbackPage, codexAuthReturnUrl } from "./CodexAuthCallbackPage.ts"; import * as Effect from "effect/Effect"; @@ -29,6 +30,7 @@ import * as ServerSettings from "../serverSettings.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; const isSetupError = Schema.is(ProviderSetupError); +const CHATGPT_REFRESH_TIMEOUT_MS = 20_000; const RESOURCE = "https://api.openai.com/v1"; const REQUIRED_SCOPE = "chatgpt.tokens.use.direct"; const DISCOVERY = "https://auth.openai.com/.well-known/openid-configuration"; @@ -106,6 +108,7 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* readonly defaultReturnUrl?: string; readonly reconnectProfile?: ChatGptReconnectProfile | null; readonly telemetryFlow?: "direct" | "primary_handoff"; + readonly expectedProfileKey?: string; }) { const analytics = yield* Effect.serviceOption(AnalyticsService.AnalyticsService); const settings = yield* Effect.serviceOption(ServerSettings.ServerSettingsService); @@ -718,6 +721,17 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* ), ); }); + // ID tokens survive renewal but change on sign-in. Bind active children to the + // launch profile without putting an identity token or subject in their argv. + const profileKey = (record: CodexChatGptCredentials) => + crypto + .digest( + "SHA-256", + new TextEncoder().encode( + JSON.stringify([record.clientId, record.subject, record.idToken ?? null]), + ), + ) + .pipe(Effect.map(Hex.encode), Effect.orDie); const access = lock.withPermit( withSessionLock( Effect.uninterruptible( @@ -726,6 +740,11 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* if (Option.isNone(saved)) return yield* failure("access", "Sign in with ChatGPT to use managed Codex."); let record = saved.value; + if ( + options.expectedProfileKey !== undefined && + (yield* profileKey(record)) !== options.expectedProfileKey + ) + return yield* failure("access", "The ChatGPT connection changed. Start a new session."); if (!record.scopes.includes(REQUIRED_SCOPE)) return yield* failure( "sharing", @@ -753,7 +772,7 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* }), ).pipe( Effect.interruptible, - Effect.timeout("20 seconds"), + Effect.timeout(CHATGPT_REFRESH_TIMEOUT_MS), Effect.mapError((error) => error._tag === "ProviderSetupError" && error.operation === "client" ? error @@ -984,6 +1003,7 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* controller: { ...controller, reconnectProfile, importProfile }, read, access, + profileKey, 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 69d27f1377b6..a46d44c40999 100644 --- a/apps/server/src/provider/CodexManagedRuntime.test.ts +++ b/apps/server/src/provider/CodexManagedRuntime.test.ts @@ -15,6 +15,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 * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; import { makeCodexManagedRuntime } from "./CodexManagedRuntime.ts"; import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; import { codexAppServerArgs } from "./codexLaunchArgs.ts"; @@ -44,17 +45,19 @@ 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.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", @@ -113,11 +116,13 @@ it.effect.each( email: null, }); 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", + ACCESS_TOKEN: "ambient-token-must-not-reach-child", PATH: "/usr/bin", }; const runtime = yield* makeCodexManagedRuntime({ @@ -137,7 +142,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,26 +154,31 @@ 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.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.include( - args, - 'model_providers.openai_token_sharing.model_catalog_url="https://api.openai.com/v1/models"', - ); + const provider = args.find((arg) => + arg.startsWith("model_providers.openai_token_sharing_"), + )!; + assert.include(provider, 'base_url="https://api.openai.com/v1"'); + assert.include(provider, 'model_catalog_url="https://api.openai.com/v1/models"'); + assert.include(provider, "__managed-chatgpt-token"); + assert.include(provider, "refresh_interval_ms=0"); + assert.include(provider, "timeout_ms=80000"); + assert.notInclude(provider, "env_key"); 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, "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.include(provider, "supports_websockets=false"); + assert.include(provider, "requires_openai_auth=false"); assert.notInclude(effective.config.launchArgs, "dummy-owned-access"); assert.strictEqual(ambient.CODEX_HOME, "/user/.codex"); }).pipe(Effect.scoped); @@ -186,6 +196,10 @@ it.effect.each( ), }), Effect.provide(layerInstaller), + Effect.provideService(CodexManagedTokenBroker.CodexManagedTokenBroker, { + endpoint: Effect.succeed("private-pipe"), + awaitIdle: Effect.void, + }), Effect.provideService( HttpClient.HttpClient, HttpClient.make((request) => diff --git a/apps/server/src/provider/CodexManagedRuntime.ts b/apps/server/src/provider/CodexManagedRuntime.ts index fd8f3d7a9cd9..070a71f2bd14 100644 --- a/apps/server/src/provider/CodexManagedRuntime.ts +++ b/apps/server/src/provider/CodexManagedRuntime.ts @@ -5,8 +5,12 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as ServerConfig from "../config.ts"; +import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import * as CodexInstallation from "./CodexInstallation.ts"; import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import { COMMAND_TIMEOUT_MS } from "./CodexManagedTokenProtocol.ts"; import { materializeCodexShadowHome } from "./Drivers/CodexHomeLayout.ts"; export interface CodexEffectiveRuntime { @@ -16,21 +20,18 @@ export interface CodexEffectiveRuntime { } const decodeSettings = Schema.decodeSync(CodexSettings); // Managed sign-in stores tokens in T3's credential store and never writes native auth.json. -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) => + [ + `model_provider="openai_token_sharing_${profileKey}"`, + // Codex recursively merges -c tables. A profile-specific key avoids inheriting + // auth fields or headers from a shared openai_token_sharing provider. + `model_providers.openai_token_sharing_${profileKey}={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=${COMMAND_TIMEOUT_MS},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; @@ -42,6 +43,10 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func const config = yield* ServerConfig.ServerConfig; const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; + const broker = yield* CodexManagedTokenBroker.CodexManagedTokenBroker; + const environmentIdentity = yield* ServerEnvironment.ServerEnvironmentIdentity; + const environmentId = yield* environmentIdentity.getEnvironmentId; + const invocation = yield* resolveSelfInvocation(); const auth = yield* makeCodexChatGptAuth({ instanceId: options.instanceId, defaultReturnUrl: new URL( @@ -56,6 +61,19 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func ); const homePath = homeLayout.effectiveHomePath ?? homeLayout.sharedHomePath; const resolve = Effect.gen(function* () { + // Establish renewal availability before auth.access can consume a refresh token. + const endpoint = yield* broker.endpoint.pipe( + Effect.mapError( + (cause) => + new ProviderSetupError({ + instanceId: options.instanceId, + operation: "runtime", + detail: + "Could not prepare the private managed Codex token listener. Restart T3 and retry.", + cause, + }), + ), + ); const executable = yield* installation.acquire().pipe( Effect.mapError( () => @@ -67,6 +85,14 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func ), ); const credentials = yield* auth.access; + const profileKey = yield* auth.profileKey(credentials); + const authArgs = selfInvocationArgs(invocation, [ + "__managed-chatgpt-token", + endpoint, + options.instanceId, + environmentId, + profileKey, + ]); yield* materializeCodexShadowHome(homeLayout).pipe( Effect.provideService(FileSystem.FileSystem, fs), Effect.provideService(Path.Path, path), @@ -92,9 +118,9 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func // 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.ACCESS_TOKEN; delete environment.T3CODE_CODEX_LAUNCH_ARGS; delete environment.OPENAI_API_KEY; delete environment.OPENAI_BASE_URL; @@ -104,10 +130,10 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func setupMode: "managed", binaryPath: executable.executablePath, homePath, - launchArgs: managedCodexLaunchArgs, + launchArgs: managedCodexLaunchArgs(profileKey, invocation.command, authArgs), }), environment, - revision: credentials.accessToken, + revision: profileKey, } satisfies CodexEffectiveRuntime; }); return { auth, resolve, installation, homePath, homeLayout }; diff --git a/apps/server/src/provider/CodexManagedTokenBroker.test.ts b/apps/server/src/provider/CodexManagedTokenBroker.test.ts new file mode 100644 index 000000000000..5e850e7e02e9 --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenBroker.test.ts @@ -0,0 +1,671 @@ +// @effect-diagnostics nodeBuiltinImport:off - Disposable private IPC clients and file-store commit barrier. +import * as NodeNet from "node:net"; +import * as NodeFSP from "node:fs/promises"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as NodePath from "node:path"; +import * as NodeOS from "node:os"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it, vi } from "@effect/vitest"; +import { CodexSettings, EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Scope from "effect/Scope"; +import * as Schema from "effect/Schema"; +import { HttpClient, HttpClientResponse } from "effect/http"; +import * as ServerConfig from "../config.ts"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import * as Broker from "./CodexManagedTokenBroker.ts"; +import * as Protocol from "./CodexManagedTokenProtocol.ts"; +import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; +import { withChatGptSessionLock } from "./CodexChatGptSessionLock.ts"; +import { makeCodexManagedRuntime } from "./CodexManagedRuntime.ts"; +import * as CodexInstallation from "./CodexInstallation.ts"; +import { codexAppServerArgs } from "./codexLaunchArgs.ts"; + +const decodeSettings = Schema.decodeSync(CodexSettings); + +const instanceId = ProviderInstanceId.make("codex-broker-test"); +const environmentId = EnvironmentId.make("00000000-0000-4000-8000-000000000001"); +const credentials = { + clientId: "oaiapp_test", + subject: "subject", + idToken: "identity", + accessToken: "old-access", + refreshToken: "once-only-refresh", + expiresAt: 0, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + email: null, +}; +const connect = (endpoint: string) => + Effect.acquireRelease( + Effect.callback<{ socket: NodeNet.Socket; closed: Promise }, Error>((resume) => { + const socket = NodeNet.createConnection(endpoint); + const closed = new Promise((resolve) => socket.once("close", () => resolve())); + socket.on("error", () => {}); + socket.once("error", (error) => resume(Effect.fail(error))); + socket.once("connect", () => resume(Effect.succeed({ socket, closed }))); + return Effect.sync(() => { + if (socket.connecting) socket.destroy(); + }); + }), + ({ socket }) => Effect.sync(() => socket.destroy()), + ); + +const makeFixture = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const config = yield* ServerConfig.ServerConfig; + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const oauthEntered = yield* Deferred.make(); + const oauthRelease = yield* Deferred.make(); + const revoked: string[] = []; + let pauseOAuth = false; + let exchanges = 0; + const fileSystem: FileSystem.FileSystem = { + ...fs, + rename: (from, to) => + Effect.gen(function* () { + if (from.endsWith(".tmp") && to.includes("provider-auth-")) { + const text = yield* fs.readFileString(from); + if (text.includes("rotated-refresh")) { + assert.equal(exchanges, 1); + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + } + } + yield* fs.rename(from, to); + }), + }; + const http = HttpClient.make((request) => + Effect.gen(function* () { + if (request.url.endsWith("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", + revocation_endpoint: "https://auth.openai.com/revoke", + }), + ); + assert.equal(request.body._tag, "Uint8Array"); + const body = + request.body._tag === "Uint8Array" + ? new URLSearchParams(new TextDecoder().decode(request.body.body)) + : new URLSearchParams(); + if (request.url.endsWith("/revoke")) { + revoked.push(body.get("token")!); + return HttpClientResponse.fromWeb(request, new Response(null, { status: 200 })); + } + assert.equal(request.url, "https://auth.openai.com/token"); + assert.equal(body.get("refresh_token"), "once-only-refresh"); + exchanges++; + assert.equal(exchanges, 1, "a consumed refresh may never be replayed"); + yield* Deferred.succeed(oauthEntered, undefined); + if (pauseOAuth) yield* Deferred.await(oauthRelease); + return HttpClientResponse.fromWeb( + request, + Response.json({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + token_type: "Bearer", + scope: "chatgpt.tokens.use.direct", + }), + ); + }), + ); + return 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 profileKey = yield* auth.profileKey(credentials); + const services = Context.omit(Scope.Scope)( + yield* Effect.context< + Exclude< + Effect.Services>, + Broker.CodexManagedTokenBroker | CodexInstallation.CodexInstallation + > + >(), + ); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer((exit) => Scope.close(scope, exit)); + const context = yield* Layer.buildWithScope(Broker.layer, scope); + const broker = Context.get(context, Broker.CodexManagedTokenBroker); + yield* Effect.addFinalizer(() => + Deferred.succeed(release, undefined).pipe( + Effect.andThen(Deferred.succeed(oauthRelease, undefined)), + ), + ); + const request = JSON.stringify({ version: 1, instanceId, environmentId, profileKey }) + "\n"; + const installer = yield* Layer.build( + Layer.mock(CodexInstallation.CodexInstallation)({ + managedDirectory: "unused-test-installation", + acquire: () => + Effect.succeed({ + executablePath: "/mock/codex", + source: "local" as const, + version: "0.156.1", + managedVersionDirectory: null, + }), + }), + ); + const runtimeServices = Context.merge( + Context.add(services, Broker.CodexManagedTokenBroker, broker), + installer, + ); + const clean = Effect.gen(function* () { + assert.deepEqual( + (yield* fs.readDirectory(config.secretsDir)).filter( + (n) => n.endsWith(".tmp") || n.endsWith(".lock"), + ), + [], + ); + }); + return { + broker, + scope, + request, + entered, + release, + oauthEntered, + oauthRelease, + auth, + store, + revoked, + clean, + exchanges: () => exchanges, + pauseOAuth: () => { + pauseOAuth = true; + }, + freshRuntime: makeCodexManagedRuntime({ + instanceId, + enabled: true, + environment: {}, + config: decodeSettings({ + setupMode: "managed", + homePath: NodePath.join(config.stateDir, "mock-codex-home"), + }), + }).pipe(Effect.provide(runtimeServices)), + switchProfile: (ready: Deferred.Deferred, proceed: Deferred.Deferred) => + withChatGptSessionLock( + config.secretsDir, + store.binding.key, + instanceId, + Effect.gen(function* () { + yield* Deferred.succeed(ready, undefined); + yield* Deferred.await(proceed); + yield* store.set( + new TextEncoder().encode( + JSON.stringify({ + ...credentials, + subject: "new-profile", + idToken: "new-identity", + accessToken: "other-access", + refreshToken: "other-refresh", + expiresAt: Number.MAX_SAFE_INTEGER, + }), + ), + ); + }), + ), + }; + }).pipe( + Effect.provide(ServerSecretStore.layer), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(HttpClient.HttpClient, http), + ); +}); +const fixture = makeFixture.pipe( + Effect.flatMap((h) => + h.broker.endpoint.pipe(Effect.map((endpoint) => ({ ...h, broker: { ...h.broker, endpoint } }))), + ), +); +const provide = (program: Effect.Effect) => + program.pipe( + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed(environmentId), + }), + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-broker-test-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ), + ), + ); + +it.live.each(["post-oauth-pre-rename", "during-oauth"] as const)( + "disconnect %s leaves admitted rotation owned by server", + (stage) => + provide( + Effect.gen(function* () { + const h = yield* fixture; + if (stage === "during-oauth") h.pauseOAuth(); + const client = yield* connect(h.broker.endpoint); + client.socket.write(h.request); + yield* Deferred.await(stage === "during-oauth" ? h.oauthEntered : h.entered); + client.socket.destroy(); + yield* Effect.promise(() => client.closed); + if (stage === "during-oauth") { + yield* Deferred.succeed(h.oauthRelease, undefined); + yield* Deferred.await(h.entered); + } + yield* Deferred.succeed(h.release, undefined); + yield* h.broker.awaitIdle; + assert.equal(Option.getOrThrow(yield* h.auth.read).refreshToken, "rotated-refresh"); + assert.equal((yield* h.auth.access).accessToken, "new-access"); + assert.equal(h.exchanges(), 1); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live("normal server close drains accepted commits before releasing listener resources", () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const client = yield* connect(h.broker.endpoint); + client.socket.write(h.request); + yield* Deferred.await(h.entered); + const close = yield* Scope.close(h.scope, Exit.void).pipe(Effect.forkChild); + yield* Effect.promise(() => client.closed); + assert.isUndefined(close.pollUnsafe()); + yield* Deferred.succeed(h.release, undefined); + yield* Fiber.join(close); + assert.equal(Option.getOrThrow(yield* h.auth.read).refreshToken, "rotated-refresh"); + yield* h.clean; + if ((yield* HostProcess.Platform) !== "win32") + assert.isFalse( + yield* Effect.promise(() => + NodeFSP.stat(NodePath.dirname(h.broker.endpoint)).then( + () => true, + () => false, + ), + ), + ); + }).pipe(Effect.scoped), + ), +); + +it.live("logout during commit revokes the newest token and cannot resurrect credentials", () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const client = yield* connect(h.broker.endpoint); + client.socket.write(h.request); + yield* Deferred.await(h.entered); + const logout = yield* h.auth.controller.logout(Effect.void).pipe(Effect.forkChild); + yield* Effect.yieldNow; + assert.isUndefined(logout.pollUnsafe()); + client.socket.destroy(); + yield* Deferred.succeed(h.release, undefined); + yield* Fiber.join(logout); + yield* h.broker.awaitIdle; + assert.deepEqual(h.revoked, ["rotated-refresh"]); + assert.isTrue(Option.isNone(yield* h.auth.read)); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live( + "controller rebuild does not retire the server endpoint or old-session profile binding", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const controllerScope = yield* Scope.make(); + yield* h.store.set( + new TextEncoder().encode( + JSON.stringify({ ...credentials, expiresAt: Number.MAX_SAFE_INTEGER }), + ), + ); + const oldRuntime = yield* Scope.provide(controllerScope)(h.freshRuntime); + const oldEffective = yield* Scope.provide(controllerScope)(oldRuntime.resolve); + const table = codexAppServerArgs(oldEffective.config.launchArgs).find((arg) => + arg.startsWith("model_providers.openai_token_sharing_"), + )!; + const args: string[] = JSON.parse(/args=(\[.*\]),timeout_ms=/u.exec(table)![1]!); + assert.equal(args.at(-4), h.broker.endpoint); + const old = oldRuntime.auth; + const parentScope = yield* Scope.make(); + yield* Effect.addFinalizer((exit) => Scope.close(parentScope, exit)); + // Real auth controller semantics: admitted session survives controller closure. + const admitted = yield* Scope.provide(parentScope)( + old.controller.withAccess!(Effect.succeed("session")), + ); + assert.equal(admitted, "session"); + yield* Scope.close(controllerScope, Exit.void); + const replacement = yield* h.freshRuntime; + assert.equal(replacement.homePath, oldRuntime.homePath); + yield* h.store.set(new TextEncoder().encode(JSON.stringify(credentials))); + const client = yield* connect(args.at(-4)!); + client.socket.write( + JSON.stringify({ + version: 1, + instanceId: args.at(-3), + environmentId: args.at(-2), + profileKey: args.at(-1), + }) + "\n", + ); + yield* Deferred.await(h.entered); + client.socket.destroy(); + yield* Deferred.succeed(h.release, undefined); + yield* h.broker.awaitIdle; + assert.equal((yield* h.auth.access).accessToken, "new-access"); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live( + "reconnect start closes admitted sessions; cancelling reconnect cannot cancel a broker commit", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const client = yield* connect(h.broker.endpoint); + const sessionClosed = yield* Deferred.make(); + yield* h.auth.controller.withAccess!( + Effect.addFinalizer(() => + Effect.sync(() => client.socket.destroy()).pipe( + Effect.andThen(Deferred.succeed(sessionClosed, undefined)), + ), + ), + ); + client.socket.write(h.request); + yield* Deferred.await(h.entered); + const state = yield* h.auth.controller.start( + "owner", + Effect.void, + "chatgpt", + "http://localhost:7001/welcome", + "client", + ); + yield* Deferred.await(sessionClosed); + yield* Effect.promise(() => client.closed); + assert.isNotNull(state.flowId); + yield* h.auth.controller.cancel("owner", state.flowId!); + yield* Deferred.succeed(h.release, undefined); + yield* h.broker.awaitIdle; + assert.equal((yield* h.auth.access).refreshToken, "rotated-refresh"); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live( + "old-profile requests admitted during a locked profile switch fail closed after contention", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const ready = yield* Deferred.make(); + const proceed = yield* Deferred.make(); + const switching = yield* h.switchProfile(ready, proceed).pipe(Effect.forkChild); + yield* Deferred.await(ready); + const peers = yield* Effect.forEach(Array.from({ length: 16 }), () => + Effect.gen(function* () { + const client = yield* connect(h.broker.endpoint); + client.socket.write(h.request); + let bytes = 0; + client.socket.on("data", (chunk) => { + bytes += chunk.length; + }); + return { ...client, bytes: () => bytes }; + }), + ); + const overflow = yield* connect(h.broker.endpoint); + overflow.socket.write(h.request); + yield* Effect.promise(() => overflow.closed); + yield* Deferred.succeed(proceed, undefined); + yield* Fiber.join(switching); + yield* Effect.forEach(peers, (p) => Effect.promise(() => p.closed)); + yield* h.broker.awaitIdle; + assert.deepEqual( + peers.map((p) => p.bytes()), + peers.map(() => 0), + ); + assert.equal(Option.getOrThrow(yield* h.auth.read).refreshToken, "other-refresh"); + assert.equal(h.exchanges(), 0); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live( + "invalid, oversized, extra-operation, wrong-environment and disconnected partial requests receive zero token bytes", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + for (const request of [ + "not json\n", + "x".repeat(Protocol.REQUEST_BYTES + 1), + h.request.replace('"version":1', '"version":2'), + h.request.replace('"version":1', '"operation":"read-store","version":1'), + h.request.replace(environmentId, "00000000-0000-4000-8000-000000000002"), + ]) { + const client = yield* connect(h.broker.endpoint); + let bytes = 0; + client.socket.on("data", (chunk) => { + bytes += chunk.length; + }); + client.socket.write(request); + yield* Effect.promise(() => client.closed); + assert.equal(bytes, 0); + } + const client = yield* connect(h.broker.endpoint); + client.socket.write(h.request.slice(0, 10)); + client.socket.destroy(); + yield* Effect.promise(() => client.closed); + yield* h.broker.awaitIdle; + assert.equal(h.exchanges(), 0); + }).pipe(Effect.scoped), + ), +); + +it.live( + "requestless read-only peers expire with zero bytes; incomplete sockets are bounded", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const peers = yield* Effect.forEach(Array.from({ length: 32 }), () => + connect(h.broker.endpoint), + ); + const overflow = yield* connect(h.broker.endpoint); + yield* Effect.promise(() => overflow.closed); + const byteCounts = peers.map(() => 0); + peers.forEach((p, index) => + p.socket.on("data", (chunk) => { + byteCounts[index]! += chunk.length; + }), + ); + yield* Effect.forEach(peers, (p) => Effect.promise(() => p.closed), { + concurrency: "unbounded", + }); + assert.deepEqual( + byteCounts, + peers.map(() => 0), + ); + assert.equal(h.exchanges(), 0); + }).pipe(Effect.scoped), + ), + 15_000, +); + +it.live.skipIf(HostProcess.Platform.defaultValue() === "win32")( + "Unix endpoint is private and cleanup never removes unowned contents", + () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const directory = NodePath.dirname(h.broker.endpoint); + const [parent, socket] = yield* Effect.promise(() => + Promise.all([NodeFSP.lstat(directory), NodeFSP.lstat(h.broker.endpoint)]), + ); + assert.isFalse(parent.isSymbolicLink()); + assert.equal(parent.uid, yield* HostProcess.UserId); + assert.equal(parent.mode & 0o777, 0o700); + assert.isTrue(socket.isSocket()); + assert.equal(socket.mode & 0o777, 0o600); + const sentinel = NodePath.join(directory, "not-the-brokers-file"); + yield* Effect.promise(() => NodeFSP.writeFile(sentinel, "keep")); + yield* Scope.close(h.scope, Exit.void); + assert.equal(yield* Effect.promise(() => NodeFSP.readFile(sentinel, "utf8")), "keep"); + yield* Effect.promise(async () => { + await NodeFSP.unlink(sentinel); + await NodeFSP.rmdir(directory); + }); + }).pipe(Effect.scoped), + ), +); + +it.live.each(["error-event", "throw"] as const)( + "unavailable IPC (%s) fails managed setup before rotating credentials, not root construction", + (failure) => + provide( + Effect.gen(function* () { + const listenerError = new Error("private-listener-cause"); + const listen = vi.spyOn(NodeNet.Server.prototype, "listen").mockImplementation(function ( + this: NodeNet.Server, + ) { + if (failure === "throw") throw listenerError; + queueMicrotask(() => this.emit("error", listenerError)); + return this; + }); + yield* Effect.addFinalizer(() => Effect.sync(() => listen.mockRestore())); + const h = yield* makeFixture; + const runtime = yield* h.freshRuntime; + yield* Deferred.succeed(h.release, undefined); + const result = yield* runtime.resolve.pipe(Effect.result); + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal(result.failure._tag, "ProviderSetupError"); + assert.equal(result.failure.operation, "runtime"); + assert.notInclude(result.failure.detail, "private-listener-cause"); + assert.notInclude(result.failure.message, "private-listener-cause"); + assert.nestedPropertyVal(result.failure, "cause._tag", "ManagedTokenBrokerStartError"); + assert.nestedPropertyVal(result.failure, "cause.cause", listenerError); + } + assert.equal(h.exchanges(), 0); + assert.equal(Option.getOrThrow(yield* h.auth.read).refreshToken, "once-only-refresh"); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live.skipIf(HostProcess.Platform.defaultValue() === "win32")( + "long trusted Unix temp parent selects a short protected endpoint", + () => + provide( + Effect.gen(function* () { + const root = yield* Effect.promise(() => NodeFSP.mkdtemp("/tmp/t3-startup-")); + yield* Effect.addFinalizer(() => + Effect.promise(() => NodeFSP.rm(root, { recursive: true, force: true })), + ); + const long = NodePath.join(root, "x".repeat(90)); + yield* Effect.promise(() => NodeFSP.mkdir(long, { mode: 0o700 })); + const tmpdir = vi.spyOn(NodeOS, "tmpdir").mockReturnValue(long); + yield* Effect.addFinalizer(() => Effect.sync(() => tmpdir.mockRestore())); + const h = yield* fixture; + assert.isAtMost(Buffer.byteLength(h.broker.endpoint), 100); + const canonical = yield* Effect.promise(() => NodeFSP.realpath("/tmp")); + assert.equal(NodePath.dirname(NodePath.dirname(h.broker.endpoint)), canonical); + const directory = yield* Effect.promise(() => + NodeFSP.lstat(NodePath.dirname(h.broker.endpoint)), + ); + assert.equal(directory.uid, yield* HostProcess.UserId); + assert.equal(directory.mode & 0o777, 0o700); + assert.equal( + (yield* Effect.promise(() => NodeFSP.lstat(h.broker.endpoint))).mode & 0o777, + 0o600, + ); + yield* Scope.close(h.scope, Exit.void); + assert.deepEqual(yield* Effect.promise(() => NodeFSP.readdir(long)), []); + assert.isFalse( + yield* Effect.promise(() => + NodeFSP.lstat(NodePath.dirname(h.broker.endpoint)).then( + () => true, + () => false, + ), + ), + ); + }).pipe(Effect.scoped), + ), +); + +it.live.skipIf(HostProcess.Platform.defaultValue() === "win32")( + "failed Unix acquisition closes a bound listener and removes its directory before root close", + () => + provide( + Effect.gen(function* () { + const chmod = NodeFSP.chmod; + const endpoints: string[] = []; + const permissions = vi.spyOn(NodeFSP, "chmod").mockImplementation(async (path, mode) => { + if (String(path).endsWith("/s")) { + endpoints.push(String(path)); + throw new Error("socket chmod unavailable"); + } + return chmod(path, mode); + }); + const listen = vi.spyOn(NodeNet.Server.prototype, "listen"); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + permissions.mockRestore(); + listen.mockRestore(); + }), + ); + const h = yield* makeFixture; + assert.equal((yield* h.broker.endpoint.pipe(Effect.result))._tag, "Failure"); + assert.lengthOf(listen.mock.contexts, 1); + assert.isFalse((listen.mock.contexts[0] as NodeNet.Server).listening); + assert.lengthOf(endpoints, 1); + assert.isFalse( + yield* Effect.promise(() => + NodeFSP.lstat(NodePath.dirname(endpoints[0]!)).then( + () => true, + () => false, + ), + ), + ); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); + +it.live("job overload rejects immediately without cancelling already admitted rotation", () => + provide( + Effect.gen(function* () { + const h = yield* fixture; + const peers = yield* Effect.forEach(Array.from({ length: 16 }), () => + Effect.gen(function* () { + const p = yield* connect(h.broker.endpoint); + p.socket.write(h.request); + return p; + }), + ); + yield* Deferred.await(h.entered); + // A final connection is ordered after the first 16 complete frames on the + // same event loop. Rejected EOF is the overload witness, not a sleep. + const overflow = yield* connect(h.broker.endpoint); + overflow.socket.write(h.request); + yield* Effect.promise(() => overflow.closed); + for (const p of peers) p.socket.destroy(); + yield* Deferred.succeed(h.release, undefined); + yield* h.broker.awaitIdle; + assert.equal(h.exchanges(), 1); + yield* h.clean; + }).pipe(Effect.scoped), + ), +); diff --git a/apps/server/src/provider/CodexManagedTokenBroker.ts b/apps/server/src/provider/CodexManagedTokenBroker.ts new file mode 100644 index 000000000000..ca00636cddc8 --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenBroker.ts @@ -0,0 +1,229 @@ +// @effect-diagnostics nodeBuiltinImport:off - Server-owned private pipe/Unix socket, never a public listener. +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 NodeCrypto from "node:crypto"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FiberSet from "effect/FiberSet"; +import * as Layer from "effect/Layer"; +import * as Exit from "effect/Exit"; +import * as Scope from "effect/Scope"; +import * as Schema from "effect/Schema"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import * as Protocol from "./CodexManagedTokenProtocol.ts"; + +// No queue: cap both requestless sockets and admitted jobs (including lock wait, +// commit and response delivery). A disconnected job retains its slot until done. +const MAX_CONNECTIONS = 32; +const MAX_JOBS = 16; +// Stay below macOS's 104-byte and Linux's 108-byte sockaddr_un limits, including NUL. +const UNIX_SOCKET_PATH_BYTES = 100; +const encodeResponse = Schema.encodeSync(Schema.fromJsonString(Protocol.Response)); +class BrokerStartError extends Schema.TaggedError()( + "ManagedTokenBrokerStartError", + { + cause: Schema.Defect(), + }, +) { + override get message() { + return "Could not start the private managed token listener."; + } +} +export class CodexManagedTokenBroker extends Context.Service< + CodexManagedTokenBroker, + { + readonly endpoint: Effect.Effect; + readonly awaitIdle: Effect.Effect; + } +>()("t3/provider/CodexManagedTokenBroker") {} + +const io = (f: () => Promise) => + Effect.tryPromise({ try: f, catch: (cause) => new BrokerStartError({ cause }) }); +const acquireListener = Effect.gen(function* () { + const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; + const environmentId = yield* identity.getEnvironmentId; + const platform = yield* HostProcess.Platform; + const uid = yield* HostProcess.UserId; + const owned = yield* Effect.acquireRelease( + io(async () => { + if (platform === "win32") + return { endpoint: `\\\\.\\pipe\\t3-token-${NodeCrypto.randomUUID()}` }; + if (uid === undefined) throw new Error("Missing IPC owner"); + const acquireDirectory = async (candidate: string) => { + // Canonicalize each OS-standard parent independently. Ambient TMPDIR + // can be trusted yet too long for a Unix socket; /tmp must pass the same + // boundary checks, not inherit trust from the ambient parent. + const parent = await NodeFSP.realpath(candidate); + const stat = await NodeFSP.lstat(parent); + if ( + !stat.isDirectory() || + stat.isSymbolicLink() || + !( + (stat.uid === uid && (stat.mode & 0o022) === 0) || + (stat.uid === 0 && ((stat.mode & 0o022) === 0 || (stat.mode & 0o1000) !== 0)) + ) + ) + throw new Error("Untrusted IPC parent"); + const directory = await NodeFSP.mkdtemp(NodePath.join(parent, "t3-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 too long"); + return { endpoint, directory, created }; + } catch (error) { + await NodeFSP.rmdir(directory); + throw error; + } + }; + return acquireDirectory(NodeOS.tmpdir()).catch(() => acquireDirectory("/tmp")); + }), + (owned) => + Effect.promise(async () => { + if (!("directory" in owned)) return; + // Never recursively remove a replaced directory or another process's files. + const current = await NodeFSP.lstat(owned.directory).catch(() => undefined); + if ( + current && + current.dev === owned.created.dev && + current.ino === owned.created.ino && + !current.isSymbolicLink() + ) + await NodeFSP.rmdir(owned.directory).catch(() => undefined); + }), + ); + const jobs = yield* FiberSet.make(); + const runJob = + yield* FiberSet.runtime(jobs)< + Exclude< + Effect.Services>, + import("effect/Scope").Scope + > + >(); + const connections = yield* FiberSet.make(); + const runConnection = yield* FiberSet.runtime(connections)(); + const sockets = new Set(); + let closing = false; + let admitted = 0; + const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Protocol.Request), { + onExcessProperty: "error", + }); + const server = NodeNet.createServer((socket) => { + socket.on("error", () => {}); + if (closing || sockets.size >= MAX_CONNECTIONS) { + socket.destroy(); + return; + } + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + runConnection( + Effect.gen(function* () { + let transferred = false; + yield* Effect.gen(function* () { + // Windows defaults are NOT owner-only: Everyone/Anonymous have READ. + // Authorization relies on WRITE/duplex being limited to owner/Admin/System. + // https://learn.microsoft.com/windows/win32/ipc/named-pipe-security-and-access-rights + // Do not use readableAll/writableAll. Never send anything on connect: a + // read-only peer cannot supply this complete request and gets zero bytes. + const request = yield* Protocol.readFrame( + socket, + Protocol.REQUEST_BYTES, + Protocol.SOCKET_IDLE_MS, + ).pipe(Effect.flatMap(decode)); + if (closing || admitted >= MAX_JOBS || request.environmentId !== environmentId) return; + admitted++; + transferred = true; + runJob( + Effect.scoped( + Effect.gen(function* () { + // Construct against the captured server store/identity, not a controller + // registry or instance scope: existing sessions outlive controller rebuilds. + const auth = yield* makeCodexChatGptAuth({ + instanceId: request.instanceId, + expectedProfileKey: request.profileKey, + }); + const credentials = yield* auth.access; + yield* Protocol.writeFrame( + socket, + encodeResponse({ token: credentials.accessToken }), + Protocol.RESPONSE_BYTES, + ); + }), + ).pipe( + Effect.ignoreCause, + Effect.ensuring( + Effect.sync(() => { + admitted--; + socket.destroy(); + }), + ), + ), + ); + }).pipe( + Effect.ignoreCause, + Effect.ensuring( + Effect.sync(() => { + if (!transferred) socket.destroy(); + }), + ), + ); + }), + ); + }); + server.on("error", () => {}); + // Registered AFTER the FiberSets: close admission and sockets, then drain + // accepted writes BEFORE FiberSet's interrupting finalizers can run. No job + // deadline or socket-close race may cancel an accepted rotating-token commit. + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + closing = true; + for (const socket of sockets) socket.destroy(); + yield* Effect.promise(() => new Promise((resolve) => server.close(() => resolve()))); + yield* FiberSet.awaitEmpty(connections); + yield* FiberSet.awaitEmpty(jobs); + }), + ); + yield* Effect.callback((resume) => { + const onError = (cause: Error) => resume(Effect.fail(new BrokerStartError({ cause }))); + server.once("error", onError); + try { + server.listen(owned.endpoint, () => resume(Effect.void)); + } catch (cause) { + resume(Effect.fail(new BrokerStartError({ cause }))); + } + return Effect.sync(() => server.off("error", onError)); + }); + if (platform !== "win32") yield* io(() => NodeFSP.chmod(owned.endpoint, 0o600)); + return { endpoint: owned.endpoint, awaitIdle: FiberSet.awaitEmpty(jobs) }; +}); +const make = Effect.gen(function* () { + // Root-owned, not controller-owned. Failed acquisition releases even partially + // acquired listeners/FiberSets immediately; success stays attached to the server + // with the listener's LIFO drain-before-interruption ordering intact. + const scope = yield* Scope.fork(yield* Effect.scope); + const acquired = yield* acquireListener.pipe(Scope.provide(scope), Effect.result); + if (acquired._tag === "Failure") { + yield* Scope.close(scope, Exit.fail(acquired.failure)); + return CodexManagedTokenBroker.of({ + endpoint: Effect.fail(acquired.failure), + awaitIdle: Effect.void, + }); + } + return CodexManagedTokenBroker.of({ + endpoint: Effect.succeed(acquired.success.endpoint), + awaitIdle: acquired.success.awaitIdle, + }); +}); +export const layer = Layer.effect(CodexManagedTokenBroker, make); diff --git a/apps/server/src/provider/CodexManagedTokenProtocol.test.ts b/apps/server/src/provider/CodexManagedTokenProtocol.test.ts new file mode 100644 index 000000000000..808b7422ef1e --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenProtocol.test.ts @@ -0,0 +1,105 @@ +// @effect-diagnostics nodeBuiltinImport:off - Feed transport fragments without opening a listener. +import * as NodeNet from "node:net"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; +import * as Protocol from "./CodexManagedTokenProtocol.ts"; + +const decodeRequest = Schema.decodeUnknownEffect(Protocol.Request); +const decodeResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(Protocol.Response)); + +it.effect( + "maximum contract instance ID fits the request budget; wire parser does not invent a new ID contract", + () => + Effect.gen(function* () { + const request = { + version: 1, + instanceId: "A".repeat(64), + environmentId: "00000000-0000-4000-8000-000000000001", + profileKey: "a".repeat(64), + }; + const text = JSON.stringify(request); + assert.isBelow(Buffer.byteLength(text) + 1, Protocol.REQUEST_BYTES); + assert.equal((yield* decodeRequest(request)).instanceId.length, 64); + assert.isTrue( + Exit.isFailure( + yield* Effect.exit( + decodeRequest({ + ...request, + instanceId: "A".repeat(65), + }), + ), + ), + ); + }), +); + +it.effect( + "frame parser assembles fragments, rejects trailing frames and never truncates an oversized response", + () => + Effect.gen(function* () { + for (const [chunks, succeeds] of [ + [['{"token":', '"mock"}\n'], true], + [["{}\n{}\n"], false], + [["x".repeat(Protocol.RESPONSE_BYTES), "\n"], false], + ] as const) { + const socket = new NodeNet.Socket(); + const read = yield* Protocol.readFrame( + socket, + Protocol.RESPONSE_BYTES, + Protocol.SOCKET_IDLE_MS, + ).pipe(Effect.forkChild); + yield* Effect.yieldNow; + for (const text of chunks) socket.emit("data", Buffer.from(text)); + const result = yield* Fiber.await(read); + assert.equal(Exit.isSuccess(result), succeeds); + if (Exit.isSuccess(result)) + assert.deepEqual(yield* decodeResponse(result.value), { token: "mock" }); + socket.destroy(); + } + }), +); + +it.effect("incomplete frame deadline is absolute, not renewed by each byte", () => + Effect.gen(function* () { + const socket = new NodeNet.Socket(); + const read = yield* Protocol.readFrame( + socket, + Protocol.REQUEST_BYTES, + Protocol.SOCKET_IDLE_MS, + ).pipe(Effect.forkChild); + yield* TestClock.adjust(Protocol.SOCKET_IDLE_MS - 1); + socket.emit("data", Buffer.from("{")); + assert.isUndefined(read.pollUnsafe()); + yield* TestClock.adjust(1); + assert.isTrue(Exit.isFailure(yield* Fiber.await(read))); + socket.destroy(); + }), +); + +it.effect("disconnected and oversized writes fail without writing a partial token", () => + Effect.gen(function* () { + const socket = new NodeNet.Socket(); + let writes = 0; + socket.on("data", () => { + writes++; + }); + assert.isTrue( + Exit.isFailure( + yield* Effect.exit( + Protocol.writeFrame(socket, "x".repeat(Protocol.RESPONSE_BYTES), Protocol.RESPONSE_BYTES), + ), + ), + ); + socket.destroy(); + assert.isTrue( + Exit.isFailure( + yield* Effect.exit(Protocol.writeFrame(socket, "token", Protocol.RESPONSE_BYTES)), + ), + ); + assert.equal(writes, 0); + }), +); diff --git a/apps/server/src/provider/CodexManagedTokenProtocol.ts b/apps/server/src/provider/CodexManagedTokenProtocol.ts new file mode 100644 index 000000000000..9f7686b73aae --- /dev/null +++ b/apps/server/src/provider/CodexManagedTokenProtocol.ts @@ -0,0 +1,72 @@ +// @effect-diagnostics nodeBuiltinImport:off - Private local IPC framing, shared by server and lean CLI. +import type * as NodeNet from "node:net"; +import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +export const Request = Schema.Struct({ + version: Schema.Literal(1), + instanceId: ProviderInstanceId, + environmentId: EnvironmentId, + profileKey: Schema.String.check(Schema.isPattern(/^[a-f0-9]{64}$/u)), +}); +export const Response = Schema.Struct({ token: Schema.NonEmptyString }); +// A request contains only a slug, UUID and SHA-256 binding. Tokens can be larger +// JWTs. Never truncate either frame into something that could be accepted. +export const REQUEST_BYTES = 1024; +export const RESPONSE_BYTES = 64 * 1024; +// Idle/incomplete peers and blocked response consumers must not retain sockets. +export const SOCKET_IDLE_MS = 5_000; +// Existing lock retry (40s), OAuth (20s), plus startup/store headroom (20s). +export const COMMAND_TIMEOUT_MS = 80_000; +export class ProtocolError extends Schema.TaggedError()( + "ManagedTokenProtocolError", + {}, +) {} + +export const readFrame = (socket: NodeNet.Socket, maxBytes: number, timeoutMs: number) => + Effect.callback((resume) => { + let bytes = Buffer.alloc(0); + const fail = () => resume(Effect.fail(new ProtocolError())); + const onData = (chunk: Buffer) => { + if (bytes.length + chunk.length > maxBytes) return fail(); + bytes = Buffer.concat([bytes, chunk]); + const end = bytes.indexOf(10); + if (end < 0) return; + if (end !== bytes.length - 1) return fail(); + resume(Effect.succeed(bytes.subarray(0, end).toString("utf8"))); + }; + socket.on("data", onData); + socket.once("error", fail); + socket.once("close", fail); + socket.once("end", fail); + return Effect.sync(() => { + socket.off("data", onData); + socket.off("error", fail); + socket.off("close", fail); + socket.off("end", fail); + }); + }).pipe( + Effect.timeoutOrElse({ duration: timeoutMs, orElse: () => Effect.fail(new ProtocolError()) }), + ); + +export const writeFrame = (socket: NodeNet.Socket, frame: string, maxBytes: number) => + Effect.callback((resume) => { + if (socket.destroyed || Buffer.byteLength(frame) + 1 > maxBytes) { + resume(Effect.fail(new ProtocolError())); + return; + } + const fail = () => resume(Effect.fail(new ProtocolError())); + socket.once("error", fail); + socket.once("close", fail); + socket.end(`${frame}\n`, () => resume(Effect.void)); + return Effect.sync(() => { + socket.off("error", fail); + socket.off("close", fail); + }); + }).pipe( + Effect.timeoutOrElse({ + duration: SOCKET_IDLE_MS, + orElse: () => Effect.fail(new ProtocolError()), + }), + ); diff --git a/apps/server/src/provider/Drivers/CodexDriver.test.ts b/apps/server/src/provider/Drivers/CodexDriver.test.ts index 7ac25a2e6bc1..3e87fe7161b5 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.test.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.test.ts @@ -1,11 +1,13 @@ import * as CodexInstallation from "../CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../CodexManagedTokenBroker.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; // @effect-diagnostics nodeBuiltinImport:off import * as NodeServices from "@effect/platform-node/NodeServices"; import * as NodeOS from "node:os"; +import * as NodeNet from "node:net"; import * as NodePath from "node:path"; -import { expect, it } from "@effect/vitest"; +import { expect, it, vi } from "@effect/vitest"; import { EnvironmentId, ProviderInstanceId, ProviderSessionId, ThreadId } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Effect from "effect/Effect"; @@ -58,6 +60,11 @@ const layerDeps = ServerConfig.layerTest(process.cwd(), { }), ), Layer.provideMerge(Layer.mock(ServerSecretStore.ServerSecretStore)({})), + Layer.provideMerge( + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), + ), Layer.provideMerge( Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { getEnvironmentId: Effect.succeed(EnvironmentId.make("00000000-0000-4000-8000-000000000001")), @@ -96,10 +103,11 @@ const noSpawn = ChildProcessSpawner.make(() => const encodeCredentials = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); it.layer(layerTest)("CodexDriver", (it) => { - it.effect("disconnect refreshes a restored managed account while its auth flow is idle", () => + const checkManagedAccount = (profileChange: "unchanged" | "before-probe" | "during-probe") => Effect.gen(function* () { const instanceId = ProviderInstanceId.make("restored-managed-account"); const credentials = new Map(); + const catalogRequests: Array = []; const secrets = ServerSecretStore.ServerSecretStore.of({ get: (key) => Effect.sync(() => Option.fromUndefinedOr(credentials.get(key))), set: (key, value) => @@ -126,6 +134,19 @@ it.layer(layerTest)("CodexDriver", (it) => { email: "account@example.test", }); yield* store.set(new TextEncoder().encode(json)); + const replacement = yield* encodeCredentials({ + clientId: "oaiapp_other", + accessToken: "dummy-other-access", + refreshToken: "dummy-other-refresh", + expiresAt: Number.MAX_SAFE_INTEGER, + earliestRefreshAt: null, + scopes: ["chatgpt.tokens.use.direct"], + subject: "other-user", + email: "other@example.test", + }); + const replaceProfile = store.set(new TextEncoder().encode(replacement)); + const probeEntered = yield* Deferred.make(); + const resumeProbe = yield* Deferred.make(); const executable = { executablePath: "/user/bin/codex", managedVersionDirectory: null, @@ -145,7 +166,10 @@ it.layer(layerTest)("CodexDriver", (it) => { const startupChecksInterrupted = yield* Deferred.make(); const acquire = () => Effect.suspend(() => { - if (interruptedChecks === 2) return Effect.succeed(executable); + if (interruptedChecks === 2) + return ( + profileChange === "before-probe" ? replaceProfile.pipe(Effect.orDie) : Effect.void + ).pipe(Effect.as(executable)); interruptedChecks += 1; return ( interruptedChecks === 2 @@ -181,13 +205,15 @@ it.layer(layerTest)("CodexDriver", (it) => { Effect.provideService( ChildProcessSpawner.ChildProcessSpawner, ChildProcessSpawner.make(() => - Effect.fail( - PlatformError.badArgument({ + Effect.gen(function* () { + yield* Deferred.succeed(probeEntered, undefined); + yield* Deferred.await(resumeProbe); + return yield* PlatformError.badArgument({ module: "ChildProcessSpawner", method: "spawn", description: "The fixture app-server is unavailable", - }), - ), + }); + }), ), ), ); @@ -203,14 +229,27 @@ it.layer(layerTest)("CodexDriver", (it) => { Effect.forkScoped, ); yield* Deferred.await(startupChecksInterrupted); - const restored = yield* instance.snapshot.refresh; + const refresh = yield* instance.snapshot.refresh.pipe(Effect.forkChild); + yield* Deferred.await(probeEntered); + if (profileChange === "during-probe") yield* replaceProfile; + yield* Deferred.succeed(resumeProbe, undefined); + const restored = yield* Fiber.join(refresh); + if (profileChange !== "unchanged") { + expect(catalogRequests).toEqual([]); + expect(restored.auth.email).toBe("other@example.test"); + expect(restored.auth.profileId).toBe("oaiapp_other"); + expect(restored.models).toEqual([]); + expect(restored.message).toContain("Could not check Codex right now"); + return; + } + expect(catalogRequests).toEqual(["Bearer dummy-owned-access"]); 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); - // Sessions launch the T3-installed Codex with the account's token, not ambient credentials. + // Sessions launch Codex with an account-bound refresh command, not a static token. const threadId = ThreadId.make("managed-account-thread"); yield* instance.orchestrationAdapter .openSession({ @@ -228,7 +267,9 @@ 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.settings.launchArgs).not.toContain("dummy-owned-access"); + expect(launch.environment.ACCESS_TOKEN).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); @@ -251,8 +292,10 @@ it.layer(layerTest)("CodexDriver", (it) => { "https://auth.openai.com/revoke", "https://api.openai.com/v1/models", ]).toContain(request.url); - if (request.url.endsWith("/models")) + if (request.url.endsWith("/models")) { + catalogRequests.push(request.headers.authorization); return HttpClientResponse.fromWeb(request, Response.json({ models: [] })); + } return HttpClientResponse.fromWeb( request, request.url.endsWith("/revoke") @@ -269,7 +312,10 @@ it.layer(layerTest)("CodexDriver", (it) => { ), ), ); - }).pipe(Effect.scoped), + }).pipe(Effect.scoped); + it.effect.each(["unchanged", "before-probe", "during-probe"] as const)( + "managed account check keeps token and label consistent: %s", + checkManagedAccount, ); it.effect.skipIf(windowsHost)( @@ -313,6 +359,36 @@ it.layer(layerTest)("CodexDriver", (it) => { ), ); + it.effect("constructs external Codex with a root broker whose listener cannot start", () => + Effect.gen(function* () { + const listen = vi.spyOn(NodeNet.Server.prototype, "listen").mockImplementation(function ( + this: NodeNet.Server, + ) { + queueMicrotask(() => this.emit("error", new Error("IPC unavailable"))); + return this; + }); + yield* Effect.addFinalizer(() => Effect.sync(() => listen.mockRestore())); + yield* Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const instance = yield* CodexDriver.create({ + instanceId: ProviderInstanceId.make("external-without-ipc"), + displayName: "External Codex", + enabled: false, + environment: [], + config: { + ...CodexDriver.defaultConfig(), + setupMode: "existing", + binaryPath: NodePath.join(config.stateDir, "missing-codex"), + homePath: NodePath.join(config.stateDir, "external-codex-home"), + }, + }); + expect(instance.instanceId).toBe("external-without-ipc"); + expect((yield* instance.snapshot.resolveMaintenance()).update).toBeNull(); + }).pipe(Effect.provide(CodexManagedTokenBroker.layer)); + expect(listen).toHaveBeenCalledOnce(); + }).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, noSpawn), Effect.scoped), + ); + it.effect("stays manual-only when the configured executable does not exist", () => Effect.gen(function* () { const instance = yield* CodexDriver.create({ diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 09d5413f39d5..e91b7e32293a 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -75,6 +75,7 @@ import { import { makeManagedCodexProvider } from "./CodexManagedProvider.ts"; import { codexUsageReader } from "./codexUsage.ts"; import * as CodexInstallation from "../CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "../CodexManagedTokenBroker.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -125,7 +126,8 @@ export type CodexDriverEnv = | ServerConfig.ServerConfig | ServerSecretStore.ServerSecretStore | ServerEnvironment.ServerEnvironmentIdentity - | CodexInstallation.CodexInstallation; + | CodexInstallation.CodexInstallation + | CodexManagedTokenBroker.CodexManagedTokenBroker; export const CodexDriver: ProviderDriver = { driverKind: DRIVER_KIND, diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index 374f177fd9ca..4df518ce53b6 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"; @@ -128,10 +133,21 @@ 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.access; + // Another server can replace the saved profile while the native probe runs. + // Never send that account's token with the launch profile's catalog labels. + if ( + (yield* runtime.auth.profileKey(credentials)) !== effective.revision || + (yield* runtime.auth.profileKey(saved.value)) !== 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), + ); return { ...draft, models }; }), ), @@ -226,7 +242,7 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu Effect.forkScoped, ); const resolveRuntime = runtime.auth.controller.withAccess!(runtime.resolve); - // Launch settings resolve per session from the signed-in token. The registry + // Launch settings bind the refresh command to the signed-in profile. The registry // already wraps openSession in withAccess, so resolve without re-entering it. const orchestrationAdapter = yield* createCodexAdapterV2(input, { onUsageLimits: (update) => snapshot.applyUsageLimits(update), diff --git a/apps/server/src/provider/ProviderInstanceRegistry.test.ts b/apps/server/src/provider/ProviderInstanceRegistry.test.ts index eb3dc4109b85..da37ace71c67 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.test.ts @@ -1,5 +1,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; /** * Multi-instance validation slices for the live `ProviderInstanceRegistry`. @@ -208,6 +209,11 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { const layerBaseDeps = ServerConfig.layerTest(process.cwd(), { prefix: "provider-instance-registry-test", }).pipe( + Layer.provideMerge( + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), + ), Layer.provideMerge(NodeServices.layer), Layer.provideMerge( Layer.mock(CodexInstallation.CodexInstallation)({ @@ -618,6 +624,11 @@ describe("ProviderInstanceRegistry — all drivers slice", () => { // surfaced; that merged layer then provides `ServerConfig.layerTest`'s // `FileSystem` dep while keeping everything else surfaced to the test. const layerInfra = OpenCodeRuntime.layer.pipe( + Layer.provideMerge( + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), + ), Layer.provide(OpenCodeServerLedger.layerTest), Layer.provideMerge(NodeServices.layer), Layer.provideMerge( diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index d0c1f348a736..1d19fe0b8a02 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1,5 +1,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; +import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, it, assert, expect } from "@effect/vitest"; @@ -403,6 +404,9 @@ const awaitPersistedProvider = ( const layerTestNodeServices = Layer.mergeAll( NodeServices.layer, + Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ + endpoint: Effect.succeed("unused-private-pipe"), + }), Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", }), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 23981a932e05..e2298d08cadb 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -119,6 +119,7 @@ import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as AuthHttp from "./auth/http.ts"; import * as ReplayMarkers from "./auth/replayMarkers.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; +import * as CodexManagedTokenBroker from "./provider/CodexManagedTokenBroker.ts"; import * as WebhookRoute from "./scheduledTasks/webhookRoute.ts"; import * as RelayDeliveryProof from "./scheduledTasks/RelayDeliveryProof.ts"; import * as HeldHooksWaker from "./relay/HeldHooksWaker.ts"; @@ -659,6 +660,7 @@ const layerRuntimeCoreDependencies = layerRuntimeCoreDependenciesBase.pipe( Layer.provideMerge(ProjectEnrichmentService.layer), Layer.provideMerge(Layer.mergeAll(NativeAppIconResolver.layer, layerProjectFaviconResolver)), Layer.provideMerge(layerRepositoryIdentityResolver), + Layer.provideMerge(CodexManagedTokenBroker.layer), Layer.provideMerge(layerServerEnvironment), Layer.provideMerge(layerAuth), Layer.provideMerge(ServerSecretStore.layer),