net/quic/p2p/quic_p2p_session_test.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 "net/quic/p2p/quic_p2p_session.h"
   6
   7 #include "base/callback_helpers.h"
   8 #include "base/location.h"
   9 #include "base/memory/weak_ptr.h"
  10 #include "base/run_loop.h"
  11 #include "base/single_thread_task_runner.h"
  12 #include "base/thread_task_runner_handle.h"
  13 #include "net/base/io_buffer.h"
  14 #include "net/base/net_errors.h"
  15 #include "net/quic/crypto/quic_random.h"
  16 #include "net/quic/p2p/quic_p2p_crypto_config.h"
  17 #include "net/quic/p2p/quic_p2p_stream.h"
  18 #include "net/quic/quic_connection_helper.h"
  19 #include "net/quic/quic_default_packet_writer.h"
  20 #include "net/socket/socket.h"
  21 #include "testing/gtest/include/gtest/gtest.h"
  22
  23 namespace net {
  24
  25 namespace {
  26
  27 const char kTestSharedKey[] = "Shared key exchanged out of bound.";
  28
  29 class FakeP2PDatagramSocket : public Socket {
  30  public:
  31   FakeP2PDatagramSocket() : weak_factory_(this) {}
  32   ~FakeP2PDatagramSocket() override {}
  33
  34   void ConnectWith(FakeP2PDatagramSocket* peer_socket) {
  35     peer_socket_ = peer_socket->weak_factory_.GetWeakPtr();
  36     peer_socket->peer_socket_ = weak_factory_.GetWeakPtr();
  37   }
  38
  39   void SetReadError(int error) {
  40     read_error_ = error;
  41     if (!read_callback_.is_null()) {
  42       base::ResetAndReturn(&read_callback_).Run(error);
  43     }
  44   }
  45
  46   void SetWriteError(int error) { write_error_ = error; }
  47
  48   void AppendInputPacket(const std::vector<char>& data) {
  49     if (!read_callback_.is_null()) {
  50       int size = std::min(read_buffer_size_, static_cast<int>(data.size()));
  51       memcpy(read_buffer_->data(), &data[0], data.size());
  52       read_buffer_ = nullptr;
  53       base::ResetAndReturn(&read_callback_).Run(size);
  54     } else {
  55       incoming_packets_.push_back(data);
  56     }
  57   }
  58
  59   // Socket interface.
  60   int Read(IOBuffer* buf,
  61            int buf_len,
  62            const CompletionCallback& callback) override {
  63     DCHECK(read_callback_.is_null());
  64
  65     if (read_error_ != OK) {
  66       return read_error_;
  67     }
  68
  69     if (!incoming_packets_.empty()) {
  70       scoped_refptr<IOBuffer> buffer(buf);
  71       int size =
  72           std::min(static_cast<int>(incoming_packets_.front().size()), buf_len);
  73       memcpy(buffer->data(), &*incoming_packets_.front().begin(), size);
  74       incoming_packets_.pop_front();
  75       return size;
  76     } else {
  77       read_callback_ = callback;
  78       read_buffer_ = buf;
  79       read_buffer_size_ = buf_len;
  80       return ERR_IO_PENDING;
  81     }
  82   }
  83
  84   int Write(IOBuffer* buf,
  85             int buf_len,
  86             const CompletionCallback& callback) override {
  87     if (write_error_ != OK) {
  88       return write_error_;
  89     }
  90
  91     if (peer_socket_) {
  92       base::ThreadTaskRunnerHandle::Get()->PostTask(
  93           FROM_HERE,
  94           base::Bind(&FakeP2PDatagramSocket::AppendInputPacket, peer_socket_,
  95                      std::vector<char>(buf->data(), buf->data() + buf_len)));
  96     }
  97
  98     return buf_len;
  99   }
 100
 101   int SetReceiveBufferSize(int32 size) override {
 102     NOTIMPLEMENTED();
 103     return ERR_NOT_IMPLEMENTED;
 104   }
 105   int SetSendBufferSize(int32 size) override {
 106     NOTIMPLEMENTED();
 107     return ERR_NOT_IMPLEMENTED;
 108   }
 109
 110  private:
 111   int read_error_ = OK;
 112   int write_error_ = OK;
 113
 114   scoped_refptr<IOBuffer> read_buffer_;
 115   int read_buffer_size_;
 116   CompletionCallback read_callback_;
 117
 118   std::deque<std::vector<char>> incoming_packets_;
 119
 120   base::WeakPtr<FakeP2PDatagramSocket> peer_socket_;
 121
 122   base::WeakPtrFactory<FakeP2PDatagramSocket> weak_factory_;
 123 };
 124
 125 class DefaultPacketWriterFactory : public QuicConnection::PacketWriterFactory {
 126  public:
 127   explicit DefaultPacketWriterFactory(Socket* socket) : socket_(socket) {}
 128   ~DefaultPacketWriterFactory() override {}
 129
 130   QuicPacketWriter* Create(QuicConnection* connection) const override {
 131     scoped_ptr<net::QuicDefaultPacketWriter> writer(
 132         new net::QuicDefaultPacketWriter(socket_));
 133     writer->SetConnection(connection);
 134     return writer.release();
 135   }
 136
 137  private:
 138   Socket* socket_;
 139 };
 140
 141 class TestP2PStreamDelegate : public QuicP2PStream::Delegate {
 142  public:
 143   TestP2PStreamDelegate() {}
 144   ~TestP2PStreamDelegate() override {}
 145
 146   void OnDataReceived(const char* data, int length) override {
 147     received_data_.append(data, length);
 148   }
 149   void OnClose(QuicErrorCode error) override {
 150     is_closed_ = true;
 151     error_ = error;
 152   }
 153
 154   const std::string& received_data() { return received_data_; }
 155   bool is_closed() { return is_closed_; }
 156   QuicErrorCode error() { return error_; }
 157
 158  private:
 159   std::string received_data_;
 160   bool is_closed_ = false;
 161   QuicErrorCode error_ = QUIC_NO_ERROR;
 162
 163   DISALLOW_COPY_AND_ASSIGN(TestP2PStreamDelegate);
 164 };
 165
 166 class TestP2PSessionDelegate : public QuicP2PSession::Delegate {
 167  public:
 168   void OnIncomingStream(QuicP2PStream* stream) override {
 169     last_incoming_stream_ = stream;
 170     stream->SetDelegate(next_incoming_stream_delegate_);
 171     next_incoming_stream_delegate_ = nullptr;
 172     if (!on_incoming_stream_callback_.is_null())
 173       base::ResetAndReturn(&on_incoming_stream_callback_).Run();
 174   }
 175
 176   void OnConnectionClosed(QuicErrorCode error) override {
 177     is_closed_ = true;
 178     error_ = error;
 179   }
 180
 181   void set_next_incoming_stream_delegate(QuicP2PStream::Delegate* delegate) {
 182     next_incoming_stream_delegate_ = delegate;
 183   }
 184   void set_on_incoming_stream_callback(const base::Closure& callback) {
 185     on_incoming_stream_callback_ = callback;
 186   }
 187   QuicP2PStream* last_incoming_stream() { return last_incoming_stream_; }
 188   bool is_closed() { return is_closed_; }
 189   QuicErrorCode error() { return error_; }
 190
 191  private:
 192   QuicP2PStream::Delegate* next_incoming_stream_delegate_ = nullptr;
 193   base::Closure on_incoming_stream_callback_;
 194   QuicP2PStream* last_incoming_stream_ = nullptr;
 195   bool is_closed_ = false;
 196   QuicErrorCode error_ = QUIC_NO_ERROR;
 197 };
 198
 199 }  // namespace
 200
 201 class QuicP2PSessionTest : public ::testing::Test {
 202  public:
 203   void OnWriteResult(int result);
 204
 205  protected:
 206   QuicP2PSessionTest()
 207       : quic_helper_(base::ThreadTaskRunnerHandle::Get().get(),
 208                      &quic_clock_,
 209                      net::QuicRandom::GetInstance()) {
 210     // Simulate out-of-bound config handshake.
 211     CryptoHandshakeMessage hello_message;
 212     config_.ToHandshakeMessage(&hello_message);
 213     std::string error_detail;
 214     EXPECT_EQ(QUIC_NO_ERROR,
 215               config_.ProcessPeerHello(hello_message, CLIENT, &error_detail));
 216   }
 217
 218   void CreateSessions() {
 219     socket1_ = new FakeP2PDatagramSocket();
 220     socket2_ = new FakeP2PDatagramSocket();
 221     socket1_->ConnectWith(socket2_);
 222
 223     QuicP2PCryptoConfig crypto_config(kTestSharedKey);
 224
 225     session1_ = CreateP2PSession(make_scoped_ptr(socket1_), crypto_config,
 226                                  Perspective::IS_SERVER);
 227     session2_ = CreateP2PSession(make_scoped_ptr(socket2_), crypto_config,
 228                                  Perspective::IS_CLIENT);
 229   }
 230
 231   scoped_ptr<QuicP2PSession> CreateP2PSession(scoped_ptr<Socket> socket,
 232                                               QuicP2PCryptoConfig crypto_config,
 233                                               Perspective perspective) {
 234     DefaultPacketWriterFactory writer_factory(socket.get());
 235     net::IPAddressNumber ip(net::kIPv4AddressSize, 0);
 236     scoped_ptr<QuicConnection> quic_connection1(
 237         new QuicConnection(0, net::IPEndPoint(ip, 0), &quic_helper_,
 238                            writer_factory, true /* owns_writer */, perspective,
 239                            true /* is_secuire */, QuicSupportedVersions()));
 240
 241     scoped_ptr<QuicP2PSession> result(new QuicP2PSession(
 242         config_, crypto_config, quic_connection1.Pass(), socket.Pass()));
 243     result->Initialize();
 244     return result.Pass();
 245   }
 246
 247   void TestStreamConnection(QuicP2PSession* from_session,
 248                             QuicP2PSession* to_session,
 249                             QuicStreamId expected_stream_id);
 250
 251   QuicClock quic_clock_;
 252   QuicConnectionHelper quic_helper_;
 253   QuicConfig config_;
 254
 255   FakeP2PDatagramSocket* socket1_;
 256   scoped_ptr<QuicP2PSession> session1_;
 257
 258   FakeP2PDatagramSocket* socket2_;
 259   scoped_ptr<QuicP2PSession> session2_;
 260 };
 261
 262 void QuicP2PSessionTest::OnWriteResult(int result) {
 263   EXPECT_EQ(OK, result);
 264 }
 265
 266 void QuicP2PSessionTest::TestStreamConnection(QuicP2PSession* from_session,
 267                                               QuicP2PSession* to_session,
 268                                               QuicStreamId expected_stream_id) {
 269   QuicP2PStream* outgoing_stream = from_session->CreateOutgoingDynamicStream();
 270   EXPECT_TRUE(outgoing_stream);
 271   TestP2PStreamDelegate outgoing_stream_delegate;
 272   outgoing_stream->SetDelegate(&outgoing_stream_delegate);
 273   EXPECT_EQ(expected_stream_id, outgoing_stream->id());
 274
 275   // Send a test message to the client.
 276   const char kTestMessage[] = "Hi";
 277   const char kTestResponse[] = "Response";
 278   outgoing_stream->Write(
 279       std::string(kTestMessage),
 280       base::Bind(&QuicP2PSessionTest::OnWriteResult, base::Unretained(this)));
 281
 282   // Wait for the incoming stream to be created.
 283   TestP2PStreamDelegate incoming_stream_delegate;
 284   base::RunLoop run_loop;
 285   TestP2PSessionDelegate session_delegate;
 286   session_delegate.set_next_incoming_stream_delegate(&incoming_stream_delegate);
 287   session_delegate.set_on_incoming_stream_callback(
 288       base::Bind(&base::RunLoop::Quit, base::Unretained(&run_loop)));
 289   to_session->SetDelegate(&session_delegate);
 290   run_loop.Run();
 291   to_session->SetDelegate(nullptr);
 292
 293   QuicP2PStream* incoming_stream = session_delegate.last_incoming_stream();
 294   ASSERT_TRUE(incoming_stream);
 295   EXPECT_EQ(expected_stream_id, incoming_stream->id());
 296   EXPECT_EQ(kTestMessage, incoming_stream_delegate.received_data());
 297
 298   incoming_stream->Write(
 299       std::string(kTestResponse),
 300       base::Bind(&QuicP2PSessionTest::OnWriteResult, base::Unretained(this)));
 301   base::RunLoop().RunUntilIdle();
 302   EXPECT_EQ(kTestResponse, outgoing_stream_delegate.received_data());
 303
 304   from_session->CloseStream(outgoing_stream->id());
 305   base::RunLoop().RunUntilIdle();
 306
 307   EXPECT_TRUE(outgoing_stream_delegate.is_closed());
 308   EXPECT_TRUE(incoming_stream_delegate.is_closed());
 309 }
 310
 311 TEST_F(QuicP2PSessionTest, ClientToServer) {
 312   CreateSessions();
 313   TestStreamConnection(session2_.get(), session1_.get(), 3);
 314 }
 315
 316 TEST_F(QuicP2PSessionTest, ServerToClient) {
 317   CreateSessions();
 318   TestStreamConnection(session1_.get(), session2_.get(), 2);
 319 }
 320
 321 TEST_F(QuicP2PSessionTest, TransportWriteError) {
 322   CreateSessions();
 323
 324   TestP2PSessionDelegate session_delegate;
 325   session1_->SetDelegate(&session_delegate);
 326
 327   QuicP2PStream* stream = session1_->CreateOutgoingDynamicStream();
 328   EXPECT_TRUE(stream);
 329   TestP2PStreamDelegate stream_delegate;
 330   stream->SetDelegate(&stream_delegate);
 331   EXPECT_EQ(2U, stream->id());
 332
 333   socket1_->SetWriteError(ERR_INTERNET_DISCONNECTED);
 334
 335   const char kTestMessage[] = "Hi";
 336   stream->Write(
 337       std::string(kTestMessage),
 338       base::Bind(&QuicP2PSessionTest::OnWriteResult, base::Unretained(this)));
 339
 340   base::RunLoop().RunUntilIdle();
 341
 342   EXPECT_TRUE(stream_delegate.is_closed());
 343   EXPECT_EQ(QUIC_PACKET_WRITE_ERROR, stream_delegate.error());
 344   EXPECT_TRUE(session_delegate.is_closed());
 345   EXPECT_EQ(QUIC_PACKET_WRITE_ERROR, session_delegate.error());
 346 }
 347
 348 TEST_F(QuicP2PSessionTest, TransportReceiveError) {
 349   CreateSessions();
 350
 351   TestP2PSessionDelegate session_delegate;
 352   session1_->SetDelegate(&session_delegate);
 353
 354   QuicP2PStream* stream = session1_->CreateOutgoingDynamicStream();
 355   EXPECT_TRUE(stream);
 356   TestP2PStreamDelegate stream_delegate;
 357   stream->SetDelegate(&stream_delegate);
 358   EXPECT_EQ(2U, stream->id());
 359
 360   socket1_->SetReadError(ERR_INTERNET_DISCONNECTED);
 361
 362   base::RunLoop().RunUntilIdle();
 363
 364   EXPECT_TRUE(stream_delegate.is_closed());
 365   EXPECT_EQ(QUIC_PACKET_READ_ERROR, stream_delegate.error());
 366   EXPECT_TRUE(session_delegate.is_closed());
 367   EXPECT_EQ(QUIC_PACKET_READ_ERROR, session_delegate.error());
 368 }
 369
 370 }  // namespace net