components/cast_channel/cast_socket.cc

   1 // Copyright 2014 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 "components/cast_channel/cast_socket.h"
   6
   7 #include <stdlib.h>
   8 #include <string.h>
   9
  10 #include <memory>
  11 #include <utility>
  12
  13 #include "base/bind.h"
  14 #include "base/callback_helpers.h"
  15 #include "base/format_macros.h"
  16 #include "base/lazy_instance.h"
  17 #include "base/location.h"
  18 #include "base/memory/ptr_util.h"
  19 #include "base/numerics/safe_conversions.h"
  20 #include "base/single_thread_task_runner.h"
  21 #include "base/strings/string_number_conversions.h"
  22 #include "base/strings/stringprintf.h"
  23 #include "base/sys_byteorder.h"
  24 #include "base/task/post_task.h"
  25 #include "base/threading/thread_task_runner_handle.h"
  26 #include "base/time/time.h"
  27 #include "components/cast_channel/cast_auth_util.h"
  28 #include "components/cast_channel/cast_framer.h"
  29 #include "components/cast_channel/cast_message_util.h"
  30 #include "components/cast_channel/cast_transport.h"
  31 #include "components/cast_channel/keep_alive_delegate.h"
  32 #include "components/cast_channel/logger.h"
  33 #include "components/cast_channel/mojo_data_pump.h"
  34 #include "components/cast_channel/proto/cast_channel.pb.h"
  35 #include "content/public/browser/browser_task_traits.h"
  36 #include "content/public/browser/browser_thread.h"
  37 #include "net/base/address_list.h"
  38 #include "net/base/host_port_pair.h"
  39 #include "net/base/net_errors.h"
  40 #include "net/ssl/ssl_info.h"
  41 #include "net/traffic_annotation/network_traffic_annotation.h"
  42
  43 // Helper for logging data with remote host IP and authentication state.
  44 // Assumes |ip_endpoint_| of type net::IPEndPoint and |channel_auth_| of enum
  45 // type ChannelAuthType are available in the current scope.
  46 #define CONNECTION_INFO()                                             \
  47   "[" << open_params_.ip_endpoint.ToString() << ", auth=SSL_VERIFIED" \
  48       << "] "
  49 #define VLOG_WITH_CONNECTION(level) VLOG(level) << CONNECTION_INFO()
  50 #define LOG_WITH_CONNECTION(level) LOG(level) << CONNECTION_INFO()
  51
  52 namespace cast_channel {
  53 namespace {
  54
  55 bool IsTerminalState(ConnectionState state) {
  56   return state == ConnectionState::FINISHED ||
  57          state == ConnectionState::CONNECT_ERROR ||
  58          state == ConnectionState::TIMEOUT;
  59 }
  60
  61 void OnConnected(
  62     network::mojom::NetworkContext::CreateTCPConnectedSocketCallback callback,
  63     int result,
  64     const base::Optional<net::IPEndPoint>& local_addr,
  65     const base::Optional<net::IPEndPoint>& peer_addr,
  66     mojo::ScopedDataPipeConsumerHandle receive_stream,
  67     mojo::ScopedDataPipeProducerHandle send_stream) {
  68   DCHECK_CURRENTLY_ON(content::BrowserThread::UI);
  69   base::PostTaskWithTraits(
  70       FROM_HERE, {content::BrowserThread::IO},
  71       base::BindOnce(std::move(callback), result, local_addr, peer_addr,
  72                      std::move(receive_stream), std::move(send_stream)));
  73 }
  74
  75 void ConnectOnUIThread(
  76     CastSocketImpl::NetworkContextGetter network_context_getter,
  77     const net::AddressList& remote_address_list,
  78     network::mojom::TCPConnectedSocketRequest request,
  79     network::mojom::NetworkContext::CreateTCPConnectedSocketCallback callback) {
  80   network_context_getter.Run()->CreateTCPConnectedSocket(
  81       base::nullopt /* local_addr */, remote_address_list,
  82       nullptr /* tcp_connected_socket_options */,
  83       net::MutableNetworkTrafficAnnotationTag(
  84           CastSocketImpl::GetNetworkTrafficAnnotationTag()),
  85       std::move(request), nullptr /* observer */,
  86       base::BindOnce(OnConnected, std::move(callback)));
  87 }
  88
  89 }  // namespace
  90
  91 void CastSocket::Observer::OnReadyStateChanged(const CastSocket& socket) {}
  92
  93 CastSocketImpl::CastSocketImpl(NetworkContextGetter network_context_getter,
  94                                const CastSocketOpenParams& open_params,
  95                                const scoped_refptr<Logger>& logger)
  96     : CastSocketImpl(network_context_getter,
  97                      open_params,
  98                      logger,
  99                      AuthContext::Create()) {}
 100
 101 CastSocketImpl::CastSocketImpl(NetworkContextGetter network_context_getter,
 102                                const CastSocketOpenParams& open_params,
 103                                const scoped_refptr<Logger>& logger,
 104                                const AuthContext& auth_context)
 105     : channel_id_(0),
 106       open_params_(open_params),
 107       logger_(logger),
 108       network_context_getter_(network_context_getter),
 109       auth_context_(auth_context),
 110       connect_timeout_timer_(new base::OneShotTimer),
 111       is_canceled_(false),
 112       audio_only_(false),
 113       connect_state_(ConnectionState::START_CONNECT),
 114       error_state_(ChannelError::NONE),
 115       ready_state_(ReadyState::NONE),
 116       auth_delegate_(nullptr),
 117       weak_factory_(this) {
 118   DCHECK(open_params.ip_endpoint.address().IsValid());
 119 }
 120
 121 CastSocketImpl::~CastSocketImpl() {
 122   // Ensure that resources are freed but do not run pending callbacks that
 123   // would result in re-entrancy.
 124   CloseInternal();
 125
 126   error_state_ = ChannelError::UNKNOWN;
 127   for (auto& connect_callback : connect_callbacks_)
 128     std::move(connect_callback).Run(this);
 129   connect_callbacks_.clear();
 130 }
 131
 132 ReadyState CastSocketImpl::ready_state() const {
 133   return ready_state_;
 134 }
 135
 136 ChannelError CastSocketImpl::error_state() const {
 137   return error_state_;
 138 }
 139
 140 const net::IPEndPoint& CastSocketImpl::ip_endpoint() const {
 141   return open_params_.ip_endpoint;
 142 }
 143
 144 int CastSocketImpl::id() const {
 145   return channel_id_;
 146 }
 147
 148 void CastSocketImpl::set_id(int id) {
 149   channel_id_ = id;
 150 }
 151
 152 bool CastSocketImpl::keep_alive() const {
 153   return open_params_.liveness_timeout > base::TimeDelta();
 154 }
 155
 156 bool CastSocketImpl::audio_only() const {
 157   return audio_only_;
 158 }
 159
 160 bool CastSocketImpl::VerifyChannelPolicy(const AuthResult& result) {
 161   audio_only_ = (result.channel_policies & AuthResult::POLICY_AUDIO_ONLY) != 0;
 162   if (audio_only_ && (open_params_.device_capabilities &
 163                       CastDeviceCapability::VIDEO_OUT) != 0) {
 164     LOG_WITH_CONNECTION(ERROR)
 165         << "Audio only channel policy enforced for video out capable device";
 166     return false;
 167   }
 168   return true;
 169 }
 170
 171 bool CastSocketImpl::VerifyChallengeReply() {
 172   DCHECK(peer_cert_);
 173   AuthResult result =
 174       AuthenticateChallengeReply(*challenge_reply_, *peer_cert_, auth_context_);
 175   logger_->LogSocketChallengeReplyEvent(channel_id_, result);
 176   if (result.success()) {
 177     VLOG(1) << result.error_message;
 178     if (!VerifyChannelPolicy(result)) {
 179       return false;
 180     }
 181   }
 182   return result.success();
 183 }
 184
 185 void CastSocketImpl::SetTransportForTesting(
 186     std::unique_ptr<CastTransport> transport) {
 187   transport_ = std::move(transport);
 188 }
 189
 190 void CastSocketImpl::SetPeerCertForTesting(
 191     scoped_refptr<net::X509Certificate> peer_cert) {
 192   peer_cert_ = peer_cert;
 193 }
 194
 195 void CastSocketImpl::Connect(OnOpenCallback callback) {
 196   switch (ready_state_) {
 197     case ReadyState::NONE:
 198       connect_callbacks_.push_back(std::move(callback));
 199       Connect();
 200       break;
 201     case ReadyState::CONNECTING:
 202       connect_callbacks_.push_back(std::move(callback));
 203       break;
 204     case ReadyState::OPEN:
 205       error_state_ = ChannelError::NONE;
 206       std::move(callback).Run(this);
 207       break;
 208     case ReadyState::CLOSED:
 209       error_state_ = ChannelError::CONNECT_ERROR;
 210       std::move(callback).Run(this);
 211       break;
 212     default:
 213       NOTREACHED() << "Unknown ReadyState: "
 214                    << ReadyStateToString(ready_state_);
 215   }
 216 }
 217
 218 void CastSocketImpl::Connect() {
 219   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
 220   VLOG_WITH_CONNECTION(1) << "Connect readyState = "
 221                           << ReadyStateToString(ready_state_);
 222   DCHECK_EQ(ReadyState::NONE, ready_state_);
 223   DCHECK_EQ(ConnectionState::START_CONNECT, connect_state_);
 224
 225   delegate_ = std::make_unique<CastSocketMessageDelegate>(this);
 226
 227   SetReadyState(ReadyState::CONNECTING);
 228   SetConnectState(ConnectionState::TCP_CONNECT);
 229
 230   // Set up connection timeout.
 231   if (open_params_.connect_timeout.InMicroseconds() > 0) {
 232     DCHECK(connect_timeout_callback_.IsCancelled());
 233     connect_timeout_callback_.Reset(
 234         base::Bind(&CastSocketImpl::OnConnectTimeout, base::Unretained(this)));
 235     GetTimer()->Start(FROM_HERE, open_params_.connect_timeout,
 236                       connect_timeout_callback_.callback());
 237   }
 238
 239   DoConnectLoop(net::OK);
 240 }
 241
 242 CastTransport* CastSocketImpl::transport() const {
 243   return transport_.get();
 244 }
 245
 246 void CastSocketImpl::AddObserver(Observer* observer) {
 247   DCHECK(observer);
 248   if (!observers_.HasObserver(observer))
 249     observers_.AddObserver(observer);
 250 }
 251
 252 void CastSocketImpl::RemoveObserver(Observer* observer) {
 253   DCHECK(observer);
 254   observers_.RemoveObserver(observer);
 255 }
 256
 257 net::NetworkTrafficAnnotationTag
 258 CastSocketImpl::GetNetworkTrafficAnnotationTag() {
 259   return net::DefineNetworkTrafficAnnotation("cast_socket", R"(
 260         semantics {
 261           sender: "Cast Socket"
 262           description:
 263             "Requests to a Cast device."
 264           trigger:
 265             "A new Cast device has been discovered via mDNS in the local "
 266             "network or after it's connected."
 267           data:
 268             "A serialized Cast protocol or application-level protobuf message. "
 269             "A non-exhaustive list of Cast protocol messages:\n"
 270             "- nonce challenge,\n"
 271             "- ping/pong data,\n"
 272             "- Virtual connection requests,\n"
 273             "- App availability / media status / receiver status requests,\n"
 274             "- Launch / stop Cast session requests,\n"
 275             "- Media commands, such as play/pause.\n"
 276             "Application-level messages may contain data specific to the Cast "
 277             "application."
 278           destination: OTHER
 279           destination_other:
 280             "Data will be sent to a Cast device in local network."
 281         }
 282         policy {
 283           cookies_allowed: NO
 284           setting:
 285             "This request cannot be disabled, but it would not be sent if user "
 286             "does not connect a Cast device to the local network."
 287           policy_exception_justification: "Not implemented."
 288         })");
 289 }
 290
 291 void CastSocketImpl::OnConnectTimeout() {
 292   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
 293   // Stop all pending connection setup tasks and report back to the client.
 294   is_canceled_ = true;
 295   VLOG_WITH_CONNECTION(1) << "Timeout while establishing a connection.";
 296   SetErrorState(ChannelError::CONNECT_TIMEOUT);
 297   DoConnectCallback();
 298 }
 299
 300 void CastSocketImpl::ResetConnectLoopCallback() {
 301   DCHECK(connect_loop_callback_.IsCancelled());
 302   connect_loop_callback_.Reset(
 303       base::Bind(&CastSocketImpl::DoConnectLoop, base::Unretained(this)));
 304 }
 305
 306 void CastSocketImpl::PostTaskToStartConnectLoop(int result) {
 307   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
 308
 309   ResetConnectLoopCallback();
 310   base::ThreadTaskRunnerHandle::Get()->PostTask(
 311       FROM_HERE, base::BindOnce(connect_loop_callback_.callback(), result));
 312 }
 313
 314 // This method performs the state machine transitions for connection flow.
 315 // There are two entry points to this method:
 316 // 1. Connect method: this starts the flow
 317 // 2. Callback from network operations that finish asynchronously.
 318 void CastSocketImpl::DoConnectLoop(int result) {
 319   connect_loop_callback_.Cancel();
 320   if (is_canceled_) {
 321     LOG_WITH_CONNECTION(ERROR) << "CANCELLED - Aborting DoConnectLoop.";
 322     return;
 323   }
 324
 325   // Network operations can either finish synchronously or asynchronously.
 326   // This method executes the state machine transitions in a loop so that
 327   // correct state transitions happen even when network operations finish
 328   // synchronously.
 329   int rv = result;
 330   do {
 331     ConnectionState state = connect_state_;
 332     connect_state_ = ConnectionState::UNKNOWN;
 333     switch (state) {
 334       case ConnectionState::TCP_CONNECT:
 335         rv = DoTcpConnect();
 336         break;
 337       case ConnectionState::TCP_CONNECT_COMPLETE:
 338         rv = DoTcpConnectComplete(rv);
 339         break;
 340       case ConnectionState::SSL_CONNECT:
 341         DCHECK_EQ(net::OK, rv);
 342         rv = DoSslConnect();
 343         break;
 344       case ConnectionState::SSL_CONNECT_COMPLETE:
 345         rv = DoSslConnectComplete(rv);
 346         break;
 347       case ConnectionState::AUTH_CHALLENGE_SEND:
 348         rv = DoAuthChallengeSend();
 349         break;
 350       case ConnectionState::AUTH_CHALLENGE_SEND_COMPLETE:
 351         rv = DoAuthChallengeSendComplete(rv);
 352         break;
 353       case ConnectionState::AUTH_CHALLENGE_REPLY_COMPLETE:
 354         rv = DoAuthChallengeReplyComplete(rv);
 355         DCHECK(IsTerminalState(connect_state_));
 356         break;
 357       default:
 358         NOTREACHED() << "Unknown state in connect flow: " << AsInteger(state);
 359         SetConnectState(ConnectionState::FINISHED);
 360         SetErrorState(ChannelError::UNKNOWN);
 361         DoConnectCallback();
 362         return;
 363     }
 364   } while (rv != net::ERR_IO_PENDING && !IsTerminalState(connect_state_));
 365   // Exit the state machine if an asynchronous network operation is pending
 366   // or if the state machine is in the terminal "finished" state.
 367
 368   if (IsTerminalState(connect_state_)) {
 369     DCHECK_NE(rv, net::ERR_IO_PENDING);
 370     GetTimer()->Stop();
 371     DoConnectCallback();
 372   } else {
 373     DCHECK_EQ(rv, net::ERR_IO_PENDING);
 374   }
 375 }
 376
 377 int CastSocketImpl::DoTcpConnect() {
 378   DCHECK(connect_loop_callback_.IsCancelled());
 379   VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
 380   SetConnectState(ConnectionState::TCP_CONNECT_COMPLETE);
 381
 382   base::PostTaskWithTraits(
 383       FROM_HERE, {content::BrowserThread::UI},
 384       base::BindOnce(ConnectOnUIThread, network_context_getter_,
 385                      net::AddressList(open_params_.ip_endpoint),
 386                      mojo::MakeRequest(&tcp_socket_),
 387                      base::BindOnce(&CastSocketImpl::OnConnect,
 388                                     weak_factory_.GetWeakPtr())));
 389
 390   return net::ERR_IO_PENDING;
 391 }
 392
 393 int CastSocketImpl::DoTcpConnectComplete(int connect_result) {
 394   VLOG_WITH_CONNECTION(1) << "DoTcpConnectComplete: " << connect_result;
 395   logger_->LogSocketEventWithRv(
 396       channel_id_, ChannelEvent::TCP_SOCKET_CONNECT_COMPLETE, connect_result);
 397   if (connect_result == net::OK) {
 398     SetConnectState(ConnectionState::SSL_CONNECT);
 399   } else if (connect_result == net::ERR_CONNECTION_TIMED_OUT) {
 400     SetConnectState(ConnectionState::FINISHED);
 401     SetErrorState(ChannelError::CONNECT_TIMEOUT);
 402   } else {
 403     SetConnectState(ConnectionState::FINISHED);
 404     SetErrorState(ChannelError::CONNECT_ERROR);
 405   }
 406   return connect_result;
 407 }
 408
 409 int CastSocketImpl::DoSslConnect() {
 410   DCHECK(connect_loop_callback_.IsCancelled());
 411   VLOG_WITH_CONNECTION(1) << "DoSslConnect";
 412   SetConnectState(ConnectionState::SSL_CONNECT_COMPLETE);
 413
 414   net::HostPortPair host_port_pair =
 415       net::HostPortPair::FromIPEndPoint(open_params_.ip_endpoint);
 416   network::mojom::TLSClientSocketOptionsPtr options =
 417       network::mojom::TLSClientSocketOptions::New();
 418   // Cast code does its own authentication after SSL handshake since the devices
 419   // don't have a known hostname.
 420   options->unsafely_skip_cert_verification = true;
 421   tcp_socket_->UpgradeToTLS(
 422       host_port_pair, std::move(options),
 423       net::MutableNetworkTrafficAnnotationTag(GetNetworkTrafficAnnotationTag()),
 424       mojo::MakeRequest(&socket_), nullptr /* observer */,
 425       base::BindOnce(&CastSocketImpl::OnUpgradeToTLS,
 426                      weak_factory_.GetWeakPtr()));
 427
 428   return net::ERR_IO_PENDING;
 429 }
 430
 431 int CastSocketImpl::DoSslConnectComplete(int result) {
 432   logger_->LogSocketEventWithRv(
 433       channel_id_, ChannelEvent::SSL_SOCKET_CONNECT_COMPLETE, result);
 434   VLOG_WITH_CONNECTION(1) << "DoSslConnectComplete: " << result;
 435   if (result == net::OK) {
 436     if (!peer_cert_) {
 437       LOG_WITH_CONNECTION(WARNING) << "Could not extract peer cert.";
 438       SetConnectState(ConnectionState::FINISHED);
 439       SetErrorState(ChannelError::AUTHENTICATION_ERROR);
 440       return net::ERR_CERT_INVALID;
 441     }
 442
 443     // SSL connection succeeded.
 444     if (!transport_) {
 445       // Create a channel transport if one wasn't already set (e.g. by test
 446       // code).
 447       transport_.reset(new CastTransportImpl(mojo_data_pump_.get(), channel_id_,
 448                                              open_params_.ip_endpoint,
 449                                              logger_));
 450     }
 451     auth_delegate_ = new AuthTransportDelegate(this);
 452     transport_->SetReadDelegate(base::WrapUnique(auth_delegate_));
 453     SetConnectState(ConnectionState::AUTH_CHALLENGE_SEND);
 454   } else if (result == net::ERR_CONNECTION_TIMED_OUT) {
 455     SetConnectState(ConnectionState::FINISHED);
 456     SetErrorState(ChannelError::CONNECT_TIMEOUT);
 457   } else {
 458     SetConnectState(ConnectionState::FINISHED);
 459     SetErrorState(ChannelError::AUTHENTICATION_ERROR);
 460   }
 461   return result;
 462 }
 463
 464 int CastSocketImpl::DoAuthChallengeSend() {
 465   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSend";
 466   SetConnectState(ConnectionState::AUTH_CHALLENGE_SEND_COMPLETE);
 467
 468   CastMessage challenge_message;
 469   CreateAuthChallengeMessage(&challenge_message, auth_context_);
 470   VLOG_WITH_CONNECTION(1) << "Sending challenge: "
 471                           << CastMessageToString(challenge_message);
 472
 473   ResetConnectLoopCallback();
 474
 475   transport_->SendMessage(challenge_message, connect_loop_callback_.callback());
 476
 477   // Always return IO_PENDING since the result is always asynchronous.
 478   return net::ERR_IO_PENDING;
 479 }
 480
 481 int CastSocketImpl::DoAuthChallengeSendComplete(int result) {
 482   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
 483   if (result < 0) {
 484     SetConnectState(ConnectionState::CONNECT_ERROR);
 485     SetErrorState(ChannelError::CAST_SOCKET_ERROR);
 486     logger_->LogSocketEventWithRv(
 487         channel_id_, ChannelEvent::SEND_AUTH_CHALLENGE_FAILED, result);
 488     return result;
 489   }
 490   transport_->Start();
 491   SetConnectState(ConnectionState::AUTH_CHALLENGE_REPLY_COMPLETE);
 492   return net::ERR_IO_PENDING;
 493 }
 494
 495 CastSocketImpl::AuthTransportDelegate::AuthTransportDelegate(
 496     CastSocketImpl* socket)
 497     : socket_(socket), error_state_(ChannelError::NONE) {
 498   DCHECK(socket);
 499 }
 500
 501 ChannelError CastSocketImpl::AuthTransportDelegate::error_state() const {
 502   return error_state_;
 503 }
 504
 505 LastError CastSocketImpl::AuthTransportDelegate::last_error() const {
 506   return last_error_;
 507 }
 508
 509 void CastSocketImpl::AuthTransportDelegate::OnError(ChannelError error_state) {
 510   error_state_ = error_state;
 511   socket_->PostTaskToStartConnectLoop(net::ERR_CONNECTION_FAILED);
 512 }
 513
 514 void CastSocketImpl::AuthTransportDelegate::OnMessage(
 515     const CastMessage& message) {
 516   if (!IsAuthMessage(message)) {
 517     error_state_ = ChannelError::TRANSPORT_ERROR;
 518     socket_->PostTaskToStartConnectLoop(net::ERR_INVALID_RESPONSE);
 519   } else {
 520     socket_->challenge_reply_.reset(new CastMessage(message));
 521     socket_->PostTaskToStartConnectLoop(net::OK);
 522   }
 523 }
 524
 525 void CastSocketImpl::AuthTransportDelegate::Start() {}
 526
 527 int CastSocketImpl::DoAuthChallengeReplyComplete(int result) {
 528   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeReplyComplete: " << result;
 529
 530   if (auth_delegate_->error_state() != ChannelError::NONE) {
 531     SetErrorState(auth_delegate_->error_state());
 532     SetConnectState(ConnectionState::CONNECT_ERROR);
 533     return net::ERR_CONNECTION_FAILED;
 534   }
 535   auth_delegate_ = nullptr;
 536
 537   if (result < 0) {
 538     SetConnectState(ConnectionState::CONNECT_ERROR);
 539     return result;
 540   }
 541
 542   if (!VerifyChallengeReply()) {
 543     SetErrorState(ChannelError::AUTHENTICATION_ERROR);
 544     SetConnectState(ConnectionState::CONNECT_ERROR);
 545     return net::ERR_CONNECTION_FAILED;
 546   }
 547   VLOG_WITH_CONNECTION(1) << "Auth challenge verification succeeded";
 548
 549   SetConnectState(ConnectionState::FINISHED);
 550   return net::OK;
 551 }
 552
 553 void CastSocketImpl::OnConnect(
 554     int result,
 555     const base::Optional<net::IPEndPoint>& local_addr,
 556     const base::Optional<net::IPEndPoint>& peer_addr,
 557     mojo::ScopedDataPipeConsumerHandle receive_stream,
 558     mojo::ScopedDataPipeProducerHandle send_stream) {
 559   DoConnectLoop(result);
 560 }
 561
 562 void CastSocketImpl::OnUpgradeToTLS(
 563     int result,
 564     mojo::ScopedDataPipeConsumerHandle receive_stream,
 565     mojo::ScopedDataPipeProducerHandle send_stream,
 566     const base::Optional<net::SSLInfo>& ssl_info) {
 567   if (result == net::OK) {
 568     mojo_data_pump_ = std::make_unique<MojoDataPump>(std::move(receive_stream),
 569                                                      std::move(send_stream));
 570   }
 571   if (ssl_info.has_value() && ssl_info->cert)
 572     peer_cert_ = ssl_info->cert;
 573   DoConnectLoop(result);
 574 }
 575
 576 void CastSocketImpl::DoConnectCallback() {
 577   VLOG(1) << "DoConnectCallback (error_state = "
 578           << ChannelErrorToString(error_state_) << ")";
 579   if (connect_callbacks_.empty()) {
 580     DLOG(FATAL) << "Connection callback invoked multiple times.";
 581     return;
 582   }
 583
 584   if (error_state_ == ChannelError::NONE) {
 585     SetReadyState(ReadyState::OPEN);
 586     if (keep_alive()) {
 587       auto* keep_alive_delegate = new KeepAliveDelegate(
 588           this, logger_, std::move(delegate_), open_params_.ping_interval,
 589           open_params_.liveness_timeout);
 590       delegate_.reset(keep_alive_delegate);
 591     }
 592     transport_->SetReadDelegate(std::move(delegate_));
 593   } else {
 594     CloseInternal();
 595   }
 596
 597   for (auto& connect_callback : connect_callbacks_)
 598     std::move(connect_callback).Run(this);
 599   connect_callbacks_.clear();
 600 }
 601
 602 void CastSocketImpl::Close(const net::CompletionCallback& callback) {
 603   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
 604   CloseInternal();
 605   // Run this callback last.  It may delete the socket.
 606   callback.Run(net::OK);
 607 }
 608
 609 void CastSocketImpl::CloseInternal() {
 610   // TODO(mfoltz): Enforce this when CastChannelAPITest is rewritten to create
 611   // and free sockets on the same thread.  crbug.com/398242
 612   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
 613   if (ready_state_ == ReadyState::CLOSED) {
 614     return;
 615   }
 616
 617   VLOG_WITH_CONNECTION(1) << "Close ReadyState = "
 618                           << ReadyStateToString(ready_state_);
 619   observers_.Clear();
 620   delegate_.reset();
 621   mojo_data_pump_.reset();
 622   transport_.reset();
 623   tcp_socket_.reset();
 624   socket_.reset();
 625   if (GetTimer()) {
 626     GetTimer()->Stop();
 627   }
 628
 629   // Cancel callbacks that we queued ourselves to re-enter the connect or read
 630   // loops.
 631   connect_loop_callback_.Cancel();
 632   connect_timeout_callback_.Cancel();
 633   SetReadyState(ReadyState::CLOSED);
 634 }
 635
 636 base::OneShotTimer* CastSocketImpl::GetTimer() {
 637   return connect_timeout_timer_.get();
 638 }
 639
 640 void CastSocketImpl::SetConnectState(ConnectionState connect_state) {
 641   if (connect_state_ != connect_state) {
 642     connect_state_ = connect_state;
 643   }
 644 }
 645
 646 void CastSocketImpl::SetReadyState(ReadyState ready_state) {
 647   if (ready_state_ != ready_state) {
 648     ready_state_ = ready_state;
 649     for (auto& observer : observers_)
 650       observer.OnReadyStateChanged(*this);
 651   }
 652 }
 653
 654 void CastSocketImpl::SetErrorState(ChannelError error_state) {
 655   VLOG_WITH_CONNECTION(1) << "SetErrorState "
 656                           << ChannelErrorToString(error_state);
 657   DCHECK_EQ(ChannelError::NONE, error_state_);
 658   error_state_ = error_state;
 659   delegate_->OnError(error_state_);
 660 }
 661
 662 CastSocketImpl::CastSocketMessageDelegate::CastSocketMessageDelegate(
 663     CastSocketImpl* socket)
 664     : socket_(socket) {
 665   DCHECK(socket_);
 666 }
 667
 668 CastSocketImpl::CastSocketMessageDelegate::~CastSocketMessageDelegate() {}
 669
 670 // CastTransport::Delegate implementation.
 671 void CastSocketImpl::CastSocketMessageDelegate::OnError(
 672     ChannelError error_state) {
 673   for (auto& observer : socket_->observers_)
 674     observer.OnError(*socket_, error_state);
 675 }
 676
 677 void CastSocketImpl::CastSocketMessageDelegate::OnMessage(
 678     const CastMessage& message) {
 679   for (auto& observer : socket_->observers_)
 680     observer.OnMessage(*socket_, message);
 681 }
 682
 683 void CastSocketImpl::CastSocketMessageDelegate::Start() {}
 684
 685 CastSocketOpenParams::CastSocketOpenParams(const net::IPEndPoint& ip_endpoint,
 686                                            base::TimeDelta connect_timeout)
 687     : ip_endpoint(ip_endpoint),
 688       connect_timeout(connect_timeout),
 689       device_capabilities(cast_channel::CastDeviceCapability::NONE) {}
 690
 691 CastSocketOpenParams::CastSocketOpenParams(const net::IPEndPoint& ip_endpoint,
 692                                            base::TimeDelta connect_timeout,
 693                                            base::TimeDelta liveness_timeout,
 694                                            base::TimeDelta ping_interval,
 695                                            uint64_t device_capabilities)
 696     : ip_endpoint(ip_endpoint),
 697       connect_timeout(connect_timeout),
 698       liveness_timeout(liveness_timeout),
 699       ping_interval(ping_interval),
 700       device_capabilities(device_capabilities) {}
 701
 702 }  // namespace cast_channel
 703 #undef VLOG_WITH_CONNECTION