diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst index a0d0475909f59..59a9461c08753 100644 --- a/docs/root/version_history/current.rst +++ b/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 diff --git a/source/extensions/transport_sockets/tls/ssl_socket.cc b/source/extensions/transport_sockets/tls/ssl_socket.cc index 4854684430963..bd54be7f932a3 100644 --- a/source/extensions/transport_sockets/tls/ssl_socket.cc +++ b/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; + } } } } diff --git a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc b/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc index db9b0afd9ec58..3543f44434221 100644 --- a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc +++ b/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 + testFragmentedRequestWithBufferLimit(std::list 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(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 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 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 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 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 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 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< diff --git a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.h b/test/extensions/transport_sockets/tls/integration/ssl_integration_test.h index 133e73bd433e9..5af886e851bf3 100644 --- a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.h +++ b/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 context_manager_; }; diff --git a/test/extensions/transport_sockets/tls/ssl_socket_test.cc b/test/extensions/transport_sockets/tls/ssl_socket_test.cc index 92e1fcd98af63..7ef72d33b9dd6 100644 --- a/test/extensions/transport_sockets/tls/ssl_socket_test.cc +++ b/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(); } - 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(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); } diff --git a/test/integration/base_integration_test.h b/test/integration/base_integration_test.h index 299a976c293e8..6ecb2677061de 100644 --- a/test/integration/base_integration_test.h +++ b/test/integration/base_integration_test.h @@ -289,6 +289,22 @@ class BaseIntegrationTest : protected Logger::Loggable { *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 createConnectionDriver( + uint32_t port, RawConnectionDriver::DoWriteCallback write_request_cb, + std::function&& data_callback, + Network::TransportSocketPtr transport_socket = nullptr) { + return std::make_unique(port, write_request_cb, data_callback, version_, + *dispatcher_, std::move(transport_socket)); + } + protected: bool initialized() const { return initialized_; } diff --git a/test/integration/utility.cc b/test/integration/utility.cc index a0e93000cb653..b4808ff0a6ff2 100644 --- a/test/integration/utility.cc +++ b/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(); + 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(); + callbacks_ = std::make_unique( + [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(); } diff --git a/test/integration/utility.h b/test/integration/utility.h index 6ff69ad27a831..497fe872472b4 100644 --- a/test/integration/utility.h +++ b/test/integration/utility.h @@ -63,10 +63,17 @@ using BufferingStreamDecoderPtr = std::unique_ptr; */ class RawConnectionDriver { public: + using DoWriteCallback = std::function; using ReadCallback = std::function; - RawConnectionDriver(uint32_t port, Buffer::Instance& initial_data, ReadCallback data_callback, - Network::Address::IpVersion version, Event::Dispatcher& dispatcher, + RawConnectionDriver(uint32_t port, DoWriteCallback write_request_callback, + ReadCallback response_data_callback, Network::Address::IpVersion version, + Event::Dispatcher& dispatcher, + Network::TransportSocketPtr transport_socket = nullptr); + // Similar to the constructor above but accepts the request as a constructor argument. + RawConnectionDriver(uint32_t port, Buffer::Instance& request_data, + ReadCallback response_data_callback, Network::Address::IpVersion version, + Event::Dispatcher& dispatcher, Network::TransportSocketPtr transport_socket = nullptr); ~RawConnectionDriver(); const Network::Connection& connection() { return *client_; } @@ -83,37 +90,44 @@ class RawConnectionDriver { private: struct ForwardingFilter : public Network::ReadFilterBaseImpl { ForwardingFilter(RawConnectionDriver& parent, ReadCallback cb) - : parent_(parent), data_callback_(cb) {} + : parent_(parent), response_data_callback_(cb) {} // Network::ReadFilter Network::FilterStatus onData(Buffer::Instance& data, bool) override { - data_callback_(*parent_.client_, data); + response_data_callback_(*parent_.client_, data); data.drain(data.length()); return Network::FilterStatus::StopIteration; } RawConnectionDriver& parent_; - ReadCallback data_callback_; + ReadCallback response_data_callback_; }; struct ConnectionCallbacks : public Network::ConnectionCallbacks { + using WriteCb = std::function; + ConnectionCallbacks(WriteCb write_cb) : write_cb_(write_cb) {} bool connected() const { return connected_; } bool closed() const { return closed_; } // Network::ConnectionCallbacks void onEvent(Network::ConnectionEvent event) override { + if (!connected_ && event == Network::ConnectionEvent::Connected) { + write_cb_(); + } + last_connection_event_ = event; closed_ |= (event == Network::ConnectionEvent::RemoteClose || event == Network::ConnectionEvent::LocalClose); connected_ |= (event == Network::ConnectionEvent::Connected); } void onAboveWriteBufferHighWatermark() override {} - void onBelowWriteBufferLowWatermark() override {} + void onBelowWriteBufferLowWatermark() override { write_cb_(); } Network::ConnectionEvent last_connection_event_; private: + WriteCb write_cb_; bool connected_{false}; bool closed_{false}; };