remoting/jingle_glue/xmpp_socket_adapter.cc

   1 // Copyright (c) 2010 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/jingle_glue/xmpp_socket_adapter.h"
   6
   7 #include <iomanip>
   8 #include <string>
   9
  10 #include "base/logging.h"
  11 #include "remoting/jingle_glue/ssl_adapter.h"
  12 #include "third_party/libjingle/source/talk/base/byteorder.h"
  13 #include "third_party/libjingle/source/talk/base/common.h"
  14 #include "third_party/libjingle/source/talk/base/firewallsocketserver.h"
  15 #include "third_party/libjingle/source/talk/base/logging.h"
  16 #include "third_party/libjingle/source/talk/base/socketadapters.h"
  17 #include "third_party/libjingle/source/talk/base/ssladapter.h"
  18 #include "third_party/libjingle/source/talk/base/thread.h"
  19 #include "third_party/libjingle/source/talk/xmpp/xmppengine.h"
  20
  21 namespace remoting {
  22
  23 XmppSocketAdapter::XmppSocketAdapter(const buzz::XmppClientSettings& xcs,
  24                                      bool allow_unverified_certs)
  25   : state_(STATE_CLOSED),
  26     error_(ERROR_NONE),
  27     wsa_error_(0),
  28     socket_(NULL),
  29     protocol_(xcs.protocol()),
  30     firewall_(false),
  31     write_buffer_(NULL),
  32     write_buffer_length_(0),
  33     write_buffer_capacity_(0),
  34     allow_unverified_certs_(allow_unverified_certs) {
  35   proxy_.type = xcs.proxy();
  36   proxy_.address.SetIP(xcs.proxy_host());
  37   proxy_.address.SetPort(xcs.proxy_port());
  38   proxy_.username = xcs.proxy_user();
  39   proxy_.password = xcs.proxy_pass();
  40 }
  41
  42 XmppSocketAdapter::~XmppSocketAdapter() {
  43   FreeState();
  44
  45   // Clean up any previous socket - cannot delete socket on close because close
  46   // happens during the child socket's stack callback.
  47   if (socket_) {
  48     delete socket_;
  49     socket_ = NULL;
  50   }
  51 }
  52
  53 buzz::AsyncSocket::State XmppSocketAdapter::state() {
  54   return state_;
  55 }
  56
  57 buzz::AsyncSocket::Error XmppSocketAdapter::error() {
  58   return error_;
  59 }
  60
  61 int XmppSocketAdapter::GetError() {
  62   return wsa_error_;
  63 }
  64
  65 bool XmppSocketAdapter::FreeState() {
  66   int code = 0;
  67
  68   // Clean up the socket.
  69   if (socket_ && !(state_ == STATE_CLOSED || state_ == STATE_CLOSING)) {
  70     code = socket_->Close();
  71   }
  72
  73   delete[] write_buffer_;
  74   write_buffer_ = NULL;
  75   write_buffer_length_ = 0;
  76   write_buffer_capacity_ = 0;
  77
  78   if (code) {
  79     SetWSAError(code);
  80     return false;
  81   }
  82   return true;
  83 }
  84
  85 bool XmppSocketAdapter::Connect(const talk_base::SocketAddress& addr) {
  86   if (state_ != STATE_CLOSED) {
  87     SetError(ERROR_WRONGSTATE);
  88     return false;
  89   }
  90
  91   VLOG(1) << "XmppSocketAdapter::Connect(" << addr.ToString() << ")";
  92
  93   // Clean up any previous socket - cannot delete socket on close because close
  94   // happens during the child socket's stack callback.
  95   if (socket_) {
  96     delete socket_;
  97     socket_ = NULL;
  98   }
  99
 100   talk_base::AsyncSocket* socket =
 101       talk_base::Thread::Current()->socketserver()->CreateAsyncSocket(
 102           SOCK_STREAM);
 103   if (!socket) {
 104     SetWSAError(WSA_NOT_ENOUGH_MEMORY);
 105     return false;
 106   }
 107
 108   if (firewall_) {
 109     // TODO(sync): Change this to make WSAAsyncSockets support current thread
 110     // socket server.
 111     talk_base::FirewallSocketServer* fw =
 112         static_cast<talk_base::FirewallSocketServer*>(
 113             talk_base::Thread::Current()->socketserver());
 114     socket = fw->WrapSocket(socket, SOCK_STREAM);
 115   }
 116
 117   if (proxy_.type) {
 118     talk_base::AsyncSocket* proxy_socket = 0;
 119     if (proxy_.type == talk_base::PROXY_SOCKS5) {
 120       proxy_socket = new talk_base::AsyncSocksProxySocket(
 121           socket, proxy_.address, proxy_.username, proxy_.password);
 122     } else {
 123       // Note: we are trying unknown proxies as HTTPS currently.
 124       proxy_socket = new talk_base::AsyncHttpsProxySocket(socket,
 125           "chromoting", proxy_.address, proxy_.username,
 126           proxy_.password);
 127     }
 128     if (!proxy_socket) {
 129       SetWSAError(WSA_NOT_ENOUGH_MEMORY);
 130       delete socket;
 131       return false;
 132     }
 133     socket = proxy_socket;  // For our purposes the proxy is now the socket.
 134   }
 135
 136   if (protocol_ == cricket::PROTO_SSLTCP) {
 137     talk_base::AsyncSocket *fake_ssl_socket =
 138         new talk_base::AsyncSSLSocket(socket);
 139     if (!fake_ssl_socket) {
 140       SetWSAError(WSA_NOT_ENOUGH_MEMORY);
 141       delete socket;
 142       return false;
 143     }
 144     socket = fake_ssl_socket;  // For our purposes the SSL socket is the socket.
 145   }
 146
 147 #if defined(FEATURE_ENABLE_SSL)
 148   talk_base::SSLAdapter* ssl_adapter = remoting::CreateSSLAdapter(socket);
 149   socket = ssl_adapter;  // For our purposes the SSL adapter is the socket.
 150 #endif
 151
 152   socket->SignalReadEvent.connect(this, &XmppSocketAdapter::OnReadEvent);
 153   socket->SignalWriteEvent.connect(this, &XmppSocketAdapter::OnWriteEvent);
 154   socket->SignalConnectEvent.connect(this, &XmppSocketAdapter::OnConnectEvent);
 155   socket->SignalCloseEvent.connect(this, &XmppSocketAdapter::OnCloseEvent);
 156
 157   // The linux implementation of socket::Connect returns an error when the
 158   // connect didn't complete yet.  This can be distinguished from a failure
 159   // because socket::IsBlocking is true.  Perhaps, the linux implementation
 160   // should be made to behave like the windows version which doesn't do this,
 161   // but it seems to be a pattern with these methods that they return an error
 162   // if the operation didn't complete in a sync fashion and one has to check
 163   // IsBlocking to tell if was a "real" error.
 164   if (socket->Connect(addr) == SOCKET_ERROR && !socket->IsBlocking()) {
 165     SetWSAError(socket->GetError());
 166     delete socket;
 167     return false;
 168   }
 169
 170   socket_ = socket;
 171   state_ = STATE_CONNECTING;
 172   return true;
 173 }
 174
 175 bool XmppSocketAdapter::Read(char* data, size_t len, size_t* len_read) {
 176   if (len_read)
 177     *len_read = 0;
 178
 179   if (state_ <= STATE_CLOSING) {
 180     SetError(ERROR_WRONGSTATE);
 181     return false;
 182   }
 183
 184   DCHECK(socket_);
 185
 186   if (IsOpen()) {
 187     int result = socket_->Recv(data, len);
 188     if (result < 0) {
 189       if (!socket_->IsBlocking()) {
 190         SetWSAError(socket_->GetError());
 191         return false;
 192       }
 193
 194       result = 0;
 195     }
 196
 197     if (len_read)
 198       *len_read = result;
 199   }
 200
 201   return true;
 202 }
 203
 204 bool XmppSocketAdapter::Write(const char* data, size_t len) {
 205   if (state_ <= STATE_CLOSING) {
 206     // There may be data in a buffer that gets lost.  Too bad!
 207     SetError(ERROR_WRONGSTATE);
 208     return false;
 209   }
 210
 211   DCHECK(socket_);
 212
 213   size_t sent = 0;
 214
 215   // Try an immediate write when there is no buffer and we aren't in SSL mode
 216   // or opening the connection.
 217   if (write_buffer_length_ == 0 && IsOpen()) {
 218     int result = socket_->Send(data, len);
 219     if (result < 0) {
 220       if (!socket_->IsBlocking()) {
 221         SetWSAError(socket_->GetError());
 222         return false;
 223       }
 224       result = 0;
 225     }
 226
 227     sent = static_cast<size_t>(result);
 228   }
 229
 230   // Buffer what we didn't send.
 231   if (sent < len) {
 232     QueueWriteData(data + sent, len - sent);
 233   }
 234
 235   // Service the socket right away to push the written data out in SSL mode.
 236   return HandleWritable();
 237 }
 238
 239 bool XmppSocketAdapter::Close() {
 240   if (state_ == STATE_CLOSING) {
 241     return false;  // Avoid recursion, but not unexpected.
 242   }
 243   if (state_ == STATE_CLOSED) {
 244     // In theory should not be trying to re-InternalClose.
 245     SetError(ERROR_WRONGSTATE);
 246     return false;
 247   }
 248
 249   // TODO(sync): deal with flushing close (flush, don't do reads, clean ssl).
 250
 251   // If we've gotten to the point where we really do have a socket underneath
 252   // then close it.  It should call us back to tell us it is closed, and
 253   // NotifyClose will be called.  We indicate "closing" state so that we
 254   // do not recusively try to keep closing the socket.
 255   if (socket_) {
 256     state_ = STATE_CLOSING;
 257     socket_->Close();
 258   }
 259
 260   // If we didn't get the callback, then we better make sure we signal
 261   // closed.
 262   if (state_ != STATE_CLOSED) {
 263     // The socket was closed manually, not directly due to error.
 264     if (error_ != ERROR_NONE) {
 265       VLOG(1) << "XmppSocketAdapter::Close - previous Error: " << error_
 266               << " WSAError: " << wsa_error_;
 267       error_ = ERROR_NONE;
 268       wsa_error_ = 0;
 269     }
 270     NotifyClose();
 271   }
 272   return true;
 273 }
 274
 275 void XmppSocketAdapter::NotifyClose() {
 276   if (state_ == STATE_CLOSED) {
 277     SetError(ERROR_WRONGSTATE);
 278   } else {
 279     VLOG(1) << "XmppSocketAdapter::NotifyClose - Error: " << error_
 280             << " WSAError: " << wsa_error_;
 281     state_ = STATE_CLOSED;
 282     SignalClosed();
 283     FreeState();
 284   }
 285 }
 286
 287 void XmppSocketAdapter::OnConnectEvent(talk_base::AsyncSocket *socket) {
 288   if (state_ == STATE_CONNECTING) {
 289     state_ = STATE_OPEN;
 290     VLOG(1) << "XmppSocketAdapter::OnConnectEvent - STATE_OPEN";
 291     SignalConnected();
 292 #if defined(FEATURE_ENABLE_SSL)
 293   } else if (state_ == STATE_TLS_CONNECTING) {
 294     state_ = STATE_TLS_OPEN;
 295     VLOG(1) << "XmppSocketAdapter::OnConnectEvent - STATE_TLS_OPEN";
 296     SignalSSLConnected();
 297     if (write_buffer_length_ > 0) {
 298       HandleWritable();
 299     }
 300 #endif  // defined(FEATURE_ENABLE_SSL)
 301   } else {
 302     LOG(DFATAL) << "unexpected XmppSocketAdapter::OnConnectEvent state: "
 303                 << state_;
 304   }
 305 }
 306
 307 void XmppSocketAdapter::OnReadEvent(talk_base::AsyncSocket *socket) {
 308   HandleReadable();
 309 }
 310
 311 void XmppSocketAdapter::OnWriteEvent(talk_base::AsyncSocket *socket) {
 312   HandleWritable();
 313 }
 314
 315 void XmppSocketAdapter::OnCloseEvent(talk_base::AsyncSocket *socket,
 316                                      int error) {
 317   VLOG(1) << "XmppSocketAdapter::OnCloseEvent(" << error << ")";
 318   SetWSAError(error);
 319   if (error == SOCKET_EACCES) {
 320     SignalAuthenticationError();  // Proxy needs authentication.
 321   }
 322   NotifyClose();
 323 }
 324
 325 #if defined(FEATURE_ENABLE_SSL)
 326 bool XmppSocketAdapter::StartTls(const std::string& verify_host_name) {
 327   if (state_ != STATE_OPEN) {
 328     SetError(ERROR_WRONGSTATE);
 329     return false;
 330   }
 331
 332   state_ = STATE_TLS_CONNECTING;
 333
 334   DCHECK_EQ(write_buffer_length_, 0U);
 335
 336   talk_base::SSLAdapter* ssl_adapter =
 337       static_cast<talk_base::SSLAdapter*>(socket_);
 338
 339   if (allow_unverified_certs_) {
 340     ssl_adapter->set_ignore_bad_cert(true);
 341   }
 342
 343   if (ssl_adapter->StartSSL(verify_host_name.c_str(), false) != 0) {
 344     state_ = STATE_OPEN;
 345     SetError(ERROR_SSL);
 346     return false;
 347   }
 348
 349   return true;
 350 }
 351 #endif  // defined(FEATURE_ENABLE_SSL)
 352
 353 void XmppSocketAdapter::QueueWriteData(const char* data, size_t len) {
 354   // Expand buffer if needed.
 355   if (write_buffer_length_ + len > write_buffer_capacity_) {
 356     size_t new_capacity = 1024;
 357     while (new_capacity < write_buffer_length_ + len) {
 358       new_capacity = new_capacity * 2;
 359     }
 360     char* new_buffer = new char[new_capacity];
 361     DCHECK_LE(write_buffer_length_, 64000U);
 362     memcpy(new_buffer, write_buffer_, write_buffer_length_);
 363     delete[] write_buffer_;
 364     write_buffer_ = new_buffer;
 365     write_buffer_capacity_ = new_capacity;
 366   }
 367
 368   // Copy data into the end of buffer.
 369   memcpy(write_buffer_ + write_buffer_length_, data, len);
 370   write_buffer_length_ += len;
 371 }
 372
 373 void XmppSocketAdapter::FlushWriteQueue(Error* error, int* wsa_error) {
 374   DCHECK(error);
 375   DCHECK(wsa_error);
 376
 377   size_t flushed = 0;
 378   while (flushed < write_buffer_length_) {
 379     int sent = socket_->Send(write_buffer_ + flushed,
 380                              static_cast<int>(write_buffer_length_ - flushed));
 381     if (sent < 0) {
 382       if (!socket_->IsBlocking()) {
 383         *error = ERROR_WINSOCK;
 384         *wsa_error = socket_->GetError();
 385       }
 386       break;
 387     }
 388     flushed += static_cast<size_t>(sent);
 389   }
 390
 391   // Remove flushed memory.
 392   write_buffer_length_ -= flushed;
 393   memmove(write_buffer_, write_buffer_ + flushed, write_buffer_length_);
 394
 395   // When everything is flushed, deallocate the buffer if it's gotten big.
 396   if (write_buffer_length_ == 0) {
 397     if (write_buffer_capacity_ > 8192) {
 398       delete[] write_buffer_;
 399       write_buffer_ = NULL;
 400       write_buffer_capacity_ = 0;
 401     }
 402   }
 403 }
 404
 405 void XmppSocketAdapter::SetError(Error error) {
 406   if (error_ == ERROR_NONE) {
 407     error_ = error;
 408   }
 409 }
 410
 411 void XmppSocketAdapter::SetWSAError(int error) {
 412   if (error_ == ERROR_NONE && error != 0) {
 413     error_ = ERROR_WINSOCK;
 414     wsa_error_ = error;
 415   }
 416 }
 417
 418 bool XmppSocketAdapter::HandleReadable() {
 419   if (!IsOpen())
 420     return false;
 421
 422   SignalRead();
 423   return true;
 424 }
 425
 426 bool XmppSocketAdapter::HandleWritable() {
 427   if (!IsOpen())
 428     return false;
 429
 430   Error error = ERROR_NONE;
 431   int wsa_error = 0;
 432   FlushWriteQueue(&error, &wsa_error);
 433   if (error != ERROR_NONE) {
 434     Close();
 435     return false;
 436   }
 437   return true;
 438 }
 439
 440 }  // namespace remoting