TeaSpeak/Teaspeak-Server
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

A lot of file transfer updates

Browse Source
This commit is contained in:
WolverinDEV
2020-05-13 11:32:08 +02:00
parent 1a2dd4a008
commit 90b1646876
45 changed files with 1989 additions and 2806 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/client/ConnectedClient.cpp
+3 -2
View File
@@ -11,7 +11,6 @@
#include "src/VirtualServer.h"
#include "voice/VoiceClient.h"
#include "../server/VoiceServer.h"
#include "src/server/file/LocalFileServer.h"
#include "../InstanceHandler.h"
#include "ConnectedClient.h"
@@ -143,7 +142,8 @@ void ConnectedClient::updateChannelClientProperties(bool lock_channel_tree, bool
iconId = value.value;
}
logTrace(this->getServerId(), "[CLIENT] Updating client icon from " + to_string(this->properties()[property::CLIENT_ICON_ID].as<IconId>()) + " to " + to_string(iconId));
if(this->properties()[property::CLIENT_ICON_ID].as<IconId>() != iconId){
if(this->properties()[property::CLIENT_ICON_ID].as<IconId>() != iconId) {
#if 0
if(server_ref && iconId != 0) {
auto dir = serverInstance->getFileServer()->iconDirectory(server_ref);
if(!serverInstance->getFileServer()->iconExists(server_ref, iconId)) {
@@ -151,6 +151,7 @@ void ConnectedClient::updateChannelClientProperties(bool lock_channel_tree, bool
iconId = 0;
}
}
#endif
if(this->properties()[property::CLIENT_ICON_ID].as<IconId>() != iconId) {
this->properties()[property::CLIENT_ICON_ID] = (IconId) iconId;
notifyList.emplace_back(property::CLIENT_ICON_ID);
server/src/client/ConnectedClient.h
+13 -6
View File
@@ -13,7 +13,11 @@
#define CLIENT_STR_LOG_PREFIX CLIENT_STR_LOG_PREFIX_(this)
#define CMD_REQ_SERVER \
if(!this->server) return command_result{error::server_invalid_id};
do { \
if(!this->server) { \
return command_result{error::server_invalid_id}; \
} \
} while(0)
/* TODO: Play lock the server here with read? So the client dosn't get kicked within that moment */
#define CMD_REF_SERVER(variable_name) \
@@ -33,8 +37,10 @@ if(!cmd[0].has(parm)) return command_result{error::parameter_not_found};
//the message here is show to the manager!
#define CMD_CHK_AND_INC_FLOOD_POINTS(num) \
this->increaseFloodPoints(num); \
if(this->shouldFloodBlock()) return command_result{error::ban_flooding};
do {\
this->increaseFloodPoints(num); \
if(this->shouldFloodBlock()) return command_result{error::ban_flooding}; \
} while(0)
#define CMD_CHK_PARM_COUNT(count) \
if(cmd.bulkCount() != count) return command_result{error::parameter_invalid_count};
@@ -453,10 +459,11 @@ namespace ts {
command_result handleCommandFTDeleteFile(Command&);
command_result handleCommandFTInitUpload(Command&);
command_result handleCommandFTInitDownload(Command&);
command_result handleCommandFTGetFileInfo(Command&);
//CMD_TODO handleCommandFTGetFileInfo -> 5 points
command_result handleCommandFTGetFileInfo(Command&);
command_result handleCommandFTRenameFile(Command&);
command_result handleCommandFTList(Command&);
command_result handleCommandFTStop(Command&);
//CMD_TODO handleCommandFTStop -> 5 points
//CMD_TODO handleCommandFTRenameFile -> 5 points
//CMD_TODO handleCommandFTList -> 5 points
command_result handleCommandBanList(Command&);
server/src/client/ConnectedClientNotifyHandler.cpp
-1
View File
@@ -3,7 +3,6 @@
#include <algorithm>
#include "ConnectedClient.h"
#include "voice/VoiceClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../server/VoiceServer.h"
#include "../InstanceHandler.h"
#include "../server/QueryServer.h"
server/src/client/DataClient.cpp
+2 -1
View File
@@ -4,7 +4,6 @@
#include <misc/hex.h>
#include "DataClient.h"
#include "ConnectedClient.h"
#include "src/server/file/LocalFileServer.h"
#include "src/InstanceHandler.h"
#include "misc/base64.h"
@@ -115,6 +114,7 @@ bool DataClient::loadDataForCurrentServer() { //TODO for query
this->clientPermissions = serverInstance->databaseHelper()->loadClientPermissionManager(ref_server, this->getClientDatabaseId());
//Setup / fix stuff
#if 0
if(!this->properties()[property::CLIENT_FLAG_AVATAR].as<string>().empty()){
if(
!ref_server ||
@@ -123,6 +123,7 @@ bool DataClient::loadDataForCurrentServer() { //TODO for query
this->properties()[property::CLIENT_FLAG_AVATAR] = "";
}
}
#endif
if(ref_server)
this->properties()[property::CLIENT_UNREAD_MESSAGES] = ref_server->letters->unread_letter_count(this->getUid());
server/src/client/command_handler/channel.cpp
+1 -2
View File
@@ -5,13 +5,11 @@
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../../server/VoiceServer.h"
#include "../voice/VoiceClient.h"
#include "PermissionManager.h"
#include "../../InstanceHandler.h"
#include "../../server/QueryServer.h"
#include "../file/FileClient.h"
#include "../music/MusicClient.h"
#include "../query/QueryClient.h"
#include "../../weblist/WebListManager.h"
@@ -182,6 +180,7 @@ command_result ConnectedClient::handleCommandChannelGroupCopy(Command &cmd) {
return permission::b_serverinstance_modify_querygroup;
break;
case GroupType::GROUP_TYPE_NORMAL:
default:
break;
}
server/src/client/command_handler/client.cpp
-27
View File
@@ -6,12 +6,10 @@
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../voice/VoiceClient.h"
#include "PermissionManager.h"
#include "../../InstanceHandler.h"
#include "../../server/QueryServer.h"
#include "../file/FileClient.h"
#include "../music/MusicClient.h"
#include "../query/QueryClient.h"
#include "../../weblist/WebListManager.h"
@@ -349,31 +347,6 @@ command_result ConnectedClient::handleCommandClientChatClosed(Command &cmd) {
return command_result{error::ok};
}
//ftgetfilelist cid=1 cpw path=\/ return_code=1:x
//Answer:
//1 .. n
// notifyfilelist cid=1 path=\/ return_code=1:x name=testFile size=35256 datetime=1509459767 type=1|name=testDir size=0 datetime=1509459741 type=0|name=testDir_2 size=0 datetime=1509459763 type=0
//notifyfilelistfinished cid=1 path=\/
inline void cmd_filelist_append_files(ServerId sid, Command &command, vector<std::shared_ptr<file::FileEntry>> files) {
int index = 0;
logTrace(sid, "Sending file list for path {}", command["path"].string());
for (const auto& fileEntry : files) {
logTrace(sid, " - {} ({})", fileEntry->name, fileEntry->type == file::FileType::FILE ? "file" : "directory");
command[index]["name"] = fileEntry->name;
command[index]["datetime"] = std::chrono::duration_cast<std::chrono::seconds>(fileEntry->lastChanged.time_since_epoch()).count();
command[index]["type"] = fileEntry->type;
if (fileEntry->type == file::FileType::FILE)
command[index]["size"] = static_pointer_cast<file::File>(fileEntry)->fileSize;
else
command[index]["size"] = 0;
index++;
}
}
#define CMD_REQ_FSERVER if(!serverInstance->getFileServer()) return command_result{error::vs_critical, "file server not started yet!"}
//start=0 duration=10
//pattern=%asd%
server/src/client/command_handler/file.cpp
+836 -331
View File
File diff suppressed because it is too large Load Diff
server/src/client/command_handler/helpers.h
+3 -1
View File
@@ -79,7 +79,8 @@ if (permission::resolvePermissionData(permType)->type == permission::PermissionT
}
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wswitch-enum"
inline bool permission_require_granted_value(ts::permission::PermissionType type) {
using namespace ts;
/*
@@ -192,6 +193,7 @@ inline bool permission_is_client_property(ts::permission::PermissionType type) {
return false;
}
}
#pragma GCC diagnostic pop
inline ssize_t count_characters(const std::string& in) {
server/src/client/command_handler/misc.cpp
+3 -2
View File
@@ -13,13 +13,11 @@
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../../server/VoiceServer.h"
#include "../voice/VoiceClient.h"
#include "PermissionManager.h"
#include "../../InstanceHandler.h"
#include "../../server/QueryServer.h"
#include "../file/FileClient.h"
#include "../music/MusicClient.h"
#include "../query/QueryClient.h"
#include "../../weblist/WebListManager.h"
@@ -136,6 +134,9 @@ command_result ConnectedClient::handleCommand(Command &cmd) {
else if (command == "ftinitupload") return this->handleCommandFTInitUpload(cmd);
else if (command == "ftinitdownload") return this->handleCommandFTInitDownload(cmd);
else if (command == "ftgetfileinfo") return this->handleCommandFTGetFileInfo(cmd);
else if (command == "ftrenamefile") return this->handleCommandFTRenameFile(cmd);
else if (command == "ftlist") return this->handleCommandFTList(cmd);
else if (command == "ftstop") return this->handleCommandFTStop(cmd);
//Banlist
else if (command == "banlist") return this->handleCommandBanList(cmd);
else if (command == "banadd") return this->handleCommandBanAdd(cmd);
server/src/client/command_handler/music.cpp
-2
View File
@@ -13,13 +13,11 @@
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../../server/VoiceServer.h"
#include "../voice/VoiceClient.h"
#include "PermissionManager.h"
#include "../../InstanceHandler.h"
#include "../../server/QueryServer.h"
#include "../file/FileClient.h"
#include "../music/MusicClient.h"
#include "../query/QueryClient.h"
#include "../../weblist/WebListManager.h"
server/src/client/command_handler/server.cpp
-2
View File
@@ -10,12 +10,10 @@
#include "../../build.h"
#include "../ConnectedClient.h"
#include "../InternalClient.h"
#include "src/server/file/LocalFileServer.h"
#include "../../server/VoiceServer.h"
#include "../voice/VoiceClient.h"
#include "../../InstanceHandler.h"
#include "../../server/QueryServer.h"
#include "../file/FileClient.h"
#include "../music/MusicClient.h"
#include "../query/QueryClient.h"
#include "../../weblist/WebListManager.h"
server/src/client/file/FileClient.cpp
-719
View File
@@ -1,719 +0,0 @@
#include <algorithm>
#include <src/server/file/LocalFileServer.h>
#include <log/LogUtils.h>
#include "FileClient.h"
#include <src/InstanceHandler.h>
#include <experimental/filesystem>
#include <misc/memtracker.h>
#include <misc/base64.h>
#include "src/client/ConnectedClient.h"
#include <netinet/tcp.h>
using namespace std;
using namespace std::chrono;
using namespace ts::server;
namespace fs = std::experimental::filesystem;
#define BUFFER_SIZE (size_t) 2048
FileClient::FileClient(LocalFileServer* handle, int socketFd) : handle(handle), clientFd(socketFd) {
memtrack::allocated<FileClient>(this);
this->last_io_action = system_clock::now();
int enabled = 1;
if(setsockopt(socketFd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof enabled) < 0)
logError(LOG_FT, "{} Cant enable TCP no delay for socket {}. Error: {}/{}", this->client_prefix(), socketFd, errno, strerror(errno));
this->readEvent = event_new(this->handle->ioLoop, socketFd, EV_READ|EV_PERSIST, [](int a, short b, void* c){ ((FileClient*) c)->handleMessageRead(a, b, c); }, this);
this->writeEvent = event_new(this->handle->ioLoop, socketFd, EV_TIMEOUT | EV_WRITE, [](int a, short b, void* c){ ((FileClient*) c)->handleMessageWrite(a, b, c); }, this);
this->ssl_handler.direct_process(pipes::PROCESS_DIRECTION_OUT, true);
this->ssl_handler.direct_process(pipes::PROCESS_DIRECTION_IN, true);
this->ssl_handler.callback_write(bind(&FileClient::sendRawMessage, this, placeholders::_1));
this->ssl_handler.callback_data(bind(&FileClient::handle_ssl_message, this, placeholders::_1));
/*
this->ssl_handler.callback_data([&](const pipes::buffer_view& msg) {
if(this->ftType == FTType::TeaWeb_HTTPS) {
this->handle_http_message(msg);
} else if(this->ftType == FTType::TeaWeb_SSL_WS) {
this->ws_handler.process_incoming_data(msg);
} else {
logError(LOG_FT, "{} Decoded SSL packet, but transfer type isn't SSL!", this->client_prefix());
}
});
*/
//FIXME init ssl error handler?
this->ws_handler.direct_process(pipes::PROCESS_DIRECTION_OUT, true);
this->ws_handler.direct_process(pipes::PROCESS_DIRECTION_IN, true);
/*
this->ws_handler.callback_data([&](const pipes::WSMessage& message) {
if(this->state_transfere == T_INITIALIZE) {
this->applyKey(message.data.string()); //Got the key :)
} else if(this->state_transfere == T_TRANSFER) {
if(this->pendingKey->upload)
this->uploadWriteBytes(message.data.string());
else {
logError(LOG_FT, "{} Invalid write (Just download)", this->client_prefix());
this->disconnect();
}
}
});
*/
this->ws_handler.callback_data(bind(&FileClient::handle_ws_message,this, placeholders::_1));
this->ws_handler.callback_write([&](const pipes::buffer_view& data) {
this->ssl_handler.send(data);
});
this->ws_handler.on_connect = [&]() {};
this->ws_handler.on_disconnect = [&](const std::string&) {};
}
FileClient::~FileClient() {
if(this->thread_flush.joinable()){
this->state_connection = C_DISCONNECTED;
if(this->thread_flush.get_id() != this_thread::get_id())
this->thread_flush.join();
else
this->thread_flush.detach();
}
memtrack::freed<FileClient>(this);
}
size_t FileClient::used_bandwidth() {
auto now = system_clock::now();
size_t tranfarred_bytes = 0;
auto timeout = now - seconds(1);
{
lock_guard<recursive_mutex> lock(this->bandwidth_lock);
for(auto it = this->bandwidth.end(); it != this->bandwidth.begin();) {
--it;
if((*it)->timestamp < timeout) {
this->bandwidth.erase(this->bandwidth.begin(), it);
break;
}
tranfarred_bytes += (*it)->length;
}
}
return tranfarred_bytes;
}
std::string FileClient::client_prefix() {
bool hide_ip = config::server::disable_ip_saving;
if(!hide_ip) {
auto client = this->client;
if(client) {
auto server = client->getServer();
if(server) {
hide_ip = server->disable_ip_saving();
}
}
}
std::string address = "";
if(hide_ip)
address = "X.X.X.X:" + to_string(net::port(this->remote_address));
else
address = net::to_string(this->remote_address);
if(this->client) return "[" + to_string(this->client->getServerId()) + "|" + address + "| " + this->client->getDisplayName() + "]";
return "[0|" + address + "|unconnected]";
}
size_t FileClient::transferred_bytes() {
return this->bytesHandled;
}
size_t FileClient::remaining_bytes() {
threads::MutexLock lock(this->tickLock);
if(!this->pendingKey) return 0;
return this->pendingKey->size - this->bytesHandled;
}
void FileClient::disconnect(std::chrono::milliseconds timeout) {
auto own_lock = _this.lock();
if(!own_lock) return;
bool apply_flush = timeout.count() > 0;
debugMessage(LOG_FT, "{} Disconnecting client. Connection state: {} Flush IO: {}", this->client_prefix(), (int) this->state_connection, apply_flush);
unique_lock<threads::Mutex> tick_lock(this->tickLock);
if(this->state_connection == C_DISCONNECTED) return; /* we're already disconnected */
if(apply_flush && this->state_connection == C_DISCONNECTING) return; /* we're already flushing the IO */
if(this->ftType == FTType::TeaWeb_SSL_WS) {
this->ws_handler.disconnect(1000, "disconnected");
};
this->state_connection = apply_flush ? C_DISCONNECTING : C_DISCONNECTED;
if(this->readEvent)
event_del_noblock(this->readEvent);
if(apply_flush){
lock_guard flush_lock(this->thread_flush_lock);
assert(!this->thread_flush.joinable());
this->thread_flush = std::thread([own_lock, timeout]{
auto beg = system_clock::now();
while(own_lock->state_connection == C_DISCONNECTING && beg + timeout > system_clock::now()){
{
lock_guard buffer_lock(own_lock->bufferLock);
if(own_lock->read_queue.empty() && own_lock->write_queue.empty()) break;
}
usleep(10 * 1000);
}
unique_lock<threads::Mutex> l(own_lock->tickLock);
if(own_lock->state_connection != C_DISCONNECTING) return;
own_lock->disconnectFinal(l, true);
});
{
auto native_handle = this->thread_flush.native_handle();
pthread_setname_np(native_handle, "FileClient flush");
}
} else {
unique_lock flush_lock(this->thread_flush_lock);
if(this->thread_flush.joinable()) {
flush_lock.unlock();
this->thread_flush.join();
flush_lock.lock();
}
disconnectFinal(tick_lock, false);
}
}
void FileClient::disconnectFinal(unique_lock<threads::Mutex>& l, bool lock_flush_thread) {
auto ownLock = _this.lock();
unique_lock l1(this->bufferLock);
this->state_connection = C_DISCONNECTED;
{
unique_lock flush_lock(this->thread_flush_lock, try_to_lock);
if(flush_lock.owns_lock() || !lock_flush_thread) {
if(this->thread_flush.joinable()) {
if(this->thread_flush.get_id() == this_thread::get_id())
this->thread_flush.detach();
else
this->thread_flush.join();
}
}
}
if(this->readEvent){
auto event = this->readEvent;
this->readEvent = nullptr;
l1.unlock();
l.unlock();
event_del_block(event);
event_free(event);
l.lock();
l1.lock();
}
if(this->writeEvent){
auto event = this->writeEvent;
this->writeEvent = nullptr;
l1.unlock();
l.unlock();
event_del_block(event);
event_free(event);
l.lock();
l1.lock();
}
if(this->clientFd > 0) {
shutdown(this->clientFd, SHUT_RDWR);
close(this->clientFd);
}
this->clientFd = -1;
{
threads::MutexLock l2(this->handle->clientLock);
auto& clList = this->handle->connectedClients;
auto elm = find(clList.begin(), clList.end(), _this.lock());
if(elm != clList.end()) clList.erase(elm);
else logError(LOG_FT, "{} Invalid ft client list!", this->client_prefix());
}
this->read_queue.clear();
this->write_queue.clear();
if(this->fstream){
this->fstream->flush();
this->fstream->close();
delete this->fstream;
this->fstream = nullptr;
}
this->pendingKey = nullptr;
}
bool FileClient::tick() {
lock_guard<threads::Mutex> l(this->tickLock);
if(this->state_connection == C_DISCONNECTED) return false;
if(this->state_connection != C_DISCONNECTING) {
if(last_io_action.time_since_epoch().count() == 0)
last_io_action = system_clock::now();
else if(last_io_action + minutes(1) < system_clock::now()) {
logMessage(LOG_FT, "{} Timed out after one minute of silence!", this->client_prefix());
this->disconnect();
return false;
}
}
/* decode incomming stuff */
bool flag_reexecute = false;
{
/* types which require an extra layer of decode */
if(this->ftType == FTType::TeaWeb_SSL || this->ftType == FTType::TeaWeb_SSL_HTTP || this->ftType == FTType::TeaWeb_SSL_WS || this->ftType == FTType::TeaWeb_HTTP) {
pipes::buffer buffer;
{
lock_guard buffer_lock(this->bufferLock);
if(this->read_queue.empty()) {
flag_reexecute = false;
} else {
flag_reexecute = true;
buffer = this->read_queue.front();
this->read_queue.pop_front();
}
}
if(flag_reexecute) {
if(this->ftType == FTType::TeaWeb_HTTP)
this->handle_http_message(buffer);
else
this->ssl_handler.process_incoming_data(buffer);
}
} else if(this->ftType == FTType::TeamSpeak) {
flag_reexecute |= this->handle_ts_message();
} else if(this->ftType == FTType::Unknown) {
/* we need at least 16 bytes to detect any type */
if(this->availableBytes() >= 16) {
auto header = this->peekBytes(16);
if(header.find("GET") != -1 || header.find("POST") != -1 || header.find("OPTIONS") != -1) {
debugMessage(LOG_FT, "{} Using HTTP only!", this->client_prefix());
this->ftType = FTType::TeaWeb_HTTP;
return true;
} else if(pipes::SSL::isSSLHeader(header) && serverInstance->sslManager()->web_ssl_options()) {
debugMessage(LOG_FT, "{} Encrypting pipe with SSL", this->client_prefix());
this->ftType = FTType::TeaWeb_SSL;
this->ssl_handler.initialize(serverInstance->sslManager()->web_ssl_options());
return true;
} else {
debugMessage(LOG_FT, "{} Transferring data with the TS protocol", this->client_prefix());
this->ftType = FTType::TeamSpeak;
return true;
}
}
} else {
logError(LOG_FT, "{} Ticked client with unknown protocol type. Closing connection.", this->client_prefix());
this->disconnect();
return false;
}
}
if(this->state_transfer == T_TRANSFER) {
if(this->pendingKey) {
/* test for download */
if(!this->pendingKey->upload) {
/* lets send some data :) */
if(!fstream) {
logError(LOG_FT, "{} Missing file stream! Disconnecting client...", this->client_prefix());
this->disconnect();
return false;
}
size_t count = 0;
{
lock_guard lock(this->bufferLock);
count = this->write_queue.size();
}
if(count <= (1024 * 512) / BUFFER_SIZE){ //Max buffer 500Kb
if(!fstream->good()){
logError(LOG_FT, "{} Cant finish file download. File isn't good anymore!", this->client_prefix());
this->disconnect();
return false;
}
pipes::buffer writeBuffer(BUFFER_SIZE);
auto read = fstream->readsome((char*) writeBuffer.data_ptr(), writeBuffer.length());
if(read < 0){
logError(LOG_FT, "{} Invalid file read. Read {} bytes of max {}. Index {}/{}", this->client_prefix(), read, writeBuffer.length(), this->bytesHandled, this->pendingKey->size);
this->disconnect();
return false;
} else if(read == 0){
if(this->bytesHandled != this->pendingKey->size){
logError(LOG_FT, "{} Invalid end of file. Expected {} bytes and attempted to read at {}", this->client_prefix(), this->pendingKey->size, this->bytesHandled);
this->disconnect();
return false;
}
}
this->bytesHandled += read;
this->client->getConnectionStatistics()->logFileTransferOut(read);
if(read > 0){
writeBuffer.resize(read);
this->sendMessage(writeBuffer);
flag_reexecute |= true;
}
}
}
/* lets test if we're done */
if(this->bytesHandled == this->pendingKey->size){
auto time = duration_cast<milliseconds>(system_clock::now() - this->connect_timestamp).count();
logMessage(LOG_FT, "{} File transfer completed. Transferred {} bytes in {} milliseconds. Waiting for disconnect.", this->client_prefix(), this->bytesHandled, time);
this->close_file_handle();
this->pendingKey.reset();
this->state_transfer = T_DONE;
this->finished_timestamp = system_clock::now();
/* we expect TS3 to hangup the connection */
if(this->ftType != FTType::TeamSpeak)
this->disconnect(seconds(5));
return false;
}
}
} else if(this->state_transfer == T_DONE) {
if(this->finished_timestamp + seconds(2) < system_clock::now() && this->state_connection == C_CONNECTED) {
debugMessage(LOG_FT, "{} Disconnecting client after 2 seconds after finish!", this->client_prefix());
this->disconnect(seconds(5));
}
}
return flag_reexecute;
}
bool FileClient::applyKey(const string &key) {
{
threads::MutexLock lock(this->handle->keylock);
for(const auto& pkey : this->handle->pendingKeys) //needs a copy
if(pkey && pkey->key == key){
this->pendingKey = pkey;
this->handle->pendingKeys.erase(std::find(this->handle->pendingKeys.begin(), this->handle->pendingKeys.end(), pkey));
break;
}
}
if(!this->pendingKey){
logError(LOG_FT, "{} Tried to apply an non existing key! (Key: {})", this->client_prefix(), key);
this->disconnect();
return false;
}
this->client = this->pendingKey->owner.lock();
if(!this->client) {
logError(LOG_FT, "{} Tried connect with an invalid key (client offline)! (Key: {})", this->client_prefix(), key);
this->disconnect();
return false;
}
debugMessage(LOG_FT, "{} Initialized file transfer for file '{}' (Type: {})", this->client_prefix(), this->pendingKey->targetFile, pendingKey->upload ? "upload" : "download");
if(pendingKey->offset == 0 && pendingKey->upload) {
try {
fs::remove(fs::u8path(pendingKey->targetFile));
} catch (std::exception& e) {}
}
fstream = new std::fstream();
fstream->open(pendingKey->targetFile, (pendingKey->upload ? ios::out : ios::in) | ios::binary | ios::app);
if(!*fstream) {
logError(LOG_FT, "{} Failed to open target file {} for {}", this->client_prefix(), this->pendingKey->targetFile, pendingKey->upload ? "upload" : "download");
delete fstream;
this->fstream = nullptr;
return false;
}
fstream->seekg(0, std::ios::beg);
auto fsize = fstream->tellg();
fstream->seekg(0, std::ios::end);
fsize = fstream->tellg() - fsize;
fstream->seekg(this->pendingKey->offset, std::ios::beg);
if(!*fstream) {
logError(LOG_FT, "{} Failed to seek within file {} for {}", this->client_prefix(), this->pendingKey->targetFile, pendingKey->upload ? "upload" : "download");
delete fstream;
this->fstream = nullptr;
return false;
}
debugMessage(LOG_FT, "{} Received local file size {}. Target size if {}", this->client_prefix(), fsize, pendingKey->size);
if(this->state_transfer == T_INITIALIZE)
this->state_transfer = T_TRANSFER;
this->connect_timestamp = system_clock::now();
return true;
}
bool FileClient::uploadWriteBytes(const pipes::buffer_view& message) {
this->client->getConnectionStatistics()->logFileTransferIn(message.length());
if(!fstream->good()){
logError(LOG_FT, "{} uploadWriteBytes(...) called with invalid fstream!", this->client_prefix());
return false;
}
fstream->write(message.data_ptr<char>(), message.length());
if(!fstream->good()){
logError(LOG_FT, "{} Invalid file write! ({})", this->client_prefix(), message.length());
this->disconnect();
return false;
}
this->bytesHandled += message.length();
return true;
}
void FileClient::close_file_handle() {
if(this->fstream) {
this->fstream->flush();
this->fstream->close();
delete this->fstream;
this->fstream = nullptr;
}
}
void FileClient::sendMessage(const pipes::buffer_view& message) {
switch(this->ftType) {
case FTType::TeamSpeak:
case FTType::TeaWeb_HTTP:
this->sendRawMessage(message);
break;
case FTType::TeaWeb_SSL:
case FTType::TeaWeb_SSL_HTTP:
this->ssl_handler.send(message);
break;
case FTType::TeaWeb_SSL_WS:
this->ws_handler.send({pipes::BINARY, message.own_buffer()});
break;
default:
/* Dont log an error because the timeout disconnect function uses this without knowing which proto */
__asm__("nop");
return;
}
}
void FileClient::handle_ssl_message(const pipes::buffer_view &buffer) {
if(this->ftType == FTType::TeaWeb_SSL_HTTP)
this->handle_http_message(buffer);
else if(this->ftType == FTType::TeaWeb_SSL_WS)
this->ws_handler.process_incoming_data(buffer);
else if(this->ftType == FTType::TeaWeb_SSL) { /* lets detect if we have HTTP or WebSocket */
this->read_buffer += buffer;
this->handle_http_header();
}
}
void FileClient::handle_http_header() {
auto header_end = this->read_buffer.find("\r\n\r\n");
if(header_end == string::npos) {
if(this->read_buffer.length() > 1024 * 1024 * 4) {
this->read_buffer = pipes::buffer{};
logMessage(LOG_FT, "{} Client tried to fillup server memory. Disconnecting client.", this->client_prefix());
this->disconnect();
return;
}
return;
}
auto raw_request = this->read_buffer.view(0, header_end).string();
http::HttpRequest request{};
if(!http::parse_request(raw_request, request)) {
logError(LOG_FT, "{} Failed to parse HTTP request. Disconnecting client.", this->client_prefix());
this->disconnect();
return;
}
auto header_upgrade = request.findHeader("Upgrade");
if(header_upgrade && header_upgrade.values[0] == "websocket") {
debugMessage(LOG_FT, "{} Received WebSocket upgrade. Upgrading connection.", this->client_prefix(), raw_request);
if(this->ftType == FTType::TeaWeb_SSL)
this->ftType = FTType::TeaWeb_SSL_WS;
else {
//TODO: WebSocket only!
}
this->ws_handler.initialize();
this->ws_handler.process_incoming_data(this->read_buffer);
this->read_buffer = pipes::buffer{};
this->handle->tickFileClient(_this.lock()); /* we require a manual reticking */
return;
} else {
/* process the http request header */
if(this->ftType == FTType::TeaWeb_SSL)
this->ftType = FTType::TeaWeb_SSL_HTTP;
http::HttpResponse response{};
response.setHeader("Connection", {"close"}); /* close the connection instance, we dont want multiple requests */
response.setHeader("Access-Control-Allow-Methods", {"GET, POST"});
response.setHeader("Access-Control-Allow-Origin", {"*"});
response.setHeader("Access-Control-Allow-Headers", request.findHeader("Access-Control-Request-Headers").values); //access-control-allow-headers
response.setHeader("Access-Control-Max-Age", {"86400"});
response.setHeader("Access-Control-Expose-Headers", {"*, Content-Length, X-media-bytes, Content-Disposition"});
/* test for a preflight request: https://developer.mozilla.org/en-US/docs/Glossary/preflight_request */
auto request_method = request.findHeader("Access-Control-Request-Method");
if(request_method) {
debugMessage(LOG_FT, "{} Received preflight request. Sending close response with allow and let the client reconnect.", this->client_prefix());
response.code = http::code::_200;
this->read_buffer = this->read_buffer.range(header_end + 4);
auto raw_response = response.build();
this->sendMessage(pipes::buffer_view{raw_response.data(), raw_response.length()});
this->disconnect(seconds(5)); /* write our response & flush */
return;
} else {
auto transfer_key = request.findHeader("transfer-key");
auto download_name = request.findHeader("download-name");
if(!transfer_key) {
response.code = http::code::code(400, "Bad Request");
debugMessage(LOG_FT, "{} Received invalid HTTP request (Missing transfer key).", this->client_prefix());
} else {
if(!this->applyKey(transfer_key.values[0]) || !this->pendingKey) {
response.code = http::code::code(404, "Not Found");
debugMessage(LOG_FT, "{} Received invalid HTTP request (Invalid transfer key: {}).", this->client_prefix(), transfer_key.values[0]);
} else {
response.code = http::code::code(200, "OK");
response.setHeader("Content-Length", {to_string(this->pendingKey->size)});
if(!this->pendingKey->upload) {
response.setHeader("Content-type", {"application/octet-stream; "});
response.setHeader("Content-Transfer-Encoding", {"binary"});
response.setHeader("Content-Disposition", {"attachment; filename=\"" + http::encode_url(download_name ? download_name.values[0] : "TeaWeb Download") + "\""});
if(this->pendingKey->size > 1) {
char header_buffer[64];
auto read = fstream->readsome(header_buffer, 64);
if(read > 0)
response.setHeader("X-media-bytes", {base64::encode(header_buffer, read)});
fstream->seekg(this->pendingKey->offset, std::ios::beg);
}
}
}
}
if(!this->pendingKey || !this->pendingKey->upload) {
auto raw_response = response.build();
this->sendMessage(pipes::buffer_view{raw_response.data(), raw_response.length()});
}
if(response.code->code != 200) {
this->disconnect(seconds(5)) /* write our response & flush */;
return;
}
this->http_init = true;
auto overhead = this->read_buffer.range(header_end + 4);
this->read_buffer = pipes::buffer{}; /* reset the read buffer */
if(overhead.length() > 0)
this->handle_http_message(overhead);
this->handle->tickFileClient(_this.lock()); /* we require a manual reticking */
}
}
}
bool FileClient::handle_ts_message() {
if(this->state_transfer == T_INITIALIZE) {
if(this->availableBytes() >= 16) {
return this->applyKey(this->getBytes(16));
} else
return false;
} else if(this->state_transfer == T_TRANSFER) {
if(!this->pendingKey)
return false;
if(!this->pendingKey->upload)
return false; /* should never happen! */
bool reexecute = false;
pipes::buffer buffer;
{
lock_guard buffer_lock(this->bufferLock);
if(this->read_queue.empty())
return false; /* nothing to upload */
buffer = this->read_queue.front();
this->read_queue.pop_front();
reexecute |= !this->read_queue.empty();
}
if(!this->uploadWriteBytes(buffer))
return false; /* error already handeled by uploadWriteBytes(...) */
return reexecute;
}
return false;
}
void FileClient::handle_ws_message(const pipes::WSMessage &message) {
if(this->state_transfer == T_INITIALIZE) {
debugMessage(LOG_FT, "{} Received transfer key: {}", this->client_prefix(), message.data.string());
if(this->applyKey(message.data.string()))
this->handle->tickFileClient(_this.lock()); /* we require a manual reticking */
return;
} else if(this->state_transfer == T_TRANSFER) {
if(this->pendingKey->upload)
this->uploadWriteBytes(message.data);
else {
logError(LOG_FT, "{} Invalid write (Just download)", this->client_prefix());
this->disconnect();
}
}
}
void FileClient::handle_http_message(const pipes::buffer_view &message) {
if(!this->http_init) {
this->read_buffer += message;
this->handle_http_header();
return;
}
if(!https_upload_init) {
//------WebKitFormBoundaryaWP8XAzMBnMOJznv\r\nContent-Disposition: form-data; name=\"file\"; filename=\"blob\"\r\nContent-Type: application/octet-stream\r\n\r\n
this->read_buffer += message;
auto header_end = this->read_buffer.find("\r\n\r\n");
if(header_end == string::npos) {
if(this->read_buffer.length() > 1024 * 1024 * 4) {
this->read_buffer = pipes::buffer{};
logMessage(LOG_FT, "{} Client tried to fillup server memory. Disconnecting client.", this->client_prefix());
this->disconnect();
return;
}
return;
}
https_upload_init = true;
auto overhead = this->read_buffer.view(header_end + 4);
this->read_buffer = pipes::buffer{};
if(!overhead.empty())
this->handle_http_message(overhead);
return;
}
if(!this->pendingKey || !this->pendingKey->upload) {
logError(LOG_FT, "{} HTTP Invalid request", this->client_prefix());
return;
}
auto bytes_to_write = this->pendingKey->size - this->bytesHandled; /* ignore boundaries */
if(bytes_to_write < message.length())
this->uploadWriteBytes(message.view(0, bytes_to_write));
else
this->uploadWriteBytes(message);
if(this->bytesHandled == this->pendingKey->size){
http::HttpResponse response{};
response.setHeader("Connection", {"close"}); /* close the connection instance, we dont want multiple requests */
response.setHeader("Access-Control-Allow-Methods", {"GET, POST"});
response.setHeader("Access-Control-Allow-Origin", {"*"});
response.setHeader("Access-Control-Allow-Headers", {"*"});
response.setHeader("Access-Control-Max-Age", {"86400"});
response.setHeader("Access-Control-Expose-Headers", {"*"});
auto raw_response = response.build();
this->sendMessage(pipes::buffer_view{raw_response.data(), raw_response.length()});
}
}
server/src/client/file/FileClient.h
-133
View File
@@ -1,133 +0,0 @@
#pragma once
#include <protocol/buffers.h>
#include <poll.h>
#include <fstream>
#include <src/server/file/LocalFileServer.h>
#include <event.h>
#include <pipes/ws.h>
#include <pipes/ssl.h>
#include "src/VirtualServer.h"
namespace ts {
namespace server {
class ConnectedClient;
class ConnectedClient;
class LocalFileServer;
enum FTType {
Unknown,
TeamSpeak,
TeaWeb_SSL, /* not yet decided if the protocol is HTTP or WebSocket */
TeaWeb_SSL_WS,
TeaWeb_SSL_HTTP,
TeaWeb_HTTP
};
class FileClient {
friend class LocalFileServer;
public:
enum TransferState {
T_INITIALIZE,
T_TRANSFER,
T_DONE
};
enum ConnectionState {
C_CONNECTED,
C_DISCONNECTING,
C_DISCONNECTED
};
struct BandwidthEntry {
std::chrono::system_clock::time_point timestamp;
uint16_t length = 0;
};
FileClient(LocalFileServer* handle, int socketFd);
~FileClient();
void disconnect(std::chrono::milliseconds = std::chrono::milliseconds(5000));
FTType getFTType(){ return this->ftType; }
size_t used_bandwidth();
std::string client_prefix();
size_t transferred_bytes();
size_t remaining_bytes();
protected:
void disconnectFinal(std::unique_lock<threads::Mutex>& /* tick lock */, bool /* handle flush thread */);
bool tick();
bool uploadWriteBytes(const pipes::buffer_view&);
void close_file_handle();
bool applyKey(const std::string &);
void sendMessage(const pipes::buffer_view&);
//Direct methods & IO stuff
void sendRawMessage(const pipes::buffer_view&);
void handleMessageRead(int, short, void*);
void handleMessageWrite(int, short, void*);
void handle_ssl_message(const pipes::buffer_view&); /* handeles all decoded SSL messages */
/* http header parser. header must be stored with read buffer! */
void handle_http_header();
/* Final protocol handlers */
void handle_http_message(const pipes::buffer_view&);
void handle_ws_message(const pipes::WSMessage&);
bool handle_ts_message();
private:
LocalFileServer* handle;
std::weak_ptr<FileClient> _this;
std::recursive_mutex bandwidth_lock;
std::deque<std::unique_ptr<BandwidthEntry>> bandwidth;
sockaddr_storage remote_address;
int clientFd;
bool event_read_hold = false;
::event* readEvent = nullptr;
bool event_write_hold = false;
::event* writeEvent = nullptr;
threads::Mutex tickLock;
std::recursive_timed_mutex bufferLock;
std::deque<pipes::buffer> write_queue;
std::deque<pipes::buffer> read_queue;
pipes::buffer read_buffer; /* buffer which contains fragments of decoded data, e.g. HTTP request. Access only within tickLock locked */
FTType ftType = FTType::Unknown;
pipes::WebSocket ws_handler;
pipes::SSL ssl_handler;
bool https_upload_init = false;
bool http_init = false; /* general flag if the HTTP header has been parsed */
std::shared_ptr<ConnectedClient> client;
std::shared_ptr<file::FileTransfereKey> pendingKey = nullptr;
std::chrono::time_point<std::chrono::system_clock> last_io_action;
std::chrono::time_point<std::chrono::system_clock> connect_timestamp;
std::chrono::time_point<std::chrono::system_clock> finished_timestamp;
std::fstream* fstream = nullptr;
size_t bytesHandled = 0;
ConnectionState state_connection = ConnectionState::C_CONNECTED;
TransferState state_transfer = TransferState::T_INITIALIZE;
size_t availableBytes();
std::string getBytes(size_t);
std::string peekBytes(size_t);
std::mutex thread_flush_lock;
std::thread thread_flush;
};
}
}
server/src/client/file/FileClientIO.cpp
-248
View File
@@ -1,248 +0,0 @@
#include <algorithm>
#include <memory>
#include <src/server/file/LocalFileServer.h>
#include <log/LogUtils.h>
#include <misc/std_unique_ptr.h>
#include <pipes/buffer.h>
#include "FileClient.h"
using namespace std;
using namespace std::chrono;
using namespace ts::server;
void FileClient::sendRawMessage(const pipes::buffer_view &message) {
lock_guard lock(this->bufferLock);
this->write_queue.push_back(message.own_buffer());
if(!this->event_write_hold)
event_add(this->writeEvent, nullptr);
else {
__asm__("nop");
}
}
void FileClient::handleMessageWrite(int fd, short, void *) {
auto self = this->_this.lock();
if(!self) return;
decltype(this->pendingKey) pending_key;
{
lock_guard<threads::Mutex> l(this->tickLock);
pending_key = this->pendingKey;
}
unique_lock lock(this->bufferLock, defer_lock);
if(!lock.try_lock_for(milliseconds(5))) {
logWarning(LOG_FT, "{} Buffer lock locked, could not write data!", this->client_prefix());
return; /* somebody else doing a action and will (hopefully) readd the event */
}
if(self->state_connection == C_DISCONNECTED) return;
auto used = this->used_bandwidth();
if(pending_key) { //Delay the write <3 :P
if(pending_key->max_bandwhidth >= 0 && used > pending_key->max_bandwhidth) {
event_write_hold = true;
logTrace(LOG_FT, "{} Exceeded bandwidth limit of {} bytes (Used {} bytes). Temporary skipping write event!", this->client_prefix(), pending_key->max_bandwhidth, this->used_bandwidth());
return;
}
}
event_write_hold = false;
pipes::buffer* buffer = nullptr;
while(true) {
if(this->write_queue.empty()) {
lock.unlock(); /* unlock write buffer because we're ticking */
if(pending_key && !pending_key->upload)
this->handle->tickFileClient(_this.lock()); //We have to fill up again
return;
}
buffer = &this->write_queue.front();
if(buffer->empty()) {
this->write_queue.pop_front();
buffer = nullptr;
continue;
}
break;
}
ssize_t length = buffer->length();
if(pending_key && pending_key->max_bandwhidth >= 0) {
if(pending_key->max_bandwhidth < length && pending_key->max_bandwhidth > used) length = pending_key->max_bandwhidth - used;
}
length = send(fd, buffer->data_ptr(), length, MSG_NOSIGNAL);
//logTrace(LOG_FT, "{} Wrote {} bytes", this->client_prefix(), length);
if(length == -1) {
if (errno == EINTR || errno == EAGAIN) {
event_add(this->writeEvent, nullptr);
return;
}
else {
logError(LOG_FT, "{} Failed to write some data! ({}/{})", this->client_prefix(), errno, strerror(errno));
lock.unlock();
self->disconnect();
return;
}
} else {
*buffer = buffer->range(length);
auto entry = make_unique<BandwidthEntry>();
entry->length = length ;
entry->timestamp = system_clock::now();
{
lock_guard<recursive_mutex> b_lock(this->bandwidth_lock);
this->bandwidth.push_back(move(entry));
}
}
last_io_action = system_clock::now();
if(buffer->empty()) {
this->write_queue.pop_front();
buffer = nullptr;
}
if(pending_key && this->bytesHandled != pending_key->size) {
if(this->write_queue.size() < 4) this->handle->tickFileClient(_this.lock());
}
if(!this->write_queue.empty()) {
event_add(this->writeEvent, nullptr);
}
self.reset();
}
void FileClient::handleMessageRead(int fd, short, void *) {
auto self = this->_this.lock();
if(self->state_connection != C_CONNECTED) return;
decltype(this->pendingKey) pending_key;
{
lock_guard<threads::Mutex> l(this->tickLock);
pending_key = this->pendingKey;
}
size_t buffer_length = 1024;
if(pending_key && pending_key->max_bandwhidth >= 0) {
auto used = this->used_bandwidth();
if(used < pending_key->max_bandwhidth) {
buffer_length = pending_key->max_bandwhidth - used;
if(buffer_length > 1024) buffer_length = 1024;
} else {
logTrace(LOG_FT, "{} Exceeded bandwidth limit of {} bytes (Used {} bytes). Temporary removing read event!", this->client_prefix(), pending_key->max_bandwhidth, used);
{
lock_guard lock(this->bufferLock);
if(this->readEvent)
event_del_noblock(this->readEvent);
}
this->event_read_hold = true;
return;
}
}
pipes::buffer buffer(buffer_length);
auto length = recv(fd, buffer.data_ptr(), buffer.length(), 0);
if(length < 0){
if(errno == EINTR || errno == EAGAIN)
;//event_add(this->readEvent, nullptr);
else {
{
lock_guard lock(this->bufferLock);
if(this->readEvent)
event_del_noblock(this->readEvent);
}
if(this->state_connection == C_CONNECTED) {
logError(LOG_FT, "{} Failed to read some data! ({}/{})", this->client_prefix(), errno, strerror(errno));
self->disconnect();
}
}
return;
} else if(length == 0){
{
lock_guard lock(this->bufferLock);
if(this->readEvent)
event_del_noblock(this->readEvent);
}
if(this->state_connection == C_CONNECTED) {
if(this->state_transfer == T_TRANSFER)
logWarning(LOG_FT, "{} Transfer hang up. Remote peer closed the connection.", this->client_prefix());
else
logMessage(LOG_FT, "{} Remote peer has closed the connection before initializing a transfer.", this->client_prefix());
self->disconnect(seconds(3));
}
return;
}
last_io_action = system_clock::now();
buffer.resize(length);
{
lock_guard lock(this->bufferLock);
if(self->state_connection != C_CONNECTED) return; /* drop the buffer because we're not connected anymore */
this->read_queue.push_back(std::move(buffer));
}
this->handle->tickFileClient(_this.lock());
auto entry = make_unique<BandwidthEntry>();
entry->length = length ;
entry->timestamp = system_clock::now();
//logTrace(LOG_FT, "{} Readed {} bytes", this->client_prefix(), length);
{
lock_guard<recursive_mutex> lock(this->bandwidth_lock);
this->bandwidth.push_back(move(entry));
}
}
size_t FileClient::availableBytes() {
lock_guard lock(this->bufferLock);
size_t available = 0;
for(const auto& buf : this->read_queue)
available += buf.length();
return available;
}
std::string FileClient::peekBytes(size_t size) {
ssize_t required = size;
lock_guard lock(this->bufferLock);
string result;
result.reserve(size);
for(pipes::buffer& buf : this->read_queue) {
if(required <= 0) break;
if(buf.length() > required) {
result += string(buf.data_ptr<const char>(), required);
required = 0;
} else {
result += string(buf.data_ptr<const char>(), buf.length());
required -= buf.length();
}
}
return result;
}
std::string FileClient::getBytes(size_t size) {
lock_guard lock(this->bufferLock);
string result;
result.reserve(size);
while(!this->read_queue.empty()) {
if(size <= 0) break;
auto& buf = this->read_queue.front();
if(buf.length() > size) {
result += string((char*) buf.pipes::buffer_view::data_ptr(), size);
buf = buf.range(size);
size = 0;
} else {
result += string((char*) buf.data_ptr(), buf.length());
size -= buf.length();
this->read_queue.pop_front();
}
}
return result;
}
server/src/client/music/internal_provider/channel_replay/ChannelProvider.cpp
+8 -2
View File
@@ -1,7 +1,6 @@
#include "ChannelProvider.h"
#include "../../MusicClient.h"
#include "../../../../InstanceHandler.h"
#include "src/server/file/LocalFileServer.h"
#include "../../../../../../music/providers/ffmpeg/FFMpegProvider.h"
@@ -22,6 +21,7 @@ threads::Future<std::shared_ptr<::music::MusicPlayer>> ChannelProvider::createPl
auto server = ((VirtualServer*) ptr_server)->ref();
threads::Future<std::shared_ptr<::music::MusicPlayer>> future;
#if 0
if(server) {
std::thread([future, server, url, ptr_server]{
auto f_server = serverInstance->getFileServer();
@@ -116,7 +116,9 @@ threads::Future<std::shared_ptr<::music::MusicPlayer>> ChannelProvider::createPl
} else {
future.executionFailed("invalid bot");
}
#else
future.executionFailed("channel file playback is currently not supported");
#endif
return future;
}
@@ -165,6 +167,7 @@ threads::Future<shared_ptr<UrlInfo>> ChannelProvider::query_info(const std::stri
auto server = ((VirtualServer*) ptr_server)->ref();
threads::Future<shared_ptr<UrlInfo>> future;
#if 0
if(server) {
std::thread([future, server, url, ptr_server]{
auto f_server = serverInstance->getFileServer();
@@ -264,6 +267,9 @@ threads::Future<shared_ptr<UrlInfo>> ChannelProvider::query_info(const std::stri
future.executionFailed("invalid bot");
}
#else
future.executionFailed("channel file playback is currently not supported");
#endif
return future;
}