fix: validate and re-cap all inbound worker-port traffic (Codex round 1)

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).
2026-07-08 11:42:59 +08:00
parent 466a052159
commit aa2a7f9a8a
6 changed files with 192 additions and 13 deletions
@@ -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`

@@ -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.

@@ -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
@@ -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<string>(['console', 'stdout', 'stderr'])
+const LOG_LEVELS = new Set<string>(['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<string, unknown>
+  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<string, unknown>
+      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<CodeLogEntry['level'], undefined> } : {},
+          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<string, unknown>).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<number>()
      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)
      })
@@ -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`
+}
@@ -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({