From 401a7b4430cf8d5b8a4fd689750575b49e922497 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Mon, 29 Jun 2026 03:46:21 -0300 Subject: [PATCH] fix(auth): compare-and-swap guard on OAuth refresh persist (#4038) (#5294) Integrated into release/v3.8.40 --- CHANGELOG.md | 1 + config/quality/file-size-baseline.json | 3 +- open-sse/handlers/chatCore.ts | 20 ++- open-sse/services/tokenRefresh.ts | 87 ++++++++++++- .../unit/token-refresh-cas-guard-4038.test.ts | 114 ++++++++++++++++++ 5 files changed, 222 insertions(+), 3 deletions(-) create mode 100644 tests/unit/token-refresh-cas-guard-4038.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 57d5f9c419..05eccb6dfb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,7 @@ _In development β€” bullets added per PR; finalized at release._ ### πŸ”§ Bug Fixes +- **fix(auth): compare-and-swap guard on the OAuth refresh persist** β€” under multi-agent load, the per-connection refresh mutex makes `[network refresh + DB write]` atomic for **one** connection, but it does not protect against a **third** writer (a sibling request, a concurrent HealthCheck, or a replica) landing a fresher `refresh_token` rotation on the same `connection_id` between the staleness read and the persist. Overwriting that fresher row reverts the sibling's rotation; the next caller then loads the now-consumed token, Auth0/Anthropic flag it as `refresh_token_reused`, and the whole token family gets revoked (the 1352Γ— claude/`aa5dd5cf` invalidation storm). `getAccessToken` now re-reads the row's current `refresh_token` immediately before persisting (inside the mutex) and **skips the write** when it has rotated past the token the caller presented β€” the caller still receives the freshly-issued access token, only the DB overwrite is skipped. Opt-in via `runWithCasGuard` (no active guard β‡’ byte-identical behavior); skip/persist counters exposed via `getCasGuardStats()`. Regression guard: `tests/unit/token-refresh-cas-guard-4038.test.ts`. ([#4038](https://github.com/diegosouzapw/OmniRoute/issues/4038) β€” thanks @KooshaPari for the root-cause diagnosis) - **mcp:** break the `schemas/tools.ts ↔ schemas/toolSearch.ts` import cycle introduced when the `tool_search` defs (#5269) were extracted into their own module β€” `toolSearch.ts` imported `McpToolDefinition` from `tools.ts` while `tools.ts` imported `toolSearchTool` from `toolSearch.ts`, failing `check:cycles` on `release/v3.8.40`. The shared `AuditLevel` + `McpToolDefinition` types now live in a leaf `schemas/toolDefinition.ts` that both import; `tools.ts` re-exports them for backward compatibility. - **compression (analytics):** record attempted-but-no-op compression runs so Stacked is no longer invisible when it saves nothing. Previously a `compression_analytics` row was written only on a net-positive saving, so a Stacked (RTKβ†’Caveman) pipeline that ran on already-compact context produced no row β€” indistinguishable from "never dispatched" (`byMode.stacked.count` stayed flat while Ultra climbed). Such runs are now recorded with `skip_reason` and surfaced as a per-mode `skipped` count plus `totalSkipped`/`bySkipReason` in the analytics summary and the Mode Breakdown; the existing net-saving totals/averages are unchanged (skip rows are excluded from them) (#4268 β€” thanks @abdulkadirozyurt, @androw) - **cli (tray):** fix `omniroute server --tray` showing no tray on macOS/Linux with no error printed. The wired Unix tray path loaded `systray2` through an inline loader that called `require("module")` inside an ESM `.mjs` file (`"type":"module"`) β†’ `ReferenceError: require is not defined`, silently swallowed (regressed in v3.8.34); even if it had loaded, `systray2` isn't in `node_modules` (it's lazily installed into `~/.omniroute/runtime`). The loader now delegates to the runtime loader, the icon path (`icon.png`) is corrected, `isTemplateIcon` is `false` (the full-color icon rendered as a white square under macOS template mode), and tray start failures are surfaced to stderr instead of being swallowed (#4605 β€” thanks @ProgMEM-CC) diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 39b7396d12..700c30c2f2 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -165,7 +165,8 @@ "_rebaseline_2026_06_29_5286_memoization": "PR #5286 own growth: strategySelector.ts 899->960 (+61 = the opt-in result-memoization branches in applyCompression/applyCompressionAsync β€” principal+determinism gate, makeMemoKey lookup/store with model+supportsVision folded into the key, recompute-with-memo-off). Default off (memoizeCompressionResults), so zero behavior change. The memo helpers live in the leaf resultMemo.ts (2181 (+78 = the compare-and-swap guard on the refresh persist β€” runWithCasGuard/getActiveCasGuard AsyncLocalStorage pair mirroring runWithOnPersist, casGuardShouldSkipPersist that rereads the row right before persisting and skips the write when a concurrent writer already rotated the refresh_token past the one presented, plus getCasGuardStats counters). Fixes the sibling-rotation-revert β†’ token-family-revocation storm. Gated behind an active guard (opt-in; no guard => byte-identical). Wiring lives at the two persist chokepoints inside getAccessToken; the comparison reuses wasRefreshTokenRotated from refreshSerializer. Not extractable without splitting the refresh hot path.", + "open-sse/services/tokenRefresh.ts": 2181, "open-sse/services/usage.ts": 3454, "open-sse/translator/request/openai-to-gemini.ts": 906, "open-sse/translator/request/openai-to-kiro.ts": 842, diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 8472cb280a..0d22e66837 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -82,6 +82,7 @@ import { refreshWithRetry, isUnrecoverableRefreshError, runWithOnPersist, + runWithCasGuard, } from "../services/tokenRefresh.ts"; import { createRequestLogger } from "../utils/requestLogger.ts"; import { createPreparedRequestLogger, runWithCapture } from "../utils/providerRequestLogging.ts"; @@ -2903,8 +2904,25 @@ export async function handleChatCore({ } : undefined; + // #4038: build a compare-and-swap reread so getAccessToken can skip the persist if a + // concurrent writer (sibling request / HealthCheck / replica) already rotated this + // connection's refresh_token past the one we presented β€” overwriting would revert it + // and revoke the token family. No connectionId β‡’ no guard (behavior unchanged). + const casConnectionId = + typeof credentials?.connectionId === "string" ? credentials.connectionId.trim() : ""; + const casReread = casConnectionId + ? async () => { + const latest = await getProviderConnectionById(casConnectionId); + return typeof latest?.refreshToken === "string" ? latest.refreshToken : null; + } + : null; + const newCredentials = (await refreshWithRetry( - () => runWithOnPersist(persistFn, () => executor.refreshCredentials(credentials, log)), + () => + runWithCasGuard( + casReread ? { expectedRefreshToken: attemptedRefreshToken, reread: casReread } : null, + () => runWithOnPersist(persistFn, () => executor.refreshCredentials(credentials, log)) + ), 3, log, provider // Explicitly pass the provider to avoid universally tripping the "unknown" circuit breaker diff --git a/open-sse/services/tokenRefresh.ts b/open-sse/services/tokenRefresh.ts index 4d436f4d34..338ddca604 100755 --- a/open-sse/services/tokenRefresh.ts +++ b/open-sse/services/tokenRefresh.ts @@ -4,7 +4,7 @@ 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 { serializeRefresh } from "./refreshSerializer.ts"; +import { serializeRefresh, wasRefreshTokenRotated } from "./refreshSerializer.ts"; import { WINDSURF_CONFIG } from "@/lib/oauth/constants/oauth"; import { buildGitLabOAuthEndpoints, resolveGitLabOAuthBaseUrl } from "@/lib/oauth/gitlab"; @@ -171,6 +171,81 @@ export function getActiveOnPersist(): RefreshPersistFn | undefined { return onPersistStore.getStore(); } +// ── #4038: compare-and-swap (CAS) guard on the refresh persist ─────────────── +// Fix A makes [network refresh + DB write] atomic *for a single connection's +// mutex*. It does NOT protect against a THIRD writer (a sibling process, a +// concurrent HealthCheck, or a replica) landing a fresher rotation on the same +// `connection_id` between the moment the caller read the row and the moment this +// persist runs. Overwriting that fresher row reverts the sibling's rotation, the +// next caller loads the reverted (now-consumed) refresh_token, and Auth0/Anthropic +// revoke the whole token family (the 1352Γ— claude/aa5dd5cf invalidation storm). +// +// The CAS guard carries the refresh_token the caller PRESENTED (the version token, +// since refresh_tokens rotate on every refresh) plus a `reread` of the row's +// current refresh_token. Right before persisting, `getAccessToken` re-reads and, if +// a concurrent writer already rotated the row past the presented token, SKIPS the +// persist so the DB stays at the fresher state. The caller still receives the new +// accessToken β€” upstream already authenticated the request; only the DB write is +// skipped. No active guard β‡’ behavior is byte-identical to before (opt-in). +type CasGuard = { + /** The refresh_token the caller presented for this refresh (CAS version token). */ + expectedRefreshToken: string | null; + /** Re-reads the CURRENT persisted refresh_token for this connection (decrypted). */ + reread: () => Promise; +}; +const casGuardStore = new AsyncLocalStorage(); +const casGuardStats = { skipped: 0, persisted: 0 }; + +export function runWithCasGuard( + guard: CasGuard | undefined | null, + fn: () => Promise +): Promise { + if (!guard) return fn(); + return casGuardStore.run(guard, fn); +} + +export function getActiveCasGuard(): CasGuard | undefined { + return casGuardStore.getStore(); +} + +/** Skip/persist counters for observability + tests. */ +export function getCasGuardStats(): { skipped: number; persisted: number } { + return { ...casGuardStats }; +} + +/** Test-only: reset the CAS counters between cases. */ +export function _resetCasGuardStats(): void { + casGuardStats.skipped = 0; + casGuardStats.persisted = 0; +} + +/** + * Returns true when the persist should be SKIPPED because a concurrent writer + * already rotated the row's refresh_token past the one we presented (CAS mismatch). + * Best-effort: any reread failure falls through to persist (never blocks recovery). + */ +async function casGuardShouldSkipPersist(log?: RefreshLogger): Promise { + const guard = getActiveCasGuard(); + if (!guard || !guard.expectedRefreshToken) return false; + let current: string | null | undefined; + try { + current = await guard.reread(); + } catch { + return false; // reread failed β€” fall through to persist (best-effort) + } + // wasRefreshTokenRotated is true iff both are non-empty AND current !== expected. + if (wasRefreshTokenRotated(guard.expectedRefreshToken, current)) { + casGuardStats.skipped++; + log?.warn?.( + "TOKEN_REFRESH", + "CAS guard: skipping persist β€” a concurrent writer already rotated the refresh_token (#4038)" + ); + return true; + } + casGuardStats.persisted++; + return false; +} + type RefreshLogger = { info?: (tag: string, message: string, data?: Record) => void; warn?: (tag: string, message: string, data?: Record) => void; @@ -1670,6 +1745,11 @@ export async function getAccessToken( // 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) { + // #4038: skip the persist if a concurrent writer already rotated this row past the + // refresh_token we presented (compare-and-swap) β€” overwriting would revert it. + if (await casGuardShouldSkipPersist(log)) { + return result; + } try { await effectiveOnPersist(result); } catch (persistErr) { @@ -1707,6 +1787,11 @@ export async function getAccessToken( ) .then(async (result) => { if (result?.accessToken && effectiveOnPersist) { + // #4038: same compare-and-swap guard as Layer 1 β€” skip the persist if a concurrent + // writer already rotated this row past the refresh_token we presented. + if (await casGuardShouldSkipPersist(log)) { + return result; + } try { await effectiveOnPersist(result); } catch (persistErr) { diff --git a/tests/unit/token-refresh-cas-guard-4038.test.ts b/tests/unit/token-refresh-cas-guard-4038.test.ts new file mode 100644 index 0000000000..20305b6926 --- /dev/null +++ b/tests/unit/token-refresh-cas-guard-4038.test.ts @@ -0,0 +1,114 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { + getAccessToken, + runWithCasGuard, + getCasGuardStats, + _resetCasGuardStats, + _clearTokenRotationMap, +} from "../../open-sse/services/tokenRefresh.ts"; + +// #4038: the per-connection mutex makes [refresh + persist] atomic for ONE connection, +// but a THIRD writer (sibling request / HealthCheck / replica) can land a fresher +// refresh_token rotation between our staleness read and our persist. Overwriting it +// reverts the sibling's rotation β†’ the next caller loads a now-consumed token β†’ Auth0 +// revokes the whole family (the 1352Γ— claude invalidation storm). The CAS guard +// re-reads the row right before persisting and SKIPS the write when the row's +// refresh_token has rotated past the one we presented. + +const silentLog = { info() {}, warn() {}, error() {} }; + +// Mock the Anthropic token endpoint so claude's refresh succeeds without a network call. +function withMockedRefresh(newRefreshToken: string, fn: () => Promise): Promise { + const realFetch = globalThis.fetch; + globalThis.fetch = (async () => + new Response( + JSON.stringify({ + access_token: "NEW_ACCESS_TOKEN", + refresh_token: newRefreshToken, + expires_in: 3600, + }), + { status: 200, headers: { "content-type": "application/json" } } + )) as never; + return fn().finally(() => { + globalThis.fetch = realFetch; + }); +} + +test("#4038 CAS guard SKIPS the persist when a concurrent writer rotated the refresh_token", async () => { + _resetCasGuardStats(); + _clearTokenRotationMap(); + let persisted = false; + const onPersist = async () => { + persisted = true; + }; + + const result = await withMockedRefresh("ROTATED_BY_US", () => + runWithCasGuard( + // The row's CURRENT refresh_token is NOT the one we presented (R0): a sibling + // already rotated it to R_CONCURRENT while our network refresh was in flight. + { expectedRefreshToken: "R0", reread: async () => "R_CONCURRENT" }, + () => + getAccessToken( + "claude", + { refreshToken: "R0", connectionId: "conn-cas-skip" }, + silentLog, + null, + onPersist + ) + ) + ); + + assert.equal(persisted, false, "persist MUST be skipped when the row was rotated concurrently"); + assert.equal(getCasGuardStats().skipped, 1, "the skip must be counted"); + assert.ok(result?.accessToken, "caller still receives the freshly-issued access token"); +}); + +test("#4038 CAS guard PERSISTS when the row is unchanged (no concurrent rotation)", async () => { + _resetCasGuardStats(); + _clearTokenRotationMap(); + let persisted = false; + const onPersist = async () => { + persisted = true; + }; + + await withMockedRefresh("NEW_REFRESH_TOKEN", () => + runWithCasGuard( + // The row still holds R0 β€” the exact token we presented β€” so our persist is safe. + { expectedRefreshToken: "R0", reread: async () => "R0" }, + () => + getAccessToken( + "claude", + { refreshToken: "R0", connectionId: "conn-cas-pass" }, + silentLog, + null, + onPersist + ) + ) + ); + + assert.equal(persisted, true, "persist MUST run when the row still holds the presented token"); + assert.equal(getCasGuardStats().persisted, 1, "the pass must be counted"); +}); + +test("#4038 no CAS guard β‡’ persist always runs (opt-in; zero behavior change)", async () => { + _resetCasGuardStats(); + _clearTokenRotationMap(); + let persisted = false; + const onPersist = async () => { + persisted = true; + }; + + await withMockedRefresh("NEW_REFRESH_TOKEN", () => + getAccessToken( + "claude", + { refreshToken: "R0", connectionId: "conn-no-guard" }, + silentLog, + null, + onPersist + ) + ); + + assert.equal(persisted, true, "without a guard the persist always runs (unchanged behavior)"); + assert.equal(getCasGuardStats().skipped, 0, "no guard β‡’ nothing skipped"); +});