websocket: tunneling websockets (and upgrades in general) over H2

api/envoy/api/v2/core/protocol.proto

@@ -76,6 +76,20 @@ message Http2ProtocolOptions {
   // window. Currently, this has the same minimum/maximum/default as *initial_stream_window_size*.
   google.protobuf.UInt32Value initial_connection_window_size = 4
       [(validate.rules).uint32 = {gte: 65535, lte: 2147483647}];
+
+  // [#not-implemented-hide:] Hiding until nghttp2 has native support.
+  //
+  // Allows proxying Websocket and other upgrades over H2 connect.
+  //
+  // THIS IS NOT SAFE TO USE IN PRODUCTION
+  //
+  // This currently works via disabling all HTTP sanity checks for H2 traffic
+  // which is a much larger hammer than we'd like to use. Eventually when
+  // https://github.com/nghttp2/nghttp2/issues/1181 is resolved, this will work
+  // with simply enabling CONNECT for H2. This may require some tweaks to the
+  // headers making pre-CONNECT-support proxying not backwards compatible with
+  // post-CONNECT-support proxying.
+  google.protobuf.BoolValue allow_connect = 5;
 }
 
 // [#not-implemented-hide:]

docs/root/intro/arch_overview/websocket.rst

@@ -20,6 +20,31 @@ one can set up custom
 for the given upgrade type, up to and including only using the router filter to send the WebSocket
 data upstream.
 
+Handling H2 hops (implementation in progress)
+---------------------------------------------
+
+One oft requested feature for Envoy was to allow WebSocket to traverse HTTP/2 hops, where there
+was a set-up such as
+
+Client ---- HTTP/1.1 ---- Frontline Envoy ---- HTTP/2 ---- Second tier Envoy ---- H1  ---- Upstream
+
+In this case, if a client is for example using WebSocket, we want the Websocket to arive at the
+upstream server functionally intact, which means it needs to traverse the HTTP/2 hop.
+
+TODO(alyssawilk) copy the warnings from the config here, or just land the docs when we unhide.
+
+This is accomplished via
+`extended CONNECT <https://tools.ietf.org/html/draft-mcmanus-httpbis-h2-websockets`_ support. The
+WebSocket request will be transformed into an HTTP/2 CONNECT stream, with :protocol header
+indicating the original upgrade, traverse the HTTP/2 hop, and be downgraded back into an HTTP/1
+WebSocket Upgrade. This same Upgrade-CONNECT-Upgrade transformation will be performed on any
+HTTP/2 hop, with the documented flaw that the HTTP/1.1 method is always assumed to be GET.
+Non-WebSocket upgrades are allowed to use any valid HTTP method (i.e. POST) and the current
+upgrade/downgrade mechanism will drop the original method and transform the Upgrade request to
+a GET method on the final Envoy-Upstream hop.
+
+TODO(alyssawilk) link to the config changes required to enable upgrade upstream/downstream
+
 Old style WebSocket support
 ===========================
 

include/envoy/http/codec.h

@@ -208,6 +208,7 @@ struct Http2Settings {
   uint32_t max_concurrent_streams_{DEFAULT_MAX_CONCURRENT_STREAMS};
   uint32_t initial_stream_window_size_{DEFAULT_INITIAL_STREAM_WINDOW_SIZE};
   uint32_t initial_connection_window_size_{DEFAULT_INITIAL_CONNECTION_WINDOW_SIZE};
+  bool allow_connect_{DEFAULT_ALLOW_CONNECT};
 
   // disable HPACK compression
   static const uint32_t MIN_HPACK_TABLE_SIZE = 0;
@@ -241,6 +242,8 @@ struct Http2Settings {
   // our default connection-level window also equals to our stream-level
   static const uint32_t DEFAULT_INITIAL_CONNECTION_WINDOW_SIZE = 256 * 1024 * 1024;
   static const uint32_t MAX_INITIAL_CONNECTION_WINDOW_SIZE = (1U << 31) - 1;
+  // By default both nghttp2 and Envoy do now allow CONNECT over H2.
+  static const bool DEFAULT_ALLOW_CONNECT = false;
 };
 
 /**

include/envoy/http/header_map.h

@@ -277,6 +277,7 @@ class HeaderEntry {
   HEADER_FUNC(Origin)                                                                              \
   HEADER_FUNC(OtSpanContext)                                                                       \
   HEADER_FUNC(Path)                                                                                \
+  HEADER_FUNC(Protocol)                                                                            \
   HEADER_FUNC(ProxyConnection)                                                                     \
   HEADER_FUNC(Referer)                                                                             \
   HEADER_FUNC(RequestId)                                                                           \

source/common/http/conn_manager_impl.cc

@@ -576,7 +576,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
 
   // Modify the downstream remote address depending on configuration and headers.
   request_info_.setDownstreamRemoteAddress(ConnectionManagerUtility::mutateRequestHeaders(
-      *request_headers_, protocol, connection_manager_.read_callbacks_->connection(),
+      *request_headers_, connection_manager_.read_callbacks_->connection(),
       connection_manager_.config_, *snapped_route_config_, connection_manager_.random_generator_,
       connection_manager_.runtime_, connection_manager_.local_info_));
   ASSERT(request_info_.downstreamRemoteAddress() != nullptr);

source/common/http/conn_manager_utility.cc

@@ -19,13 +19,13 @@ namespace Envoy {
 namespace Http {
 
 Network::Address::InstanceConstSharedPtr ConnectionManagerUtility::mutateRequestHeaders(
-    Http::HeaderMap& request_headers, Protocol protocol, Network::Connection& connection,
+    Http::HeaderMap& request_headers, Network::Connection& connection,
     ConnectionManagerConfig& config, const Router::Config& route_config,
     Runtime::RandomGenerator& random, Runtime::Loader& runtime,
     const LocalInfo::LocalInfo& local_info) {
   // If this is a Upgrade request, do not remove the Connection and Upgrade headers,
   // as we forward them verbatim to the upstream hosts.
-  if (protocol == Protocol::Http11 && Utility::isUpgrade(request_headers)) {
+  if (Utility::isUpgrade(request_headers)) {
     // The current WebSocket implementation re-uses the HTTP1 codec to send upgrade headers to
     // the upstream host. This adds the "transfer-encoding: chunked" request header if the stream
     // has not ended and content-length does not exist. In HTTP1.1, if transfer-encoding and

source/common/http/conn_manager_utility.h

@@ -28,10 +28,10 @@ class ConnectionManagerUtility {
    *         existence of the x-forwarded-for header. Again see the method for more details.
    */
   static Network::Address::InstanceConstSharedPtr
-  mutateRequestHeaders(Http::HeaderMap& request_headers, Protocol protocol,
-                       Network::Connection& connection, ConnectionManagerConfig& config,
-                       const Router::Config& route_config, Runtime::RandomGenerator& random,
-                       Runtime::Loader& runtime, const LocalInfo::LocalInfo& local_info);
+  mutateRequestHeaders(Http::HeaderMap& request_headers, Network::Connection& connection,
+                       ConnectionManagerConfig& config, const Router::Config& route_config,
+                       Runtime::RandomGenerator& random, Runtime::Loader& runtime,
+                       const LocalInfo::LocalInfo& local_info);
 
   static void mutateResponseHeaders(Http::HeaderMap& response_headers,
                                     const Http::HeaderMap* request_headers, const std::string& via);

source/common/http/headers.h

@@ -76,6 +76,7 @@ class HeaderValues {
   const LowerCaseString Origin{"origin"};
   const LowerCaseString OtSpanContext{"x-ot-span-context"};
   const LowerCaseString Path{":path"};
+  const LowerCaseString Protocol{":protocol"};
   const LowerCaseString ProxyConnection{"proxy-connection"};
   const LowerCaseString Referer{"referer"};
   const LowerCaseString RequestId{"x-request-id"};
@@ -158,6 +159,7 @@ class HeaderValues {
   } ExpectValues;
 
   struct {
+    const std::string Connect{"CONNECT"};
     const std::string Get{"GET"};
     const std::string Head{"HEAD"};
     const std::string Post{"POST"};

source/common/http/http2/codec_impl.cc

@@ -90,7 +90,19 @@ void ConnectionImpl::StreamImpl::encode100ContinueHeaders(const HeaderMap& heade
 
 void ConnectionImpl::StreamImpl::encodeHeaders(const HeaderMap& headers, bool end_stream) {
   std::vector<nghttp2_nv> final_headers;
-  buildHeaders(final_headers, headers);
+
+  if (Http::Utility::isUpgrade(headers)) {
+    HeaderMapImpl modified_headers(headers);
+    upgrade_type_ = headers.Upgrade()->value().c_str();
+    if (headers.Status()) {
+      Http::Utility::transformUpgradeResponseFromH1toH2(modified_headers);
+    } else {
+      Http::Utility::transformUpgradeRequestFromH1toH2(modified_headers);
+    }
+    buildHeaders(final_headers, modified_headers);
+  } else {
+    buildHeaders(final_headers, headers);
+  }
 
   nghttp2_data_provider provider;
   if (!end_stream) {
@@ -151,6 +163,15 @@ void ConnectionImpl::StreamImpl::pendingRecvBufferLowWatermark() {
   readDisable(false);
 }
 
+void ConnectionImpl::StreamImpl::decodeHeaders() {
+  if (Http::Utility::isH2UpgradeRequest(*headers_)) {
+    Http::Utility::transformUpgradeRequestFromH2toH1(*headers_);
+  } else if (!upgrade_type_.empty() && headers_->Status()) {
+    Http::Utility::transformUpgradeResponseFromH2toH1(*headers_, upgrade_type_);
+  }
+  decoder_->decodeHeaders(std::move(headers_), remote_end_stream_);
+}
+
 void ConnectionImpl::StreamImpl::pendingSendBufferHighWatermark() {
   ENVOY_CONN_LOG(debug, "send buffer over limit ", parent_.connection_);
   ASSERT(!pending_send_buffer_high_watermark_called_);
@@ -366,13 +387,13 @@ int ConnectionImpl::onFrameReceived(const nghttp2_frame* frame) {
         ASSERT(!stream->remote_end_stream_);
         stream->decoder_->decode100ContinueHeaders(std::move(stream->headers_));
       } else {
-        stream->decoder_->decodeHeaders(std::move(stream->headers_), stream->remote_end_stream_);
+        stream->decodeHeaders();
       }
       break;
     }
 
     case NGHTTP2_HCAT_REQUEST: {
-      stream->decoder_->decodeHeaders(std::move(stream->headers_), stream->remote_end_stream_);
+      stream->decodeHeaders();
       break;
     }
 
@@ -401,7 +422,7 @@ int ConnectionImpl::onFrameReceived(const nghttp2_frame* frame) {
           // start out with. In this case, raise as headers. nghttp2 message checking guarantees
           // proper flow here.
           ASSERT(!stream->headers_->Status() || stream->headers_->Status()->value() != "100");
-          stream->decoder_->decodeHeaders(std::move(stream->headers_), stream->remote_end_stream_);
+          stream->decodeHeaders();
         }
       }
 
@@ -735,6 +756,10 @@ ConnectionImpl::Http2Options::Http2Options(const Http2Settings& http2_settings)
   if (http2_settings.hpack_table_size_ != NGHTTP2_DEFAULT_HEADER_TABLE_SIZE) {
     nghttp2_option_set_max_deflate_dynamic_table_size(options_, http2_settings.hpack_table_size_);
   }
+  if (http2_settings.allow_connect_) {
+    // TODO(alyssawilk) change to ENABLE_CONNECT_PROTOCOL when it's available.
+    nghttp2_option_set_no_http_messaging(options_, 1);
+  }
 }
 
 ConnectionImpl::Http2Options::~Http2Options() { nghttp2_option_del(options_); }

source/common/http/http2/codec_impl.h

@@ -187,11 +187,16 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
     // I don't fully understand.
     static const uint64_t MAX_HEADER_SIZE = 63 * 1024;
 
+    // Does any necessary WebSocket/Upgrade conversion, then passes the headers
+    // to the decoder_.
+    void decodeHeaders();
+
     bool buffers_overrun() const { return read_disable_count_ > 0; }
 
     ConnectionImpl& parent_;
     HeaderMapImplPtr headers_;
     StreamDecoder* decoder_{};
+    std::string upgrade_type_;
     int32_t stream_id_{-1};
     uint32_t unconsumed_bytes_{0};
     uint32_t read_disable_count_{0};

source/common/http/utility.cc

@@ -209,6 +209,12 @@ bool Utility::isUpgrade(const HeaderMap& headers) {
                                            Http::Headers::get().ConnectionValues.Upgrade.c_str()));
 }
 
+bool Utility::isH2UpgradeRequest(const HeaderMap& headers) {
+  return headers.Method() &&
+         headers.Method()->value().c_str() == Http::Headers::get().MethodValues.Connect &&
+         headers.Protocol() && !headers.Protocol()->value().empty();
+}
+
 bool Utility::isWebSocketUpgradeRequest(const HeaderMap& headers) {
   return (isUpgrade(headers) && (0 == StringUtil::caseInsensitiveCompare(
                                           headers.Upgrade()->value().c_str(),
@@ -227,6 +233,8 @@ Utility::parseHttp2Settings(const envoy::api::v2::core::Http2ProtocolOptions& co
   ret.initial_connection_window_size_ =
       PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, initial_connection_window_size,
                                       Http::Http2Settings::DEFAULT_INITIAL_CONNECTION_WINDOW_SIZE);
+  ret.allow_connect_ = PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, allow_connect,
+                                                       Http::Http2Settings::DEFAULT_ALLOW_CONNECT);
   return ret;
 }
 
@@ -392,5 +400,50 @@ std::string Utility::queryParamsToString(const QueryParams& params) {
   return out;
 }
 
+void Utility::transformUpgradeRequestFromH1toH2(HeaderMap& headers) {
+  ASSERT(Utility::isUpgrade(headers));
+
+  const HeaderString& upgrade = headers.Upgrade()->value();
+  headers.insertMethod().value().setReference(Http::Headers::get().MethodValues.Connect);
+  headers.insertProtocol().value().setCopy(upgrade.c_str(), upgrade.size());
+  headers.removeUpgrade();
+  headers.removeConnection();
+  if (headers.ContentLength() == nullptr) {
+    headers.insertTransferEncoding().value().setReference(
+        Http::Headers::get().TransferEncodingValues.Chunked);
+  }
+}
+
+void Utility::transformUpgradeResponseFromH1toH2(HeaderMap& headers) {
+  if (getResponseStatus(headers) == 101) {
+    headers.insertStatus().value().setCopy("200", 3);
+  }
+  headers.removeUpgrade();
+  headers.removeConnection();
+  if (headers.ContentLength() == nullptr) {
+    headers.insertTransferEncoding().value().setReference(
+        Http::Headers::get().TransferEncodingValues.Chunked);
+  }
+}
+
+void Utility::transformUpgradeRequestFromH2toH1(HeaderMap& headers) {
+  ASSERT(Utility::isH2UpgradeRequest(headers));
+
+  const HeaderString& protocol = headers.Protocol()->value();
+  headers.insertMethod().value().setReference(Http::Headers::get().MethodValues.Get);
+  headers.insertUpgrade().value().setCopy(protocol.c_str(), protocol.size());
+  headers.insertConnection().value().setReference(Http::Headers::get().ConnectionValues.Upgrade);
+  headers.removeProtocol();
+}
+
+void Utility::transformUpgradeResponseFromH2toH1(HeaderMap& headers, absl::string_view upgrade) {
+  if (getResponseStatus(headers) == 200) {
+    headers.insertStatus().value().setCopy("101", 3);
+  }
+  // TODO(alyssawilk) what should we do for websocket responses on the failure path?
+  headers.insertUpgrade().value().setCopy(upgrade.data(), upgrade.size());
+  headers.insertConnection().value().setReference(Http::Headers::get().ConnectionValues.Upgrade);
+}
+
 } // namespace Http
 } // namespace Envoy

source/common/http/utility.h

@@ -101,6 +101,11 @@ uint64_t getResponseStatus(const HeaderMap& headers);
  */
 bool isUpgrade(const HeaderMap& headers);
 
+/**
+ * @return true if this is a CONNECT request with a :protocol header present, false otherwise.
+ */
+bool isH2UpgradeRequest(const HeaderMap& headers);
+
 /**
  * Determine whether this is a WebSocket Upgrade request.
  * This function returns true if the following HTTP headers and values are present:
@@ -198,6 +203,34 @@ MessagePtr prepareHeaders(const ::envoy::api::v2::core::HttpUri& http_uri);
  */
 std::string queryParamsToString(const QueryParams& query_params);
 
+/**
+ * Transforms the supplied headers from an HTTP/1 Upgrade request to an H2 style upgrade.
+ * Changes the method to connection, moves the Upgrade to a :protocol header,
+ * @param headers the headers to convert.
+ */
+void transformUpgradeRequestFromH1toH2(HeaderMap& headers);
+
+/**
+ * Transforms the supplied headers from an HTTP/1 Upgrade response to an H2 style upgrade response.
+ * Changes the 101 upgrade response to a 200 for the CONNECT response.
+ * @param headers the headers to convert.
+ */
+void transformUpgradeResponseFromH1toH2(HeaderMap& headers);
+
+/**
+ * Transforms the supplied headers from an H2 "CONNECT"-with-:protocol-header to an HTTP/1 style
+ * Upgrade response.
+ * @param headers the headers to convert.
+ */
+void transformUpgradeRequestFromH2toH1(HeaderMap& headers);
+
+/**
+ * Transforms the supplied headers from an H2 "CONNECT success" to an HTTP/1 style Upgrade response.
+ * The caller is responsible for ensuring this only happens on upgraded streams.
+ * @param headers the headers to convert.
+ */
+void transformUpgradeResponseFromH2toH1(HeaderMap& headers, absl::string_view upgrade);
+
 } // namespace Utility
 } // namespace Http
 } // namespace Envoy

test/common/http/conn_manager_utility_test.cc

@@ -88,11 +88,11 @@ class ConnectionManagerUtilityTest : public testing::Test {
   // This is a convenience method used to call mutateRequestHeaders(). It is done in this
   // convoluted way to force tests to check both the final downstream address as well as whether
   // the request is internal/external, given the importance of these two pieces of data.
-  MutateRequestRet callMutateRequestHeaders(HeaderMap& headers, Protocol protocol) {
+  MutateRequestRet callMutateRequestHeaders(HeaderMap& headers, Protocol) {
     MutateRequestRet ret;
     ret.downstream_address_ =
-        ConnectionManagerUtility::mutateRequestHeaders(
-            headers, protocol, connection_, config_, route_config_, random_, runtime_, local_info_)
+        ConnectionManagerUtility::mutateRequestHeaders(headers, connection_, config_, route_config_,
+                                                       random_, runtime_, local_info_)
             ->asString();
     ret.internal_ = headers.EnvoyInternalRequest() != nullptr;
     return ret;
@@ -536,16 +536,6 @@ TEST_F(ConnectionManagerUtilityTest, RemoveConnectionUpgradeForNonWebSocketReque
   EXPECT_FALSE(headers.has("upgrade"));
 }
 
-// Make sure we remove connections headers for a WS request over h2.
-TEST_F(ConnectionManagerUtilityTest, RemoveConnectionUpgradeForHttp2Requests) {
-  TestHeaderMapImpl headers{{"connection", "upgrade"}, {"upgrade", "websocket"}};
-
-  EXPECT_EQ((MutateRequestRet{"10.0.0.3:50000", false}),
-            callMutateRequestHeaders(headers, Protocol::Http2));
-  EXPECT_FALSE(headers.has("connection"));
-  EXPECT_FALSE(headers.has("upgrade"));
-}
-
 // Test cleaning response headers.
 TEST_F(ConnectionManagerUtilityTest, MutateResponseHeaders) {
   TestHeaderMapImpl response_headers{