diff --git a/packages/onebot/src/config.ts b/packages/onebot/src/config.ts index 5f7c5c58..9864cd8b 100644 --- a/packages/onebot/src/config.ts +++ b/packages/onebot/src/config.ts @@ -462,18 +462,8 @@ function validateRestoreAdapterArray( invalid(`${pathAt}.reportSelfMessage must be a boolean`); } - if (kind !== 'httpClients' && raw.maxPayloadBytes !== undefined) { - const maxPayload = parseRestoreInteger(raw.maxPayloadBytes); - - if ( - maxPayload === null || - maxPayload < MAX_PAYLOAD_BYTES_MIN || - maxPayload > MAX_PAYLOAD_BYTES_MAX - ) { - invalid( - `${pathAt}.maxPayloadBytes must be an integer between ${MAX_PAYLOAD_BYTES_MIN} and ${MAX_PAYLOAD_BYTES_MAX}`, - ); - } + if (kind !== 'httpClients') { + parseMaxPayloadBytes(raw.maxPayloadBytes, pathAt); } if (kind === 'httpServers' || kind === 'wsServers') { @@ -679,7 +669,7 @@ export function assertValidOneBotConfig(value: unknown): asserts value is OneBot const serverBindings = new Map(); validateNetworkList(value.networks, 'httpServers', seen, (item, at) => { validateServer(item, at); - validateMaxPayloadBytes(item, at); + parseMaxPayloadBytes(item.maxPayloadBytes, at); if (item.enableWebSocket !== undefined && typeof item.enableWebSocket !== 'boolean') { invalid(`${at}.enableWebSocket must be a boolean`); @@ -690,10 +680,7 @@ export function assertValidOneBotConfig(value: unknown): asserts value is OneBot validateNetworkList(value.networks, 'httpClients', seen, (item, at) => { validateClient(item, at, new Set(['http:', 'https:'])); - // A frame cap on the pure HTTP POST adapter would be saved and ignored. - if (item.maxPayloadBytes !== undefined) { - invalid(`${at}.maxPayloadBytes is only supported by WebSocket adapters`); - } + rejectHttpClientMaxPayloadBytes(item, at); if ( item.timeoutMs !== undefined && @@ -709,12 +696,12 @@ export function assertValidOneBotConfig(value: unknown): asserts value is OneBot validateServer(item, at); validateServerBinding(item, at, serverBindings); validateRole(item.role, `${at}.role`); - validateMaxPayloadBytes(item, at); + parseMaxPayloadBytes(item.maxPayloadBytes, at); }); validateNetworkList(value.networks, 'wsClients', seen, (item, at) => { validateClient(item, at, new Set(['ws:', 'wss:'])); validateRole(item.role, `${at}.role`); - validateMaxPayloadBytes(item, at); + parseMaxPayloadBytes(item.maxPayloadBytes, at); if ( item.reconnectIntervalMs !== undefined && @@ -847,22 +834,31 @@ function validateNetworkBase(item: JsonObject, at: string): void { } } -/** Strict gate for the WebSocket-only frame cap; `httpClients` must not carry it. */ -function validateMaxPayloadBytes(item: JsonObject, at: string): void { - if (item.maxPayloadBytes === undefined) { - return; +/** Apply the same limit validation to files, restores and API writes. */ +function parseMaxPayloadBytes(value: unknown, at: string): number | undefined { + if (value === undefined) { + return undefined; } if ( - typeof item.maxPayloadBytes !== 'number' || - !Number.isSafeInteger(item.maxPayloadBytes) || - item.maxPayloadBytes < MAX_PAYLOAD_BYTES_MIN || - item.maxPayloadBytes > MAX_PAYLOAD_BYTES_MAX + typeof value !== 'number' || + !Number.isSafeInteger(value) || + value < MAX_PAYLOAD_BYTES_MIN || + value > MAX_PAYLOAD_BYTES_MAX ) { invalid( `${at}.maxPayloadBytes must be an integer between ${MAX_PAYLOAD_BYTES_MIN} and ${MAX_PAYLOAD_BYTES_MAX}`, ); } + + return value; +} + +function rejectHttpClientMaxPayloadBytes(item: JsonObject, at: string): void { + // A frame cap on the pure HTTP POST adapter would be saved and ignored. + if (item.maxPayloadBytes !== undefined) { + invalid(`${at}.maxPayloadBytes is only supported by WebSocket adapters`); + } } function validateServer(item: JsonObject, at: string): void { @@ -1415,23 +1411,9 @@ function parseBase(value: JsonObject, defaults: AdapterDefaults) { }; } -/** - * Compatibility parser for the optional frame cap. Clamps like its neighbours - * (`timeoutMs`, `reconnectIntervalMs`) so an on-disk value written by an - * older/newer build still loads; `assertValidOneBotConfig` is the strict gate - * for values saved through the API/WebUI. - */ -function parseMaxPayloadBytes(value: unknown): number | undefined { - const parsed = asNumber(value, 0); - - if (parsed <= 0) { - return undefined; - } - - return Math.min(Math.max(parsed, MAX_PAYLOAD_BYTES_MIN), MAX_PAYLOAD_BYTES_MAX); -} - function parseHttpServer(value: JsonObject, defaults: AdapterDefaults): HttpServerNetwork | null { + const maxPayloadBytes = parseMaxPayloadBytes(value.maxPayloadBytes, 'http server'); + if (value.enableWebSocket !== undefined && typeof value.enableWebSocket !== 'boolean') { invalid('http server enableWebSocket must be a boolean'); } @@ -1448,11 +1430,12 @@ function parseHttpServer(value: JsonObject, defaults: AdapterDefaults): HttpServ port, path: asString(value.path, '/'), enableWebSocket: value.enableWebSocket === true, - maxPayloadBytes: parseMaxPayloadBytes(value.maxPayloadBytes), + maxPayloadBytes, }); } function parseHttpClient(value: JsonObject, defaults: AdapterDefaults): HttpClientNetwork | null { + rejectHttpClientMaxPayloadBytes(value, 'http client'); const url = asString(value.url); const base = parseBase(value, defaults); @@ -1469,6 +1452,7 @@ function parseHttpClient(value: JsonObject, defaults: AdapterDefaults): HttpClie } function parseWsServer(value: JsonObject, defaults: AdapterDefaults): WsServerNetwork | null { + const maxPayloadBytes = parseMaxPayloadBytes(value.maxPayloadBytes, 'websocket server'); const port = asNumber(value.port, 0); if (port <= 0) { @@ -1481,11 +1465,12 @@ function parseWsServer(value: JsonObject, defaults: AdapterDefaults): WsServerNe port, path: asString(value.path, '/'), role: asRole(value.role, 'Universal'), - maxPayloadBytes: parseMaxPayloadBytes(value.maxPayloadBytes), + maxPayloadBytes, }); } function parseWsClient(value: JsonObject, defaults: AdapterDefaults): WsClientNetwork | null { + const maxPayloadBytes = parseMaxPayloadBytes(value.maxPayloadBytes, 'websocket client'); const url = asString(value.url); const base = parseBase(value, defaults); @@ -1499,7 +1484,7 @@ function parseWsClient(value: JsonObject, defaults: AdapterDefaults): WsClientNe url, role: asRole(value.role, 'Universal'), reconnectIntervalMs: Math.max(1000, reconnectIntervalMs), - maxPayloadBytes: parseMaxPayloadBytes(value.maxPayloadBytes), + maxPayloadBytes, }); } diff --git a/packages/onebot/tests/config.test.ts b/packages/onebot/tests/config.test.ts index fda68176..916df1d0 100644 --- a/packages/onebot/tests/config.test.ts +++ b/packages/onebot/tests/config.test.ts @@ -437,11 +437,43 @@ describe('loadOneBotConfig', () => { const junk = structuredClone(onDisk) as { networks: { wsClients: Array> }; }; - junk.networks.wsClients[0]!.maxPayloadBytes = 0; - expect(() => prepareOneBotConfigForRestore(junk, 'per-uin')).toThrow(/maxPayloadBytes/); + + for (const value of [0, '268435456', 1024.5, null]) { + junk.networks.wsClients[0]!.maxPayloadBytes = value; + expect(() => prepareOneBotConfigForRestore(junk, 'per-uin')).toThrow(/maxPayloadBytes/); + } }); - it('persists maxPayloadBytes for httpServers and drops it for httpClients', () => { + it.each(['httpServers', 'wsServers', 'wsClients'] as const)( + 'rejects invalid on-disk maxPayloadBytes for %s without rewriting the file', + (kind) => { + const dir = path.join(tempDir, 'config'); + fs.mkdirSync(dir, { recursive: true }); + const file = path.join(dir, 'onebot_10023.json'); + + for (const value of [-1, 0, 1023, 1024.5, 2 * 1024 ** 3 + 1, '2048', null, true]) { + const raw = JSON.stringify({ + mode: 'snapshot', + networks: { + [kind]: [ + { + name: 'invalid-limit', + port: 3001, + url: 'ws://127.0.0.1:3001', + enableWebSocket: true, + maxPayloadBytes: value, + }, + ], + }, + }); + fs.writeFileSync(file, raw, 'utf8'); + expect(() => loadOneBotConfig('10023'), String(value)).toThrow(/maxPayloadBytes/); + expect(fs.readFileSync(file, 'utf8')).toBe(raw); + } + }, + ); + + it('persists maxPayloadBytes for httpServers and rejects it for httpClients', () => { const uin = '10021'; const config = makeDefaultOneBotConfig(); config.networks.httpServers[0].enableWebSocket = true; @@ -469,8 +501,8 @@ describe('loadOneBotConfig', () => { /maxPayloadBytes is not supported/, ); - // A hand-written file is parsed permissively, but the key is not carried - // into the canonical config (nor written back). + // A hand-written file must reject the same unsupported field instead + // of discarding it before the canonical validation runs. const dir = path.join(tempDir, 'config'); fs.writeFileSync( path.join(dir, 'onebot_10022.json'), @@ -493,10 +525,9 @@ describe('loadOneBotConfig', () => { }), 'utf8', ); - const loaded = loadOneBotConfig('10022'); - expect( - (loaded.networks.httpClients[0] as unknown as Record).maxPayloadBytes, - ).toBeUndefined(); + expect(() => loadOneBotConfig('10022')).toThrow( + /maxPayloadBytes is only supported by WebSocket adapters/, + ); }); it('replays a saved per-UIN snapshot without reviving deleted global adapters', () => { diff --git a/packages/websocket/src/websocket.ts b/packages/websocket/src/websocket.ts index 4b2c35f2..3f14440e 100644 --- a/packages/websocket/src/websocket.ts +++ b/packages/websocket/src/websocket.ts @@ -119,8 +119,8 @@ const DIAG_BUFFER_LIMIT = 64 * 1024; * payload exceeds `maxPayload`. * * The native parser rejects an oversized frame without saying where it is or how - * large it is, so the reporting path re-reads the bytes buffered since the last - * delivered message. Walking from a frame boundary removes the "the first bytes of + * large it is, so the reporting path re-reads the retained bytes from the first + * unparsed frame. Walking from a frame boundary removes the "the first bytes of * this chunk are the offending header" assumption: earlier data frames, interleaved * control frames and a header split across chunks are all stepped over correctly. * Returns `null` when the walk cannot reach a verdict, in which case the caller @@ -298,9 +298,9 @@ export class WebSocket extends EventEmitter { protected _msgCompressed = false; // Diagnostic re-read of inbound bytes, kept only so an oversized-frame error - // can name the offending frame. It never feeds parsing, is cleared whenever a - // message is delivered, and downgrades the error text (instead of growing) - // once the small budget would be exceeded. + // can name the offending frame. It follows the parser's frame boundaries, + // retaining any partial next frame, and downgrades the error text once the + // small budget would be exceeded. It never feeds parsing. private _diagChunks: Buffer[] = []; private _diagBytes = 0; private _diagOverflow = false; @@ -368,6 +368,7 @@ export class WebSocket extends EventEmitter { } for (const f of res.frames) { + this._consumeDiagnosticFrame(f); this._handleFrame(f); if (this._readyState === CLOSED) { @@ -401,21 +402,57 @@ export class WebSocket extends EventEmitter { return `${message}: ${info.payloadBytes} B ${opcodeLabel(info.opcode)} frame exceeds the ${limit}`; } - /** Keep a bounded copy of the bytes received since the last delivered message. */ + /** Keep bounded diagnostic bytes starting at the first unparsed frame. */ private _recordDiagnosticBytes(chunk: Buffer): void { + this._diagBytes += chunk.length; + if (this._diagOverflow) { return; } - if (this._diagBytes + chunk.length > DIAG_BUFFER_LIMIT) { + if (this._diagBytes > DIAG_BUFFER_LIMIT) { this._diagOverflow = true; this._diagChunks = []; - this._diagBytes = 0; return; } this._diagChunks.push(chunk); - this._diagBytes += chunk.length; + } + + private _consumeDiagnosticFrame(frame: ParsedFrame): void { + const payloadBytes = frame.payload.length; + // The parser enforces minimal length encoding and the masking direction, + // so every accepted frame has this exact wire length, including controls. + const lengthBytes = payloadBytes <= 125 ? 0 : payloadBytes <= 65535 ? 2 : 8; + let remaining = 2 + lengthBytes + (this._isServer ? 4 : 0) + payloadBytes; + this._diagBytes -= remaining; + + if (this._diagBytes === 0) { + this._resetDiagnosticBytes(); + return; + } + + // Once bytes are discarded, wait until the parser drains all pending + // bytes before recording again; a new read may start inside a frame. + if (this._diagOverflow) { + return; + } + + let consumedChunks = 0; + + while (remaining > 0) { + const chunk = this._diagChunks[consumedChunks]!; + + if (remaining < chunk.length) { + this._diagChunks[consumedChunks] = chunk.subarray(remaining); + break; + } + + remaining -= chunk.length; + consumedChunks += 1; + } + + this._diagChunks.splice(0, consumedChunks); } private _resetDiagnosticBytes(): void { @@ -572,7 +609,6 @@ export class WebSocket extends EventEmitter { this._msgSize = 0; this._msgValidator = null; this._msgCompressed = false; - this._resetDiagnosticBytes(); this.emit('message', data, isBinary); } } diff --git a/packages/websocket/tests/websocket.test.ts b/packages/websocket/tests/websocket.test.ts index 62aaec55..40d28d70 100644 --- a/packages/websocket/tests/websocket.test.ts +++ b/packages/websocket/tests/websocket.test.ts @@ -946,6 +946,43 @@ describe.skipIf(!nativeIo)('WebSocket incoming frames', () => { expect(errors[0]!.code).toBe(1009); }); + it.each([ + { isServer: false, previousPayloadBytes: 1 }, + { isServer: false, previousPayloadBytes: 126 }, + { isServer: true, previousPayloadBytes: 1 }, + { isServer: true, previousPayloadBytes: 126 }, + ])('preserves a split header after delivering the preceding message: %j', (options) => { + const { isServer, previousPayloadBytes } = options; + const { ws, socket } = createWs({ isServer, maxPayload: 1024 }); + const errors: Array = []; + const messages: Buffer[] = []; + ws.on('error', (err) => errors.push(err)); + ws.on('message', (data) => messages.push(data)); + const payload = Buffer.alloc(previousPayloadBytes, 0x61); + const precedingFrame = isServer + ? clientFrame(OP_TEXT, payload) + : serverFrame(OP_TEXT, payload); + const header = Buffer.from([0x82, isServer ? 0xfe : 0x7e, 0x82, 0x7e]); + socket.emit('data', Buffer.concat([precedingFrame, header.subarray(0, 2)])); + expect(messages).toEqual([payload]); + expect(errors).toHaveLength(0); + + // The length bytes resemble a new frame header. Losing the buffered + // prefix would invent a size from the mask or payload bytes that follow. + socket.emit( + 'data', + Buffer.concat([ + header.subarray(2), + isServer ? MASK : Buffer.alloc(0), + Buffer.from([0x20, 0x00]), + ]), + ); + expect(errors[0]!.message).toBe( + 'WebSocket protocol error: Payload exceeds configured max: 33406 B binary frame exceeds the 1024 byte limit (maxPayloadBytes)', + ); + expect(errors[0]!.code).toBe(1009); + }); + it('skips interleaved control frames while locating the oversized frame', () => { const { ws, socket } = createWs({ isServer: true }); const errors: Array = []; @@ -972,13 +1009,15 @@ describe.skipIf(!nativeIo)('WebSocket incoming frames', () => { ws.on('error', (err) => { errors.push(err); }); - // Build up more than the 64 KiB diagnostic budget with a fragmented - // message, so the offending header never gets buffered. - const fragment = Buffer.alloc(8 * 1024, 0x61); - for (let i = 0; i < 12; i += 1) { - socket.emit('data', clientFrame(i === 0 ? OP_TEXT : OP_CONT, fragment, false)); - } - socket.emit('data', maskedOversizedHeader(OP_TEXT, 2 * 1024 * 1024)); + // This single read exceeds the diagnostic budget before the parser + // reports the error, so no reliable header is retained. + socket.emit( + 'data', + Buffer.concat([ + clientFrame(OP_TEXT, Buffer.alloc(96 * 1024, 0x61)), + maskedOversizedHeader(OP_TEXT, 2 * 1024 * 1024), + ]), + ); expect(errors[0]!.message).toBe( 'WebSocket protocol error: Payload exceeds configured max: an inbound frame exceeds the 1048576 byte limit (maxPayloadBytes)', ); @@ -986,6 +1025,56 @@ describe.skipIf(!nativeIo)('WebSocket incoming frames', () => { expect(errors[0]!.code).toBe(1009); }); + it('releases parsed fragments and control frames from the diagnostic budget', () => { + const maxPayload = 1024 * 1024; + const { ws, socket } = createWs({ isServer: true, maxPayload }); + const errors: Error[] = []; + ws.on('error', (err) => errors.push(err)); + const fragment = Buffer.alloc(8 * 1024, 0x61); + + for (let i = 0; i < 12; i += 1) { + socket.emit('data', clientFrame(i === 0 ? OP_TEXT : OP_CONT, fragment, i === 11)); + socket.emit('data', clientFrame(OP_PING, Buffer.from('hb'))); + } + + expect(errors).toHaveLength(0); + socket.emit('data', maskedOversizedHeader(OP_BIN, 2 * 1024 * 1024)); + expect(errors[0]!.message).toContain('2097152 B binary frame'); + }); + + it('recovers diagnostics after a complete frame exceeds the diagnostic budget', () => { + const { ws, socket } = createWs({ isServer: true, maxPayload: 1024 * 1024 }); + const errors: Error[] = []; + const messages: Buffer[] = []; + ws.on('error', (err) => errors.push(err)); + ws.on('message', (data) => messages.push(data)); + const payload = Buffer.alloc(96 * 1024, 0x61); + socket.emit('data', clientFrame(OP_TEXT, payload)); + expect(messages).toEqual([payload]); + expect(errors).toHaveLength(0); + socket.emit('data', maskedOversizedHeader(OP_BIN, 2 * 1024 * 1024)); + expect(errors[0]!.message).toContain('2097152 B binary frame'); + }); + + it('keeps size unknown when a partial next header was discarded by the budget', () => { + const { ws, socket } = createWs({ isServer: true, maxPayload: 1024 * 1024 }); + const errors: Error[] = []; + ws.on('error', (err) => errors.push(err)); + const header = maskedOversizedHeader(OP_BIN, 2 * 1024 * 1024); + socket.emit( + 'data', + Buffer.concat([ + clientFrame(OP_TEXT, Buffer.alloc(96 * 1024, 0x61)), + header.subarray(0, 2), + ]), + ); + expect(errors).toHaveLength(0); + socket.emit('data', header.subarray(2)); + expect(errors[0]!.message).toBe( + 'WebSocket protocol error: Payload exceeds configured max: an inbound frame exceeds the 1048576 byte limit (maxPayloadBytes)', + ); + }); + it('names the message type when a binary frame exceeds the limit', () => { const { ws, socket } = createWs({ isServer: true, maxPayload: 4 }); const errors: Array = [];