tls: fix SslSocket read resumption after readDisable when processing the SSL record that contains the last bytes of the HTTP message

docs/root/version_history/current.rst

@@ -58,6 +58,7 @@ Bug Fixes
 * listener: fixed crash at listener inplace update when connetion load balancer is set.
 * rocketmq_proxy network-level filter: fixed an issue involving incorrect header lengths. In debug mode it causes crash and in release mode it causes underflow.
 * thrift_proxy: fixed crashing bug on request overflow.
+* tls: fix read resumption after triggering buffer high-watermark and all remaining request/response bytes are stored in the SSL connection's internal buffers.
 * udp_proxy: fixed a crash due to UDP packets being processed after listener removal.
 
 Removed Config or Runtime

source/extensions/transport_sockets/tls/ssl_socket.cc

@@ -121,12 +121,13 @@ Network::IoResult SslSocket::doRead(Buffer::Instance& read_buffer) {
   bool end_stream = false;
   PostIoAction action = PostIoAction::KeepOpen;
   uint64_t bytes_read = 0;
+  uint64_t reserve_size = 16384;
   while (keep_reading) {
     // We use 2 slices here so that we can use the remainder of an existing buffer chain element
     // if there is extra space. 16K read is arbitrary and can be tuned later.
     Buffer::RawSlice slices[2];
     uint64_t slices_to_commit = 0;
-    uint64_t num_slices = read_buffer.reserve(16384, slices, 2);
+    uint64_t num_slices = read_buffer.reserve(reserve_size, slices, 2);
     for (uint64_t i = 0; i < num_slices; i++) {
       auto result = sslReadIntoSlice(slices[i]);
       if (result.commit_slice_) {
@@ -157,8 +158,24 @@ Network::IoResult SslSocket::doRead(Buffer::Instance& read_buffer) {
     if (slices_to_commit > 0) {
       read_buffer.commit(slices, slices_to_commit);
       if (callbacks_->shouldDrainReadBuffer()) {
-        callbacks_->setReadBufferReady();
-        keep_reading = false;
+        // Verify that SSL_get_read_ahead is disabled. SSL_pending does not provide an accurate
+        // answer when read-ahead is enabled. As far as we can tell BoringSSL does not implement
+        // read-ahead, so the odds of us ever failing this sanity check are effectively zero.
+        ASSERT(!SSL_get_read_ahead(rawSsl()));
+
+        // Query the SSL implementation for the number of bytes available for immediate read from
+        // internal buffers, and do one last read iteration if bytes are available. This is
+        // important to ensure read resumption works correctly after calls to
+        // Network::Connection::readDisable().
+        int pending_bytes = SSL_pending(rawSsl());
+        ASSERT(pending_bytes < 16 * 1024, "SSL record should be at most 16KB");
+        if (pending_bytes > 0) {
+          ASSERT(keep_reading);
+          reserve_size = pending_bytes;
+        } else {
+          callbacks_->setReadBufferReady();
+          keep_reading = false;
+        }
       }
     }
   }

test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc

@@ -19,6 +19,7 @@
 #include "extensions/transport_sockets/tls/context_config_impl.h"
 #include "extensions/transport_sockets/tls/context_manager_impl.h"
 
+#include "test/integration/autonomous_upstream.h"
 #include "test/integration/integration.h"
 #include "test/integration/utility.h"
 #include "test/test_common/network_utility.h"
@@ -177,6 +178,103 @@ TEST_P(SslIntegrationTest, AdminCertEndpoint) {
   EXPECT_EQ("200", response->headers().getStatusValue());
 }
 
+class RawWriteSslIntegrationTest : public SslIntegrationTest {
+protected:
+  std::unique_ptr<Http::TestRequestHeaderMapImpl>
+  testFragmentedRequestWithBufferLimit(std::list<std::string> request_chunks,
+                                       uint32_t buffer_limit) {
+    autonomous_upstream_ = true;
+    config_helper_.setBufferLimits(buffer_limit, buffer_limit);
+    initialize();
+
+    // write_request_cb will write each of the items in request_chunks as a separate SSL_write.
+    auto write_request_cb = [&request_chunks](Network::ClientConnection& client) {
+      if (!request_chunks.empty()) {
+        Buffer::OwnedImpl buffer(request_chunks.front());
+        client.write(buffer, false);
+        request_chunks.pop_front();
+      }
+    };
+
+    auto client_transport_socket_factory_ptr =
+        createClientSslTransportSocketFactory({}, *context_manager_, *api_);
+    std::string response;
+    auto connection = createConnectionDriver(
+        lookupPort("http"), write_request_cb,
+        [&](Network::ClientConnection&, const Buffer::Instance& data) -> void {
+          response.append(data.toString());
+        },
+        client_transport_socket_factory_ptr->createTransportSocket({}));
+
+    // Drive the connection until we get a response.
+    while (response.empty()) {
+      connection->run(Event::Dispatcher::RunType::NonBlock);
+    }
+    EXPECT_THAT(response, testing::HasSubstr("HTTP/1.1 200 OK\r\n"));
+
+    connection->close();
+    return reinterpret_cast<AutonomousUpstream*>(fake_upstreams_.front().get())
+        ->lastRequestHeaders();
+  }
+};
+
+INSTANTIATE_TEST_SUITE_P(IpVersions, RawWriteSslIntegrationTest,
+                         testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
+                         TestUtility::ipTestParamsToString);
+
+// Regression test for https://github.com/envoyproxy/envoy/issues/12304
+TEST_P(RawWriteSslIntegrationTest, HighWatermarkReadResumptionProcessingHeaders) {
+  // The raw writer will perform a separate SSL_write for each of the chunks below. Chunk sizes were
+  // picked such that the connection's high watermark will trigger while processing the last SSL
+  // record containing the request headers. Verify that read resumption works correctly after
+  // hitting the receive buffer high watermark.
+  std::list<std::string> request_chunks = {
+      "GET / HTTP/1.1\r\nHost: host\r\n",
+      "key1:" + std::string(14000, 'a') + "\r\n",
+      "key2:" + std::string(16000, 'b') + "\r\n\r\n",
+  };
+
+  std::unique_ptr<Http::TestRequestHeaderMapImpl> upstream_headers =
+      testFragmentedRequestWithBufferLimit(request_chunks, 15 * 1024);
+  ASSERT_TRUE(upstream_headers != nullptr);
+  EXPECT_EQ(upstream_headers->Host()->value(), "host");
+  EXPECT_EQ(std::string(14000, 'a'),
+            upstream_headers->get(Envoy::Http::LowerCaseString("key1"))->value().getStringView());
+  EXPECT_EQ(std::string(16000, 'b'),
+            upstream_headers->get(Envoy::Http::LowerCaseString("key2"))->value().getStringView());
+}
+
+// Regression test for https://github.com/envoyproxy/envoy/issues/12304
+TEST_P(RawWriteSslIntegrationTest, HighWatermarkReadResumptionProcesingBody) {
+  // The raw writer will perform a separate SSL_write for each of the chunks below. Chunk sizes were
+  // picked such that the connection's high watermark will trigger while processing the last SSL
+  // record containing the POST body. Verify that read resumption works correctly after hitting the
+  // receive buffer high watermark.
+  std::list<std::string> request_chunks = {
+      "POST / HTTP/1.1\r\nHost: host\r\ncontent-length: 30000\r\n\r\n",
+      std::string(14000, 'a'),
+      std::string(16000, 'a'),
+  };
+
+  std::unique_ptr<Http::TestRequestHeaderMapImpl> upstream_headers =
+      testFragmentedRequestWithBufferLimit(request_chunks, 15 * 1024);
+  ASSERT_TRUE(upstream_headers != nullptr);
+}
+
+// Regression test for https://github.com/envoyproxy/envoy/issues/12304
+TEST_P(RawWriteSslIntegrationTest, HighWatermarkReadResumptionProcesingLargerBody) {
+  std::list<std::string> request_chunks = {
+      "POST / HTTP/1.1\r\nHost: host\r\ncontent-length: 150000\r\n\r\n",
+  };
+  for (int i = 0; i < 10; ++i) {
+    request_chunks.push_back(std::string(15000, 'a'));
+  }
+
+  std::unique_ptr<Http::TestRequestHeaderMapImpl> upstream_headers =
+      testFragmentedRequestWithBufferLimit(request_chunks, 16 * 1024);
+  ASSERT_TRUE(upstream_headers != nullptr);
+}
+
 // Validate certificate selection across different certificate types and client TLS versions.
 class SslCertficateIntegrationTest
     : public testing::TestWithParam<

test/extensions/transport_sockets/tls/integration/ssl_integration_test.h

@@ -36,8 +36,6 @@ class SslIntegrationTestBase : public HttpIntegrationTest {
   // Set this true to debug SSL handshake issues with openssl s_client. The
   // verbose trace will be in the logs, openssl must be installed separately.
   bool debug_with_s_client_{false};
-
-private:
   std::unique_ptr<ContextManager> context_manager_;
 };
 

test/extensions/transport_sockets/tls/ssl_socket_test.cc

@@ -12,6 +12,7 @@
 #include "common/event/dispatcher_impl.h"
 #include "common/json/json_loader.h"
 #include "common/network/address_impl.h"
+#include "common/network/connection_impl.h"
 #include "common/network/listen_socket_impl.h"
 #include "common/network/transport_socket_options_impl.h"
 #include "common/network/utility.h"
@@ -4567,8 +4568,10 @@ class SslReadBufferLimitTest : public SslSocketTest {
     read_filter_ = std::make_shared<Network::MockReadFilter>();
   }
 
-  void readBufferLimitTest(uint32_t read_buffer_limit, uint32_t expected_chunk_size,
-                           uint32_t write_size, uint32_t num_writes, bool reserve_write_space) {
+  void readBufferLimitTest(uint32_t read_buffer_limit, uint32_t min_expected_chunk_size,
+                           uint32_t max_expected_chunk_size, uint32_t write_size,
+                           uint32_t num_writes, bool reserve_write_space,
+                           bool bypass_client_connection) {
     initialize();
 
     EXPECT_CALL(listener_callbacks_, onAccept_(_))
@@ -4592,7 +4595,8 @@ class SslReadBufferLimitTest : public SslSocketTest {
     EXPECT_CALL(*read_filter_, onNewConnection());
     EXPECT_CALL(*read_filter_, onData(_, _))
         .WillRepeatedly(Invoke([&](Buffer::Instance& data, bool) -> Network::FilterStatus {
-          EXPECT_GE(expected_chunk_size, data.length());
+          EXPECT_LE(min_expected_chunk_size, data.length());
+          EXPECT_GE(max_expected_chunk_size, data.length());
           filter_seen += data.length();
           data.drain(data.length());
           if (filter_seen == (write_size * num_writes)) {
@@ -4619,7 +4623,15 @@ class SslReadBufferLimitTest : public SslSocketTest {
         data.commit(iovecs, 2);
       }
 
-      client_connection_->write(data, false);
+      if (bypass_client_connection) {
+        auto result = client_transport_socket_->doWrite(data, false);
+        ASSERT_EQ(Network::PostIoAction::KeepOpen, result.action_);
+        ASSERT_EQ(write_size, result.bytes_processed_);
+        ASSERT_FALSE(result.end_stream_read_);
+      } else {
+        client_connection_->write(data, false);
+        dynamic_cast<Network::ConnectionImpl*>(client_connection_.get())->flushWriteBuffer();
+      }
     }
 
     dispatcher_->run(Event::Dispatcher::RunType::Block);
@@ -4741,17 +4753,23 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, SslReadBufferLimitTest,
                          TestUtility::ipTestParamsToString);
 
 TEST_P(SslReadBufferLimitTest, NoLimit) {
-  readBufferLimitTest(0, 256 * 1024, 256 * 1024, 1, false);
+  readBufferLimitTest(0, 1, 256 * 1024, 256 * 1024, 1, false, false);
 }
 
-TEST_P(SslReadBufferLimitTest, NoLimitReserveSpace) { readBufferLimitTest(0, 512, 512, 1, true); }
+TEST_P(SslReadBufferLimitTest, NoLimitReserveSpace) {
+  readBufferLimitTest(0, 512, 512, 512, 1, true, false);
+}
 
 TEST_P(SslReadBufferLimitTest, NoLimitSmallWrites) {
-  readBufferLimitTest(0, 256 * 1024, 1, 256 * 1024, false);
+  readBufferLimitTest(0, 1, 256 * 1024, 1, 256 * 1024, false, false);
 }
 
 TEST_P(SslReadBufferLimitTest, SomeLimit) {
-  readBufferLimitTest(32 * 1024, 32 * 1024, 256 * 1024, 1, false);
+  readBufferLimitTest(32 * 1024, 32 * 1024, 32 * 1024, 256 * 1024, 1, false, false);
+}
+
+TEST_P(SslReadBufferLimitTest, DrainToSslRecordBoundary) {
+  readBufferLimitTest(16 * 1024, 20 * 1024, 20 * 1024, 10 * 1024, 10, false, true);
 }
 
 TEST_P(SslReadBufferLimitTest, WritesSmallerThanBufferLimit) { singleWriteTest(5 * 1024, 1024); }

test/integration/base_integration_test.h

@@ -289,6 +289,22 @@ class BaseIntegrationTest : protected Logger::Loggable<Logger::Id::testing> {
                                                  *dispatcher_);
   }
 
+  /**
+   * Helper to create ConnectionDriver.
+   *
+   * @param port the port to connect to.
+   * @param write_request_cb callback used to send data.
+   * @param data_callback the callback on the received data.
+   * @param transport_socket transport socket to use for the client connection
+   **/
+  std::unique_ptr<RawConnectionDriver> createConnectionDriver(
+      uint32_t port, RawConnectionDriver::DoWriteCallback write_request_cb,
+      std::function<void(Network::ClientConnection&, const Buffer::Instance&)>&& data_callback,
+      Network::TransportSocketPtr transport_socket = nullptr) {
+    return std::make_unique<RawConnectionDriver>(port, write_request_cb, data_callback, version_,
+                                                 *dispatcher_, std::move(transport_socket));
+  }
+
 protected:
   bool initialized() const { return initialized_; }
 

test/integration/utility.cc

@@ -27,6 +27,21 @@
 #include "absl/strings/match.h"
 
 namespace Envoy {
+namespace {
+
+RawConnectionDriver::DoWriteCallback writeBufferCallback(Buffer::Instance& data) {
+  auto shared_data = std::make_shared<Buffer::OwnedImpl>();
+  shared_data->move(data);
+  return [shared_data](Network::ClientConnection& client) {
+    if (shared_data->length() > 0) {
+      client.write(*shared_data, false);
+      shared_data->drain(shared_data->length());
+    }
+  };
+}
+
+} // namespace
+
 void BufferingStreamDecoder::decodeHeaders(Http::ResponseHeaderMapPtr&& headers, bool end_stream) {
   ASSERT(!complete_);
   complete_ = end_stream;
@@ -112,15 +127,24 @@ IntegrationUtil::makeSingleRequest(uint32_t port, const std::string& method, con
   return makeSingleRequest(addr, method, url, body, type, host, content_type);
 }
 
-RawConnectionDriver::RawConnectionDriver(uint32_t port, Buffer::Instance& initial_data,
-                                         ReadCallback data_callback,
+RawConnectionDriver::RawConnectionDriver(uint32_t port, Buffer::Instance& request_data,
+                                         ReadCallback response_data_callback,
+                                         Network::Address::IpVersion version,
+                                         Event::Dispatcher& dispatcher,
+                                         Network::TransportSocketPtr transport_socket)
+    : RawConnectionDriver(port, writeBufferCallback(request_data), response_data_callback, version,
+                          dispatcher, std::move(transport_socket)) {}
+
+RawConnectionDriver::RawConnectionDriver(uint32_t port, DoWriteCallback write_request_callback,
+                                         ReadCallback response_data_callback,
                                          Network::Address::IpVersion version,
                                          Event::Dispatcher& dispatcher,
                                          Network::TransportSocketPtr transport_socket)
     : dispatcher_(dispatcher) {
   api_ = Api::createApiForTest(stats_store_);
   Event::GlobalTimeSystem time_system;
-  callbacks_ = std::make_unique<ConnectionCallbacks>();
+  callbacks_ = std::make_unique<ConnectionCallbacks>(
+      [this, write_request_callback]() { write_request_callback(*client_); });
 
   if (transport_socket == nullptr) {
     transport_socket = Network::Test::createRawBufferSocket();
@@ -130,9 +154,13 @@ RawConnectionDriver::RawConnectionDriver(uint32_t port, Buffer::Instance& initia
       Network::Utility::resolveUrl(
           fmt::format("tcp://{}:{}", Network::Test::getLoopbackAddressUrlString(version), port)),
       Network::Address::InstanceConstSharedPtr(), std::move(transport_socket), nullptr);
+  // ConnectionCallbacks will call write_request_callback from the connect and low-watermark
+  // callbacks. Set a small buffer limit so high-watermark is triggered after every write and
+  // low-watermark is triggered every time the buffer is drained.
+  client_->setBufferLimits(1);
   client_->addConnectionCallbacks(*callbacks_);
-  client_->addReadFilter(Network::ReadFilterSharedPtr{new ForwardingFilter(*this, data_callback)});
-  client_->write(initial_data, false);
+  client_->addReadFilter(
+      Network::ReadFilterSharedPtr{new ForwardingFilter(*this, response_data_callback)});
   client_->connect();
 }