github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

fix: Refresh Agent MCP OAuth credentials (no-changelog) (#37828)

Browse Source
This commit is contained in:
Michael Drury 2026-09-07 09:27:56 +00:00 committed by GitHub
parent bb5b910ea1
commit a0df5b34aa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 954 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
packages/@n8n/agents/src/__tests__/integration/mcp-client-boundary.test.ts
View File

@ -43,7 +43,7 @@ describe('McpClient.listTools()', () => {
expect(tools).toHaveLength(0);
});
it('skips an unreachable server, records the failure, and keeps the cache', async () => {
it('skips an unreachable server and records the failure', async () => {
const client = new McpClient([{ name: 'dead', url: 'http://127.0.0.1:1/sse' }]);
const tools = await client.listTools();

21
packages/@n8n/agents/src/runtime/__tests__/mcp-connection.test.ts
View File

@ -234,6 +234,27 @@ describe('McpClient — connection failure handling', () => {
// The transport that opened before listTools threw was closed.
expect(clientClose).toHaveBeenCalledTimes(1);
});
it('retries a failed server on the next tool-list request', async () => {
clientConnect
.mockRejectedValueOnce(new Error('temporary failure'))
.mockResolvedValue(undefined);
clientListTools.mockResolvedValue({
tools: [{ name: 'echo', description: '', inputSchema: { type: 'object' } }],
});
clientClose.mockResolvedValue(undefined);
const client = new McpClient([
{ name: 'flaky', url: 'https://example.test/mcp', transport: 'streamableHttp' },
]);
expect(await client.listTools()).toEqual([]);
const tools = await client.listTools();
expect(tools.map((tool) => tool.name)).toEqual(['flaky_echo']);
expect(clientConnect).toHaveBeenCalledTimes(2);
expect(client.getConnectionFailures()).toEqual([]);
});
});
describe('McpClient — tool name normalization', () => {

40
packages/@n8n/agents/src/sdk/__tests__/agent-isolated-runtime.test.ts
View File

@ -6,6 +6,7 @@ import type { AgentEventBus } from '../../runtime/state/event-bus';
import { AgentEvent } from '../../runtime/state/event-bus';
import type { StreamChunk } from '../../types';
import { Agent } from '../agent';
import { McpClient } from '../mcp-client';
vi.mock('@ai-sdk/openai', () => ({
createOpenAI: () => () => ({ provider: 'openai', modelId: 'mock', specificationVersion: 'v3' }),
@ -87,6 +88,45 @@ describe('Agent isolated runtimes', () => {
);
});
it('rebuilds before the next run after an MCP connection failure', async () => {
generateText.mockResolvedValue(makeGenerateSuccess('ok'));
const client = new McpClient([]);
const listTools = vi.spyOn(client, 'listTools').mockResolvedValue([]);
vi.spyOn(client, 'getConnectionFailures')
.mockReturnValueOnce([{ server: 'linear', error: 'temporary failure' }])
.mockReturnValue([]);
const agent = new Agent('agent').model('openai/gpt-4o-mini').instructions('test').mcp(client);
await agent.generate('first');
await agent.generate('second');
expect(listTools).toHaveBeenCalledTimes(2);
});
it('keeps in-memory checkpoints when an MCP connection failure triggers a rebuild', async () => {
const client = new McpClient([]);
const listTools = vi.spyOn(client, 'listTools').mockResolvedValue([]);
vi.spyOn(client, 'getConnectionFailures')
.mockReturnValueOnce([{ server: 'linear', error: 'temporary failure' }])
.mockReturnValue([]);
const agent = new Agent('agent')
.model('openai/gpt-4o-mini')
.instructions('test')
.mcp(client)
.checkpoint('memory');
const internals = agent as unknown as AgentInternals;
const firstConfig = await internals.ensureBuilt();
await firstConfig.runState?.suspend('run-1', { status: 'suspended' } as never);
const secondConfig = await internals.ensureBuilt();
expect(await secondConfig.runState?.resume('run-1')).toEqual(
expect.objectContaining({ status: 'suspended' }),
);
expect(secondConfig.runState).toBe(firstConfig.runState);
expect(listTools).toHaveBeenCalledTimes(2);
});
it('applies event handler changes to active runtimes', async () => {
const agent = new Agent('agent').model('openai/gpt-4o-mini').instructions('test');
const internals = agent as unknown as AgentInternals;

24
packages/@n8n/agents/src/sdk/agent.ts
View File

@ -199,6 +199,8 @@ export class Agent implements BuiltAgent, AgentBuilder {
private checkpointStore?: 'memory' | CheckpointStore;
private runStateManager?: RunStateManager;
private thinkingConfig?: ThinkingConfig;
private reasoningLevel?: ReasoningLevel;
@ -430,6 +432,9 @@ export class Agent implements BuiltAgent, AgentBuilder {
* ```
*/
checkpoint(storage: 'memory' | CheckpointStore): this {
if (this.checkpointStore !== storage) {
this.runStateManager = undefined;
}
this.checkpointStore = storage;
return this;
}
@ -874,16 +879,23 @@ export class Agent implements BuiltAgent, AgentBuilder {
/**
* @internal Lazy-build the agent on first use. Stores the promise so
* concurrent callers share one build operation. On error the promise is
* cleared so the caller can retry.
* concurrent callers share one build operation. An error clears the promise.
* An MCP connection failure also clears it so the next run can retry discovery.
*/
private async ensureBuilt(): Promise<AgentRuntimeConfig> {
if (!this.buildPromise) {
const p = this.build();
this.buildPromise = p;
p.catch(() => {
if (this.buildPromise === p) this.buildPromise = undefined;
});
void p.then(
(config) => {
if (config.mcpConnectionFailures?.length && this.buildPromise === p) {
this.buildPromise = undefined;
}
},
() => {
if (this.buildPromise === p) this.buildPromise = undefined;
},
);
}
return await this.buildPromise;
}
@ -1071,7 +1083,7 @@ export class Agent implements BuiltAgent, AgentBuilder {
finalDeferredTools.length > 0 && this.deferredToolSearchTopK !== undefined
? { topK: this.deferredToolSearchTopK }
: undefined;
const runState = new RunStateManager(this.checkpointStore);
const runState = (this.runStateManager ??= new RunStateManager(this.checkpointStore));
allTools = this.completeInlineDelegateTools(allTools, {
deferredTools: finalDeferredTools,

23
packages/@n8n/agents/src/sdk/mcp-client.ts
View File

@ -102,20 +102,27 @@ export class McpClient {
/**
* Connect to all servers (if not already connected) and return the full
* flat list of tools. Subsequent calls return the cached list without
* additional network round-trips. Servers that fail to connect are
* flat list of tools. Subsequent calls return the cached list when all
* servers connect. Servers that fail to connect are
* skipped (their tools are omitted and the failure is recorded via
* `getConnectionFailures()`); the call still resolves with the remaining
* servers' tools. On a hard error (e.g. tool-name collision) the cache is
* cleared so the caller can retry.
* `getConnectionFailures()`). The call still resolves with the remaining
* servers' tools. A failed server retries on the next call. A hard error
* (for example, a tool-name collision) also clears the cache.
*/
async listTools(): Promise<BuiltTool[]> {
if (!this.listToolsPromise) {
const p = this.doListTools();
this.listToolsPromise = p;
p.catch(() => {
if (this.listToolsPromise === p) this.listToolsPromise = undefined;
});
void p.then(
() => {
if (this.connectionFailures.length > 0 && this.listToolsPromise === p) {
this.listToolsPromise = undefined;
}
},
() => {
if (this.listToolsPromise === p) this.listToolsPromise = undefined;
},
);
}
return await this.listToolsPromise;
}

53
packages/@n8n/ai-utilities/src/__tests__/utils/refreshing-auth-fetch.test.ts
View File

@ -259,4 +259,57 @@ describe('createRefreshingAuthFetch', () => {
]);
});
});
describe('proactive refresh', () => {
it('refreshes before sending a request when the token is close to expiry', async () => {
const baseFetch = vi.fn().mockResolvedValue(new Response('ok'));
let refreshDue = true;
const refreshHeaders = vi.fn().mockImplementation(async () => {
refreshDue = false;
return { Authorization: 'Bearer fresh' };
});
const fetchWithAuth = createRefreshingAuthFetch({
baseFetch,
initialHeaders: { Authorization: 'Bearer stale' },
refreshHeaders,
shouldRefresh: () => refreshDue,
});
const response = await fetchWithAuth('https://example.com/mcp');
expect(response.status).toBe(200);
expect(refreshHeaders).toHaveBeenCalledTimes(1);
expect(baseFetch).toHaveBeenCalledTimes(1);
expect(authorizationOf(baseFetch.mock.calls[0])).toBe('Bearer fresh');
});
it('shares one proactive refresh between concurrent requests', async () => {
const baseFetch = vi.fn().mockResolvedValue(new Response('ok'));
let finishRefresh: ((headers: HeadersInit) => void) | undefined;
const refreshHeaders = vi.fn().mockImplementation(
async () =>
await new Promise<HeadersInit>((resolve) => {
finishRefresh = resolve;
}),
);
const fetchWithAuth = createRefreshingAuthFetch({
baseFetch,
initialHeaders: { Authorization: 'Bearer stale' },
refreshHeaders,
shouldRefresh: () => true,
});
const requests = [
fetchWithAuth('https://example.com/first'),
fetchWithAuth('https://example.com/second'),
];
await vi.waitFor(() => expect(refreshHeaders).toHaveBeenCalledTimes(1));
finishRefresh?.({ Authorization: 'Bearer fresh' });
await Promise.all(requests);
expect(refreshHeaders).toHaveBeenCalledTimes(1);
expect(baseFetch).toHaveBeenCalledTimes(2);
expect(baseFetch.mock.calls.map(authorizationOf)).toEqual(['Bearer fresh', 'Bearer fresh']);
});
});
});

6
packages/@n8n/ai-utilities/src/utils/refreshing-auth-fetch.ts
View File

@ -4,6 +4,7 @@ export interface RefreshingAuthFetchOptions {
baseFetch: typeof fetch;
initialHeaders?: HeadersInit;
refreshHeaders?: (current: Headers) => Promise<HeadersInit | null>;
shouldRefresh?: () => boolean;
assertAllowedUrl?: (url: string) => void | Promise<void>;
}
@ -21,6 +22,7 @@ export function createRefreshingAuthFetch({
baseFetch,
initialHeaders,
refreshHeaders,
shouldRefresh,
assertAllowedUrl,
}: RefreshingAuthFetchOptions): typeof fetch {
let authHeaders = new Headers(initialHeaders);
@ -51,6 +53,10 @@ export function createRefreshingAuthFetch({
requestInput: RequestInfo | URL,
requestInit?: RequestInit,
): Promise<Response> => {
if (sendAuth && refreshHeaders && shouldRefresh?.()) {
await refresh();
}
const requestAuthVersion = authVersion;
const execute = async () =>
await baseFetch(requestInput instanceof Request ? requestInput.clone() : requestInput, {

4
packages/@n8n/backend-common/src/index.ts
View File

@ -20,6 +20,10 @@ export { CliParser } from './cli-parser';
export { TypedEmitter } from './typed-emitter';
export { LockService } from './locking/lock.service';
export {
SingleFlightLease,
type SingleFlightLeaseOptions,
} from './locking/single-flight-lease';
export {
type ILockService,
LockNamespace,

94
packages/@n8n/backend-common/src/locking/__tests__/single-flight-lease.test.ts Normal file
View File

@ -0,0 +1,94 @@
import { mock } from 'vitest-mock-extended';
import { InProcessLockService } from '../in-process-lock.service';
import {
LockAcquisitionTimeoutError,
type ILockService,
LockNamespace,
} from '../lock-service.interface';
import { LockService } from '../lock.service';
import { SingleFlightLease } from '../single-flight-lease';
describe('SingleFlightLease', () => {
it('serializes same-key work across coordinators', async () => {
const lockService = new LockService(new InProcessLockService());
const firstCoordinator = new SingleFlightLease<string>();
const secondCoordinator = new SingleFlightLease<string>();
let finishFirst: (() => void) | undefined;
const firstOperation = vi.fn(
async () =>
await new Promise<string>((resolve) => {
finishFirst = () => resolve('first');
}),
);
const secondOperation = vi.fn().mockResolvedValue('second');
const options = { lockService, namespace: LockNamespace.CREDENTIALS };
const first = firstCoordinator.run('credential', firstOperation, options);
const second = secondCoordinator.run('credential', secondOperation, options);
await new Promise((resolve) => setImmediate(resolve));
expect(firstOperation).toHaveBeenCalledTimes(1);
expect(secondOperation).not.toHaveBeenCalled();
finishFirst?.();
await expect(Promise.all([first, second])).resolves.toEqual(['first', 'second']);
});
it('does not coalesce the same key across namespaces', async () => {
const lockService = new LockService(new InProcessLockService());
const coordinator = new SingleFlightLease<string>();
let finishFirst: (() => void) | undefined;
const firstOperation = vi.fn(
async () =>
await new Promise<string>((resolve) => {
finishFirst = () => resolve('first');
}),
);
const secondOperation = vi.fn().mockResolvedValue('second');
const first = coordinator.run('shared-key', firstOperation, {
lockService,
namespace: LockNamespace.CREDENTIALS,
});
const second = coordinator.run('shared-key', secondOperation, {
lockService,
namespace: LockNamespace.KNOWN_LOCKS,
});
await new Promise((resolve) => setImmediate(resolve));
finishFirst?.();
await expect(Promise.all([first, second])).resolves.toEqual(['first', 'second']);
expect(firstOperation).toHaveBeenCalledTimes(1);
expect(secondOperation).toHaveBeenCalledTimes(1);
});
it('runs work without a lease when lease acquisition times out', async () => {
const lockService = mock<ILockService>();
const timeout = new LockAcquisitionTimeoutError('Timed out waiting for lock');
lockService.withLease.mockRejectedValueOnce(timeout);
const coordinator = new SingleFlightLease<string>();
const operation = vi.fn().mockResolvedValue('result');
const onLeaseTimeout = vi.fn();
await expect(
coordinator.run('credential', operation, {
lockService,
namespace: LockNamespace.CREDENTIALS,
onLeaseTimeout,
}),
).resolves.toBe('result');
expect(operation).toHaveBeenCalledTimes(1);
expect(onLeaseTimeout).toHaveBeenCalledWith(timeout);
lockService.withLease.mockImplementationOnce(
async (_namespace, _key, fn) => await fn(new AbortController().signal),
);
await expect(
coordinator.run('credential', operation, {
lockService,
namespace: LockNamespace.CREDENTIALS,
}),
).resolves.toBe('result');
expect(operation).toHaveBeenCalledTimes(2);
});
});

42
packages/@n8n/backend-common/src/locking/single-flight-lease.ts Normal file
View File

@ -0,0 +1,42 @@
import {
LockAcquisitionTimeoutError,
type ILockService,
type LockNamespace,
} from './lock-service.interface';
export interface SingleFlightLeaseOptions {
lockService: ILockService;
namespace: LockNamespace;
waitTimeoutMs?: number;
leaseTtlMs?: number;
/** Called before `fn` runs without a lease after an acquisition timeout. */
onLeaseTimeout?: (error: LockAcquisitionTimeoutError) => void;
}
/** Coalesce same-key work in this process and coordinate it across processes with a lease. */
export class SingleFlightLease<T> {
private readonly inFlight = new Map<string, Promise<T>>();
async run(
key: string,
fn: (signal: AbortSignal) => Promise<T>,
options: SingleFlightLeaseOptions,
): Promise<T> {
const { lockService, namespace, waitTimeoutMs, leaseTtlMs, onLeaseTimeout } = options;
const inFlightKey = `${namespace}:${key}`;
const inFlight = this.inFlight.get(inFlightKey);
if (inFlight) return await inFlight;
const promise = lockService
.withLease(namespace, key, fn, { waitTimeoutMs, leaseTtlMs })
.catch(async (error: unknown) => {
if (!(error instanceof LockAcquisitionTimeoutError)) throw error;
onLeaseTimeout?.(error);
return await fn(new AbortController().signal);
})
.finally(() => this.inFlight.delete(inFlightKey));
this.inFlight.set(inFlightKey, promise);
return await promise;
}
}

20
packages/@n8n/nodes-langchain/nodes/mcp/shared/__test__/utils.test.ts
View File

@ -225,6 +225,26 @@ describe('utils', () => {
expect(ctx.helpers.refreshOAuth2Token).toHaveBeenCalledWith('mcpOAuth2Api');
});
it('should refresh an expiring client credentials token without a refresh token', async () => {
const now = 1_700_000_000_000;
vi.spyOn(Date, 'now').mockReturnValue(now);
const ctx = mockDeep<IExecuteFunctions>();
ctx.getCredentials.mockResolvedValue({
grantType: 'clientCredentials',
oauthTokenData: {
access_token: 'access-token',
expires_in: 3600,
n8n_expires_at: String(now + 60_000),
},
});
ctx.helpers.refreshOAuth2Token.mockResolvedValue({ access_token: 'new-access-token' });
const result = await getAuthHeaders(ctx, 'mcpOAuth2Api');
expect(result.headers).toEqual({ Authorization: 'Bearer new-access-token' });
expect(ctx.helpers.refreshOAuth2Token).toHaveBeenCalledWith('mcpOAuth2Api');
});
it('should not refresh mcpOAuth2Api credentials when the access token is still valid', async () => {
const now = 1_700_000_000_000;
vi.spyOn(Date, 'now').mockReturnValue(now);

31
packages/@n8n/nodes-langchain/nodes/mcp/shared/utils.ts
View File

@ -21,6 +21,7 @@ import {
assertUrlAllowed,
getMcpAuthHeaders,
NodeOperationError,
shouldRefreshMcpOAuth2Token,
} from 'n8n-workflow';
import {
@ -82,13 +83,6 @@ type OnUnauthorizedHandler = (
headers?: Record<string, string>,
) => Promise<Record<string, string> | null>;
const OAUTH2_REFRESH_BUFFER_MS = 2 * 60 * 1000;
const OAUTH2_REFRESH_BUFFER_RATIO = 0.1;
type McpOAuth2Credentials = ICredentialDataDecryptedObject & {
oauthTokenData?: ClientOAuth2TokenData;
};
type ConnectMcpClientError =
| { type: 'invalid_url'; error: Error }
| { type: 'connection'; error: Error }
@ -326,24 +320,6 @@ function createAuthFetch(
});
}
function shouldRefreshOAuth2Token(credentials: McpOAuth2Credentials): boolean {
const tokenData = credentials.oauthTokenData;
if (!tokenData?.refresh_token) return false;
const expiresAt = Number(tokenData.n8n_expires_at);
if (!Number.isFinite(expiresAt)) {
return false;
}
const expiresInMs = Number(tokenData.expires_in) * 1000;
const refreshBufferMs =
Number.isFinite(expiresInMs) && expiresInMs > 0
? Math.min(OAUTH2_REFRESH_BUFFER_MS, expiresInMs * OAUTH2_REFRESH_BUFFER_RATIO)
: OAUTH2_REFRESH_BUFFER_MS;
return Date.now() + refreshBufferMs >= expiresAt;
}
export async function getAuthHeaders(
ctx: IExecuteFunctions | ISupplyDataFunctions | ILoadOptionsFunctions,
authentication: McpAuthenticationOption,
@ -371,7 +347,10 @@ export async function getAuthHeaders(
.catch(() => null);
if (!credentials) return {};
if (isMcpOAuth2Authentication(authentication) && shouldRefreshOAuth2Token(credentials)) {
if (
isMcpOAuth2Authentication(authentication) &&
shouldRefreshMcpOAuth2Token(credentials.oauthTokenData, credentials.grantType)
) {
const refreshedHeaders = await tryRefreshOAuth2Token(ctx, authentication);
if (refreshedHeaders) return { headers: refreshedHeaders, credentials };
}

83
packages/cli/src/modules/agents/json-config/__tests__/mcp-client-factory.test.ts
View File

@ -176,7 +176,7 @@ describe('buildMcpClientForServer — OAuth2 refresh on 401', () => {
const oauthService = mock<OauthService>();
oauthService.refreshOAuth2CredentialById.mockResolvedValue({
Authorization: 'Bearer fresh-token',
headers: { Authorization: 'Bearer fresh-token' },
});
await buildMcpClientForServer(
@ -189,7 +189,9 @@ describe('buildMcpClientForServer — OAuth2 refresh on 401', () => {
const res = await fetchFn('https://example.test/mcp');
expect(res.status).toBe(200);
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith('cred-1', 'proj-1');
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith('cred-1', 'proj-1', {
accessToken: 'stale-token',
});
// First call uses the stale header, second uses the refreshed one.
const [, firstInit] = proxyFetchMock.mock.calls[0] as [unknown, RequestInit];
const [, secondInit] = proxyFetchMock.mock.calls[1] as [unknown, RequestInit];
@ -197,6 +199,77 @@ describe('buildMcpClientForServer — OAuth2 refresh on 401', () => {
expect(new Headers(secondInit.headers).get('authorization')).toBe('Bearer fresh-token');
});
it('refreshes an expiring OAuth2 token before the first request', async () => {
proxyFetchMock.mockResolvedValue(makeOk());
const credentialProvider = mock<CredentialProvider>();
credentialProvider.resolve.mockResolvedValue({
oauthTokenData: {
access_token: 'stale-token',
refresh_token: 'refresh-token',
expires_in: 3600,
n8n_expires_at: String(Date.now() + 60_000),
},
} as never);
const oauthService = mock<OauthService>();
oauthService.refreshOAuth2CredentialById.mockResolvedValue({
headers: { Authorization: 'Bearer fresh-token' },
expiresAt: Date.now() + 60_000,
expiresInSeconds: 60,
});
await buildMcpClientForServer(
makeServer({ authentication: 'mcpOAuth2Api', credential: 'cred-1' }),
{ credentialProvider, oauthService, projectId: 'proj-1', proxyFetch },
);
const [configs] = mcpClientCtor.mock.calls[0] as [Array<{ fetch: typeof fetch }>];
await configs[0].fetch('https://example.test/mcp');
await configs[0].fetch('https://example.test/mcp');
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledTimes(1);
expect(proxyFetchMock).toHaveBeenCalledTimes(2);
for (const [, init] of proxyFetchMock.mock.calls as Array<[unknown, RequestInit]>) {
expect(new Headers(init.headers).get('authorization')).toBe('Bearer fresh-token');
}
});
it('refreshes an expiring client credentials token before the first request', async () => {
proxyFetchMock.mockResolvedValue(makeOk());
const expiresAt = Date.now() + 60_000;
const credentialProvider = mock<CredentialProvider>();
credentialProvider.resolve.mockResolvedValue({
grantType: 'clientCredentials',
oauthTokenData: {
access_token: 'stale-token',
expires_in: 3600,
n8n_expires_at: String(expiresAt),
},
} as never);
const oauthService = mock<OauthService>();
oauthService.refreshOAuth2CredentialById.mockResolvedValue({
headers: { Authorization: 'Bearer fresh-token' },
expiresAt: Date.now() + 3_600_000,
expiresInSeconds: 3600,
});
await buildMcpClientForServer(
makeServer({ authentication: 'mcpOAuth2Api', credential: 'cred-1' }),
{ credentialProvider, oauthService, projectId: 'proj-1', proxyFetch },
);
const [configs] = mcpClientCtor.mock.calls[0] as [Array<{ fetch: typeof fetch }>];
await configs[0].fetch('https://example.test/mcp');
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith('cred-1', 'proj-1', {
accessToken: 'stale-token',
expiresAt,
});
expect(proxyFetchMock).toHaveBeenCalledTimes(1);
});
it('does NOT call refreshOAuth2CredentialById for non-OAuth2 auth schemes', async () => {
proxyFetchMock.mockResolvedValueOnce(make401());
@ -428,7 +501,7 @@ describe('buildMcpClientForServer — service-specific McpOAuth2Api refresh', ()
const oauthService = mock<OauthService>();
oauthService.refreshOAuth2CredentialById.mockResolvedValue({
Authorization: 'Bearer fresh',
headers: { Authorization: 'Bearer fresh' },
});
await buildMcpClientForServer(
@ -457,7 +530,9 @@ describe('buildMcpClientForServer — service-specific McpOAuth2Api refresh', ()
const res = await configs[0].fetch('https://example.test/mcp');
expect(res.status).toBe(200);
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith('cred-1', 'proj-1');
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith('cred-1', 'proj-1', {
accessToken: 'stale',
});
});
it('does NOT wire an onUnauthorized handler when credential is absent for mcpOAuth2Api', async () => {

52
packages/cli/src/modules/agents/json-config/mcp-client-factory.ts
View File

@ -2,7 +2,13 @@ import type { CredentialProvider, McpClient, McpServerConfig } from '@n8n/agents
import type { AgentJsonMcpServerConfig } from '@n8n/api-types';
import type { CustomFetch } from '@n8n/backend-network';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import { getMcpAuthHeaders, isMcpOAuth2Authentication, OperationalError } from 'n8n-workflow';
import { isRecord } from '@n8n/utils/is-record';
import {
getMcpAuthHeaders,
isMcpOAuth2Authentication,
OperationalError,
shouldRefreshMcpOAuth2Token,
} from 'n8n-workflow';
import type { ICredentialDataDecryptedObject, McpRegistryConnection } from 'n8n-workflow';
import {
@ -13,6 +19,7 @@ import type { OauthService } from '@/oauth/oauth.service';
import {
type AuthFetchDomainPolicy,
createAuthFetch,
getBearerTokenRevision,
resolveAllowedDomains,
} from '@/utils/auth-fetch';
@ -47,8 +54,8 @@ type DerivedAuth = {
* have to depend on `@n8n/nodes-langchain`.
*
* For any supported OAuth2 credential type, the Bearer header is computed from
* the already-stored `oauthTokenData.access_token`. Refresh-on-401 is handled
* by `createAuthFetch` below; this function only computes the initial set.
* the already-stored `oauthTokenData.access_token`. `createAuthFetch` refreshes
* the token before expiry and after a 401 response.
*/
async function deriveAuthHeaders(
server: AgentJsonMcpServerConfig,
@ -107,7 +114,7 @@ export interface BuildMcpClientDeps {
credentialProvider: CredentialProvider;
resolveRegistryConnection?: (nodeTypeName: string) => Promise<McpRegistryConnection | undefined>;
/**
* Used to refresh OAuth2 tokens on a 401 response without an
* Used to refresh OAuth2 tokens before expiry or after a 401 response without an
* `IExecuteFunctions` workflow context. Only invoked when
* `server.authentication` is a supported OAuth2 credential type.
*/
@ -193,14 +200,36 @@ export async function buildMcpClientForServer(
}
}
const onUnauthorized =
const oauthTokenData = isRecord(credentialData?.oauthTokenData)
? { ...credentialData.oauthTokenData }
: undefined;
const grantType = credentialData?.grantType;
const refreshAuthHeaders =
isMcpOAuth2Authentication(server.authentication) && server.credential
? async () => {
? async (currentHeaders: Record<string, string>) => {
const credentialId = server.credential;
if (!credentialId) return null;
return await oauthService
.refreshOAuth2CredentialById(credentialId, projectId)
.catch(() => null);
const result = await oauthService.refreshOAuth2CredentialById(
credentialId,
projectId,
getBearerTokenRevision(currentHeaders, oauthTokenData?.n8n_expires_at),
);
if (!result) return null;
if (oauthTokenData) {
if (result.expiresAt === undefined) {
delete oauthTokenData.n8n_expires_at;
} else {
oauthTokenData.n8n_expires_at = String(result.expiresAt);
}
if (result.expiresInSeconds === undefined) {
delete oauthTokenData.expires_in;
} else {
oauthTokenData.expires_in = result.expiresInSeconds;
}
}
return result.headers;
}
: undefined;
@ -231,7 +260,10 @@ export async function buildMcpClientForServer(
fetch: createAuthFetch({
baseFetch: proxyFetch,
initialHeaders,
onUnauthorized,
onUnauthorized: refreshAuthHeaders,
...(refreshAuthHeaders
? { shouldRefresh: () => shouldRefreshMcpOAuth2Token(oauthTokenData, grantType) }
: {}),
allowedDomains,
}),
};

4
packages/cli/src/modules/instance-ai/mcp/__tests__/instance-ai-mcp-registry.service.test.ts
View File

@ -356,7 +356,7 @@ describe('InstanceAiMcpRegistryService', () => {
.mockResolvedValueOnce(new Response('unauthorized', { status: 401 }))
.mockResolvedValueOnce(new Response('ok', { status: 200 }));
oauthService.refreshOAuth2CredentialById.mockResolvedValue({
Authorization: 'Bearer fresh-token',
headers: { Authorization: 'Bearer fresh-token' },
});
const [server] = await service.getRegistryMcpServers(user);
@ -371,6 +371,7 @@ describe('InstanceAiMcpRegistryService', () => {
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith(
credential.id,
'project-1',
{ accessToken: 'stale-token' },
);
});
@ -401,6 +402,7 @@ describe('InstanceAiMcpRegistryService', () => {
expect(oauthService.refreshOAuth2CredentialById).toHaveBeenCalledWith(
credential.id,
'project-1',
{ accessToken: 'stale-token' },
);
});

30
packages/cli/src/modules/instance-ai/mcp/instance-ai-mcp-registry.service.ts
View File

@ -11,6 +11,7 @@ import { OutboundHttp } from '@n8n/backend-network';
import { isUniqueConstraintError, type CredentialsEntity, type User } from '@n8n/db';
import { Service } from '@n8n/di';
import type { McpServerConfig } from '@n8n/instance-ai';
import { isRecord } from '@n8n/utils/is-record';
import type { ICredentialDataDecryptedObject, LiteralMcpRegistryConnection } from 'n8n-workflow';
import { randomUUID } from 'node:crypto';
@ -31,7 +32,7 @@ import { McpRegistryService } from '@/modules/mcp-registry/registry/mcp-registry
import type { McpRegistryServer } from '@/modules/mcp-registry/registry/mcp-registry.types';
import { OauthService } from '@/oauth/oauth.service';
import { createAiMcpFetch } from '@/utils/ai-proxy-fetch';
import { createAuthFetch } from '@/utils/auth-fetch';
import { createAuthFetch, getBearerTokenRevision } from '@/utils/auth-fetch';
import type {
InstanceAiMcpRegistryConnection,
@ -493,13 +494,32 @@ export class InstanceAiMcpRegistryService {
}
const projectId = credentialWithData.credential.shared?.[0]?.projectId ?? null;
const storedTokenData = credentialWithData.data.oauthTokenData;
const oauthTokenData = isRecord(storedTokenData) ? { ...storedTokenData } : undefined;
return createAuthFetch({
baseFetch,
initialHeaders: prepared.value.headers,
onUnauthorized: async () =>
projectId
? await this.oauthService.refreshOAuth2CredentialById(config.credentialId, projectId)
: null,
onUnauthorized: async (currentHeaders) => {
if (!projectId) return null;
const result = await this.oauthService.refreshOAuth2CredentialById(
config.credentialId,
projectId,
getBearerTokenRevision(currentHeaders, oauthTokenData?.n8n_expires_at),
);
if (result && oauthTokenData) {
if (result.expiresAt === undefined) {
delete oauthTokenData.n8n_expires_at;
} else {
oauthTokenData.n8n_expires_at = String(result.expiresAt);
}
if (result.expiresInSeconds === undefined) {
delete oauthTokenData.expires_in;
} else {
oauthTokenData.expires_in = result.expiresInSeconds;
}
}
return result?.headers ?? null;
},
allowedDomains: {
mode: 'domains',
domains: prepared.value.allowedDomains,

7
packages/cli/src/oauth/__tests__/oauth.service.integration.test.ts
View File

@ -1,4 +1,4 @@
import type { Logger } from '@n8n/backend-common';
import type { LockService, Logger } from '@n8n/backend-common';
import { OutboundHttp, type SsrfProtectionService } from '@n8n/backend-network';
import { type LocalServer, startServer } from '@n8n/backend-network/testing';
import type { GlobalConfig, SsrfProtectionConfig } from '@n8n/config';
@ -34,11 +34,16 @@ interface Received {
*/
function buildService() {
const ssrf = mock<SsrfProtectionService>();
const lockService = mock<LockService>();
ssrf.validateUrl.mockResolvedValue({ ok: true, result: undefined });
ssrf.validateConnectionHost.mockReturnValue({ ok: true, result: undefined });
lockService.withLease.mockImplementation(
async (_namespace, _key, operation) => await operation(new AbortController().signal),
);
return new OauthService(
mock<Logger>(),
lockService,
mock<CredentialsHelper>(),
mock<CredentialsRepository>(),
mock<CredentialsFinderService>(),

219
packages/cli/src/oauth/__tests__/oauth.service.test.ts
View File

@ -1,4 +1,4 @@
import { Logger } from '@n8n/backend-common';
import { LockAcquisitionTimeoutError, LockService, Logger } from '@n8n/backend-common';
import { OutboundHttp, SsrfProtectionService, type HttpRequestClient } from '@n8n/backend-network';
import { mockInstance } from '@n8n/backend-test-utils';
import type { OAuth2CredentialData } from '@n8n/client-oauth2';
@ -80,6 +80,7 @@ const requestMock = vi.fn(async (options: IHttpRequestOptions) => {
describe('OauthService', () => {
const logger = mockInstance(Logger);
const lockService = mockInstance(LockService);
const credentialsHelper = mockInstance(CredentialsHelper);
const credentialsRepository = mockInstance(CredentialsRepository);
const credentialsFinderService = mockInstance(CredentialsFinderService);
@ -110,6 +111,9 @@ describe('OauthService', () => {
// path opt in explicitly; the rest fall back to the legacy URL-encoded state.
cacheService.get.mockResolvedValue(undefined);
credentialsHelper.getCredentialsProperties.mockReturnValue([]);
lockService.withLease.mockImplementation(
async (_namespace, _key, operation) => await operation(new AbortController().signal),
);
globalConfig.endpoints = { rest: 'rest' } as any;
urlService.getInstanceBaseUrl.mockReturnValue('http://localhost:5678');
@ -137,6 +141,7 @@ describe('OauthService', () => {
service = new OauthService(
logger,
lockService,
credentialsHelper,
credentialsRepository,
credentialsFinderService,
@ -5379,10 +5384,176 @@ describe('OauthService', () => {
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ Authorization: 'Bearer new-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer new-token' } });
expect(mockToken.refresh).toHaveBeenCalledTimes(1);
});
it('stores and returns the new token expiry time', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
const refreshed = {
data: { access_token: 'new-token', token_type: 'bearer', expires_in: 3600 },
accessToken: 'new-token',
};
const mockToken = { refresh: vi.fn().mockResolvedValue(refreshed), client: {} };
const credential = makeCredential({ isGlobal: true });
vi.mocked(ClientOAuth2).mockImplementation(function () {
return { createToken: vi.fn().mockReturnValue(mockToken) } as never;
});
credentialsRepository.findOne.mockResolvedValue(credential as never);
vi.spyOn(service, 'getOAuthCredentials').mockResolvedValue({
clientId: 'id',
clientSecret: 'secret',
accessTokenUrl: 'https://example.com/token',
grantType: 'authorizationCode',
authentication: 'header',
oauthTokenData: {
access_token: 'stale',
refresh_token: 'refresh-token',
expires_in: 3600,
n8n_expires_at: String(timestamp - 1),
},
} as unknown as OAuth2CredentialData);
vi.spyOn(service, 'encryptAndSaveData').mockResolvedValue(undefined);
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
const expiresAt = timestamp + 3_600_000;
expect(result).toEqual({
headers: { Authorization: 'Bearer new-token' },
expiresAt,
expiresInSeconds: 3600,
});
expect(service.encryptAndSaveData).toHaveBeenCalledWith(credential, {
oauthTokenData: expect.objectContaining({
access_token: 'new-token',
n8n_expires_at: String(expiresAt),
}),
});
});
it('shares one refresh between concurrent callers', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
let finishRefresh: ((value: unknown) => void) | undefined;
const refresh = vi.fn().mockImplementation(
async () =>
await new Promise((resolve) => {
finishRefresh = resolve;
}),
);
vi.mocked(ClientOAuth2).mockImplementation(function () {
return { createToken: vi.fn().mockReturnValue({ refresh, client: {} }) } as never;
});
credentialsRepository.findOne.mockResolvedValue(makeCredential({ isGlobal: true }) as never);
vi.spyOn(service, 'getOAuthCredentials').mockResolvedValue({
clientId: 'id',
clientSecret: 'secret',
accessTokenUrl: 'https://example.com/token',
grantType: 'authorizationCode',
authentication: 'header',
oauthTokenData: { access_token: 'stale', refresh_token: 'refresh-token' },
} as unknown as OAuth2CredentialData);
vi.spyOn(service, 'encryptAndSaveData').mockResolvedValue(undefined);
const results = [
service.refreshOAuth2CredentialById(credentialId, projectId),
service.refreshOAuth2CredentialById(credentialId, projectId),
];
await vi.waitFor(() => expect(refresh).toHaveBeenCalledTimes(1));
finishRefresh?.({ data: { access_token: 'new-token' }, accessToken: 'new-token' });
await expect(Promise.all(results)).resolves.toEqual([
{ headers: { Authorization: 'Bearer new-token' } },
{ headers: { Authorization: 'Bearer new-token' } },
]);
expect(lockService.withLease).toHaveBeenCalledTimes(1);
});
it('uses token data refreshed by another lock holder', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
credentialsRepository.findOne.mockResolvedValue(makeCredential({ isGlobal: true }) as never);
vi.spyOn(service, 'getOAuthCredentials')
.mockResolvedValueOnce({
clientId: 'id',
accessTokenUrl: 'https://example.com/token',
oauthTokenData: { access_token: 'stale', refresh_token: 'old-refresh' },
} as unknown as OAuth2CredentialData)
.mockResolvedValueOnce({
clientId: 'id',
accessTokenUrl: 'https://example.com/token',
oauthTokenData: { access_token: 'fresh', refresh_token: 'new-refresh' },
} as unknown as OAuth2CredentialData);
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ headers: { Authorization: 'Bearer fresh' } });
expect(ClientOAuth2).not.toHaveBeenCalled();
});
it('uses the stored token when the caller has an older token revision', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
const expiresAt = timestamp + 3_600_000;
credentialsRepository.findOne.mockResolvedValue(makeCredential({ isGlobal: true }) as never);
vi.spyOn(service, 'getOAuthCredentials').mockResolvedValue({
clientId: 'id',
accessTokenUrl: 'https://example.com/token',
oauthTokenData: {
access_token: 'fresh',
refresh_token: 'new-refresh',
expires_in: 3600,
n8n_expires_at: String(expiresAt),
},
} as unknown as OAuth2CredentialData);
const result = await service.refreshOAuth2CredentialById(credentialId, projectId, {
accessToken: 'stale',
expiresAt: timestamp - 1,
});
expect(result).toEqual({
headers: { Authorization: 'Bearer fresh' },
expiresAt,
expiresInSeconds: 3600,
});
expect(ClientOAuth2).not.toHaveBeenCalled();
});
it('refreshes without a credential lease when lease acquisition times out', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
const mockToken = {
refresh: vi.fn().mockResolvedValue({
data: { access_token: 'new-token', token_type: 'bearer' },
accessToken: 'new-token',
}),
client: {},
};
vi.mocked(ClientOAuth2).mockImplementation(function () {
return { createToken: vi.fn().mockReturnValue(mockToken) } as never;
});
lockService.withLease.mockRejectedValue(
new LockAcquisitionTimeoutError('Timed out waiting for the credential lock'),
);
credentialsRepository.findOne.mockResolvedValue(makeCredential({ isGlobal: true }) as never);
vi.spyOn(service, 'getOAuthCredentials').mockResolvedValue({
clientId: 'id',
accessTokenUrl: 'https://example.com/token',
grantType: 'authorizationCode',
oauthTokenData: { access_token: 'stale', refresh_token: 'refresh-token' },
} as unknown as OAuth2CredentialData);
vi.spyOn(service, 'encryptAndSaveData').mockResolvedValue(undefined);
const result = await service.refreshOAuth2CredentialById(credentialId, projectId, {
accessToken: 'stale',
});
expect(result).toEqual({ headers: { Authorization: 'Bearer new-token' } });
expect(mockToken.refresh).toHaveBeenCalledTimes(1);
expect(service.encryptAndSaveData).toHaveBeenCalledTimes(1);
expect(logger.warn).toHaveBeenCalledWith(
'Refreshing the OAuth2 credential without cross-process coordination',
expect.objectContaining({ credentialId }),
);
});
describe('outbound network policy', () => {
const captureRefreshClientOptions = async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
@ -5431,7 +5602,7 @@ describe('OauthService', () => {
const { capturedOptions, result } = await captureRefreshClientOptions();
expect(capturedOptions.ssrfBridge).toBe(ssrfProtectionService);
expect(result).toEqual({ Authorization: 'Bearer new-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer new-token' } });
});
it('should build the refresh client without a bridge when the guard is disabled', async () => {
@ -5441,7 +5612,7 @@ describe('OauthService', () => {
expect(capturedOptions.ssrfBridge).toBeUndefined();
// The refresh must still succeed, so instances that leave the guard off are unaffected.
expect(result).toEqual({ Authorization: 'Bearer new-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer new-token' } });
});
});
@ -5478,7 +5649,7 @@ describe('OauthService', () => {
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ Authorization: 'Bearer new-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer new-token' } });
expect(capturedOptions).toEqual(
expect.objectContaining({
clientCredentialType: 'certificate',
@ -5542,7 +5713,7 @@ describe('OauthService', () => {
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ Authorization: 'Bearer cc-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer cc-token' } });
expect(getToken).toHaveBeenCalledTimes(1);
expect(mockToken.refresh).not.toHaveBeenCalled();
});
@ -5573,7 +5744,7 @@ describe('OauthService', () => {
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ Authorization: 'Bearer cc-token' });
expect(result).toEqual({ headers: { Authorization: 'Bearer cc-token' } });
expect(capturedOptions).toEqual(expect.objectContaining({ resource }));
expect(service.encryptAndSaveData).toHaveBeenCalledWith(credential, {
oauthTokenData: {
@ -5612,7 +5783,33 @@ describe('OauthService', () => {
);
});
it('still returns the auth header even when persisting the new token data fails', async () => {
it('asks the user to reconnect when the refresh grant is invalid', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
const mockToken = {
refresh: vi
.fn()
.mockRejectedValue(new OAuth2AuthError('invalid grant', { error: 'invalid_grant' })),
client: {},
};
vi.mocked(ClientOAuth2).mockImplementation(function () {
return { createToken: vi.fn().mockReturnValue(mockToken) } as never;
});
credentialsRepository.findOne.mockResolvedValue(makeCredential({ isGlobal: true }) as never);
vi.spyOn(service, 'getOAuthCredentials').mockResolvedValue({
clientId: 'id',
clientSecret: 'secret',
accessTokenUrl: 'https://example.com/token',
grantType: 'authorizationCode',
authentication: 'header',
oauthTokenData: { access_token: 'stale', refresh_token: 'refresh-token' },
} as unknown as OAuth2CredentialData);
await expect(service.refreshOAuth2CredentialById(credentialId, projectId)).rejects.toThrow(
'This credential needs to be reconnected.',
);
});
it('rejects the refreshed token when persisting the new token data fails', async () => {
const { ClientOAuth2 } = await import('@n8n/client-oauth2');
const refreshed = { data: { access_token: 'new-token' }, accessToken: 'new-token' };
const mockToken = { refresh: vi.fn().mockResolvedValue(refreshed), client: {} };
@ -5631,9 +5828,9 @@ describe('OauthService', () => {
} as unknown as OAuth2CredentialData);
vi.spyOn(service, 'encryptAndSaveData').mockRejectedValue(new Error('db write error'));
const result = await service.refreshOAuth2CredentialById(credentialId, projectId);
expect(result).toEqual({ Authorization: 'Bearer new-token' });
await expect(service.refreshOAuth2CredentialById(credentialId, projectId)).rejects.toThrow(
'Could not save the refreshed OAuth2 token.',
);
expect(logger.warn).toHaveBeenCalledWith(
'Refreshed OAuth2 token but failed to persist new token data',
expect.objectContaining({ credentialId }),

198
packages/cli/src/oauth/oauth.service.ts
View File

@ -1,4 +1,4 @@
import { Logger } from '@n8n/backend-common';
import { LockNamespace, LockService, Logger, SingleFlightLease } from '@n8n/backend-common';
import {
OutboundHttp,
SsrfProtectionService,
@ -9,6 +9,7 @@ import { GlobalConfig, SsrfProtectionConfig } from '@n8n/config';
import type { AuthenticatedRequest, CredentialsEntity, ICredentialsDb } from '@n8n/db';
import { CredentialsRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { isRecord } from '@n8n/utils/is-record';
import Csrf from 'csrf';
import type { Request, Response } from 'express';
import { Credentials, Cipher } from 'n8n-core';
@ -94,6 +95,17 @@ export type OauthFlowState = {
const OAUTH_FLOW_CACHE_PREFIX = 'oauth:flow:';
const OAUTH_REQUEST_TIMEOUT_MS = 30 * Time.seconds.toMilliseconds; // This might be added to a OAuth Config (there is currently none)
export interface OAuth2CredentialRefreshResult {
headers: Record<string, string>;
expiresAt?: number;
expiresInSeconds?: number;
}
export interface OAuth2CredentialTokenRevision {
accessToken?: string;
expiresAt?: number;
}
export function shouldSkipAuthOnOAuthCallback() {
const value = process.env.N8N_SKIP_AUTH_ON_OAUTH_CALLBACK?.toLowerCase() ?? 'false';
return value === 'true';
@ -119,8 +131,11 @@ export class InvalidOAuthUrlError extends BadRequestError {
@Service()
export class OauthService {
private readonly oauth2Refreshes = new SingleFlightLease<OAuth2CredentialRefreshResult | null>();
constructor(
protected readonly logger: Logger,
private readonly lockService: LockService,
private readonly credentialsHelper: CredentialsHelper,
private readonly credentialsRepository: CredentialsRepository,
private readonly credentialsFinderService: CredentialsFinderService,
@ -717,21 +732,75 @@ export class OauthService {
refreshedData: ClientOAuth2TokenData,
resource?: string,
) {
return {
const merged = {
...oauthTokenData,
...refreshedData,
...(!refreshedData.resource && resource ? { resource } : {}),
};
const expiresInSeconds = Number(merged.expires_in);
return Number.isFinite(expiresInSeconds) && expiresInSeconds > 0
? {
...merged,
n8n_expires_at: String(Date.now() + expiresInSeconds * 1000),
}
: merged;
}
private getOAuth2AccessToken(tokenData: unknown): string | undefined {
if (!isRecord(tokenData)) return undefined;
const accessToken = tokenData.access_token ?? tokenData.accessToken;
return typeof accessToken === 'string' && accessToken.length > 0 ? accessToken : undefined;
}
private getOAuth2TokenRevision(tokenData: unknown): OAuth2CredentialTokenRevision {
const accessToken = this.getOAuth2AccessToken(tokenData);
const expiresAt = isRecord(tokenData) ? Number(tokenData.n8n_expires_at) : Number.NaN;
return {
...(accessToken ? { accessToken } : {}),
...(Number.isFinite(expiresAt) ? { expiresAt } : {}),
};
}
private oauth2TokenRevisionChanged(
initial: OAuth2CredentialTokenRevision,
current: ClientOAuth2TokenData,
): boolean {
const currentRevision = this.getOAuth2TokenRevision(current);
return (
initial.accessToken !== currentRevision.accessToken ||
initial.expiresAt !== currentRevision.expiresAt
);
}
private buildOAuth2RefreshResult(
tokenData: ClientOAuth2TokenData,
accessToken = this.getOAuth2AccessToken(tokenData),
): OAuth2CredentialRefreshResult | null {
if (!accessToken) return null;
const expiresAt = Number(tokenData.n8n_expires_at);
const expiresInSeconds = Number(tokenData.expires_in);
return {
headers: { Authorization: `Bearer ${accessToken}` },
...(Number.isFinite(expiresAt) ? { expiresAt } : {}),
...(Number.isFinite(expiresInSeconds) && expiresInSeconds > 0 ? { expiresInSeconds } : {}),
};
}
private isRevokedOAuth2GrantError(error: unknown): boolean {
if (!(error instanceof OAuth2AuthError)) return false;
return (error.body as { error?: unknown } | undefined)?.error === 'invalid_grant';
}
/**
* Refresh the OAuth2 token stored on a credential by id, persist the refreshed token data,
* and return the new auth headers to inject into outbound requests.
* Refresh the OAuth2 token stored on a credential by ID. Save the new token data.
* Return the auth headers and expiry time for outbound requests.
*/
async refreshOAuth2CredentialById(
credentialId: string,
projectId: string,
): Promise<Record<string, string> | null> {
knownTokenRevision?: OAuth2CredentialTokenRevision,
): Promise<OAuth2CredentialRefreshResult | null> {
const credential = await this.credentialsRepository.findOne({
where: { id: credentialId, usageScope: 'project' },
relations: { shared: true },
@ -743,49 +812,88 @@ export class OauthService {
const oauthCredentials = await this.getOAuthCredentials<OAuth2CredentialData>(credential);
const oauthTokenData = oauthCredentials.oauthTokenData as ClientOAuth2TokenData | undefined;
if (!oauthTokenData) return null;
const tokenRevision = knownTokenRevision ?? this.getOAuth2TokenRevision(oauthTokenData);
const resource = this.resolveOAuth2Resource(oauthCredentials, oauthTokenData);
const oAuthClient = this.createOAuth2ClientForRefresh(oauthCredentials, resource);
const runRefresh = async (): Promise<OAuth2CredentialRefreshResult | null> => {
const currentCredential = await this.credentialsRepository.findOne({
where: { id: credentialId, usageScope: 'project' },
relations: { shared: true },
});
if (!currentCredential) return null;
if (!this.credentialIsAccessibleToProject(currentCredential, projectId)) return null;
const token = oAuthClient.createToken(
{
...oauthTokenData,
...(oauthTokenData.access_token ? { access_token: oauthTokenData.access_token } : {}),
...(oauthTokenData.refresh_token ? { refresh_token: oauthTokenData.refresh_token } : {}),
const currentOAuthCredentials =
await this.getOAuthCredentials<OAuth2CredentialData>(currentCredential);
const currentTokenData = currentOAuthCredentials.oauthTokenData as
| ClientOAuth2TokenData
| undefined;
if (!currentTokenData) return null;
if (this.oauth2TokenRevisionChanged(tokenRevision, currentTokenData)) {
return this.buildOAuth2RefreshResult(currentTokenData);
}
const resource = this.resolveOAuth2Resource(currentOAuthCredentials, currentTokenData);
const oAuthClient = this.createOAuth2ClientForRefresh(currentOAuthCredentials, resource);
const token = oAuthClient.createToken(
{
...currentTokenData,
...(currentTokenData.access_token ? { access_token: currentTokenData.access_token } : {}),
...(currentTokenData.refresh_token
? { refresh_token: currentTokenData.refresh_token }
: {}),
},
currentTokenData.token_type,
);
let refreshed;
try {
refreshed =
currentOAuthCredentials.grantType === 'clientCredentials'
? await token.client.credentials.getToken()
: await token.refresh();
} catch (error) {
if (this.isRevokedOAuth2GrantError(error)) {
throw new UserError('This credential needs to be reconnected.', { cause: error });
}
this.logger.warn('Failed to refresh OAuth2 token for credential', {
credentialId,
error: error instanceof Error ? error.message : String(error),
});
return null;
}
const refreshedTokenData = this.mergeRefreshedOAuthTokenData(
currentTokenData,
refreshed.data,
resource,
);
try {
await this.encryptAndSaveData(currentCredential, { oauthTokenData: refreshedTokenData });
} catch (error) {
this.logger.warn('Refreshed OAuth2 token but failed to persist new token data', {
credentialId,
error: error instanceof Error ? error.message : String(error),
});
throw new OperationalError('Could not save the refreshed OAuth2 token.', { cause: error });
}
return this.buildOAuth2RefreshResult(refreshedTokenData, refreshed.accessToken);
};
return await this.oauth2Refreshes.run(credentialId, runRefresh, {
lockService: this.lockService,
namespace: LockNamespace.CREDENTIALS,
waitTimeoutMs: 10_000,
leaseTtlMs: 30_000,
onLeaseTimeout: (error) => {
this.logger.warn('Refreshing the OAuth2 credential without cross-process coordination', {
credentialId,
error: error.message,
});
},
oauthTokenData.token_type,
);
let refreshed;
try {
refreshed =
oauthCredentials.grantType === 'clientCredentials'
? await token.client.credentials.getToken()
: await token.refresh();
} catch (error) {
this.logger.warn('Failed to refresh OAuth2 token for credential', {
credentialId,
error: error instanceof Error ? error.message : String(error),
});
return null;
}
const refreshedTokenData = this.mergeRefreshedOAuthTokenData(
oauthTokenData,
refreshed.data,
resource,
);
try {
await this.encryptAndSaveData(credential, { oauthTokenData: refreshedTokenData });
} catch (error) {
this.logger.warn('Refreshed OAuth2 token but failed to persist new token data', {
credentialId,
error: error instanceof Error ? error.message : String(error),
});
}
return { Authorization: `Bearer ${refreshed.accessToken}` };
});
}
/**

1
packages/cli/src/utils/__tests__/auth-fetch.test.ts
View File

@ -73,6 +73,7 @@ describe('createAuthFetch', () => {
const res = await fetchFn('https://example.test/mcp');
expect(res.status).toBe(200);
expect(onUnauthorized).toHaveBeenCalledWith({ authorization: 'Bearer A' });
expect(baseFetchMock).toHaveBeenCalledTimes(2);
const [, init2] = baseFetchMock.mock.calls[1] as [unknown, RequestInit];
expect(new Headers(init2.headers).get('authorization')).toBe('Bearer B');

35
packages/cli/src/utils/auth-fetch.ts
View File

@ -5,6 +5,21 @@ import type { DomainRestrictionMode, ICredentialDataDecryptedObject } from 'n8n-
export type AuthFetchDomainPolicy = { mode: 'domains'; domains: string } | { mode: 'none' };
export function getBearerTokenRevision(
headers: Record<string, string>,
expiresAtValue?: unknown,
): { accessToken?: string; expiresAt?: number } {
const authorization = Object.entries(headers).find(
([name]) => name.toLowerCase() === 'authorization',
)?.[1];
const accessToken = authorization?.match(/^Bearer\s+(.+)$/i)?.[1];
const expiresAt = Number(expiresAtValue);
return {
...(accessToken ? { accessToken } : {}),
...(Number.isFinite(expiresAt) ? { expiresAt } : {}),
};
}
interface CreateAuthFetchOptions {
/** Proxy-aware base `fetch` every request routes through (see `createAiProxyFetch`). */
baseFetch: CustomFetch;
@ -14,7 +29,11 @@ interface CreateAuthFetchOptions {
* `null` if the refresh failed. The returned headers replace the cached
* set used by subsequent requests.
*/
onUnauthorized?: () => Promise<Record<string, string> | null>;
onUnauthorized?: (
currentHeaders: Record<string, string>,
) => Promise<Record<string, string> | null>;
/** Return true when auth must refresh before the next request. */
shouldRefresh?: () => boolean;
/**
* Domain policy from the credential. When set, the initial request and every
* redirect hop are validated so credentials are never sent to an
@ -59,9 +78,10 @@ function assertDomainPolicyAllowsUrl(url: string, policy: AuthFetchDomainPolicy)
* 1. routes through the supplied proxy-aware `baseFetch` (so corporate
* HTTP_PROXY settings apply uniformly),
* 2. injects the latest auth headers on every request,
* 3. on a single 401, calls `onUnauthorized` to refresh the token and
* 3. refreshes before a request when `shouldRefresh` returns true,
* 4. on a single 401, calls `onUnauthorized` to refresh the token and
* retries the request once with the new headers,
* 4. when a domain policy is set, follows redirects manually, validating
* 5. when a domain policy is set, follows redirects manually, validating
* every hop and withholding the auth headers once a hop crosses origins.
*
* This mirrors the langchain MCP node's `createAuthFetch` so an agent's MCP
@ -71,12 +91,19 @@ export function createAuthFetch({
baseFetch,
initialHeaders,
onUnauthorized,
shouldRefresh,
allowedDomains,
}: CreateAuthFetchOptions): typeof fetch {
return createRefreshingAuthFetch({
baseFetch,
initialHeaders,
...(onUnauthorized ? { refreshHeaders: async () => await onUnauthorized() } : {}),
...(onUnauthorized
? {
refreshHeaders: async (current: Headers) =>
await onUnauthorized(Object.fromEntries(current.entries())),
}
: {}),
...(shouldRefresh ? { shouldRefresh } : {}),
...(allowedDomains
? {
assertAllowedUrl: (url: string) => assertDomainPolicyAllowsUrl(url, allowedDomains),

21
packages/core/src/execution-engine/node-execution-context/utils/request-helpers/__tests__/oauth.test.ts
View File

@ -1,4 +1,4 @@
import { LockNamespace, LockService } from '@n8n/backend-common';
import { LockAcquisitionTimeoutError, LockNamespace, LockService } from '@n8n/backend-common';
import type { SsrfBridge } from '@n8n/backend-network';
import { Container } from '@n8n/di';
import FormData from 'form-data';
@ -1298,6 +1298,25 @@ describe('requestOAuth2 - concurrent refresh serialization', () => {
await expect(call()).resolves.toEqual({ success: true });
});
test('refreshes without a lease when lease acquisition times out', async () => {
const withLeaseSpy = vi
.spyOn(Container.get(LockService), 'withLease')
.mockRejectedValueOnce(new LockAcquisitionTimeoutError('Timed out waiting for lock'));
const tokenScope = nock(tokenUrl)
.post('/token')
.reply(200, { access_token: 'new-token', token_type: 'bearer' });
mockThis.helpers.httpRequest
.mockRejectedValueOnce(error401())
.mockResolvedValue({ success: true });
await expect(call()).resolves.toEqual({ success: true });
expect(tokenScope.isDone()).toBe(true);
expect(
mockAdditionalData.credentialsHelper.updateCredentialsOauthTokenData,
).toHaveBeenCalledTimes(1);
withLeaseSpy.mockRestore();
});
test('wraps the refresh in LockService.withLease keyed by the credential id', async () => {
// In-process the coalescing map masks the lock, so this is the only proof the
// (cross-process) lock path is actually wired.

48
packages/core/src/execution-engine/node-execution-context/utils/request-helpers/oauth.ts
View File

@ -4,7 +4,7 @@
/* eslint-disable @typescript-eslint/no-unsafe-return */
/* eslint-disable @typescript-eslint/no-unsafe-member-access */
import { LockNamespace, LockAcquisitionTimeoutError, LockService } from '@n8n/backend-common';
import { LockNamespace, LockService, SingleFlightLease } from '@n8n/backend-common';
import { isFormDataInstance, removeEmptyBody, type SsrfBridge } from '@n8n/backend-network';
import type {
ClientOAuth2Options,
@ -146,7 +146,7 @@ function buildOAuth2ReconnectError(node: INode, credentialsType: string): NodeOp
);
}
const inFlightRefreshes = new Map<string, Promise<ClientOAuth2Token>>();
const inFlightRefreshes = new SingleFlightLease<ClientOAuth2Token>();
async function refreshOrFetchToken(ctx: RefreshOAuth2TokenContext): Promise<ClientOAuth2Token> {
const nodeCredentials = ctx.node.credentials?.[ctx.credentialsType];
@ -159,15 +159,6 @@ async function refreshOrFetchToken(ctx: RefreshOAuth2TokenContext): Promise<Clie
});
}
// Critical section from here until the in-flight promise is stored in the map.
// This map is process-LOCAL: it coalesces concurrent refreshes *within this instance*
// so they share one network call. Cross-instance serialization is handled by the lease below.
// We must not yield to the event loop before `inFlightRefreshes.set(...)`, or a second concurrent
// caller could pass the `has` check before the promise is registered and trigger a duplicate refresh.
if (inFlightRefreshes.has(credentialId)) {
return await inFlightRefreshes.get(credentialId)!;
}
const runRefresh = async () => {
const {
credentials,
@ -302,29 +293,18 @@ async function refreshOrFetchToken(ctx: RefreshOAuth2TokenContext): Promise<Clie
return signingToken;
};
const promise = Container.get(LockService)
.withLease(LockNamespace.CREDENTIALS, credentialId, runRefresh, {
waitTimeoutMs: 10_000,
leaseTtlMs: 30_000,
})
.catch(async (error) => {
if (error instanceof LockAcquisitionTimeoutError) {
// The lease is an efficiency primitive, not a correctness one. If the lock
// backend is unavailable/contended, refresh without cross-process coordination
// rather than stalling the execution; runRefresh's read-back check still avoids
// a redundant network call when another holder already rotated the token.
ctx.logger.warn(
`Could not acquire refresh lock for credential "${credentialId}"; refreshing without cross-process coordination`,
{ error },
);
return await runRefresh();
}
throw error;
})
.finally(() => inFlightRefreshes.delete(credentialId));
inFlightRefreshes.set(credentialId, promise);
return await promise;
return await inFlightRefreshes.run(credentialId, runRefresh, {
lockService: Container.get(LockService),
namespace: LockNamespace.CREDENTIALS,
waitTimeoutMs: 10_000,
leaseTtlMs: 30_000,
onLeaseTimeout: (error) => {
ctx.logger.warn(
`Could not acquire refresh lock for credential "${credentialId}"; refreshing without cross-process coordination`,
{ error },
);
},
});
}
function resolveTokenExpiredStatusCode(

22
packages/workflow/src/mcp-helpers.ts
View File

@ -2,6 +2,9 @@ import { isRecord } from '@n8n/utils/is-record';
import type { ICredentialDataDecryptedObject } from './interfaces';
const OAUTH2_REFRESH_BUFFER_MS = 2 * 60 * 1000;
const OAUTH2_REFRESH_BUFFER_RATIO = 0.1;
/** Covers MCP-specific and existing native OAuth2 credential type names. */
export type McpOAuth2CredentialType = 'oAuth2Api' | `${string}OAuth2Api` | `${string}OAuth2`;
@ -105,6 +108,25 @@ export function isMcpOAuth2Authentication(
);
}
/** Return true when an MCP OAuth2 token is close enough to expiry to refresh it. */
export function shouldRefreshMcpOAuth2Token(tokenData: unknown, grantType?: unknown): boolean {
if (!isRecord(tokenData)) return false;
if (grantType !== 'clientCredentials' && !(tokenData.refresh_token ?? tokenData.refreshToken)) {
return false;
}
const expiresAt = Number(tokenData.n8n_expires_at);
if (!Number.isFinite(expiresAt)) return false;
const expiresInMs = Number(tokenData.expires_in) * 1000;
const refreshBufferMs =
Number.isFinite(expiresInMs) && expiresInMs > 0
? Math.min(OAUTH2_REFRESH_BUFFER_MS, expiresInMs * OAUTH2_REFRESH_BUFFER_RATIO)
: OAUTH2_REFRESH_BUFFER_MS;
return Date.now() + refreshBufferMs >= expiresAt;
}
export function getMcpAuthHeaders(
authentication: string,
credentialData: ICredentialDataDecryptedObject,

32
packages/workflow/test/mcp-helpers.test.ts
View File

@ -1,4 +1,4 @@
import { isMcpOAuth2Authentication } from '../src/mcp-helpers';
import { isMcpOAuth2Authentication, shouldRefreshMcpOAuth2Token } from '../src/mcp-helpers';
describe('isMcpOAuth2Authentication', () => {
it('returns true for the canonical "mcpOAuth2Api" type', () => {
@ -28,3 +28,33 @@ describe('isMcpOAuth2Authentication', () => {
expect(isMcpOAuth2Authentication('')).toBe(false);
});
});
describe('shouldRefreshMcpOAuth2Token', () => {
const now = 1_700_000_000_000;
beforeEach(() => {
vi.spyOn(Date, 'now').mockReturnValue(now);
});
afterEach(() => {
vi.restoreAllMocks();
});
it('refreshes an expiring client credentials token without a refresh token', () => {
expect(
shouldRefreshMcpOAuth2Token(
{ expires_in: 3600, n8n_expires_at: String(now + 60_000) },
'clientCredentials',
),
).toBe(true);
});
it('does not refresh an authorization code token without a refresh token', () => {
expect(
shouldRefreshMcpOAuth2Token(
{ expires_in: 3600, n8n_expires_at: String(now + 60_000) },
'authorizationCode',
),
).toBe(false);
});
});