diff --git a/CHANGELOG.md b/CHANGELOG.md index 02df606fe..14a752478 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased - Major] +### Added + +- `agent-relay cloud worker` can run Codex and Grok jobs with the workspace's subscription logins when the control plane provides them; the login files exist only while the run does. +- `agent-relay cloud worker` saves refreshed Codex and Grok login tokens for later runs. + ### Changed - The core (`agent-relay-probe`) talks to one Agent Relay site, fixed when it is built: `AGENT_RELAY_SITE_URL` at build time (`https://agentrelay.com` when unset; an https origin, or `http://localhost[:port]` / `http://127.0.0.1[:port]`), and a build for a site the core could not trust fails to build. Nothing changes the site at run time, in any build. diff --git a/packages/cli/src/cli/commands/cloud-worker.ts b/packages/cli/src/cli/commands/cloud-worker.ts index 604a429bf..ca3506ea8 100644 --- a/packages/cli/src/cli/commands/cloud-worker.ts +++ b/packages/cli/src/cli/commands/cloud-worker.ts @@ -20,6 +20,14 @@ import { } from '@agent-relay/cloud'; import { defaultExit } from '../lib/exit.js'; +import { + isCliCredentialsPayload, + cliLoginHome, + materializeCliCredentials, + removeLoginHome, + writeBackCliCredentials, + type MaterializedCliCredentials, +} from '../lib/cloud-worker-cli-credentials.js'; type ExitFn = (code: number) => never; @@ -80,11 +88,15 @@ function safePathSegment(value: string): string { function buildWorkerRuntimeEnv( payload: WorkerWorkflowPayload, - deps: CloudWorkerDependencies + deps: CloudWorkerDependencies, + cliCredentialEnv: Record = {} ): NodeJS.ProcessEnv { return { ...deps.env, ...payload.envSecrets, + // Materialised subscription logins (CODEX_HOME/GROK_HOME pointing into the + // run dir) win over anything inherited or delivered under those names. + ...cliCredentialEnv, AGENT_RELAY_CLOUD_WORKER_RUN_ID: payload.runId, RELAY_WORKSPACE_ID: payload.relayWorkspaceId, RELAY_API_KEY: payload.relaycastApiKey, @@ -217,6 +229,9 @@ export function createDefaultAssignmentRunner(deps: CloudWorkerDependencies): Ex // directory/env and then invokes relayflows with the equivalent run flags. const runDir = path.join(cloudWorkerStateDir(deps.env), 'runs', payload.runId); const keepRunDir = deps.env.AGENT_RELAY_WORKER_KEEP_RUN_DIR === '1'; + const cliCredentials = isCliCredentialsPayload(payload.cliCredentials) ? payload.cliCredentials : null; + let materialized: MaterializedCliCredentials | null = null; + let writeBackDone = false; try { await fsp.mkdir(runDir, { recursive: true, mode: 0o700 }); await fsp.chmod(runDir, 0o700).catch(() => undefined); @@ -225,12 +240,23 @@ export function createDefaultAssignmentRunner(deps: CloudWorkerDependencies): Ex const workflowPath = path.join(runDir, safeFileName(payload.workflowFileName)); await writeSecretFile(workflowPath, payload.workflow); + if (cliCredentials) { + materialized = await materializeCliCredentials({ + worker, + credentials: cliCredentials, + runDir, + fetchImpl: deps.fetchImpl, + signal, + log: deps.log, + }); + } + const relayflowsCli = deps.resolveRelayflowsCliEntrypoint(); const result = await runChild({ command: process.execPath, args: relayflowsArgs(relayflowsCli, workflowPath, payload), cwd: runDir, - env: buildWorkerRuntimeEnv(payload, deps), + env: buildWorkerRuntimeEnv(payload, deps, materialized?.env), deps, signal, }); @@ -239,16 +265,56 @@ export function createDefaultAssignmentRunner(deps: CloudWorkerDependencies): Ex throw new Error(`Workflow runner exited with code ${result.exitCode}`); } + const failedWriteBacks = await storeRefreshedLogins(); return { exitCode: result.exitCode, durationMs: result.durationMs, - summary: `Workflow ${payload.runId} completed.`, + summary: failedWriteBacks.length + ? `Workflow ${payload.runId} completed. The refreshed ${failedWriteBacks.join(', ')} login could not be stored; sign in again if the next run fails.` + : `Workflow ${payload.runId} completed.`, }; + } catch (error) { + // A failed or cancelled run still stores a refreshed login (Grok rotates + // its single-use refresh token), and says so when it could not. + const failedWriteBacks = await storeRefreshedLogins(); + if (failedWriteBacks.length && error instanceof Error) { + const warning = `The refreshed ${failedWriteBacks.join(', ')} login could not be stored; sign in again if the next run fails.`; + // Assignment reports keep the first 500 characters: lead with the + // warning when the original error would push it out. + error.message = + error.message.length + warning.length + 1 > 500 + ? `${warning} ${error.message}` + : `${error.message} ${warning}`; + } + throw error; } finally { + await storeRefreshedLogins(); + // The login files are removed even when the run directory is kept for + // debugging, and even when materialisation failed partway. + if (cliCredentials) { + const loginHome = materialized?.home ?? cliLoginHome(runDir); + await removeLoginHome(loginHome).catch(() => { + deps.error("[cloud-worker] could not remove the run's CLI login directory; delete it manually"); + }); + } if (!keepRunDir) { await fsp.rm(runDir, { recursive: true, force: true }).catch(() => undefined); } } + + /** Writes back refreshed logins once; returns the CLIs whose login could not be stored. */ + async function storeRefreshedLogins(): Promise { + if (!cliCredentials || !materialized?.files.length || writeBackDone) return []; + writeBackDone = true; + const outcome = await writeBackCliCredentials({ + worker, + credentials: cliCredentials, + materialized, + fetchImpl: deps.fetchImpl, + log: deps.log, + }).catch(() => ({ written: 0, failed: materialized?.files ?? [] })); + return outcome.failed.map((file) => file.cli); + } }; } diff --git a/packages/cli/src/cli/lib/cloud-worker-cli-credentials.test.ts b/packages/cli/src/cli/lib/cloud-worker-cli-credentials.test.ts new file mode 100644 index 000000000..16a9564ce --- /dev/null +++ b/packages/cli/src/cli/lib/cloud-worker-cli-credentials.test.ts @@ -0,0 +1,472 @@ +import { EventEmitter } from 'node:events'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { createDefaultAssignmentRunner } from '../commands/cloud-worker.js'; +import { + cliLoginHome, + materializeCliCredentials, + removeLoginHome, + SUBSCRIPTION_LOGINS_ENV, + writeBackCliCredentials, +} from './cloud-worker-cli-credentials.js'; + +const posix = process.platform !== 'win32'; + +function workerFor(baseUrl = 'https://control.test') { + return { + baseUrl, + workerId: 'wrk_1', + workerToken: 'sh_wrk_secret', + name: 'w', + heartbeatIntervalMs: 30_000, + registeredAt: '2026-11-03T00:00:00.000Z', + updatedAt: '2026-11-03T00:00:00.000Z', + } as never; +} +const worker = workerFor(); + +const CODEX = JSON.stringify({ auth_mode: 'chatgpt', tokens: { refresh_token: 'rt-secret-1' } }); +const GROK = JSON.stringify({ 'https://auth.x.ai::c': { key: 'k', refresh_token: 'xrt-secret' } }); +const credentials = { + ref: 'sh_clicred_ref', + path: '/api/v1/workers/{workerId}/assignments/run_1/cli-credentials', + credentials: [ + { credentialId: 'c-codex', provider: 'openai', cli: 'codex' }, + { credentialId: 'c-grok', provider: 'xai', cli: 'grok' }, + ], +}; + +const codexFile = { + credentialId: 'c-codex', + provider: 'openai', + cli: 'codex', + homeEnv: 'CODEX_HOME', + directory: '.codex', + file: 'auth.json', + content: CODEX, +}; +const grokFile = { + credentialId: 'c-grok', + provider: 'xai', + cli: 'grok', + homeEnv: 'GROK_HOME', + directory: '.grok', + file: 'auth.json', + content: GROK, +}; + +function controlPlane(files: unknown[], putStatus = 200) { + const calls: Array<{ url: string; method: string; headers: Headers; body?: string }> = []; + const fetchImpl = vi.fn(async (url: string, init: RequestInit = {}) => { + calls.push({ + url, + method: init.method ?? 'GET', + headers: new Headers(init.headers), + body: typeof init.body === 'string' ? init.body : undefined, + }); + if ((init.method ?? 'GET') === 'GET') { + return new Response(JSON.stringify({ files, skipped: [] }), { + headers: { 'content-type': 'application/json' }, + }); + } + return new Response(JSON.stringify({ updated: putStatus === 200 }), { + status: putStatus, + headers: { 'content-type': 'application/json' }, + }); + }); + return { calls, fetchImpl: fetchImpl as unknown as typeof fetch }; +} + +// Each run dir lives in its own temp parent, so the sibling login dir is +// removed with it after every test. +const tempParents: string[] = []; +function tmp(prefix: string) { + const parent = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + tempParents.push(parent); + // /runs/run, so the login dir (/cli-logins/run) stays inside the parent. + const runDir = path.join(parent, 'runs', 'run'); + fs.mkdirSync(runDir, { recursive: true, mode: 0o700 }); + return runDir; +} +afterEach(() => { + for (const parent of tempParents.splice(0)) fs.rmSync(parent, { recursive: true, force: true }); +}); + +describe('cloud worker CLI logins', () => { + it('writes each auth file 0600 in a 0700 dir inside the run dir and points the CLI at it', async () => { + const runDir = tmp('cli-creds-'); + try { + const cp = controlPlane([codexFile, grokFile]); + const log = vi.fn(); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + log, + }); + expect(cp.calls[0]!.url).toBe( + 'https://control.test/api/v1/workers/wrk_1/assignments/run_1/cli-credentials' + ); + expect(cp.calls[0]!.headers.get('authorization')).toBe('Bearer sh_wrk_secret'); + expect(cp.calls[0]!.headers.get('x-cli-credential-ref')).toBe('sh_clicred_ref'); + expect(materialized.home).toBe(cliLoginHome(runDir)); + expect(materialized.env).toEqual({ + CODEX_HOME: path.join(cliLoginHome(runDir), '.codex'), + GROK_HOME: path.join(cliLoginHome(runDir), '.grok'), + [SUBSCRIPTION_LOGINS_ENV]: 'codex,grok', + }); + const codexPath = path.join(cliLoginHome(runDir), '.codex', 'auth.json'); + expect(fs.readFileSync(codexPath, 'utf-8')).toBe(CODEX); + if (posix) { + expect(fs.statSync(codexPath).mode & 0o777).toBe(0o600); + expect(fs.statSync(path.join(cliLoginHome(runDir), '.grok', 'auth.json')).mode & 0o777).toBe(0o600); + expect(fs.statSync(path.dirname(codexPath)).mode & 0o777).toBe(0o700); + expect(fs.statSync(cliLoginHome(runDir)).mode & 0o777).toBe(0o700); + } + expect(JSON.stringify(log.mock.calls)).not.toContain('rt-secret'); + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('mounts only files that match a credential the assignment listed, with its own CLI layout', async () => { + const runDir = tmp('cli-creds-'); + try { + const cp = controlPlane([ + { ...codexFile, homeEnv: 'GROK_HOME', directory: '.grok' }, + { ...codexFile, credentialId: 'c-grok' }, + { ...codexFile, credentialId: 'not-listed' }, + { ...codexFile, homeEnv: 'PATH' }, + { ...codexFile, directory: '../../etc' }, + { ...codexFile, cli: 'claude' }, + ]); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + }); + expect(materialized.files).toEqual([]); + // No login mounted: each CLI home is an empty run-scoped dir, never the host's. + expect(materialized.env).toEqual({ + CODEX_HOME: path.join(cliLoginHome(runDir), '.codex'), + GROK_HOME: path.join(cliLoginHome(runDir), '.grok'), + [SUBSCRIPTION_LOGINS_ENV]: '', + }); + expect(fs.readdirSync(path.join(cliLoginHome(runDir), '.grok'))).toEqual([]); + expect(fs.existsSync(path.join(runDir, 'etc'))).toBe(false); + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('keeps the base path prefix and refuses any path but the credential route', async () => { + const runDir = tmp('cli-creds-'); + try { + const prefixed = controlPlane([codexFile]); + await materializeCliCredentials({ + worker: workerFor('https://agentrelay.test/cloud/'), + credentials, + runDir, + fetchImpl: prefixed.fetchImpl, + }); + expect(prefixed.calls[0]!.url).toBe( + 'https://agentrelay.test/cloud/api/v1/workers/wrk_1/assignments/run_1/cli-credentials' + ); + for (const hostile of [ + 'https://evil.test/steal', + '/api/\\evil.example/x', + '/api/v1/workers/{workerId}/assignments/run_1/../../../admin', + '/api/v1/workers/{workerId}/assignments/run_1/cli-credentials?x=1', + '//evil.test/api/v1/workers/{workerId}/assignments/run_1/cli-credentials', + ]) { + const cp = controlPlane([]); + await expect( + materializeCliCredentials({ + worker, + credentials: { ...credentials, path: hostile }, + runDir, + fetchImpl: cp.fetchImpl, + }) + ).rejects.toThrow(/credential route/); + expect(cp.calls).toHaveLength(0); + } + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('writes back only changed regular files, never through a symlink', async () => { + const runDir = tmp('cli-creds-'); + try { + const cp = controlPlane([codexFile, grokFile]); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + }); + const refreshed = JSON.stringify({ auth_mode: 'chatgpt', tokens: { refresh_token: 'rt-secret-2' } }); + fs.writeFileSync(materialized.files[0]!.filePath, refreshed); + if (posix) { + const secret = path.join(runDir, 'host-secret.txt'); + fs.writeFileSync(secret, 'not a login'); + fs.rmSync(materialized.files[1]!.filePath); + fs.symlinkSync(secret, materialized.files[1]!.filePath); + } + const outcome = await writeBackCliCredentials({ + worker, + credentials, + materialized, + fetchImpl: cp.fetchImpl, + }); + expect(outcome).toEqual({ written: 1, failed: [] }); + const puts = cp.calls.filter((call) => call.method === 'PUT'); + expect(puts).toHaveLength(1); + expect(puts[0]!.url).toBe( + 'https://control.test/api/v1/workers/wrk_1/assignments/run_1/cli-credentials/c-codex' + ); + expect(puts[0]!.headers.get('x-cli-credential-ref')).toBe('sh_clicred_ref'); + expect(JSON.parse(puts[0]!.body!)).toEqual({ content: refreshed }); + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('retries a failed write-back and reports it when it cannot be stored', async () => { + const runDir = tmp('cli-creds-'); + try { + const cp = controlPlane([codexFile], 503); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + }); + fs.writeFileSync(materialized.files[0]!.filePath, '{"changed":true}'); + const outcome = await writeBackCliCredentials({ + worker, + credentials, + materialized, + fetchImpl: cp.fetchImpl, + retryDelayMs: 1, + }); + expect(outcome).toEqual({ written: 0, failed: [{ credentialId: 'c-codex', cli: 'codex' }] }); + expect(cp.calls.filter((call) => call.method === 'PUT')).toHaveLength(3); + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('the assignment runner mounts logins for relayflows, writes back, and removes the logins', async () => { + const home = tmp('cli-creds-home-'); + try { + const cp = controlPlane([codexFile]); + let seenEnv: NodeJS.ProcessEnv = {}; + let fileDuringRun = ''; + const spawnProcess = vi.fn((_command: string, _args: string[], options: { env: NodeJS.ProcessEnv }) => { + seenEnv = options.env; + const authPath = path.join(options.env.CODEX_HOME!, 'auth.json'); + fileDuringRun = fs.readFileSync(authPath, 'utf-8'); + // The CLI refreshes its token during the run. + fs.writeFileSync( + authPath, + JSON.stringify({ auth_mode: 'chatgpt', tokens: { refresh_token: 'rt-secret-3' } }) + ); + const child = new EventEmitter() as EventEmitter & { killed: boolean; kill: () => boolean }; + child.killed = false; + child.kill = () => true; + queueMicrotask(() => child.emit('exit', 0, null)); + return child; + }) as never; + const runner = createDefaultAssignmentRunner({ + log: vi.fn(), + error: vi.fn(), + exit: vi.fn() as never, + // The run dir is kept, but the logins must still be removed. + env: { + AGENT_RELAY_HOME: home, + CODEX_HOME: '/home/user/.codex', + AGENT_RELAY_WORKER_KEEP_RUN_DIR: '1', + }, + spawnProcess, + now: () => new Date(), + cwd: () => home, + fetchImpl: cp.fetchImpl, + resolveRelayflowsCliEntrypoint: () => '/opt/relayflows/dist/cli.js', + }); + const result = await runner({ + assignment: { runId: 'run_1' } as never, + payload: { + runId: 'run_1', + workspaceId: 'ws', + relayWorkspaceId: 'rw', + relaycastApiKey: 'rk', + relayfileUrl: '', + relayfileToken: '', + workflow: 'version: "1.0"\n', + fileType: 'yaml', + sourceFileType: 'yaml', + workflowFileName: 'workflow.yaml', + cliCredentials: credentials, + }, + worker, + signal: new AbortController().signal, + }); + const runDir = path.join(home, 'cloud-workers', 'runs', 'run_1'); + expect(seenEnv.CODEX_HOME).toBe(path.join(cliLoginHome(runDir), '.codex')); + expect(seenEnv[SUBSCRIPTION_LOGINS_ENV]).toBe('codex'); + expect(fileDuringRun).toBe(CODEX); + const puts = cp.calls.filter((call) => call.method === 'PUT'); + expect(puts).toHaveLength(1); + expect(JSON.parse(puts[0]!.body!).content).toContain('rt-secret-3'); + expect(result.summary).toBe('Workflow run_1 completed.'); + expect(fs.existsSync(runDir)).toBe(true); + expect(fs.existsSync(cliLoginHome(runDir))).toBe(false); + } finally { + fs.rmSync(home, { recursive: true, force: true }); + } + }); + + it('fails clearly when the reference has no credential list, and skips malformed entries', async () => { + const runDir = tmp('cli-creds-'); + try { + const cp = controlPlane([codexFile]); + await expect( + materializeCliCredentials({ + worker, + credentials: { ref: credentials.ref, path: credentials.path }, + runDir, + fetchImpl: cp.fetchImpl, + }) + ).rejects.toThrow(/must list the CLI logins/); + const malformed = controlPlane([null, 7, codexFile]); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: malformed.fetchImpl, + }); + expect(materialized.files.map((file) => file.credentialId)).toEqual(['c-codex']); + expect(materialized.env[SUBSCRIPTION_LOGINS_ENV]).toBe('codex'); + expect(materialized.env.GROK_HOME).toBe(path.join(cliLoginHome(runDir), '.grok')); + // The list is checked before any request; an unreadable response fails the run. + expect(cp.calls).toHaveLength(0); + const html = vi.fn( + async () => new Response('oops', { status: 200 }) + ) as unknown as typeof fetch; + await expect( + materializeCliCredentials({ worker, credentials, runDir, fetchImpl: html }) + ).rejects.toThrow(/unreadable/); + const shapeless = vi.fn( + async () => new Response('{"files":"nope"}', { status: 200 }) + ) as unknown as typeof fetch; + await expect( + materializeCliCredentials({ worker, credentials, runDir, fetchImpl: shapeless }) + ).rejects.toThrow(/invalid/); + const proto = controlPlane([ + { ...codexFile, cli: '__proto__' }, + { ...codexFile, cli: 'toString' }, + ]); + expect( + (await materializeCliCredentials({ worker, credentials, runDir, fetchImpl: proto.fetchImpl })).files + ).toEqual([]); + } finally { + fs.rmSync(runDir, { recursive: true, force: true }); + } + }); + + it('never writes a login through a symlink planted in the run dir', async () => { + if (!posix) return; + const runDir = tmp('cli-creds-'); + const outside = tmp('cli-creds-outside-'); + try { + fs.mkdirSync(path.dirname(cliLoginHome(runDir)), { recursive: true }); + fs.symlinkSync(outside, cliLoginHome(runDir)); + const cp = controlPlane([codexFile]); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + }); + expect(fs.readdirSync(outside)).toEqual([]); + expect(fs.lstatSync(cliLoginHome(runDir)).isSymbolicLink()).toBe(false); + expect(fs.readFileSync(materialized.files[0]!.filePath, 'utf-8')).toBe(CODEX); + } finally { + fs.rmSync(cliLoginHome(runDir), { recursive: true, force: true }); + fs.rmSync(runDir, { recursive: true, force: true }); + fs.rmSync(outside, { recursive: true, force: true }); + } + }); + + it('removes the login dir and reports an unstored login when the run fails', async () => { + const home = tmp('cli-creds-home-'); + try { + const cp = controlPlane([codexFile], 400); + const spawnProcess = vi.fn((_command: string, _args: string[], options: { env: NodeJS.ProcessEnv }) => { + fs.writeFileSync(path.join(options.env.CODEX_HOME!, 'auth.json'), '{"refreshed":true}'); + const child = new EventEmitter() as EventEmitter & { killed: boolean; kill: () => boolean }; + child.killed = false; + child.kill = () => true; + queueMicrotask(() => child.emit('exit', 1, null)); + return child; + }) as never; + const runner = createDefaultAssignmentRunner({ + log: vi.fn(), + error: vi.fn(), + exit: vi.fn() as never, + env: { AGENT_RELAY_HOME: home, AGENT_RELAY_WORKER_KEEP_RUN_DIR: '1' }, + spawnProcess, + now: () => new Date(), + cwd: () => home, + fetchImpl: cp.fetchImpl, + resolveRelayflowsCliEntrypoint: () => '/opt/relayflows/dist/cli.js', + }); + await expect( + runner({ + assignment: { runId: 'run_2' } as never, + payload: { + runId: 'run_2', + workspaceId: 'ws', + relayWorkspaceId: 'rw', + relaycastApiKey: 'rk', + relayfileUrl: '', + relayfileToken: '', + workflow: 'version: "1.0"\n', + fileType: 'yaml', + sourceFileType: 'yaml', + workflowFileName: 'workflow.yaml', + cliCredentials: credentials, + }, + worker, + signal: new AbortController().signal, + }) + ).rejects.toThrow(/exited with code 1.*codex login could not be stored/); + expect(fs.existsSync(cliLoginHome(path.join(home, 'cloud-workers', 'runs', 'run_2')))).toBe(false); + } finally { + fs.rmSync(home, { recursive: true, force: true }); + } + }); + + it('removes a login dir the workflow made unreadable', async () => { + if (!posix || process.getuid?.() === 0) return; + const runDir = tmp('cli-creds-'); + const cp = controlPlane([codexFile]); + const materialized = await materializeCliCredentials({ + worker, + credentials, + runDir, + fetchImpl: cp.fetchImpl, + }); + fs.chmodSync(path.join(materialized.home, '.codex'), 0o000); + await removeLoginHome(materialized.home); + expect(fs.existsSync(materialized.home)).toBe(false); + expect(materialized.home).toBe(path.join(path.dirname(path.dirname(runDir)), 'cli-logins', 'run')); + }); +}); diff --git a/packages/cli/src/cli/lib/cloud-worker-cli-credentials.ts b/packages/cli/src/cli/lib/cloud-worker-cli-credentials.ts new file mode 100644 index 000000000..7378e1efe --- /dev/null +++ b/packages/cli/src/cli/lib/cloud-worker-cli-credentials.ts @@ -0,0 +1,364 @@ +// Subscription-login materialisation for `agent-relay cloud worker`. +// +// A control plane that holds Codex/Grok subscription logins for the workspace +// (self-hosted relaycontrol; hosted Cloud mounts the same files in its own +// sandboxes) puts `cliCredentials` in the assignment payload: a short-lived, +// run-scoped reference, never the files. Before spawning relayflows, the worker +// exchanges the reference (with its worker token) for the auth files, writes +// each one 0600 into a 0700 directory inside the run's private dir, and points +// the CLI at it (CODEX_HOME / GROK_HOME). After the run it writes back a file +// the CLI refreshed, so the stored login keeps a live refresh token. The login +// directory is deleted after the run (even when the run dir is kept). +// +// Nothing here logs a file's content or the reference. +import fs from 'node:fs'; +import fsp from 'node:fs/promises'; +import path from 'node:path'; +import { execFile } from 'node:child_process'; +import { promisify } from 'node:util'; + +const execFileAsync = promisify(execFile); + +import type { CloudWorkerRecord } from '@agent-relay/cloud'; + +/** The `cliCredentials` field of an assignment payload. */ +export type CliCredentialsPayload = { + ref: string; + expiresAt?: string; + /** `/api/v1/workers/{workerId}/assignments//cli-credentials`. */ + path: string; + credentials?: Array<{ credentialId: string; provider: string; cli: string }>; +}; + +type CredentialFile = { + credentialId: string; + provider: string; + cli: string; + homeEnv: string; + directory: string; + file: string; + content: string; +}; + +/** What was mounted for a run. */ +export type MaterializedCliCredentials = { + /** Env for the relayflows process (CODEX_HOME/GROK_HOME, AGENT_RELAY_SUBSCRIPTION_LOGINS). */ + env: Record; + /** The private directory holding every mounted login; delete it after the run. */ + home: string; + files: Array<{ credentialId: string; cli: string; filePath: string; original: string }>; +}; + +/** Result of a write-back pass. */ +export type CliCredentialWriteBack = { + written: number; + failed: Array<{ credentialId: string; cli: string }>; +}; + +/** The only layouts a login may use: each CLI's own home variable, directory and file. */ +const CLI_LAYOUTS: Record = { + codex: { homeEnv: 'CODEX_HOME', directory: '.codex', file: 'auth.json' }, + grok: { homeEnv: 'GROK_HOME', directory: '.grok', file: 'auth.json' }, +}; + +/** + * Lists the mounted CLIs. relayflows reads it to give the subscription + * precedence over API keys (AgentWorkforce/relayflows#65); the precedence + * applies once the relayflows release containing it is the one this package + * pins. Codex and Grok themselves read only CODEX_HOME/GROK_HOME. + */ +export const SUBSCRIPTION_LOGINS_ENV = 'AGENT_RELAY_SUBSCRIPTION_LOGINS'; + +const CREDENTIAL_PATH = + /^\/api\/v1\/workers\/\{workerId\}\/assignments\/[A-Za-z0-9_-]{1,128}\/cli-credentials$/; +const CREDENTIAL_ID = /^[A-Za-z0-9_-]{1,128}$/; +const WRITE_BACK_TIMEOUT_MS = 15_000; +const WRITE_BACK_ATTEMPTS = 3; + +/** + * Whether an assignment payload's `cliCredentials` has the expected shape. + * + * @param value - The raw `payload.cliCredentials` value. + * @returns True when it carries a reference and a credential path. + */ +export function isCliCredentialsPayload(value: unknown): value is CliCredentialsPayload { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const candidate = value as Record; + return typeof candidate.ref === 'string' && candidate.ref.length > 0 && typeof candidate.path === 'string'; +} + +/** The listed credentials, or an error: a reference without its list cannot be matched safely. */ +function offeredCredentials(credentials: CliCredentialsPayload): Map { + const list: unknown = credentials.credentials; + if ( + !Array.isArray(list) || + !list.every( + (entry) => + entry !== null && + typeof entry === 'object' && + typeof (entry as Record).credentialId === 'string' && + typeof (entry as Record).cli === 'string' + ) + ) { + throw new Error('cliCredentials.credentials must list the CLI logins offered for this run.'); + } + return new Map(list.map((entry: { credentialId: string; cli: string }) => [entry.credentialId, entry.cli])); +} + +/** + * The run's login directory: `/cli-logins/`, outside `runs/`, + * so neither a code-archive entry nor another run's directory can collide + * with it. Pure path helper. + * + * @param runDir - The run's private directory. + * @returns The login directory path. + */ +export function cliLoginHome(runDir: string): string { + return path.join(path.dirname(path.dirname(runDir)), 'cli-logins', path.basename(runDir)); +} + +/** + * Creates the login directory fresh: anything already there (a directory or a + * symlink) is removed first, so every path below it is created by the worker + * and no write follows a link. + */ +async function freshLoginHome(runDir: string): Promise { + const home = cliLoginHome(runDir); + await fsp.mkdir(path.dirname(home), { recursive: true, mode: 0o700 }); + await fsp.rm(home, { recursive: true, force: true }); + await fsp.mkdir(home, { mode: 0o700 }); + await fsp.chmod(home, 0o700).catch(() => undefined); + const stat = await fsp.lstat(home); + if (!stat.isDirectory() || stat.isSymbolicLink()) + throw new Error('Could not create the CLI login directory'); + return home; +} + +/** + * Builds a credential URL under the worker's API base, keeping any base path + * prefix (e.g. `/cloud`). Only the exact credential route is accepted, so the + * worker token and reference can never be sent to another origin or route. + */ +function credentialUrl(worker: CloudWorkerRecord, template: string, credentialId?: string): string { + if (!CREDENTIAL_PATH.test(template)) { + throw new Error('cliCredentials.path is not the worker credential route'); + } + if (credentialId !== undefined && !CREDENTIAL_ID.test(credentialId)) { + throw new Error('Invalid CLI credential id'); + } + const base = new URL(worker.baseUrl); + const relative = template.replace('{workerId}', encodeURIComponent(worker.workerId)); + const url = new URL( + `${worker.baseUrl.replace(/\/+$/, '')}${relative}${credentialId === undefined ? '' : `/${credentialId}`}` + ); + if (url.origin !== base.origin || url.search || url.hash) { + throw new Error('cliCredentials.path resolves outside the control plane'); + } + return url.toString(); +} + +function headers(worker: CloudWorkerRecord, ref: string, json = false): Headers { + const result = new Headers({ Authorization: `Bearer ${worker.workerToken}`, 'X-Cli-Credential-Ref': ref }); + if (json) result.set('Content-Type', 'application/json'); + return result; +} + +/** + * Fetches the run's subscription logins and writes each auth file 0600 into a + * 0700 directory in the run's login dir (a sibling of `runDir`). + * + * A file is mounted only when it matches a credential the assignment payload + * listed, with that credential's CLI and the CLI's own layout; anything else is + * skipped (and logged by CLI name only). + * + * @param input.worker - The authenticated worker (token sent with the request). + * @param input.credentials - The payload's `cliCredentials`. + * @param input.runDir - The run's private directory. + * @returns The env for relayflows, the login directory, and the mounted files. + * @throws When the control plane refuses the reference or the path is not the credential route. + */ +export async function materializeCliCredentials(input: { + worker: CloudWorkerRecord; + credentials: CliCredentialsPayload; + runDir: string; + fetchImpl: typeof fetch; + signal?: AbortSignal; + log?: (...args: unknown[]) => void; +}): Promise { + // Validate the payload before any request, so a bad list is reported as such. + const expected = offeredCredentials(input.credentials); + const response = await input.fetchImpl(credentialUrl(input.worker, input.credentials.path), { + method: 'GET', + headers: headers(input.worker, input.credentials.ref), + signal: input.signal, + // A redirect would resend the worker token and reference elsewhere. + redirect: 'error', + }); + if (!response.ok) { + throw new Error(`Could not load the run's CLI logins (HTTP ${response.status}).`); + } + // An unreadable response fails the run: starting without the offered login + // could silently fall back to inherited credentials. + let body: { files?: unknown; skipped?: unknown }; + try { + body = (await response.json()) as { files?: unknown; skipped?: unknown }; + } catch { + throw new Error('The control plane returned an unreadable CLI login response.'); + } + if (!body || typeof body !== 'object' || !Array.isArray(body.files)) { + throw new Error('The control plane returned an invalid CLI login response.'); + } + const home = await freshLoginHome(input.runDir); + const env: Record = {}; + const files: MaterializedCliCredentials['files'] = []; + const entries: Array | null> = body.files; + for (const entry of entries) { + if (!entry || typeof entry !== 'object') { + input.log?.('[cloud-worker] skipping a malformed CLI login entry'); + continue; + } + const layout = + typeof entry.cli === 'string' && Object.hasOwn(CLI_LAYOUTS, entry.cli) + ? CLI_LAYOUTS[entry.cli] + : undefined; + const valid = + layout !== undefined && + typeof entry.credentialId === 'string' && + CREDENTIAL_ID.test(entry.credentialId) && + expected.get(entry.credentialId) === entry.cli && + entry.homeEnv === layout.homeEnv && + entry.directory === layout.directory && + entry.file === layout.file && + typeof entry.content === 'string' && + !(layout.homeEnv in env); + if (!valid || !layout) { + input.log?.( + `[cloud-worker] skipping a CLI login that does not match the assignment (${String(entry.cli)})` + ); + continue; + } + const directory = path.join(home, layout.directory); + await fsp.mkdir(directory, { mode: 0o700 }); + await fsp.chmod(directory, 0o700).catch(() => undefined); + const filePath = path.join(directory, layout.file); + // 'wx' = O_CREAT | O_EXCL: fails on anything already at the path, a symlink included. + await fsp.writeFile(filePath, entry.content as string, { encoding: 'utf-8', mode: 0o600, flag: 'wx' }); + await fsp.chmod(filePath, 0o600).catch(() => undefined); + env[layout.homeEnv] = directory; + files.push({ + credentialId: entry.credentialId as string, + cli: entry.cli as string, + filePath, + original: entry.content as string, + }); + } + // Every CLI home points into the run's login dir, empty when nothing mounted, + // so a CLI never falls back to a login on the worker host (or its ~/.codex). + for (const layout of Object.values(CLI_LAYOUTS)) { + if (layout.homeEnv in env) continue; + const directory = path.join(home, layout.directory); + await fsp.mkdir(directory, { mode: 0o700 }); + env[layout.homeEnv] = directory; + } + // Always set, so an inherited value never claims a login this run lacks. + env[SUBSCRIPTION_LOGINS_ENV] = [...new Set(files.map((file) => file.cli))].sort().join(','); + if (Array.isArray(body.skipped) && body.skipped.length) { + input.log?.(`[cloud-worker] ${body.skipped.length} CLI login(s) unavailable for this run`); + } + return { env, home, files }; +} + +/** Reads a regular file without following a symlink a workflow may have planted. */ +async function readRegularFile(filePath: string): Promise { + const noFollow = fs.constants.O_NOFOLLOW ?? 0; + if (!noFollow) { + const stat = await fsp.lstat(filePath).catch(() => null); + if (!stat?.isFile()) return null; + } + let handle: fsp.FileHandle; + try { + handle = await fsp.open(filePath, fs.constants.O_RDONLY | noFollow); + } catch { + return null; + } + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size > 256 * 1024) return null; + return await handle.readFile({ encoding: 'utf-8' }); + } finally { + await handle.close(); + } +} + +/** + * After the run, sends back each auth file the CLI rewrote (its own token + * refresh), so the stored login keeps a live refresh token. Each PUT has its + * own timeout and is retried; it never waits on the run's abort signal, so a + * cancelled run still stores a rotated (single-use) refresh token. + * + * @returns How many files were stored, and which changed files could not be. + */ +export async function writeBackCliCredentials(input: { + worker: CloudWorkerRecord; + credentials: CliCredentialsPayload; + materialized: MaterializedCliCredentials; + fetchImpl: typeof fetch; + log?: (...args: unknown[]) => void; + retryDelayMs?: number; +}): Promise { + const result: CliCredentialWriteBack = { written: 0, failed: [] }; + for (const file of input.materialized.files) { + const current = await readRegularFile(file.filePath); + if (current === null || current === file.original) continue; + let stored = false; + for (let attempt = 1; attempt <= WRITE_BACK_ATTEMPTS && !stored; attempt += 1) { + try { + const response = await input.fetchImpl( + credentialUrl(input.worker, input.credentials.path, file.credentialId), + { + method: 'PUT', + headers: headers(input.worker, input.credentials.ref, true), + body: JSON.stringify({ content: current }), + signal: AbortSignal.timeout(WRITE_BACK_TIMEOUT_MS), + redirect: 'error', + } + ); + if (response.ok) stored = true; + // A 4xx will not succeed on retry. + else if (response.status < 500 && response.status !== 429) break; + } catch { + // Network failure or timeout: retry. + } + if (!stored && attempt < WRITE_BACK_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, (input.retryDelayMs ?? 1_000) * attempt)); + } + } + if (stored) { + result.written += 1; + } else { + result.failed.push({ credentialId: file.credentialId, cli: file.cli }); + input.log?.( + `[cloud-worker] the refreshed ${file.cli} login could not be stored; sign in again if the next run fails` + ); + } + } + return result; +} + +/** Makes a login tree removable again (a workflow may have chmod-ed it) and removes it. */ +export async function removeLoginHome(home: string): Promise { + try { + await fsp.rm(home, { recursive: true, force: true }); + return; + } catch { + // Restore owner permissions and retry once. + } + // Restore owner permissions with chmod(1) rather than a check-then-chmod in + // this process: `chmod -R` (default -P) never follows symlinks, applies the + // mode to a directory before descending, so a mode-000 directory made by + // the workflow is repaired too, and leaves no check/use race in this code. + if (process.platform !== 'win32') { + await execFileAsync('chmod', ['-R', 'u+rwX', '--', home]).catch(() => undefined); + } + await fsp.rm(home, { recursive: true, force: true }); +} diff --git a/packages/cloud/src/worker.ts b/packages/cloud/src/worker.ts index 21f420805..11024c768 100644 --- a/packages/cloud/src/worker.ts +++ b/packages/cloud/src/worker.ts @@ -21,6 +21,17 @@ export type WorkerWorkflowPayload = { sourceFileType: WorkerFileType | 'workflow'; workflowFileName: string; envSecrets?: Record; + /** + * Subscription logins for the run: a short-lived reference the worker + * exchanges, with its worker token, for CLI auth files (see + * packages/cli/src/cli/commands/cloud-worker-cli-credentials.ts). + */ + cliCredentials?: { + ref: string; + expiresAt?: string; + path: string; + credentials?: Array<{ credentialId: string; provider: string; cli: string }>; + }; metadata?: Record; s3CodeKey?: string; paths?: Array<{