diff --git a/src/stream_base.cc b/src/stream_base.cc index a73a39513fb..a4f2d3c573c 100644 --- a/src/stream_base.cc +++ b/src/stream_base.cc @@ -18,6 +18,7 @@ namespace node { using v8::Array; using v8::ArrayBuffer; +using v8::ArrayBufferView; using v8::BackingStore; using v8::BackingStoreInitializationMode; using v8::ConstructorBehavior; @@ -162,8 +163,8 @@ int StreamBase::ReadStopJS(const FunctionCallbackInfo& args) { int StreamBase::UseUserBuffer(const FunctionCallbackInfo& args) { CHECK(Buffer::HasInstance(args[0])); - uv_buf_t buf = uv_buf_init(Buffer::Data(args[0]), Buffer::Length(args[0])); - PushStreamListener(new CustomBufferJSListener(buf)); + PushStreamListener( + new CustomBufferJSListener(args.This(), args[0].As())); return 0; } @@ -765,9 +766,44 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { stream->CallJSOnreadMethod(nread, ab, offset); } +CustomBufferJSListener::CustomBufferJSListener(Local object, + Local view) { + SetBuffer(object, view); +} + +void CustomBufferJSListener::SetBuffer(Local object, + Local view) { + // Keep the view's lifetime tied to the wrapper without a strong global root. + object->SetInternalField(StreamBase::kUserBufferField, view); + backing_store_ = view->Buffer()->GetBackingStore(); +} uv_buf_t CustomBufferJSListener::OnStreamAlloc(size_t suggested_size) { - return buffer_; + CHECK_NOT_NULL(stream_); + StreamBase* stream = static_cast(stream_); + Environment* env = stream->stream_env(); + Isolate* isolate = env->isolate(); + HandleScope handle_scope(isolate); + + Local view = + stream->GetAsyncWrap() + ->object() + ->GetInternalField(StreamBase::kUserBufferField) + .As(); + if (!backing_store_->IsImmutable()) { + // SetBuffer() materializes the backing store, so this cannot copy. + const auto contents = view->GetContents({}); + if (!contents.empty()) { + buffer_ = uv_buf_init(reinterpret_cast(contents.data()), + contents.size()); + return buffer_; + } + } + + // Allocation must return a usable buffer. Report the invalid view when the + // read completes, without writing into the user's former backing store. + using_fallback_buffer_ = true; + return env->allocate_managed_buffer(suggested_size); } @@ -779,6 +815,13 @@ void CustomBufferJSListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) { HandleScope handle_scope(env->isolate()); Context::Scope context_scope(env->context()); + if (using_fallback_buffer_) { + using_fallback_buffer_ = false; + env->recycle_managed_buffer(env->release_managed_buffer(buf)); + stream->CallJSOnreadMethod(UV_EINVAL, Local()); + return; + } + // In the case that there's an error and buf is null, return immediately. // This can happen on unices when POLLHUP is received and UV_EOF is returned // or when getting an error while performing a UV_HANDLE_ZERO_READ on Windows. @@ -795,8 +838,9 @@ void CustomBufferJSListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) { StreamBase::SKIP_NREAD_CHECKS); Local next_buf_v; if (ret.ToLocal(&next_buf_v) && !next_buf_v->IsUndefined()) { - buffer_.base = Buffer::Data(next_buf_v); - buffer_.len = Buffer::Length(next_buf_v); + CHECK(Buffer::HasInstance(next_buf_v)); + SetBuffer(stream->GetAsyncWrap()->object(), + next_buf_v.As()); } } diff --git a/src/stream_base.h b/src/stream_base.h index 450205a1b86..0a45270f76e 100644 --- a/src/stream_base.h +++ b/src/stream_base.h @@ -11,6 +11,7 @@ #include "v8.h" #include // INT_MIN +#include namespace node { @@ -211,10 +212,16 @@ class CustomBufferJSListener : public ReportWritesToJSStreamListener { void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override; void OnStreamDestroy() override { delete this; } - explicit CustomBufferJSListener(uv_buf_t buffer) : buffer_(buffer) {} + CustomBufferJSListener(v8::Local object, + v8::Local view); private: - uv_buf_t buffer_; + void SetBuffer(v8::Local object, + v8::Local view); + + std::shared_ptr backing_store_; + uv_buf_t buffer_ = uv_buf_init(nullptr, 0); + bool using_fallback_buffer_ = false; }; @@ -314,6 +321,7 @@ class StreamBase : public StreamResource { kSlot = BaseObject::kSlot, kStreamBaseField = AsyncWrap::kInternalFieldCount, kOnReadFunctionField, + kUserBufferField, kInternalFieldCount }; diff --git a/test/parallel/test-net-onread-buffer-lifetime.js b/test/parallel/test-net-onread-buffer-lifetime.js new file mode 100644 index 00000000000..4dc48257a73 --- /dev/null +++ b/test/parallel/test-net-onread-buffer-lifetime.js @@ -0,0 +1,157 @@ +// Flags: --js-immutable-arraybuffer +'use strict'; + +// Reads must not modify an immutable ArrayBuffer created by transferring a +// registered onread buffer. + +const common = require('../common'); +const assert = require('node:assert'); +const { once } = require('node:events'); +const net = require('node:net'); +const { test } = require('node:test'); + +for (const mode of ['before first read', 'between reads', 'replacement buffer']) { + test(`onread buffer transferred ${mode}`, async (t) => { + const backing = new ArrayBuffer(8); + new Uint8Array(backing).fill(42); + const buffer = mode === 'replacement buffer' ? + Buffer.from(backing, 2, 4) : new Uint8Array(backing, 2, 4); + const firstBuffer = new Uint8Array(1); + const connections = new Set(); + let peer; + let socket; + let immutable; + let expected; + let factoryCalls = 0; + let readError; + const closed = Promise.withResolvers(); + + const server = net.createServer(common.mustCall((connection) => { + peer = connection; + connection.on('error', (error) => { + if (error.code !== 'ECONNRESET') throw error; + }); + if (mode === 'before first read') { + connection.end('test'); + } else { + connection.write('a'); + } + })); + server.on('connection', (connection) => { + connections.add(connection); + connection.on('close', () => connections.delete(connection)); + }); + t.after(async () => { + socket?.destroy(); + for (const connection of connections) connection.destroy(); + if (server.listening) { + await new Promise((resolve) => server.close(resolve)); + } + }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + + function transfer() { + expected = [...new Uint8Array(backing)]; + immutable = backing.transferToImmutable(); + assert.strictEqual(buffer.byteLength, 0); + assert.strictEqual(immutable.immutable, true); + } + + socket = net.connect({ + host: '127.0.0.1', + port: server.address().port, + onread: { + buffer: mode === 'replacement buffer' ? common.mustCall(() => { + return ++factoryCalls === 1 ? firstBuffer : buffer; + }, 2) : buffer, + callback: common.mustCall((nread, view) => { + try { + if (immutable !== undefined) { + assert.deepStrictEqual([...new Uint8Array(immutable)], expected); + assert.fail('onread called after the buffer was transferred'); + } + assert.strictEqual(nread, 1); + assert.strictEqual(view, mode === 'replacement buffer' ? firstBuffer : buffer); + assert.strictEqual(view[0], 97); + if (mode === 'replacement buffer') { + // The factory's next view is registered after this callback. + setImmediate(common.mustCall(() => { + transfer(); + peer.end('test'); + })); + } else { + transfer(); + peer.end('test'); + } + } catch (error) { + closed.reject(error); + return false; + } + }, mode === 'before first read' ? 0 : 1), + }, + }); + socket.on('data', common.mustNotCall()); + socket.on('end', common.mustNotCall()); + socket.on('error', common.mustCall((error) => { + readError = error; + })); + socket.on('close', common.mustCall(closed.resolve)); + if (mode === 'before first read') transfer(); + assert.strictEqual(await closed.promise, true); + assert.strictEqual(readError.code, 'EINVAL'); + assert.strictEqual(readError.syscall, 'read'); + assert.deepStrictEqual([...new Uint8Array(immutable)], expected); + }); +} + +test('onread factory replaces buffers transferred after each read', async (t) => { + const message = Buffer.from('test'); + const chunks = []; + const closed = Promise.withResolvers(); + let peer; + let socket; + + const server = net.createServer(common.mustCall((connection) => { + peer = connection; + connection.on('error', (error) => { + if (error.code !== 'ECONNRESET') throw error; + }); + connection.end(message); + })); + t.after(async () => { + socket?.destroy(); + peer?.destroy(); + if (server.listening) { + await new Promise((resolve) => server.close(resolve)); + } + }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + + socket = net.connect({ + host: '127.0.0.1', + port: server.address().port, + onread: { + buffer: common.mustCall(() => new Uint8Array(1), message.length + 1), + callback: common.mustCall((nread, view) => { + try { + assert.strictEqual(nread, 1); + const immutable = view.buffer.transferToImmutable(); + assert.strictEqual(view.byteLength, 0); + assert.strictEqual(immutable.immutable, true); + chunks.push(immutable); + } catch (error) { + closed.reject(error); + return false; + } + }, message.length), + }, + }); + socket.on('data', common.mustNotCall()); + socket.on('end', common.mustCall()); + socket.on('error', closed.reject); + socket.on('close', common.mustCall(closed.resolve)); + assert.strictEqual(await closed.promise, false); + assert.deepStrictEqual(chunks.flatMap((chunk) => [...new Uint8Array(chunk)]), [...message]); +}); diff --git a/test/parallel/test-tls-onread-buffer-lifetime.js b/test/parallel/test-tls-onread-buffer-lifetime.js new file mode 100644 index 00000000000..5c54fca2070 --- /dev/null +++ b/test/parallel/test-tls-onread-buffer-lifetime.js @@ -0,0 +1,118 @@ +// Flags: --js-immutable-arraybuffer +'use strict'; + +// TLS reads must not modify an immutable ArrayBuffer created by transferring a +// registered onread buffer. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); + +const fixtures = require('../common/fixtures'); +const assert = require('node:assert'); +const { once } = require('node:events'); +const { test } = require('node:test'); +const tls = require('node:tls'); + +const options = { + key: fixtures.readKey('agent2-key.pem'), + cert: fixtures.readKey('agent2-cert.pem'), +}; + +for (const mode of ['before first read', 'between reads', 'replacement buffer']) { + test(`tls onread buffer transferred ${mode}`, async (t) => { + const backing = new ArrayBuffer(8); + new Uint8Array(backing).fill(42); + const buffer = mode === 'replacement buffer' ? + Buffer.from(backing, 2, 4) : new Uint8Array(backing, 2, 4); + const firstBuffer = new Uint8Array(1); + const connections = new Set(); + let peer; + let socket; + let immutable; + let expected; + let factoryCalls = 0; + let readError; + const closed = Promise.withResolvers(); + + const server = tls.createServer(options, common.mustCall((connection) => { + peer = connection; + connections.add(connection); + connection.on('close', () => connections.delete(connection)); + connection.on('error', (error) => { + if (error.code !== 'ECONNRESET') throw error; + }); + if (mode === 'before first read') { + connection.end('test'); + } else { + connection.write('a'); + } + })); + server.on('connection', (connection) => { + connections.add(connection); + connection.on('close', () => connections.delete(connection)); + }); + t.after(async () => { + socket?.destroy(); + for (const connection of connections) connection.destroy(); + if (server.listening) { + await new Promise((resolve) => server.close(resolve)); + } + }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + + function transfer() { + expected = [...new Uint8Array(backing)]; + immutable = backing.transferToImmutable(); + assert.strictEqual(buffer.byteLength, 0); + assert.strictEqual(immutable.immutable, true); + } + + socket = tls.connect({ + host: '127.0.0.1', + port: server.address().port, + rejectUnauthorized: false, + onread: { + buffer: mode === 'replacement buffer' ? common.mustCall(() => { + return ++factoryCalls === 1 ? firstBuffer : buffer; + }, 2) : buffer, + callback: common.mustCall((nread, view) => { + try { + if (immutable !== undefined) { + assert.deepStrictEqual([...new Uint8Array(immutable)], expected); + assert.fail('onread called after the buffer was transferred'); + } + assert.strictEqual(nread, 1); + assert.strictEqual(view, mode === 'replacement buffer' ? firstBuffer : buffer); + assert.strictEqual(view[0], 97); + if (mode === 'replacement buffer') { + // The factory's next view is registered after this callback. + setImmediate(common.mustCall(() => { + transfer(); + peer.end('test'); + })); + } else { + transfer(); + peer.end('test'); + } + } catch (error) { + closed.reject(error); + return false; + } + }, mode === 'before first read' ? 0 : 1), + }, + }); + socket.on('data', common.mustNotCall()); + socket.on('end', common.mustNotCall()); + socket.on('error', common.mustCall((error) => { + readError = error; + })); + socket.on('close', common.mustCall(closed.resolve)); + if (mode === 'before first read') transfer(); + assert.strictEqual(await closed.promise, true); + assert.strictEqual(readError.code, 'EINVAL'); + assert.strictEqual(readError.syscall, 'read'); + assert.deepStrictEqual([...new Uint8Array(immutable)], expected); + }); +}