TcpTunneling: Fix scheme header setting in TcpTunneling with upstream…

… http filters (#38438)

previously attempted to fix in #36711

Issue is that if method is POST and tunnel is tls enabled, `scheme`
header was still being forwarded as `http` instead of https.
In upstreamHttpFilters case, tcppoxy::CombinedStream needs upstream
connection info from Router::UpstreamRequest. Router::UpstreamRequest
fills streamInfo with upstream connection info once connection pool is
ready. After that UpstreamRequest invokes callback
`onUpstreamHostSelected` at tcpproxy::CombinedStream.
Thats when now changes in this PR is updating downstream_headers

Risk Level: low
Testing: integration test added

---------

Signed-off-by: Vikas Choudhary (vikasc) <[email protected]>

source/common/router/upstream_request.cc

@@ -595,8 +595,6 @@ void UpstreamRequest::onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
 
   host->outlierDetector().putResult(Upstream::Outlier::Result::LocalOriginConnectSuccess);
 
-  onUpstreamHostSelected(host, true);
-
   if (protocol) {
     stream_info_.protocol(protocol.value());
   } else {
@@ -627,6 +625,11 @@ void UpstreamRequest::onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
   upstream_info.setUpstreamRemoteAddress(address_provider.remoteAddress());
   upstream_info.setUpstreamSslConnection(info.downstreamAddressProvider().sslConnection());
 
+  // Invoke the onUpstreamHostSelected after setting ssl_connection_info_ in upstream_info.
+  // This is because the onUpstreamHostSelected callback may need to access the ssl_connection_info
+  // to determine the scheme of the upstream connection.
+  onUpstreamHostSelected(host, true);
+
   if (info.downstreamAddressProvider().connectionID().has_value()) {
     upstream_info.setUpstreamConnectionId(info.downstreamAddressProvider().connectionID().value());
   }

source/common/tcp_proxy/upstream.cc

@@ -367,6 +367,7 @@ void HttpConnPool::onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPt
   }
   combined_upstream_->setConnPoolCallbacks(std::make_unique<HttpConnPool::Callbacks>(
       *this, host, downstream_info_.downstreamAddressProvider().sslConnection()));
+  combined_upstream_->recordUpstreamSslConnection();
 }
 
 void HttpConnPool::onPoolReady(Http::RequestEncoder& request_encoder,
@@ -411,17 +412,14 @@ CombinedUpstream::CombinedUpstream(HttpConnPool& http_conn_pool,
     : config_(config), downstream_info_(downstream_info), parent_(http_conn_pool),
       decoder_filter_callbacks_(decoder_callbacks), response_decoder_(*this),
       upstream_callbacks_(callbacks) {
-  auto is_ssl = downstream_info_.downstreamAddressProvider().sslConnection();
+  type_ = parent_.codecType();
   downstream_headers_ = Http::createHeaderMap<Http::RequestHeaderMapImpl>({
       {Http::Headers::get().Method, config_.usePost() ? "POST" : "CONNECT"},
       {Http::Headers::get().Host, config_.host(downstream_info_)},
   });
 
   if (config_.usePost()) {
-    const std::string& scheme =
-        is_ssl ? Http::Headers::get().SchemeValues.Https : Http::Headers::get().SchemeValues.Http;
     downstream_headers_->addReference(Http::Headers::get().Path, config_.postPath());
-    downstream_headers_->addReference(Http::Headers::get().Scheme, scheme);
   }
 
   config_.headerEvaluator().evaluateHeaders(
@@ -470,7 +468,7 @@ CombinedUpstream::onDownstreamEvent(Network::ConnectionEvent event) {
 }
 
 bool CombinedUpstream::isValidResponse(const Http::ResponseHeaderMap& headers) {
-  switch (parent_.codecType()) {
+  switch (type_) {
   case Http::CodecType::HTTP1:
     // According to RFC7231 any 2xx response indicates that the connection is
     // established.
@@ -528,6 +526,18 @@ void CombinedUpstream::onUpstreamTrailers(Http::ResponseTrailerMapPtr&& trailers
 
 Http::RequestHeaderMap* CombinedUpstream::downstreamHeaders() { return downstream_headers_.get(); }
 
+void CombinedUpstream::recordUpstreamSslConnection() {
+  if ((type_ != Http::CodecType::HTTP1) && (config_.usePost())) {
+    auto is_ssl = upstream_request_->streamInfo().upstreamInfo()->upstreamSslConnection();
+    const std::string& scheme =
+        is_ssl ? Http::Headers::get().SchemeValues.Https : Http::Headers::get().SchemeValues.Http;
+    if (downstream_headers_->Scheme()) {
+      downstream_headers_->removeScheme();
+    }
+    downstream_headers_->addReference(Http::Headers::get().Scheme, scheme);
+  }
+}
+
 void CombinedUpstream::doneReading() {
   read_half_closed_ = true;
   if (write_half_closed_) {

source/common/tcp_proxy/upstream.h

@@ -302,6 +302,7 @@ class CombinedUpstream : public GenericUpstream, public Envoy::Router::RouterFil
   void setConnPoolCallbacks(std::unique_ptr<HttpConnPool::Callbacks>&& callbacks) {
     conn_pool_callbacks_ = std::move(callbacks);
   }
+  void recordUpstreamSslConnection();
   void addBytesSentCallback(Network::Connection::BytesSentCb) override{};
   // HTTP upstream must not implement converting upstream transport
   // socket from non-secure to secure mode.
@@ -402,6 +403,7 @@ class CombinedUpstream : public GenericUpstream, public Envoy::Router::RouterFil
   // upstream_request_ has to be destroyed first as they may use CombinedUpstream parent
   // during destruction.
   UpstreamRequestPtr upstream_request_;
+  Http::CodecType type_;
 };
 
 } // namespace TcpProxy

test/integration/tcp_tunneling_integration_test.cc

@@ -1007,6 +1007,45 @@ TEST_P(TcpTunnelingIntegrationTest, UpstreamHttpFiltersPauseAndResume) {
   }
 }
 
+TEST_P(TcpTunnelingIntegrationTest, SchemeHeader) {
+  if (!(GetParam().upstream_protocol == Http::CodecType::HTTP2)) {
+    return;
+  }
+  envoy::extensions::filters::network::tcp_proxy::v3::TcpProxy proxy_config;
+  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void {
+    envoy::extensions::filters::network::tcp_proxy::v3::TcpProxy proxy_config;
+    proxy_config.set_stat_prefix("tcp_stats");
+    proxy_config.set_cluster("cluster_0");
+    proxy_config.mutable_tunneling_config()->set_hostname("foo.lyft.com:80");
+    proxy_config.mutable_tunneling_config()->set_use_post(true);
+
+    auto* listeners = bootstrap.mutable_static_resources()->mutable_listeners();
+    for (auto& listener : *listeners) {
+      if (listener.name() != "tcp_proxy") {
+        continue;
+      }
+      auto* filter_chain = listener.mutable_filter_chains(0);
+      auto* filter = filter_chain->mutable_filters(0);
+      filter->mutable_typed_config()->PackFrom(proxy_config);
+      break;
+    }
+  });
+
+  upstream_tls_ = true;
+  config_helper_.configureUpstreamTls();
+
+  initialize();
+
+  setUpConnection(fake_upstream_connection_);
+  sendBidiData(fake_upstream_connection_);
+  EXPECT_EQ(Http::Headers::get().SchemeValues.Https,
+            upstream_request_->headers()
+                .get(Http::LowerCaseString(Http::Headers::get().Scheme))[0]
+                ->value()
+                .getStringView());
+  closeConnection(fake_upstream_connection_);
+}
+
 TEST_P(TcpTunnelingIntegrationTest, FlowControlOnAndGiantBody) {
   downstream_buffer_limit_ = 1024;
   config_helper_.setBufferLimits(1024, 2024);