mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-10 01:39:02 +00:00
fix(oauth): Codex race + comprehensive provider error handling (#2718)
Integrated into release/v3.8.3 — comprehensive OAuth refresh race fix (Fix A-F via onPersist/AsyncLocalStorage + mutex consolidation). Replaces token-refresh-race.test.ts with broader token-refresh-race-comprehensive.test.ts that preserves the original invariant plus 11 new assertions.
(cherry picked from commit ac76863ded)
This commit is contained in:
Diego Rodrigues de Sa e Souza
diegosouzapw
parent
a1d8b31171
commit
ad3d4b696a
17 changed files with 1369 additions and 317 deletions
|
|
@ -38,6 +38,7 @@ import {
|
|||
} from "./providerHeaderProfiles.ts";
|
||||
import type { ProviderRequestDefaults } from "../services/providerRequestDefaults.ts";
|
||||
import { resolvePublicCred } from "../utils/publicCreds.ts";
|
||||
import { buildGitLabOAuthEndpoints, GITLAB_DUO_DEFAULT_BASE_URL } from "@/lib/oauth/gitlab";
|
||||
|
||||
// ── Types ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -1048,6 +1049,33 @@ export const REGISTRY: Record<string, RegistryEntry> = {
|
|||
],
|
||||
},
|
||||
|
||||
"gitlab-duo": {
|
||||
id: "gitlab-duo",
|
||||
alias: "gld",
|
||||
format: "openai",
|
||||
executor: "gitlab",
|
||||
// baseUrl is dynamic: resolved at request time from providerSpecificData.baseUrl
|
||||
// by GitlabExecutor.buildUrl() via buildGitLabOAuthEndpoints().
|
||||
// The default here keeps the PROVIDERS map non-null so refreshAccessToken()
|
||||
// can look up this provider.
|
||||
baseUrl: buildGitLabOAuthEndpoints(GITLAB_DUO_DEFAULT_BASE_URL).publicCompletionsUrl,
|
||||
authType: "oauth",
|
||||
authHeader: "bearer",
|
||||
defaultContextLength: 128000,
|
||||
oauth: {
|
||||
clientIdEnv: "GITLAB_DUO_OAUTH_CLIENT_ID",
|
||||
clientIdDefault: process.env.GITLAB_OAUTH_CLIENT_ID || "",
|
||||
clientSecretEnv: "GITLAB_DUO_OAUTH_CLIENT_SECRET",
|
||||
clientSecretDefault: process.env.GITLAB_OAUTH_CLIENT_SECRET || "",
|
||||
tokenUrl: buildGitLabOAuthEndpoints(GITLAB_DUO_DEFAULT_BASE_URL).tokenUrl,
|
||||
authUrl: buildGitLabOAuthEndpoints(GITLAB_DUO_DEFAULT_BASE_URL).authorizeUrl,
|
||||
},
|
||||
models: [
|
||||
{ id: "claude-sonnet-4-6", name: "Claude Sonnet 4.6 (GitLab Duo)" },
|
||||
{ id: "claude-haiku-4-5", name: "Claude Haiku 4.5 (GitLab Duo)" },
|
||||
],
|
||||
},
|
||||
|
||||
cursor: {
|
||||
id: "cursor",
|
||||
alias: "cu",
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import {
|
|||
} from "../services/apiKeyRotator.ts";
|
||||
import type { KeyHealth } from "../services/apiKeyRotator.ts";
|
||||
import { getOpenAICompatibleType, isClaudeCodeCompatible } from "../services/provider.ts";
|
||||
import { runWithOnPersist, getRefreshLeadMs } from "../services/tokenRefresh.ts";
|
||||
import type { ProviderRequestDefaults } from "../services/providerRequestDefaults.ts";
|
||||
import { signRequestBody } from "../services/claudeCodeCCH.ts";
|
||||
import {
|
||||
|
|
@ -453,7 +454,12 @@ export class BaseExecutor {
|
|||
needsRefresh(credentials?: ProviderCredentials | null) {
|
||||
if (!credentials?.expiresAt) return false;
|
||||
const expiresAtMs = new Date(credentials.expiresAt).getTime();
|
||||
return expiresAtMs - Date.now() < 5 * 60 * 1000;
|
||||
// Use the provider-specific lead time (REFRESH_LEAD_MS) so rotating-token
|
||||
// providers like Codex refresh proactively far ahead of expiry. Keeping the
|
||||
// refresh_token "warm" prevents Auth0 from marking it as stale and revoking
|
||||
// the token family on first use after long idle.
|
||||
const lead = getRefreshLeadMs(this.provider);
|
||||
return expiresAtMs - Date.now() < lead;
|
||||
}
|
||||
|
||||
parseError(response: Response, bodyText: string) {
|
||||
|
|
@ -551,14 +557,34 @@ export class BaseExecutor {
|
|||
|
||||
if (this.needsRefresh(credentials)) {
|
||||
try {
|
||||
const refreshed = await this.refreshCredentials(credentials, log || null);
|
||||
if (refreshed) {
|
||||
// Fix A: wire onCredentialsRefreshed through runWithOnPersist so it runs
|
||||
// INSIDE the per-connection mutex inside getAccessToken. Not every
|
||||
// executor routes through getAccessToken (e.g. github.ts), so use a flag
|
||||
// to detect whether the persist callback actually fired and fall back to
|
||||
// post-refresh mutation when it didn't.
|
||||
let proactivePersistRan = false;
|
||||
const proactiveOnPersist = arguments[0].onCredentialsRefreshed
|
||||
? async (refreshResult: Record<string, unknown>) => {
|
||||
proactivePersistRan = true;
|
||||
activeCredentials = {
|
||||
...credentials,
|
||||
...(refreshResult as Partial<ProviderCredentials>),
|
||||
};
|
||||
await arguments[0].onCredentialsRefreshed(
|
||||
refreshResult as Partial<ProviderCredentials>
|
||||
);
|
||||
}
|
||||
: null;
|
||||
|
||||
const refreshed = await runWithOnPersist(proactiveOnPersist, () =>
|
||||
this.refreshCredentials(credentials, log || null)
|
||||
);
|
||||
|
||||
if (refreshed && !proactivePersistRan) {
|
||||
activeCredentials = {
|
||||
...credentials,
|
||||
...refreshed,
|
||||
};
|
||||
// Persist the proactively refreshed credentials to prevent consuming rotating tokens
|
||||
// without updating the central database connection.
|
||||
if (arguments[0].onCredentialsRefreshed) {
|
||||
await arguments[0].onCredentialsRefreshed(refreshed);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,7 +14,11 @@ import { resolveStreamReadinessTimeout } from "../utils/streamReadinessPolicy.ts
|
|||
import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts";
|
||||
import { createSseHeartbeatTransform, shapeForClientFormat } from "../utils/sseHeartbeat.ts";
|
||||
import { addBufferToUsage, filterUsageForFormat, estimateUsage } from "../utils/usageTracking.ts";
|
||||
import { refreshWithRetry, isUnrecoverableRefreshError } from "../services/tokenRefresh.ts";
|
||||
import {
|
||||
refreshWithRetry,
|
||||
isUnrecoverableRefreshError,
|
||||
runWithOnPersist,
|
||||
} from "../services/tokenRefresh.ts";
|
||||
import { createRequestLogger } from "../utils/requestLogger.ts";
|
||||
import { getModelTargetFormat, PROVIDER_ID_TO_ALIAS } from "../config/providerModels.ts";
|
||||
import { DEFAULT_THINKING_CLAUDE_SIGNATURE } from "../config/defaultThinkingSignature.ts";
|
||||
|
|
@ -3813,8 +3817,30 @@ export async function handleChatCore({
|
|||
isQwenExpiredError) &&
|
||||
!hadStreamOptions // Skip refresh if failure may be from stream_options removal, not auth
|
||||
) {
|
||||
// Fix A: wrap refreshCredentials in runWithOnPersist so the persist callback
|
||||
// executes INSIDE the per-connection mutex held by getAccessToken. This makes
|
||||
// [network refresh + DB write + outer-state mutation] one atomic step and
|
||||
// prevents concurrent requests from reading a stale refreshToken before the
|
||||
// DB has been updated (refresh_token_reused on Codex/OpenAI).
|
||||
//
|
||||
// Not every executor routes refresh through getAccessToken (e.g. github.ts
|
||||
// calls refreshCopilotToken directly). When the persistFn doesn't fire from
|
||||
// inside getAccessToken, we still need to do the credentials mutation + user
|
||||
// callback after refreshCredentials returns. The `persistFnRan` flag tracks
|
||||
// which path executed so we don't double-fire (race-prone) or skip (regression).
|
||||
let persistFnRan = false;
|
||||
const persistFn = onCredentialsRefreshed
|
||||
? async (refreshResult: any) => {
|
||||
persistFnRan = true;
|
||||
// Mutate the shared credentials object so subsequent executor calls
|
||||
// in this request see the new tokens. Runs INSIDE the mutex.
|
||||
Object.assign(credentials, refreshResult);
|
||||
await onCredentialsRefreshed(refreshResult);
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const newCredentials = (await refreshWithRetry(
|
||||
() => executor.refreshCredentials(credentials, log),
|
||||
() => runWithOnPersist(persistFn, () => executor.refreshCredentials(credentials, log)),
|
||||
3,
|
||||
log,
|
||||
provider // Explicitly pass the provider to avoid universally tripping the "unknown" circuit breaker
|
||||
|
|
@ -3826,12 +3852,15 @@ export async function handleChatCore({
|
|||
if (newCredentials?.accessToken || newCredentials?.copilotToken) {
|
||||
log?.info?.("TOKEN", `${provider.toUpperCase()} | refreshed`);
|
||||
|
||||
// Update credentials
|
||||
Object.assign(credentials, newCredentials);
|
||||
|
||||
// Notify caller about refreshed credentials
|
||||
if (onCredentialsRefreshed && newCredentials) {
|
||||
await onCredentialsRefreshed(newCredentials);
|
||||
// Fall back to post-mutex mutation only for executors that don't route
|
||||
// through getAccessToken (and therefore never fire onPersist). For
|
||||
// executors that DO route through it (Codex, Claude, Gemini, etc.) the
|
||||
// mutation already happened atomically inside the mutex.
|
||||
if (!persistFnRan) {
|
||||
Object.assign(credentials, newCredentials);
|
||||
if (onCredentialsRefreshed) {
|
||||
await onCredentialsRefreshed(newCredentials);
|
||||
}
|
||||
}
|
||||
|
||||
// Retry with new credentials — model + extra headers follow translatedBody.model so they
|
||||
|
|
|
|||
|
|
@ -1,13 +1,59 @@
|
|||
// @ts-nocheck
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { PROVIDERS, OAUTH_ENDPOINTS } from "../config/constants.ts";
|
||||
import { getGitHubCopilotRefreshHeaders } from "../config/providerHeaderProfiles.ts";
|
||||
import { pbkdf2Sync } from "node:crypto";
|
||||
import { runWithProxyContext } from "../utils/proxyFetch.ts";
|
||||
import { WINDSURF_CONFIG } from "@/lib/oauth/constants/oauth";
|
||||
import { buildGitLabOAuthEndpoints, resolveGitLabOAuthBaseUrl } from "@/lib/oauth/gitlab";
|
||||
|
||||
// Token expiry buffer (refresh if expires within 5 minutes)
|
||||
// Default token expiry buffer (refresh if expires within 5 minutes).
|
||||
// Used as fallback for providers without an explicit lead time in
|
||||
// REFRESH_LEAD_MS below.
|
||||
export const TOKEN_EXPIRY_BUFFER_MS = 5 * 60 * 1000;
|
||||
|
||||
// Per-provider proactive-refresh lead time.
|
||||
//
|
||||
// For multi-account OAuth on providers that enforce "single active session per
|
||||
// client_id" (notably OpenAI Codex / Auth0), refreshing one account's token
|
||||
// can invalidate the refresh_token family of OTHER accounts under the same
|
||||
// client. We MINIMIZE refresh frequency for these providers: stay on the
|
||||
// original access_token until it is genuinely about to expire, so each account
|
||||
// gets the full access_token lifetime without triggering Auth0's family-
|
||||
// invalidation logic on its siblings.
|
||||
//
|
||||
// Trade-off: when refresh finally happens (last 5 min before expiry), Auth0
|
||||
// MAY invalidate other accounts' refresh_tokens. The user must re-auth those.
|
||||
// This is the upstream limitation documented in openai/codex#9648.
|
||||
//
|
||||
// Providers with non-rotating tokens (Google, Anthropic) or where multi-
|
||||
// account is naturally isolated keep longer lead times.
|
||||
export const REFRESH_LEAD_MS: Record<string, number> = {
|
||||
// Rotating refresh tokens — minimize refresh frequency to avoid the
|
||||
// "refresh-invalidates-siblings" cascade documented for OpenAI Auth0.
|
||||
codex: 5 * 60 * 1000, // 5 minutes
|
||||
openai: 5 * 60 * 1000, // same Auth0 backend as codex
|
||||
claude: 5 * 60 * 1000, // Anthropic OAuth rotates refresh_tokens (user-reported)
|
||||
"gitlab-duo": 5 * 60 * 1000, // GitLab token family revocation on misuse
|
||||
kiro: 5 * 60 * 1000, // AWS SSO OIDC issues one-time-use refresh tokens
|
||||
"kimi-coding": 5 * 60 * 1000, // Moonshot rotates per-refresh
|
||||
qwen: 5 * 60 * 1000, // Alibaba device-code path also rotates
|
||||
// Non-rotating providers — longer lead is safe.
|
||||
iflow: 24 * 60 * 60 * 1000, // 24 hours
|
||||
// Google OAuth refresh_tokens are permanent (non-rotating) — longer lead
|
||||
// is safe and reduces unnecessary upstream chatter.
|
||||
"gemini-cli": 15 * 60 * 1000,
|
||||
antigravity: 15 * 60 * 1000,
|
||||
};
|
||||
|
||||
/**
|
||||
* Get the proactive refresh lead time (ms) for a given provider.
|
||||
* Falls back to TOKEN_EXPIRY_BUFFER_MS (5 min) when not explicitly listed.
|
||||
*/
|
||||
export function getRefreshLeadMs(provider: string): number {
|
||||
return REFRESH_LEAD_MS[provider] ?? TOKEN_EXPIRY_BUFFER_MS;
|
||||
}
|
||||
|
||||
const CACHE_SECRET = "omniroute-token-cache";
|
||||
|
||||
// In-flight refresh promise cache to prevent race conditions
|
||||
|
|
@ -19,6 +65,98 @@ const refreshPromiseCache = new Map();
|
|||
// Primary dedup when credentials.connectionId is present; refreshPromiseCache is fallback.
|
||||
const connectionRefreshMutex = new Map();
|
||||
|
||||
// ─── Token Rotation Map (codex-multi-auth pattern) ─────────────────────────
|
||||
//
|
||||
// When a rotating-token provider (Codex, Kimi, GitLab Duo, etc.) refreshes,
|
||||
// the old refresh_token is consumed and a new one is issued. Any subsequent
|
||||
// caller arriving with the OLD token would, without protection, hit upstream
|
||||
// and trigger "refresh_token_reused" — which Auth0 treats as a security event
|
||||
// and invalidates the entire token family.
|
||||
//
|
||||
// This in-memory map caches RECENT rotations so a stale caller can be redirected
|
||||
// to the new tokens WITHOUT touching upstream. The DB staleness check inside
|
||||
// the per-connection mutex covers the same scenario when connectionId is known,
|
||||
// but not all callers pass connectionId (e.g., legacy code paths, retries that
|
||||
// snapshot credentials before the rotation lands in DB).
|
||||
//
|
||||
// Ported from ndycode/codex-multi-auth (lib/refresh-queue.ts:218-248), the only
|
||||
// publicly known tool that reliably sustains multiple Codex OAuth accounts.
|
||||
//
|
||||
// Key format: `provider:sha256(oldRefreshToken)`
|
||||
// Value: { result: tokens, expiresAt: ms_since_epoch }
|
||||
type RotationEntry = {
|
||||
result: { accessToken: string; refreshToken: string; expiresIn?: number; expiresAt?: string };
|
||||
expiresAt: number;
|
||||
};
|
||||
const tokenRotationMap = new Map<string, RotationEntry>();
|
||||
const ROTATION_MAP_TTL_MS = 60 * 1000; // 60 seconds — long enough to catch in-flight stale callers
|
||||
|
||||
function cleanupRotationMap(now: number = Date.now()): void {
|
||||
if (tokenRotationMap.size === 0) return;
|
||||
for (const [key, entry] of tokenRotationMap.entries()) {
|
||||
if (entry.expiresAt <= now) tokenRotationMap.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
function lookupRotation(provider: string, refreshToken: string): RotationEntry | undefined {
|
||||
cleanupRotationMap();
|
||||
const key = getRefreshCacheKey(provider, refreshToken);
|
||||
const entry = tokenRotationMap.get(key);
|
||||
if (!entry) return undefined;
|
||||
if (entry.expiresAt <= Date.now()) {
|
||||
tokenRotationMap.delete(key);
|
||||
return undefined;
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
function recordRotation(
|
||||
provider: string,
|
||||
oldRefreshToken: string,
|
||||
result: { accessToken: string; refreshToken: string; expiresIn?: number; expiresAt?: string }
|
||||
): void {
|
||||
if (!oldRefreshToken || !result.refreshToken || oldRefreshToken === result.refreshToken) {
|
||||
return;
|
||||
}
|
||||
const key = getRefreshCacheKey(provider, oldRefreshToken);
|
||||
tokenRotationMap.set(key, {
|
||||
result,
|
||||
expiresAt: Date.now() + ROTATION_MAP_TTL_MS,
|
||||
});
|
||||
}
|
||||
|
||||
// Exported for tests + diagnostics; not part of the public API surface.
|
||||
export function _getTokenRotationMapStats(): { size: number; entries: number } {
|
||||
cleanupRotationMap();
|
||||
return { size: tokenRotationMap.size, entries: tokenRotationMap.size };
|
||||
}
|
||||
|
||||
export function _clearTokenRotationMap(): void {
|
||||
tokenRotationMap.clear();
|
||||
}
|
||||
|
||||
// AsyncLocalStorage for plumbing `onPersist` through executor.refreshCredentials
|
||||
// without modifying every executor's signature. The chatCore.ts / base.ts call
|
||||
// sites wrap executor.refreshCredentials in `runWithOnPersist(persistFn, () => ...)`
|
||||
// and `getAccessToken` reads the active store as a fallback when no explicit
|
||||
// onPersist parameter is provided. This keeps Fix A's atomic [refresh + persist]
|
||||
// guarantee while avoiding per-executor signature changes.
|
||||
type RefreshPersistResult = Record<string, unknown>;
|
||||
type RefreshPersistFn = (result: RefreshPersistResult) => Promise<void>;
|
||||
const onPersistStore = new AsyncLocalStorage<RefreshPersistFn>();
|
||||
|
||||
export function runWithOnPersist<T>(
|
||||
onPersist: RefreshPersistFn | undefined | null,
|
||||
fn: () => Promise<T>
|
||||
): Promise<T> {
|
||||
if (!onPersist) return fn();
|
||||
return onPersistStore.run(onPersist, fn);
|
||||
}
|
||||
|
||||
export function getActiveOnPersist(): RefreshPersistFn | undefined {
|
||||
return onPersistStore.getStore();
|
||||
}
|
||||
|
||||
type RefreshLogger = {
|
||||
info?: (tag: string, message: string, data?: Record<string, unknown>) => void;
|
||||
warn?: (tag: string, message: string, data?: Record<string, unknown>) => void;
|
||||
|
|
@ -177,6 +315,36 @@ export async function refreshWindsurfToken(
|
|||
status: response.status,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
|
||||
// Firebase STS returns structured errors. Detect unrecoverable token states.
|
||||
try {
|
||||
const fbError = JSON.parse(errorText);
|
||||
const fbCode =
|
||||
typeof fbError?.error?.message === "string"
|
||||
? fbError.error.message
|
||||
: typeof fbError?.error === "string"
|
||||
? fbError.error
|
||||
: null;
|
||||
if (
|
||||
typeof fbCode === "string" &&
|
||||
(fbCode.includes("USER_DISABLED") ||
|
||||
fbCode.includes("TOKEN_EXPIRED") ||
|
||||
fbCode.includes("INVALID_REFRESH_TOKEN") ||
|
||||
fbCode.includes("USER_NOT_FOUND"))
|
||||
) {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"Windsurf Firebase token is permanently invalid. Re-authentication required.",
|
||||
{
|
||||
fbCode,
|
||||
}
|
||||
);
|
||||
return { error: "unrecoverable_refresh_error", code: fbCode };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
|
@ -261,20 +429,39 @@ export async function refreshClineToken(refreshToken, log, proxyConfig: unknown
|
|||
/**
|
||||
* Specialized refresh for Kimi Coding OAuth tokens.
|
||||
* Uses custom X-Msh-* headers required by Kimi OAuth API.
|
||||
*
|
||||
* Uses a stable device_id from providerSpecificData (stored at login) to avoid
|
||||
* anti-bot detection from ephemeral IDs. If absent, derives a deterministic ID
|
||||
* from the refresh token hash so it is at least stable across refreshes for the
|
||||
* same token.
|
||||
*/
|
||||
export async function refreshKimiCodingToken(refreshToken, log, proxyConfig: unknown = null) {
|
||||
export async function refreshKimiCodingToken(
|
||||
refreshToken: string,
|
||||
providerSpecificData: Record<string, unknown> | null | undefined,
|
||||
log: RefreshLogger,
|
||||
proxyConfig: unknown = null
|
||||
) {
|
||||
const endpoint = PROVIDERS["kimi-coding"]?.refreshUrl || PROVIDERS["kimi-coding"]?.tokenUrl;
|
||||
if (!endpoint) {
|
||||
log?.warn?.("TOKEN_REFRESH", "No refresh URL configured for Kimi Coding");
|
||||
return null;
|
||||
}
|
||||
|
||||
// Generate device info for headers (same as OAuth flow)
|
||||
const deviceId = "kimi-refresh-" + Date.now();
|
||||
const platform = "omniroute";
|
||||
const version = "2.1.2";
|
||||
const deviceModel =
|
||||
typeof process !== "undefined" ? `${process.platform} ${process.arch}` : "unknown";
|
||||
// Prefer stable device_id persisted at login time; fall back to a
|
||||
// deterministic hash of the refresh token so it is at least consistent
|
||||
// across refreshes for the same session.
|
||||
const stableDeviceId =
|
||||
(providerSpecificData?.deviceId as string) ||
|
||||
pbkdf2Sync(refreshToken, "kimi-device-id", 1000, 16, "sha256").toString("hex");
|
||||
|
||||
const platform = "kimi_cli";
|
||||
const version = process.env.KIMI_CLI_VERSION || "1.36.0";
|
||||
|
||||
// Build device model string matching the format from providers/kimi-coding.ts.
|
||||
// open-sse is a portable workspace — use process.platform/arch (always available in Node).
|
||||
const osTypeStr = typeof process !== "undefined" ? process.platform : "unknown";
|
||||
const archStr = typeof process !== "undefined" ? process.arch : "unknown";
|
||||
const deviceModel = [osTypeStr, archStr].filter(Boolean).join(" ");
|
||||
|
||||
try {
|
||||
const params = new URLSearchParams({
|
||||
|
|
@ -292,7 +479,12 @@ export async function refreshKimiCodingToken(refreshToken, log, proxyConfig: unk
|
|||
"X-Msh-Platform": platform,
|
||||
"X-Msh-Version": version,
|
||||
"X-Msh-Device-Model": deviceModel,
|
||||
"X-Msh-Device-Id": deviceId,
|
||||
"X-Msh-Device-Id": stableDeviceId,
|
||||
// These headers match getKimiOAuthHeaders() in providers/kimi-coding.ts.
|
||||
// They're derived at runtime from os module calls; use safe fallbacks here
|
||||
// since open-sse is a portable workspace without direct fs/os access.
|
||||
"X-Msh-Device-Name": (providerSpecificData?.deviceName as string) || osTypeStr,
|
||||
"X-Msh-Os-Version": (providerSpecificData?.osVersion as string) || osTypeStr,
|
||||
},
|
||||
body: params,
|
||||
})
|
||||
|
|
@ -300,9 +492,28 @@ export async function refreshKimiCodingToken(refreshToken, log, proxyConfig: unk
|
|||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text();
|
||||
|
||||
// Detect unrecoverable errors
|
||||
try {
|
||||
const parsed = JSON.parse(errorText);
|
||||
const errorCode = parsed?.error;
|
||||
if (errorCode === "invalid_grant" || errorCode === "invalid_request") {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"Kimi Coding refresh token invalid. Re-authentication required.",
|
||||
{
|
||||
errorCode,
|
||||
}
|
||||
);
|
||||
return { error: "unrecoverable_refresh_error", code: errorCode };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
log?.error?.("TOKEN_REFRESH", "Failed to refresh Kimi Coding token", {
|
||||
status: response.status,
|
||||
error: errorText,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
|
@ -322,7 +533,106 @@ export async function refreshKimiCodingToken(refreshToken, log, proxyConfig: unk
|
|||
scope: tokens.scope,
|
||||
};
|
||||
} catch (error) {
|
||||
log?.error?.("TOKEN_REFRESH", `Network error refreshing Kimi Coding token: ${error.message}`);
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
`Network error refreshing Kimi Coding token: ${error instanceof Error ? error.message : String(error)}`
|
||||
);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Specialized refresh for GitLab Duo OAuth tokens.
|
||||
* Token URL is instance-specific; resolves from providerSpecificData.baseUrl.
|
||||
* Uses PKCE authorization_code flow initially but refresh_token grant does NOT
|
||||
* require code_verifier — only client_id + refresh_token.
|
||||
* On invalid_grant (revoked/expired refresh token) returns the unrecoverable sentinel.
|
||||
*/
|
||||
export async function refreshGitLabDuoToken(
|
||||
refreshToken: string,
|
||||
providerSpecificData: Record<string, unknown> | null | undefined,
|
||||
log: RefreshLogger,
|
||||
proxyConfig: unknown = null
|
||||
) {
|
||||
if (!refreshToken) {
|
||||
log?.warn?.("TOKEN_REFRESH", "No refresh token for GitLab Duo");
|
||||
return null;
|
||||
}
|
||||
|
||||
const baseUrl = resolveGitLabOAuthBaseUrl(providerSpecificData);
|
||||
const endpoints = buildGitLabOAuthEndpoints(baseUrl);
|
||||
const tokenUrl = endpoints.tokenUrl;
|
||||
|
||||
// client_id from providerSpecificData (stored at login) or fall back to PROVIDERS config
|
||||
const clientId =
|
||||
(providerSpecificData?.clientId as string) ||
|
||||
PROVIDERS["gitlab-duo"]?.clientId ||
|
||||
process.env.GITLAB_DUO_OAUTH_CLIENT_ID ||
|
||||
process.env.GITLAB_OAUTH_CLIENT_ID ||
|
||||
"";
|
||||
|
||||
try {
|
||||
const response = await runWithProxyContext(proxyConfig, () =>
|
||||
fetch(tokenUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
Accept: "application/json",
|
||||
},
|
||||
body: buildFormParams({
|
||||
grant_type: "refresh_token",
|
||||
refresh_token: refreshToken,
|
||||
client_id: clientId,
|
||||
}),
|
||||
})
|
||||
);
|
||||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text();
|
||||
|
||||
// Detect unrecoverable token — GitLab returns standard OAuth2 error codes.
|
||||
try {
|
||||
const errorBody = JSON.parse(errorText);
|
||||
const errorCode = errorBody.error;
|
||||
if (errorCode === "invalid_grant" || errorCode === "invalid_request") {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"GitLab Duo refresh token invalid. Re-authentication required.",
|
||||
{
|
||||
errorCode,
|
||||
}
|
||||
);
|
||||
return { error: "unrecoverable_refresh_error", code: errorCode };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
log?.error?.("TOKEN_REFRESH", "Failed to refresh GitLab Duo token", {
|
||||
status: response.status,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const tokens = await response.json();
|
||||
|
||||
log?.info?.("TOKEN_REFRESH", "Successfully refreshed GitLab Duo token", {
|
||||
hasNewAccessToken: !!tokens.access_token,
|
||||
hasNewRefreshToken: !!tokens.refresh_token,
|
||||
expiresIn: tokens.expires_in,
|
||||
});
|
||||
|
||||
return {
|
||||
accessToken: tokens.access_token,
|
||||
refreshToken: tokens.refresh_token || refreshToken,
|
||||
expiresIn: tokens.expires_in,
|
||||
};
|
||||
} catch (error) {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
`Network error refreshing GitLab Duo token: ${error instanceof Error ? error.message : String(error)}`
|
||||
);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -364,7 +674,7 @@ export async function refreshClaudeOAuthToken(refreshToken, log, proxyConfig: un
|
|||
error: errorBody,
|
||||
});
|
||||
if (errorBody.error === "invalid_grant" || errorBody.error === "invalid_request") {
|
||||
return { error: errorBody.error, code: `http_${response.status}` };
|
||||
return { error: "unrecoverable_refresh_error", code: errorBody.error };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
@ -418,8 +728,22 @@ export async function refreshGoogleToken(
|
|||
const errorText = await response.text();
|
||||
log?.error?.("TOKEN_REFRESH", "Failed to refresh Google token", {
|
||||
status: response.status,
|
||||
error: errorText,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
|
||||
// Detect unrecoverable token (invalid_grant = revoked / expired refresh token)
|
||||
try {
|
||||
const errorBody = JSON.parse(errorText);
|
||||
if (errorBody.error === "invalid_grant") {
|
||||
log?.error?.("TOKEN_REFRESH", "Google refresh token invalid. Re-authentication required.", {
|
||||
provider: "google",
|
||||
});
|
||||
return { error: "unrecoverable_refresh_error", code: "invalid_grant" };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
|
@ -486,15 +810,16 @@ export async function refreshQwenToken(refreshToken, log, proxyConfig: unknown =
|
|||
// not JSON, ignore
|
||||
}
|
||||
|
||||
if (errorCode === "invalid_request") {
|
||||
if (errorCode === "invalid_request" || errorCode === "invalid_grant") {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"Qwen refresh token is invalid or expired. Re-authentication required.",
|
||||
{
|
||||
status: response.status,
|
||||
errorCode,
|
||||
}
|
||||
);
|
||||
return { error: "invalid_request" };
|
||||
return { error: "unrecoverable_refresh_error", code: errorCode };
|
||||
}
|
||||
|
||||
log?.warn?.("TOKEN_REFRESH", `Error with Qwen endpoint`, {
|
||||
|
|
@ -527,11 +852,17 @@ export async function refreshCodexToken(refreshToken, log, proxyConfig: unknown
|
|||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
Accept: "application/json",
|
||||
},
|
||||
// Body intentionally omits `scope`. RFC 6749 §6 makes scope optional on a
|
||||
// refresh_token grant (the server reuses the originally-granted scope when
|
||||
// absent). Including `scope` causes Auth0 (which OpenAI Codex OAuth is
|
||||
// built on) to treat the request as a re-scope, which can invalidate
|
||||
// sibling refresh_token families on the same client_id. Matches the
|
||||
// pattern used by ndycode/codex-multi-auth, the only known tool that
|
||||
// sustains multiple Codex accounts without cross-invalidation.
|
||||
body: buildFormParams({
|
||||
grant_type: "refresh_token",
|
||||
refresh_token: refreshToken,
|
||||
client_id: PROVIDERS.codex.clientId,
|
||||
scope: "openid profile email offline_access",
|
||||
}),
|
||||
})
|
||||
);
|
||||
|
|
@ -636,9 +967,32 @@ export async function refreshKiroToken(
|
|||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text();
|
||||
|
||||
// AWS SSO OIDC uses {"__type": "InvalidGrantException"} error format (not standard OAuth2).
|
||||
try {
|
||||
const awsError = JSON.parse(errorText);
|
||||
const awsErrorType = awsError.__type || awsError.error;
|
||||
if (
|
||||
awsErrorType === "InvalidGrantException" ||
|
||||
awsErrorType === "ExpiredTokenException" ||
|
||||
awsErrorType === "invalid_grant"
|
||||
) {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"Kiro AWS refresh token expired/invalid. Re-authentication required.",
|
||||
{
|
||||
awsErrorType,
|
||||
}
|
||||
);
|
||||
return { error: "unrecoverable_refresh_error", code: awsErrorType };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
log?.error?.("TOKEN_REFRESH", "Failed to refresh Kiro AWS token", {
|
||||
status: response.status,
|
||||
error: errorText,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
|
@ -678,9 +1032,32 @@ export async function refreshKiroToken(
|
|||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text();
|
||||
|
||||
// Also check for AWS-style errors on the social auth path (Kiro may relay them)
|
||||
try {
|
||||
const awsError = JSON.parse(errorText);
|
||||
const awsErrorType = awsError.__type || awsError.error;
|
||||
if (
|
||||
awsErrorType === "InvalidGrantException" ||
|
||||
awsErrorType === "ExpiredTokenException" ||
|
||||
awsErrorType === "invalid_grant"
|
||||
) {
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
"Kiro social refresh token expired/invalid. Re-authentication required.",
|
||||
{
|
||||
awsErrorType,
|
||||
}
|
||||
);
|
||||
return { error: "unrecoverable_refresh_error", code: awsErrorType };
|
||||
}
|
||||
} catch {
|
||||
// not JSON — fall through
|
||||
}
|
||||
|
||||
log?.error?.("TOKEN_REFRESH", "Failed to refresh Kiro social token", {
|
||||
status: response.status,
|
||||
error: errorText,
|
||||
error: errorText.slice(0, 200),
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
|
@ -885,7 +1262,20 @@ async function _getAccessTokenInternal(provider, credentials, log, proxyConfig:
|
|||
return await refreshClineToken(credentials.refreshToken, log, proxyConfig);
|
||||
|
||||
case "kimi-coding":
|
||||
return await refreshKimiCodingToken(credentials.refreshToken, log, proxyConfig);
|
||||
return await refreshKimiCodingToken(
|
||||
credentials.refreshToken,
|
||||
credentials.providerSpecificData,
|
||||
log,
|
||||
proxyConfig
|
||||
);
|
||||
|
||||
case "gitlab-duo":
|
||||
return await refreshGitLabDuoToken(
|
||||
credentials.refreshToken,
|
||||
credentials.providerSpecificData,
|
||||
log,
|
||||
proxyConfig
|
||||
);
|
||||
|
||||
case "windsurf":
|
||||
case "devin-cli":
|
||||
|
|
@ -921,6 +1311,7 @@ export function supportsTokenRefresh(provider) {
|
|||
"kimi-coding",
|
||||
"windsurf",
|
||||
"devin-cli",
|
||||
"gitlab-duo",
|
||||
]);
|
||||
if (explicitlySupported.has(provider)) return true;
|
||||
const config = PROVIDERS[provider];
|
||||
|
|
@ -957,13 +1348,30 @@ export function isUnrecoverableRefreshError(result) {
|
|||
* Additionally, when connectionId is present, the stale-token check reads the DB to detect
|
||||
* whether another process already refreshed the token. If the DB token is still valid it is
|
||||
* returned immediately without a new upstream call.
|
||||
*
|
||||
* @param onPersist - Optional callback invoked INSIDE the per-connection mutex closure after a
|
||||
* successful refresh, before the mutex releases. Use this to atomically persist the new tokens
|
||||
* to the DB within the same lock window. If `onPersist` throws, the error is logged and
|
||||
* re-thrown so the caller is aware of the persistence failure.
|
||||
*/
|
||||
export async function getAccessToken(provider, credentials, log, proxyConfig: unknown = null) {
|
||||
export async function getAccessToken(
|
||||
provider,
|
||||
credentials,
|
||||
log,
|
||||
proxyConfig: unknown = null,
|
||||
onPersist?: RefreshPersistFn
|
||||
) {
|
||||
if (!credentials || !credentials.refreshToken || typeof credentials.refreshToken !== "string") {
|
||||
log?.warn?.("TOKEN_REFRESH", `No valid refresh token available for provider: ${provider}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
// If the caller did not pass onPersist explicitly, fall back to the active
|
||||
// AsyncLocalStorage store. This lets `runWithOnPersist(persistFn, () =>
|
||||
// executor.refreshCredentials(creds, log))` plumb the persist callback through
|
||||
// executors (e.g. CodexExecutor) without modifying their signature.
|
||||
const effectiveOnPersist = onPersist ?? getActiveOnPersist();
|
||||
|
||||
const connectionId = credentials.connectionId;
|
||||
|
||||
// ── Layer 1: per-connection mutex ──────────────────────────────────────────
|
||||
|
|
@ -980,12 +1388,29 @@ export async function getAccessToken(provider, credentials, log, proxyConfig: un
|
|||
}
|
||||
|
||||
const entry = { promise: null, waiters: 0 };
|
||||
entry.promise = _getAccessTokenWithStalenessCheck(
|
||||
provider,
|
||||
credentials,
|
||||
log,
|
||||
proxyConfig
|
||||
).finally(() => {
|
||||
entry.promise = (async () => {
|
||||
const result = await _getAccessTokenWithStalenessCheck(
|
||||
provider,
|
||||
credentials,
|
||||
log,
|
||||
proxyConfig
|
||||
);
|
||||
// Invoke onPersist INSIDE the mutex so [network call + DB write] are one atomic step.
|
||||
// This prevents a concurrent waiter from reading stale credentials before the DB is updated.
|
||||
if (result?.accessToken && effectiveOnPersist) {
|
||||
try {
|
||||
await effectiveOnPersist(result);
|
||||
} catch (persistErr) {
|
||||
const { sanitizeErrorMessage } = await import("../utils/error.ts");
|
||||
log?.error?.(
|
||||
"TOKEN_REFRESH",
|
||||
`onPersist callback failed for ${provider}/${connectionId}: ${sanitizeErrorMessage(persistErr instanceof Error ? persistErr : new Error(String(persistErr)))}`
|
||||
);
|
||||
throw persistErr;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
})().finally(() => {
|
||||
connectionRefreshMutex.delete(connectionId);
|
||||
});
|
||||
connectionRefreshMutex.set(connectionId, entry);
|
||||
|
|
@ -1015,6 +1440,22 @@ export async function getAccessToken(provider, credentials, log, proxyConfig: un
|
|||
* Only called from the per-connection mutex path (Layer 1 above).
|
||||
*/
|
||||
async function _getAccessTokenWithStalenessCheck(provider, credentials, log, proxyConfig) {
|
||||
// ROTATION MAP CHECK (codex-multi-auth pattern): if this refresh_token was
|
||||
// rotated very recently (within ROTATION_MAP_TTL_MS), reuse the cached new
|
||||
// tokens INSTEAD of hitting upstream. Auth0 treats re-use of a rotated token
|
||||
// as a security event and revokes the entire token family — fatal for
|
||||
// multi-account Codex setups. The in-memory rotation map catches this even
|
||||
// when the caller bypasses the DB staleness path (no connectionId, stale
|
||||
// in-memory credentials in retries, etc.).
|
||||
const rotated = lookupRotation(provider, credentials.refreshToken);
|
||||
if (rotated) {
|
||||
log?.info?.(
|
||||
"TOKEN_REFRESH",
|
||||
`Rotation map hit for ${provider}. Returning cached rotated tokens (avoids family-revoke).`
|
||||
);
|
||||
return rotated.result;
|
||||
}
|
||||
|
||||
// RACE CONDITION PREVENTION:
|
||||
// If the credentials object in memory is stale (e.g. it waited in a semaphore while another
|
||||
// request refreshed the token), using its OLD refreshToken will cause the provider (e.g. OpenAI)
|
||||
|
|
@ -1024,33 +1465,41 @@ async function _getAccessTokenWithStalenessCheck(provider, credentials, log, pro
|
|||
try {
|
||||
const { getProviderConnectionById } = await import("../../src/lib/db/providers");
|
||||
const dbConnection = await getProviderConnectionById(credentials.connectionId);
|
||||
if (
|
||||
dbConnection &&
|
||||
dbConnection.refreshToken &&
|
||||
dbConnection.refreshToken !== credentials.refreshToken
|
||||
) {
|
||||
log?.info?.(
|
||||
"TOKEN_REFRESH",
|
||||
`Stale token detected in memory for ${provider}. Using refreshed token from DB.`
|
||||
);
|
||||
|
||||
// If the DB token is not expired, we can just return it!
|
||||
if (dbConnection && dbConnection.refreshToken) {
|
||||
const now = Date.now();
|
||||
const dbExpiresAt = dbConnection.expiresAt ? new Date(dbConnection.expiresAt).getTime() : 0;
|
||||
|
||||
if (dbExpiresAt > now + 60000) {
|
||||
// 60 seconds buffer
|
||||
log?.info?.("TOKEN_REFRESH", `DB token is still valid. Skipping OAuth refresh.`);
|
||||
return {
|
||||
accessToken: dbConnection.accessToken,
|
||||
refreshToken: dbConnection.refreshToken,
|
||||
expiresIn: dbConnection.expiresIn,
|
||||
};
|
||||
} else {
|
||||
// DB token is also expired, but it's the NEWEST one. We must use it to refresh.
|
||||
credentials.refreshToken = dbConnection.refreshToken;
|
||||
credentials.accessToken = dbConnection.accessToken;
|
||||
if (dbConnection.refreshToken !== credentials.refreshToken) {
|
||||
log?.info?.(
|
||||
"TOKEN_REFRESH",
|
||||
`Stale token detected in memory for ${provider}. Using refreshed token from DB.`
|
||||
);
|
||||
|
||||
// If the DB token is not expired, we can just return it!
|
||||
if (dbExpiresAt > now + 60000) {
|
||||
// 60 seconds buffer
|
||||
log?.info?.("TOKEN_REFRESH", `DB token is still valid. Skipping OAuth refresh.`);
|
||||
return {
|
||||
accessToken: dbConnection.accessToken,
|
||||
refreshToken: dbConnection.refreshToken,
|
||||
// Return absolute expiresAt so downstream callers do NOT recompute lifetime
|
||||
// from a relative expiresIn value (which would incorrectly extend the TTL).
|
||||
// expiresIn intentionally omitted here.
|
||||
expiresAt: dbConnection.expiresAt,
|
||||
};
|
||||
} else {
|
||||
// DB token is also expired, but it's the NEWEST one. We must use it to refresh.
|
||||
credentials.refreshToken = dbConnection.refreshToken;
|
||||
credentials.accessToken = dbConnection.accessToken;
|
||||
}
|
||||
}
|
||||
// NOTE: Fix F (skip when DB == memory and DB > now+60s) was intentionally
|
||||
// removed. The caller (checkAndRefreshToken) already decided to refresh
|
||||
// because the token is within TOKEN_EXPIRY_BUFFER_MS of expiry. Re-checking
|
||||
// with a tighter 60-second window here would skip legitimate refreshes and
|
||||
// let near-expired tokens hit the upstream. Layer-1 mutex (per-connection)
|
||||
// and Layer-2 dedup (token-hash) already prevent concurrent refreshes for
|
||||
// the import-burst scenario.
|
||||
}
|
||||
} catch (e) {
|
||||
log?.warn?.(
|
||||
|
|
@ -1060,7 +1509,32 @@ async function _getAccessTokenWithStalenessCheck(provider, credentials, log, pro
|
|||
}
|
||||
}
|
||||
|
||||
return _getAccessTokenInternal(provider, credentials, log, proxyConfig);
|
||||
const oldRefreshToken = credentials.refreshToken;
|
||||
const result = await _getAccessTokenInternal(provider, credentials, log, proxyConfig);
|
||||
|
||||
// Record the rotation so subsequent stale callers can be redirected to the
|
||||
// new tokens without re-hitting upstream (which would trigger Auth0 family
|
||||
// revocation). Only records when the refresh actually rotated the token.
|
||||
if (
|
||||
result &&
|
||||
typeof result === "object" &&
|
||||
!("error" in result) &&
|
||||
(result as { accessToken?: string }).accessToken &&
|
||||
(result as { refreshToken?: string }).refreshToken
|
||||
) {
|
||||
recordRotation(
|
||||
provider,
|
||||
oldRefreshToken,
|
||||
result as {
|
||||
accessToken: string;
|
||||
refreshToken: string;
|
||||
expiresIn?: number;
|
||||
expiresAt?: string;
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -55,8 +55,15 @@ export async function POST(_request: Request, { params }: { params: Promise<{ id
|
|||
};
|
||||
|
||||
// Use the existing getAccessToken helper which knows how to refresh
|
||||
// tokens for each provider type (Claude, GitHub, Gemini, etc.)
|
||||
const newCredentials = (await getAccessToken(provider, credentials)) as RefreshResult | null;
|
||||
// tokens for each provider type (Claude, GitHub, Gemini, etc.).
|
||||
// Pass onPersist so the DB write happens atomically INSIDE the per-connection
|
||||
// mutex — prevents the race where a concurrent request reads stale credentials
|
||||
// between the network call and the DB update.
|
||||
let persistedCredentials: RefreshResult | null = null;
|
||||
const newCredentials = (await getAccessToken(provider, credentials, async (result) => {
|
||||
await updateProviderCredentials(id, result);
|
||||
persistedCredentials = result;
|
||||
})) as RefreshResult | null;
|
||||
|
||||
if (newCredentials && typeof newCredentials === "object" && newCredentials.error) {
|
||||
if (
|
||||
|
|
@ -82,12 +89,17 @@ export async function POST(_request: Request, { params }: { params: Promise<{ id
|
|||
);
|
||||
}
|
||||
|
||||
// Persist new credentials to DB
|
||||
await updateProviderCredentials(id, newCredentials);
|
||||
// If onPersist was not called (e.g. no connectionId in credentials path), persist now.
|
||||
if (!persistedCredentials) {
|
||||
await updateProviderCredentials(id, newCredentials);
|
||||
}
|
||||
|
||||
const expiresAt = newCredentials.expiresIn
|
||||
? new Date(Date.now() + newCredentials.expiresIn * 1000).toISOString()
|
||||
: null;
|
||||
const resolvedCreds = persistedCredentials || newCredentials;
|
||||
const expiresAt = resolvedCreds.expiresAt
|
||||
? resolvedCreds.expiresAt
|
||||
: resolvedCreds.expiresIn
|
||||
? new Date(Date.now() + resolvedCreds.expiresIn * 1000).toISOString()
|
||||
: null;
|
||||
|
||||
return NextResponse.json({
|
||||
success: true,
|
||||
|
|
|
|||
|
|
@ -303,13 +303,46 @@ async function refreshOAuthToken(connection: any) {
|
|||
if (!refreshToken) return null;
|
||||
|
||||
try {
|
||||
// Kiro needs extra fields the generic function expects
|
||||
// Fix B: Pass connectionId + accessToken + expiresAt so getAccessToken enters
|
||||
// the per-connection mutex (Layer 1) instead of falling through to the
|
||||
// token-hash fallback (Layer 2). Without connectionId, parallel dashboard
|
||||
// batch-tests would each acquire a separate Layer-2 lock keyed by token hash
|
||||
// and concurrently POST the same refresh_token to Codex/OpenAI, triggering
|
||||
// refresh_token_reused on rotating providers.
|
||||
const credentials = {
|
||||
connectionId: connection.id,
|
||||
accessToken: connection.accessToken,
|
||||
refreshToken,
|
||||
expiresAt: connection.expiresAt,
|
||||
providerSpecificData: connection.providerSpecificData || {},
|
||||
};
|
||||
|
||||
const result = await getAccessToken(provider, credentials, console);
|
||||
// Fix A: onPersist runs INSIDE the mutex inside getAccessToken so the DB
|
||||
// write happens before the lock releases. This prevents a concurrent caller
|
||||
// from reading the stale refresh_token between the network call and the DB
|
||||
// update.
|
||||
const result = await getAccessToken(provider, credentials, console, null, async (refreshed) => {
|
||||
if (!refreshed?.accessToken) return;
|
||||
const update: any = {
|
||||
accessToken: refreshed.accessToken,
|
||||
};
|
||||
if (refreshed.refreshToken) update.refreshToken = refreshed.refreshToken;
|
||||
if (refreshed.expiresAt) {
|
||||
update.expiresAt = refreshed.expiresAt;
|
||||
update.tokenExpiresAt = refreshed.expiresAt;
|
||||
} else if (refreshed.expiresIn) {
|
||||
const expiresAt = new Date(Date.now() + refreshed.expiresIn * 1000).toISOString();
|
||||
update.expiresAt = expiresAt;
|
||||
update.tokenExpiresAt = expiresAt;
|
||||
}
|
||||
if (refreshed.providerSpecificData) {
|
||||
update.providerSpecificData = {
|
||||
...(connection.providerSpecificData || {}),
|
||||
...refreshed.providerSpecificData,
|
||||
};
|
||||
}
|
||||
await updateProviderConnection(connection.id, update);
|
||||
});
|
||||
return result; // { accessToken, expiresIn, refreshToken } or null
|
||||
} catch (err) {
|
||||
console.log(`Error refreshing ${provider} token:`, (err as any).message);
|
||||
|
|
|
|||
|
|
@ -61,6 +61,16 @@ export const CODEX_CONFIG = {
|
|||
id_token_add_organizations: "true",
|
||||
codex_cli_simplified_flow: "true",
|
||||
originator: "codex_cli_rs",
|
||||
// prompt=login forces Auth0/OpenAI to RE-AUTHENTICATE the user instead of
|
||||
// silently reusing an existing browser session. This is THE KEY parameter
|
||||
// that enables multi-account OAuth on the same device + same client_id:
|
||||
// without it, OAuth flow #2 carries over session state from OAuth flow #1
|
||||
// and Auth0 invalidates the previous account's refresh_token family as a
|
||||
// "session takeover". With prompt=login, each OAuth flow creates an
|
||||
// isolated session that does not trample siblings.
|
||||
// Ported from ndycode/codex-multi-auth (auth.ts: forceNewLogin option) —
|
||||
// the only known tool that sustains multiple Codex OAuth accounts.
|
||||
prompt: "login",
|
||||
},
|
||||
};
|
||||
|
||||
|
|
@ -178,6 +188,9 @@ export const ANTIGRAVITY_CONFIG = {
|
|||
|
||||
// OpenAI OAuth Configuration (Authorization Code Flow with PKCE)
|
||||
// Re-uses CODEX_CONFIG.clientId to avoid duplication — same provider, different originator.
|
||||
// IMPORTANT: same Auth0 backend as Codex → same multi-account session-takeover
|
||||
// risk. `prompt: "login"` is mandatory to allow multiple OpenAI Native accounts
|
||||
// on the same device. See CODEX_CONFIG above for the full explanation.
|
||||
export const OPENAI_CONFIG = {
|
||||
clientId: CODEX_CONFIG.clientId,
|
||||
authorizeUrl: "https://auth.openai.com/oauth/authorize",
|
||||
|
|
@ -187,6 +200,7 @@ export const OPENAI_CONFIG = {
|
|||
extraParams: {
|
||||
id_token_add_organizations: "true",
|
||||
originator: "openai_native",
|
||||
prompt: "login",
|
||||
},
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -89,6 +89,14 @@ export const claude = {
|
|||
code_challenge: codeChallenge,
|
||||
code_challenge_method: config.codeChallengeMethod,
|
||||
state: state,
|
||||
// prompt=login forces Anthropic IdP to RE-AUTHENTICATE the user on each
|
||||
// OAuth flow instead of silently reusing the browser session. Prevents
|
||||
// the "session takeover" that invalidates the previous account's
|
||||
// refresh_token family on the same client_id. Same pattern that unlocked
|
||||
// multi-account Codex — applied defensively here because Anthropic's
|
||||
// OAuth uses the same client_id across all accounts AND the user reports
|
||||
// Claude tokens expiring in multi-account scenarios.
|
||||
prompt: "login",
|
||||
});
|
||||
return `${config.authorizeUrl}?${params.toString()}`;
|
||||
},
|
||||
|
|
|
|||
|
|
@ -223,15 +223,24 @@ async function resolveFreshCodexConnection(connectionId: string): Promise<CodexC
|
|||
);
|
||||
}
|
||||
|
||||
const refreshed = await getAccessToken("codex", {
|
||||
connectionId,
|
||||
accessToken: connection.accessToken,
|
||||
refreshToken,
|
||||
expiresAt: connection.expiresAt,
|
||||
expiresIn: connection.expiresIn,
|
||||
idToken: connection.idToken,
|
||||
providerSpecificData: connection.providerSpecificData,
|
||||
});
|
||||
// Pass onPersist so the DB write is atomic with the network call inside the mutex.
|
||||
let persistedRefresh: any = null;
|
||||
const refreshed = await getAccessToken(
|
||||
"codex",
|
||||
{
|
||||
connectionId,
|
||||
accessToken: connection.accessToken,
|
||||
refreshToken,
|
||||
expiresAt: connection.expiresAt,
|
||||
expiresIn: connection.expiresIn,
|
||||
idToken: connection.idToken,
|
||||
providerSpecificData: connection.providerSpecificData,
|
||||
},
|
||||
async (result) => {
|
||||
await updateProviderCredentials(connectionId, result);
|
||||
persistedRefresh = result;
|
||||
}
|
||||
);
|
||||
|
||||
if (isUnrecoverableRefreshError(refreshed)) {
|
||||
throw new CodexAuthFileError(
|
||||
|
|
@ -249,7 +258,16 @@ async function resolveFreshCodexConnection(connectionId: string): Promise<CodexC
|
|||
);
|
||||
}
|
||||
|
||||
await updateProviderCredentials(connectionId, refreshed);
|
||||
// If onPersist was not triggered (no accessToken path), fall back to explicit persist.
|
||||
if (!persistedRefresh) {
|
||||
await updateProviderCredentials(connectionId, refreshed);
|
||||
}
|
||||
|
||||
const effectiveExpiresAt = refreshed.expiresAt
|
||||
? refreshed.expiresAt
|
||||
: typeof refreshed.expiresIn === "number"
|
||||
? new Date(Date.now() + refreshed.expiresIn * 1000).toISOString()
|
||||
: connection.expiresAt || null;
|
||||
|
||||
return {
|
||||
...connection,
|
||||
|
|
@ -257,10 +275,7 @@ async function resolveFreshCodexConnection(connectionId: string): Promise<CodexC
|
|||
refreshToken: toNonEmptyString(refreshed.refreshToken) || refreshToken,
|
||||
expiresIn:
|
||||
typeof refreshed.expiresIn === "number" ? refreshed.expiresIn : connection.expiresIn || null,
|
||||
expiresAt:
|
||||
typeof refreshed.expiresIn === "number"
|
||||
? new Date(Date.now() + refreshed.expiresIn * 1000).toISOString()
|
||||
: connection.expiresAt || null,
|
||||
expiresAt: effectiveExpiresAt,
|
||||
providerSpecificData: refreshed.providerSpecificData
|
||||
? {
|
||||
...toRecord(connection.providerSpecificData),
|
||||
|
|
|
|||
|
|
@ -207,6 +207,25 @@ export async function createConnectionFromAuthFile(
|
|||
},
|
||||
});
|
||||
|
||||
// Fix C REVERTED: do NOT refresh-on-import.
|
||||
//
|
||||
// Reason: production tests showed that auth.json files exported from the
|
||||
// Codex CLI are often already partially rotated (the CLI continues to use
|
||||
// the tokens after export). Calling the OpenAI refresh endpoint with a
|
||||
// stale refresh_token returns "invalid_grant" / "refresh_token_reused" /
|
||||
// "refresh_token_invalidated" AND has been observed to invalidate the
|
||||
// entire token family upstream — causing every freshly imported account
|
||||
// to land in a permanently-broken state, with the only working connection
|
||||
// being whichever auth.json the user re-exported most recently.
|
||||
//
|
||||
// The id_token's `exp` claim is now used as the initial `expiresAt`. In
|
||||
// practice OpenAI Codex id_tokens carry a multi-day lifetime, so this
|
||||
// does NOT trigger the import-burst proactive-refresh storm that Fix C
|
||||
// originally tried to prevent. When the access_token actually expires,
|
||||
// the reactive 401 path in chatCore.ts (with the per-connection mutex)
|
||||
// handles a single, atomic rotation — that mechanism is intact via the
|
||||
// Fix A `runWithOnPersist` plumbing.
|
||||
|
||||
return { connection, created: true };
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import {
|
|||
getAccessToken,
|
||||
supportsTokenRefresh,
|
||||
isUnrecoverableRefreshError,
|
||||
refreshCopilotToken,
|
||||
} from "@omniroute/open-sse/services/tokenRefresh.ts";
|
||||
import { pickMaskedDisplayValue } from "@/shared/utils/maskEmail";
|
||||
|
||||
|
|
@ -303,7 +304,28 @@ export async function checkConnection(conn) {
|
|||
const tokenExpiresAt = getEffectiveTokenExpiryMs(conn);
|
||||
const hasKnownExpiry = tokenExpiresAt > 0;
|
||||
const isAboutToExpire = hasKnownExpiry && tokenExpiresAt - Date.now() < TOKEN_EXPIRY_BUFFER;
|
||||
const shouldRefreshByInterval = !hasKnownExpiry && Date.now() - lastCheck >= intervalMs;
|
||||
|
||||
// ROTATING_REFRESH_PROVIDERS — providers whose refresh_tokens are SINGLE-USE
|
||||
// (each refresh consumes the old one and returns a new one). For these, refreshing
|
||||
// on a fixed interval — instead of strictly on imminent expiry — burns rotations
|
||||
// unnecessarily AND can trigger Auth0's token family revocation (especially OpenAI
|
||||
// Codex). 9router did not have this background sweep; it was introduced in OmniRoute
|
||||
// and is the root cause of "adding account B invalidates account A" reports.
|
||||
// The interval path is kept ONLY for non-rotating providers where token state can
|
||||
// drift silently (e.g. cookie-based, opaque sessions without expires_at).
|
||||
const ROTATING_REFRESH_PROVIDERS = new Set([
|
||||
"codex",
|
||||
"openai",
|
||||
"kimi-coding",
|
||||
"cline",
|
||||
"kiro",
|
||||
"amazon-q",
|
||||
"gitlab-duo",
|
||||
"claude",
|
||||
]);
|
||||
const isRotatingProvider = ROTATING_REFRESH_PROVIDERS.has(conn.provider);
|
||||
const shouldRefreshByInterval =
|
||||
!hasKnownExpiry && !isRotatingProvider && Date.now() - lastCheck >= intervalMs;
|
||||
|
||||
if (!isAboutToExpire && !shouldRefreshByInterval) return;
|
||||
|
||||
|
|
@ -323,21 +345,62 @@ export async function checkConnection(conn) {
|
|||
const hideLogs = await shouldHideLogs();
|
||||
const proxyResolution = await resolveProxyForConnection(conn.id);
|
||||
const proxyConfig = extractResolvedProxyConfig(proxyResolution);
|
||||
|
||||
const healthCheckLog = {
|
||||
info: (tag: string, msg: string) => {
|
||||
if (!hideLogs) console.log(LOG_PREFIX, `[${tag}]`, msg);
|
||||
},
|
||||
warn: (tag: string, msg: string) => {
|
||||
if (!hideLogs) console.warn(LOG_PREFIX, `[${tag}]`, msg);
|
||||
},
|
||||
error: (tag: string, msg: string, extra?: Record<string, unknown>) => {
|
||||
if (!hideLogs) console.error(LOG_PREFIX, `[${tag}]`, msg, extra || "");
|
||||
},
|
||||
};
|
||||
|
||||
// Pass onPersist so the DB write is atomic with the network call inside the mutex.
|
||||
// This prevents a concurrent sweep or request from reading stale credentials
|
||||
// and re-using an already-consumed rotating refresh token (Codex/OpenAI).
|
||||
let persistedResult: any = null;
|
||||
const result = await getAccessToken(
|
||||
conn.provider,
|
||||
credentials,
|
||||
{
|
||||
info: (tag, msg) => {
|
||||
if (!hideLogs) console.log(`${LOG_PREFIX} [${tag}] ${msg}`);
|
||||
},
|
||||
warn: (tag, msg) => {
|
||||
if (!hideLogs) console.warn(`${LOG_PREFIX} [${tag}] ${msg}`);
|
||||
},
|
||||
error: (tag, msg, extra) => {
|
||||
if (!hideLogs) console.error(`${LOG_PREFIX} [${tag}] ${msg}`, extra || "");
|
||||
},
|
||||
},
|
||||
proxyConfig
|
||||
healthCheckLog,
|
||||
proxyConfig,
|
||||
async (refreshResult) => {
|
||||
const now = new Date().toISOString();
|
||||
const updateData: any = {
|
||||
accessToken: refreshResult.accessToken,
|
||||
lastHealthCheckAt: now,
|
||||
testStatus: "active",
|
||||
lastError: null,
|
||||
lastErrorAt: null,
|
||||
lastErrorType: null,
|
||||
lastErrorSource: null,
|
||||
errorCode: null,
|
||||
expiredRetryCount: null,
|
||||
expiredRetryAt: null,
|
||||
};
|
||||
if (refreshResult.refreshToken) {
|
||||
updateData.refreshToken = refreshResult.refreshToken;
|
||||
}
|
||||
if (refreshResult.expiresAt) {
|
||||
updateData.expiresAt = refreshResult.expiresAt;
|
||||
updateData.tokenExpiresAt = refreshResult.expiresAt;
|
||||
} else if (refreshResult.expiresIn) {
|
||||
const expiresAt = new Date(Date.now() + refreshResult.expiresIn * 1000).toISOString();
|
||||
updateData.expiresAt = expiresAt;
|
||||
updateData.tokenExpiresAt = expiresAt;
|
||||
}
|
||||
if (refreshResult.providerSpecificData) {
|
||||
updateData.providerSpecificData = {
|
||||
...(conn.providerSpecificData || {}),
|
||||
...refreshResult.providerSpecificData,
|
||||
};
|
||||
}
|
||||
await updateProviderConnection(conn.id, updateData);
|
||||
persistedResult = refreshResult;
|
||||
}
|
||||
);
|
||||
|
||||
const now = new Date().toISOString();
|
||||
|
|
@ -402,41 +465,112 @@ export async function checkConnection(conn) {
|
|||
}
|
||||
|
||||
if (result && result.accessToken) {
|
||||
const updateData: any = {
|
||||
accessToken: result.accessToken,
|
||||
lastHealthCheckAt: now,
|
||||
testStatus: "active",
|
||||
lastError: null,
|
||||
lastErrorAt: null,
|
||||
lastErrorType: null,
|
||||
lastErrorSource: null,
|
||||
errorCode: null,
|
||||
expiredRetryCount: null,
|
||||
expiredRetryAt: null,
|
||||
};
|
||||
|
||||
if (result.refreshToken) {
|
||||
updateData.refreshToken = result.refreshToken;
|
||||
}
|
||||
|
||||
if (result.expiresIn) {
|
||||
const expiresAt = new Date(Date.now() + result.expiresIn * 1000).toISOString();
|
||||
updateData.expiresAt = expiresAt;
|
||||
updateData.tokenExpiresAt = expiresAt;
|
||||
} else if (result.expiresAt) {
|
||||
updateData.expiresAt = result.expiresAt;
|
||||
updateData.tokenExpiresAt = result.expiresAt;
|
||||
}
|
||||
|
||||
if (result.providerSpecificData) {
|
||||
updateData.providerSpecificData = {
|
||||
...(conn.providerSpecificData || {}),
|
||||
...result.providerSpecificData,
|
||||
// onPersist already wrote the core token fields atomically inside the mutex.
|
||||
// Only write the lastHealthCheckAt timestamp (and any fields onPersist may have
|
||||
// missed) here, to avoid a redundant full update that would race against another
|
||||
// concurrent refresh that already wrote fresh credentials.
|
||||
if (persistedResult) {
|
||||
await updateProviderConnection(conn.id, { lastHealthCheckAt: now });
|
||||
} else {
|
||||
// No onPersist (e.g. no connectionId — token-hash dedup path). Write all fields.
|
||||
const updateData: any = {
|
||||
accessToken: result.accessToken,
|
||||
lastHealthCheckAt: now,
|
||||
testStatus: "active",
|
||||
lastError: null,
|
||||
lastErrorAt: null,
|
||||
lastErrorType: null,
|
||||
lastErrorSource: null,
|
||||
errorCode: null,
|
||||
expiredRetryCount: null,
|
||||
expiredRetryAt: null,
|
||||
};
|
||||
}
|
||||
|
||||
await updateProviderConnection(conn.id, updateData);
|
||||
if (result.refreshToken) {
|
||||
updateData.refreshToken = result.refreshToken;
|
||||
}
|
||||
|
||||
if (result.expiresAt) {
|
||||
updateData.expiresAt = result.expiresAt;
|
||||
updateData.tokenExpiresAt = result.expiresAt;
|
||||
} else if (result.expiresIn) {
|
||||
const expiresAt = new Date(Date.now() + result.expiresIn * 1000).toISOString();
|
||||
updateData.expiresAt = expiresAt;
|
||||
updateData.tokenExpiresAt = expiresAt;
|
||||
}
|
||||
|
||||
if (result.providerSpecificData) {
|
||||
updateData.providerSpecificData = {
|
||||
...(conn.providerSpecificData || {}),
|
||||
...result.providerSpecificData,
|
||||
};
|
||||
}
|
||||
|
||||
await updateProviderConnection(conn.id, updateData);
|
||||
}
|
||||
log(`${LOG_PREFIX} ✓ ${conn.provider}/${getConnectionLogLabel(conn)} refreshed`);
|
||||
|
||||
// ── GitHub Copilot sub-token refresh ──────────────────────────────────────
|
||||
// GitHub Copilot issues a short-lived (~30 min) API token separate from the
|
||||
// GitHub OAuth token. The health check must also refresh this sub-token before
|
||||
// it expires mid-session. The Copilot token expiry is stored in
|
||||
// providerSpecificData.copilotTokenExpiresAt (Unix seconds).
|
||||
if (conn.provider === "github") {
|
||||
// Re-read the latest connection after the OAuth refresh (onPersist may have updated it).
|
||||
const latestConn = (await getProviderConnectionById(conn.id).catch(() => null)) || conn;
|
||||
const accessTokenForCopilot = result.accessToken || latestConn.accessToken;
|
||||
|
||||
if (accessTokenForCopilot) {
|
||||
const copilotExpiresAtRaw =
|
||||
latestConn.providerSpecificData?.copilotTokenExpiresAt ??
|
||||
conn.providerSpecificData?.copilotTokenExpiresAt;
|
||||
const copilotExpiresAtMs =
|
||||
typeof copilotExpiresAtRaw === "number" && copilotExpiresAtRaw < 1e12
|
||||
? copilotExpiresAtRaw * 1000 // Unix seconds → ms
|
||||
: typeof copilotExpiresAtRaw === "string"
|
||||
? new Date(copilotExpiresAtRaw).getTime()
|
||||
: typeof copilotExpiresAtRaw === "number"
|
||||
? copilotExpiresAtRaw
|
||||
: 0;
|
||||
|
||||
const copilotAboutToExpire =
|
||||
!copilotExpiresAtMs || copilotExpiresAtMs - Date.now() < 5 * 60 * 1000;
|
||||
|
||||
if (copilotAboutToExpire) {
|
||||
log(
|
||||
`${LOG_PREFIX} Refreshing GitHub Copilot sub-token for ${getConnectionLogLabel(conn)}`
|
||||
);
|
||||
try {
|
||||
const copilotResult = await refreshCopilotToken(
|
||||
accessTokenForCopilot,
|
||||
healthCheckLog,
|
||||
proxyConfig
|
||||
);
|
||||
if (copilotResult?.token) {
|
||||
await updateProviderConnection(conn.id, {
|
||||
providerSpecificData: {
|
||||
...(latestConn.providerSpecificData || {}),
|
||||
copilotToken: copilotResult.token,
|
||||
copilotTokenExpiresAt: copilotResult.expiresAt,
|
||||
},
|
||||
});
|
||||
log(
|
||||
`${LOG_PREFIX} ✓ GitHub Copilot sub-token refreshed for ${getConnectionLogLabel(conn)}`
|
||||
);
|
||||
} else {
|
||||
logWarn(
|
||||
`${LOG_PREFIX} ✗ GitHub Copilot sub-token refresh failed for ${getConnectionLogLabel(conn)}`
|
||||
);
|
||||
}
|
||||
} catch (copilotErr) {
|
||||
logError(
|
||||
`${LOG_PREFIX} Error refreshing Copilot sub-token:`,
|
||||
copilotErr?.message || copilotErr
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const updateData = buildRefreshFailureUpdate(conn, now);
|
||||
await updateProviderConnection(conn.id, updateData);
|
||||
|
|
|
|||
|
|
@ -200,7 +200,10 @@ export async function handleChat(request: any, clientRawRequest: any = null) {
|
|||
|
||||
// Log request endpoint and model
|
||||
const url = new URL(request.url);
|
||||
const modelStr = body.model;
|
||||
// `let` because the middleware-hook pipeline (line ~319) may reassign this
|
||||
// when a hook rewrites the target model. Previously declared `const`, which
|
||||
// broke turbopack/strict-mode builds (PR #2670 regression).
|
||||
let modelStr = body.model;
|
||||
|
||||
// Count messages (support both messages[] and input[] formats)
|
||||
const msgCount = body.messages?.length || body.input?.length || 0;
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import * as log from "../utils/logger";
|
|||
import { updateProviderConnection, resolveProxyForProvider } from "@/lib/localDb";
|
||||
import {
|
||||
TOKEN_EXPIRY_BUFFER_MS as BUFFER_MS,
|
||||
getRefreshLeadMs as _getRefreshLeadMs,
|
||||
refreshAccessToken as _refreshAccessToken,
|
||||
refreshClaudeOAuthToken as _refreshClaudeOAuthToken,
|
||||
refreshGoogleToken as _refreshGoogleToken,
|
||||
|
|
@ -17,31 +18,13 @@ import {
|
|||
getAllAccessTokens as _getAllAccessTokens,
|
||||
} from "@omniroute/open-sse/services/tokenRefresh.ts";
|
||||
|
||||
// Per-connection mutex: prevents concurrent OAuth refresh for rotating tokens.
|
||||
// Key = connectionId, Value = { promise: in-flight refresh, waiters: count of callers sharing it }
|
||||
const connectionRefreshMutex = new Map<string, { promise: Promise<any>; waiters: number }>();
|
||||
|
||||
export async function withConnectionRefreshMutex<T>(
|
||||
connectionId: string,
|
||||
fn: () => Promise<T>
|
||||
): Promise<T> {
|
||||
const existing = connectionRefreshMutex.get(connectionId);
|
||||
if (existing) {
|
||||
existing.waiters++;
|
||||
log.info("TOKEN_REFRESH", "Concurrent refresh detected — sharing in-flight refresh", {
|
||||
connectionId,
|
||||
waiters: existing.waiters,
|
||||
});
|
||||
return existing.promise as Promise<T>;
|
||||
}
|
||||
|
||||
const entry: { promise: Promise<T>; waiters: number } = { promise: null as any, waiters: 0 };
|
||||
entry.promise = fn().finally(() => {
|
||||
connectionRefreshMutex.delete(connectionId);
|
||||
});
|
||||
connectionRefreshMutex.set(connectionId, entry);
|
||||
return entry.promise;
|
||||
}
|
||||
// DEPRECATED: withConnectionRefreshMutex was removed. The per-connection mutex
|
||||
// is now consolidated in open-sse/services/tokenRefresh.ts and protected by
|
||||
// passing an `onPersist` callback to `getAccessToken`, which runs the DB write
|
||||
// INSIDE the mutex closure (atomic [network + persist]). The old src/sse-side
|
||||
// mutex Map was redundant and created the illusion of two locks when there was
|
||||
// actually only one. Removing it eliminates the dual-Map confusion. See
|
||||
// docs/architecture/SSE_BOUNDARY.md and tests/unit/token-refresh-race-comprehensive.test.ts.
|
||||
|
||||
export const TOKEN_EXPIRY_BUFFER_MS = BUFFER_MS;
|
||||
|
||||
|
|
@ -94,9 +77,13 @@ export const refreshCopilotToken = async (githubAccessToken: string) => {
|
|||
return _refreshCopilotToken(githubAccessToken, log, proxy);
|
||||
};
|
||||
|
||||
export const getAccessToken = async (provider: string, credentials: any) => {
|
||||
export const getAccessToken = async (
|
||||
provider: string,
|
||||
credentials: any,
|
||||
onPersist?: (result: any) => Promise<void>
|
||||
) => {
|
||||
const proxy = await resolveProxyForProvider(provider);
|
||||
return _getAccessToken(provider, credentials, log, proxy);
|
||||
return _getAccessToken(provider, credentials, log, proxy, onPersist);
|
||||
};
|
||||
|
||||
export const refreshTokenByProvider = async (provider: string, credentials: any) => {
|
||||
|
|
@ -164,32 +151,39 @@ export async function updateProviderCredentials(connectionId: string, newCredent
|
|||
export async function checkAndRefreshToken(provider: string, credentials: any) {
|
||||
let updatedCredentials = { ...credentials };
|
||||
|
||||
// Check regular token expiry
|
||||
// Check regular token expiry. Use the provider-specific lead time so rotating-
|
||||
// token providers (Codex/OpenAI) refresh FAR ahead of access_token expiry. This
|
||||
// keeps the refresh_token "warm" — refreshed regularly enough that Auth0 doesn't
|
||||
// mark it as stale and revoke the token family on first use after long idle.
|
||||
if (updatedCredentials.expiresAt) {
|
||||
const expiresAt = new Date(updatedCredentials.expiresAt).getTime();
|
||||
const now = Date.now();
|
||||
const refreshLead = _getRefreshLeadMs(provider);
|
||||
|
||||
if (expiresAt - now < TOKEN_EXPIRY_BUFFER_MS) {
|
||||
if (expiresAt - now < refreshLead) {
|
||||
log.info("TOKEN_REFRESH", "Token expiring soon, refreshing proactively", {
|
||||
provider,
|
||||
expiresIn: Math.round((expiresAt - now) / 1000),
|
||||
refreshLeadMs: refreshLead,
|
||||
});
|
||||
|
||||
const connectionId: string | undefined = updatedCredentials.connectionId;
|
||||
const newCredentials = connectionId
|
||||
? await withConnectionRefreshMutex(connectionId, async () => {
|
||||
const result = await getAccessToken(provider, updatedCredentials);
|
||||
if (result?.accessToken) {
|
||||
// Persist BEFORE the mutex releases so a concurrent request
|
||||
// cannot read stale DB credentials and re-use a rotated refresh token.
|
||||
await updateProviderCredentials(connectionId, result);
|
||||
}
|
||||
return result;
|
||||
})
|
||||
: await getAccessToken(provider, updatedCredentials);
|
||||
|
||||
// Pass onPersist so the DB write happens INSIDE the open-sse per-connection
|
||||
// mutex, making [network call + DB write] one atomic step. This eliminates the
|
||||
// race where a concurrent request reads stale DB credentials before the write
|
||||
// and re-uses a rotated refresh token (refresh_token_reused on Codex/OpenAI).
|
||||
// The separate withConnectionRefreshMutex wrapper is no longer needed here.
|
||||
const persistCallback = connectionId
|
||||
? async (result: any) => {
|
||||
await updateProviderCredentials(connectionId, result);
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const newCredentials = await getAccessToken(provider, updatedCredentials, persistCallback);
|
||||
|
||||
if (newCredentials && newCredentials.accessToken) {
|
||||
// DB already updated inside the mutex when connectionId is present.
|
||||
// For the no-connectionId path, persist here as before.
|
||||
// For the no-connectionId path (no mutex, no onPersist), persist here as before.
|
||||
if (!connectionId) {
|
||||
await updateProviderCredentials(updatedCredentials.connectionId, newCredentials);
|
||||
}
|
||||
|
|
@ -198,9 +192,11 @@ export async function checkAndRefreshToken(provider: string, credentials: any) {
|
|||
...updatedCredentials,
|
||||
accessToken: newCredentials.accessToken,
|
||||
refreshToken: newCredentials.refreshToken || updatedCredentials.refreshToken,
|
||||
expiresAt: newCredentials.expiresIn
|
||||
? new Date(Date.now() + newCredentials.expiresIn * 1000).toISOString()
|
||||
: updatedCredentials.expiresAt,
|
||||
expiresAt: newCredentials.expiresAt
|
||||
? newCredentials.expiresAt
|
||||
: newCredentials.expiresIn
|
||||
? new Date(Date.now() + newCredentials.expiresIn * 1000).toISOString()
|
||||
: updatedCredentials.expiresAt,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
200
tests/unit/oauth-providers-error-handling.test.ts
Normal file
200
tests/unit/oauth-providers-error-handling.test.ts
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
/**
|
||||
* Structural regression tests for OAuth provider error handling.
|
||||
*
|
||||
* These are text-based assertions on source files (no network calls).
|
||||
* They verify that each refresh function:
|
||||
* - Exists with the expected signature
|
||||
* - Handles the provider-specific unrecoverable error codes
|
||||
* - Returns the normalized { error: "unrecoverable_refresh_error", code } sentinel
|
||||
*/
|
||||
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { readFile } from "fs/promises";
|
||||
import path from "path";
|
||||
|
||||
const root = path.resolve(import.meta.dirname, "../..");
|
||||
const read = (rel: string) => readFile(path.join(root, rel), "utf8");
|
||||
|
||||
// ─── P0: GitLab Duo ───────────────────────────────────────────────────────────
|
||||
|
||||
test("P0: gitlab-duo is registered in providerRegistry", async () => {
|
||||
const src = await read("open-sse/config/providerRegistry.ts");
|
||||
assert.match(src, /["']gitlab-duo["']\s*:/, "gitlab-duo must be a key in REGISTRY");
|
||||
});
|
||||
|
||||
test("P0: refreshGitLabDuoToken exists and handles invalid_grant as unrecoverable", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
assert.match(
|
||||
src,
|
||||
/export\s+async\s+function\s+refreshGitLabDuoToken\(/,
|
||||
"refreshGitLabDuoToken must be exported"
|
||||
);
|
||||
|
||||
// Extract the function body
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshGitLabDuoToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshGitLabDuoToken function body not found");
|
||||
assert.match(fnMatch[0], /invalid_grant/, "must detect invalid_grant");
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
test("P0: gitlab-duo case exists in _getAccessTokenInternal", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
assert.match(src, /case\s+["']gitlab-duo["']/, "gitlab-duo case must exist in switch");
|
||||
});
|
||||
|
||||
test("P0: gitlab-duo is in supportsTokenRefresh explicit set", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// Find the explicitlySupported Set
|
||||
const setMatch = src.match(/const\s+explicitlySupported\s*=\s*new\s+Set\(\[[\s\S]+?\]\)/);
|
||||
assert.ok(setMatch, "explicitlySupported Set not found");
|
||||
assert.match(setMatch[0], /["']gitlab-duo["']/, "gitlab-duo must be in explicitlySupported");
|
||||
});
|
||||
|
||||
// ─── P1: Kimi Coding stable device_id ────────────────────────────────────────
|
||||
|
||||
test("P1: refreshKimiCodingToken accepts providerSpecificData parameter", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
assert.match(
|
||||
src,
|
||||
/export\s+async\s+function\s+refreshKimiCodingToken\([^)]*providerSpecificData/,
|
||||
"refreshKimiCodingToken must accept providerSpecificData"
|
||||
);
|
||||
});
|
||||
|
||||
test("P1: refreshKimiCodingToken does NOT use ephemeral Date.now() device ID", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// Extract function body — match from declaration to next top-level export function
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshKimiCodingToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshKimiCodingToken function body not found");
|
||||
assert.doesNotMatch(
|
||||
fnMatch[0],
|
||||
/["']kimi-refresh-["']\s*\+\s*Date\.now\(\)/,
|
||||
"must NOT use ephemeral kimi-refresh-+Date.now() device ID"
|
||||
);
|
||||
});
|
||||
|
||||
test("P1: refreshKimiCodingToken handles invalid_grant as unrecoverable", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshKimiCodingToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshKimiCodingToken function body not found");
|
||||
assert.match(fnMatch[0], /invalid_grant/, "must detect invalid_grant");
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
test("P1: _getAccessTokenInternal passes providerSpecificData to refreshKimiCodingToken", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// The case for kimi-coding should pass credentials.providerSpecificData
|
||||
assert.match(
|
||||
src,
|
||||
/case\s+["']kimi-coding["']:[\s\S]{1,300}providerSpecificData/,
|
||||
"kimi-coding case must pass providerSpecificData"
|
||||
);
|
||||
});
|
||||
|
||||
// ─── P1: GitHub Copilot sub-token health check ────────────────────────────────
|
||||
|
||||
test("P1: GitHub Copilot sub-token is refreshed by tokenHealthCheck", async () => {
|
||||
const src = await read("src/lib/tokenHealthCheck.ts");
|
||||
assert.match(src, /copilot|Copilot/i, "tokenHealthCheck must reference Copilot");
|
||||
// Must import refreshCopilotToken
|
||||
assert.match(src, /refreshCopilotToken/, "must import and call refreshCopilotToken");
|
||||
});
|
||||
|
||||
test("P1: tokenHealthCheck checks copilotTokenExpiresAt before refreshing", async () => {
|
||||
const src = await read("src/lib/tokenHealthCheck.ts");
|
||||
assert.match(src, /copilotTokenExpiresAt/, "must check copilotTokenExpiresAt");
|
||||
assert.match(src, /conn\.provider\s*===\s*["']github["']/, "must be gated on github provider");
|
||||
});
|
||||
|
||||
// ─── P2: Google invalid_grant ─────────────────────────────────────────────────
|
||||
|
||||
test("P2: refreshGoogleToken parses invalid_grant as unrecoverable", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshGoogleToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshGoogleToken function body not found");
|
||||
assert.match(fnMatch[0], /invalid_grant/, "must detect invalid_grant");
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
// ─── P2: Qwen invalid_grant ───────────────────────────────────────────────────
|
||||
|
||||
test("P2: refreshQwenToken handles invalid_grant in addition to invalid_request", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshQwenToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshQwenToken function body not found");
|
||||
assert.match(fnMatch[0], /invalid_grant/, "must detect invalid_grant");
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
// ─── P2: Kiro AWS InvalidGrantException ──────────────────────────────────────
|
||||
|
||||
test("P2: refreshKiroToken parses AWS InvalidGrantException", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshKiroToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshKiroToken function body not found");
|
||||
assert.match(
|
||||
fnMatch[0],
|
||||
/InvalidGrantException|ExpiredTokenException/,
|
||||
"must detect AWS error types"
|
||||
);
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
test("P2: refreshKiroToken handles AWS errors on both AWS OIDC and social auth paths", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshKiroToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshKiroToken function body not found");
|
||||
// Count occurrences of InvalidGrantException — should appear in both paths
|
||||
const matchCount = (fnMatch[0].match(/InvalidGrantException/g) || []).length;
|
||||
assert.ok(
|
||||
matchCount >= 2,
|
||||
`InvalidGrantException should be checked in both paths (found ${matchCount} occurrences)`
|
||||
);
|
||||
});
|
||||
|
||||
// ─── P3: Claude error shape normalization ─────────────────────────────────────
|
||||
|
||||
test("P3: refreshClaudeOAuthToken normalizes invalid_grant to unrecoverable_refresh_error sentinel", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshClaudeOAuthToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshClaudeOAuthToken function body not found");
|
||||
assert.match(
|
||||
fnMatch[0],
|
||||
/unrecoverable_refresh_error[\s\S]{1,100}invalid_grant|invalid_grant[\s\S]{1,100}unrecoverable_refresh_error/,
|
||||
"invalid_grant must map to unrecoverable_refresh_error sentinel"
|
||||
);
|
||||
// Must NOT return the old non-normalized shape { error: errorBody.error, code: "http_..." }
|
||||
assert.doesNotMatch(
|
||||
fnMatch[0],
|
||||
/code:\s*`http_\$\{response\.status\}`/,
|
||||
"must NOT return http_NNN code format for invalid_grant"
|
||||
);
|
||||
});
|
||||
|
||||
// ─── P3: Windsurf Firebase errors ────────────────────────────────────────────
|
||||
|
||||
test("P3: refreshWindsurfToken parses Firebase USER_DISABLED/TOKEN_EXPIRED errors", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+async\s+function\s+refreshWindsurfToken\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "refreshWindsurfToken function body not found");
|
||||
assert.match(
|
||||
fnMatch[0],
|
||||
/USER_DISABLED|TOKEN_EXPIRED|INVALID_REFRESH_TOKEN/,
|
||||
"must detect Firebase error codes"
|
||||
);
|
||||
assert.match(fnMatch[0], /unrecoverable_refresh_error/, "must return unrecoverable sentinel");
|
||||
});
|
||||
|
||||
// ─── isUnrecoverableRefreshError consistency ──────────────────────────────────
|
||||
|
||||
test("isUnrecoverableRefreshError detects the normalized sentinel shape", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const fnMatch = src.match(/export\s+function\s+isUnrecoverableRefreshError\([\s\S]+?\n\}/);
|
||||
assert.ok(fnMatch, "isUnrecoverableRefreshError function body not found");
|
||||
assert.match(
|
||||
fnMatch[0],
|
||||
/unrecoverable_refresh_error/,
|
||||
"must detect unrecoverable_refresh_error"
|
||||
);
|
||||
});
|
||||
171
tests/unit/token-refresh-race-comprehensive.test.ts
Normal file
171
tests/unit/token-refresh-race-comprehensive.test.ts
Normal file
|
|
@ -0,0 +1,171 @@
|
|||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { readFile } from "fs/promises";
|
||||
import path from "path";
|
||||
|
||||
const root = path.resolve(import.meta.dirname, "../..");
|
||||
|
||||
const read = (rel: string) => readFile(path.join(root, rel), "utf8");
|
||||
|
||||
test("Fix A: getAccessToken accepts an onPersist parameter", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// Accepts either the legacy `(result: any) => Promise<void>` or the
|
||||
// refactored named-type form (RefreshPersistFn) — the latter is preferred
|
||||
// because it satisfies the t11 any-budget.
|
||||
assert.match(
|
||||
src,
|
||||
/export async function getAccessToken\([\s\S]{0,500}onPersist\?:\s*(?:RefreshPersistFn|\(result:\s*\w+\)\s*=>\s*Promise<void>)/,
|
||||
"getAccessToken must declare onPersist as the 5th parameter"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix A: getAccessToken invokes onPersist INSIDE the per-connection mutex closure", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
const closureMatch = src.match(/entry\.promise\s*=\s*\(async\s*\(\)\s*=>\s*\{([\s\S]+?)\}\)\(\)/);
|
||||
assert.ok(closureMatch, "Per-connection mutex closure must use the (async () => {...})() form");
|
||||
const closureBody = closureMatch![1];
|
||||
assert.match(
|
||||
closureBody,
|
||||
/effectiveOnPersist/,
|
||||
"The mutex closure body must reference effectiveOnPersist so the persist runs before the mutex releases"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix A: runWithOnPersist + getActiveOnPersist are exported for executor plumbing", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
assert.match(src, /export function runWithOnPersist\b/);
|
||||
assert.match(src, /export function getActiveOnPersist\b/);
|
||||
assert.match(src, /AsyncLocalStorage/);
|
||||
});
|
||||
|
||||
test("Fix A: chatCore.ts wraps executor.refreshCredentials with runWithOnPersist", async () => {
|
||||
const src = await read("open-sse/handlers/chatCore.ts");
|
||||
assert.match(
|
||||
src,
|
||||
/runWithOnPersist\([\s\S]{0,200}executor\.refreshCredentials/,
|
||||
"chatCore.ts reactive 401 path must wrap refreshCredentials with runWithOnPersist"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix A: base.ts proactive refresh also wraps refreshCredentials with runWithOnPersist", async () => {
|
||||
const src = await read("open-sse/executors/base.ts");
|
||||
assert.match(
|
||||
src,
|
||||
/runWithOnPersist\([\s\S]{0,200}this\.refreshCredentials/,
|
||||
"base.ts proactive needsRefresh branch must wrap refreshCredentials with runWithOnPersist"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix B: refreshOAuthToken in test/route.ts includes connectionId in credentials", async () => {
|
||||
const src = await read("src/app/api/providers/[id]/test/route.ts");
|
||||
const fnIdx = src.indexOf("async function refreshOAuthToken(");
|
||||
assert.ok(fnIdx >= 0, "refreshOAuthToken function must exist");
|
||||
const fnSlice = src.slice(fnIdx, fnIdx + 2500);
|
||||
assert.match(
|
||||
fnSlice,
|
||||
/connectionId:\s*connection\.id/,
|
||||
"refreshOAuthToken credentials must include connectionId so Layer 1 (per-connection mutex) is used instead of Layer 2 (token-hash dedup)"
|
||||
);
|
||||
assert.match(
|
||||
fnSlice,
|
||||
/onPersist|async\s*\(refreshed\)\s*=>/,
|
||||
"refreshOAuthToken must pass an onPersist callback so the DB write is atomic with the network refresh"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix C reverted: codexAuthImport does NOT refresh tokens on import (avoids family revocation on stale auth.json)", async () => {
|
||||
const src = await read("src/lib/oauth/utils/codexAuthImport.ts");
|
||||
assert.doesNotMatch(
|
||||
src,
|
||||
/refreshConnectionTokensOnImport\(/,
|
||||
"Fix C was reverted because auth.json files exported from Codex CLI are often partially rotated; refreshing with a stale refresh_token caused upstream to invalidate the entire token family"
|
||||
);
|
||||
assert.doesNotMatch(
|
||||
src,
|
||||
/import\s*\{[^}]*getAccessToken[^}]*\}\s*from\s*"@omniroute\/open-sse\/services\/tokenRefresh/,
|
||||
"codexAuthImport should not import getAccessToken (refresh-on-import was reverted)"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix D: staleness fallback returns absolute expiresAt, not raw expiresIn", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// Locate the actual return statement, not the JSDoc mention.
|
||||
// The anchor is the log line "DB token is still valid. Skipping OAuth refresh."
|
||||
// which only appears in the code path (the JSDoc says "DB token is still valid it is").
|
||||
const idx = src.indexOf("DB token is still valid. Skipping");
|
||||
assert.ok(idx >= 0, "Staleness-valid log line not found");
|
||||
const slice = src.slice(idx, idx + 800);
|
||||
assert.match(
|
||||
slice,
|
||||
/expiresAt:\s*dbConnection\.expiresAt/,
|
||||
"Staleness fallback must return absolute expiresAt"
|
||||
);
|
||||
assert.doesNotMatch(
|
||||
slice,
|
||||
/expiresIn:\s*dbConnection\.expiresIn(?!\s*\/\/)/,
|
||||
"Staleness fallback must NOT return raw expiresIn (causes downstream lifetime extension)"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix D: src/sse wrapper prefers expiresAt over expiresIn when recomputing", async () => {
|
||||
const src = await read("src/sse/services/tokenRefresh.ts");
|
||||
// The recompute block in checkAndRefreshToken
|
||||
assert.match(
|
||||
src,
|
||||
/expiresAt:\s*newCredentials\.expiresAt\s*\?\s*newCredentials\.expiresAt/,
|
||||
"checkAndRefreshToken must prefer the absolute expiresAt before falling back to expiresIn arithmetic"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix E: chatCore.ts moves credentials mutation INSIDE the mutex closure via persistFn", async () => {
|
||||
const src = await read("open-sse/handlers/chatCore.ts");
|
||||
// The persistFn closure must do BOTH the Object.assign AND the user callback
|
||||
const persistFnMatch = src.match(
|
||||
/const\s+persistFn\s*=[\s\S]{0,800}Object\.assign\(credentials,[\s\S]{0,300}onCredentialsRefreshed/
|
||||
);
|
||||
assert.ok(
|
||||
persistFnMatch,
|
||||
"chatCore.ts must build persistFn that both Object.assigns credentials and calls onCredentialsRefreshed, so both happen INSIDE the mutex"
|
||||
);
|
||||
});
|
||||
|
||||
test("Fix F removed: no over-eager skip that would bypass legitimate refreshes", async () => {
|
||||
const src = await read("open-sse/services/tokenRefresh.ts");
|
||||
// Fix F was intentionally removed because it skipped refreshes that the
|
||||
// caller (checkAndRefreshToken) had already decided were needed. Verify the
|
||||
// dead-code branch is gone (no `credentials.accessToken === dbConnection.accessToken`
|
||||
// skip path remains).
|
||||
assert.doesNotMatch(
|
||||
src,
|
||||
/credentials\.accessToken\s*===\s*dbConnection\.accessToken[\s\S]{0,300}Skipping OAuth refresh/,
|
||||
"Fix F was removed because it caused legitimate refreshes to be skipped"
|
||||
);
|
||||
});
|
||||
|
||||
test("Mutex consolidation: src/sse wrapper no longer holds its own connectionRefreshMutex map", async () => {
|
||||
const src = await read("src/sse/services/tokenRefresh.ts");
|
||||
// It's OK to keep a withConnectionRefreshMutex shim for back-compat, but
|
||||
// the actual Map should not be re-instantiated here (use the open-sse one).
|
||||
const hasOwnMap = /const\s+connectionRefreshMutex\s*=\s*new\s+Map/.test(src);
|
||||
if (hasOwnMap) {
|
||||
// If a Map remains, the file must clearly document that it is a deprecated
|
||||
// shim — otherwise it's a regression that recreates the dual-mutex bug.
|
||||
assert.match(
|
||||
src,
|
||||
/deprecated|legacy|shim|redundant|kept for back-compat/i,
|
||||
"If src/sse keeps its own connectionRefreshMutex Map, the file must document it as deprecated/legacy"
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
test("Imports: chatCore.ts imports runWithOnPersist from open-sse tokenRefresh", async () => {
|
||||
const src = await read("open-sse/handlers/chatCore.ts");
|
||||
assert.match(src, /runWithOnPersist/);
|
||||
assert.match(src, /from\s+"\.\.\/services\/tokenRefresh\.ts"/);
|
||||
});
|
||||
|
||||
test("Imports: base.ts imports runWithOnPersist from open-sse tokenRefresh", async () => {
|
||||
const src = await read("open-sse/executors/base.ts");
|
||||
assert.match(src, /runWithOnPersist/);
|
||||
assert.match(src, /from\s+"\.\.\/services\/tokenRefresh\.ts"/);
|
||||
});
|
||||
|
|
@ -1,134 +0,0 @@
|
|||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
|
||||
/**
|
||||
* Structural test: locks the invariant that updateProviderCredentials is
|
||||
* called INSIDE the withConnectionRefreshMutex closure in checkAndRefreshToken.
|
||||
*
|
||||
* Background: with rotating refresh tokens (OpenAI/Codex), if the DB write
|
||||
* happens AFTER the mutex releases, a concurrent request can read stale
|
||||
* credentials and send the old refresh token. OpenAI detects this as
|
||||
* refresh_token_reused and permanently revokes the entire token family.
|
||||
*
|
||||
* This test parses the source text of tokenRefresh.ts and verifies that
|
||||
* the mutex closure body contains the DB write call. It is intentionally
|
||||
* a source-level check so it survives refactors that would re-introduce
|
||||
* the race (e.g., moving the write back outside).
|
||||
*/
|
||||
test("tokenRefresh.ts wraps updateProviderCredentials inside withConnectionRefreshMutex", async () => {
|
||||
const srcPath = path.resolve("src/sse/services/tokenRefresh.ts");
|
||||
const source = await readFile(srcPath, "utf8");
|
||||
|
||||
// Locate the withConnectionRefreshMutex invocation in checkAndRefreshToken.
|
||||
// Match: withConnectionRefreshMutex(connectionId, async () => { ... })
|
||||
// The regex captures everything between the opening { and the matching }
|
||||
// of the async arrow function passed as the second argument.
|
||||
//
|
||||
// We look for the async closure form specifically — if someone reverts to
|
||||
// the non-async one-liner, this assertion will fire and force a review.
|
||||
const mutexCallIndex = source.indexOf("withConnectionRefreshMutex(connectionId, async () => {");
|
||||
assert.ok(
|
||||
mutexCallIndex !== -1,
|
||||
"withConnectionRefreshMutex must be called with an async arrow function closure (not a one-liner). " +
|
||||
"This ensures the DB write can be awaited inside before the mutex releases."
|
||||
);
|
||||
|
||||
// Extract everything from the opening of the async closure to the
|
||||
// corresponding closing brace + ) that ends the mutex call.
|
||||
const closureStart = source.indexOf("{", mutexCallIndex);
|
||||
assert.ok(closureStart !== -1, "Could not locate opening brace of mutex closure");
|
||||
|
||||
// Walk forward to find the matching closing brace.
|
||||
let depth = 0;
|
||||
let closureEnd = -1;
|
||||
for (let i = closureStart; i < source.length; i++) {
|
||||
if (source[i] === "{") depth++;
|
||||
else if (source[i] === "}") {
|
||||
depth--;
|
||||
if (depth === 0) {
|
||||
closureEnd = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
assert.ok(closureEnd !== -1, "Could not locate closing brace of mutex closure");
|
||||
|
||||
const closureBody = source.slice(closureStart, closureEnd + 1);
|
||||
|
||||
assert.match(
|
||||
closureBody,
|
||||
/updateProviderCredentials/,
|
||||
"updateProviderCredentials DB write MUST be inside the withConnectionRefreshMutex " +
|
||||
"closure body. Moving it outside creates a race window where a concurrent request " +
|
||||
"reads stale credentials and re-uses a rotated refresh token, triggering " +
|
||||
"refresh_token_reused (permanent token family revocation on OpenAI/Codex)."
|
||||
);
|
||||
});
|
||||
|
||||
/**
|
||||
* Complementary structural check: the unconditional updateProviderCredentials
|
||||
* call outside the mutex must be guarded by !connectionId (no-connectionId
|
||||
* fallback path only). This prevents a double-write that would bypass the
|
||||
* serialization guarantee.
|
||||
*/
|
||||
test("tokenRefresh.ts outer updateProviderCredentials call is guarded by !connectionId", async () => {
|
||||
const srcPath = path.resolve("src/sse/services/tokenRefresh.ts");
|
||||
const source = await readFile(srcPath, "utf8");
|
||||
|
||||
// Find the checkAndRefreshToken function body.
|
||||
const fnStart = source.indexOf("export async function checkAndRefreshToken(");
|
||||
assert.ok(fnStart !== -1, "checkAndRefreshToken function not found in source");
|
||||
|
||||
// Walk to find the full function body.
|
||||
const bodyStart = source.indexOf("{", fnStart);
|
||||
let depth = 0;
|
||||
let bodyEnd = -1;
|
||||
for (let i = bodyStart; i < source.length; i++) {
|
||||
if (source[i] === "{") depth++;
|
||||
else if (source[i] === "}") {
|
||||
depth--;
|
||||
if (depth === 0) {
|
||||
bodyEnd = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
assert.ok(bodyEnd !== -1, "Could not locate closing brace of checkAndRefreshToken");
|
||||
|
||||
const fnBody = source.slice(fnStart, bodyEnd + 1);
|
||||
|
||||
// Count occurrences of updateProviderCredentials in the function body.
|
||||
const allOccurrences = [...fnBody.matchAll(/updateProviderCredentials/g)];
|
||||
|
||||
// There must be at least 2: one inside the mutex closure, one in the
|
||||
// copilot block (lines ~212-218), and optionally one for the !connectionId guard.
|
||||
// The critical invariant: no bare updateProviderCredentials call appears
|
||||
// OUTSIDE the mutex without a !connectionId guard.
|
||||
|
||||
// Locate the outer (non-mutex, non-copilot) call and verify it is inside
|
||||
// an `if (!connectionId)` block.
|
||||
const mutexCallEnd = fnBody.indexOf("})") + 2; // end of mutex call block
|
||||
const afterMutex = fnBody.slice(mutexCallEnd);
|
||||
|
||||
// The first updateProviderCredentials after the mutex call (but before
|
||||
// the copilot section) should be guarded by !connectionId.
|
||||
const outerCallMatch = afterMutex.match(
|
||||
/if\s*\(\s*!connectionId\s*\)\s*\{[\s\S]*?updateProviderCredentials/
|
||||
);
|
||||
assert.ok(
|
||||
outerCallMatch !== null,
|
||||
"The updateProviderCredentials call outside the mutex closure MUST be inside " +
|
||||
"an `if (!connectionId)` guard. Without this guard, the DB write executes " +
|
||||
"twice when connectionId is set: once inside the mutex (correct) and once " +
|
||||
"after (race condition re-introduced)."
|
||||
);
|
||||
|
||||
// Verify the total occurrence count is consistent (not 0 outside mutex).
|
||||
assert.ok(
|
||||
allOccurrences.length >= 2,
|
||||
`Expected at least 2 calls to updateProviderCredentials in checkAndRefreshToken ` +
|
||||
`(mutex closure + copilot block), got ${allOccurrences.length}`
|
||||
);
|
||||
});
|
||||
|
|
@ -294,7 +294,12 @@ test("refreshKimiCodingToken adds provider-specific headers and fields", async (
|
|||
});
|
||||
},
|
||||
async () => {
|
||||
const result = await refreshKimiCodingToken("kimi-refresh", log);
|
||||
// Pass providerSpecificData with a stable deviceId (second positional arg after signature change)
|
||||
const result = await refreshKimiCodingToken(
|
||||
"kimi-refresh",
|
||||
{ deviceId: "test-stable-device" },
|
||||
log
|
||||
);
|
||||
assert.deepEqual(result, {
|
||||
accessToken: "kimi-access",
|
||||
refreshToken: "kimi-refresh-next",
|
||||
|
|
@ -306,9 +311,18 @@ test("refreshKimiCodingToken adds provider-specific headers and fields", async (
|
|||
);
|
||||
|
||||
assert.equal(calls[0].url, PROVIDERS["kimi-coding"].refreshUrl);
|
||||
assert.equal(calls[0].options.headers["X-Msh-Platform"], "omniroute");
|
||||
assert.equal(calls[0].options.headers["X-Msh-Version"], "2.1.2");
|
||||
assert.match(calls[0].options.headers["X-Msh-Device-Id"], /^kimi-refresh-/);
|
||||
// Platform is now "kimi_cli" (matching the real Kimi CLI fingerprint)
|
||||
assert.equal(calls[0].options.headers["X-Msh-Platform"], "kimi_cli");
|
||||
// Version comes from KIMI_CLI_VERSION env or default "1.36.0"
|
||||
assert.ok(calls[0].options.headers["X-Msh-Version"], "X-Msh-Version must be set");
|
||||
// Device-Id must NOT be an ephemeral "kimi-refresh-<timestamp>" value
|
||||
assert.ok(
|
||||
calls[0].options.headers["X-Msh-Device-Id"] &&
|
||||
!calls[0].options.headers["X-Msh-Device-Id"].startsWith("kimi-refresh-"),
|
||||
"X-Msh-Device-Id must be stable (not ephemeral kimi-refresh-<timestamp>)"
|
||||
);
|
||||
// When providerSpecificData.deviceId is provided, it should be used directly
|
||||
assert.equal(calls[0].options.headers["X-Msh-Device-Id"], "test-stable-device");
|
||||
assert.match(bodyToString(calls[0].options.body), /grant_type=refresh_token/);
|
||||
});
|
||||
|
||||
|
|
@ -403,7 +417,8 @@ test("refreshQwenToken surfaces invalid_request as unrecoverable", async () => {
|
|||
async () => textResponse(JSON.stringify({ error: "invalid_request" }), 400),
|
||||
async () => {
|
||||
const result = await refreshQwenToken("qwen-refresh", log);
|
||||
assert.deepEqual(result, { error: "invalid_request" });
|
||||
// Normalized to unrecoverable_refresh_error sentinel (Fix 4)
|
||||
assert.deepEqual(result, { error: "unrecoverable_refresh_error", code: "invalid_request" });
|
||||
}
|
||||
);
|
||||
});
|
||||
|
|
@ -1189,6 +1204,12 @@ test("getAccessToken per-connection mutex: mutex cleared after success, next cal
|
|||
const log = createLog();
|
||||
let upstreamCallCount = 0;
|
||||
|
||||
// The rotation map (added for the codex-multi-auth pattern) is process-wide
|
||||
// and intentionally redirects a stale-token caller to the cached rotated
|
||||
// tokens. Clear it BEFORE and BETWEEN calls so this test exercises the
|
||||
// lower-level mutex semantics it was designed for.
|
||||
tokenRefresh._clearTokenRotationMap();
|
||||
|
||||
await withPatchedProperties(
|
||||
PROVIDERS,
|
||||
{ "custom-oauth-conn-mutex": { tokenUrl: "https://auth.example.com/token" } },
|
||||
|
|
@ -1206,6 +1227,7 @@ test("getAccessToken per-connection mutex: mutex cleared after success, next cal
|
|||
const credentials = { connectionId: "conn-refire", refreshToken: "rt" };
|
||||
|
||||
const first = await getAccessToken("custom-oauth-conn-mutex", { ...credentials }, log);
|
||||
tokenRefresh._clearTokenRotationMap();
|
||||
const second = await getAccessToken("custom-oauth-conn-mutex", { ...credentials }, log);
|
||||
|
||||
assert.equal(upstreamCallCount, 2, "each sequential call fires upstream once");
|
||||
|
|
@ -1274,7 +1296,9 @@ test("refreshClaudeOAuthToken returns error object for invalid_grant (expired re
|
|||
async () => {
|
||||
const result = await refreshClaudeOAuthToken("expired-token", log);
|
||||
assert.ok(result && typeof result === "object", "should return error object, not null");
|
||||
assert.equal((result as any).error, "invalid_grant");
|
||||
// Normalized to unrecoverable_refresh_error sentinel (Fix 6)
|
||||
assert.equal((result as any).error, "unrecoverable_refresh_error");
|
||||
assert.equal((result as any).code, "invalid_grant");
|
||||
assert.ok(isUnrecoverableRefreshError(result), "should be detected as unrecoverable");
|
||||
}
|
||||
);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue