reverse_tunnels: Fix some flakes in the reverse connection cluster integration test

docs/root/_configs/reverse_connection/responder-envoy.yaml

@@ -26,6 +26,7 @@ static_resources:
           "@type": >-
             type.googleapis.com/envoy.extensions.filters.network.reverse_tunnel.v3.ReverseTunnel
           ping_interval: 2s
+          auto_close_connections: true
 
   # Listener that will route the downstream request to the reverse connection cluster
   - name: egress_listener

source/extensions/bootstrap/reverse_tunnel/common/BUILD

@@ -21,6 +21,7 @@ envoy_cc_extension(
         "//source/common/common:logger_lib",
         "//source/common/http:header_map_lib",
         "//source/common/http:headers_lib",
+        "//source/common/tls:ssl_handshaker_lib",
     ],
 )
 

source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.cc

@@ -3,8 +3,10 @@
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/common/assert.h"
 #include "source/common/common/logger.h"
+#include "source/common/tls/ssl_handshaker.h"
 
 #include "absl/strings/str_cat.h"
+#include "openssl/ssl.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -93,6 +95,25 @@ bool PingMessageHandler::processPingMessage(absl::string_view data,
   return false;
 }
 
+void ReverseConnectionUtility::applySslQuietClose(Network::Connection& connection) {
+  auto ssl_conn = connection.ssl();
+
+  if (ssl_conn) {
+    ENVOY_CONN_LOG(
+        trace,
+        "reverse_tunnel: Setting quiet shutdown on SSL connection to prevent close_notify alert",
+        connection);
+    const auto* ssl_handshaker =
+        dynamic_cast<const Extensions::TransportSockets::Tls::SslHandshakerImpl*>(ssl_conn.get());
+    if (ssl_handshaker && ssl_handshaker->ssl()) {
+      SSL_set_quiet_shutdown(ssl_handshaker->ssl(), 1);
+      ENVOY_CONN_LOG(trace, "reverse_tunnel: Quiet shutdown enabled for connection", connection);
+    } else {
+      ENVOY_LOG(warn, "reverse_tunnel: Failed to cast to SslHandshakerImpl or ssl() returned null");
+    }
+  }
+}
+
 } // namespace ReverseConnection
 } // namespace Bootstrap
 } // namespace Extensions

source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h

@@ -49,6 +49,8 @@ class ReverseConnectionUtility : public Logger::Loggable<Logger::Id::connection>
   static std::string buildTenantScopedIdentifier(absl::string_view tenant,
                                                  absl::string_view identifier);
 
+  static void applySslQuietClose(Network::Connection& conn);
+
 private:
   ReverseConnectionUtility() = delete;
 };

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_connection_io_handle.cc

@@ -1174,22 +1174,7 @@ void ReverseConnectionIOHandle::onConnectionDone(const std::string& error,
 
     // Set quiet shutdown since we are duplicating the socket and closing the original socket. When
     // the original socket is closed, a TLS close_notify alert is otherwise sent.
-    if (connection->ssl()) {
-      ENVOY_LOG(
-          trace,
-          "reverse_tunnel: Setting quiet shutdown on SSL connection to prevent close_notify alert");
-      const Extensions::TransportSockets::Tls::SslHandshakerImpl* ssl_handshaker =
-          dynamic_cast<const Extensions::TransportSockets::Tls::SslHandshakerImpl*>(
-              connection->ssl().get());
-      if (ssl_handshaker && ssl_handshaker->ssl()) {
-        SSL_set_quiet_shutdown(ssl_handshaker->ssl(), 1);
-        ENVOY_LOG(trace, "reverse_tunnel: Quiet shutdown enabled for connection {}",
-                  connection_key);
-      } else {
-        ENVOY_LOG(warn,
-                  "reverse_tunnel: Failed to cast to SslHandshakerImpl or ssl() returned null");
-      }
-    }
+    ReverseConnectionUtility::applySslQuietClose(*connection);
 
     Network::ClientConnectionPtr released_conn = wrapper->releaseConnection();
 

source/extensions/filters/network/reverse_tunnel/reverse_tunnel_filter.cc

@@ -463,7 +463,9 @@ void ReverseTunnelFilter::RequestDecoderImpl::processIfComplete(bool end_stream)
 
   // Close the connection if configured to do so after handling the request.
   if (parent_.config_->autoCloseConnections()) {
-    parent_.read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite);
+    auto& connection = parent_.read_callbacks_->connection();
+    Bootstrap::ReverseConnection::ReverseConnectionUtility::applySslQuietClose(connection);
+    connection.close(Network::ConnectionCloseType::FlushWrite);
   }
 }
 

test/common/tls/BUILD

@@ -217,6 +217,14 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "mock_ssl_handshaker_lib",
+    hdrs = ["mock_ssl_handshaker.h"],
+    deps = [
+        "//source/common/tls:ssl_handshaker_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "test_private_key_method_provider_test_lib",
     srcs = [

test/common/tls/mock_ssl_handshaker.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include "source/common/tls/ssl_handshaker.h"
+
+#include "openssl/ssl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace TransportSockets {
+namespace Tls {
+
+/**
+ * Test helper that subclasses the real TLS handshaker implementation so dynamic casts in production
+ * code succeed.
+ */
+class MockSslHandshakerImpl : public SslHandshakerImpl {
+public:
+  explicit MockSslHandshakerImpl(SSL* ssl)
+      : SslHandshakerImpl(bssl::UniquePtr<SSL>(ssl), 0, nullptr), mock_ssl_(ssl) {}
+
+  SSL* ssl() const override { return mock_ssl_; }
+
+private:
+  SSL* mock_ssl_{nullptr};
+};
+
+} // namespace Tls
+} // namespace TransportSockets
+} // namespace Extensions
+} // namespace Envoy

test/extensions/bootstrap/reverse_tunnel/common/BUILD

@@ -16,7 +16,10 @@ envoy_cc_test(
         "//source/common/buffer:buffer_lib",
         "//source/common/network:connection_lib",
         "//source/extensions/bootstrap/reverse_tunnel/common:reverse_connection_utility_lib",
+        "//test/common/tls:mock_ssl_handshaker_lib",
         "//test/mocks/network:network_mocks",
+        "//test/mocks/ssl:ssl_mocks",
+        "//test/test_common:logging_lib",
         "//test/test_common:test_runtime_lib",
     ],
 )

test/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility_test.cc

@@ -2,12 +2,16 @@
 #include "source/common/network/connection_impl.h"
 #include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
 
+#include "test/common/tls/mock_ssl_handshaker.h"
 #include "test/mocks/network/mocks.h"
+#include "test/mocks/ssl/mocks.h"
+#include "test/test_common/logging.h"
 #include "test/test_common/test_runtime.h"
 
 #include "absl/strings/str_cat.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
+#include "openssl/ssl.h"
 
 using testing::_;
 using testing::NiceMock;
@@ -18,6 +22,8 @@ namespace Extensions {
 namespace Bootstrap {
 namespace ReverseConnection {
 
+using TransportSockets::Tls::MockSslHandshakerImpl;
+
 class ReverseConnectionUtilityTest : public testing::Test {
 protected:
   ReverseConnectionUtilityTest() = default;
@@ -267,6 +273,30 @@ TEST_F(ReverseConnectionUtilityTest, BuildTenantScopedIdentifierWithoutTenant) {
   EXPECT_EQ(composite, "node-1");
 }
 
+TEST_F(ReverseConnectionUtilityTest, ApplySslQuietCloseWithoutSsl) {
+  NiceMock<Network::MockConnection> connection;
+  EXPECT_CALL(connection, ssl()).WillOnce(Return(nullptr));
+
+  ReverseConnectionUtility::applySslQuietClose(connection);
+}
+
+TEST_F(ReverseConnectionUtilityTest, ApplySslQuietCloseOnValidSslHandshaker) {
+  NiceMock<Network::MockConnection> connection;
+
+  bssl::UniquePtr<SSL_CTX> ctx(SSL_CTX_new(TLS_method()));
+  ASSERT_NE(ctx, nullptr);
+  SSL* ssl = SSL_new(ctx.get());
+  ASSERT_NE(ssl, nullptr);
+  auto mock_ssl_handshaker = std::make_shared<MockSslHandshakerImpl>(ssl);
+
+  EXPECT_CALL(connection, ssl()).WillOnce(Return(mock_ssl_handshaker));
+  EXPECT_EQ(0, SSL_get_quiet_shutdown(ssl));
+
+  ReverseConnectionUtility::applySslQuietClose(connection);
+
+  EXPECT_EQ(1, SSL_get_quiet_shutdown(ssl));
+}
+
 } // namespace ReverseConnection
 } // namespace Bootstrap
 } // namespace Extensions

test/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/BUILD

@@ -58,6 +58,7 @@ envoy_cc_test(
         "//source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface:reverse_connection_io_handle_lib",
         "//source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface:reverse_tunnel_extension_lib",
         "//source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface:reverse_tunnel_initiator_lib",
+        "//test/common/tls:mock_ssl_handshaker_lib",
         "//test/mocks/api:api_mocks",
         "//test/mocks/event:event_mocks",
         "//test/mocks/server:factory_context_mocks",

test/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_connection_io_handle_test.cc

@@ -9,12 +9,12 @@
 
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/network/address_impl.h"
-#include "source/common/tls/ssl_handshaker.h"
 #include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_connection_io_handle.h"
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_tunnel_initiator.h"
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_tunnel_initiator_extension.h"
 
+#include "test/common/tls/mock_ssl_handshaker.h"
 #include "test/mocks/api/mocks.h"
 #include "test/mocks/event/mocks.h"
 #include "test/mocks/server/factory_context.h"
@@ -39,21 +39,7 @@ namespace Extensions {
 namespace Bootstrap {
 namespace ReverseConnection {
 
-// Mock SslHandshakerImpl for testing SSL quiet shutdown functionality.
-// This extends the real SslHandshakerImpl so dynamic_cast will succeed.
-class MockSslHandshakerImpl : public Extensions::TransportSockets::Tls::SslHandshakerImpl {
-public:
-  // Constructor that takes an SSL object to pass to the base class.
-  explicit MockSslHandshakerImpl(SSL* ssl)
-      : Extensions::TransportSockets::Tls::SslHandshakerImpl(bssl::UniquePtr<SSL>(ssl), 0, nullptr),
-        mock_ssl_(ssl) {}
-
-  // Override ssl() to return our mock SSL pointer.
-  SSL* ssl() const override { return mock_ssl_; }
-
-private:
-  SSL* mock_ssl_{nullptr};
-};
+using TransportSockets::Tls::MockSslHandshakerImpl;
 
 // ReverseConnectionIOHandle Test Class.
 
@@ -3056,7 +3042,7 @@ TEST_F(ReverseConnectionIOHandleTest, OnConnectionDoneTlsConnectionQuietShutdown
   auto mock_ssl_handshaker = std::make_shared<MockSslHandshakerImpl>(mock_ssl);
 
   // Mock ssl() to return MockSslHandshakerImpl.
-  EXPECT_CALL(*mock_connection, ssl()).WillRepeatedly(Return(mock_ssl_handshaker));
+  EXPECT_CALL(*mock_connection, ssl()).WillOnce(Return(mock_ssl_handshaker));
 
   Upstream::MockHost::MockCreateConnectionData success_conn_data;
   success_conn_data.connection_ = mock_connection.get();

test/extensions/clusters/reverse_connection/reverse_connection_cluster_integration_test.cc

@@ -151,7 +151,7 @@ name: envoy.bootstrap.reverse_tunnel.downstream_socket_interface
     // Configure the reverse tunnel filter.
     envoy::extensions::filters::network::reverse_tunnel::v3::ReverseTunnel rt_config;
     rt_config.mutable_ping_interval()->set_seconds(60);
-    rt_config.set_auto_close_connections(false);
+    rt_config.set_auto_close_connections(true);
     rt_config.set_request_path("/reverse_connections/request");
     rt_config.set_request_method(envoy::config::core::v3::GET);
     rt_filter->mutable_typed_config()->PackFrom(rt_config);
@@ -251,6 +251,7 @@ name: envoy.bootstrap.reverse_tunnel.downstream_socket_interface
     // using the rc:// address format.
     auto* init_listener = bootstrap.mutable_static_resources()->add_listeners();
     init_listener->set_name("reverse_conn_listener");
+    init_listener->set_stat_prefix("reverse_conn_listener");
     init_listener->mutable_listener_filters_timeout()->set_seconds(0);
 
     // Use rc:// address format to encode reverse connection metadata.
@@ -344,6 +345,9 @@ TEST_P(ReverseConnectionClusterIntegrationTest, EndToEndReverseTunnelTest) {
   // Wait for reverse tunnel to establish.
   test_server_->waitForCounterGe("reverse_tunnel.handshake.accepted", 1,
                                  std::chrono::milliseconds(5000));
+  // Wait for the listener to accept a downstream connection.
+  test_server_->waitForCounterGe("listener.reverse_conn_listener.downstream_cx_total", 1,
+                                 std::chrono::milliseconds(5000));
 
   // Verify reverse tunnel stats.
   test_server_->waitForGaugeGe("reverse_tunnel_acceptor.nodes.test-node-id", 1);
@@ -599,6 +603,8 @@ TEST_P(ReverseConnectionClusterIntegrationTest, EndToEndReverseTunnelTestWithMut
   // Wait for reverse tunnel to establish with mTLS.
   test_server_->waitForCounterGe("reverse_tunnel.handshake.accepted", 1,
                                  std::chrono::milliseconds(5000));
+  test_server_->waitForCounterGe("listener.reverse_conn_listener.downstream_cx_total", 1,
+                                 std::chrono::milliseconds(5000));
 
   // Verify reverse tunnel stats.
   test_server_->waitForGaugeGe("reverse_tunnel_acceptor.nodes.test-node-id", 1);
@@ -729,6 +735,7 @@ TEST_P(ReverseConnectionClusterIntegrationTest, ReverseTunnelResiliencyTest) {
     auto build_initiator_listener = [&](envoy::config::listener::v3::Listener& listener, int node,
                                         int cloud) {
       listener.set_name(fmt::format("node_{}_to_cloud_{}", node, cloud));
+      listener.set_stat_prefix("reverse_conn_listener");
       listener.mutable_listener_filters_timeout()->set_seconds(0);
       listener.set_drain_type(envoy::config::listener::v3::Listener::DEFAULT);
 
@@ -786,7 +793,7 @@ TEST_P(ReverseConnectionClusterIntegrationTest, ReverseTunnelResiliencyTest) {
 
       envoy::extensions::filters::network::reverse_tunnel::v3::ReverseTunnel rt_config;
       rt_config.mutable_ping_interval()->set_seconds(60);
-      rt_config.set_auto_close_connections(false);
+      rt_config.set_auto_close_connections(true);
       rt_config.set_request_path("/reverse_connections/request");
       rt_config.set_request_method(envoy::config::core::v3::GET);
       rt_filter->mutable_typed_config()->PackFrom(rt_config);
@@ -925,6 +932,8 @@ name: envoy.bootstrap.reverse_tunnel.downstream_socket_interface
   // Wait for all 4 tunnels (2 nodes x 2 clouds).
   test_server_->waitForCounterGe("reverse_tunnel.handshake.accepted", 4,
                                  std::chrono::milliseconds(10000));
+  test_server_->waitForCounterGe("listener.reverse_conn_listener.downstream_cx_total", 4,
+                                 std::chrono::milliseconds(5000));
 
   test_server_->waitForGaugeGe("reverse_tunnel_acceptor.nodes.node-1", 2);
   test_server_->waitForGaugeGe("reverse_tunnel_acceptor.nodes.node-2", 2);
@@ -1022,6 +1031,8 @@ name: envoy.bootstrap.reverse_tunnel.downstream_socket_interface
   ENVOY_LOG_MISC(info, "Waiting for node-1 tunnels to re-establish.");
   test_server_->waitForCounterGe("reverse_tunnel.handshake.accepted", 6,
                                  std::chrono::milliseconds(10000)); // 4 initial + 2 reconnect
+  test_server_->waitForCounterGe("listener.reverse_conn_listener.downstream_cx_total", 6,
+                                 std::chrono::milliseconds(5000));
 
   test_server_->waitForGaugeGe("reverse_tunnel_acceptor.nodes.node-1", 2);
   test_server_->waitForGaugeEq("reverse_tunnel_acceptor.nodes.node-2", 2);
@@ -1127,6 +1138,8 @@ TEST_P(ReverseConnectionClusterIntegrationTest, MultiWorkerEndToEndReverseTunnel
   // Each of the 4 workers should establish 1 connection, so we expect 4 total handshakes.
   test_server_->waitForCounterGe("reverse_tunnel.handshake.accepted", 4,
                                  std::chrono::milliseconds(10000));
+  test_server_->waitForCounterGe("listener.reverse_conn_listener.downstream_cx_total", 4,
+                                 std::chrono::milliseconds(5000));
 
   // Verify total node connections. Since all workers use the same node-id (test-node-id),
   // the acceptor should show 4 connections from the same logical node.

test/extensions/filters/network/reverse_tunnel/BUILD

@@ -34,6 +34,7 @@ envoy_extension_cc_test(
         "//source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface:reverse_tunnel_acceptor_lib",
         "//source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface:upstream_socket_manager_lib",
         "//source/extensions/filters/network/reverse_tunnel:reverse_tunnel_filter_lib",
+        "//test/common/tls:mock_ssl_handshaker_lib",
         "//test/mocks/event:event_mocks",
         "//test/mocks/network:network_mocks",
         "//test/mocks/reverse_tunnel_reporting_service:reporter_mocks",