router: adding CONNECT support to upstreams (not prod-ready)

docs/root/intro/arch_overview/http/http.rst

@@ -7,5 +7,5 @@ HTTP
   http_connection_management
   http_filters
   http_routing
-  websocket
+  upgrades
   http_proxy

docs/root/intro/arch_overview/http/websocket.rst → docs/root/intro/arch_overview/http/upgrades.rst

@@ -1,19 +1,19 @@
 .. _arch_overview_websocket:
 
-WebSocket and HTTP upgrades
+HTTP upgrades
 ===========================
 
-Envoy Upgrade support is intended mainly for WebSocket but may be used for non-WebSocket
-upgrades as well. Upgrades pass both the HTTP headers and the upgrade payload
+Envoy Upgrade support is intended mainly for WebSocket and CONNECT support, but may be used for
+arbitrary upgrades as well. Upgrades pass both the HTTP headers and the upgrade payload
 through an HTTP filter chain. One may configure the
 :ref:`upgrade_configs <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.upgrade_configs>`
 with or without custom filter chains. If only the
 :ref:`upgrade_type <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.UpgradeConfig.upgrade_type>`
-is specified, both the upgrade headers, any request and response body, and WebSocket payload will
+is specified, both the upgrade headers, any request and response body, and HTTP data payload will
 pass through the default HTTP filter chain. To avoid the use of HTTP-only filters for upgrade payload,
 one can set up custom
 :ref:`filters <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.UpgradeConfig.filters>`
-for the given upgrade type, up to and including only using the router filter to send the WebSocket
+for the given upgrade type, up to and including only using the router filter to send the HTTP
 data upstream.
 
 Upgrades can be enabled or disabled on a :ref:`per-route <envoy_api_field_route.RouteAction.upgrade_configs>` basis.
@@ -32,12 +32,12 @@ laid out below, but custom filter chains can only be configured on a per-HttpCon
 | F                     | F                       | F                 |
 +-----------------------+-------------------------+-------------------+
 
-Note that the statistics for upgrades are all bundled together so WebSocket
+Note that the statistics for upgrades are all bundled together so WebSocket and other upgrades
 :ref:`statistics <config_http_conn_man_stats>` are tracked by stats such as
 downstream_cx_upgrades_total and downstream_cx_upgrades_active
 
-Handling HTTP/2 hops
-^^^^^^^^^^^^^^^^^^^^
+Websocket over HTTP/2 hops
+^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 While HTTP/2 support for WebSockets is off by default, Envoy does support tunneling WebSockets over
 HTTP/2 streams for deployments that prefer a uniform HTTP/2 mesh throughout; this enables, for example,
@@ -61,3 +61,31 @@ a GET method on the final Envoy-Upstream hop.
 
 Note that the HTTP/2 upgrade path has very strict HTTP/1.1 compliance, so will not proxy WebSocket
 upgrade requests or responses with bodies.
+
+.. TODO(alyssawilk) unhide this when unhiding config
+.. CONNECT support
+.. ^^^^^^^^^^^^^^^
+
+.. Envoy CONNECT support is off by default (Envoy will send an internally generated 403 in response to
+.. CONNECT requests). CONNECT support can be enabled via the upgrade options described above, setting
+.. the upgrade value to the special keyword "CONNECT".
+
+.. While for HTTP/2, CONNECT request may have a path, in general and for HTTP/1.1 CONNECT requests do
+.. not have a path, and can only be matched using a
+.. :ref:`connect_matcher <envoy_api_field_route.RouteMatch.connect_matcher>`
+..
+.. Envoy can handle CONNECT in one of two ways, either proxying the CONNECT headers through as if they
+.. were any other request, and letting the upstream terminate the CONNECT request, or by terminating the
+.. CONNECT request, and forwarding the payload as raw TCP data. When CONNECT upgrade configuration is
+.. set up, the default behavior is to proxy the CONNECT request, treating it like any other request using
+.. the upgrade path.
+.. If termination is desired, this can be accomplished by setting
+.. :ref:`connect_config <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.UpgradeConfig.connect_config>`
+.. If it that message is present for CONNECT requests, the router filter will strip the request headers,
+.. and forward the HTTP payload upstream. On receipt of initial TCP data from upstream, the router
+.. will synthesize 200 response headers, and then forward the TCP data as the HTTP response body.
+
+.. .. warning::
+.. This mode of CONNECT support can create major security holes if configured correctly, as the upstream
+.. will be forwarded *unsanitized* headers if they are in the body payload. Please use with caution
+

source/common/router/router.cc

@@ -115,6 +115,16 @@ bool convertRequestHeadersForInternalRedirect(Http::RequestHeaderMap& downstream
 
 constexpr uint64_t TimeoutPrecisionFactor = 100;
 
+Http::ConnectionPool::Instance*
+httpPool(absl::variant<Http::ConnectionPool::Instance*, Tcp::ConnectionPool::Instance*> pool) {
+  return absl::get<Http::ConnectionPool::Instance*>(pool);
+}
+
+Tcp::ConnectionPool::Instance*
+tcpPool(absl::variant<Http::ConnectionPool::Instance*, Tcp::ConnectionPool::Instance*> pool) {
+  return absl::get<Tcp::ConnectionPool::Instance*>(pool);
+}
+
 const absl::string_view getPath(const Http::RequestHeaderMap& headers) {
   return headers.Path() ? headers.Path()->value().getStringView() : "";
 }
@@ -549,12 +559,10 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
     }
   }
 
-  Http::ConnectionPool::Instance* http_pool = getHttpConnPool();
   Upstream::HostDescriptionConstSharedPtr host;
+  Filter::HttpOrTcpPool conn_pool = createConnPool(host);
 
-  if (http_pool) {
-    host = http_pool->host();
-  } else {
+  if (!host) {
     sendNoHealthyUpstreamResponse();
     return Http::FilterHeadersStatus::StopIteration;
   }
@@ -644,8 +652,7 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
   // Hang onto the modify_headers function for later use in handling upstream responses.
   modify_headers_ = modify_headers;
 
-  UpstreamRequestPtr upstream_request =
-      std::make_unique<UpstreamRequest>(*this, std::make_unique<HttpConnPool>(*http_pool));
+  UpstreamRequestPtr upstream_request = createUpstreamRequest(conn_pool);
   upstream_request->moveIntoList(std::move(upstream_request), upstream_requests_);
   upstream_requests_.front()->encodeHeaders(end_stream);
   if (end_stream) {
@@ -655,6 +662,38 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
   return Http::FilterHeadersStatus::StopIteration;
 }
 
+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;
+
+  if (!should_tcp_proxy) {
+    conn_pool = getHttpConnPool();
+    if (httpPool(conn_pool)) {
+      host = httpPool(conn_pool)->host();
+    }
+  } else {
+    transport_socket_options_ = Network::TransportSocketOptionsUtility::fromFilterState(
+        *callbacks_->streamInfo().filterState());
+    conn_pool = config_.cm_.tcpConnPoolForCluster(route_entry_->clusterName(),
+                                                  Upstream::ResourcePriority::Default, this);
+    if (tcpPool(conn_pool)) {
+      host = tcpPool(conn_pool)->host();
+    }
+  }
+  return conn_pool;
+}
+
+UpstreamRequestPtr Filter::createUpstreamRequest(Filter::HttpOrTcpPool conn_pool) {
+  if (absl::holds_alternative<Http::ConnectionPool::Instance*>(conn_pool)) {
+    return std::make_unique<UpstreamRequest>(*this,
+                                             std::make_unique<HttpConnPool>(*httpPool(conn_pool)));
+  }
+  return std::make_unique<UpstreamRequest>(*this,
+                                           std::make_unique<TcpConnPool>(tcpPool(conn_pool)));
+}
+
 Http::ConnectionPool::Instance* Filter::getHttpConnPool() {
   // Choose protocol based on cluster configuration and downstream connection
   // Note: Cluster may downgrade HTTP2 to HTTP1 based on runtime configuration.
@@ -1454,19 +1493,15 @@ void Filter::doRetry() {
   attempt_count_++;
   ASSERT(pending_retries_ > 0);
   pending_retries_--;
-  UpstreamRequestPtr upstream_request;
-
-  Http::ConnectionPool::Instance* conn_pool = getHttpConnPool();
-  if (conn_pool) {
-    upstream_request =
-        std::make_unique<UpstreamRequest>(*this, std::make_unique<HttpConnPool>(*conn_pool));
-  }
 
-  if (!upstream_request) {
+  Upstream::HostDescriptionConstSharedPtr host;
+  Filter::HttpOrTcpPool conn_pool = createConnPool(host);
+  if (!host) {
     sendNoHealthyUpstreamResponse();
     cleanup();
     return;
   }
+  UpstreamRequestPtr upstream_request = createUpstreamRequest(conn_pool);
 
   if (include_attempt_count_in_request_) {
     downstream_headers_->setEnvoyAttemptCount(attempt_count_);

source/common/router/router.h

@@ -467,6 +467,12 @@ class Filter : Logger::Loggable<Logger::Id::router>,
                    const Upstream::ClusterInfo& cluster, const VirtualCluster* vcluster,
                    Runtime::Loader& runtime, Runtime::RandomGenerator& random,
                    Event::Dispatcher& dispatcher, Upstream::ResourcePriority priority) PURE;
+
+  using HttpOrTcpPool =
+      absl::variant<Http::ConnectionPool::Instance*, Tcp::ConnectionPool::Instance*>;
+  HttpOrTcpPool createConnPool(Upstream::HostDescriptionConstSharedPtr& host);
+  UpstreamRequestPtr createUpstreamRequest(Filter::HttpOrTcpPool conn_pool);
+
   Http::ConnectionPool::Instance* getHttpConnPool();
   void maybeDoShadowing();
   bool maybeRetryReset(Http::StreamResetReason reset_reason, UpstreamRequest& upstream_request);

source/common/router/upstream_request.cc

@@ -490,6 +490,16 @@ void HttpConnPool::newStream(GenericConnectionPoolCallbacks* callbacks) {
   }
 }
 
+void TcpConnPool::onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr&& conn_data,
+                              Upstream::HostDescriptionConstSharedPtr host) {
+  upstream_handle_ = nullptr;
+  Network::Connection& latched_conn = conn_data->connection();
+  auto upstream =
+      std::make_unique<TcpUpstream>(callbacks_->upstreamRequest(), std::move(conn_data));
+  callbacks_->onPoolReady(std::move(upstream), host, latched_conn.localAddress(),
+                          latched_conn.streamInfo());
+}
+
 bool HttpConnPool::cancelAnyPendingRequest() {
   if (conn_pool_stream_handle_) {
     conn_pool_stream_handle_->cancel();
@@ -516,5 +526,68 @@ void HttpConnPool::onPoolReady(Http::RequestEncoder& request_encoder,
                           request_encoder.getStream().connectionLocalAddress(), info);
 }
 
+TcpUpstream::TcpUpstream(UpstreamRequest* upstream_request,
+                         Tcp::ConnectionPool::ConnectionDataPtr&& upstream)
+    : upstream_request_(upstream_request), upstream_conn_data_(std::move(upstream)) {
+  upstream_conn_data_->connection().enableHalfClose(true);
+  upstream_conn_data_->addUpstreamCallbacks(*this);
+}
+
+void TcpUpstream::encodeData(Buffer::Instance& data, bool end_stream) {
+  upstream_conn_data_->connection().write(data, end_stream);
+}
+
+void TcpUpstream::encodeHeaders(const Http::RequestHeaderMap&, bool end_stream) {
+  if (end_stream) {
+    Buffer::OwnedImpl data;
+    upstream_conn_data_->connection().write(data, true);
+  }
+}
+
+void TcpUpstream::encodeTrailers(const Http::RequestTrailerMap&) {
+  Buffer::OwnedImpl data;
+  upstream_conn_data_->connection().write(data, true);
+}
+
+void TcpUpstream::readDisable(bool disable) {
+  if (upstream_conn_data_->connection().state() != Network::Connection::State::Open) {
+    return;
+  }
+  upstream_conn_data_->connection().readDisable(disable);
+}
+
+void TcpUpstream::resetStream() {
+  upstream_request_ = nullptr;
+  upstream_conn_data_->connection().close(Network::ConnectionCloseType::NoFlush);
+}
+
+void TcpUpstream::onUpstreamData(Buffer::Instance& data, bool end_stream) {
+  if (!sent_headers_) {
+    Http::ResponseHeaderMapPtr headers{
+        Http::createHeaderMap<Http::ResponseHeaderMapImpl>({{Http::Headers::get().Status, "200"}})};
+    upstream_request_->decodeHeaders(std::move(headers), false);
+    sent_headers_ = true;
+  }
+  upstream_request_->decodeData(data, end_stream);
+}
+
+void TcpUpstream::onEvent(Network::ConnectionEvent event) {
+  if (event != Network::ConnectionEvent::Connected && upstream_request_) {
+    upstream_request_->onResetStream(Http::StreamResetReason::ConnectionTermination, "");
+  }
+}
+
+void TcpUpstream::onAboveWriteBufferHighWatermark() {
+  if (upstream_request_) {
+    upstream_request_->disableDataFromDownstreamForFlowControl();
+  }
+}
+
+void TcpUpstream::onBelowWriteBufferLowWatermark() {
+  if (upstream_request_) {
+    upstream_request_->enableDataFromDownstreamForFlowControl();
+  }
+}
+
 } // namespace Router
 } // namespace Envoy

source/common/router/upstream_request.h

@@ -202,6 +202,41 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
   GenericConnectionPoolCallbacks* callbacks_{};
 };
 
+class TcpConnPool : public GenericConnPool, public Tcp::ConnectionPool::Callbacks {
+public:
+  TcpConnPool(Tcp::ConnectionPool::Instance* conn_pool) : conn_pool_(conn_pool) {}
+
+  void newStream(GenericConnectionPoolCallbacks* callbacks) override {
+    callbacks_ = callbacks;
+    upstream_handle_ = conn_pool_->newConnection(*this);
+  }
+
+  bool cancelAnyPendingRequest() override {
+    if (upstream_handle_) {
+      upstream_handle_->cancel(Tcp::ConnectionPool::CancelPolicy::Default);
+      upstream_handle_ = nullptr;
+      return true;
+    }
+    return false;
+  }
+  absl::optional<Http::Protocol> protocol() const override { return absl::nullopt; }
+
+  // Tcp::ConnectionPool::Callbacks
+  void onPoolFailure(ConnectionPool::PoolFailureReason reason,
+                     Upstream::HostDescriptionConstSharedPtr host) override {
+    upstream_handle_ = nullptr;
+    callbacks_->onPoolFailure(reason, "", host);
+  }
+
+  void onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr&& conn_data,
+                   Upstream::HostDescriptionConstSharedPtr host) override;
+
+private:
+  Tcp::ConnectionPool::Instance* conn_pool_;
+  Tcp::ConnectionPool::Cancellable* upstream_handle_{};
+  GenericConnectionPoolCallbacks* callbacks_{};
+};
+
 // A generic API which covers common functionality between HTTP and TCP upstreams.
 class GenericUpstream {
 public:
@@ -261,5 +296,29 @@ class HttpUpstream : public GenericUpstream, public Http::StreamCallbacks {
   Http::RequestEncoder* request_encoder_{};
 };
 
+class TcpUpstream : public GenericUpstream, public Tcp::ConnectionPool::UpstreamCallbacks {
+public:
+  TcpUpstream(UpstreamRequest* upstream_request, Tcp::ConnectionPool::ConnectionDataPtr&& upstream);
+
+  // GenericUpstream
+  void encodeData(Buffer::Instance& data, bool end_stream) override;
+  void encodeMetadata(const Http::MetadataMapVector&) override {}
+  void encodeHeaders(const Http::RequestHeaderMap&, bool end_stream) override;
+  void encodeTrailers(const Http::RequestTrailerMap&) override;
+  void readDisable(bool disable) override;
+  void resetStream() override;
+
+  // Tcp::ConnectionPool::UpstreamCallbacks
+  void onUpstreamData(Buffer::Instance& data, bool end_stream) override;
+  void onEvent(Network::ConnectionEvent event) override;
+  void onAboveWriteBufferHighWatermark() override;
+  void onBelowWriteBufferLowWatermark() override;
+
+private:
+  UpstreamRequest* upstream_request_;
+  Tcp::ConnectionPool::ConnectionDataPtr upstream_conn_data_;
+  bool sent_headers_{};
+};
+
 } // namespace Router
 } // namespace Envoy

test/common/router/BUILD

@@ -319,6 +319,26 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "upstream_request_test",
+    srcs = ["upstream_request_test.cc"],
+    deps = [
+        "//source/common/buffer:buffer_lib",
+        "//source/common/router:router_lib",
+        "//source/common/upstream:upstream_includes",
+        "//source/common/upstream:upstream_lib",
+        "//test/common/http:common_lib",
+        "//test/mocks/http:http_mocks",
+        "//test/mocks/network:network_mocks",
+        "//test/mocks/router:router_mocks",
+        "//test/mocks/server:server_mocks",
+        "//test/mocks/upstream:upstream_mocks",
+        "//test/test_common:environment_lib",
+        "//test/test_common:simulated_time_system_lib",
+        "//test/test_common:utility_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "header_formatter_test",
     srcs = ["header_formatter_test.cc"],