Merge pull request #531 from capnproto/websocket

Add WebSocket support to HTTP library

c++/src/kj/compat/http.h

@@ -86,7 +86,10 @@ namespace kj {
   MACRO(te, "TE") \
   MACRO(trailer, "Trailer") \
   MACRO(transferEncoding, "Transfer-Encoding") \
-  MACRO(upgrade, "Upgrade")
+  MACRO(upgrade, "Upgrade") \
+  MACRO(websocketKey, "Sec-WebSocket-Key") \
+  MACRO(websocketVersion, "Sec-WebSocket-Version") \
+  MACRO(websocketAccept, "Sec-WebSocket-Accept")
 
 enum class HttpMethod {
   // Enum of known HTTP methods.
@@ -368,13 +371,54 @@ private:
   //   also add direct accessors for those headers.
 };
 
+class EntropySource {
+  // Interface for an object that generates entropy. Typically, cryptographically-random entropy
+  // is expected.
+  //
+  // TODO(cleanup): Put this somewhere more general.
+
+public:
+  virtual void generate(kj::ArrayPtr<byte> buffer) = 0;
+};
+
 class WebSocket {
+  // Interface representincg an open WebSocket session.
+  //
+  // Each side can send and receive data and "close" messages.
+  //
+  // Ping/Pong and message fragmentation are not exposed through this interface. These features of
+  // the underlying WebSocket protocol are not exposed by the browser-level Javascript API either,
+  // and thus applications typically need to implement these features at the application protocol
+  // level instead. The implementation is, however, expected to reply to Ping messages it receives.
+
 public:
-  WebSocket(kj::Own<kj::AsyncIoStream> stream);
-  // Create a WebSocket wrapping the given I/O stream.
+  virtual kj::Promise<void> send(kj::ArrayPtr<const byte> message) = 0;
+  virtual kj::Promise<void> send(kj::ArrayPtr<const char> message) = 0;
+  // Send a message (binary or text). The underlying buffer must remain valid, and you must not
+  // call send() again, until the returned promise resolves.
+
+  virtual kj::Promise<void> close(uint16_t code, kj::StringPtr reason) = 0;
+  // Send a Close message.
+  //
+  // Note that the returned Promise resolves once the message has been sent -- it does NOT wait
+  // for the other end to send a Close reply. The application should await a reply before dropping
+  // the WebSocket object.
+
+  virtual kj::Promise<void> disconnect() = 0;
+  // Sends EOF on the underlying connection without sending a "close" message. This is NOT a clean
+  // shutdown, but is sometimes useful when you want the other end to trigger whatever behavior
+  // it normally triggers when a connection is dropped.
+
+  struct Close {
+    uint16_t code;
+    kj::String reason;
+  };
 
-  kj::Promise<void> send(kj::ArrayPtr<const byte> message);
-  kj::Promise<void> send(kj::ArrayPtr<const char> message);
+  typedef kj::OneOf<kj::String, kj::Array<byte>, Close> Message;
+
+  virtual kj::Promise<Message> receive() = 0;
+  // Read one message from the WebSocket and return it. Can only call once at a time. Do not call
+  // again after EndOfStream is received.
 };
 
 class HttpClient {
@@ -392,7 +436,7 @@ public:
     kj::StringPtr statusText;
     const HttpHeaders* headers;
     kj::Own<kj::AsyncInputStream> body;
-    // `statusText` and `headers` remain valid until `body` is dropped.
+    // `statusText` and `headers` remain valid until `body` is dropped or read from.
   };
 
   struct Request {
@@ -424,14 +468,15 @@ public:
     uint statusCode;
     kj::StringPtr statusText;
     const HttpHeaders* headers;
-    kj::OneOf<kj::Own<kj::AsyncInputStream>, kj::Own<WebSocket>> upstreamOrBody;
-    // `statusText` and `headers` remain valid until `upstreamOrBody` is dropped.
+    kj::OneOf<kj::Own<kj::AsyncInputStream>, kj::Own<WebSocket>> webSocketOrBody;
+    // `statusText` and `headers` remain valid until `webSocketOrBody` is dropped or read from.
   };
   virtual kj::Promise<WebSocketResponse> openWebSocket(
-      kj::StringPtr url, const HttpHeaders& headers, kj::Own<WebSocket> downstream);
+      kj::StringPtr url, const HttpHeaders& headers);
   // Tries to open a WebSocket. Default implementation calls send() and never returns a WebSocket.
   //
-  // `url` and `headers` are invalidated when the returned promise resolves.
+  // `url` and `headers` need only remain valid until `openWebSocket()` returns (they can be
+  // stack-allocated).
 
   virtual kj::Promise<kj::Own<kj::AsyncIoStream>> connect(kj::StringPtr host);
   // Handles CONNECT requests. Only relevant for proxy clients. Default implementation throws
@@ -478,13 +523,10 @@ public:
 
   class WebSocketResponse: public Response {
   public:
-    kj::Own<WebSocket> startWebSocket(
-        uint statusCode, kj::StringPtr statusText, const HttpHeaders& headers,
-        WebSocket& upstream);
-    // Begin the response.
+    virtual kj::Own<WebSocket> acceptWebSocket(const HttpHeaders& headers) = 0;
+    // Accept and open the WebSocket.
     //
-    // `statusText` and `headers` need only remain valid until startWebSocket() returns (they can
-    // be stack-allocated).
+    // `headers` need only remain valid until acceptWebSocket() returns (it can be stack-allocated).
   };
 
   virtual kj::Promise<void> openWebSocket(
@@ -500,15 +542,25 @@ public:
 };
 
 kj::Own<HttpClient> newHttpClient(HttpHeaderTable& responseHeaderTable, kj::Network& network,
-                                  kj::Maybe<kj::Network&> tlsNetwork = nullptr);
+                                  kj::Maybe<kj::Network&> tlsNetwork = nullptr,
+                                  kj::Maybe<EntropySource&> entropySource = nullptr);
 // Creates a proxy HttpClient that connects to hosts over the given network.
 //
 // `responseHeaderTable` is used when parsing HTTP responses. Requests can use any header table.
 //
 // `tlsNetwork` is required to support HTTPS destination URLs. Otherwise, only HTTP URLs can be
 // fetched.
-
-kj::Own<HttpClient> newHttpClient(HttpHeaderTable& responseHeaderTable, kj::AsyncIoStream& stream);
+//
+// `entropySource` must be provided in order to use `openWebSocket`. If you don't need WebSockets,
+// `entropySource` can be omitted. The WebSocket protocol uses random values to avoid triggering
+// flaws (including security flaws) in certain HTTP proxy software. Specifically, entropy is used
+// to generate the `Sec-WebSocket-Key` header and to generate frame masks. If you know that there
+// are no broken or vulnerable proxies between you and the server, you can provide an dummy entropy
+// source that doesn't generate real entropy (e.g. returning the same value every time). Otherwise,
+// you must provide a cryptographically-random entropy source.
+
+kj::Own<HttpClient> newHttpClient(HttpHeaderTable& responseHeaderTable, kj::AsyncIoStream& stream,
+                                  kj::Maybe<EntropySource&> entropySource = nullptr);
 // Creates an HttpClient that speaks over the given pre-established connection. The client may
 // be used as a proxy client or a host client depending on whether the peer is operating as
 // a proxy.
@@ -518,11 +570,33 @@ kj::Own<HttpClient> newHttpClient(HttpHeaderTable& responseHeaderTable, kj::Asyn
 // fail as well. If the destination server chooses to close the connection after a response,
 // subsequent requests will fail. If a response takes a long time, it blocks subsequent responses.
 // If a WebSocket is opened successfully, all subsequent requests fail.
+//
+// `entropySource` must be provided in order to use `openWebSocket`. If you don't need WebSockets,
+// `entropySource` can be omitted. The WebSocket protocol uses random values to avoid triggering
+// flaws (including security flaws) in certain HTTP proxy software. Specifically, entropy is used
+// to generate the `Sec-WebSocket-Key` header and to generate frame masks. If you know that there
+// are no broken or vulnerable proxies between you and the server, you can provide an dummy entropy
+// source that doesn't generate real entropy (e.g. returning the same value every time). Otherwise,
+// you must provide a cryptographically-random entropy source.
 
 kj::Own<HttpClient> newHttpClient(HttpService& service);
 kj::Own<HttpService> newHttpService(HttpClient& client);
 // Adapts an HttpClient to an HttpService and vice versa.
 
+kj::Own<WebSocket> newWebSocket(kj::Own<kj::AsyncIoStream> stream,
+                                kj::Maybe<EntropySource&> maskEntropySource);
+// Create a new WebSocket on top of the given stream. It is assumed that the HTTP -> WebSocket
+// upgrade handshake has already occurred (or is not needed), and messages can immediately be
+// sent and received on the stream. Normally applications would not call this directly.
+//
+// `maskEntropySource` is used to generate cryptographically-random frame masks. If null, outgoing
+// frames will not be masked. Servers are required NOT to mask their outgoing frames, but clients
+// ARE required to do so. So, on the client side, you MUST specify an entropy source. The mask
+// must be crytographically random if the data being sent on the WebSocket may be malicious. The
+// purpose of the mask is to prevent badly-written HTTP proxies from interpreting "things that look
+// like HTTP requests" in a message as being actual HTTP requests, which could result in cache
+// poisoning. See RFC6455 section 10.3.
+
 struct HttpServerSettings {
   kj::Duration headerTimeout = 15 * kj::SECONDS;
   // After initial connection open, or after receiving the first byte of a pipelined request,

c++/src/kj/one-of-test.c++

@@ -98,4 +98,41 @@ TEST(OneOf, Copy) {
   EXPECT_STREQ("foo", var2.get<const char*>());
 }
 
+TEST(OneOf, Switch) {
+  OneOf<int, float, const char*> var;
+  var = "foo";
+  uint count = 0;
+
+  {
+    KJ_SWITCH_ONEOF(var) {
+      KJ_CASE_ONEOF(i, int) {
+        KJ_FAIL_ASSERT("expected char*, got int", i);
+      }
+      KJ_CASE_ONEOF(s, const char*) {
+        KJ_EXPECT(kj::StringPtr(s) == "foo");
+        ++count;
+      }
+      KJ_CASE_ONEOF(n, float) {
+        KJ_FAIL_ASSERT("expected char*, got float", n);
+      }
+    }
+  }
+
+  KJ_EXPECT(count == 1);
+
+  {
+    KJ_SWITCH_ONEOF(kj::cp(var)) {
+      KJ_CASE_ONEOF(i, int) {
+        KJ_FAIL_ASSERT("expected char*, got int", i);
+      }
+      KJ_CASE_ONEOF(s, const char*) {
+        KJ_EXPECT(kj::StringPtr(s) == "foo");
+      }
+      KJ_CASE_ONEOF(n, float) {
+        KJ_FAIL_ASSERT("expected char*, got float", n);
+      }
+    }
+  }
+}
+
 }  // namespace kj

c++/src/kj/one-of.h

@@ -37,6 +37,31 @@ struct TypeIndex_ { static constexpr uint value = TypeIndex_<i + 1, Key, Rest...
 template <uint i, typename Key, typename... Rest>
 struct TypeIndex_<i, Key, Key, Rest...> { static constexpr uint value = i; };
 
+enum class Variants0 {};
+enum class Variants1 { _variant0 };
+enum class Variants2 { _variant0, _variant1 };
+enum class Variants3 { _variant0, _variant1, _variant2 };
+enum class Variants4 { _variant0, _variant1, _variant2, _variant3 };
+enum class Variants5 { _variant0, _variant1, _variant2, _variant3, _variant4 };
+enum class Variants6 { _variant0, _variant1, _variant2, _variant3, _variant4, _variant5 };
+enum class Variants7 { _variant0, _variant1, _variant2, _variant3, _variant4, _variant5, _variant6 };
+enum class Variants8 { _variant0, _variant1, _variant2, _variant3, _variant4, _variant5, _variant6,
+                       _variant7 };
+
+template <uint i> struct Variants_;
+template <> struct Variants_<0> { typedef Variants0 Type; };
+template <> struct Variants_<1> { typedef Variants1 Type; };
+template <> struct Variants_<2> { typedef Variants2 Type; };
+template <> struct Variants_<3> { typedef Variants3 Type; };
+template <> struct Variants_<4> { typedef Variants4 Type; };
+template <> struct Variants_<5> { typedef Variants5 Type; };
+template <> struct Variants_<6> { typedef Variants6 Type; };
+template <> struct Variants_<7> { typedef Variants7 Type; };
+template <> struct Variants_<8> { typedef Variants8 Type; };
+
+template <uint i>
+using Variants = typename Variants_<i>::Type;
+
 }  // namespace _ (private)
 
 template <typename... Variants>
@@ -48,7 +73,12 @@ class OneOf {
 public:
   inline OneOf(): tag(0) {}
   OneOf(const OneOf& other) { copyFrom(other); }
+  OneOf(OneOf& other) { copyFrom(other); }
   OneOf(OneOf&& other) { moveFrom(other); }
+  template <typename T>
+  OneOf(T&& other): tag(typeIndex<Decay<T>>()) {
+    ctor(*reinterpret_cast<Decay<T>*>(space), kj::fwd<T>(other));
+  }
   ~OneOf() { destroy(); }
 
   OneOf& operator=(const OneOf& other) { if (tag != 0) destroy(); copyFrom(other); return *this; }
@@ -96,6 +126,22 @@ public:
   // block call allHandled<n>() where n is the number of variants. This will fail to compile
   // if new variants are added in the future.
 
+  typedef _::Variants<sizeof...(Variants)> Tag;
+
+  Tag which() {
+    KJ_IREQUIRE(tag != 0, "Can't KJ_SWITCH_ONEOF() on uninitialized value.");
+    return static_cast<Tag>(tag - 1);
+  }
+
+  template <typename T>
+  static constexpr Tag tagFor() {
+    return static_cast<Tag>(typeIndex<T>() - 1);
+  }
+
+  OneOf* _switchSubject() & { return this; }
+  const OneOf* _switchSubject() const& { return this; }
+  _::NullableValue<OneOf> _switchSubject() && { return kj::mv(*this); }
+
 private:
   uint tag;
 
@@ -150,6 +196,20 @@ private:
     doAll(copyVariantFrom<Variants>(other)...);
   }
 
+  template <typename T>
+  inline bool copyVariantFrom(OneOf& other) {
+    if (other.is<T>()) {
+      ctor(*reinterpret_cast<T*>(space), other.get<T>());
+    }
+    return false;
+  }
+  void copyFrom(OneOf& other) {
+    // Initialize as a copy of `other`.  Expects that `this` starts out uninitialized, so the tag
+    // is invalid.
+    tag = other.tag;
+    doAll(copyVariantFrom<Variants>(other)...);
+  }
+
   template <typename T>
   inline bool moveVariantFrom(OneOf& other) {
     if (other.is<T>()) {
@@ -176,6 +236,53 @@ void OneOf<Variants...>::allHandled() {
   KJ_UNREACHABLE;
 }
 
+#if __cplusplus > 201402L
+#define KJ_SWITCH_ONEOF(value) \
+  switch (auto _kj_switch_subject = value._switchSubject(); _kj_switch_subject->which())
+#else
+#define KJ_SWITCH_ONEOF(value) \
+  /* Without C++17, we can only support one switch per containing block. Deal with it. */ \
+  auto _kj_switch_subject = value._switchSubject(); \
+  switch (_kj_switch_subject->which())
+#endif
+#define KJ_CASE_ONEOF(name, ...) \
+    break; \
+  case ::kj::Decay<decltype(*_kj_switch_subject)>::tagFor<__VA_ARGS__>(): \
+    for (auto& name = _kj_switch_subject->get<__VA_ARGS__>(), *_kj_switch_done = &name; \
+         _kj_switch_done; _kj_switch_done = nullptr)
+#define KJ_CASE_ONEOF_DEFAULT break; default:
+// Allows switching over a OneOf.
+//
+// Example:
+//
+//     kj::OneOf<int, float, const char*> variant;
+//     KJ_SWITCH_ONEOF(variant) {
+//       KJ_CASE_ONEOF(i, int) {
+//         doSomethingWithInt(i);
+//       }
+//       KJ_CASE_ONEOF(s, const char*) {
+//         doSomethingWithString(s);
+//       }
+//       KJ_CASE_ONEOF_DEFAULT {
+//         doSomethingElse();
+//       }
+//     }
+//
+// Notes:
+// - If you don't handle all possible types and don't include a default branch, you'll get a
+//   compiler warning, just like a regular switch() over an enum where one of the enum values is
+//   missing.
+// - There's no need for a `break` statement in a KJ_CASE_ONEOF; it is implied.
+// - Under C++11 and C++14, only one KJ_SWITCH_ONEOF() can appear in a block. Wrap the switch in
+//   a pair of braces if you need a second switch in the same block. If C++17 is enabled, this is
+//   not an issue.
+//
+// Implementation notes:
+// - The use of __VA_ARGS__ is to account for template types that have commas separating type
+//   parameters, since macros don't recognize <> as grouping.
+// - _kj_switch_done is really used as a boolean flag to prevent the for() loop from actually
+//   looping, but it's defined as a pointer since that's all we can define in this context.
+
 }  // namespace kj
 
 #endif  // KJ_ONE_OF_H_

super-test.sh

@@ -11,6 +11,9 @@ QUICK=
 
 PARALLEL=$(nproc 2>/dev/null || echo 1)
 
+# Have automake dump test failure to stdout. Important for CI.
+export VERBOSE=true
+
 while [ $# -gt 0 ]; do
   case "$1" in
     -j* )