src/common/remotepeer.cpp

   1 /***************************************************************************
   2  *   Copyright (C) 2005-2018 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 <QtEndian>
  22
  23 #include <QHostAddress>
  24 #include <QTimer>
  25
  26 #ifdef HAVE_SSL
  27 #  include <QSslSocket>
  28 #else
  29 #  include <QTcpSocket>
  30 #endif
  31
  32 #include "remotepeer.h"
  33 #include "util.h"
  34
  35 using namespace Protocol;
  36
  37 const quint32 maxMessageSize = 64 * 1024 * 1024; // This is uncompressed size. 64 MB should be enough for any sort of initData or backlog chunk
  38
  39 RemotePeer::RemotePeer(::AuthHandler *authHandler, QTcpSocket *socket, Compressor::CompressionLevel level, QObject *parent)
  40     : Peer(authHandler, parent),
  41     _socket(socket),
  42     _compressor(new Compressor(socket, level, this)),
  43     _signalProxy(nullptr),
  44     _heartBeatTimer(new QTimer(this)),
  45     _heartBeatCount(0),
  46     _lag(0),
  47     _msgSize(0)
  48 {
  49     socket->setParent(this);
  50     connect(socket, &QAbstractSocket::stateChanged, this, &RemotePeer::onSocketStateChanged);
  51     connect(socket, selectOverload<QAbstractSocket::SocketError>(&QAbstractSocket::error), this, &RemotePeer::onSocketError);
  52     connect(socket, &QAbstractSocket::disconnected, this, &Peer::disconnected);
  53
  54 #ifdef HAVE_SSL
  55     auto *sslSocket = qobject_cast<QSslSocket *>(socket);
  56     if (sslSocket) {
  57         connect(sslSocket, &QSslSocket::encrypted, this, [this]() { emit secureStateChanged(true); });
  58     }
  59 #endif
  60
  61     connect(_compressor, &Compressor::readyRead, this, &RemotePeer::onReadyRead);
  62     connect(_compressor, &Compressor::error, this, &RemotePeer::onCompressionError);
  63
  64     connect(_heartBeatTimer, &QTimer::timeout, this, &RemotePeer::sendHeartBeat);
  65 }
  66
  67
  68 void RemotePeer::onSocketStateChanged(QAbstractSocket::SocketState state)
  69 {
  70     if (state == QAbstractSocket::ClosingState) {
  71         emit statusMessage(tr("Disconnecting..."));
  72     }
  73 }
  74
  75
  76 void RemotePeer::onSocketError(QAbstractSocket::SocketError error)
  77 {
  78     emit socketError(error, socket()->errorString());
  79 }
  80
  81
  82 void RemotePeer::onCompressionError(Compressor::Error error)
  83 {
  84     close(QString("Compression error %1").arg(error));
  85 }
  86
  87
  88 QString RemotePeer::description() const
  89 {
  90     if (socket())
  91         return socket()->peerAddress().toString();
  92
  93     return QString();
  94 }
  95
  96 QString RemotePeer::address() const
  97 {
  98     if (socket())
  99         return socket()->peerAddress().toString();
 100
 101     return QString();
 102 }
 103
 104 quint16 RemotePeer::port() const
 105 {
 106     if (socket())
 107         return socket()->peerPort();
 108
 109     return 0;
 110 }
 111
 112
 113 ::SignalProxy *RemotePeer::signalProxy() const
 114 {
 115     return _signalProxy;
 116 }
 117
 118
 119 void RemotePeer::setSignalProxy(::SignalProxy *proxy)
 120 {
 121     if (proxy == _signalProxy)
 122         return;
 123
 124     if (!proxy) {
 125         _heartBeatTimer->stop();
 126         disconnect(signalProxy(), nullptr, this, nullptr);
 127         _signalProxy = nullptr;
 128         if (isOpen())
 129             close();
 130     }
 131     else {
 132         if (signalProxy()) {
 133             qWarning() << Q_FUNC_INFO << "Setting another SignalProxy not supported, ignoring!";
 134             return;
 135         }
 136         _signalProxy = proxy;
 137         connect(proxy, &SignalProxy::heartBeatIntervalChanged, this, &RemotePeer::changeHeartBeatInterval);
 138         _heartBeatTimer->setInterval(proxy->heartBeatInterval() * 1000);
 139         _heartBeatTimer->start();
 140     }
 141 }
 142
 143
 144 void RemotePeer::changeHeartBeatInterval(int secs)
 145 {
 146     if(secs <= 0)
 147         _heartBeatTimer->stop();
 148     else {
 149         _heartBeatTimer->setInterval(secs * 1000);
 150         _heartBeatTimer->start();
 151     }
 152 }
 153
 154
 155 int RemotePeer::lag() const
 156 {
 157     return _lag;
 158 }
 159
 160
 161 QTcpSocket *RemotePeer::socket() const
 162 {
 163     return _socket;
 164 }
 165
 166
 167 bool RemotePeer::isSecure() const
 168 {
 169     if (socket()) {
 170         if (isLocal())
 171             return true;
 172 #ifdef HAVE_SSL
 173         auto *sslSocket = qobject_cast<QSslSocket *>(socket());
 174         if (sslSocket && sslSocket->isEncrypted())
 175             return true;
 176 #endif
 177     }
 178     return false;
 179 }
 180
 181
 182 bool RemotePeer::isLocal() const
 183 {
 184     if (socket()) {
 185         if (socket()->peerAddress() == QHostAddress::LocalHost || socket()->peerAddress() == QHostAddress::LocalHostIPv6)
 186             return true;
 187     }
 188     return false;
 189 }
 190
 191
 192 bool RemotePeer::isOpen() const
 193 {
 194     return socket() && socket()->state() == QTcpSocket::ConnectedState;
 195 }
 196
 197
 198 void RemotePeer::close(const QString &reason)
 199 {
 200     if (!reason.isEmpty()) {
 201         qWarning() << "Disconnecting:" << reason;
 202     }
 203
 204     if (socket() && socket()->state() != QTcpSocket::UnconnectedState) {
 205         socket()->disconnectFromHost();
 206     }
 207 }
 208
 209
 210 void RemotePeer::onReadyRead()
 211 {
 212     QByteArray msg;
 213     while (readMessage(msg)) {
 214         if (SignalProxy::current())
 215             SignalProxy::current()->setSourcePeer(this);
 216
 217         processMessage(msg);
 218
 219         if (SignalProxy::current())
 220             SignalProxy::current()->setSourcePeer(nullptr);
 221     }
 222 }
 223
 224
 225 bool RemotePeer::readMessage(QByteArray &msg)
 226 {
 227     if (_msgSize == 0) {
 228         if (_compressor->bytesAvailable() < 4)
 229             return false;
 230         _compressor->read((char*)&_msgSize, 4);
 231         _msgSize = qFromBigEndian<quint32>(_msgSize);
 232
 233         if (_msgSize > maxMessageSize) {
 234             close("Peer tried to send package larger than max package size!");
 235             return false;
 236         }
 237
 238         if (_msgSize == 0) {
 239             close("Peer tried to send an empty message!");
 240             return false;
 241         }
 242     }
 243
 244     if (_compressor->bytesAvailable() < _msgSize) {
 245         emit transferProgress(socket()->bytesAvailable(), _msgSize);
 246         return false;
 247     }
 248
 249     emit transferProgress(_msgSize, _msgSize);
 250
 251     msg.resize(_msgSize);
 252     qint64 bytesRead = _compressor->read(msg.data(), _msgSize);
 253     if (bytesRead != _msgSize) {
 254         close("Premature end of data stream!");
 255         return false;
 256     }
 257
 258     _msgSize = 0;
 259     return true;
 260 }
 261
 262
 263 void RemotePeer::writeMessage(const QByteArray &msg)
 264 {
 265     auto size = qToBigEndian<quint32>(msg.size());
 266     _compressor->write((const char*)&size, 4, Compressor::NoFlush);
 267     _compressor->write(msg.constData(), msg.size());
 268 }
 269
 270
 271 void RemotePeer::handle(const HeartBeat &heartBeat)
 272 {
 273     dispatch(HeartBeatReply(heartBeat.timestamp));
 274 }
 275
 276
 277 void RemotePeer::handle(const HeartBeatReply &heartBeatReply)
 278 {
 279     _heartBeatCount = 0;
 280     emit lagUpdated(heartBeatReply.timestamp.msecsTo(QDateTime::currentDateTime().toUTC()) / 2);
 281 }
 282
 283
 284 void RemotePeer::sendHeartBeat()
 285 {
 286     if (signalProxy()->maxHeartBeatCount() > 0 && _heartBeatCount >= signalProxy()->maxHeartBeatCount()) {
 287         qWarning() << "Disconnecting peer:" << description()
 288                    << "(didn't receive a heartbeat for over" << _heartBeatCount *_heartBeatTimer->interval() / 1000 << "seconds)";
 289         socket()->close();
 290         _heartBeatTimer->stop();
 291         return;
 292     }
 293
 294     if (_heartBeatCount > 0) {
 295         _lag = _heartBeatCount * _heartBeatTimer->interval();
 296         emit lagUpdated(_lag);
 297     }
 298
 299     dispatch(HeartBeat(QDateTime::currentDateTime().toUTC()));
 300     ++_heartBeatCount;
 301 }