diff --git a/docker-compose.yml b/docker-compose.yml index 276cc175f..8398df654 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,6 +11,12 @@ services: # Longer than the backend's 10 s internal shutdown deadline, so Docker # never SIGKILLs the backend while it removes its sandbox containers. stop_grace_period: 15s + # Bounded on-disk log: Docker's json-file default keeps every line forever. + logging: + driver: json-file + options: + max-size: "${UNOSIM_LOG_MAX_SIZE:-10m}" + max-file: "${UNOSIM_LOG_MAX_FILE:-5}" group_add: - "${DOCKER_GID:?Set DOCKER_GID to the numeric GID of /var/run/docker.sock}" environment: diff --git a/docs/INSTALL_SERVER.md b/docs/INSTALL_SERVER.md index 57db2cb67..f62d0356c 100644 --- a/docs/INSTALL_SERVER.md +++ b/docs/INSTALL_SERVER.md @@ -141,6 +141,7 @@ muss diese Werte nicht ändern. Für größere Installationen können sie über | `SANDBOX_MEMORY_MB` | Memory-Limit pro Sandbox | | `SANDBOX_CPU_LIMIT` | CPU-Limit pro Sandbox | | `UNOSIM_GITHUB_TOKEN` | optionaler Read-only-Token nur für `api.github.com` (Ref-Auflösung der Kursinhalte); ohne Token gilt das GitHub-Limit von 60 Abfragen pro Stunde und Server-IP | +| `UNOSIM_LOG_MAX_SIZE`, `UNOSIM_LOG_MAX_FILE` | Rotation des Backend-Logs (Docker `json-file`); Standard `10m` je Datei und `5` Dateien | | `WS_HEARTBEAT_INTERVAL_MS` | Ping-Intervall der WebSocket-Verbindungen; eine Verbindung ohne Antwort auf den vorherigen Ping wird getrennt und gibt ihre Simulation frei; Standard `30000` ms | Die Defaults der Anwendung stehen in `server/config.ts` (unter anderem diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 8af61e663..c1e5c789d 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -134,6 +134,13 @@ development startup reject the flag. - Each WebSocket connection is limited to 500 messages per second (burst 1000); excess messages are dropped. At most 1 MiB of sketch input may wait in a sketch's stdin. Connections that miss a heartbeat pong are closed. +- Sketch code controls the simulation's stdout and stderr. Plain runtime text + reaches the client at up to 50 lines per second after a burst of 200 lines + per run; further lines are dropped and counted, the client is told once and + the log reports drops aggregated (at most every 10 s), never per line. At + most 4 I/O-registry cycles per run are processed; the generated program + emits one. Compose rotates the backend log (`json-file`, 5 × 10 MB by + default). - Compilers stop after 50 errors (`-fmax-errors`) and after 8 MiB of captured output. At most 256 KiB of compiler diagnostics reach the client; the rest is reported as omitted. diff --git a/server/routes/simulation.ws.ts b/server/routes/simulation.ws.ts index da471cfed..30451f249 100644 --- a/server/routes/simulation.ws.ts +++ b/server/routes/simulation.ws.ts @@ -326,8 +326,10 @@ export function registerSimulationWebSocket( outputBuffer.sendSerialOutputBatched(ws, line, isComplete); }; + // Runtime diagnostics are sketch-controlled and already bounded per run; + // a log entry per line would let a sketch flood the backend log. const onError = (err: string) => { - logger.warn(`[Client WS][ERR] diagnostic received (${Buffer.byteLength(err)} bytes)`); + logger.debug(`[Client WS][ERR] diagnostic received (${Buffer.byteLength(err)} bytes)`); outputBuffer.flushSerialOutputBuffer(ws); sendMessageToClient(ws, { type: WSMessageType.SERIAL_OUTPUT, diff --git a/server/services/registry-manager.ts b/server/services/registry-manager.ts index 51e02498f..6b1637339 100644 --- a/server/services/registry-manager.ts +++ b/server/services/registry-manager.ts @@ -48,12 +48,23 @@ interface RegistryManagerConfig { * It provides debouncing to minimize WebSocket traffic and change detection * to avoid sending duplicate registry data. */ +/** + * The generated main() emits one registry per run. Registry markers are written + * to stdout/stderr, which sketch code controls, so further cycles are ignored: + * each would otherwise send, log and persist a registry. + */ +export const MAX_REGISTRY_COLLECTIONS_PER_RUN = 4; + export class RegistryManager { private registry: IOPinRecord[] = []; /** Runtime pin modes observed before the first IO_REGISTRY_START marker. */ private preCollectionRegistry: IOPinRecord[] = []; private hasStartedCollection = false; private isCollecting = false; + private collectionsThisRun = 0; + /** Inside a registry cycle beyond the per-run bound; its markers are ignored. */ + private ignoringCollection = false; + private ignoredRegistryCycles = 0; private registryHash = ""; private debounceTimer: NodeJS.Timeout | null = null; private waitTimer: NodeJS.Timeout | null = null; @@ -390,6 +401,11 @@ export class RegistryManager { */ startCollection(): void { if (this.destroyed) return; + if (this.collectionsThisRun >= MAX_REGISTRY_COLLECTIONS_PER_RUN) { + this.ignoreCollection(); + return; + } + this.collectionsThisRun++; // Collection start (not logged individually — too noisy). // ROBUSTNESS: Flush current registry state before clearing @@ -436,6 +452,18 @@ export class RegistryManager { } } + /** Skips a registry cycle beyond the per-run bound; logs only the first one. */ + private ignoreCollection(): void { + this.ignoringCollection = true; + this.isCollecting = false; + this.ignoredRegistryCycles++; + if (this.ignoredRegistryCycles === 1) { + this.logger.warn( + `Ignoring further I/O registry cycles of this run (bound: ${MAX_REGISTRY_COLLECTIONS_PER_RUN})`, + ); + } + } + /** * Detect and annotate conflicts on a single pin record in-place. * @@ -455,10 +483,8 @@ export class RegistryManager { */ addPin(pinRecord: IOPinRecord): void { if (this.destroyed) return; - if (!this.isCollecting) { - this.logger.warn("Received pin record while not collecting - ignoring"); - return; - } + // Sketch-controlled marker outside a collection: ignored without a log per line. + if (!this.isCollecting) return; // Individual pin additions are not logged (20 per start is too noisy). const existingIndex = this.registry.findIndex((p) => p.pin === pinRecord.pin); if (existingIndex >= 0) { @@ -501,6 +527,10 @@ export class RegistryManager { */ finishCollection(): void { if (this.destroyed) return; + if (this.ignoringCollection) { + this.ignoringCollection = false; + return; + } this.logger.debug( `Registry collection complete: ${this.registry.length} pins`, ); @@ -720,6 +750,9 @@ export class RegistryManager { * Reset the manager state (called when simulation stops) */ reset(): void { + this.collectionsThisRun = 0; + this.ignoringCollection = false; + this.ignoredRegistryCycles = 0; this.registry = []; this.preCollectionRegistry = []; this.hasStartedCollection = false; diff --git a/server/services/sandbox-runner.ts b/server/services/sandbox-runner.ts index e75b20e27..851b2066f 100644 --- a/server/services/sandbox-runner.ts +++ b/server/services/sandbox-runner.ts @@ -21,6 +21,7 @@ import { StreamHandler } from "./sandbox/stream-handler"; import { FilesystemHelper } from "./sandbox/filesystem-helper"; import { ExecutionManager, type ExecutionState, SimulationState, SANDBOX_CONFIG } from "./sandbox/execution-manager"; import { cleanupExecutionContainer, flushMessageQueue } from "./sandbox/execution-phases/cleanup-phase"; +import { delegateParsedLineToStreamHandler } from "./sandbox/execution-phases/stream-phase"; import { config } from "../config"; export class SandboxRunner { @@ -89,21 +90,11 @@ export class SandboxRunner { this.processController, stderrParser, this.timeoutManager, - (parsed, callbacks) => { - // Delegate parsed line to stream handler - if (this.executionState) { - const streamState = { - pinStateBatcher: this.executionState.pinStateBatcher, - serialOutputBatcher: this.executionState.serialOutputBatcher, - backpressurePaused: this.executionState.backpressurePaused, - isPaused: this.executionState.state === SimulationState.PAUSED, - baudrate: this.executionState.baudrate, - registryManager: this.registryManager, - }; - this.streamHandler.handleParsedLine(parsed, streamState, callbacks); - this.executionState.backpressurePaused = streamState.backpressurePaused; - } - }, + // Same routing (and run budget) as the local path. + (parsed, callbacks) => delegateParsedLineToStreamHandler(parsed, this.executionState, callbacks, { + registryManager: this.registryManager, + streamHandler: this.streamHandler, + }), () => { this.executionState.terminationRequested = true; }, ); this.streamHandler = new StreamHandler(this.processController); diff --git a/server/services/sandbox/execution-manager.ts b/server/services/sandbox/execution-manager.ts index c8f1f0412..8b9716b55 100644 --- a/server/services/sandbox/execution-manager.ts +++ b/server/services/sandbox/execution-manager.ts @@ -24,6 +24,7 @@ import { normalizeBaudrate, normalizeSimulationTimeout } from "@shared/input-lim import { canTransition } from "../simulation-state-machine"; import { createProcessExecutionPort, type ProcessExecution } from "../process-execution-port"; import { OutputCollector } from "../output-collector"; +import { RuntimeTextLimiter } from "./runtime-text-limiter"; import { flushMessageQueue, flushBatchers, cleanupExecutionContainer } from "./execution-phases/cleanup-phase"; import { scheduleExecutionTimeout } from "./execution-phases/timeout-phase"; import { createStreamCallbacks, delegateParsedLineToStreamHandler, handleStderrFallbackData } from "./execution-phases/stream-phase"; @@ -138,6 +139,8 @@ export interface ExecutionState { runGeneration?: number; /** Aborted when the current run is stopped or superseded. */ runAbort?: AbortController | null; + /** Budget of the current run for sketch-controlled runtime text lines. */ + runtimeTextLimiter?: RuntimeTextLimiter | null; } /** @@ -384,6 +387,7 @@ export class ExecutionManager { state.isSendingOutput = false; state.totalOutputBytes = 0; state.outputCollector = new OutputCollector(SANDBOX_CONFIG.maxOutputBytes); + state.runtimeTextLimiter = new RuntimeTextLimiter(); state.onOutputCallback = onOutput; state.ioRegistryCallback = onIORegistry; } diff --git a/server/services/sandbox/execution-phases/stream-phase.ts b/server/services/sandbox/execution-phases/stream-phase.ts index c017cbf3c..e5da00eaa 100644 --- a/server/services/sandbox/execution-phases/stream-phase.ts +++ b/server/services/sandbox/execution-phases/stream-phase.ts @@ -94,6 +94,7 @@ export function delegateParsedLineToStreamHandler( isPaused: state.state === "paused", baudrate: state.baudrate, registryManager: deps.registryManager, + runtimeTextLimiter: state.runtimeTextLimiter ?? undefined, }; deps.streamHandler.handleParsedLine(parsed, streamState, callbacks); diff --git a/server/services/sandbox/runtime-text-limiter.ts b/server/services/sandbox/runtime-text-limiter.ts new file mode 100644 index 000000000..d0f7c6234 --- /dev/null +++ b/server/services/sandbox/runtime-text-limiter.ts @@ -0,0 +1,87 @@ +import { Logger } from "@shared/logger"; + +/** Sustained runtime text lines per second that reach the client. */ +export const RUNTIME_TEXT_LINES_PER_SECOND = 50; +/** Lines a run may emit at once before the sustained rate applies. */ +export const RUNTIME_TEXT_LINE_BURST = 200; +/** A run that keeps dropping lines reports the count at most this often. */ +export const RUNTIME_TEXT_DROP_LOG_INTERVAL_MS = 10_000; +/** Sent to the client once per run, when the first line is dropped. */ +export const RUNTIME_TEXT_LIMIT_NOTICE = + `Runtime output rate limit exceeded (${RUNTIME_TEXT_LINES_PER_SECOND} lines/s); further lines are dropped`; + +export type RuntimeTextDecision = "forward" | "notice" | "drop"; + +interface RuntimeTextLimiterOptions { + linesPerSecond?: number; + burst?: number; + logIntervalMs?: number; + now?: () => number; + logger?: Pick; +} + +/** + * Budget of one run for runtime text lines: stdout/stderr output that is neither + * a simulator protocol marker nor serial output. Sketch code controls these + * lines and can write them far faster than any client reads them, so a token + * bucket admits a burst and a sustained rate; every further line is dropped and + * counted. The client is told once, the log only aggregated. + */ +export class RuntimeTextLimiter { + private readonly linesPerSecond: number; + private readonly burst: number; + private readonly logIntervalMs: number; + private readonly now: () => number; + private readonly logger: Pick; + private tokens: number; + private refilledAt: number; + private dropped = 0; + private droppedSinceLog = 0; + private loggedAt = 0; + + constructor(options: RuntimeTextLimiterOptions = {}) { + this.linesPerSecond = options.linesPerSecond ?? RUNTIME_TEXT_LINES_PER_SECOND; + this.burst = options.burst ?? RUNTIME_TEXT_LINE_BURST; + this.logIntervalMs = options.logIntervalMs ?? RUNTIME_TEXT_DROP_LOG_INTERVAL_MS; + this.now = options.now ?? Date.now; + this.logger = options.logger ?? new Logger("RuntimeTextLimiter"); + this.tokens = this.burst; + this.refilledAt = this.now(); + } + + /** Decides one line: forward it, replace it by the one-time notice, or drop it. */ + admit(): RuntimeTextDecision { + const now = this.now(); + this.refill(now); + if (this.tokens >= 1) { + this.tokens -= 1; + return "forward"; + } + this.dropped += 1; + this.droppedSinceLog += 1; + if (this.dropped === 1) { + this.logDrops(now, "Runtime output rate limit exceeded; dropping excess lines of this run"); + return "notice"; + } + if (now - this.loggedAt >= this.logIntervalMs) { + this.logDrops(now, "Runtime output still exceeds its rate limit"); + } + return "drop"; + } + + get droppedLines(): number { + return this.dropped; + } + + private refill(now: number): void { + const elapsedMs = Math.max(0, now - this.refilledAt); + this.refilledAt = now; + this.tokens = Math.min(this.burst, this.tokens + (elapsedMs * this.linesPerSecond) / 1_000); + } + + private logDrops(now: number, message: string): void { + this.logger.warn(`${message} (${this.droppedSinceLog} dropped since last report, ${this.dropped} total)`); + this.loggedAt = now; + this.droppedSinceLog = 0; + } +} diff --git a/server/services/sandbox/stream-handler.ts b/server/services/sandbox/stream-handler.ts index 8a4eef5eb..1ae017e47 100644 --- a/server/services/sandbox/stream-handler.ts +++ b/server/services/sandbox/stream-handler.ts @@ -9,6 +9,7 @@ import type { SerialOutputBatcher } from "../serial-output-batcher"; import type { RegistryManager } from "../registry-manager"; import type { ParsedStderrOutput } from "../arduino-output-parser"; import { Logger } from "@shared/logger"; +import { RUNTIME_TEXT_LIMIT_NOTICE, type RuntimeTextLimiter } from "./runtime-text-limiter"; import type { PinStateChange } from "@shared/types/arduino.types"; interface StreamHandlerCallbacks { @@ -24,6 +25,8 @@ interface StreamHandlerState { isPaused: boolean; baudrate: number; registryManager: RegistryManager; + /** Budget of the current run for runtime text lines; absent means unbounded. */ + runtimeTextLimiter?: Pick; } export class StreamHandler { @@ -119,11 +122,19 @@ export class StreamHandler { break; case "text": - if (callbacks.onError) { - this.logger.warn(`[STDERR] diagnostic received (${Buffer.byteLength(parsed.line)} bytes)`); - callbacks.onError(parsed.line); - } + this.handleRuntimeText(parsed.line, state, callbacks); break; } } + + /** + * Runtime text is sketch-controlled. The run's limiter bounds what reaches the + * client; nothing is logged per line (the limiter logs drops aggregated). + */ + private handleRuntimeText(line: string, state: StreamHandlerState, callbacks: StreamHandlerCallbacks): void { + if (!callbacks.onError) return; + const decision = state.runtimeTextLimiter?.admit() ?? "forward"; + if (decision === "forward") callbacks.onError(line); + else if (decision === "notice") callbacks.onError(RUNTIME_TEXT_LIMIT_NOTICE); + } } diff --git a/tests/integration/runtime-output-storm.test.ts b/tests/integration/runtime-output-storm.test.ts new file mode 100644 index 000000000..e433d9cca --- /dev/null +++ b/tests/integration/runtime-output-storm.test.ts @@ -0,0 +1,80 @@ +/** + * Integration Test: runtime text output storm (real Docker sandbox) + * + * A sketch that writes plain text to stderr in a tight loop must not make the + * backend forward or log every line: the client receives a bounded number of + * diagnostics and the log stays aggregated, independent of the line count. + */ + +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { SandboxRunner } from "../../server/services/sandbox-runner"; +import { Logger } from "../../shared/logger"; + +const skipHeavy = process.env.SKIP_HEAVY_TESTS !== "0" && process.env.SKIP_HEAVY_TESTS !== "false"; +const maybeDescribe = skipHeavy ? describe.skip : describe; + +const FLOOD_SECONDS = 2; +const FLOOD_SKETCH = String.raw` +#include +#include + +void setup() {} + +void loop() { + static unsigned long start = millis(); + if (millis() - start > ${FLOOD_SECONDS * 1000}) { + fputs("FLOOD_END\n", stderr); + exit(0); + } + for (int i = 0; i < 2000; i++) fputs("flood\n", stderr); +} +`.trim(); + +maybeDescribe("runtime output storm", () => { + let runner: SandboxRunner; + + beforeEach(() => { + runner = new SandboxRunner(); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + await runner.stop(); + }); + + test("forwards and logs a bounded number of lines while a sketch floods stderr", async () => { + const logCalls: string[] = []; + for (const level of ["warn", "info", "error"] as const) { + const original = Logger.prototype[level]; + vi.spyOn(Logger.prototype, level).mockImplementation(function (this: Logger, message: string) { + logCalls.push(message); + return original.call(this, message); + }); + } + const errors: string[] = []; + const exited = new Promise((resolve) => { + void runner.runSketch({ + code: FLOOD_SKETCH, + onOutput: () => {}, + onError: (line) => errors.push(line), + onExit: () => resolve(), + onCompileError: (error) => { + errors.push(`COMPILE: ${error}`); + resolve(); + }, + timeoutSec: 30, + }); + }); + await exited; + + expect(errors.some((line) => line.startsWith("COMPILE:"))).toBe(false); + const floodLines = errors.filter((line) => line === "flood").length; + // Without a budget every one of the >100 000 lines reached the client. + expect(floodLines).toBeGreaterThan(0); + expect(floodLines).toBeLessThan(1_000); + expect(errors.filter((line) => line.includes("Runtime output rate limit"))).toHaveLength(1); + // Without aggregation every line produced its own diagnostic log entry. + expect(logCalls.filter((message) => message.includes("STDERR")).length).toBe(0); + expect(logCalls.length).toBeLessThan(200); + }, 90_000); +}); diff --git a/tests/server/compose-log-rotation.test.ts b/tests/server/compose-log-rotation.test.ts new file mode 100644 index 000000000..7516cbecb --- /dev/null +++ b/tests/server/compose-log-rotation.test.ts @@ -0,0 +1,19 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { parse } from "yaml"; + +const repoRoot = join(__dirname, "../.."); + +describe("docker-compose backend logging", () => { + it("rotates the backend's json-file log", () => { + const compose = parse(readFileSync(join(repoRoot, "docker-compose.yml"), "utf8")) as { + services: Record } }>; + }; + const logging = compose.services["unosim-backend"].logging; + + expect(logging?.driver).toBe("json-file"); + expect(logging?.options?.["max-size"]).toBe("${UNOSIM_LOG_MAX_SIZE:-10m}"); + expect(logging?.options?.["max-file"]).toBe("${UNOSIM_LOG_MAX_FILE:-5}"); + }); +}); diff --git a/tests/server/routes/simulation-runtime-diagnostics.test.ts b/tests/server/routes/simulation-runtime-diagnostics.test.ts new file mode 100644 index 000000000..0dbb264b4 --- /dev/null +++ b/tests/server/routes/simulation-runtime-diagnostics.test.ts @@ -0,0 +1,66 @@ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { WebSocket } from "ws"; +import type { Logger } from "@shared/logger"; +import { registerSimulationWebSocket } from "../../../server/routes/simulation.ws"; +import type { SandboxRunner } from "../../../server/services/sandbox-runner"; +import type { SandboxRunnerPool } from "../../../server/services/sandbox-runner-pool"; +import type { RunSketchOptions } from "../../../server/services/run-sketch-types"; +import { SimulationAdmissionController } from "../../../server/services/simulation-admission-controller"; + +const SECRET = "a-secure-gateway-secret-with-32-characters"; +const ORIGIN = "https://classroom.example"; +let server: Server | undefined; +let socket: WebSocket | undefined; + +afterEach(async () => { + socket?.terminate(); + await new Promise((resolve) => (server ? server.close(() => resolve()) : resolve())); +}); + +describe("simulation runtime diagnostics", () => { + it("does not write a log entry per runtime diagnostic line", async () => { + const logged: string[] = []; + const record = (message: unknown) => logged.push(String(message)); + const logger = { debug: vi.fn(), info: vi.fn(record), warn: vi.fn(record), error: vi.fn(record) } as unknown as Logger; + let runOptions: RunSketchOptions | undefined; + const runSketch = vi.fn(async (options: RunSketchOptions) => { + runOptions = options; + return true; + }); + server = createServer(); + registerSimulationWebSocket(server, { + SandboxRunner: class {} as typeof SandboxRunner, + getSimulationRateLimiter: () => ({ checkLimit: () => ({ allowed: true }) }), + getSimulationAdmissionController: () => new SimulationAdmissionController(25), + shouldSendSimulationEndMessage: () => true, + getLastCompiledCode: () => null, + logger, + runnerPool: { + acquireRunner: vi.fn(async () => ({ runSketch, stop: vi.fn().mockResolvedValue(undefined) })), + releaseRunner: vi.fn().mockResolvedValue(undefined), + getRunnerIndex: vi.fn(() => 0), + getStats: vi.fn(() => ({ availableRunners: 1, totalRunners: 1, maxRunners: 1, queuedRequests: 0 })), + } as unknown as SandboxRunnerPool, + trust: { mode: "gateway", gatewaySecret: SECRET, trustedProxy: "127.0.0.1" }, + allowedWebSocketOrigins: [ORIGIN], + disableRateLimit: true, + }); + await new Promise((resolve) => server!.listen(0, "127.0.0.1", resolve)); + socket = new WebSocket(`ws://127.0.0.1:${(server.address() as AddressInfo).port}/ws`, { + headers: { Origin: ORIGIN, "X-UnoSim-Gateway-Secret": SECRET, "X-UnoSim-Subject": "student", "X-UnoSim-Roles": "user" }, + }); + await new Promise((resolve, reject) => { socket!.once("open", resolve); socket!.once("error", reject); }); + socket.send(JSON.stringify({ type: "start_simulation", code: "void setup(){} void loop(){}" })); + const deadline = Date.now() + 2_000; + while (!runOptions && Date.now() < deadline) await new Promise((r) => setTimeout(r, 5)); + expect(runOptions).toBeDefined(); + logged.length = 0; + + for (let index = 0; index < 10_000; index++) runOptions!.onError("diagnostic line"); + + // Without aggregation each diagnostic line produced its own warning. + expect(logged.length).toBeLessThan(10); + }); +}); diff --git a/tests/server/services/runtime-output-bounds.test.ts b/tests/server/services/runtime-output-bounds.test.ts new file mode 100644 index 000000000..6b45034ca --- /dev/null +++ b/tests/server/services/runtime-output-bounds.test.ts @@ -0,0 +1,138 @@ +// Bounds for untrusted runtime output: sketch code controls stdout/stderr and can +// emit plain text and spoofed protocol markers at full speed. + +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ArduinoOutputParser } from "../../../server/services/arduino-output-parser"; +import { RegistryManager } from "../../../server/services/registry-manager"; +import { delegateParsedLineToStreamHandler } from "../../../server/services/sandbox/execution-phases/stream-phase"; +import type { ExecutionState } from "../../../server/services/sandbox/execution-manager"; +import { + RUNTIME_TEXT_LIMIT_NOTICE, + RUNTIME_TEXT_LINE_BURST, + RUNTIME_TEXT_LINES_PER_SECOND, + RuntimeTextLimiter, +} from "../../../server/services/sandbox/runtime-text-limiter"; +import { StreamHandler } from "../../../server/services/sandbox/stream-handler"; +import { Logger } from "../../../shared/logger"; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +function spyOnLogs(): string[] { + const messages: string[] = []; + for (const level of ["debug", "info", "warn", "error"] as const) { + vi.spyOn(Logger.prototype, level).mockImplementation((message: string) => { + messages.push(message); + }); + } + return messages; +} + +describe("RegistryManager bounds for spoofed registry markers", () => { + it("sends a bounded number of registries when a sketch repeats registry cycles", () => { + const onUpdate = vi.fn(); + const manager = new RegistryManager({ onUpdate }); + const logs = spyOnLogs(); + + for (let cycle = 0; cycle < 1_000; cycle++) { + manager.startCollection(); + manager.addPin({ pin: String(cycle % 2), defined: true, pinMode: 1, usedAt: [] }); + manager.finishCollection(); + } + + // Without a bound every cycle with changed content sent and logged a registry. + expect(onUpdate.mock.calls.length).toBeGreaterThan(0); + expect(onUpdate.mock.calls.length).toBeLessThanOrEqual(4); + expect(logs.length).toBeLessThan(20); + manager.destroy(); + }); + + it("does not log every pin record received outside a collection", () => { + const manager = new RegistryManager({ onUpdate: vi.fn() }); + const logs = spyOnLogs(); + + for (let index = 0; index < 1_000; index++) { + manager.addPin({ pin: "13", defined: true, pinMode: 1, usedAt: [] }); + } + + expect(manager.getRegistry()).toHaveLength(0); + expect(logs.length).toBeLessThan(5); + manager.destroy(); + }); + + it("starts a fresh budget for the next run after reset", () => { + const onUpdate = vi.fn(); + const manager = new RegistryManager({ onUpdate }); + spyOnLogs(); + for (let cycle = 0; cycle < 50; cycle++) { + manager.startCollection(); + manager.addPin({ pin: String(cycle % 2), defined: true, pinMode: 1, usedAt: [] }); + manager.finishCollection(); + } + manager.reset(); + onUpdate.mockClear(); + + manager.startCollection(); + manager.addPin({ pin: "7", defined: true, pinMode: 1, usedAt: [] }); + manager.finishCollection(); + + expect(onUpdate).toHaveBeenCalledTimes(1); + manager.destroy(); + }); +}); + +describe("RuntimeTextLimiter", () => { + it("admits a burst and the sustained rate, then drops with one notice and aggregated logs", () => { + let now = 0; + const logger = { warn: vi.fn() }; + const limiter = new RuntimeTextLimiter({ now: () => now, logger }); + const decisions = Array.from({ length: 100_000 }, () => limiter.admit()); + + expect(decisions.filter((decision) => decision === "forward")).toHaveLength(RUNTIME_TEXT_LINE_BURST); + expect(decisions.filter((decision) => decision === "notice")).toHaveLength(1); + expect(limiter.droppedLines).toBe(100_000 - RUNTIME_TEXT_LINE_BURST); + expect(logger.warn).toHaveBeenCalledTimes(1); + + now += 1_000; + const nextSecond = Array.from({ length: 1_000 }, () => limiter.admit()); + expect(nextSecond.filter((decision) => decision === "forward")).toHaveLength(RUNTIME_TEXT_LINES_PER_SECOND); + expect(nextSecond).not.toContain("notice"); + expect(logger.warn).toHaveBeenCalledTimes(1); + + now += 10_000; + for (let index = 0; index < 1_000; index++) limiter.admit(); + expect(logger.warn).toHaveBeenCalledTimes(2); + expect(logger.warn.mock.calls[1][0]).toContain("total"); + }); +}); + +describe("runtime text storm through the shared stream path", () => { + it("forwards a bounded number of lines and logs nothing per line", () => { + const logs = spyOnLogs(); + const registryManager = new RegistryManager({ onUpdate: vi.fn() }); + const state = { + pinStateBatcher: null, + serialOutputBatcher: null, + backpressurePaused: false, + state: "running", + baudrate: 9600, + runtimeTextLimiter: new RuntimeTextLimiter({ now: () => 0 }), + } as unknown as ExecutionState; + const parser = new ArduinoOutputParser(); + const streamHandler = new StreamHandler({} as never); + const onError = vi.fn(); + + for (let index = 0; index < 100_000; index++) { + delegateParsedLineToStreamHandler(parser.parseStderrLine("flood", 0), state, { onOutput: vi.fn(), onError, onPinState: vi.fn() }, { + registryManager, + streamHandler, + }); + } + + expect(onError).toHaveBeenCalledTimes(RUNTIME_TEXT_LINE_BURST + 1); + expect(onError).toHaveBeenLastCalledWith(RUNTIME_TEXT_LIMIT_NOTICE); + expect(logs.length).toBeLessThan(5); + registryManager.destroy(); + }); +}); diff --git a/vitest.config.ts b/vitest.config.ts index 2137035db..101a902e3 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -125,6 +125,7 @@ export default defineConfig({ "tests/server/services/serial-backpressure.test.ts", "tests/integration/docker-security-contract.test.ts", "tests/integration/sandbox-orphan-sweep.test.ts", + "tests/integration/runtime-output-storm.test.ts", "tests/server/telemetry-heartbeat-integration.test.ts", "tests/core/sandbox-stress.test.ts", ...serializedHttpUnitTests, @@ -175,6 +176,7 @@ export default defineConfig({ "tests/server/services/serial-backpressure.test.ts", "tests/integration/docker-security-contract.test.ts", "tests/integration/sandbox-orphan-sweep.test.ts", + "tests/integration/runtime-output-storm.test.ts", ], environment: "node", setupFiles: ["./tests/setup.node.ts"],