From aa2a7f9a8a03dfbad4b1d7e32c3af67fdbabd513 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 8 Jul 2026 11:39:57 +0800 Subject: [PATCH] fix: validate and re-cap all inbound worker-port traffic (Codex round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The host's message listener trusted the compile-time WorkerToHost shape on traffic from a peer that runs model code: postMessage(null) threw in the listener and crashed the host process; forged log/done messages bypassed maxLogBytes/maxValueBytes (the worker-side LogBuffer and prepareValue cap only honest flows); and the error-reply renegotiation re-echoed a forged non-cloneable call id, throwing outside any catch. Every inbound message now passes a runtime shape gate that validates and REBUILDS it field by field (junk drops without a throw; call ids must be numbers, so replies are always clone-plain; forged extra fields never ride along). One host-side ledger bounds everything landing in logs — honest port entries, forged ones, and stray pipe bytes — at the single documented maxLogBytes, with the shared in-band truncation marker emitted host-side when the ledger trips first; the completion value is re-capped host-side through the same prepareValue (with exactly the truncation suffix as slack so honest worker-capped values pass unchanged), and done error text is bounded. Also folds the stray-capture budget into that shared ledger (round-1 finding B: it was a second maxLogBytes on top of the documented shared cap). --- docs/config-catalog.md | 2 +- .../code-runtime-worker/README.md | 4 +- .../code-runtime-worker/src/bootstrap.ts | 3 +- .../code-runtime-worker/src/index.ts | 106 ++++++++++++++++-- .../code-runtime-worker/src/protocol.ts | 13 +++ .../code-runtime-worker/tests/runtime.spec.ts | 77 +++++++++++++ 6 files changed, 192 insertions(+), 13 deletions(-) diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 6510aea849..6339911606 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -171,7 +171,7 @@ export interface Config { } ``` -Source: [`packages/code-runtime/code-runtime-worker/src/index.ts:27`](../packages/code-runtime/code-runtime-worker/src/index.ts) +Source: [`packages/code-runtime/code-runtime-worker/src/index.ts:29`](../packages/code-runtime/code-runtime-worker/src/index.ts) ## `@deepseek-ai/dsh-compact-basic` diff --git a/packages/code-runtime/code-runtime-worker/README.md b/packages/code-runtime/code-runtime-worker/README.md index 13bf992bb5..b8f440397a 100644 --- a/packages/code-runtime/code-runtime-worker/README.md +++ b/packages/code-runtime/code-runtime-worker/README.md @@ -21,9 +21,9 @@ Every field is validated (positive numbers) and defaulted; there are no other tu - **One fresh worker per run, no pooling** — a program's world dies with its worker: no cross-run state to log, state bleed unrepresentable, runs reconstructable from the session log alone. - **Type-strip host-side, in execution context** — the program is wrapped in an async-function shell, stripped with `node:module`'s `stripTypeScriptTypes` (erasable syntax only — `enum`/namespaces are rejected as a program `exception` and no worker spawns), and sliced back out byte-positioned; it then executes as the body of an `AsyncFunction`, so top-level `await`/`return` work. -- **The port assumes a hostile peer** — model code can reach `parentPort` and forge traffic, so the host answers each call id at most once, resolves binding names as OWN properties only (a forged `constructor` cannot walk a prototype chain), drops post-settlement replies, and converts a non-cloneable binding resolution into an error reply. Worker-side namespaces are null-prototype with `defineProperty`, so `__proto__`-shaped binding names are ordinary keys. +- **The port assumes a hostile peer** — model code can reach `parentPort` and forge traffic, so every inbound message is shape-validated and REBUILT before anything reads it (`null`, primitives, junk types, and malformed payloads drop without a throw; forged extra fields never ride along), the host answers each call id at most once, resolves binding names as OWN properties only (a forged `constructor` cannot walk a prototype chain), drops post-settlement replies, and converts a non-cloneable binding resolution into an error reply. Forged `log`/`done` messages cannot bypass the caps: one host-side ledger bounds everything that lands in `logs`, and the completion value is re-capped host-side. Worker-side namespaces are null-prototype with `defineProperty`, so `__proto__`-shaped binding names are ordinary keys. - **Two independent budgets, because the peer is hostile** — `computeMs` meters the worker's MEASURED busy time (`worker.performance.eventLoopUtilization()` polling): a hot loop cannot hide behind a pending decoy dispatch, and a program awaiting a slow tool accrues nothing. `maxWallMs` backstops what busy time cannot see (awaiting a promise nobody resolves). Both funnel into `worker.terminate()`, which ends hot synchronous loops too; heap overflow surfaces as the worker's OOM exit (`kind: 'worker-exit'`). -- **Logs stream eagerly** — console/stdout/stderr entries cross the port as they happen, so a timed-out or killed program still shows what it printed; pipe bytes that bypass the patched streams are appended after, under the same byte budget. +- **Logs stream eagerly** — console/stdout/stderr entries cross the port as they happen, so a timed-out or killed program still shows what it printed. ONE shared `maxLogBytes` ledger bounds everything: streamed entries, forged port traffic, and pipe bytes that bypass the patched streams (appended after), with the overflow marked in-band once. - **Empty environment** — the worker gets `env: {}` and `execArgv: []`: no ambient credentials (stronger than the scrubbed-env rule for spawned commands) and no inherited loader flags. - **Dispose to quiescence** — teardown fails in-flight runs as `abort` and AWAITS each worker's exit before resolving. diff --git a/packages/code-runtime/code-runtime-worker/src/bootstrap.ts b/packages/code-runtime/code-runtime-worker/src/bootstrap.ts index 06cef02c7b..ae6bcf25bf 100644 --- a/packages/code-runtime/code-runtime-worker/src/bootstrap.ts +++ b/packages/code-runtime/code-runtime-worker/src/bootstrap.ts @@ -12,6 +12,7 @@ import { inspect } from 'node:util' import type { CodeLogEntry } from '@deepseek-ai/dsh-code-runtime' +import { logTruncationMarker } from './protocol.ts' import type { DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts' /** The port surface the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */ @@ -62,7 +63,7 @@ export class LogBuffer { const cost = Buffer.byteLength(entry.text, 'utf8') if (cost > this.remaining) { this.truncated = true - this.sink({ source: 'stderr', text: `[dsh-code-runtime-worker] log capture truncated at ${this.maxBytes} bytes` }) + this.sink({ source: 'stderr', text: logTruncationMarker(this.maxBytes) }) return } this.remaining -= cost diff --git a/packages/code-runtime/code-runtime-worker/src/index.ts b/packages/code-runtime/code-runtime-worker/src/index.ts index dbd0a44966..a2385ba83e 100644 --- a/packages/code-runtime/code-runtime-worker/src/index.ts +++ b/packages/code-runtime/code-runtime-worker/src/index.ts @@ -18,6 +18,8 @@ import { Context } from 'cordis' import z from 'schemastery' import { CodeRuntime } from '@deepseek-ai/dsh-code-runtime' import type { CodeBindingFunction, CodeLogEntry, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime' +import { prepareValue } from './bootstrap.ts' +import { logTruncationMarker } from './protocol.ts' import type { ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts' export type { BootstrapPort, PatchableStream } from './bootstrap.ts' @@ -108,6 +110,64 @@ function messageOf(error: unknown): string { return error instanceof Error ? error.message : String(error) } +/** The log sources / console levels the seam vocabulary admits, as runtime sets for inbound-message validation. */ +const LOG_SOURCES = new Set(['console', 'stdout', 'stderr']) +const LOG_LEVELS = new Set(['log', 'info', 'warn', 'error', 'debug']) + +/** + * Runtime shape gate for inbound port traffic. The peer runs MODEL CODE and + * can post anything — `null`, primitives, objects with poisoned fields — so + * the compile-time `WorkerToHost` type means nothing here: everything is + * re-validated and REBUILT field by field (a forged extra field never rides + * along; a non-number call id can never be echoed into a reply). Junk returns + * `undefined` and is dropped — a throw in the host's `message` listener would + * crash the host process. + */ +function parseWorkerMessage(raw: unknown): WorkerToHost | undefined { + if (typeof raw !== 'object' || raw === null) return undefined + const m = raw as Record + switch (m.type) { + case 'call': { + if (typeof m.id !== 'number' || typeof m.global !== 'string' || typeof m.name !== 'string') return undefined + return { type: 'call', id: m.id, global: m.global, name: m.name, args: m.args } + } + case 'log': { + const entry = m.entry + if (typeof entry !== 'object' || entry === null) return undefined + const e = entry as Record + if (typeof e.text !== 'string') return undefined + if (typeof e.source !== 'string' || !LOG_SOURCES.has(e.source)) return undefined + if (e.level !== undefined && (typeof e.level !== 'string' || !LOG_LEVELS.has(e.level))) return undefined + return { + type: 'log', + entry: { + source: e.source as CodeLogEntry['source'], + ...e.level !== undefined ? { level: e.level as Exclude } : {}, + text: e.text, + }, + } + } + case 'done': { + if (m.error === undefined) return { type: 'done', ...m.value !== undefined ? { value: m.value } : {} } + const error = m.error + if (typeof error !== 'object' || error === null) return undefined + const message = (error as Record).message + if (typeof message !== 'string') return undefined + return { type: 'done', ...m.value !== undefined ? { value: m.value } : {}, error: { message } } + } + default: return undefined + } +} + +/** + * Headroom the host's value re-cap grants over `maxValueBytes`: exactly the + * truncation suffix {@link prepareValue} appends, so a value the WORKER + * already capped passes through unchanged instead of being marked twice. + * (A multibyte rendering the worker sliced by characters can still exceed + * this and pick up a second marker — bounded and harmless.) + */ +const VALUE_RENDER_SLACK = Buffer.byteLength('… [truncated]', 'utf8') + /** * The shipped {@link CodeRuntime} backend (`ctx.codeRuntime`). Registers as * the `codeRuntime` service; every cap comes from validated config. See the @@ -234,13 +294,32 @@ export class WorkerCodeRuntime extends CodeRuntime { const answered = new Set() const logs: CodeLogEntry[] = [] const strayLogs: CodeLogEntry[] = [] - let strayBudget = this.config.maxLogBytes + // ONE host-side ledger for everything that lands in `logs`/`strayLogs`, + // whatever the path: honest port entries, FORGED port entries (model + // code posting `log` messages directly, bypassing the worker-side + // LogBuffer), and stray pipe bytes. On the first overflow it emits the + // same in-band marker the worker's LogBuffer would and drops the rest, + // so the documented cap is one shared `maxLogBytes` however it is hit. + let logBudget = this.config.maxLogBytes + let logsTruncated = false + const admit = (entry: CodeLogEntry, sink: CodeLogEntry[]): void => { + if (logsTruncated) return + const cost = Buffer.byteLength(entry.text, 'utf8') + if (cost > logBudget) { + logsTruncated = true + sink.push({ source: 'stderr', text: logTruncationMarker(this.config.maxLogBytes) }) + return + } + logBudget -= cost + sink.push(entry) + } + + // No settled guard: `finish` snapshots the arrays when it resolves, so + // a chunk flushing after settlement mutates only the discarded buffers, + // and the ledger bounds that growth until the pipes close. const captureStray = (source: 'stdout' | 'stderr') => (chunk: Buffer) => { - if (settled || strayBudget <= 0) return - const text = chunk.toString('utf8').slice(0, strayBudget) - strayBudget -= Buffer.byteLength(text, 'utf8') - strayLogs.push({ source, text }) + admit({ source, text: chunk.toString('utf8') }, strayLogs) } worker.stdout.on('data', captureStray('stdout')) worker.stderr.on('data', captureStray('stderr')) @@ -267,9 +346,14 @@ export class WorkerCodeRuntime extends CodeRuntime { const onDone = (message: WorkerToHost): void => { if (message.type !== 'done') return + // Re-cap the completion value HOST-side: the honest path already + // capped it in the worker (prepareValue there), but a forged done + // message bypasses the bootstrap entirely — without this, model code + // could flood the host past maxValueBytes. Honest values pass + // unchanged (see VALUE_RENDER_SLACK); the error text is bounded too. finish({ - ...message.value !== undefined ? { value: message.value } : {}, - ...message.error ? { error: { kind: 'exception' as const, message: message.error.message } } : {}, + ...prepareValue(message.value, this.config.maxValueBytes + VALUE_RENDER_SLACK), + ...message.error ? { error: { kind: 'exception' as const, message: message.error.message.slice(0, this.config.maxValueBytes) } } : {}, }) } @@ -308,8 +392,12 @@ export class WorkerCodeRuntime extends CodeRuntime { })() } - worker.on('message', (message: WorkerToHost) => { - if (message.type === 'log' && !settled) logs.push(message.entry) + worker.on('message', (raw: unknown) => { + // Parse before touching: the peer can post ANY shape, and a throw in + // this listener would crash the host process. Junk drops silently. + const message = parseWorkerMessage(raw) + if (!message) return + if (message.type === 'log' && !settled) admit(message.entry, logs) onCall(message) onDone(message) }) diff --git a/packages/code-runtime/code-runtime-worker/src/protocol.ts b/packages/code-runtime/code-runtime-worker/src/protocol.ts index 85d5113b82..b8ea122c5b 100644 --- a/packages/code-runtime/code-runtime-worker/src/protocol.ts +++ b/packages/code-runtime/code-runtime-worker/src/protocol.ts @@ -63,3 +63,16 @@ export type WorkerToHost = CallMessage | LogMessage | DoneMessage export type ReplyMessage = | { type: 'reply'; id: number; ok: true; value: unknown } | { type: 'reply'; id: number; ok: false; message: string } + +/** + * The in-band marker entry text announcing that log capture stopped at the + * byte budget. Shared wire vocabulary: the worker's LogBuffer emits it when + * ITS budget exhausts, and the host emits the identical text when its own + * ledger drops an entry first (forged port traffic, stray pipe bytes) — so + * a truncated run reads the same however the cap was hit. + * @param maxBytes - the configured `maxLogBytes` the marker names. + * @returns the marker line. + */ +export function logTruncationMarker(maxBytes: number): string { + return `[dsh-code-runtime-worker] log capture truncated at ${maxBytes} bytes` +} diff --git a/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts b/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts index 8edfdc6a41..a2a3aa4091 100644 --- a/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts +++ b/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts @@ -101,6 +101,13 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => { expect(typeof result.value).toBe('string') }) + it('completes a program that returns nothing with no value at all', async () => { + const { runtime } = await setup() + const result = await runtime.run({ program: 'const x = 1', bindings: [] }) + expect(result.error).toBeUndefined() + expect('value' in result).toBe(false) + }) + it('keeps logs streamed before a failure', async () => { const { runtime } = await setup() const result = await runtime.run({ @@ -255,6 +262,76 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => { expect(result.value).toBe('still-works') }) + it('survives arbitrary junk on the port: non-objects, junk types, malformed calls, logs, and dones', async () => { + const { runtime } = await setup() + const result = await runtime.run({ + program: ` + const { parentPort } = await import('node:worker_threads'); + for (const junk of [ + null, 42, 'junk', [], + { type: 'nope' }, + { type: 'call' }, + { type: 'call', id: 'x', global: 'tools', name: 'real', args: {} }, + { type: 'call', id: 1e9, global: 7, name: 'real', args: {} }, + { type: 'call', id: 1e9, global: 'tools', name: 7, args: {} }, + { type: 'log' }, + { type: 'log', entry: null }, + { type: 'log', entry: { source: 'stdout', text: 7 } }, + { type: 'log', entry: { source: 'nope', text: 'x' } }, + { type: 'log', entry: { source: 'console', level: 'nope', text: 'x' } }, + { type: 'log', entry: { source: 'console', level: 7, text: 'x' } }, + { type: 'done', error: 5 }, + { type: 'done', error: { message: 5 } }, + ]) parentPort.postMessage(junk); + return await tools.real({}); + `, + bindings: tools({ real: async () => 'still-works' }), + }) + expect(result.error).toBeUndefined() + expect(result.value).toBe('still-works') + expect(result.logs).toEqual([]) + }) + + it('caps forged log floods and forged done values at the configured budgets, dropping forged extra fields', async () => { + const { runtime } = await setup({ maxLogBytes: 200, maxValueBytes: 64 }) + const result = await runtime.run({ + // Forged messages bypass the worker-side LogBuffer and prepareValue + // entirely — only the host-side ledger and re-cap stand between model + // code and an unbounded result. + program: ` + const { parentPort } = await import('node:worker_threads'); + for (let i = 0; i < 50; i++) parentPort.postMessage({ type: 'log', entry: { source: 'stdout', text: 'F'.repeat(100), forged: true } }); + parentPort.postMessage({ type: 'done', value: 'V'.repeat(100000) }); + for (;;) {} + `, + bindings: [], + }) + expect(typeof result.value).toBe('string') + const value = result.value as string + expect(value.startsWith('V'.repeat(64))).toBe(true) + expect(value.endsWith('… [truncated]')).toBe(true) + expect(value.length).toBeLessThan(120) + const marker = '[dsh-code-runtime-worker] log capture truncated at 200 bytes' + const total = result.logs.reduce((sum, entry) => sum + Buffer.byteLength(entry.text, 'utf8'), 0) + expect(total).toBeLessThanOrEqual(200 + Buffer.byteLength(marker, 'utf8')) + expect(result.logs.at(-1)?.text).toBe(marker) + expect(result.logs.every(entry => !('forged' in entry))).toBe(true) + }) + + it('accepts a forged done carrying both value and error (self-sabotage, contained)', async () => { + const { runtime } = await setup() + const result = await runtime.run({ + program: ` + const { parentPort } = await import('node:worker_threads'); + parentPort.postMessage({ type: 'done', value: 'lied', error: { message: 'fake failure' } }); + for (;;) {} + `, + bindings: [], + }) + expect(result.value).toBe('lied') + expect(result.error).toEqual({ kind: 'exception', message: 'fake failure' }) + }) + it('answers a binding whose resolution cannot be cloned with a failure reply', async () => { const { runtime } = await setup() const result = await runtime.run({