diff --git a/src/auth/bearer-headers.ts b/src/auth/bearer-headers.ts new file mode 100644 index 000000000..9cda856aa --- /dev/null +++ b/src/auth/bearer-headers.ts @@ -0,0 +1,10 @@ +export interface BearerToken { + readonly access: string; +} + +export function bearerHeaders( + token: BearerToken, + extra: Record = {}, +): Record { + return { authorization: `Bearer ${token.access}`, ...extra }; +} diff --git a/src/auth/codex/auth-headers.ts b/src/auth/codex/auth-headers.ts index 7b02f2e64..c41e03568 100644 --- a/src/auth/codex/auth-headers.ts +++ b/src/auth/codex/auth-headers.ts @@ -1,3 +1,4 @@ +import { bearerHeaders } from "../bearer-headers.js"; import { CODEX_CLIENT_VERSION, CODEX_ORIGINATOR } from "./constants.js"; export function codexAuthHeadersForToken( @@ -7,11 +8,10 @@ export function codexAuthHeadersForToken( }, commandName: string, ): Record { - const headers: Record = { - 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; diff --git a/src/auth/codex/callback-server.ts b/src/auth/codex/callback-server.ts index 5a492f37d..8d9eaa2d5 100644 --- a/src/auth/codex/callback-server.ts +++ b/src/auth/codex/callback-server.ts @@ -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, @@ -19,13 +18,15 @@ export async function startCodexCallbackServer( expectedState: string, copy: CallbackPageCopy, ): Promise { - 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, + ); } diff --git a/src/auth/codex/login.ts b/src/auth/codex/login.ts index 10d755125..fb0635e4b 100644 --- a/src/auth/codex/login.ts +++ b/src/auth/codex/login.ts @@ -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"; @@ -25,10 +24,9 @@ export async function startCodexLogin( opts: StartCodexLoginOptions, ): Promise { const { home, copy, ...loginOpts } = opts; - return startOAuthLogin(loginOpts, { + return startProviderLogin(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), diff --git a/src/auth/codex/refresh-lock.test.ts b/src/auth/codex/refresh-lock.test.ts index 3eda8cab3..d9b440e0d 100644 --- a/src/auth/codex/refresh-lock.test.ts +++ b/src/auth/codex/refresh-lock.test.ts @@ -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) => { + if (args[0] === failingReadPath) throw new Error("lock read failed"); + return real.readFile(...args); + }, + stat: async (...args: Parameters) => { + 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 { return mkdtemp(join(tmpdir(), "cl8628-lock-")); @@ -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 }); + } + }); }); diff --git a/src/auth/codex/refresh-lock.ts b/src/auth/codex/refresh-lock.ts index 01aecfdc2..99dc0ee4d 100644 --- a/src/auth/codex/refresh-lock.ts +++ b/src/auth/codex/refresh-lock.ts @@ -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>(); // Stale horizon for legacy locks that carry no holder PID (foreign writers, // pre-tagging locks). Kept far above any legitimate hold and contender @@ -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 { - 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 { - 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 { - 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, - lockPath: string, - timeoutMs: number, - deadline: number, -): Promise { - let timer: ReturnType | undefined; - try { - await Promise.race([ - previous.catch(() => undefined), - new Promise((_, 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( lockPath: string, work: () => Promise, options: CodexRefreshLockOptions = {}, ): Promise { 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((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(); - } } diff --git a/src/auth/codex/session.ts b/src/auth/codex/session.ts index a7b4b7298..db4b9baed 100644 --- a/src/auth/codex/session.ts +++ b/src/auth/codex/session.ts @@ -1,31 +1,26 @@ -import { - createTokenSession, - isTokenExpired, - OAuthProfileNotFoundError, - OAuthRefreshFailedError, - type TokenSession, -} from "@corbits/oauth-core"; +import { OAuthProfileNotFoundError } from "@corbits/oauth-core"; import { CODEX_REFRESH_SKEW_MS, refreshCodexTokens, type CodexTokens, } from "@corbits/codex-provider"; +import type { TokenSession } from "@corbits/oauth-core"; +import type { InferenceErrorLike } from "../../inference-gateway-error.js"; import { - codexAuthPath, loadCodexProfile, + codexAuthPath, updateCodexTokens, } from "../../config/oauth-stores.js"; -import type { InferenceErrorLike } from "../../inference-gateway-error.js"; import { - replaceMutableTokens, - sanitizedRefreshFailure, -} from "../token-session-boundary.js"; -import { - CodexRefreshLockTimeoutError, withCodexRefreshLock, + CodexRefreshLockTimeoutError, } from "./refresh-lock.js"; import { withDefaultCodexExpiry } from "./store.js"; +import { + createProviderTokenSession, + type TokenSessionCallContext, +} from "../provider-token-session.js"; // Raised when a Codex profile cannot yield a usable access token: it is gone, // or its refresh token was revoked/expired. Carries the profile name and a @@ -70,111 +65,53 @@ export interface CodexAccess { accountId?: string | undefined; } -function wrapCodexAuthError(name: string, err: unknown): never { - if (err instanceof OAuthProfileNotFoundError) { - throw new CodexAuthError( - name, - "missing", - `Codex profile "${name}" is not authorized. Log in again.`, - ); - } - if (err instanceof OAuthRefreshFailedError) { - const cause = err.cause; - throw new CodexAuthError( - name, - "refresh-failed", - `Codex profile "${name}" could not be refreshed (${cause instanceof Error ? cause.message : String(cause)}). Log in again.`, - ); - } - throw err; -} - -const sessions = new Map>(); - -async function refreshCodexTokensForStore( - refreshToken: string, - now: number, - previous: CodexTokens, -): Promise { - return withDefaultCodexExpiry( - await refreshCodexTokens(refreshToken, now, previous), - now, - ); -} - -/** - * Independent Codex token session for a home directory. Each session carries - * its own in-flight deduplication, so two sessions over one home behave like - * two processes sharing one credential store. - */ -export function createCodexTokenSession( - home?: string, -): TokenSession { - const refreshBasis = new WeakMap(); - const inner = createTokenSession({ - skewMs: CODEX_REFRESH_SKEW_MS, - loadProfile: (name) => loadCodexProfile(name, home), - updateTokens: async (name, tokens) => { - const winner = await updateCodexTokens( - name, - tokens, - home, - refreshBasis.get(tokens), - ); - if (winner === undefined) throw new OAuthProfileNotFoundError(name); - replaceMutableTokens(tokens, winner.tokens); - }, - // createTokenSession only passes (refresh, now); mergeRefreshed supplies - // the prior tokens the package helper needs to keep chatgpt-account-id. - refreshTokens: async (refreshToken, now) => { - try { - const refreshed = await refreshCodexTokensForStore(refreshToken, now, { - access: "", - refresh: refreshToken, - }); - refreshBasis.set(refreshed, refreshToken); - return refreshed; - } catch (error) { - throw sanitizedRefreshFailure(error, refreshToken); - } - }, - toAccess: (tokens) => ({ - access: tokens.access, - accountId: tokens.accountId, - }), - mergeRefreshed: (refreshed, previous) => { - const merged = - refreshed.accountId === undefined && previous.accountId !== undefined - ? { ...refreshed, accountId: previous.accountId } - : refreshed; - const expectedRefreshToken = refreshBasis.get(refreshed); - if (expectedRefreshToken !== undefined) - refreshBasis.set(merged, expectedRefreshToken); - return merged; - }, - }); - return { - isExpired: inner.isExpired, - getValidToken: (name, now = Date.now()) => - withSerializedCodexRefresh(home, name, now, () => - inner.getValidToken(name, now), +export const codexTokensSession = createProviderTokenSession< + CodexTokens, + CodexAccess +>({ + errorClass: CodexAuthError, + providerLabel: "Codex", + skewMs: CODEX_REFRESH_SKEW_MS, + loadProfile: loadCodexProfile, + updateTokens: updateCodexTokens, + toAccess: (tokens) => ({ + access: tokens.access, + accountId: tokens.accountId, + }), + refreshTokens: async (refreshToken, now, previous) => + // The session-driven refresh has no prior tokens to lean on; the mapper + // only needs the refresh grant itself. Staged refreshes pass the real + // prior tokens through. + withDefaultCodexExpiry( + await refreshCodexTokens( + refreshToken, + now, + previous ?? { access: "", refresh: refreshToken }, ), - }; -} + now, + ), + // Keep the stored chatgpt account id when the refresh response omits + // id_token; the shared factory carries the refresh basis for the store CAS. + mergeRefreshed: (refreshed, previous) => + refreshed.accountId === undefined && previous.accountId !== undefined + ? { ...refreshed, accountId: previous.accountId } + : refreshed, + serializeRefresh: (ctx) => serializeCodexRefresh(ctx), +}); // Refreshes for one shared credential store serialize on a lock file so two // headless runs (or two sessions in one process) cannot hold overlapping // refresh grants and revoke each other under token rotation. Fresh tokens // resolve before the lock: a stalled refresh never blocks healthy readers. -async function withSerializedCodexRefresh( - home: string | undefined, - name: string, - now: number, - refresh: () => Promise, -): Promise { +async function serializeCodexRefresh({ + home, + name, + now, + getValid, +}: TokenSessionCallContext): Promise { const profile = await loadCodexProfile(name, home); if (profile === undefined) throw new OAuthProfileNotFoundError(name); - if (!isCodexTokenExpired(profile.tokens, now)) + if (!codexTokensSession.isExpired(profile.tokens, now)) return { access: profile.tokens.access, accountId: profile.tokens.accountId, @@ -182,7 +119,7 @@ async function withSerializedCodexRefresh( try { return await withCodexRefreshLock( `${codexAuthPath(home)}.refresh.lock`, - refresh, + getValid, ); } catch (err) { // A lock-blocked refresh is contention/crash hygiene, not a dead @@ -210,17 +147,19 @@ export function codexAuthFailureDiagnostic( return null; } -function sessionFor(home?: string): TokenSession { - const key = home ?? ""; - const existing = sessions.get(key); - if (existing !== undefined) return existing; - const created = createCodexTokenSession(home); - sessions.set(key, created); - return created; +/** + * Independent Codex token session for a home directory. Each session carries + * its own in-flight deduplication and serializes inter-process refreshes on + * the credential file lock. + */ +export function createCodexTokenSession( + home?: string, +): TokenSession { + return codexTokensSession.createSession(home); } export function isCodexTokenExpired(tokens: CodexTokens, now: number): boolean { - return isTokenExpired(tokens, now, CODEX_REFRESH_SKEW_MS); + return codexTokensSession.isExpired(tokens, now); } export async function getValidCodexToken( @@ -228,24 +167,12 @@ export async function getValidCodexToken( now?: number, home?: string, ): Promise { - try { - return await sessionFor(home).getValidToken(name, now); - } catch (err) { - wrapCodexAuthError(name, err); - } + return codexTokensSession.getValidToken(name, now, home); } export async function refreshStagedCodexTokens( tokens: CodexTokens, now: number = Date.now(), ): Promise { - if (!isCodexTokenExpired(tokens, now)) return tokens; - let refreshed: CodexTokens; - try { - refreshed = await refreshCodexTokensForStore(tokens.refresh, now, tokens); - } catch (error) { - throw sanitizedRefreshFailure(error, tokens.refresh); - } - replaceMutableTokens(tokens, refreshed); - return tokens; + return codexTokensSession.refreshStaged(tokens, now); } diff --git a/src/auth/codex/store.ts b/src/auth/codex/store.ts index 039178a37..6b2e196fc 100644 --- a/src/auth/codex/store.ts +++ b/src/auth/codex/store.ts @@ -2,7 +2,7 @@ import { type } from "arktype"; import type { AuthProfile } from "@corbits/oauth-core"; import type { CodexTokens } from "@corbits/codex-provider"; -import { createAuthStore } from "../store.js"; +import { defineProviderAuthStore } from "../provider-auth-store.js"; // On-disk store for Codex OAuth profiles: a user may hold multiple // subscriptions (personal, work, ...), so credentials are keyed by a @@ -12,17 +12,6 @@ export type { CodexTokens }; export type CodexProfile = AuthProfile; -const CodexTokensShape = type({ - access: "string", - refresh: "string", - expiresAt: "number", - "accountId?": "string", -}); - -function isCodexTokens(value: unknown): value is CodexTokens { - return !(CodexTokensShape(value) instanceof type.errors); -} - // The package mapper leaves expiresAt unset when the token endpoint omits // expires_in; disk profiles need a concrete expiry so the arktype guard can // load them. 3600s matches the previous host mapper. @@ -38,10 +27,12 @@ export function withDefaultCodexExpiry( export const CODEX_AUTH_FILENAME = "codex-auth.json"; -export function createCodexAuthStore(settingsDirName: string) { - return createAuthStore({ - filename: CODEX_AUTH_FILENAME, - settingsDirName, - isTokens: isCodexTokens, - }); -} +export const createCodexAuthStore = defineProviderAuthStore({ + filename: CODEX_AUTH_FILENAME, + tokensShape: type({ + access: "string", + refresh: "string", + expiresAt: "number", + "accountId?": "string", + }), +}); diff --git a/src/auth/file-lock.ts b/src/auth/file-lock.ts new file mode 100644 index 000000000..376f0fa0e --- /dev/null +++ b/src/auth/file-lock.ts @@ -0,0 +1,269 @@ +import { + mkdir, + open, + readFile, + stat, + unlink, + type FileHandle, +} from "node:fs/promises"; +import { dirname } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; + +export function isErrnoCode(err: unknown, code: string): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + err.code === code + ); +} + +function isEnoent(err: unknown): boolean { + return isErrnoCode(err, "ENOENT"); +} + +// Per-waiter unique claim (pid + counter): pid alone is not unique per call, +// so the counter leg tells our claim from a concurrent winner's. +let lockClaimCounter = 0; + +function newLockClaim(): string { + return `${String(process.pid)}:${String((lockClaimCounter += 1))}`; +} + +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: success or EPERM means alive; ESRCH/EINVAL mean +// dead. Any other failure reads as alive — never steal a live holder's lock +// on a confused signal check; the waiter times out with a recovery hint. +function isPidAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (err) { + return !isErrnoCode(err, "ESRCH") && !isErrnoCode(err, "EINVAL"); + } +} + +// Null when the lock vanished under the waiter (a release raced the read); +// anything but ENOENT propagates — permission and disk errors must surface. +async function readLockContent(lockPath: string): Promise { + try { + return await readFile(lockPath, "utf8"); + } catch (err) { + if (isEnoent(err)) return null; + throw err; + } +} + +// True when the lock is safe to take over: a tagged holder with a dead PID +// (crashed), or a legacy untagged lock older than the stale horizon. A +// vanished lock reads as not stale; the acquire loop retries the create. +async function isLockStale( + lockPath: string, + staleMs: number, + content: string, +): Promise { + const pid = holderPid(content); + if (pid !== null) return !isPidAlive(pid); + try { + const info = await stat(lockPath); + return Date.now() - info.mtimeMs > staleMs; + } catch (err) { + if (isEnoent(err)) return false; + throw err; + } +} + +// Same-process callers queue on an in-memory tail per lock path so a backed-up +// queue stalls no longer than its own deadline. +const tails = new Map>(); + +// Called when the deadline passes without acquiring the lock. Each consumer +// carries its own failure shape (message text, cause, recovery hint). +export type FileLockTimeoutError = ( + lockPath: string, + timeoutMs: number, + cause: unknown, +) => Error; + +export interface FileLockOptions { + /** Maximum wait for the lock before giving up. */ + timeoutMs: number; + /** Poll interval while contending. Defaults to 25. */ + retryMs?: number; + /** + * Age at which an untagged holder is presumed crashed and the lock is taken + * over. Must stay far above the longest legitimate hold and any contender + * timeout, or a waiter steals a live holder's lock. PID-tagged locks ignore + * this: a dead PID takes over, a live one never does. + */ + staleMs: number; + timeoutError: FileLockTimeoutError; +} + +export async function withFileLock( + lockPath: string, + work: () => Promise, + options: FileLockOptions, +): Promise { + const retryMs = options.retryMs ?? 25; + const deadline = Date.now() + options.timeoutMs; + + const previous = tails.get(lockPath) ?? Promise.resolve(); + let releaseTail!: () => void; + const current = new Promise((resolve) => { + releaseTail = resolve; + }); + tails.set(lockPath, current); + + try { + // Bound the queue wait by the same deadline as file contention. + let timer: ReturnType | undefined; + try { + await Promise.race([ + previous.catch(() => undefined), + new Promise((_, reject) => { + const remaining = deadline - Date.now(); + if (remaining <= 0) { + reject( + options.timeoutError(lockPath, options.timeoutMs, undefined), + ); + return; + } + timer = setTimeout(() => { + reject( + options.timeoutError(lockPath, options.timeoutMs, undefined), + ); + }, remaining); + }), + ]); + } finally { + clearTimeout(timer); + } + + const claim = newLockClaim(); + let lock: FileHandle | undefined; + + await mkdir(dirname(lockPath), { recursive: true, mode: 0o700 }); + for (;;) { + try { + const handle = await open(lockPath, "wx", 0o600); + try { + await handle.writeFile(claim, "utf8"); + } catch (writeError) { + await closeQuietly(handle); + // A half-written lock would read as an untakeable claim; remove it + // and surface the original failure. + try { + await unlink(lockPath); + } catch (unlinkError) { + if (!isEnoent(unlinkError)) throw unlinkError; + } + throw writeError; + } + // A steal may have replaced our fresh file between create and write; + // the path then names a live winner. Close and re-contend — never + // unlink. + if ((await readLockContent(lockPath)) !== claim) { + await closeQuietly(handle); + await pace(options, lockPath, retryMs, deadline, undefined); + continue; + } + lock = handle; + break; + } catch (error) { + if (!isErrnoCode(error, "EEXIST")) throw error; + // A crashed holder never releases: take over a stale lock rather + // than keep waiting. A lock that vanished under the read is not + // stale — retry the exclusive create. + const content = await readLockContent(lockPath); + if ( + content !== null && + (await isLockStale(lockPath, options.staleMs, content)) + ) { + // Re-check before unlinking so a concurrent winner's fresh claim + // is never mistaken for the stale entry just observed — claims + // are unique per waiter. A steal can still interleave between + // re-read and unlink; the post-create check above detects the + // loser and re-contends. + if ((await readLockContent(lockPath)) === content) { + try { + await unlink(lockPath); + } catch (unlinkError) { + if (!isEnoent(unlinkError)) throw unlinkError; + } + } + } + // Fall through to the deadline/sleep path: a steal that lost to a + // concurrent winner paces like any other contention. + await pace(options, lockPath, retryMs, deadline, error); + } + } + + const releaseClaim = async (): Promise => { + if (lock === undefined) return; + try { + await lock.close(); + } catch { + // The work outcome owns precedence; a close failure must not mask it. + } + try { + // Fenced release: remove the file only while it still carries our + // claim, or a stale-takeover steal would hand two holders grants. + if ((await readLockContent(lockPath)) === claim) { + await unlink(lockPath); + } + } catch (error) { + // A steal racing the release removes the file first — that is a + // normal outcome; anything else surfaces after the work value. + if (!isEnoent(error)) throw error; + } + }; + + let workOutcome: + | { ok: true; value: TResult } + | { ok: false; error: unknown }; + try { + workOutcome = { ok: true, value: await work() }; + } catch (error) { + workOutcome = { ok: false, error }; + } + let releaseOutcome: { ok: true } | { ok: false; error: unknown } = { + ok: true, + }; + try { + await releaseClaim(); + } catch (error) { + releaseOutcome = { ok: false, error }; + } + if (!workOutcome.ok) throw workOutcome.error; + if (!releaseOutcome.ok) throw releaseOutcome.error; + return workOutcome.value; + } finally { + if (tails.get(lockPath) === current) tails.delete(lockPath); + releaseTail(); + } +} + +async function pace( + options: FileLockOptions, + lockPath: string, + retryMs: number, + deadline: number, + cause: unknown, +): Promise { + if (Date.now() >= deadline) + throw options.timeoutError(lockPath, options.timeoutMs, cause); + await delay(retryMs); +} + +async function closeQuietly(handle: FileHandle): Promise { + try { + await handle.close(); + } catch { + // Close failures never mask the contended state of the lock file. + } +} diff --git a/src/auth/oauth-scope-check.ts b/src/auth/oauth-scope-check.ts index e4ad2bd92..4c5690150 100644 --- a/src/auth/oauth-scope-check.ts +++ b/src/auth/oauth-scope-check.ts @@ -106,21 +106,13 @@ function classifyStatus( return unavailable(providerLabel, `provider HTTP ${status}`); } -async function checkCodexScope( - tokens: CodexTokens, - commandName: string, +async function probeOAuthScope( + providerLabel: string, + authorizeHeaders: () => Promise>, + fetchEndpoint: (headers: Record) => Promise, ): Promise { - const providerLabel = "Codex"; try { - const headers = codexAuthHeadersForToken( - await refreshStagedCodexTokens(tokens), - commandName, - ); - const url = `${CODEX_BASE_URL}${CODEX_MODELS_PATH}?client_version=${encodeURIComponent(CODEX_CLIENT_VERSION)}`; - const res = await fetch(url, { - headers, - signal: AbortSignal.timeout(SCOPE_CHECK_TIMEOUT_MS), - }); + const res = await fetchEndpoint(await authorizeHeaders()); if (res.ok) return { status: "ok" }; return classifyStatus(res.status, providerLabel); } catch (err) { @@ -133,28 +125,37 @@ async function checkCodexScope( } } +async function checkCodexScope( + tokens: CodexTokens, + commandName: string, +): Promise { + return probeOAuthScope( + "Codex", + async () => + codexAuthHeadersForToken( + await refreshStagedCodexTokens(tokens), + commandName, + ), + (headers) => + fetch( + `${CODEX_BASE_URL}${CODEX_MODELS_PATH}?client_version=${encodeURIComponent(CODEX_CLIENT_VERSION)}`, + { headers, signal: AbortSignal.timeout(SCOPE_CHECK_TIMEOUT_MS) }, + ), + ); +} + async function checkXaiScope( tokens: XaiTokens, ): Promise { - const providerLabel = "Grok"; - try { - const headers = xaiAuthHeadersForToken( - await refreshStagedXaiTokens(tokens), - ); - const res = await fetch(`${XAI_BASE_URL}/models`, { - headers, - signal: AbortSignal.timeout(XAI_TOKEN_TIMEOUT_MS), - }); - if (res.ok) return { status: "ok" }; - return classifyStatus(res.status, providerLabel); - } catch (err) { - if (isDefinitiveRefreshAuthRejection(err)) - return invalidCredentials(providerLabel); - return unavailable( - providerLabel, - isTimeout(err) ? "probe timed out" : undefined, - ); - } + return probeOAuthScope( + "Grok", + async () => xaiAuthHeadersForToken(await refreshStagedXaiTokens(tokens)), + (headers) => + fetch(`${XAI_BASE_URL}/models`, { + headers, + signal: AbortSignal.timeout(XAI_TOKEN_TIMEOUT_MS), + }), + ); } // Probe an OAuth-issued token against the provider's own catalog/list diff --git a/src/auth/provider-auth-store.ts b/src/auth/provider-auth-store.ts new file mode 100644 index 000000000..8f2657ec2 --- /dev/null +++ b/src/auth/provider-auth-store.ts @@ -0,0 +1,19 @@ +import { type, type Type } from "arktype"; +import type { BaseTokens } from "@corbits/oauth-core"; + +import { createAuthStore, type AuthStore } from "./store.js"; + +export interface ProviderAuthStoreDefinition { + filename: string; + tokensShape: Type; +} + +export function defineProviderAuthStore( + definition: ProviderAuthStoreDefinition, +): (settingsDirName: string) => AuthStore { + const { filename, tokensShape } = definition; + const isTokens = (value: unknown): value is TTokens => + !(tokensShape(value) instanceof type.errors); + return (settingsDirName: string) => + createAuthStore({ filename, settingsDirName, isTokens }); +} diff --git a/src/auth/provider-callback-server.ts b/src/auth/provider-callback-server.ts new file mode 100644 index 000000000..c23d35a68 --- /dev/null +++ b/src/auth/provider-callback-server.ts @@ -0,0 +1,32 @@ +import { startCallbackServer, type CallbackServer } from "@corbits/oauth-core"; + +import { + authorizationDoneHtml, + callbackPageHtml, + type CallbackPageCopy, +} from "./callback-page.js"; + +export type { CallbackServer }; + +export interface ProviderCallbackServerConfig { + port: number; + host?: string | undefined; + path: string; +} + +export async function startProviderCallbackServer( + providerLabel: string, + config: ProviderCallbackServerConfig, + expectedState: string, + copy: CallbackPageCopy, +): Promise { + return startCallbackServer(expectedState, { + port: config.port, + // exactOptionalPropertyTypes: host only rides when a provider selects one. + ...(config.host === undefined ? {} : { host: config.host }), + path: config.path, + doneHtml: authorizationDoneHtml(providerLabel, copy), + failedHtml: (reason) => + callbackPageHtml({ subject: providerLabel, error: reason }, copy), + }); +} diff --git a/src/auth/provider-login.ts b/src/auth/provider-login.ts new file mode 100644 index 000000000..3999ee681 --- /dev/null +++ b/src/auth/provider-login.ts @@ -0,0 +1,38 @@ +import { + buildAuthorizeUrl, + startOAuthLogin, + type CallbackServer, + type OAuthClientConfig, + type OAuthLoginHandle, + type StartOAuthLoginOptions, +} from "@corbits/oauth-core"; +import type { BaseTokens } from "@corbits/oauth-core"; + +export interface ProviderLoginSpec { + oauthConfig: OAuthClientConfig; + startCallbackServer: (expectedState: string) => Promise; + exchangeCode: ( + code: string, + verifier: string, + now: number, + ) => Promise; + saveProfile: (profile: { + name: string; + tokens: TTokens; + createdAt: number; + }) => Promise; +} + +export function startProviderLogin( + opts: StartOAuthLoginOptions, + spec: ProviderLoginSpec, +): Promise> { + return startOAuthLogin(opts, { + startCallbackServer: spec.startCallbackServer, + buildAuthorizeUrl: (pkce, state) => + buildAuthorizeUrl(spec.oauthConfig, pkce, state), + exchangeCode: (code, verifier, now) => + spec.exchangeCode(code, verifier, now), + saveProfile: (profile) => spec.saveProfile(profile), + }); +} diff --git a/src/auth/provider-token-session.test.ts b/src/auth/provider-token-session.test.ts new file mode 100644 index 000000000..1d4799cf2 --- /dev/null +++ b/src/auth/provider-token-session.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, test } from "bun:test"; +import { createProviderTokenSession } from "./provider-token-session.js"; + +type TestTokens = { + access: string; + refresh: string; + expiresAt: number; +}; + +class TestAuthError extends Error { + readonly profile: string; + readonly reason: "missing" | "refresh-failed"; + + constructor( + profile: string, + reason: "missing" | "refresh-failed", + message: string, + ) { + super(message); + this.name = "TestAuthError"; + this.profile = profile; + this.reason = reason; + } +} + +describe("createProviderTokenSession", () => { + test("staged refresh hands the prior tokens to the provider hook", async () => { + const seen: { + refreshToken: string; + previousAccess: string | undefined; + }[] = []; + const api = createProviderTokenSession({ + errorClass: TestAuthError, + providerLabel: "Test", + skewMs: 0, + loadProfile: async () => undefined, + updateTokens: async () => undefined, + toAccess: (tokens) => ({ access: tokens.access }), + refreshTokens: async (refreshToken, now, previous) => { + seen.push({ refreshToken, previousAccess: previous?.access }); + return { access: "fresh", refresh: "r2", expiresAt: now + 10_000 }; + }, + }); + + expect( + await api.refreshStaged( + { access: "old2", refresh: "r1", expiresAt: 1_000 }, + 5_000, + ), + ).toEqual({ access: "fresh", refresh: "r2", expiresAt: 15_000 }); + expect(seen).toEqual([{ refreshToken: "r1", previousAccess: "old2" }]); + }); +}); diff --git a/src/auth/provider-token-session.ts b/src/auth/provider-token-session.ts new file mode 100644 index 000000000..937a6ca8a --- /dev/null +++ b/src/auth/provider-token-session.ts @@ -0,0 +1,208 @@ +import { + createTokenSession, + isTokenExpired, + OAuthProfileNotFoundError, + OAuthRefreshFailedError, + type AuthProfile, + type BaseTokens, + type TokenSession, +} from "@corbits/oauth-core"; + +import { + replaceMutableTokens, + sanitizedRefreshFailure, +} from "./token-session-boundary.js"; + +export type ProviderAuthErrorReason = "missing" | "refresh-failed"; + +export type ProviderAuthError = Error & { + readonly profile: string; + readonly reason: ProviderAuthErrorReason; +}; + +export type ProviderAuthErrorClass = new ( + profile: string, + reason: ProviderAuthErrorReason, + message: string, +) => ProviderAuthError; + +// Where the session posts refreshed tokens back to the store: the winner is +// this update, a concurrent refresh's value, or undefined after removal. +export type ProviderTokenStoreUpdate = ( + name: string, + tokens: TTokens, + home: string | undefined, + expectedRefreshToken: string | undefined, +) => Promise | undefined>; + +// Context one getValidToken call hands to the reconcile hook: `session` is +// the unwrapped inner session, and `getValid` the raw store-driven path. +export interface TokenSessionCallContext { + home?: string | undefined; + name: string; + now: number; + session: TokenSession; + getValid: () => Promise; +} + +export interface ProviderTokenSessionSpec { + errorClass: ProviderAuthErrorClass; + /** Profile label in re-login messaging, e.g. "Codex" or "xAI". */ + providerLabel: string; + skewMs: number; + loadProfile: ( + name: string, + home?: string, + ) => Promise | undefined>; + updateTokens: ProviderTokenStoreUpdate; + toAccess: (tokens: TTokens) => TAccess; + // Fresh tokens from a stored refresh token. `previous` rides along for + // staged refreshes (full prior tokens) and is absent on the session's + // endpoint-driven refresh. + refreshTokens: ( + refreshToken: string, + now: number, + previous?: TTokens, + ) => Promise; + mergeRefreshed?: (refreshed: TTokens, previous: TTokens) => TTokens; + serializeRefresh?: ( + ctx: TokenSessionCallContext, + ) => Promise; +} + +export interface ProviderTokenSessionApi { + createSession: (home?: string) => TokenSession; + sessionFor: (home?: string) => TokenSession; + isExpired: (tokens: TTokens, now: number) => boolean; + getValidToken: ( + name: string, + now?: number, + home?: string, + ) => Promise; + refreshStaged: (tokens: TTokens, now?: number) => Promise; +} + +export function createProviderTokenSession( + spec: ProviderTokenSessionSpec, +): ProviderTokenSessionApi { + const isExpired = (tokens: TTokens, now: number): boolean => + isTokenExpired(tokens, now, spec.skewMs); + + const sessions = new Map>(); + + // Each session carries its own in-flight deduplication, so two sessions + // over one home behave like two processes sharing one credential store. + const createSession = (home?: string): TokenSession => { + // Map each refreshed token to the raw refresh token that produced it, + // so updateTokens can CAS-compare at the store; it survives the merge + // and the mutable-token swap inside the store update. + const refreshBasis = new WeakMap(); + + const inner = createTokenSession({ + skewMs: spec.skewMs, + loadProfile: (name) => spec.loadProfile(name, home), + updateTokens: async (name, tokens) => { + const winner = await spec.updateTokens( + name, + tokens, + home, + refreshBasis.get(tokens), + ); + if (winner === undefined) throw new OAuthProfileNotFoundError(name); + replaceMutableTokens(tokens, winner.tokens); + }, + refreshTokens: async (refreshToken, now) => { + try { + const refreshed = await spec.refreshTokens(refreshToken, now); + refreshBasis.set(refreshed, refreshToken); + return refreshed; + } catch (error) { + throw sanitizedRefreshFailure(error, refreshToken); + } + }, + toAccess: spec.toAccess, + mergeRefreshed: (refreshed, previous) => { + const merged = spec.mergeRefreshed + ? spec.mergeRefreshed(refreshed, previous) + : refreshed; + const basis = refreshBasis.get(refreshed); + if (basis !== undefined) refreshBasis.set(merged, basis); + return merged; + }, + }); + + if (spec.serializeRefresh === undefined) return inner; + const serialize = spec.serializeRefresh; + const nowOrDefault = (now: number | undefined): number => + now === undefined ? Date.now() : now; + return { + isExpired: inner.isExpired, + getValidToken: (name, now) => { + const at = nowOrDefault(now); + return serialize({ + home, + name, + now: at, + session: inner, + getValid: () => inner.getValidToken(name, at), + }); + }, + }; + }; + + const sessionFor = (home?: string): TokenSession => { + const key = home ?? ""; + const existing = sessions.get(key); + if (existing !== undefined) return existing; + const created = createSession(home); + sessions.set(key, created); + return created; + }; + + // Projects the core's own errors onto the provider error: anything else is + // not ours to reclassify (e.g. a refresh lock timeout carries its own + // recovery path and must never compose into credential_failure). + const wrapAuthFailure = (name: string, err: unknown): unknown => { + if (err instanceof OAuthProfileNotFoundError) { + return new spec.errorClass( + name, + "missing", + `${spec.providerLabel} profile "${name}" is not authorized. Log in again.`, + ); + } + if (err instanceof OAuthRefreshFailedError) { + const cause = err.cause; + return new spec.errorClass( + name, + "refresh-failed", + `${spec.providerLabel} profile "${name}" could not be refreshed ` + + `(${cause instanceof Error ? cause.message : String(cause)}). Log in again.`, + ); + } + return err; + }; + + return { + createSession, + sessionFor, + isExpired, + getValidToken: async (name, now, home) => { + try { + return await sessionFor(home).getValidToken(name, now); + } catch (err) { + throw wrapAuthFailure(name, err); + } + }, + refreshStaged: async (tokens, now = Date.now()) => { + if (!isExpired(tokens, now)) return tokens; + let refreshed: TTokens; + try { + refreshed = await spec.refreshTokens(tokens.refresh, now, tokens); + } catch (error) { + throw sanitizedRefreshFailure(error, tokens.refresh); + } + replaceMutableTokens(tokens, refreshed); + return tokens; + }, + }; +} diff --git a/src/auth/store.ts b/src/auth/store.ts index 2b7bce936..52c8e436a 100644 --- a/src/auth/store.ts +++ b/src/auth/store.ts @@ -1,17 +1,9 @@ -import { - mkdir, - open, - readFile, - rename, - stat, - unlink, - writeFile, -} from "node:fs/promises"; +import { mkdir, readFile, rename, writeFile } from "node:fs/promises"; import { homedir } from "node:os"; import { dirname, join } from "node:path"; -import { setTimeout as delay } from "node:timers/promises"; import { type } from "arktype"; import type { AuthProfile, BaseTokens } from "@corbits/oauth-core"; +import { isErrnoCode, withFileLock } from "./file-lock.js"; export type { AuthProfile, BaseTokens }; @@ -63,10 +55,6 @@ const LOCK_STALE_MS = 5_000; // writeAuthFile ever overlaps in-process. let tmpWriteCounter = 0; -// Per-waiter unique lock claim (pid + counter): a matching steal re-read -// names the same file, and the post-create check tells our claim from a winner's. -let lockClaimCounter = 0; - // Same-process ops on one auth file queue here so a caller's lock deadline // starts when it runs, not when it was invoked — one held-past-timeout lock // then fails only its first waiter, not the whole burst. @@ -82,15 +70,6 @@ const ProfileShape = type({ tokens: "unknown", }); -function isErrnoCode(err: unknown, code: string): boolean { - return ( - typeof err === "object" && - err !== null && - "code" in err && - err.code === code - ); -} - function isProfile( value: unknown, isTokens: (value: unknown) => value is TTokens, @@ -100,53 +79,6 @@ function isProfile( return isTokens(parsed.tokens); } -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: success or EPERM means alive; ESRCH/EINVAL mean -// dead. Any other failure reads as alive — never steal a live holder's lock -// on a confused signal check; the waiter times out with a recovery hint. -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"; - } -} - -// Null when the lock vanished under the waiter (a release raced the read); -// anything but ENOENT propagates — permission and disk errors must surface. -async function readLockContent(lockPath: string): Promise { - try { - return await readFile(lockPath, "utf8"); - } catch (err) { - if (isErrnoCode(err, "ENOENT")) return null; - throw err; - } -} - -// True when the lock is safe to take over: a tagged holder with a dead PID -// (crashed), or a legacy untagged lock older than the stale horizon. A -// vanished lock reads as not stale; the acquire loop retries the create. -async function isLockStale( - lockPath: string, - content: string, -): Promise { - const pid = holderPid(content); - if (pid !== null) return !isPidAlive(pid); - try { - const info = await stat(lockPath); - return Date.now() - info.mtimeMs > LOCK_STALE_MS; - } catch (err) { - if (isErrnoCode(err, "ENOENT")) return false; - throw err; - } -} - function lockTimeoutError(lockPath: string, cause: unknown): Error { return new Error( `Timed out waiting for OAuth credential lock ${lockPath}. ` + @@ -155,17 +87,6 @@ function lockTimeoutError(lockPath: string, cause: unknown): Error { ); } -// Pace one contention round: throw when the deadline passed, else sleep a -// retry interval. All wait paths funnel here so steal contention never hot-spins. -async function paceLockWait( - deadline: number, - lockPath: string, - cause: unknown, -): Promise { - if (Date.now() >= deadline) throw lockTimeoutError(lockPath, cause); - await delay(LOCK_RETRY_MS); -} - export function createAuthStore( options: AuthStoreOptions, ): AuthStore { @@ -213,103 +134,14 @@ export function createAuthStore( home: string, callback: () => Promise, ): Promise { - const path = authPath(home); - const lockPath = `${path}.lock`; - await mkdir(dirname(path), { recursive: true, mode: 0o700 }); - const deadline = Date.now() + (options.lockTimeoutMs ?? LOCK_TIMEOUT_MS); - // Counter leg keeps every waiter's claim distinct. - const claim = `${process.pid}:${(lockClaimCounter += 1)}`; - let lock; - - while (true) { - try { - const handle = await open(lockPath, "wx", 0o600); - try { - await handle.writeFile(claim, "utf8"); - } catch (writeError) { - try { - await handle.close(); - } catch { - // Ignore close errors on the cleanup path; the write error below is the one the caller must see. - } - try { - await unlink(lockPath); - } catch (unlinkError) { - if (!isErrnoCode(unlinkError, "ENOENT")) throw unlinkError; - } - throw writeError; - } - // A steal may have replaced our fresh file between create and write; - // the path then names a live winner. Close and re-contend — never unlink. - if ((await readLockContent(lockPath)) !== claim) { - try { - await handle.close(); - } catch { - // The path names a live winner; close outcome must not mask the paced retry below. - } - await paceLockWait(deadline, lockPath, undefined); - continue; - } - lock = handle; - break; - } catch (error) { - if (!isErrnoCode(error, "EEXIST")) throw error; - // A crashed holder never releases: take over a stale lock rather - // than brick the store. A lock that vanished under the read is not - // stale — retry the exclusive create. - const content = await readLockContent(lockPath); - if (content !== null && (await isLockStale(lockPath, content))) { - // Re-check before unlinking so a concurrent winner's fresh claim - // is never mistaken for the stale entry just observed — claims - // are unique per waiter. A steal can still interleave between - // re-read and unlink; the post-create check then detects the - // loser and re-contends. - if ((await readLockContent(lockPath)) === content) { - try { - await unlink(lockPath); - } catch (unlinkError) { - // A concurrent winner unlinked first; retry the create. - if (!isErrnoCode(unlinkError, "ENOENT")) throw unlinkError; - } - } - // Fall through to the deadline/sleep path: a steal that lost to - // a concurrent winner paces like any other contention. - } - await paceLockWait(deadline, lockPath, error); - } - } - - let callbackOutcome: - | { ok: true; value: TResult } - | { ok: false; error: unknown }; - try { - callbackOutcome = { ok: true, value: await callback() }; - } catch (error) { - callbackOutcome = { ok: false, error }; - } - - try { - await lock.close(); - } catch { - // The callback outcome owns precedence; a close failure must not mask it. - // The unlink below still runs. - } - - let releaseOutcome: { ok: true } | { ok: false; error: unknown } = { - ok: true, - }; - try { - await unlink(lockPath); - } catch (error) { - // A stale-takeover steal may legitimately remove the file first. - if (!isErrnoCode(error, "ENOENT")) { - releaseOutcome = { ok: false, error }; - } - } - - if (!callbackOutcome.ok) throw callbackOutcome.error; - if (!releaseOutcome.ok) throw releaseOutcome.error; - return callbackOutcome.value; + const lockPath = `${authPath(home)}.lock`; + return withFileLock(lockPath, callback, { + timeoutMs: options.lockTimeoutMs ?? LOCK_TIMEOUT_MS, + retryMs: LOCK_RETRY_MS, + staleMs: LOCK_STALE_MS, + timeoutError: (lockPath, _timeoutMs, cause) => + lockTimeoutError(lockPath, cause), + }); } function enqueueAuthFileOp( diff --git a/src/auth/xai/auth-headers.ts b/src/auth/xai/auth-headers.ts index d66796dcf..660d63077 100644 --- a/src/auth/xai/auth-headers.ts +++ b/src/auth/xai/auth-headers.ts @@ -1,3 +1,4 @@ +import { bearerHeaders } from "../bearer-headers.js"; import { XAI_CLIENT_IDENTIFIER, XAI_CLIENT_VERSION, @@ -8,12 +9,11 @@ import { export function xaiAuthHeadersForToken(token: { readonly access: string; }): Record { - const headers: Record = { - authorization: `Bearer ${token.access}`, + const headers = bearerHeaders(token, { "user-agent": XAI_USER_AGENT, "x-grok-client-identifier": XAI_CLIENT_IDENTIFIER, "x-grok-client-version": XAI_CLIENT_VERSION, - }; + }); const userId = xaiUserIdFromAccessToken(token.access); if (userId !== undefined) headers["x-grok-user-id"] = userId; return headers; diff --git a/src/auth/xai/callback-server.ts b/src/auth/xai/callback-server.ts index f3788bbf7..23e9123ab 100644 --- a/src/auth/xai/callback-server.ts +++ b/src/auth/xai/callback-server.ts @@ -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 { XAI_CALLBACK_HOST, XAI_CALLBACK_PATH, @@ -17,12 +16,14 @@ export async function startXaiCallbackServer( expectedState: string, copy: CallbackPageCopy, ): Promise { - return startCallbackServer(expectedState, { - port: XAI_CALLBACK_PORT, - host: XAI_CALLBACK_HOST, - path: XAI_CALLBACK_PATH, - doneHtml: authorizationDoneHtml("xAI", copy), - failedHtml: (reason) => - callbackPageHtml({ subject: "xAI", error: reason }, copy), - }); + return startProviderCallbackServer( + "xAI", + { + port: XAI_CALLBACK_PORT, + host: XAI_CALLBACK_HOST, + path: XAI_CALLBACK_PATH, + }, + expectedState, + copy, + ); } diff --git a/src/auth/xai/login.ts b/src/auth/xai/login.ts index 7659f103f..8a2e0dce7 100644 --- a/src/auth/xai/login.ts +++ b/src/auth/xai/login.ts @@ -1,14 +1,13 @@ -import { - buildAuthorizeUrl, - startOAuthLogin, - type OAuthLoginHandle, - type StartOAuthLoginOptions, -} from "@corbits/oauth-core"; import { exchangeXaiCode, xaiOAuthConfig, type XaiTokens, } from "@corbits/xai-provider"; +import type { + OAuthLoginHandle, + StartOAuthLoginOptions, +} from "@corbits/oauth-core"; +import { startProviderLogin } from "../provider-login.js"; import type { CallbackPageCopy } from "../callback-page.js"; import { saveXaiProfile } from "../../config/oauth-stores.js"; @@ -24,10 +23,9 @@ export async function startXaiLogin( opts: StartXaiLoginOptions, ): Promise { const { home, copy, ...loginOpts } = opts; - return startOAuthLogin(loginOpts, { + return startProviderLogin(loginOpts, { + oauthConfig: xaiOAuthConfig, startCallbackServer: (state) => startXaiCallbackServer(state, copy), - buildAuthorizeUrl: (pkce, state) => - buildAuthorizeUrl(xaiOAuthConfig, pkce, state), exchangeCode: exchangeXaiCode, saveProfile: (profile) => saveXaiProfile(profile, home), }); diff --git a/src/auth/xai/session.ts b/src/auth/xai/session.ts index f18cc2676..d602e3279 100644 --- a/src/auth/xai/session.ts +++ b/src/auth/xai/session.ts @@ -1,7 +1,4 @@ import { - createTokenSession, - isTokenExpired, - OAuthProfileNotFoundError, OAuthRefreshFailedError, type TokenSession, } from "@corbits/oauth-core"; @@ -13,9 +10,9 @@ import { import { loadXaiProfile, updateXaiTokens } from "../../config/oauth-stores.js"; import { - replaceMutableTokens, - sanitizedRefreshFailure, -} from "../token-session-boundary.js"; + createProviderTokenSession, + type TokenSessionCallContext, +} from "../provider-token-session.js"; export class XaiAuthError extends Error { readonly profile: string; @@ -37,89 +34,63 @@ export interface XaiAccess { access: string; } -function wrapXaiAuthError(name: string, err: unknown): never { - if (err instanceof OAuthProfileNotFoundError) { - throw new XaiAuthError( - name, - "missing", - `xAI profile "${name}" is not authorized. Log in again.`, - ); - } - if (err instanceof OAuthRefreshFailedError) { - const cause = err.cause; - throw new XaiAuthError( - name, - "refresh-failed", - `xAI profile "${name}" could not be refreshed (${cause instanceof Error ? cause.message : String(cause)}). Log in again.`, - ); +export const xaiTokensSession = createProviderTokenSession< + XaiTokens, + XaiAccess +>({ + errorClass: XaiAuthError, + providerLabel: "xAI", + skewMs: XAI_REFRESH_SKEW_MS, + loadProfile: loadXaiProfile, + updateTokens: updateXaiTokens, + toAccess: (tokens) => ({ access: tokens.access }), + refreshTokens: (refreshToken, now) => refreshXaiTokens(refreshToken, now), + // No inter-process lock here: xAI never rotates refresh tokens, so + // overlapping grants cannot revoke each other. What IS needed is the + // committed-winner recovery: a lost invalid_grant race must return the + // rotation another session committed instead of a re-login error. + serializeRefresh: (ctx) => recoverCommittedXaiWinner(ctx), +}); + +// On a shared credential two independent sessions may refresh at once; the +// loser's grant is rejected with invalid_grant. If a newer committed winner +// exists on disk by the time we fail, that is the shared credential's fresh +// access token — return it instead of a re-login error. +async function recoverCommittedXaiWinner({ + home, + name, + now, + getValid, + session, +}: TokenSessionCallContext): Promise { + const basis = await loadXaiProfile(name, home); + try { + return await getValid(); + } catch (error) { + if (error instanceof OAuthRefreshFailedError && basis !== undefined) { + const winner = await loadXaiProfile(name, home); + if ( + winner !== undefined && + !session.isExpired(winner.tokens, now) && + // A committed rotation always changes the access token or its + // lifetime; an unchanged profile is not a winner to recover. + (winner.tokens.access !== basis.tokens.access || + winner.tokens.expiresAt !== basis.tokens.expiresAt) + ) + return { access: winner.tokens.access }; + } + throw error; } - throw err; } -const sessions = new Map>(); - export function createXaiTokenSession( home?: string, ): TokenSession { - const refreshBasis = new WeakMap(); - const inner = createTokenSession({ - skewMs: XAI_REFRESH_SKEW_MS, - loadProfile: (name) => loadXaiProfile(name, home), - updateTokens: async (name, tokens) => { - const winner = await updateXaiTokens( - name, - tokens, - home, - refreshBasis.get(tokens), - ); - if (winner === undefined) throw new OAuthProfileNotFoundError(name); - replaceMutableTokens(tokens, winner.tokens); - }, - refreshTokens: async (refreshToken, now) => { - try { - const refreshed = await refreshXaiTokens(refreshToken, now); - refreshBasis.set(refreshed, refreshToken); - return refreshed; - } catch (error) { - throw sanitizedRefreshFailure(error, refreshToken); - } - }, - toAccess: (tokens) => ({ access: tokens.access }), - }); - return { - isExpired: inner.isExpired, - getValidToken: async (name, now = Date.now()) => { - const basis = await loadXaiProfile(name, home); - try { - return await inner.getValidToken(name, now); - } catch (error) { - if (error instanceof OAuthRefreshFailedError && basis !== undefined) { - const winner = await loadXaiProfile(name, home); - if ( - winner !== undefined && - !inner.isExpired(winner.tokens, now) && - (winner.tokens.access !== basis.tokens.access || - winner.tokens.expiresAt !== basis.tokens.expiresAt) - ) - return { access: winner.tokens.access }; - } - throw error; - } - }, - }; -} - -function sessionFor(home?: string): TokenSession { - const key = home ?? ""; - const existing = sessions.get(key); - if (existing !== undefined) return existing; - const created = createXaiTokenSession(home); - sessions.set(key, created); - return created; + return xaiTokensSession.createSession(home); } export function isXaiTokenExpired(tokens: XaiTokens, now: number): boolean { - return isTokenExpired(tokens, now, XAI_REFRESH_SKEW_MS); + return xaiTokensSession.isExpired(tokens, now); } export async function getValidXaiToken( @@ -127,24 +98,12 @@ export async function getValidXaiToken( now?: number, home?: string, ): Promise { - try { - return await sessionFor(home).getValidToken(name, now); - } catch (err) { - wrapXaiAuthError(name, err); - } + return xaiTokensSession.getValidToken(name, now, home); } export async function refreshStagedXaiTokens( tokens: XaiTokens, now: number = Date.now(), ): Promise { - if (!isXaiTokenExpired(tokens, now)) return tokens; - let refreshed: XaiTokens; - try { - refreshed = await refreshXaiTokens(tokens.refresh, now); - } catch (error) { - throw sanitizedRefreshFailure(error, tokens.refresh); - } - replaceMutableTokens(tokens, refreshed); - return tokens; + return xaiTokensSession.refreshStaged(tokens, now); } diff --git a/src/auth/xai/store.ts b/src/auth/xai/store.ts index e7c8b1053..c76baf9d6 100644 --- a/src/auth/xai/store.ts +++ b/src/auth/xai/store.ts @@ -1,30 +1,21 @@ +import { type } from "arktype"; import type { AuthProfile } from "@corbits/oauth-core"; import type { XaiTokens } from "@corbits/xai-provider"; -import { type } from "arktype"; -import { createAuthStore } from "../store.js"; +import { defineProviderAuthStore } from "../provider-auth-store.js"; export type { XaiTokens }; export type XaiProfile = AuthProfile; -const XaiTokensShape = type({ - access: "string", - refresh: "string", - expiresAt: "number", - "idToken?": "string", -}); - -function isXaiTokens(value: unknown): value is XaiTokens { - return !(XaiTokensShape(value) instanceof type.errors); -} - export const XAI_AUTH_FILENAME = "xai-auth.json"; -export function createXaiAuthStore(settingsDirName: string) { - return createAuthStore({ - filename: XAI_AUTH_FILENAME, - settingsDirName, - isTokens: isXaiTokens, - }); -} +export const createXaiAuthStore = defineProviderAuthStore({ + filename: XAI_AUTH_FILENAME, + tokensShape: type({ + access: "string", + refresh: "string", + expiresAt: "number", + "idToken?": "string", + }), +});