luulam/9router
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(proxy): add proxy pool and per-connection binding + strictProxy support

Browse Source 
- Centralize proxy management with reusable proxy pools
- Per-connection proxy binding with legacy fallback
- Add strictProxy option: fail hard instead of silently falling back to direct
- Resolve alicode-intl conflict: keep alicode-intl support + proxy support

Made-with: Cursor
This commit is contained in:
decolua
2026-03-09 15:46:06 +07:00
parent 4c469291a1
commit 880f4eca91
22 changed files with 1811 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
open-sse/executors/antigravity.js
View File

@@ -2,6 +2,7 @@ import crypto from "crypto";
import { BaseExecutor } from "./base.js";
import { PROVIDERS, OAUTH_ENDPOINTS, HTTP_STATUS, ANTIGRAVITY_HEADERS, INTERNAL_REQUEST_HEADER } from "../config/constants.js";
import { deriveSessionId } from "../utils/sessionManager.js";
import { proxyAwareFetch } from "../utils/proxyFetch.js";
const MAX_RETRY_AFTER_MS = 10000;
@@ -161,7 +162,7 @@ export class AntigravityExecutor extends BaseExecutor {
return totalMs > 0 ? totalMs : null;
}
async execute({ model, body, stream, credentials, signal, log }) {
async execute({ model, body, stream, credentials, signal, log, proxyOptions = null }) {
const fallbackCount = this.getFallbackCount();
let lastError = null;
let lastStatus = 0;
@@ -180,12 +181,12 @@ export class AntigravityExecutor extends BaseExecutor {
}
try {
const response = await fetch(url, {
const response = await proxyAwareFetch(url, {
method: "POST",
headers,
body: JSON.stringify(transformedBody),
signal
});
}, proxyOptions);
if (response.status === HTTP_STATUS.RATE_LIMITED || response.status === HTTP_STATUS.SERVICE_UNAVAILABLE) {
// Try to get retry time from headers first

7
open-sse/executors/base.js
View File

@@ -1,4 +1,5 @@
import { HTTP_STATUS } from "../config/constants.js";
import { proxyAwareFetch } from "../utils/proxyFetch.js";
/**
* BaseExecutor - Base class for provider executors
@@ -75,7 +76,7 @@ export class BaseExecutor {
return { status: response.status, message: bodyText || `HTTP ${response.status}` };
}
async execute({ model, body, stream, credentials, signal, log }) {
async execute({ model, body, stream, credentials, signal, log, proxyOptions = null }) {
const fallbackCount = this.getFallbackCount();
let lastError = null;
let lastStatus = 0;
@@ -86,12 +87,12 @@ export class BaseExecutor {
const transformedBody = this.transformRequest(model, body, stream, credentials);
try {
const response = await fetch(url, {
const response = await proxyAwareFetch(url, {
method: "POST",
headers,
body: JSON.stringify(transformedBody),
signal
});
}, proxyOptions);
if (this.shouldRetry(response.status, urlIndex)) {
log?.debug?.("RETRY", `${response.status} on ${url}, trying fallback ${urlIndex + 1}`);

14
open-sse/executors/cursor.js
View File

@@ -8,6 +8,7 @@ import {
import { estimateUsage } from "../utils/usageTracking.js";
import { FORMATS } from "../translator/formats.js";
import { buildCursorRequest } from "../translator/request/openai-to-cursor.js";
import { proxyAwareFetch } from "../utils/proxyFetch.js";
import crypto from "crypto";
import { v5 as uuidv5 } from "uuid";
import zlib from "zlib";
@@ -163,13 +164,13 @@ export class CursorExecutor extends BaseExecutor {
return generateCursorBody(messages, model, tools, reasoningEffort);
}
async makeFetchRequest(url, headers, body, signal) {
const response = await fetch(url, {
async makeFetchRequest(url, headers, body, signal, proxyOptions = null) {
const response = await proxyAwareFetch(url, {
method: "POST",
headers,
body,
signal
});
}, proxyOptions);
return {
status: response.status,
@@ -227,15 +228,16 @@ export class CursorExecutor extends BaseExecutor {
});
}
async execute({ model, body, stream, credentials, signal, log }) {
async execute({ model, body, stream, credentials, signal, log, proxyOptions = null }) {
const url = this.buildUrl();
const headers = this.buildHeaders(credentials);
const transformedBody = this.transformRequest(model, body, stream, credentials);
try {
const response = http2
const shouldForceFetch = proxyOptions?.enabled === true || proxyOptions?.connectionProxyEnabled === true;
const response = (http2 && !shouldForceFetch)
? await this.makeHttp2Request(url, headers, transformedBody, signal)
: await this.makeFetchRequest(url, headers, transformedBody, signal);
: await this.makeFetchRequest(url, headers, transformedBody, signal, proxyOptions);
if (response.status !== 200) {
const errorText = response.body?.toString() || "Unknown error";

9
open-sse/executors/github.js
View File

@@ -4,6 +4,7 @@ import { openaiToOpenAIResponsesRequest } from "../translator/request/openai-res
import { openaiResponsesToOpenAIResponse } from "../translator/response/openai-responses.js";
import { initState } from "../translator/index.js";
import { parseSSELine, formatSSE } from "../utils/streamHelpers.js";
import { proxyAwareFetch } from "../utils/proxyFetch.js";
import crypto from "crypto";
export class GithubExecutor extends BaseExecutor {
@@ -86,7 +87,7 @@ export class GithubExecutor extends BaseExecutor {
body: this.sanitizeMessagesForChatCompletions(options.body)
};
const result = await super.execute(sanitizedOptions);
const result = await super.execute({ ...sanitizedOptions, proxyOptions: options.proxyOptions || null });
if (result.response.status === HTTP_STATUS.BAD_REQUEST) {
const errorBody = await result.response.clone().text();
@@ -101,7 +102,7 @@ export class GithubExecutor extends BaseExecutor {
return result;
}
async executeWithResponsesEndpoint({ model, body, stream, credentials, signal, log }) {
async executeWithResponsesEndpoint({ model, body, stream, credentials, signal, log, proxyOptions = null }) {
const url = this.config.responsesUrl;
const headers = this.buildHeaders(credentials, stream);
@@ -109,12 +110,12 @@ export class GithubExecutor extends BaseExecutor {
log?.debug("GITHUB", "Sending translated request to /responses");
const response = await fetch(url, {
const response = await proxyAwareFetch(url, {
method: "POST",
headers,
body: JSON.stringify(transformedBody),
signal
});
}, proxyOptions);
if (!response.ok) {
return { response, url, headers, transformedBody };

7
open-sse/executors/kiro.js
View File

@@ -2,6 +2,7 @@ import { BaseExecutor } from "./base.js";
import { PROVIDERS } from "../config/constants.js";
import { v4 as uuidv4 } from "uuid";
import { refreshKiroToken } from "../services/tokenRefresh.js";
import { proxyAwareFetch } from "../utils/proxyFetch.js";
/**
* KiroExecutor - Executor for Kiro AI (AWS CodeWhisperer)
@@ -33,17 +34,17 @@ export class KiroExecutor extends BaseExecutor {
/**
* Custom execute for Kiro - handles AWS EventStream binary response
*/
async execute({ model, body, stream, credentials, signal, log }) {
async execute({ model, body, stream, credentials, signal, log, proxyOptions = null }) {
const url = this.buildUrl(model, stream, 0);
const headers = this.buildHeaders(credentials, stream);
const transformedBody = this.transformRequest(model, body, stream, credentials);
const response = await fetch(url, {
const response = await proxyAwareFetch(url, {
method: "POST",
headers,
body: JSON.stringify(transformedBody),
signal
});
}, proxyOptions);
if (!response.ok) {
return { response, url, headers, transformedBody };

32
open-sse/handlers/chatCore.js
View File

@@ -67,10 +67,38 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
log, provider, model
});
const proxyOptions = {
connectionProxyEnabled: credentials?.providerSpecificData?.connectionProxyEnabled === true,
connectionProxyUrl: credentials?.providerSpecificData?.connectionProxyUrl || "",
connectionNoProxy: credentials?.providerSpecificData?.connectionNoProxy || "",
};
if (proxyOptions.connectionProxyEnabled && proxyOptions.connectionProxyUrl) {
let maskedProxyUrl = proxyOptions.connectionProxyUrl;
try {
const parsed = new URL(proxyOptions.connectionProxyUrl);
const host = parsed.hostname || "";
const port = parsed.port ? `:${parsed.port}` : "";
const protocol = parsed.protocol || "http:";
maskedProxyUrl = `${protocol}//${host}${port}`;
} catch {
// Keep raw if URL parsing fails
}
const poolId = credentials?.providerSpecificData?.connectionProxyPoolId || "none";
const connectionName = credentials?.connectionName || credentials?.connectionId || "unknown";
log?.info?.("PROXY", `${provider.toUpperCase()} | ${model} | conn=${connectionName} | pool=${poolId} | url=${maskedProxyUrl}`);
}
if (proxyOptions.connectionProxyEnabled && proxyOptions.connectionNoProxy) {
const connectionName = credentials?.connectionName || credentials?.connectionId || "unknown";
log?.debug?.("PROXY", `${provider.toUpperCase()} | ${model} | conn=${connectionName} | no_proxy=${proxyOptions.connectionNoProxy}`);
}
// Execute request
let providerResponse, providerUrl, providerHeaders, finalBody;
try {
const result = await executor.execute({ model, body: translatedBody, stream, credentials, signal: streamController.signal, log });
const result = await executor.execute({ model, body: translatedBody, stream, credentials, signal: streamController.signal, log, proxyOptions });
providerResponse = result.response;
providerUrl = result.url;
providerHeaders = result.headers;
@@ -106,7 +134,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
Object.assign(credentials, newCredentials);
if (onCredentialsRefreshed) await onCredentialsRefreshed(newCredentials);
try {
const retryResult = await executor.execute({ model, body: translatedBody, stream, credentials, signal: streamController.signal, log });
const retryResult = await executor.execute({ model, body: translatedBody, stream, credentials, signal: streamController.signal, log, proxyOptions });
if (retryResult.response.ok) { providerResponse = retryResult.response; providerUrl = retryResult.url; }
} catch { log?.warn?.("TOKEN", `${provider.toUpperCase()} | retry after refresh failed`); }
} else {

124
open-sse/utils/proxyFetch.js
View File

@@ -1,8 +1,7 @@
const isCloud = typeof caches !== "undefined" && typeof caches === "object";
const originalFetch = globalThis.fetch;
let proxyDispatcher = null;
let proxyDispatcherUrl = null;
const proxyDispatchers = new Map();
// Constants
const DNS_CACHE = {};
@@ -14,12 +13,17 @@ const HTTPS_PORT = 443;
const HTTP_SUCCESS_MIN = 200;
const HTTP_SUCCESS_MAX = 300;
function normalizeString(value) {
if (value === undefined || value === null) return "";
return String(value).trim();
}
/**
* Resolve real IP using Google DNS (bypass system DNS)
*/
async function resolveRealIP(hostname) {
if (DNS_CACHE[hostname]) return DNS_CACHE[hostname];
try {
const dns = await import("dns");
const { promisify } = await import("util");
@@ -40,11 +44,11 @@ async function resolveRealIP(hostname) {
*/
function shouldBypassMitmDns(url, options) {
if (!options?.headers) return false;
const headers = options.headers;
const hasLocalMarker = headers[MITM_BYPASS_HEADER] === MITM_BYPASS_VALUE ||
const hasLocalMarker = headers[MITM_BYPASS_HEADER] === MITM_BYPASS_VALUE ||
headers[MITM_BYPASS_HEADER.charAt(0).toUpperCase() + MITM_BYPASS_HEADER.slice(1)] === MITM_BYPASS_VALUE;
if (!hasLocalMarker) {
// Debug: log when bypass is not triggered
const hostname = new URL(url).hostname;
@@ -53,37 +57,39 @@ function shouldBypassMitmDns(url, options) {
}
return false;
}
const hostname = new URL(url).hostname;
return MITM_BYPASS_HOSTS.some(host => hostname.includes(host));
}
function shouldBypassByNoProxy(targetUrl, noProxyValue) {
const noProxy = normalizeString(noProxyValue);
if (!noProxy) return false;
const hostname = new URL(targetUrl).hostname.toLowerCase();
const patterns = noProxy.split(",").map((p) => p.trim().toLowerCase()).filter(Boolean);
return patterns.some((pattern) => {
if (pattern === "*") return true;
if (pattern.startsWith(".")) return hostname.endsWith(pattern) || hostname === pattern.slice(1);
return hostname === pattern || hostname.endsWith(`.${pattern}`);
});
}
/**
* Get proxy URL from environment
*/
function getProxyUrl(targetUrl) {
function getEnvProxyUrl(targetUrl) {
const noProxy = process.env.NO_PROXY || process.env.no_proxy;
if (noProxy) {
const hostname = new URL(targetUrl).hostname.toLowerCase();
const patterns = noProxy.split(",").map(p => p.trim().toLowerCase());
const shouldBypass = patterns.some(pattern => {
if (pattern === "*") return true;
if (pattern.startsWith(".")) return hostname.endsWith(pattern) || hostname === pattern.slice(1);
return hostname === pattern || hostname.endsWith(`.${pattern}`);
});
if (shouldBypass) return null;
}
if (shouldBypassByNoProxy(targetUrl, noProxy)) return null;
const protocol = new URL(targetUrl).protocol;
if (protocol === "https:") {
return process.env.HTTPS_PROXY || process.env.https_proxy ||
return process.env.HTTPS_PROXY || process.env.https_proxy ||
process.env.ALL_PROXY || process.env.all_proxy;
}
return process.env.HTTP_PROXY || process.env.http_proxy ||
process.env.ALL_PROXY || process.env.all_proxy;
}
@@ -92,33 +98,45 @@ function getProxyUrl(targetUrl) {
* Normalize proxy URL (allow host:port)
*/
function normalizeProxyUrl(proxyUrl) {
if (!proxyUrl) return null;
const normalizedInput = normalizeString(proxyUrl);
if (!normalizedInput) return null;
try {
// eslint-disable-next-line no-new
new URL(proxyUrl);
return proxyUrl;
new URL(normalizedInput);
return normalizedInput;
} catch {
// Allow "127.0.0.1:7890" style values
return `http://${proxyUrl}`;
return `http://${normalizedInput}`;
}
}
function resolveConnectionProxyUrl(targetUrl, proxyOptions) {
const enabled = proxyOptions?.enabled === true || proxyOptions?.connectionProxyEnabled === true;
if (!enabled) return null;
const proxyUrlRaw = normalizeString(proxyOptions?.url ?? proxyOptions?.connectionProxyUrl);
if (!proxyUrlRaw) return null;
const noProxy = normalizeString(proxyOptions?.noProxy ?? proxyOptions?.connectionNoProxy);
if (noProxy && shouldBypassByNoProxy(targetUrl, noProxy)) return null;
return normalizeProxyUrl(proxyUrlRaw);
}
/**
* Create proxy dispatcher lazily (undici-compatible)
* Closes old dispatcher when proxy URL changes to prevent connection pool leak
*/
async function getDispatcher(proxyUrl) {
const normalized = normalizeProxyUrl(proxyUrl);
if (!normalized) return null;
if (!proxyDispatcher || proxyDispatcherUrl !== normalized) {
try { proxyDispatcher?.close?.(); } catch { /* ignore */ }
if (!proxyDispatchers.has(normalized)) {
const { ProxyAgent } = await import("undici");
proxyDispatcher = new ProxyAgent({ uri: normalized });
proxyDispatcherUrl = normalized;
proxyDispatchers.set(normalized, new ProxyAgent({ uri: normalized }));
}
return proxyDispatcher;
return proxyDispatchers.get(normalized);
}
/**
@@ -128,10 +146,10 @@ async function createBypassRequest(parsedUrl, realIP, options) {
const https = await import("https");
const net = await import("net");
const { Readable } = require("stream");
return new Promise((resolve, reject) => {
const socket = new net.Socket();
socket.connect(HTTPS_PORT, realIP, () => {
const reqOptions = {
socket,
@@ -144,7 +162,7 @@ async function createBypassRequest(parsedUrl, realIP, options) {
Host: parsedUrl.hostname,
},
};
const req = https.request(reqOptions, (res) => {
const response = {
ok: res.statusCode >= HTTP_SUCCESS_MIN && res.statusCode < HTTP_SUCCESS_MAX,
@@ -161,24 +179,21 @@ async function createBypassRequest(parsedUrl, realIP, options) {
};
resolve(response);
});
req.on("error", reject);
if (options.body) {
req.write(typeof options.body === "string" ? options.body : JSON.stringify(options.body));
}
req.end();
});
socket.on("error", reject);
});
}
/**
* Patched fetch with proxy support and MITM DNS bypass
*/
async function patchedFetch(url, options = {}) {
export async function proxyAwareFetch(url, options = {}, proxyOptions = null) {
const targetUrl = typeof url === "string" ? url : url.toString();
// MITM DNS bypass: resolve real IP for googleapis.com when x-request-source: local
if (shouldBypassMitmDns(targetUrl, options)) {
try {
@@ -189,22 +204,35 @@ async function patchedFetch(url, options = {}) {
console.warn(`[ProxyFetch] MITM bypass failed: ${error.message}`);
}
}
// Normal proxy handling
const proxyUrl = normalizeProxyUrl(getProxyUrl(targetUrl));
const connectionProxyUrl = resolveConnectionProxyUrl(targetUrl, proxyOptions);
const envProxyUrl = connectionProxyUrl ? null : normalizeProxyUrl(getEnvProxyUrl(targetUrl));
const proxyUrl = connectionProxyUrl || envProxyUrl;
if (proxyUrl) {
try {
const dispatcher = await getDispatcher(proxyUrl);
return await originalFetch(url, { ...options, dispatcher });
} catch (proxyError) {
// If strictProxy is enabled, fail hard instead of falling back to direct
if (proxyOptions?.strictProxy === true) {
throw new Error(`[ProxyFetch] Proxy required but failed (strictProxy=true): ${proxyError.message}`);
}
console.warn(`[ProxyFetch] Proxy failed, falling back to direct: ${proxyError.message}`);
return originalFetch(url, options);
}
}
return originalFetch(url, options);
}
/**
* Patched global fetch with env-proxy support and MITM DNS bypass
*/
async function patchedFetch(url, options = {}) {
return proxyAwareFetch(url, options, null);
}
// Idempotency guard — only patch once to avoid wrapping multiple times
if (!isCloud && globalThis.fetch !== patchedFetch) {
globalThis.fetch = patchedFetch;