diff --git a/extensions/telegram/src/api-fetch.test.ts b/extensions/telegram/src/api-fetch.test.ts index 4778d05cfe4..abaf4fe8b40 100644 --- a/extensions/telegram/src/api-fetch.test.ts +++ b/extensions/telegram/src/api-fetch.test.ts @@ -3,6 +3,35 @@ import { createRequire } from "node:module"; import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { fetchTelegramChatId } from "./api-fetch.js"; +const TELEGRAM_GETCHAT_JSON_CAP_BYTES = 4 * 1024 * 1024; + +function getChatOkResponse(id: number | string): Response { + return new Response(JSON.stringify({ ok: true, result: { id } }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); +} + +function oversizedTelegramGetChatJsonResponse(onCancel: () => void): Response { + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(TELEGRAM_GETCHAT_JSON_CAP_BYTES + 1)); + }, + cancel() { + onCancel(); + }, + }), + { headers: { "content-type": "application/json" }, status: 200 }, + ); + Object.defineProperty(response, "json", { + value: async () => { + throw new Error("unbounded json reader was used"); + }, + }); + return response; +} + const require = createRequire(import.meta.url); const EnvHttpProxyAgent = require("undici/lib/dispatcher/env-http-proxy-agent.js") as { new (opts?: Record): Record; @@ -67,18 +96,12 @@ describe("fetchTelegramChatId", () => { const cases = [ { name: "returns stringified id when Telegram getChat succeeds", - fetchImpl: vi.fn(async () => ({ - ok: true, - json: async () => ({ ok: true, result: { id: 12345 } }), - })), + fetchImpl: vi.fn(async () => getChatOkResponse(12345)), expected: "12345", }, { name: "returns null when response is not ok", - fetchImpl: vi.fn(async () => ({ - ok: false, - json: async () => ({}), - })), + fetchImpl: vi.fn(async () => new Response("{}", { status: 404 })), expected: null, }, { @@ -104,10 +127,7 @@ describe("fetchTelegramChatId", () => { } it("calls Telegram getChat endpoint", async () => { - const fetchMock = vi.fn(async () => ({ - ok: true, - json: async () => ({ ok: true, result: { id: 12345 } }), - })); + const fetchMock = vi.fn(async () => getChatOkResponse(12345)); vi.stubGlobal("fetch", fetchMock); await fetchTelegramChatId({ token: "abc", chatId: "@user" }); @@ -118,10 +138,7 @@ describe("fetchTelegramChatId", () => { }); it("uses caller-provided fetch impl when present", async () => { - const customFetch = vi.fn(async () => ({ - ok: true, - json: async () => ({ ok: true, result: { id: 12345 } }), - })); + const customFetch = vi.fn(async () => getChatOkResponse(12345)); vi.stubGlobal( "fetch", vi.fn(async () => { @@ -140,6 +157,24 @@ describe("fetchTelegramChatId", () => { undefined, ); }); + + it("returns null for oversized getChat JSON responses and cancels the stream", async () => { + let cancelCount = 0; + const fetchImpl = vi.fn(async () => + oversizedTelegramGetChatJsonResponse(() => { + cancelCount += 1; + }), + ); + + await expect( + fetchTelegramChatId({ + token: "abc", + chatId: "@user", + fetchImpl: fetchImpl as unknown as typeof fetch, + }), + ).resolves.toBeNull(); + expect(cancelCount).toBe(1); + }); }); describe("undici env proxy semantics", () => { diff --git a/extensions/telegram/src/api-fetch.ts b/extensions/telegram/src/api-fetch.ts index f5233aaf48a..4958392bfa9 100644 --- a/extensions/telegram/src/api-fetch.ts +++ b/extensions/telegram/src/api-fetch.ts @@ -1,8 +1,16 @@ // Telegram plugin module implements api fetch behavior. import type { TelegramNetworkConfig } from "openclaw/plugin-sdk/config-contracts"; +import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { resolveTelegramApiBase, resolveTelegramFetch } from "./fetch.js"; import { makeProxyFetch } from "./proxy.js"; +const TELEGRAM_BOT_API_MAX_RESPONSE_BYTES = 4 * 1024 * 1024; + +type TelegramGetChatResponse = { + ok?: boolean; + result?: { id?: number | string }; +}; + export function resolveTelegramChatLookupFetch(params?: { proxyUrl?: string; network?: TelegramNetworkConfig; @@ -47,10 +55,14 @@ export async function fetchTelegramChatId(params: { if (!res.ok) { return null; } - const data = (await res.json().catch(() => null)) as { - ok?: boolean; - result?: { id?: number | string }; - } | null; + let data: TelegramGetChatResponse | null = null; + try { + data = JSON.parse( + (await readResponseWithLimit(res, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES)).toString("utf8"), + ) as TelegramGetChatResponse; + } catch { + return null; + } const id = data?.ok ? data?.result?.id : undefined; if (typeof id === "number" || typeof id === "string") { return String(id); diff --git a/scripts/proof-telegram-getchat-bound.mjs b/scripts/proof-telegram-getchat-bound.mjs new file mode 100644 index 00000000000..516f1b840a2 --- /dev/null +++ b/scripts/proof-telegram-getchat-bound.mjs @@ -0,0 +1,159 @@ +import { once } from "node:events"; +// Loopback proof: fetchTelegramChatId (getChat) through the production Bot API read path. +import { createServer } from "node:http"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const pkgRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +const TELEGRAM_BOT_API_MAX_RESPONSE_BYTES = 4 * 1024 * 1024; + +const { fetchTelegramChatId } = await import(`${pkgRoot}/extensions/telegram/src/api-fetch.ts`); + +const CAP = TELEGRAM_BOT_API_MAX_RESPONSE_BYTES; +const STREAM_SIZE = 24 * 1024 * 1024; + +let allPassed = true; +function check(label, val) { + console.log(` ${val ? "ok" : "FAIL"}: ${label}`); + if (!val) { + allPassed = false; + } +} + +let serverBytesWritten = 0; + +function createStreamingServer() { + return createServer((req, res) => { + if (req.url === "/huge") { + res.writeHead(200, { "Content-Type": "application/json" }); + const chunk = Buffer.alloc(65536, 120); + const header = Buffer.from('{"ok":true,"result":{"id":1'); + res.write(header); + serverBytesWritten += header.length; + let sent = header.length; + const writeNext = () => { + if (sent >= STREAM_SIZE) { + const tail = Buffer.from("}}"); + res.write(tail); + serverBytesWritten += tail.length; + res.end(); + return; + } + const ok = res.write(chunk); + serverBytesWritten += chunk.length; + sent += chunk.length; + if (ok) { + setImmediate(writeNext); + } else { + res.once("drain", writeNext); + } + }; + writeNext(); + return; + } + if (req.url?.includes("/getChat")) { + const chatId = new URL(req.url, "http://127.0.0.1").searchParams.get("chat_id"); + if (chatId === "OVERSIZED") { + res.writeHead(200, { "Content-Type": "application/json" }); + const chunk = Buffer.alloc(65536, 120); + const header = Buffer.from('{"ok":true,"result":{"id":1'); + res.write(header); + serverBytesWritten += header.length; + let sent = header.length; + const writeNext = () => { + if (sent >= STREAM_SIZE) { + const tail = Buffer.from("}}"); + res.write(tail); + serverBytesWritten += tail.length; + res.end(); + return; + } + const ok = res.write(chunk); + serverBytesWritten += chunk.length; + sent += chunk.length; + if (ok) { + setImmediate(writeNext); + } else { + res.once("drain", writeNext); + } + }; + writeNext(); + return; + } + const body = JSON.stringify({ ok: true, result: { id: 123456789 } }); + res.writeHead(200, { + "Content-Type": "application/json", + "Content-Length": String(Buffer.byteLength(body)), + }); + res.end(body); + serverBytesWritten += Buffer.byteLength(body); + return; + } + res.writeHead(404); + res.end(); + }); +} + +async function withServer(fn) { + serverBytesWritten = 0; + const server = createStreamingServer(); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const { port } = server.address(); + try { + await fn(port); + } finally { + await new Promise((resolveDone) => { + server.close(resolveDone); + }); + } +} + +console.log(`\n[proof] fetchTelegramChatId (getChat) production path`); +console.log(` cap=${CAP} bytes (4 MiB), would-stream≈${STREAM_SIZE} bytes (24 MiB)\n`); + +await withServer(async (port) => { + serverBytesWritten = 0; + const chatId = await fetchTelegramChatId({ + token: "proof-token", + chatId: "OVERSIZED", + apiRoot: `http://127.0.0.1:${port}`, + }); + await new Promise((r) => { + setTimeout(r, 50); + }); + check( + "oversized getChat fails closed with null (production fetchTelegramChatId)", + chatId === null, + ); + check( + `server wrote ${serverBytesWritten} bytes, stopped before full 24 MiB stream`, + serverBytesWritten < STREAM_SIZE && serverBytesWritten > CAP, + ); +}); + +await withServer(async (port) => { + serverBytesWritten = 0; + const res = await fetch(`http://127.0.0.1:${port}/huge`); + await res.json().catch(() => undefined); + await new Promise((r) => { + setTimeout(r, 50); + }); + check( + `negative control: unbounded .json() wrote ${serverBytesWritten} bytes (>> ${CAP})`, + serverBytesWritten > CAP, + ); +}); + +await withServer(async (port) => { + serverBytesWritten = 0; + const chatId = await fetchTelegramChatId({ + token: "proof-token", + chatId: "-100123", + apiRoot: `http://127.0.0.1:${port}`, + }); + check(`small getChat parsed through production path (id=${chatId})`, chatId === "123456789"); +}); + +console.log(allPassed ? "\nALL PROOF ASSERTIONS PASSED" : "\nSOME ASSERTIONS FAILED"); +process.exit(allPassed ? 0 : 1);