quic: Fix use after free in next

Author: martenrichter

src/dataqueue/queue.cc

@@ -36,6 +36,7 @@ class NonIdempotentDataQueueReader;
 class EntryImpl : public DataQueue::Entry {
  public:
   virtual std::shared_ptr<DataQueue::Reader> get_reader() = 0;
+  virtual void clearPendingNext() = 0;
 };
 
 class DataQueueImpl final : public DataQueue,
@@ -159,6 +160,12 @@ class DataQueueImpl final : public DataQueue,
     return std::nullopt;
   }
 
+  void clearPendingNext() override {
+     for (const auto& entry : entries_) {
+      entry->clearPendingNext();
+     }
+  }
+
   void MemoryInfo(node::MemoryTracker* tracker) const override {
     tracker->TrackField(
         "entries", entries_, "std::vector<std::unique_ptr<Entry>>");
@@ -590,6 +597,10 @@ class EmptyEntry final : public EntryImpl {
     return std::make_unique<EmptyEntry>();
   }
 
+  void clearPendingNext() override {
+    // this is a noop for an empty object
+  }
+
   std::optional<uint64_t> size() const override { return 0; }
 
   bool is_idempotent() const override { return true; }
@@ -702,6 +713,10 @@ class InMemoryEntry final : public EntryImpl {
     return makeEntry(start, byte_length_ - start);
   }
 
+  void clearPendingNext() override {
+    // this is a noop for an object, that immediately calls next.
+  }
+
   std::optional<uint64_t> size() const override { return byte_length_; }
 
   bool is_idempotent() const override { return true; }
@@ -750,6 +765,11 @@ class DataQueueEntry : public EntryImpl {
     return std::make_unique<DataQueueEntry>(std::move(sliced));
   }
 
+  void clearPendingNext() override {
+    // this is just calls clearPendingNext of the queue
+    data_queue_->clearPendingNext();
+  }
+
   // Returns the number of bytes represented by this Entry if it is
   // known. Certain types of entries, such as those backed by streams
   // might not know the size in advance and therefore cannot provide
@@ -868,6 +888,10 @@ class FdEntry final : public EntryImpl {
     return std::make_unique<FdEntry>(env_, path_, stat_, new_start, new_end);
   }
 
+  void clearPendingNext() override {
+    clearPendingNextImpl();
+  }
+
   std::optional<uint64_t> size() const override { return end_ - start_; }
 
   bool is_idempotent() const override { return true; }
@@ -885,6 +909,16 @@ class FdEntry final : public EntryImpl {
   uint64_t start_ = 0;
   uint64_t end_ = 0;
 
+  class ReaderImpl;
+  std::unordered_set<ReaderImpl*> readers_;
+
+  void clearPendingNextImpl() {
+    // this should clear all pending pulls from the FD reader
+    for (ReaderImpl* p : readers_) {
+      p->ClearAllPendingPulls();
+    }
+  }
+
   bool is_modified(const uv_stat_t& other) {
     return other.st_size != stat_.st_size ||
            other.st_mtim.tv_nsec != stat_.st_mtim.tv_nsec;
@@ -932,12 +966,14 @@ class FdEntry final : public EntryImpl {
         : env_(handle->env()), handle_(std::move(handle)), entry_(entry) {
       handle_->PushStreamListener(this);
       handle_->env()->AddCleanupHook(cleanup, this);
+      entry_->readers_.insert(this);
     }
 
     ~ReaderImpl() override {
       handle_->env()->RemoveCleanupHook(cleanup, this);
       DrainAndClose();
       handle_->RemoveStreamListener(this);
+      entry_->readers_.erase(this);
     }
 
     uv_buf_t OnStreamAlloc(size_t suggested_size) override {
@@ -1062,6 +1098,10 @@ class FdEntry final : public EntryImpl {
       return std::move(pending_pulls_.front());
     }
 
+    void ClearAllPendingPulls() {
+      pending_pulls_.clear();
+    }
+
     friend class FdEntry;
   };
 
@@ -1094,6 +1134,10 @@ class FeederEntry final : public EntryImpl {
   }
 
   bool is_idempotent() const override { return false; }
+ 
+  void clearPendingNext() override {
+    if (feeder_) feeder_->clearPendingNext();
+  }
 
   SET_NO_MEMORY_INFO()
   SET_MEMORY_INFO_NAME(FeederEntry)

src/dataqueue/queue.h

@@ -190,6 +190,10 @@ class DataQueue : public MemoryRetainer {
     // idempotent and cannot preserve that quality, subsequent reads
     // must fail with an error when a variance is detected.
     virtual bool is_idempotent() const = 0;
+
+    // calls if required readers to invalidate next function's
+    // which may hold captures to stream and session objects
+    virtual void clearPendingNext() = 0;
   };
 
   // Creates an idempotent DataQueue with a pre-established collection
@@ -299,6 +303,10 @@ class DataQueue : public MemoryRetainer {
   // been set, maybeCapRemaining() will return std::nullopt.
   virtual std::optional<uint64_t> maybeCapRemaining() const = 0;
 
+  // calls all entries and readers to invalidate next function's
+  // which may hold captures to stream and session objects
+  virtual void clearPendingNext() = 0;
+
   // BackpressureListeners only work on non-idempotent DataQueues.
   virtual void addBackpressureListener(BackpressureListener* listener) = 0;
   virtual void removeBackpressureListener(BackpressureListener* listener) = 0;

src/quic/streams.cc

@@ -402,6 +402,13 @@ class Stream::Outbound final : public MemoryRetainer {
         queue_(std::move(queue)),
         reader_(queue_->get_reader()) {}
 
+  ~Outbound() {
+    // we need to clear all pending next's from the queue, as it holds pointers
+    // to Stream and Session, which will be invalidated and may cause use after
+    // free otherwise
+    queue_->clearPendingNext();
+  }
+
   void Acknowledge(size_t amount) {
     size_t remaining = std::min(amount, total_ - uncommitted_);
     while (remaining > 0 && head_ != nullptr) {

src/quic/streams.h

@@ -433,6 +433,10 @@ class DataQueueFeeder final : public AsyncWrap {
     }
   }
 
+  void clearPendingNext() {
+    pendingPulls_.clear();
+  }
+
   struct PendingPull {
     Next next;
     explicit PendingPull(Next next) : next(std::move(next)) {}