From ccda5440e180bd860b1fdfe8b4e01f53474bb043 Mon Sep 17 00:00:00 2001 From: Alix-007 Date: Thu, 16 Jul 2026 15:46:07 +0800 Subject: [PATCH] fix(discord): bound stalled probe response reads (#106036) * fix(discord): bound stalled probe response reads * fix(discord): bound application probe bodies * fix(discord): preserve probe result within status budget * style(discord): format probe timeout test * chore(deadcode): allowlist test-only timeout constant exports * refactor(discord): keep status probe budget in loader * fix(discord): enforce total probe deadline Co-authored-by: Alix-007 --------- Co-authored-by: Peter Steinberger --- extensions/discord/src/channel.loaders.ts | 13 ++ extensions/discord/src/channel.test.ts | 24 ++- extensions/discord/src/channel.ts | 5 +- extensions/discord/src/probe.intents.test.ts | 166 +++++++++++++++++++ extensions/discord/src/probe.ts | 117 +++++++++---- 5 files changed, 286 insertions(+), 39 deletions(-) diff --git a/extensions/discord/src/channel.loaders.ts b/extensions/discord/src/channel.loaders.ts index 799d40b2ab1b..9cf852602bef 100644 --- a/extensions/discord/src/channel.loaders.ts +++ b/extensions/discord/src/channel.loaders.ts @@ -1,5 +1,6 @@ // Discord plugin module implements channel.loaders behavior. import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; +import type { DiscordProbe } from "./probe.js"; export const loadDiscordDirectoryConfigModule = createLazyRuntimeModule( () => import("./directory-config.js"), @@ -23,6 +24,18 @@ export const loadDiscordProviderRuntime = createLazyRuntimeModule( export const loadDiscordProbeRuntime = createLazyRuntimeModule(() => import("./probe.runtime.js")); +export async function probeDiscordStatusAccount(params: { + token: string; + timeoutMs: number; +}): Promise { + const startedAtMs = Date.now(); + const runtime = await loadDiscordProbeRuntime(); + // The gateway starts its hook deadline before lazy plugin loading. Carry only the remaining + // budget into the probe or a cold import can let optional metadata outrun the caller. + const remainingMs = Math.max(1, params.timeoutMs - Math.max(0, Date.now() - startedAtMs)); + return await runtime.probeDiscord(params.token, remainingMs, { includeApplication: true }); +} + export const loadDiscordAuditModule = createLazyRuntimeModule(() => import("./audit.js")); export const loadDiscordSendModule = createLazyRuntimeModule(() => import("./send.js")); diff --git a/extensions/discord/src/channel.test.ts b/extensions/discord/src/channel.test.ts index f7dfab7382c5..30dcfbceb763 100644 --- a/extensions/discord/src/channel.test.ts +++ b/extensions/discord/src/channel.test.ts @@ -569,12 +569,34 @@ describe("discordPlugin outbound", () => { cfg, }); - expect(probeDiscordMock).toHaveBeenCalledWith("discord-token", 5000, { + expect(probeDiscordMock).toHaveBeenCalledWith("discord-token", expect.any(Number), { includeApplication: true, }); + const forwardedTimeoutMs = Number(argAt(probeDiscordMock, 0, 1)); + expect(forwardedTimeoutMs).toBeGreaterThan(0); + expect(forwardedTimeoutMs).toBeLessThanOrEqual(5_000); expect(runtimeProbeDiscord).not.toHaveBeenCalled(); }); + it("subtracts lazy probe loading from the status budget", async () => { + const nowSpy = vi.spyOn(Date, "now").mockReturnValueOnce(1_000).mockReturnValueOnce(1_200); + probeDiscordMock.mockResolvedValue({ ok: true, elapsedMs: 1 }); + try { + const cfg = createCfg(); + await discordPlugin.status!.probeAccount!({ + account: resolveAccount(cfg), + timeoutMs: 5_000, + cfg, + }); + + expect(probeDiscordMock).toHaveBeenCalledWith("discord-token", 4_800, { + includeApplication: true, + }); + } finally { + nowSpy.mockRestore(); + } + }); + it("reports missing voice permissions in targeted capabilities diagnostics", async () => { const fetchPermissionsSpy = vi .spyOn(sendModule, "fetchChannelPermissionsDiscord") diff --git a/extensions/discord/src/channel.ts b/extensions/discord/src/channel.ts index 7f8e76da0204..6cab1e3f270c 100644 --- a/extensions/discord/src/channel.ts +++ b/extensions/discord/src/channel.ts @@ -61,6 +61,7 @@ import { loadDiscordSendModule, loadDiscordTargetResolverModule, loadDiscordThreadBindingsManagerModule, + probeDiscordStatusAccount, } from "./channel.loaders.js"; import { openDiscordCommandDeployHashStore } from "./command-deploy-store.js"; import { shouldSuppressLocalDiscordExecApprovalPrompt } from "./exec-approvals.js"; @@ -531,9 +532,7 @@ export const discordPlugin: ChannelPlugin buildChannelSummary: ({ snapshot }) => buildTokenChannelStatusSummary(snapshot, { includeMode: false }), probeAccount: async ({ account, timeoutMs }) => - (await loadDiscordProbeRuntime()).probeDiscord(account.token, timeoutMs, { - includeApplication: true, - }), + await probeDiscordStatusAccount({ token: account.token, timeoutMs }), formatCapabilitiesProbe: ({ probe }) => { const discordProbe = probe as DiscordProbe | undefined; const lines = []; diff --git a/extensions/discord/src/probe.intents.test.ts b/extensions/discord/src/probe.intents.test.ts index 61029134cc07..241c9999e2e9 100644 --- a/extensions/discord/src/probe.intents.test.ts +++ b/extensions/discord/src/probe.intents.test.ts @@ -31,6 +31,61 @@ function oversizedDiscordProbeJsonResponse(onCancel: () => void): Response { return response; } +function abortableTricklingDiscordProbeJsonResponse( + signal: AbortSignal | null | undefined, + onAbort: () => void, +): Response { + let interval: ReturnType | undefined; + return new Response( + new ReadableStream({ + start(controller) { + const abort = () => { + if (interval) { + clearInterval(interval); + } + onAbort(); + controller.error(signal?.reason ?? new Error("request aborted")); + }; + if (signal?.aborted) { + abort(); + return; + } + controller.enqueue(new TextEncoder().encode('{"id":"bot-1"')); + interval = setInterval(() => controller.enqueue(new Uint8Array([0x20])), 10); + signal?.addEventListener("abort", abort, { once: true }); + }, + cancel() { + if (interval) { + clearInterval(interval); + } + }, + }), + { headers: { "content-type": "application/json" }, status: 200 }, + ); +} + +function abortableStalledDiscordJsonResponse( + signal: AbortSignal | null | undefined, + onAbort: () => void, +): Response { + return new Response( + new ReadableStream({ + start(controller) { + const abort = () => { + onAbort(); + controller.error(signal?.reason ?? new Error("request aborted")); + }; + if (signal?.aborted) { + abort(); + return; + } + signal?.addEventListener("abort", abort, { once: true }); + }, + }), + { headers: { "content-type": "application/json" }, status: 200 }, + ); +} + describe("resolveDiscordPrivilegedIntentsFromFlags", () => { beforeEach(() => { vi.useRealTimers(); @@ -141,6 +196,117 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { expect(cancelCount).toBe(1); }); + it("times out and cancels stalled getMe probe JSON response bodies", async () => { + vi.useFakeTimers(); + try { + let abortCount = 0; + const fetcher = withFetchPreconnect(async (_input, init) => + abortableStalledDiscordJsonResponse(init?.signal, () => { + abortCount += 1; + }), + ); + + const probe = probeDiscord("MTIz.abc.def", 50, { fetcher }); + const assertion = expect(probe).resolves.toMatchObject({ + ok: false, + error: "discord.probe.getMe: JSON response timed out after 50ms", + }); + + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(50); + await assertion; + expect(abortCount).toBe(1); + } finally { + vi.useRealTimers(); + } + }); + + it("uses one total deadline across getMe headers and a trickling JSON body", async () => { + vi.useFakeTimers(); + try { + let abortCount = 0; + const fetcher = withFetchPreconnect(async (_input, init) => { + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + return abortableTricklingDiscordProbeJsonResponse(init?.signal, () => { + abortCount += 1; + }); + }); + + const probe = probeDiscord("MTIz.abc.def", 50, { fetcher }); + const assertion = expect(probe).resolves.toMatchObject({ + ok: false, + error: "discord.probe.getMe: JSON response timed out after 50ms", + }); + + await vi.advanceTimersByTimeAsync(50); + await assertion; + expect(abortCount).toBe(1); + } finally { + vi.useRealTimers(); + } + }); + + it("bounds stalled application-summary response bodies during probes", async () => { + vi.useFakeTimers(); + try { + let abortCount = 0; + const fetcher = withFetchPreconnect(async (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url.endsWith("/users/@me")) { + return jsonResponse({ id: "bot-1", username: "openclaw" }); + } + return abortableStalledDiscordJsonResponse(init?.signal, () => { + abortCount += 1; + }); + }); + + const probe = probeDiscord("MTIz.abc.def", 50, { + fetcher, + includeApplication: true, + }); + const outerStatusResult = Promise.race([ + probe, + new Promise<{ ok: false; timedOut: true }>((resolve) => { + setTimeout(() => resolve({ ok: false, timedOut: true }), 50); + }), + ]); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(50); + + await expect(outerStatusResult).resolves.toMatchObject({ + ok: true, + bot: { id: "bot-1", username: "openclaw" }, + }); + expect(abortCount).toBe(1); + } finally { + vi.useRealTimers(); + } + }); + + it("bounds stalled application-id response bodies", async () => { + vi.useFakeTimers(); + try { + let abortCount = 0; + const fetcher = withFetchPreconnect(async (_input, init) => + abortableStalledDiscordJsonResponse(init?.signal, () => { + abortCount += 1; + }), + ); + + const lookup = fetchDiscordApplicationId("unparseable.token", 50, fetcher); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(50); + + await expect(lookup).resolves.toBeUndefined(); + expect(abortCount).toBe(1); + } finally { + vi.useRealTimers(); + } + }); + it("derives application id from parseable tokens before probing REST", async () => { let calls = 0; const fetcher = withFetchPreconnect(async () => { diff --git a/extensions/discord/src/probe.ts b/extensions/discord/src/probe.ts index a8bfb40cd7f6..b942def5cdd5 100644 --- a/extensions/discord/src/probe.ts +++ b/extensions/discord/src/probe.ts @@ -1,13 +1,17 @@ // Discord plugin module implements probe behavior. import type { BaseProbeResult } from "openclaw/plugin-sdk/channel-contract"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { buildTimeoutAbortSignal } from "openclaw/plugin-sdk/extension-shared"; import { resolveFetch } from "openclaw/plugin-sdk/fetch-runtime"; -import { readProviderJsonResponse } from "openclaw/plugin-sdk/provider-http"; +import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { fetchWithTimeout } from "openclaw/plugin-sdk/text-utility-runtime"; import { DiscordApiError, fetchDiscord } from "./api.js"; import { normalizeDiscordToken } from "./token.js"; const DISCORD_API_BASE = "https://discord.com/api/v10"; +const DISCORD_PROBE_GET_ME_LABEL = "discord.probe.getMe"; +const DISCORD_PROBE_JSON_MAX_BYTES = 16 * 1024 * 1024; +const DISCORD_PROBE_COMPLETION_RESERVE_MAX_MS = 25; export type DiscordProbe = BaseProbeResult & { status?: number | null; @@ -50,25 +54,14 @@ async function fetchDiscordApplicationMe( return await fetchDiscord<{ id?: string; flags?: number }>( "/oauth2/applications/@me", normalized, - createDiscordTimeoutFetch(fetcher, timeoutMs), - { retry: { attempts: 1 } }, + fetcher, + { retry: { attempts: 1 }, timeoutMs }, ); } catch { return undefined; } } -function createDiscordTimeoutFetch(fetcher: typeof fetch, timeoutMs: number): typeof fetch { - const fetchImpl = getResolvedFetch(fetcher); - return ((input: RequestInfo | URL, init?: RequestInit) => - fetchWithTimeout( - typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url, - init ?? {}, - timeoutMs, - fetchImpl, - )) as typeof fetch; -} - export function resolveDiscordPrivilegedIntentsFromFlags( flags: number, ): DiscordPrivilegedIntentsSummary { @@ -121,6 +114,24 @@ function getResolvedFetch(fetcher: typeof fetch): typeof fetch { return fetchImpl; } +async function readDiscordProbeGetMeJson( + response: Response, + timeoutMs: number, +): Promise<{ id?: string; username?: string }> { + const bytes = await readResponseWithLimit(response, DISCORD_PROBE_JSON_MAX_BYTES, { + chunkTimeoutMs: timeoutMs, + onIdleTimeout: ({ chunkTimeoutMs }) => + new Error(`${DISCORD_PROBE_GET_ME_LABEL}: JSON response stalled after ${chunkTimeoutMs}ms`), + onOverflow: ({ maxBytes }) => + new Error(`${DISCORD_PROBE_GET_ME_LABEL}: JSON response exceeds ${maxBytes} bytes`), + }); + try { + return JSON.parse(new TextDecoder().decode(bytes)) as { id?: string; username?: string }; + } catch (cause) { + throw new Error(`${DISCORD_PROBE_GET_ME_LABEL}: malformed JSON response`, { cause }); + } +} + export async function probeDiscord( token: string, timeoutMs: number, @@ -145,29 +156,64 @@ export async function probeDiscord( } let res: Response | undefined; try { - res = await fetchWithTimeout( - `${DISCORD_API_BASE}/users/@me`, - { headers: { Authorization: `Bot ${normalized}` } }, + const getMeUrl = `${DISCORD_API_BASE}/users/@me`; + const getMeTimeout = buildTimeoutAbortSignal({ timeoutMs, - getResolvedFetch(fetcher), - ); - if (!res.ok) { - result.status = res.status; - result.error = `getMe failed (${res.status})`; - return { ...result, elapsedMs: Date.now() - started }; + operation: DISCORD_PROBE_GET_ME_LABEL, + url: getMeUrl, + }); + try { + res = await fetchWithTimeout( + getMeUrl, + { + headers: { Authorization: `Bot ${normalized}` }, + signal: getMeTimeout.signal, + }, + timeoutMs, + getResolvedFetch(fetcher), + ); + if (!res.ok) { + result.status = res.status; + result.error = `getMe failed (${res.status})`; + return { ...result, elapsedMs: Date.now() - started }; + } + let json: { id?: string; username?: string }; + try { + json = await readDiscordProbeGetMeJson(res, timeoutMs); + } catch (error) { + if (getMeTimeout.signal?.aborted) { + const message = `${DISCORD_PROBE_GET_ME_LABEL}: JSON response timed out after ${timeoutMs}ms`; + if (error instanceof Error) { + error.message = message; + throw error; + } + throw new Error(message, { cause: error }); + } + throw error; + } + result.ok = true; + result.bot = { + id: json.id ?? null, + username: json.username ?? null, + }; + } finally { + // The timeout must outlive header receipt so the same caller budget covers body reads. + getMeTimeout.cleanup(); } - const json = await readProviderJsonResponse<{ id?: string; username?: string }>( - res, - "discord.probe.getMe", - ); - result.ok = true; - result.bot = { - id: json.id ?? null, - username: json.username ?? null, - }; if (includeApplication) { - result.application = - (await fetchDiscordApplicationSummary(normalized, timeoutMs, fetcher)) ?? undefined; + // Application metadata is optional. Keep its deadline inside the outer status budget so a + // stalled secondary response cannot discard the already-resolved bot identity. + const elapsedMs = Math.max(0, Date.now() - started); + const completionReserveMs = Math.min( + DISCORD_PROBE_COMPLETION_RESERVE_MAX_MS, + Math.max(1, Math.floor(timeoutMs / 10)), + ); + const applicationTimeoutMs = Math.floor(timeoutMs - elapsedMs - completionReserveMs); + if (applicationTimeoutMs > 0) { + result.application = + (await fetchDiscordApplicationSummary(normalized, applicationTimeoutMs, fetcher)) ?? + undefined; + } } return { ...result, elapsedMs: Date.now() - started }; } catch (err) { @@ -229,7 +275,8 @@ export async function fetchDiscordApplicationId( const json = await fetchDiscord<{ id?: string }>( "/oauth2/applications/@me", normalized, - createDiscordTimeoutFetch(fetcher, timeoutMs), + fetcher, + { timeoutMs }, ); if (json?.id) { return json.id;