net: validate onread buffers before each read

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<Value>& args) {
 int StreamBase::UseUserBuffer(const FunctionCallbackInfo<Value>& 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<ArrayBufferView>()));
   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> object,
+                                               Local<ArrayBufferView> view) {
+  SetBuffer(object, view);
+}
+
+void CustomBufferJSListener::SetBuffer(Local<Object> object,
+                                       Local<ArrayBufferView> 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<StreamBase*>(stream_);
+  Environment* env = stream->stream_env();
+  Isolate* isolate = env->isolate();
+  HandleScope handle_scope(isolate);
+
+  Local<ArrayBufferView> view =
+      stream->GetAsyncWrap()
+          ->object()
+          ->GetInternalField(StreamBase::kUserBufferField)
+          .As<ArrayBufferView>();
+  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<char*>(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<ArrayBuffer>());
+    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<Value> 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<ArrayBufferView>());
   }
 }
 

src/stream_base.h

@@ -11,6 +11,7 @@
 #include "v8.h"
 
 #include <climits>  // INT_MIN
+#include <memory>
 
 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<v8::Object> object,
+                         v8::Local<v8::ArrayBufferView> view);
 
  private:
-  uv_buf_t buffer_;
+  void SetBuffer(v8::Local<v8::Object> object,
+                 v8::Local<v8::ArrayBufferView> view);
+
+  std::shared_ptr<v8::BackingStore> 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
   };
 

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]);
+});