fix(daemon): bound NDJSON and binary-frame ingress

The daemon's two wire parsers each accept peer-controlled input without a
ceiling on what a single message can turn into.

Binary frames: the parser reads a 32-bit declared payload length and waits
for that many bytes. A peer that declares 4 GiB makes the parser buffer every
byte it sends, unbounded, with no frame ever completing. FRAME_MAX_PAYLOAD
existed but was only enforced on the send path. Reject from the header
instead: a declared length over the cap is unsatisfiable, so there is nothing
to wait for.

NDJSON: a line under the 16 MiB cap can still expand into an object graph
far larger than its wire size. 4 MiB of nested arrays parses into ~109 MiB of
heap — 27x amplification, and the cap admits that per line, repeatedly.
Scan the line's structure before JSON.parse and reject amplified input on
token count and nesting depth, so the ceiling covers what the message costs
in memory rather than only what it costs on the wire.

The parser also accumulates into a byte buffer rather than a JS string, which
lets it report the admitted byte count without re-serializing the parsed value.

Verified: 1,123 daemon tests pass; 4,782 across shared + daemon + cli, three
consecutive runs. Realistic daemon PTY traffic (20k messages, 40 MiB, 64 KiB
socket reads) parses at 795 MiB/s, well above real load.
This commit is contained in:
Neil
2026-07-26 19:37:45 -07:00
parent 06b298f67e
commit 2d5fd8def8
4 changed files with 107 additions and 13 deletions
+17
View File
@@ -1,5 +1,6 @@
import { describe, expect, it, vi } from 'vitest'
import { encodeFrame, createFrameParser, FrameType, FRAME_HEADER_SIZE } from './binary-frame'
import { FRAME_MAX_PAYLOAD } from './types'
describe('encodeFrame', () => {
it('encodes a data frame with correct header', () => {
@@ -130,6 +131,22 @@ describe('createFrameParser', () => {
expect(onFrame.mock.calls[0][1].length).toBe(0)
})
it('rejects an oversized declared payload from the header alone', () => {
const onFrame = vi.fn()
const parser = createFrameParser(onFrame)
const header = Buffer.alloc(FRAME_HEADER_SIZE)
header[0] = FrameType.Data
header.writeUInt32BE(FRAME_MAX_PAYLOAD + 1, 1)
expect(() => parser.feed(header)).toThrow(
`Frame payload ${FRAME_MAX_PAYLOAD + 1} exceeds max ${FRAME_MAX_PAYLOAD}`
)
expect(onFrame).not.toHaveBeenCalled()
parser.feed(encodeFrame(FrameType.Data, Buffer.from('fresh')))
expect(onFrame).toHaveBeenCalledOnce()
})
it('parses different frame types correctly', () => {
const onFrame = vi.fn()
const parser = createFrameParser(onFrame)
+4
View File
@@ -29,6 +29,10 @@ export function createFrameParser(
function parse(): void {
while (buffer.length >= FRAME_HEADER_SIZE) {
const payloadLength = buffer.readUInt32BE(1)
if (payloadLength > FRAME_MAX_PAYLOAD) {
buffer = Buffer.alloc(0)
throw new Error(`Frame payload ${payloadLength} exceeds max ${FRAME_MAX_PAYLOAD}`)
}
const totalLength = FRAME_HEADER_SIZE + payloadLength
if (buffer.length < totalLength) {
+54 -1
View File
@@ -1,7 +1,20 @@
import { describe, expect, it, vi } from 'vitest'
import { encodeNdjson, createNdjsonParser, NDJSON_MAX_LINE_BYTES } from './ndjson'
import {
encodeBoundedNdjson,
encodeNdjson,
createNdjsonParser,
NDJSON_MAX_LINE_BYTES,
NDJSON_MAX_STRUCTURAL_TOKENS
} from './ndjson'
describe('encodeNdjson', () => {
it('bounds response serialization while preserving admitted wire bytes', () => {
expect(encodeBoundedNdjson({ ok: true }, 12)).toBe('{"ok":true}\n')
expect(() => encodeBoundedNdjson({ value: 'x'.repeat(100) }, 32)).toThrow(
'JSON output exceeds 31 bytes'
)
})
it('encodes an object as a JSON line ending with newline', () => {
const result = encodeNdjson({ type: 'hello', version: 1 })
expect(result).toBe('{"type":"hello","version":1}\n')
@@ -32,6 +45,32 @@ describe('createNdjsonParser', () => {
expect(onError).not.toHaveBeenCalled()
})
it('optionally reports each parsed line byte length without re-serializing it', () => {
const onMessage = vi.fn()
const parser = createNdjsonParser(onMessage, undefined, { includeLineBytes: true })
parser.feed('{"text":"é"}\n')
expect(onMessage).toHaveBeenCalledWith({ text: 'é' }, Buffer.byteLength('{"text":"é"}'))
})
it('rejects structurally amplified lines before parsing', () => {
const onMessage = vi.fn()
const onError = vi.fn()
const parser = createNdjsonParser(onMessage, onError)
const parseSpy = vi.spyOn(JSON, 'parse')
try {
parser.feed(`{"values":[${'0,'.repeat(NDJSON_MAX_STRUCTURAL_TOKENS)}0]}\n`)
expect(onMessage).not.toHaveBeenCalled()
expect(onError).toHaveBeenCalledWith(
expect.objectContaining({ message: expect.stringContaining('JSON structure exceeds') })
)
expect(parseSpy).not.toHaveBeenCalled()
} finally {
parseSpy.mockRestore()
}
})
it('parses multiple messages in a single chunk', () => {
const onMessage = vi.fn()
const parser = createNdjsonParser(onMessage)
@@ -56,6 +95,20 @@ describe('createNdjsonParser', () => {
expect(onMessage).toHaveBeenCalledWith({ type: 'hello', version: 1 })
})
it('parses a line delivered in more than 100,000 one-character fragments', () => {
const onMessage = vi.fn()
const parser = createNdjsonParser(onMessage, undefined, { includeLineBytes: true })
const message = { value: 'x'.repeat(100_000) }
const line = JSON.stringify(message)
for (const character of line) {
parser.feed(character)
}
parser.feed('\n')
expect(onMessage).toHaveBeenCalledWith(message, Buffer.byteLength(line))
})
it('handles a chunk that ends mid-line followed by more data', () => {
const onMessage = vi.fn()
const parser = createNdjsonParser(onMessage)
+32 -12
View File
@@ -1,8 +1,21 @@
import { assertJsonTextStructureWithinLimits } from '../../shared/memory-safety/json-text-structure-limit'
import { GrowingByteBuffer } from '../../shared/growing-byte-buffer'
import { stringifyJsonWithinByteLimit } from '../../shared/memory-safety/node-bounded-json-stringify'
export function encodeNdjson(msg: unknown): string {
return `${JSON.stringify(msg)}\n`
}
export function encodeBoundedNdjson(msg: unknown, maxBytes: number): string {
if (!Number.isSafeInteger(maxBytes) || maxBytes < 1) {
throw new RangeError('NDJSON byte limit must be a positive safe integer')
}
return `${stringifyJsonWithinByteLimit(msg, maxBytes - 1).serialized}\n`
}
export const NDJSON_MAX_LINE_BYTES = 16 * 1024 * 1024
export const NDJSON_MAX_STRUCTURAL_TOKENS = 1_000_000
export const NDJSON_MAX_NESTING_DEPTH = 128
export type NdjsonParser = {
feed(chunk: string): void
@@ -11,21 +24,20 @@ export type NdjsonParser = {
export type NdjsonParserOptions = {
maxLineBytes?: number
includeLineBytes?: boolean
}
export function createNdjsonParser(
onMessage: (msg: unknown) => void,
onMessage: (msg: unknown, lineBytes?: number) => void,
onError?: (err: Error) => void,
options: NdjsonParserOptions = {}
): NdjsonParser {
let buffer = ''
let bufferBytes = 0
const buffer = new GrowingByteBuffer()
let discardingOversizedLine = false
const maxLineBytes = Math.max(1, options.maxLineBytes ?? NDJSON_MAX_LINE_BYTES)
const clearBuffer = (): void => {
buffer = ''
bufferBytes = 0
buffer.clear()
}
const reportOversizedLine = (observedBytes: number): void => {
@@ -53,8 +65,8 @@ export function createNdjsonParser(
return
}
const segmentBytes = Buffer.byteLength(segment, 'utf8')
const nextLineBytes = bufferBytes + segmentBytes
const segmentBytes = Buffer.from(segment, 'utf8')
const nextLineBytes = buffer.byteLength + segmentBytes.byteLength
// Why: daemon sockets are local but persistent; a peer that never sends
// a newline must not grow the parser buffer without bound.
if (nextLineBytes > maxLineBytes) {
@@ -67,21 +79,29 @@ export function createNdjsonParser(
continue
}
buffer += segment
bufferBytes = nextLineBytes
buffer.append(segmentBytes)
if (!hasNewline) {
return
}
const line = buffer
clearBuffer()
const lineBytes = buffer.byteLength
const line = buffer.takeString('utf8')
if (line.length === 0) {
continue
}
try {
onMessage(JSON.parse(line))
assertJsonTextStructureWithinLimits(line, {
structuralTokens: NDJSON_MAX_STRUCTURAL_TOKENS,
nestingDepth: NDJSON_MAX_NESTING_DEPTH
})
const parsed = JSON.parse(line)
if (options.includeLineBytes) {
onMessage(parsed, lineBytes)
} else {
onMessage(parsed)
}
} catch (err) {
onError?.(err instanceof Error ? err : new Error(String(err)))
}