From 80261e52246094b0b2f9f5abef39911dc3d28a55 Mon Sep 17 00:00:00 2001 From: Robert Date: Mon, 1 Jun 2026 09:51:45 -0300 Subject: [PATCH] Coordinate a2amcp credential refresh --- package-lock.json | 4 +- package.json | 2 +- src/oauth.ts | 109 ++++++++++++++++++++++++++++++++++++++++++-- tests/oauth.test.ts | 86 ++++++++++++++++++++++++++++++++++ 4 files changed, 193 insertions(+), 8 deletions(-) diff --git a/package-lock.json b/package-lock.json index 2e84e86..84dcec8 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "a2amcp", - "version": "0.1.9", + "version": "0.1.10", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "a2amcp", - "version": "0.1.9", + "version": "0.1.10", "license": "Apache-2.0", "dependencies": { "@modelcontextprotocol/sdk": "^1.0.4", diff --git a/package.json b/package.json index 8ae9dd8..82852d2 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "a2amcp", - "version": "0.1.9", + "version": "0.1.10", "description": "MCP gateway for a2acloud agents. Run as a local MCP server; expose any number of deployed A2A agents to Claude Code, Cursor, and other MCP clients.", "type": "module", "bin": { diff --git a/src/oauth.ts b/src/oauth.ts index ade371a..85f7470 100644 --- a/src/oauth.ts +++ b/src/oauth.ts @@ -1,5 +1,8 @@ import crypto from "node:crypto"; +import { promises as fs } from "node:fs"; import http from "node:http"; +import os from "node:os"; +import path from "node:path"; import { spawn } from "node:child_process"; import type { AddressInfo } from "node:net"; @@ -7,6 +10,7 @@ import { ControlPlaneClient } from "./api.js"; import { DEFAULT_API_URL, type Credentials, + loadCredentials, saveCredentials, } from "./credentials.js"; @@ -15,6 +19,9 @@ export const DEFAULT_OAUTH_CLIENT_ID = "a2acloud-cli"; export const DEFAULT_OAUTH_SCOPE = "openid email offline_access mcp:invoke agent:read"; export const DEFAULT_REDIRECT_PORT = 41873; const CALLBACK_PATH = "/callback"; +const REFRESH_LOCK_WAIT_MS = 10_000; +const REFRESH_LOCK_STALE_MS = 30_000; +const REFRESH_LOCK_POLL_MS = 100; type OpenIdConfiguration = { authorization_endpoint: string; @@ -116,20 +123,40 @@ export async function refreshCredentialsIfNeeded( minTtlSeconds = 60, ): Promise { if (!creds.refreshToken || !creds.issuer || !creds.clientId) return creds; - if (creds.expiresAt && creds.expiresAt - nowSeconds() > minTtlSeconds) return creds; + if (credentialsFresh(creds, minTtlSeconds)) return creds; - const cfg = await discoverOpenIdConfiguration(creds.issuer); + return withRefreshLock(async () => { + const latest = await loadCredentials(); + if (latest && sameOAuthClient(latest, creds)) { + if (credentialsFresh(latest, minTtlSeconds)) return latest; + if (latest.refreshToken && latest.issuer && latest.clientId) creds = latest; + } + + return refreshCredentials(creds); + }); +} + +async function refreshCredentials(creds: Credentials): Promise { + const cfg = await discoverOpenIdConfiguration(creds.issuer!); const body = new URLSearchParams({ grant_type: "refresh_token", - client_id: creds.clientId, - refresh_token: creds.refreshToken, + client_id: creds.clientId!, + refresh_token: creds.refreshToken!, }); const resp = await fetch(cfg.token_endpoint, { method: "POST", headers: { "content-type": "application/x-www-form-urlencoded" }, body, }); - const token = await parseTokenResponse(resp); + let token: TokenResponse; + try { + token = await parseTokenResponse(resp); + } catch (err) { + const detail = err instanceof Error ? err.message : String(err); + throw new Error( + `Login expired or revoked. Run \`a2amcp login\` and retry. (${detail})`, + ); + } const next: Credentials = { ...creds, token: token.access_token!, @@ -141,6 +168,74 @@ export async function refreshCredentialsIfNeeded( return next; } +function credentialsFresh(creds: Credentials, minTtlSeconds: number): boolean { + return Boolean(creds.expiresAt && creds.expiresAt - nowSeconds() > minTtlSeconds); +} + +function sameOAuthClient(a: Credentials, b: Credentials): boolean { + return ( + a.apiUrl === b.apiUrl && + a.issuer === b.issuer && + a.clientId === b.clientId + ); +} + +async function withRefreshLock(fn: () => Promise): Promise { + const release = await acquireRefreshLock(); + try { + return await fn(); + } finally { + await release(); + } +} + +async function acquireRefreshLock(): Promise<() => Promise> { + const file = refreshLockFile(); + await fs.mkdir(path.dirname(file), { recursive: true }); + const started = Date.now(); + + while (true) { + try { + const handle = await fs.open(file, "wx", 0o600); + try { + await handle.writeFile( + JSON.stringify({ pid: process.pid, created_at: new Date().toISOString() }), + ); + } finally { + await handle.close(); + } + return async () => { + try { + await fs.unlink(file); + } catch (err: any) { + if (err?.code !== "ENOENT") throw err; + } + }; + } catch (err: any) { + if (err?.code !== "EEXIST") throw err; + await removeStaleRefreshLock(file); + if (Date.now() - started > REFRESH_LOCK_WAIT_MS) { + throw new Error("timed out waiting for credential refresh lock"); + } + await sleep(REFRESH_LOCK_POLL_MS); + } + } +} + +async function removeStaleRefreshLock(file: string): Promise { + try { + const stat = await fs.stat(file); + if (Date.now() - stat.mtimeMs <= REFRESH_LOCK_STALE_MS) return; + await fs.unlink(file); + } catch (err: any) { + if (err?.code !== "ENOENT") throw err; + } +} + +function refreshLockFile(): string { + return path.join(os.homedir(), ".a2a", "credentials-refresh.lock"); +} + export async function discoverOpenIdConfiguration( issuer: string, ): Promise { @@ -338,6 +433,10 @@ function nowSeconds(): number { return Math.floor(Date.now() / 1000); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function parseJsonObject(text: string): Record { if (!text.trim()) return {}; try { diff --git a/tests/oauth.test.ts b/tests/oauth.test.ts index d33a0de..c31de21 100644 --- a/tests/oauth.test.ts +++ b/tests/oauth.test.ts @@ -93,3 +93,89 @@ test("refreshCredentialsIfNeeded refreshes expired Keycloak access token", async assert.equal((await loadCredentials())!.token, "fresh-token"); assert.ok(calls.some((call) => call.body?.includes("grant_type=refresh_token"))); }); + +test("refreshCredentialsIfNeeded serializes concurrent refreshes", async () => { + let tokenCalls = 0; + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = String(input); + if (url.endsWith("/.well-known/openid-configuration")) { + return new Response( + JSON.stringify({ + authorization_endpoint: "https://auth.example.test/auth", + token_endpoint: "https://auth.example.test/token", + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + tokenCalls += 1; + await new Promise((resolve) => setTimeout(resolve, 25)); + return new Response( + JSON.stringify({ + access_token: `fresh-token-${tokenCalls}`, + refresh_token: `fresh-refresh-${tokenCalls}`, + expires_in: 3600, + scope: "openid email mcp:invoke", + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }) as typeof fetch; + + await saveCredentials({ + apiUrl: "https://api.example.test", + token: "old-token", + refreshToken: "refresh-token", + expiresAt: 1, + email: "dev@example.test", + issuer: "https://auth.example.test", + clientId: "a2acloud-cli", + }); + + const stale = (await loadCredentials())!; + const [first, second] = await Promise.all([ + refreshCredentialsIfNeeded(stale), + refreshCredentialsIfNeeded({ ...stale }), + ]); + + assert.equal(tokenCalls, 1); + assert.equal(first.token, "fresh-token-1"); + assert.equal(second.token, "fresh-token-1"); + assert.equal((await loadCredentials())!.refreshToken, "fresh-refresh-1"); +}); + +test("refreshCredentialsIfNeeded reports stale Keycloak sessions clearly", async () => { + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = String(input); + if (url.endsWith("/.well-known/openid-configuration")) { + return new Response( + JSON.stringify({ + authorization_endpoint: "https://auth.example.test/auth", + token_endpoint: "https://auth.example.test/token", + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + return new Response( + JSON.stringify({ + error: "invalid_grant", + error_description: "Session doesn't have required client", + }), + { status: 400, headers: { "content-type": "application/json" } }, + ); + }) as typeof fetch; + + await saveCredentials({ + apiUrl: "https://api.example.test", + token: "old-token", + refreshToken: "refresh-token", + expiresAt: 1, + email: "dev@example.test", + issuer: "https://auth.example.test", + clientId: "a2acloud-cli", + }); + + await assert.rejects( + refreshCredentialsIfNeeded((await loadCredentials())!), + /Login expired or revoked\. Run `a2amcp login` and retry\. .*Session doesn't have required client/, + ); + assert.equal((await loadCredentials())!.token, "old-token"); +});