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