From 1aebcfd55db33ce740537528cc1a0d06ff64e10c Mon Sep 17 00:00:00 2001 From: TapTap Date: Tue, 1 Sep 2026 21:21:24 +0200 Subject: [PATCH] refactor: consolidate transfer infrastructure --- CMakeLists.txt | 135 +++++++++----- src/server/server.c | 123 +------------ src/shared/config.c | 341 +++++++++-------------------------- src/shared/config.h | 47 +++++ src/shared/multiprocessing.c | 228 +++++++++++------------ src/shared/multiprocessing.h | 5 + src/shared/transport_tcp.c | 6 +- src/shared/transport_tcp.h | 1 + src/shared/transport_tls.c | 49 +---- 9 files changed, 348 insertions(+), 587 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 42c2353..6f5bec2 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -8,34 +8,30 @@ set(CMAKE_C_STANDARD_REQUIRED ON) add_compile_options(-Wall -g -O3) -# --- Sanitizer option --- set(SANITIZER "none" CACHE STRING "Sanitizer to enable (address, thread, undefined, none)") set_property(CACHE SANITIZER PROPERTY STRINGS address thread undefined none) - if(SANITIZER STREQUAL "address") - add_compile_options(-fsanitize=address -fno-omit-frame-pointer -g) - add_link_options(-fsanitize=address) + add_compile_options(-fsanitize=address -fno-omit-frame-pointer -g) + add_link_options(-fsanitize=address) elseif(SANITIZER STREQUAL "thread") - add_compile_options(-fsanitize=thread -fno-omit-frame-pointer -g) - add_link_options(-fsanitize=thread) + add_compile_options(-fsanitize=thread -fno-omit-frame-pointer -g) + add_link_options(-fsanitize=thread) elseif(SANITIZER STREQUAL "undefined") - add_compile_options(-fsanitize=undefined -fno-omit-frame-pointer -g) - add_link_options(-fsanitize=undefined) + add_compile_options(-fsanitize=undefined -fno-omit-frame-pointer -g) + add_link_options(-fsanitize=undefined) elseif(NOT SANITIZER STREQUAL "none") - message(FATAL_ERROR "Unknown sanitizer: ${SANITIZER}. Supported values: address, thread, undefined, none") + message(FATAL_ERROR "Unknown sanitizer: ${SANITIZER}. Supported values: address, thread, undefined, none") endif() -# --- Strict warnings option --- option(STRICT_WARNINGS "Enable strict warnings (Wextra, Wpedantic, Werror)" OFF) if(STRICT_WARNINGS) - add_compile_options(-Wextra -Wpedantic -Werror) + add_compile_options(-Wextra -Wpedantic -Werror) endif() -# --- Coverage option --- option(ENABLE_COVERAGE "Enable gcov coverage" OFF) if(ENABLE_COVERAGE) - add_compile_options(--coverage -fprofile-arcs -ftest-coverage -O0 -g) - add_link_options(--coverage) + add_compile_options(--coverage -fprofile-arcs -ftest-coverage -O0 -g) + add_link_options(--coverage) endif() include(FetchContent) @@ -49,55 +45,102 @@ FetchContent_MakeAvailable(xxhash) set(THREADS_PREFER_PTHREAD_FLAG ON) find_package(Threads REQUIRED) - find_library(ZSTD_LIBRARY zstd) if(NOT ZSTD_LIBRARY) - message(FATAL_ERROR "zstd library not found. Ensure it is in your nix-shell!") + message(FATAL_ERROR "zstd library not found. Ensure it is in your nix-shell!") endif() - find_package(OpenSSL REQUIRED) -file(GLOB SHARED_SRCS "src/shared/*.c") -file(GLOB SERVER_SRCS "src/server/*.c") -file(GLOB CLIENT_SRCS "src/client/*.c") +set(SHARED_SRCS + src/shared/array_list.c + src/shared/chunk.c + src/shared/compression.c + src/shared/config.c + src/shared/data.c + src/shared/delta.c + src/shared/file.c + src/shared/log.c + src/shared/metadata.c + src/shared/multiprocessing.c + src/shared/protocol.c + src/shared/queue.c + src/shared/transport_ssh.c + src/shared/transport_tcp.c + src/shared/transport_tls.c + src/shared/utils.c +) +set(SERVER_SRCS src/server/server.c) +set(CLIENT_SRCS + src/client/client_cli.c + src/client/client_send.c + src/client/scanner.c + src/client/usage.c +) + +function(configure_fastsync_target target) + target_include_directories(${target} PRIVATE src/shared src/server src/client) + target_link_libraries(${target} PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash) +endfunction() -# --- Main executables --- add_executable(server ${SERVER_SRCS} ${SHARED_SRCS}) -target_include_directories(server PRIVATE src/shared src/server src/client) -target_link_libraries(server PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash) - +configure_fastsync_target(server) add_executable(client ${CLIENT_SRCS} ${SHARED_SRCS}) -target_include_directories(client PRIVATE src/shared src/server src/client) -target_link_libraries(client PRIVATE Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash) +configure_fastsync_target(client) -# --- Testing --- enable_testing() - -# Common test libraries -set(TEST_LIBS Threads::Threads ${ZSTD_LIBRARY} OpenSSL::SSL OpenSSL::Crypto xxhash) -set(TEST_INCLUDES tests src/shared src/server src/client) - -# Monolithic test binary (backward compatible) -file(GLOB TEST_SRCS "tests/test_*.c" "tests/runner.c") +set(TEST_SRCS + tests/runner.c + tests/test_array_list.c + tests/test_chunk.c + tests/test_client_cli.c + tests/test_compression.c + tests/test_config.c + tests/test_data.c + tests/test_delta.c + tests/test_file.c + tests/test_file_sendfile.c + tests/test_fuzz_smoke.c + tests/test_glob.c + tests/test_log.c + tests/test_metadata.c + tests/test_multiprocessing.c + tests/test_property.c + tests/test_protocol.c + tests/test_queue.c + tests/test_robustness.c + tests/test_scanner.c + tests/test_server.c + tests/test_shared_utils.c + tests/test_stress.c + tests/test_transport_ssh.c + tests/test_transport_tcp.c + tests/test_transport_tls.c +) add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} src/client/scanner.c src/client/client_cli.c) -target_include_directories(tests PRIVATE ${TEST_INCLUDES}) +configure_fastsync_target(tests) +target_include_directories(tests PRIVATE tests) target_compile_definitions(tests PRIVATE FASTSYNC_TEST_BUILD) -target_link_libraries(tests PRIVATE ${TEST_LIBS}) add_test(NAME unit_all COMMAND tests) -# --- Fuzz targets (requires clang) --- option(ENABLE_FUZZ "Build fuzz targets (requires clang)" OFF) if(ENABLE_FUZZ) if(NOT CMAKE_C_COMPILER_ID MATCHES "Clang") message(FATAL_ERROR "ENABLE_FUZZ requires Clang (compiler is ${CMAKE_C_COMPILER_ID})") endif() - file(GLOB FUZZ_SRCS "tests/fuzz/*.c") - foreach(FUZZ_SRC ${FUZZ_SRCS}) - get_filename_component(FUZZ_NAME ${FUZZ_SRC} NAME_WE) - add_executable(${FUZZ_NAME} ${FUZZ_SRC} ${SHARED_SRCS}) - target_include_directories(${FUZZ_NAME} PRIVATE ${TEST_INCLUDES}) - target_compile_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer) - target_link_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined) - target_link_libraries(${FUZZ_NAME} PRIVATE ${TEST_LIBS}) - endforeach() + set(FUZZ_SRCS + tests/fuzz/fuzz_chunk_deserialize.c + tests/fuzz/fuzz_compress_decompress.c + tests/fuzz/fuzz_delta_deserialize.c + tests/fuzz/fuzz_delta_signature_deserialize.c + tests/fuzz/fuzz_glob_match.c + tests/fuzz/fuzz_metadata_from_buf.c + ) + foreach(FUZZ_SRC ${FUZZ_SRCS}) + get_filename_component(FUZZ_NAME ${FUZZ_SRC} NAME_WE) + add_executable(${FUZZ_NAME} ${FUZZ_SRC} ${SHARED_SRCS}) + configure_fastsync_target(${FUZZ_NAME}) + target_include_directories(${FUZZ_NAME} PRIVATE tests) + target_compile_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer) + target_link_options(${FUZZ_NAME} PRIVATE -fsanitize=fuzzer,address,undefined) + endforeach() endif() diff --git a/src/server/server.c b/src/server/server.c index ee34957..a0eb0f6 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -1,7 +1,4 @@ -#include "array_list.h" -#include "chunk.h" #include "config.h" -#include "data.h" #include "file.h" #include "log.h" #include "multiprocessing.h" @@ -29,9 +26,12 @@ static bool path_is_within(const char* root, const char* path) { return strncmp(root, path, n) == 0 && (path[n] == '\0' || path[n] == '/'); } -static bool valid_batch_path(const char* path) { - return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path) && - strchr(path, '\0') == path + strlen(path); +static bool save_received_file(File* file, void* context) { + Config* config = context; + if (config->save_to_disk && !file_save_to_disk(config->receive_root_directory, file, config)) + return false; + file_destroy(file); + return true; } static bool __attribute__((unused)) configure_authorization(const char* root) { @@ -53,116 +53,7 @@ static bool __attribute__((unused)) configure_authorization(const char* root) { } int receive_files(Config* config, int fd) { - Status status; - if (!receive_status(fd, &status)) - return -1; - - while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK || - status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) { - if (status == STATUS_KEEPALIVE) { - send_status(fd, STATUS_KEEPALIVE); - goto next; - } - if (status == STATUS_ABORT) { - log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up"); - return -1; - } - if (status == STATUS_CHECK) { - bool skipped; - File* file = receive_incremental_check(fd, config, &skipped); - if (skipped) - goto next; - if (file == NULL && !skipped) - return -1; - if (config->save_to_disk && - !file_save_to_disk(config->receive_root_directory, file, config)) { - file_destroy(file); - send_status(fd, STATUS_ERROR); - return -1; - } - file_destroy(file); - } else if (status == STATUS_CHUNK) { - Chunk* chunk = receive_chunk_data(fd, config); - if (chunk == NULL) { - send_status(fd, STATUS_ERROR); - return -1; - } - for (int i = 0; i < chunk->element_count; i++) { - if (config->save_to_disk && - !file_save_to_disk(config->receive_root_directory, chunk->items[i], config)) { - chunk_destroy(chunk); - send_status(fd, STATUS_ERROR); - return -1; - } - } - chunk_destroy(chunk); - } else if (status == STATUS_CHECK_BATCH) { - int count; - /* Batch framing has no checksum field yet; never silently downgrade a - checksum-enabled transfer into mtime-only matching. */ - if (config->checksum || !receive_int(fd, &count) || count < 0 || count > MAX_MANIFEST_ENTRIES) - return -1; - for (int i = 0; i < count; i++) { - char* check_path = receive_str(fd); - if (!check_path) - return -1; - unsigned long long check_size; - long long check_mtime; - if (!receive_n_data(fd, &check_size, sizeof(check_size)) || - !receive_n_data(fd, &check_mtime, sizeof(check_mtime))) { - free(check_path); - return -1; - } - if (!valid_batch_path(check_path)) { - free(check_path); - send_status(fd, STATUS_ERROR); - return -1; - } - char* full_path = path_cat(config->receive_root_directory, check_path); - struct stat st; - bool has_old = full_path && lstat(full_path, &st) == 0; - bool match = has_old && (unsigned long long)st.st_size == check_size && - (long long)st.st_mtime == check_mtime; - bool sent = send_status(fd, match ? STATUS_OK : STATUS_NEXT); - free(full_path); - free(check_path); - if (!sent) - return -1; - } - goto next; - } else { - File* file = file_receive(config, fd); - if (file == NULL) { - log_message(LOG_LEVEL_ERROR, "Failed to receive file"); - send_status(fd, STATUS_ERROR); - return -1; - } - if (config->save_to_disk && - !file_save_to_disk(config->receive_root_directory, file, config)) { - file_destroy(file); - send_status(fd, STATUS_ERROR); - return -1; - } - file_destroy(file); - } - next: - if (!receive_status(fd, &status)) { - send_status(fd, STATUS_ERROR); - return -1; - } - } - - if (status == STATUS_MANIFEST) { - if (receive_manifest(fd, config, &status) != 0) - return -1; - } - if (status != STATUS_FINISHED) { - log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status"); - send_status(fd, STATUS_ERROR); - return -1; - } - send_status(fd, STATUS_OK); - return 0; + return receive_files_common(config, fd, save_received_file, config, true); } void handler(int file_descriptor) { diff --git a/src/shared/config.c b/src/shared/config.c index 4c1a3a0..28fe5d4 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -173,103 +173,39 @@ void config_delete(Config* config) { free(config); } -/* Wire format order (must match config_receive and be updated when PROTOCOL_VERSION bumps): - * version, send_directory, receive_root_directory, save_to_disk, use_multithreading, - * use_chunk_serialization, use_compression, use_metadata, compression_level, chunk_size, - * use_sendfile, use_delete, use_incremental, use_delta, delta_block_size, delta_max_file_size, - * backup, backup_dir, follow_symlinks, copy_links, safe_links, copy_unsafe_links, - * preserve_hard_links, preserve_acls, preserve_xattrs, preserve_devices, preserve_sparse, - * update, inplace, append, append_verify, delete_excluded, delete_after, max_delete, relative, - * prune_empty_dirs, temp_dir, partial, partial_dir, suffix, delete_before, checksum, - * compress_choice, status - */ +static bool config_send_string(int file_descriptor, const char* value, bool optional) { + return send_str(file_descriptor, value ? value : (optional ? "" : NULL)); +} + +#define CONFIG_SEND_VERSION(field) \ + do { \ + if (!config_send_string(file_descriptor, config->field, false)) \ + return false; \ + } while (0); +#define CONFIG_SEND_STRING(field) \ + do { \ + if (!config_send_string(file_descriptor, config->field, false)) \ + return false; \ + } while (0); +#define CONFIG_SEND_OPTIONAL_STRING(field) \ + do { \ + if (!config_send_string(file_descriptor, config->field, true)) \ + return false; \ + } while (0); +#define CONFIG_SEND_INTEGER(field) \ + do { \ + if (!send_int(file_descriptor, config->field)) \ + return false; \ + } while (0); +#define CONFIG_SEND_DATA(field) \ + do { \ + if (!send_n_data(file_descriptor, &config->field, sizeof(config->field))) \ + return false; \ + } while (0); + bool config_send(int file_descriptor, const Config* config) { - if (!send_str(file_descriptor, config->version)) - return false; - if (!send_str(file_descriptor, config->send_directory)) - return false; - if (!send_str(file_descriptor, config->receive_root_directory)) - return false; - if (!send_int(file_descriptor, config->save_to_disk)) - return false; - if (!send_int(file_descriptor, config->use_multithreading)) - return false; - if (!send_int(file_descriptor, config->use_chunk_serialization)) - return false; - if (!send_int(file_descriptor, config->use_compression)) - return false; - if (!send_int(file_descriptor, config->use_metadata)) - return false; - if (!send_int(file_descriptor, config->compression_level)) - return false; - if (!send_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size))) - return false; - if (!send_int(file_descriptor, config->use_sendfile)) - return false; - if (!send_int(file_descriptor, config->use_delete)) - return false; - if (!send_int(file_descriptor, config->use_incremental)) - return false; - if (!send_int(file_descriptor, config->use_delta)) - return false; - if (!send_n_data(file_descriptor, &config->delta_block_size, sizeof(config->delta_block_size))) - return false; - if (!send_n_data(file_descriptor, &config->delta_max_file_size, sizeof(unsigned long long))) - return false; - if (!send_int(file_descriptor, config->backup)) - return false; - if (!send_str(file_descriptor, config->backup_dir ? config->backup_dir : "")) - return false; - if (!send_int(file_descriptor, config->follow_symlinks)) - return false; - if (!send_int(file_descriptor, config->copy_links)) - return false; - if (!send_int(file_descriptor, config->safe_links)) - return false; - if (!send_int(file_descriptor, config->copy_unsafe_links)) - return false; - if (!send_int(file_descriptor, config->preserve_hard_links)) - return false; - if (!send_int(file_descriptor, config->preserve_acls)) - return false; - if (!send_int(file_descriptor, config->preserve_xattrs)) - return false; - if (!send_int(file_descriptor, config->preserve_devices)) - return false; - if (!send_int(file_descriptor, config->preserve_sparse)) - return false; - if (!send_int(file_descriptor, config->update)) - return false; - if (!send_int(file_descriptor, config->inplace)) - return false; - if (!send_int(file_descriptor, config->append)) - return false; - if (!send_int(file_descriptor, config->append_verify)) - return false; - if (!send_int(file_descriptor, config->delete_excluded)) - return false; - if (!send_int(file_descriptor, config->delete_after)) - return false; - if (!send_n_data(file_descriptor, &config->max_delete, sizeof(config->max_delete))) - return false; - if (!send_int(file_descriptor, config->relative)) - return false; - if (!send_int(file_descriptor, config->prune_empty_dirs)) - return false; - if (!send_str(file_descriptor, config->temp_dir ? config->temp_dir : "")) - return false; - if (!send_int(file_descriptor, config->partial)) - return false; - if (!send_str(file_descriptor, config->partial_dir ? config->partial_dir : "")) - return false; - if (!send_str(file_descriptor, config->suffix ? config->suffix : "")) - return false; - if (!send_int(file_descriptor, config->delete_before)) - return false; - if (!send_int(file_descriptor, config->checksum)) - return false; - if (!send_str(file_descriptor, config->compress_choice ? config->compress_choice : "")) - return false; + CONFIG_WIRE_FIELDS(CONFIG_SEND_VERSION, CONFIG_SEND_STRING, CONFIG_SEND_OPTIONAL_STRING, + CONFIG_SEND_INTEGER, CONFIG_SEND_DATA) Status status; if (!receive_status(file_descriptor, &status)) return false; @@ -280,160 +216,60 @@ bool config_send(int file_descriptor, const Config* config) { return true; } -/* Wire format order: see the comment above config_send. */ +#undef CONFIG_SEND_VERSION +#undef CONFIG_SEND_STRING +#undef CONFIG_SEND_OPTIONAL_STRING +#undef CONFIG_SEND_INTEGER +#undef CONFIG_SEND_DATA + +static bool config_receive_string(int file_descriptor, char** destination) { + char* value = receive_str(file_descriptor); + if (!value) + return false; + *destination = value; + return true; +} + +#define CONFIG_RECEIVE_VERSION(field) \ + do { \ + free(config->field); \ + config->field = receive_str(file_descriptor); \ + if (!config->field) \ + goto error; \ + if (strcmp(config->field, PROTOCOL_VERSION) != 0) { \ + fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", config->field, \ + PROTOCOL_VERSION); \ + config_delete(config); \ + send_status(file_descriptor, STATUS_ERROR); \ + return NULL; \ + } \ + } while (0); +#define CONFIG_RECEIVE_STRING(field) \ + do { \ + if (!config_receive_string(file_descriptor, &config->field)) \ + goto error; \ + } while (0); +#define CONFIG_RECEIVE_OPTIONAL_STRING(field) CONFIG_RECEIVE_STRING(field) +#define CONFIG_RECEIVE_INTEGER(field) \ + do { \ + if (!receive_int(file_descriptor, &tmp)) \ + goto error; \ + config->field = tmp; \ + } while (0); +#define CONFIG_RECEIVE_DATA(field) \ + do { \ + if (!receive_n_data(file_descriptor, &config->field, sizeof(config->field))) \ + goto error; \ + } while (0); + Config* config_receive(int file_descriptor) { Config* config = (Config*)malloc(sizeof(Config)); if (config == NULL) return NULL; config_set_defaults(config); - free(config->version); - config->version = receive_str(file_descriptor); - if (!config->version) { - free(config->server_host); - free(config); - return NULL; - } - if (strcmp(config->version, PROTOCOL_VERSION) != 0) { - fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", config->version, - PROTOCOL_VERSION); - free(config->version); - free(config->server_host); - free(config); - send_status(file_descriptor, STATUS_ERROR); - return NULL; - } - config->send_directory = receive_str(file_descriptor); - if (!config->send_directory) { - free(config->version); - free(config->server_host); - free(config); - return NULL; - } - config->receive_root_directory = receive_str(file_descriptor); - if (!config->receive_root_directory) { - free(config->version); - free(config->send_directory); - free(config->server_host); - free(config); - return NULL; - } int tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->save_to_disk = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_multithreading = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_chunk_serialization = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_compression = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_metadata = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->compression_level = tmp; - if (!receive_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size))) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_sendfile = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_delete = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_incremental = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->use_delta = tmp; - if (!receive_n_data(file_descriptor, &config->delta_block_size, sizeof(config->delta_block_size))) - goto error; - if (!receive_n_data(file_descriptor, &config->delta_max_file_size, sizeof(unsigned long long))) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->backup = tmp; - config->backup_dir = receive_str(file_descriptor); - if (config->backup_dir == NULL) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->follow_symlinks = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->copy_links = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->safe_links = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->copy_unsafe_links = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->preserve_hard_links = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->preserve_acls = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->preserve_xattrs = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->preserve_devices = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->preserve_sparse = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->update = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->inplace = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->append = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->append_verify = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->delete_excluded = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->delete_after = tmp; - if (!receive_n_data(file_descriptor, &config->max_delete, sizeof(config->max_delete))) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->relative = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->prune_empty_dirs = tmp; - config->temp_dir = receive_str(file_descriptor); - if (config->temp_dir == NULL) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->partial = tmp; - config->partial_dir = receive_str(file_descriptor); - if (config->partial_dir == NULL) - goto error; - config->suffix = receive_str(file_descriptor); - if (config->suffix == NULL) - goto error; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->delete_before = tmp; - if (!receive_int(file_descriptor, &tmp)) - goto error; - config->checksum = tmp; - config->compress_choice = receive_str(file_descriptor); - if (config->compress_choice == NULL) - goto error; + CONFIG_WIRE_FIELDS(CONFIG_RECEIVE_VERSION, CONFIG_RECEIVE_STRING, CONFIG_RECEIVE_OPTIONAL_STRING, + CONFIG_RECEIVE_INTEGER, CONFIG_RECEIVE_DATA) if (config->compress_choice[0] != '\0' && strcmp(config->compress_choice, "zstd") != 0 && strcmp(config->compress_choice, "none") != 0) { fprintf(stderr, "Unsupported compression choice: %s\n", config->compress_choice); @@ -445,15 +281,12 @@ Config* config_receive(int file_descriptor) { return config; error: - free(config->version); - free(config->send_directory); - free(config->receive_root_directory); - free(config->server_host); - free(config->backup_dir); - free(config->temp_dir); - free(config->partial_dir); - free(config->suffix); - free(config->compress_choice); - free(config); + config_delete(config); return NULL; } + +#undef CONFIG_RECEIVE_VERSION +#undef CONFIG_RECEIVE_STRING +#undef CONFIG_RECEIVE_OPTIONAL_STRING +#undef CONFIG_RECEIVE_INTEGER +#undef CONFIG_RECEIVE_DATA diff --git a/src/shared/config.h b/src/shared/config.h index 986a5c4..83e4495 100644 --- a/src/shared/config.h +++ b/src/shared/config.h @@ -128,6 +128,53 @@ typedef struct Config { char* compress_choice; } Config; +/* Keep the on-wire field order in one place. The first three strings require + * values when sent; optional strings are encoded as empty strings when NULL. */ +#define CONFIG_WIRE_FIELDS(VERSION, STRING, OPTIONAL_STRING, INTEGER, DATA) \ + VERSION(version) \ + STRING(send_directory) \ + STRING(receive_root_directory) \ + INTEGER(save_to_disk) \ + INTEGER(use_multithreading) \ + INTEGER(use_chunk_serialization) \ + INTEGER(use_compression) \ + INTEGER(use_metadata) \ + INTEGER(compression_level) \ + DATA(chunk_size) \ + INTEGER(use_sendfile) \ + INTEGER(use_delete) \ + INTEGER(use_incremental) \ + INTEGER(use_delta) \ + DATA(delta_block_size) \ + DATA(delta_max_file_size) \ + INTEGER(backup) \ + OPTIONAL_STRING(backup_dir) \ + INTEGER(follow_symlinks) \ + INTEGER(copy_links) \ + INTEGER(safe_links) \ + INTEGER(copy_unsafe_links) \ + INTEGER(preserve_hard_links) \ + INTEGER(preserve_acls) \ + INTEGER(preserve_xattrs) \ + INTEGER(preserve_devices) \ + INTEGER(preserve_sparse) \ + INTEGER(update) \ + INTEGER(inplace) \ + INTEGER(append) \ + INTEGER(append_verify) \ + INTEGER(delete_excluded) \ + INTEGER(delete_after) \ + DATA(max_delete) \ + INTEGER(relative) \ + INTEGER(prune_empty_dirs) \ + OPTIONAL_STRING(temp_dir) \ + INTEGER(partial) \ + OPTIONAL_STRING(partial_dir) \ + OPTIONAL_STRING(suffix) \ + INTEGER(delete_before) \ + INTEGER(checksum) \ + OPTIONAL_STRING(compress_choice) + #define PROTOCOL_VERSION "2.2.0" #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024) diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 5edbec2..7b9c2f0 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -13,11 +13,107 @@ #include #include #include +#include static bool valid_batch_path(const char* path) { return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path); } +static bool handle_batch_checks(int file_descriptor, const Config* config) { + int count; + if (config->checksum || !receive_int(file_descriptor, &count) || count < 0 || + count > MAX_MANIFEST_ENTRIES) + return false; + for (int i = 0; i < count; i++) { + char* check_path = receive_str(file_descriptor); + if (!check_path) + return false; + unsigned long long check_size; + long long check_mtime; + bool received = receive_n_data(file_descriptor, &check_size, sizeof(check_size)) && + receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime)); + if (!received || !valid_batch_path(check_path)) { + free(check_path); + if (received) + send_status(file_descriptor, STATUS_ERROR); + return false; + } + char* full_path = path_cat(config->receive_root_directory, check_path); + struct stat st; + bool has_old = full_path && lstat(full_path, &st) == 0; + bool match = has_old && (unsigned long long)st.st_size == check_size && + (long long)st.st_mtime == check_mtime; + bool sent = send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT); + free(full_path); + free(check_path); + if (!sent) + return false; + } + return true; +} + +int receive_files_common(const Config* config, int file_descriptor, ReceivedFileHandler handler, + void* context, bool send_completion_status) { + Status status; + if (!receive_status(file_descriptor, &status)) + return -1; + while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK || + status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) { + if (status == STATUS_KEEPALIVE) { + if (!send_status(file_descriptor, STATUS_KEEPALIVE)) + return -1; + } else if (status == STATUS_ABORT) { + log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up"); + return -1; + } else if (status == STATUS_CHECK_BATCH) { + if (!handle_batch_checks(file_descriptor, config)) + return -1; + } else { + if (status == STATUS_CHUNK) { + Chunk* chunk = receive_chunk_data(file_descriptor, config); + if (!chunk) + return -1; + for (int i = 0; i < chunk->element_count; i++) { + File* file = chunk->items[i]; + chunk->items[i] = NULL; + if (!handler(file, context)) { + file_destroy(file); + chunk_destroy(chunk); + return -1; + } + } + chunk_destroy(chunk); + } else { + bool skipped = false; + File* file = status == STATUS_CHECK + ? receive_incremental_check(file_descriptor, config, &skipped) + : file_receive(config, file_descriptor); + if (!skipped) { + if (!file || !handler(file, context)) { + file_destroy(file); + if (status == STATUS_NEXT) + log_message(LOG_LEVEL_ERROR, "Failed to receive file"); + return -1; + } + } + } + } + if (!receive_status(file_descriptor, &status)) + return -1; + } + if (status == STATUS_MANIFEST && receive_manifest(file_descriptor, config, &status) != 0) + return -1; + if (status != STATUS_FINISHED) { + log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status"); + if (send_completion_status) + send_status(file_descriptor, STATUS_ERROR); + return -1; + } + if (send_completion_status && !send_status(file_descriptor, STATUS_OK)) + return -1; + return 0; +} + PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner, Queue* queue_loader) { PipelineContextSender* context = malloc(sizeof(PipelineContextSender)); @@ -137,41 +233,14 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver* context) { free(context); } -static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* context) { - Chunk* chunk = receive_chunk_data(file_descriptor, context->config); - if (chunk == NULL) - return false; - - for (int i = 0; i < chunk->element_count; i++) { - File* file = chunk->items[i]; - chunk->items[i] = NULL; - if (!queue_enqueue_multithreaded_cancel(context->queue, file, &context->mutex, - &context->condition_not_empty, - &context->condition_not_full, &context->cancelled)) { - file_destroy(file); - chunk_destroy(chunk); - return false; - } - } - chunk_destroy(chunk); - return true; -} - -static void receiver_thread_fail(PipelineContextReceiver* context) { - mtx_lock(&context->mutex); - atomic_store(&context->cancelled, true); - context->receiver_done = true; - cnd_broadcast(&context->condition_not_empty); - cnd_broadcast(&context->condition_not_full); - mtx_unlock(&context->mutex); +static bool enqueue_received_file(File* file, void* context) { + PipelineContextReceiver* receiver = context; + return queue_enqueue_multithreaded_cancel(receiver->queue, file, &receiver->mutex, + &receiver->condition_not_empty, + &receiver->condition_not_full, &receiver->cancelled); } int receive_thread(void* pipeline_context) { -#define RECEIVE_THREAD_FAIL() \ - do { \ - receiver_thread_fail(context); \ - return thrd_error; \ - } while (0) PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context; if (context->ssl) io_set_ssl(context->ssl); @@ -180,100 +249,15 @@ int receive_thread(void* pipeline_context) { const Config* config = context->config; mtx_unlock(&context->mutex); - Status status; - if (!receive_status(file_descriptor, &status)) - RECEIVE_THREAD_FAIL(); - while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK || - status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) { - if (status == STATUS_KEEPALIVE) { - if (!send_status(file_descriptor, STATUS_KEEPALIVE)) - RECEIVE_THREAD_FAIL(); - goto next; - } - if (status == STATUS_ABORT) { - log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up"); - RECEIVE_THREAD_FAIL(); - } - if (status == STATUS_CHECK) { - bool skipped; - File* file = receive_incremental_check(file_descriptor, config, &skipped); - if (!skipped) { - if (file == NULL) - RECEIVE_THREAD_FAIL(); - if (!queue_enqueue_multithreaded_cancel( - context->queue, file, &context->mutex, &context->condition_not_empty, - &context->condition_not_full, &context->cancelled)) { - file_destroy(file); - RECEIVE_THREAD_FAIL(); - } - } - } else if (status == STATUS_CHUNK) { - if (!receive_chunk_enqueue(file_descriptor, context)) - RECEIVE_THREAD_FAIL(); - } else if (status == STATUS_CHECK_BATCH) { - int count; - if (config->checksum || !receive_int(file_descriptor, &count) || count < 0 || - count > MAX_MANIFEST_ENTRIES) - RECEIVE_THREAD_FAIL(); - for (int i = 0; i < count; i++) { - char* check_path = receive_str(file_descriptor); - if (!check_path) - RECEIVE_THREAD_FAIL(); - unsigned long long check_size; - long long check_mtime; - if (!receive_n_data(file_descriptor, &check_size, sizeof(check_size)) || - !receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime))) { - free(check_path); - RECEIVE_THREAD_FAIL(); - } - if (!valid_batch_path(check_path)) { - free(check_path); - if (!send_status(file_descriptor, STATUS_ERROR)) - RECEIVE_THREAD_FAIL(); - RECEIVE_THREAD_FAIL(); - } - char* full_path = path_cat(config->receive_root_directory, check_path); - struct stat st; - bool has_old = full_path && lstat(full_path, &st) == 0; - bool match = has_old && (unsigned long long)st.st_size == check_size && - (long long)st.st_mtime == check_mtime; - if (!send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT)) - RECEIVE_THREAD_FAIL(); - free(full_path); - free(check_path); - } - goto next; - } else { - File* file = file_receive(config, file_descriptor); - if (file) { - if (!queue_enqueue_multithreaded_cancel( - context->queue, file, &context->mutex, &context->condition_not_empty, - &context->condition_not_full, &context->cancelled)) { - file_destroy(file); - receiver_thread_fail(context); - return thrd_error; - } - } else { - log_message(LOG_LEVEL_ERROR, "Failed to receive file"); - RECEIVE_THREAD_FAIL(); - } - } - next: - if (!receive_status(file_descriptor, &status)) - RECEIVE_THREAD_FAIL(); - } - if (status == STATUS_MANIFEST) { - if (receive_manifest(file_descriptor, config, &status) != 0) - RECEIVE_THREAD_FAIL(); - } - if (status != STATUS_FINISHED) - RECEIVE_THREAD_FAIL(); + int result = receive_files_common(config, file_descriptor, enqueue_received_file, context, false); mtx_lock(&context->mutex); + if (result != 0) + atomic_store(&context->cancelled, true); context->receiver_done = true; cnd_signal(&context->condition_not_empty); + cnd_broadcast(&context->condition_not_full); mtx_unlock(&context->mutex); -#undef RECEIVE_THREAD_FAIL - return thrd_success; + return result == 0 ? thrd_success : thrd_error; } int write_thread(void* pipeline_context) { diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index 348b955..cfdb516 100644 --- a/src/shared/multiprocessing.h +++ b/src/shared/multiprocessing.h @@ -42,6 +42,11 @@ typedef struct PipelineContextReceiver { atomic_bool cancelled; } PipelineContextReceiver; +typedef bool (*ReceivedFileHandler)(File* file, void* context); + +int receive_files_common(const Config* config, int file_descriptor, ReceivedFileHandler handler, + void* context, bool send_completion_status); + PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner, Queue* queue_loader); void pipeline_context_sender_destroy(PipelineContextSender* context); diff --git a/src/shared/transport_tcp.c b/src/shared/transport_tcp.c index 49c2fb9..bbd6621 100644 --- a/src/shared/transport_tcp.c +++ b/src/shared/transport_tcp.c @@ -173,7 +173,7 @@ Client* client_create() { return client; } -bool client_connect(Client* client, char* host, int port) { +bool tcp_connect_socket(Client* client, const char* host, int port) { struct addrinfo hints; struct addrinfo* result; memset(&hints, 0, sizeof(hints)); @@ -226,6 +226,10 @@ bool client_connect(Client* client, char* host, int port) { return true; } +bool client_connect(Client* client, char* host, int port) { + return tcp_connect_socket(client, host, port); +} + void client_disconnect(Client* client) { if (client->ssl) { SSL_shutdown(client->ssl); diff --git a/src/shared/transport_tcp.h b/src/shared/transport_tcp.h index d74d694..09cf6b9 100644 --- a/src/shared/transport_tcp.h +++ b/src/shared/transport_tcp.h @@ -29,6 +29,7 @@ void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* chil const char* log_fmt); void server_delete(Server** server); Client* client_create(); +bool tcp_connect_socket(Client* client, const char* host, int port); bool client_connect(Client* client, char* host, int port); void client_disconnect(Client* client); void client_delete(Client* client); diff --git a/src/shared/transport_tls.c b/src/shared/transport_tls.c index 49fc554..c294e41 100644 --- a/src/shared/transport_tls.c +++ b/src/shared/transport_tls.c @@ -2,8 +2,6 @@ #include "log.h" #include "protocol.h" #include "transport_tcp.h" -#include -#include #include #include #include @@ -149,53 +147,8 @@ bool server_listen_tls(Server* server, void (*handler)(int file_descriptor)) { bool client_connect_tls(Client* client, char* host, int port, const char* cert_path, const char* key_path, const char* ca_path) { - struct addrinfo hints; - struct addrinfo* result; - memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_UNSPEC; - hints.ai_socktype = SOCK_STREAM; - hints.ai_protocol = IPPROTO_TCP; - - char port_str[16]; - snprintf(port_str, sizeof(port_str), "%d", port); - - int err = getaddrinfo(host, port_str, &hints, &result); - if (err != 0 || result == NULL) { - fprintf(stderr, "Could not resolve host: %s (%s)\n", host, gai_strerror(err)); + if (!tcp_connect_socket(client, host, port)) return false; - } - - struct addrinfo* rp; - bool connected = false; - for (rp = result; rp != NULL; rp = rp->ai_next) { - if (client->file_descriptor >= 0) - close(client->file_descriptor); - - client->file_descriptor = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); - if (client->file_descriptor < 0) - continue; - - struct timeval ct; - ct.tv_sec = tcp_get_contimeout_sec(); - ct.tv_usec = 0; - setsockopt(client->file_descriptor, SOL_SOCKET, SO_RCVTIMEO, &ct, sizeof(ct)); - setsockopt(client->file_descriptor, SOL_SOCKET, SO_SNDTIMEO, &ct, sizeof(ct)); - - memcpy(&client->address, rp->ai_addr, rp->ai_addrlen); - client->address_length = rp->ai_addrlen; - - if (connect(client->file_descriptor, (struct sockaddr*)&client->address, - client->address_length) == 0) { - connected = true; - break; - } - } - freeaddrinfo(result); - - if (!connected) { - perror("Could not connect to Server!"); - return false; - } SSL_CTX* ctx = create_ssl_ctx(false, cert_path, key_path, ca_path); if (!ctx) -- 2.54.0