diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index 026a64098641..7f0901ec1a6e 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -331,8 +331,11 @@ ObjectDefineProperty(ZlibBase.prototype, "bytesRead", { }); ZlibBase.prototype.reset = function (this: $Node.ZlibBase) { - $assert(this._handle, "zlib binding closed"); - return this._handle.reset(); + const handle = this._handle; + if (!handle) { + throw $ERR_INTERNAL_ASSERTION("zlib binding closed" + require("internal/shared").kInternalAssertionSuffix); + } + return handle.reset(); }; // This is the _flush function called by the transform class, internally, when the last chunk has been written. diff --git a/src/runtime/node/zlib/NativeZlib.rs b/src/runtime/node/zlib/NativeZlib.rs index 4f9f5618bb77..7c674cc83c0a 100644 --- a/src/runtime/node/zlib/NativeZlib.rs +++ b/src/runtime/node/zlib/NativeZlib.rs @@ -218,6 +218,8 @@ pub(crate) struct Context { pub flush: c::FlushValue, pub dictionary: Vec, pub(crate) gzip_id_bytes_read: u8, + /// GZIP only: Node's `output_emitted_ && !stream_complete_` (https://github.com/nodejs/node/blob/v26.12.1/src/node_zlib.cc#L1127-L1137). + output_emitted: bool, } impl Default for Context { @@ -229,6 +231,7 @@ impl Default for Context { flush: c::FlushValue::NoFlush, dictionary: Vec::new(), gzip_id_bytes_read: 0, + output_emitted: false, } } } @@ -253,6 +256,7 @@ impl Context { use c::NodeMode::*; self.flush = c::FlushValue::NoFlush; self.err = c::ReturnCode::Ok; + self.output_emitted = false; let window_bits_actual = match self.mode { NONE => unreachable!(), @@ -364,6 +368,19 @@ impl Context { } pub(crate) fn reset(&mut self) -> Error { + // https://github.com/nodejs/node/blob/v26.12.1/src/node_zlib.cc#L1282-L1294 + if self.output_emitted { + return Error::init( + c"Cannot reset a gzip stream with an incomplete member; end the stream or start a new gzip compressor".as_ptr(), + c::ReturnCode::StreamError as c_int, + c"ERR_ZLIB_INCOMPLETE_FRAME".as_ptr(), + ); + } + self.reset_z_stream() + } + + /// `reset()` without the check for an open gzip member. The gunzip loop over members calls it. + fn reset_z_stream(&mut self) -> Error { use c::NodeMode::*; self.err = c::ReturnCode::Ok; match self.mode { @@ -423,9 +440,12 @@ impl Context { // of room. If there was avail_out left over, then it means // that all of the input was consumed. match self.mode { - DEFLATE | GZIP | DEFLATERAW => { + DEFLATE | DEFLATERAW => { return self.do_work_deflate(); } + GZIP => { + return self.do_work_gzip(); + } UNZIP => { if self.state.avail_in() > 0 { next_expected_header_byte = Some(self.state.next_in()); @@ -479,6 +499,18 @@ impl Context { self.err = unsafe { c::deflate(self.state.as_ptr(), self.flush) }; } + // Out of line, so `do_work` stays a tail call for the modes that `reset()` never refuses. + #[inline(never)] + fn do_work_gzip(&mut self) { + let avail_out = self.state.avail_out(); + self.do_work_deflate(); + if self.err == c::ReturnCode::StreamEnd { + self.output_emitted = false; + } else if self.state.avail_out() < avail_out { + self.output_emitted = true; + } + } + fn do_work_inflate(&mut self) { // SAFETY: FFI — state is an initialized inflate stream. self.err = unsafe { c::inflate(self.state.as_ptr(), self.flush) }; @@ -511,8 +543,8 @@ impl Context { { // Bytes remain in input buffer. Perhaps this is another compressed member in the same archive, or just trailing garbage. // Trailing zero bytes are okay, though, since they are frequently used for padding. - let _ = self.reset(); - // SAFETY: FFI — state was just re-initialized by reset(). + let _ = self.reset_z_stream(); + // SAFETY: FFI — state was just re-initialized by reset_z_stream(). self.err = unsafe { c::inflate(self.state.as_ptr(), self.flush) }; } } diff --git a/test/js/node/__snapshots__/members-of-prototype-objects-for-any-receiver.test.ts.snap b/test/js/node/__snapshots__/members-of-prototype-objects-for-any-receiver.test.ts.snap index b1a9e912caa4..d6ae60cdc755 100644 --- a/test/js/node/__snapshots__/members-of-prototype-objects-for-any-receiver.test.ts.snap +++ b/test/js/node/__snapshots__/members-of-prototype-objects-for-any-receiver.test.ts.snap @@ -4919,10 +4919,12 @@ node:zlib.Deflate.prototype, 2 above bytesRead, stored to throws TypeError: undefined is not an object <- undefined throws TypeError: null is not an object <- null node:zlib.Deflate.prototype, 2 above reset() - throws TypeError: undefined is not an object <- every object, a number, a string, undefined + throws Error ERR_INTERNAL_ASSERTION: zlib binding closed <- every object, a number, a string + throws TypeError: undefined is not an object <- undefined throws TypeError: null is not an object <- null node:zlib.Deflate.prototype, 2 above reset("a", 1) - throws TypeError: undefined is not an object <- every object, a number, a string, undefined + throws Error ERR_INTERNAL_ASSERTION: zlib binding closed <- every object, a number, a string + throws TypeError: undefined is not an object <- undefined throws TypeError: null is not an object <- null node:zlib.Deflate.prototype, 2 above _flush() throws TypeError: what is called is not a function <- {}, an object without a prototype, an instance of another class, a proxy of {}, a number, a string diff --git a/test/js/node/test/parallel/test-zlib-reset-incomplete-output.js b/test/js/node/test/parallel/test-zlib-reset-incomplete-output.js new file mode 100644 index 000000000000..84190b32d528 --- /dev/null +++ b/test/js/node/test/parallel/test-zlib-reset-incomplete-output.js @@ -0,0 +1,154 @@ +'use strict'; + +// Tests that reset() refuses to run on gzip once an incomplete member has +// already emitted output. +// +// deflateReset() is equivalent to deflateEnd + deflateInit. gzip writes a +// header on the first write, so those bytes cannot be taken back. +// +// zlib-wrapped deflate still allows reset after flush: official +// test-zlib-dictionary.js discards the first member and reuses the +// compressor. Raw deflate has no wrapper header: a small write may emit +// nothing, and flush+reset still concatenates. + +require('../common'); +const assert = require('assert'); +const { finished } = require('stream/promises'); +const test = require('node:test'); +const zlib = require('zlib'); + +async function writeHello(stream) { + await new Promise((resolve, reject) => { + stream.write(Buffer.from('hello'), (err) => { + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); +} + +test('Gzip reset throws when write has emitted wrapper output', async () => { + const stream = zlib.createGzip(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + await writeHello(stream); + assert.ok(Buffer.concat(chunks).length > 0); + + stream.reset(); + stream.end(Buffer.from('world')); + + await assert.rejects(finished(stream), { + code: 'ERR_ZLIB_INCOMPLETE_FRAME', + }); +}); + +test('Gzip reset throws when flush has emitted incomplete output', async () => { + const stream = zlib.createGzip(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + assert.ok(Buffer.concat(chunks).length > 0); + + stream.reset(); + stream.end(Buffer.from('world')); + + await assert.rejects(finished(stream), { + code: 'ERR_ZLIB_INCOMPLETE_FRAME', + }); +}); + +test('Gzip flush followed by end still produces a valid stream', async () => { + const stream = zlib.createGzip(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.gunzipSync(Buffer.concat(chunks)).toString(), + 'helloworld', + ); +}); + +test('Gzip reset before any write still works', async () => { + const stream = zlib.createGzip(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.reset(); + stream.end(Buffer.from('hello')); + await finished(stream); + + assert.strictEqual( + zlib.gunzipSync(Buffer.concat(chunks)).toString(), + 'hello', + ); +}); + +test('Deflate reset after flush still works when first output is discarded', + async () => { + const stream = zlib.createDeflate(); + const chunks = []; + let take = false; + stream.on('data', (chunk) => { + if (take) { + chunks.push(chunk); + } + }); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + stream.reset(); + take = true; + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.inflateSync(Buffer.concat(chunks)).toString(), + 'world', + ); + }); + +test('DeflateRaw reset after write without emitted output still works', + async () => { + const stream = zlib.createDeflateRaw(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + await writeHello(stream); + assert.strictEqual(Buffer.concat(chunks).length, 0); + + stream.reset(); + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.inflateRawSync(Buffer.concat(chunks)).toString(), + 'world', + ); + }); + +test('DeflateRaw flush followed by reset still concatenates', async () => { + const stream = zlib.createDeflateRaw(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + stream.reset(); + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.inflateRawSync(Buffer.concat(chunks)).toString(), + 'helloworld', + ); +}); diff --git a/test/js/node/zlib/zlib-reset-incomplete-frame.test.ts b/test/js/node/zlib/zlib-reset-incomplete-frame.test.ts new file mode 100644 index 000000000000..431b61b617e2 --- /dev/null +++ b/test/js/node/zlib/zlib-reset-incomplete-frame.test.ts @@ -0,0 +1,347 @@ +// reset() of a gzip compressor whose open member has already given output. +// Node.js refuses it since v26.11.0 (nodejs/node#66179), because the bytes that are already out +// cannot be taken back. This file passes unchanged under Node.js v26.12.1. +import assert from "node:assert"; +import { once } from "node:events"; +import { finished } from "node:stream/promises"; +import { describe, test } from "node:test"; +import zlib from "node:zlib"; + +const { Z_DEFAULT_STRATEGY, Z_FINISH, Z_NO_FLUSH, Z_STREAM_ERROR, Z_SYNC_FLUSH } = zlib.constants; + +// The tests reach `_handle`, `_writeState` and `_processChunk`, and the types have no `reset()` for most kinds. +type Stream = any; +type Handle = any; +type Report = { message: string; errno: number; code: string }; + +const refusal: Report = { + message: "Cannot reset a gzip stream with an incomplete member; end the stream or start a new gzip compressor", + errno: Z_STREAM_ERROR, + code: "ERR_ZLIB_INCOMPLETE_FRAME", +}; + +// Without an 'error' listener, a stream that reset() destroys throws its error from process.nextTick, +// into the test that runs by then. Each test reads the error from finished(). +function quiet(stream: Stream): Stream { + return stream.on("error", () => {}); +} + +// A write that fails destroys the stream and does not call its callback, so 'error' and 'close' +// settle the promise too. +function settled(stream: Stream, start: (callback: (error?: Error | null) => void) => void) { + const { promise, resolve, reject } = Promise.withResolvers(); + const onError = (error: Error) => reject(error); + const onClose = () => reject(new Error("the stream closed before the callback ran")); + stream.once("error", onError); + stream.once("close", onClose); + start(error => (error ? reject(error) : resolve())); + return promise.finally(() => { + stream.off("error", onError); + stream.off("close", onClose); + }); +} + +function flushed(stream: Stream, kind?: number) { + return settled(stream, callback => (kind === undefined ? stream.flush(callback) : stream.flush(kind, callback))); +} + +function written(stream: Stream, chunk: string | Buffer) { + return settled(stream, callback => stream.write(chunk, callback)); +} + +function collect(stream: Stream): Buffer[] { + const chunks: Buffer[] = []; + stream.on("data", (chunk: Buffer) => chunks.push(chunk)); + return chunks; +} + +function reportsOf(handle: Handle, onReport?: (error: Error) => void): Report[] { + const reports: Report[] = []; + handle.onerror = (message: string, errno: number, code: string) => { + reports.push({ message, errno, code }); + onReport?.(Object.assign(new Error(message), { errno, code })); + }; + return reports; +} + +describe("Gzip: reset() in an open member", () => { + test("after a flush, destroys the stream before it returns", async () => { + const stream = quiet(zlib.createGzip()); + const chunks = collect(stream); + stream.write("hello"); + await flushed(stream); + assert.ok(Buffer.concat(chunks).length > 0); + + assert.strictEqual(stream.reset(), undefined); + assert.strictEqual(stream.destroyed, true); + await assert.rejects(finished(stream), { name: "Error", ...refusal }); + }); + + test("after output and then a write that gives none", async () => { + const stream = quiet(zlib.createGzip()); + const chunks = collect(stream); + stream.write("hello"); + await flushed(stream); + const given = Buffer.concat(chunks).length; + assert.ok(given > 0); + // This write stays in the compressor. The member is still open, and part of it is out. + await written(stream, "world"); + assert.strictEqual(Buffer.concat(chunks).length, given); + + stream.reset(); + assert.strictEqual(stream.destroyed, true); + await assert.rejects(finished(stream), { code: "ERR_ZLIB_INCOMPLETE_FRAME" }); + }); + + test("is allowed once a finish flush completed it, and the next one is guarded again", async () => { + const stream = quiet(zlib.createGzip()); + const chunks = collect(stream); + stream.write("hello"); + // This output makes it open. The finish flush then completes it. + await flushed(stream); + assert.ok(Buffer.concat(chunks).length > 0); + await flushed(stream, Z_FINISH); + stream.reset(); + assert.strictEqual(stream.destroyed, false); + stream.write("world"); + await flushed(stream, Z_FINISH); + // Two complete members decode as one. + assert.strictEqual(zlib.gunzipSync(Buffer.concat(chunks)).toString(), "helloworld"); + + stream.reset(); + stream.write("again"); + await flushed(stream); + stream.reset(); + assert.strictEqual(stream.destroyed, true); + await assert.rejects(finished(stream), { code: "ERR_ZLIB_INCOMPLETE_FRAME" }); + }); + + test("inside a 'data' listener", async () => { + const stream = quiet(zlib.createGzip()); + const done = finished(stream); + const { promise: destroyed, resolve } = Promise.withResolvers(); + stream.once("data", () => { + stream.reset(); + resolve(stream.destroyed); + }); + stream.write("hello"); + stream.flush(); + // A stream that ends before its first 'data' settles `done`, and 'data' never comes. + const early = done.then(() => Promise.reject(new Error("the stream finished before 'data'"))); + assert.strictEqual(await Promise.race([destroyed, early]), true); + await assert.rejects(done, { code: "ERR_ZLIB_INCOMPLETE_FRAME" }); + }); + + test("after a synchronous _processChunk()", async () => { + const stream = quiet(zlib.createGzip()); + const handle = stream._handle; + // A synchronous _processChunk() closes the handle when it returns. minizlib keeps it open this way. + const close = handle.close; + handle.close = () => {}; + const output = stream._processChunk(Buffer.from("hello"), Z_SYNC_FLUSH); + stream._handle = handle; + handle.close = close; + assert.ok(output.length > 0); + + stream.reset(); + assert.strictEqual(stream.destroyed, true); + await assert.rejects(finished(stream), { code: "ERR_ZLIB_INCOMPLETE_FRAME" }); + }); + + test("after params(), which flushes the header", async () => { + const stream = quiet(zlib.createGzip()); + await settled(stream, callback => stream.params(1, Z_DEFAULT_STRATEGY, callback)); + stream.reset(); + assert.strictEqual(stream.destroyed, true); + await assert.rejects(finished(stream), { code: "ERR_ZLIB_INCOMPLETE_FRAME" }); + }); + + test("on the native handle, reports to onerror and keeps what is open", () => { + const stream = quiet(zlib.createGzip()); + const handle = stream._handle; + const reports = reportsOf(handle); + const input = Buffer.from("hello hello hello hello"); + const output = Buffer.alloc(4096); + try { + // 4 bytes of room: the output starts, and its end does not fit. + handle.writeSync(Z_FINISH, input, 0, input.length, output, 0, 4); + handle.reset(); + assert.deepStrictEqual(reports, [refusal]); + + // The refused reset() dropped nothing: the rest arrives, and all of it decodes. + const unread = stream._writeState[1]; + handle.writeSync(Z_FINISH, input, input.length - unread, unread, output, 4, output.length - 4); + const length = output.length - stream._writeState[0]; + assert.strictEqual(zlib.gunzipSync(output.subarray(0, length)).toString(), input.toString()); + + // It is complete now, so reset() is allowed. + handle.reset(); + assert.strictEqual(reports.length, 1); + } finally { + stream.destroy(); + } + }); + + test("on the native handle, after output and then a write that gives none", () => { + const stream = quiet(zlib.createGzip()); + const handle = stream._handle; + const reports = reportsOf(handle); + const input = Buffer.from("hello"); + const output = Buffer.alloc(4096); + try { + handle.writeSync(Z_SYNC_FLUSH, input, 0, input.length, output, 0, output.length); + assert.ok(stream._writeState[0] < output.length); + handle.writeSync(Z_NO_FLUSH, input, 0, input.length, output, 0, output.length); + assert.deepStrictEqual([...stream._writeState], [output.length, 0]); + handle.reset(); + assert.deepStrictEqual(reports, [refusal]); + } finally { + stream.destroy(); + } + }); + + test("on the native handle, after a write() on the thread pool", async () => { + const stream = quiet(zlib.createGzip()); + const handle = new stream._handle.constructor(zlib.constants.GZIP); + stream.destroy(); + const { promise: wrote, resolve, reject } = Promise.withResolvers(); + handle.init(15, 6, 8, Z_DEFAULT_STRATEGY, new Uint32Array(2), resolve, undefined); + const reports = reportsOf(handle, reject); + const input = Buffer.from("hello"); + const output = Buffer.alloc(4096); + try { + handle.write(Z_SYNC_FLUSH, input, 0, input.length, output, 0, output.length); + await wrote; + handle.reset(); + assert.deepStrictEqual(reports, [refusal]); + } finally { + handle.close(); + } + }); +}); + +describe("Gzip: reset() is allowed while no part of a member is out", () => { + const input = Buffer.from("hello"); + + test("a finish write completes the member only if its end fits in the output", () => { + // At level 0 a member is 10 bytes of header, one stored block of 5 + n bytes, and 8 bytes of trailer. + for (const [length, expected] of [ + [64 - 23, []], + [64 - 22, [refusal]], + ] as const) { + const stream = quiet(zlib.createGzip({ level: 0 })); + const handle = stream._handle; + const reports = reportsOf(handle); + const output = Buffer.alloc(64); + try { + handle.writeSync(Z_FINISH, Buffer.alloc(length, "a"), 0, length, output, 0, output.length); + // The output is full in both cases. + assert.deepStrictEqual([...stream._writeState], [0, 0]); + handle.reset(); + assert.deepStrictEqual(reports, expected); + } finally { + stream.destroy(); + } + } + }); + + test("a write with no room for the header gives no output", () => { + const stream = quiet(zlib.createGzip()); + const handle = stream._handle; + const reports = reportsOf(handle); + try { + handle.writeSync(Z_NO_FLUSH, input, 0, input.length, Buffer.alloc(64), 0, 0); + assert.deepStrictEqual([...stream._writeState], [0, input.length]); + handle.reset(); + assert.deepStrictEqual(reports, []); + } finally { + stream.destroy(); + } + }); + + test("a write after the end of the member fails and gives no output", () => { + const stream = quiet(zlib.createGzip()); + const handle = stream._handle; + const reports = reportsOf(handle); + const output = Buffer.alloc(64); + try { + handle.writeSync(Z_FINISH, input, 0, input.length, output, 0, output.length); + assert.deepStrictEqual(reports, []); + handle.writeSync(Z_NO_FLUSH, input, 0, input.length, output, 0, output.length); + assert.deepStrictEqual(reports, [{ message: "stream error", errno: Z_STREAM_ERROR, code: "Z_STREAM_ERROR" }]); + handle.reset(); + assert.strictEqual(reports.length, 1); + } finally { + stream.destroy(); + } + }); +}); + +// Node.js refuses none of these. +const neverRefused: [name: string, create: () => Stream][] = [ + ["Deflate", () => zlib.createDeflate()], + ["DeflateRaw", () => zlib.createDeflateRaw()], + ["BrotliCompress", () => zlib.createBrotliCompress()], +]; + +for (const [name, create] of neverRefused) { + test(`${name}: reset() after a flush is allowed`, async () => { + const stream = quiet(create()); + const chunks = collect(stream); + stream.write("hello"); + await flushed(stream); + assert.ok(Buffer.concat(chunks).length > 0); + stream.reset(); + assert.strictEqual(stream.destroyed, false); + stream.end("world"); + await finished(stream); + }); +} + +const decompressors: [name: string, create: () => Stream, compress: (data: Buffer) => Buffer][] = [ + ["Gunzip", () => zlib.createGunzip(), zlib.gzipSync], + ["Inflate", () => zlib.createInflate(), zlib.deflateSync], + ["InflateRaw", () => zlib.createInflateRaw(), zlib.deflateRawSync], + ["Unzip", () => zlib.createUnzip(), zlib.gzipSync], + ["BrotliDecompress", () => zlib.createBrotliDecompress(), zlib.brotliCompressSync], + ["ZstdDecompress", () => zlib.createZstdDecompress(), zlib.zstdCompressSync], +]; + +for (const [name, create, compress] of decompressors) { + test(`${name}: reset() in the middle of the input is allowed`, async () => { + const data = Buffer.alloc(100_000, "abcdefghij"); + const frame = compress(data); + const stream = quiet(create()); + const chunks = collect(stream); + await written(stream, frame.subarray(0, frame.length >> 1)); + stream.reset(); + assert.strictEqual(stream.destroyed, false); + + chunks.length = 0; + stream.end(frame); + await once(stream, "end"); + assert.ok(Buffer.concat(chunks).equals(data)); + }); +} + +const everyKind: [name: string, create: () => Stream][] = [ + ["Gzip", () => zlib.createGzip()], + ...neverRefused, + ["ZstdCompress", () => zlib.createZstdCompress()], + ...decompressors.map(([name, create]): [string, () => Stream] => [name, create]), +]; + +for (const [name, create] of everyKind) { + test(`${name}: reset() of a destroyed stream throws ERR_INTERNAL_ASSERTION`, () => { + const stream = create(); + stream.destroy(); + assert.throws(() => stream.reset(), { + name: "Error", + code: "ERR_INTERNAL_ASSERTION", + message: + "zlib binding closed\n" + + "This is caused by either a bug in Node.js or incorrect usage of Node.js internals.\n" + + "Please open an issue with this stack trace at https://github.com/nodejs/node/issues\n", + }); + }); +}