WebSocketChannel implementation

net/websockets/websocket_channel.cc

+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/websockets/websocket_channel.h"
+
+#include <algorithm>
+
+#include "base/basictypes.h"  // for size_t
+#include "base/bind.h"
+#include "base/safe_numerics.h"
+#include "base/strings/string_util.h"
+#include "net/base/big_endian.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_request_info.h"
+#include "net/http/http_stream_factory.h"
+#include "net/ssl/ssl_config_service.h"
+#include "net/websockets/websocket_errors.h"
+#include "net/websockets/websocket_event_interface.h"
+#include "net/websockets/websocket_frame.h"
+#include "net/websockets/websocket_mux.h"
+#include "net/websockets/websocket_stream.h"
+
+namespace net {
+
+namespace {
+
+const int kDefaultSendQuotaLowWaterMark = 1 << 16;
+const int kDefaultSendQuotaHighWaterMark = 1 << 17;
+const size_t kWebSocketCloseCodeLength = 2;
+
+// Concatenate the data from two IOBufferWithSize objects into a single one.
+IOBufferWithSize* ConcatenateIOBuffers(
+    const scoped_refptr<IOBufferWithSize>& part1,
+    const scoped_refptr<IOBufferWithSize>& part2) {
+  int newsize = part1->size() + part2->size();
+  IOBufferWithSize* newbuffer = new IOBufferWithSize(newsize);
+  std::copy(part1->data(), part1->data() + part1->size(), newbuffer->data());
+  std::copy(part2->data(),
+            part2->data() + part2->size(),
+            newbuffer->data() + part1->size());
+  return newbuffer;
+}
+
+}  // namespace
+
+struct WebSocketChannel::SendBuffer {
+  SendBuffer() : total_bytes(0) {}
+  ScopedVector<WebSocketFrameChunk> frames;
+  size_t total_bytes;
+};
+
+// Implementation of WebSocketStream::ConnectDelegate that simply forwards the
+// calls on to the WebSocketChannel that created it.
+class WebSocketChannel::ConnectDelegate
+    : public WebSocketStream::ConnectDelegate {
+ public:
+  explicit ConnectDelegate(WebSocketChannel* creator) : creator_(creator) {}
+
+  virtual void OnSuccess(scoped_ptr<WebSocketStream> stream) OVERRIDE {
+    creator_->OnConnectSuccess(stream.Pass());
+  }
+
+  virtual void OnFailure(unsigned short websocket_error) OVERRIDE {
+    creator_->OnConnectFailure(websocket_error);
+  }
+
+ private:
+  // A pointer to the WebSocketChannel that created us. We do not need to worry
+  // about this pointer being stale, because deleting WebSocketChannel cancels
+  // the connect process, deleting this object and preventing its callbacks from
+  //  being called.
+  WebSocketChannel* const creator_;
+
+  DISALLOW_COPY_AND_ASSIGN(ConnectDelegate);
+};
+
+WebSocketChannel::WebSocketChannel(
+    const GURL& socket_url,
+    scoped_ptr<WebSocketEventInterface> event_interface)
+    : socket_url_(socket_url),
+      event_interface_(event_interface.Pass()),
+      send_quota_low_water_mark_(kDefaultSendQuotaLowWaterMark),
+      send_quota_high_water_mark_(kDefaultSendQuotaHighWaterMark),
+      current_send_quota_(0),
+      state_(FRESHLY_CONSTRUCTED),
+      weak_factory_(this) {}
+
+WebSocketChannel::~WebSocketChannel() {
+  // The stream may hold a pointer to read_frame_chunks_, and so it needs to be
+  // destroyed first.
+  stream_.reset();
+}
+
+void WebSocketChannel::SendAddChannelRequest(
+    const std::vector<std::string>& requested_subprotocols,
+    const GURL& origin,
+    URLRequestContext* url_request_context) {
+  // Delegate to the tested version.
+  SendAddChannelRequestWithFactory(
+      requested_subprotocols,
+      origin,
+      url_request_context,
+      base::Bind(&WebSocketStream::CreateAndConnectStream));
+}
+
+void WebSocketChannel::SendAddChannelRequestWithFactory(
+    const std::vector<std::string>& requested_subprotocols,
+    const GURL& origin,
+    URLRequestContext* url_request_context,
+    base::Callback<scoped_ptr<WebSocketStreamRequest>(
+        const GURL&,
+        const std::vector<std::string>&,
+        const GURL&,
+        URLRequestContext*,
+        const BoundNetLog&,
+        scoped_ptr<WebSocketStream::ConnectDelegate>)> factory) {
+  DCHECK_EQ(FRESHLY_CONSTRUCTED, state_);
+  scoped_ptr<WebSocketStream::ConnectDelegate> connect_delegate(
+      new WebSocketChannel::ConnectDelegate(this));
+  stream_request_ = factory.Run(socket_url_,
+                                requested_subprotocols,
+                                origin,
+                                url_request_context,
+                                BoundNetLog(),
+                                connect_delegate.Pass());
+  state_ = CONNECTING;
+}
+
+void WebSocketChannel::OnConnectSuccess(scoped_ptr<WebSocketStream> stream) {
+  DCHECK(stream);
+  DCHECK_EQ(CONNECTING, state_);
+  stream_ = stream.Pass();
+  event_interface_->OnAddChannelResponse(false, stream_->GetSubProtocol());

[tyoshino (SeeGerritForStatus), 2013/07/01 05:28:34]

what action are the implementors of this interface allowed to take? they never
call back to this instance from this (OnAddChannelResponse()) method?

basically i think it's better that we finish update of variables before calling
this. i.e. state_ = CONNECTED, etc. but it depends on semantics of this
interface you give.

[Adam Rice, 2013/07/01 07:59:05]

OnAddChannelResponse(true, ...) ie. "Connection Failed" is permitted to delete
the WebSocketChannel object.

OnDropChannel() should probably also permit deleting the WebSocketChannel
object, except that with the current implementation that will break RFC6455
semantics, because it will cause the browser to close the connection without
waiting for a server response. I haven't decided what to do about that yet.

For testing purposes it may be useful to call WebSocketChannel methods
re-entrantly, so I have reordered the code in this and a few other places to
make that safer.

+  // TODO(ricea): Get flow control information from the WebSocketStream once we
+  // have a multiplexing WebSocketStream.
+  event_interface_->OnFlowControl(send_quota_high_water_mark_);
+  current_send_quota_ = send_quota_high_water_mark_;
+  state_ = CONNECTED;
+  // We don't need this any more.
+  stream_request_.reset();
+  ReadFrames();
+}
+
+void WebSocketChannel::OnConnectFailure(unsigned short web_socket_error) {
+  DCHECK_EQ(CONNECTING, state_);
+  event_interface_->OnAddChannelResponse(true, "");
+  stream_request_.reset();
+  state_ = CLOSED;
+}
+
+void WebSocketChannel::SendFrame(bool fin,
+                                 WebSocketFrameHeader::OpCode op_code,
+                                 const std::vector<char>& data) {
+  if (data.size() > INT_MAX) {
+    NOTREACHED() << "Frame size sanity check failed";
+    return;
+  }
+  if (stream_ == NULL) {
+    LOG(DFATAL) << "Got SendFrame without a connection established; "
+                << "misbehaving renderer? fin=" << fin << " op_code=" << op_code
+                << " data.size()=" << data.size();
+    return;
+  }
+  if (state_ == SEND_CLOSED || state_ == RECV_CLOSED || state_ == CLOSED) {
+    VLOG(1) << "SendFrame called in state " << state_
+            << ". This may be a bug, or a harmless race.";
+    return;
+  }
+  if (state_ != CONNECTED) {
+    NOTREACHED() << "SendFrame() called in state " << state_;
+    return;
+  }
+  if (data.size() > base::checked_numeric_cast<size_t>(current_send_quota_)) {
+    FailChannel(SEND_INTERNAL_ERROR,
+                kWebSocketMuxErrorSendQuotaViolation,
+                "Send quota exceeded");
+    return;
+  }
+  if (!WebSocketFrameHeader::IsKnownDataOpCode(op_code)) {
+    LOG(DFATAL) << "Got SendFrame with bogus op_code " << op_code
+                << "; misbehaving renderer? fin=" << fin
+                << " data.size()=" << data.size();
+    return;
+  }
+  current_send_quota_ -= data.size();
+  // TODO(ricea): If current_send_quota_ has dropped below
+  // send_quota_low_water_mark_, we may want to consider increasing the "low
+  // water mark" and "high water mark", but only if we think we are not
+  // saturating the link to the WebSocket server.
+  // TODO(ricea): For kOpCodeText, do UTF-8 validation?
+  scoped_refptr<IOBufferWithSize> buffer(new IOBufferWithSize(data.size()));
+  std::copy(data.begin(), data.end(), buffer->data());
+  SendIOBufferWithSize(fin, op_code, buffer);
+}
+
+void WebSocketChannel::SendIOBufferWithSize(
+    bool fin,
+    WebSocketFrameHeader::OpCode op_code,
+    const scoped_refptr<IOBufferWithSize>& buffer) {
+  DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
+  DCHECK(stream_);
+  scoped_ptr<WebSocketFrameHeader> header(new WebSocketFrameHeader(op_code));
+  header->final = fin;
+  header->masked = true;
+  header->payload_length = buffer->size();
+  scoped_ptr<WebSocketFrameChunk> chunk(new WebSocketFrameChunk());
+  chunk->header = header.Pass();
+  chunk->final_chunk = true;
+  chunk->data = buffer;
+  if (currently_sending_) {
+    // Either the link to the WebSocket server is saturated, or we are simply
+    // processing a batch of messages.
+    // TODO(ricea): We need to keep some statistics to work out which situation
+    // we are in and adjust quota appropriately.
+    if (!send_next_) {
+      send_next_.reset(new SendBuffer);
+    }
+    send_next_->frames.push_back(chunk.release());
+    send_next_->total_bytes += buffer->size();
+  } else {
+    currently_sending_.reset(new SendBuffer);
+    currently_sending_->frames.push_back(chunk.release());
+    currently_sending_->total_bytes += buffer->size();
+    WriteFrames();
+  }
+}
+
+void WebSocketChannel::WriteFrames() {
+  // This is safe because we own the WebSocketStream and destroying it cancels
+  // all callbacks.
+  int result = stream_->WriteFrames(
+      &(currently_sending_->frames),
+      base::Bind(&WebSocketChannel::OnWriteDone, base::Unretained(this)));
+  if (result != ERR_IO_PENDING) {
+    OnWriteDone(result);
+  }
+}
+
+void WebSocketChannel::OnWriteDone(int result) {
+  DCHECK(state_ != FRESHLY_CONSTRUCTED && state_ != CONNECTING);
+  DCHECK_NE(ERR_IO_PENDING, result);
+  DCHECK(currently_sending_);
+  switch (result) {
+    case OK:
+      if (send_next_) {
+        currently_sending_ = send_next_.Pass();
+        WriteFrames();
+      } else {
+        currently_sending_.reset();
+        if (current_send_quota_ < send_quota_low_water_mark_) {
+          // TODO(ricea): Increase low_water_mark and high_water_mark if
+          // throughput is high, reduce them if throughput is low.  Low water
+          // mark needs to be >= the bandwidth delay product *of the IPC
+          // channel*. Because factors like context-switch time, thread wake-up
+          // time, and bus speed come into play it is complex and probably needs
+          // to be determined empirically.
+          DCHECK_LE(send_quota_low_water_mark_, send_quota_high_water_mark_);
+          // TODO(ricea): Truncate quota by the quota specified by the remote
+          // server, if the protocol in use supports quota.
+          int fresh_quota = send_quota_high_water_mark_ - current_send_quota_;
+          event_interface_->OnFlowControl(fresh_quota);
+          current_send_quota_ += fresh_quota;
+        }
+      }
+      break;
+
+    // If a recoverable error condition existed, it would go here.
+
+    default:
+      DCHECK_LT(result, 0)
+          << "WriteFrames() should only return OK or ERR_ codes";
+      stream_->Close();
+      state_ = CLOSED;
+      event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                      "Abnormal Closure");
+      break;
+  }
+}
+
+void WebSocketChannel::ReadFrames() {
+  // This use if base::Unretained is safe because we own the WebSocketStream,

[tyoshino (SeeGerritForStatus), 2013/07/01 05:05:18]

if -> of?

[Adam Rice, 2013/07/01 07:59:05]

Sorry, yes. Fixed.

+  // and any pending reads will be cancelled when it is destroyed.
+  int result = stream_->ReadFrames(
+      &read_frame_chunks_,
+      base::Bind(&WebSocketChannel::OnReadDone, base::Unretained(this)));
+  if (result != ERR_IO_PENDING) {
+    OnReadDone(result);
+  }
+}
+
+void WebSocketChannel::OnReadDone(int result) {
+  DCHECK(state_ != FRESHLY_CONSTRUCTED && state_ != CONNECTING);
+  DCHECK_NE(ERR_IO_PENDING, result);
+  switch (result) {
+    case OK:
+      // ReadFrames() must use ERR_CONNECTION_CLOSED for a closed connection
+      // with no data read, not an empty response.
+      DCHECK(!read_frame_chunks_.empty())
+          << "ReadFrames() returned OK, but nothing was read.";
+      for (size_t i = 0; i < read_frame_chunks_.size(); ++i) {
+        scoped_ptr<WebSocketFrameChunk> chunk(read_frame_chunks_[i]);
+        read_frame_chunks_[i] = NULL;
+        ProcessFrameChunk(chunk.Pass());
+      }
+      read_frame_chunks_.clear();
+      // We need to always keep a call to ReadFrames pending.
+      ReadFrames();
+      return;
+
+    case ERR_CONNECTION_CLOSED: {
+      State old_state = state_;
+      state_ = CLOSED;
+      if (old_state != RECV_CLOSED && old_state != CLOSED) {
+        // We need to inform the render process of the unexpected closure.
+        event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                        "Abnormal Closure");
+      }
+      return;
+    }
+
+    default: {
+      DCHECK_LT(result, 0)
+          << "ReadFrames() should only return OK or ERR_ codes";
+      stream_->Close();
+      State old_state = state_;
+      state_ = CLOSED;
+      if (old_state != RECV_CLOSED && old_state != CLOSED) {
+        event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                        "Abnormal Closure");
+      }
+      return;
+    }
+  }
+}
+
+// TODO(ricea): This method is too long. Break it up.
+void WebSocketChannel::ProcessFrameChunk(
+    scoped_ptr<WebSocketFrameChunk> chunk) {
+  bool first_chunk = false;
+  if (chunk->header) {
+    first_chunk = true;
+    current_frame_header_.swap(chunk->header);
+    if (current_frame_header_->masked) {
+      // RFC6455 Section 5.1 "A client MUST close a connection if it detects a
+      // masked frame."
+      FailChannel(SEND_REAL_ERROR,
+                  kWebSocketErrorProtocolError,
+                  "Masked frame from server");
+      return;
+    }
+  }
+  if (!current_frame_header_) {
+    DCHECK(state_ != CONNECTED) << "Unexpected header-less frame received "
+                                << "(final_chunk = " << chunk->final_chunk
+                                << ", data size = " << chunk->data->size()
+                                << ")";
+    return;
+  }
+  scoped_refptr<IOBufferWithSize> data_buffer;
+  data_buffer.swap(chunk->data);
+  WebSocketFrameHeader::OpCode opcode = current_frame_header_->opcode;
+  if (WebSocketFrameHeader::IsKnownControlOpCode(opcode)) {
+    if (chunk->final_chunk) {
+      if (incomplete_control_frame_body_) {
+        VLOG(2) << "Rejoining a split control frame, opcode "
+                << opcode;
+        data_buffer = ConcatenateIOBuffers(incomplete_control_frame_body_,
+                                           data_buffer);
+        incomplete_control_frame_body_ = NULL;
+      }
+    } else {
+      // TODO(ricea): Enforce a maximum size of 125 bytes on the control frames
+      // we accept.
+      VLOG(2) << "Encountered a split control frame, opcode "
+              << opcode;
+      if (incomplete_control_frame_body_) {
+        // The really horrid case. We need to create a new IOBufferWithSize
+        // combining the new one and the old one. This should virtually never
+        // happen.
+        // TODO(ricea): This algorithm is O(N^2). Use a fixed 127-byte byffer
+        // instead.
+        VLOG(3) << "Hit the really horrid case";
+        incomplete_control_frame_body_ =
+            ConcatenateIOBuffers(incomplete_control_frame_body_, data_buffer);
+      } else {
+        // The merely horrid case. Store the IOBufferWithSize to use when the
+        // rest of the control frame arrives.
+        incomplete_control_frame_body_.swap(data_buffer);
+      }
+      return;
+    }
+  }
+
+  switch (opcode) {
+    case WebSocketFrameHeader::kOpCodeText:  // fall-thru
+    case WebSocketFrameHeader::kOpCodeBinary:
+      if (!first_chunk) {
+        opcode = WebSocketFrameHeader::kOpCodeContinuation;
+      }
+    // fall-thru
+    case WebSocketFrameHeader::kOpCodeContinuation:
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Data packet received after close");
+        return;
+      } else if (state_ == CONNECTED) {
+        const bool final = chunk->final_chunk && current_frame_header_->final;
+        // TODO(ricea): Can this copy be eliminated?
+        const char* const data_begin = data_buffer->data();
+        const char* const data_end = data_begin + data_buffer->size();
+        const std::vector<char> data(data_begin, data_end);
+        // TODO(ricea): Handle the (improbable) case when ReadFrames returns far
+        // more data at once than we want to send in a single IPC (in which case
+        // we need to buffer the data and return to the event loop with a
+        // callback to send the rest in 32K chunks).
+
+        // Send the received frame to the renderer process.
+        event_interface_->OnDataFrame(final, opcode, data);
+      } else {
+        VLOG(3) << "Ignored data packet received in state " << state_;
+      }
+      break;
+
+    case WebSocketFrameHeader::kOpCodePing:
+      VLOG(1) << "Got Ping of size " << data_buffer->size();
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Ping received after Close");
+        return;
+      } else if (state_ == CONNECTED) {
+        SendIOBufferWithSize(
+            true, WebSocketFrameHeader::kOpCodePong, data_buffer);
+      } else {
+        VLOG(3) << "Ignored ping in state " << state_;
+      }
+      break;
+
+    case WebSocketFrameHeader::kOpCodePong:
+      VLOG(1) << "Got Pong of size " << data_buffer->size();
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Pong received after Close");
+        return;
+      }
+      // We do not need to do anything with pong messages.
+      break;
+
+    case WebSocketFrameHeader::kOpCodeClose: {
+      unsigned short code = kWebSocketNormalClosure;
+      std::string reason;
+      ParseClose(data_buffer, &code, &reason);
+      // TODO(ricea): Find a way to safely log the message from the close
+      // message (escape control codes and so on).
+      VLOG(1) << "Got Close with code " << code;
+      switch (state_) {
+        case CONNECTED:
+          state_ = RECV_CLOSED;
+          SendClose(code, reason);
+          event_interface_->OnDropChannel(code, reason);
+          break;
+
+        case RECV_CLOSED:
+          FailChannel(SEND_REAL_ERROR,
+                      kWebSocketErrorProtocolError,
+                      "Close received after Close");
+          break;
+
+        case SEND_CLOSED:
+          state_ = CLOSED;
+          event_interface_->OnDropChannel(code, reason);
+          break;
+
+        default:
+          LOG(DFATAL) << "Got Close in unexpected state " << state_;
+          break;
+      }
+      break;
+    }
+
+    default:
+      FailChannel(SEND_REAL_ERROR,
+                  kWebSocketErrorProtocolError,
+                  "Unknown opcode");
+      break;
+  }
+  if (chunk->final_chunk) {
+    // Make sure we do not apply this frame header to any future chunks.
+    current_frame_header_.reset();
+  }
+}
+
+void WebSocketChannel::SendFlowControl(int64 quota) {
+  DCHECK_EQ(CONNECTED, state_);
+  // TODO(ricea): Add interface to WebSocketStream and implement.
+  // stream_->SendFlowControl(quota);
+}
+
+void WebSocketChannel::SendDropChannel(unsigned short code,
+                                       const std::string& reason) {
+  if (state_ == SEND_CLOSED || state_ == CLOSED) {
+    VLOG(1) << "SendDropChannel called in state " << state_
+            << ". This may be a bug, or a harmless race.";
+    return;
+  }
+  DCHECK_EQ(CONNECTED, state_);
+  // TODO(ricea): Validate |code|? Check that |reason| is valid UTF-8?
+  // TODO(ricea): There should be a timeout for the closing handshake.
+  SendClose(code, reason);
+}
+
+void WebSocketChannel::FailChannel(ExposeError expose,
+                                   unsigned short code,
+                                   const std::string& reason) {
+  // TODO(ricea): Logging.
+  State old_state = state_;
+  if (state_ == CONNECTED) {
+    unsigned short send_code = kWebSocketErrorGoingAway;
+    std::string send_reason = "Internal Error";
+    if (expose == SEND_REAL_ERROR) {
+      send_code = code;
+      send_reason = reason;
+    }
+    SendClose(send_code, send_reason);
+  }
+  // This method is mostly called in response to an invalid frame, in
+  // which case we should not re-use the header.
+  current_frame_header_.reset();
+  if (old_state != RECV_CLOSED && old_state != CLOSED) {
+    event_interface_->OnDropChannel(code, reason);
+  }
+}
+
+void WebSocketChannel::SendClose(unsigned short code,
+                                 const std::string& reason) {
+  DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
+  uint64 payload_length = kWebSocketCloseCodeLength + reason.length();
+  scoped_refptr<IOBufferWithSize> body = new IOBufferWithSize(payload_length);
+  WriteBigEndian(body->data(), code);
+  COMPILE_ASSERT(sizeof(code) == kWebSocketCloseCodeLength,
+                 they_should_both_be_two);
+  std::copy(reason.begin(),
+            reason.end(),
+            body->data() + kWebSocketCloseCodeLength);
+  SendIOBufferWithSize(true, WebSocketFrameHeader::kOpCodeClose, body);
+  state_ = state_ == CONNECTED ? SEND_CLOSED : CLOSED;
+}
+
+void WebSocketChannel::ParseClose(const scoped_refptr<IOBufferWithSize>& buffer,
+                                  unsigned short* code,
+                                  std::string* reason) {
+  const char* data = buffer->data();
+  size_t size = base::checked_numeric_cast<size_t>(buffer->size());
+  reason->clear();
+  if (size < kWebSocketCloseCodeLength) {
+    *code = kWebSocketErrorNoStatusReceived;
+    if (size != 0) {
+      VLOG(1) << "Close frame with payload size " << size << " received "
+              << "(the first byte is " << std::hex << static_cast<int>(data[0])
+              << ")";
+      return;
+    }
+    return;
+  }
+  unsigned short unchecked_code = 0;
+  ReadBigEndian(data, &unchecked_code);
+  COMPILE_ASSERT(sizeof(unchecked_code) == kWebSocketCloseCodeLength,
+                 they_should_both_be_two_bytes);
+  if (unchecked_code >=
+          static_cast<unsigned short>(kWebSocketNormalClosure) &&
+      unchecked_code <=
+          static_cast<unsigned short>(kWebSocketErrorPrivateReservedMax)) {
+    *code = unchecked_code;
+  } else {
+    VLOG(1) << "Close frame contained code outside of the valid range: "
+            << unchecked_code;
+    *code = kWebSocketErrorProtocolError;
+  }
+  std::string text(data + kWebSocketCloseCodeLength, data + size);
+  // TODO(ricea): Is this check strict enough? In particular, check the
+  // "Security Considerations" from RFC3629.
+  if (IsStringUTF8(text)) {
+    using std::swap;
+    swap(*reason, text);
+  }
+}
+
+}  // namespace net