fix(mcp): harden OAuth request isolation

Author: waleedlatif1

apps/sim/lib/mcp/client.test.ts

@@ -465,7 +465,7 @@ describe('McpClient notification handler', () => {
     expect(logged).not.toContain('test-session')
   })
 
-  it('passes configured headers for OAuth transports as well as header auth transports', () => {
+  it('scopes configured headers to the MCP endpoint for OAuth transports', () => {
     const authProvider = {} as unknown as NonNullable<McpClientOptions['authProvider']>
     new McpClient({
       config: {
@@ -479,10 +479,13 @@ describe('McpClient notification handler', () => {
 
     expect(StreamableHTTPClientTransport).toHaveBeenCalledWith(
       new URL('https://test.example.com/mcp'),
-      {
+      expect.objectContaining({
         authProvider,
-        requestInit: { headers: { 'X-Sim-Via': 'workflow' } },
-      }
+        fetch: expect.any(Function),
+      })
+    )
+    expect(vi.mocked(StreamableHTTPClientTransport).mock.calls.at(-1)?.[1]).not.toHaveProperty(
+      'requestInit'
     )
   })
 })

apps/sim/lib/mcp/client.ts

@@ -13,7 +13,10 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { getMcpSafeErrorDiagnostics } from '@/lib/mcp/error-diagnostics'
 import { McpOauthRedirectRequired } from '@/lib/mcp/oauth'
-import { createCoordinatedMcpOauthFetch } from '@/lib/mcp/oauth/coordinated-fetch'
+import {
+  createCoordinatedMcpOauthFetch,
+  createMcpEndpointFetch,
+} from '@/lib/mcp/oauth/coordinated-fetch'
 import { createGuardedMcpFetch, createPinnedPrivateMcpFetch } from '@/lib/mcp/pinned-fetch'
 import {
   type McpClientOptions,
@@ -127,17 +130,22 @@ export class McpClient {
         : createGuardedMcpFetch(this.config.url)
       : undefined
     this.closeGuardedTransport = guarded?.close
+    const oauthFetch = useOauth
+      ? createMcpEndpointFetch(guarded?.fetch ?? fetch, {
+          serverUrl: this.config.url,
+          headers: this.config.headers,
+        })
+      : undefined
     const transportFetch =
-      useOauth && options.oauthCredentials
+      options.oauthCredentials && oauthFetch
         ? createCoordinatedMcpOauthFetch(options.oauthCredentials, {
             serverUrl: this.config.url,
-            fetch: guarded?.fetch ?? fetch,
-            requestInit: { headers: this.config.headers },
+            fetch: oauthFetch,
           })
-        : guarded?.fetch
+        : (oauthFetch ?? guarded?.fetch)
     this.transport = new StreamableHTTPClientTransport(new URL(this.config.url), {
       authProvider: useOauth ? this.authProvider : undefined,
-      requestInit: { headers: this.config.headers },
+      ...(useOauth ? {} : { requestInit: { headers: this.config.headers } }),
       ...(transportFetch ? { fetch: transportFetch } : {}),
     })
 

apps/sim/lib/mcp/oauth/coordinated-fetch.test.ts

@@ -8,7 +8,10 @@ import type { OAuthTokens } from '@modelcontextprotocol/sdk/shared/auth.js'
 import type { FetchLike } from '@modelcontextprotocol/sdk/shared/transport.js'
 import { encryptionMock, redisConfigMockFns, resetRedisConfigMock } from '@sim/testing'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
-import { createCoordinatedMcpOauthFetch } from '@/lib/mcp/oauth/coordinated-fetch'
+import {
+  createCoordinatedMcpOauthFetch,
+  createMcpEndpointFetch,
+} from '@/lib/mcp/oauth/coordinated-fetch'
 import { withMcpOauthRefreshLock } from '@/lib/mcp/oauth/storage'
 
 vi.mock('@/lib/core/security/encryption', () => encryptionMock)
@@ -447,7 +450,25 @@ describe('coordinated MCP OAuth with the real SDK and refresh mutex', () => {
     expect(new Headers(request.mock.calls[0][1]?.headers).has('authorization')).toBe(false)
   })
 
-  it('does not refresh or retry a request cancelled while waiting for the lock', async () => {
+  it('applies configured headers only to the MCP endpoint', async () => {
+    const request = vi.fn<FetchLike>().mockResolvedValue(new Response(null, { status: 200 }))
+    const scoped = createMcpEndpointFetch(request, {
+      serverUrl: SERVER,
+      headers: { 'x-mcp-credential': 'configured-value' },
+    })
+
+    await scoped(SERVER, { headers: { accept: 'application/json' } })
+    await scoped(TOKEN_URL, { headers: { 'content-type': 'application/x-www-form-urlencoded' } })
+
+    const mcpHeaders = new Headers(request.mock.calls[0][1]?.headers)
+    expect(mcpHeaders.get('x-mcp-credential')).toBe('configured-value')
+    expect(mcpHeaders.get('accept')).toBe('application/json')
+    const oauthHeaders = new Headers(request.mock.calls[1][1]?.headers)
+    expect(oauthHeaders.has('x-mcp-credential')).toBe(false)
+    expect(oauthHeaders.get('content-type')).toBe('application/x-www-form-urlencoded')
+  })
+
+  it('rejects promptly without refreshing when cancelled while waiting for the lock', async () => {
     const entered = deferred()
     const finish = deferred()
     const holder = withMcpOauthRefreshLock('shared-grant', async () => {
@@ -464,12 +485,15 @@ describe('coordinated MCP OAuth with the real SDK and refresh mutex', () => {
     const abort = new AbortController()
     const call = fetchFor(grant, request)(SERVER, { signal: abort.signal })
     const outcome = expect(call).rejects.toThrow('cancelled')
-    await rejected.promise
-    abort.abort(new Error('cancelled'))
-    finish.resolve()
-    await holder
-    await outcome
-    expect(grant.tokenRequest).not.toHaveBeenCalled()
-    expect(request).toHaveBeenCalledTimes(1)
+    try {
+      await rejected.promise
+      abort.abort(new Error('cancelled'))
+      await outcome
+      expect(grant.tokenRequest).not.toHaveBeenCalled()
+      expect(request).toHaveBeenCalledTimes(1)
+    } finally {
+      finish.resolve()
+      await holder
+    }
   })
 })

apps/sim/lib/mcp/oauth/coordinated-fetch.ts

@@ -3,7 +3,7 @@ import {
   type OAuthClientProvider,
   UnauthorizedError,
 } from '@modelcontextprotocol/sdk/client/auth.js'
-import { createFetchWithInit, type FetchLike } from '@modelcontextprotocol/sdk/shared/transport.js'
+import type { FetchLike } from '@modelcontextprotocol/sdk/shared/transport.js'
 import { mcpAuthGuarded } from '@/lib/mcp/oauth/auth'
 import { withMcpOauthRefreshLock } from '@/lib/mcp/oauth/storage'
 
@@ -19,6 +19,23 @@ export interface McpOauthSession extends McpOauthCredentials {
   initialProvider: OAuthClientProvider
 }
 
+/** Applies configured headers only to the MCP endpoint, never to OAuth discovery or token URLs. */
+export function createMcpEndpointFetch(
+  fetchFn: FetchLike,
+  options: { serverUrl: string; headers?: HeadersInit }
+): FetchLike {
+  if (!options.headers) return fetchFn
+  const serverUrl = new URL(options.serverUrl)
+  const configuredHeaders = new Headers(options.headers)
+
+  return (input, init) => {
+    if (new URL(input).href !== serverUrl.href) return fetchFn(input, init)
+    const headers = new Headers(init?.headers)
+    configuredHeaders.forEach((value, key) => headers.set(key, value))
+    return fetchFn(input, { ...init, headers })
+  }
+}
+
 /**
  * Coordinates the SDK's public OAuth flow across clients without locking MCP requests.
  * The transport must omit authProvider so it cannot refresh outside this boundary.
@@ -27,10 +44,9 @@ export interface McpOauthSession extends McpOauthCredentials {
  */
 export function createCoordinatedMcpOauthFetch(
   { credentialId, loadProvider, initialProvider }: McpOauthSession,
-  options: { serverUrl: string; fetch: FetchLike; requestInit?: RequestInit }
+  options: { serverUrl: string; fetch: FetchLike }
 ): FetchLike {
   const serverUrl = new URL(options.serverUrl)
-  const authFetch = createFetchWithInit(options.fetch, options.requestInit)
   let currentProvider = initialProvider
 
   return async (input, init) => {
@@ -63,27 +79,31 @@ export function createCoordinatedMcpOauthFetch(
       if (authenticatedChallenges.has(challengeKey)) return response
 
       await response.body?.cancel()
-      await withMcpOauthRefreshLock(credentialId, async () => {
-        init?.signal?.throwIfAborted()
-        const current = await loadProvider()
-        const latestTokens = await current.tokens()
-        const refreshedElsewhere =
-          latestTokens && latestTokens.access_token !== tokens?.access_token
+      await withMcpOauthRefreshLock(
+        credentialId,
+        async () => {
+          init?.signal?.throwIfAborted()
+          const current = await loadProvider()
+          const latestTokens = await current.tokens()
+          const refreshedElsewhere =
+            latestTokens && latestTokens.access_token !== tokens?.access_token
 
-        init?.signal?.throwIfAborted()
-        if (!refreshedElsewhere) {
-          const result = await mcpAuthGuarded(current, {
-            serverUrl,
-            resourceMetadataUrl: challenge.resourceMetadataUrl,
-            scope: challenge.scope,
-            fetchFn: authFetch,
-          })
-          if (result !== 'AUTHORIZED') throw new UnauthorizedError()
-          authenticatedChallenges.add(challengeKey)
-        }
-        provider = current
-        currentProvider = current
-      })
+          init?.signal?.throwIfAborted()
+          if (!refreshedElsewhere) {
+            const result = await mcpAuthGuarded(current, {
+              serverUrl,
+              resourceMetadataUrl: challenge.resourceMetadataUrl,
+              scope: challenge.scope,
+              fetchFn: options.fetch,
+            })
+            if (result !== 'AUTHORIZED') throw new UnauthorizedError()
+            authenticatedChallenges.add(challengeKey)
+          }
+          provider = current
+          currentProvider = current
+        },
+        init?.signal ?? undefined
+      )
     }
   }
 }

apps/sim/lib/mcp/oauth/storage.test.ts

@@ -150,6 +150,20 @@ describe('withMcpOauthRefreshLock', () => {
     expect(fn).toHaveBeenCalledTimes(1)
   })
 
+  it('stops waiting for a cross-process lock when the caller aborts', async () => {
+    mockAcquireLock.mockResolvedValue(false)
+    const fn = vi.fn(async () => 'should-not-run')
+    const controller = new AbortController()
+    const pending = withMcpOauthRefreshLock('row-abort', fn, controller.signal)
+    const assertion = expect(pending).rejects.toThrow('cancelled')
+
+    await vi.waitFor(() => expect(mockAcquireLock).toHaveBeenCalled())
+    controller.abort(new Error('cancelled'))
+
+    await assertion
+    expect(fn).not.toHaveBeenCalled()
+  })
+
   it('falls open when Redis is unavailable on acquire', async () => {
     mockAcquireLock.mockRejectedValueOnce(new Error('Redis connection refused'))
     const fn = vi.fn(async () => 'uncoordinated')

apps/sim/lib/mcp/oauth/storage.ts

@@ -7,7 +7,7 @@ import { db } from '@sim/db'
 import { mcpServerOauth } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
+import { interruptibleSleep } from '@sim/utils/helpers'
 import { generateId, generateShortId } from '@sim/utils/id'
 import { and, eq, gt } from 'drizzle-orm'
 import { acquireLock, extendLock, releaseLock } from '@/lib/core/config/redis'
@@ -275,7 +275,12 @@ const REFRESH_QUEUE_WAIT_TIMEOUT_MS = 90_000
 
 const inflightChains = new Map<string, Promise<unknown>>()
 
-export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promise<T>): Promise<T> {
+export async function withMcpOauthRefreshLock<T>(
+  rowId: string,
+  fn: () => Promise<T>,
+  signal?: AbortSignal
+): Promise<T> {
+  signal?.throwIfAborted()
   const lockKey = `mcp:oauth:refresh:${rowId}`
   const prev = inflightChains.get(lockKey) ?? Promise.resolve()
   const prevSettled = prev.catch(() => undefined)
@@ -285,7 +290,8 @@ export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promis
     if (queueTimedOut) {
       throw new Error(`MCP OAuth refresh queue for ${rowId} abandoned after timeout`)
     }
-    return runWithRedisMutex(lockKey, rowId, fn)
+    signal?.throwIfAborted()
+    return runWithRedisMutex(lockKey, rowId, fn, signal)
   })
   inflightChains.set(lockKey, next)
   const cleanup = () => {
@@ -305,11 +311,25 @@ export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promis
     }, REFRESH_QUEUE_WAIT_TIMEOUT_MS)
     queueTimer.unref?.()
   })
+  let abortListener: (() => void) | undefined
+  const queueAbort = new Promise<never>((_resolve, reject) => {
+    if (!signal) return
+    abortListener = () => {
+      try {
+        signal.throwIfAborted()
+      } catch (error) {
+        reject(error)
+      }
+    }
+    signal.addEventListener('abort', abortListener, { once: true })
+    if (signal.aborted) abortListener()
+  })
 
   try {
-    await Promise.race([prevSettled, queueDeadline])
+    await Promise.race([prevSettled, queueDeadline, queueAbort])
   } finally {
     clearTimeout(queueTimer)
+    if (signal && abortListener) signal.removeEventListener('abort', abortListener)
   }
 
   return next
@@ -318,16 +338,19 @@ export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promis
 async function runWithRedisMutex<T>(
   lockKey: string,
   rowId: string,
-  fn: () => Promise<T>
+  fn: () => Promise<T>,
+  signal?: AbortSignal
 ): Promise<T> {
   const ownerToken = generateShortId()
   const deadline = Date.now() + REFRESH_MAX_WAIT_MS
 
   while (true) {
+    signal?.throwIfAborted()
     let acquired = false
     try {
       acquired = await acquireLock(lockKey, ownerToken, REFRESH_LOCK_TTL_SEC)
     } catch (error) {
+      signal?.throwIfAborted()
       logger.warn('Redis unavailable, running OAuth flow uncoordinated', {
         rowId,
         error: toError(error).message,
@@ -345,6 +368,7 @@ async function runWithRedisMutex<T>(
         })
       }, REFRESH_LOCK_EXTEND_INTERVAL_MS)
       try {
+        signal?.throwIfAborted()
         return await fn()
       } finally {
         clearInterval(watchdog)
@@ -357,6 +381,7 @@ async function runWithRedisMutex<T>(
       }
     }
 
+    signal?.throwIfAborted()
     if (Date.now() >= deadline) {
       // Lock still held by another process AND its watchdog is keeping it
       // alive — falling open would let us refresh concurrently and race the
@@ -367,6 +392,6 @@ async function runWithRedisMutex<T>(
         `MCP OAuth refresh lock for ${rowId} held longer than ${REFRESH_MAX_WAIT_MS}ms`
       )
     }
-    await sleep(REFRESH_POLL_INTERVAL_MS)
+    await interruptibleSleep(REFRESH_POLL_INTERVAL_MS, signal)
   }
 }