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

Some file transfer updates

Browse Source
This commit is contained in:
WolverinDEV 2020-05-10 16:23:02 +02:00
parent dd4a871bf0
commit 4e3921502d
12 changed files with 742 additions and 233 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
file/local_server/LocalFileProvider.cpp
View File

@ -8,10 +8,58 @@
using namespace ts::server; using namespace ts::server;
using LocalFileServer = file::LocalFileProvider; using LocalFileServer = file::LocalFileProvider;
EVP_PKEY* ssl_generate_key() {
auto key = std::unique_ptr<EVP_PKEY, decltype(&EVP_PKEY_free)>(EVP_PKEY_new(), ::EVP_PKEY_free);
auto rsa = RSA_new();
auto e = std::unique_ptr<BIGNUM, decltype(&BN_free)>(BN_new(), ::BN_free);
BN_set_word(e.get(), RSA_F4);
if(!RSA_generate_key_ex(rsa, 2048, e.get(), nullptr)) return nullptr;
EVP_PKEY_assign_RSA(key.get(), rsa);
return key.release();
}
X509* ssl_generate_certificate(EVP_PKEY* key) {
auto cert = X509_new();
X509_set_pubkey(cert, key);
ASN1_INTEGER_set(X509_get_serialNumber(cert), 3);
X509_gmtime_adj(X509_get_notBefore(cert), 0);
X509_gmtime_adj(X509_get_notAfter(cert), 31536000L);
X509_NAME* name = nullptr;
name = X509_get_subject_name(cert);
//for(const auto& subject : this->subjects)
// X509_NAME_add_entry_by_txt(name, subject.first.c_str(), MBSTRING_ASC, (unsigned char *) subject.second.c_str(), subject.second.length(), -1, 0);
X509_set_subject_name(cert, name);
name = X509_get_issuer_name(cert);
//for(const auto& subject : this->issues)
// X509_NAME_add_entry_by_txt(name, subject.first.c_str(), MBSTRING_ASC, (unsigned char *) subject.second.c_str(), subject.second.length(), -1, 0);
X509_set_issuer_name(cert, name);
X509_sign(cert, key, EVP_sha512());
return cert;
}
std::shared_ptr<LocalFileServer> server_instance{}; std::shared_ptr<LocalFileServer> server_instance{};
bool file::initialize(std::string &error) { bool file::initialize(std::string &error) {
server_instance = std::make_shared<LocalFileProvider>(); server_instance = std::make_shared<LocalFileProvider>();
if(!server_instance->initialize(error)) {
auto options = std::make_shared<pipes::SSL::Options>();
options->verbose_io = true;
options->context_method = SSLv23_method();
options->free_unused_keypairs = false;
{
std::shared_ptr<EVP_PKEY> pkey{ssl_generate_key(), ::EVP_PKEY_free};
std::shared_ptr<X509> cert{ssl_generate_certificate(&*pkey), ::X509_free};
options->default_keypair({pkey, cert});
}
if(!server_instance->initialize(error, options)) {
server_instance = nullptr; server_instance = nullptr;
return false; return false;
} }
@ -32,7 +80,7 @@ std::shared_ptr<file::AbstractFileServer> file::server() {
LocalFileServer::LocalFileProvider() : file_system_{}, file_transfer_{this->file_system_} {} LocalFileServer::LocalFileProvider() : file_system_{}, file_transfer_{this->file_system_} {}
LocalFileServer::~LocalFileProvider() {} LocalFileServer::~LocalFileProvider() {}
bool LocalFileServer::initialize(std::string &error) { bool LocalFileServer::initialize(std::string &error, const std::shared_ptr<pipes::SSL::Options>& ssl_options) {
if(!this->file_system_.initialize(error, "file-root/")) if(!this->file_system_.initialize(error, "file-root/"))
return false; return false;
@ -51,7 +99,7 @@ bool LocalFileServer::initialize(std::string &error) {
bindings.push_back(std::move(binding)); bindings.push_back(std::move(binding));
} }
if(!this->file_transfer_.start(bindings)) { if(!this->file_transfer_.start(bindings, ssl_options)) {
error = "transfer server startup failed"; error = "transfer server startup failed";
this->file_system_.finalize(); this->file_system_.finalize();
return false; return false;

69
file/local_server/LocalFileProvider.h
View File

@ -164,6 +164,12 @@ namespace ts::server::file {
if(this->bytes_send < this->max_bytes) return this->max_bytes - this->bytes_send; if(this->bytes_send < this->max_bytes) return this->max_bytes - this->bytes_send;
return 0; return 0;
} }
[[nodiscard]] inline std::chrono::milliseconds expected_writing_time(size_t bytes) const {
if(this->max_bytes <= 0) return std::chrono::milliseconds{0};
return std::chrono::seconds{bytes / (this->max_bytes * (1000 / kThrottleTimespanMs))};
}
}; };
/* all variables are locked via the state_mutex */ /* all variables are locked via the state_mutex */
@ -185,13 +191,17 @@ namespace ts::server::file {
PROTOCOL_TS_V1 PROTOCOL_TS_V1
}; };
enum HTTPUploadState {
HTTP_AWAITING_HEADER,
HTTP_STATE_AWAITING_BOUNDARY,
HTTP_STATE_UPLOADING,
HTTP_STATE_DOWNLOADING
};
struct { struct {
bool file_locked{false}; bool file_locked{false};
std::string absolute_path{}; std::string absolute_path{};
#if 0
struct event* io_process{nullptr}; /* either a read event or write event */
#endif
int file_descriptor{0}; int file_descriptor{0};
bool currently_processing{false}; bool currently_processing{false};
@ -231,6 +241,9 @@ namespace ts::server::file {
NetworkThrottle throttle; NetworkThrottle throttle;
pipes::SSL pipe_ssl{}; pipes::SSL pipe_ssl{};
bool pipe_ssl_init{false};
std::unique_ptr<Buffer, decltype(free_buffer)*> http_header_buffer{nullptr, free_buffer};
HTTPUploadState http_state{HTTPUploadState::HTTP_AWAITING_HEADER};
/* Only read the transfer key length at the beginning. We than have the actual limit which will be set via throttle */ /* Only read the transfer key length at the beginning. We than have the actual limit which will be set via throttle */
size_t max_read_buffer_size{TRANSFER_KEY_LENGTH}; size_t max_read_buffer_size{TRANSFER_KEY_LENGTH};
@ -272,6 +285,7 @@ namespace ts::server::file {
void add_network_read_event(bool /* ignore bandwidth limits */); void add_network_read_event(bool /* ignore bandwidth limits */);
bool send_file_bytes(const void* /* buffer */, size_t /* length */); bool send_file_bytes(const void* /* buffer */, size_t /* length */);
bool enqueue_buffer_bytes(const void* /* buffer */, size_t /* length */);
[[nodiscard]] inline std::string log_prefix() const { return "[" + net::to_string(this->networking.address) + "]"; } [[nodiscard]] inline std::string log_prefix() const { return "[" + net::to_string(this->networking.address) + "]"; }
}; };
@ -347,7 +361,29 @@ namespace ts::server::file {
enum struct TransferKeyApplyResult { enum struct TransferKeyApplyResult {
SUCCESS, SUCCESS,
FILE_ERROR, FILE_ERROR,
UNKNOWN_KEY UNKNOWN_KEY,
INTERNAL_ERROR
};
enum struct TransferUploadRawResult {
MORE_DATA_TO_RECEIVE,
FINISH,
/* UNKNOWN ERROR */
};
enum struct TransferUploadHTTPResult {
MORE_DATA_TO_RECEIVE,
FINISH,
BOUNDARY_MISSING,
BOUNDARY_TOKEN_INVALID,
BOUNDARY_INVALID,
MISSING_CONTENT_TYPE,
INVALID_CONTENT_TYPE
/* UNKNOWN ERROR */
}; };
struct NetworkBinding : std::enable_shared_from_this<NetworkBinding> { struct NetworkBinding : std::enable_shared_from_this<NetworkBinding> {
@ -365,9 +401,17 @@ namespace ts::server::file {
explicit LocalFileTransfer(filesystem::LocalFileSystem&); explicit LocalFileTransfer(filesystem::LocalFileSystem&);
~LocalFileTransfer(); ~LocalFileTransfer();
[[nodiscard]] bool start(const std::deque<std::shared_ptr<NetworkBinding>>& /* bindings */); [[nodiscard]] bool start(const std::deque<std::shared_ptr<NetworkBinding>>& /* bindings */, const std::shared_ptr<pipes::SSL::Options>& /* ssl options */);
void stop(); void stop();
[[nodiscard]] inline const auto& ssl_options() const {
return this->ssl_options_;
}
inline void set_ssl_options(const std::shared_ptr<pipes::SSL::Options>& options) {
this->ssl_options_ = options;
}
std::shared_ptr<ExecuteResponse<TransferInitError, std::shared_ptr<Transfer>>> std::shared_ptr<ExecuteResponse<TransferInitError, std::shared_ptr<Transfer>>>
initialize_channel_transfer(Transfer::Direction direction, ServerId id, ChannelId channelId, initialize_channel_transfer(Transfer::Direction direction, ServerId id, ChannelId channelId,
const TransferInfo &info) override; const TransferInfo &info) override;
@ -390,7 +434,6 @@ namespace ts::server::file {
filesystem::LocalFileSystem& file_system_; filesystem::LocalFileSystem& file_system_;
std::atomic<transfer_id> current_transfer_id{0}; std::atomic<transfer_id> current_transfer_id{0};
std::mt19937 transfer_random_token_generator{std::random_device{}()}; std::mt19937 transfer_random_token_generator{std::random_device{}()};
std::mutex result_notify_mutex{}; std::mutex result_notify_mutex{};
@ -400,6 +443,8 @@ namespace ts::server::file {
std::deque<std::shared_ptr<FileClient>> transfers_{}; std::deque<std::shared_ptr<FileClient>> transfers_{};
std::deque<std::shared_ptr<Transfer>> pending_transfers{}; std::deque<std::shared_ptr<Transfer>> pending_transfers{};
std::shared_ptr<pipes::SSL::Options> ssl_options_{};
enum ServerState { enum ServerState {
STOPPED, STOPPED,
RUNNING RUNNING
@ -457,10 +502,17 @@ namespace ts::server::file {
[[nodiscard]] FileInitializeResult initialize_file_io(const std::shared_ptr<FileClient>& /* client */); [[nodiscard]] FileInitializeResult initialize_file_io(const std::shared_ptr<FileClient>& /* client */);
void finalize_file_io(const std::shared_ptr<FileClient>& /* client */, std::unique_lock<std::shared_mutex>& /* state lock */); void finalize_file_io(const std::shared_ptr<FileClient>& /* client */, std::unique_lock<std::shared_mutex>& /* state lock */);
[[nodiscard]] bool initialize_client_ssl(const std::shared_ptr<FileClient>& /* client */);
void finalize_client_ssl(const std::shared_ptr<FileClient>& /* client */);
void enqueue_disk_io(const std::shared_ptr<FileClient>& /* client */); void enqueue_disk_io(const std::shared_ptr<FileClient>& /* client */);
void execute_disk_io(const std::shared_ptr<FileClient>& /* client */); void execute_disk_io(const std::shared_ptr<FileClient>& /* client */);
void report_transfer_statistics(const std::shared_ptr<FileClient>& /* client */); void report_transfer_statistics(const std::shared_ptr<FileClient>& /* client */);
[[nodiscard]] TransferUploadRawResult handle_transfer_upload_raw(const std::shared_ptr<FileClient>& /* client */, const char * /* buffer */, size_t /* length */);
[[nodiscard]] TransferUploadHTTPResult handle_transfer_upload_http(const std::shared_ptr<FileClient>& /* client */, const char * /* buffer */, size_t /* length */);
void send_http_response(const std::shared_ptr<FileClient>& /* client */, http::HttpResponse& /* response */);
static void callback_transfer_network_write(int, short, void*); static void callback_transfer_network_write(int, short, void*);
static void callback_transfer_network_read(int, short, void*); static void callback_transfer_network_read(int, short, void*);
@ -472,7 +524,8 @@ namespace ts::server::file {
static void dispatch_loop_disk_io(void*); static void dispatch_loop_disk_io(void*);
size_t handle_transfer_read(const std::shared_ptr<FileClient>& /* client */, const char* /* buffer */, size_t /* bytes */); size_t handle_transfer_read(const std::shared_ptr<FileClient>& /* client */, const char* /* buffer */, size_t /* bytes */);
[[nodiscard]] TransferKeyApplyResult handle_transfer_key_provided(const std::shared_ptr<FileClient>& /* client */); size_t handle_transfer_read_raw(const std::shared_ptr<FileClient>& /* client */, const char* /* buffer */, size_t /* bytes */);
[[nodiscard]] TransferKeyApplyResult handle_transfer_key_provided(const std::shared_ptr<FileClient>& /* client */, std::string& /* error */);
}; };
} }
@ -481,7 +534,7 @@ namespace ts::server::file {
LocalFileProvider(); LocalFileProvider();
virtual ~LocalFileProvider(); virtual ~LocalFileProvider();
[[nodiscard]] bool initialize(std::string& /* error */); [[nodiscard]] bool initialize(std::string& /* error */, const std::shared_ptr<pipes::SSL::Options>& /* ssl options */);
void finalize(); void finalize();
filesystem::AbstractProvider &file_system() override; filesystem::AbstractProvider &file_system() override;

6
file/local_server/LocalFileTransfer.cpp
View File

@ -16,6 +16,7 @@ Buffer* transfer::allocate_buffer(size_t size) {
auto total_size = sizeof(Buffer) + size; auto total_size = sizeof(Buffer) + size;
auto buffer = (Buffer*) malloc(total_size); auto buffer = (Buffer*) malloc(total_size);
new (buffer) Buffer{}; new (buffer) Buffer{};
buffer->capacity = size;
return buffer; return buffer;
} }
@ -45,7 +46,10 @@ FileClient::~FileClient() {
LocalFileTransfer::LocalFileTransfer(filesystem::LocalFileSystem &fs) : file_system_{fs} {} LocalFileTransfer::LocalFileTransfer(filesystem::LocalFileSystem &fs) : file_system_{fs} {}
LocalFileTransfer::~LocalFileTransfer() = default; LocalFileTransfer::~LocalFileTransfer() = default;
bool LocalFileTransfer::start(const std::deque<std::shared_ptr<NetworkBinding>>& bindings) { bool LocalFileTransfer::start(const std::deque<std::shared_ptr<NetworkBinding>>& bindings, const std::shared_ptr<pipes::SSL::Options>& ssl_options) {
assert(ssl_options);
this->ssl_options_ = ssl_options;
(void) this->start_client_worker(); (void) this->start_client_worker();
{ {

19
file/local_server/LocalFileTransferClientWorker.cpp
View File

@ -6,6 +6,7 @@
#include <event2/event.h> #include <event2/event.h>
#include <log/LogUtils.h> #include <log/LogUtils.h>
#include "./LocalFileProvider.h" #include "./LocalFileProvider.h"
#include "LocalFileProvider.h"
using namespace ts::server::file; using namespace ts::server::file;
using namespace ts::server::file::transfer; using namespace ts::server::file::transfer;
@ -45,7 +46,17 @@ void LocalFileTransfer::disconnect_client(const std::shared_ptr<FileClient> &cli
client->state = flush ? FileClient::STATE_DISCONNECTING : FileClient::STATE_DISCONNECTED; client->state = flush ? FileClient::STATE_DISCONNECTING : FileClient::STATE_DISCONNECTED;
client->timings.disconnecting = std::chrono::system_clock::now(); client->timings.disconnecting = std::chrono::system_clock::now();
if(flush) { if(flush) {
if(client->transfer->direction == Transfer::DIRECTION_UPLOAD) { const auto network_flush_time = client->networking.throttle.expected_writing_time(client->buffer.bytes) + std::chrono::seconds{10};
if(!client->transfer) {
del_ev_noblock(client->networking.event_read);
del_ev_noblock(client->networking.event_throttle);
client->add_network_write_event_nolock(false);
/* max flush 10 seconds */
client->networking.disconnect_timeout = std::chrono::system_clock::now() + network_flush_time;
debugMessage(LOG_FT, "{} Disconnecting client. Flushing pending bytes (max {} seconds)", client->log_prefix(), std::chrono::floor<std::chrono::seconds>(network_flush_time).count());
} else if(client->transfer->direction == Transfer::DIRECTION_UPLOAD) {
del_ev_noblock(client->networking.event_read); del_ev_noblock(client->networking.event_read);
del_ev_noblock(client->networking.event_write); del_ev_noblock(client->networking.event_write);
del_ev_noblock(client->networking.event_throttle); del_ev_noblock(client->networking.event_throttle);
@ -57,7 +68,8 @@ void LocalFileTransfer::disconnect_client(const std::shared_ptr<FileClient> &cli
client->add_network_write_event_nolock(false); client->add_network_write_event_nolock(false);
/* max flush 10 seconds */ /* max flush 10 seconds */
client->networking.disconnect_timeout = std::chrono::system_clock::now() + std::chrono::seconds{10}; client->networking.disconnect_timeout = std::chrono::system_clock::now() + network_flush_time;
debugMessage(LOG_FT, "{} Disconnecting client. Flushing pending bytes (max {} seconds)", client->log_prefix(), std::chrono::floor<std::chrono::seconds>(network_flush_time).count());
} }
} else { } else {
del_ev_noblock(client->networking.event_read); del_ev_noblock(client->networking.event_read);
@ -146,6 +158,8 @@ void LocalFileTransfer::dispatch_loop_client_worker(void *ptr_transfer) {
return t->timings.last_write + std::chrono::seconds{5} < now; return t->timings.last_write + std::chrono::seconds{5} < now;
} }
} else if(t->state == FileClient::STATE_DISCONNECTING) { } else if(t->state == FileClient::STATE_DISCONNECTING) {
if(t->networking.disconnect_timeout.time_since_epoch().count() > 0)
return t->networking.disconnect_timeout + std::chrono::seconds{5} < now;
return t->timings.disconnecting + std::chrono::seconds{30} < now; return t->timings.disconnecting + std::chrono::seconds{30} < now;
} }
return false; return false;
@ -175,6 +189,7 @@ void LocalFileTransfer::dispatch_loop_client_worker(void *ptr_transfer) {
std::unique_lock slock{client->state_mutex}; std::unique_lock slock{client->state_mutex};
client->state = FileClient::STATE_DISCONNECTED; client->state = FileClient::STATE_DISCONNECTED;
provider->finalize_file_io(client, slock); provider->finalize_file_io(client, slock);
provider->finalize_client_ssl(client);
provider->finalize_networking(client, slock); provider->finalize_networking(client, slock);
} }

2
file/local_server/LocalFileTransferDisk.cpp
View File

@ -408,6 +408,8 @@ void LocalFileTransfer::execute_disk_io(const std::shared_ptr<FileClient> &clien
client->add_network_read_event(false); client->add_network_read_event(false);
} }
} else if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD) { } else if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD) {
if(client->state == FileClient::STATE_DISCONNECTING) return;
while(true) { while(true) {
constexpr auto buffer_capacity{4096}; constexpr auto buffer_capacity{4096};
char buffer[buffer_capacity]; char buffer[buffer_capacity];

662
file/local_server/LocalFileTransferNetwork.cpp
View File

@ -9,7 +9,6 @@
#include <misc/net.h> #include <misc/net.h>
#include "./LocalFileProvider.h" #include "./LocalFileProvider.h"
#include "./duration_utils.h" #include "./duration_utils.h"
#include "LocalFileProvider.h"
#if defined(TCP_CORK) && !defined(TCP_NOPUSH) #if defined(TCP_CORK) && !defined(TCP_NOPUSH)
#define TCP_NOPUSH TCP_CORK #define TCP_NOPUSH TCP_CORK
@ -18,6 +17,8 @@
using namespace ts::server::file; using namespace ts::server::file;
using namespace ts::server::file::transfer; using namespace ts::server::file::transfer;
#define MAX_HTTP_HEADER_SIZE (4096)
inline void add_network_event(FileClient& transfer, event* ev, bool& ev_throttle_readd_flag, bool ignore_bandwidth) { inline void add_network_event(FileClient& transfer, event* ev, bool& ev_throttle_readd_flag, bool ignore_bandwidth) {
timeval tv{0, 1}, *ptv{nullptr}; timeval tv{0, 1}, *ptv{nullptr};
{ {
@ -87,16 +88,21 @@ void FileClient::add_network_write_event_nolock(bool ignore_bandwidth) {
return; return;
case STATE_DISCONNECTING: case STATE_DISCONNECTING:
if(this->transfer->direction == Transfer::DIRECTION_UPLOAD) if(this->transfer && this->transfer->direction == Transfer::DIRECTION_UPLOAD)
return; return;
/* flush our write buffer */ /* flush our write buffer */
break; break;
case STATE_AWAITING_KEY:
if(this->networking.protocol != FileClient::PROTOCOL_HTTPS) {
assert(false);
return;
}
break;
case STATE_TRANSFERRING: case STATE_TRANSFERRING:
break; break;
//case STATE_PENDING:
//case STATE_AWAITING_KEY:
default: default:
assert(false); assert(false);
break; break;
@ -107,22 +113,7 @@ void FileClient::add_network_write_event_nolock(bool ignore_bandwidth) {
bool FileClient::send_file_bytes(const void *snd_buffer, size_t size) { bool FileClient::send_file_bytes(const void *snd_buffer, size_t size) {
if(this->networking.protocol == FileClient::PROTOCOL_TS_V1) { if(this->networking.protocol == FileClient::PROTOCOL_TS_V1) {
auto tbuffer = allocate_buffer(size); return this->enqueue_buffer_bytes(snd_buffer, size);
tbuffer->length = size;
tbuffer->offset = 0;
memcpy(tbuffer->data, snd_buffer, size);
size_t buffer_size;
{
std::lock_guard block{this->buffer.mutex};
*this->buffer.buffer_tail = tbuffer;
this->buffer.buffer_tail = &tbuffer->next;
buffer_size = (this->buffer.bytes += size);
}
this->add_network_write_event(false);
return buffer_size > TRANSFER_MAX_CACHED_BYTES;
} else if(this->networking.protocol == FileClient::PROTOCOL_HTTPS) { } else if(this->networking.protocol == FileClient::PROTOCOL_HTTPS) {
this->networking.pipe_ssl.send(pipes::buffer_view{snd_buffer, size}); this->networking.pipe_ssl.send(pipes::buffer_view{snd_buffer, size});
return this->buffer.bytes > TRANSFER_MAX_CACHED_BYTES; return this->buffer.bytes > TRANSFER_MAX_CACHED_BYTES;
@ -131,6 +122,25 @@ bool FileClient::send_file_bytes(const void *snd_buffer, size_t size) {
} }
} }
bool FileClient::enqueue_buffer_bytes(const void *snd_buffer, size_t size) {
auto tbuffer = allocate_buffer(size);
tbuffer->length = size;
tbuffer->offset = 0;
memcpy(tbuffer->data, snd_buffer, size);
size_t buffer_size;
{
std::lock_guard block{this->buffer.mutex};
*this->buffer.buffer_tail = tbuffer;
this->buffer.buffer_tail = &tbuffer->next;
buffer_size = (this->buffer.bytes += size);
}
this->add_network_write_event(false);
return buffer_size > TRANSFER_MAX_CACHED_BYTES;
}
NetworkingStartResult LocalFileTransfer::start_networking() { NetworkingStartResult LocalFileTransfer::start_networking() {
assert(!this->network.active); assert(!this->network.active);
@ -311,6 +321,74 @@ void LocalFileTransfer::finalize_networking(const std::shared_ptr<FileClient> &c
client->networking.file_descriptor = 0; client->networking.file_descriptor = 0;
} }
#if 0
void dp_log(void* ptr, pipes::Logger::LogLevel level, const std::string& name, const std::string& message, ...) {
auto max_length = 1024 * 8;
char buffer[max_length];
va_list args;
va_start(args, message);
max_length = vsnprintf(buffer, max_length, message.c_str(), args);
va_end(args);
debugMessage(LOG_GENERAL, "[{}][{}] {}", level, name, std::string{buffer});
}
#endif
bool LocalFileTransfer::initialize_client_ssl(const std::shared_ptr<FileClient> &client) {
std::string error;
auto& ssl_pipe = client->networking.pipe_ssl;
if(!ssl_pipe.initialize(this->ssl_options_, error)) {
logWarning(0, "{} Failed to initialize client SSL pipe ({}). Disconnecting client.", client->log_prefix(), error);
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return false;
}
#if 0
auto logger = std::make_shared<pipes::Logger>();
logger->callback_log = dp_log;
ssl_pipe.logger(logger);
#endif
ssl_pipe.direct_process(pipes::PROCESS_DIRECTION_IN, true);
ssl_pipe.direct_process(pipes::PROCESS_DIRECTION_OUT, true);
ssl_pipe.callback_initialized = [client] {
logTrace(LOG_FT, "{} SSL layer has been initialized", client->log_prefix());
};
ssl_pipe.callback_data([&, client](const pipes::buffer_view& message) {
client->handle->handle_transfer_read(client, message.data_ptr<char>(), message.length());
});
ssl_pipe.callback_error([client](int error, const std::string & error_detail) {
logMessage(LOG_FT, "{} Received SSL error ({}/{}). Closing connection.", client->log_prefix(), error, error_detail);
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
});
ssl_pipe.callback_write([client](const pipes::buffer_view& buffer) {
client->enqueue_buffer_bytes(buffer.data_ptr(), buffer.length());
client->add_network_write_event(false);
});
return true;
}
void LocalFileTransfer::finalize_client_ssl(const std::shared_ptr<FileClient> &client) {
auto& ssl_pipe = client->networking.pipe_ssl;
ssl_pipe.callback_initialized = []{};
ssl_pipe.callback_write([](const pipes::buffer_view&){});
ssl_pipe.callback_error([](auto, const auto&){});
ssl_pipe.callback_data([](const auto&){});
}
void LocalFileTransfer::callback_transfer_network_accept(int fd, short, void *ptr_binding) { void LocalFileTransfer::callback_transfer_network_accept(int fd, short, void *ptr_binding) {
auto binding = reinterpret_cast<NetworkBinding*>(ptr_binding); auto binding = reinterpret_cast<NetworkBinding*>(ptr_binding);
auto transfer = binding->handle; auto transfer = binding->handle;
@ -456,10 +534,10 @@ void LocalFileTransfer::callback_transfer_network_read(int fd, short events, voi
size_t bytes_buffered{0}; size_t bytes_buffered{0};
if(transfer->state == FileClient::STATE_AWAITING_KEY) { if(transfer->state == FileClient::STATE_AWAITING_KEY) {
bytes_buffered = transfer->handle->handle_transfer_read(transfer->shared_from_this(), buffer, read); bytes_buffered = transfer->handle->handle_transfer_read_raw(transfer->shared_from_this(), buffer, read);
} else if(transfer->state == FileClient::STATE_TRANSFERRING) { } else if(transfer->state == FileClient::STATE_TRANSFERRING) {
if(transfer->transfer->direction == Transfer::DIRECTION_UPLOAD) { if(transfer->transfer->direction == Transfer::DIRECTION_UPLOAD) {
bytes_buffered = transfer->handle->handle_transfer_read(transfer->shared_from_this(), buffer, read); bytes_buffered = transfer->handle->handle_transfer_read_raw(transfer->shared_from_this(), buffer, read);
} else { } else {
debugMessage(LOG_FT, "{} Received {} bytes without any need. Dropping them.", transfer->log_prefix(), read); debugMessage(LOG_FT, "{} Received {} bytes without any need. Dropping them.", transfer->log_prefix(), read);
} }
@ -500,18 +578,20 @@ void LocalFileTransfer::callback_transfer_network_write(int fd, short events, vo
if((unsigned) events & (unsigned) EV_WRITE) { if((unsigned) events & (unsigned) EV_WRITE) {
if(transfer->state != FileClient::STATE_DISCONNECTING && transfer->state != FileClient::STATE_TRANSFERRING) { if(transfer->state != FileClient::STATE_DISCONNECTING && transfer->state != FileClient::STATE_TRANSFERRING) {
debugMessage(LOG_FT, "{} Tried to write data to send only stream. Dropping buffers.", transfer->log_prefix()); if(!(transfer->state == FileClient::STATE_AWAITING_KEY && transfer->networking.protocol == FileClient::PROTOCOL_HTTPS)) {
debugMessage(LOG_FT, "{} Tried to write data to send only stream. Dropping buffers.", transfer->log_prefix());
std::unique_lock block{transfer->buffer.mutex}; std::unique_lock block{transfer->buffer.mutex};
auto head = std::exchange(transfer->buffer.buffer_head, nullptr); auto head = std::exchange(transfer->buffer.buffer_head, nullptr);
transfer->buffer.buffer_tail = &transfer->buffer.buffer_head; transfer->buffer.buffer_tail = &transfer->buffer.buffer_head;
while(head) { while(head) {
auto next = head->next; auto next = head->next;
free_buffer(head); free_buffer(head);
head = next; head = next;
}
return;
} }
return;
} }
Buffer* buffer{nullptr}; Buffer* buffer{nullptr};
@ -598,7 +678,7 @@ void LocalFileTransfer::callback_transfer_network_write(int fd, short events, vo
if(buffer_left_size > 0) if(buffer_left_size > 0)
transfer->add_network_write_event(false); transfer->add_network_write_event(false);
else if(transfer->state == FileClient::STATE_DISCONNECTING) { else if(transfer->state == FileClient::STATE_DISCONNECTING) {
if(transfer->statistics.file_bytes_transferred + transfer->transfer->file_offset == transfer->transfer->expected_file_size) { if(transfer->transfer && transfer->statistics.file_bytes_transferred + transfer->transfer->file_offset == transfer->transfer->expected_file_size) {
logMessage(LOG_FT, "{} Finished file transfer within {}. Closing connection.", transfer->log_prefix(), duration_to_string(std::chrono::system_clock::now() - transfer->timings.key_received)); logMessage(LOG_FT, "{} Finished file transfer within {}. Closing connection.", transfer->log_prefix(), duration_to_string(std::chrono::system_clock::now() - transfer->timings.key_received));
transfer->handle->report_transfer_statistics(transfer->shared_from_this()); transfer->handle->report_transfer_statistics(transfer->shared_from_this());
if(auto callback{transfer->handle->callback_transfer_finished}; callback) if(auto callback{transfer->handle->callback_transfer_finished}; callback)
@ -624,64 +704,241 @@ inline std::string transfer_key_to_string(char key[TRANSFER_KEY_LENGTH]) {
return result; return result;
} }
size_t LocalFileTransfer::handle_transfer_read_raw(const std::shared_ptr<FileClient> &client, const char *buffer, size_t length) {
if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) {
return this->handle_transfer_read(client, buffer, length);
} else if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
client->networking.pipe_ssl.process_incoming_data(pipes::buffer_view{buffer, length});
return client->buffer.bytes;
} else if(client->networking.protocol != FileClient::PROTOCOL_UNKNOWN) {
assert(false);
logWarning(LOG_FT, "{} Read bytes with unknown protocol. Closing connection.", client->log_prefix());
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return (size_t) -1;
}
if(client->state != FileClient::STATE_AWAITING_KEY) {
logWarning(LOG_FT, "{} Read bytes with unknown protocol but having not awaiting key state. Closing connection.", client->log_prefix());
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return (size_t) -1;
}
/* lets read the key bytes (16) and then decide */
if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH) {
const auto bytes_write = std::min(TRANSFER_KEY_LENGTH - client->transfer_key.provided_bytes, length);
memcpy(client->transfer_key.key + client->transfer_key.provided_bytes, buffer, bytes_write);
length -= bytes_write;
buffer += bytes_write;
client->transfer_key.provided_bytes += bytes_write;
}
if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH)
return 0; /* we need more data */
if(pipes::SSL::is_ssl((uint8_t*) client->transfer_key.key, client->transfer_key.provided_bytes)) {
client->networking.protocol = FileClient::PROTOCOL_HTTPS;
client->networking.http_header_buffer.reset(allocate_buffer(MAX_HTTP_HEADER_SIZE)); /* max 8k header */
client->networking.max_read_buffer_size = (size_t) MAX_HTTP_HEADER_SIZE; /* HTTP-Header are sometimes a bit bigger. Dont cap max bandwidth here. */
debugMessage(LOG_FT, "{} Using protocol HTTPS for file transfer.", client->log_prefix());
char first_bytes[TRANSFER_KEY_LENGTH];
memcpy(first_bytes, client->transfer_key.key, TRANSFER_KEY_LENGTH);
client->transfer_key.provided_bytes = 0;
if(!this->initialize_client_ssl(client))
return (size_t) -1;
client->networking.pipe_ssl.process_incoming_data(pipes::buffer_view{first_bytes, TRANSFER_KEY_LENGTH});
if(length > 0)
client->networking.pipe_ssl.process_incoming_data(pipes::buffer_view{buffer, length});
return client->buffer.bytes;
} else {
client->networking.protocol = FileClient::PROTOCOL_TS_V1;
debugMessage(LOG_FT, "{} Using protocol RAWv1 for file transfer.", client->log_prefix());
std::string error_detail{};
auto key_result = this->handle_transfer_key_provided(client, error_detail);
switch(key_result) {
case TransferKeyApplyResult::SUCCESS:
if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD)
this->enqueue_disk_io(client); /* we've to take initiative */
return length ? this->handle_transfer_read(client, buffer, length) : 0;
case TransferKeyApplyResult::UNKNOWN_KEY:
logMessage(LOG_FT, "{} Disconnecting client because we don't recognise his key ({}).", client->log_prefix(), transfer_key_to_string(client->transfer_key.key));
break;
case TransferKeyApplyResult::FILE_ERROR:
assert(client->transfer);
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_aborted}; callback)
callback(client->transfer, { TransferError::DISK_INITIALIZE_ERROR, error_detail });
logMessage(LOG_FT, "{} Disconnecting client because we failed to open the target file.", client->log_prefix());
break;
default:
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_aborted}; client->transfer && callback)
callback(client->transfer, { TransferError::UNKNOWN, error_detail });
logMessage(LOG_FT, "{} Disconnecting client because of an unknown key initialize error ({}).", client->log_prefix(), (int) key_result);
break;
}
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return (size_t) -1;
}
return 0;
}
size_t LocalFileTransfer::handle_transfer_read(const std::shared_ptr<FileClient> &client, const char *buffer, size_t length) { size_t LocalFileTransfer::handle_transfer_read(const std::shared_ptr<FileClient> &client, const char *buffer, size_t length) {
if(client->state == FileClient::STATE_AWAITING_KEY) { if(client->state == FileClient::STATE_AWAITING_KEY) {
if(client->networking.protocol == FileClient::PROTOCOL_UNKNOWN) { if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
/* lets read the key bytes (16) and then decide */ assert(client->networking.http_header_buffer);
auto header_buffer = &*client->networking.http_header_buffer;
if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH) { http::HttpResponse response{};
const auto bytes_write = std::min(TRANSFER_KEY_LENGTH - client->transfer_key.provided_bytes, length); size_t overhead_length{0};
memcpy(client->transfer_key.key + client->transfer_key.provided_bytes, buffer, bytes_write); char* overhead_data_ptr{nullptr};
length -= bytes_write;
buffer += bytes_write; if(header_buffer->offset + length > header_buffer->capacity) {
client->transfer_key.provided_bytes += bytes_write; logMessage(LOG_FT, "{} Closing connection due to an too long HTTP(S) header (over {} bytes)", client->log_prefix(), header_buffer->capacity);
response.code = http::code::code(413, "Entity Too Large");
response.setHeader("x-error-message", { "header exceeds max size of " + std::to_string(header_buffer->capacity) });
goto send_response_exit;
} }
if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH) {
return 0; /* we need more data */ http::HttpRequest request{};
if(pipes::SSL::is_ssl((uint8_t*) client->transfer_key.key, client->transfer_key.provided_bytes)) { const auto old_offset = header_buffer->offset;
client->networking.protocol = FileClient::PROTOCOL_HTTPS; memcpy(header_buffer->data + header_buffer->offset, buffer, length);
client->networking.max_read_buffer_size = (size_t) 1024 * 2; /* HTTPS-Header are sometimes a bit bigger. Dont cap max bandwidth here. */ header_buffer->offset += length;
debugMessage(LOG_FT, "{} Using protocol HTTPS for file transfer.", client->log_prefix());
char first_bytes[TRANSFER_KEY_LENGTH]; constexpr static std::string_view header_end_token{"\r\n\r\n"};
memcpy(first_bytes, client->transfer_key.key, TRANSFER_KEY_LENGTH); auto header_view = std::string_view{header_buffer->data, header_buffer->offset};
client->transfer_key.provided_bytes = 0; auto header_end = header_view.find(header_end_token, old_offset > 3 ? old_offset - 3 : 0);
if(header_end == std::string::npos) return 0;
this->handle_transfer_read(client, first_bytes, TRANSFER_KEY_LENGTH); debugMessage(LOG_FT, "{} Received clients HTTP header.", client->log_prefix());
return this->handle_transfer_read(client, buffer, length); if(!http::parse_request(std::string{header_view.data(), header_end}, request)) {
} else { logError(LOG_FT, "{} Failed to parse HTTP request. Disconnecting client.", client->log_prefix());
client->networking.protocol = FileClient::PROTOCOL_TS_V1;
debugMessage(LOG_FT, "{} Using protocol RAWv1 for file transfer.", client->log_prefix());
return this->handle_transfer_read(client, buffer, length); std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, true);
return (size_t) -1;
}
const auto transfer_key_header = request.findHeader("transfer-key");
if(!transfer_key_header || transfer_key_header.values.empty()) {
logMessage(0, "{} Missing transfer key header. Disconnecting client.", client->log_prefix());
response.code = http::code::code(510, "Not Extended");
response.setHeader("x-error-message", { "missing transfer key" });
goto send_response_exit;
}
const auto& transfer_key = transfer_key_header.values[0];
if(transfer_key.length() != TRANSFER_KEY_LENGTH) {
logMessage(0, "{} Received too short/long transfer key. Expected {} but received {}. Disconnecting client.", client->log_prefix(), TRANSFER_KEY_LENGTH, transfer_key.length());
response.code = http::code::code(510, "Not Extended");
response.setHeader("x-error-message", { "key too short/long" });
goto send_response_exit;
}
client->transfer_key.provided_bytes = TRANSFER_KEY_LENGTH;
memcpy(client->transfer_key.key, transfer_key.data(), TRANSFER_KEY_LENGTH);
std::string error_detail{};
auto key_result = this->handle_transfer_key_provided(client, error_detail);
switch(key_result) {
case TransferKeyApplyResult::SUCCESS:
break;
case TransferKeyApplyResult::FILE_ERROR:
assert(client->transfer);
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_aborted}; callback)
callback(client->transfer, { TransferError::DISK_INITIALIZE_ERROR, error_detail });
logMessage(LOG_FT, "{} Disconnecting client because we failed to open the target file.", client->log_prefix());
response.code = http::code::code(500, "Internal Server Error");
response.setHeader("x-error-message", { error_detail });
goto send_response_exit;
case TransferKeyApplyResult::UNKNOWN_KEY:
logMessage(LOG_FT, "{} Disconnecting client because we don't recognise his key ({}).", client->log_prefix(), transfer_key_to_string(client->transfer_key.key));
response.code = http::code::code(406, "Not Acceptable");
response.setHeader("x-error-message", { "unknown key" });
goto send_response_exit;
default:
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_aborted}; client->transfer && callback)
callback(client->transfer, { TransferError::UNKNOWN, error_detail });
logMessage(LOG_FT, "{} Disconnecting client because of an unknown key initialize error ({}).", client->log_prefix(), (int) key_result);
response.code = http::code::code(500, "Internal Server Error");
response.setHeader("x-error-message", { error_detail.empty() ? "failed to initialize transfer" : error_detail });
goto send_response_exit;
}
response.code = http::code::_200;
if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD) {
const auto download_name = request.findHeader("download-name");
response.setHeader("Content-Length", { std::to_string(client->transfer->expected_file_size - client->transfer->file_offset) });
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.empty() ? download_name.values[0] : "TeaWeb Download") + "\""
});
/* TODO: X-media-bytes */
#if 0
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);
}
#endif
client->networking.http_state = FileClient::HTTP_STATE_DOWNLOADING;
goto send_response_exit;
} else {
response.setHeader("Content-Length", { std::to_string(client->transfer->expected_file_size) });
client->networking.http_state = FileClient::HTTP_STATE_AWAITING_BOUNDARY;
}
overhead_length = header_buffer->offset - header_end - header_end_token.length();
overhead_data_ptr = header_buffer->data + header_end + header_end_token.length();
} }
} else if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) {
auto key_result = this->handle_transfer_key_provided(client);
switch(key_result) {
case TransferKeyApplyResult::SUCCESS:
return length ? this->handle_transfer_read(client, buffer, length) : 0;
case TransferKeyApplyResult::FILE_ERROR: send_response_exit:
logMessage(LOG_FT, "{} Disconnecting client because we failed to open the target file.", client->log_prefix()); this->send_http_response(client, response);
return (size_t) -1; if(response.code->code != 200) {
std::unique_lock slock{client->state_mutex};
case TransferKeyApplyResult::UNKNOWN_KEY: client->handle->disconnect_client(client->shared_from_this(), slock, true);
logMessage(LOG_FT, "{} Disconnecting client because we don't recognise his key ({}).", client->log_prefix(), transfer_key_to_string(client->transfer_key.key)); return (size_t) -1;
return (size_t) -1;
default:
logMessage(LOG_FT, "{} Disconnecting client because of an unknown key initialize error ({}).", client->log_prefix(), (int) key_result);
return (size_t) -1;
} }
} else if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
/* TODO! */ if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD)
/* 1. Await HTTP header this->enqueue_disk_io(client); /* we've to take initiative */
* 2. Test for file transfer key & if its upload or download
*/ header_buffer->offset = 0;
return overhead_length == 0 ? 0 : this->handle_transfer_read(client, overhead_data_ptr, overhead_length);
} else { } else {
logError(LOG_FT, "{} Protocol variable contains unknown protocol. Disconnecting client.", client->log_prefix()); logError(LOG_FT, "{} Protocol variable contains invalid protocol for awaiting key state. Disconnecting client.", client->log_prefix());
std::unique_lock slock{client->state_mutex}; std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, true); client->handle->disconnect_client(client->shared_from_this(), slock, true);
@ -695,47 +952,73 @@ size_t LocalFileTransfer::handle_transfer_read(const std::shared_ptr<FileClient>
} }
if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) { if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
client->networking.pipe_ssl.process_incoming_data(pipes::buffer_view{buffer, length}); std::string error_message{};
return client->buffer.bytes; /* a bit unexact but the best we could get away with it */ const auto upload_result = client->handle->handle_transfer_upload_http(client, buffer, length);
switch(upload_result) {
case TransferUploadHTTPResult::FINISH: {this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_finished}; callback)
callback(client->transfer);
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, true);
return client->buffer.bytes; /* a bit unexact but the best we could get away with it */
}
case TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE:
return client->buffer.bytes; /* a bit unexact but the best we could get away with it */
case TransferUploadHTTPResult::MISSING_CONTENT_TYPE:
logMessage(LOG_FT, "{} Missing boundary content type. Disconnecting client.", client->log_prefix());
error_message = "invalid boundary content type";
break;
case TransferUploadHTTPResult::INVALID_CONTENT_TYPE:
logMessage(LOG_FT, "{} Invalid boundary content type. Disconnecting client.", client->log_prefix());
error_message = "missing boundary content type";
break;
case TransferUploadHTTPResult::BOUNDARY_MISSING:
logMessage(LOG_FT, "{} Missing boundary token. Disconnecting client.", client->log_prefix());
error_message = "missing boundary token";
break;
case TransferUploadHTTPResult::BOUNDARY_INVALID:
logMessage(LOG_FT, "{} Invalid boundary. Disconnecting client.", client->log_prefix());
error_message = "invalid boundary";
break;
case TransferUploadHTTPResult::BOUNDARY_TOKEN_INVALID:
logMessage(LOG_FT, "{} Invalid boundary token. Disconnecting client.", client->log_prefix());
error_message = "invalid boundary token";
break;
}
http::HttpResponse response{};
response.code = http::code::code(510, "Not Extended");
response.setHeader("x-error-message", { error_message });
client->handle->send_http_response(client, response);
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, true);
return (size_t) -1;
} else if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) { } else if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) {
client->statistics.file_bytes_transferred += length; auto result = this->handle_transfer_upload_raw(client, buffer, length);
bool transfer_finished{false}; switch (result) {
auto writte_offset = client->statistics.file_bytes_transferred + client->transfer->file_offset; case TransferUploadRawResult::FINISH: {this->report_transfer_statistics(client);
if(writte_offset > client->transfer->expected_file_size) { if(auto callback{this->callback_transfer_finished}; callback)
logMessage(LOG_FT, "{} Client send {} too many bytes (Transfer length was {}). Dropping them, flushing the disk IO and closing the transfer.", client->log_prefix(), writte_offset - client->transfer->expected_file_size, duration_to_string(std::chrono::system_clock::now() - client->timings.key_received)); callback(client->transfer);
length -= writte_offset - client->transfer->expected_file_size;
transfer_finished = true; std::unique_lock slock{client->state_mutex};
} else if(writte_offset == client->transfer->expected_file_size) { client->handle->disconnect_client(client->shared_from_this(), slock, true);
logMessage(LOG_FT, "{} File upload has been completed in {}. Flushing disk IO and closing the connection.", client->log_prefix(), duration_to_string(std::chrono::system_clock::now() - client->timings.key_received)); return (size_t) -1;
transfer_finished = true; }
case TransferUploadRawResult::MORE_DATA_TO_RECEIVE:
return client->buffer.bytes; /* a bit unexact but the best we could get away with it */
} }
auto tbuffer = allocate_buffer(length);
tbuffer->offset = 0;
tbuffer->length = length;
memcpy(tbuffer->data, buffer, length);
size_t buffered_bytes;
{
std::lock_guard block{client->buffer.mutex};
*client->buffer.buffer_tail = tbuffer;
client->buffer.buffer_tail = &tbuffer->next;
buffered_bytes = (client->buffer.bytes += length);
}
this->enqueue_disk_io(client);
if(transfer_finished) {
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_finished}; callback)
callback(client->transfer);
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, true);
}
return buffered_bytes;
} else { } else {
logWarning(LOG_FT, "{} Read message for client with unknown protocol. Dropping {} bytes.", client->log_prefix(), length); logWarning(LOG_FT, "{} Read message for client with unknown protocol. Dropping {} bytes.", client->log_prefix(), length);
return 0; return 0;
@ -746,7 +1029,7 @@ size_t LocalFileTransfer::handle_transfer_read(const std::shared_ptr<FileClient>
return 0; return 0;
} }
TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std::shared_ptr<FileClient> &client) { TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std::shared_ptr<FileClient> &client, std::string& error_detail) {
{ {
std::lock_guard tlock{this->transfers_mutex}; std::lock_guard tlock{this->transfers_mutex};
for(auto it = this->pending_transfers.begin(); it != this->pending_transfers.end(); it++) { for(auto it = this->pending_transfers.begin(); it != this->pending_transfers.end(); it++) {
@ -758,14 +1041,8 @@ TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std
} }
} }
if(!client->transfer) { if(!client->transfer)
logMessage(LOG_FT, "{} Received invalid transfer key. Disconnecting client.", client->log_prefix());
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return TransferKeyApplyResult::UNKNOWN_KEY; return TransferKeyApplyResult::UNKNOWN_KEY;
}
{ {
std::string absolute_path{}; std::string absolute_path{};
@ -788,15 +1065,8 @@ TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std
break; break;
default: default:
logError(LOG_FT, "{} Tried to initialize client with unknown file target type ({}). Dropping transfer.", client->log_prefix(), (int) transfer->target_type); logError(LOG_FT, "{} Tried to initialize client with unknown file target type ({}). Dropping transfer.", client->log_prefix(), (int) transfer->target_type);
this->report_transfer_statistics(client); error_detail = "invalid transfer target type";
if(auto callback{this->callback_transfer_aborted}; callback) return TransferKeyApplyResult::INTERNAL_ERROR;
callback(client->transfer, { TransferError::UNKNOWN, "invalid transfer type" });
{
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
}
return TransferKeyApplyResult::FILE_ERROR;
} }
debugMessage(LOG_FT, "{} Absolute file path: {}", client->log_prefix(), absolute_path); debugMessage(LOG_FT, "{} Absolute file path: {}", client->log_prefix(), absolute_path);
client->file.absolute_path = absolute_path; client->file.absolute_path = absolute_path;
@ -812,14 +1082,7 @@ TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std
if(io_init_result != FileInitializeResult::SUCCESS) { if(io_init_result != FileInitializeResult::SUCCESS) {
logMessage(LOG_FT, "{} Failed to initialize file {}: {}/{}. Disconnecting client.", logMessage(LOG_FT, "{} Failed to initialize file {}: {}/{}. Disconnecting client.",
client->log_prefix(), client->transfer->direction == Transfer::DIRECTION_UPLOAD ? "writer" : "reader", (int) io_init_result, kFileInitializeResultMessages[(int) io_init_result]); client->log_prefix(), client->transfer->direction == Transfer::DIRECTION_UPLOAD ? "writer" : "reader", (int) io_init_result, kFileInitializeResultMessages[(int) io_init_result]);
error_detail = std::to_string((int) io_init_result) + "/" + std::string{kFileInitializeResultMessages[(int) io_init_result]};
this->report_transfer_statistics(client);
if(auto callback{this->callback_transfer_aborted}; callback)
callback(client->transfer, { TransferError::DISK_INITIALIZE_ERROR, std::to_string((int) io_init_result) + "/" + std::string{kFileInitializeResultMessages[(int) io_init_result]} });
std::unique_lock slock{client->state_mutex};
client->handle->disconnect_client(client->shared_from_this(), slock, false);
return TransferKeyApplyResult::FILE_ERROR; return TransferKeyApplyResult::FILE_ERROR;
} }
@ -835,9 +1098,122 @@ TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std
callback(client->transfer); callback(client->transfer);
client->timings.key_received = std::chrono::system_clock::now(); client->timings.key_received = std::chrono::system_clock::now();
if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD)
this->enqueue_disk_io(client); /* we've to take initiative */
return TransferKeyApplyResult::SUCCESS; return TransferKeyApplyResult::SUCCESS;
} }
TransferUploadRawResult LocalFileTransfer::handle_transfer_upload_raw(const std::shared_ptr<FileClient> &client, const char *buffer, size_t length) {
client->statistics.file_bytes_transferred += length;
bool transfer_finished{false};
auto writte_offset = client->statistics.file_bytes_transferred + client->transfer->file_offset;
if(writte_offset > client->transfer->expected_file_size) {
logMessage(LOG_FT, "{} Client send {} too many bytes (Transfer length was {}). Dropping them, flushing the disk IO and closing the transfer.", client->log_prefix(), writte_offset - client->transfer->expected_file_size, duration_to_string(std::chrono::system_clock::now() - client->timings.key_received));
length -= writte_offset - client->transfer->expected_file_size;
transfer_finished = true;
} else if(writte_offset == client->transfer->expected_file_size) {
logMessage(LOG_FT, "{} File upload has been completed in {}. Flushing disk IO and closing the connection.", client->log_prefix(), duration_to_string(std::chrono::system_clock::now() - client->timings.key_received));
transfer_finished = true;
}
auto tbuffer = allocate_buffer(length);
tbuffer->offset = 0;
tbuffer->length = length;
memcpy(tbuffer->data, buffer, length);
{
std::lock_guard block{client->buffer.mutex};
*client->buffer.buffer_tail = tbuffer;
client->buffer.buffer_tail = &tbuffer->next;
client->buffer.bytes += length;
}
this->enqueue_disk_io(client);
return transfer_finished ? TransferUploadRawResult::FINISH : TransferUploadRawResult::MORE_DATA_TO_RECEIVE;
}
//Example boundary:
//------WebKitFormBoundaryaWP8XAzMBnMOJznv\r\nContent-Disposition: form-data; name="file"; filename="blob"\r\nContent-Type: application/octet-stream\r\n\r\n
TransferUploadHTTPResult LocalFileTransfer::handle_transfer_upload_http(const std::shared_ptr<FileClient> &client,
const char *buffer, size_t length) {
constexpr static std::string_view boundary_end_token{"\r\n\r\n"};
constexpr static std::string_view boundary_token_end_token{"\r\n"};
if(client->networking.http_state == FileClient::HTTP_STATE_AWAITING_BOUNDARY) {
assert(client->networking.http_header_buffer);
/* Notice: The buffer ptr might be some data within our header buffer! But since its somewhere in the back its okey */
auto boundary_buffer = &*client->networking.http_header_buffer;
if(boundary_buffer->offset + length > boundary_buffer->capacity)
return TransferUploadHTTPResult::BOUNDARY_MISSING;
const auto old_offset = boundary_buffer->offset;
memcpy(boundary_buffer->data + boundary_buffer->offset, buffer, length);
boundary_buffer->offset += length;
auto boundary_view = std::string_view{boundary_buffer->data, boundary_buffer->offset};
auto boundary_end = boundary_view.find(boundary_end_token, old_offset > 3 ? old_offset - 3 : 0);
if(boundary_end == std::string::npos)
return TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE;
auto boundary_token_end = boundary_view.find(boundary_token_end_token);
if(boundary_token_end >= boundary_end)
return TransferUploadHTTPResult::BOUNDARY_TOKEN_INVALID;
const auto boundary_token = boundary_view.substr(0, boundary_token_end);
debugMessage(LOG_FT, "{} Received clients HTTP file boundary ({}).", client->log_prefix(), boundary_token);
const auto boundary_payload = boundary_view.substr(boundary_token_end + boundary_token_end_token.size());
http::HttpRequest boundary{};
if(!http::parse_request(std::string{boundary_payload}, boundary))
return TransferUploadHTTPResult::BOUNDARY_INVALID;
const auto content_type = boundary.findHeader("Content-Type");
if(!content_type || content_type.values.empty())
return TransferUploadHTTPResult::MISSING_CONTENT_TYPE;
else if(content_type.values[0] != "application/octet-stream")
return TransferUploadHTTPResult::INVALID_CONTENT_TYPE;
const auto overhead_length = boundary_buffer->offset - boundary_end - boundary_end_token.length();
const auto overhead_data_ptr = boundary_buffer->data + boundary_end + boundary_end_token.length();
client->networking.http_state = FileClient::HTTP_STATE_UPLOADING;
boundary_buffer->offset = 0;
return overhead_length == 0 ? TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE : this->handle_transfer_upload_http(client, overhead_data_ptr, overhead_length);
} else if(client->networking.http_state == FileClient::HTTP_STATE_UPLOADING) {
auto result = this->handle_transfer_upload_raw(client, buffer, length);
switch(result) {
case TransferUploadRawResult::MORE_DATA_TO_RECEIVE:
return TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE;
case TransferUploadRawResult::FINISH:
/* TODO: Try to read the end boundary! */
return TransferUploadHTTPResult::FINISH;
default:
assert(false);
return TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE;
}
} else {
logWarning(0, "{} Received HTTP(S) data, for an invalid HTTP state ({}).", client->log_prefix(), (int) client->networking.http_state);
return TransferUploadHTTPResult::MORE_DATA_TO_RECEIVE;
}
}
inline void apply_cors_and_connection_headers(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.findHeader("Access-Control-Request-Headers").values); //access-control-allow-headers
response.setHeader("Access-Control-Max-Age", {"86400"});
}
void LocalFileTransfer::send_http_response(const std::shared_ptr<FileClient> &client, http::HttpResponse &response) {
apply_cors_and_connection_headers(response);
response.setHeader("Access-Control-Expose-Headers", {"*, x-error-message, Content-Length, X-media-bytes, Content-Disposition"});
const auto payload = response.build();
client->send_file_bytes(payload.data(), payload.length());
}

4
file/test/main.cpp
View File

@ -160,12 +160,12 @@ int main() {
}; };
{ {
auto response = ft.initialize_channel_transfer(file::transfer::Transfer::DIRECTION_DOWNLOAD, 0, 2, { auto response = ft.initialize_channel_transfer(file::transfer::Transfer::DIRECTION_UPLOAD, 0, 2, {
"test2.txt", "test2.txt",
false, false,
4, 4,
120, 120,
16 32
}); });
response->wait(); response->wait();
print_ft_response("Download test.txt", response); print_ft_response("Download test.txt", response);

2
file/test/upload-content.bin Normal file
View File

@ -0,0 +1,2 @@
Hello World
xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

2
git-teaspeak

@ -1 +1 @@
Subproject commit 99293e3af97cdc4041282a29495841b8ae3d2a4d Subproject commit ee5090232867df795348b84f1c6d96561ea24d90

1
server/src/VirtualServer.cpp
View File

@ -122,6 +122,7 @@ bool VirtualServer::initialize(bool test_properties) {
channelTree = new ServerChannelTree(self.lock(), this->sql); channelTree = new ServerChannelTree(self.lock(), this->sql);
channelTree->loadChannelsFromDatabase(); channelTree->loadChannelsFromDatabase();
channelTree->deleteSemiPermanentChannels();
this->groups = new GroupManager(self.lock(), this->sql, serverInstance->getGroupManager()); this->groups = new GroupManager(self.lock(), this->sql, serverInstance->getGroupManager());
if(!this->groups->loadGroupFormDatabase()){ //TODO exception etc if(!this->groups->loadGroupFormDatabase()){ //TODO exception etc

150
server/src/client/SpeakingClient.cpp
View File

@ -121,7 +121,9 @@ enum WhisperType {
SERVER_GROUP = 0, SERVER_GROUP = 0,
CHANNEL_GROUP = 1, CHANNEL_GROUP = 1,
CHANNEL_COMMANDER = 2, CHANNEL_COMMANDER = 2,
ALL = 3 ALL = 3,
ECHO_TEXT = 0x10,
}; };
enum WhisperTarget { enum WhisperTarget {
@ -181,87 +183,93 @@ void SpeakingClient::handlePacketVoiceWhisper(const pipes::buffer_view& data, bo
#endif #endif
deque<shared_ptr<SpeakingClient>> available_clients; deque<shared_ptr<SpeakingClient>> available_clients;
for(const auto& client : this->server->getClients()) { if(type == WhisperType::ECHO_TEXT) {
auto speakingClient = dynamic_pointer_cast<SpeakingClient>(client); available_clients.push_back(dynamic_pointer_cast<SpeakingClient>(this->ref()));
if(!speakingClient || client == this) continue; } else {
if(!speakingClient->currentChannel) continue; for(const auto& client : this->server->getClients()) {
auto speakingClient = dynamic_pointer_cast<SpeakingClient>(client);
if(!speakingClient || client == this) continue;
if(!speakingClient->currentChannel) continue;
if(type == WhisperType::ALL) { if(type == WhisperType::ALL) {
available_clients.push_back(speakingClient);
} else if(type == WhisperType::SERVER_GROUP) {
if(type_id == 0)
available_clients.push_back(speakingClient); available_clients.push_back(speakingClient);
else { } else if(type == WhisperType::SERVER_GROUP) {
shared_lock client_lock(this->channel_lock); if(type_id == 0)
for(const auto& id : client->cached_server_groups) { available_clients.push_back(speakingClient);
if(id == type_id) { else {
available_clients.push_back(speakingClient); shared_lock client_lock(this->channel_lock);
break; for(const auto& id : client->cached_server_groups) {
if(id == type_id) {
available_clients.push_back(speakingClient);
break;
}
} }
} }
} else if(type == WhisperType::CHANNEL_GROUP) {
if(client->cached_channel_group == type_id)
available_clients.push_back(speakingClient);
} else if(type == WhisperType::CHANNEL_COMMANDER) {
if(client->properties()[property::CLIENT_IS_CHANNEL_COMMANDER].as<bool>())
available_clients.push_back(speakingClient);
} }
} else if(type == WhisperType::CHANNEL_GROUP) {
if(client->cached_channel_group == type_id)
available_clients.push_back(speakingClient);
} else if(type == WhisperType::CHANNEL_COMMANDER) {
if(client->properties()[property::CLIENT_IS_CHANNEL_COMMANDER].as<bool>())
available_clients.push_back(speakingClient);
} }
}
if(target == WhisperTarget::CHANNEL_CURRENT) {
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
return target->currentChannel != this->currentChannel;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_PARENT) {
auto current_parent = this->currentChannel->parent();
if(!current_parent) return;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
return target->currentChannel != current_parent; if(target == WhisperTarget::CHANNEL_CURRENT) {
}), available_clients.end()); available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
} else if(target == WhisperTarget::CHANNEL_ALL_PARENT) { return target->currentChannel != this->currentChannel;
shared_ptr<BasicChannel> current_parent; }), available_clients.end());
{ } else if(target == WhisperTarget::CHANNEL_PARENT) {
current_parent = this->currentChannel->parent(); auto current_parent = this->currentChannel->parent();
if(!current_parent) return; if(!current_parent) return;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
return target->currentChannel != current_parent;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_ALL_PARENT) {
shared_ptr<BasicChannel> current_parent;
{
current_parent = this->currentChannel->parent();
if(!current_parent) return;
}
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_parent = current_parent;
while(tmp_parent && tmp_parent != target->currentChannel)
tmp_parent = tmp_parent->parent();
return target->currentChannel != tmp_parent;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_FAMILY) {
shared_ptr<BasicChannel> current = this->currentChannel;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_current = target->currentChannel;
while(tmp_current && tmp_current != current)
tmp_current = tmp_current->parent();
return current != tmp_current;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_COMPLETE_FAMILY) {
shared_ptr<BasicChannel> current = this->currentChannel;
while(current && current->parent()) current = current->parent();
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_current = target->currentChannel;
while(tmp_current && tmp_current != current)
tmp_current = tmp_current->parent();
return current != tmp_current;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_SUBCHANNELS) {
shared_ptr<BasicChannel> current = this->currentChannel;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
return target->currentChannel->parent() != current;
}), available_clients.end());
} }
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_parent = current_parent; auto self_lock = this->_this.lock();
while(tmp_parent && tmp_parent != target->currentChannel) available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const std::shared_ptr<ConnectedClient>& cl) {
tmp_parent = tmp_parent->parent(); auto speakingClient = dynamic_pointer_cast<SpeakingClient>(cl);
return target->currentChannel != tmp_parent; return !speakingClient->shouldReceiveVoiceWhisper(self_lock);
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_FAMILY) {
shared_ptr<BasicChannel> current = this->currentChannel;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_current = target->currentChannel;
while(tmp_current && tmp_current != current)
tmp_current = tmp_current->parent();
return current != tmp_current;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_COMPLETE_FAMILY) {
shared_ptr<BasicChannel> current = this->currentChannel;
while(current && current->parent()) current = current->parent();
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
auto tmp_current = target->currentChannel;
while(tmp_current && tmp_current != current)
tmp_current = tmp_current->parent();
return current != tmp_current;
}), available_clients.end());
} else if(target == WhisperTarget::CHANNEL_SUBCHANNELS) {
shared_ptr<BasicChannel> current = this->currentChannel;
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const shared_ptr<SpeakingClient>& target) {
return target->currentChannel->parent() != current;
}), available_clients.end()); }), available_clients.end());
} }
auto self_lock = this->_this.lock();
available_clients.erase(std::remove_if(available_clients.begin(), available_clients.end(), [&](const std::shared_ptr<ConnectedClient>& cl) {
auto speakingClient = dynamic_pointer_cast<SpeakingClient>(cl);
return !speakingClient->shouldReceiveVoiceWhisper(self_lock);
}), available_clients.end());
if(available_clients.empty()) { if(available_clients.empty()) {
if(update_whisper_error(this->speak_last_no_whisper_target)) { if(update_whisper_error(this->speak_last_no_whisper_target)) {
command_result result{error::whisper_no_targets}; command_result result{error::whisper_no_targets};
@ -289,7 +297,7 @@ void SpeakingClient::handlePacketVoiceWhisper(const pipes::buffer_view& data, bo
VoicePacketFlags flags{}; VoicePacketFlags flags{};
auto data = pipes::buffer_view(packet_buffer, OUT_WHISPER_PKT_OFFSET + data_length); auto data = pipes::buffer_view(packet_buffer, OUT_WHISPER_PKT_OFFSET + data_length);
for(const auto& cl : available_clients){ for(const auto& cl : available_clients){
cl->send_voice_whisper_packet(data, flags); cl->send_voice_whisper_packet(data, flags);
} }
this->updateSpeak(false, system_clock::now()); this->updateSpeak(false, system_clock::now());
@ -385,7 +393,7 @@ auto regex_wildcard = std::regex(".*");
#define S(x) #x #define S(x) #x
#define HWID_REGEX(name, pattern) \ #define HWID_REGEX(name, pattern) \
auto regex_hwid_ ##name = [](){ \ auto regex_hwid_ ##name = []() noexcept { \
try { \ try { \
return std::regex(pattern); \ return std::regex(pattern); \
} catch (std::exception& ex) { \ } catch (std::exception& ex) { \

4
server/src/client/command_handler/bulk_parsers.h
View File

@ -78,7 +78,7 @@ namespace ts::command::bulk_parser {
inline void apply_to(const std::shared_ptr<permission::v2::PermissionManager>& manager, permission::v2::PermissionUpdateType mode) const { inline void apply_to(const std::shared_ptr<permission::v2::PermissionManager>& manager, permission::v2::PermissionUpdateType mode) const {
if(this->is_grant_permission()) { if(this->is_grant_permission()) {
manager->set_permission(this->permission_type(), permission::v2::empty_permission_values, permission::v2::PermissionUpdateType::do_nothing, mode); manager->set_permission(this->permission_type(), { this->value(), true }, permission::v2::PermissionUpdateType::do_nothing, mode);
} else { } else {
manager->set_permission( manager->set_permission(
this->permission_type(), this->permission_type(),
@ -91,7 +91,7 @@ namespace ts::command::bulk_parser {
inline void apply_to_channel(const std::shared_ptr<permission::v2::PermissionManager>& manager, permission::v2::PermissionUpdateType mode, ChannelId channel_id) const { inline void apply_to_channel(const std::shared_ptr<permission::v2::PermissionManager>& manager, permission::v2::PermissionUpdateType mode, ChannelId channel_id) const {
if(this->is_grant_permission()) { if(this->is_grant_permission()) {
manager->set_channel_permission(this->permission_type(), channel_id, permission::v2::empty_permission_values, permission::v2::PermissionUpdateType::do_nothing, mode); manager->set_channel_permission(this->permission_type(), channel_id, { this->value(), true }, permission::v2::PermissionUpdateType::do_nothing, mode);
} else { } else {
manager->set_channel_permission( manager->set_channel_permission(
this->permission_type(), this->permission_type(),