From 2a6a92cb7ba6067ce00072163d792f6abd974079 Mon Sep 17 00:00:00 2001 From: Xia Chao Date: Mon, 21 Sep 2026 17:36:42 +0800 Subject: [PATCH] zlib: reject reset while a zstd frame is incomplete Resetting a ZstdCompress stream while a frame is still in progress dropped the frame state, but any bytes already written out stayed at the start of the output stream. The next frame was then appended to that fragment, so the resulting stream could not be decompressed. The failure was silent: the compressor reported no error at all. ZSTD_reset_session_only cancels unflushed internal data, so refuse reset only when the incomplete frame has already emitted output. Signed-off-by: Xia Chao --- doc/api/zlib.md | 6 + src/node_zlib.cc | 48 +++++-- .../test-zlib-zstd-reset-incomplete-frame.js | 129 ++++++++++++++++++ 3 files changed, 175 insertions(+), 8 deletions(-) create mode 100644 test/parallel/test-zlib-zstd-reset-incomplete-frame.js diff --git a/doc/api/zlib.md b/doc/api/zlib.md index 656331d9b08e..3664cb6080da 100644 --- a/doc/api/zlib.md +++ b/doc/api/zlib.md @@ -2200,6 +2200,12 @@ configured for a Zstd compressor, it applies again to the next frame. Calling `reset()` while a write is in progress throws an `Error`. +Resetting an incomplete Zstd compression frame after it has emitted output +causes the stream to error with `ERR_ZLIB_INCOMPLETE_FRAME`. Resetting at +that point would discard the frame state while the bytes already written +out remain at the start of the output stream, leaving it undecodable. Call +`.end()`, or start over with a new stream, instead. + ## Class: `ZstdOptions` > Stability: 1 - Experimental diff --git a/src/node_zlib.cc b/src/node_zlib.cc index 810df1c779fd..7d6ed93242db 100644 --- a/src/node_zlib.cc +++ b/src/node_zlib.cc @@ -352,6 +352,14 @@ class ZstdCompressContext final : public ZstdContext { uint64_t pledged_src_size_ = ZSTD_CONTENTSIZE_UNKNOWN; std::optional consumed_src_size_; + + // A frame is complete once ZSTD_compressStream2() has been called with + // ZSTD_e_end and has returned 0. Resetting an incomplete frame is only unsafe + // once some of that frame has already been written out: those bytes cannot be + // discarded, and the next frame would be appended to the fragment. Unflushed + // internal state alone is cancelled by ZSTD_reset_session_only. + bool frame_complete_ = true; + bool frame_output_emitted_ = false; }; class ZstdDecompressContext final : public ZstdContext { @@ -1678,6 +1686,8 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size, std::string_view dictionary, bool) { pledged_src_size_ = pledged_src_size; + frame_complete_ = true; + frame_output_emitted_ = false; if (pledged_src_size == ZSTD_CONTENTSIZE_UNKNOWN) { consumed_src_size_.reset(); } else { @@ -1718,6 +1728,17 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size, } CompressionError ZstdCompressContext::ResetStream() { + // ZSTD_reset_session_only cancels unflushed internal data. Bytes that have + // already been written out cannot be taken back, so refuse reset only when + // the current incomplete frame has already emitted output. + if (!frame_complete_ && frame_output_emitted_) { + return CompressionError( + "Cannot reset a zstd stream with an incomplete frame; end the frame " + "or discard the output produced so far", + "ERR_ZLIB_INCOMPLETE_FRAME", + ZSTD_error_stage_wrong); + } + size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only); if (ZSTD_isError(result)) { const ZSTD_ErrorCode error = ZSTD_getErrorCode(result); @@ -1737,6 +1758,8 @@ CompressionError ZstdCompressContext::ResetStream() { } else { consumed_src_size_ = 0; } + frame_complete_ = true; + frame_output_emitted_ = false; error_ = ZSTD_error_no_error; error_string_.clear(); error_code_string_.clear(); @@ -1751,19 +1774,28 @@ void ZstdCompressContext::DoThreadPoolWork() { if (consumed_src_size_.has_value()) { *consumed_src_size_ += input_.pos - input_pos; } + if (output_.pos > 0) { + frame_output_emitted_ = true; + } if (ZSTD_isError(remaining)) { error_ = ZSTD_getErrorCode(remaining); error_code_string_ = ZstdStrerror(error_); error_string_ = ZSTD_getErrorString(error_); - } else if (remaining == 0 && flush_ == ZSTD_e_end && - consumed_src_size_.has_value()) { - uint64_t const consumed_src_size = *consumed_src_size_; - consumed_src_size_.reset(); - if (consumed_src_size != pledged_src_size_) { - error_ = ZSTD_error_srcSize_wrong; - error_code_string_ = ZstdStrerror(error_); - error_string_ = ZSTD_getErrorString(error_); + frame_complete_ = false; + } else if (remaining == 0 && flush_ == ZSTD_e_end) { + frame_complete_ = true; + frame_output_emitted_ = false; + if (consumed_src_size_.has_value()) { + uint64_t const consumed_src_size = *consumed_src_size_; + consumed_src_size_.reset(); + if (consumed_src_size != pledged_src_size_) { + error_ = ZSTD_error_srcSize_wrong; + error_code_string_ = ZstdStrerror(error_); + error_string_ = ZSTD_getErrorString(error_); + } } + } else { + frame_complete_ = false; } } diff --git a/test/parallel/test-zlib-zstd-reset-incomplete-frame.js b/test/parallel/test-zlib-zstd-reset-incomplete-frame.js new file mode 100644 index 000000000000..587155d26c45 --- /dev/null +++ b/test/parallel/test-zlib-zstd-reset-incomplete-frame.js @@ -0,0 +1,129 @@ +'use strict'; + +// Tests that reset() refuses to run only when an incomplete zstd frame has +// already emitted output. +// +// ZSTD_reset_session_only cancels unflushed internal data, so write()-then- +// reset() with no emitted bytes is safe. Once flush() (or a write that filled +// the output buffer) has written a fragment out, resetting would append the +// next frame to that fragment and produce an undecodable stream. + +require('../common'); +const assert = require('assert'); +const { finished } = require('stream/promises'); +const test = require('node:test'); +const zlib = require('zlib'); + +test('ZstdCompress reset throws when an incomplete frame has emitted output', + async () => { + const stream = zlib.createZstdCompress(); + 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); + + // A fragment of the frame is already outside the compressor, so reset() + // must refuse instead of silently producing a stream that cannot be + // decoded. + stream.reset(); + stream.end(Buffer.from('world')); + + await assert.rejects(finished(stream), { + code: 'ERR_ZLIB_INCOMPLETE_FRAME', + }); + }); + +test('ZstdCompress reset throws when write itself emitted frame output', + async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + // Fill a buffer that is large enough for zstd to emit compressed bytes + // during write(), without an explicit flush(). + const input = Buffer.allocUnsafe(512 * 1024); + for (let i = 0; i < input.length; i++) { + input[i] = i & 0xff; + } + await new Promise((resolve, reject) => { + stream.write(input, (err) => { + if (err) { + reject(err); + } else { + 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('ZstdCompress reset after write without emitted output still works', + async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + // Small writes often stay buffered inside zstd until flush/end. + await new Promise((resolve, reject) => { + stream.write(Buffer.from('hello'), (err) => { + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); + assert.strictEqual(Buffer.concat(chunks).length, 0); + + // No bytes have left the compressor, so session reset is allowed. + stream.reset(); + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(), + 'world', + ); + }); + +test('ZstdCompress flush followed by end still produces a valid stream', + async () => { + const stream = zlib.createZstdCompress(); + 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.zstdDecompressSync(Buffer.concat(chunks)).toString(), + 'helloworld', + ); + }); + +test('ZstdCompress reset before any write still works', async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + // No frame has been started yet, so reset() is allowed. + stream.reset(); + stream.end(Buffer.from('hello')); + await finished(stream); + + assert.strictEqual( + zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(), + 'hello', + ); +});