http2: pause input processing if sending output

If we are waiting for the ability to send more output, we should not
process more input. This commit a) makes us send output earlier,
during processing of input, if we accumulate a lot and b) allows
interrupting the call into nghttp2 that processes input data
and resuming it at a later time, if we do find ourselves in a position
where we are waiting to be able to send more output.

This is part of mitigating CVE-2019-9511/CVE-2019-9517.

PR-URL: https://github.com/nodejs/node/pull/29122
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

1 files changed, 87 insertions, 41 deletions

diff --git a/src/node_http2.cc b/src/node_http2.cc
index 0fec129373..7e32f7f14b 100644
--- a/src/node_http2.cc
+++ b/src/node_http2.cc
@@ -868,31 +868,51 @@ ssize_t Http2Session::OnCallbackPadding(size_t frameLen,
 // various callback functions. Each of these will typically result in a call
 // out to JavaScript so this particular function is rather hot and can be
 // quite expensive. This is a potential performance optimization target later.
-ssize_t Http2Session::Write(const uv_buf_t* bufs, size_t nbufs) {
-  size_t total = 0;
-  // Note that nghttp2_session_mem_recv is a synchronous operation that
-  // will trigger a number of other callbacks. Those will, in turn have
+ssize_t Http2Session::ConsumeHTTP2Data() {
+  CHECK_NOT_NULL(stream_buf_.base);
+  CHECK_LT(stream_buf_offset_, stream_buf_.len);
+  size_t read_len = stream_buf_.len - stream_buf_offset_;
+
   // multiple side effects.
-  for (size_t n = 0; n < nbufs; n++) {
-    Debug(this, "receiving %d bytes [wants data? %d]",
-          bufs[n].len,
-          nghttp2_session_want_read(session_));
-    ssize_t ret =
-      nghttp2_session_mem_recv(session_,
-                               reinterpret_cast<uint8_t*>(bufs[n].base),
-                               bufs[n].len);
-    CHECK_NE(ret, NGHTTP2_ERR_NOMEM);
-
-    if (ret < 0)
-      return ret;
+  Debug(this, "receiving %d bytes [wants data? %d]",
+        read_len,
+        nghttp2_session_want_read(session_));
+  flags_ &= ~SESSION_STATE_NGHTTP2_RECV_PAUSED;
+  ssize_t ret =
+    nghttp2_session_mem_recv(session_,
+                             reinterpret_cast<uint8_t*>(stream_buf_.base) +
+                                 stream_buf_offset_,
+                             read_len);
+  CHECK_NE(ret, NGHTTP2_ERR_NOMEM);
 
-    total += ret;
+  if (flags_ & SESSION_STATE_NGHTTP2_RECV_PAUSED) {
+    CHECK_NE(flags_ & SESSION_STATE_READING_STOPPED, 0);
+
+    CHECK_GT(ret, 0);
+    CHECK_LE(static_cast<size_t>(ret), read_len);
+
+    if (static_cast<size_t>(ret) < read_len) {
+      // Mark the remainder of the data as available for later consumption.
+      stream_buf_offset_ += ret;
+      return ret;
+    }
   }
+
+  // We are done processing the current input chunk.
+  DecrementCurrentSessionMemory(stream_buf_.len);
+  stream_buf_offset_ = 0;
+  stream_buf_ab_.Reset();
+  stream_buf_allocation_.clear();
+  stream_buf_ = uv_buf_init(nullptr, 0);
+
+  if (ret < 0)
+    return ret;
+
   // Send any data that was queued up while processing the received data.
   if (!IsDestroyed()) {
     SendPendingData();
   }
-  return total;
+  return ret;
 }
 
 
@@ -1194,8 +1214,18 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle,
       nghttp2_session_consume_stream(handle, id, avail);
     else
       stream->inbound_consumed_data_while_paused_ += avail;
+
+    // If we have a gathered a lot of data for output, try sending it now.
+    if (session->outgoing_length_ > 4096) session->SendPendingData();
   } while (len != 0);
 
+  // If we are currently waiting for a write operation to finish, we should
+  // tell nghttp2 that we want to wait before we process more input data.
+  if (session->flags_ & SESSION_STATE_WRITE_IN_PROGRESS) {
+    session->flags_ |= SESSION_STATE_NGHTTP2_RECV_PAUSED;
+    return NGHTTP2_ERR_PAUSE;
+  }
+
   return 0;
 }
 
@@ -1283,6 +1313,7 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   size_t offset = buf.base - session->stream_buf_.base;
 
   // Verify that the data offset is inside the current read buffer.
+  CHECK_GE(offset, session->stream_buf_offset_);
   CHECK_LE(offset, session->stream_buf_.len);
   CHECK_LE(offset + buf.len, session->stream_buf_.len);
 
@@ -1554,6 +1585,11 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
     stream_->ReadStart();
   }
 
+  // If there is more incoming data queued up, consume it.
+  if (stream_buf_offset_ > 0) {
+    ConsumeHTTP2Data();
+  }
+
   if (!(flags_ & SESSION_STATE_WRITE_SCHEDULED)) {
     // Schedule a new write if nghttp2 wants to send data.
     MaybeScheduleWrite();
@@ -1609,6 +1645,7 @@ void Http2Session::ClearOutgoing(int status) {
 
   if (outgoing_buffers_.size() > 0) {
     outgoing_storage_.clear();
+    outgoing_length_ = 0;
 
     std::vector<nghttp2_stream_write> current_outgoing_buffers_;
     current_outgoing_buffers_.swap(outgoing_buffers_);
@@ -1639,6 +1676,11 @@ void Http2Session::ClearOutgoing(int status) {
   }
 }
 
+void Http2Session::PushOutgoingBuffer(nghttp2_stream_write&& write) {
+  outgoing_length_ += write.buf.len;
+  outgoing_buffers_.emplace_back(std::move(write));
+}
+
 // Queue a given block of data for sending. This always creates a copy,
 // so it is used for the cases in which nghttp2 requests sending of a
 // small chunk of data.
@@ -1651,7 +1693,7 @@ void Http2Session::CopyDataIntoOutgoing(const uint8_t* src, size_t src_length) {
   // of the outgoing_buffers_ vector may invalidate the pointer.
   // The correct base pointers will be set later, before writing to the
   // underlying socket.
-  outgoing_buffers_.emplace_back(nghttp2_stream_write {
+  PushOutgoingBuffer(nghttp2_stream_write {
     uv_buf_init(nullptr, src_length)
   });
 }
@@ -1774,13 +1816,13 @@ int Http2Session::OnSendData(
     if (write.buf.len <= length) {
       // This write does not suffice by itself, so we can consume it completely.
       length -= write.buf.len;
-      session->outgoing_buffers_.emplace_back(std::move(write));
+      session->PushOutgoingBuffer(std::move(write));
       stream->queue_.pop();
       continue;
     }
 
     // Slice off `length` bytes of the first write in the queue.
-    session->outgoing_buffers_.emplace_back(nghttp2_stream_write {
+    session->PushOutgoingBuffer(nghttp2_stream_write {
       uv_buf_init(write.buf.base, length)
     });
     write.buf.base += length;
@@ -1790,7 +1832,7 @@ int Http2Session::OnSendData(
 
   if (frame->data.padlen > 0) {
     // Send padding if that was requested.
-    session->outgoing_buffers_.emplace_back(nghttp2_stream_write {
+    session->PushOutgoingBuffer(nghttp2_stream_write {
       uv_buf_init(const_cast<char*>(zero_bytes_256), frame->data.padlen - 1)
     });
   }
@@ -1827,8 +1869,6 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   Http2Scope h2scope(this);
   CHECK_NOT_NULL(stream_);
   Debug(this, "receiving %d bytes", nread);
-  CHECK_EQ(stream_buf_allocation_.size(), 0);
-  CHECK(stream_buf_ab_.IsEmpty());
   AllocatedBuffer buf(env(), buf_);
 
   // Only pass data on if nread > 0
@@ -1839,24 +1879,31 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
     return;
   }
 
-  // Shrink to the actual amount of used data.
-  buf.Resize(nread);
+  statistics_.data_received += nread;
 
-  IncrementCurrentSessionMemory(nread);
-  OnScopeLeave on_scope_leave([&]() {
-    // Once finished handling this write, reset the stream buffer.
-    // The memory has either been free()d or was handed over to V8.
-    // We use `nread` instead of `buf.size()` here, because the buffer is
-    // cleared as part of the `.ToArrayBuffer()` call below.
-    DecrementCurrentSessionMemory(nread);
+  if (UNLIKELY(stream_buf_offset_ > 0)) {
+    // This is a very unlikely case, and should only happen if the ReadStart()
+    // call in OnStreamAfterWrite() immediately provides data. If that does
+    // happen, we concatenate the data we received with the already-stored
+    // pending input data, slicing off the already processed part.
+    AllocatedBuffer new_buf = env()->AllocateManaged(
+        stream_buf_.len - stream_buf_offset_ + nread);
+    memcpy(new_buf.data(),
+           stream_buf_.base + stream_buf_offset_,
+           stream_buf_.len - stream_buf_offset_);
+    memcpy(new_buf.data() + stream_buf_.len - stream_buf_offset_,
+           buf.data(),
+           nread);
+    buf = std::move(new_buf);
+    nread = buf.size();
+    stream_buf_offset_ = 0;
     stream_buf_ab_.Reset();
-    stream_buf_allocation_.clear();
-    stream_buf_ = uv_buf_init(nullptr, 0);
-  });
+    DecrementCurrentSessionMemory(stream_buf_offset_);
+  }
 
-  // Make sure that there was no read previously active.
-  CHECK_NULL(stream_buf_.base);
-  CHECK_EQ(stream_buf_.len, 0);
+  // Shrink to the actual amount of used data.
+  buf.Resize(nread);
+  IncrementCurrentSessionMemory(nread);
 
   // Remember the current buffer, so that OnDataChunkReceived knows the
   // offset of a DATA frame's data into the socket read buffer.
@@ -1869,8 +1916,7 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   // to copy memory.
   stream_buf_allocation_ = std::move(buf);
 
-  statistics_.data_received += nread;
-  ssize_t ret = Write(&stream_buf_, 1);
+  ssize_t ret = ConsumeHTTP2Data();
 
   if (UNLIKELY(ret < 0)) {
     Debug(this, "fatal error receiving data: %d", ret);