Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/auth/bearer-headers.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
export interface BearerToken {
readonly access: string;
}

export function bearerHeaders(
token: BearerToken,
extra: Record<string, string> = {},
): Record<string, string> {
return { authorization: `Bearer ${token.access}`, ...extra };
}
6 changes: 3 additions & 3 deletions src/auth/codex/auth-headers.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { bearerHeaders } from "../bearer-headers.js";
import { CODEX_CLIENT_VERSION, CODEX_ORIGINATOR } from "./constants.js";

export function codexAuthHeadersForToken(
Expand All @@ -7,11 +8,10 @@ export function codexAuthHeadersForToken(
},
commandName: string,
): Record<string, string> {
const headers: Record<string, string> = {
authorization: `Bearer ${token.access}`,
const headers = bearerHeaders(token, {
originator: CODEX_ORIGINATOR,
"user-agent": `${commandName} (${CODEX_ORIGINATOR}/${CODEX_CLIENT_VERSION})`,
};
});
// An empty account id carries no identity — sending it as a header value
// would label the request with a meaningless id. Only a non-empty id rides.
if (token.accountId) headers["chatgpt-account-id"] = token.accountId;
Expand Down
31 changes: 16 additions & 15 deletions src/auth/codex/callback-server.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
import { startCallbackServer, type CallbackServer } from "@corbits/oauth-core";

import {
authorizationDoneHtml,
callbackPageHtml,
type CallbackPageCopy,
} from "../callback-page.js";
startProviderCallbackServer,
type CallbackServer,
} from "../provider-callback-server.js";

import type { CallbackPageCopy } from "../callback-page.js";
import {
CODEX_CALLBACK_HOST,
CODEX_CALLBACK_PATH,
Expand All @@ -19,13 +18,15 @@ export async function startCodexCallbackServer(
expectedState: string,
copy: CallbackPageCopy,
): Promise<CodexCallbackServer> {
return startCallbackServer(expectedState, {
port: CODEX_CALLBACK_PORT,
// Codex's registered redirect_uri uses localhost (not 127.0.0.1).
host: CODEX_CALLBACK_HOST,
path: CODEX_CALLBACK_PATH,
doneHtml: authorizationDoneHtml("Codex", copy),
failedHtml: (reason) =>
callbackPageHtml({ subject: "Codex", error: reason }, copy),
});
return startProviderCallbackServer(
"Codex",
{
port: CODEX_CALLBACK_PORT,
// Codex's registered redirect_uri uses localhost (not 127.0.0.1).
host: CODEX_CALLBACK_HOST,
path: CODEX_CALLBACK_PATH,
},
expectedState,
copy,
);
}
18 changes: 8 additions & 10 deletions src/auth/codex/login.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,13 @@
import {
buildAuthorizeUrl,
startOAuthLogin,
type OAuthLoginHandle,
type StartOAuthLoginOptions,
} from "@corbits/oauth-core";
import {
codexOAuthConfig,
exchangeCodexCode,
codexOAuthConfig,
type CodexTokens,
} from "@corbits/codex-provider";
import type {
OAuthLoginHandle,
StartOAuthLoginOptions,
} from "@corbits/oauth-core";
import { startProviderLogin } from "../provider-login.js";

import type { CallbackPageCopy } from "../callback-page.js";
import { saveCodexProfile } from "../../config/oauth-stores.js";
Expand All @@ -25,10 +24,9 @@ export async function startCodexLogin(
opts: StartCodexLoginOptions,
): Promise<CodexLoginHandle> {
const { home, copy, ...loginOpts } = opts;
return startOAuthLogin(loginOpts, {
return startProviderLogin<CodexTokens>(loginOpts, {
oauthConfig: codexOAuthConfig,
startCallbackServer: (state) => startCodexCallbackServer(state, copy),
buildAuthorizeUrl: (pkce, state) =>
buildAuthorizeUrl(codexOAuthConfig, pkce, state),
exchangeCode: async (code, verifier, now) =>
withDefaultCodexExpiry(await exchangeCodexCode(code, verifier, now), now),
saveProfile: (profile) => saveCodexProfile(profile, home),
Expand Down
74 changes: 70 additions & 4 deletions src/auth/codex/refresh-lock.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,30 @@ import {
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, test } from "bun:test";
import {
CodexRefreshLockTimeoutError,
withCodexRefreshLock,
} from "./refresh-lock.js";
import { CodexRefreshLockTimeoutError } from "./refresh-lock.js";

import { withMockedModule } from "../../../testkit/mock-module.js";

// Non-ENOENT read and stat failures must surface raw instead of being swallowed.
// Flags gate each failure to its own test; undefined keeps every call on the real module.
let failingReadPath: string | undefined;
let failingStatPath: string | undefined;
await withMockedModule(
import.meta.resolve("node:fs/promises"),
(real: typeof import("node:fs/promises")) => ({
...real,
readFile: async (...args: Parameters<typeof real.readFile>) => {
if (args[0] === failingReadPath) throw new Error("lock read failed");
return real.readFile(...args);
},
stat: async (...args: Parameters<typeof real.stat>) => {
if (args[0] === failingStatPath) throw new Error("lock stat failed");
return real.stat(...args);
},
}),
);

const { withCodexRefreshLock } = await import("./refresh-lock.js");

async function tempDir(): Promise<string> {
return mkdtemp(join(tmpdir(), "cl8628-lock-"));
Expand Down Expand Up @@ -214,4 +234,50 @@ describe("codex refresh lock", () => {
await rm(dir, { recursive: true, force: true });
}
});

test("a non-ENOENT read failure while contending propagates raw", async () => {
const dir = await tempDir();
try {
const lock = join(dir, "refresh.lock");
await writeFile(lock, "");
failingReadPath = lock;
try {
// The unified lock re-reads the holder content while contending; a
// permission/disk failure there must surface, not read as "no lock".
await expect(
withCodexRefreshLock(lock, async () => "never", {
timeoutMs: 1_000,
retryMs: 10,
}),
).rejects.toThrow("lock read failed");
} finally {
failingReadPath = undefined;
}
} finally {
await rm(dir, { recursive: true, force: true });
}
});

test("a non-ENOENT stat failure while deciding staleness propagates raw", async () => {
const dir = await tempDir();
try {
const lock = join(dir, "refresh.lock");
// Untagged content routes the staleness decision through mtime stat;
// a non-ENOENT stat failure must surface, not read as "not stale".
await writeFile(lock, "");
failingStatPath = lock;
try {
await expect(
withCodexRefreshLock(lock, async () => "never", {
timeoutMs: 1_000,
retryMs: 10,
}),
).rejects.toThrow("lock stat failed");
} finally {
failingStatPath = undefined;
}
} finally {
await rm(dir, { recursive: true, force: true });
}
});
});
169 changes: 11 additions & 158 deletions src/auth/codex/refresh-lock.ts
Original file line number Diff line number Diff line change
@@ -1,20 +1,16 @@
import { mkdir, open, readFile, stat, unlink } from "node:fs/promises";
import { randomUUID } from "node:crypto";
import { dirname } from "node:path";
import { withFileLock } from "../file-lock.js";

// Serializes Codex OAuth refresh-and-persist sections across processes that
// share one credential store. Same-process callers chain on an in-memory
// tail; cross-process callers contend on an exclusive lock file with
// crashed-holder takeover via PID liveness; the inner TokenSession dedups
// overlapping refreshes. At most one refresh grant is in flight, so a second
// refresher sees the persisted result instead of racing refresh-token
// rotation and revoking its sibling.
// tail; cross-process callers contend on an exclusive lock file with a
// crashed-holder takeover via PID liveness. At most one refresh grant is in
// flight, so a second refresher sees the persisted result instead of racing
// refresh-token rotation and revoking its sibling.
//
// Accepted limits: PID liveness only means something on one host in one PID
// namespace (ESRCH would steal a live holder's lock, a recycled PID stalls
// recovery), and one lock path per store (two spellings contend on two
// files). Same-host headless runs satisfy both.
const tails = new Map<string, Promise<void>>();

// Stale horizon for legacy locks that carry no holder PID (foreign writers,
// pre-tagging locks). Kept far above any legitimate hold and contender
Expand Down Expand Up @@ -54,160 +50,17 @@ export class CodexRefreshLockTimeoutError extends Error {
}
}

// Holder tag: the creating PID (for crashed-holder liveness) plus a unique
// token, so only the creator removes the file — a belated release after a
// stale-takeover steal would otherwise hand two holders overlapping grants.
function holderTag(): string {
return `${String(process.pid)}:${randomUUID()}`;
}

function holderPid(content: string): number | null {
const pid = Number(content.split(":")[0]?.trim());
return Number.isInteger(pid) && pid > 0 ? pid : null;
}

// kill(pid, 0) liveness: EPERM is alive (other user), ESRCH/EINVAL dead,
// anything else treated as alive so a confused signal check never steals a
// live holder's lock. Only meaningful within one PID namespace.
function isPidAlive(pid: number): boolean {
try {
process.kill(pid, 0);
return true;
} catch (err) {
const code = (err as NodeJS.ErrnoException)?.code;
return code !== "ESRCH" && code !== "EINVAL";
}
}

async function readLockContent(lockPath: string): Promise<string | null> {
try {
return await readFile(lockPath, "utf8");
} catch {
return null;
}
}

// Safe to take over: a tagged holder with a dead PID, or an untagged lock
// older than the stale horizon.
async function isLockStale(
lockPath: string,
staleMs: number,
content: string | null,
): Promise<boolean> {
if (content === null) return false;
const pid = holderPid(content);
if (pid !== null) return !isPidAlive(pid);
try {
const info = await stat(lockPath);
return Date.now() - info.mtimeMs > staleMs;
} catch {
// The lock vanished between read and stat: not ours to take over.
return false;
}
}

async function acquireLockFile(
lockPath: string,
timeoutMs: number,
retryMs: number,
staleMs: number,
deadline: number,
): Promise<string> {
await mkdir(dirname(lockPath), { recursive: true, mode: 0o700 });
for (;;) {
const tag = holderTag();
try {
const handle = await open(lockPath, "wx", 0o600);
await handle.writeFile(tag, "utf8");
await handle.close();
return tag;
} catch (err) {
if ((err as NodeJS.ErrnoException)?.code !== "EEXIST") throw err;
}
// A crashed holder never releases: unlink and take over. Re-check the
// content first so a concurrent winner's fresh lock is not unlinked.
const content = await readLockContent(lockPath);
if (content !== null && (await isLockStale(lockPath, staleMs, content))) {
if ((await readLockContent(lockPath)) === content) {
await unlink(lockPath).catch(() => undefined);
}
continue;
}
if (Date.now() >= deadline)
throw new CodexRefreshLockTimeoutError(lockPath, timeoutMs);
await new Promise((resolve) => setTimeout(resolve, retryMs));
}
}

// Bound the in-memory queue wait by the same deadline as file contention,
// or a backed-up queue stalls far longer than timeoutMs promises.
async function waitForTail(
previous: Promise<void>,
lockPath: string,
timeoutMs: number,
deadline: number,
): Promise<void> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
await Promise.race([
previous.catch(() => undefined),
new Promise<never>((_, reject) => {
const remaining = deadline - Date.now();
if (remaining <= 0) {
reject(new CodexRefreshLockTimeoutError(lockPath, timeoutMs));
return;
}
timer = setTimeout(() => {
reject(new CodexRefreshLockTimeoutError(lockPath, timeoutMs));
}, remaining);
}),
]);
} finally {
clearTimeout(timer);
}
}

/**
* Runs `work` while holding the refresh lock at `lockPath`. Same-process
* callers queue in memory, other processes on the lock file; one deadline
* covers both, so the call holds or throws within ~timeoutMs. The creating
* holder removes the file. Callers must pass one canonical path per store.
*/
export async function withCodexRefreshLock<T>(
lockPath: string,
work: () => Promise<T>,
options: CodexRefreshLockOptions = {},
): Promise<T> {
const timeoutMs = options.timeoutMs ?? 30_000;
const retryMs = options.retryMs ?? 25;
const staleMs = options.staleMs ?? DEFAULT_STALE_MS;
const deadline = Date.now() + timeoutMs;
const previous = tails.get(lockPath) ?? Promise.resolve();
let releaseTail!: () => void;
const current = new Promise<void>((resolve) => {
releaseTail = resolve;
return withFileLock(lockPath, work, {
timeoutMs,
retryMs: options.retryMs ?? 25,
staleMs: options.staleMs ?? DEFAULT_STALE_MS,
timeoutError: (lockPath: string, timeoutMs: number) =>
new CodexRefreshLockTimeoutError(lockPath, timeoutMs),
});
tails.set(lockPath, current);
try {
await waitForTail(previous, lockPath, timeoutMs, deadline);
const tag = await acquireLockFile(
lockPath,
timeoutMs,
retryMs,
staleMs,
deadline,
);
try {
return await work();
} finally {
// Fenced release: remove the file only while it still carries our
// tag, or a stale-takeover steal would hand two holders grants.
if ((await readLockContent(lockPath)) === tag) {
await unlink(lockPath).catch(() => undefined);
}
}
} finally {
if (tails.get(lockPath) === current) tails.delete(lockPath);
releaseTail();
}
}
Loading
Loading