mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-15 04:27:20 +03:00
fix(core): keep MCP OAuth credentials when a refresh fails temporarily (#63012)
This commit is contained in:
committed by
GitHub
Unverified
parent
dc4a1a74f6
commit
2a9a3ed97c
@@ -3,6 +3,8 @@ export * as McpClient from "./client.js"
|
||||
import { pathToFileURL } from "node:url"
|
||||
import {
|
||||
Client,
|
||||
OAuthError,
|
||||
OAuthErrorCode,
|
||||
ProtocolError,
|
||||
SdkError,
|
||||
SdkErrorCode,
|
||||
@@ -583,6 +585,8 @@ function repeats(error: unknown): boolean {
|
||||
(error instanceof SdkError && error.code === SdkErrorCode.RequestTimeout) ||
|
||||
// Refused credentials, even without OAuth: any connection with them would be refused the same way.
|
||||
(error instanceof SdkHttpError && (error.status === 401 || error.status === 403)) ||
|
||||
// The authorization server could not refresh the token for now; see McpOAuth.
|
||||
(error instanceof OAuthError && error.code === OAuthErrorCode.TemporarilyUnavailable) ||
|
||||
isTransient(error) ||
|
||||
PlatformError.isPlatformError(error)
|
||||
)
|
||||
|
||||
@@ -50,6 +50,16 @@ class RefreshRejected extends Data.TaggedError("McpOAuth.RefreshRejected")<{
|
||||
readonly body: string
|
||||
}> {}
|
||||
|
||||
// The SDK takes a refresh failure it cannot classify (network error, timeout, 5xx) as a lost login and
|
||||
// starts a new authorization, which a connection reports as needs_auth until the user signs in again.
|
||||
// temporarily_unavailable is an OAuth error the SDK surfaces as is, so the attempt fails and the
|
||||
// credential stays for the next one.
|
||||
const unavailable = (reason: string) =>
|
||||
Response.json(
|
||||
{ error: "temporarily_unavailable", error_description: `Could not refresh the OAuth token: ${reason}` },
|
||||
{ status: 503 },
|
||||
)
|
||||
|
||||
const refreshKey = (url: string | URL, init: RequestInit | undefined) => {
|
||||
if (!(init?.body instanceof URLSearchParams) || init.body.get("grant_type") !== "refresh_token") return undefined
|
||||
return [String(url), init.body.get("client_id") ?? "", init.body.get("refresh_token") ?? ""].join("\u0000")
|
||||
@@ -343,9 +353,12 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
|
||||
return current.value
|
||||
}
|
||||
// Bounds reading the body too, so a server that stalls after its headers cannot hold the lock.
|
||||
const { response, body } = yield* Effect.tryPromise(async (signal) => {
|
||||
const response = await send(url, { ...init, signal })
|
||||
return { response, body: await response.text() }
|
||||
const { response, body } = yield* Effect.tryPromise({
|
||||
try: async (signal) => {
|
||||
const response = await send(url, { ...init, signal })
|
||||
return { response, body: await response.text() }
|
||||
},
|
||||
catch: (error) => error,
|
||||
}).pipe(Effect.timeout("30 seconds"))
|
||||
const parsed = OAuthTokensSchema.safeParse(
|
||||
response.ok ? Option.getOrUndefined(decodeJson(body)) : undefined,
|
||||
@@ -377,11 +390,12 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
|
||||
: // The credential was signed out meanwhile; the SDK drops it and reports needs_auth.
|
||||
Response.json({ error: "invalid_grant" }, { status: 400 }),
|
||||
),
|
||||
Effect.catch((error) =>
|
||||
error instanceof RefreshRejected
|
||||
? Effect.succeed(new Response(error.body, { status: error.status, headers: error.headers }))
|
||||
: Effect.fail(error),
|
||||
),
|
||||
Effect.catch((error) => {
|
||||
if (!(error instanceof RefreshRejected))
|
||||
return Effect.succeed(unavailable(error instanceof Error ? error.message : String(error)))
|
||||
if (error.status === 429 || error.status >= 500) return Effect.succeed(unavailable(`HTTP ${error.status}`))
|
||||
return Effect.succeed(new Response(error.body, { status: error.status, headers: error.headers }))
|
||||
}),
|
||||
),
|
||||
)
|
||||
const result = provider({
|
||||
|
||||
@@ -9,6 +9,7 @@ import { Integration } from "@opencode/core/integration"
|
||||
import { McpClient } from "@opencode/core/mcp/client"
|
||||
import { McpOAuth } from "@opencode/core/mcp/oauth"
|
||||
import { Cause, Effect, Exit, ManagedRuntime } from "effect"
|
||||
import { TestClock } from "effect/testing"
|
||||
import { hostEnvironmentLayer } from "./fixture/environment"
|
||||
import { tmpdir } from "./fixture/tmpdir"
|
||||
|
||||
@@ -854,6 +855,152 @@ describe("MCP OAuth refresh across processes", () => {
|
||||
])
|
||||
})
|
||||
|
||||
describe("when the refresh fails", () => {
|
||||
// A server whose MCP endpoint accepts only the refreshed token and whose token endpoint the test switches.
|
||||
const guardedServer = () => {
|
||||
const state = {
|
||||
token: "ok" as "ok" | "unavailable" | "invalid_grant",
|
||||
tokenEndpoint: undefined as string | undefined,
|
||||
sent: [] as Array<string | null>,
|
||||
}
|
||||
const server = Bun.serve({
|
||||
port: 0,
|
||||
async fetch(request) {
|
||||
const url = new URL(request.url)
|
||||
if (url.pathname === "/.well-known/oauth-authorization-server")
|
||||
return Response.json({
|
||||
issuer: url.origin,
|
||||
authorization_endpoint: `${url.origin}/authorize`,
|
||||
token_endpoint: state.tokenEndpoint ?? `${url.origin}/token`,
|
||||
response_types_supported: ["code"],
|
||||
})
|
||||
if (request.method === "POST" && url.pathname === "/token") {
|
||||
state.sent.push(new URLSearchParams(await request.text()).get("refresh_token"))
|
||||
if (state.token === "unavailable") return new Response("upstream unavailable", { status: 503 })
|
||||
if (state.token === "invalid_grant") return Response.json({ error: "invalid_grant" }, { status: 400 })
|
||||
return Response.json({ access_token: "a2", token_type: "Bearer", refresh_token: "r2", expires_in: 3600 })
|
||||
}
|
||||
if (request.method === "POST" && url.pathname === "/mcp") {
|
||||
if (request.headers.get("authorization") !== "Bearer a2") return new Response(null, { status: 401 })
|
||||
const body = (await request.json()) as {
|
||||
id: number
|
||||
method: string
|
||||
params?: { protocolVersion?: string }
|
||||
}
|
||||
if (body.method !== "initialize") return new Response(null, { status: 202 })
|
||||
return Response.json({
|
||||
jsonrpc: "2.0",
|
||||
id: body.id,
|
||||
result: {
|
||||
protocolVersion: body.params?.protocolVersion,
|
||||
capabilities: {},
|
||||
serverInfo: { name: "test", version: "1" },
|
||||
},
|
||||
})
|
||||
}
|
||||
return new Response(null, { status: 404 })
|
||||
},
|
||||
})
|
||||
return { server, state, url: `${server.url.origin}/mcp` }
|
||||
}
|
||||
|
||||
const connect = async (runtime: ReturnType<typeof instance>, url: string) => {
|
||||
const current = await connection(runtime, url)
|
||||
return Effect.runPromise(
|
||||
Effect.scoped(
|
||||
McpClient.connect({
|
||||
server: "test",
|
||||
config: remote(url),
|
||||
directory: import.meta.dir,
|
||||
authProvider: current.provider,
|
||||
}),
|
||||
).pipe(Effect.provide(hostEnvironmentLayer), Effect.exit),
|
||||
)
|
||||
}
|
||||
|
||||
const failure = (exit: Awaited<ReturnType<typeof connect>>) => Exit.isFailure(exit) && Cause.squash(exit.cause)
|
||||
|
||||
test("fails without signing out when the token endpoint answers 503", async () => {
|
||||
const guarded = guardedServer()
|
||||
await using processes = await setup(guarded.url)
|
||||
guarded.state.token = "unavailable"
|
||||
|
||||
const failed = failure(await connect(processes.first, guarded.url))
|
||||
expect(failed).toBeInstanceOf(McpClient.ConnectError)
|
||||
expect(failed instanceof McpClient.ConnectError && [failed.message, failed.answered]).toEqual([
|
||||
"Could not refresh the OAuth token: HTTP 503 (temporarily_unavailable)",
|
||||
false,
|
||||
])
|
||||
expect(await stored(processes.second)).toEqual({ access: "a1", refresh: "r1" })
|
||||
|
||||
guarded.state.token = "ok"
|
||||
const exit = await connect(processes.first, guarded.url).finally(() => guarded.server.stop(true))
|
||||
expect(Exit.isSuccess(exit)).toBe(true)
|
||||
expect(guarded.state.sent).toEqual(["r1", "r1"])
|
||||
expect(await stored(processes.second)).toEqual({ access: "a2", refresh: "r2" })
|
||||
})
|
||||
|
||||
test("fails without signing out when the token endpoint is unreachable", async () => {
|
||||
const guarded = guardedServer()
|
||||
const closed = Bun.serve({ port: 0, fetch: () => new Response(null, { status: 404 }) })
|
||||
closed.stop(true)
|
||||
guarded.state.tokenEndpoint = `${closed.url.origin}/token`
|
||||
await using processes = await setup(guarded.url)
|
||||
|
||||
const failed = failure(await connect(processes.first, guarded.url))
|
||||
expect(failed).toBeInstanceOf(McpClient.ConnectError)
|
||||
expect(failed instanceof McpClient.ConnectError && failed.answered).toBe(false)
|
||||
expect(await stored(processes.second)).toEqual({ access: "a1", refresh: "r1" })
|
||||
|
||||
guarded.state.tokenEndpoint = undefined
|
||||
const exit = await connect(processes.first, guarded.url).finally(() => guarded.server.stop(true))
|
||||
expect(Exit.isSuccess(exit)).toBe(true)
|
||||
})
|
||||
|
||||
test("signs out when the token endpoint answers invalid_grant", async () => {
|
||||
const guarded = guardedServer()
|
||||
guarded.state.token = "invalid_grant"
|
||||
await using processes = await setup(guarded.url)
|
||||
|
||||
const failed = failure(await connect(processes.first, guarded.url))
|
||||
guarded.server.stop(true)
|
||||
expect(failed).toBeInstanceOf(McpClient.NeedsAuthError)
|
||||
expect(await stored(processes.second)).toBeUndefined()
|
||||
})
|
||||
|
||||
test("keeps the credential when the refresh times out", async () => {
|
||||
const received = Promise.withResolvers<void>()
|
||||
const hung = Bun.serve({
|
||||
port: 0,
|
||||
fetch: () => {
|
||||
received.resolve()
|
||||
return new Promise<Response>(() => {})
|
||||
},
|
||||
})
|
||||
await using processes = await setup(hung.url.href)
|
||||
|
||||
const error = await processes.first
|
||||
.runPromise(
|
||||
Effect.gen(function* () {
|
||||
const credentials = yield* Credential.Service
|
||||
const provider = yield* McpOAuth.connectProvider({
|
||||
config: remote(hung.url.href),
|
||||
credential: yield* credentials.get(id),
|
||||
})
|
||||
const fetchFn = yield* McpOAuth.loggedFetch({ server: "test" }, provider)
|
||||
const pending = refresh(hung, fetchFn, "r1").catch((error: unknown) => error)
|
||||
yield* Effect.promise(() => received.promise)
|
||||
yield* TestClock.adjust("30 seconds")
|
||||
return yield* Effect.promise(() => pending)
|
||||
}).pipe(Effect.provide(TestClock.layer())),
|
||||
)
|
||||
.finally(() => hung.stop(true))
|
||||
|
||||
expect(error instanceof OAuthError && error.code).toBe("temporarily_unavailable")
|
||||
expect(await stored(processes.second)).toEqual({ access: "a1", refresh: "r1" })
|
||||
})
|
||||
})
|
||||
|
||||
test("hands a rejected refresh back to the SDK and keeps the credential", async () => {
|
||||
const server = Bun.serve({
|
||||
port: 0,
|
||||
@@ -895,8 +1042,7 @@ describe("MCP OAuth refresh across processes", () => {
|
||||
const current = await connection(processes.first, endpoint.server.url.href)
|
||||
|
||||
const error = await refresh(unreachable, current.fetchFn, "r1").catch((error: unknown) => error)
|
||||
expect(error).toBeInstanceOf(Error)
|
||||
expect(error).not.toBeInstanceOf(OAuthError)
|
||||
expect(error instanceof OAuthError && error.code).toBe("temporarily_unavailable")
|
||||
expect(await stored(processes.first)).toEqual({ access: "a1", refresh: "r1" })
|
||||
|
||||
const tokens = await refresh(endpoint.server, current.fetchFn, "r1").finally(() => endpoint.server.stop(true))
|
||||
|
||||
Reference in New Issue
Block a user