Teaspeak-Server/server/src/client/voice/VoiceClientConnection.cpp

#include <algorithm>
#include <log/LogUtils.h>
#include "../../server/VoiceServer.h"
#include <misc/memtracker.h>
#include <protocol/Packet.h>
#include <ThreadPool/Timer.h>
#include "VoiceClientConnection.h"
#include "VoiceClient.h"


//#define LOG_AUTO_ACK_AUTORESPONSE
//#define FUZZING_TESTING_INCOMMING
//#define FUZZING_TESTING_OUTGOING
//#define FIZZING_TESTING_DISABLE_HANDSHAKE
#define FUZZING_TESTING_DROP 8
#define FUZZING_TESTING_DROP_MAX 10

//#define CONNECTION_NO_STATISTICS

#define QLZ_COMPRESSION_LEVEL 1
#include "qlz/QuickLZ.h"

using namespace std;
using namespace std::chrono;
using namespace ts;
using namespace ts::connection;
using namespace ts::protocol;
using namespace ts::server;

VoiceClientConnection::VoiceClientConnection(VoiceClient* client)
    : client{client}, crypt_handler{}, packet_decoder_{&this->crypt_handler}, packet_encoder_{&this->crypt_handler, &this->packet_statistics_} {
    memtrack::allocated<VoiceClientConnection>(this);

    this->packet_decoder_.callback_argument = this;
    this->packet_decoder_.callback_decoded_packet = VoiceClientConnection::callback_packet_decoded;
    this->packet_decoder_.callback_decoded_command = VoiceClientConnection::callback_command_decoded;
    this->packet_decoder_.callback_send_acknowledge = VoiceClientConnection::callback_send_acknowledge;

    this->packet_encoder_.callback_data = this;
    this->packet_encoder_.callback_request_write = VoiceClientConnection::callback_request_write;
    this->packet_encoder_.callback_crypt_error = VoiceClientConnection::callback_encode_crypt_error;
    this->packet_encoder_.callback_resend_failed = VoiceClientConnection::callback_resend_failed;
    this->packet_encoder_.callback_resend_stats = VoiceClientConnection::callback_resend_statistics;
    this->packet_encoder_.callback_connection_stats = VoiceClientConnection::callback_outgoing_connection_statistics;

    debugMessage(client->getServer()->getServerId(), "Allocated new voice client connection at {}", (void*) this);
}

VoiceClientConnection::~VoiceClientConnection() {
    this->reset();
    this->client = nullptr;
    memtrack::freed<VoiceClientConnection>(this);
}

void VoiceClientConnection::triggerWrite() {
    if(this->client->voice_server)
        this->client->voice_server->triggerWrite(dynamic_pointer_cast<VoiceClient>(this->client->_this.lock()));
}

#ifdef CLIENT_LOG_PREFIX
    #undef CLIENT_LOG_PREFIX
#endif
#define CLIENT_LOG_PREFIX "[" << this->client->getPeerIp() << ":" << this->client->getPeerPort() << " | " << this->client->getDisplayName() << "]"

//Message handle methods
void VoiceClientConnection::handle_incoming_datagram(const pipes::buffer_view& buffer) {
    ClientPacketParser packet_parser{buffer};
    if(!packet_parser.valid())
        return;

#ifndef CONNECTION_NO_STATISTICS
    if(this->client) {
        auto stats = this->client->connectionStatistics;
        stats->logIncomingPacket(stats::ConnectionStatistics::category::from_type(packet_parser.type()), buffer.length() + 96); /* 96 for the UDP packet overhead */
    }
    this->packet_statistics().received_packet((protocol::PacketType) packet_parser.type(), packet_parser.full_packet_id());
#endif

    std::string error{};
    auto result = this->packet_decoder_.process_incoming_data(packet_parser, error);
    using PacketProcessResult = server::server::udp::PacketProcessResult;
    switch (result) {
        case PacketProcessResult::SUCCESS:
        case PacketProcessResult::FUZZ_DROPPED: /* maybe some kind of log? */
        case PacketProcessResult::DECRYPT_FAILED: /* Silently drop this packet */
        case PacketProcessResult::DUPLICATED_PACKET:  /* no action needed, acknowledge should be send already  */
            break;

        case PacketProcessResult::DECRYPT_KEY_GEN_FAILED:
            /* no action needed, acknowledge should be send */
            logCritical(this->client->getServerId(), "{} Failed to generate decrypt key. Dropping packet.", CLIENT_STR_LOG_PREFIX_(this->client));
            break;

        case PacketProcessResult::BUFFER_OVERFLOW:
        case PacketProcessResult::BUFFER_UNDERFLOW:
            debugMessage(this->client->getServerId(), "{} Dropping command packet because command assembly buffer has an {}: {}",
                         CLIENT_STR_LOG_PREFIX_(this->client),
                         result == PacketProcessResult::BUFFER_UNDERFLOW ? "underflow" : "overflow",
                         error
            );
            break;

        case PacketProcessResult::UNKNOWN_ERROR:
            logCritical(this->client->getServerId(), "{} Having an unknown error while processing a incoming packet: {}",
                        CLIENT_STR_LOG_PREFIX_(this->client),
                        error
            );
            goto disconnect_client;

        case PacketProcessResult::COMMAND_BUFFER_OVERFLOW:
            debugMessage(this->client->getServerId(), "{} Having a command buffer overflow. This might cause the client to drop.", CLIENT_STR_LOG_PREFIX_(this->client));
            break;

        case PacketProcessResult::COMMAND_DECOMPRESS_FAILED:
            logWarning(this->client->getServerId(), "{} Failed to decompress a command packet. Dropping command.", CLIENT_STR_LOG_PREFIX_(this->client));
            break;

        case PacketProcessResult::COMMAND_TOO_LARGE:
            logWarning(this->client->getServerId(), "{} Received a too large command. Dropping client.", CLIENT_STR_LOG_PREFIX_(this->client));
            goto disconnect_client;

        case PacketProcessResult::COMMAND_SEQUENCE_LENGTH_TOO_LONG:
            logWarning(this->client->getServerId(), "{} Received a too long command sequence. Dropping client.", CLIENT_STR_LOG_PREFIX_(this->client));
            goto disconnect_client;

        default:
            assert(false);
            break;
    }

    return;

    disconnect_client:;
    /* FIXME: Disconnect the client */
}

void VoiceClientConnection::callback_send_acknowledge(void *ptr_this, uint16_t packet_id, bool command_low) {
    /* FIXME: Move this to the connection! */
    reinterpret_cast<VoiceClientConnection*>(ptr_this)->client->sendAcknowledge(packet_id, command_low);
}

void VoiceClientConnection::callback_packet_decoded(void *ptr_this, const ts::protocol::ClientPacketParser &packet) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);
    switch (packet.type()) {
        case protocol::VOICE:
            connection->client->handlePacketVoice(packet);
            break;

        case protocol::VOICE_WHISPER:
            connection->client->handlePacketVoiceWhisper(packet);
            break;

        case protocol::ACK:
        case protocol::ACK_LOW:
            connection->client->handlePacketAck(packet);
            break;

        case protocol::PING:
        case protocol::PONG:
            connection->client->handlePacketPing(packet);
            break;

        default:
            assert(false);
            logError(connection->client->getServerId(), "{} Received hand decoded packet, but we've no method to handle it. Dropping packet.", CLIENT_STR_LOG_PREFIX_(connection->client));
            break;
    }
}

void VoiceClientConnection::callback_command_decoded(void *ptr_this, ts::server::server::udp::ReassembledCommand *&command) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);

    /* we're exchanging the command so we're taking the ownership */
    connection->enqueue_command_execution(std::exchange(command, nullptr));
}

bool VoiceClientConnection::verify_encryption(const pipes::buffer_view &buffer /* incl. mac etc */) {
    return this->packet_decoder_.verify_encryption(buffer);
}

void VoiceClientConnection::enqueue_command_execution(ReassembledCommand *command) {
    assert(!command->next_command);

    bool command_handling_scheduled{false};
    {
        std::lock_guard pc_lock{this->pending_commands_lock};
        *this->pending_commands_tail = command;
        this->pending_commands_tail = &command->next_command;

        command_handling_scheduled = std::exchange(this->has_command_handling_scheduled, true);
    }

    if(!command_handling_scheduled) {
        auto voice_server = this->client->voice_server;
        if(voice_server)
            voice_server->schedule_command_handling(this->client);
    }
}

void VoiceClientConnection::execute_handle_command_packets(const std::chrono::system_clock::time_point& /* scheduled */) {
    if(this->client->state >= ConnectionState::DISCONNECTING || !this->client->getServer())
        return;

    std::unique_ptr<ReassembledCommand, void(*)(ReassembledCommand*)> pending_command{nullptr, ReassembledCommand::free};
    while(true) {
        {
            std::lock_guard pc_lock{this->pending_commands_lock};
            pending_command.reset(this->pending_commands_head);
            if(!pending_command) {
                this->has_command_handling_scheduled = false;
                return;
            } else if(pending_command->next_command) {
                this->pending_commands_head = pending_command->next_command;
            } else {
                this->pending_commands_head = nullptr;
                this->pending_commands_tail = &this->pending_commands_head;
            }
        }

        auto startTime = system_clock::now();
        try {
            this->client->handlePacketCommand(pipes::buffer_view{pending_command->command(), pending_command->length()});
        } catch (std::exception& ex) {
            logCritical(this->client->getServerId(), "{} Exception reached root tree! {}", CLIENT_STR_LOG_PREFIX_(this->client), ex.what());
        }

        auto end = system_clock::now();
        if(end - startTime > milliseconds(10)) {
            logError(this->client->getServerId(),
                     "{} Handling of command packet needs more than 10ms ({}ms)",
                     CLIENT_STR_LOG_PREFIX_(this->client),
                     duration_cast<milliseconds>(end - startTime).count()
            );
        }

        break; /* Maybe handle more than one command? Maybe some kind of time limit? */
    }

    auto voice_server = this->client->voice_server;
    if(voice_server)
        voice_server->schedule_command_handling(this->client);
}


bool VoiceClientConnection::wait_empty_write_and_prepare_queue(chrono::time_point<chrono::system_clock> until) {
    return this->packet_encoder_.wait_empty_write_and_prepare_queue(until);
}

void VoiceClientConnection::reset() {
    this->crypt_handler.reset();

    {
        std::unique_lock pc_lock{this->pending_commands_lock};
        auto head = std::exchange(this->pending_commands_head, nullptr);
        this->pending_commands_tail = &this->pending_commands_head;
        pc_lock.unlock();

        while(head) {
            auto cmd = head->next_command;
            ReassembledCommand::free(head);
            head = cmd;
        }
    }

    this->packet_decoder_.reset();
    this->packet_encoder_.reset();
}

void VoiceClientConnection::force_insert_command(const pipes::buffer_view &buffer) {
    auto command = ReassembledCommand::allocate(buffer.length());
    memcpy(command->command(), buffer.data_ptr(), command->length());
    this->enqueue_command_execution(command);
}

void VoiceClientConnection::send_packet(protocol::OutgoingServerPacket *packet) {
    this->packet_encoder_.send_packet(packet);

    auto statistics = this->client ? this->client->connectionStatistics : nullptr;
    if(statistics) {
        auto category = stats::ConnectionStatistics::category::from_type(packet->packet_type());
        statistics->logOutgoingPacket(category, packet->packet_length() + 96); /* 96 for the UDP packet overhead */
    }
}

void VoiceClientConnection::send_packet(protocol::PacketType type, protocol::PacketFlag::PacketFlags flag, const void *payload, size_t payload_size) {
    auto packet = protocol::allocate_outgoing_packet(payload_size);

    packet->type_and_flags = (uint8_t) type | (uint8_t) flag;
    memcpy(packet->payload, payload, payload_size);

    this->send_packet(packet);
}

void VoiceClientConnection::send_command(const std::string_view &cmd, bool b, std::unique_ptr<threads::Future<bool>> cb) {
    this->packet_encoder_.send_command(cmd, b, std::move(cb));
}

void VoiceClientConnection::callback_encode_crypt_error(void *ptr_this,
                                                        const PacketEncoder::CryptError &error,
                                                        const std::string &detail) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);
    switch (error) {
        case PacketEncoder::CryptError::ENCRYPT_FAILED:
            logError(connection->client->getServerId(), "{} Failed to encrypt packet. Error: {}", CLIENT_STR_LOG_PREFIX_(connection->client), detail);
            break;

        case PacketEncoder::CryptError::KEY_GENERATION_FAILED:
            logError(connection->client->getServerId(), "{} Failed to generate crypt key/nonce for sending a packet. This should never happen! Dropping packet.", CLIENT_STR_LOG_PREFIX_(connection->client));
            break;

        default:
            assert(false);
            return;
    }
}

void VoiceClientConnection::callback_request_write(void *ptr_this) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);
    connection->triggerWrite();
}

void VoiceClientConnection::callback_resend_failed(void *ptr_this, const shared_ptr<AcknowledgeManager::Entry> &entry) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);

    debugMessage(connection->client->getServerId(), "{} Failed to execute packet resend of packet {}. Dropping connection.", CLIENT_STR_LOG_PREFIX_(connection->client), entry->packet_full_id);

    if(connection->client->state == ConnectionState::CONNECTED) {
        connection->client->disconnect(ViewReasonId::VREASON_TIMEOUT, config::messages::timeout::packet_resend_failed, nullptr, true);
    } else {
        connection->client->close_connection(system_clock::now() + seconds(1));
    }
}

void VoiceClientConnection::callback_resend_statistics(void *ptr_this, size_t send_count) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);

    logTrace(connection->client->getServerId(), "{} Resending {} packets.", CLIENT_STR_LOG_PREFIX_(connection->client), send_count);
}

void VoiceClientConnection::callback_outgoing_connection_statistics(void *ptr_this,
                                                                    ts::stats::ConnectionStatistics::category::value category,
                                                                    size_t send_count) {
    auto connection = reinterpret_cast<VoiceClientConnection*>(ptr_this);
    auto statistics = connection->client->connectionStatistics;
    if(!statistics) return;

    statistics->logOutgoingPacket(category, send_count);
}