src/common/protocols/datastream/datastreampeer.cpp

   1 /***************************************************************************
   2  *   Copyright (C) 2005-2014 by the Quassel Project                        *
   3  *   devel@quassel-irc.org                                                 *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) version 3.                                           *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.         *
  19  ***************************************************************************/
  20
  21 #include <QHostAddress>
  22 #include <QTcpSocket>
  23
  24 #include "datastreampeer.h"
  25 #include "quassel.h"
  26
  27 /* version.inc is no longer used for this */
  28 const uint protocolVersion = 10;
  29 const uint coreNeedsProtocol = protocolVersion;
  30 const uint clientNeedsProtocol = protocolVersion;
  31
  32 using namespace Protocol;
  33
  34 DataStreamPeer::DataStreamPeer(::AuthHandler *authHandler, QTcpSocket *socket, quint16 features, QObject *parent)
  35     : RemotePeer(authHandler, socket, parent),
  36     _blockSize(0),
  37     _useCompression(false)
  38 {
  39     Q_UNUSED(features);
  40
  41     _stream.setDevice(socket);
  42     _stream.setVersion(QDataStream::Qt_4_2);
  43 }
  44
  45
  46 void DataStreamPeer::setSignalProxy(::SignalProxy *proxy)
  47 {
  48     RemotePeer::setSignalProxy(proxy);
  49
  50     // FIXME only in compat mode
  51     if (proxy) {
  52         // enable compression now if requested - the initial handshake is uncompressed in the legacy protocol!
  53         _useCompression = socket()->property("UseCompression").toBool();
  54         if (_useCompression)
  55             qDebug() << "Using compression for peer:" << qPrintable(socket()->peerAddress().toString());
  56     }
  57
  58 }
  59
  60
  61 quint16 DataStreamPeer::supportedFeatures()
  62 {
  63     return 0;
  64 }
  65
  66
  67 bool DataStreamPeer::acceptsFeatures(quint16 peerFeatures)
  68 {
  69     Q_UNUSED(peerFeatures);
  70     return true;
  71 }
  72
  73
  74 quint16 DataStreamPeer::enabledFeatures() const
  75 {
  76     return 0;
  77 }
  78
  79
  80 void DataStreamPeer::onSocketDataAvailable()
  81 {
  82     QVariant item;
  83     while (readSocketData(item)) {
  84         // if no sigproxy is set, we're in handshake mode and let the data be handled elsewhere
  85         if (!signalProxy())
  86             handleHandshakeMessage(item);
  87         else
  88             handlePackedFunc(item);
  89     }
  90 }
  91
  92
  93 bool DataStreamPeer::readSocketData(QVariant &item)
  94 {
  95     if (_blockSize == 0) {
  96         if (socket()->bytesAvailable() < 4)
  97             return false;
  98         _stream >> _blockSize;
  99     }
 100
 101     if (_blockSize > 1 << 22) {
 102         close("Peer tried to send package larger than max package size!");
 103         return false;
 104     }
 105
 106     if (_blockSize == 0) {
 107         close("Peer tried to send 0 byte package!");
 108         return false;
 109     }
 110
 111     if (socket()->bytesAvailable() < _blockSize) {
 112         emit transferProgress(socket()->bytesAvailable(), _blockSize);
 113         return false;
 114     }
 115
 116     emit transferProgress(_blockSize, _blockSize);
 117
 118     _blockSize = 0;
 119
 120     if (_useCompression) {
 121         QByteArray rawItem;
 122         _stream >> rawItem;
 123
 124         int nbytes = rawItem.size();
 125         if (nbytes <= 4) {
 126             const char *data = rawItem.constData();
 127             if (nbytes < 4 || (data[0] != 0 || data[1] != 0 || data[2] != 0 || data[3] != 0)) {
 128                 close("Peer sent corrupted compressed data!");
 129                 return false;
 130             }
 131         }
 132
 133         rawItem = qUncompress(rawItem);
 134
 135         QDataStream itemStream(&rawItem, QIODevice::ReadOnly);
 136         itemStream.setVersion(QDataStream::Qt_4_2);
 137         itemStream >> item;
 138     }
 139     else {
 140         _stream >> item;
 141     }
 142
 143     if (!item.isValid()) {
 144         close("Peer sent corrupt data: unable to load QVariant!");
 145         return false;
 146     }
 147
 148     return true;
 149 }
 150
 151
 152 void DataStreamPeer::writeSocketData(const QVariant &item)
 153 {
 154     if (!socket()->isOpen()) {
 155         qWarning() << Q_FUNC_INFO << "Can't write to a closed socket!";
 156         return;
 157     }
 158
 159     QByteArray block;
 160     QDataStream out(&block, QIODevice::WriteOnly);
 161     out.setVersion(QDataStream::Qt_4_2);
 162
 163     if (_useCompression) {
 164         QByteArray rawItem;
 165         QDataStream itemStream(&rawItem, QIODevice::WriteOnly);
 166         itemStream.setVersion(QDataStream::Qt_4_2);
 167         itemStream << item;
 168
 169         rawItem = qCompress(rawItem);
 170
 171         out << rawItem;
 172     }
 173     else {
 174         out << item;
 175     }
 176
 177     _stream << block;  // also writes the length as part of the serialization format
 178 }
 179
 180
 181 /*** Handshake messages ***/
 182
 183 /* These messages are transmitted during handshake phase, which in case of the legacy protocol means they have
 184  * a structure different from those being used after the handshake.
 185  * Also, the legacy handshake does not fully match the redesigned one, so we'll have to do various mappings here.
 186  */
 187
 188 void DataStreamPeer::handleHandshakeMessage(const QVariant &msg)
 189 {
 190     QVariantMap m = msg.toMap();
 191
 192     QString msgType = m["MsgType"].toString();
 193     if (msgType.isEmpty()) {
 194         emit protocolError(tr("Invalid handshake message!"));
 195         return;
 196     }
 197
 198     if (msgType == "ClientInit") {
 199         // FIXME only in compat mode
 200         uint ver = m["ProtocolVersion"].toUInt();
 201         if (ver < coreNeedsProtocol) {
 202             emit protocolVersionMismatch((int)ver, (int)coreNeedsProtocol);
 203             return;
 204         }
 205
 206 #ifndef QT_NO_COMPRESS
 207         // FIXME only in compat mode
 208         if (m["UseCompression"].toBool()) {
 209             socket()->setProperty("UseCompression", true);
 210         }
 211 #endif
 212         handle(RegisterClient(m["ClientVersion"].toString(), m["UseSsl"].toBool()));
 213     }
 214
 215     else if (msgType == "ClientInitReject") {
 216         handle(ClientDenied(m["Error"].toString()));
 217     }
 218
 219     else if (msgType == "ClientInitAck") {
 220         // FIXME only in compat mode
 221         uint ver = m["ProtocolVersion"].toUInt(); // actually an UInt
 222         if (ver < clientNeedsProtocol) {
 223             emit protocolVersionMismatch((int)ver, (int)clientNeedsProtocol);
 224             return;
 225         }
 226 #ifndef QT_NO_COMPRESS
 227         if (m["SupportsCompression"].toBool())
 228             socket()->setProperty("UseCompression", true);
 229 #endif
 230
 231         handle(ClientRegistered(m["CoreFeatures"].toUInt(), m["Configured"].toBool(), m["StorageBackends"].toList(), m["SupportSsl"].toBool(), QDateTime()));
 232     }
 233
 234     else if (msgType == "CoreSetupData") {
 235         QVariantMap map = m["SetupData"].toMap();
 236         handle(SetupData(map["AdminUser"].toString(), map["AdminPasswd"].toString(), map["Backend"].toString(), map["ConnectionProperties"].toMap()));
 237     }
 238
 239     else if (msgType == "CoreSetupReject") {
 240         handle(SetupFailed(m["Error"].toString()));
 241     }
 242
 243     else if (msgType == "CoreSetupAck") {
 244         handle(SetupDone());
 245     }
 246
 247     else if (msgType == "ClientLogin") {
 248         handle(Login(m["User"].toString(), m["Password"].toString()));
 249     }
 250
 251     else if (msgType == "ClientLoginReject") {
 252         handle(LoginFailed(m["Error"].toString()));
 253     }
 254
 255     else if (msgType == "ClientLoginAck") {
 256         handle(LoginSuccess());
 257     }
 258
 259     else if (msgType == "SessionInit") {
 260         QVariantMap map = m["SessionState"].toMap();
 261         handle(SessionState(map["Identities"].toList(), map["BufferInfos"].toList(), map["NetworkIds"].toList()));
 262     }
 263
 264     else {
 265         emit protocolError(tr("Unknown protocol message of type %1").arg(msgType));
 266     }
 267 }
 268
 269
 270 void DataStreamPeer::dispatch(const RegisterClient &msg) {
 271     QVariantMap m;
 272     m["MsgType"] = "ClientInit";
 273     m["ClientVersion"] = msg.clientVersion;
 274     m["ClientDate"] = Quassel::buildInfo().buildDate;
 275
 276     // FIXME only in compat mode
 277     m["ProtocolVersion"] = protocolVersion;
 278     m["UseSsl"] = msg.sslSupported;
 279 #ifndef QT_NO_COMPRESS
 280     m["UseCompression"] = true;
 281 #else
 282     m["UseCompression"] = false;
 283 #endif
 284
 285     writeSocketData(m);
 286 }
 287
 288
 289 void DataStreamPeer::dispatch(const ClientDenied &msg) {
 290     QVariantMap m;
 291     m["MsgType"] = "ClientInitReject";
 292     m["Error"] = msg.errorString;
 293
 294     writeSocketData(m);
 295 }
 296
 297
 298 void DataStreamPeer::dispatch(const ClientRegistered &msg) {
 299     QVariantMap m;
 300     m["MsgType"] = "ClientInitAck";
 301     m["CoreFeatures"] = msg.coreFeatures;
 302     m["StorageBackends"] = msg.backendInfo;
 303
 304     // FIXME only in compat mode
 305     m["ProtocolVersion"] = protocolVersion;
 306     m["SupportSsl"] = msg.sslSupported;
 307     m["SupportsCompression"] = socket()->property("UseCompression").toBool(); // this property gets already set in the ClientInit handler
 308
 309     // This is only used for old v10 clients (pre-0.5)
 310     int uptime = msg.coreStartTime.secsTo(QDateTime::currentDateTime().toUTC());
 311     int updays = uptime / 86400; uptime %= 86400;
 312     int uphours = uptime / 3600; uptime %= 3600;
 313     int upmins = uptime / 60;
 314     m["CoreInfo"] = tr("<b>Quassel Core Version %1</b><br>"
 315                        "Built: %2<br>"
 316                        "Up %3d%4h%5m (since %6)").arg(Quassel::buildInfo().fancyVersionString)
 317                        .arg(Quassel::buildInfo().buildDate)
 318                        .arg(updays).arg(uphours, 2, 10, QChar('0')).arg(upmins, 2, 10, QChar('0')).arg(msg.coreStartTime.toString(Qt::TextDate));
 319
 320     m["LoginEnabled"] = m["Configured"] = msg.coreConfigured;
 321
 322     writeSocketData(m);
 323 }
 324
 325
 326 void DataStreamPeer::dispatch(const SetupData &msg)
 327 {
 328     QVariantMap map;
 329     map["AdminUser"] = msg.adminUser;
 330     map["AdminPasswd"] = msg.adminPassword;
 331     map["Backend"] = msg.backend;
 332     map["ConnectionProperties"] = msg.setupData;
 333
 334     QVariantMap m;
 335     m["MsgType"] = "CoreSetupData";
 336     m["SetupData"] = map;
 337     writeSocketData(m);
 338 }
 339
 340
 341 void DataStreamPeer::dispatch(const SetupFailed &msg)
 342 {
 343     QVariantMap m;
 344     m["MsgType"] = "CoreSetupReject";
 345     m["Error"] = msg.errorString;
 346
 347     writeSocketData(m);
 348 }
 349
 350
 351 void DataStreamPeer::dispatch(const SetupDone &msg)
 352 {
 353     Q_UNUSED(msg)
 354
 355     QVariantMap m;
 356     m["MsgType"] = "CoreSetupAck";
 357
 358     writeSocketData(m);
 359 }
 360
 361
 362 void DataStreamPeer::dispatch(const Login &msg)
 363 {
 364     QVariantMap m;
 365     m["MsgType"] = "ClientLogin";
 366     m["User"] = msg.user;
 367     m["Password"] = msg.password;
 368
 369     writeSocketData(m);
 370 }
 371
 372
 373 void DataStreamPeer::dispatch(const LoginFailed &msg)
 374 {
 375     QVariantMap m;
 376     m["MsgType"] = "ClientLoginReject";
 377     m["Error"] = msg.errorString;
 378
 379     writeSocketData(m);
 380 }
 381
 382
 383 void DataStreamPeer::dispatch(const LoginSuccess &msg)
 384 {
 385     Q_UNUSED(msg)
 386
 387     QVariantMap m;
 388     m["MsgType"] = "ClientLoginAck";
 389
 390     writeSocketData(m);
 391 }
 392
 393
 394 void DataStreamPeer::dispatch(const SessionState &msg)
 395 {
 396     QVariantMap m;
 397     m["MsgType"] = "SessionInit";
 398
 399     QVariantMap map;
 400     map["BufferInfos"] = msg.bufferInfos;
 401     map["NetworkIds"] = msg.networkIds;
 402     map["Identities"] = msg.identities;
 403     m["SessionState"] = map;
 404
 405     writeSocketData(m);
 406 }
 407
 408
 409 /*** Standard messages ***/
 410
 411 void DataStreamPeer::handlePackedFunc(const QVariant &packedFunc)
 412 {
 413     QVariantList params(packedFunc.toList());
 414
 415     if (params.isEmpty()) {
 416         qWarning() << Q_FUNC_INFO << "Received incompatible data:" << packedFunc;
 417         return;
 418     }
 419
 420     // TODO: make sure that this is a valid request type
 421     RequestType requestType = (RequestType)params.takeFirst().value<int>();
 422     switch (requestType) {
 423         case Sync: {
 424             if (params.count() < 3) {
 425                 qWarning() << Q_FUNC_INFO << "Received invalid sync call:" << params;
 426                 return;
 427             }
 428             QByteArray className = params.takeFirst().toByteArray();
 429             QString objectName = params.takeFirst().toString();
 430             QByteArray slotName = params.takeFirst().toByteArray();
 431             handle(Protocol::SyncMessage(className, objectName, slotName, params));
 432             break;
 433         }
 434         case RpcCall: {
 435             if (params.empty()) {
 436                 qWarning() << Q_FUNC_INFO << "Received empty RPC call!";
 437                 return;
 438             }
 439             QByteArray slotName = params.takeFirst().toByteArray();
 440             handle(Protocol::RpcCall(slotName, params));
 441             break;
 442         }
 443         case InitRequest: {
 444             if (params.count() != 2) {
 445                 qWarning() << Q_FUNC_INFO << "Received invalid InitRequest:" << params;
 446                 return;
 447             }
 448             QByteArray className = params[0].toByteArray();
 449             QString objectName = params[1].toString();
 450             handle(Protocol::InitRequest(className, objectName));
 451             break;
 452         }
 453         case InitData: {
 454             if (params.count() != 3) {
 455                 qWarning() << Q_FUNC_INFO << "Received invalid InitData:" << params;
 456                 return;
 457             }
 458             QByteArray className = params[0].toByteArray();
 459             QString objectName = params[1].toString();
 460             QVariantMap initData = params[2].toMap();
 461             handle(Protocol::InitData(className, objectName, initData));
 462             break;
 463         }
 464         case HeartBeat: {
 465             if (params.count() != 1) {
 466                 qWarning() << Q_FUNC_INFO << "Received invalid HeartBeat:" << params;
 467                 return;
 468             }
 469             // The legacy protocol would only send a QTime, no QDateTime
 470             // so we assume it's sent today, which works in exactly the same cases as it did in the old implementation
 471             QDateTime dateTime = QDateTime::currentDateTime().toUTC();
 472             dateTime.setTime(params[0].toTime());
 473             handle(Protocol::HeartBeat(dateTime));
 474             break;
 475         }
 476         case HeartBeatReply: {
 477             if (params.count() != 1) {
 478                 qWarning() << Q_FUNC_INFO << "Received invalid HeartBeat:" << params;
 479                 return;
 480             }
 481             // The legacy protocol would only send a QTime, no QDateTime
 482             // so we assume it's sent today, which works in exactly the same cases as it did in the old implementation
 483             QDateTime dateTime = QDateTime::currentDateTime().toUTC();
 484             dateTime.setTime(params[0].toTime());
 485             handle(Protocol::HeartBeatReply(dateTime));
 486             break;
 487         }
 488
 489     }
 490 }
 491
 492
 493 void DataStreamPeer::dispatch(const Protocol::SyncMessage &msg)
 494 {
 495     dispatchPackedFunc(QVariantList() << (qint16)Sync << msg.className << msg.objectName << msg.slotName << msg.params);
 496 }
 497
 498
 499 void DataStreamPeer::dispatch(const Protocol::RpcCall &msg)
 500 {
 501     dispatchPackedFunc(QVariantList() << (qint16)RpcCall << msg.slotName << msg.params);
 502 }
 503
 504
 505 void DataStreamPeer::dispatch(const Protocol::InitRequest &msg)
 506 {
 507     dispatchPackedFunc(QVariantList() << (qint16)InitRequest << msg.className << msg.objectName);
 508 }
 509
 510
 511 void DataStreamPeer::dispatch(const Protocol::InitData &msg)
 512 {
 513     dispatchPackedFunc(QVariantList() << (qint16)InitData << msg.className << msg.objectName << msg.initData);
 514 }
 515
 516
 517 void DataStreamPeer::dispatch(const Protocol::HeartBeat &msg)
 518 {
 519     dispatchPackedFunc(QVariantList() << (qint16)HeartBeat << msg.timestamp.time());
 520 }
 521
 522
 523 void DataStreamPeer::dispatch(const Protocol::HeartBeatReply &msg)
 524 {
 525     dispatchPackedFunc(QVariantList() << (qint16)HeartBeatReply << msg.timestamp.time());
 526 }
 527
 528
 529 void DataStreamPeer::dispatchPackedFunc(const QVariantList &packedFunc)
 530 {
 531     writeSocketData(QVariant(packedFunc));
 532 }