http: pausing when proxying CONNECT requests for security reasons

source/common/router/router.cc

@@ -664,9 +664,9 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
 
 Filter::HttpOrTcpPool Filter::createConnPool(Upstream::HostDescriptionConstSharedPtr& host) {
   Filter::HttpOrTcpPool conn_pool;
-  bool should_tcp_proxy = route_entry_->connectConfig().has_value() &&
-                          downstream_headers_->Method()->value().getStringView() ==
-                              Http::Headers::get().MethodValues.Connect;
+  const bool should_tcp_proxy = route_entry_->connectConfig().has_value() &&
+                                downstream_headers_->Method()->value().getStringView() ==
+                                    Http::Headers::get().MethodValues.Connect;
 
   if (!should_tcp_proxy) {
     conn_pool = getHttpConnPool();

source/common/router/upstream_request.cc

@@ -48,7 +48,7 @@ UpstreamRequest::UpstreamRequest(RouterFilterInterface& parent,
       calling_encode_headers_(false), upstream_canary_(false), decode_complete_(false),
       encode_complete_(false), encode_trailers_(false), retried_(false), awaiting_headers_(true),
       outlier_detection_timeout_recorded_(false),
-      create_per_try_timeout_on_request_complete_(false),
+      create_per_try_timeout_on_request_complete_(false), paused_for_connect_(false),
       record_timeout_budget_(parent_.cluster()->timeoutBudgetStats().has_value()) {
   if (parent_.config().start_child_span_) {
     span_ = parent_.callbacks()->activeSpan().spawnChild(
@@ -127,6 +127,12 @@ void UpstreamRequest::decodeHeaders(Http::ResponseHeaderMapPtr&& headers, bool e
   }
   const uint64_t response_code = Http::Utility::getResponseStatus(*headers);
   stream_info_.response_code_ = static_cast<uint32_t>(response_code);
+
+  if (paused_for_connect_ && response_code == 200) {
+    encodeBodyAndTrailers();
+    paused_for_connect_ = false;
+  }
+
   parent_.onUpstreamHeaders(response_code, std::move(headers), *this, end_stream);
 }
 
@@ -177,7 +183,7 @@ void UpstreamRequest::encodeData(Buffer::Instance& data, bool end_stream) {
   ASSERT(!encode_complete_);
   encode_complete_ = end_stream;
 
-  if (!upstream_) {
+  if (!upstream_ || paused_for_connect_) {
     ENVOY_STREAM_LOG(trace, "buffering {} bytes", *parent_.callbacks(), data.length());
     if (!buffered_request_body_) {
       buffered_request_body_ = std::make_unique<Buffer::WatermarkBuffer>(
@@ -357,6 +363,7 @@ void UpstreamRequest::onPoolReady(
   parent_.callbacks()->addDownstreamWatermarkCallbacks(downstream_watermark_manager_);
 
   calling_encode_headers_ = true;
+  auto* headers = parent_.downstreamHeaders();
   if (parent_.routeEntry()->autoHostRewrite() && !host->hostname().empty()) {
     parent_.downstreamHeaders()->setHost(host->hostname());
   }
@@ -367,13 +374,22 @@ void UpstreamRequest::onPoolReady(
 
   upstream_timing_.onFirstUpstreamTxByteSent(parent_.callbacks()->dispatcher().timeSource());
 
-  const bool end_stream = !buffered_request_body_ && encode_complete_ && !encode_trailers_;
-  // If end_stream is set in headers, and there are metadata to send, delays end_stream. The case
-  // only happens when decoding headers filters return ContinueAndEndStream.
-  const bool delay_headers_end_stream = end_stream && !downstream_metadata_map_vector_.empty();
-  upstream_->encodeHeaders(*parent_.downstreamHeaders(), end_stream && !delay_headers_end_stream);
+  // Make sure that when we are forwarding CONNECT payload we do not do so until
+  // the upstream has accepted the CONNECT request.
+  if (conn_pool_->protocol().has_value() && headers->Method() &&
+      headers->Method()->value().getStringView() == Http::Headers::get().MethodValues.Connect) {
+    paused_for_connect_ = true;
+  }
+
+  upstream_->encodeHeaders(*parent_.downstreamHeaders(), shouldSendEndStream());
   calling_encode_headers_ = false;
 
+  if (!paused_for_connect_) {
+    encodeBodyAndTrailers();
+  }
+}
+
+void UpstreamRequest::encodeBodyAndTrailers() {
   // It is possible to get reset in the middle of an encodeHeaders() call. This happens for
   // example in the HTTP/2 codec if the frame cannot be encoded for some reason. This should never
   // happen but it's unclear if we have covered all cases so protect against it and test for it.
@@ -388,7 +404,7 @@ void UpstreamRequest::onPoolReady(
                        downstream_metadata_map_vector_);
       upstream_->encodeMetadata(downstream_metadata_map_vector_);
       downstream_metadata_map_vector_.clear();
-      if (delay_headers_end_stream) {
+      if (shouldSendEndStream()) {
         Buffer::OwnedImpl empty_data("");
         upstream_->encodeData(empty_data, true);
       }

source/common/router/upstream_request.h

@@ -119,6 +119,7 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   };
 
   void readEnable();
+  void encodeBodyAndTrailers();
 
   // Getters and setters
   Upstream::HostDescriptionConstSharedPtr& upstreamHost() { return upstream_host_; }
@@ -140,6 +141,13 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   }
 
 private:
+  bool shouldSendEndStream() {
+    // Only encode end stream if the full request has been received, the body
+    // has been sent, and any trailers or metadata have also been sent.
+    return encode_complete_ && !buffered_request_body_ && !encode_trailers_ &&
+           downstream_metadata_map_vector_.empty();
+  }
+
   RouterFilterInterface& parent_;
   std::unique_ptr<GenericConnPool> conn_pool_;
   bool grpc_rq_success_deferred_;
@@ -172,6 +180,9 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   // Tracks whether we deferred a per try timeout because the downstream request
   // had not been completed yet.
   bool create_per_try_timeout_on_request_complete_ : 1;
+  // True if the CONNECT headers have been sent but proxying payload is paused
+  // waiting for response headers.
+  bool paused_for_connect_ : 1;
 
   // Sentinel to indicate if timeout budget tracking is configured for the cluster,
   // and if so, if the per-try histogram should record a value.

test/common/router/router_test.cc

@@ -5667,6 +5667,71 @@ TEST_F(RouterTest, ApplicationProtocols) {
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
 }
 
+// Verify that CONNECT payload is not sent upstream until :200 response headers
+// are received.
+TEST_F(RouterTest, ConnectPauseAndResume) {
+  NiceMock<Http::MockRequestEncoder> encoder;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+  expectResponseTimerCreate();
+
+  EXPECT_CALL(encoder, encodeHeaders(_, false));
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  headers.setMethod("CONNECT");
+  router_.decodeHeaders(headers, false);
+
+  // Make sure any early data does not go upstream.
+  EXPECT_CALL(encoder, encodeData(_, _)).Times(0);
+  Buffer::OwnedImpl data;
+  router_.decodeData(data, true);
+
+  // Now send the response headers, and ensure the deferred payload is proxied.
+  EXPECT_CALL(encoder, encodeData(_, _));
+  Http::ResponseHeaderMapPtr response_headers(
+      new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  response_decoder->decodeHeaders(std::move(response_headers), true);
+}
+
+// Verify that CONNECT payload is not sent upstream if non-200 response headers are received.
+TEST_F(RouterTest, ConnectPauseNoResume) {
+  NiceMock<Http::MockRequestEncoder> encoder;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+  expectResponseTimerCreate();
+
+  EXPECT_CALL(encoder, encodeHeaders(_, false));
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  headers.setMethod("CONNECT");
+  router_.decodeHeaders(headers, false);
+
+  // Make sure any early data does not go upstream.
+  EXPECT_CALL(encoder, encodeData(_, _)).Times(0);
+  Buffer::OwnedImpl data;
+  router_.decodeData(data, true);
+
+  // Now send the response headers, and ensure the deferred payload is not proxied.
+  EXPECT_CALL(encoder, encodeData(_, _)).Times(0);
+  Http::ResponseHeaderMapPtr response_headers(
+      new Http::TestResponseHeaderMapImpl{{":status", "400"}});
+  response_decoder->decodeHeaders(std::move(response_headers), true);
+}
+
 class WatermarkTest : public RouterTest {
 public:
   void sendRequest(bool header_only_request = true, bool pool_ready = true) {

test/integration/integration_test.cc

@@ -1336,8 +1336,10 @@ TEST_P(IntegrationTest, ConnectWithChunkedBody) {
       });
   initialize();
 
+  // Send the payload early so we can regression test that body data does not
+  // get proxied until after the response headers are sent.
   IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("http"));
-  tcp_client->write("CONNECT host.com:80 HTTP/1.1\r\nHost: host\r\n\r\n", false);
+  tcp_client->write("CONNECT host.com:80 HTTP/1.1\r\nHost: host\r\n\r\npayload", false);
 
   FakeRawConnectionPtr fake_upstream_connection;
   ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
@@ -1347,6 +1349,8 @@ TEST_P(IntegrationTest, ConnectWithChunkedBody) {
   // No transfer-encoding: chunked or connection: close
   EXPECT_FALSE(absl::StrContains(data, "hunked")) << data;
   EXPECT_FALSE(absl::StrContains(data, "onnection")) << data;
+  // The payload should not be present as the response headers have not been sent.
+  EXPECT_FALSE(absl::StrContains(data, "payload")) << data;
 
   ASSERT_TRUE(fake_upstream_connection->write(
       "HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\nb\r\nHello World\r\n0\r\n\r\n"));
@@ -1356,8 +1360,7 @@ TEST_P(IntegrationTest, ConnectWithChunkedBody) {
   EXPECT_TRUE(absl::StrContains(tcp_client->data(), "\r\n\r\nb\r\nHello World\r\n0\r\n\r\n"))
       << tcp_client->data();
 
-  // Make sure the following payload is proxied without chunks or any other modifications.
-  tcp_client->write("payload");
+  // Make sure the early payload is proxied without chunks or any other modifications.
   ASSERT_TRUE(fake_upstream_connection->waitForData(
       FakeRawConnection::waitForInexactMatch("\r\n\r\npayload")));
 

test/integration/tcp_tunneling_integration_test.cc

@@ -80,8 +80,6 @@ class ConnectTerminationIntegrationTest
   bool enable_timeout_{};
 };
 
-// TODO(alyssawilk) make sure that if data is sent with the connect it does not go upstream
-// until the 200 headers are sent before unhiding ANY config.
 TEST_P(ConnectTerminationIntegrationTest, Basic) {
   initialize();