src/common/remotepeer.cpp

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