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