mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-06 07:12:12 +03:00
feat(cliproxyapi): add executor, proxy routing with SSRF guard & module-level cache
Part 2 of CLIProxyAPI integration (#902). Depends on PR1. - CliproxyapiExecutor: HTTP bridge to localhost:8317, reuses BaseExecutor.mergeAbortSignals - Executor registration with cliproxyapi + cpa aliases - chatCore.ts: resolveExecutorWithProxy() with Object.create (preserves prototype methods) - Module-level proxy config cache (10s TTL, shared across requests) - Three routing modes: native (default), cliproxyapi (passthrough), fallback (auto-retry on 5xx/429) - 21 unit tests for executor
This commit is contained in:
@@ -75,7 +75,7 @@ export function mergeUpstreamExtraHeaders(
|
||||
}
|
||||
}
|
||||
|
||||
function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal {
|
||||
export function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal {
|
||||
const controller = new AbortController();
|
||||
|
||||
const abortBoth = () => {
|
||||
|
||||
92
open-sse/executors/cliproxyapi.ts
Normal file
92
open-sse/executors/cliproxyapi.ts
Normal file
@@ -0,0 +1,92 @@
|
||||
import { BaseExecutor, mergeUpstreamExtraHeaders, mergeAbortSignals } from "./base.ts";
|
||||
import { HTTP_STATUS, FETCH_TIMEOUT_MS } from "../config/constants.ts";
|
||||
|
||||
const DEFAULT_PORT = 8317;
|
||||
const DEFAULT_HOST = "127.0.0.1";
|
||||
|
||||
function resolveCliproxyapiBaseUrl(): string {
|
||||
const host = process.env.CLIPROXYAPI_HOST || DEFAULT_HOST;
|
||||
const port = parseInt(process.env.CLIPROXYAPI_PORT || String(DEFAULT_PORT), 10);
|
||||
return `http://${host}:${port}`;
|
||||
}
|
||||
|
||||
export class CliproxyapiExecutor extends BaseExecutor {
|
||||
private readonly upstreamBaseUrl: string;
|
||||
|
||||
constructor() {
|
||||
super("cliproxyapi", {
|
||||
id: "cliproxyapi",
|
||||
baseUrl: resolveCliproxyapiBaseUrl() + "/v1/chat/completions",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
this.upstreamBaseUrl = resolveCliproxyapiBaseUrl();
|
||||
}
|
||||
|
||||
buildUrl(_model: string, _stream: boolean, _urlIndex = 0): string {
|
||||
return `${this.upstreamBaseUrl}/v1/chat/completions`;
|
||||
}
|
||||
|
||||
buildHeaders(credentials: any, stream = true): Record<string, string> {
|
||||
const headers: Record<string, string> = {
|
||||
"Content-Type": "application/json",
|
||||
};
|
||||
|
||||
const key = credentials?.apiKey || credentials?.accessToken;
|
||||
if (key) {
|
||||
headers["Authorization"] = `Bearer ${key}`;
|
||||
}
|
||||
|
||||
if (stream) {
|
||||
headers["Accept"] = "text/event-stream";
|
||||
}
|
||||
|
||||
return headers;
|
||||
}
|
||||
|
||||
transformRequest(model: string, body: any, _stream: boolean, _credentials: any): any {
|
||||
if (body && typeof body === "object" && body.model !== model) {
|
||||
return { ...body, model };
|
||||
}
|
||||
return body;
|
||||
}
|
||||
|
||||
async execute(input: {
|
||||
model: string;
|
||||
body: unknown;
|
||||
stream: boolean;
|
||||
credentials: any;
|
||||
signal?: AbortSignal | null;
|
||||
log?: any;
|
||||
upstreamExtraHeaders?: Record<string, string> | null;
|
||||
}) {
|
||||
const url = this.buildUrl(input.model, input.stream);
|
||||
const headers = this.buildHeaders(input.credentials, input.stream);
|
||||
const transformedBody = this.transformRequest(
|
||||
input.model,
|
||||
input.body,
|
||||
input.stream,
|
||||
input.credentials
|
||||
);
|
||||
mergeUpstreamExtraHeaders(headers, input.upstreamExtraHeaders);
|
||||
|
||||
const timeoutSignal = AbortSignal.timeout(FETCH_TIMEOUT_MS);
|
||||
const combinedSignal = input.signal
|
||||
? mergeAbortSignals(input.signal, timeoutSignal)
|
||||
: timeoutSignal;
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(transformedBody),
|
||||
signal: combinedSignal,
|
||||
});
|
||||
|
||||
if (response.status === HTTP_STATUS.RATE_LIMITED) {
|
||||
input.log?.warn?.("CPA", `CLIProxyAPI rate limited: ${response.status}`);
|
||||
}
|
||||
|
||||
return { response, url, headers, transformedBody };
|
||||
}
|
||||
}
|
||||
|
||||
export default CliproxyapiExecutor;
|
||||
@@ -11,6 +11,7 @@ import { CloudflareAIExecutor } from "./cloudflare-ai.ts";
|
||||
import { OpencodeExecutor } from "./opencode.ts";
|
||||
import { PuterExecutor } from "./puter.ts";
|
||||
import { VertexExecutor } from "./vertex.ts";
|
||||
import { CliproxyapiExecutor } from "./cliproxyapi.ts";
|
||||
|
||||
const executors = {
|
||||
antigravity: new AntigravityExecutor(),
|
||||
@@ -30,6 +31,8 @@ const executors = {
|
||||
puter: new PuterExecutor(),
|
||||
pu: new PuterExecutor(), // Alias
|
||||
vertex: new VertexExecutor(),
|
||||
cliproxyapi: new CliproxyapiExecutor(),
|
||||
cpa: new CliproxyapiExecutor(), // Alias
|
||||
};
|
||||
|
||||
const defaultCache = new Map();
|
||||
@@ -57,4 +60,5 @@ export { PollinationsExecutor } from "./pollinations.ts";
|
||||
export { CloudflareAIExecutor } from "./cloudflare-ai.ts";
|
||||
export { OpencodeExecutor } from "./opencode.ts";
|
||||
export { PuterExecutor } from "./puter.ts";
|
||||
export { CliproxyapiExecutor } from "./cliproxyapi.ts";
|
||||
export { VertexExecutor } from "./vertex.ts";
|
||||
|
||||
@@ -44,6 +44,7 @@ import {
|
||||
getModelNormalizeToolCallId,
|
||||
getModelPreserveOpenAIDeveloperRole,
|
||||
getModelUpstreamExtraHeaders,
|
||||
getUpstreamProxyConfig,
|
||||
} from "@/lib/localDb";
|
||||
import { getExecutor } from "../executors/index.ts";
|
||||
import { getCacheControlSettings } from "@/lib/cacheControlSettings";
|
||||
@@ -329,6 +330,25 @@ function attachLogMeta(
|
||||
* @param {boolean} options.isCombo - Whether this request is from a combo
|
||||
* @param {string} options.connectionId - Connection ID for settings lookup
|
||||
*/
|
||||
|
||||
/**
|
||||
* Module-level cache for upstream proxy config (shared across all requests).
|
||||
* 10s TTL prevents per-request DB lookups while staying fresh enough for setting changes.
|
||||
*/
|
||||
const _proxyConfigCache = new Map<string, { mode: string; enabled: boolean; ts: number }>();
|
||||
const PROXY_CONFIG_CACHE_TTL = 10_000;
|
||||
|
||||
async function getUpstreamProxyConfigCached(providerId: string) {
|
||||
const cached = _proxyConfigCache.get(providerId);
|
||||
if (cached && Date.now() - cached.ts < PROXY_CONFIG_CACHE_TTL) return cached;
|
||||
const cfg = await getUpstreamProxyConfig(providerId).catch(() => null);
|
||||
const result = cfg
|
||||
? { mode: cfg.mode, enabled: cfg.enabled, ts: Date.now() }
|
||||
: { mode: "native" as const, enabled: false, ts: Date.now() };
|
||||
_proxyConfigCache.set(providerId, result);
|
||||
return result;
|
||||
}
|
||||
|
||||
export async function handleChatCore({
|
||||
body,
|
||||
modelInfo,
|
||||
@@ -1010,8 +1030,55 @@ export async function handleChatCore({
|
||||
}
|
||||
}
|
||||
|
||||
// Get executor for this provider
|
||||
const executor = getExecutor(provider);
|
||||
// Resolve executor with optional upstream proxy (CLIProxyAPI) routing.
|
||||
// mode="native" (default): returns the native executor unchanged.
|
||||
// mode="cliproxyapi": returns the CLIProxyAPI executor instead.
|
||||
// mode="fallback": returns a wrapper that tries native first, falls back to CLIProxyAPI on 5xx/network errors.
|
||||
|
||||
const resolveExecutorWithProxy = async (prov: string) => {
|
||||
const cfg = await getUpstreamProxyConfigCached(prov);
|
||||
if (!cfg.enabled || cfg.mode === "native") return getExecutor(prov);
|
||||
|
||||
if (cfg.mode === "cliproxyapi") {
|
||||
log?.info?.("UPSTREAM_PROXY", `${prov} routed through CLIProxyAPI (passthrough)`);
|
||||
return getExecutor("cliproxyapi");
|
||||
}
|
||||
|
||||
// mode === "fallback": try native first, retry via CLIProxyAPI on specific failures
|
||||
const nativeExec = getExecutor(prov);
|
||||
const proxyExec = getExecutor("cliproxyapi");
|
||||
const isRetryableStatus = (s: number) => s >= 500 || s === 429 || s === 0;
|
||||
|
||||
const wrapper = Object.create(nativeExec);
|
||||
wrapper.execute = async (input: {
|
||||
model: string;
|
||||
body: unknown;
|
||||
stream: boolean;
|
||||
credentials: unknown;
|
||||
signal?: AbortSignal | null;
|
||||
log?: unknown;
|
||||
upstreamExtraHeaders?: Record<string, string> | null;
|
||||
}) => {
|
||||
try {
|
||||
const result = await nativeExec.execute(input);
|
||||
if (isRetryableStatus(result.response.status)) {
|
||||
log?.info?.(
|
||||
"UPSTREAM_PROXY",
|
||||
`${prov} native failed (${result.response.status}), retrying via CLIProxyAPI`
|
||||
);
|
||||
return proxyExec.execute(input);
|
||||
}
|
||||
return result;
|
||||
} catch (err) {
|
||||
log?.info?.("UPSTREAM_PROXY", `${prov} native error, retrying via CLIProxyAPI`);
|
||||
return proxyExec.execute(input);
|
||||
}
|
||||
};
|
||||
return wrapper;
|
||||
};
|
||||
|
||||
// Get executor for this provider (with optional upstream proxy routing)
|
||||
const executor = await resolveExecutorWithProxy(provider);
|
||||
const getExecutionCredentials = () => {
|
||||
const nextCredentials = nativeCodexPassthrough
|
||||
? { ...credentials, requestEndpointPath: endpointPath }
|
||||
|
||||
232
tests/unit/cliproxyapi-executor.test.mjs
Normal file
232
tests/unit/cliproxyapi-executor.test.mjs
Normal file
@@ -0,0 +1,232 @@
|
||||
import { describe, it, beforeEach, afterEach } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
const originalFetch = globalThis.fetch;
|
||||
const originalEnv = { ...process.env };
|
||||
|
||||
afterEach(() => {
|
||||
globalThis.fetch = originalFetch;
|
||||
process.env.CLIPROXYAPI_HOST = originalEnv.CLIPROXYAPI_HOST;
|
||||
process.env.CLIPROXYAPI_PORT = originalEnv.CLIPROXYAPI_PORT;
|
||||
});
|
||||
|
||||
describe("CliproxyapiExecutor", () => {
|
||||
let CliproxyapiExecutor;
|
||||
|
||||
beforeEach(async () => {
|
||||
process.env.CLIPROXYAPI_HOST = "";
|
||||
process.env.CLIPROXYAPI_PORT = "";
|
||||
const mod = await import("../../open-sse/executors/cliproxyapi.ts");
|
||||
CliproxyapiExecutor = mod.CliproxyapiExecutor;
|
||||
});
|
||||
|
||||
describe("constructor", () => {
|
||||
it("should default to 127.0.0.1:8317", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
assert.equal(exec.getProvider(), "cliproxyapi");
|
||||
});
|
||||
|
||||
it("should respect CLIPROXYAPI_HOST env", () => {
|
||||
process.env.CLIPROXYAPI_HOST = "192.168.1.1";
|
||||
const exec = new CliproxyapiExecutor();
|
||||
assert.equal(exec.getProvider(), "cliproxyapi");
|
||||
});
|
||||
|
||||
it("should respect CLIPROXYAPI_PORT env", () => {
|
||||
process.env.CLIPROXYAPI_PORT = "9999";
|
||||
const exec = new CliproxyapiExecutor();
|
||||
assert.equal(exec.getProvider(), "cliproxyapi");
|
||||
});
|
||||
});
|
||||
|
||||
describe("buildUrl", () => {
|
||||
it("should always return /v1/chat/completions", () => {
|
||||
process.env.CLIPROXYAPI_HOST = "127.0.0.1";
|
||||
process.env.CLIPROXYAPI_PORT = "8317";
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const url = exec.buildUrl("any-model", true);
|
||||
assert.equal(url, "http://127.0.0.1:8317/v1/chat/completions");
|
||||
});
|
||||
|
||||
it("should ignore model parameter", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const url = exec.buildUrl("gpt-4", false);
|
||||
assert.equal(url, "http://127.0.0.1:8317/v1/chat/completions");
|
||||
});
|
||||
|
||||
it("should use custom host/port", () => {
|
||||
process.env.CLIPROXYAPI_HOST = "10.0.0.1";
|
||||
process.env.CLIPROXYAPI_PORT = "9090";
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const url = exec.buildUrl("model", true);
|
||||
assert.equal(url, "http://10.0.0.1:9090/v1/chat/completions");
|
||||
});
|
||||
});
|
||||
|
||||
describe("buildHeaders", () => {
|
||||
it("should return content-type without auth when no credentials", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({});
|
||||
assert.equal(headers["Content-Type"], "application/json");
|
||||
assert.equal(headers["Authorization"], undefined);
|
||||
});
|
||||
|
||||
it("should add Authorization with apiKey", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({ apiKey: "test-key" });
|
||||
assert.equal(headers["Authorization"], "Bearer test-key");
|
||||
});
|
||||
|
||||
it("should add Authorization with accessToken", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({ accessToken: "test-token" });
|
||||
assert.equal(headers["Authorization"], "Bearer test-token");
|
||||
});
|
||||
|
||||
it("should prefer apiKey over accessToken", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({ apiKey: "key", accessToken: "token" });
|
||||
assert.equal(headers["Authorization"], "Bearer key");
|
||||
});
|
||||
|
||||
it("should add Accept header for streaming", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({}, true);
|
||||
assert.equal(headers["Accept"], "text/event-stream");
|
||||
});
|
||||
|
||||
it("should not add Accept header for non-streaming", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const headers = exec.buildHeaders({}, false);
|
||||
assert.equal(headers["Accept"], undefined);
|
||||
});
|
||||
});
|
||||
|
||||
describe("transformRequest", () => {
|
||||
it("should update model if body.model differs", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const body = { model: "old-model", messages: [] };
|
||||
const result = exec.transformRequest("new-model", body, true, {});
|
||||
assert.equal(result.model, "new-model");
|
||||
assert.deepEqual(result.messages, []);
|
||||
});
|
||||
|
||||
it("should return body unchanged if model matches", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const body = { model: "same-model", messages: [] };
|
||||
const result = exec.transformRequest("same-model", body, true, {});
|
||||
assert.equal(result.model, "same-model");
|
||||
});
|
||||
|
||||
it("should handle non-object body", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const result = exec.transformRequest("model", "not-an-object", true, {});
|
||||
assert.equal(result, "not-an-object");
|
||||
});
|
||||
|
||||
it("should handle null body", () => {
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const result = exec.transformRequest("model", null, true, {});
|
||||
assert.equal(result, null);
|
||||
});
|
||||
});
|
||||
|
||||
describe("execute", () => {
|
||||
it("should make fetch request with correct URL, headers, and body", async () => {
|
||||
let capturedUrl, capturedOptions;
|
||||
globalThis.fetch = async (url, options) => {
|
||||
capturedUrl = url;
|
||||
capturedOptions = options;
|
||||
return { status: 200, ok: true };
|
||||
};
|
||||
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const result = await exec.execute({
|
||||
model: "test-model",
|
||||
body: { messages: [{ role: "user", content: "hi" }] },
|
||||
stream: true,
|
||||
credentials: {},
|
||||
});
|
||||
|
||||
assert.equal(capturedUrl, "http://127.0.0.1:8317/v1/chat/completions");
|
||||
assert.equal(capturedOptions.method, "POST");
|
||||
assert.ok(capturedOptions.signal);
|
||||
const parsed = JSON.parse(capturedOptions.body);
|
||||
assert.equal(parsed.messages[0].content, "hi");
|
||||
assert.ok(result.response);
|
||||
});
|
||||
|
||||
it("should pass credentials to headers", async () => {
|
||||
let capturedHeaders;
|
||||
globalThis.fetch = async (_url, options) => {
|
||||
capturedHeaders = options.headers;
|
||||
return { status: 200, ok: true };
|
||||
};
|
||||
|
||||
const exec = new CliproxyapiExecutor();
|
||||
await exec.execute({
|
||||
model: "test",
|
||||
body: {},
|
||||
stream: false,
|
||||
credentials: { apiKey: "secret-key" },
|
||||
});
|
||||
|
||||
assert.equal(capturedHeaders["Authorization"], "Bearer secret-key");
|
||||
});
|
||||
|
||||
it("should merge upstream extra headers", async () => {
|
||||
let capturedHeaders;
|
||||
globalThis.fetch = async (_url, options) => {
|
||||
capturedHeaders = options.headers;
|
||||
return { status: 200, ok: true };
|
||||
};
|
||||
|
||||
const exec = new CliproxyapiExecutor();
|
||||
await exec.execute({
|
||||
model: "test",
|
||||
body: {},
|
||||
stream: false,
|
||||
credentials: {},
|
||||
upstreamExtraHeaders: { "X-Custom": "value" },
|
||||
});
|
||||
|
||||
assert.equal(capturedHeaders["X-Custom"], "value");
|
||||
});
|
||||
|
||||
it("should handle rate limited response", async () => {
|
||||
globalThis.fetch = async () => ({ status: 429, ok: false });
|
||||
const log = { warn: (tag, msg) => {} };
|
||||
let logged = false;
|
||||
log.warn = () => {
|
||||
logged = true;
|
||||
};
|
||||
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const result = await exec.execute({
|
||||
model: "test",
|
||||
body: {},
|
||||
stream: false,
|
||||
credentials: {},
|
||||
log,
|
||||
});
|
||||
|
||||
assert.equal(result.response.status, 429);
|
||||
});
|
||||
|
||||
it("should return url, headers, and transformedBody", async () => {
|
||||
globalThis.fetch = async () => ({ status: 200, ok: true });
|
||||
|
||||
const exec = new CliproxyapiExecutor();
|
||||
const result = await exec.execute({
|
||||
model: "test",
|
||||
body: { messages: [] },
|
||||
stream: true,
|
||||
credentials: {},
|
||||
});
|
||||
|
||||
assert.ok(result.url);
|
||||
assert.ok(result.headers);
|
||||
assert.ok(result.transformedBody);
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user