From 994a438cfd803e83a9ae582dd680fecfc979ec82 Mon Sep 17 00:00:00 2001 From: jubnl Date: Sat, 11 Jul 2026 21:58:05 +0200 Subject: [PATCH] fix(plugins): throttle the plugin log channel to prevent host-thread DoS The per-plugin RpcRateLimiter only guarded the ctx.* (req) channel; ctx.log.*, stdout/stderr and unknown evt topics reached a synchronous INSERT+prune on the host thread unthrottled, so a while(true) ctx.log.error(...) loop could freeze the instance. Route every plugin-driven log path through a per-plugin log token bucket; excess lines are dropped with a summary line on resume. --- server/src/nest/plugins/host/rate-limit.ts | 10 +++++ .../plugins/supervisor/plugin-supervisor.ts | 35 ++++++++++++--- .../unit/plugins/supervisor-lifecycle.test.ts | 43 ++++++++++++++++++- 3 files changed, 82 insertions(+), 6 deletions(-) diff --git a/server/src/nest/plugins/host/rate-limit.ts b/server/src/nest/plugins/host/rate-limit.ts index 32d7cc01b..091dc5fea 100644 --- a/server/src/nest/plugins/host/rate-limit.ts +++ b/server/src/nest/plugins/host/rate-limit.ts @@ -57,6 +57,16 @@ export const DEFAULT_RPC_LIMIT: RpcLimitConfig = { maxInFlight: Number(process.env.TREK_PLUGIN_RPC_INFLIGHT) || 16, }; +// Plugin log volume (ctx.log.*, stdout/stderr, unknown evt topics) reaches the +// host's synchronous error-log INSERT+prune, and — unlike ctx.* calls — bypasses +// the RpcRateLimiter above. A separate, generous bucket throttles it so a +// `while (true) ctx.log.error(...)` loop can't pin the host thread; a legitimate +// plugin logs far below this and never notices. Excess lines are dropped. +export const DEFAULT_LOG_LIMIT = { + burst: Number(process.env.TREK_PLUGIN_LOG_BURST) || 50, + perSec: Number(process.env.TREK_PLUGIN_LOG_PER_SEC) || 10, +}; + /** Per-plugin limiter: a token bucket + an in-flight gauge. */ export class RpcRateLimiter { private readonly bucket: TokenBucket; diff --git a/server/src/nest/plugins/supervisor/plugin-supervisor.ts b/server/src/nest/plugins/supervisor/plugin-supervisor.ts index 352765d9e..1cc8b05ea 100644 --- a/server/src/nest/plugins/supervisor/plugin-supervisor.ts +++ b/server/src/nest/plugins/supervisor/plugin-supervisor.ts @@ -6,7 +6,7 @@ import type { Envelope, RpcError, RpcRequest } from '../protocol/envelope'; import type { PluginRpcHost } from '../host/rpc-host'; import { scheduleJobs, stopJobs, type ScheduledJob } from '../host/plugin-jobs'; import { SNAPSHOT_GRANT, type PluginEventMeta } from '../../../plugin-event-sink'; -import { RpcRateLimiter, DEFAULT_RPC_LIMIT } from '../host/rate-limit'; +import { RpcRateLimiter, DEFAULT_RPC_LIMIT, TokenBucket, DEFAULT_LOG_LIMIT } from '../host/rate-limit'; export interface PluginRouteInfo { i: number; @@ -60,6 +60,8 @@ interface Supervised { pending: Map; // host→child invokes awaiting a response invocations: Map; // reqId -> acting user of that invoke (undefined = no user, e.g. a job) rpcLimiter: RpcRateLimiter; // caps this plugin's ctx.* call rate + concurrency (host-loop DoS guard) + logLimiter: TokenBucket; // caps this plugin's log/stderr volume (host-loop DoS guard, separate from rpcLimiter) + droppedLogs: number; // count of log lines dropped by logLimiter since the last one got through activation?: { resolve: () => void; reject: (e: Error) => void }; activationTimer?: ReturnType; // deadline for the plugin to reach 'active' respawnTimer?: ReturnType; // crash-backoff restart timer (identity-guarded) @@ -162,6 +164,8 @@ export class PluginSupervisor { pending: new Map(), invocations: new Map(), rpcLimiter: new RpcRateLimiter(DEFAULT_RPC_LIMIT, Date.now()), + logLimiter: new TokenBucket(DEFAULT_LOG_LIMIT.burst, DEFAULT_LOG_LIMIT.perSec, Date.now()), + droppedLogs: 0, }; this.running.set(id, sup); this.ensureSweep(); @@ -486,8 +490,29 @@ export class PluginSupervisor { child.on('message', (raw: unknown) => this.onMessage(sup, raw as Envelope)); child.on('exit', (code, signal) => this.onExit(sup, code, signal)); child.on('error', (e) => this.hooks.onLog?.(sup.id, 'error', `child error: ${e.message}`)); - child.stdout?.on('data', (b) => this.hooks.onLog?.(sup.id, 'info', String(b).trimEnd())); - child.stderr?.on('data', (b) => this.hooks.onLog?.(sup.id, 'error', String(b).trimEnd())); + child.stdout?.on('data', (b) => this.recordLog(sup, 'info', String(b).trimEnd())); + child.stderr?.on('data', (b) => this.recordLog(sup, 'error', String(b).trimEnd())); + } + + /** + * Funnel a plugin-driven log line (ctx.log.*, stdout/stderr, unknown evt topics) + * to the host log sink through a per-plugin token bucket. onLog persists warn/error + * with a synchronous INSERT + prune, and this path is NOT covered by rpcLimiter, so + * an unthrottled log flood (`while (true) ctx.log.error(...)`) would pin the host + * thread. Excess lines are dropped; when logging resumes, one summary line reports + * how many were dropped so the throttling is visible to an operator. + */ + private recordLog(sup: Supervised, level: string, msg: string, meta?: unknown): void { + if (!sup.logLimiter.take(Date.now())) { + sup.droppedLogs += 1; + return; + } + if (sup.droppedLogs > 0) { + const dropped = sup.droppedLogs; + sup.droppedLogs = 0; + this.hooks.onLog?.(sup.id, 'warn', `[trek] ${dropped} log line(s) dropped (plugin log rate limit exceeded)`); + } + this.hooks.onLog?.(sup.id, level, msg, meta); } private async onMessage(sup: Supervised, msg: Envelope): Promise { @@ -598,7 +623,7 @@ export class PluginSupervisor { } case 'log': { const d = msg.data as { level?: string; msg?: string; meta?: unknown }; - this.hooks.onLog?.(sup.id, d.level || 'info', d.msg || '', d.meta); + this.recordLog(sup, d.level || 'info', d.msg || '', d.meta); break; } default: @@ -608,7 +633,7 @@ export class PluginSupervisor { // than forward it blindly. `onEvent` stays on SupervisorHooks, reserved // for a future SDK-sanctioned custom-event channel, but is not invoked // for arbitrary topics. - this.hooks.onLog?.(sup.id, 'warn', `dropped unknown plugin event topic: ${msg.topic}`); + this.recordLog(sup, 'warn', `dropped unknown plugin event topic: ${msg.topic}`); } } } diff --git a/server/tests/unit/plugins/supervisor-lifecycle.test.ts b/server/tests/unit/plugins/supervisor-lifecycle.test.ts index 6149761fb..ce780c653 100644 --- a/server/tests/unit/plugins/supervisor-lifecycle.test.ts +++ b/server/tests/unit/plugins/supervisor-lifecycle.test.ts @@ -6,7 +6,7 @@ */ import { describe, it, expect, vi, afterEach } from 'vitest'; import { PluginSupervisor } from '../../../src/nest/plugins/supervisor/plugin-supervisor'; -import { RpcRateLimiter } from '../../../src/nest/plugins/host/rate-limit'; +import { RpcRateLimiter, TokenBucket } from '../../../src/nest/plugins/host/rate-limit'; function makeSupervisor() { const dispose = vi.fn(); @@ -269,3 +269,44 @@ describe('supervisor rate-limits ctx.* dispatch', () => { expect(last).toMatchObject({ k: 'res', id: 'r2', ok: false, error: { code: 'HOST_ERROR' } }); }); }); + +describe('supervisor throttles plugin log/stderr volume (host-thread DoS guard)', () => { + // Regression: the rpcLimiter only covers the `req` (ctx.*) channel. Log lines + // (ctx.log.*, stderr, unknown evt topics) reach a SYNCHRONOUS INSERT+prune on the + // host thread through a different branch, so a `while(true) ctx.log.error(...)` loop + // would freeze the instance unless that path is throttled too. + it('drops evt/log lines beyond the per-plugin bucket so a ctx.log flood cannot pin the host thread', async () => { + const onLog = vi.fn(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {}); + const sup = { id: 'p', status: 'active', logLimiter: new TokenBucket(3, 0, Date.now()), droppedLogs: 0 }; + // 10 rapid error lines through the real evt→log path; the bucket holds only 3 + // tokens (refill 0/sec for the test), so only 3 reach onLog's INSERT+prune. + for (let i = 0; i < 10; i++) { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + await (s as any).onMessage(sup, { k: 'evt', topic: 'log', data: { level: 'error', msg: `line ${i}` } }); + } + expect(onLog).toHaveBeenCalledTimes(3); + expect(sup.droppedLogs).toBe(7); + }); + + it('surfaces a single "N dropped" summary when logging resumes after throttling', () => { + const onLog = vi.fn(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {}); + const sup = { id: 'p', logLimiter: new TokenBucket(1, 0, Date.now()), droppedLogs: 0 }; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const rec = (level: string, msg: string) => (s as any).recordLog(sup, level, msg); + rec('error', 'a'); // spends the only token + rec('error', 'b'); // dropped + rec('error', 'c'); // dropped + expect(sup.droppedLogs).toBe(2); + // Simulate a refill; the next accepted line first reports the drop, then itself. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (sup.logLimiter as any).tokens = 1; + rec('error', 'd'); + expect(onLog).toHaveBeenCalledWith('p', 'warn', expect.stringContaining('2 log line(s) dropped')); + expect(onLog).toHaveBeenCalledWith('p', 'error', 'd', undefined); + expect(sup.droppedLogs).toBe(0); + }); +});