tcp tunneling: optionally move response header to downstream info

api/envoy/extensions/filters/network/tcp_proxy/v3/tcp_proxy.proto

@@ -99,6 +99,10 @@ message TcpProxy {
     // Neither ``:-prefixed`` pseudo-headers nor the Host: header can be overridden.
     repeated config.core.v3.HeaderValueOption headers_to_add = 3
         [(validate.rules).repeated = {max_items: 1000}];
+
+    // Save the response headers to the downstream info filter state for consumption
+    // by the network filters. The filter state key is ``envoy.tcp_proxy.propagate_response_headers``.
+    bool propagate_response_headers = 4;
   }
 
   message OnDemand {

changelogs/current.yaml

@@ -223,6 +223,11 @@ new_features:
 - area: redis
   change: |
     added support for quit command to the redis proxy.
+- area: tcp_proxy
+  change: |
+    added support for propagating the response headers in :ref:`TunnelingConfig
+    <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.TunnelingConfig.propagate_response_headers>` to
+    the downstream info filter state.
 - area: access_log
   change: |
     log ``duration``, ``upstream_request_attempt_count``, ``connection_termination_details`` and tls ``ja3`` field in the grpc access log

envoy/tcp/upstream.h

@@ -35,6 +35,11 @@ class TunnelingConfigHelper {
 
   // The evaluator to add additional HTTP request headers to the upstream request.
   virtual Envoy::Http::HeaderEvaluator& headerEvaluator() const PURE;
+
+  // Save HTTP response headers to the downstream filter state.
+  virtual void
+  propagateResponseHeaders(Http::ResponseHeaderMapPtr&& headers,
+                           const StreamInfo::FilterStateSharedPtr& filter_state) const PURE;
 };
 
 using TunnelingConfigHelperOptConstRef = OptRef<const TunnelingConfigHelper>;
@@ -142,13 +147,15 @@ class GenericConnPoolFactory : public Envoy::Config::TypedFactory {
    * @param config the tunneling config, if doing connect tunneling.
    * @param context the load balancing context for this connection.
    * @param upstream_callbacks the callbacks to provide to the connection if successfully created.
+   * @param downstream_info is the downstream connection stream info.
    * @return may be null if there is no cluster with the given name.
    */
   virtual GenericConnPoolPtr
   createGenericConnPool(Upstream::ThreadLocalCluster& thread_local_cluster,
                         TunnelingConfigHelperOptConstRef config,
                         Upstream::LoadBalancerContext* context,
-                        Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks) const PURE;
+                        Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks,
+                        StreamInfo::StreamInfo& downstream_info) const PURE;
 };
 
 using GenericConnPoolFactoryPtr = std::unique_ptr<GenericConnPoolFactory>;

source/common/tcp_proxy/BUILD

@@ -78,6 +78,7 @@ envoy_cc_library(
         "//source/common/upstream:load_balancer_lib",
         "//source/extensions/upstreams/tcp/generic:config",
         "@envoy_api//envoy/config/accesslog/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/filters/network/tcp_proxy/v3:pkg_cc_proto",
     ],
 )

source/common/tcp_proxy/tcp_proxy.cc

@@ -6,6 +6,7 @@
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/config/accesslog/v3/accesslog.pb.h"
+#include "envoy/config/core/v3/base.pb.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/event/timer.h"
 #include "envoy/extensions/filters/network/tcp_proxy/v3/tcp_proxy.pb.h"
@@ -468,7 +469,7 @@ bool Filter::maybeTunnel(Upstream::ThreadLocalCluster& cluster) {
   }
 
   generic_conn_pool_ = factory->createGenericConnPool(cluster, config_->tunnelingConfigHelper(),
-                                                      this, *upstream_callbacks_);
+                                                      this, *upstream_callbacks_, getStreamInfo());
   if (generic_conn_pool_) {
     connecting_ = true;
     connect_attempts_++;
@@ -542,12 +543,28 @@ const Router::MetadataMatchCriteria* Filter::metadataMatchCriteria() {
   }
 }
 
+ProtobufTypes::MessagePtr TunnelResponseHeaders::serializeAsProto() const {
+  auto proto_out = std::make_unique<envoy::config::core::v3::HeaderMap>();
+  response_headers_->iterate([&proto_out](const Http::HeaderEntry& e) -> Http::HeaderMap::Iterate {
+    auto* new_header = proto_out->add_headers();
+    new_header->set_key(std::string(e.key().getStringView()));
+    new_header->set_value(std::string(e.value().getStringView()));
+    return Http::HeaderMap::Iterate::Continue;
+  });
+  return proto_out;
+}
+
+const std::string& TunnelResponseHeaders::key() {
+  CONSTRUCT_ON_FIRST_USE(std::string, "envoy.tcp_proxy.propagate_response_headers");
+}
+
 TunnelingConfigHelperImpl::TunnelingConfigHelperImpl(
     const envoy::extensions::filters::network::tcp_proxy::v3::TcpProxy_TunnelingConfig&
         config_message,
     Server::Configuration::FactoryContext& context)
     : use_post_(config_message.use_post()),
-      header_parser_(Envoy::Router::HeaderParser::configure(config_message.headers_to_add())) {
+      header_parser_(Envoy::Router::HeaderParser::configure(config_message.headers_to_add())),
+      propagate_response_headers_(config_message.propagate_response_headers()) {
   envoy::config::core::v3::SubstitutionFormatString substitution_format_config;
   substitution_format_config.mutable_text_format_source()->set_inline_string(
       config_message.hostname());
@@ -562,6 +579,17 @@ std::string TunnelingConfigHelperImpl::host(const StreamInfo::StreamInfo& stream
                                absl::string_view());
 }
 
+void TunnelingConfigHelperImpl::propagateResponseHeaders(
+    Http::ResponseHeaderMapPtr&& headers,
+    const StreamInfo::FilterStateSharedPtr& filter_state) const {
+  if (!propagate_response_headers_) {
+    return;
+  }
+  filter_state->setData(
+      TunnelResponseHeaders::key(), std::make_shared<TunnelResponseHeaders>(std::move(headers)),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::Connection);
+}
+
 void Filter::onConnectTimeout() {
   ENVOY_CONN_LOG(debug, "connect timeout", read_callbacks_->connection());
   read_callbacks_->upstreamHost()->outlierDetector().putResult(

source/common/tcp_proxy/tcp_proxy.h

@@ -112,7 +112,23 @@ using RouteConstSharedPtr = std::shared_ptr<const Route>;
 using TunnelingConfig =
     envoy::extensions::filters::network::tcp_proxy::v3::TcpProxy_TunnelingConfig;
 
-class TunnelingConfigHelperImpl : public TunnelingConfigHelper {
+/**
+ * Response headers for the tunneling connections.
+ */
+class TunnelResponseHeaders : public StreamInfo::FilterState::Object {
+public:
+  TunnelResponseHeaders(Http::ResponseHeaderMapPtr&& response_headers)
+      : response_headers_(std::move(response_headers)) {}
+  const Http::ResponseHeaderMap& value() const { return *response_headers_; }
+  ProtobufTypes::MessagePtr serializeAsProto() const override;
+  static const std::string& key();
+
+private:
+  const Http::ResponseHeaderMapPtr response_headers_;
+};
+
+class TunnelingConfigHelperImpl : public TunnelingConfigHelper,
+                                  protected Logger::Loggable<Logger::Id::filter> {
 public:
   TunnelingConfigHelperImpl(
       const envoy::extensions::filters::network::tcp_proxy::v3::TcpProxy_TunnelingConfig&
@@ -121,11 +137,15 @@ class TunnelingConfigHelperImpl : public TunnelingConfigHelper {
   std::string host(const StreamInfo::StreamInfo& stream_info) const override;
   bool usePost() const override { return use_post_; }
   Envoy::Http::HeaderEvaluator& headerEvaluator() const override { return *header_parser_; }
+  void
+  propagateResponseHeaders(Http::ResponseHeaderMapPtr&& headers,
+                           const StreamInfo::FilterStateSharedPtr& filter_state) const override;
 
 private:
   const bool use_post_;
   std::unique_ptr<Envoy::Router::HeaderParser> header_parser_;
   Formatter::FormatterPtr hostname_fmt_;
+  const bool propagate_response_headers_;
 };
 
 /**

source/common/tcp_proxy/upstream.cc

@@ -59,7 +59,7 @@ TcpUpstream::onDownstreamEvent(Network::ConnectionEvent event) {
 
 HttpUpstream::HttpUpstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
                            const TunnelingConfigHelper& config,
-                           const StreamInfo::StreamInfo& downstream_info)
+                           StreamInfo::StreamInfo& downstream_info)
     : config_(config), downstream_info_(downstream_info), response_decoder_(*this),
       upstream_callbacks_(callbacks) {}
 
@@ -197,9 +197,9 @@ HttpConnPool::HttpConnPool(Upstream::ThreadLocalCluster& thread_local_cluster,
                            Upstream::LoadBalancerContext* context,
                            const TunnelingConfigHelper& config,
                            Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks,
-                           Http::CodecType type)
+                           Http::CodecType type, StreamInfo::StreamInfo& downstream_info)
     : config_(config), type_(type), upstream_callbacks_(upstream_callbacks),
-      downstream_info_(context->downstreamConnection()->streamInfo()) {
+      downstream_info_(downstream_info) {
   absl::optional<Http::Protocol> protocol;
   if (type_ == Http::CodecType::HTTP3) {
     protocol = Http::Protocol::Http3;
@@ -259,7 +259,7 @@ void HttpConnPool::onGenericPoolReady(Upstream::HostDescriptionConstSharedPtr& h
 
 Http2Upstream::Http2Upstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
                              const TunnelingConfigHelper& config,
-                             const StreamInfo::StreamInfo& downstream_info)
+                             StreamInfo::StreamInfo& downstream_info)
     : HttpUpstream(callbacks, config, downstream_info) {}
 
 bool Http2Upstream::isValidResponse(const Http::ResponseHeaderMap& headers) {
@@ -298,7 +298,7 @@ void Http2Upstream::setRequestEncoder(Http::RequestEncoder& request_encoder, boo
 
 Http1Upstream::Http1Upstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
                              const TunnelingConfigHelper& config,
-                             const StreamInfo::StreamInfo& downstream_info)
+                             StreamInfo::StreamInfo& downstream_info)
     : HttpUpstream(callbacks, config, downstream_info) {}
 
 void Http1Upstream::setRequestEncoder(Http::RequestEncoder& request_encoder, bool) {

source/common/tcp_proxy/upstream.h

@@ -47,7 +47,8 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
 public:
   HttpConnPool(Upstream::ThreadLocalCluster& thread_local_cluster,
                Upstream::LoadBalancerContext* context, const TunnelingConfigHelper& config,
-               Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks, Http::CodecType type);
+               Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks, Http::CodecType type,
+               StreamInfo::StreamInfo& downstream_info);
   ~HttpConnPool() override;
 
   bool valid() const { return conn_pool_data_.has_value(); }
@@ -100,7 +101,7 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
   GenericConnectionPoolCallbacks* callbacks_{};
   Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks_;
   std::unique_ptr<HttpUpstream> upstream_;
-  const StreamInfo::StreamInfo& downstream_info_;
+  StreamInfo::StreamInfo& downstream_info_;
 };
 
 class TcpUpstream : public GenericUpstream {
@@ -149,15 +150,15 @@ class HttpUpstream : public GenericUpstream, protected Http::StreamCallbacks {
 
 protected:
   HttpUpstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
-               const TunnelingConfigHelper& config, const StreamInfo::StreamInfo& downstream_info);
+               const TunnelingConfigHelper& config, StreamInfo::StreamInfo& downstream_info);
   void resetEncoder(Network::ConnectionEvent event, bool inform_downstream = true);
 
   // The encoder offered by the upstream http client.
   Http::RequestEncoder* request_encoder_{};
   // The config object that is owned by the downstream network filter chain factory.
   const TunnelingConfigHelper& config_;
   // The downstream info that is owned by the downstream connection.
-  const StreamInfo::StreamInfo& downstream_info_;
+  StreamInfo::StreamInfo& downstream_info_;
 
 private:
   class DecoderShim : public Http::ResponseDecoder {
@@ -169,6 +170,8 @@ class HttpUpstream : public GenericUpstream, protected Http::StreamCallbacks {
       if (!parent_.isValidResponse(*headers) || end_stream) {
         parent_.resetEncoder(Network::ConnectionEvent::LocalClose);
       } else if (parent_.conn_pool_callbacks_ != nullptr) {
+        parent_.config_.propagateResponseHeaders(std::move(headers),
+                                                 parent_.downstream_info_.filterState());
         parent_.conn_pool_callbacks_->onSuccess(*parent_.request_encoder_);
         parent_.conn_pool_callbacks_.reset();
       }
@@ -201,7 +204,7 @@ class HttpUpstream : public GenericUpstream, protected Http::StreamCallbacks {
 class Http1Upstream : public HttpUpstream {
 public:
   Http1Upstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
-                const TunnelingConfigHelper& config, const StreamInfo::StreamInfo& downstream_info);
+                const TunnelingConfigHelper& config, StreamInfo::StreamInfo& downstream_info);
 
   void encodeData(Buffer::Instance& data, bool end_stream) override;
   void setRequestEncoder(Http::RequestEncoder& request_encoder, bool is_ssl) override;
@@ -211,7 +214,7 @@ class Http1Upstream : public HttpUpstream {
 class Http2Upstream : public HttpUpstream {
 public:
   Http2Upstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,
-                const TunnelingConfigHelper& config, const StreamInfo::StreamInfo& downstream_info);
+                const TunnelingConfigHelper& config, StreamInfo::StreamInfo& downstream_info);
 
   void setRequestEncoder(Http::RequestEncoder& request_encoder, bool is_ssl) override;
   bool isValidResponse(const Http::ResponseHeaderMap& headers) override;

source/extensions/upstreams/tcp/generic/config.cc

@@ -14,7 +14,8 @@ namespace Generic {
 TcpProxy::GenericConnPoolPtr GenericConnPoolFactory::createGenericConnPool(
     Upstream::ThreadLocalCluster& thread_local_cluster,
     TcpProxy::TunnelingConfigHelperOptConstRef config, Upstream::LoadBalancerContext* context,
-    Envoy::Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks) const {
+    Envoy::Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks,
+    StreamInfo::StreamInfo& downstream_info) const {
   if (config.has_value()) {
     Http::CodecType pool_type;
     if ((thread_local_cluster.info()->features() & Upstream::ClusterInfo::Features::HTTP2) != 0) {
@@ -25,8 +26,8 @@ TcpProxy::GenericConnPoolPtr GenericConnPoolFactory::createGenericConnPool(
     } else {
       pool_type = Http::CodecType::HTTP1;
     }
-    auto ret = std::make_unique<TcpProxy::HttpConnPool>(thread_local_cluster, context, *config,
-                                                        upstream_callbacks, pool_type);
+    auto ret = std::make_unique<TcpProxy::HttpConnPool>(
+        thread_local_cluster, context, *config, upstream_callbacks, pool_type, downstream_info);
     return (ret->valid() ? std::move(ret) : nullptr);
   }
   auto ret =

source/extensions/upstreams/tcp/generic/config.h

@@ -17,10 +17,12 @@ class GenericConnPoolFactory : public TcpProxy::GenericConnPoolFactory {
 public:
   std::string name() const override { return "envoy.filters.connection_pools.tcp.generic"; }
   std::string category() const override { return "envoy.upstreams"; }
-  TcpProxy::GenericConnPoolPtr createGenericConnPool(
-      Upstream::ThreadLocalCluster& thread_local_cluster,
-      TcpProxy::TunnelingConfigHelperOptConstRef config, Upstream::LoadBalancerContext* context,
-      Envoy::Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks) const override;
+  TcpProxy::GenericConnPoolPtr
+  createGenericConnPool(Upstream::ThreadLocalCluster& thread_local_cluster,
+                        TcpProxy::TunnelingConfigHelperOptConstRef config,
+                        Upstream::LoadBalancerContext* context,
+                        Envoy::Tcp::ConnectionPool::UpstreamCallbacks& upstream_callbacks,
+                        StreamInfo::StreamInfo& downstream_info) const override;
 
   ProtobufTypes::MessagePtr createEmptyConfigProto() override {
     return std::make_unique<

test/extensions/upstreams/tcp/generic/config_test.cc

@@ -23,7 +23,6 @@ namespace Generic {
 class TcpConnPoolTest : public ::testing::Test {
 public:
   TcpConnPoolTest() {
-    EXPECT_CALL(connection_, streamInfo()).WillRepeatedly(ReturnRef(downstream_stream_info_));
     EXPECT_CALL(lb_context_, downstreamConnection()).WillRepeatedly(Return(&connection_));
   }
   NiceMock<Upstream::MockThreadLocalCluster> thread_local_cluster_;
@@ -42,7 +41,7 @@ TEST_F(TcpConnPoolTest, TestNoConnPool) {
   EXPECT_CALL(thread_local_cluster_, httpConnPool(_, _, _)).WillOnce(Return(absl::nullopt));
   EXPECT_EQ(nullptr, factory_.createGenericConnPool(
                          thread_local_cluster_, TcpProxy::TunnelingConfigHelperOptConstRef(config),
-                         &lb_context_, callbacks_));
+                         &lb_context_, callbacks_, downstream_stream_info_));
 }
 
 TEST_F(TcpConnPoolTest, Http2Config) {
@@ -56,7 +55,7 @@ TEST_F(TcpConnPoolTest, Http2Config) {
   EXPECT_CALL(thread_local_cluster_, httpConnPool(_, _, _)).WillOnce(Return(absl::nullopt));
   EXPECT_EQ(nullptr, factory_.createGenericConnPool(
                          thread_local_cluster_, TcpProxy::TunnelingConfigHelperOptConstRef(config),
-                         &lb_context_, callbacks_));
+                         &lb_context_, callbacks_, downstream_stream_info_));
 }
 
 TEST_F(TcpConnPoolTest, Http3Config) {
@@ -71,7 +70,7 @@ TEST_F(TcpConnPoolTest, Http3Config) {
   EXPECT_CALL(thread_local_cluster_, httpConnPool(_, _, _)).WillOnce(Return(absl::nullopt));
   EXPECT_EQ(nullptr, factory_.createGenericConnPool(
                          thread_local_cluster_, TcpProxy::TunnelingConfigHelperOptConstRef(config),
-                         &lb_context_, callbacks_));
+                         &lb_context_, callbacks_, downstream_stream_info_));
 }
 
 } // namespace Generic

test/integration/BUILD

@@ -1453,6 +1453,7 @@ envoy_cc_test(
         "//source/extensions/filters/network/tcp_proxy:config",
         "//source/extensions/upstreams/http/tcp:config",
         "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/access_loggers/file/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/filters/network/tcp_proxy/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/upstreams/http/tcp/v3:pkg_cc_proto",
     ],