diff --git a/.env.example b/.env.example index 67097b7104..f259d4d854 100644 --- a/.env.example +++ b/.env.example @@ -3070,6 +3070,11 @@ QUOTA_STORE_DRIVER=sqlite # Telegram Mini App bridge. The update endpoint remains disabled while the bot # token is unset. Used by: src/lib/telegram/* and src/app/api/telegram/update/route.ts. # TELEGRAM_BOT_TOKEN= +# Shared secret registered with setWebhook and echoed back by Telegram as the +# X-Telegram-Bot-Api-Secret-Token header. REQUIRED for the webhook path: without +# it the webhook is rejected with 503, because an unauthenticated update lets any +# caller mint API keys and spend upstream quota. The Mini App path does not use it. +# TELEGRAM_WEBHOOK_SECRET= # TELEGRAM_DEFAULT_MODEL=auto/chat # TELEGRAM_BOT_API_BASE=https://api.telegram.org # TELEGRAM_WEBHOOK_TIMEOUT_MS=60000 diff --git a/changelog.d/fixes/13095-acp-buffer-cap.md b/changelog.d/fixes/13095-acp-buffer-cap.md new file mode 100644 index 0000000000..61473dca6d --- /dev/null +++ b/changelog.d/fixes/13095-acp-buffer-cap.md @@ -0,0 +1 @@ +- **fix(acp):** bound the ACP session output buffers — `stdoutBuffer` and `stderrBuffer` now cap at 1 MiB keeping the most recent output behind a visible `[...output truncated...]` marker, and `stderrBuffer` is reset per prompt instead of accumulating for the lifetime of the session. diff --git a/changelog.d/fixes/13095-acp-sendprompt-listener-leak.md b/changelog.d/fixes/13095-acp-sendprompt-listener-leak.md new file mode 100644 index 0000000000..56473cc8c2 --- /dev/null +++ b/changelog.d/fixes/13095-acp-sendprompt-listener-leak.md @@ -0,0 +1 @@ +- **fix(acp):** release the `stdout`/`exit` listeners and the idle timer that a `sendPrompt` timeout used to leave attached to the `acpManager` singleton, and drop sessions that exited on their own from the session map instead of keeping them forever. diff --git a/changelog.d/fixes/13103-badge-sse-aborted-signal.md b/changelog.d/fixes/13103-badge-sse-aborted-signal.md new file mode 100644 index 0000000000..000e3ddb45 --- /dev/null +++ b/changelog.d/fixes/13103-badge-sse-aborted-signal.md @@ -0,0 +1 @@ +- **fix(gamification):** close the badge notification SSE stream when the request signal is already aborted before the stream starts — a client that disconnects while the route is still awaiting auth used to leave both the 2s unlock poll and the 15s heartbeat running for the lifetime of the process. diff --git a/changelog.d/fixes/13108-nodesqlite-process-listener-leak.md b/changelog.d/fixes/13108-nodesqlite-process-listener-leak.md new file mode 100644 index 0000000000..93875c856f --- /dev/null +++ b/changelog.d/fixes/13108-nodesqlite-process-listener-leak.md @@ -0,0 +1 @@ +- **fix(db):** release the `beforeExit`/`SIGINT`/`SIGTERM` handlers when a `node:sqlite` adapter closes, so a closed adapter and its database handle are no longer pinned to `process` for the lifetime of the run — the same treatment #7494 gave the sql.js adapter. diff --git a/changelog.d/fixes/13113-logstream-timer-leak.md b/changelog.d/fixes/13113-logstream-timer-leak.md new file mode 100644 index 0000000000..6bd6ced117 --- /dev/null +++ b/changelog.d/fixes/13113-logstream-timer-leak.md @@ -0,0 +1 @@ +- **fix(cli-helper):** clear the `createLogStream` timeout on the abort path — `stop()` aborts the in-flight fetch and returned through the `signal.aborted` branch, which skipped `clearTimeout` and left an armed timer per stopped stream. The stream reader is now also cancelled when the read loop exits early. diff --git a/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md b/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md new file mode 100644 index 0000000000..4c5249f4fd --- /dev/null +++ b/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md @@ -0,0 +1 @@ +- **fix(traffic-inspector):** the WebSocket route no longer leaks a traffic-buffer subscriber and a 30s ping timer when the client socket is already closed at handler time — listeners are attached before any resource is acquired, a destroyed socket bails out early, and the ping interval stops on a dead socket where `write()` never throws ([#13155](https://github.com/diegosouzapw/OmniRoute/pull/13155)) diff --git a/changelog.d/fixes/13165-telegram-keycache-unbounded.md b/changelog.d/fixes/13165-telegram-keycache-unbounded.md new file mode 100644 index 0000000000..2672fb93be --- /dev/null +++ b/changelog.d/fixes/13165-telegram-keycache-unbounded.md @@ -0,0 +1 @@ +- **fix(telegram):** bound the per-user API key cache in the Telegram chat proxy so a burst of distinct chat ids can no longer grow the process heap without limit ([#13165](https://github.com/diegosouzapw/OmniRoute/issues/13165)) diff --git a/changelog.d/fixes/13169-jsonbody-sniff-reader-leak.md b/changelog.d/fixes/13169-jsonbody-sniff-reader-leak.md new file mode 100644 index 0000000000..b0ad5629c3 --- /dev/null +++ b/changelog.d/fixes/13169-jsonbody-sniff-reader-leak.md @@ -0,0 +1 @@ +- **fix(stream):** cancel the upstream response body when the JSON-to-SSE sniff unwinds on a body timeout, so a stalled upstream no longer pins the connection ([#13169](https://github.com/diegosouzapw/OmniRoute/issues/13169)) diff --git a/changelog.d/fixes/13172-telegram-webhook-secret.md b/changelog.d/fixes/13172-telegram-webhook-secret.md new file mode 100644 index 0000000000..49f14b55fb --- /dev/null +++ b/changelog.d/fixes/13172-telegram-webhook-secret.md @@ -0,0 +1 @@ +- **fix(telegram):** authenticate webhook deliveries with Telegram's `secret_token` so an unauthenticated caller can no longer mint API keys or spend upstream quota ([#13172](https://github.com/diegosouzapw/OmniRoute/issues/13172)) diff --git a/changelog.d/fixes/compression-pool-idle-terminate.md b/changelog.d/fixes/compression-pool-idle-terminate.md new file mode 100644 index 0000000000..d2a280f1ba --- /dev/null +++ b/changelog.d/fixes/compression-pool-idle-terminate.md @@ -0,0 +1 @@ +- fix(compression): terminate idle worker threads on eviction so long-running instances stop leaking OS threads and MessagePorts diff --git a/changelog.d/fixes/llmlingua-worker-spawn.md b/changelog.d/fixes/llmlingua-worker-spawn.md new file mode 100644 index 0000000000..865f19a32e --- /dev/null +++ b/changelog.d/fixes/llmlingua-worker-spawn.md @@ -0,0 +1 @@ +- fix(compression): spawn the LLMLingua worker with a file URL object so compression actually runs instead of silently failing open on Node diff --git a/changelog.d/fixes/local-test-concurrency.md b/changelog.d/fixes/local-test-concurrency.md new file mode 100644 index 0000000000..70a46e3984 --- /dev/null +++ b/changelog.d/fixes/local-test-concurrency.md @@ -0,0 +1 @@ +- **fix(test):** run the local `test` and `test:unit` scripts at concurrency 4 so a full-suite run no longer exhausts the machine's commit charge and kills unrelated processes diff --git a/changelog.d/fixes/plugin-sigkill-listener-leak.md b/changelog.d/fixes/plugin-sigkill-listener-leak.md new file mode 100644 index 0000000000..a4baa47066 --- /dev/null +++ b/changelog.d/fixes/plugin-sigkill-listener-leak.md @@ -0,0 +1 @@ +- fix(plugins): stop leaking an exit listener per plugin hook timeout, which triggered MaxListenersExceededWarning on plugins that ignore SIGTERM diff --git a/changelog.d/fixes/webdav-test-windows-path.md b/changelog.d/fixes/webdav-test-windows-path.md new file mode 100644 index 0000000000..2ba7400611 --- /dev/null +++ b/changelog.d/fixes/webdav-test-windows-path.md @@ -0,0 +1 @@ +- **fix(test):** resolve the WebDAV handler path with `fileURLToPath` so the suite's 37 WebDAV tests run on Windows instead of failing with a doubled `C:\C:\` drive prefix diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index f357f73584..13b70dd4e3 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -1623,6 +1623,7 @@ These settings were introduced after the previous environment-contract snapshot. | `ADOBE_FIREFLY_CHROME_HEADLESS` | `0` | `open-sse/services/adobeFireflyBrowserLogin.ts` | Debug-only true-headless mode; Adobe colligo normally rejects the resulting risk session. | | `CHROME_PATH` | auto-detect | `open-sse/executors/cloudflare-playground.ts`, `open-sse/executors/chatgpt-web-codex.ts` | Optional absolute Chrome executable used by the browser-driven executors when platform auto-detection is insufficient. | | `TELEGRAM_BOT_TOKEN` | _(unset)_ | `src/lib/telegram/config.ts` | BotFather token that enables the inbound webhook and signs Mini App `initData`. | +| `TELEGRAM_WEBHOOK_SECRET` | _(unset)_ | `src/lib/telegram/config.ts` | Shared secret registered via `setWebhook` and verified against the `X-Telegram-Bot-Api-Secret-Token` header on every webhook delivery. Required for the webhook path; unset means webhook deliveries are refused with 503. | | `TELEGRAM_DEFAULT_MODEL` | `auto/chat` | `src/lib/telegram/chatProxy.ts` | Model used for Telegram chat replies. | | `TELEGRAM_BOT_API_BASE` | `https://api.telegram.org` | `src/lib/telegram/config.ts` | Bot API base URL override for proxies or self-hosted Bot API servers. | | `TELEGRAM_WEBHOOK_TIMEOUT_MS` | `60000` | `src/lib/telegram/config.ts` | Timeout in milliseconds for outbound Bot API calls. | diff --git a/open-sse/handlers/chatCore/jsonBodyToSse.ts b/open-sse/handlers/chatCore/jsonBodyToSse.ts index 66c9688807..97bf840ee9 100644 --- a/open-sse/handlers/chatCore/jsonBodyToSse.ts +++ b/open-sse/handlers/chatCore/jsonBodyToSse.ts @@ -88,33 +88,46 @@ async function sniffJsonBodyForSse( let sniffed = ""; let sniffedBytes = 0; const maxSniffBytes = 4096; - while (sniffedBytes < maxSniffBytes) { - const chunk = await deps.withBodyTimeout>(reader.read()); - if (chunk.done || !chunk.value) break; - bufferedChunks.push(chunk.value); - sniffedBytes += chunk.value.byteLength; - sniffed += decoder.decode(chunk.value, { stream: true }); + // The two success paths below hand this still-open reader to + // prependBufferedChunks(), so the reader must NOT be cancelled on the happy + // path. Any other unwind (notably a withBodyTimeout rejection on a stalled + // upstream) would otherwise abandon the body with no cancellation, pinning + // the connection for the lifetime of the socket. + let handedOff = false; + try { + while (sniffedBytes < maxSniffBytes) { + const chunk = await deps.withBodyTimeout>(reader.read()); + if (chunk.done || !chunk.value) break; + bufferedChunks.push(chunk.value); + sniffedBytes += chunk.value.byteLength; + sniffed += decoder.decode(chunk.value, { stream: true }); - if (classifyBodyPrefix(sniffed) === "sse") { - const rebuiltHeaders = new Headers(providerResponse.headers); - rebuiltHeaders.delete("content-length"); - rebuiltHeaders.set("content-type", "text/event-stream"); - ctx.log?.debug?.( - "STREAM", - `Upstream returned SSE bytes with application/json content-type — preserving streaming body (${ctx.provider}/${ctx.model})` - ); - return { - sseResponse: new Response(prependBufferedChunks(bufferedChunks, reader), { - status: providerResponse.status, - statusText: providerResponse.statusText, - headers: rebuiltHeaders, - }), - jsonBody: new Response(null), - }; + if (classifyBodyPrefix(sniffed) === "sse") { + const rebuiltHeaders = new Headers(providerResponse.headers); + rebuiltHeaders.delete("content-length"); + rebuiltHeaders.set("content-type", "text/event-stream"); + ctx.log?.debug?.( + "STREAM", + `Upstream returned SSE bytes with application/json content-type — preserving streaming body (${ctx.provider}/${ctx.model})` + ); + handedOff = true; + return { + sseResponse: new Response(prependBufferedChunks(bufferedChunks, reader), { + status: providerResponse.status, + statusText: providerResponse.statusText, + headers: rebuiltHeaders, + }), + jsonBody: new Response(null), + }; + } } - } - return { jsonBody: new Response(prependBufferedChunks(bufferedChunks, reader)) }; + handedOff = true; + return { jsonBody: new Response(prependBufferedChunks(bufferedChunks, reader)) }; + } finally { + // Cancellation is best-effort: the body may already be errored or closed. + if (!handedOff) void reader.cancel().catch(() => {}); + } } export async function maybeConvertJsonBodyToSse( diff --git a/open-sse/services/compression/compressionWorkerPool.ts b/open-sse/services/compression/compressionWorkerPool.ts index 352aabd39c..7402294c39 100644 --- a/open-sse/services/compression/compressionWorkerPool.ts +++ b/open-sse/services/compression/compressionWorkerPool.ts @@ -131,7 +131,7 @@ export class CompressionWorkerPool { } async close(): Promise { for (const job of this.queue.splice(0)) job.resolve(unchanged(job.originalBody)); - await Promise.all([...this.workers].map((slot) => this.remove(slot, true))); + await Promise.all([...this.workers].map((slot) => this.remove(slot))); } private spawn(): PoolWorker { const slot: PoolWorker = { @@ -185,7 +185,10 @@ export class CompressionWorkerPool { slot.timeout = null; slot.job = null; job.resolve(result); - slot.idle = setTimeout(() => void this.remove(slot, false), this.idleMs); + // Idle eviction MUST terminate. Dropping the slot from the set only releases our + // reference - the thread, its MessagePort and its private heap outlive the pool + // for the whole process lifetime, invisible to process.memoryUsage(). (#12812) + slot.idle = setTimeout(() => void this.remove(slot), this.idleMs); slot.idle.unref(); this.dispatch(); } @@ -193,13 +196,15 @@ export class CompressionWorkerPool { const job = slot.job; if (job) job.resolve(unchanged(job.originalBody)); slot.job = null; - void this.remove(slot, true).finally(() => this.dispatch()); + void this.remove(slot).finally(() => this.dispatch()); } - private async remove(slot: PoolWorker, terminate: boolean): Promise { + /** Drop a slot and release its OS thread. Removal always terminates: a pooled worker + * has no other owner, so skipping terminate() strands the thread permanently. */ + private async remove(slot: PoolWorker): Promise { if (!this.workers.delete(slot)) return; if (slot.timeout) clearTimeout(slot.timeout); if (slot.idle) clearTimeout(slot.idle); - if (terminate) await slot.worker.terminate().catch(() => undefined); + await slot.worker.terminate().catch(() => undefined); } } diff --git a/open-sse/services/compression/engines/llmlingua/worker.ts b/open-sse/services/compression/engines/llmlingua/worker.ts index 00ba3e1255..0994e1a935 100644 --- a/open-sse/services/compression/engines/llmlingua/worker.ts +++ b/open-sse/services/compression/engines/llmlingua/worker.ts @@ -234,7 +234,12 @@ function ensureWorker(): Worker { const { workerFile, execArgv } = resolveWorkerFile(); const absoluteWorkerFile = path.resolve(workerFile); - const w = new Worker(pathToFileURL(absoluteWorkerFile).href, { execArgv }); + // Pass the URL OBJECT, not `.href`. `new Worker()` treats a plain string as a + // filesystem path, so a "file://..." string is looked up literally and throws + // ERR_WORKER_PATH (a string arg must start with ./ or ../). Only a URL instance + // is interpreted as a file: URL. Spawn failures are swallowed by pump()'s catch, + // so getting this wrong silently disables compression instead of erroring. + const w = new Worker(pathToFileURL(absoluteWorkerFile), { execArgv }); w.on("message", (reply: WorkerReply) => { const entry = pending.get(reply.id); diff --git a/package.json b/package.json index 2d5ae328ae..ccebb40b09 100644 --- a/package.json +++ b/package.json @@ -125,8 +125,8 @@ "electron:build:mac": "npm run build && cd electron && npm run build:mac", "electron:build:linux": "npm run build && cd electron && npm run build:linux", "electron:smoke:packaged": "node scripts/dev/smoke-electron-packaged.mjs", - "test": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-concurrency=20 tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-concurrency=20 \"tests/unit/dashboard/**/*.test.ts\"", - "test:unit": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=20 tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=20 \"tests/unit/dashboard/**/*.test.ts\" && npm run test:unit:serial", + "test": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 \"tests/unit/dashboard/**/*.test.ts\"", + "test:unit": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 \"tests/unit/dashboard/**/*.test.ts\" && npm run test:unit:serial", "test:unit:ci": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 \"tests/unit/dashboard/**/*.test.ts\" && npm run test:unit:serial", "test:unit:ci:shard": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 --test-shard=$TEST_SHARD tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=4 --test-shard=$TEST_SHARD \"tests/unit/dashboard/**/*.test.ts\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-concurrency=1 --test-shard=$TEST_SHARD \"tests/unit/serial/**/*.test.ts\"", "test:unit:fast": "cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-isolation=none tests/unit/*.test.ts \"tests/unit/{api,auth,authz,build,cli,cli-helper,combo,compression,correctness,cors,db,db-adapters,docs,gamification,guardrails,lib,mcp,memory,runtime,security,services,settings,shared,translator,ui,usage}/**/*.test.ts\" \"tests/unit/**/*.test.mjs\" && cross-env DISABLE_SQLITE_AUTO_BACKUP=true node --max-old-space-size=8192 --import tsx --import ./open-sse/utils/setupPolyfill.ts --import ./tests/_setup/isolateDataDir.ts --test --test-force-exit --test-isolation=none \"tests/unit/dashboard/**/*.test.ts\" && npm run test:unit:serial", diff --git a/src/app/api/telegram/update/route.ts b/src/app/api/telegram/update/route.ts index 1fd3582da9..c2194a5895 100644 --- a/src/app/api/telegram/update/route.ts +++ b/src/app/api/telegram/update/route.ts @@ -13,12 +13,18 @@ * 3. Handles /start (returns the Mini App deep link) and everything else * as a chat prompt proxied through the OmniRoute pipeline. */ +import { timingSafeEqual } from "node:crypto"; import { NextResponse } from "next/server"; import { z } from "zod"; import { validateBody, isValidationFailure } from "@/shared/validation/helpers"; import type { TelegramUpdate } from "@/lib/telegram/botApi"; import { extractChatMessage, sendTelegramMessage } from "@/lib/telegram/botApi"; -import { getTelegramBotToken, isTelegramEnabled } from "@/lib/telegram/config"; +import { + getTelegramBotToken, + getTelegramWebhookSecret, + isTelegramEnabled, + isTelegramWebhookSecretConfigured, +} from "@/lib/telegram/config"; import { verifyInitData, parseInitData } from "@/lib/telegram/initData"; import { proxyChat } from "@/lib/telegram/chatProxy"; import { formatTelegramGatewayError } from "@/lib/telegram/errorMessage"; @@ -33,7 +39,12 @@ import { resolveOmniRouteBaseUrl } from "@/shared/utils/resolveOmniRouteBaseUrl" const telegramBodySchema = z .object({ initData: z.string().optional(), - message: z.string().optional(), + // `message` is a STRING on the Mini App path ({ initData, message }) and an + // OBJECT on the webhook path (a Telegram update). Constraining it to a + // string rejected every real webhook delivery with 400 before any auth or + // routing ran, so accept either shape here and let each branch validate the + // shape it actually needs. + message: z.union([z.string(), z.record(z.string(), z.unknown())]).optional(), update_id: z.number().optional(), // allow unknown update fields }) @@ -103,6 +114,21 @@ export async function POST(request: Request) { } // ── Bot webhook path: TelegramUpdate ───────────────────────────────────── + // Unlike the Mini App branch above (which verifies the initData HMAC), a + // webhook body carries no proof of origin: `chat.id` is attacker-chosen and + // reaches proxyChat(), which mints a real API key and spends upstream quota. + // Telegram's `secret_token` echo is the only authentication available here. + if (!isTelegramWebhookSecretConfigured()) { + return NextResponse.json( + { ok: false, error: "Telegram webhook secret not configured" }, + { status: 503 } + ); + } + const presentedSecret = request.headers.get("x-telegram-bot-api-secret-token") || ""; + if (!webhookSecretMatches(presentedSecret, getTelegramWebhookSecret())) { + return NextResponse.json({ ok: false, error: "Unauthorized" }, { status: 401 }); + } + const update = body as unknown as TelegramUpdate; const chat = extractChatMessage(update); if (!chat) { @@ -117,6 +143,22 @@ export async function POST(request: Request) { return NextResponse.json({ ok: true }); } +/** + * Constant-time comparison of the presented webhook secret against the + * configured one. A plain `===` short-circuits on the first differing byte and + * leaks the shared-prefix length through response timing; `timingSafeEqual` + * does not. It requires equal-length buffers, so a length mismatch is rejected + * up front (the length itself is not secret). + * + * Exported as a test seam only — not part of the route contract. + */ +export function webhookSecretMatches(presented: string, expected: string): boolean { + const a = Buffer.from(presented); + const b = Buffer.from(expected); + if (a.length !== b.length) return false; + return timingSafeEqual(a, b); +} + async function handleAndReply(chatId: number, text: string, messageId?: number): Promise { try { const trimmed = text.trim(); diff --git a/src/app/api/tools/traffic-inspector/ws/route.ts b/src/app/api/tools/traffic-inspector/ws/route.ts index b32a5d7cc5..a9545e221f 100644 --- a/src/app/api/tools/traffic-inspector/ws/route.ts +++ b/src/app/api/tools/traffic-inspector/ws/route.ts @@ -96,6 +96,14 @@ export async function GET(request: Request): Promise { } const acceptHeader = acceptKey(clientKey); + + // The client can vanish during the upgrade round trip. `close` has then + // ALREADY fired, so the listeners below would never run and every resource + // acquired past this point would be held with no path to release it. + if (socket.destroyed) { + return new Response(null, { status: 101 }); + } + socket.write( [ "HTTP/1.1 101 Switching Protocols", @@ -106,21 +114,17 @@ export async function GET(request: Request): Promise { ].join("\r\n") ); - const unsubscribe = globalTrafficBuffer.subscribe((ev) => { - sendText(socket, ev); - }); - - const pingTimer = setInterval(() => { - try { - socket.write(encodeWsFrame(0x09)); // ping - } catch { - cleanup(); - } - }, PING_INTERVAL_MS); + let unsubscribe: (() => void) | null = null; + let pingTimer: ReturnType | null = null; + let cleanedUp = false; function cleanup(): void { - clearInterval(pingTimer); - unsubscribe(); + if (cleanedUp) return; + cleanedUp = true; + if (pingTimer) clearInterval(pingTimer); + pingTimer = null; + unsubscribe?.(); + unsubscribe = null; try { socket.destroy(); } catch { @@ -128,14 +132,43 @@ export async function GET(request: Request): Promise { } } - socket.once("close", cleanup); - socket.once("error", cleanup); - - // Never resolve — the socket is the response channel. - await new Promise((resolve) => { + // Attached BEFORE any resource is acquired, so there is no window in which a + // subscriber or timer exists without a live path to cleanup(). + const settled = new Promise((resolve) => { socket.once("close", resolve); socket.once("error", resolve); }); + socket.once("close", cleanup); + socket.once("error", cleanup); + + // Re-check: `close` may have fired while we were writing the handshake, in + // which case the listeners above already ran and cleanup() is a no-op we + // still must not skip. + if (socket.destroyed) { + cleanup(); + return new Response(null, { status: 101 }); + } + + unsubscribe = globalTrafficBuffer.subscribe((ev) => { + sendText(socket, ev); + }); + + pingTimer = setInterval(() => { + // `socket.write()` does NOT throw synchronously on a destroyed socket, so + // the destroyed check — not the catch — is what stops a dead interval. + if (socket.destroyed) { + cleanup(); + return; + } + try { + socket.write(encodeWsFrame(0x09)); // ping + } catch { + cleanup(); + } + }, PING_INTERVAL_MS); + + // Never resolve — the socket is the response channel. + await settled; cleanup(); return new Response(null, { status: 101 }); diff --git a/src/lib/acp/manager.ts b/src/lib/acp/manager.ts index 85bc05e720..099239e9b1 100644 --- a/src/lib/acp/manager.ts +++ b/src/lib/acp/manager.ts @@ -30,6 +30,34 @@ export interface AcpSession { createdAt: Date; } +/** + * Upper bound for each per-session output buffer. + * + * Both buffers grow on every chunk a CLI agent writes and are only reset when + * the next prompt starts, so a chatty or looping agent can grow them without + * limit while the session stays alive. 1 MiB is far above a realistic agent + * response while keeping a stuck session's footprint bounded. + */ +const MAX_BUFFER_CHARS = 1_048_576; + +const TRUNCATION_NOTICE = "\n[...output truncated...]\n"; + +/** + * Append to a buffer, keeping the most recent output when the cap is exceeded. + * + * The tail is what callers care about: `sendPrompt` resolves with the stdout + * collected since the prompt was written, and stderr is read for diagnostics + * after a failure. Dropping from the front keeps both useful. + */ +function appendCapped(buffer: string, chunk: string): string { + const combined = buffer + chunk; + if (combined.length <= MAX_BUFFER_CHARS) return combined; + + const keep = MAX_BUFFER_CHARS - TRUNCATION_NOTICE.length; + if (keep <= 0) return combined.slice(-MAX_BUFFER_CHARS); + return TRUNCATION_NOTICE + combined.slice(-keep); +} + /** * ACP Session Manager * @@ -79,17 +107,21 @@ export class AcpManager extends EventEmitter { }; child.stdout?.on("data", (chunk: Buffer) => { - session.stdoutBuffer += chunk.toString(); + session.stdoutBuffer = appendCapped(session.stdoutBuffer, chunk.toString()); this.emit("stdout", { sessionId, data: chunk.toString() }); }); child.stderr?.on("data", (chunk: Buffer) => { - session.stderrBuffer += chunk.toString(); + session.stderrBuffer = appendCapped(session.stderrBuffer, chunk.toString()); this.emit("stderr", { sessionId, data: chunk.toString() }); }); child.on("exit", (code, signal) => { session.alive = false; + // Only kill() used to remove entries, so any agent that exited on its own + // stayed in the map forever. getActiveSessions() filters on `alive`, which + // hid the growth from callers. + this.sessions.delete(sessionId); this.emit("exit", { sessionId, code, signal }); }); @@ -121,39 +153,46 @@ export class AcpManager extends EventEmitter { const session = this.sessions.get(sessionId); if (!session?.alive) throw new Error(`Session ${sessionId} is not alive`); - // Clear buffer before sending + // Clear buffers before sending. stderr is reset too: it was previously only + // ever appended to, so diagnostics for one prompt carried stale output from + // every earlier prompt in the session. session.stdoutBuffer = ""; + session.stderrBuffer = ""; // Send prompt this.sendInput(sessionId, prompt + "\n"); // Wait for response (collect until process goes idle or timeout) return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - reject(new Error(`ACP timeout after ${timeoutMs}ms`)); - }, timeoutMs); + let idleTimer: ReturnType | undefined; - let idleTimer: ReturnType; + // Every outcome -- idle, exit, or timeout -- has to release the same + // resources. `acpManager` is a module-level singleton, so a branch that + // skips this leaks a listener per call for the lifetime of the process. + const settle = (finish: () => void) => { + clearTimeout(timer); + clearTimeout(idleTimer); + this.removeListener("stdout", onData); + this.removeListener("exit", onExit); + finish(); + }; + + const timer = setTimeout(() => { + settle(() => reject(new Error(`ACP timeout after ${timeoutMs}ms`))); + }, timeoutMs); const onData = ({ sessionId: sid }: { sessionId: string }) => { if (sid !== sessionId) return; // Reset idle timer on new data clearTimeout(idleTimer); idleTimer = setTimeout(() => { - clearTimeout(timer); - this.removeListener("stdout", onData); - this.removeListener("exit", onExit); - resolve(session.stdoutBuffer); + settle(() => resolve(session.stdoutBuffer)); }, 2000); // 2s idle = response complete }; const onExit = ({ sessionId: sid }: { sessionId: string }) => { if (sid !== sessionId) return; - clearTimeout(timer); - clearTimeout(idleTimer); - this.removeListener("stdout", onData); - this.removeListener("exit", onExit); - resolve(session.stdoutBuffer); + settle(() => resolve(session.stdoutBuffer)); }; this.on("stdout", onData); diff --git a/src/lib/cli-helper/log-streamer.ts b/src/lib/cli-helper/log-streamer.ts index 1fdc151848..06dbffbb2d 100644 --- a/src/lib/cli-helper/log-streamer.ts +++ b/src/lib/cli-helper/log-streamer.ts @@ -38,30 +38,37 @@ export function createLogStream(options: LogStreamOptions = {}): LogStream { if (!response.ok) { controller.error(new Error(`HTTP ${response.status}: ${response.statusText}`)); - clearTimeout(timeoutId); return; } if (!response.body) { controller.error(new Error("Response body is null")); - clearTimeout(timeoutId); return; } const reader = response.body.getReader(); - while (true) { - const { done, value } = await reader.read(); - if (done) break; - if (signal.aborted) break; - controller.enqueue(value); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (signal.aborted) break; + controller.enqueue(value); + } + } finally { + // Leaving the loop early (abort/throw) otherwise keeps the body locked + // and its socket held until GC. + await reader.cancel().catch(() => {}); } controller.close(); - clearTimeout(timeoutId); } catch (err) { if (signal.aborted) return; // Expected stop controller.error(err instanceof Error ? err : new Error(String(err))); + } finally { + // `stop()` aborts mid-fetch and returns through the `signal.aborted` + // branch above, so clearing the timer on the individual exit paths + // misses the one path stop() is built to take. clearTimeout(timeoutId); } }, diff --git a/src/lib/db/adapters/nodeSqliteAdapter.ts b/src/lib/db/adapters/nodeSqliteAdapter.ts index 73c3aeee60..a422d1c8c5 100644 --- a/src/lib/db/adapters/nodeSqliteAdapter.ts +++ b/src/lib/db/adapters/nodeSqliteAdapter.ts @@ -35,26 +35,34 @@ export async function createNodeSqliteAdapter(filePath: string): Promise { + adapter.close(); + }; + const onSignal = () => { + adapter.close(); + process.exit(0); + }; + function gracefulClose() { clearInterval(checkpointTimer as unknown as NodeJS.Timeout); try { db.exec("PRAGMA wal_checkpoint(TRUNCATE)"); } catch {} + process.removeListener("beforeExit", onBeforeExit); + process.removeListener("SIGINT", onSignal); + process.removeListener("SIGTERM", onSignal); } const adapter = createNodeSqliteAdapterFromDatabase(db, filePath, gracefulClose); - process.once("beforeExit", () => { - adapter.close(); - }); - process.once("SIGINT", () => { - adapter.close(); - process.exit(0); - }); - process.once("SIGTERM", () => { - adapter.close(); - process.exit(0); - }); + process.once("beforeExit", onBeforeExit); + process.once("SIGINT", onSignal); + process.once("SIGTERM", onSignal); return adapter; } diff --git a/src/lib/gamification/notifications.ts b/src/lib/gamification/notifications.ts index 4226cb9977..ef3daf1dab 100644 --- a/src/lib/gamification/notifications.ts +++ b/src/lib/gamification/notifications.ts @@ -110,6 +110,13 @@ export function createBadgeNotificationStream( } }; + // A client that disconnects while the route is still awaiting auth + // arrives here already aborted, and "abort" will never fire again -- + // the timers above would then run for the lifetime of the process. + if (signal?.aborted) { + cleanup(); + return; + } if (signal) { signal.addEventListener("abort", cleanup); } diff --git a/src/lib/plugins/loader.ts b/src/lib/plugins/loader.ts index d4bcd2739d..71e9223b00 100644 --- a/src/lib/plugins/loader.ts +++ b/src/lib/plugins/loader.ts @@ -9,6 +9,7 @@ */ import { spawn } from "child_process"; +import type { ChildProcess } from "child_process"; import { writeFile, readFile } from "fs/promises"; import { rmSync } from "fs"; import { join } from "path"; @@ -105,6 +106,37 @@ function forwardChildOutput( * against process exit — under `node --test --test-force-exit` the runner exits * before the promise settles, leaking one temp .mjs per plugin load. */ +/** Children already escalating to SIGKILL. Prevents re-arming a second timer + listener + * for a child that is already being killed. */ +const escalating = new WeakSet(); + +/** + * SIGTERM has already been sent; escalate to SIGKILL if the child ignores it. + * + * Must be idempotent per child. Every hook timeout hits this path, and a plugin that + * traps SIGTERM keeps taking calls, so re-arming would add one exit listener plus one + * killTimer closure per timeout — Node starts printing MaxListenersExceededWarning at 11. + * One pending kill per child is also all that is useful: SIGKILL cannot be ignored, so a + * second timer would only re-signal a corpse. (#12819) + */ +function escalateToSigkill(child: ChildProcess): void { + if (escalating.has(child)) return; + escalating.add(child); + + const onExit = () => { + clearTimeout(killTimer); + escalating.delete(child); + }; + const killTimer = setTimeout(() => { + child.removeListener("exit", onExit); + escalating.delete(child); + try { + child.kill("SIGKILL"); + } catch {} + }, SIGKILL_GRACE_MS); + child.once("exit", onExit); +} + function removeHostScript(path: string): void { try { rmSync(path, { force: true }); @@ -293,12 +325,7 @@ export async function loadPlugin( } child.kill("SIGTERM"); // Escalate to SIGKILL if plugin ignores SIGTERM - const killTimer = setTimeout(() => { - try { - child.kill("SIGKILL"); - } catch {} - }, SIGKILL_GRACE_MS); - child.once("exit", () => clearTimeout(killTimer)); + escalateToSigkill(child); reject(new Error(`Plugin hook '${hook}' timed out after ${timeout}ms`)); }, timeout); @@ -399,12 +426,7 @@ export async function loadPlugin( const cleanup = () => { child.kill("SIGTERM"); // Escalate to SIGKILL after grace period - const killTimer = setTimeout(() => { - try { - child.kill("SIGKILL"); - } catch {} - }, SIGKILL_GRACE_MS); - child.once("exit", () => clearTimeout(killTimer)); + escalateToSigkill(child); removeHostScript(hostScriptPath); log.info("loader.cleanup", { name: manifest.name }); }; diff --git a/src/lib/telegram/botApi.ts b/src/lib/telegram/botApi.ts index 4bdc50071a..c1a0e5e048 100644 --- a/src/lib/telegram/botApi.ts +++ b/src/lib/telegram/botApi.ts @@ -5,7 +5,12 @@ * replies and setWebhook for webhook registration. Streaming is emulated * by the caller via progressive edits (sendMessage / editMessageText). */ -import { getTelegramBotApiBase, getTelegramBotToken, getTelegramWebhookTimeoutMs } from "./config"; +import { + getTelegramBotApiBase, + getTelegramBotToken, + getTelegramWebhookTimeoutMs, + getTelegramWebhookSecret, +} from "./config"; export interface TelegramSendMessageParams { chat_id: number | string; @@ -92,7 +97,15 @@ export async function setTelegramWebhook( opts: { dropPending?: boolean } = {} ): Promise<{ url: string; pending_update_count?: number }> { if (url) { - return botFetch("setWebhook", { url, drop_pending_updates: opts.dropPending ?? true }); + // Register the shared secret so Telegram echoes it back as + // X-Telegram-Bot-Api-Secret-Token on every delivery; the webhook route + // rejects deliveries that do not carry it (#13172). + const secret = getTelegramWebhookSecret(); + return botFetch("setWebhook", { + url, + drop_pending_updates: opts.dropPending ?? true, + ...(secret ? { secret_token: secret } : {}), + }); } return botFetch("deleteWebhook", { drop_pending_updates: opts.dropPending ?? true }); } diff --git a/src/lib/telegram/chatProxy.ts b/src/lib/telegram/chatProxy.ts index d2b136954e..724ccd2b00 100644 --- a/src/lib/telegram/chatProxy.ts +++ b/src/lib/telegram/chatProxy.ts @@ -21,11 +21,31 @@ const DEFAULT_MODEL = process.env.TELEGRAM_DEFAULT_MODEL || "auto/chat"; * Resolve (and lazily mint) an OmniRoute API key for a Telegram user. * Returns the plaintext key value, cached per user id. */ +// Bounded LRU. The webhook path passes a caller-supplied chat id, so the key +// space is not limited to the real user population and an uncapped Map would +// grow for the lifetime of the process. Insertion order is the recency order: +// a hit re-inserts, and the oldest entry is dropped once the cap is reached. +const KEY_CACHE_MAX_ENTRIES = 1000; const keyCache = new Map(); +function rememberUserApiKey(telegramUserId: number, key: string): void { + // Re-insert so this id becomes the most recently used entry. + keyCache.delete(telegramUserId); + keyCache.set(telegramUserId, key); + while (keyCache.size > KEY_CACHE_MAX_ENTRIES) { + const oldest = keyCache.keys().next(); + if (oldest.done) break; + keyCache.delete(oldest.value); + } +} + export async function resolveUserApiKey(telegramUserId: number): Promise { const cached = keyCache.get(telegramUserId); - if (cached) return cached; + if (cached) { + // Refresh recency so an active user is not evicted by a burst of new ids. + rememberUserApiKey(telegramUserId, cached); + return cached; + } const machineId = (await getConsistentMachineId().catch(() => null)) || "0000000000000000"; @@ -39,12 +59,12 @@ export async function resolveUserApiKey(telegramUserId: number): Promise ); const matchKey = (match as { key?: string } | undefined)?.key; if (typeof matchKey === "string" && matchKey.length > 0) { - keyCache.set(telegramUserId, matchKey); + rememberUserApiKey(telegramUserId, matchKey); return matchKey; } const created = await createApiKey(`telegram:${telegramUserId}`, machineId); - keyCache.set(telegramUserId, created.key); + rememberUserApiKey(telegramUserId, created.key); return created.key; } diff --git a/src/lib/telegram/config.ts b/src/lib/telegram/config.ts index 421739ef5e..817641e9be 100644 --- a/src/lib/telegram/config.ts +++ b/src/lib/telegram/config.ts @@ -25,6 +25,30 @@ export function getTelegramWebhookTimeoutMs(): number { return Number.isInteger(parsed) && parsed > 0 ? parsed : DEFAULT_WEBHOOK_TIMEOUT_MS; } +/** + * Shared secret for authenticating Telegram webhook deliveries. + * + * Telegram echoes the `secret_token` passed to `setWebhook` back on every + * delivery in the `X-Telegram-Bot-Api-Secret-Token` header, which is the only + * way to prove a webhook POST actually came from Telegram. Kept in the + * environment alongside the bot token so it is never stored in the DB. + */ +export function getTelegramWebhookSecret(): string { + return process.env.TELEGRAM_WEBHOOK_SECRET || ""; +} + +/** + * Whether webhook deliveries are authenticated. + * + * When no secret is configured the webhook path is rejected outright rather + * than served unauthenticated: an open path mints API keys and spends upstream + * quota for any caller (see #13172). The Mini App path is unaffected — it + * authenticates with the initData HMAC and does not use this secret. + */ +export function isTelegramWebhookSecretConfigured(): boolean { + return getTelegramWebhookSecret().length > 0; +} + export function getTelegramBotApiBase(): string { return process.env.TELEGRAM_BOT_API_BASE || "https://api.telegram.org"; } diff --git a/tests/unit/acp-manager-buffer-cap-13095.test.ts b/tests/unit/acp-manager-buffer-cap-13095.test.ts new file mode 100644 index 0000000000..31ac053625 --- /dev/null +++ b/tests/unit/acp-manager-buffer-cap-13095.test.ts @@ -0,0 +1,143 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { AcpManager } = await import("../../src/lib/acp/manager.ts"); +const { setCustomAgents } = await import("../../src/lib/acp/registry.ts"); + +const AGENT_ID = "buffer-cap-probe"; +const CAP = 1_048_576; + +/** + * Spawn a node process that writes `bytes` of stdout (or stderr) and stays alive, + * so the buffers can be inspected while the session is still running. + */ +function makeAgent(stream: "stdout" | "stderr", bytes: number) { + setCustomAgents([ + { + id: AGENT_ID, + name: "Buffer cap probe", + binary: process.execPath, + acpSpawnable: true, + }, + ]); + const script = ` + const chunk = "x".repeat(64 * 1024); + let written = 0; + const target = ${bytes}; + while (written < target) { + process.${stream}.write(chunk); + written += chunk.length; + } + setInterval(() => {}, 1000); + `; + return ["-e", script]; +} + +async function waitForOutput(session: { stdoutBuffer: string; stderrBuffer: string }) { + // Give the child time to flush everything it intends to write. + for (let i = 0; i < 60; i++) { + await new Promise((r) => setTimeout(r, 50)); + if (session.stdoutBuffer.length > CAP / 2 || session.stderrBuffer.length > CAP / 2) break; + } + await new Promise((r) => setTimeout(r, 300)); +} + +test("stdout buffer stays bounded when an agent floods it (#13095)", async () => { + const mgr = new AcpManager(); + const session = mgr.spawn(AGENT_ID, process.execPath, makeAgent("stdout", 4 * CAP)); + try { + await waitForOutput(session); + assert.ok( + session.stdoutBuffer.length > 0, + "precondition: the probe agent must have written something" + ); + assert.ok( + session.stdoutBuffer.length <= CAP, + `stdoutBuffer grew to ${session.stdoutBuffer.length} chars, above the ${CAP} cap` + ); + } finally { + mgr.kill(session.id); + } +}); + +test("stderr buffer stays bounded when an agent floods it (#13095)", async () => { + const mgr = new AcpManager(); + const session = mgr.spawn(AGENT_ID, process.execPath, makeAgent("stderr", 4 * CAP)); + try { + await waitForOutput(session); + assert.ok( + session.stderrBuffer.length > 0, + "precondition: the probe agent must have written something" + ); + assert.ok( + session.stderrBuffer.length <= CAP, + `stderrBuffer grew to ${session.stderrBuffer.length} chars, above the ${CAP} cap` + ); + } finally { + mgr.kill(session.id); + } +}); + +test("truncation keeps the most recent output, not the oldest (#13095)", async () => { + setCustomAgents([ + { + id: AGENT_ID, + name: "Buffer cap probe", + binary: process.execPath, + acpSpawnable: true, + }, + ]); + const script = ` + const chunk = "x".repeat(64 * 1024); + let written = 0; + while (written < ${2 * CAP}) { process.stdout.write(chunk); written += chunk.length; } + process.stdout.write("FINAL-MARKER"); + setInterval(() => {}, 1000); + `; + const mgr = new AcpManager(); + const session = mgr.spawn(AGENT_ID, process.execPath, ["-e", script]); + try { + await waitForOutput(session); + // The tail is the part callers use: sendPrompt resolves with stdout, and + // stderr is read for diagnostics after a failure. + assert.ok( + session.stdoutBuffer.endsWith("FINAL-MARKER"), + "the newest output must survive truncation" + ); + assert.ok(session.stdoutBuffer.length <= CAP, "buffer must still respect the cap"); + } finally { + mgr.kill(session.id); + } +}); + +test("stderr is reset between prompts so diagnostics are per-prompt (#13095)", async () => { + setCustomAgents([ + { + id: AGENT_ID, + name: "Buffer cap probe", + binary: process.execPath, + acpSpawnable: true, + }, + ]); + // Echoes stdin back on stdout, and writes a fixed line to stderr per prompt. + const script = ` + process.stdin.on("data", (d) => { + process.stderr.write("warn:" + d.toString().trim() + "\\n"); + process.stdout.write("ok\\n"); + }); + setInterval(() => {}, 1000); + `; + const mgr = new AcpManager(); + const session = mgr.spawn(AGENT_ID, process.execPath, ["-e", script]); + try { + await mgr.sendPrompt(session.id, "first", 6000); + await mgr.sendPrompt(session.id, "second", 6000); + assert.ok( + !session.stderrBuffer.includes("warn:first"), + `stderr from an earlier prompt leaked into the next one: ${JSON.stringify(session.stderrBuffer)}` + ); + assert.ok(session.stderrBuffer.includes("warn:second"), "current prompt's stderr must be kept"); + } finally { + mgr.kill(session.id); + } +}); diff --git a/tests/unit/acp-manager-sendprompt-leak-13095.test.ts b/tests/unit/acp-manager-sendprompt-leak-13095.test.ts new file mode 100644 index 0000000000..d3b5dce291 --- /dev/null +++ b/tests/unit/acp-manager-sendprompt-leak-13095.test.ts @@ -0,0 +1,101 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { AcpManager } = await import("../../src/lib/acp/manager.ts"); +const { setCustomAgents } = await import("../../src/lib/acp/registry.ts"); + +// A registered agent whose binary is just node running a script that stays quiet, +// so sendPrompt() reliably hits its timeout instead of resolving on data/exit. +const AGENT_ID = "acp-leak-probe"; +setCustomAgents([ + { + id: AGENT_ID, + name: "ACP leak probe", + binary: process.execPath, + description: "test-only agent", + }, +]); + +function spawnIdleSession(manager) { + // Keeps stdin open and never writes to stdout: the prompt can only time out. + return manager.spawn(AGENT_ID, process.execPath, [ + "-e", + "process.stdin.resume(); setTimeout(() => {}, 60_000);", + ]); +} + +test("sendPrompt timeout does not leak listeners on the manager (#13095)", async () => { + const manager = new AcpManager(); + const session = spawnIdleSession(manager); + + try { + const before = { + stdout: manager.listenerCount("stdout"), + exit: manager.listenerCount("exit"), + }; + + // Each of these must reject on the timeout path. + for (let i = 0; i < 12; i++) { + await assert.rejects( + () => manager.sendPrompt(session.id, "ping", 15), + /ACP timeout after 15ms/, + `attempt ${i + 1} should time out` + ); + } + + // The timeout branch has to tear down both listeners it registered. Before the + // fix these grew by one per timed-out prompt and were never released, which + // matters because `acpManager` is a module-level singleton. + assert.equal( + manager.listenerCount("stdout"), + before.stdout, + "stdout listeners must return to the pre-prompt count" + ); + assert.equal( + manager.listenerCount("exit"), + before.exit, + "exit listeners must return to the pre-prompt count" + ); + } finally { + manager.killAll(); + } +}); + +test("sendPrompt timeout clears its idle timer so the process can settle (#13095)", async () => { + const manager = new AcpManager(); + const session = spawnIdleSession(manager); + + try { + await assert.rejects( + () => manager.sendPrompt(session.id, "ping", 15), + /ACP timeout after 15ms/ + ); + + // A leaked idle timer keeps a 2s handle (and the captured session) alive after + // the promise already rejected. Nothing should be pending on the manager. + assert.equal(manager.listenerCount("stdout"), 0); + assert.equal(manager.listenerCount("exit"), 0); + } finally { + manager.killAll(); + } +}); + +test("exited sessions are removed from the session map (#13095)", async () => { + const manager = new AcpManager(); + // Exits immediately on its own; nothing calls kill() for it. + const session = manager.spawn(AGENT_ID, process.execPath, ["-e", "process.exit(0)"]); + + await new Promise((resolve) => { + manager.on("exit", ({ sessionId }) => { + if (sessionId === session.id) resolve(); + }); + }); + // Let the exit handler finish its bookkeeping. + await new Promise((resolve) => setTimeout(resolve, 50)); + + assert.equal( + manager.getSession(session.id), + undefined, + "a session that exited on its own must not stay in the map" + ); +}); diff --git a/tests/unit/badge-sse-aborted-signal-13103.test.ts b/tests/unit/badge-sse-aborted-signal-13103.test.ts new file mode 100644 index 0000000000..fbe5a91ed7 --- /dev/null +++ b/tests/unit/badge-sse-aborted-signal-13103.test.ts @@ -0,0 +1,79 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { createBadgeNotificationStream } = + await import("../../src/lib/gamification/notifications.ts"); + +/** + * Count timers created while `fn` runs and are still armed afterwards. + * The stream owns its handles privately, so this is the only way to observe them. + */ +async function withTimerAccounting( + fn: () => Promise | T +): Promise<{ result: T; live: number }> { + const live = new Set(); + const realSet = globalThis.setInterval; + const realClear = globalThis.clearInterval; + + globalThis.setInterval = ((...args: Parameters) => { + const handle = realSet(...args); + live.add(handle); + return handle; + }) as typeof realSet; + + globalThis.clearInterval = ((handle: Parameters[0]) => { + if (handle !== undefined) live.delete(handle); + return realClear(handle); + }) as typeof realClear; + + try { + const result = await fn(); + // Let any pending abort/microtask cleanup run. + await new Promise((r) => setTimeout(r, 50)); + // Stop whatever survived so a failing test cannot hang the runner. + for (const handle of live) realClear(handle as Parameters[0]); + return { result, live: live.size }; + } finally { + globalThis.setInterval = realSet; + globalThis.clearInterval = realClear; + } +} + +test("aborting after the stream starts clears both intervals (#13103)", async () => { + const controller = new AbortController(); + const { live } = await withTimerAccounting(async () => { + createBadgeNotificationStream("key-normal", controller.signal); + controller.abort(); + }); + assert.equal(live, 0, "the normal lifecycle must clean up (baseline for the next test)"); +}); + +test("a signal already aborted before start() must not leave timers running (#13103)", async () => { + const controller = new AbortController(); + // The route awaits auth before building the stream, so a client that + // disconnects during that round-trip arrives here already aborted. + controller.abort(); + + const { live } = await withTimerAccounting(() => { + createBadgeNotificationStream("key-preaborted", controller.signal); + }); + + assert.equal( + live, + 0, + `an already-aborted signal left ${live} interval(s) running for the lifetime of the process` + ); +}); + +test("an already-aborted stream is closed rather than left enqueuing (#13103)", async () => { + const controller = new AbortController(); + controller.abort(); + + const stream = createBadgeNotificationStream("key-closed", controller.signal); + const reader = stream.getReader(); + + // enqueue() into an unread stream only buffers -- it does not throw -- so a + // stream left open here would keep filling its queue with nobody draining it. + const { done } = await reader.read(); + assert.equal(done, true, "the stream must be closed when the signal was already aborted"); +}); diff --git a/tests/unit/compression/llmlingua-worker-spawn-12822.test.ts b/tests/unit/compression/llmlingua-worker-spawn-12822.test.ts new file mode 100644 index 0000000000..360a1f2037 --- /dev/null +++ b/tests/unit/compression/llmlingua-worker-spawn-12822.test.ts @@ -0,0 +1,77 @@ +/** + * Regression guard for #12822: the LLMLingua worker must actually spawn on Node. + * + * Root cause: `new Worker(pathToFileURL(file).href, ...)` passes a STRING. Node treats a + * string argument as a filesystem path (it must start with ./ or ../), so a "file://..." + * string is looked up literally and throws ERR_WORKER_PATH. Only a URL INSTANCE is + * interpreted as a file: URL. + * + * Why it was invisible: pump() wraps ensureWorker() in `catch {}` and fails open, so the + * spawn crash silently degraded every compression call to a passthrough instead of erroring. + * + * This test asserts the Node contract directly against a real Worker, so it fails on the + * old `.href` spelling and passes on the URL object. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { Worker } from "node:worker_threads"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const WORKER_SRC = path.resolve( + here, + "../../../open-sse/services/compression/engines/llmlingua/worker.ts" +); + +function spawnWith(arg: string | URL): Promise { + return new Promise((resolve, reject) => { + let w: Worker; + try { + w = new Worker(arg, {}); + } catch (err) { + reject(err); + return; + } + w.on("error", reject); + w.on("exit", () => resolve()); + }); +} + +test("a file: URL STRING is rejected by node:worker_threads (the #12822 crash)", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-worker-")); + const child = path.join(dir, "child.mjs"); + fs.writeFileSync(child, "process.exit(0);\n"); + + await assert.rejects( + () => spawnWith(pathToFileURL(child).href), + (err: NodeJS.ErrnoException) => err.code === "ERR_WORKER_PATH", + "passing .href must fail — this is exactly what shipped and was swallowed by the fail-open catch" + ); + + fs.rmSync(dir, { recursive: true, force: true }); +}); + +test("a file: URL OBJECT spawns cleanly", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-worker-")); + const child = path.join(dir, "child.mjs"); + fs.writeFileSync(child, "process.exit(0);\n"); + + await spawnWith(pathToFileURL(child)); + + fs.rmSync(dir, { recursive: true, force: true }); +}); + +test("worker.ts passes the URL object, not .href", () => { + const code = fs.readFileSync(WORKER_SRC, "utf8"); + assert.ok( + /new Worker\(\s*pathToFileURL\([A-Za-z0-9_]+\)\s*,/.test(code), + "ensureWorker must pass the URL instance to new Worker()" + ); + assert.ok( + !/new Worker\(\s*pathToFileURL\([A-Za-z0-9_]+\)\.href/.test(code), + "ensureWorker must not pass pathToFileURL(...).href — that throws ERR_WORKER_PATH" + ); +}); diff --git a/tests/unit/compression/worker-pool-idle-eviction-12812.test.ts b/tests/unit/compression/worker-pool-idle-eviction-12812.test.ts new file mode 100644 index 0000000000..c95eb181ed --- /dev/null +++ b/tests/unit/compression/worker-pool-idle-eviction-12812.test.ts @@ -0,0 +1,69 @@ +/** + * Regression guard for #12812: idle eviction must terminate the worker thread. + * + * Root cause: finish() scheduled `remove(slot, false)`, so the idle timer dropped the slot + * from the pool WITHOUT calling worker.terminate(). The OS thread, its MessagePort and its + * private heap then survived for the whole process lifetime. Nothing in + * process.memoryUsage() reports that, which is why a 16h instance showed rss=660MB while + * holding 5.7GB of commit charge. + * + * The assertion measures the real thing: a worker that was evicted must no longer be able + * to run code. A live-but-unreferenced thread still responds; a terminated one cannot. + */ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import type { Worker } from "node:worker_threads"; +import { CompressionWorkerPool } from "../../../open-sse/services/compression/compressionWorkerPool.ts"; + +const body = { + model: "gpt-test", + messages: [{ role: "user", content: "please kindly actually simplify this text ".repeat(40) }], +}; + +/** Reach into the pool's private slot set — the leak is only observable there. */ +function slotsOf(pool: CompressionWorkerPool): Set<{ worker: Worker }> { + return (pool as unknown as { workers: Set<{ worker: Worker }> }).workers; +} + +describe("compression worker pool idle eviction (#12812)", () => { + it("terminates the worker thread when the idle timer fires", async () => { + // Idle window short enough to fire during the test. + const pool = new CompressionWorkerPool({ size: 1, idleMs: 50 }); + + await pool.run(body, "stacked", undefined, undefined); + + const slots = [...slotsOf(pool)]; + assert.equal(slots.length, 1, "one worker should have been spawned"); + const { worker } = slots[0]; + + // The observable difference between 'evicted' and 'terminated' is the exit event: + // a leaked thread stays alive and never emits it. Arm the listener BEFORE the idle + // window so we cannot miss the event. + const exited = new Promise((resolve) => { + worker.once("exit", () => resolve(true)); + setTimeout(() => resolve(false), 3_000).unref?.(); + }); + + await new Promise((r) => setTimeout(r, 400)); + assert.equal(slotsOf(pool).size, 0, "slot should be evicted from the pool"); + + assert.equal( + await exited, + true, + "idle eviction must terminate the thread, not just drop the reference (#12812)" + ); + + await pool.close(); + }); + + it("close() terminates every pooled worker", async () => { + const pool = new CompressionWorkerPool({ size: 2, idleMs: 60_000 }); + await Promise.all([ + pool.run(body, "stacked", undefined, undefined), + pool.run(body, "stacked", undefined, undefined), + ]); + assert.ok(slotsOf(pool).size >= 1, "pool should hold workers before close"); + await pool.close(); + assert.equal(slotsOf(pool).size, 0, "close() must drain the pool"); + }); +}); diff --git a/tests/unit/jsonbody-sniff-reader-leak-13169.test.ts b/tests/unit/jsonbody-sniff-reader-leak-13169.test.ts new file mode 100644 index 0000000000..7518f4087c --- /dev/null +++ b/tests/unit/jsonbody-sniff-reader-leak-13169.test.ts @@ -0,0 +1,101 @@ +/** + * Regression test for #13169: the JSON-to-SSE sniff must release the upstream + * body when it unwinds abnormally. + * + * `sniffJsonBodyForSse()` reads the upstream body under `withBodyTimeout()`. + * On a stalled upstream that rejects, an un-cancelled reader keeps the + * connection pinned. The upstream stream declares an explicit `cancel()` hook, + * so the assertions observe real cancellation rather than an incidental close. + */ +import { describe, test } from "node:test"; +import assert from "node:assert/strict"; + +import { maybeConvertJsonBodyToSse } from "../../open-sse/handlers/chatCore/jsonBodyToSse.ts"; + +type Deps = Parameters[2]; + +/** Upstream that serves `first` and then stalls forever, tracking cancellation. */ +function stallingUpstream(first: string) { + const state = { cancelled: false }; + let pulls = 0; + const body = new ReadableStream({ + pull(controller) { + pulls += 1; + if (pulls === 1) { + controller.enqueue(new TextEncoder().encode(first)); + return; + } + return new Promise(() => {}); + }, + cancel() { + state.cancelled = true; + }, + }); + return { body, state }; +} + +function timeoutDeps(ms: number): Deps { + return { + withBodyTimeout: ((p: Promise) => + Promise.race([ + p, + new Promise((_, reject) => + setTimeout(() => { + const err = new Error(`Response body read timeout after ${ms}ms`); + err.name = "BodyTimeoutError"; + reject(err); + }, ms) + ), + ])) as Deps["withBodyTimeout"], + synthesizeOpenAiSseFromJson: () => null, + } as Deps; +} + +describe("jsonBodyToSse upstream body release (#13169)", () => { + test("cancels the upstream body when the sniff times out", async () => { + const { body, state } = stallingUpstream('{"choices":['); + const providerResponse = new Response(body, { + status: 200, + headers: { "content-type": "application/json" }, + }); + + await assert.rejects( + () => + maybeConvertJsonBodyToSse(providerResponse, { provider: "p", model: "m" }, timeoutDeps(50)), + (err: Error) => err.name === "BodyTimeoutError" + ); + + // Let any async cancellation settle before observing. + await new Promise((r) => setTimeout(r, 50)); + + assert.equal(state.cancelled, true, "upstream body should be cancelled after the timeout"); + }); + + test("does NOT cancel the body on the success path", async () => { + // A complete SSE-looking body: the sniff hands the reader onward, so + // cancelling here would truncate a healthy stream. + const state = { cancelled: false }; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: {}\n\n")); + controller.close(); + }, + cancel() { + state.cancelled = true; + }, + }); + const providerResponse = new Response(body, { + status: 200, + headers: { "content-type": "application/json" }, + }); + + const out = await maybeConvertJsonBodyToSse( + providerResponse, + { provider: "p", model: "m" }, + timeoutDeps(5000) + ); + + assert.ok(out instanceof Response, "sniff should return a Response"); + assert.equal(state.cancelled, false, "a healthy body must not be cancelled by the sniff"); + }); +}); diff --git a/tests/unit/logstream-timer-leak-13113.test.ts b/tests/unit/logstream-timer-leak-13113.test.ts new file mode 100644 index 0000000000..111ad34c9c --- /dev/null +++ b/tests/unit/logstream-timer-leak-13113.test.ts @@ -0,0 +1,98 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import type { AddressInfo } from "node:net"; + +import { createLogStream } from "../../src/lib/cli-helper/log-streamer.ts"; + +function armedTimers(): number { + return process.getActiveResourcesInfo().filter((r) => r === "Timeout").length; +} + +async function startServer(): Promise<{ port: number; close: () => Promise }> { + const open: http.ServerResponse[] = []; + const server = http.createServer((_req, res) => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.write("log line\n"); + // Deliberately left open: stop() must land while the stream is still live. + open.push(res); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + return { + port, + close: async () => { + for (const res of open) res.end(); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} + +test("stop() clears the stream timeout timer", async () => { + const server = await startServer(); + try { + const before = armedTimers(); + + const streams = Array.from({ length: 8 }, () => + createLogStream({ + baseUrl: `http://127.0.0.1:${server.port}`, + follow: true, + // Long enough that a leaked timer is still armed when we measure. + timeout: 120_000, + }) + ); + + // Begin consuming so start() runs and the fetch is in flight. + for (const s of streams) { + void s.stream + .getReader() + .read() + .catch(() => {}); + } + await new Promise((r) => setTimeout(r, 300)); + + for (const s of streams) s.stop(); + await new Promise((r) => setTimeout(r, 500)); + + const after = armedTimers(); + assert.ok( + after <= before, + `stopping 8 streams retained ${after - before} armed timer(s) ` + + `(before=${before} after=${after}); stop() must clear the timeout` + ); + } finally { + await server.close(); + } +}); + +test("a stream that ends normally still clears its timer", async () => { + const finished = http.createServer((_req, res) => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end("done\n"); + }); + await new Promise((resolve) => finished.listen(0, "127.0.0.1", resolve)); + const { port } = finished.address() as AddressInfo; + + try { + const before = armedTimers(); + const { stream } = createLogStream({ + baseUrl: `http://127.0.0.1:${port}`, + follow: false, + timeout: 120_000, + }); + + const reader = stream.getReader(); + while (true) { + const { done } = await reader.read(); + if (done) break; + } + await new Promise((r) => setTimeout(r, 200)); + + assert.ok( + armedTimers() <= before, + "a normally-completed stream must not leave its timeout armed" + ); + } finally { + await new Promise((resolve) => finished.close(() => resolve())); + } +}); diff --git a/tests/unit/nodesqlite-process-listener-leak-13108.test.ts b/tests/unit/nodesqlite-process-listener-leak-13108.test.ts new file mode 100644 index 0000000000..500461a0c1 --- /dev/null +++ b/tests/unit/nodesqlite-process-listener-leak-13108.test.ts @@ -0,0 +1,63 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { tmpdir } from "node:os"; +import { mkdtempSync, rmSync } from "node:fs"; +import { join } from "node:path"; + +const { createNodeSqliteAdapter } = await import("../../src/lib/db/adapters/nodeSqliteAdapter.ts"); + +const SIGNALS = ["beforeExit", "SIGINT", "SIGTERM"] as const; + +function counts(): Record { + return Object.fromEntries(SIGNALS.map((s) => [s, process.listenerCount(s)])); +} + +function delta(before: Record, after: Record) { + return Object.fromEntries(SIGNALS.map((s) => [s, after[s] - before[s]])); +} + +test("closing a node:sqlite adapter releases its process listeners (#13108)", async () => { + const dir = mkdtempSync(join(tmpdir(), "omniroute-dbleak-")); + const before = counts(); + + try { + // Short-lived adapters are a real pattern: POST /api/db-backups/import + // opens one per request purely to validate the uploaded file. + const N = 12; + for (let i = 0; i < N; i++) { + const adapter = await createNodeSqliteAdapter(join(dir, `probe-${i}.sqlite`)); + adapter.close(); + } + + const leaked = delta(before, counts()); + for (const signal of SIGNALS) { + assert.equal( + leaked[signal], + 0, + `${N} open+close cycles retained ${leaked[signal]} "${signal}" listener(s) on process` + ); + } + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("an open node:sqlite adapter keeps its shutdown listeners registered (#13108)", async () => { + const dir = mkdtempSync(join(tmpdir(), "omniroute-dbleak-open-")); + const before = counts(); + let adapter: Awaited> | null = null; + + try { + adapter = await createNodeSqliteAdapter(join(dir, "open.sqlite")); + + // The fix must not detach eagerly: these handlers are what checkpoint the + // WAL on Ctrl-C, so they have to stay armed for as long as the db is open. + const armed = delta(before, counts()); + for (const signal of SIGNALS) { + assert.equal(armed[signal], 1, `an open adapter must keep its "${signal}" handler`); + } + } finally { + adapter?.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/tests/unit/plugins-sigkill-listener-leak-12819.test.ts b/tests/unit/plugins-sigkill-listener-leak-12819.test.ts new file mode 100644 index 0000000000..df6a39638a --- /dev/null +++ b/tests/unit/plugins-sigkill-listener-leak-12819.test.ts @@ -0,0 +1,100 @@ +// Regression test for #12819 — loadPlugin() leaked one "exit" listener per hook timeout. +// +// Root cause: on the SIGTERM→SIGKILL escalation path the loader attached a fresh +// `child.once("exit", () => clearTimeout(killTimer))`. `once` only detaches when exit +// actually FIRES, so a plugin that ignores SIGTERM leaves the listener (and its killTimer +// closure) attached on every hook timeout. Node then prints MaxListenersExceededWarning +// once 11 accumulate. +// +// The plugin below traps SIGTERM and keeps running, which is exactly the condition the +// bug needs. We drive several hook timeouts and assert the listener count stays bounded. +import { test, describe, after } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, writeFileSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +const { loadPlugin } = await import("../../src/lib/plugins/loader.ts"); + +const dirs: string[] = []; +after(() => { + for (const d of dirs) rmSync(d, { recursive: true, force: true }); +}); + +/** A plugin that ignores SIGTERM and never answers a hook, forcing the escalation path. */ +function writeStubbornPlugin(): string { + const dir = mkdtempSync(join(tmpdir(), "omniroute-plugin-12819-")); + dirs.push(dir); + const entry = join(dir, "index.mjs"); + writeFileSync( + entry, + [ + // Trap SIGTERM so the loader has to escalate to SIGKILL. + 'process.on("SIGTERM", () => {});', + "export default {", + " // Never resolves → every call hits the hook timeout.", + " onRequest: () => new Promise(() => {}),", + "};", + "", + ].join("\n") + ); + return entry; +} + +describe("plugin loader SIGKILL escalation (#12819)", () => { + test("does not accumulate an exit listener per hook timeout", async () => { + const entryPoint = writeStubbornPlugin(); + const loaded = await loadPlugin( + entryPoint, + { + name: "sigkill-listener-leak", + version: "1.0.0", + license: "MIT", + main: "index.mjs", + source: "local", + tags: [], + requires: { permissions: [] }, + hooks: { onRequest: true, onResponse: false, onError: false }, + skills: [], + enabledByDefault: false, + configSchema: {}, + } as never, + { hookTimeoutMs: 120 } + ); + + const onRequest = ( + loaded.plugin as unknown as { + onRequest?: (ctx: unknown) => Promise; + } + ).onRequest; + assert.ok(onRequest, "onRequest hook should be registered"); + + // `child` is private to the loader, so observe the leak the way a user does: Node + // itself emits MaxListenersExceededWarning once an emitter passes 10 listeners. + const warnings: string[] = []; + const onWarning = (w: Error) => { + if (w.name === "MaxListenersExceededWarning") warnings.push(w.message); + }; + process.on("warning", onWarning); + + try { + // 12 timeouts: comfortably past Node's default limit of 10, so the pre-fix code + // trips the warning while the fixed code stays flat. + for (let i = 0; i < 12; i++) { + await onRequest({ body: {} }).catch(() => undefined); + } + // Warnings are delivered on the next tick; let them land before asserting. + await new Promise((r) => setTimeout(r, 50)); + } finally { + process.removeListener("warning", onWarning); + } + + assert.deepEqual( + warnings, + [], + `hook timeouts must not accumulate exit listeners (#12819): ${warnings[0] ?? ""}` + ); + + loaded.cleanup?.(); + }); +}); diff --git a/tests/unit/telegram-keycache-bounded-13165.test.ts b/tests/unit/telegram-keycache-bounded-13165.test.ts new file mode 100644 index 0000000000..46305744c5 --- /dev/null +++ b/tests/unit/telegram-keycache-bounded-13165.test.ts @@ -0,0 +1,123 @@ +/** + * Regression test for #13165: the Telegram per-user key cache must stay bounded. + * + * `resolveUserApiKey()` is reachable from the webhook path of + * POST /api/telegram/update with a caller-supplied chat id, so an uncapped Map + * grows for the lifetime of the process. The cache is module-private, so this + * asserts the observable LRU contract: a cold id is re-minted after a burst of + * distinct ids (proving eviction), while a recently used id survives it. + * + * Runner: node:test (tests/unit/*.test.ts), so DB access is stubbed through a + * module mock rather than vi.mock. + */ +import { test, describe, before, beforeEach } from "node:test"; +import assert from "node:assert/strict"; +import { register } from "node:module"; +import { pathToFileURL } from "node:url"; + +const CAP = 1000; + +/** Names passed to createApiKey — one entry per real mint (i.e. per cache miss). */ +const minted: string[] = []; + +let resolveUserApiKey: (id: number) => Promise; + +before(async () => { + // Stub the DB + machine-id modules so nothing touches SQLite. The loader + // matches the specifiers used by chatProxy.ts. The stub must export every + // name the real module exports: chatProxy pulls in the chat handler, which + // imports other members of this module, and a missing export is a module-load + // SyntaxError that would look like a failing assertion. + const dbExports = [ + "clearApiKeyCaches", + "deleteApiKey", + "getApiKeyById", + "getApiKeyMetadata", + "getApiKeysCount", + "getExclusiveLeaseConnectionIds", + "isModelAllowedForKey", + "pickApiKeyForInternalUse", + "regenerateApiKey", + "resetApiKeyState", + "revokeApiKey", + "setApiKeyExpiry", + "updateApiKeyPermissions", + "validateApiKey", + ]; + + const dbStub = ` + export async function getApiKeys() { return []; } + export async function createApiKey(name) { + globalThis.__mintedKeys.push(name); + return { key: "sk-omni-" + "x".repeat(32) + "-" + name }; + } + ${dbExports.map((n) => `export async function ${n}() { return null; }`).join("\n")} + `; + const machineStub = ` + export async function getConsistentMachineId() { return "0000000000000000"; } + `; + + (globalThis as Record).__mintedKeys = minted; + + const loader = ` + export async function resolve(spec, ctx, next) { + if (spec.includes("db/apiKeys")) { + return { url: "data:text/javascript,${encodeURIComponent(dbStub)}", shortCircuit: true }; + } + if (spec.includes("machineId")) { + return { url: "data:text/javascript,${encodeURIComponent(machineStub)}", shortCircuit: true }; + } + return next(spec, ctx); + } + `; + register("data:text/javascript," + encodeURIComponent(loader), pathToFileURL("./")); + + ({ resolveUserApiKey } = await import("../../src/lib/telegram/chatProxy.ts")); +}); + +describe("telegram keyCache bounding (#13165)", () => { + beforeEach(() => { + minted.length = 0; + }); + + test("evicts a cold id once the cap is exceeded", async () => { + const victim = 7_000_001; + const beforeFirstResolve = minted.length; + await resolveUserApiKey(victim); + assert.equal(minted.length - beforeFirstResolve, 1, "first resolve should mint exactly once"); + + // Never touch `victim` again: it must fall out of a CAP-sized cache. + for (let i = 0; i < CAP + 50; i++) await resolveUserApiKey(600_000 + i); + + // Measure the victim's own resolve in isolation. Comparing against the + // running total would be dominated by the burst's own mints and would pass + // even with an unbounded cache. + const beforeVictimResolve = minted.length; + await resolveUserApiKey(victim); + const mintedForVictim = minted.length - beforeVictimResolve; + + // Evicted => cache miss => exactly one fresh mint for this id. + assert.equal( + mintedForVictim, + 1, + `expected victim to be re-minted after eviction, got ${mintedForVictim} mint(s)` + ); + }); + + test("keeps a recently used id alive across a burst of new ids", async () => { + const active = 8_000_001; + const first = await resolveUserApiKey(active); + + // Touch the active id throughout the burst so it stays most-recently-used. + for (let i = 0; i < CAP * 2; i++) { + await resolveUserApiKey(500_000 + i); + if (i % 100 === 0) await resolveUserApiKey(active); + } + + const mintsBefore = minted.length; + const again = await resolveUserApiKey(active); + + assert.equal(again, first, "active id should keep its cached key"); + assert.equal(minted.length, mintsBefore, "active id should not be re-minted"); + }); +}); diff --git a/tests/unit/telegram-webhook-secret-13172.test.ts b/tests/unit/telegram-webhook-secret-13172.test.ts new file mode 100644 index 0000000000..f525ee745a --- /dev/null +++ b/tests/unit/telegram-webhook-secret-13172.test.ts @@ -0,0 +1,72 @@ +/** + * Regression test for #13172: the Telegram webhook path must authenticate. + * + * Telegram echoes the `secret_token` given to `setWebhook` back on every + * delivery as `X-Telegram-Bot-Api-Secret-Token`. Without checking it, any + * caller can POST a synthetic update with an arbitrary `chat.id`, which reaches + * proxyChat() and mints a real API key plus upstream spend. + * + * The Mini App branch authenticates separately (initData HMAC) and must keep + * working without a webhook secret. + */ +import { describe, test, before, after } from "node:test"; +import assert from "node:assert/strict"; + +const BOT_TOKEN = "123456:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; +const SECRET = "s3cret-webhook-token"; + +let POST: (req: Request) => Promise; +let webhookSecretMatches: (a: string, b: string) => boolean; +const proxied: number[] = []; + +before(async () => { + process.env.TELEGRAM_BOT_TOKEN = BOT_TOKEN; + process.env.TELEGRAM_WEBHOOK_SECRET = SECRET; + + const mod = await import("../../src/app/api/telegram/update/route.ts"); + POST = mod.POST as typeof POST; + webhookSecretMatches = mod.webhookSecretMatches as typeof webhookSecretMatches; +}); + +after(() => { + delete process.env.TELEGRAM_WEBHOOK_SECRET; +}); + +function webhookRequest(headers: Record = {}): Request { + return new Request("https://example.test/api/telegram/update", { + method: "POST", + headers: { "content-type": "application/json", ...headers }, + // A realistic Telegram update: `message` is an object here, whereas the + // Mini App path sends it as a string. Both shapes must reach their branch. + body: JSON.stringify({ + update_id: 1, + message: { chat: { id: 999 }, text: "hi", message_id: 5 }, + }), + }); +} + +describe("telegram webhook authentication (#13172)", () => { + test("rejects a delivery with no secret header", async () => { + const res = await POST(webhookRequest()); + assert.equal(res.status, 401, "unauthenticated webhook must be rejected"); + assert.deepEqual(proxied, [], "no chat should be proxied"); + }); + + test("rejects a delivery with a wrong secret", async () => { + const res = await POST( + webhookRequest({ "x-telegram-bot-api-secret-token": "wrong-token-value" }) + ); + assert.equal(res.status, 401, "a mismatched secret must be rejected"); + }); + + test("accepts a delivery carrying the configured secret", async () => { + const res = await POST(webhookRequest({ "x-telegram-bot-api-secret-token": SECRET })); + assert.equal(res.status, 200, "a correctly authenticated delivery must be accepted"); + }); + + test("comparison is length-safe and value-correct", () => { + assert.equal(webhookSecretMatches(SECRET, SECRET), true); + assert.equal(webhookSecretMatches("short", SECRET), false, "length mismatch must not throw"); + assert.equal(webhookSecretMatches("", ""), true, "equal empties compare equal"); + }); +}); diff --git a/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts b/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts new file mode 100644 index 0000000000..230a3f28d2 --- /dev/null +++ b/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts @@ -0,0 +1,134 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import net from "node:net"; +import type { AddressInfo } from "node:net"; + +import { GET } from "@/app/api/tools/traffic-inspector/ws/route"; +import { globalTrafficBuffer } from "@/mitm/inspector/buffer"; + +const DEAD_UPGRADES = 6; + +function armedTimers(): number { + return process.getActiveResourcesInfo().filter((r) => r === "Timeout").length; +} + +function upgradeRequest(socket: net.Socket): Request { + const req = new Request("http://127.0.0.1/api/tools/traffic-inspector/ws", { + headers: { + upgrade: "websocket", + "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ==", + }, + }); + Object.defineProperty(req, "socket", { value: socket, configurable: true }); + return req; +} + +async function deadSocket(port: number): Promise { + const sock = net.connect(port, "127.0.0.1"); + await new Promise((r) => sock.once("connect", () => r())); + sock.on("error", () => {}); + sock.destroy(); + await new Promise((r) => setTimeout(r, 20)); + return sock; +} + +test("an already-closed socket leaves no subscriber and no ping timer", async () => { + const accepted: net.Socket[] = []; + const server = net.createServer((c) => { + accepted.push(c); + c.on("error", () => {}); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const { port } = server.address() as AddressInfo; + + try { + const timersBefore = armedTimers(); + const subsBefore = globalTrafficBuffer.subscriberCount(); + + const handlers: Promise[] = []; + for (let i = 0; i < DEAD_UPGRADES; i++) { + // Catch at creation time: the route answers a hijacked upgrade with a 101 + // Response, which undici rejects off a real server. Left unattached, that + // rejection would sit through the next await and trip Node's unhandled + // rejection detection. Either settlement proves the handler released its + // resources instead of hanging, which is what this test measures. + handlers.push(GET(upgradeRequest(await deadSocket(port))).catch(() => undefined)); + } + + // Own the race timer so it can be cleared before measuring; otherwise the + // test's own armed timeout is counted as a leaked one. + let raceTimer: ReturnType | undefined; + const outcome = await Promise.race([ + Promise.all(handlers).then(() => "settled"), + new Promise((r) => { + raceTimer = setTimeout(() => r("hung"), 2000); + }), + ]); + if (raceTimer) clearTimeout(raceTimer); + assert.equal( + outcome, + "settled", + "each handler must return instead of hanging forever on a dead socket" + ); + + const timersAfter = armedTimers(); + assert.ok( + timersAfter <= timersBefore, + `${DEAD_UPGRADES} dead upgrades retained ${timersAfter - timersBefore} ping timer(s)` + ); + + // Measure the subscriber set directly; counting fan-out to our own probe + // says nothing about whether the dead sockets stayed subscribed. + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore, + `${DEAD_UPGRADES} dead upgrades left ${globalTrafficBuffer.subscriberCount() - subsBefore} subscriber(s) behind` + ); + } finally { + // close() only fires once every accepted connection is gone. + for (const c of accepted) c.destroy(); + await new Promise((r) => server.close(() => r())); + } +}); + +test("a live socket keeps its subscription until the socket closes", async () => { + const accepted: net.Socket[] = []; + const server = net.createServer((c) => { + accepted.push(c); + c.on("error", () => {}); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const { port } = server.address() as AddressInfo; + + const sock = net.connect(port, "127.0.0.1"); + await new Promise((r) => sock.once("connect", () => r())); + sock.on("error", () => {}); + + try { + const subsBefore = globalTrafficBuffer.subscriberCount(); + + const handler = GET(upgradeRequest(sock)).catch(() => undefined); + await new Promise((r) => setTimeout(r, 100)); + + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore + 1, + "a live upgrade must register exactly one traffic subscriber" + ); + + // Closing the socket resolves the handler's `settled` promise, which is the + // only path that releases the subscriber. + sock.destroy(); + await handler; + + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore, + "closing the socket must release the subscriber" + ); + } finally { + // close() only fires once every accepted connection is gone. + for (const c of accepted) c.destroy(); + await new Promise((r) => server.close(() => r())); + } +}); diff --git a/tests/unit/webdav-server-3485.test.ts b/tests/unit/webdav-server-3485.test.ts index f30d5824e3..291d25dc03 100644 --- a/tests/unit/webdav-server-3485.test.ts +++ b/tests/unit/webdav-server-3485.test.ts @@ -30,14 +30,19 @@ import path from "node:path"; import http from "node:http"; import { EventEmitter } from "node:events"; import { createCipheriv, randomBytes, scryptSync } from "node:crypto"; -import { pathToFileURL } from "node:url"; +import { fileURLToPath, pathToFileURL } from "node:url"; // ───────────────────────────────────────────────────────────────────────────── // Helpers // ───────────────────────────────────────────────────────────────────────────── +// `URL.pathname` is a URL path, not an OS path: on Windows it yields +// "/C:/..." — a leading slash before the drive letter. `path.resolve` does not +// treat that as absolute, so it prepends the CWD and produces "C:\C:\...", +// which fails to import. `fileURLToPath` decodes to a real OS path on every +// platform (it also un-escapes %20 in paths containing spaces). const HANDLER_PATH = path.resolve( - path.dirname(new URL(import.meta.url).pathname), + path.dirname(fileURLToPath(import.meta.url)), "../../scripts/dev/webdav-handler.mjs" );