Cockatrice/servatrice/src/isl_interface.cpp

#include "isl_interface.h"

#include "libcockatrice/protocol/pb/commands.pb.h"
#include "libcockatrice/protocol/pb/room_commands.pb.h"
#include "libcockatrice/protocol/pb/room_event.pb.h"
#include "libcockatrice/protocol/pb/serverinfo_game.pb.h"
#include "libcockatrice/protocol/pb/serverinfo_room.pb.h"
#include "libcockatrice/protocol/pb/serverinfo_user.pb.h"
#include "libcockatrice/protocol/pb/session_event.pb.h"
#include "main.h"
#include "servatrice.h"
#include "server_abstractuserinterface.h"
#include "server_logger.h"
#include "serverinfo_user_container.h"

#include <QDebug>
#include <QHostAddress>
#include <QList>
#include <QLoggingCategory>
#include <QMap>
#include <QMapIterator>
#include <QMessageLogger>
#include <QMutexLocker>
#include <QReadLocker>
#include <QReadWriteLock>
#include <QSharedPointer>
#include <QSslError>
#include <QSslSocket>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>
#include <google/protobuf/stubs/common.h>
#include <libcockatrice/protocol/debug_pb_message.h>
#include <libcockatrice/protocol/get_pb_extension.h>
#include <libcockatrice/protocol/pb/event_game_joined.pb.h>
#include <libcockatrice/protocol/pb/event_join_room.pb.h>
#include <libcockatrice/protocol/pb/event_leave_room.pb.h>
#include <libcockatrice/protocol/pb/event_list_games.pb.h>
#include <libcockatrice/protocol/pb/event_remove_messages.pb.h>
#include <libcockatrice/protocol/pb/event_room_channel_say.pb.h>
#include <libcockatrice/protocol/pb/event_room_say.pb.h>
#include <libcockatrice/protocol/pb/event_server_complete_list.pb.h>
#include <libcockatrice/protocol/pb/event_user_joined.pb.h>
#include <libcockatrice/protocol/pb/event_user_left.pb.h>
#include <libcockatrice/protocol/pb/isl_message.pb.h>
#include <qnamespace.h>
#include <qobjectdefs.h>
#include <server_protocolhandler.h>
#include <server_room.h>

// Maximum size of a single ISL message.
static const int MAX_ISL_MESSAGE_SIZE = 4 * 1024 * 1024;

inline Q_LOGGING_CATEGORY(IslInterfaceLog, "isl_interface");

void IslInterface::sharedCtor(const QSslCertificate &cert, const QSslKey &privateKey)
{
    socket = new QSslSocket(this);
    socket->setLocalCertificate(cert);
    socket->setPrivateKey(privateKey);

    connect(socket, &QSslSocket::readyRead, this, &IslInterface::readClient, Qt::QueuedConnection);
    connect(socket, &QAbstractSocket::errorOccurred, this, &IslInterface::catchSocketError);
    connect(this, &IslInterface::outputBufferChanged, this, &IslInterface::flushOutputBuffer, Qt::QueuedConnection);
}

IslInterface::IslInterface(int _socketDescriptor,
                           const QSslCertificate &cert,
                           const QSslKey &privateKey,
                           Servatrice *_server)
    : QObject(), serverId(-1), socketDescriptor(_socketDescriptor), server(_server), messageInProgress(false)
{
    sharedCtor(cert, privateKey);
}

IslInterface::IslInterface(int _serverId,
                           const QString &_peerHostName,
                           const QString &_peerAddress,
                           int _peerPort,
                           const QSslCertificate &_peerCert,
                           const QSslCertificate &cert,
                           const QSslKey &privateKey,
                           Servatrice *_server)
    : QObject(), serverId(_serverId), peerHostName(_peerHostName), peerAddress(_peerAddress), peerPort(_peerPort),
      peerCert(_peerCert), server(_server), messageInProgress(false)
{
    sharedCtor(cert, privateKey);
}

IslInterface::~IslInterface()
{
    logger->logMessage("[ISL] session ended", this);

    flushOutputBuffer();

    // As these signals are connected with Qt::QueuedConnection implicitly,
    // we don't need to worry about them modifying the lists while we're iterating.

    server->roomsLock.lockForRead();
    QMapIterator<int, Server_Room *> roomIterator(server->getRooms());
    while (roomIterator.hasNext()) {
        Server_Room *room = roomIterator.next().value();
        room->usersLock.lockForRead();
        QMapIterator<QString, ServerInfo_User_Container> roomUsers(room->getExternalUsers());
        while (roomUsers.hasNext()) {
            roomUsers.next();
            if (roomUsers.value().getUserInfo()->server_id() == serverId) {
                emit externalRoomUserLeft(room->getId(), roomUsers.key());
            }
        }
        room->usersLock.unlock();
    }
    server->roomsLock.unlock();

    server->clientsLock.lockForRead();
    QMapIterator<QString, Server_AbstractUserInterface *> extUsers(server->getExternalUsers());
    while (extUsers.hasNext()) {
        extUsers.next();
        if (extUsers.value()->getUserInfo()->server_id() == serverId) {
            emit externalUserLeft(extUsers.key());
        }
    }
    server->clientsLock.unlock();
}

void IslInterface::initServer()
{
    socket->setSocketDescriptor(socketDescriptor);

    logger->logMessage(QString("[ISL] incoming connection: %1").arg(socket->peerAddress().toString()));

    QList<ServerProperties> serverList = server->getServerList();
    int listIndex = -1;
    for (int i = 0; i < serverList.size(); ++i) {
        if (serverList[i].address == socket->peerAddress()) {
            listIndex = i;
            break;
        }
    }
    if (listIndex == -1) {
        logger->logMessage(
            QString("[ISL] address %1 unknown, terminating connection").arg(socket->peerAddress().toString()));
        deleteLater();
        return;
    }

    socket->startServerEncryption();
    if (!socket->waitForEncrypted(5000)) {
        QList<QSslError> sslErrors(socket->sslHandshakeErrors());
        if (sslErrors.isEmpty()) {
            qCDebug(IslInterfaceLog) << "SSL handshake timeout, terminating connection";
        } else {
            qCWarning(IslInterfaceLog) << "SSL errors:" << sslErrors;
        }
        deleteLater();
        return;
    }

    if (serverList[listIndex].cert == socket->peerCertificate()) {
        logger->logMessage(QString("[ISL] Peer authenticated as " + serverList[listIndex].hostname));
    } else {
        logger->logMessage(QString("[ISL] Authentication failed, terminating connection"));
        deleteLater();
        return;
    }
    serverId = serverList[listIndex].id;

    Event_ServerCompleteList event;
    event.set_server_id(server->getServerID());

    server->clientsLock.lockForRead();
    QMapIterator<QString, Server_ProtocolHandler *> userIterator(server->getUsers());
    while (userIterator.hasNext()) {
        event.add_user_list()->CopyFrom(userIterator.next().value()->copyUserInfo(true, true));
    }
    server->clientsLock.unlock();

    server->roomsLock.lockForRead();
    QMapIterator<int, Server_Room *> roomIterator(server->getRooms());
    while (roomIterator.hasNext()) {
        Server_Room *room = roomIterator.next().value();
        room->usersLock.lockForRead();
        room->gamesLock.lockForRead();
        room->getInfo(*event.add_room_list(), true, true, false);
    }

    IslMessage message;
    message.set_message_type(IslMessage::SESSION_EVENT);
    SessionEvent *sessionEvent = message.mutable_session_event();
    sessionEvent->GetReflection()
        ->MutableMessage(sessionEvent, event.GetDescriptor()->FindExtensionByName("ext"))
        ->CopyFrom(event);

    server->islLock.lockForWrite();
    if (server->islConnectionExists(serverId)) {
        qCDebug(IslInterfaceLog) << "Duplicate connection to #" << serverId << "terminating connection";
        deleteLater();
    } else {
        transmitMessage(message);
        server->addIslInterface(serverId, this);
    }
    server->islLock.unlock();

    roomIterator.toFront();
    while (roomIterator.hasNext()) {
        roomIterator.next();
        roomIterator.value()->gamesLock.unlock();
        roomIterator.value()->usersLock.unlock();
    }
    server->roomsLock.unlock();
}

void IslInterface::initClient()
{
    QList<QSslError> expectedErrors;
    expectedErrors.append(QSslError(QSslError::SelfSignedCertificate, peerCert));
    socket->ignoreSslErrors(expectedErrors);

    qCDebug(IslInterfaceLog) << "Connecting to #" << serverId << ":" << peerAddress << ":" << peerPort;

    socket->connectToHostEncrypted(peerAddress, peerPort, peerHostName);
    if (!socket->waitForConnected(5000)) {
        qCDebug(IslInterfaceLog) << "Socket error:" << socket->errorString();
        deleteLater();
        return;
    }
    if (!socket->waitForEncrypted(5000)) {
        QList<QSslError> sslErrors(socket->sslHandshakeErrors());
        if (sslErrors.isEmpty()) {
            qCDebug(IslInterfaceLog) << "SSL handshake timeout, terminating connection";
        } else {
            qCWarning(IslInterfaceLog) << "SSL errors:" << sslErrors;
        }
        deleteLater();
        return;
    }

    server->islLock.lockForWrite();
    if (server->islConnectionExists(serverId)) {
        qCDebug(IslInterfaceLog) << "Duplicate connection to #" << serverId << "terminating connection";
        deleteLater();
        return;
    }

    server->addIslInterface(serverId, this);
    server->islLock.unlock();
}

void IslInterface::flushOutputBuffer()
{
    QMutexLocker locker(&outputBufferMutex);
    if (outputBuffer.isEmpty()) {
        return;
    }
    server->incTxBytes(outputBuffer.size());
    socket->write(outputBuffer);
    socket->flush();
    outputBuffer.clear();
}

void IslInterface::readClient()
{
    QByteArray data = socket->readAll();
    server->incRxBytes(data.size());
    inputBuffer.append(data);

    // Parse messages straight out of inputBuffer at readOffset. Removing each
    // consumed message from the front would memmove the remaining bytes once
    // per message; the consumed prefix is dropped once parsing stops.
    qsizetype readOffset = 0;

    while (true) {
        if (!messageInProgress) {
            if (inputBuffer.size() - readOffset >= 4) {
                const auto *header = reinterpret_cast<const unsigned char *>(inputBuffer.constData() + readOffset);
                messageLength = (static_cast<quint32>(header[0]) << 24) + (static_cast<quint32>(header[1]) << 16) +
                                (static_cast<quint32>(header[2]) << 8) + static_cast<quint32>(header[3]);
                // Reject implausible lengths instead of buffering until they arrive.
                // This also handles lengths that overflow into negative int values.
                if (messageLength < 0 || messageLength > MAX_ISL_MESSAGE_SIZE) {
                    qCWarning(IslInterfaceLog) << "Invalid message length" << messageLength << "from" << peerAddress;
                    socket->disconnectFromHost();
                    return;
                }
                readOffset += 4;
                messageInProgress = true;
            } else {
                break;
            }
        }
        if (inputBuffer.size() - readOffset < messageLength) {
            break;
        }

        IslMessage newMessage;
        bool ok = newMessage.ParseFromArray(inputBuffer.constData() + readOffset, messageLength);
        readOffset += messageLength;
        messageInProgress = false;

        if (ok) {
            processMessage(newMessage);
        } else {
            qCWarning(IslInterfaceLog) << "parsing error!";
        }
    }

    // Drop the consumed prefix once, instead of shifting the buffer per message.
    if (readOffset > 0) {
        inputBuffer.remove(0, readOffset);
    }
}

void IslInterface::catchSocketError(QAbstractSocket::SocketError socketError)
{
    qCWarning(IslInterfaceLog) << "Socket error:" << socketError;

    server->islLock.lockForWrite();
    server->removeIslInterface(serverId);
    server->islLock.unlock();

    deleteLater();
}

void IslInterface::transmitMessage(const IslMessage &item)
{
    QByteArray buf;
#if GOOGLE_PROTOBUF_VERSION > 3001000
    unsigned int size = static_cast<unsigned int>(item.ByteSizeLong());
#else
    unsigned int size = static_cast<unsigned int>(item.ByteSize());
#endif
    buf.resize(size + 4);
    if (!item.SerializeToArray(buf.data() + 4, size)) {
        qCWarning(IslInterfaceLog) << "transmit error!";
        return;
    }
    buf.data()[3] = (unsigned char)size;
    buf.data()[2] = (unsigned char)(size >> 8);
    buf.data()[1] = (unsigned char)(size >> 16);
    buf.data()[0] = (unsigned char)(size >> 24);

    outputBufferMutex.lock();
    outputBuffer.append(buf);
    outputBufferMutex.unlock();
    emit outputBufferChanged();
}

void IslInterface::sessionEvent_ServerCompleteList(const Event_ServerCompleteList &event)
{
    for (int i = 0; i < event.user_list_size(); ++i) {
        ServerInfo_User temp(event.user_list(i));
        temp.set_server_id(serverId);
        emit externalUserJoined(temp);
    }
    for (int i = 0; i < event.room_list_size(); ++i) {
        const ServerInfo_Room &room = event.room_list(i);
        for (int j = 0; j < room.user_list_size(); ++j) {
            ServerInfo_User userInfo(room.user_list(j));
            userInfo.set_server_id(serverId);
            emit externalRoomUserJoined(room.room_id(), userInfo);
        }
        for (int j = 0; j < room.game_list_size(); ++j) {
            ServerInfo_Game gameInfo(room.game_list(j));
            gameInfo.set_server_id(serverId);
            emit externalRoomGameListChanged(room.room_id(), gameInfo);
        }
    }
}

void IslInterface::sessionEvent_UserJoined(const Event_UserJoined &event)
{
    ServerInfo_User userInfo(event.user_info());
    userInfo.set_server_id(serverId);
    emit externalUserJoined(userInfo);
}

void IslInterface::sessionEvent_UserLeft(const Event_UserLeft &event)
{
    emit externalUserLeft(QString::fromStdString(event.name()));
}

void IslInterface::roomEvent_UserJoined(int roomId, const Event_JoinRoom &event)
{
    ServerInfo_User userInfo(event.user_info());
    userInfo.set_server_id(serverId);
    emit externalRoomUserJoined(roomId, userInfo);
}

void IslInterface::roomEvent_UserLeft(int roomId, const Event_LeaveRoom &event)
{
    emit externalRoomUserLeft(roomId, QString::fromStdString(event.name()));
}

void IslInterface::roomEvent_Say(int roomId, const Event_RoomSay &event)
{
    emit externalRoomSay(roomId, QString::fromStdString(event.name()), QString::fromStdString(event.message()),
                         QString());
}

void IslInterface::roomEvent_ChannelSay(int roomId, const Event_RoomChannelSay &event)
{
    emit externalRoomSay(roomId, QString::fromStdString(event.name()), QString::fromStdString(event.message()),
                         QString::fromStdString(event.channel_id()));
}

void IslInterface::roomEvent_ListGames(int roomId, const Event_ListGames &event)
{
    for (int i = 0; i < event.game_list_size(); ++i) {
        ServerInfo_Game gameInfo(event.game_list(i));
        gameInfo.set_server_id(serverId);
        emit externalRoomGameListChanged(roomId, gameInfo);
    }
}

void IslInterface::roomEvent_RemoveMessages(int roomId, const Event_RemoveMessages &event)
{
    emit externalRoomRemoveMessages(roomId, QString::fromStdString(event.name()), event.amount());
}

void IslInterface::roomCommand_JoinGame(const Command_JoinGame &cmd, int cmdId, int roomId, qint64 sessionId)
{
    emit joinGameCommandReceived(cmd, cmdId, roomId, serverId, sessionId);
}

void IslInterface::processSessionEvent(const SessionEvent &event, qint64 sessionId)
{
    switch (getPbExtension(event)) {
        case SessionEvent::SERVER_COMPLETE_LIST:
            sessionEvent_ServerCompleteList(event.GetExtension(Event_ServerCompleteList::ext));
            break;
        case SessionEvent::USER_JOINED:
            sessionEvent_UserJoined(event.GetExtension(Event_UserJoined::ext));
            break;
        case SessionEvent::USER_LEFT:
            sessionEvent_UserLeft(event.GetExtension(Event_UserLeft::ext));
            break;
        case SessionEvent::GAME_JOINED: {
            QReadLocker clientsLocker(&server->clientsLock);
            Server_AbstractUserInterface *client = server->getUsersBySessionId().value(sessionId);
            if (!client) {
                qCDebug(IslInterfaceLog) << "IslInterface::processSessionEvent: session id" << sessionId << "not found";
                break;
            }
            const Event_GameJoined &gameJoined = event.GetExtension(Event_GameJoined::ext);
            client->playerAddedToGame(gameJoined.game_info().game_id(), gameJoined.game_info().room_id(),
                                      gameJoined.player_id());
            client->sendProtocolItem(event);
            break;
        }
        case SessionEvent::USER_MESSAGE:
        case SessionEvent::REPLAY_ADDED: {
            QReadLocker clientsLocker(&server->clientsLock);
            Server_AbstractUserInterface *client = server->getUsersBySessionId().value(sessionId);
            if (!client) {
                qCWarning(IslInterfaceLog)
                    << "IslInterface::processSessionEvent: session id" << sessionId << "not found";
                break;
            }

            client->sendProtocolItem(event);
            break;
        }
        default:;
    }
}

void IslInterface::processRoomEvent(const RoomEvent &event)
{
    switch (getPbExtension(event)) {
        case RoomEvent::JOIN_ROOM:
            roomEvent_UserJoined(event.room_id(), event.GetExtension(Event_JoinRoom::ext));
            break;
        case RoomEvent::LEAVE_ROOM:
            roomEvent_UserLeft(event.room_id(), event.GetExtension(Event_LeaveRoom::ext));
            break;
        case RoomEvent::ROOM_SAY:
            roomEvent_Say(event.room_id(), event.GetExtension(Event_RoomSay::ext));
            break;
        case RoomEvent::ROOM_CHANNEL_SAY:
            roomEvent_ChannelSay(event.room_id(), event.GetExtension(Event_RoomChannelSay::ext));
            break;
        case RoomEvent::LIST_GAMES:
            roomEvent_ListGames(event.room_id(), event.GetExtension(Event_ListGames::ext));
            break;
        case RoomEvent::REMOVE_MESSAGES:
            roomEvent_RemoveMessages(event.room_id(), event.GetExtension(Event_RemoveMessages::ext));
            break;
        default:;
    }
}

void IslInterface::processRoomCommand(const CommandContainer &cont, qint64 sessionId)
{
    for (int i = 0; i < cont.room_command_size(); ++i) {
        const RoomCommand &roomCommand = cont.room_command(i);
        switch (static_cast<RoomCommand::RoomCommandType>(getPbExtension(roomCommand))) {
            case RoomCommand::JOIN_GAME:
                roomCommand_JoinGame(roomCommand.GetExtension(Command_JoinGame::ext), cont.cmd_id(), cont.room_id(),
                                     sessionId);
            default:;
        }
    }
}

void IslInterface::processMessage(const IslMessage &item)
{
    if (IslInterfaceLog().isDebugEnabled()) {
        qCDebug(IslInterfaceLog) << getSafeDebugString(item);
    }

    switch (item.message_type()) {
        case IslMessage::ROOM_COMMAND_CONTAINER: {
            processRoomCommand(item.room_command(), item.session_id());
            break;
        }
        case IslMessage::GAME_COMMAND_CONTAINER: {
            emit gameCommandContainerReceived(item.game_command(), item.player_id(), serverId, item.session_id());
            break;
        }
        case IslMessage::SESSION_EVENT: {
            processSessionEvent(item.session_event(), item.session_id());
            break;
        }
        case IslMessage::RESPONSE: {
            emit responseReceived(item.response(), item.session_id());
            break;
        }
        case IslMessage::GAME_EVENT_CONTAINER: {
            emit gameEventContainerReceived(item.game_event_container(), item.session_id());
            break;
        }
        case IslMessage::ROOM_EVENT: {
            processRoomEvent(item.room_event());
            break;
        }
        default:;
    }
}