fix: harden codex runtime routing and oauth recovery
This commit is contained in:
Cita
@@ -500,6 +500,153 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('keeps the existing codex connection intact when a rebind callback fails model discovery', async () => {
|
||||
const originalJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-existing',
|
||||
chatgpt_plan_type: 'team',
|
||||
},
|
||||
});
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'ChatGPT Codex OAuth',
|
||||
url: 'https://chatgpt.com/backend-api/codex',
|
||||
platform: 'codex',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
const existing = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'codex-existing@example.com',
|
||||
accessToken: 'stable-access-token',
|
||||
apiToken: null,
|
||||
status: 'active',
|
||||
oauthProvider: 'codex',
|
||||
oauthAccountKey: 'chatgpt-account-existing',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-existing',
|
||||
accountKey: 'chatgpt-account-existing',
|
||||
email: 'codex-existing@example.com',
|
||||
planType: 'team',
|
||||
refreshToken: 'stable-refresh-token',
|
||||
idToken: originalJwt,
|
||||
},
|
||||
}),
|
||||
}).returning().get();
|
||||
|
||||
const reboundJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-rebound',
|
||||
chatgpt_plan_type: 'plus',
|
||||
},
|
||||
});
|
||||
fetchMock
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
access_token: 'rebound-access-token',
|
||||
refresh_token: 'rebound-refresh-token',
|
||||
id_token: reboundJwt,
|
||||
expires_in: 3600,
|
||||
token_type: 'Bearer',
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
ok: false,
|
||||
status: 403,
|
||||
json: async () => ({ error: 'forbidden' }),
|
||||
text: async () => 'forbidden',
|
||||
});
|
||||
|
||||
const startResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/oauth/providers/codex/start',
|
||||
headers: {
|
||||
host: 'metapi.example',
|
||||
'x-forwarded-proto': 'https',
|
||||
},
|
||||
payload: {
|
||||
accountId: existing.id,
|
||||
},
|
||||
});
|
||||
const startBody = startResponse.json() as { state: string };
|
||||
|
||||
const callbackResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/oauth/sessions/${encodeURIComponent(startBody.state)}/manual-callback`,
|
||||
payload: {
|
||||
callbackUrl: `http://localhost:1455/auth/callback?state=${encodeURIComponent(startBody.state)}&code=oauth-code-rebind-fail`,
|
||||
},
|
||||
});
|
||||
expect(callbackResponse.statusCode).toBe(500);
|
||||
|
||||
const stored = await db.select().from(schema.accounts).where(eq(schema.accounts.id, existing.id)).get();
|
||||
expect(stored).toMatchObject({
|
||||
id: existing.id,
|
||||
username: 'codex-existing@example.com',
|
||||
accessToken: 'stable-access-token',
|
||||
oauthAccountKey: 'chatgpt-account-existing',
|
||||
});
|
||||
expect(JSON.parse(stored?.extraConfig || '{}')).toMatchObject({
|
||||
oauth: {
|
||||
accountId: 'chatgpt-account-existing',
|
||||
refreshToken: 'stable-refresh-token',
|
||||
idToken: originalJwt,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('fails codex oauth onboarding when token exchange does not expose chatgpt_account_id', async () => {
|
||||
const jwt = buildJwt({
|
||||
email: 'codex-no-account@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_plan_type: 'plus',
|
||||
},
|
||||
});
|
||||
fetchMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
access_token: 'oauth-access-token',
|
||||
refresh_token: 'oauth-refresh-token',
|
||||
id_token: jwt,
|
||||
expires_in: 3600,
|
||||
token_type: 'Bearer',
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
});
|
||||
|
||||
const startResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/oauth/providers/codex/start',
|
||||
headers: {
|
||||
host: 'metapi.example',
|
||||
'x-forwarded-proto': 'https',
|
||||
},
|
||||
});
|
||||
const startBody = startResponse.json() as { state: string };
|
||||
|
||||
const callbackResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/oauth/sessions/${encodeURIComponent(startBody.state)}/manual-callback`,
|
||||
payload: {
|
||||
callbackUrl: `http://localhost:1455/auth/callback?state=${encodeURIComponent(startBody.state)}&code=oauth-code-no-account-id`,
|
||||
},
|
||||
});
|
||||
expect(callbackResponse.statusCode).toBe(500);
|
||||
expect(callbackResponse.json()).toMatchObject({
|
||||
message: expect.stringContaining('chatgpt_account_id'),
|
||||
});
|
||||
|
||||
const accounts = await db.select().from(schema.accounts).all();
|
||||
expect(accounts).toEqual([]);
|
||||
});
|
||||
|
||||
it('marks gemini oauth session as error when token exchange fails before account persistence', async () => {
|
||||
fetchMock.mockResolvedValueOnce({
|
||||
ok: false,
|
||||
|
||||
@@ -361,4 +361,47 @@ describe('chat proxy codex oauth compatibility', () => {
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok after refresh');
|
||||
});
|
||||
|
||||
it('retries oauth chat requests after a 403 auth failure', async () => {
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: { message: 'forbidden account mismatch', type: 'invalid_request_error' },
|
||||
}), {
|
||||
status: 403,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
id: 'chatcmpl_refreshed_403',
|
||||
object: 'chat.completion',
|
||||
created: 1706000000,
|
||||
model: 'gpt-4o-mini',
|
||||
choices: [
|
||||
{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: 'ok after forbidden refresh' },
|
||||
finish_reason: 'stop',
|
||||
},
|
||||
],
|
||||
usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
messages: [{ role: 'user', content: 'hello oauth' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok after forbidden refresh');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -186,6 +186,7 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
'user-agent': 'CodexClient/1.0',
|
||||
'Chatgpt-Account-Id': 'spoofed-account',
|
||||
},
|
||||
payload: {
|
||||
model: 'gpt-5.2-codex',
|
||||
@@ -214,6 +215,46 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
expect(response.json()?.output_text).toContain('ok after codex token refresh');
|
||||
});
|
||||
|
||||
it('refreshes codex oauth token and retries the same responses request on 403', async () => {
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: { message: 'forbidden account mismatch', type: 'invalid_request_error' },
|
||||
}), {
|
||||
status: 403,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
id: 'resp_codex_refreshed_403',
|
||||
object: 'response',
|
||||
model: 'gpt-5.2-codex',
|
||||
status: 'completed',
|
||||
output_text: 'ok after codex forbidden refresh',
|
||||
usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
'user-agent': 'CodexClient/1.0',
|
||||
},
|
||||
payload: {
|
||||
model: 'gpt-5.2-codex',
|
||||
input: 'hello codex',
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.output_text).toContain('ok after codex forbidden refresh');
|
||||
});
|
||||
|
||||
it('retries oauth responses requests with a normalized upstream URL after refresh', async () => {
|
||||
selectChannelMock.mockReturnValue({
|
||||
channel: { id: 11, routeId: 22 },
|
||||
@@ -316,7 +357,7 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('preserves explicit prompt_cache_key for codex responses requests', async () => {
|
||||
it('preserves explicit prompt_cache_key for codex responses requests without converting it into codex session headers', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp_codex_with_cache_key',
|
||||
object: 'response',
|
||||
@@ -344,8 +385,8 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
|
||||
const [, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
const forwardedBody = JSON.parse(options.body);
|
||||
expect(options.headers.Session_id || options.headers.session_id).toBe('codex-cache-123');
|
||||
expect(options.headers.Conversation_id || options.headers.conversation_id).toBe('codex-cache-123');
|
||||
expect(String(options.headers.Session_id || options.headers.session_id || '')).toMatch(/^[0-9a-f-]{36}$/i);
|
||||
expect(options.headers.Conversation_id || options.headers.conversation_id).toBeUndefined();
|
||||
expect(forwardedBody.prompt_cache_key).toBe('codex-cache-123');
|
||||
});
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import { createServer, type Server } from 'node:http';
|
||||
import { AddressInfo } from 'node:net';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { config } from '../../config.js';
|
||||
|
||||
const fetchMock = vi.fn();
|
||||
const selectChannelMock = vi.fn();
|
||||
@@ -10,6 +11,8 @@ const selectNextChannelMock = vi.fn();
|
||||
const previewSelectedChannelMock = vi.fn();
|
||||
const recordSuccessMock = vi.fn();
|
||||
const recordFailureMock = vi.fn();
|
||||
const authorizeDownstreamTokenMock = vi.fn();
|
||||
const consumeManagedKeyRequestMock = vi.fn();
|
||||
const refreshModelsAndRebuildRoutesMock = vi.fn();
|
||||
const reportProxyAllFailedMock = vi.fn();
|
||||
const reportTokenExpiredMock = vi.fn();
|
||||
@@ -47,6 +50,22 @@ vi.mock('../../services/alertService.js', () => ({
|
||||
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/downstreamApiKeyService.js', () => ({
|
||||
authorizeDownstreamToken: (...args: unknown[]) => authorizeDownstreamTokenMock(...args),
|
||||
consumeManagedKeyRequest: (...args: unknown[]) => consumeManagedKeyRequestMock(...args),
|
||||
isModelAllowedByPolicyOrAllowedRoutes: async (
|
||||
model: string,
|
||||
policy: { supportedModels?: string[]; allowedRouteIds?: number[]; denyAllWhenEmpty?: boolean },
|
||||
) => {
|
||||
const supportedModels = Array.isArray(policy?.supportedModels) ? policy.supportedModels : [];
|
||||
const allowedRouteIds = Array.isArray(policy?.allowedRouteIds) ? policy.allowedRouteIds : [];
|
||||
if (supportedModels.length === 0 && allowedRouteIds.length === 0) {
|
||||
return policy?.denyAllWhenEmpty === true ? false : true;
|
||||
}
|
||||
return supportedModels.includes(model);
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock('../../services/alertRules.js', () => ({
|
||||
isTokenExpiredError: () => false,
|
||||
}));
|
||||
@@ -175,7 +194,17 @@ function waitForSocketMessages(socket: WebSocket, count: number, timeoutMs = 100
|
||||
});
|
||||
}
|
||||
|
||||
function createClientSocket(baseUrl: string, headers: Record<string, string> = {}) {
|
||||
return new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
Authorization: 'Bearer sk-global-proxy-token',
|
||||
...headers,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
describe('responses websocket transport', () => {
|
||||
const originalCodexResponsesWebsocketBeta = config.codexResponsesWebsocketBeta;
|
||||
let app: FastifyInstance;
|
||||
let baseUrl: string;
|
||||
let upstreamServer: WebSocketServer;
|
||||
@@ -186,6 +215,9 @@ describe('responses websocket transport', () => {
|
||||
let upstreamMessageHandler: (socket: WebSocket, parsed: Record<string, unknown>, requestIndex: number) => void;
|
||||
let rejectedUpgradeServer: Server;
|
||||
let rejectedUpgradeSiteUrl: string;
|
||||
let rejectedUpgradeStatus: number;
|
||||
let rejectedUpgradeStatusText: string;
|
||||
let rejectedUpgradeBody: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
const { responsesProxyRoute } = await import('./responses.js');
|
||||
@@ -214,13 +246,14 @@ describe('responses websocket transport', () => {
|
||||
|
||||
rejectedUpgradeServer = createServer();
|
||||
rejectedUpgradeServer.on('upgrade', (_request, socket) => {
|
||||
const body = rejectedUpgradeBody;
|
||||
socket.write(
|
||||
'HTTP/1.1 426 Upgrade Required\r\n'
|
||||
`HTTP/1.1 ${rejectedUpgradeStatus} ${rejectedUpgradeStatusText}\r\n`
|
||||
+ 'Content-Type: text/plain\r\n'
|
||||
+ 'Content-Length: 16\r\n'
|
||||
+ `Content-Length: ${Buffer.byteLength(body)}\r\n`
|
||||
+ 'Connection: close\r\n'
|
||||
+ '\r\n'
|
||||
+ 'Upgrade Required',
|
||||
+ body,
|
||||
);
|
||||
socket.destroy();
|
||||
});
|
||||
@@ -236,6 +269,8 @@ describe('responses websocket transport', () => {
|
||||
previewSelectedChannelMock.mockReset();
|
||||
recordSuccessMock.mockReset();
|
||||
recordFailureMock.mockReset();
|
||||
authorizeDownstreamTokenMock.mockReset();
|
||||
consumeManagedKeyRequestMock.mockReset();
|
||||
refreshModelsAndRebuildRoutesMock.mockReset();
|
||||
reportProxyAllFailedMock.mockReset();
|
||||
reportTokenExpiredMock.mockReset();
|
||||
@@ -249,6 +284,21 @@ describe('responses websocket transport', () => {
|
||||
upstreamConnectionCount = 0;
|
||||
upstreamUpgradeHeaders = {};
|
||||
upstreamRequests = [];
|
||||
(config as any).codexResponsesWebsocketBeta = originalCodexResponsesWebsocketBeta;
|
||||
rejectedUpgradeStatus = 426;
|
||||
rejectedUpgradeStatusText = 'Upgrade Required';
|
||||
rejectedUpgradeBody = 'Upgrade Required';
|
||||
authorizeDownstreamTokenMock.mockResolvedValue({
|
||||
ok: true,
|
||||
source: 'global',
|
||||
token: 'sk-global-proxy-token',
|
||||
key: null,
|
||||
policy: {
|
||||
supportedModels: [],
|
||||
allowedRouteIds: [],
|
||||
siteWeightMultipliers: {},
|
||||
},
|
||||
});
|
||||
upstreamMessageHandler = (socket, parsed, requestIndex) => {
|
||||
const responseId = `resp_upstream_${requestIndex}`;
|
||||
socket.send(JSON.stringify({
|
||||
@@ -331,7 +381,7 @@ describe('responses websocket transport', () => {
|
||||
}));
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 4);
|
||||
|
||||
@@ -362,10 +412,8 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
|
||||
it('echoes x-codex-turn-state on websocket upgrade responses', async () => {
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
},
|
||||
const socket = createClientSocket(baseUrl, {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
});
|
||||
|
||||
const [upgrade] = await Promise.all([
|
||||
@@ -378,17 +426,16 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
|
||||
it('reuses one upstream codex websocket session across sequential websocket turns', async () => {
|
||||
(config as any).codexResponsesWebsocketBeta = 'responses_websockets=2099-01-01';
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: upstreamSiteUrl,
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
'x-codex-beta-features': 'feature-a,feature-b',
|
||||
},
|
||||
const socket = createClientSocket(baseUrl, {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
'x-codex-beta-features': 'feature-a,feature-b',
|
||||
});
|
||||
await waitForSocketOpen(socket);
|
||||
const firstMessagesPromise = waitForSocketMessages(socket, 1);
|
||||
@@ -427,7 +474,7 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
expect(upstreamUpgradeHeaders['x-codex-turn-state']).toBe('turn-state-123');
|
||||
expect(upstreamUpgradeHeaders['x-codex-beta-features']).toBe('feature-a,feature-b');
|
||||
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=');
|
||||
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=2099-01-01');
|
||||
});
|
||||
|
||||
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 426', async () => {
|
||||
@@ -442,7 +489,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -460,6 +507,80 @@ describe('responses websocket transport', () => {
|
||||
expect(messages[0]?.response?.id).toBe('resp_http_fallback');
|
||||
});
|
||||
|
||||
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 401', async () => {
|
||||
rejectedUpgradeStatus = 401;
|
||||
rejectedUpgradeStatusText = 'Unauthorized';
|
||||
rejectedUpgradeBody = JSON.stringify({
|
||||
error: {
|
||||
message: 'expired token',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
});
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: rejectedUpgradeSiteUrl,
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
fetchMock.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_http_fallback_401","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
|
||||
expect(messages[0]?.response?.id).toBe('resp_http_fallback_401');
|
||||
});
|
||||
|
||||
it('rejects websocket turns whose model is blocked by the downstream key policy before channel selection', async () => {
|
||||
authorizeDownstreamTokenMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
source: 'managed',
|
||||
token: 'sk-managed-denied',
|
||||
key: {
|
||||
id: 99,
|
||||
name: 'limited-key',
|
||||
},
|
||||
policy: {
|
||||
supportedModels: ['gpt-4.1'],
|
||||
allowedRouteIds: [],
|
||||
siteWeightMultipliers: {},
|
||||
},
|
||||
});
|
||||
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(messages[0]).toMatchObject({
|
||||
type: 'error',
|
||||
status: 403,
|
||||
});
|
||||
expect(selectChannelMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('merges follow-up response.create payloads when the selected upstream does not support incremental mode', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
sitePlatform: 'openai',
|
||||
@@ -483,7 +604,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
@@ -609,7 +730,7 @@ describe('responses websocket transport', () => {
|
||||
}));
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
@@ -694,7 +815,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -753,7 +874,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const prewarmMessagesPromise = waitForSocketMessages(socket, 2);
|
||||
@@ -805,7 +926,7 @@ describe('responses websocket transport', () => {
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -852,7 +973,7 @@ describe('responses websocket transport', () => {
|
||||
socket.close();
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 3, 400);
|
||||
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { IncomingMessage } from 'node:http';
|
||||
import type { Duplex } from 'node:stream';
|
||||
import { WebSocketServer, type RawData, type WebSocket } from 'ws';
|
||||
import { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } from '../../proxy-core/runtime/codexWebsocketRuntime.js';
|
||||
import {
|
||||
authorizeDownstreamToken,
|
||||
consumeManagedKeyRequest,
|
||||
isModelAllowedByPolicyOrAllowedRoutes,
|
||||
type DownstreamTokenAuthSuccess,
|
||||
} from '../../services/downstreamApiKeyService.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { buildOauthProviderHeaders } from '../../services/oauth/service.js';
|
||||
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
|
||||
@@ -15,6 +22,7 @@ const RESPONSES_WEBSOCKET_TRANSPORT_HEADER = 'x-metapi-responses-websocket-trans
|
||||
const codexWebsocketRuntime = createCodexWebsocketRuntime();
|
||||
|
||||
type SelectedChannel = NonNullable<Awaited<ReturnType<typeof tokenRouter.selectChannel>>>;
|
||||
type ResponsesWebsocketAuthContext = DownstreamTokenAuthSuccess;
|
||||
|
||||
type NormalizedResponsesWebsocketRequest =
|
||||
| {
|
||||
@@ -131,8 +139,8 @@ function shouldReuseSelectedChannel(
|
||||
}
|
||||
|
||||
function deriveCodexExplicitSessionId(body: Record<string, unknown>, sessionId: string): string {
|
||||
const promptCacheKey = asTrimmedString(body.prompt_cache_key);
|
||||
return promptCacheKey || sessionId;
|
||||
void body;
|
||||
return sessionId;
|
||||
}
|
||||
|
||||
function parseJsonObject(raw: RawData): Record<string, unknown> | null {
|
||||
@@ -417,15 +425,46 @@ function buildInjectHeaders(request: IncomingMessage): Record<string, string | s
|
||||
return headers;
|
||||
}
|
||||
|
||||
function extractWebsocketAuthToken(request: IncomingMessage, url: URL): string {
|
||||
const auth = headerValueToTrimmedString(request.headers.authorization);
|
||||
if (auth) return auth.replace(/^Bearer\s+/i, '').trim();
|
||||
const apiKey = headerValueToTrimmedString(request.headers['x-api-key']);
|
||||
if (apiKey) return apiKey;
|
||||
const googApiKey = headerValueToTrimmedString(request.headers['x-goog-api-key']);
|
||||
if (googApiKey) return googApiKey;
|
||||
return asTrimmedString(url.searchParams.get('key'));
|
||||
}
|
||||
|
||||
function writeUpgradeHttpError(socket: Duplex, status: number, message: string): void {
|
||||
const statusText = status === 401
|
||||
? 'Unauthorized'
|
||||
: status === 403
|
||||
? 'Forbidden'
|
||||
: status === 400
|
||||
? 'Bad Request'
|
||||
: 'Error';
|
||||
const body = JSON.stringify({ error: message });
|
||||
socket.write(
|
||||
`HTTP/1.1 ${status} ${statusText}\r\n`
|
||||
+ 'Content-Type: application/json\r\n'
|
||||
+ `Content-Length: ${Buffer.byteLength(body)}\r\n`
|
||||
+ 'Connection: close\r\n'
|
||||
+ '\r\n'
|
||||
+ body,
|
||||
);
|
||||
socket.destroy();
|
||||
}
|
||||
|
||||
async function supportsResponsesWebsocketIncrementalInput(
|
||||
parsed: Record<string, unknown>,
|
||||
lastRequest: Record<string, unknown> | null,
|
||||
authContext: ResponsesWebsocketAuthContext,
|
||||
): Promise<boolean> {
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
if (!requestModel) return false;
|
||||
|
||||
try {
|
||||
const selected = await tokenRouter.previewSelectedChannel(requestModel);
|
||||
const selected = await tokenRouter.previewSelectedChannel(requestModel, authContext.policy);
|
||||
return selectedChannelSupportsIncrementalInput(selected, requestModel);
|
||||
} catch {
|
||||
return false;
|
||||
@@ -436,6 +475,7 @@ async function handleResponsesWebsocketConnection(
|
||||
app: FastifyInstance,
|
||||
socket: WebSocket,
|
||||
request: IncomingMessage,
|
||||
authContext: ResponsesWebsocketAuthContext,
|
||||
) {
|
||||
const websocketSessionId = headerValueToTrimmedString(request.headers['session_id'])
|
||||
|| headerValueToTrimmedString(request.headers['session-id'])
|
||||
@@ -449,129 +489,141 @@ async function handleResponsesWebsocketConnection(
|
||||
});
|
||||
|
||||
socket.on('message', async (raw) => {
|
||||
const parsed = parseJsonObject(raw);
|
||||
if (!parsed) {
|
||||
writeResponsesWebsocketError(socket, 400, 'Invalid websocket JSON payload');
|
||||
return;
|
||||
}
|
||||
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|
||||
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
|
||||
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed,
|
||||
lastRequest,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
const normalized = normalizeResponsesWebsocketRequest(
|
||||
parsed,
|
||||
lastRequest,
|
||||
lastResponseOutput,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
if (!normalized.ok) {
|
||||
writeResponsesWebsocketError(socket, normalized.status, normalized.message);
|
||||
return;
|
||||
}
|
||||
|
||||
lastRequest = normalized.nextRequestSnapshot;
|
||||
if (shouldHandleLocalPrewarm) {
|
||||
lastResponseOutput = [];
|
||||
for (const payload of synthesizePrewarmResponsePayloads(normalized.request)) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
try {
|
||||
const parsed = parseJsonObject(raw);
|
||||
if (!parsed) {
|
||||
writeResponsesWebsocketError(socket, 400, 'Invalid websocket JSON payload');
|
||||
return;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
|
||||
selectedChannel = requestModel
|
||||
? await tokenRouter.selectChannel(requestModel)
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
if (requestModel && !await isModelAllowedByPolicyOrAllowedRoutes(requestModel, authContext.policy)) {
|
||||
writeResponsesWebsocketError(socket, 403, 'model is not allowed for this downstream key');
|
||||
return;
|
||||
}
|
||||
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|
||||
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest, authContext);
|
||||
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed,
|
||||
lastRequest,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
const normalized = normalizeResponsesWebsocketRequest(
|
||||
parsed,
|
||||
lastRequest,
|
||||
lastResponseOutput,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
if (!normalized.ok) {
|
||||
writeResponsesWebsocketError(socket, normalized.status, normalized.message);
|
||||
return;
|
||||
}
|
||||
|
||||
if (authContext.source === 'managed' && authContext.key?.id) {
|
||||
await consumeManagedKeyRequest(authContext.key.id);
|
||||
}
|
||||
|
||||
lastRequest = normalized.nextRequestSnapshot;
|
||||
if (shouldHandleLocalPrewarm) {
|
||||
lastResponseOutput = [];
|
||||
for (const payload of synthesizePrewarmResponsePayloads(normalized.request)) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
|
||||
selectedChannel = requestModel
|
||||
? await tokenRouter.selectChannel(requestModel, authContext.policy)
|
||||
: null;
|
||||
}
|
||||
|
||||
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
|
||||
? selectedChannel
|
||||
: null;
|
||||
}
|
||||
|
||||
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
|
||||
? selectedChannel
|
||||
: null;
|
||||
|
||||
if (codexWebsocketChannel) {
|
||||
const downstreamHeaders: Record<string, unknown> = {
|
||||
...(request.headers as Record<string, unknown>),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
};
|
||||
const providerHeaders = buildOauthProviderHeaders({
|
||||
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
|
||||
? codexWebsocketChannel.account.extraConfig
|
||||
: null,
|
||||
downstreamHeaders,
|
||||
});
|
||||
const prepared = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
|
||||
stream: true,
|
||||
tokenValue: codexWebsocketChannel.tokenValue,
|
||||
sitePlatform: codexWebsocketChannel.site.platform,
|
||||
siteUrl: codexWebsocketChannel.site.url,
|
||||
openaiBody: normalized.request,
|
||||
downstreamFormat: 'responses',
|
||||
responsesOriginalBody: normalized.request,
|
||||
downstreamHeaders,
|
||||
providerHeaders,
|
||||
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
|
||||
});
|
||||
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
|
||||
|
||||
try {
|
||||
const runtimeResult = await codexWebsocketRuntime.sendRequest({
|
||||
sessionId: websocketSessionId,
|
||||
requestUrl,
|
||||
headers: prepared.headers,
|
||||
body: prepared.body,
|
||||
if (codexWebsocketChannel) {
|
||||
const downstreamHeaders: Record<string, unknown> = {
|
||||
...(request.headers as Record<string, unknown>),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
};
|
||||
const providerHeaders = buildOauthProviderHeaders({
|
||||
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
|
||||
? codexWebsocketChannel.account.extraConfig
|
||||
: null,
|
||||
downstreamHeaders,
|
||||
});
|
||||
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
|
||||
for (const payload of runtimeResult.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
} catch (error) {
|
||||
const runtimeError = error instanceof CodexWebsocketRuntimeError
|
||||
? error
|
||||
: new CodexWebsocketRuntimeError('upstream websocket request failed');
|
||||
if (runtimeError.status === 426) {
|
||||
const fallbackOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: false,
|
||||
});
|
||||
if (fallbackOutput) {
|
||||
lastResponseOutput = fallbackOutput;
|
||||
}
|
||||
return;
|
||||
}
|
||||
lastResponseOutput = collectResponsesOutput(runtimeError.events);
|
||||
for (const payload of runtimeError.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
writeResponsesWebsocketError(
|
||||
socket,
|
||||
runtimeError.status || 408,
|
||||
runtimeError.message,
|
||||
runtimeError.payload,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const prepared = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
|
||||
stream: true,
|
||||
tokenValue: codexWebsocketChannel.tokenValue,
|
||||
sitePlatform: codexWebsocketChannel.site.platform,
|
||||
siteUrl: codexWebsocketChannel.site.url,
|
||||
openaiBody: normalized.request,
|
||||
downstreamFormat: 'responses',
|
||||
responsesOriginalBody: normalized.request,
|
||||
downstreamHeaders,
|
||||
providerHeaders,
|
||||
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
|
||||
});
|
||||
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
|
||||
|
||||
const forwardedOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: supportsIncrementalInput,
|
||||
});
|
||||
if (forwardedOutput) {
|
||||
lastResponseOutput = forwardedOutput;
|
||||
try {
|
||||
const runtimeResult = await codexWebsocketRuntime.sendRequest({
|
||||
sessionId: websocketSessionId,
|
||||
requestUrl,
|
||||
headers: prepared.headers,
|
||||
body: prepared.body,
|
||||
});
|
||||
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
|
||||
for (const payload of runtimeResult.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
} catch (error) {
|
||||
const runtimeError = error instanceof CodexWebsocketRuntimeError
|
||||
? error
|
||||
: new CodexWebsocketRuntimeError('upstream websocket request failed');
|
||||
if (runtimeError.status && runtimeError.events.length === 0) {
|
||||
const fallbackOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: false,
|
||||
});
|
||||
if (fallbackOutput) {
|
||||
lastResponseOutput = fallbackOutput;
|
||||
}
|
||||
return;
|
||||
}
|
||||
lastResponseOutput = collectResponsesOutput(runtimeError.events);
|
||||
for (const payload of runtimeError.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
writeResponsesWebsocketError(
|
||||
socket,
|
||||
runtimeError.status || 408,
|
||||
runtimeError.message,
|
||||
runtimeError.payload,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const forwardedOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: supportsIncrementalInput,
|
||||
});
|
||||
if (forwardedOutput) {
|
||||
lastResponseOutput = forwardedOutput;
|
||||
}
|
||||
} catch {
|
||||
writeResponsesWebsocketError(socket, 500, 'internal websocket proxy error');
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -588,10 +640,24 @@ export function ensureResponsesWebsocketTransport(app: FastifyInstance) {
|
||||
});
|
||||
|
||||
app.server.on('upgrade', (request, socket, head) => {
|
||||
const url = new URL(request.url || '/', 'http://localhost');
|
||||
if (url.pathname !== '/v1/responses') return;
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
void handleResponsesWebsocketConnection(app, client, request);
|
||||
void (async () => {
|
||||
const url = new URL(request.url || '/', 'http://localhost');
|
||||
if (url.pathname !== '/v1/responses') return;
|
||||
const token = extractWebsocketAuthToken(request, url);
|
||||
if (!token) {
|
||||
writeUpgradeHttpError(socket, 401, 'Missing Authorization, x-api-key, x-goog-api-key, or key query parameter');
|
||||
return;
|
||||
}
|
||||
const authResult = await authorizeDownstreamToken(token);
|
||||
if (!authResult.ok) {
|
||||
writeUpgradeHttpError(socket, authResult.statusCode, authResult.error);
|
||||
return;
|
||||
}
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
void handleResponsesWebsocketConnection(app, client, request, authResult);
|
||||
});
|
||||
})().catch(() => {
|
||||
writeUpgradeHttpError(socket, 500, 'internal websocket proxy error');
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -581,7 +581,7 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
expect(request.body.prompt_cache_key).toBeUndefined();
|
||||
});
|
||||
|
||||
it('preserves explicit prompt_cache_key for native codex responses requests', () => {
|
||||
it('preserves explicit prompt_cache_key for native codex responses requests without mapping it into codex session headers', () => {
|
||||
const request = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: 'gpt-5.4',
|
||||
@@ -599,11 +599,10 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
providerHeaders: {
|
||||
Originator: 'codex_cli_rs',
|
||||
},
|
||||
codexExplicitSessionId: 'codex-cache-123',
|
||||
} as any);
|
||||
|
||||
expect(request.headers.Session_id).toBe('codex-cache-123');
|
||||
expect(request.headers.Conversation_id).toBe('codex-cache-123');
|
||||
expect(request.headers.Session_id).toMatch(/^[0-9a-f-]{36}$/i);
|
||||
expect(request.headers.Conversation_id).toBeUndefined();
|
||||
expect(request.body.prompt_cache_key).toBe('codex-cache-123');
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user