Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions src/js/node/zlib.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
38 changes: 35 additions & 3 deletions src/runtime/node/zlib/NativeZlib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,8 @@ pub(crate) struct Context {
pub flush: c::FlushValue,
pub dictionary: Vec<u8>,
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 {
Expand All @@ -229,6 +231,7 @@ impl Default for Context {
flush: c::FlushValue::NoFlush,
dictionary: Vec::new(),
gzip_id_bytes_read: 0,
output_emitted: false,
}
}
}
Expand All @@ -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!(),
Comment thread
robobun marked this conversation as resolved.
Resolved
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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) };
Expand Down Expand Up @@ -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) };
}
}
Expand Down
154 changes: 154 additions & 0 deletions test/js/node/test/parallel/test-zlib-reset-incomplete-output.js
Original file line number Diff line number Diff line change
@@ -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',
);
});
Loading
Loading