fix(web-search): bind each search provider attempt to its connection proxy (#9201)

* fix(web-search): bind each search provider attempt to its connection proxy (#9201) The search path resolved credentials but never resolved the connection proxy, so the upstream fetch always egressed directly. The connection-test path already used the proxy correctly, proving the gap was in the data-plane transport binding. - Resolve the connection proxy before each upstream attempt using the existing resolveProxyForConnection(connectionId, apiKeyId, providerId) precedence chain, then wrap the fetch in runWithProxyContext so the patched globalThis.fetch routes through the configured proxy. - Resolve and bind the alternate connection proxy independently during failover, so the primary account's context never leaks into the fallback. - Carry connectionId and apiKeyId through SearchHandlerOptions into the route and executeWebSearch callers. - Add connectionId to all saveCallLog entries in tryProvider, so the regular call log identifies the account. - Emit a sanitized logProxyEvent per real upstream search attempt with provider, connection ID, proxy level, status, duration, and target origin/path (no query, API key, or proxy credentials). - Cover both POST /v1/search and executeWebSearch() consumers (MCP, internal, skills) since both bypassed the same proxy binding. * fix(sse): extract search proxy binding into leaf module to fit file-size cap Move the per-attempt proxy resolution, proxied fetch, sanitized proxy-event emission, and response handling for web search providers out of open-sse/handlers/search.ts into a new open-sse/handlers/search/searchProxy.ts, so the provider-dispatch chokepoint (tryProvider) stays a thin wiring call and search.ts fits back under the frozen file-size cap (1536 lines). --------- Co-authored-by: diegosouzapw <diegosouzapw@users.noreply.github.com>
2026-08-15 19:52:50 +03:00 · 2026-08-07 18:09:08 -03:00
parent 09e4c150c1
commit df64220087
6 changed files with 431 additions and 104 deletions
--- a/changelog.d/fixes/9201-web-search-proxy-bind.plan.md
+++ b/changelog.d/fixes/9201-web-search-proxy-bind.plan.md
@@ -0,0 +1 @@
+- fix(web-search): bind each search provider attempt to its connection proxy (#9201)
--- a/open-sse/handlers/search.ts
+++ b/open-sse/handlers/search.ts
@@ -27,6 +27,7 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js";
 import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
 import { z } from "zod";
 import { sanitizeErrorMessage } from "../utils/error.ts";
+import { resolveSearchProxy, executeProviderFetch } from "./search/searchProxy.ts";

 export interface SearchResult {
  title: string;
@@ -96,6 +97,9 @@ interface SearchHandlerOptions {
  alternateProvider?: string;
  alternateCredentials?: Record<string, any> | null;
  log?: any;
+  /** Connection ID (proxy resolution + call-log attribution) and API key ID (per-key proxy). */
+  connectionId?: string;
+  apiKeyId?: string;
 }

 // ── Constants ────────────────────────────────────────────────────────────
@@ -1195,6 +1199,8 @@ export async function handleSearch(options: SearchHandlerOptions): Promise<Searc
    alternateProvider,
    alternateCredentials,
    log,
+    connectionId,
+    apiKeyId,
  } = options;
  const startTime = Date.now();

@@ -1238,7 +1244,15 @@ export async function handleSearch(options: SearchHandlerOptions): Promise<Searc
  }

  // 4. Try primary provider
-  const result = await tryProvider(primaryConfig, requestParams, credentials, startTime, log);
+  const result = await tryProvider(
+    primaryConfig,
+    requestParams,
+    credentials,
+    startTime,
+    log,
+    connectionId,
+    apiKeyId
+  );

  if (result.success) return result;

@@ -1256,12 +1270,15 @@ export async function handleSearch(options: SearchHandlerOptions): Promise<Searc
      );
    }

+    // Resolve alternate connection proxy independently so primary context never leaks
    const fallbackResult = await tryProvider(
      alternateConfig,
      requestParams,
      alternateCredentials,
      startTime,
-      log
+      log,
+      alternateCredentials?.connectionId,
+      apiKeyId
    );

    if (fallbackResult.success) return fallbackResult;
@@ -1374,7 +1391,9 @@ async function tryProvider(
  params: Omit<SearchRequestParams, "token">,
  credentials: Record<string, any>,
  globalStartTime: number,
-  log?: any
+  log?: any,
+  connectionId?: string,
+  apiKeyId?: string
 ): Promise<SearchHandlerResult> {
  const startTime = Date.now();
  const providerSpecificData =
@@ -1421,6 +1440,10 @@ async function tryProvider(
    };
  }

+  // Resolve proxy for the selected connection (see search/searchProxy.ts for the
+  // resolveProxyForConnection precedence chain: per-key, account, provider, combo, global).
+  const { proxy, proxyLevel } = await resolveSearchProxy(connectionId, apiKeyId, config.id);
+
  // Timeout: min of provider timeout and remaining global timeout
  const remainingGlobal = GLOBAL_TIMEOUT_MS - (Date.now() - globalStartTime);
  const timeout = Math.min(config.timeoutMs, Math.max(remainingGlobal, 1000));
@@ -1431,105 +1454,22 @@ async function tryProvider(
    log.info("SEARCH", `${config.id} | query: "${query.slice(0, 80)}" | type: ${searchType}`);
  }

-  try {
-    const response = await fetch(url, { ...init, signal: controller.signal });
-    clearTimeout(timer);
-
-    if (!response.ok) {
-      const errorText = await response.text();
-      if (log) {
-        log.error("SEARCH", `${config.id} error ${response.status}: ${errorText.slice(0, 200)}`);
-      }
-
-      saveCallLog({
-        method: config.method,
-        path: "/v1/search",
-        status: response.status,
-        model: config.id,
-        provider: config.id,
-        duration: Date.now() - startTime,
-        requestType: "search",
-        error: errorText.slice(0, 500),
-        requestBody: {
-          query: query.slice(0, 200),
-          search_type: searchType,
-          max_results: maxResults,
-        },
-      }).catch(() => {
-        /* non-critical — logging must not block search response */
-      });
-
-      return {
-        success: false,
-        status: response.status,
-        error: `Search provider ${config.id} returned ${response.status}`,
-      };
-    }
-
-    const data = await response.json();
-    const normalized = normalizeResponse(config.id, data, query, searchType);
-    // Enforce max_results — some providers return more than requested
-    const results = normalized.results.slice(0, maxResults);
-    const totalResults = normalized.totalResults;
-    const duration = Date.now() - startTime;
-
-    saveCallLog({
-      method: config.method,
-      path: "/v1/search",
-      status: 200,
-      model: config.id,
-      provider: config.id,
-      duration,
-      requestType: "search",
-      tokens: { prompt_tokens: 0, completion_tokens: 0 },
-      requestBody: { query: query.slice(0, 200), search_type: searchType, max_results: maxResults },
-      responseBody: { results_count: results.length, cached: false },
-    }).catch(() => {
-      /* non-critical — logging must not block search response */
-    });
-
-    return {
-      success: true,
-      data: {
-        provider: config.id,
-        query,
-        results,
-        answer: null,
-        usage: { queries_used: 1, search_cost_usd: config.costPerQuery },
-        metrics: {
-          response_time_ms: duration,
-          upstream_latency_ms: duration,
-          total_results_available: totalResults,
-        },
-        errors: [],
-      },
-    };
-  } catch (err: any) {
-    clearTimeout(timer);
-
-    const isTimeout = err.name === "AbortError";
-    if (log) {
-      log.error("SEARCH", `${config.id} ${isTimeout ? "timeout" : "fetch error"}: ${err.message}`);
-    }
-
-    saveCallLog({
-      method: config.method,
-      path: "/v1/search",
-      status: isTimeout ? 504 : 502,
-      model: config.id,
-      provider: config.id,
-      duration: Date.now() - startTime,
-      requestType: "search",
-      error: err.message,
-      requestBody: { query: query.slice(0, 200), search_type: searchType, max_results: maxResults },
-    }).catch(() => {
-      /* non-critical — logging must not block search response */
-    });
-
-    return {
-      success: false,
-      status: isTimeout ? 504 : 502,
-      error: `Search provider ${isTimeout ? "timeout" : "error"}: ${sanitizeErrorMessage(err.message)}`,
-    };
-  }
+  // Delegate the fetch + response handling (proxy fetch, call-log, sanitized
+  // proxy event, result shaping) to the shared chokepoint in searchProxy.ts.
+  return executeProviderFetch({
+    config,
+    url,
+    init,
+    controller,
+    timer,
+    query,
+    searchType,
+    maxResults,
+    startTime,
+    connectionId,
+    proxy,
+    proxyLevel,
+    log,
+    normalize: normalizeResponse,
+  });
 }
--- a/open-sse/handlers/search/searchProxy.ts
+++ b/open-sse/handlers/search/searchProxy.ts
@@ -0,0 +1,245 @@
+/**
+ * Per-attempt proxy binding for web search provider calls.
+ *
+ * Extracted from ../search.ts (tryProvider) to keep the provider-dispatch
+ * chokepoint under the frozen file-size cap. Resolves the proxy for a given
+ * connection/apiKey/provider triple, wraps a fetch in that proxy context,
+ * and emits a sanitized proxy event for observability (never includes
+ * query, API key, or proxy credentials).
+ */
+
+import { saveCallLog } from "@/lib/usageDb";
+import { sanitizeErrorMessage } from "../../utils/error.ts";
+import type { SearchProviderConfig } from "../../config/searchRegistry.ts";
+import type { SearchResult } from "../search.ts";
+
+/** Resolved proxy binding for a single provider attempt. */
+export interface ResolvedSearchProxy {
+  proxy: unknown;
+  proxyLevel: string;
+}
+
+/**
+ * Resolve the proxy for the selected connection. Uses the existing
+ * resolveProxyForConnection(connectionId, apiKeyId, providerId) precedence
+ * chain so per-key, account, provider, combo, and global proxy rules apply
+ * consistently with other data-plane routes.
+ *
+ * Never throws — proxy resolution failure must not block the search.
+ */
+export async function resolveSearchProxy(
+  connectionId: string | undefined,
+  apiKeyId: string | undefined,
+  providerId: string
+): Promise<ResolvedSearchProxy> {
+  if (!connectionId) {
+    return { proxy: null, proxyLevel: "direct" };
+  }
+  try {
+    const { resolveProxyForConnection } = await import("@/lib/db/settings");
+    const proxyInfo = await resolveProxyForConnection(connectionId, apiKeyId, providerId);
+    return { proxy: proxyInfo.proxy, proxyLevel: proxyInfo.level || "direct" };
+  } catch {
+    return { proxy: null, proxyLevel: "direct" };
+  }
+}
+
+/**
+ * Run a fetch, routed through the resolved proxy context when one is set.
+ * Wraps the patched globalThis.fetch so the upstream call egresses via the
+ * configured proxy instead of directly.
+ */
+export async function fetchWithSearchProxy(
+  proxy: unknown,
+  doFetch: () => Promise<Response>
+): Promise<Response> {
+  if (!proxy) return doFetch();
+  const { runWithProxyContext } = await import("../../utils/proxyFetch.ts");
+  return runWithProxyContext(proxy, doFetch);
+}
+
+/**
+ * Emit a sanitized proxy event for a search provider attempt.
+ * Never includes query, API key, proxy username, or proxy password.
+ */
+export async function emitSearchProxyEvent(
+  provider: string,
+  connectionId: string | undefined,
+  proxy: unknown,
+  proxyLevel: string,
+  targetUrl: string,
+  startTime: number,
+  status: string
+): Promise<void> {
+  try {
+    const { logProxyEvent } = await import("@/lib/proxyLogger");
+    let targetOrigin = "";
+    let targetPath = "";
+    try {
+      const u = new URL(targetUrl);
+      targetOrigin = u.origin;
+      targetPath = u.pathname;
+    } catch {
+      targetOrigin = targetUrl.slice(0, 80);
+    }
+    const proxyRecord =
+      proxy && typeof proxy === "object" ? (proxy as Record<string, unknown>) : null;
+    const proxyInfo = proxyRecord
+      ? {
+          type: String(proxyRecord.type || "http"),
+          host: String(proxyRecord.host || ""),
+          port: Number(proxyRecord.port || 0),
+        }
+      : null;
+    logProxyEvent({
+      status,
+      proxy: proxyInfo,
+      level: proxyLevel,
+      levelId: connectionId || null,
+      provider: provider || null,
+      targetUrl: `${targetOrigin}${targetPath}`,
+      latencyMs: Date.now() - startTime,
+      connectionId: connectionId || null,
+      account: connectionId ? connectionId.slice(0, 8) : null,
+    });
+  } catch {
+    // Non-critical — proxy logging must not block search response
+  }
+}
+
+/** Loose result shape mirroring SearchHandlerResult in ../search.ts. */
+export interface ProviderFetchResult {
+  success: boolean;
+  status?: number;
+  error?: string;
+  data?: {
+    provider: string;
+    query: string;
+    results: SearchResult[];
+    answer: null;
+    usage: { queries_used: number; search_cost_usd: number };
+    metrics: { response_time_ms: number; upstream_latency_ms: number; total_results_available: number | null };
+    errors: [];
+  };
+}
+
+/** Minimal logger shape used by the search handlers (pino-compatible). */
+export interface SearchLog {
+  info: (tag: string, message: string) => void;
+  error: (tag: string, message: string) => void;
+  warn?: (tag: string, message: string) => void;
+}
+
+export interface ExecuteProviderFetchParams {
+  config: SearchProviderConfig;
+  url: string;
+  init: RequestInit;
+  controller: AbortController;
+  timer: ReturnType<typeof setTimeout>;
+  query: string;
+  searchType: string;
+  maxResults: number;
+  startTime: number;
+  connectionId?: string;
+  proxy: unknown;
+  proxyLevel: string;
+  log?: SearchLog;
+  normalize: (
+    providerId: string,
+    data: unknown,
+    query: string,
+    searchType: string
+  ) => { results: SearchResult[]; totalResults: number | null };
+}
+
+/**
+ * Perform the upstream search HTTP call (through the resolved proxy, if any),
+ * then handle the success/error/exception branches: call-log persistence,
+ * sanitized proxy-event emission, and SearchHandlerResult construction.
+ * This is the single chokepoint tryProvider() delegates to after building
+ * the request and resolving the proxy — keeps search.ts to wiring only.
+ */
+export async function executeProviderFetch(p: ExecuteProviderFetchParams): Promise<ProviderFetchResult> {
+  const { config, url, init, controller, timer, query, searchType, maxResults, startTime } = p;
+  const { connectionId, proxy, proxyLevel, log, normalize } = p;
+  const emitEvent = (status: string) =>
+    emitSearchProxyEvent(config.id, connectionId, proxy, proxyLevel, url, startTime, status);
+  const logCall = (fields: Record<string, unknown>) =>
+    saveCallLog({
+      method: config.method,
+      path: "/v1/search",
+      model: config.id,
+      provider: config.id,
+      connectionId: connectionId || null,
+      requestType: "search",
+      requestBody: { query: query.slice(0, 200), search_type: searchType, max_results: maxResults },
+      ...fields,
+    }).catch(() => {
+      /* non-critical — logging must not block search response */
+    });
+
+  try {
+    const response = await fetchWithSearchProxy(proxy, () =>
+      fetch(url, { ...init, signal: controller.signal })
+    );
+    clearTimeout(timer);
+
+    if (!response.ok) {
+      const errorText = await response.text();
+      if (log) {
+        log.error("SEARCH", `${config.id} error ${response.status}: ${errorText.slice(0, 200)}`);
+      }
+      logCall({ status: response.status, duration: Date.now() - startTime, error: errorText.slice(0, 500) });
+      await emitEvent("error");
+      return {
+        success: false,
+        status: response.status,
+        error: `Search provider ${config.id} returned ${response.status}`,
+      };
+    }
+
+    const data = await response.json();
+    const normalized = normalize(config.id, data, query, searchType);
+    const results = normalized.results.slice(0, maxResults);
+    const duration = Date.now() - startTime;
+
+    logCall({
+      status: 200,
+      duration,
+      tokens: { prompt_tokens: 0, completion_tokens: 0 },
+      responseBody: { results_count: results.length, cached: false },
+    });
+    await emitEvent("success");
+
+    return {
+      success: true,
+      data: {
+        provider: config.id,
+        query,
+        results,
+        answer: null,
+        usage: { queries_used: 1, search_cost_usd: config.costPerQuery },
+        metrics: {
+          response_time_ms: duration,
+          upstream_latency_ms: duration,
+          total_results_available: normalized.totalResults,
+        },
+        errors: [],
+      },
+    };
+  } catch (err: unknown) {
+    clearTimeout(timer);
+    const error = err instanceof Error ? err : new Error(String(err));
+    const isTimeout = error.name === "AbortError";
+    if (log) {
+      log.error("SEARCH", `${config.id} ${isTimeout ? "timeout" : "fetch error"}: ${error.message}`);
+    }
+    logCall({ status: isTimeout ? 504 : 502, duration: Date.now() - startTime, error: error.message });
+    await emitEvent(isTimeout ? "timeout" : "error");
+    return {
+      success: false,
+      status: isTimeout ? 504 : 502,
+      error: `Search provider ${isTimeout ? "timeout" : "error"}: ${sanitizeErrorMessage(error.message)}`,
+    };
+  }
+}
--- a/src/app/api/v1/search/route.ts
+++ b/src/app/api/v1/search/route.ts
@@ -300,6 +300,8 @@ async function postHandler(request: Request, context: unknown) {
        alternateProvider: alternateProviderId,
        alternateCredentials,
        log,
+        connectionId: credentials?.connectionId || undefined,
+        apiKeyId: policy.apiKeyInfo?.id || undefined,
      });

      if (!result.success) {
--- a/src/lib/search/executeWebSearch.ts
+++ b/src/lib/search/executeWebSearch.ts
@@ -249,6 +249,8 @@ export async function executeWebSearch(
      alternateProvider: alternateProviderId,
      alternateCredentials,
      log,
+      connectionId: credentials?.connectionId || undefined,
+      apiKeyId: input.apiKeyId || undefined,
    });

    if (!result.success || !result.data) {
--- a/tests/unit/9201-search-proxy-bypass.test.ts
+++ b/tests/unit/9201-search-proxy-bypass.test.ts
@@ -0,0 +1,137 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import http from "node:http";
+import os from "node:os";
+import path from "node:path";
+
+const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-9201-search-proxy-"));
+process.env.DATA_DIR = dataDir;
+process.env.REQUIRE_API_KEY = "false";
+process.env.DASHBOARD_PASSWORD = "";
+process.env.INITIAL_PASSWORD = "";
+delete process.env.JWT_SECRET;
+delete process.env.HTTP_PROXY;
+delete process.env.HTTPS_PROXY;
+delete process.env.ALL_PROXY;
+delete process.env.http_proxy;
+delete process.env.https_proxy;
+delete process.env.all_proxy;
+process.env.NO_PROXY = "";
+process.env.no_proxy = "";
+
+const core = await import("../../src/lib/db/core.ts");
+const providersDb = await import("../../src/lib/db/providers.ts");
+const proxiesDb = await import("../../src/lib/db/proxies.ts");
+const searchRegistry = await import("../../open-sse/config/searchRegistry.ts");
+const searchRoute = await import("../../src/app/api/v1/search/route.ts");
+
+let proxyServer: http.Server;
+let proxyPort = 0;
+let connectionId = "";
+const originalSerperBaseUrl = searchRegistry.SEARCH_PROVIDERS["serper-search"].baseUrl;
+
+function listen(server: http.Server): Promise<number> {
+  return new Promise((resolve) => {
+    server.listen(0, "127.0.0.1", () => {
+      const address = server.address();
+      if (!address || typeof address === "string") throw new Error("proxy did not bind");
+      resolve(address.port);
+    });
+  });
+}
+
+test.before(async () => {
+  proxyServer = http.createServer();
+  proxyPort = await listen(proxyServer);
+
+  const connection = await providersDb.createProviderConnection({
+    provider: "serper-search",
+    authType: "apikey",
+    name: "serper-proxy-probe",
+    apiKey: "probe-serper-key",
+    isActive: true,
+    testStatus: "active",
+  });
+  connectionId = String(connection.id);
+  await proxiesDb.createProxyAndAssign(
+    { name: "search-probe-proxy", type: "http", host: "127.0.0.1", port: proxyPort },
+    { scope: "account", scopeId: connectionId }
+  );
+
+  searchRegistry.SEARCH_PROVIDERS["serper-search"].baseUrl = "http://search-probe.invalid";
+});
+
+test.after(async () => {
+  searchRegistry.SEARCH_PROVIDERS["serper-search"].baseUrl = originalSerperBaseUrl;
+  await new Promise<void>((resolve) => proxyServer.close(() => resolve()));
+  core.resetDbInstance();
+  fs.rmSync(dataDir, { recursive: true, force: true });
+});
+
+function installProxyResponseCounter() {
+  let proxyRequests = 0;
+  const payload = JSON.stringify({
+    organic: [
+      {
+        title: "Proxy-served result",
+        link: "https://example.com/proxy-served",
+        snippet: "The configured connection proxy received this request.",
+      },
+    ],
+    searchParameters: { totalResults: 1 },
+  });
+  proxyServer.removeAllListeners("request");
+  proxyServer.removeAllListeners("connect");
+  proxyServer.on("request", (_request, response) => {
+    proxyRequests += 1;
+    response.statusCode = 200;
+    response.setHeader("content-type", "application/json");
+    response.end(payload);
+  });
+  proxyServer.on("connect", (_request, socket, head) => {
+    proxyRequests += 1;
+    socket.write("HTTP/1.1 200 Connection Established\r\n\r\n");
+    const reply = () => {
+      socket.end(
+        `HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: ${Buffer.byteLength(payload)}\r\nConnection: close\r\n\r\n${payload}`
+      );
+    };
+    if (head.length > 0) reply();
+    else socket.once("data", reply);
+  });
+  return () => proxyRequests;
+}
+
+async function postSearch(query: string) {
+  return searchRoute.POST(
+    new Request("http://localhost/v1/search", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        query,
+        provider: "serper-search",
+        max_results: 1,
+        search_type: "web",
+      }),
+    })
+  );
+}
+
+test("POST /v1/search sends a connection's provider request through its configured proxy", async () => {
+  const getProxyRequests = installProxyResponseCounter();
+
+  const response = await postSearch(`proxy probe red ${Date.now()}`);
+  const body = (await response.json()) as { results?: unknown[]; error?: unknown };
+
+  assert.deepEqual(
+    {
+      status: response.status,
+      proxyRequests: getProxyRequests(),
+      resultCount: Array.isArray(body.results) ? body.results.length : 0,
+    },
+    { status: 200, proxyRequests: 1, resultCount: 1 },
+    JSON.stringify(body)
+  );
+  assert.equal(connectionId.length > 0, true);
+});
				`@@ -0,0 +1 @@`
				`- fix(web-search): bind each search provider attempt to its connection proxy (#9201)`