From 24478135dc255f49ab10a89c00bbb5f6830dcfb2 Mon Sep 17 00:00:00 2001 From: "Kai (Tam Nhu) Tran" <61256810+kaitranntt@users.noreply.github.com> Date: Tue, 19 May 2026 08:13:40 -0400 Subject: [PATCH] fix(proxy): scope insecure TLS to routed profile Squash merge PR #1299 into dev. --- src/proxy/server/messages-route.ts | 36 +++- tests/unit/proxy/messages-route.test.ts | 217 +++++++++++++++++++++--- 2 files changed, 231 insertions(+), 22 deletions(-) diff --git a/src/proxy/server/messages-route.ts b/src/proxy/server/messages-route.ts index 317f2256..5165ae2b 100644 --- a/src/proxy/server/messages-route.ts +++ b/src/proxy/server/messages-route.ts @@ -1,4 +1,5 @@ import * as http from 'http'; +import { Agent } from 'undici'; import type { Dispatcher } from 'undici'; import type { OpenAICompatProfileConfig } from '../profile-router'; import { resolveProxyRequestRoute } from '../request-router'; @@ -267,13 +268,29 @@ export async function handleProxyMessagesRequest( } ); + const useSharedInsecureDispatcher = + insecureDispatcher !== undefined && + upstream.route.profile.profileName === profile.profileName; + const useProfileInsecureTls = upstream.route.profile.insecure === true; + const ephemeralInsecureDispatcher = + useProfileInsecureTls && !useSharedInsecureDispatcher + ? new Agent({ connect: { rejectUnauthorized: false } }) + : undefined; + const dispatcher = useSharedInsecureDispatcher + ? insecureDispatcher + : useProfileInsecureTls + ? ephemeralInsecureDispatcher + : undefined; + try { logger.stage('dispatch', 'upstream.dispatch', 'Dispatching upstream fetch', { profileName: profile.profileName, + routedProfileName: upstream.route.profile.profileName, + insecureTls: dispatcher !== undefined, }); const upstreamResponse = await fetch( resolveOpenAIChatCompletionsUrl(upstream.route.profile.baseUrl), - buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher) + buildFetchInit(upstream.route.profile, upstream.body, controller.signal, dispatcher) ); logger.stage('upstream', 'upstream.response', 'Received upstream response', { profileName: profile.profileName, @@ -288,6 +305,23 @@ export async function handleProxyMessagesRequest( } finally { clearTimeout(timeout); cleanupDisconnectHandlers(); + if (ephemeralInsecureDispatcher) { + try { + await ephemeralInsecureDispatcher.close(); + } catch (closeError) { + logger.stage( + 'cleanup', + 'request.dispatcher_close_failed', + 'Failed to close per-request insecure dispatcher', + { + profileName: profile.profileName, + routedProfileName: upstream.route.profile.profileName, + error: closeError instanceof Error ? closeError.message : String(closeError), + }, + { level: 'warn' } + ); + } + } } } catch (error) { const message = error instanceof Error ? error.message : 'Unknown proxy error'; diff --git a/tests/unit/proxy/messages-route.test.ts b/tests/unit/proxy/messages-route.test.ts index f118a794..b4fcbbe9 100644 --- a/tests/unit/proxy/messages-route.test.ts +++ b/tests/unit/proxy/messages-route.test.ts @@ -1,34 +1,129 @@ -import { describe, expect, it } from 'bun:test'; +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; import { EventEmitter } from 'events'; -import { attachDisconnectAbortHandlers } from '../../../src/proxy/server/messages-route'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { PassThrough } from 'stream'; +import { Agent } from 'undici'; +import { resolveOpenAICompatProfileConfig } from '../../../src/proxy/profile-router'; +import { + attachDisconnectAbortHandlers, + handleProxyMessagesRequest, +} from '../../../src/proxy/server/messages-route'; +import { loadSettings } from '../../../src/utils/config-manager'; class FakeSocket extends EventEmitter { destroyed = false; } -class FakeRequest extends EventEmitter { - destroyed = false; +class FakeRequest extends PassThrough { + headers: Record; + method = 'POST'; + url = '/v1/messages'; socket = new FakeSocket(); + + constructor(headers: Record = {}) { + super(); + this.headers = headers; + } } -class FakeResponse extends EventEmitter { - destroyed = false; - writableEnded = false; +class FakeResponse extends PassThrough { + headers = new Map(); + statusCode = 200; socket = new FakeSocket(); + + setHeader(name: string, value: string): this { + this.headers.set(name.toLowerCase(), value); + return this; + } + + writeHead(statusCode: number, headers: Record): this { + this.statusCode = statusCode; + for (const [name, value] of Object.entries(headers)) { + this.setHeader(name, value); + } + return this; + } } +let originalCcsHome: string | undefined; +let originalFetch: typeof globalThis.fetch; +let originalAgentClose: typeof Agent.prototype.close; +let tempDir: string; + +function writeSettings(profileName: string, env: Record): string { + const settingsPath = path.join(tempDir, '.ccs', `${profileName}.settings.json`); + fs.writeFileSync(settingsPath, JSON.stringify({ env }, null, 2), 'utf8'); + return settingsPath; +} + +function buildProfile(profileName: string) { + const settingsPath = path.join(tempDir, '.ccs', `${profileName}.settings.json`); + const profile = resolveOpenAICompatProfileConfig( + profileName, + settingsPath, + loadSettings(settingsPath).env || {} + ); + expect(profile).toBeTruthy(); + return profile!; +} + +beforeEach(() => { + originalCcsHome = process.env.CCS_HOME; + originalFetch = globalThis.fetch; + originalAgentClose = Agent.prototype.close; + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccs-messages-route-')); + fs.mkdirSync(path.join(tempDir, '.ccs'), { recursive: true }); + process.env.CCS_HOME = tempDir; + + const profiles = { + hf: writeSettings('hf', { + ANTHROPIC_BASE_URL: 'https://router.huggingface.co/v1', + ANTHROPIC_AUTH_TOKEN: 'hf_token', + ANTHROPIC_MODEL: 'hf-default', + CCS_DROID_PROVIDER: 'generic-chat-completion-api', + }), + deepseek: writeSettings('deepseek', { + ANTHROPIC_BASE_URL: 'https://api.deepseek.com/v1', + ANTHROPIC_AUTH_TOKEN: 'deepseek_token', + ANTHROPIC_MODEL: 'deepseek-chat', + CCS_DROID_PROVIDER: 'generic-chat-completion-api', + CCS_OPENAI_PROXY_INSECURE: 'true', + }), + search: writeSettings('search', { + ANTHROPIC_BASE_URL: 'https://search.example.com/v1', + ANTHROPIC_AUTH_TOKEN: 'search_token', + ANTHROPIC_MODEL: 'sonar-pro', + CCS_DROID_PROVIDER: 'generic-chat-completion-api', + }), + }; + + fs.writeFileSync( + path.join(tempDir, '.ccs', 'config.json'), + JSON.stringify({ profiles, proxy: { routing: {} } }, null, 2), + 'utf8' + ); +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + Agent.prototype.close = originalAgentClose; + if (originalCcsHome !== undefined) { + process.env.CCS_HOME = originalCcsHome; + } else { + delete process.env.CCS_HOME; + } + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + describe('attachDisconnectAbortHandlers', () => { it('cleans up registered listeners after the request completes', () => { const req = new FakeRequest(); const res = new FakeResponse(); const controller = new AbortController(); - const cleanup = attachDisconnectAbortHandlers( - req as never, - res as never, - controller, - () => {} - ); + const cleanup = attachDisconnectAbortHandlers(req as never, res as never, controller, () => {}); expect(req.listenerCount('aborted')).toBe(1); expect(req.listenerCount('close')).toBe(0); @@ -49,14 +144,9 @@ describe('attachDisconnectAbortHandlers', () => { const controller = new AbortController(); let disconnectCount = 0; - const cleanup = attachDisconnectAbortHandlers( - req as never, - res as never, - controller, - () => { - disconnectCount += 1; - } - ); + const cleanup = attachDisconnectAbortHandlers(req as never, res as never, controller, () => { + disconnectCount += 1; + }); req.emit('aborted'); req.socket.emit('close'); @@ -68,3 +158,88 @@ describe('attachDisconnectAbortHandlers', () => { cleanup(); }); }); + +describe('handleProxyMessagesRequest', () => { + it('uses a per-request insecure dispatcher for routed profiles and closes it on failure', async () => { + const activeProfile = buildProfile('hf'); + const sharedDispatcher = { name: 'shared-insecure-dispatcher' } as never; + let closeCalls = 0; + let capturedDispatcher: unknown; + + Agent.prototype.close = function close(): Promise { + closeCalls += 1; + return Promise.resolve(); + }; + + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + capturedDispatcher = (init as RequestInit & { dispatcher?: unknown })?.dispatcher; + throw new Error('upstream exploded'); + }) as typeof globalThis.fetch; + + const req = new FakeRequest({ + 'x-api-key': 'local-token', + }); + const res = new FakeResponse(); + const pending = handleProxyMessagesRequest( + req as never, + res as never, + activeProfile, + 'local-token', + sharedDispatcher + ); + req.end( + JSON.stringify({ + model: 'deepseek:deepseek-chat', + stream: true, + messages: [{ role: 'user', content: 'route this' }], + }) + ); + await pending; + + expect(capturedDispatcher).toBeInstanceOf(Agent); + expect(capturedDispatcher).not.toBe(sharedDispatcher); + expect(closeCalls).toBe(1); + expect(res.statusCode).toBe(502); + }); + + it('reuses the shared insecure dispatcher when the request stays on the active profile', async () => { + const activeProfile = buildProfile('hf'); + const sharedDispatcher = { name: 'shared-insecure-dispatcher' } as never; + let closeCalls = 0; + let capturedDispatcher: unknown; + + Agent.prototype.close = function close(): Promise { + closeCalls += 1; + return Promise.resolve(); + }; + + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + capturedDispatcher = (init as RequestInit & { dispatcher?: unknown })?.dispatcher; + throw new Error('upstream exploded'); + }) as typeof globalThis.fetch; + + const req = new FakeRequest({ + 'x-api-key': 'local-token', + }); + const res = new FakeResponse(); + const pending = handleProxyMessagesRequest( + req as never, + res as never, + activeProfile, + 'local-token', + sharedDispatcher + ); + req.end( + JSON.stringify({ + model: 'hf-default', + stream: true, + messages: [{ role: 'user', content: 'stay local' }], + }) + ); + await pending; + + expect(capturedDispatcher).toBe(sharedDispatcher); + expect(closeCalls).toBe(0); + expect(res.statusCode).toBe(502); + }); +});