Merge pull request #915 from oyi77/feat/cliproxyapi-2-routing

feat(cliproxyapi): executor, proxy routing with SSRF guard & module-level cache

open-sse/executors/base.ts

@@ -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 = () => {

open-sse/executors/cliproxyapi.ts

@@ -0,0 +1,95 @@
+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 { resolveCliproxyapiBaseUrl };
+
+export class CliproxyapiExecutor extends BaseExecutor {
+  private readonly upstreamBaseUrl: string;
+
+  constructor(baseUrl?: string) {
+    const effectiveBase = baseUrl ?? resolveCliproxyapiBaseUrl();
+    super("cliproxyapi", {
+      id: "cliproxyapi",
+      baseUrl: effectiveBase + "/v1/chat/completions",
+      headers: { "Content-Type": "application/json" },
+    });
+    this.upstreamBaseUrl = effectiveBase;
+  }
+
+  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;

open-sse/executors/index.ts

@@ -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";

open-sse/handlers/chatCore.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,68 @@ 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`
+          );
+          try {
+            return await proxyExec.execute(input);
+          } catch (proxyErr) {
+            const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr);
+            log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`);
+            throw proxyErr;
+          }
+        }
+        return result;
+      } catch (err) {
+        const errMsg = err instanceof Error ? err.message : String(err);
+        log?.info?.("UPSTREAM_PROXY", `${prov} native error (${errMsg}), retrying via CLIProxyAPI`);
+        try {
+          return await proxyExec.execute(input);
+        } catch (proxyErr) {
+          const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr);
+          log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`);
+          throw proxyErr;
+        }
+      }
+    };
+    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 }

tests/unit/cliproxyapi-executor.test.mjs

@@ -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);
+    });
+  });
+});