diff --git a/.changeset/oauth-refresh-cross-session.md b/.changeset/oauth-refresh-cross-session.md new file mode 100644 index 000000000..e3d94dd52 --- /dev/null +++ b/.changeset/oauth-refresh-cross-session.md @@ -0,0 +1,13 @@ +--- +"@executor-js/sdk": patch +--- + +Share the OAuth refresh gate across execution stacks so a rotating refresh token is redeemed once. + +The in-flight refresh gate was built inside `createExecutor`, so it only covered one execution stack. A host builds a fresh stack per MCP session, and now per request, so two sessions resolving the same connection each read the same stored refresh token and each believed they were the refresh winner. Against a provider that rotates refresh tokens, the loser redeems a token the winner already spent, and a provider that detects reuse revokes the whole token family: the connection dies and the user has to reauthorize. The first refresh always succeeds, so the fault stayed invisible until a later expiry. + +The gate now hangs off the root database handle, which is the object hosts already share across sessions and requests, so every stack over one handle converges on one gate. Its key includes the tenant, because a gate that spans tenants would otherwise let two tenants collide on one entry. + +The grant also runs on its own detached fiber that callers await, rather than on whichever caller registered it. Sharing an entry across stacks would otherwise share the first caller's cancellation: a disconnected MCP client or an execution deadline would fail every peer waiting on that entry, and would abandon a refresh token the authorization server had already rotated, which is itself a dead connection. A cancelled peer now detaches without touching the grant, and a grant nobody is left waiting on still settles and still persists the rotated token. + +Deduplication covers one database handle in one process. A host that builds a fresh handle per request or per session, and any multi-instance or multi-replica deployment, is out of scope here and still needs database-backed coordination, such as a compare-and-swap on the stored refresh token. diff --git a/e2e/selfhost/oauth-refresh-cross-session.test.ts b/e2e/selfhost/oauth-refresh-cross-session.test.ts new file mode 100644 index 000000000..6819b4158 --- /dev/null +++ b/e2e/selfhost/oauth-refresh-cross-session.test.ts @@ -0,0 +1,275 @@ +// Selfhost-only: two MCP sessions that hit an expired token at the same moment +// must share ONE refresh-token grant, never race two. +// +// Issue #1520: the in-flight refresh gate lived inside a single execution +// stack, but the self-host builds a fresh stack per MCP session, so each +// session believed it was the refresh winner and redeemed the same stored +// refresh token. Providers that rotate refresh tokens answer the second +// redemption with `invalid_grant: refresh token reuse detected` and may revoke +// the whole token family — the connection dies and the user must reauthorize. +// The first refresh cycle succeeds, so the bug stays invisible until a later +// expiry. +// +// The journey: an OpenAPI integration completes a real authorization-code flow +// against a live test authorization server; the upstream then rejects both +// sessions' first call with a 401 at the same instant (it holds both requests +// until both have arrived, so the contention is forced rather than left to the +// scheduler); and the authorization server's own request ledger proves exactly +// one refresh grant was issued and both retries carried the same new bearer. +import { randomBytes } from "node:crypto"; +import { createServer, type ServerResponse } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, +} from "@executor-js/sdk/shared"; +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +/** Both sessions call once, are rejected together, then retry once. */ +const SESSIONS = 2; + +type UpstreamHandle = { + readonly url: string; + readonly bearers: () => readonly string[]; + readonly close: () => void; +}; + +/** + * Upstream that rejects the whole first wave at once. + * + * The barrier is the point: it holds every session's first call until all have + * arrived, then 401s them together, which forces both sessions into a genuinely + * simultaneous refresh instead of hoping the scheduler interleaves them. + */ +const serveUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + const bearers: string[] = []; + const held: ServerResponse[] = []; + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith("/issues")) { + bearers.push((request.headers.authorization ?? "").replace(/^Bearer\s+/i, "")); + if (held.length < SESSIONS) { + held.push(response); + if (held.length === SESSIONS) { + for (const rejected of held) { + rejected.writeHead(401, { "content-type": "application/json" }); + rejected.end(JSON.stringify({ error: "invalid_token" })); + } + } + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ issues: [] })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + bearers: () => [...bearers], + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const spec = ( + baseUrl: string, + oauth: { readonly authorizationEndpoint: string; readonly tokenEndpoint: string }, +): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Issues API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/issues": { + get: { + operationId: "listIssues", + security: [{ oauth: ["issues.read"] }], + responses: { "200": { description: "issues" } }, + }, + }, + }, + components: { + securitySchemes: { + oauth: { + type: "oauth2", + flows: { + authorizationCode: { + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: { "issues.read": "Read issues" }, + }, + }, + }, + }, + }, + }); + +const invokeByAddressCode = (address: string) => ` +const segments = ${JSON.stringify(address)}.split(".").slice(1); +let node = tools; +for (const segment of segments) node = node[segment]; +const result = await node({}); +return JSON.stringify(result); +`; + +const completeAuthorization = (authorizationUrl: string) => + Effect.promise(async () => { + const authorize = await fetch(authorizationUrl, { redirect: "manual" }); + const loginUrl = authorize.headers.get("location"); + if (!loginUrl) return null; + const login = await fetch(loginUrl, { + method: "POST", + headers: { authorization: `Basic ${Buffer.from("alice:password").toString("base64")}` }, + redirect: "manual", + }); + const callbackUrl = login.headers.get("location"); + if (!callbackUrl) return null; + return new URL(callbackUrl).searchParams.get("code"); + }); + +scenario( + "OAuth refresh · separate MCP sessions share one rotating-token refresh grant", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream(); + const oauth = yield* serveOAuthTestServer({ scopes: ["issues.read"] }); + const slug = unique("refreshcrosssession"); + const clientSlug = OAuthClientSlug.make(unique("refreshcrosssessionc")); + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: spec(upstream.url, oauth) }, + slug, + baseUrl: upstream.url, + authenticationTemplate: [ + { + slug: "oauth", + kind: "oauth2", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: ["issues.read"], + }, + ], + }, + }); + yield* client.oauth.createClient({ + payload: { + owner: "org", + slug: clientSlug, + grant: "authorization_code", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + clientId: "test-client", + clientSecret: "test-secret", + originIntegration: IntegrationSlug.make(slug), + }, + }); + const started = yield* client.oauth.start({ + payload: { + client: clientSlug, + clientOwner: "org", + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make(slug), + template: AuthTemplateSlug.make("oauth"), + }, + }); + expect(started.status, "oauth.start redirects to the authorization server").toBe( + "redirect", + ); + if (started.status !== "redirect") return yield* Effect.die("no redirect"); + const code = yield* completeAuthorization(started.authorizationUrl); + expect(code, "the authorization server issued a callback code").toBeDefined(); + if (code == null) return yield* Effect.die("no authorization code"); + yield* client.oauth.complete({ payload: { state: started.state, code } }); + + const address = (yield* client.tools.list({ query: {} })) + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((tool) => tool.endsWith("listIssues")); + expect(address, "the OAuth-protected tool is in the catalog").toBeDefined(); + if (!address) return yield* Effect.die("no listIssues tool"); + yield* oauth.clearRequests; + + const sessions = Array.from({ length: SESSIONS }, () => mcp.session(identity)); + const call = (session: (typeof sessions)[number]) => + Effect.gen(function* () { + let result = yield* session.call("execute", { code: invokeByAddressCode(address) }); + let approvals = 0; + while (result.text.includes("executionId:") && approvals < 10) { + result = yield* session.approvePaused(result.text); + approvals += 1; + } + // Without a shared gate the loser redeems a retired refresh token + // and the authorization server answers invalid_grant, so this is + // the assertion that carries the user-visible failure. + expect(result.ok, `MCP execute completed: ${result.text.slice(0, 400)}`).toBe(true); + }); + + yield* Effect.all(sessions.map(call), { concurrency: "unbounded" }); + + const refreshGrants = (yield* oauth.requests).filter( + (request) => + request.path === "/token" && request.body.includes("grant_type=refresh_token"), + ); + expect(refreshGrants, "both sessions joined one refresh grant").toHaveLength(1); + const bearers = upstream.bearers(); + expect(bearers, "both rejected calls retried after the refresh").toHaveLength( + SESSIONS * 2, + ); + expect(bearers[2], "the first retry used a new bearer").not.toBe(bearers[0]); + expect(bearers[3], "both retries used the refreshed bearer").toBe(bearers[2]); + }), + Effect.gen(function* () { + yield* client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("main"), + }, + }) + .pipe(Effect.ignore); + yield* client.oauth + .removeClient({ params: { slug: clientSlug }, payload: { owner: "org" } }) + .pipe(Effect.ignore); + yield* client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore); + }), + ); + }), + ), +); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 7c3c25929..e58ca0fee 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -199,6 +199,47 @@ const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90; const PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE = 90; const MAX_APPROVAL_ARGUMENT_PREVIEW_CHARS = 4_000; +// --------------------------------------------------------------------------- +// In-flight OAuth refresh gate — a CROSS-STACK resource. +// +// Concurrent resolves of one connection must share a single refresh-token +// grant: the authorization server rotates the refresh token, so a second +// grant redeems a token the first already consumed, and a provider that +// detects reuse revokes the whole token family. The first refresh cycle still +// succeeds, so the fault hides until a later expiry. +// +// The gate therefore cannot live on a single execution stack. A host builds a +// fresh scoped executor per MCP session (and, since request-scoped stack +// builds, per request), so a per-`createExecutor` map put every session in its +// own gate and deduplicated nothing. Hanging it off the root DB handle instead +// converges every stack over one handle on one map — the same object hosts +// already treat as their shared, process-lived resource. +// +// SCOPE OF THE GUARANTEE: dedup reaches exactly as far as one root DB handle +// in one process. A host that hands every scoped executor a FRESH handle keys +// a different map each time and gets no dedup — silently, because an unshared +// gate still behaves correctly for the one caller holding it. Multi-instance +// deployments are outside it for the same reason: a process-local map cannot +// see a peer isolate or replica. Both need database-backed coordination +// (compare-and-swap on the stored refresh token) rather than a wider map. +// +// Weakly keyed so the map dies with the handle and a host that opens and drops +// handles does not leak one gate per handle. +type RefreshGate = Map< + string, + Deferred.Deferred +>; + +const refreshGateByRootDb = new WeakMap(); + +const refreshGateFor = (rootDb: object): RefreshGate => { + const existing = refreshGateByRootDb.get(rootDb); + if (existing) return existing; + const created: RefreshGate = new Map(); + refreshGateByRootDb.set(rootDb, created); + return created; +}; + // --------------------------------------------------------------------------- // Elicitation handler — resolved once at `createExecutor({ onElicitation })` // and overridable per `execute`. A tool that requests user input mid-execution @@ -1697,6 +1738,9 @@ export const createExecutor = { validateExecutorDbTables(tables, rootDbUntyped.internal.tables); @@ -1920,17 +1964,18 @@ export const createExecutor = - >(); - + // Key for the shared in-flight refresh gate (`refreshInFlight`, bound at + // the top of `createExecutor`). The tenant leads because the gate spans + // every execution stack over one DB handle, so it spans tenants too: + // without it, two tenants whose rows agree on owner/subject/integration/ + // name would collide on one entry and one tenant's caller could be handed + // the other's access token. JSON.stringify keeps the components + // unambiguous: tenant, subject, integration, and name are opaque strings + // that may contain any delimiter, so a delimiter-joined key would let + // tenant "a" + subject "user:b" collide with tenant "a:user" + subject + // "b" — the same cross-tenant bleed by another route. const connectionKey = (row: ConnectionRow): string => - `${row.owner}:${row.subject}:${row.integration}:${row.name}`; + JSON.stringify([tenant, row.owner, row.subject, row.integration, row.name]); const loadOAuthClientRow = ( owner: Owner, @@ -2526,26 +2571,42 @@ export const createExecutor = { const key = connectionKey(row); // Joining an in-flight grant is correct for BOTH triggers: whatever // that peer mints is newer than the token this fiber just saw rejected, // which is exactly what a reactive retry wants. The gate is cleared on // settle, so a 401 arriving after a refresh completed starts a fresh - // grant rather than replaying the stale memoized one. + // grant rather than replaying a stale result. const existing = refreshInFlight.get(key); - if (existing) return yield* existing; - // `Effect.cached` memoizes the grant onto a deferred: it runs once and - // replays to every awaiter sharing this entry. - const memoized = yield* Effect.cached(performTokenRefresh(row, provider, trigger)); - const gated = memoized.pipe( - Effect.ensuring(Effect.sync(() => refreshInFlight.delete(key))), + if (existing) return Deferred.await(existing); + + // The grant runs on a DETACHED fiber and every caller — including this + // one — only awaits its deferred. The entry is shared across execution + // stacks, so the fiber that registers it is merely the first arrival, + // not an owner. Running the grant ON that fiber would hand it that + // caller's interruption: a disconnected MCP client, an execution + // deadline or a cancelled tool call would fail every peer awaiting the + // same entry with an interrupt none of them caused and none can act on. + // Awaiting is per-caller, so a cancelled peer detaches without touching + // the grant or its siblings, and a grant nobody is left waiting on + // still settles and still persists the rotated token — which is what + // keeps the next caller off a consumed one. Token requests are bounded + // by `AbortSignal.timeout`, so the detached fiber cannot outlive its + // request. + const deferred = Deferred.makeUnsafe< + string | null, + StorageFailure | CredentialResolutionError + >(); + // Nothing suspends between the lookup above and this registration, so + // check-and-set is atomic against peer fibers and cannot double-fire. + refreshInFlight.set(key, deferred); + const run = performTokenRefresh(row, provider, trigger).pipe( + Effect.exit, + Effect.flatMap((exit) => Deferred.done(deferred, exit)), + Effect.ensuring(Effect.sync(() => void refreshInFlight.delete(key))), ); - // Re-check after building (a peer fiber may have registered first while - // we built ours) so everyone converges on the same shared grant. - const winner = refreshInFlight.get(key) ?? gated; - if (winner === gated) refreshInFlight.set(key, gated); - return yield* winner; + return Effect.forkDetach(run).pipe(Effect.andThen(Deferred.await(deferred))); }); // Resolve every named input of a connection (`variable → value`). A diff --git a/packages/core/sdk/src/oauth-flow.test.ts b/packages/core/sdk/src/oauth-flow.test.ts index e7a8e6d65..7c86fa32d 100644 --- a/packages/core/sdk/src/oauth-flow.test.ts +++ b/packages/core/sdk/src/oauth-flow.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "@effect/vitest"; import { Deferred, Effect, Fiber, Predicate } from "effect"; +import { withQueryContext } from "@executor-js/fumadb/query"; import { AuthTemplateSlug, @@ -8,6 +9,8 @@ import { OAuthClientSlug, OAuthState, ProviderKey, + Subject, + Tenant, ToolAddress, ToolName, } from "./ids"; @@ -68,6 +71,69 @@ const oauthPlugin = definePlugin(() => ({ const plugins = [memoryCredentialsPlugin(), oauthPlugin] as const; +// Stated explicitly where a test builds a SECOND root database handle by hand: +// both handles must carry the same owner-policy context to address one +// connection, so the values cannot be left to `makeTestConfig`'s defaults. +const SHARED_STORE_TENANT = "test-tenant"; +const SHARED_STORE_SUBJECT = "test-subject"; + +/** The URL a `fetch` double was handed, however the caller spelled it. */ +const fetchTarget = (input: Parameters[0]): string => + typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + +/** + * A `fetch` that holds token-endpoint requests open AFTER the authorization + * server has answered them. + * + * Parking after the response is the point. The refresh token has been rotated + * upstream by then, and the in-flight gate entry is still registered, so a peer + * arriving during the park has to resolve against an OPEN grant rather than a + * settled one. Parking before the response would prove nothing: the grant would + * never reach the server, and a peer that went on to run its own grant would + * find the stored token still live and succeed. + * + * Idle until `arm()`, so connection setup (the authorization-code exchange) + * runs through untouched. + */ +const makeTokenRequestPark = () => { + let armed = false; + let onSeen: (() => void) | null = null; + const seen = new Promise((resolve) => { + onSeen = resolve; + }); + let onRelease: (() => void) | null = null; + const parked = new Promise((resolve) => { + onRelease = resolve; + }); + // oxlint-disable-next-line executor/no-raw-fetch -- test boundary: the park wraps the platform fetch and must delegate back to it, which is the only seam that can hold a token request open mid-grant. + const platformFetch: typeof globalThis.fetch = globalThis.fetch; + const fetch: typeof globalThis.fetch = async (input, init) => { + const response = await platformFetch(input, init); + if (armed && new URL(fetchTarget(input)).pathname === "/token") { + onSeen?.(); + await parked; + } + return response; + }; + return { + fetch, + arm: () => { + armed = true; + }, + /** Resolves once a token request has been answered and is being held. */ + seen, + release: () => onRelease?.(), + }; +}; + +/** Every refresh-token grant the authorization server was asked for. */ +const refreshGrantsIn = ( + requests: ReadonlyArray<{ readonly path: string; readonly body: string }>, +) => + requests.filter( + (request) => request.path === "/token" && request.body.includes("grant_type=refresh_token"), + ); + interface TokenEndpointCall { readonly host: string; readonly grantType: string | null; @@ -887,14 +953,344 @@ describe("oauth token refresh in resolveConnectionValue", () => { ), ); + // Issue #1520, in one process. A self-host builds a FRESH execution stack per + // MCP session over ONE database handle, so two sessions resolving the same + // connection each read the same stored refresh token and each believe they + // are the refresh winner. The authorization server rotates that token, so the + // loser redeems one the winner already spent, and a server that detects reuse + // revokes the whole family: the connection dies and the user must + // reauthorize. The first refresh always succeeds, which is why the fault + // stays invisible until a later expiry. + it.effect("two execution stacks over one host database share a single refresh grant", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const park = makeTokenRequestPark(); + + // One database handle and one credential store under two execution + // stacks — what a self-host holds while two MCP sessions are open. + const config = { ...makeTestConfig({ plugins }), fetch: park.fetch }; + const sessionA = yield* createExecutor(config); + const sessionB = yield* createExecutor(config); + yield* Effect.addFinalizer(() => sessionA.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => sessionB.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => + Effect.promise(() => config.testDb.close()).pipe(Effect.ignore), + ); + + yield* sessionA.acme.seed(); + yield* sessionA.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + const started = yield* sessionA.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* sessionA.oauth.complete({ state: started.state, code: callback.code }); + + const address = ToolAddress.make("tools.acme.org.main.whoami"); + const original = (yield* sessionA.execute(address, {})) as { token: string }; + + // Expire the access token so BOTH stacks must refresh. + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("name", "=", "main"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + yield* server.clearRequests; + park.arm(); + + const first = yield* Effect.forkChild(sessionA.execute(address, {})); + const second = yield* Effect.forkChild(sessionB.execute(address, {})); + // Release only once a grant has been answered and is being held open, + // so the peer resolves against a grant that is still in flight. Without + // the park the peer could arrive after the winner had already settled, + // find a fresh token, refresh nothing, and pass this test for the wrong + // reason. + yield* Effect.promise(() => park.seen); + park.release(); + + const firstToken = (yield* Fiber.join(first)) as { token: string }; + const secondToken = (yield* Fiber.join(second)) as { token: string }; + + expect(firstToken.token, "the refresh minted a new access token").not.toBe(original.token); + expect(secondToken.token, "both stacks resolved the SAME refreshed token").toBe( + firstToken.token, + ); + expect( + refreshGrantsIn(yield* server.requests), + "one refresh grant for the connection, not one per execution stack", + ).toHaveLength(1); + }), + ), + ); + + // The gate spans tenants (one map per root DB handle), so its key must keep + // tenant and subject unambiguous. Both are opaque strings that may contain + // any delimiter: under a colon-joined key, tenant "a" + subject "user:b" and + // tenant "a:user" + subject "b" both flatten to "a:user:user:b:…", so two + // DIFFERENT tenants' refreshes would share one gate entry and one tenant's + // caller would be handed the other tenant's access token. + it.effect( + "colliding tenant/subject pairs never share a refresh gate entry", + () => + Effect.scoped( + Effect.gen(function* () { + const serverA = yield* serveOAuthTestServer({ scopes: ["read"] }); + const serverB = yield* serveOAuthTestServer({ scopes: ["read"] }); + const parkA = makeTokenRequestPark(); + const parkB = makeTokenRequestPark(); + + // ONE root DB handle under TWO tenants — the shape a multi-tenant host + // holds — so both executors share one refresh gate. `shared.db` stays + // bound to tenant A's owner-policy context; tenant B's row edits below + // build their own scoped handle by hand. Each tenant gets its OWN + // credential store instance: the memory store keys items without a + // tenant, so sharing one across tenants would cross their tokens at + // the store layer and mask the gate-key collision this test is about. + const pluginsA = [memoryCredentialsPlugin(), oauthPlugin] as const; + const pluginsB = [memoryCredentialsPlugin(), oauthPlugin] as const; + const shared = makeTestConfig({ plugins: pluginsA, tenant: "a", subject: "user:b" }); + const configA = { ...shared, fetch: parkA.fetch }; + const configB = { + ...shared, + plugins: pluginsB, + tenant: Tenant.make("a:user"), + subject: Subject.make("b"), + fetch: parkB.fetch, + }; + const sessionA = yield* createExecutor(configA); + const sessionB = yield* createExecutor(configB); + yield* Effect.addFinalizer(() => sessionA.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => sessionB.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => + Effect.promise(() => shared.testDb.close()).pipe(Effect.ignore), + ); + + // Each tenant mints its own USER-owned connection (user rows carry the + // session subject, which is what the colliding pair needs) against its + // own authorization server, so token provenance is observable. + const connect = (session: typeof sessionA, server: typeof serverA) => + Effect.gen(function* () { + yield* session.acme.seed(); + yield* session.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + const started = yield* session.oauth.start({ + owner: "user", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("mine"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* session.oauth.complete({ state: started.state, code: callback.code }); + }); + yield* connect(sessionA, serverA); + yield* connect(sessionB, serverB); + + const address = ToolAddress.make("tools.acme.user.mine.whoami"); + const originalA = (yield* sessionA.execute(address, {})) as { token: string }; + const originalB = (yield* sessionB.execute(address, {})) as { token: string }; + expect(originalB.token).not.toBe(originalA.token); + + // Expire BOTH rows so both tenants must refresh. `shared.db` is bound + // to tenant A; tenant B's partition needs its own scoped handle. + const dbB = withQueryContext(shared.testDb.db, { tenant: "a:user", subject: "b" }); + yield* Effect.promise(() => + shared.db.updateMany("connection", { + where: (b) => b("name", "=", "mine"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + yield* Effect.promise(() => + dbB.updateMany("connection", { + where: (b) => b("name", "=", "mine"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + + parkA.arm(); + parkB.arm(); + const first = yield* Effect.forkChild(sessionA.execute(address, {})); + // Hold tenant A's grant open so its gate entry is still registered + // when tenant B performs its lookup of the would-be colliding key. + yield* Effect.promise(() => parkA.seen); + const second = yield* Effect.forkChild(sessionB.execute(address, {})); + // With a collision-free key, tenant B misses the gate and sends its + // OWN grant. Under a colliding key it would await tenant A's deferred + // and never reach its server, so cap the wait with a real timer (the + // test clock is virtual, so Effect.sleep would never fire) instead of + // hanging the suite; the assertions below then report the bleed. + yield* Effect.promise(() => + Promise.race([parkB.seen, new Promise((resolve) => setTimeout(resolve, 2_000))]), + ); + parkA.release(); + parkB.release(); + + const tokenA = (yield* Fiber.join(first)) as { token: string }; + const tokenB = (yield* Fiber.join(second)) as { token: string }; + + expect(tokenA.token, "tenant A refreshed to a new token").not.toBe(originalA.token); + expect(tokenB.token, "tenant B refreshed to a new token").not.toBe(originalB.token); + expect(tokenB.token, "no cross-tenant token bleed").not.toBe(tokenA.token); + expect(yield* serverA.acceptsAccessToken(tokenA.token)).toBe(true); + expect( + yield* serverB.acceptsAccessToken(tokenB.token), + "tenant B's token was minted by tenant B's own authorization server", + ).toBe(true); + expect( + refreshGrantsIn(yield* serverA.requests), + "tenant A ran its own refresh grant", + ).toHaveLength(1); + expect( + refreshGrantsIn(yield* serverB.requests), + "tenant B ran its own refresh grant — two distinct refresh executions", + ).toHaveLength(1); + }), + ), + // Two authorization servers, two full mint flows and two refreshes — about + // twice the cost of the single-tenant gate tests, which sits on the 5s + // default when the test runs cold. + 20_000, + ); + + // The gate entry is shared, so the stack that REGISTERS a grant is only the + // first arrival, not its owner. Running the grant on that caller's fiber + // would hand it that caller's interruption — a disconnected MCP client, an + // execution deadline, a cancelled tool call — and abandon a refresh token the + // authorization server has ALREADY rotated. What the store still holds is + // then dead, and the next grant is answered invalid_grant: the interruption + // would have killed the connection. So the grant runs detached, and callers + // only await it. + it.effect("an interrupted first arrival still settles the grant and persists its token", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const park = makeTokenRequestPark(); + + const config = { ...makeTestConfig({ plugins }), fetch: park.fetch }; + const sessionA = yield* createExecutor(config); + const sessionB = yield* createExecutor(config); + yield* Effect.addFinalizer(() => sessionA.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => sessionB.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => + Effect.promise(() => config.testDb.close()).pipe(Effect.ignore), + ); + + yield* sessionA.acme.seed(); + yield* sessionA.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + const started = yield* sessionA.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* sessionA.oauth.complete({ state: started.state, code: callback.code }); + + const address = ToolAddress.make("tools.acme.org.main.whoami"); + const original = (yield* sessionA.execute(address, {})) as { token: string }; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("name", "=", "main"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + yield* server.clearRequests; + park.arm(); + + // The first arrival registers the grant, and its session drops the + // instant the authorization server has rotated the token — the worst + // possible moment, and the one a disconnecting MCP client picks. + const arrival = yield* Effect.forkChild(Effect.exit(sessionA.execute(address, {}))); + yield* Effect.promise(() => park.seen); + yield* Fiber.interrupt(arrival); + park.release(); + + // That the grant finishes at all, with nobody left waiting on it, is + // the property under test: a rotated token that is never persisted is a + // dead connection. + const persisted = yield* Effect.promise(async () => { + for (let attempt = 0; attempt < 500; attempt += 1) { + const row = await config.db.findFirst("connection", { + where: (b) => b("name", "=", "main"), + }); + const expiresAt = row?.expires_at; + if (expiresAt != null && Number(expiresAt) > Date.now()) return true; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return false; + }); + expect(persisted, "the detached grant settled and persisted its rotated token").toBe(true); + + const recovered = (yield* sessionB.execute(address, {})) as { token: string }; + expect(recovered.token, "the peer stack resolved the rotated token").not.toBe( + original.token, + ); + expect( + yield* server.acceptsAccessToken(recovered.token), + "and the authorization server still honours it", + ).toBe(true); + expect( + refreshGrantsIn(yield* server.requests), + "the interrupted arrival's grant settled, so no second grant was needed", + ).toHaveLength(1); + }), + ), + ); + // Two product instances, one connection, one credential store. The in-flight - // refresh gate serialises refreshes WITHIN an instance and cannot see across - // them, so nothing but the store itself stands between two refreshers and - // the same rotated token. The provider below opens a seam exactly where the - // danger is — between the read of the stored refresh token and whatever the - // reader writes next — because a probe that "tests" the store by rewriting - // the value it just read would put the spent token back over the peer's - // rotated one, and kill the connection it was added to protect. + // refresh gate serialises refreshes across every execution stack over ONE + // root database handle and cannot see past it, so between two INSTANCES — + // two replicas, two isolates — nothing but the store itself stands between + // two refreshers and the same rotated token. The provider below opens a seam + // exactly where the danger is — between the read of the stored refresh token + // and whatever the reader writes next — because a probe that "tests" the + // store by rewriting the value it just read would put the spent token back + // over the peer's rotated one, and kill the connection it was added to + // protect. it.effect( "a refresher paused after reading the stored token never writes it back over a peer's rotated one", () => @@ -936,12 +1332,30 @@ describe("oauth token refresh in resolveConnectionValue", () => { // One database and one credential store, two executors over them — // the deployment this race needs and the one a single harness // cannot express. + // + // Each executor gets its OWN root database handle onto that one + // database, because that handle is what identifies an instance: the + // in-flight refresh gate is shared per handle, so two executors over + // the SAME handle are two execution stacks in one instance and the + // second would simply join the first's grant — closing the very + // window this test exists to open. A second replica holds a second + // handle, which is what the extra `withQueryContext` wrapper is. const config = { - ...makeTestConfig({ plugins: [oauthPlugin] as const }), + ...makeTestConfig({ + plugins: [oauthPlugin] as const, + tenant: SHARED_STORE_TENANT, + subject: SHARED_STORE_SUBJECT, + }), providers: [sharedStore], }; const instanceA = yield* createExecutor(config); - const instanceB = yield* createExecutor(config); + const instanceB = yield* createExecutor({ + ...config, + db: withQueryContext(config.testDb.db, { + tenant: SHARED_STORE_TENANT, + subject: SHARED_STORE_SUBJECT, + }), + }); yield* Effect.addFinalizer(() => Effect.promise(() => config.testDb.close()).pipe(Effect.ignore), );