fix(minimax): add timeouts to OAuth HTTP requests (#102862)

* fix(minimax): add timeouts to OAuth HTTP requests * fix(minimax): remove duplicate OAuth abort signals * refactor(minimax): keep OAuth timeout internal * test(minimax): make OAuth timeout proof deterministic --------- Co-authored-by: llagy009 <0668001470@xydigit.com> Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-08-12 21:53:00 -06:00 · 2026-07-11 13:02:34 +08:00
parent b09f9b0482
commit 8f8aad9ae6
2 changed files with 299 additions and 0 deletions
@@ -1,8 +1,12 @@
 // Minimax tests cover oauth plugin behavior.
+import { createServer } from "node:http";
+import type { Socket } from "node:net";
 import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { loginMiniMaxPortalOAuth, normalizeOAuthExpires } from "./oauth.js";

+const MINIMAX_OAUTH_FETCH_TIMEOUT_MS = 30_000;
+
 function cancelTrackedResponse(
  text: string,
  init: ResponseInit,
@@ -25,6 +29,190 @@ function cancelTrackedResponse(
  };
 }

+function timeoutResult<T>(value: T, timeoutMs: number): Promise<T> {
+  return new Promise((resolve) => {
+    setTimeout(() => resolve(value), timeoutMs);
+  });
+}
+
+function captureMiniMaxOAuthFetchTimeout() {
+  const originalSetTimeout = globalThis.setTimeout;
+  let fireTimeout: (() => void) | undefined;
+  const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout").mockImplementation(((
+    callback: (...args: unknown[]) => void,
+    timeout?: number,
+    ...args: unknown[]
+  ) => {
+    if (timeout === MINIMAX_OAUTH_FETCH_TIMEOUT_MS) {
+      fireTimeout = () => callback(...args);
+      return 0 as unknown as ReturnType<typeof setTimeout>;
+    }
+    return originalSetTimeout(() => callback(...args), timeout);
+  }) as typeof setTimeout);
+  return {
+    setTimeoutSpy,
+    fire() {
+      if (!fireTimeout) {
+        throw new Error("expected MiniMax OAuth fetch timeout to be scheduled");
+      }
+      const callback = fireTimeout;
+      fireTimeout = undefined;
+      callback();
+    },
+  };
+}
+
+async function listenOnLoopback(server: ReturnType<typeof createServer>): Promise<number> {
+  return await new Promise((resolve, reject) => {
+    const onError = (err: Error) => reject(err);
+    server.once("error", onError);
+    server.listen(0, "127.0.0.1", () => {
+      server.off("error", onError);
+      const address = server.address();
+      if (!address || typeof address === "string") {
+        reject(new Error("expected loopback TCP address"));
+        return;
+      }
+      resolve(address.port);
+    });
+  });
+}
+
+async function startHangingLoopbackServer(): Promise<{
+  origin: string;
+  requests: string[];
+  waitForRequestCount: (count: number) => Promise<void>;
+  close: () => Promise<void>;
+}> {
+  type RequestWaiter = {
+    count: number;
+    resolve: () => void;
+    reject: (error: Error) => void;
+    timer?: ReturnType<typeof setTimeout>;
+  };
+
+  const sockets = new Set<Socket>();
+  const requests: string[] = [];
+  const waiters: RequestWaiter[] = [];
+
+  const resolveWaiters = () => {
+    for (let index = waiters.length - 1; index >= 0; index -= 1) {
+      const waiter = waiters[index];
+      if (!waiter || requests.length < waiter.count) {
+        continue;
+      }
+      waiters.splice(index, 1);
+      if (waiter.timer) {
+        clearTimeout(waiter.timer);
+      }
+      waiter.resolve();
+    }
+  };
+
+  const server = createServer((req, _res) => {
+    requests.push(req.url ?? "");
+    req.resume();
+    resolveWaiters();
+  });
+  server.on("connection", (socket) => {
+    sockets.add(socket);
+    socket.on("close", () => sockets.delete(socket));
+  });
+
+  const port = await listenOnLoopback(server);
+  return {
+    origin: `http://127.0.0.1:${port}`,
+    requests,
+    waitForRequestCount: async (count: number) => {
+      if (requests.length >= count) {
+        return;
+      }
+      await new Promise<void>((resolve, reject) => {
+        const waiter: RequestWaiter = {
+          count,
+          resolve,
+          reject,
+        };
+        waiter.timer = setTimeout(() => {
+          const index = waiters.indexOf(waiter);
+          if (index >= 0) {
+            waiters.splice(index, 1);
+          }
+          reject(new Error(`server received ${requests.length} request(s), expected ${count}`));
+        }, 2_000);
+        waiters.push(waiter);
+      });
+    },
+    close: async () => {
+      for (const waiter of waiters.splice(0)) {
+        if (waiter.timer) {
+          clearTimeout(waiter.timer);
+        }
+        waiter.reject(new Error("server closed"));
+      }
+      for (const socket of sockets) {
+        socket.destroy();
+      }
+      await new Promise<void>((resolve, reject) => {
+        server.close((err) => {
+          if (err) {
+            reject(err);
+            return;
+          }
+          resolve();
+        });
+      });
+    },
+  };
+}
+
+async function expectFetchWithoutDeadlineToStayPending(params: {
+  url: string;
+  init?: RequestInit;
+  waitForRequest: () => Promise<void>;
+}) {
+  const controller = new AbortController();
+  const request = fetch(params.url, { ...params.init, signal: controller.signal });
+  request.catch(() => undefined);
+  await params.waitForRequest();
+
+  const result = await Promise.race([
+    request.then(
+      () => "settled" as const,
+      () => "settled" as const,
+    ),
+    timeoutResult("pending" as const, 30),
+  ]);
+
+  controller.abort();
+  await request.catch(() => undefined);
+  expect(result).toBe("pending");
+}
+
+async function loginOutcomeWithin(
+  promise: Promise<unknown>,
+  timeoutMs: number,
+): Promise<
+  | { status: "pending" }
+  | { status: "resolved" }
+  | {
+      status: "rejected";
+      error: unknown;
+    }
+> {
+  return await Promise.race([
+    promise.then(
+      () => ({ status: "resolved" as const }),
+      (error: unknown) => ({ status: "rejected" as const, error }),
+    ),
+    timeoutResult({ status: "pending" as const }, timeoutMs),
+  ]);
+}
+
+function expectAbortOrTimeoutError(error: unknown) {
+  expect(error).toHaveProperty("name", expect.stringMatching(/^(AbortError|TimeoutError)$/));
+}
+
 afterEach(() => {
  vi.useRealTimers();
  vi.restoreAllMocks();
@@ -52,6 +240,114 @@ describe("normalizeOAuthExpires", () => {
 });

 describe("loginMiniMaxPortalOAuth", () => {
+  it("times out authorization code HTTP requests against a hanging loopback server", async () => {
+    const realFetch = fetch;
+    const server = await startHangingLoopbackServer();
+    const oauthTimeout = captureMiniMaxOAuthFetchTimeout();
+    let loginPromise: Promise<unknown> | undefined;
+
+    try {
+      await expectFetchWithoutDeadlineToStayPending({
+        url: `${server.origin}/control`,
+        init: { method: "POST", body: "response_type=code" },
+        waitForRequest: () => server.waitForRequestCount(1),
+      });
+
+      const fetchMock = vi.fn(
+        async (_input: RequestInfo | URL, init?: RequestInit) =>
+          await realFetch(`${server.origin}/device-code`, init),
+      );
+      vi.stubGlobal("fetch", fetchMock);
+
+      loginPromise = loginMiniMaxPortalOAuth({
+        openUrl: vi.fn(async () => undefined),
+        note: vi.fn(async () => undefined),
+        progress: { update: vi.fn(), stop: vi.fn() },
+      });
+      loginPromise.catch(() => undefined);
+
+      await server.waitForRequestCount(2);
+      oauthTimeout.fire();
+      const result = await loginOutcomeWithin(loginPromise, 2_000);
+      if (result.status !== "rejected") {
+        throw new Error(`expected authorization code request to reject, got ${result.status}`);
+      }
+      expectAbortOrTimeoutError(result.error);
+      expect(server.requests).toContain("/device-code");
+      expect(fetchMock.mock.calls[0]?.[1]?.signal).toBeInstanceOf(AbortSignal);
+      expect(
+        oauthTimeout.setTimeoutSpy.mock.calls.some(
+          ([, timeout]) => timeout === MINIMAX_OAUTH_FETCH_TIMEOUT_MS,
+        ),
+      ).toBe(true);
+    } finally {
+      await server.close();
+      await loginPromise?.catch(() => undefined);
+    }
+  });
+
+  it("times out token polling HTTP requests against a hanging loopback server", async () => {
+    const realFetch = fetch;
+    const server = await startHangingLoopbackServer();
+    const oauthTimeout = captureMiniMaxOAuthFetchTimeout();
+    let loginPromise: Promise<unknown> | undefined;
+
+    try {
+      await expectFetchWithoutDeadlineToStayPending({
+        url: `${server.origin}/control`,
+        init: { method: "POST", body: "grant_type=user_code" },
+        waitForRequest: () => server.waitForRequestCount(1),
+      });
+
+      let callCount = 0;
+      const fetchMock = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+        callCount += 1;
+        const body =
+          init?.body instanceof URLSearchParams
+            ? init.body
+            : new URLSearchParams(typeof init?.body === "string" ? init.body : "");
+        if (callCount === 1) {
+          return new Response(
+            JSON.stringify({
+              user_code: "CODE",
+              verification_uri: "https://example.com/device",
+              expired_in: Date.now() + 10_000,
+              state: body.get("state"),
+            }),
+            { status: 200, headers: { "Content-Type": "application/json" } },
+          );
+        }
+        return await realFetch(`${server.origin}/token`, init);
+      });
+      vi.stubGlobal("fetch", fetchMock);
+
+      loginPromise = loginMiniMaxPortalOAuth({
+        openUrl: vi.fn(async () => undefined),
+        note: vi.fn(async () => undefined),
+        progress: { update: vi.fn(), stop: vi.fn() },
+      });
+      loginPromise.catch(() => undefined);
+
+      await server.waitForRequestCount(2);
+      oauthTimeout.fire();
+      const result = await loginOutcomeWithin(loginPromise, 2_000);
+      if (result.status !== "rejected") {
+        throw new Error(`expected token polling request to reject, got ${result.status}`);
+      }
+      expectAbortOrTimeoutError(result.error);
+      expect(server.requests).toContain("/token");
+      expect(fetchMock.mock.calls[1]?.[1]?.signal).toBeInstanceOf(AbortSignal);
+      expect(
+        oauthTimeout.setTimeoutSpy.mock.calls.some(
+          ([, timeout]) => timeout === MINIMAX_OAUTH_FETCH_TIMEOUT_MS,
+        ),
+      ).toBe(true);
+    } finally {
+      await server.close();
+      await loginPromise?.catch(() => undefined);
+    }
+  });
+
  it("bounds authorization error bodies without using response.text()", async () => {
    const tracked = cancelTrackedResponse(
      `${"minimax authorization unavailable ".repeat(1024)}tail`,
@@ -34,6 +34,7 @@ const MINIMAX_OAUTH_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:user_code";
 const MINIMAX_RELATIVE_EXPIRY_SECONDS_THRESHOLD = 1_000_000_000;
 const MINIMAX_ABSOLUTE_EXPIRY_MS_THRESHOLD = 1_000_000_000_000;
 const MINIMAX_OAUTH_ERROR_BODY_LIMIT_BYTES = 8 * 1024;
+const MINIMAX_OAUTH_FETCH_TIMEOUT_MS = 30_000;

 function getOAuthEndpoints(region: MiniMaxRegion) {
  const config = MINIMAX_OAUTH_CONFIG[region];
@@ -115,6 +116,7 @@ async function requestOAuthCode(params: {
        state: params.state,
      }),
    },
+    timeoutMs: MINIMAX_OAUTH_FETCH_TIMEOUT_MS,
    policy: { allowedHostnames: [endpoints.hostname] },
    auditContext: "minimax.oauth.code",
  });
@@ -168,6 +170,7 @@ async function pollOAuthToken(params: {
        code_verifier: params.verifier,
      }),
    },
+    timeoutMs: MINIMAX_OAUTH_FETCH_TIMEOUT_MS,
    policy: { allowedHostnames: [endpoints.hostname] },
    auditContext: "minimax.oauth.token",
  });