[conn_pool] fix use after free in H/1 connection pool

source/common/conn_pool/conn_pool_base.cc

@@ -15,7 +15,11 @@ ConnPoolImplBase::ConnPoolImplBase(
     const Network::TransportSocketOptionsSharedPtr& transport_socket_options,
     Upstream::ClusterConnectivityState& state)
     : state_(state), host_(host), priority_(priority), dispatcher_(dispatcher),
-      socket_options_(options), transport_socket_options_(transport_socket_options) {}
+      socket_options_(options), transport_socket_options_(transport_socket_options),
+      upstream_ready_cb_(dispatcher_.createSchedulableCallback([this]() {
+        upstream_ready_enabled_ = false;
+        onUpstreamReady();
+      })) {}
 
 ConnPoolImplBase::~ConnPoolImplBase() {
   ASSERT(ready_clients_.empty());
@@ -214,6 +218,13 @@ bool ConnPoolImplBase::maybePrefetch(float global_prefetch_ratio) {
   return tryCreateNewConnection(global_prefetch_ratio);
 }
 
+void ConnPoolImplBase::scheduleOnUpstreamReady() {
+  if (!pending_streams_.empty() && !upstream_ready_enabled_) {
+    upstream_ready_enabled_ = true;
+    upstream_ready_cb_->scheduleCallbackCurrentIteration();
+  }
+}
+
 void ConnPoolImplBase::onUpstreamReady() {
   while (!pending_streams_.empty() && !ready_clients_.empty()) {
     ActiveClientPtr& client = ready_clients_.front();

source/common/conn_pool/conn_pool_base.h

@@ -154,7 +154,8 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
                          Network::ConnectionEvent event);
   // See if the drain process has started and/or completed.
   void checkForDrained();
-  void onUpstreamReady();
+  // Schedule onUpstreamReady.
+  void scheduleOnUpstreamReady();
   ConnectionPool::Cancellable* newStream(AttachContext& context);
   // Called if this pool is likely to be picked soon, to determine if it's worth
   // prefetching a connection.
@@ -241,6 +242,10 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   // Clients that are not ready to handle additional streams because they are CONNECTING.
   std::list<ActiveClientPtr> connecting_clients_;
 
+  void onUpstreamReady();
+  Event::SchedulableCallbackPtr upstream_ready_cb_;
+  bool upstream_ready_enabled_{false};
+
 private:
   std::list<PendingStreamPtr> pending_streams_;
 

source/common/http/http1/conn_pool.cc

@@ -76,7 +76,7 @@ void ActiveClient::StreamWrapper::onDecodeComplete() {
     parent_.codec_client_->close();
   } else {
     auto* pool = &parent_.parent();
-    pool->dispatcher().post([pool]() -> void { pool->onUpstreamReady(); });
+    pool->scheduleOnUpstreamReady();
     parent_.stream_wrapper_.reset();
 
     pool->checkForDrained();

test/common/http/http1/conn_pool_test.cc

@@ -52,7 +52,8 @@ class ConnPoolImplForTest : public Event::TestUsingSimulatedTime, public FixedHt
 public:
   ConnPoolImplForTest(Event::MockDispatcher& dispatcher,
                       Upstream::ClusterInfoConstSharedPtr cluster,
-                      Random::RandomGenerator& random_generator)
+                      Random::RandomGenerator& random_generator,
+                      NiceMock<Event::MockSchedulableCallback>* upstream_ready_cb)
       : FixedHttpConnPoolImpl(
             Upstream::makeTestHost(cluster, "tcp://127.0.0.1:9000", dispatcher.timeSource()),
             Upstream::ResourcePriority::Default, dispatcher, nullptr, nullptr, random_generator,
@@ -62,7 +63,8 @@ class ConnPoolImplForTest : public Event::TestUsingSimulatedTime, public FixedHt
               return nullptr; // Not used: createCodecClient overloaded.
             },
             std::vector<Protocol>{Protocol::Http11}),
-        api_(Api::createApiForTest()), mock_dispatcher_(dispatcher) {}
+        api_(Api::createApiForTest()), mock_dispatcher_(dispatcher),
+        mock_upstream_ready_cb_(upstream_ready_cb) {}
 
   ~ConnPoolImplForTest() override {
     EXPECT_EQ(0U, ready_clients_.size());
@@ -118,15 +120,22 @@ class ConnPoolImplForTest : public Event::TestUsingSimulatedTime, public FixedHt
   }
 
   void expectEnableUpstreamReady() {
-    EXPECT_CALL(mock_dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb_));
+    EXPECT_FALSE(upstream_ready_enabled_);
+    EXPECT_CALL(*mock_upstream_ready_cb_, scheduleCallbackCurrentIteration())
+        .Times(1)
+        .RetiresOnSaturation();
   }
 
-  void expectAndRunUpstreamReady() { post_cb_(); }
+  void expectAndRunUpstreamReady() {
+    EXPECT_TRUE(upstream_ready_enabled_);
+    mock_upstream_ready_cb_->invokeCallback();
+    EXPECT_FALSE(upstream_ready_enabled_);
+  }
 
   Upstream::ClusterConnectivityState state_;
   Api::ApiPtr api_;
   Event::MockDispatcher& mock_dispatcher_;
-  Event::PostCb post_cb_;
+  NiceMock<Event::MockSchedulableCallback>* mock_upstream_ready_cb_;
   std::vector<TestCodecClient> test_clients_;
 };
 
@@ -136,7 +145,9 @@ class ConnPoolImplForTest : public Event::TestUsingSimulatedTime, public FixedHt
 class Http1ConnPoolImplTest : public testing::Test {
 public:
   Http1ConnPoolImplTest()
-      : conn_pool_(std::make_unique<ConnPoolImplForTest>(dispatcher_, cluster_, random_)) {}
+      : upstream_ready_cb_(new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)),
+        conn_pool_(std::make_unique<ConnPoolImplForTest>(dispatcher_, cluster_, random_,
+                                                         upstream_ready_cb_)) {}
 
   ~Http1ConnPoolImplTest() override {
     EXPECT_EQ("", TestUtility::nonZeroedGauges(cluster_->stats_store_.gauges()));
@@ -145,6 +156,7 @@ class Http1ConnPoolImplTest : public testing::Test {
   NiceMock<Random::MockRandomGenerator> random_;
   NiceMock<Event::MockDispatcher> dispatcher_;
   std::shared_ptr<Upstream::MockClusterInfo> cluster_{new NiceMock<Upstream::MockClusterInfo>()};
+  NiceMock<Event::MockSchedulableCallback>* upstream_ready_cb_;
   std::unique_ptr<ConnPoolImplForTest> conn_pool_;
   NiceMock<Runtime::MockLoader> runtime_;
 };
@@ -293,7 +305,9 @@ TEST_F(Http1ConnPoolImplTest, VerifyAlpnFallback) {
 
   // Recreate the conn pool so that the host re-evaluates the transport socket match, arriving at
   // our test transport socket factory.
-  conn_pool_ = std::make_unique<ConnPoolImplForTest>(dispatcher_, cluster_, random_);
+  new NiceMock<Event::MockSchedulableCallback>(&dispatcher_);
+  conn_pool_ =
+      std::make_unique<ConnPoolImplForTest>(dispatcher_, cluster_, random_, upstream_ready_cb_);
   NiceMock<MockResponseDecoder> outer_decoder;
   ConnPoolCallbacks callbacks;
   conn_pool_->expectClientCreate(Protocol::Http11);
@@ -646,7 +660,6 @@ TEST_F(Http1ConnPoolImplTest, MaxConnections) {
   inner_decoder->decodeHeaders(std::move(response_headers), true);
 
   conn_pool_->expectAndRunUpstreamReady();
-  conn_pool_->expectEnableUpstreamReady();
   EXPECT_TRUE(
       callbacks2.outer_encoder_
           ->encodeHeaders(TestRequestHeaderMapImpl{{":path", "/"}, {":method", "GET"}}, true)
@@ -716,7 +729,6 @@ TEST_F(Http1ConnPoolImplTest, ConnectionCloseWithoutHeader) {
   EXPECT_CALL(callbacks2.pool_ready_, ready());
   conn_pool_->test_clients_[0].connection_->raiseEvent(Network::ConnectionEvent::Connected);
 
-  conn_pool_->expectEnableUpstreamReady();
   EXPECT_TRUE(
       callbacks2.outer_encoder_
           ->encodeHeaders(TestRequestHeaderMapImpl{{":path", "/"}, {":method", "GET"}}, true)
@@ -978,9 +990,7 @@ TEST_F(Http1ConnPoolImplTest, ConcurrentConnections) {
   r3.startRequest();
   EXPECT_EQ(3U, cluster_->stats_.upstream_rq_total_.value());
 
-  conn_pool_->expectEnableUpstreamReady();
   r2.completeResponse(false);
-  conn_pool_->expectEnableUpstreamReady();
   r3.completeResponse(false);
 
   // Disconnect both clients.
@@ -1125,6 +1135,56 @@ TEST_F(Http1ConnPoolImplTest, PendingRequestIsConsideredActive) {
   EXPECT_EQ(1U, cluster_->stats_.upstream_cx_destroy_local_.value());
 }
 
+class ConnPoolImplNoDestructForTest : public ConnPoolImplForTest {
+public:
+  ConnPoolImplNoDestructForTest(Event::MockDispatcher& dispatcher,
+                                Upstream::ClusterInfoConstSharedPtr cluster,
+                                Random::RandomGenerator& random_generator,
+                                NiceMock<Event::MockSchedulableCallback>* upstream_ready_cb)
+      : ConnPoolImplForTest(dispatcher, cluster, random_generator, upstream_ready_cb) {}
+
+  ~ConnPoolImplNoDestructForTest() override { destructAllConnections(); }
+};
+
+// Regression test for use after free when dispatcher executes onUpstreamReady after connection pool
+// is destroyed.
+TEST_F(Http1ConnPoolImplTest, CbAfterConnPoolDestroyed) {
+  // Recreate conn pool without destructor checks.
+  new NiceMock<Event::MockSchedulableCallback>(&dispatcher_);
+  conn_pool_ = std::make_unique<ConnPoolImplNoDestructForTest>(dispatcher_, cluster_, random_,
+                                                               upstream_ready_cb_);
+
+  InSequence s;
+
+  NiceMock<MockResponseDecoder> outer_decoder;
+  ConnPoolCallbacks callbacks;
+  conn_pool_->expectClientCreate();
+  Http::ConnectionPool::Cancellable* handle = conn_pool_->newStream(outer_decoder, callbacks);
+  EXPECT_NE(nullptr, handle);
+
+  NiceMock<MockRequestEncoder> request_encoder;
+  ResponseDecoder* inner_decoder;
+  EXPECT_CALL(*conn_pool_->test_clients_[0].codec_, newStream(_))
+      .WillOnce(DoAll(SaveArgAddress(&inner_decoder), ReturnRef(request_encoder)));
+  EXPECT_CALL(callbacks.pool_ready_, ready());
+  EXPECT_CALL(*conn_pool_->test_clients_[0].connect_timer_, disableTimer());
+  conn_pool_->test_clients_[0].connection_->raiseEvent(Network::ConnectionEvent::Connected);
+
+  EXPECT_TRUE(
+      callbacks.outer_encoder_
+          ->encodeHeaders(TestRequestHeaderMapImpl{{":path", "/"}, {":method", "GET"}}, true)
+          .ok());
+
+  EXPECT_CALL(*conn_pool_, onClientDestroy());
+  dispatcher_.deferredDelete(std::move(conn_pool_));
+
+  inner_decoder->decodeHeaders(
+      ResponseHeaderMapPtr{new TestResponseHeaderMapImpl{{":status", "200"}}}, true);
+
+  // Remove connection pool.
+  dispatcher_.clearDeferredDeleteList();
+}
+
 } // namespace
 } // namespace Http1
 } // namespace Http

test/common/http/http2/conn_pool_test.cc

@@ -74,6 +74,7 @@ class Http2ConnPoolImplTest : public Event::TestUsingSimulatedTime, public testi
 
   Http2ConnPoolImplTest()
       : api_(Api::createApiForTest(stats_store_)),
+        upstream_ready_cb_(new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)),
         pool_(std::make_unique<TestConnPoolImpl>(dispatcher_, random_, host_,
                                                  Upstream::ResourcePriority::Default, nullptr,
                                                  nullptr, state_)) {
@@ -212,6 +213,7 @@ class Http2ConnPoolImplTest : public Event::TestUsingSimulatedTime, public testi
   NiceMock<Event::MockDispatcher> dispatcher_;
   std::shared_ptr<Upstream::MockClusterInfo> cluster_{new NiceMock<Upstream::MockClusterInfo>()};
   Upstream::HostSharedPtr host_{Upstream::makeTestHost(cluster_, "tcp://127.0.0.1:80", simTime())};
+  NiceMock<Event::MockSchedulableCallback>* upstream_ready_cb_;
   std::unique_ptr<TestConnPoolImpl> pool_;
   std::vector<TestCodecClient> test_clients_;
   NiceMock<Runtime::MockLoader> runtime_;
@@ -337,6 +339,7 @@ TEST_F(Http2ConnPoolImplTest, VerifyAlpnFallback) {
   // Recreate the conn pool so that the host re-evaluates the transport socket match, arriving at
   // our test transport socket factory.
   host_ = Upstream::makeTestHost(cluster_, "tcp://127.0.0.1:80", simTime());
+  new NiceMock<Event::MockSchedulableCallback>(&dispatcher_);
   pool_ = std::make_unique<TestConnPoolImpl>(
       dispatcher_, random_, host_, Upstream::ResourcePriority::Default, nullptr, nullptr, state_);
 

test/common/http/mixed_conn_pool_test.cc

@@ -39,7 +39,8 @@ class ConnPoolImplForTest : public Event::TestUsingSimulatedTime, public HttpCon
 class MixedConnPoolImplTest : public testing::Test {
 public:
   MixedConnPoolImplTest()
-      : conn_pool_(std::make_unique<ConnPoolImplForTest>(dispatcher_, state_, random_, cluster_)) {}
+      : upstream_ready_cb_(new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)),
+        conn_pool_(std::make_unique<ConnPoolImplForTest>(dispatcher_, state_, random_, cluster_)) {}
 
   ~MixedConnPoolImplTest() override {
     EXPECT_EQ("", TestUtility::nonZeroedGauges(cluster_->stats_store_.gauges()));
@@ -48,6 +49,7 @@ class MixedConnPoolImplTest : public testing::Test {
   Upstream::ClusterConnectivityState state_;
   NiceMock<Event::MockDispatcher> dispatcher_;
   std::shared_ptr<Upstream::MockClusterInfo> cluster_{new NiceMock<Upstream::MockClusterInfo>()};
+  NiceMock<Event::MockSchedulableCallback>* upstream_ready_cb_;
   std::unique_ptr<ConnPoolImplForTest> conn_pool_;
   NiceMock<Runtime::MockLoader> runtime_;
   NiceMock<Random::MockRandomGenerator> random_;