src/net/websockets/websocket_channel.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/websockets/websocket_channel.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/basictypes.h"  // for size_t
  10 #include "base/bind.h"
  11 #include "base/compiler_specific.h"
  12 #include "base/memory/weak_ptr.h"
  13 #include "base/message_loop/message_loop.h"
  14 #include "base/numerics/safe_conversions.h"
  15 #include "base/stl_util.h"
  16 #include "base/strings/stringprintf.h"
  17 #include "base/time/time.h"
  18 #include "net/base/big_endian.h"
  19 #include "net/base/io_buffer.h"
  20 #include "net/base/net_log.h"
  21 #include "net/http/http_request_headers.h"
  22 #include "net/http/http_response_headers.h"
  23 #include "net/http/http_util.h"
  24 #include "net/websockets/websocket_errors.h"
  25 #include "net/websockets/websocket_event_interface.h"
  26 #include "net/websockets/websocket_frame.h"
  27 #include "net/websockets/websocket_handshake_request_info.h"
  28 #include "net/websockets/websocket_handshake_response_info.h"
  29 #include "net/websockets/websocket_mux.h"
  30 #include "net/websockets/websocket_stream.h"
  31
  32 namespace net {
  33
  34 namespace {
  35
  36 using base::StreamingUtf8Validator;
  37
  38 const int kDefaultSendQuotaLowWaterMark = 1 << 16;
  39 const int kDefaultSendQuotaHighWaterMark = 1 << 17;
  40 const size_t kWebSocketCloseCodeLength = 2;
  41 // This timeout is based on TCPMaximumSegmentLifetime * 2 from
  42 // MainThreadWebSocketChannel.cpp in Blink.
  43 const int kClosingHandshakeTimeoutSeconds = 2 * 2 * 60;
  44
  45 typedef WebSocketEventInterface::ChannelState ChannelState;
  46 const ChannelState CHANNEL_ALIVE = WebSocketEventInterface::CHANNEL_ALIVE;
  47 const ChannelState CHANNEL_DELETED = WebSocketEventInterface::CHANNEL_DELETED;
  48
  49 // Maximum close reason length = max control frame payload -
  50 //                               status code length
  51 //                             = 125 - 2
  52 const size_t kMaximumCloseReasonLength = 125 - kWebSocketCloseCodeLength;
  53
  54 // Check a close status code for strict compliance with RFC6455. This is only
  55 // used for close codes received from a renderer that we are intending to send
  56 // out over the network. See ParseClose() for the restrictions on incoming close
  57 // codes. The |code| parameter is type int for convenience of implementation;
  58 // the real type is uint16.
  59 bool IsStrictlyValidCloseStatusCode(int code) {
  60   static const int kInvalidRanges[] = {
  61       // [BAD, OK)
  62       0,    1000,   // 1000 is the first valid code
  63       1005, 1007,   // 1005 and 1006 MUST NOT be set.
  64       1014, 3000,   // 1014 unassigned; 1015 up to 2999 are reserved.
  65       5000, 65536,  // Codes above 5000 are invalid.
  66   };
  67   const int* const kInvalidRangesEnd =
  68       kInvalidRanges + arraysize(kInvalidRanges);
  69
  70   DCHECK_GE(code, 0);
  71   DCHECK_LT(code, 65536);
  72   const int* upper = std::upper_bound(kInvalidRanges, kInvalidRangesEnd, code);
  73   DCHECK_NE(kInvalidRangesEnd, upper);
  74   DCHECK_GT(upper, kInvalidRanges);
  75   DCHECK_GT(*upper, code);
  76   DCHECK_LE(*(upper - 1), code);
  77   return ((upper - kInvalidRanges) % 2) == 0;
  78 }
  79
  80 // This function avoids a bunch of boilerplate code.
  81 void AllowUnused(ChannelState ALLOW_UNUSED unused) {}
  82
  83 // Sets |name| to the name of the frame type for the given |opcode|. Note that
  84 // for all of Text, Binary and Continuation opcode, this method returns
  85 // "Data frame".
  86 void GetFrameTypeForOpcode(WebSocketFrameHeader::OpCode opcode,
  87                            std::string* name) {
  88   switch (opcode) {
  89     case WebSocketFrameHeader::kOpCodeText:    // fall-thru
  90     case WebSocketFrameHeader::kOpCodeBinary:  // fall-thru
  91     case WebSocketFrameHeader::kOpCodeContinuation:
  92       *name = "Data frame";
  93       break;
  94
  95     case WebSocketFrameHeader::kOpCodePing:
  96       *name = "Ping";
  97       break;
  98
  99     case WebSocketFrameHeader::kOpCodePong:
 100       *name = "Pong";
 101       break;
 102
 103     case WebSocketFrameHeader::kOpCodeClose:
 104       *name = "Close";
 105       break;
 106
 107     default:
 108       *name = "Unknown frame type";
 109       break;
 110   }
 111
 112   return;
 113 }
 114
 115 }  // namespace
 116
 117 // A class to encapsulate a set of frames and information about the size of
 118 // those frames.
 119 class WebSocketChannel::SendBuffer {
 120  public:
 121   SendBuffer() : total_bytes_(0) {}
 122
 123   // Add a WebSocketFrame to the buffer and increase total_bytes_.
 124   void AddFrame(scoped_ptr<WebSocketFrame> chunk);
 125
 126   // Return a pointer to the frames_ for write purposes.
 127   ScopedVector<WebSocketFrame>* frames() { return &frames_; }
 128
 129  private:
 130   // The frames_ that will be sent in the next call to WriteFrames().
 131   ScopedVector<WebSocketFrame> frames_;
 132
 133   // The total size of the payload data in |frames_|. This will be used to
 134   // measure the throughput of the link.
 135   // TODO(ricea): Measure the throughput of the link.
 136   size_t total_bytes_;
 137 };
 138
 139 void WebSocketChannel::SendBuffer::AddFrame(scoped_ptr<WebSocketFrame> frame) {
 140   total_bytes_ += frame->header.payload_length;
 141   frames_.push_back(frame.release());
 142 }
 143
 144 // Implementation of WebSocketStream::ConnectDelegate that simply forwards the
 145 // calls on to the WebSocketChannel that created it.
 146 class WebSocketChannel::ConnectDelegate
 147     : public WebSocketStream::ConnectDelegate {
 148  public:
 149   explicit ConnectDelegate(WebSocketChannel* creator) : creator_(creator) {}
 150
 151   virtual void OnSuccess(scoped_ptr<WebSocketStream> stream) OVERRIDE {
 152     creator_->OnConnectSuccess(stream.Pass());
 153     // |this| may have been deleted.
 154   }
 155
 156   virtual void OnFailure(const std::string& message) OVERRIDE {
 157     creator_->OnConnectFailure(message);
 158     // |this| has been deleted.
 159   }
 160
 161   virtual void OnStartOpeningHandshake(
 162       scoped_ptr<WebSocketHandshakeRequestInfo> request) OVERRIDE {
 163     creator_->OnStartOpeningHandshake(request.Pass());
 164   }
 165
 166   virtual void OnFinishOpeningHandshake(
 167       scoped_ptr<WebSocketHandshakeResponseInfo> response)
 168       OVERRIDE {
 169     creator_->OnFinishOpeningHandshake(response.Pass());
 170   }
 171
 172  private:
 173   // A pointer to the WebSocketChannel that created this object. There is no
 174   // danger of this pointer being stale, because deleting the WebSocketChannel
 175   // cancels the connect process, deleting this object and preventing its
 176   // callbacks from being called.
 177   WebSocketChannel* const creator_;
 178
 179   DISALLOW_COPY_AND_ASSIGN(ConnectDelegate);
 180 };
 181
 182 class WebSocketChannel::HandshakeNotificationSender
 183     : public base::SupportsWeakPtr<HandshakeNotificationSender> {
 184  public:
 185   explicit HandshakeNotificationSender(WebSocketChannel* channel);
 186   ~HandshakeNotificationSender();
 187
 188   static void Send(base::WeakPtr<HandshakeNotificationSender> sender);
 189
 190   ChannelState SendImmediately(WebSocketEventInterface* event_interface);
 191
 192   const WebSocketHandshakeRequestInfo* handshake_request_info() const {
 193     return handshake_request_info_.get();
 194   }
 195
 196   void set_handshake_request_info(
 197       scoped_ptr<WebSocketHandshakeRequestInfo> request_info) {
 198     handshake_request_info_ = request_info.Pass();
 199   }
 200
 201   const WebSocketHandshakeResponseInfo* handshake_response_info() const {
 202     return handshake_response_info_.get();
 203   }
 204
 205   void set_handshake_response_info(
 206       scoped_ptr<WebSocketHandshakeResponseInfo> response_info) {
 207     handshake_response_info_ = response_info.Pass();
 208   }
 209
 210  private:
 211   WebSocketChannel* owner_;
 212   scoped_ptr<WebSocketHandshakeRequestInfo> handshake_request_info_;
 213   scoped_ptr<WebSocketHandshakeResponseInfo> handshake_response_info_;
 214 };
 215
 216 WebSocketChannel::HandshakeNotificationSender::HandshakeNotificationSender(
 217     WebSocketChannel* channel) : owner_(channel) {}
 218
 219 WebSocketChannel::HandshakeNotificationSender::~HandshakeNotificationSender() {}
 220
 221 void WebSocketChannel::HandshakeNotificationSender::Send(
 222     base::WeakPtr<HandshakeNotificationSender> sender) {
 223   // Do nothing if |sender| is already destructed.
 224   if (sender) {
 225     WebSocketChannel* channel = sender->owner_;
 226     AllowUnused(sender->SendImmediately(channel->event_interface_.get()));
 227   }
 228 }
 229
 230 ChannelState WebSocketChannel::HandshakeNotificationSender::SendImmediately(
 231     WebSocketEventInterface* event_interface) {
 232
 233   if (handshake_request_info_.get()) {
 234     if (CHANNEL_DELETED == event_interface->OnStartOpeningHandshake(
 235             handshake_request_info_.Pass()))
 236       return CHANNEL_DELETED;
 237   }
 238
 239   if (handshake_response_info_.get()) {
 240     if (CHANNEL_DELETED == event_interface->OnFinishOpeningHandshake(
 241             handshake_response_info_.Pass()))
 242       return CHANNEL_DELETED;
 243
 244     // TODO(yhirano): We can release |this| to save memory because
 245     // there will be no more opening handshake notification.
 246   }
 247
 248   return CHANNEL_ALIVE;
 249 }
 250
 251 WebSocketChannel::WebSocketChannel(
 252     scoped_ptr<WebSocketEventInterface> event_interface,
 253     URLRequestContext* url_request_context)
 254     : event_interface_(event_interface.Pass()),
 255       url_request_context_(url_request_context),
 256       send_quota_low_water_mark_(kDefaultSendQuotaLowWaterMark),
 257       send_quota_high_water_mark_(kDefaultSendQuotaHighWaterMark),
 258       current_send_quota_(0),
 259       timeout_(base::TimeDelta::FromSeconds(kClosingHandshakeTimeoutSeconds)),
 260       received_close_code_(0),
 261       state_(FRESHLY_CONSTRUCTED),
 262       notification_sender_(new HandshakeNotificationSender(this)),
 263       sending_text_message_(false),
 264       receiving_text_message_(false) {}
 265
 266 WebSocketChannel::~WebSocketChannel() {
 267   // The stream may hold a pointer to read_frames_, and so it needs to be
 268   // destroyed first.
 269   stream_.reset();
 270   // The timer may have a callback pointing back to us, so stop it just in case
 271   // someone decides to run the event loop from their destructor.
 272   timer_.Stop();
 273 }
 274
 275 void WebSocketChannel::SendAddChannelRequest(
 276     const GURL& socket_url,
 277     const std::vector<std::string>& requested_subprotocols,
 278     const GURL& origin) {
 279   // Delegate to the tested version.
 280   SendAddChannelRequestWithSuppliedCreator(
 281       socket_url,
 282       requested_subprotocols,
 283       origin,
 284       base::Bind(&WebSocketStream::CreateAndConnectStream));
 285 }
 286
 287 bool WebSocketChannel::InClosingState() const {
 288   // The state RECV_CLOSED is not supported here, because it is only used in one
 289   // code path and should not leak into the code in general.
 290   DCHECK_NE(RECV_CLOSED, state_)
 291       << "InClosingState called with state_ == RECV_CLOSED";
 292   return state_ == SEND_CLOSED || state_ == CLOSE_WAIT || state_ == CLOSED;
 293 }
 294
 295 void WebSocketChannel::SendFrame(bool fin,
 296                                  WebSocketFrameHeader::OpCode op_code,
 297                                  const std::vector<char>& data) {
 298   if (data.size() > INT_MAX) {
 299     NOTREACHED() << "Frame size sanity check failed";
 300     return;
 301   }
 302   if (stream_ == NULL) {
 303     LOG(DFATAL) << "Got SendFrame without a connection established; "
 304                 << "misbehaving renderer? fin=" << fin << " op_code=" << op_code
 305                 << " data.size()=" << data.size();
 306     return;
 307   }
 308   if (InClosingState()) {
 309     VLOG(1) << "SendFrame called in state " << state_
 310             << ". This may be a bug, or a harmless race.";
 311     return;
 312   }
 313   if (state_ != CONNECTED) {
 314     NOTREACHED() << "SendFrame() called in state " << state_;
 315     return;
 316   }
 317   if (data.size() > base::checked_cast<size_t>(current_send_quota_)) {
 318     // TODO(ricea): Kill renderer.
 319     AllowUnused(
 320         FailChannel("Send quota exceeded", kWebSocketErrorGoingAway, ""));
 321     // |this| has been deleted.
 322     return;
 323   }
 324   if (!WebSocketFrameHeader::IsKnownDataOpCode(op_code)) {
 325     LOG(DFATAL) << "Got SendFrame with bogus op_code " << op_code
 326                 << "; misbehaving renderer? fin=" << fin
 327                 << " data.size()=" << data.size();
 328     return;
 329   }
 330   if (op_code == WebSocketFrameHeader::kOpCodeText ||
 331       (op_code == WebSocketFrameHeader::kOpCodeContinuation &&
 332        sending_text_message_)) {
 333     StreamingUtf8Validator::State state =
 334         outgoing_utf8_validator_.AddBytes(vector_as_array(&data), data.size());
 335     if (state == StreamingUtf8Validator::INVALID ||
 336         (state == StreamingUtf8Validator::VALID_MIDPOINT && fin)) {
 337       // TODO(ricea): Kill renderer.
 338       AllowUnused(
 339           FailChannel("Browser sent a text frame containing invalid UTF-8",
 340                       kWebSocketErrorGoingAway,
 341                       ""));
 342       // |this| has been deleted.
 343       return;
 344     }
 345     sending_text_message_ = !fin;
 346     DCHECK(!fin || state == StreamingUtf8Validator::VALID_ENDPOINT);
 347   }
 348   current_send_quota_ -= data.size();
 349   // TODO(ricea): If current_send_quota_ has dropped below
 350   // send_quota_low_water_mark_, it might be good to increase the "low
 351   // water mark" and "high water mark", but only if the link to the WebSocket
 352   // server is not saturated.
 353   scoped_refptr<IOBuffer> buffer(new IOBuffer(data.size()));
 354   std::copy(data.begin(), data.end(), buffer->data());
 355   AllowUnused(SendIOBuffer(fin, op_code, buffer, data.size()));
 356   // |this| may have been deleted.
 357 }
 358
 359 void WebSocketChannel::SendFlowControl(int64 quota) {
 360   DCHECK(state_ == CONNECTING || state_ == CONNECTED || state_ == SEND_CLOSED ||
 361          state_ == CLOSE_WAIT);
 362   // TODO(ricea): Add interface to WebSocketStream and implement.
 363   // stream_->SendFlowControl(quota);
 364 }
 365
 366 void WebSocketChannel::StartClosingHandshake(uint16 code,
 367                                              const std::string& reason) {
 368   if (InClosingState()) {
 369     VLOG(1) << "StartClosingHandshake called in state " << state_
 370             << ". This may be a bug, or a harmless race.";
 371     return;
 372   }
 373   if (state_ == CONNECTING) {
 374     // Abort the in-progress handshake and drop the connection immediately.
 375     stream_request_.reset();
 376     state_ = CLOSED;
 377     AllowUnused(DoDropChannel(false, kWebSocketErrorAbnormalClosure, ""));
 378     return;
 379   }
 380   if (state_ != CONNECTED) {
 381     NOTREACHED() << "StartClosingHandshake() called in state " << state_;
 382     return;
 383   }
 384   // Javascript actually only permits 1000 and 3000-4999, but the implementation
 385   // itself may produce different codes. The length of |reason| is also checked
 386   // by Javascript.
 387   if (!IsStrictlyValidCloseStatusCode(code) ||
 388       reason.size() > kMaximumCloseReasonLength) {
 389     // "InternalServerError" is actually used for errors from any endpoint, per
 390     // errata 3227 to RFC6455. If the renderer is sending us an invalid code or
 391     // reason it must be malfunctioning in some way, and based on that we
 392     // interpret this as an internal error.
 393     AllowUnused(SendClose(kWebSocketErrorInternalServerError, ""));
 394     // |this| may have been deleted.
 395     return;
 396   }
 397   AllowUnused(SendClose(
 398       code, StreamingUtf8Validator::Validate(reason) ? reason : std::string()));
 399   // |this| may have been deleted.
 400 }
 401
 402 void WebSocketChannel::SendAddChannelRequestForTesting(
 403     const GURL& socket_url,
 404     const std::vector<std::string>& requested_subprotocols,
 405     const GURL& origin,
 406     const WebSocketStreamCreator& creator) {
 407   SendAddChannelRequestWithSuppliedCreator(
 408       socket_url, requested_subprotocols, origin, creator);
 409 }
 410
 411 void WebSocketChannel::SetClosingHandshakeTimeoutForTesting(
 412     base::TimeDelta delay) {
 413   timeout_ = delay;
 414 }
 415
 416 void WebSocketChannel::SendAddChannelRequestWithSuppliedCreator(
 417     const GURL& socket_url,
 418     const std::vector<std::string>& requested_subprotocols,
 419     const GURL& origin,
 420     const WebSocketStreamCreator& creator) {
 421   DCHECK_EQ(FRESHLY_CONSTRUCTED, state_);
 422   if (!socket_url.SchemeIsWSOrWSS()) {
 423     // TODO(ricea): Kill the renderer (this error should have been caught by
 424     // Javascript).
 425     AllowUnused(event_interface_->OnAddChannelResponse(true, "", ""));
 426     // |this| is deleted here.
 427     return;
 428   }
 429   socket_url_ = socket_url;
 430   scoped_ptr<WebSocketStream::ConnectDelegate> connect_delegate(
 431       new ConnectDelegate(this));
 432   stream_request_ = creator.Run(socket_url_,
 433                                 requested_subprotocols,
 434                                 origin,
 435                                 url_request_context_,
 436                                 BoundNetLog(),
 437                                 connect_delegate.Pass());
 438   state_ = CONNECTING;
 439 }
 440
 441 void WebSocketChannel::OnConnectSuccess(scoped_ptr<WebSocketStream> stream) {
 442   DCHECK(stream);
 443   DCHECK_EQ(CONNECTING, state_);
 444   stream_ = stream.Pass();
 445   state_ = CONNECTED;
 446   if (event_interface_->OnAddChannelResponse(
 447           false, stream_->GetSubProtocol(), stream_->GetExtensions()) ==
 448       CHANNEL_DELETED)
 449     return;
 450
 451   // TODO(ricea): Get flow control information from the WebSocketStream once we
 452   // have a multiplexing WebSocketStream.
 453   current_send_quota_ = send_quota_high_water_mark_;
 454   if (event_interface_->OnFlowControl(send_quota_high_water_mark_) ==
 455       CHANNEL_DELETED)
 456     return;
 457
 458   // |stream_request_| is not used once the connection has succeeded.
 459   stream_request_.reset();
 460   AllowUnused(ReadFrames());
 461   // |this| may have been deleted.
 462 }
 463
 464 void WebSocketChannel::OnConnectFailure(const std::string& message) {
 465   DCHECK_EQ(CONNECTING, state_);
 466   state_ = CLOSED;
 467   stream_request_.reset();
 468
 469   if (CHANNEL_DELETED ==
 470       notification_sender_->SendImmediately(event_interface_.get())) {
 471     // |this| has been deleted.
 472     return;
 473   }
 474   AllowUnused(event_interface_->OnFailChannel(message));
 475   // |this| has been deleted.
 476 }
 477
 478 void WebSocketChannel::OnStartOpeningHandshake(
 479     scoped_ptr<WebSocketHandshakeRequestInfo> request) {
 480   DCHECK(!notification_sender_->handshake_request_info());
 481
 482   // Because it is hard to handle an IPC error synchronously is difficult,
 483   // we asynchronously notify the information.
 484   notification_sender_->set_handshake_request_info(request.Pass());
 485   ScheduleOpeningHandshakeNotification();
 486 }
 487
 488 void WebSocketChannel::OnFinishOpeningHandshake(
 489     scoped_ptr<WebSocketHandshakeResponseInfo> response) {
 490   DCHECK(!notification_sender_->handshake_response_info());
 491
 492   // Because it is hard to handle an IPC error synchronously is difficult,
 493   // we asynchronously notify the information.
 494   notification_sender_->set_handshake_response_info(response.Pass());
 495   ScheduleOpeningHandshakeNotification();
 496 }
 497
 498 void WebSocketChannel::ScheduleOpeningHandshakeNotification() {
 499   base::MessageLoop::current()->PostTask(
 500       FROM_HERE,
 501       base::Bind(HandshakeNotificationSender::Send,
 502                  notification_sender_->AsWeakPtr()));
 503 }
 504
 505 ChannelState WebSocketChannel::WriteFrames() {
 506   int result = OK;
 507   do {
 508     // This use of base::Unretained is safe because this object owns the
 509     // WebSocketStream and destroying it cancels all callbacks.
 510     result = stream_->WriteFrames(
 511         data_being_sent_->frames(),
 512         base::Bind(base::IgnoreResult(&WebSocketChannel::OnWriteDone),
 513                    base::Unretained(this),
 514                    false));
 515     if (result != ERR_IO_PENDING) {
 516       if (OnWriteDone(true, result) == CHANNEL_DELETED)
 517         return CHANNEL_DELETED;
 518     }
 519   } while (result == OK && data_being_sent_);
 520   return CHANNEL_ALIVE;
 521 }
 522
 523 ChannelState WebSocketChannel::OnWriteDone(bool synchronous, int result) {
 524   DCHECK_NE(FRESHLY_CONSTRUCTED, state_);
 525   DCHECK_NE(CONNECTING, state_);
 526   DCHECK_NE(ERR_IO_PENDING, result);
 527   DCHECK(data_being_sent_);
 528   switch (result) {
 529     case OK:
 530       if (data_to_send_next_) {
 531         data_being_sent_ = data_to_send_next_.Pass();
 532         if (!synchronous)
 533           return WriteFrames();
 534       } else {
 535         data_being_sent_.reset();
 536         if (current_send_quota_ < send_quota_low_water_mark_) {
 537           // TODO(ricea): Increase low_water_mark and high_water_mark if
 538           // throughput is high, reduce them if throughput is low.  Low water
 539           // mark needs to be >= the bandwidth delay product *of the IPC
 540           // channel*. Because factors like context-switch time, thread wake-up
 541           // time, and bus speed come into play it is complex and probably needs
 542           // to be determined empirically.
 543           DCHECK_LE(send_quota_low_water_mark_, send_quota_high_water_mark_);
 544           // TODO(ricea): Truncate quota by the quota specified by the remote
 545           // server, if the protocol in use supports quota.
 546           int fresh_quota = send_quota_high_water_mark_ - current_send_quota_;
 547           current_send_quota_ += fresh_quota;
 548           return event_interface_->OnFlowControl(fresh_quota);
 549         }
 550       }
 551       return CHANNEL_ALIVE;
 552
 553     // If a recoverable error condition existed, it would go here.
 554
 555     default:
 556       DCHECK_LT(result, 0)
 557           << "WriteFrames() should only return OK or ERR_ codes";
 558       stream_->Close();
 559       DCHECK_NE(CLOSED, state_);
 560       state_ = CLOSED;
 561       return DoDropChannel(false, kWebSocketErrorAbnormalClosure, "");
 562   }
 563 }
 564
 565 ChannelState WebSocketChannel::ReadFrames() {
 566   int result = OK;
 567   do {
 568     // This use of base::Unretained is safe because this object owns the
 569     // WebSocketStream, and any pending reads will be cancelled when it is
 570     // destroyed.
 571     result = stream_->ReadFrames(
 572         &read_frames_,
 573         base::Bind(base::IgnoreResult(&WebSocketChannel::OnReadDone),
 574                    base::Unretained(this),
 575                    false));
 576     if (result != ERR_IO_PENDING) {
 577       if (OnReadDone(true, result) == CHANNEL_DELETED)
 578         return CHANNEL_DELETED;
 579     }
 580     DCHECK_NE(CLOSED, state_);
 581   } while (result == OK);
 582   return CHANNEL_ALIVE;
 583 }
 584
 585 ChannelState WebSocketChannel::OnReadDone(bool synchronous, int result) {
 586   DCHECK_NE(FRESHLY_CONSTRUCTED, state_);
 587   DCHECK_NE(CONNECTING, state_);
 588   DCHECK_NE(ERR_IO_PENDING, result);
 589   switch (result) {
 590     case OK:
 591       // ReadFrames() must use ERR_CONNECTION_CLOSED for a closed connection
 592       // with no data read, not an empty response.
 593       DCHECK(!read_frames_.empty())
 594           << "ReadFrames() returned OK, but nothing was read.";
 595       for (size_t i = 0; i < read_frames_.size(); ++i) {
 596         scoped_ptr<WebSocketFrame> frame(read_frames_[i]);
 597         read_frames_[i] = NULL;
 598         if (HandleFrame(frame.Pass()) == CHANNEL_DELETED)
 599           return CHANNEL_DELETED;
 600       }
 601       read_frames_.clear();
 602       // There should always be a call to ReadFrames pending.
 603       // TODO(ricea): Unless we are out of quota.
 604       DCHECK_NE(CLOSED, state_);
 605       if (!synchronous)
 606         return ReadFrames();
 607       return CHANNEL_ALIVE;
 608
 609     case ERR_WS_PROTOCOL_ERROR:
 610       // This could be kWebSocketErrorProtocolError (specifically, non-minimal
 611       // encoding of payload length) or kWebSocketErrorMessageTooBig, or an
 612       // extension-specific error.
 613       return FailChannel("Invalid frame header",
 614                          kWebSocketErrorProtocolError,
 615                          "WebSocket Protocol Error");
 616
 617     default:
 618       DCHECK_LT(result, 0)
 619           << "ReadFrames() should only return OK or ERR_ codes";
 620       stream_->Close();
 621       DCHECK_NE(CLOSED, state_);
 622       state_ = CLOSED;
 623       uint16 code = kWebSocketErrorAbnormalClosure;
 624       std::string reason = "";
 625       bool was_clean = false;
 626       if (received_close_code_ != 0) {
 627         code = received_close_code_;
 628         reason = received_close_reason_;
 629         was_clean = (result == ERR_CONNECTION_CLOSED);
 630       }
 631       return DoDropChannel(was_clean, code, reason);
 632   }
 633 }
 634
 635 ChannelState WebSocketChannel::HandleFrame(
 636     scoped_ptr<WebSocketFrame> frame) {
 637   if (frame->header.masked) {
 638     // RFC6455 Section 5.1 "A client MUST close a connection if it detects a
 639     // masked frame."
 640     return FailChannel(
 641         "A server must not mask any frames that it sends to the "
 642         "client.",
 643         kWebSocketErrorProtocolError,
 644         "Masked frame from server");
 645   }
 646   const WebSocketFrameHeader::OpCode opcode = frame->header.opcode;
 647   if (WebSocketFrameHeader::IsKnownControlOpCode(opcode) &&
 648       !frame->header.final) {
 649     return FailChannel(
 650         base::StringPrintf("Received fragmented control frame: opcode = %d",
 651                            opcode),
 652         kWebSocketErrorProtocolError,
 653         "Control message with FIN bit unset received");
 654   }
 655
 656   // Respond to the frame appropriately to its type.
 657   return HandleFrameByState(
 658       opcode, frame->header.final, frame->data, frame->header.payload_length);
 659 }
 660
 661 ChannelState WebSocketChannel::HandleFrameByState(
 662     const WebSocketFrameHeader::OpCode opcode,
 663     bool final,
 664     const scoped_refptr<IOBuffer>& data_buffer,
 665     size_t size) {
 666   DCHECK_NE(RECV_CLOSED, state_)
 667       << "HandleFrame() does not support being called re-entrantly from within "
 668          "SendClose()";
 669   DCHECK_NE(CLOSED, state_);
 670   if (state_ == CLOSE_WAIT) {
 671     std::string frame_name;
 672     GetFrameTypeForOpcode(opcode, &frame_name);
 673
 674     // FailChannel() won't send another Close frame.
 675     return FailChannel(
 676         frame_name + " received after close", kWebSocketErrorProtocolError, "");
 677   }
 678   switch (opcode) {
 679     case WebSocketFrameHeader::kOpCodeText:    // fall-thru
 680     case WebSocketFrameHeader::kOpCodeBinary:  // fall-thru
 681     case WebSocketFrameHeader::kOpCodeContinuation:
 682       if (state_ == CONNECTED) {
 683         if (opcode == WebSocketFrameHeader::kOpCodeText ||
 684             (opcode == WebSocketFrameHeader::kOpCodeContinuation &&
 685              receiving_text_message_)) {
 686           // This call is not redundant when size == 0 because it tells us what
 687           // the current state is.
 688           StreamingUtf8Validator::State state =
 689               incoming_utf8_validator_.AddBytes(
 690                   size ? data_buffer->data() : NULL, size);
 691           if (state == StreamingUtf8Validator::INVALID ||
 692               (state == StreamingUtf8Validator::VALID_MIDPOINT && final)) {
 693             return FailChannel("Could not decode a text frame as UTF-8.",
 694                                kWebSocketErrorProtocolError,
 695                                "Invalid UTF-8 in text frame");
 696           }
 697           receiving_text_message_ = !final;
 698           DCHECK(!final || state == StreamingUtf8Validator::VALID_ENDPOINT);
 699         }
 700         // TODO(ricea): Can this copy be eliminated?
 701         const char* const data_begin = size ? data_buffer->data() : NULL;
 702         const char* const data_end = data_begin + size;
 703         const std::vector<char> data(data_begin, data_end);
 704         // TODO(ricea): Handle the case when ReadFrames returns far
 705         // more data at once than should be sent in a single IPC. This needs to
 706         // be handled carefully, as an overloaded IO thread is one possible
 707         // cause of receiving very large chunks.
 708
 709         // Sends the received frame to the renderer process.
 710         return event_interface_->OnDataFrame(final, opcode, data);
 711       }
 712       VLOG(3) << "Ignored data packet received in state " << state_;
 713       return CHANNEL_ALIVE;
 714
 715     case WebSocketFrameHeader::kOpCodePing:
 716       VLOG(1) << "Got Ping of size " << size;
 717       if (state_ == CONNECTED)
 718         return SendIOBuffer(
 719             true, WebSocketFrameHeader::kOpCodePong, data_buffer, size);
 720       VLOG(3) << "Ignored ping in state " << state_;
 721       return CHANNEL_ALIVE;
 722
 723     case WebSocketFrameHeader::kOpCodePong:
 724       VLOG(1) << "Got Pong of size " << size;
 725       // There is no need to do anything with pong messages.
 726       return CHANNEL_ALIVE;
 727
 728     case WebSocketFrameHeader::kOpCodeClose: {
 729       uint16 code = kWebSocketNormalClosure;
 730       std::string reason;
 731       std::string message;
 732       if (!ParseClose(data_buffer, size, &code, &reason, &message)) {
 733         return FailChannel(message, code, reason);
 734       }
 735       // TODO(ricea): Find a way to safely log the message from the close
 736       // message (escape control codes and so on).
 737       VLOG(1) << "Got Close with code " << code;
 738       switch (state_) {
 739         case CONNECTED:
 740           state_ = RECV_CLOSED;
 741           if (SendClose(code, reason) ==  // Sets state_ to CLOSE_WAIT
 742               CHANNEL_DELETED)
 743             return CHANNEL_DELETED;
 744           if (event_interface_->OnClosingHandshake() == CHANNEL_DELETED)
 745             return CHANNEL_DELETED;
 746           received_close_code_ = code;
 747           received_close_reason_ = reason;
 748           break;
 749
 750         case SEND_CLOSED:
 751           state_ = CLOSE_WAIT;
 752           // From RFC6455 section 7.1.5: "Each endpoint
 753           // will see the status code sent by the other end as _The WebSocket
 754           // Connection Close Code_."
 755           received_close_code_ = code;
 756           received_close_reason_ = reason;
 757           break;
 758
 759         default:
 760           LOG(DFATAL) << "Got Close in unexpected state " << state_;
 761           break;
 762       }
 763       return CHANNEL_ALIVE;
 764     }
 765
 766     default:
 767       return FailChannel(
 768           base::StringPrintf("Unrecognized frame opcode: %d", opcode),
 769           kWebSocketErrorProtocolError,
 770           "Unknown opcode");
 771   }
 772 }
 773
 774 ChannelState WebSocketChannel::SendIOBuffer(
 775     bool fin,
 776     WebSocketFrameHeader::OpCode op_code,
 777     const scoped_refptr<IOBuffer>& buffer,
 778     size_t size) {
 779   DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
 780   DCHECK(stream_);
 781   scoped_ptr<WebSocketFrame> frame(new WebSocketFrame(op_code));
 782   WebSocketFrameHeader& header = frame->header;
 783   header.final = fin;
 784   header.masked = true;
 785   header.payload_length = size;
 786   frame->data = buffer;
 787   if (data_being_sent_) {
 788     // Either the link to the WebSocket server is saturated, or several messages
 789     // are being sent in a batch.
 790     // TODO(ricea): Keep some statistics to work out the situation and adjust
 791     // quota appropriately.
 792     if (!data_to_send_next_)
 793       data_to_send_next_.reset(new SendBuffer);
 794     data_to_send_next_->AddFrame(frame.Pass());
 795     return CHANNEL_ALIVE;
 796   }
 797   data_being_sent_.reset(new SendBuffer);
 798   data_being_sent_->AddFrame(frame.Pass());
 799   return WriteFrames();
 800 }
 801
 802 ChannelState WebSocketChannel::FailChannel(const std::string& message,
 803                                            uint16 code,
 804                                            const std::string& reason) {
 805   DCHECK_NE(FRESHLY_CONSTRUCTED, state_);
 806   DCHECK_NE(CONNECTING, state_);
 807   DCHECK_NE(CLOSED, state_);
 808   // TODO(ricea): Logging.
 809   if (state_ == CONNECTED) {
 810     if (SendClose(code, reason) ==  // Sets state_ to SEND_CLOSED
 811         CHANNEL_DELETED)
 812       return CHANNEL_DELETED;
 813   }
 814   // Careful study of RFC6455 section 7.1.7 and 7.1.1 indicates the browser
 815   // should close the connection itself without waiting for the closing
 816   // handshake.
 817   stream_->Close();
 818   state_ = CLOSED;
 819
 820   return event_interface_->OnFailChannel(message);
 821 }
 822
 823 ChannelState WebSocketChannel::SendClose(uint16 code,
 824                                          const std::string& reason) {
 825   DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
 826   DCHECK_LE(reason.size(), kMaximumCloseReasonLength);
 827   scoped_refptr<IOBuffer> body;
 828   size_t size = 0;
 829   if (code == kWebSocketErrorNoStatusReceived) {
 830     // Special case: translate kWebSocketErrorNoStatusReceived into a Close
 831     // frame with no payload.
 832     body = new IOBuffer(0);
 833   } else {
 834     const size_t payload_length = kWebSocketCloseCodeLength + reason.length();
 835     body = new IOBuffer(payload_length);
 836     size = payload_length;
 837     WriteBigEndian(body->data(), code);
 838     COMPILE_ASSERT(sizeof(code) == kWebSocketCloseCodeLength,
 839                    they_should_both_be_two);
 840     std::copy(
 841         reason.begin(), reason.end(), body->data() + kWebSocketCloseCodeLength);
 842   }
 843   // This use of base::Unretained() is safe because we stop the timer in the
 844   // destructor.
 845   timer_.Start(
 846       FROM_HERE,
 847       timeout_,
 848       base::Bind(&WebSocketChannel::CloseTimeout, base::Unretained(this)));
 849   if (SendIOBuffer(true, WebSocketFrameHeader::kOpCodeClose, body, size) ==
 850       CHANNEL_DELETED)
 851     return CHANNEL_DELETED;
 852   // SendIOBuffer() checks |state_|, so it is best not to change it until after
 853   // SendIOBuffer() returns.
 854   state_ = (state_ == CONNECTED) ? SEND_CLOSED : CLOSE_WAIT;
 855   return CHANNEL_ALIVE;
 856 }
 857
 858 bool WebSocketChannel::ParseClose(const scoped_refptr<IOBuffer>& buffer,
 859                                   size_t size,
 860                                   uint16* code,
 861                                   std::string* reason,
 862                                   std::string* message) {
 863   bool parsed_ok = true;
 864   reason->clear();
 865   if (size < kWebSocketCloseCodeLength) {
 866     *code = kWebSocketErrorNoStatusReceived;
 867     if (size != 0) {
 868       DVLOG(1) << "Close frame with payload size " << size << " received "
 869                << "(the first byte is " << std::hex
 870                << static_cast<int>(buffer->data()[0]) << ")";
 871       parsed_ok = false;
 872       *code = kWebSocketErrorProtocolError;
 873       *message =
 874           "Received a broken close frame containing an invalid size body.";
 875     }
 876     return parsed_ok;
 877   }
 878   const char* data = buffer->data();
 879   uint16 unchecked_code = 0;
 880   ReadBigEndian(data, &unchecked_code);
 881   COMPILE_ASSERT(sizeof(unchecked_code) == kWebSocketCloseCodeLength,
 882                  they_should_both_be_two_bytes);
 883   switch (unchecked_code) {
 884     case kWebSocketErrorNoStatusReceived:
 885     case kWebSocketErrorAbnormalClosure:
 886     case kWebSocketErrorTlsHandshake:
 887       *code = kWebSocketErrorProtocolError;
 888       *message =
 889           "Received a broken close frame containing a reserved status code.";
 890       parsed_ok = false;
 891       break;
 892
 893     default:
 894       *code = unchecked_code;
 895       break;
 896   }
 897   if (parsed_ok) {
 898     std::string text(data + kWebSocketCloseCodeLength, data + size);
 899     if (StreamingUtf8Validator::Validate(text)) {
 900       reason->swap(text);
 901     } else {
 902       *code = kWebSocketErrorProtocolError;
 903       *reason = "Invalid UTF-8 in Close frame";
 904       *message = "Received a broken close frame containing invalid UTF-8.";
 905       parsed_ok = false;
 906     }
 907   }
 908   return parsed_ok;
 909 }
 910
 911 ChannelState WebSocketChannel::DoDropChannel(bool was_clean,
 912                                              uint16 code,
 913                                              const std::string& reason) {
 914   if (CHANNEL_DELETED ==
 915       notification_sender_->SendImmediately(event_interface_.get()))
 916     return CHANNEL_DELETED;
 917   return event_interface_->OnDropChannel(was_clean, code, reason);
 918 }
 919
 920 void WebSocketChannel::CloseTimeout() {
 921   stream_->Close();
 922   DCHECK_NE(CLOSED, state_);
 923   state_ = CLOSED;
 924   AllowUnused(DoDropChannel(false, kWebSocketErrorAbnormalClosure, ""));
 925   // |this| has been deleted.
 926 }
 927
 928 }  // namespace net