upstream: force h1 for websocket upgrades when allow_connect is false on auto_config clusters

changelogs/current.yaml

@@ -321,6 +321,11 @@ bug_fixes:
   change: |
     Fixed an issue where a userspace IO socket could fail to drain the write buffer of a connection
     when the peer had closed for reads, causing the connection to remain open indefinitely.
+- area: websockets
+  change: |
+    Fixed a bug where WebSocket connections would be upgraded to HTTP/2 extended CONNECT even
+    when the upstream http/2 protocol options were not set to allow extended CONNECT when the
+    cluster used auto_config for protocol selection.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

source/common/upstream/cluster_manager_impl.cc

@@ -33,6 +33,7 @@
 #include "source/common/http/http1/conn_pool.h"
 #include "source/common/http/http2/conn_pool.h"
 #include "source/common/http/mixed_conn_pool.h"
+#include "source/common/http/utility.h"
 #include "source/common/network/utility.h"
 #include "source/common/protobuf/utility.h"
 #include "source/common/router/shadow_writer_impl.h"
@@ -1969,6 +1970,17 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::httpConnPoolImp
   // we end up on a connection of the correct protocol, but for simplicity we're
   // starting with something simpler.
   auto upstream_protocols = host->cluster().upstreamHttpProtocol(downstream_protocol);
+
+  // For auto_config (ALPN) clusters: if this is a WebSocket upgrade and HTTP/2 extended
+  // CONNECT is disabled, force HTTP/1.1 to avoid the upgrade-to-CONNECT transformation
+  // that would otherwise occur in the HTTP/2 codec.
+  if ((host->cluster().features() & Upstream::ClusterInfo::Features::USE_ALPN) && context &&
+      context->downstreamHeaders() &&
+      Http::Utility::isWebSocketUpgradeRequest(*context->downstreamHeaders()) &&
+      !host->cluster().httpProtocolOptions().http2Options().allow_connect()) {
+    upstream_protocols = {Http::Protocol::Http11};
+  }
+
   std::vector<uint8_t> hash_key;
   hash_key.reserve(upstream_protocols.size());
   for (auto protocol : upstream_protocols) {

test/common/upstream/cluster_manager_impl_test.cc

@@ -2703,6 +2703,169 @@ TEST_F(ClusterManagerImplTest, LocalInterfaceNameForUpstreamConnectionThrowsInWi
 }
 #endif
 
+TEST_F(ClusterManagerImplTest, AlpnClusterWebSocketForcesHttp11WhenAllowConnectDisabled) {
+  AlpnTestConfigFactory alpn_factory;
+  Registry::InjectFactory<Server::Configuration::UpstreamTransportSocketConfigFactory>
+      registered_factory(alpn_factory);
+
+  const std::string yaml = R"EOF(
+  static_resources:
+    clusters:
+    - name: alpn_cluster
+      connect_timeout: 0.250s
+      lb_policy: ROUND_ROBIN
+      load_assignment:
+        cluster_name: alpn_cluster
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                socket_address:
+                  address: 127.0.0.1
+                  port_value: 11001
+      typed_extension_protocol_options:
+        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+          "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+          auto_config:
+            http2_protocol_options: {}
+            http_protocol_options: {}
+      transport_socket:
+        name: envoy.transport_sockets.alpn
+        typed_config:
+          "@type": type.googleapis.com/google.protobuf.Struct
+  )EOF";
+  create(parseBootstrapFromV3Yaml(yaml));
+
+  // Verify the cluster has USE_ALPN.
+  auto* tlc = cluster_manager_->getThreadLocalCluster("alpn_cluster");
+  ASSERT_NE(nullptr, tlc);
+  auto host = tlc->chooseHost(nullptr).host;
+  ASSERT_NE(nullptr, host);
+  EXPECT_NE(0, host->cluster().features() & Upstream::ClusterInfo::Features::USE_ALPN);
+
+  // Set up a WebSocket upgrade context.
+  NiceMock<MockLoadBalancerContext> context;
+  Http::TestRequestHeaderMapImpl headers{{"connection", "upgrade"}, {"upgrade", "websocket"}};
+  EXPECT_CALL(context, downstreamHeaders()).WillRepeatedly(Return(&headers));
+
+  Http::ConnectionPool::MockInstance* pool = new NiceMock<Http::ConnectionPool::MockInstance>();
+  EXPECT_CALL(factory_, allocateConnPool_(_, _, _, _, _, _, _)).WillOnce(Return(pool));
+
+  auto opt_cp = tlc->httpConnPool(host, ResourcePriority::Default, Http::Protocol::Http2, &context);
+  EXPECT_TRUE(opt_cp.has_value());
+
+  // The protocol should have been forced to HTTP/1.1.
+  ASSERT_EQ(1, factory_.last_protocols_.size());
+  EXPECT_EQ(Http::Protocol::Http11, factory_.last_protocols_[0]);
+}
+
+TEST_F(ClusterManagerImplTest, AlpnClusterWebSocketPreservesProtocolWhenAllowConnectEnabled) {
+  AlpnTestConfigFactory alpn_factory;
+  Registry::InjectFactory<Server::Configuration::UpstreamTransportSocketConfigFactory>
+      registered_factory(alpn_factory);
+
+  const std::string yaml = R"EOF(
+  static_resources:
+    clusters:
+    - name: alpn_cluster
+      connect_timeout: 0.250s
+      lb_policy: ROUND_ROBIN
+      load_assignment:
+        cluster_name: alpn_cluster
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                socket_address:
+                  address: 127.0.0.1
+                  port_value: 11001
+      typed_extension_protocol_options:
+        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+          "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+          auto_config:
+            http2_protocol_options:
+              allow_connect: true
+            http_protocol_options: {}
+      transport_socket:
+        name: envoy.transport_sockets.alpn
+        typed_config:
+          "@type": type.googleapis.com/google.protobuf.Struct
+  )EOF";
+  create(parseBootstrapFromV3Yaml(yaml));
+
+  auto* tlc = cluster_manager_->getThreadLocalCluster("alpn_cluster");
+  ASSERT_NE(nullptr, tlc);
+  auto host = tlc->chooseHost(nullptr).host;
+  ASSERT_NE(nullptr, host);
+
+  NiceMock<MockLoadBalancerContext> context;
+  Http::TestRequestHeaderMapImpl headers{{"connection", "upgrade"}, {"upgrade", "websocket"}};
+  EXPECT_CALL(context, downstreamHeaders()).WillRepeatedly(Return(&headers));
+
+  Http::ConnectionPool::MockInstance* pool = new NiceMock<Http::ConnectionPool::MockInstance>();
+  EXPECT_CALL(factory_, allocateConnPool_(_, _, _, _, _, _, _)).WillOnce(Return(pool));
+
+  auto opt_cp = tlc->httpConnPool(host, ResourcePriority::Default, Http::Protocol::Http2, &context);
+  EXPECT_TRUE(opt_cp.has_value());
+
+  // With allow_connect enabled, protocols should NOT be forced to HTTP/1.1 only.
+  // The ALPN cluster should preserve its normal protocol set (HTTP/1.1 + HTTP/2).
+  EXPECT_GT(factory_.last_protocols_.size(), 1);
+}
+
+TEST_F(ClusterManagerImplTest, AlpnClusterNonWebSocketDoesNotForceHttp11) {
+  AlpnTestConfigFactory alpn_factory;
+  Registry::InjectFactory<Server::Configuration::UpstreamTransportSocketConfigFactory>
+      registered_factory(alpn_factory);
+
+  const std::string yaml = R"EOF(
+  static_resources:
+    clusters:
+    - name: alpn_cluster
+      connect_timeout: 0.250s
+      lb_policy: ROUND_ROBIN
+      load_assignment:
+        cluster_name: alpn_cluster
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                socket_address:
+                  address: 127.0.0.1
+                  port_value: 11001
+      typed_extension_protocol_options:
+        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+          "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+          auto_config:
+            http2_protocol_options: {}
+            http_protocol_options: {}
+      transport_socket:
+        name: envoy.transport_sockets.alpn
+        typed_config:
+          "@type": type.googleapis.com/google.protobuf.Struct
+  )EOF";
+  create(parseBootstrapFromV3Yaml(yaml));
+
+  auto* tlc = cluster_manager_->getThreadLocalCluster("alpn_cluster");
+  ASSERT_NE(nullptr, tlc);
+  auto host = tlc->chooseHost(nullptr).host;
+  ASSERT_NE(nullptr, host);
+
+  // Non-WebSocket request headers.
+  NiceMock<MockLoadBalancerContext> context;
+  Http::TestRequestHeaderMapImpl headers{{"content-type", "application/json"}};
+  EXPECT_CALL(context, downstreamHeaders()).WillRepeatedly(Return(&headers));
+
+  Http::ConnectionPool::MockInstance* pool = new NiceMock<Http::ConnectionPool::MockInstance>();
+  EXPECT_CALL(factory_, allocateConnPool_(_, _, _, _, _, _, _)).WillOnce(Return(pool));
+
+  auto opt_cp = tlc->httpConnPool(host, ResourcePriority::Default, Http::Protocol::Http2, &context);
+  EXPECT_TRUE(opt_cp.has_value());
+
+  // Non-WebSocket request should preserve the normal ALPN protocol set.
+  EXPECT_GT(factory_.last_protocols_.size(), 1);
+}
+
 } // namespace
 } // namespace Upstream
 } // namespace Envoy

test/common/upstream/test_cluster_manager.h

@@ -88,18 +88,23 @@ class TestClusterManagerFactory : public ClusterManagerFactory {
   ~TestClusterManagerFactory() override { server_context_.dispatcher_.to_delete_.clear(); }
 
   Http::ConnectionPool::InstancePtr allocateConnPool(
-      Event::Dispatcher&, HostConstSharedPtr host, ResourcePriority, std::vector<Http::Protocol>&,
+      Event::Dispatcher&, HostConstSharedPtr host, ResourcePriority,
+      std::vector<Http::Protocol>& protocols,
       const absl::optional<envoy::config::core::v3::AlternateProtocolsCacheOptions>&
           alternate_protocol_options,
       const Network::ConnectionSocket::OptionsSharedPtr& options,
       const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options, TimeSource&,
       ClusterConnectivityState& state, Http::PersistentQuicInfoPtr& /*quic_info*/,
       OptRef<Quic::EnvoyQuicNetworkObserverRegistry> network_observer_registry) override {
+    last_protocols_ = protocols;
     return Http::ConnectionPool::InstancePtr{
         allocateConnPool_(host, alternate_protocol_options, options, transport_socket_options,
                           state, network_observer_registry, server_context_.overloadManager())};
   }
 
+  // Protocols passed to the most recent allocateConnPool call.
+  std::vector<Http::Protocol> last_protocols_;
+
   Tcp::ConnectionPool::InstancePtr
   allocateTcpConnPool(Event::Dispatcher&, HostConstSharedPtr host, ResourcePriority,
                       const Network::ConnectionSocket::OptionsSharedPtr&,