remoting/protocol/quic_channel_factory.cc

   1 // Copyright 2015 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 "remoting/protocol/quic_channel_factory.h"
   6
   7 #include <vector>
   8
   9 #include "base/bind.h"
  10 #include "base/location.h"
  11 #include "base/single_thread_task_runner.h"
  12 #include "base/stl_util.h"
  13 #include "base/thread_task_runner_handle.h"
  14 #include "net/base/io_buffer.h"
  15 #include "net/base/net_errors.h"
  16 #include "net/quic/crypto/crypto_framer.h"
  17 #include "net/quic/crypto/crypto_handshake_message.h"
  18 #include "net/quic/crypto/crypto_protocol.h"
  19 #include "net/quic/crypto/quic_random.h"
  20 #include "net/quic/p2p/quic_p2p_crypto_config.h"
  21 #include "net/quic/p2p/quic_p2p_session.h"
  22 #include "net/quic/p2p/quic_p2p_stream.h"
  23 #include "net/quic/quic_clock.h"
  24 #include "net/quic/quic_connection_helper.h"
  25 #include "net/quic/quic_default_packet_writer.h"
  26 #include "net/socket/stream_socket.h"
  27 #include "remoting/base/constants.h"
  28 #include "remoting/protocol/datagram_channel_factory.h"
  29 #include "remoting/protocol/p2p_datagram_socket.h"
  30 #include "remoting/protocol/quic_channel.h"
  31
  32 namespace remoting {
  33 namespace protocol {
  34
  35 namespace {
  36
  37 // The maximum receive window sizes for QUIC sessions and streams. These are
  38 // the same values that are used in chrome.
  39 const int kQuicSessionMaxRecvWindowSize = 15 * 1024 * 1024;  // 15 MB
  40 const int kQuicStreamMaxRecvWindowSize = 6 * 1024 * 1024;    // 6 MB
  41
  42 class P2PQuicPacketWriter : public net::QuicPacketWriter {
  43  public:
  44   P2PQuicPacketWriter(net::QuicConnection* connection,
  45                       P2PDatagramSocket* socket)
  46       : connection_(connection), socket_(socket), weak_factory_(this) {}
  47   ~P2PQuicPacketWriter() override {}
  48
  49   // QuicPacketWriter interface.
  50   net::WriteResult WritePacket(const char* buffer,
  51                                size_t buf_len,
  52                                const net::IPAddressNumber& self_address,
  53                                const net::IPEndPoint& peer_address) override {
  54     DCHECK(!write_blocked_);
  55
  56     scoped_refptr<net::StringIOBuffer> buf(
  57         new net::StringIOBuffer(std::string(buffer, buf_len)));
  58     int result = socket_->Send(buf, buf_len,
  59                                base::Bind(&P2PQuicPacketWriter::OnSendComplete,
  60                                           weak_factory_.GetWeakPtr()));
  61     net::WriteStatus status = net::WRITE_STATUS_OK;
  62     if (result < 0) {
  63       if (result == net::ERR_IO_PENDING) {
  64         status = net::WRITE_STATUS_BLOCKED;
  65         write_blocked_ = true;
  66       } else {
  67         status = net::WRITE_STATUS_ERROR;
  68       }
  69     }
  70
  71     return net::WriteResult(status, result);
  72   }
  73   bool IsWriteBlockedDataBuffered() const override {
  74     // P2PDatagramSocket::Send() method buffer the data until the Send is
  75     // unblocked.
  76     return true;
  77   }
  78   bool IsWriteBlocked() const override { return write_blocked_; }
  79   void SetWritable() override { write_blocked_ = false; }
  80
  81  private:
  82   void OnSendComplete(int result){
  83     DCHECK_NE(result, net::ERR_IO_PENDING);
  84     write_blocked_ = false;
  85     if (result < 0) {
  86       connection_->OnWriteError(result);
  87     }
  88     connection_->OnCanWrite();
  89   }
  90
  91   net::QuicConnection* connection_;
  92   P2PDatagramSocket* socket_;
  93
  94   // Whether a write is currently in flight.
  95   bool write_blocked_ = false;
  96
  97   base::WeakPtrFactory<P2PQuicPacketWriter> weak_factory_;
  98
  99   DISALLOW_COPY_AND_ASSIGN(P2PQuicPacketWriter);
 100 };
 101
 102 class QuicPacketWriterFactory
 103     : public net::QuicConnection::PacketWriterFactory {
 104  public:
 105   explicit QuicPacketWriterFactory(P2PDatagramSocket* socket)
 106       : socket_(socket) {}
 107   ~QuicPacketWriterFactory() override {}
 108
 109   net::QuicPacketWriter* Create(
 110       net::QuicConnection* connection) const override {
 111     return new P2PQuicPacketWriter(connection, socket_);
 112   }
 113
 114  private:
 115   P2PDatagramSocket* socket_;
 116 };
 117
 118 class P2PDatagramSocketAdapter : public net::Socket {
 119  public:
 120   explicit P2PDatagramSocketAdapter(scoped_ptr<P2PDatagramSocket> socket)
 121       : socket_(socket.Pass()) {}
 122   ~P2PDatagramSocketAdapter() override {}
 123
 124   int Read(net::IOBuffer* buf, int buf_len,
 125            const net::CompletionCallback& callback) override {
 126     return socket_->Recv(buf, buf_len, callback);
 127   }
 128   int Write(net::IOBuffer* buf, int buf_len,
 129             const net::CompletionCallback& callback) override {
 130     return socket_->Send(buf, buf_len, callback);
 131   }
 132
 133   int SetReceiveBufferSize(int32_t size) override {
 134     NOTREACHED();
 135     return net::ERR_FAILED;
 136   }
 137
 138   int SetSendBufferSize(int32_t size) override {
 139     NOTREACHED();
 140     return net::ERR_FAILED;
 141   }
 142
 143  private:
 144   scoped_ptr<P2PDatagramSocket> socket_;
 145 };
 146
 147 }  // namespace
 148
 149 class QuicChannelFactory::Core : public net::QuicP2PSession::Delegate {
 150  public:
 151   Core(const std::string& session_id, bool is_server);
 152   virtual ~Core();
 153
 154   // Called from ~QuicChannelFactory() to synchronously release underlying
 155   // socket. Core is destroyed later asynchronously.
 156   void Close();
 157
 158   // Implementation of all all methods for QuicChannelFactory.
 159   const std::string& CreateSessionInitiateConfigMessage();
 160   bool ProcessSessionAcceptConfigMessage(const std::string& message);
 161
 162   bool ProcessSessionInitiateConfigMessage(const std::string& message);
 163   const std::string& CreateSessionAcceptConfigMessage();
 164
 165   void Start(DatagramChannelFactory* factory, const std::string& shared_secret);
 166
 167   void CreateChannel(const std::string& name,
 168                      const ChannelCreatedCallback& callback);
 169   void CancelChannelCreation(const std::string& name);
 170
 171  private:
 172   friend class QuicChannelFactory;
 173
 174   struct PendingChannel {
 175     PendingChannel(const std::string& name,
 176                    const ChannelCreatedCallback& callback)
 177         : name(name), callback(callback) {}
 178
 179     std::string name;
 180     ChannelCreatedCallback callback;
 181   };
 182
 183   // QuicP2PSession::Delegate interface.
 184   void OnIncomingStream(net::QuicP2PStream* stream) override;
 185   void OnConnectionClosed(net::QuicErrorCode error) override;
 186
 187   void OnBaseChannelReady(scoped_ptr<P2PDatagramSocket> socket);
 188
 189   void OnNameReceived(QuicChannel* channel);
 190
 191   void OnChannelDestroyed(int stream_id);
 192
 193   std::string session_id_;
 194   bool is_server_;
 195   DatagramChannelFactory* base_channel_factory_ = nullptr;
 196
 197   net::QuicConfig quic_config_;
 198   std::string shared_secret_;
 199   std::string session_initiate_quic_config_message_;
 200   std::string session_accept_quic_config_message_;
 201
 202   net::QuicClock quic_clock_;
 203   net::QuicConnectionHelper quic_helper_;
 204   scoped_ptr<net::QuicP2PSession> quic_session_;
 205   bool connected_ = false;
 206
 207   std::vector<PendingChannel*> pending_channels_;
 208   std::vector<QuicChannel*> unnamed_incoming_channels_;
 209
 210   base::WeakPtrFactory<Core> weak_factory_;
 211
 212   DISALLOW_COPY_AND_ASSIGN(Core);
 213 };
 214
 215 QuicChannelFactory::Core::Core(const std::string& session_id, bool is_server)
 216     : session_id_(session_id),
 217       is_server_(is_server),
 218       quic_helper_(base::ThreadTaskRunnerHandle::Get().get(),
 219                    &quic_clock_,
 220                    net::QuicRandom::GetInstance()),
 221       weak_factory_(this) {
 222   quic_config_.SetInitialSessionFlowControlWindowToSend(
 223       kQuicSessionMaxRecvWindowSize);
 224   quic_config_.SetInitialStreamFlowControlWindowToSend(
 225       kQuicStreamMaxRecvWindowSize);
 226 }
 227
 228 QuicChannelFactory::Core::~Core() {}
 229
 230 void QuicChannelFactory::Core::Close() {
 231   DCHECK(pending_channels_.empty());
 232
 233   // Cancel creation of the base channel if it hasn't finished.
 234   if (base_channel_factory_)
 235     base_channel_factory_->CancelChannelCreation(kQuicChannelName);
 236
 237   if (quic_session_ && quic_session_->connection()->connected())
 238     quic_session_->connection()->CloseConnection(net::QUIC_NO_ERROR, false);
 239
 240   DCHECK(unnamed_incoming_channels_.empty());
 241 }
 242
 243 void QuicChannelFactory::Core::Start(DatagramChannelFactory* factory,
 244                                      const std::string& shared_secret) {
 245   base_channel_factory_ = factory;
 246   shared_secret_ = shared_secret;
 247
 248   base_channel_factory_->CreateChannel(
 249       kQuicChannelName,
 250       base::Bind(&Core::OnBaseChannelReady, weak_factory_.GetWeakPtr()));
 251 }
 252
 253 const std::string&
 254 QuicChannelFactory::Core::CreateSessionInitiateConfigMessage() {
 255   DCHECK(!is_server_);
 256
 257   net::CryptoHandshakeMessage handshake_message;
 258   handshake_message.set_tag(net::kCHLO);
 259   quic_config_.ToHandshakeMessage(&handshake_message);
 260
 261   session_initiate_quic_config_message_ =
 262       handshake_message.GetSerialized().AsStringPiece().as_string();
 263   return session_initiate_quic_config_message_;
 264 }
 265
 266 bool QuicChannelFactory::Core::ProcessSessionAcceptConfigMessage(
 267     const std::string& message) {
 268   DCHECK(!is_server_);
 269
 270   session_accept_quic_config_message_ = message;
 271
 272   scoped_ptr<net::CryptoHandshakeMessage> parsed_message(
 273       net::CryptoFramer::ParseMessage(message));
 274   if (!parsed_message) {
 275     LOG(ERROR) << "Received invalid QUIC config.";
 276     return false;
 277   }
 278
 279   if (parsed_message->tag() != net::kSHLO) {
 280     LOG(ERROR) << "Received QUIC handshake message with unexpected tag "
 281                << parsed_message->tag();
 282     return false;
 283   }
 284
 285   std::string error_message;
 286   net::QuicErrorCode error = quic_config_.ProcessPeerHello(
 287       *parsed_message, net::SERVER, &error_message);
 288   if (error != net::QUIC_NO_ERROR) {
 289     LOG(ERROR) << "Failed to process QUIC handshake message: "
 290                << error_message;
 291     return false;
 292   }
 293
 294   return true;
 295 }
 296
 297 bool QuicChannelFactory::Core::ProcessSessionInitiateConfigMessage(
 298     const std::string& message) {
 299   DCHECK(is_server_);
 300
 301   session_initiate_quic_config_message_ = message;
 302
 303   scoped_ptr<net::CryptoHandshakeMessage> parsed_message(
 304       net::CryptoFramer::ParseMessage(message));
 305   if (!parsed_message) {
 306     LOG(ERROR) << "Received invalid QUIC config.";
 307     return false;
 308   }
 309
 310   if (parsed_message->tag() != net::kCHLO) {
 311     LOG(ERROR) << "Received QUIC handshake message with unexpected tag "
 312                << parsed_message->tag();
 313     return false;
 314   }
 315
 316   std::string error_message;
 317   net::QuicErrorCode error = quic_config_.ProcessPeerHello(
 318       *parsed_message, net::CLIENT, &error_message);
 319   if (error != net::QUIC_NO_ERROR) {
 320     LOG(ERROR) << "Failed to process QUIC handshake message: "
 321                << error_message;
 322     return false;
 323   }
 324
 325   return true;
 326 }
 327
 328 const std::string&
 329 QuicChannelFactory::Core::CreateSessionAcceptConfigMessage() {
 330   DCHECK(is_server_);
 331
 332   if (session_initiate_quic_config_message_.empty()) {
 333     // Don't send quic-config to the client if the client didn't include the
 334     // config in the session-initiate message.
 335     DCHECK(session_accept_quic_config_message_.empty());
 336     return session_accept_quic_config_message_;
 337   }
 338
 339   net::CryptoHandshakeMessage handshake_message;
 340   handshake_message.set_tag(net::kSHLO);
 341   quic_config_.ToHandshakeMessage(&handshake_message);
 342
 343   session_accept_quic_config_message_ =
 344       handshake_message.GetSerialized().AsStringPiece().as_string();
 345   return session_accept_quic_config_message_;
 346 }
 347
 348 // StreamChannelFactory interface.
 349 void QuicChannelFactory::Core::CreateChannel(
 350     const std::string& name,
 351     const ChannelCreatedCallback& callback) {
 352   if (quic_session_ && quic_session_->connection()->connected()) {
 353     if (!is_server_) {
 354       net::QuicP2PStream* stream = quic_session_->CreateOutgoingDynamicStream();
 355       scoped_ptr<QuicChannel> channel(new QuicClientChannel(
 356           stream, base::Bind(&Core::OnChannelDestroyed, base::Unretained(this),
 357                              stream->id()),
 358           name));
 359       callback.Run(channel.Pass());
 360     } else {
 361       // On the server side wait for the client to create a QUIC stream and
 362       // send the name. The channel will be connected in OnNameReceived().
 363       pending_channels_.push_back(new PendingChannel(name, callback));
 364     }
 365   } else if (!base_channel_factory_) {
 366     // Fail synchronously if we failed to connect transport.
 367     callback.Run(nullptr);
 368   } else {
 369     // Still waiting for the transport.
 370     pending_channels_.push_back(new PendingChannel(name, callback));
 371   }
 372 }
 373
 374 void QuicChannelFactory::Core::CancelChannelCreation(const std::string& name) {
 375   for (auto it = pending_channels_.begin(); it != pending_channels_.end();
 376        ++it) {
 377     if ((*it)->name == name) {
 378       delete *it;
 379       pending_channels_.erase(it);
 380       return;
 381     }
 382   }
 383 }
 384
 385 void QuicChannelFactory::Core::OnBaseChannelReady(
 386     scoped_ptr<P2PDatagramSocket> socket) {
 387   base_channel_factory_ = nullptr;
 388
 389   // Failed to connect underlying transport connection. Fail all pending
 390   // channel.
 391   if (!socket) {
 392     while (!pending_channels_.empty()) {
 393       scoped_ptr<PendingChannel> pending_channel(pending_channels_.front());
 394       pending_channels_.erase(pending_channels_.begin());
 395       pending_channel->callback.Run(nullptr);
 396     }
 397     return;
 398   }
 399
 400   QuicPacketWriterFactory writer_factory(socket.get());
 401   net::IPAddressNumber ip(net::kIPv4AddressSize, 0);
 402   scoped_ptr<net::QuicConnection> quic_connection(new net::QuicConnection(
 403       0, net::IPEndPoint(ip, 0), &quic_helper_, writer_factory,
 404       true /* owns_writer */,
 405       is_server_ ? net::Perspective::IS_SERVER : net::Perspective::IS_CLIENT,
 406       true /* is_secure */, net::QuicSupportedVersions()));
 407
 408   net::QuicP2PCryptoConfig quic_crypto_config(shared_secret_);
 409   quic_crypto_config.set_hkdf_input_suffix(
 410       session_id_ + '\0' + kQuicChannelName + '\0' +
 411       session_initiate_quic_config_message_ +
 412       session_accept_quic_config_message_);
 413
 414   quic_session_.reset(new net::QuicP2PSession(
 415       quic_config_, quic_crypto_config, quic_connection.Pass(),
 416       make_scoped_ptr(new P2PDatagramSocketAdapter(socket.Pass()))));
 417   quic_session_->SetDelegate(this);
 418   quic_session_->Initialize();
 419
 420   if (!is_server_) {
 421     // On the client create streams for all pending channels and send a name for
 422     // each channel.
 423     while (!pending_channels_.empty()) {
 424       scoped_ptr<PendingChannel> pending_channel(pending_channels_.front());
 425       pending_channels_.erase(pending_channels_.begin());
 426
 427       net::QuicP2PStream* stream = quic_session_->CreateOutgoingDynamicStream();
 428       scoped_ptr<QuicChannel> channel(new QuicClientChannel(
 429           stream, base::Bind(&Core::OnChannelDestroyed, base::Unretained(this),
 430                              stream->id()),
 431           pending_channel->name));
 432       pending_channel->callback.Run(channel.Pass());
 433     }
 434   }
 435 }
 436
 437 void QuicChannelFactory::Core::OnIncomingStream(net::QuicP2PStream* stream) {
 438   QuicServerChannel* channel = new QuicServerChannel(
 439       stream, base::Bind(&Core::OnChannelDestroyed, base::Unretained(this),
 440                          stream->id()));
 441   unnamed_incoming_channels_.push_back(channel);
 442   channel->ReceiveName(
 443       base::Bind(&Core::OnNameReceived, base::Unretained(this), channel));
 444 }
 445
 446 void QuicChannelFactory::Core::OnConnectionClosed(net::QuicErrorCode error) {
 447   if (error != net::QUIC_NO_ERROR)
 448     LOG(ERROR) << "QUIC connection was closed, error_code=" << error;
 449
 450   while (!pending_channels_.empty()) {
 451     scoped_ptr<PendingChannel> pending_channel(pending_channels_.front());
 452     pending_channels_.erase(pending_channels_.begin());
 453     pending_channel->callback.Run(nullptr);
 454   }
 455 }
 456
 457 void QuicChannelFactory::Core::OnNameReceived(QuicChannel* channel) {
 458   DCHECK(is_server_);
 459
 460   scoped_ptr<QuicChannel> owned_channel(channel);
 461
 462   auto it = std::find(unnamed_incoming_channels_.begin(),
 463                       unnamed_incoming_channels_.end(), channel);
 464   DCHECK(it != unnamed_incoming_channels_.end());
 465   unnamed_incoming_channels_.erase(it);
 466
 467   if (channel->name().empty()) {
 468     // Failed to read a name for incoming channel.
 469     return;
 470   }
 471
 472   for (auto it = pending_channels_.begin();
 473        it != pending_channels_.end(); ++it) {
 474     if ((*it)->name == channel->name()) {
 475       scoped_ptr<PendingChannel> pending_channel(*it);
 476       pending_channels_.erase(it);
 477       pending_channel->callback.Run(owned_channel.Pass());
 478       return;
 479     }
 480   }
 481
 482   LOG(ERROR) << "Unexpected incoming channel: " << channel->name();
 483 }
 484
 485 void QuicChannelFactory::Core::OnChannelDestroyed(int stream_id) {
 486   if (quic_session_)
 487     quic_session_->CloseStream(stream_id);
 488 }
 489
 490 QuicChannelFactory::QuicChannelFactory(const std::string& session_id,
 491                                        bool is_server)
 492     : core_(new Core(session_id, is_server)) {}
 493
 494 QuicChannelFactory::~QuicChannelFactory() {
 495   core_->Close();
 496   base::ThreadTaskRunnerHandle::Get()->DeleteSoon(FROM_HERE, core_.release());
 497 }
 498
 499 const std::string& QuicChannelFactory::CreateSessionInitiateConfigMessage() {
 500   return core_->CreateSessionInitiateConfigMessage();
 501 }
 502
 503 bool QuicChannelFactory::ProcessSessionAcceptConfigMessage(
 504     const std::string& message) {
 505   return core_->ProcessSessionAcceptConfigMessage(message);
 506 }
 507
 508 bool QuicChannelFactory::ProcessSessionInitiateConfigMessage(
 509     const std::string& message) {
 510   return core_->ProcessSessionInitiateConfigMessage(message);
 511 }
 512
 513 const std::string& QuicChannelFactory::CreateSessionAcceptConfigMessage() {
 514   return core_->CreateSessionAcceptConfigMessage();
 515 }
 516
 517 void QuicChannelFactory::Start(DatagramChannelFactory* factory,
 518                                const std::string& shared_secret) {
 519   core_->Start(factory, shared_secret);
 520 }
 521
 522 void QuicChannelFactory::CreateChannel(const std::string& name,
 523                                        const ChannelCreatedCallback& callback) {
 524   core_->CreateChannel(name, callback);
 525 }
 526
 527 void QuicChannelFactory::CancelChannelCreation(const std::string& name) {
 528   core_->CancelChannelCreation(name);
 529 }
 530
 531 net::QuicP2PSession* QuicChannelFactory::GetP2PSessionForTests() {
 532   return core_->quic_session_.get();
 533 }
 534
 535 }  // namespace protocol
 536 }  // namespace remoting