chrome/browser/extensions/api/cast_channel/cast_socket.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 "chrome/browser/extensions/api/cast_channel/cast_socket.h"
   6
   7 #include <stdlib.h>
   8 #include <string.h>
   9
  10 #include "base/bind.h"
  11 #include "base/callback_helpers.h"
  12 #include "base/lazy_instance.h"
  13 #include "base/numerics/safe_conversions.h"
  14 #include "base/strings/string_number_conversions.h"
  15 #include "base/sys_byteorder.h"
  16 #include "chrome/browser/extensions/api/cast_channel/cast_auth_util.h"
  17 #include "chrome/browser/extensions/api/cast_channel/cast_channel.pb.h"
  18 #include "chrome/browser/extensions/api/cast_channel/cast_message_util.h"
  19 #include "net/base/address_list.h"
  20 #include "net/base/host_port_pair.h"
  21 #include "net/base/net_errors.h"
  22 #include "net/base/net_util.h"
  23 #include "net/cert/cert_verifier.h"
  24 #include "net/cert/x509_certificate.h"
  25 #include "net/http/transport_security_state.h"
  26 #include "net/socket/client_socket_factory.h"
  27 #include "net/socket/client_socket_handle.h"
  28 #include "net/socket/ssl_client_socket.h"
  29 #include "net/socket/stream_socket.h"
  30 #include "net/socket/tcp_client_socket.h"
  31 #include "net/ssl/ssl_config_service.h"
  32 #include "net/ssl/ssl_info.h"
  33
  34 // Assumes |ip_endpoint_| of type net::IPEndPoint and |channel_auth_| of enum
  35 // type ChannelAuthType are available in the current scope.
  36 #define VLOG_WITH_CONNECTION(level) VLOG(level) << "[" << \
  37     ip_endpoint_.ToString() << ", auth=" << channel_auth_ << "] "
  38
  39 namespace {
  40
  41 // The default keepalive delay.  On Linux, keepalives probes will be sent after
  42 // the socket is idle for this length of time, and the socket will be closed
  43 // after 9 failed probes.  So the total idle time before close is 10 *
  44 // kTcpKeepAliveDelaySecs.
  45 const int kTcpKeepAliveDelaySecs = 10;
  46
  47 }  // namespace
  48
  49 namespace extensions {
  50
  51 static base::LazyInstance<BrowserContextKeyedAPIFactory<
  52     ApiResourceManager<api::cast_channel::CastSocket> > > g_factory =
  53     LAZY_INSTANCE_INITIALIZER;
  54
  55 // static
  56 template <>
  57 BrowserContextKeyedAPIFactory<
  58     ApiResourceManager<api::cast_channel::CastSocket> >*
  59 ApiResourceManager<api::cast_channel::CastSocket>::GetFactoryInstance() {
  60   return g_factory.Pointer();
  61 }
  62
  63 namespace api {
  64 namespace cast_channel {
  65
  66 CastSocket::CastSocket(const std::string& owner_extension_id,
  67                        const net::IPEndPoint& ip_endpoint,
  68                        ChannelAuthType channel_auth,
  69                        CastSocket::Delegate* delegate,
  70                        net::NetLog* net_log,
  71                        const base::TimeDelta& timeout) :
  72     ApiResource(owner_extension_id),
  73     channel_id_(0),
  74     ip_endpoint_(ip_endpoint),
  75     channel_auth_(channel_auth),
  76     delegate_(delegate),
  77     current_message_size_(0),
  78     current_message_(new CastMessage()),
  79     net_log_(net_log),
  80     connect_timeout_(timeout),
  81     connect_timeout_timer_(new base::OneShotTimer<CastSocket>),
  82     is_canceled_(false),
  83     connect_state_(CONN_STATE_NONE),
  84     write_state_(WRITE_STATE_NONE),
  85     read_state_(READ_STATE_NONE),
  86     error_state_(CHANNEL_ERROR_NONE),
  87     ready_state_(READY_STATE_NONE) {
  88   DCHECK(net_log_);
  89   DCHECK(channel_auth_ == CHANNEL_AUTH_TYPE_SSL ||
  90          channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED);
  91   net_log_source_.type = net::NetLog::SOURCE_SOCKET;
  92   net_log_source_.id = net_log_->NextID();
  93
  94   // Reuse these buffers for each message.
  95   header_read_buffer_ = new net::GrowableIOBuffer();
  96   header_read_buffer_->SetCapacity(MessageHeader::header_size());
  97   body_read_buffer_ = new net::GrowableIOBuffer();
  98   body_read_buffer_->SetCapacity(MessageHeader::max_message_size());
  99   current_read_buffer_ = header_read_buffer_;
 100 }
 101
 102 CastSocket::~CastSocket() { }
 103
 104 ReadyState CastSocket::ready_state() const {
 105   return ready_state_;
 106 }
 107
 108 ChannelError CastSocket::error_state() const {
 109   return error_state_;
 110 }
 111
 112 scoped_ptr<net::TCPClientSocket> CastSocket::CreateTcpSocket() {
 113   net::AddressList addresses(ip_endpoint_);
 114   return scoped_ptr<net::TCPClientSocket>(
 115       new net::TCPClientSocket(addresses, net_log_, net_log_source_));
 116   // Options cannot be set on the TCPClientSocket yet, because the
 117   // underlying platform socket will not be created until Bind()
 118   // or Connect() is called.
 119 }
 120
 121 scoped_ptr<net::SSLClientSocket> CastSocket::CreateSslSocket(
 122     scoped_ptr<net::StreamSocket> socket) {
 123   net::SSLConfig ssl_config;
 124   // If a peer cert was extracted in a previous attempt to connect, then
 125   // whitelist that cert.
 126   if (!peer_cert_.empty()) {
 127     net::SSLConfig::CertAndStatus cert_and_status;
 128     cert_and_status.cert_status = net::CERT_STATUS_AUTHORITY_INVALID;
 129     cert_and_status.der_cert = peer_cert_;
 130     ssl_config.allowed_bad_certs.push_back(cert_and_status);
 131   }
 132
 133   cert_verifier_.reset(net::CertVerifier::CreateDefault());
 134   transport_security_state_.reset(new net::TransportSecurityState);
 135   net::SSLClientSocketContext context;
 136   // CertVerifier and TransportSecurityState are owned by us, not the
 137   // context object.
 138   context.cert_verifier = cert_verifier_.get();
 139   context.transport_security_state = transport_security_state_.get();
 140
 141   scoped_ptr<net::ClientSocketHandle> connection(new net::ClientSocketHandle);
 142   connection->SetSocket(socket.Pass());
 143   net::HostPortPair host_and_port = net::HostPortPair::FromIPEndPoint(
 144       ip_endpoint_);
 145
 146   return net::ClientSocketFactory::GetDefaultFactory()->CreateSSLClientSocket(
 147       connection.Pass(), host_and_port, ssl_config, context);
 148 }
 149
 150 bool CastSocket::ExtractPeerCert(std::string* cert) {
 151   DCHECK(cert);
 152   DCHECK(peer_cert_.empty());
 153   net::SSLInfo ssl_info;
 154   if (!socket_->GetSSLInfo(&ssl_info) || !ssl_info.cert.get())
 155     return false;
 156   bool result = net::X509Certificate::GetDEREncoded(
 157      ssl_info.cert->os_cert_handle(), cert);
 158   if (result)
 159     VLOG_WITH_CONNECTION(1) << "Successfully extracted peer certificate: "
 160                             << *cert;
 161   return result;
 162 }
 163
 164 bool CastSocket::VerifyChallengeReply() {
 165   return AuthenticateChallengeReply(*challenge_reply_.get(), peer_cert_);
 166 }
 167
 168 void CastSocket::Connect(const net::CompletionCallback& callback) {
 169   DCHECK(CalledOnValidThread());
 170   VLOG_WITH_CONNECTION(1) << "Connect readyState = " << ready_state_;
 171   if (ready_state_ != READY_STATE_NONE) {
 172     callback.Run(net::ERR_CONNECTION_FAILED);
 173     return;
 174   }
 175   ready_state_ = READY_STATE_CONNECTING;
 176   connect_callback_ = callback;
 177   connect_state_ = CONN_STATE_TCP_CONNECT;
 178   if (connect_timeout_.InMicroseconds() > 0) {
 179     GetTimer()->Start(
 180         FROM_HERE,
 181         connect_timeout_,
 182         base::Bind(&CastSocket::CancelConnect, AsWeakPtr()));
 183   }
 184   DoConnectLoop(net::OK);
 185 }
 186
 187 void CastSocket::PostTaskToStartConnectLoop(int result) {
 188   DCHECK(CalledOnValidThread());
 189   base::MessageLoop::current()->PostTask(
 190       FROM_HERE,
 191       base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr(), result));
 192 }
 193
 194 void CastSocket::CancelConnect() {
 195   DCHECK(CalledOnValidThread());
 196   // Stop all pending connection setup tasks and report back to the client.
 197   is_canceled_ = true;
 198   VLOG_WITH_CONNECTION(1) << "Timeout while establishing a connection.";
 199   DoConnectCallback(net::ERR_TIMED_OUT);
 200 }
 201
 202 // This method performs the state machine transitions for connection flow.
 203 // There are two entry points to this method:
 204 // 1. Connect method: this starts the flow
 205 // 2. Callback from network operations that finish asynchronously
 206 void CastSocket::DoConnectLoop(int result) {
 207   if (is_canceled_) {
 208     LOG(ERROR) << "CANCELLED - Aborting DoConnectLoop.";
 209     return;
 210   }
 211   // Network operations can either finish synchronously or asynchronously.
 212   // This method executes the state machine transitions in a loop so that
 213   // correct state transitions happen even when network operations finish
 214   // synchronously.
 215   int rv = result;
 216   do {
 217     ConnectionState state = connect_state_;
 218     // Default to CONN_STATE_NONE, which breaks the processing loop if any
 219     // handler fails to transition to another state to continue processing.
 220     connect_state_ = CONN_STATE_NONE;
 221     switch (state) {
 222       case CONN_STATE_TCP_CONNECT:
 223         rv = DoTcpConnect();
 224         break;
 225       case CONN_STATE_TCP_CONNECT_COMPLETE:
 226         rv = DoTcpConnectComplete(rv);
 227         break;
 228       case CONN_STATE_SSL_CONNECT:
 229         DCHECK_EQ(net::OK, rv);
 230         rv = DoSslConnect();
 231         break;
 232       case CONN_STATE_SSL_CONNECT_COMPLETE:
 233         rv = DoSslConnectComplete(rv);
 234         break;
 235       case CONN_STATE_AUTH_CHALLENGE_SEND:
 236         rv = DoAuthChallengeSend();
 237         break;
 238       case CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE:
 239         rv = DoAuthChallengeSendComplete(rv);
 240         break;
 241       case CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE:
 242         rv = DoAuthChallengeReplyComplete(rv);
 243         break;
 244       default:
 245         NOTREACHED() << "BUG in connect flow. Unknown state: " << state;
 246         break;
 247     }
 248   } while (rv != net::ERR_IO_PENDING && connect_state_ != CONN_STATE_NONE);
 249   // Get out of the loop either when:
 250   // a. A network operation is pending, OR
 251   // b. The Do* method called did not change state
 252
 253   // Connect loop is finished: if there is no pending IO invoke the callback.
 254   if (rv != net::ERR_IO_PENDING) {
 255     GetTimer()->Stop();
 256     DoConnectCallback(rv);
 257   }
 258 }
 259
 260 int CastSocket::DoTcpConnect() {
 261   VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
 262   connect_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
 263   tcp_socket_ = CreateTcpSocket();
 264   return tcp_socket_->Connect(
 265       base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
 266 }
 267
 268 int CastSocket::DoTcpConnectComplete(int result) {
 269   VLOG_WITH_CONNECTION(1) << "DoTcpConnectComplete: " << result;
 270   if (result == net::OK) {
 271     // Enable TCP protocol-level keep-alive.
 272     bool result = tcp_socket_->SetKeepAlive(true, kTcpKeepAliveDelaySecs);
 273     LOG_IF(WARNING, !result) << "Failed to SetKeepAlive.";
 274     connect_state_ = CONN_STATE_SSL_CONNECT;
 275   }
 276   return result;
 277 }
 278
 279 int CastSocket::DoSslConnect() {
 280   VLOG_WITH_CONNECTION(1) << "DoSslConnect";
 281   connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
 282   socket_ = CreateSslSocket(tcp_socket_.PassAs<net::StreamSocket>());
 283   return socket_->Connect(
 284       base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
 285 }
 286
 287 int CastSocket::DoSslConnectComplete(int result) {
 288   VLOG_WITH_CONNECTION(1) << "DoSslConnectComplete: " << result;
 289   if (result == net::ERR_CERT_AUTHORITY_INVALID &&
 290       peer_cert_.empty() && ExtractPeerCert(&peer_cert_)) {
 291     connect_state_ = CONN_STATE_TCP_CONNECT;
 292   } else if (result == net::OK &&
 293              channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) {
 294     connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
 295   }
 296   return result;
 297 }
 298
 299 int CastSocket::DoAuthChallengeSend() {
 300   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSend";
 301   connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
 302   CastMessage challenge_message;
 303   CreateAuthChallengeMessage(&challenge_message);
 304   VLOG_WITH_CONNECTION(1) << "Sending challenge: "
 305                           << CastMessageToString(challenge_message);
 306   // Post a task to send auth challenge so that DoWriteLoop is not nested inside
 307   // DoConnectLoop. This is not strictly necessary but keeps the write loop
 308   // code decoupled from connect loop code.
 309   base::MessageLoop::current()->PostTask(
 310       FROM_HERE,
 311       base::Bind(&CastSocket::SendCastMessageInternal,
 312                  AsWeakPtr(),
 313                  challenge_message,
 314                  base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr())));
 315   // Always return IO_PENDING since the result is always asynchronous.
 316   return net::ERR_IO_PENDING;
 317 }
 318
 319 int CastSocket::DoAuthChallengeSendComplete(int result) {
 320   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
 321   if (result < 0)
 322     return result;
 323   connect_state_ = CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE;
 324   // Post a task to start read loop so that DoReadLoop is not nested inside
 325   // DoConnectLoop. This is not strictly necessary but keeps the read loop
 326   // code decoupled from connect loop code.
 327   PostTaskToStartReadLoop();
 328   // Always return IO_PENDING since the result is always asynchronous.
 329   return net::ERR_IO_PENDING;
 330 }
 331
 332 int CastSocket::DoAuthChallengeReplyComplete(int result) {
 333   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeReplyComplete: " << result;
 334   if (result < 0)
 335     return result;
 336   if (!VerifyChallengeReply())
 337     return net::ERR_FAILED;
 338   VLOG_WITH_CONNECTION(1) << "Auth challenge verification succeeded";
 339   return net::OK;
 340 }
 341
 342 void CastSocket::DoConnectCallback(int result) {
 343   ready_state_ = (result == net::OK) ? READY_STATE_OPEN : READY_STATE_CLOSED;
 344   if (result == net::OK) {
 345     error_state_ = CHANNEL_ERROR_NONE;
 346     PostTaskToStartReadLoop();
 347   } else if (result == net::ERR_TIMED_OUT) {
 348     error_state_ = CHANNEL_ERROR_CONNECT_TIMEOUT;
 349   } else {
 350     error_state_ = CHANNEL_ERROR_CONNECT_ERROR;
 351   }
 352   VLOG_WITH_CONNECTION(1) << "Calling Connect_Callback";
 353   base::ResetAndReturn(&connect_callback_).Run(result);
 354 }
 355
 356 void CastSocket::Close(const net::CompletionCallback& callback) {
 357   DCHECK(CalledOnValidThread());
 358   VLOG_WITH_CONNECTION(1) << "Close ReadyState = " << ready_state_;
 359   tcp_socket_.reset();
 360   socket_.reset();
 361   cert_verifier_.reset();
 362   transport_security_state_.reset();
 363   ready_state_ = READY_STATE_CLOSED;
 364   callback.Run(net::OK);
 365   // |callback| can delete |this|
 366 }
 367
 368 void CastSocket::SendMessage(const MessageInfo& message,
 369                              const net::CompletionCallback& callback) {
 370   DCHECK(CalledOnValidThread());
 371   if (ready_state_ != READY_STATE_OPEN) {
 372     callback.Run(net::ERR_FAILED);
 373     return;
 374   }
 375   CastMessage message_proto;
 376   if (!MessageInfoToCastMessage(message, &message_proto)) {
 377     callback.Run(net::ERR_FAILED);
 378     return;
 379   }
 380
 381   SendCastMessageInternal(message_proto, callback);
 382 }
 383
 384 void CastSocket::SendCastMessageInternal(
 385     const CastMessage& message,
 386     const net::CompletionCallback& callback) {
 387   WriteRequest write_request(callback);
 388   if (!write_request.SetContent(message)) {
 389     callback.Run(net::ERR_FAILED);
 390     return;
 391   }
 392
 393   write_queue_.push(write_request);
 394   if (write_state_ == WRITE_STATE_NONE) {
 395     write_state_ = WRITE_STATE_WRITE;
 396     DoWriteLoop(net::OK);
 397   }
 398 }
 399
 400 void CastSocket::DoWriteLoop(int result) {
 401   DCHECK(CalledOnValidThread());
 402   VLOG_WITH_CONNECTION(1) << "DoWriteLoop queue size: " << write_queue_.size();
 403
 404   if (write_queue_.empty()) {
 405     write_state_ = WRITE_STATE_NONE;
 406     return;
 407   }
 408
 409   // Network operations can either finish synchronously or asynchronously.
 410   // This method executes the state machine transitions in a loop so that
 411   // write state transitions happen even when network operations finish
 412   // synchronously.
 413   int rv = result;
 414   do {
 415     WriteState state = write_state_;
 416     write_state_ = WRITE_STATE_NONE;
 417     switch (state) {
 418       case WRITE_STATE_WRITE:
 419         rv = DoWrite();
 420         break;
 421       case WRITE_STATE_WRITE_COMPLETE:
 422         rv = DoWriteComplete(rv);
 423         break;
 424       case WRITE_STATE_DO_CALLBACK:
 425         rv = DoWriteCallback();
 426         break;
 427       case WRITE_STATE_ERROR:
 428         rv = DoWriteError(rv);
 429         break;
 430       default:
 431         NOTREACHED() << "BUG in write flow. Unknown state: " << state;
 432         break;
 433     }
 434   } while (!write_queue_.empty() &&
 435            rv != net::ERR_IO_PENDING &&
 436            write_state_ != WRITE_STATE_NONE);
 437
 438   // If write loop is done because the queue is empty then set write
 439   // state to NONE
 440   if (write_queue_.empty())
 441     write_state_ = WRITE_STATE_NONE;
 442
 443   // Write loop is done - if the result is ERR_FAILED then close with error.
 444   if (rv == net::ERR_FAILED)
 445     CloseWithError(error_state_);
 446 }
 447
 448 int CastSocket::DoWrite() {
 449   DCHECK(!write_queue_.empty());
 450   WriteRequest& request = write_queue_.front();
 451
 452   VLOG_WITH_CONNECTION(2) << "WriteData byte_count = "
 453                           << request.io_buffer->size() << " bytes_written "
 454                           << request.io_buffer->BytesConsumed();
 455
 456   write_state_ = WRITE_STATE_WRITE_COMPLETE;
 457
 458   return socket_->Write(
 459       request.io_buffer.get(),
 460       request.io_buffer->BytesRemaining(),
 461       base::Bind(&CastSocket::DoWriteLoop, AsWeakPtr()));
 462 }
 463
 464 int CastSocket::DoWriteComplete(int result) {
 465   DCHECK(!write_queue_.empty());
 466   if (result <= 0) {  // NOTE that 0 also indicates an error
 467     error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
 468     write_state_ = WRITE_STATE_ERROR;
 469     return result == 0 ? net::ERR_FAILED : result;
 470   }
 471
 472   // Some bytes were successfully written
 473   WriteRequest& request = write_queue_.front();
 474   scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
 475   io_buffer->DidConsume(result);
 476   if (io_buffer->BytesRemaining() == 0)  // Message fully sent
 477     write_state_ = WRITE_STATE_DO_CALLBACK;
 478   else
 479     write_state_ = WRITE_STATE_WRITE;
 480
 481   return net::OK;
 482 }
 483
 484 int CastSocket::DoWriteCallback() {
 485   DCHECK(!write_queue_.empty());
 486   WriteRequest& request = write_queue_.front();
 487   int bytes_consumed = request.io_buffer->BytesConsumed();
 488
 489   // If inside connection flow, then there should be exaclty one item in
 490   // the write queue.
 491   if (ready_state_ == READY_STATE_CONNECTING) {
 492     write_queue_.pop();
 493     DCHECK(write_queue_.empty());
 494     PostTaskToStartConnectLoop(bytes_consumed);
 495   } else {
 496     WriteRequest& request = write_queue_.front();
 497     request.callback.Run(bytes_consumed);
 498     write_queue_.pop();
 499   }
 500   write_state_ = WRITE_STATE_WRITE;
 501   return net::OK;
 502 }
 503
 504 int CastSocket::DoWriteError(int result) {
 505   DCHECK(!write_queue_.empty());
 506   DCHECK_LT(result, 0);
 507
 508   // If inside connection flow, then there should be exactly one item in
 509   // the write queue.
 510   if (ready_state_ == READY_STATE_CONNECTING) {
 511     write_queue_.pop();
 512     DCHECK(write_queue_.empty());
 513     PostTaskToStartConnectLoop(result);
 514     // Connect loop will handle the error. Return net::OK so that write flow
 515     // does not try to report error also.
 516     return net::OK;
 517   }
 518
 519   while (!write_queue_.empty()) {
 520     WriteRequest& request = write_queue_.front();
 521     request.callback.Run(result);
 522     write_queue_.pop();
 523   }
 524   return net::ERR_FAILED;
 525 }
 526
 527 void CastSocket::PostTaskToStartReadLoop() {
 528   DCHECK(CalledOnValidThread());
 529   base::MessageLoop::current()->PostTask(
 530       FROM_HERE,
 531       base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
 532 }
 533
 534 void CastSocket::StartReadLoop() {
 535   // Read loop would have already been started if read state is not NONE
 536   if (read_state_ == READ_STATE_NONE) {
 537     read_state_ = READ_STATE_READ;
 538     DoReadLoop(net::OK);
 539   }
 540 }
 541
 542 void CastSocket::DoReadLoop(int result) {
 543   DCHECK(CalledOnValidThread());
 544   // Network operations can either finish synchronously or asynchronously.
 545   // This method executes the state machine transitions in a loop so that
 546   // write state transitions happen even when network operations finish
 547   // synchronously.
 548   int rv = result;
 549   do {
 550     ReadState state = read_state_;
 551     read_state_ = READ_STATE_NONE;
 552
 553     switch (state) {
 554       case READ_STATE_READ:
 555         rv = DoRead();
 556         break;
 557       case READ_STATE_READ_COMPLETE:
 558         rv = DoReadComplete(rv);
 559         break;
 560       case READ_STATE_DO_CALLBACK:
 561         rv = DoReadCallback();
 562         break;
 563       case READ_STATE_ERROR:
 564         rv = DoReadError(rv);
 565         break;
 566       default:
 567         NOTREACHED() << "BUG in read flow. Unknown state: " << state;
 568         break;
 569     }
 570   } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE);
 571
 572   // Read loop is done - If the result is ERR_FAILED then close with error.
 573   if (rv == net::ERR_FAILED)
 574     CloseWithError(error_state_);
 575 }
 576
 577 int CastSocket::DoRead() {
 578   read_state_ = READ_STATE_READ_COMPLETE;
 579   // Figure out whether to read header or body, and the remaining bytes.
 580   uint32 num_bytes_to_read = 0;
 581   if (header_read_buffer_->RemainingCapacity() > 0) {
 582     current_read_buffer_ = header_read_buffer_;
 583     num_bytes_to_read = header_read_buffer_->RemainingCapacity();
 584     DCHECK_LE(num_bytes_to_read, MessageHeader::header_size());
 585   } else {
 586     DCHECK_GT(current_message_size_, 0U);
 587     num_bytes_to_read = current_message_size_ - body_read_buffer_->offset();
 588     current_read_buffer_ = body_read_buffer_;
 589     DCHECK_LE(num_bytes_to_read, MessageHeader::max_message_size());
 590   }
 591   DCHECK_GT(num_bytes_to_read, 0U);
 592
 593   // Read up to num_bytes_to_read into |current_read_buffer_|.
 594   return socket_->Read(
 595       current_read_buffer_.get(),
 596       num_bytes_to_read,
 597       base::Bind(&CastSocket::DoReadLoop, AsWeakPtr()));
 598 }
 599
 600 int CastSocket::DoReadComplete(int result) {
 601   VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result
 602                           << " header offset = "
 603                           << header_read_buffer_->offset()
 604                           << " body offset = " << body_read_buffer_->offset();
 605   if (result <= 0) {  // 0 means EOF: the peer closed the socket
 606     VLOG_WITH_CONNECTION(1) << "Read error, peer closed the socket";
 607     error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
 608     read_state_ = READ_STATE_ERROR;
 609     return result == 0 ? net::ERR_FAILED : result;
 610   }
 611
 612   // Some data was read.  Move the offset in the current buffer forward.
 613   DCHECK_LE(current_read_buffer_->offset() + result,
 614             current_read_buffer_->capacity());
 615   current_read_buffer_->set_offset(current_read_buffer_->offset() + result);
 616   read_state_ = READ_STATE_READ;
 617
 618   if (current_read_buffer_.get() == header_read_buffer_.get() &&
 619       current_read_buffer_->RemainingCapacity() == 0) {
 620     // A full header is read, process the contents.
 621     if (!ProcessHeader()) {
 622       error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
 623       read_state_ = READ_STATE_ERROR;
 624     }
 625   } else if (current_read_buffer_.get() == body_read_buffer_.get() &&
 626              static_cast<uint32>(current_read_buffer_->offset()) ==
 627              current_message_size_) {
 628     // Full body is read, process the contents.
 629     if (ProcessBody()) {
 630       read_state_ = READ_STATE_DO_CALLBACK;
 631     } else {
 632       error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
 633       read_state_ = READ_STATE_ERROR;
 634     }
 635   }
 636
 637   return net::OK;
 638 }
 639
 640 int CastSocket::DoReadCallback() {
 641   read_state_ = READ_STATE_READ;
 642   const CastMessage& message = *(current_message_.get());
 643   if (IsAuthMessage(message)) {
 644     // An auth message is received, check that connect flow is running.
 645     if (ready_state_ == READY_STATE_CONNECTING) {
 646       challenge_reply_.reset(new CastMessage(message));
 647       PostTaskToStartConnectLoop(net::OK);
 648     } else {
 649       read_state_ = READ_STATE_ERROR;
 650     }
 651   } else if (delegate_) {
 652     MessageInfo message_info;
 653     if (CastMessageToMessageInfo(message, &message_info))
 654       delegate_->OnMessage(this, message_info);
 655     else
 656       read_state_ = READ_STATE_ERROR;
 657   }
 658   current_message_->Clear();
 659   return net::OK;
 660 }
 661
 662 int CastSocket::DoReadError(int result) {
 663   DCHECK_LE(result, 0);
 664   // If inside connection flow, then get back to connect loop.
 665   if (ready_state_ == READY_STATE_CONNECTING) {
 666     PostTaskToStartConnectLoop(result);
 667     // does not try to report error also.
 668     return net::OK;
 669   }
 670   return net::ERR_FAILED;
 671 }
 672
 673 bool CastSocket::ProcessHeader() {
 674   DCHECK_EQ(static_cast<uint32>(header_read_buffer_->offset()),
 675             MessageHeader::header_size());
 676   MessageHeader header;
 677   MessageHeader::ReadFromIOBuffer(header_read_buffer_.get(), &header);
 678   if (header.message_size > MessageHeader::max_message_size())
 679     return false;
 680
 681   VLOG_WITH_CONNECTION(2) << "Parsed header { message_size: "
 682                           << header.message_size << " }";
 683   current_message_size_ = header.message_size;
 684   return true;
 685 }
 686
 687 bool CastSocket::ProcessBody() {
 688   DCHECK_EQ(static_cast<uint32>(body_read_buffer_->offset()),
 689             current_message_size_);
 690   if (!current_message_->ParseFromArray(
 691       body_read_buffer_->StartOfBuffer(), current_message_size_)) {
 692     return false;
 693   }
 694   current_message_size_ = 0;
 695   header_read_buffer_->set_offset(0);
 696   body_read_buffer_->set_offset(0);
 697   current_read_buffer_ = header_read_buffer_;
 698   return true;
 699 }
 700
 701 // static
 702 bool CastSocket::Serialize(const CastMessage& message_proto,
 703                            std::string* message_data) {
 704   DCHECK(message_data);
 705   message_proto.SerializeToString(message_data);
 706   size_t message_size = message_data->size();
 707   if (message_size > MessageHeader::max_message_size()) {
 708     message_data->clear();
 709     return false;
 710   }
 711   CastSocket::MessageHeader header;
 712   header.SetMessageSize(message_size);
 713   header.PrependToString(message_data);
 714   return true;
 715 }
 716
 717 void CastSocket::CloseWithError(ChannelError error) {
 718   DCHECK(CalledOnValidThread());
 719   socket_.reset(NULL);
 720   ready_state_ = READY_STATE_CLOSED;
 721   error_state_ = error;
 722   if (delegate_)
 723     delegate_->OnError(this, error);
 724 }
 725
 726 std::string CastSocket::CastUrl() const {
 727   return ((channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) ?
 728           "casts://" : "cast://") + ip_endpoint_.ToString();
 729 }
 730
 731 bool CastSocket::CalledOnValidThread() const {
 732   return thread_checker_.CalledOnValidThread();
 733 }
 734
 735 base::Timer* CastSocket::GetTimer() {
 736   return connect_timeout_timer_.get();
 737 }
 738
 739 CastSocket::MessageHeader::MessageHeader() : message_size(0) { }
 740
 741 void CastSocket::MessageHeader::SetMessageSize(size_t size) {
 742   DCHECK(size < static_cast<size_t>(kuint32max));
 743   DCHECK(size > 0);
 744   message_size = static_cast<size_t>(size);
 745 }
 746
 747 // TODO(mfoltz): Investigate replacing header serialization with base::Pickle,
 748 // if bit-for-bit compatible.
 749 void CastSocket::MessageHeader::PrependToString(std::string* str) {
 750   MessageHeader output = *this;
 751   output.message_size = base::HostToNet32(message_size);
 752   size_t header_size = base::checked_cast<size_t,uint32>(
 753       MessageHeader::header_size());
 754   scoped_ptr<char, base::FreeDeleter> char_array(
 755       static_cast<char*>(malloc(header_size)));
 756   memcpy(char_array.get(), &output, header_size);
 757   str->insert(0, char_array.get(), header_size);
 758 }
 759
 760 // TODO(mfoltz): Investigate replacing header deserialization with base::Pickle,
 761 // if bit-for-bit compatible.
 762 void CastSocket::MessageHeader::ReadFromIOBuffer(
 763     net::GrowableIOBuffer* buffer, MessageHeader* header) {
 764   uint32 message_size;
 765   size_t header_size = base::checked_cast<size_t,uint32>(
 766       MessageHeader::header_size());
 767   memcpy(&message_size, buffer->StartOfBuffer(), header_size);
 768   header->message_size = base::NetToHost32(message_size);
 769 }
 770
 771 std::string CastSocket::MessageHeader::ToString() {
 772   return "{message_size: " + base::UintToString(message_size) + "}";
 773 }
 774
 775 CastSocket::WriteRequest::WriteRequest(const net::CompletionCallback& callback)
 776   : callback(callback) { }
 777
 778 bool CastSocket::WriteRequest::SetContent(const CastMessage& message_proto) {
 779   DCHECK(!io_buffer.get());
 780   std::string message_data;
 781   if (!Serialize(message_proto, &message_data))
 782     return false;
 783   io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(message_data),
 784                                          message_data.size());
 785   return true;
 786 }
 787
 788 CastSocket::WriteRequest::~WriteRequest() { }
 789
 790 }  // namespace cast_channel
 791 }  // namespace api
 792 }  // namespace extensions
 793
 794 #undef VLOG_WITH_CONNECTION