diff --git a/CMakeLists.txt b/CMakeLists.txt index 42c2353..9b8da1c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -58,15 +58,18 @@ endif() find_package(OpenSSL REQUIRED) file(GLOB SHARED_SRCS "src/shared/*.c") +set(FILE_STORE_SRCS src/shared/file_store.c) +list(REMOVE_ITEM SHARED_SRCS ${FILE_STORE_SRCS}) file(GLOB SERVER_SRCS "src/server/*.c") +set(SERVER_RECEIVER_SRCS src/server/receiver.c) file(GLOB CLIENT_SRCS "src/client/*.c") # --- Main executables --- -add_executable(server ${SERVER_SRCS} ${SHARED_SRCS}) +add_executable(server ${SERVER_SRCS} ${SHARED_SRCS} ${FILE_STORE_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) -add_executable(client ${CLIENT_SRCS} ${SHARED_SRCS}) +add_executable(client ${CLIENT_SRCS} ${SHARED_SRCS} ${FILE_STORE_SRCS} ${SERVER_RECEIVER_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) @@ -79,7 +82,7 @@ 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") -add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} src/client/scanner.c src/client/client_cli.c) +add_executable(tests ${TEST_SRCS} ${SHARED_SRCS} ${FILE_STORE_SRCS} ${SERVER_RECEIVER_SRCS} src/client/scanner.c src/client/client_cli.c src/client/client_validation.c src/client/usage.c) target_include_directories(tests PRIVATE ${TEST_INCLUDES}) target_compile_definitions(tests PRIVATE FASTSYNC_TEST_BUILD) target_link_libraries(tests PRIVATE ${TEST_LIBS}) @@ -94,7 +97,7 @@ if(ENABLE_FUZZ) 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}) + add_executable(${FUZZ_NAME} ${FUZZ_SRC} ${SHARED_SRCS} ${FILE_STORE_SRCS} ${SERVER_RECEIVER_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) diff --git a/src/client/client_cli.c b/src/client/client_cli.c index c00a9f2..0e51ed4 100644 --- a/src/client/client_cli.c +++ b/src/client/client_cli.c @@ -1,10 +1,12 @@ #include "client_send.h" +#include "client_validation.h" #include "config.h" #include "delta.h" #include "log.h" #include "protocol.h" #include "transport_tcp.h" #include "transport_tls.h" +#include "usage.h" #include "utils.h" #include #include @@ -12,7 +14,6 @@ #include #include #include -#include "usage.c" #ifndef FASTSYNC_TEST_BUILD /* Parse environment variables for source/destination directories and save-to-disk flag. */ @@ -360,55 +361,6 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args, return 0; } -#ifndef FASTSYNC_TEST_BUILD -/* Validate config after parsing. Returns true if valid. */ -static bool validate_config(const Config* config) { - if (!config->send_directory || !config->receive_root_directory) { - fprintf(stderr, "Error: source and destination directories are required\n"); - print_usage(); - return false; - } - if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) { - fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk " - "serialization)\n"); - return false; - } - if (config->transport == TRANSPORT_SSH && config->use_sendfile) { - fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); - return false; - } - if (config->use_incremental && config->use_chunk_serialization) { - fprintf(stderr, "Error: --incremental is not supported with -s (chunk serialization)\n"); - return false; - } - if (config->use_delta && !config->use_incremental) { - fprintf(stderr, "Error: --delta requires --incremental\n"); - return false; - } - if (config->use_delta && config->use_chunk_serialization) { - fprintf(stderr, "Error: --delta cannot be combined with -s (chunk serialization)\n"); - return false; - } - if (config->use_delta && config->use_sendfile) { - fprintf(stderr, "Error: --delta cannot be combined with -f (sendfile)\n"); - return false; - } - if (config->append || config->append_verify) { - fprintf( - stderr, - "Error: --append and --append-verify are not supported yet; refusing to ignore option\n"); - return false; - } - if (config->use_tls) { - if (!config->tls_cert || !config->tls_key) { - fprintf(stderr, "Error: --tls requires --cert and --key\n"); - return false; - } - } - return true; -} -#endif /* FASTSYNC_TEST_BUILD */ - static int read_patterns_from_file(const char* filepath, char*** patterns, int* count) { FILE* fp = fopen(filepath, "r"); if (!fp) { diff --git a/src/client/client_send.c b/src/client/client_send.c index ac2bc09..b010866 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -28,6 +28,88 @@ /* Forward declaration for progress-reporting thread used in multithreaded send. */ static int progress_thread_fn(void* arg); +static ScannerOptions scanner_options_from_config(const Config* config, int num_threads) { + ScannerOptions options = { + config->use_metadata, config->chunk_size, config->exclude_patterns, + config->exclude_count, config->include_patterns, config->include_count, + config->max_size, config->min_size, config->max_depth, + num_threads, config->follow_symlinks, config->copy_links, + config->safe_links, config->copy_unsafe_links, config->checksum}; + return options; +} + +/* Select the configured transport for both transfer execution paths. */ +static Client* connect_transfer_client(const Config* config) { + if (config->transport == TRANSPORT_SSH) { + if (config->use_sendfile) { + fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); + return NULL; + } + return client_connect_ssh(config->ssh_destination, config->ssh_port, + config->fastsync_server_path); + } + + Client* client = client_create(); + if (!client) + return NULL; + bool connected; + if (config->use_tls) { + connected = client_connect_tls(client, config->server_host, config->server_port, + config->tls_cert, config->tls_key, config->tls_ca); + } else { + connected = client_connect(client, config->server_host, config->server_port); + } + if (!connected) { + client_disconnect(client); + client_delete(client); + return NULL; + } + return client; +} + +static void disconnect_transfer_client(Client* client) { + if (!client) + return; + client_disconnect(client); + client_delete(client); +} + +static ArrayList* create_transfer_manifest(const Config* config) { + return config->use_delete ? array_list_create(free) : NULL; +} + +static bool add_chunk_to_manifest(ArrayList* manifest, const Chunk* chunk) { + if (!manifest) + return true; + for (int i = 0; i < chunk->element_count; i++) { + const char* path = chunk->items[i]->path; + if (*path == '/') + path++; + char* entry = str_dup(path); + if (!entry) { + log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); + return false; + } + if (!array_list_add(manifest, entry)) { + free(entry); + return false; + } + } + return true; +} + +static bool finalize_transfer(Client* client) { + Status status; + return send_status(client->file_descriptor, STATUS_FINISHED) && + receive_status(client->file_descriptor, &status) && status == STATUS_OK; +} + +static void mark_sender_done(PipelineContextSender* context) { + mtx_lock(&context->mutex_progress); + context->sender_done = true; + mtx_unlock(&context->mutex_progress); +} + static void pipeline_cancel(PipelineContextSender* context) { mtx_lock(&context->mutex_scanner); mtx_lock(&context->mutex_loader); @@ -43,12 +125,10 @@ static void pipeline_cancel(PipelineContextSender* context) { } /* Print dry-run manifest showing files that would be transferred. Returns 0 on success. */ -static int send_dry_run_manifest(Config* config) { - DirectoryScanner* scanner = directory_scanner_create( - config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, - config->exclude_count, config->include_patterns, config->include_count, config->max_size, - config->min_size, config->max_depth, config->follow_symlinks, config->copy_links, - config->safe_links, config->copy_unsafe_links, config->checksum); +static int send_dry_run_manifest(const Config* config) { + ScannerOptions options = scanner_options_from_config(config, 0); + DirectoryScanner* scanner = + directory_scanner_create_with_options(config->send_directory, &options); if (!scanner) return -1; Chunk* chunk; @@ -129,8 +209,11 @@ static int incremental_check(Client* client, File* file, const Config* config, static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* config) { Delta* delta = delta_compute(file->data->data, file->data->size, sig, config->delta_block_size); - if (!delta) + if (!delta) { + if (!send_status(client->file_descriptor, STATUS_NEXT)) + return -1; return 1; + } if (!delta_is_worthwhile(delta, file->data->size)) { delta_destroy(delta); @@ -291,49 +374,24 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) { static int send_chunks_multithreaded(void* pipeline_context) { PipelineContextSender* context = (PipelineContextSender*)pipeline_context; - Client* client; - if (context->config->transport == TRANSPORT_SSH) { - if (context->config->use_sendfile) { - fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); - return 1; - } - client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port, - context->config->fastsync_server_path); - } else if (context->config->use_tls) { - client = client_create(); - if (!client || !client_connect_tls(client, context->config->server_host, - context->config->server_port, context->config->tls_cert, - context->config->tls_key, context->config->tls_ca)) { - if (client) - client_delete(client); - fprintf(stderr, "Error: could not connect to server via TLS\n"); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); - return thrd_error; - } - } else { - client = client_create(); - if (!client || - !client_connect(client, context->config->server_host, context->config->server_port)) { - if (client) - client_delete(client); - fprintf(stderr, "Error: could not connect to server\n"); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); - return thrd_error; - } + Client* client = connect_transfer_client(context->config); + if (!client) { + if (context->config->transport == TRANSPORT_TCP) + fprintf(stderr, "Error: could not connect to server%s\n", + context->config->use_tls ? " via TLS" : ""); + pipeline_cancel(context); + mark_sender_done(context); + return thrd_error; } + ProtocolSession session; + protocol_session_init(&session, client->file_descriptor, client->file_descriptor); + protocol_session_set_ssl(&session, (SSL*)client->ssl); + protocol_session_bind(&session); if (!config_send(client->file_descriptor, context->config)) { - client_disconnect(client); - client_delete(client); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); + pipeline_cancel(context); + disconnect_transfer_client(client); + mark_sender_done(context); + protocol_session_unbind(); return thrd_error; } @@ -342,39 +400,37 @@ static int send_chunks_multithreaded(void* pipeline_context) { context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader, &context->condition_not_full_loader, &context->loader_done); if (current_chunk == NULL) { + if (atomic_load(&context->cancelled)) { + pipeline_cancel(context); + disconnect_transfer_client(client); + mark_sender_done(context); + protocol_session_unbind(); + return thrd_error; + } if (context->config->use_delete) { if (send_delete_manifest(client->file_descriptor, context->manifest) != 0) goto send_fail; } - if (!send_status(client->file_descriptor, STATUS_FINISHED)) - goto send_fail; - Status s; - int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK; - client_disconnect(client); - client_delete(client); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); + bool ok = finalize_transfer(client); + disconnect_transfer_client(client); + mark_sender_done(context); + protocol_session_unbind(); return ok ? thrd_success : thrd_error; send_fail: pipeline_cancel(context); - client_disconnect(client); - client_delete(client); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); + disconnect_transfer_client(client); + mark_sender_done(context); + protocol_session_unbind(); return thrd_error; } if (send_chunk(client, current_chunk, context->config) != 0) { fprintf(stderr, "Error: unexpected error while sending chunk\n"); chunk_destroy(current_chunk); pipeline_cancel(context); - client_disconnect(client); - client_delete(client); - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); + disconnect_transfer_client(client); + mark_sender_done(context); + protocol_session_unbind(); return thrd_error; } if (context->config->show_progress) { @@ -393,13 +449,9 @@ static int send_chunks_multithreaded(void* pipeline_context) { static int scan_directory_multithreaded(void* pipeline_context) { PipelineContextSender* context = (PipelineContextSender*)pipeline_context; - ParallelScanner* scanner = parallel_scanner_create( - context->config->send_directory, context->config->use_metadata, context->config->chunk_size, - context->config->exclude_patterns, context->config->exclude_count, - context->config->include_patterns, context->config->include_count, context->config->max_size, - context->config->min_size, context->config->max_depth, 4, context->config->follow_symlinks, - context->config->copy_links, context->config->safe_links, context->config->copy_unsafe_links, - context->config->checksum); + ScannerOptions options = scanner_options_from_config(context->config, 4); + ParallelScanner* scanner = + parallel_scanner_create_with_options(context->config->send_directory, &options); Chunk* current_chunk; if (scanner == NULL) { @@ -410,28 +462,14 @@ static int scan_directory_multithreaded(void* pipeline_context) { while ((current_chunk = parallel_scanner_next(scanner)) != NULL) { if (context->config->use_delete) { mtx_lock(&context->mutex_scanner); - for (int i = 0; i < current_chunk->element_count; i++) { - const char* p = current_chunk->items[i]->path; - if (*p == '/') - p++; - char* manifest_entry = str_dup(p); - if (!manifest_entry) { - log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); - mtx_unlock(&context->mutex_scanner); - pipeline_cancel(context); - parallel_scanner_destroy(scanner); - return thrd_error; - } - if (!array_list_add(context->manifest, manifest_entry)) { - free(manifest_entry); - mtx_unlock(&context->mutex_scanner); - pipeline_cancel(context); - chunk_destroy(current_chunk); - parallel_scanner_destroy(scanner); - return thrd_error; - } - } + bool manifest_ok = add_chunk_to_manifest(context->manifest, current_chunk); mtx_unlock(&context->mutex_scanner); + if (!manifest_ok) { + pipeline_cancel(context); + chunk_destroy(current_chunk); + parallel_scanner_destroy(scanner); + return thrd_error; + } } if (!queue_enqueue_multithreaded_cancel( context->queue_scanner, current_chunk, &context->mutex_scanner, @@ -481,9 +519,10 @@ static int load_files_multithreaded(void* pipeline_context) { if (f->data->size > STREAM_THRESHOLD) continue; if (!file_load_data(f)) { - log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping"); - file_destroy(f); - chunk->items[i] = NULL; + log_message(LOG_LEVEL_ERROR, "Failed to load file data"); + chunk_destroy(chunk); + pipeline_cancel(context); + return thrd_error; } } } @@ -492,9 +531,7 @@ static int load_files_multithreaded(void* pipeline_context) { &context->condition_not_full_loader, &context->cancelled)) { chunk_destroy(chunk); - atomic_store(&context->cancelled, true); - cnd_broadcast(&context->condition_not_full_loader); - cnd_broadcast(&context->condition_not_empty_loader); + pipeline_cancel(context); return thrd_error; } } @@ -540,61 +577,37 @@ int send_files(Config* config) { if (config->dry_run) return send_dry_run_manifest(config); - Client* client; - if (config->transport == TRANSPORT_SSH) { - if (config->use_sendfile) { - fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); - return 1; - } - client = - client_connect_ssh(config->ssh_destination, config->ssh_port, config->fastsync_server_path); - if (!client) - return 1; - } else if (config->use_tls) { - client = client_create(); - if (!client || !client_connect_tls(client, config->server_host, config->server_port, - config->tls_cert, config->tls_key, config->tls_ca)) { - if (client) { - client_disconnect(client); - client_delete(client); - } - fprintf(stderr, "Error: could not connect to server via TLS\n"); - return 1; - } - } else { - client = client_create(); - if (!client || !client_connect(client, config->server_host, config->server_port)) { - if (client) { - client_disconnect(client); - client_delete(client); - } - fprintf(stderr, "Error: could not connect to server\n"); - return 1; - } - } - if (!config_send(client->file_descriptor, config)) { - client_disconnect(client); - client_delete(client); + Client* client = connect_transfer_client(config); + if (!client) { + if (config->transport == TRANSPORT_TCP) + fprintf(stderr, "Error: could not connect to server%s\n", config->use_tls ? " via TLS" : ""); return 1; } - DirectoryScanner* scanner = directory_scanner_create( - config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, - config->exclude_count, config->include_patterns, config->include_count, config->max_size, - config->min_size, config->max_depth, config->follow_symlinks, config->copy_links, - config->safe_links, config->copy_unsafe_links, config->checksum); + ProtocolSession session; + protocol_session_init(&session, client->file_descriptor, client->file_descriptor); + protocol_session_set_ssl(&session, (SSL*)client->ssl); + protocol_session_bind(&session); + if (!config_send(client->file_descriptor, config)) { + disconnect_transfer_client(client); + protocol_session_unbind(); + return 1; + } + ScannerOptions scanner_options = scanner_options_from_config(config, 0); + DirectoryScanner* scanner = + directory_scanner_create_with_options(config->send_directory, &scanner_options); Chunk* current_chunk; unsigned long long total_bytes = 0; int total_files = 0; time_t last_progress = 0; time_t start = time(NULL); - ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL; + ArrayList* manifest = create_transfer_manifest(config); if (!scanner || (config->use_delete && !manifest)) { if (scanner) directory_scanner_destroy(scanner); if (manifest) array_list_delete(manifest); - client_disconnect(client); - client_delete(client); + disconnect_transfer_client(client); + protocol_session_unbind(); return 1; } while ((current_chunk = directory_scanner_next(scanner)) != NULL) { @@ -602,30 +615,10 @@ int send_files(Config* config) { for (int i = 0; i < current_chunk->element_count; i++) { chunk_bytes += current_chunk->items[i]->data->size; total_files++; - if (manifest) { - const char* p = current_chunk->items[i]->path; - if (*p == '/') - p++; - char* manifest_entry = str_dup(p); - if (!manifest_entry) { - log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); - chunk_destroy(current_chunk); - array_list_delete(manifest); - directory_scanner_destroy(scanner); - client_disconnect(client); - client_delete(client); - return 1; - } - if (!array_list_add(manifest, manifest_entry)) { - free(manifest_entry); - chunk_destroy(current_chunk); - array_list_delete(manifest); - directory_scanner_destroy(scanner); - client_disconnect(client); - client_delete(client); - return 1; - } - } + } + if (!add_chunk_to_manifest(manifest, current_chunk)) { + chunk_destroy(current_chunk); + goto send_fail; } if (!config->use_sendfile) { for (int i = 0; i < current_chunk->element_count; i++) { @@ -634,7 +627,8 @@ int send_files(Config* config) { continue; if (!file_load_data(f)) { log_message(LOG_LEVEL_ERROR, "Failed to load file data"); - continue; + chunk_destroy(current_chunk); + goto send_fail; } } } @@ -664,15 +658,13 @@ int send_files(Config* config) { if (config->use_delete) { if (send_delete_manifest(client->file_descriptor, manifest) != 0) { array_list_delete(manifest); + manifest = NULL; goto send_fail; } array_list_delete(manifest); manifest = NULL; } - if (!send_status(client->file_descriptor, STATUS_FINISHED)) - goto send_fail; - Status s; - int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK; + bool ok = finalize_transfer(client); double elapsed_total = difftime(time(NULL), start); if (config->show_progress) { double rate = elapsed_total > 0 ? total_bytes / (1048576.0 * elapsed_total) : 0; @@ -684,16 +676,16 @@ int send_files(Config* config) { rate); } directory_scanner_destroy(scanner); - client_disconnect(client); - client_delete(client); + disconnect_transfer_client(client); + protocol_session_unbind(); return ok ? 0 : 1; send_fail: if (manifest) array_list_delete(manifest); directory_scanner_destroy(scanner); - client_disconnect(client); - client_delete(client); + disconnect_transfer_client(client); + protocol_session_unbind(); return 1; } @@ -729,7 +721,11 @@ int send_files_multithreaded(Config* config) { return 1; } if (config->use_delete) - context->manifest = array_list_create(free); + context->manifest = create_transfer_manifest(config); + if (config->use_delete && !context->manifest) { + pipeline_context_sender_destroy(context); + return 1; + } thrd_t scanner, loader, sender; bool scanner_created = false; diff --git a/src/client/client_validation.c b/src/client/client_validation.c new file mode 100644 index 0000000..0ac4138 --- /dev/null +++ b/src/client/client_validation.c @@ -0,0 +1,50 @@ +#include "client_validation.h" +#include "usage.h" +#include + +/* Validate config after parsing. Returns true if valid. */ +bool validate_config(const Config* config) { + if (!config->send_directory || !config->receive_root_directory) { + fprintf(stderr, "Error: source and destination directories are required\n"); + print_usage(); + return false; + } + if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) { + fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk " + "serialization)\n"); + return false; + } + if (config->transport == TRANSPORT_SSH && config->use_sendfile) { + fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); + return false; + } + if (config->use_incremental && config->use_chunk_serialization) { + fprintf(stderr, "Error: --incremental is not supported with -s (chunk serialization)\n"); + return false; + } + if (config->use_delta && !config->use_incremental) { + fprintf(stderr, "Error: --delta requires --incremental\n"); + return false; + } + if (config->use_delta && config->use_chunk_serialization) { + fprintf(stderr, "Error: --delta cannot be combined with -s (chunk serialization)\n"); + return false; + } + if (config->use_delta && config->use_sendfile) { + fprintf(stderr, "Error: --delta cannot be combined with -f (sendfile)\n"); + return false; + } + if (config->append || config->append_verify) { + fprintf( + stderr, + "Error: --append and --append-verify are not supported yet; refusing to ignore option\n"); + return false; + } + if (config->use_tls) { + if (!config->tls_cert || !config->tls_key) { + fprintf(stderr, "Error: --tls requires --cert and --key\n"); + return false; + } + } + return true; +} diff --git a/src/client/client_validation.h b/src/client/client_validation.h new file mode 100644 index 0000000..e830ccb --- /dev/null +++ b/src/client/client_validation.h @@ -0,0 +1,9 @@ +#ifndef CLIENT_VALIDATION_H +#define CLIENT_VALIDATION_H + +#include "config.h" +#include + +bool validate_config(const Config* config); + +#endif diff --git a/src/client/scanner.c b/src/client/scanner.c index a2ea713..3f50a79 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -52,13 +52,84 @@ static bool safe_relative_link(const char* source_root, const char* containing_d return safe; } -DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_metadata, - unsigned long long chunk_size, char** exclude_patterns, - int exclude_count, char** include_patterns, - int include_count, unsigned long long max_size, - unsigned long long min_size, int max_depth, - bool follow_symlinks, bool copy_links, bool safe_links, - bool copy_unsafe_links, bool checksum) { +typedef struct { + char* path; + struct stat stats; + bool is_directory; +} ScannerEntry; + +/* Inspect symlinks, resolve the entry type, and apply file filters once for both scanners. */ +static int scanner_inspect_entry(const ScannerOptions* options, const char* source_root, + const char* containing_dir, const char* name, + ScannerEntry* entry) { + entry->path = path_cat(containing_dir, name); + if (!entry->path) + return -1; + + struct stat link_stats; + if (lstat(entry->path, &link_stats) != 0) { + free(entry->path); + return 0; + } + bool is_symlink = S_ISLNK(link_stats.st_mode); + if (is_symlink && !options->follow_symlinks && !options->copy_links && !options->safe_links && + !options->copy_unsafe_links) + goto skip; + + if (is_symlink && options->safe_links) { + char link_target[4096]; + ssize_t length = readlink(entry->path, link_target, sizeof(link_target) - 1); + if (length < 0) + goto skip; + link_target[length] = '\0'; + if (link_target[0] == '/' || !safe_relative_link(source_root, containing_dir, link_target)) + goto skip; + } + + if (is_symlink && options->copy_unsafe_links && !options->copy_links) { + char link_target[4096]; + ssize_t length = readlink(entry->path, link_target, sizeof(link_target) - 1); + if (length < 0) + goto skip; + link_target[length] = '\0'; + if (link_target[0] != '/') + goto skip; + } + + if (is_symlink && options->follow_symlinks && !options->copy_links) + entry->stats = link_stats; + else if (stat(entry->path, &entry->stats) != 0) + goto skip; + + entry->is_directory = S_ISDIR(entry->stats.st_mode); + if (entry->is_directory) + return 1; + for (int i = 0; i < options->exclude_count; i++) + if (glob_match(options->exclude_patterns[i], name)) + goto skip; + if (options->include_count > 0) { + bool included = false; + for (int i = 0; i < options->include_count; i++) + if (glob_match(options->include_patterns[i], name)) + included = true; + if (!included) + goto skip; + } + if ((options->max_size > 0 && (unsigned long long)entry->stats.st_size > options->max_size) || + (options->min_size > 0 && (unsigned long long)entry->stats.st_size < options->min_size)) + goto skip; + return 1; + +skip: + free(entry->path); + entry->path = NULL; + return 0; +} + +DirectoryScanner* directory_scanner_create_with_options(const char* root_directory, + const ScannerOptions* options) { + if (!root_directory || !options) + return NULL; DirectoryScanner* scanner = calloc(1, sizeof(DirectoryScanner)); if (scanner == NULL) return NULL; @@ -69,21 +140,21 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_ } scanner->current_dir = NULL; scanner->current_path = NULL; - scanner->use_metadata = use_metadata; - scanner->chunk_size = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE; - scanner->exclude_patterns = exclude_patterns; - scanner->exclude_count = exclude_count; - scanner->include_patterns = include_patterns; - scanner->include_count = include_count; - scanner->max_size = max_size; - scanner->min_size = min_size; - scanner->max_depth = max_depth; + scanner->use_metadata = options->use_metadata; + scanner->chunk_size = options->chunk_size > 0 ? options->chunk_size : DESIRED_CHUNK_SIZE; + scanner->exclude_patterns = options->exclude_patterns; + scanner->exclude_count = options->exclude_count; + scanner->include_patterns = options->include_patterns; + scanner->include_count = options->include_count; + scanner->max_size = options->max_size; + scanner->min_size = options->min_size; + scanner->max_depth = options->max_depth; scanner->current_depth = 0; - scanner->follow_symlinks = follow_symlinks; - scanner->copy_links = copy_links; - scanner->safe_links = safe_links; - scanner->copy_unsafe_links = copy_unsafe_links; - scanner->checksum = checksum; + scanner->follow_symlinks = options->follow_symlinks; + scanner->copy_links = options->copy_links; + scanner->safe_links = options->safe_links; + scanner->copy_unsafe_links = options->copy_unsafe_links; + scanner->checksum = options->checksum; scanner->failed = false; DirEntry* root = dir_entry_create(root_directory, 0); if (!root) { @@ -100,6 +171,20 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_ return scanner; } +DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_metadata, + unsigned long long chunk_size, char** exclude_patterns, + int exclude_count, char** include_patterns, + int include_count, unsigned long long max_size, + unsigned long long min_size, int max_depth, + bool follow_symlinks, bool copy_links, bool safe_links, + bool copy_unsafe_links, bool checksum) { + ScannerOptions options = { + use_metadata, chunk_size, exclude_patterns, exclude_count, include_patterns, + include_count, max_size, min_size, max_depth, 0, + follow_symlinks, copy_links, safe_links, copy_unsafe_links, checksum}; + return directory_scanner_create_with_options(root_directory, &options); +} + void directory_scanner_destroy(DirectoryScanner* scanner) { if (scanner == NULL) return; @@ -167,7 +252,7 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { break; } - struct dirent* entry = readdir(scanner->current_dir); + const struct dirent* entry = readdir(scanner->current_dir); if (entry == NULL) { closedir(scanner->current_dir); scanner->current_dir = NULL; @@ -179,67 +264,27 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) continue; - char* cur_path = path_cat(scanner->current_path, entry->d_name); - if (!cur_path) { + ScannerOptions options = {scanner->use_metadata, scanner->chunk_size, + scanner->exclude_patterns, scanner->exclude_count, + scanner->include_patterns, scanner->include_count, + scanner->max_size, scanner->min_size, + scanner->max_depth, 0, + scanner->follow_symlinks, scanner->copy_links, + scanner->safe_links, scanner->copy_unsafe_links, + scanner->checksum}; + ScannerEntry inspected; + int inspection = scanner_inspect_entry(&options, scanner->current_path, scanner->current_path, + entry->d_name, &inspected); + if (inspection < 0) { scanner->failed = true; break; } - struct stat stats; - struct stat lstats; - bool is_symlink = false; - if (lstat(cur_path, &lstats) != 0) { - free(cur_path); + if (inspection == 0) continue; - } - is_symlink = S_ISLNK(lstats.st_mode); + char* cur_path = inspected.path; + struct stat stats = inspected.stats; - if (is_symlink && !scanner->follow_symlinks && !scanner->copy_links && !scanner->safe_links && - !scanner->copy_unsafe_links) { - free(cur_path); - continue; - } - - if (is_symlink && scanner->safe_links) { - char link_target[4096]; - ssize_t len = readlink(cur_path, link_target, sizeof(link_target) - 1); - if (len < 0) { - free(cur_path); - continue; - } - link_target[len] = '\0'; - if (link_target[0] == '/' || - !safe_relative_link(scanner->current_path, scanner->current_path, link_target)) { - free(cur_path); - continue; - } - } - - if (is_symlink && scanner->copy_unsafe_links && !scanner->copy_links) { - char link_target[4096]; - ssize_t len = readlink(cur_path, link_target, sizeof(link_target) - 1); - if (len < 0) { - free(cur_path); - continue; - } - link_target[len] = '\0'; - bool unsafe = (link_target[0] == '/'); - if (!unsafe) { - free(cur_path); - continue; - } - } - - bool use_lstat = is_symlink && scanner->follow_symlinks && !scanner->copy_links; - if (use_lstat) { - stats = lstats; - } else { - if (stat(cur_path, &stats) != 0) { - free(cur_path); - continue; - } - } - - if (S_ISDIR(stats.st_mode)) { + if (inspected.is_directory) { int next_depth = scanner->current_depth + 1; if (scanner->max_depth <= 0 || next_depth < scanner->max_depth) { DirEntry* de = dir_entry_create(cur_path, next_depth); @@ -254,38 +299,6 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { free(cur_path); continue; } - bool excluded = false; - for (int i = 0; i < scanner->exclude_count; i++) { - if (glob_match(scanner->exclude_patterns[i], entry->d_name)) { - excluded = true; - break; - } - } - if (excluded) { - free(cur_path); - continue; - } - - if (scanner->include_count > 0) { - bool included = false; - for (int i = 0; i < scanner->include_count; i++) { - if (glob_match(scanner->include_patterns[i], entry->d_name)) { - included = true; - break; - } - } - if (!included) { - free(cur_path); - continue; - } - } - - if ((scanner->max_size > 0 && (unsigned long long)stats.st_size > scanner->max_size) || - (scanner->min_size > 0 && (unsigned long long)stats.st_size < scanner->min_size)) { - free(cur_path); - continue; - } - File* file = file_create(cur_path); if (file == NULL) { free(cur_path); @@ -336,29 +349,13 @@ typedef struct { ParallelScanner* ps; char** dirs; int dir_count; - bool use_metadata; - unsigned long long chunk_size; - char** exclude_patterns; - int exclude_count; - char** include_patterns; - int include_count; - unsigned long long max_size; - unsigned long long min_size; - int max_depth; - bool follow_symlinks; - bool copy_links; - bool safe_links; - bool copy_unsafe_links; - bool checksum; + ScannerOptions options; } ParallelWorkerArg; static int parallel_worker_thread(void* arg) { ParallelWorkerArg* wa = (ParallelWorkerArg*)arg; for (int i = 0; i < wa->dir_count; i++) { - DirectoryScanner* ds = directory_scanner_create( - wa->dirs[i], wa->use_metadata, wa->chunk_size, wa->exclude_patterns, wa->exclude_count, - wa->include_patterns, wa->include_count, wa->max_size, wa->min_size, wa->max_depth, - wa->follow_symlinks, wa->copy_links, wa->safe_links, wa->copy_unsafe_links, wa->checksum); + DirectoryScanner* ds = directory_scanner_create_with_options(wa->dirs[i], &wa->options); if (!ds) { mtx_lock(&wa->ps->result_mutex); wa->ps->failed = true; @@ -366,6 +363,8 @@ static int parallel_worker_thread(void* arg) { cnd_broadcast(&wa->ps->result_not_empty); cnd_broadcast(&wa->ps->result_not_full); mtx_unlock(&wa->ps->result_mutex); + for (int j = i; j < wa->dir_count; j++) + free(wa->dirs[j]); break; } Chunk* chunk; @@ -401,13 +400,22 @@ static int parallel_worker_thread(void* arg) { return thrd_success; } -ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata, - unsigned long long chunk_size, char** exclude_patterns, - int exclude_count, char** include_patterns, - int include_count, unsigned long long max_size, - unsigned long long min_size, int max_depth, - int num_threads, bool follow_symlinks, bool copy_links, - bool safe_links, bool copy_unsafe_links, bool checksum) { +static void parallel_scanner_creation_failed(ParallelScanner* ps) { + mtx_lock(&ps->result_mutex); + ps->failed = true; + atomic_store(&ps->cancelled, true); + ps->expected_threads = ps->created_threads; + if (ps->completed >= ps->expected_threads) + ps->done = true; + cnd_broadcast(&ps->result_not_empty); + cnd_broadcast(&ps->result_not_full); + mtx_unlock(&ps->result_mutex); +} + +ParallelScanner* parallel_scanner_create_with_options(const char* root_directory, + const ScannerOptions* options) { + if (!root_directory || !options) + return NULL; ParallelScanner* ps = calloc(1, sizeof(ParallelScanner)); if (!ps) return NULL; @@ -460,105 +468,27 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata parallel_scanner_destroy(ps); return NULL; } - struct dirent* entry; + const struct dirent* entry; while ((entry = readdir(dir)) != NULL) { if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) continue; - char* cur_path = path_cat(root_directory, entry->d_name); - if (!cur_path) - continue; - struct stat lstats; - if (lstat(cur_path, &lstats) != 0) { - free(cur_path); + ScannerEntry inspected; + int inspection = + scanner_inspect_entry(options, root_directory, root_directory, entry->d_name, &inspected); + if (inspection < 0) { + ps->failed = true; continue; } - bool is_symlink = S_ISLNK(lstats.st_mode); - - // Skip symlinks unless the user explicitly enabled following/copying them. - if (is_symlink && !follow_symlinks && !copy_links && !safe_links && !copy_unsafe_links) { - free(cur_path); + if (inspection == 0) continue; - } - - // --safe-links: reject symlinks pointing outside the source tree. - if (is_symlink && safe_links) { - char link_target[4096]; - ssize_t len = readlink(cur_path, link_target, sizeof(link_target) - 1); - if (len < 0) { - free(cur_path); - continue; - } - link_target[len] = 0; - if (link_target[0] == '/' || - !safe_relative_link(root_directory, root_directory, link_target)) { - free(cur_path); - continue; - } - } - - // --copy-unsafe-links (without --copy-links): only copy absolute symlinks. - if (is_symlink && copy_unsafe_links && !copy_links) { - char link_target[4096]; - ssize_t len = readlink(cur_path, link_target, sizeof(link_target) - 1); - if (len < 0) { - free(cur_path); - continue; - } - link_target[len] = 0; - bool unsafe = (link_target[0] == '/'); - if (!unsafe) { - free(cur_path); - continue; - } - } - - // Determine whether to use lstat or stat results for the entry. - struct stat st; - bool use_lstat_res = is_symlink && follow_symlinks && !copy_links; - if (use_lstat_res) { - st = lstats; - } else { - if (stat(cur_path, &st) != 0) { - free(cur_path); - continue; - } - } - - if (S_ISDIR(st.st_mode)) { + char* cur_path = inspected.path; + struct stat st = inspected.stats; + if (inspected.is_directory) { if (!array_list_add(subdirs, cur_path)) { free(cur_path); ps->failed = true; } } else { - bool excluded = false; - for (int i = 0; i < exclude_count; i++) { - if (glob_match(exclude_patterns[i], entry->d_name)) { - excluded = true; - break; - } - } - if (excluded) { - free(cur_path); - continue; - } - if (include_count > 0) { - bool included = false; - for (int i = 0; i < include_count; i++) { - if (glob_match(include_patterns[i], entry->d_name)) { - included = true; - break; - } - } - if (!included) { - free(cur_path); - continue; - } - } - if ((max_size > 0 && (unsigned long long)st.st_size > max_size) || - (min_size > 0 && (unsigned long long)st.st_size < min_size)) { - free(cur_path); - continue; - } File* file = file_create(cur_path); free(cur_path); if (!file) { @@ -566,9 +496,9 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata continue; } file->data->size = st.st_size; - if (use_metadata) + if (options->use_metadata) file->metadata = file_metadata_create(&st); - if (use_metadata && !file->metadata) { + if (options->use_metadata && !file->metadata) { file_destroy(file); ps->failed = true; continue; @@ -581,7 +511,7 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata } closedir(dir); - unsigned long long cs = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE; + unsigned long long cs = options->chunk_size > 0 ? options->chunk_size : DESIRED_CHUNK_SIZE; if (root_files->size > 0) { ArrayList* batch = array_list_create(NULL); if (!batch) { @@ -604,7 +534,6 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata void** items = array_list_to_array(batch); if (!items) { ps->failed = true; - batch->item_destroyer = file_destroy; array_list_delete(batch); batch = NULL; break; @@ -613,11 +542,13 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata free(items); if (!c) { ps->failed = true; - batch->item_destroyer = file_destroy; array_list_delete(batch); batch = NULL; break; } + int batch_start = i - batch->size + 1; + for (int j = batch_start; j <= i; j++) + root_files->items[j] = NULL; batch->item_destroyer = NULL; array_list_delete(batch); batch = NULL; @@ -644,11 +575,10 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata array_list_delete(batch); } ps->initial_chunk = first; - root_files->item_destroyer = NULL; } array_list_delete(root_files); - int n = num_threads > 0 ? num_threads : 4; + int n = options->num_threads > 0 ? options->num_threads : 4; if (n > subdirs->size) n = subdirs->size > 0 ? subdirs->size : 1; @@ -671,14 +601,14 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata break; ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg)); if (!wa) { - ps->failed = true; + parallel_scanner_creation_failed(ps); break; } wa->ps = ps; wa->dirs = calloc(count, sizeof(char*)); if (!wa->dirs) { free(wa); - ps->failed = true; + parallel_scanner_creation_failed(ps); break; } bool dup_ok = true; @@ -692,37 +622,19 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata free(wa->dirs[j]); free(wa->dirs); free(wa); - ps->failed = true; + parallel_scanner_creation_failed(ps); break; } wa->dir_count = count; - wa->use_metadata = use_metadata; - wa->chunk_size = cs; - wa->exclude_patterns = exclude_patterns; - wa->exclude_count = exclude_count; - wa->include_patterns = include_patterns; - wa->include_count = include_count; - wa->max_size = max_size; - wa->min_size = min_size; - wa->max_depth = max_depth; - wa->follow_symlinks = follow_symlinks; - wa->copy_links = copy_links; - wa->safe_links = safe_links; - wa->copy_unsafe_links = copy_unsafe_links; - wa->checksum = checksum; + wa->options = *options; + wa->options.chunk_size = cs; start += count; if (thrd_create(&ps->threads[t], parallel_worker_thread, wa) != thrd_success) { for (int j = 0; j < count; j++) free(wa->dirs[j]); free(wa->dirs); free(wa); - ps->failed = true; - atomic_store(&ps->cancelled, true); - ps->expected_threads = ps->created_threads; - mtx_lock(&ps->result_mutex); - cnd_broadcast(&ps->result_not_empty); - cnd_broadcast(&ps->result_not_full); - mtx_unlock(&ps->result_mutex); + parallel_scanner_creation_failed(ps); break; } ps->num_threads++; @@ -733,6 +645,20 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata return ps; } +ParallelScanner* parallel_scanner_create(const char* root_directory, bool use_metadata, + unsigned long long chunk_size, char** exclude_patterns, + int exclude_count, char** include_patterns, + int include_count, unsigned long long max_size, + unsigned long long min_size, int max_depth, + int num_threads, bool follow_symlinks, bool copy_links, + bool safe_links, bool copy_unsafe_links, bool checksum) { + ScannerOptions options = {use_metadata, chunk_size, exclude_patterns, exclude_count, + include_patterns, include_count, max_size, min_size, + max_depth, num_threads, follow_symlinks, copy_links, + safe_links, copy_unsafe_links, checksum}; + return parallel_scanner_create_with_options(root_directory, &options); +} + Chunk* parallel_scanner_next(ParallelScanner* ps) { if (ps->initial_chunk) { Chunk* c = ps->initial_chunk; @@ -741,6 +667,11 @@ Chunk* parallel_scanner_next(ParallelScanner* ps) { } if (ps->num_threads == 0) { mtx_lock(&ps->result_mutex); + if (!queue_is_empty(ps->result_queue)) { + Chunk* chunk = queue_dequeue(ps->result_queue); + mtx_unlock(&ps->result_mutex); + return chunk; + } ps->done = true; mtx_unlock(&ps->result_mutex); return NULL; diff --git a/src/client/scanner.h b/src/client/scanner.h index 0bff53c..7d3ba56 100644 --- a/src/client/scanner.h +++ b/src/client/scanner.h @@ -8,6 +8,24 @@ #include #include +typedef struct { + bool use_metadata; + unsigned long long chunk_size; + char** exclude_patterns; + int exclude_count; + char** include_patterns; + int include_count; + unsigned long long max_size; + unsigned long long min_size; + int max_depth; + int num_threads; + bool follow_symlinks; + bool copy_links; + bool safe_links; + bool copy_unsafe_links; + bool checksum; +} ScannerOptions; + typedef struct { Queue* directories; DIR* current_dir; @@ -53,17 +71,21 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_ unsigned long long min_size, int max_depth, bool follow_symlinks, bool copy_links, bool safe_links, bool copy_unsafe_links, bool checksum); +DirectoryScanner* directory_scanner_create_with_options(const char* root_directory, + const ScannerOptions* options); Chunk* directory_scanner_next(DirectoryScanner* scanner); bool directory_scanner_failed(const DirectoryScanner* scanner); void directory_scanner_destroy(DirectoryScanner* scanner); -ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata, +ParallelScanner* parallel_scanner_create(const char* root_directory, bool use_metadata, unsigned long long chunk_size, char** exclude_patterns, int exclude_count, char** include_patterns, int include_count, unsigned long long max_size, unsigned long long min_size, int max_depth, int num_threads, bool follow_symlinks, bool copy_links, bool safe_links, bool copy_unsafe_links, bool checksum); +ParallelScanner* parallel_scanner_create_with_options(const char* root_directory, + const ScannerOptions* options); Chunk* parallel_scanner_next(ParallelScanner* scanner); bool parallel_scanner_failed(const ParallelScanner* scanner); void parallel_scanner_destroy(ParallelScanner* scanner); diff --git a/src/client/usage.c b/src/client/usage.c index 918c293..2f800af 100644 --- a/src/client/usage.c +++ b/src/client/usage.c @@ -1,8 +1,9 @@ -#include "stdio.h" +#include "usage.h" +#include #include #include -static __attribute__((unused)) void print_usage() { +void print_usage(void) { printf("Usage:\n"); printf(" fastsync [options] \n"); printf(" fastsync [options] --source-dir --dest-dir \n"); diff --git a/src/client/usage.h b/src/client/usage.h new file mode 100644 index 0000000..ef480c5 --- /dev/null +++ b/src/client/usage.h @@ -0,0 +1,6 @@ +#ifndef USAGE_H +#define USAGE_H + +void print_usage(void); + +#endif diff --git a/src/server/receiver.c b/src/server/receiver.c new file mode 100644 index 0000000..cd5f3a8 --- /dev/null +++ b/src/server/receiver.c @@ -0,0 +1,126 @@ +#include "receiver.h" + +#include "chunk.h" +#include "log.h" +#include "protocol.h" +#include "utils.h" +#include +#include + +static bool receiver_process_chunk(Chunk* chunk, const ReceiverSink* sink) { + for (int i = 0; i < chunk->element_count; i++) { + File* file = chunk->items[i]; + chunk->items[i] = NULL; + if (!sink->store_file(file, sink->context)) { + chunk_destroy(chunk); + return false; + } + } + chunk_destroy(chunk); + return true; +} + +static bool receiver_process_batch(Config* config, int file_descriptor) { + 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; + if (!receive_n_data(file_descriptor, &check_size, sizeof(check_size)) || + !receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime))) { + free(check_path); + return false; + } + if (!utils_valid_batch_path(check_path)) { + free(check_path); + 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 receiver_process(Config* config, int file_descriptor, const ReceiverSink* sink) { + 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; + 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(file_descriptor, config, &skipped); + if (!skipped && (!file || !sink->store_file(file, sink->context))) + goto receive_error; + } else if (status == STATUS_CHUNK) { + Chunk* chunk = receive_chunk_data(file_descriptor, config); + if (!chunk || !receiver_process_chunk(chunk, sink)) + goto receive_error; + } else if (status == STATUS_CHECK_BATCH) { + if (!receiver_process_batch(config, file_descriptor)) + return -1; + goto next; + } else { + File* file = file_receive(config, file_descriptor); + if (!file) { + log_message(LOG_LEVEL_ERROR, "Failed to receive file"); + goto receive_error; + } + if (!sink->store_file(file, sink->context)) + goto receive_error; + } + next: + if (!receive_status(file_descriptor, &status)) + goto receive_error; + } + 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"); + goto receive_error; + } + if (sink->send_success && !send_status(file_descriptor, STATUS_OK)) + return -1; + return 0; + +receive_error: + if (sink->send_error) + send_status(file_descriptor, STATUS_ERROR); + return -1; +} + +static bool receiver_save_file(File* file, void* context) { + Config* config = context; + bool success = + !config->save_to_disk || file_save_to_disk(config->receive_root_directory, file, config); + file_destroy(file); + return success; +} + +int receiver_receive_files(Config* config, int file_descriptor) { + ReceiverSink sink = {receiver_save_file, config, true, true}; + return receiver_process(config, file_descriptor, &sink); +} diff --git a/src/server/receiver.h b/src/server/receiver.h new file mode 100644 index 0000000..d619d2e --- /dev/null +++ b/src/server/receiver.h @@ -0,0 +1,19 @@ +#ifndef RECEIVER_H +#define RECEIVER_H + +#include "config.h" +#include "file.h" + +typedef bool (*ReceiverFileSink)(File* file, void* context); + +typedef struct { + ReceiverFileSink store_file; + void* context; + bool send_error; + bool send_success; +} ReceiverSink; + +int receiver_process(Config* config, int file_descriptor, const ReceiverSink* sink); +int receiver_receive_files(Config* config, int file_descriptor); + +#endif diff --git a/src/server/server.c b/src/server/server.c index ee34957..68ec538 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -1,24 +1,20 @@ -#include "array_list.h" -#include "chunk.h" #include "config.h" -#include "data.h" #include "file.h" #include "log.h" #include "multiprocessing.h" -#include "protocol.h" #include "queue.h" +#include "receiver.h" #include "transport_tcp.h" #include "transport_tls.h" #include "unistd.h" #include "utils.h" +#include +#include #include #include #include #include -#include -#include #include -#include static char* authorized_root; static int authorized_root_fd = -1; @@ -29,11 +25,6 @@ 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 __attribute__((unused)) configure_authorization(const char* root) { char resolved[PATH_MAX]; if (!root || !realpath(root, resolved)) @@ -52,131 +43,24 @@ static bool __attribute__((unused)) configure_authorization(const char* root) { return true; } -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; -} - void handler(int file_descriptor) { SSL* ssl = io_get_ssl(); + ProtocolSession session; + protocol_session_init(&session, file_descriptor, file_descriptor); + protocol_session_set_ssl(&session, ssl); + protocol_session_bind(&session); Config* config = config_receive(file_descriptor); if (config == NULL) { log_message(LOG_LEVEL_ERROR, "Failed to receive config"); close(file_descriptor); + protocol_session_unbind(); return; } if (!authorized_root) { log_message(LOG_LEVEL_ERROR, "No server-side destination root configured"); config_delete(config); close(file_descriptor); + protocol_session_unbind(); return; } char resolved_destination[PATH_MAX]; @@ -188,6 +72,7 @@ void handler(int file_descriptor) { free(canonical_destination); config_delete(config); close(file_descriptor); + protocol_session_unbind(); return; } if (canonical_destination) @@ -200,6 +85,7 @@ void handler(int file_descriptor) { if (!config->receive_root_directory) { config_delete(config); close(file_descriptor); + protocol_session_unbind(); return; } config->use_delete = config->use_delete && allow_delete; @@ -208,6 +94,7 @@ void handler(int file_descriptor) { if (q == NULL) { config_delete(config); close(file_descriptor); + protocol_session_unbind(); return; } PipelineContextReceiver* context = @@ -216,14 +103,15 @@ void handler(int file_descriptor) { queue_destroy(q); config_delete(config); close(file_descriptor); + protocol_session_unbind(); return; } + context->session.total_allocated_bytes = session.total_allocated_bytes; thrd_t receiver, writer; - bool receiver_created = false; + bool receiver_created = thrd_create(&receiver, receive_thread, context) == thrd_success; bool writer_created = false; - receiver_created = (thrd_create(&receiver, receive_thread, context) == thrd_success); if (receiver_created) - writer_created = (thrd_create(&writer, write_thread, context) == thrd_success); + writer_created = thrd_create(&writer, write_thread, context) == thrd_success; if (!receiver_created || !writer_created) { perror("Error creating Threads"); if (receiver_created) { @@ -240,22 +128,23 @@ void handler(int file_descriptor) { if (writer_created) thrd_join(writer, NULL); pipeline_context_receiver_destroy(context); + protocol_session_unbind(); return; } int receiver_result; int writer_result; thrd_join(receiver, &receiver_result); thrd_join(writer, &writer_result); - if (receiver_result == thrd_success && writer_result == thrd_success) - send_status(file_descriptor, STATUS_OK); - else - send_status(file_descriptor, STATUS_ERROR); + send_status(file_descriptor, receiver_result == thrd_success && writer_result == thrd_success + ? STATUS_OK + : STATUS_ERROR); pipeline_context_receiver_destroy(context); } else { - if (receive_files(config, file_descriptor) != 0) + if (receiver_receive_files(config, file_descriptor) != 0) log_message(LOG_LEVEL_ERROR, "Transfer failed"); config_delete(config); } + protocol_session_unbind(); close(file_descriptor); } @@ -264,16 +153,14 @@ static Server* g_server = NULL; static void cleanup(int sig) { (void)sig; - if (g_server) { + if (g_server) server_delete(&g_server); - } _exit(0); } static void print_server_usage(void) { printf("FastSync Server\n"); - printf("Usage: fastsync-server [options]\n"); - printf("\n"); + printf("Usage: fastsync-server [options]\n\n"); printf("Options:\n"); printf(" --stdio Run in stdio mode (SSH transport)\n"); printf(" -p TCP port (default: 8080, range: 1-65535)\n"); @@ -289,9 +176,7 @@ static void print_server_usage(void) { int main(int argc, char* argv[]) { bool use_tls = false; - char* tls_cert = NULL; - char* tls_key = NULL; - char* tls_ca = NULL; + char *tls_cert = NULL, *tls_key = NULL, *tls_ca = NULL; int port = 8080; const char* destination_root = "."; bool stdio_mode = false; @@ -331,11 +216,8 @@ int main(int argc, char* argv[]) { return 1; } } - - if (tls_ca && !use_tls) { + if (tls_ca && !use_tls) log_message(LOG_LEVEL_WARNING, "--ca has no effect without --tls"); - } - signal(SIGINT, cleanup); signal(SIGTERM, cleanup); if (!configure_authorization(destination_root)) { @@ -352,7 +234,7 @@ int main(int argc, char* argv[]) { return 0; } g_server = server_create(port); - if (g_server == NULL) { + if (!g_server) { log_message(LOG_LEVEL_ERROR, "Failed to create server"); return 1; } @@ -374,4 +256,4 @@ int main(int argc, char* argv[]) { } return 0; } -#endif /* !FASTSYNC_SERVER_AS_LIB */ +#endif diff --git a/src/shared/chunk.c b/src/shared/chunk.c index 4ccf0d5..a181e8e 100644 --- a/src/shared/chunk.c +++ b/src/shared/chunk.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -14,6 +15,7 @@ /* Maximum individual file data size within a chunk (64 MB) */ #define MAX_FILE_DATA_SIZE (64ULL * 1024 * 1024) +#define MAX_CHUNK_FILES (1024 * 1024) Chunk* chunk_create(File** items, int element_count) { Chunk* chunk = (Chunk*)malloc(sizeof(Chunk)); @@ -92,6 +94,11 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { size_t remaining_size = data->size; while (remaining_size > 0) { + if (files->size >= MAX_CHUNK_FILES) { + log_message(LOG_LEVEL_ERROR, "Chunk contains too many files"); + array_list_delete(files); + return NULL; + } if (remaining_size < sizeof(size_t)) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path length"); array_list_delete(files); @@ -103,7 +110,7 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { data_pointer += sizeof(size_t); remaining_size -= sizeof(size_t); - if (remaining_size < path_len) { + if (path_len > SIZE_MAX - 1 || remaining_size < path_len) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path"); array_list_delete(files); return NULL; @@ -122,10 +129,15 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { File* file = file_create(path); free(path); + if (file == NULL) { + array_list_delete(files); + return NULL; + } if (use_metadata) { if (remaining_size < sizeof(int)) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for metadata"); + file_destroy(file); array_list_delete(files); return NULL; } @@ -134,6 +146,7 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { memcpy(&present_flag, data_pointer, sizeof(int)); if (present_flag && remaining_size < sizeof(int) + FILE_METADATA_WIRE_SIZE) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for metadata body"); + file_destroy(file); array_list_delete(files); return NULL; } @@ -141,10 +154,16 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { remaining_size -= sizeof(int); if (file->metadata) remaining_size -= FILE_METADATA_WIRE_SIZE; + else if (present_flag) { + file_destroy(file); + array_list_delete(files); + return NULL; + } } if (remaining_size < sizeof(size_t)) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for data size"); + file_destroy(file); array_list_delete(files); return NULL; } @@ -156,6 +175,7 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { if (remaining_size < file_data_size) { log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for file content"); + file_destroy(file); array_list_delete(files); return NULL; } @@ -164,29 +184,48 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) { if (file_data_size > MAX_FILE_DATA_SIZE) { log_message(LOG_LEVEL_ERROR, "File data size %zu exceeds maximum %llu", file_data_size, (unsigned long long)MAX_FILE_DATA_SIZE); + file_destroy(file); array_list_delete(files); return NULL; } - void* file_data = malloc(file_data_size); + void* file_data = malloc(file_data_size > 0 ? file_data_size : 1); if (file_data == NULL) { perror("Could not allocate memory for file data"); + file_destroy(file); array_list_delete(files); return NULL; } memcpy(file_data, data_pointer, file_data_size); data_destroy(file->data); file->data = data_create(file_data, file_data_size); + if (file->data == NULL) { + file_destroy(file); + array_list_delete(files); + return NULL; + } data_pointer += file_data_size; remaining_size -= file_data_size; - array_list_add(files, file); + if (!array_list_add(files, file)) { + file_destroy(file); + array_list_delete(files); + return NULL; + } } File** file_array = (File**)array_list_to_array(files); + if (file_array == NULL) { + array_list_delete(files); + return NULL; + } Chunk* chunk = chunk_create(file_array, files->size); free(file_array); + if (chunk == NULL) { + array_list_delete(files); + return NULL; + } files->item_destroyer = NULL; array_list_delete(files); diff --git a/src/shared/config.c b/src/shared/config.c index 4c1a3a0..8938957 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -173,102 +173,157 @@ 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 - */ +/* Each helper is deliberately ordered to match the wire format. Keep the + * helper call order in config_send and config_receive unchanged when adding + * fields. */ +static bool send_core_fields(int fd, const Config* c) { + return send_str(fd, c->version) && send_str(fd, c->send_directory) && + send_str(fd, c->receive_root_directory) && send_int(fd, c->save_to_disk) && + send_int(fd, c->use_multithreading) && send_int(fd, c->use_chunk_serialization) && + send_int(fd, c->use_compression) && send_int(fd, c->use_metadata) && + send_int(fd, c->compression_level) && + send_n_data(fd, &c->chunk_size, sizeof(c->chunk_size)) && send_int(fd, c->use_sendfile); +} + +static bool send_delta_fields(int fd, const Config* c) { + return send_int(fd, c->use_delete) && send_int(fd, c->use_incremental) && + send_int(fd, c->use_delta) && + send_n_data(fd, &c->delta_block_size, sizeof(c->delta_block_size)) && + send_n_data(fd, &c->delta_max_file_size, sizeof(unsigned long long)); +} + +static bool send_file_options(int fd, const Config* c) { + return send_int(fd, c->backup) && send_str(fd, c->backup_dir ? c->backup_dir : "") && + send_int(fd, c->follow_symlinks) && send_int(fd, c->copy_links) && + send_int(fd, c->safe_links) && send_int(fd, c->copy_unsafe_links) && + send_int(fd, c->preserve_hard_links) && send_int(fd, c->preserve_acls) && + send_int(fd, c->preserve_xattrs) && send_int(fd, c->preserve_devices) && + send_int(fd, c->preserve_sparse); +} + +static bool send_selection_options(int fd, const Config* c) { + return send_int(fd, c->update) && send_int(fd, c->inplace) && send_int(fd, c->append) && + send_int(fd, c->append_verify) && send_int(fd, c->delete_excluded) && + send_int(fd, c->delete_after) && send_n_data(fd, &c->max_delete, sizeof(c->max_delete)) && + send_int(fd, c->relative) && send_int(fd, c->prune_empty_dirs); +} + +static bool send_resume_options(int fd, const Config* c) { + return send_str(fd, c->temp_dir ? c->temp_dir : "") && send_int(fd, c->partial) && + send_str(fd, c->partial_dir ? c->partial_dir : "") && + send_str(fd, c->suffix ? c->suffix : "") && send_int(fd, c->delete_before) && + send_int(fd, c->checksum) && send_str(fd, c->compress_choice ? c->compress_choice : ""); +} + +static bool receive_core_fields(int fd, Config* c) { + int value; + c->send_directory = receive_str(fd); + c->receive_root_directory = receive_str(fd); + if (!c->send_directory || !c->receive_root_directory) + return false; + if (!receive_int(fd, &value)) + return false; + c->save_to_disk = value; + if (!receive_int(fd, &value)) + return false; + c->use_multithreading = value; + if (!receive_int(fd, &value)) + return false; + c->use_chunk_serialization = value; + if (!receive_int(fd, &value)) + return false; + c->use_compression = value; + if (!receive_int(fd, &value)) + return false; + c->use_metadata = value; + if (!receive_int(fd, &value)) + return false; + c->compression_level = value; + if (!receive_n_data(fd, &c->chunk_size, sizeof(c->chunk_size))) + return false; + if (!receive_int(fd, &value)) + return false; + c->use_sendfile = value; + return true; +} + +static bool receive_delta_fields(int fd, Config* c) { + int value; + if (!receive_int(fd, &value)) + return false; + c->use_delete = value; + if (!receive_int(fd, &value)) + return false; + c->use_incremental = value; + if (!receive_int(fd, &value)) + return false; + c->use_delta = value; + return receive_n_data(fd, &c->delta_block_size, sizeof(c->delta_block_size)) && + receive_n_data(fd, &c->delta_max_file_size, sizeof(unsigned long long)); +} + +static bool receive_file_options(int fd, Config* c) { + int value; + if (!receive_int(fd, &value)) + return false; + c->backup = value; + c->backup_dir = receive_str(fd); + if (!c->backup_dir) + return false; + bool* flags[] = {&c->follow_symlinks, &c->copy_links, &c->safe_links, + &c->copy_unsafe_links, &c->preserve_hard_links, &c->preserve_acls, + &c->preserve_xattrs, &c->preserve_devices, &c->preserve_sparse}; + for (size_t i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) { + if (!receive_int(fd, &value)) + return false; + *flags[i] = value; + } + return true; +} + +static bool receive_selection_options(int fd, Config* c) { + int value; + bool* flags[] = {&c->update, &c->inplace, &c->append, + &c->append_verify, &c->delete_excluded, &c->delete_after}; + for (size_t i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) { + if (!receive_int(fd, &value)) + return false; + *flags[i] = value; + } + if (!receive_n_data(fd, &c->max_delete, sizeof(c->max_delete))) + return false; + if (!receive_int(fd, &value)) + return false; + c->relative = value; + if (!receive_int(fd, &value)) + return false; + c->prune_empty_dirs = value; + return true; +} + +static bool receive_resume_options(int fd, Config* c) { + int value; + c->temp_dir = receive_str(fd); + if (!c->temp_dir || !receive_int(fd, &value)) + return false; + c->partial = value; + c->partial_dir = receive_str(fd); + c->suffix = c->partial_dir ? receive_str(fd) : NULL; + if (!c->partial_dir || !c->suffix || !receive_int(fd, &value)) + return false; + c->delete_before = value; + if (!receive_int(fd, &value)) + return false; + c->checksum = value; + c->compress_choice = receive_str(fd); + return c->compress_choice != NULL; +} + 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 : "")) + if (!send_core_fields(file_descriptor, config) || !send_delta_fields(file_descriptor, config) || + !send_file_options(file_descriptor, config) || + !send_selection_options(file_descriptor, config) || + !send_resume_options(file_descriptor, config)) return false; Status status; if (!receive_status(file_descriptor, &status)) @@ -280,159 +335,25 @@ bool config_send(int file_descriptor, const Config* config) { return true; } -/* Wire format order: see the comment above config_send. */ Config* config_receive(int file_descriptor) { - Config* config = (Config*)malloc(sizeof(Config)); - if (config == NULL) + Config* config = config_create(); + if (!config) 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 (!config->version) + goto error; 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; + goto error; } - 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) + if (!receive_core_fields(file_descriptor, config) || + !receive_delta_fields(file_descriptor, config) || + !receive_file_options(file_descriptor, config) || + !receive_selection_options(file_descriptor, config) || + !receive_resume_options(file_descriptor, config)) goto error; if (config->compress_choice[0] != '\0' && strcmp(config->compress_choice, "zstd") != 0 && strcmp(config->compress_choice, "none") != 0) { @@ -445,15 +366,6 @@ 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; } diff --git a/src/shared/file.c b/src/shared/file.c index b333ce1..281cff2 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -19,6 +19,7 @@ #include "config.h" #include "data.h" #include "file.h" +#include "file_store.h" #include "metadata.h" #include "protocol.h" #include "utils.h" @@ -153,17 +154,8 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, return true; } -static bool to_disk_secure(const char* path, const void* data, unsigned long long data_size, - bool inplace, bool sparse, const FileMetadata* metadata); -static int open_secure_parent(const char* path, char** leaf_out); -static bool rename_secure(const char* old_path, const char* new_path); -static int authorized_root_fd = -1; -static char* authorized_root_path; - void file_set_authorized_root(int fd, const char* canonical_path) { - authorized_root_fd = fd; - free(authorized_root_path); - authorized_root_path = canonical_path ? str_dup(canonical_path) : NULL; + file_store_set_authorized_root(fd, canonical_path); } static bool path_is_within_root(const char* root, const char* path) { @@ -280,7 +272,7 @@ bool file_save_to_disk(const char* root_directory, const File* file, const Confi mkdir_r(bdir); free(backup_dir_path); } - if (!rename_secure(disk_path, backup_path)) { + if (!file_store_rename_secure(disk_path, backup_path)) { free(backup_path); free(resolved_root); free(confined_backup); @@ -335,8 +327,8 @@ bool file_save_to_disk(const char* root_directory, const File* file, const Confi free(resolved_dir); free(resolved_root); - bool ok = to_disk_secure(disk_path, file->data->data, file->data->size, inplace, sparse, - file->metadata); + bool ok = file_store_write_secure(disk_path, file->data->data, file->data->size, inplace, sparse, + file->metadata); free(confined_backup); free(confined_partial); free(disk_path); @@ -532,7 +524,7 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { int old_fd = -1; if (full_path) { char* leaf = NULL; - int parent_fd = open_secure_parent(full_path, &leaf); + int parent_fd = file_store_open_secure_parent(full_path, &leaf); if (parent_fd >= 0) { old_fd = openat(parent_fd, leaf, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); free(leaf); @@ -651,243 +643,11 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { return file; } -static int open_secure_parent(const char* path, char** leaf_out) { - char* copy = str_dup(path); - if (!copy) - return -1; - char* parent = dirname(copy); - const char* slash = strrchr(path, '/'); - char* leaf = str_dup(slash ? slash + 1 : path); - if (!leaf) { - free(copy); - return -1; - } - int fd; - if (authorized_root_fd >= 0 && authorized_root_path && path[0] == '/' && - path_is_within_root(authorized_root_path, path)) { - fd = dup(authorized_root_fd); - size_t root_len = strlen(authorized_root_path); - char* relative = str_dup(path + root_len); - if (!relative) { - free(copy); - free(leaf); - close(fd); - return -1; - } - free(copy); - copy = relative; - parent = dirname(copy); - } else { - fd = (parent[0] == '/') ? open("/", O_RDONLY | O_DIRECTORY | O_CLOEXEC) - : open(".", O_RDONLY | O_DIRECTORY | O_CLOEXEC); - } - if (fd < 0) { - free(copy); - free(leaf); - return -1; - } - char* save = NULL; - char* component = strtok_r(parent, "/", &save); - while (component) { - if (strcmp(component, ".") != 0 && strcmp(component, "..") != 0) { - int next = openat(fd, component, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (next < 0 && errno == ENOENT && mkdirat(fd, component, 0755) == 0) - next = openat(fd, component, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (next < 0) { - close(fd); - free(copy); - free(leaf); - return -1; - } - close(fd); - fd = next; - } - component = strtok_r(NULL, "/", &save); - } - free(copy); - *leaf_out = leaf; - return fd; -} - -static bool rename_secure(const char* old_path, const char* new_path) { - char *old_leaf = NULL, *new_leaf = NULL; - int old_parent = open_secure_parent(old_path, &old_leaf); - int new_parent = open_secure_parent(new_path, &new_leaf); - bool ok = old_parent >= 0 && new_parent >= 0 && - renameat(old_parent, old_leaf, new_parent, new_leaf) == 0; - if (old_parent >= 0) - close(old_parent); - if (new_parent >= 0) - close(new_parent); - free(old_leaf); - free(new_leaf); - return ok; -} - -static bool write_all(int fd, const void* data, unsigned long long size) { - const unsigned char* p = data; - unsigned long long done = 0; - while (done < size) { - ssize_t n = write(fd, p + done, (size_t)(size - done)); - if (n < 0 && errno == EINTR) - continue; - if (n <= 0) - return false; - done += (unsigned long long)n; - } - return true; -} - -static bool to_disk_secure(const char* path, const void* data, unsigned long long data_size, - bool inplace, bool sparse, const FileMetadata* metadata) { - char* leaf = NULL; - int dirfd = open_secure_parent(path, &leaf); - if (dirfd < 0) - return false; - int fd = -1; - bool ok = false; - if (inplace) { - fd = openat(dirfd, leaf, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC | O_NOFOLLOW, 0644); - if (fd >= 0) { - if (!sparse || data_size == 0 || ftruncate(fd, (off_t)data_size) == 0) - ok = write_all(fd, data, data_size); - if (ok && metadata) - ok = file_restore_metadata_fd(fd, metadata); - } - } else { - char tmp[NAME_MAX]; - for (unsigned int i = 0; i < 100 && !ok; ++i) { - snprintf(tmp, sizeof(tmp), ".%s.tmp.%ld.%u", leaf, (long)getpid(), i); - fd = openat(dirfd, tmp, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC | O_NOFOLLOW, 0600); - if (fd < 0) - continue; - if (sparse && data_size > 0) - ok = ftruncate(fd, (off_t)data_size) == 0; - if (ok || (!sparse || data_size == 0)) - ok = write_all(fd, data, data_size); - if (ok && metadata) - ok = file_restore_metadata_fd(fd, metadata); - if (close(fd) != 0) - ok = false; - fd = -1; - if (ok && renameat(dirfd, tmp, dirfd, leaf) != 0) - ok = false; - if (!ok) - unlinkat(dirfd, tmp, 0); - } - } - if (fd >= 0) - close(fd); - close(dirfd); - free(leaf); - return ok; -} - bool to_disk(const char* path, const void* data, unsigned long long data_size, bool inplace, bool sparse) { if (!path || (!data && data_size != 0) || has_path_traversal(path)) return false; - return to_disk_secure(path, data, data_size, inplace, sparse, NULL); - /* Kept below only as historical context; all writes use descriptor-relative operations. */ - char* tmp_path = NULL; - char* directory = NULL; - - char* path_dup = str_dup(path); - if (!path_dup) - return false; - const char* dir_result = dirname(path_dup); - directory = str_dup(dir_result); - free(path_dup); - if (!directory) - return false; - - bool ok = true; - if (!mkdir_r(directory)) - goto done; - - if (inplace) { - FILE* file_pointer = fopen(path, "wb"); - if (file_pointer == NULL) { - perror("Could not open file for inplace write"); - ok = false; - goto done; - } - if (sparse && data_size > 0) { - if (fseek(file_pointer, data_size - 1, SEEK_SET) != 0) { - perror("Failed to seek for sparse file"); - fclose(file_pointer); - ok = false; - goto done; - } - if (fwrite("", 1, 1, file_pointer) != 1) { - perror("Failed to write sparse file"); - fclose(file_pointer); - ok = false; - goto done; - } - rewind(file_pointer); - } - if (data_size > 0 && fwrite(data, 1, data_size, file_pointer) != data_size) { - perror("Failed to write all data to file"); - fclose(file_pointer); - ok = false; - goto done; - } - fclose(file_pointer); - free(directory); - return true; - } - - size_t path_len = strlen(path); - tmp_path = malloc(path_len + 5); - if (!tmp_path) { - ok = false; - goto done; - } - memcpy(tmp_path, path, path_len); - memcpy(tmp_path + path_len, ".tmp", 5); - - FILE* file_pointer = fopen(tmp_path, "wb"); - if (file_pointer == NULL) { - perror("Could not open temporary file"); - ok = false; - goto done; - } - if (sparse && data_size > 0) { - if (fseek(file_pointer, data_size - 1, SEEK_SET) != 0) { - perror("Failed to seek for sparse file"); - fclose(file_pointer); - ok = false; - goto done; - } - if (fwrite("", 1, 1, file_pointer) != 1) { - perror("Failed to write sparse file"); - fclose(file_pointer); - ok = false; - goto done; - } - rewind(file_pointer); - } - if (fwrite(data, 1, data_size, file_pointer) != data_size) { - perror("Failed to write all data to temporary file"); - fclose(file_pointer); - unlink(tmp_path); - ok = false; - goto done; - } - fclose(file_pointer); - - if (rename(tmp_path, path) != 0) { - perror("Failed to atomically rename temporary file"); - unlink(tmp_path); - ok = false; - goto done; - } - -done: - free(tmp_path); - free(directory); - return ok; + return file_store_write_secure(path, data, data_size, inplace, sparse, NULL); } bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level, diff --git a/src/shared/file_store.c b/src/shared/file_store.c new file mode 100644 index 0000000..2e67807 --- /dev/null +++ b/src/shared/file_store.c @@ -0,0 +1,171 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +#include "file_store.h" +#include "metadata.h" +#include "utils.h" + +static int authorized_root_fd = -1; +static char* authorized_root_path; + +static bool path_is_within_root(const char* root, const char* path) { + size_t root_length = strlen(root); + return strncmp(root, path, root_length) == 0 && + (path[root_length] == '\0' || path[root_length] == '/'); +} + +void file_store_set_authorized_root(int fd, const char* canonical_path) { + authorized_root_fd = fd; + free(authorized_root_path); + authorized_root_path = canonical_path ? str_dup(canonical_path) : NULL; +} + +int file_store_open_secure_parent(const char* path, char** leaf_out) { + char* copy = str_dup(path); + if (!copy) + return -1; + char* parent = dirname(copy); + const char* slash = strrchr(path, '/'); + char* leaf = str_dup(slash ? slash + 1 : path); + if (!leaf) { + free(copy); + return -1; + } + int fd; + if (authorized_root_fd >= 0 && authorized_root_path && path[0] == '/' && + path_is_within_root(authorized_root_path, path)) { + fd = dup(authorized_root_fd); + size_t root_length = strlen(authorized_root_path); + char* relative = str_dup(path + root_length); + if (!relative) { + free(copy); + free(leaf); + close(fd); + return -1; + } + free(copy); + copy = relative; + parent = dirname(copy); + } else { + fd = (parent[0] == '/') ? open("/", O_RDONLY | O_DIRECTORY | O_CLOEXEC) + : open(".", O_RDONLY | O_DIRECTORY | O_CLOEXEC); + } + if (fd < 0) { + free(copy); + free(leaf); + return -1; + } + char* save = NULL; + char* component = strtok_r(parent, "/", &save); + while (component) { + if (strcmp(component, ".") != 0 && strcmp(component, "..") != 0) { + int next = openat(fd, component, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (next < 0 && errno == ENOENT && mkdirat(fd, component, 0755) == 0) + next = openat(fd, component, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (next < 0) { + close(fd); + free(copy); + free(leaf); + return -1; + } + close(fd); + fd = next; + } + component = strtok_r(NULL, "/", &save); + } + free(copy); + *leaf_out = leaf; + return fd; +} + +bool file_store_rename_secure(const char* old_path, const char* new_path) { + char *old_leaf = NULL, *new_leaf = NULL; + int old_parent = file_store_open_secure_parent(old_path, &old_leaf); + int new_parent = file_store_open_secure_parent(new_path, &new_leaf); + bool ok = old_parent >= 0 && new_parent >= 0 && + renameat(old_parent, old_leaf, new_parent, new_leaf) == 0; + if (old_parent >= 0) + close(old_parent); + if (new_parent >= 0) + close(new_parent); + free(old_leaf); + free(new_leaf); + return ok; +} + +static bool write_all(int fd, const void* data, unsigned long long size) { + const unsigned char* p = data; + unsigned long long done = 0; + while (done < size) { + ssize_t n = write(fd, p + done, (size_t)(size - done)); + if (n < 0 && errno == EINTR) + continue; + if (n <= 0) + return false; + done += (unsigned long long)n; + } + return true; +} + +bool file_store_write_secure(const char* path, const void* data, unsigned long long data_size, + bool inplace, bool sparse, const FileMetadata* metadata) { + char* leaf = NULL; + int dirfd = file_store_open_secure_parent(path, &leaf); + if (dirfd < 0) + return false; + int fd = -1; + bool ok = false; + if (inplace) { + fd = openat(dirfd, leaf, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC | O_NOFOLLOW, 0644); + if (fd >= 0) { + if (!sparse || data_size == 0 || ftruncate(fd, (off_t)data_size) == 0) + ok = write_all(fd, data, data_size); + if (ok && metadata) + ok = file_restore_metadata_fd(fd, metadata); + } + } else { + int tmp_size = snprintf(NULL, 0, ".%s.tmp.%ld.%u", leaf, (long)getpid(), 99U); + if (tmp_size < 0) { + close(dirfd); + free(leaf); + return false; + } + char* tmp = malloc((size_t)tmp_size + 1); + if (!tmp) { + close(dirfd); + free(leaf); + return false; + } + for (unsigned int i = 0; i < 100 && !ok; ++i) { + snprintf(tmp, (size_t)tmp_size + 1, ".%s.tmp.%ld.%u", leaf, (long)getpid(), i); + fd = openat(dirfd, tmp, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC | O_NOFOLLOW, 0600); + if (fd < 0) + continue; + if (sparse && data_size > 0) + ok = ftruncate(fd, (off_t)data_size) == 0; + if (ok || (!sparse || data_size == 0)) + ok = write_all(fd, data, data_size); + if (ok && metadata) + ok = file_restore_metadata_fd(fd, metadata); + if (close(fd) != 0) + ok = false; + fd = -1; + if (ok && renameat(dirfd, tmp, dirfd, leaf) != 0) + ok = false; + if (!ok) + unlinkat(dirfd, tmp, 0); + } + free(tmp); + } + if (fd >= 0) + close(fd); + close(dirfd); + free(leaf); + return ok; +} diff --git a/src/shared/file_store.h b/src/shared/file_store.h new file mode 100644 index 0000000..3ae7676 --- /dev/null +++ b/src/shared/file_store.h @@ -0,0 +1,13 @@ +#ifndef FILE_STORE_H +#define FILE_STORE_H + +#include "file.h" +#include + +void file_store_set_authorized_root(int fd, const char* canonical_path); +int file_store_open_secure_parent(const char* path, char** leaf_out); +bool file_store_rename_secure(const char* old_path, const char* new_path); +bool file_store_write_secure(const char* path, const void* data, unsigned long long data_size, + bool inplace, bool sparse, const FileMetadata* metadata); + +#endif diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 5edbec2..49de752 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -1,4 +1,5 @@ #include "multiprocessing.h" +#include "receiver.h" #include "array_list.h" #include "chunk.h" @@ -14,10 +15,6 @@ #include #include -static bool valid_batch_path(const char* path) { - return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path); -} - PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner, Queue* queue_loader) { PipelineContextSender* context = malloc(sizeof(PipelineContextSender)); @@ -101,6 +98,8 @@ PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* context->queue = queue; context->file_descriptor = file_descriptor; context->ssl = ssl; + protocol_session_init(&context->session, file_descriptor, file_descriptor); + protocol_session_set_ssl(&context->session, ssl); context->receiver_done = false; atomic_init(&context->cancelled, false); int init = 0; @@ -137,24 +136,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 bool receiver_enqueue_file(File* file, void* context_pointer) { + PipelineContextReceiver* context = context_pointer; + if (queue_enqueue_multithreaded_cancel(context->queue, file, &context->mutex, + &context->condition_not_empty, + &context->condition_not_full, &context->cancelled)) + return true; + file_destroy(file); + return false; } static void receiver_thread_fail(PipelineContextReceiver* context) { @@ -167,123 +156,42 @@ static void receiver_thread_fail(PipelineContextReceiver* context) { } 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); + protocol_session_bind(&context->session); mtx_lock(&context->mutex); int file_descriptor = context->file_descriptor; 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(); + ReceiverSink sink = {receiver_enqueue_file, context, false, false}; + if (receiver_process((Config*)config, file_descriptor, &sink) != 0) { + receiver_thread_fail(context); + protocol_session_unbind(); + return thrd_error; } - if (status == STATUS_MANIFEST) { - if (receive_manifest(file_descriptor, config, &status) != 0) - RECEIVE_THREAD_FAIL(); - } - if (status != STATUS_FINISHED) - RECEIVE_THREAD_FAIL(); mtx_lock(&context->mutex); context->receiver_done = true; cnd_signal(&context->condition_not_empty); mtx_unlock(&context->mutex); -#undef RECEIVE_THREAD_FAIL + protocol_session_unbind(); return thrd_success; } int write_thread(void* pipeline_context) { PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context; - if (context->ssl) - io_set_ssl(context->ssl); mtx_lock(&context->mutex); bool save_to_disk = context->config->save_to_disk; char* root_directory = str_dup(context->config->receive_root_directory); mtx_unlock(&context->mutex); + if (save_to_disk && !root_directory) { + mtx_lock(&context->mutex); + atomic_store(&context->cancelled, true); + context->receiver_done = true; + cnd_broadcast(&context->condition_not_full); + cnd_broadcast(&context->condition_not_empty); + mtx_unlock(&context->mutex); + return thrd_error; + } while (true) { File* file = diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index 348b955..37f09d5 100644 --- a/src/shared/multiprocessing.h +++ b/src/shared/multiprocessing.h @@ -35,6 +35,7 @@ typedef struct PipelineContextReceiver { Config* config; int file_descriptor; SSL* ssl; + ProtocolSession session; mtx_t mutex; cnd_t condition_not_full; cnd_t condition_not_empty; diff --git a/src/shared/protocol.c b/src/shared/protocol.c index a3a3aed..d7d7b5f 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -18,6 +18,8 @@ static __thread int io_read_fd = -1; static __thread int io_write_fd = -1; static __thread SSL* io_ssl; +static __thread ProtocolSession* bound_session; +static __thread ProtocolSession legacy_io_session = {.read_fd = -1, .write_fd = -1}; static unsigned long long io_bwlimit = 0; static long long bw_tokens = 0; @@ -25,15 +27,41 @@ static struct timespec bw_last_refill = {0, 0}; static mtx_t bw_mutex; static once_flag bw_mutex_once = ONCE_FLAG_INIT; -static __thread unsigned long long total_allocated_bytes = 0; - void io_set_fds(int read_fd, int write_fd) { + bound_session = NULL; io_read_fd = read_fd; io_write_fd = write_fd; /* A descriptor switch starts a new transport; never reuse a TLS object belonging to a previous connection or test pipe. */ io_ssl = NULL; - total_allocated_bytes = 0; + legacy_io_session.read_fd = read_fd; + legacy_io_session.write_fd = write_fd; + legacy_io_session.ssl = NULL; + legacy_io_session.total_allocated_bytes = 0; + protocol_session_set_bwlimit(&legacy_io_session, io_bwlimit); +} + +void protocol_session_init(ProtocolSession* session, int read_fd, int write_fd) { + if (!session) + return; + memset(session, 0, sizeof(*session)); + session->read_fd = read_fd; + session->write_fd = write_fd; + if (io_bwlimit) + protocol_session_set_bwlimit(session, io_bwlimit); +} + +void protocol_session_bind(ProtocolSession* session) { + bound_session = session; +} + +void protocol_session_unbind(void) { + bound_session = NULL; +} + +void protocol_session_set_ssl(ProtocolSession* session, SSL* ssl) { + if (session) + session->ssl = ssl; } static void bw_mutex_init(void) { @@ -49,39 +77,51 @@ void io_set_bwlimit(unsigned long long bytes_per_sec) { mtx_unlock(&bw_mutex); } -static void bw_throttle(size_t bytes_written) { - if (io_bwlimit == 0) +void protocol_session_set_bwlimit(ProtocolSession* session, unsigned long long bytes_per_sec) { + if (!session) + return; + session->bwlimit = bytes_per_sec; + session->bw_tokens = (long long)bytes_per_sec; + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + session->bw_last_refill_sec = now.tv_sec; + session->bw_last_refill_nsec = now.tv_nsec; +} + +static void bw_throttle_session(ProtocolSession* session, size_t bytes_written) { + if (session->bwlimit == 0) return; - call_once(&bw_mutex_once, bw_mutex_init); - mtx_lock(&bw_mutex); struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); - long long elapsed_ns = - (now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL + (now.tv_nsec - bw_last_refill.tv_nsec); - bw_last_refill = now; + long long elapsed_ns = (now.tv_sec - session->bw_last_refill_sec) * 1000000000LL + + (now.tv_nsec - session->bw_last_refill_nsec); + session->bw_last_refill_sec = now.tv_sec; + session->bw_last_refill_nsec = now.tv_nsec; - long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0); - bw_tokens += tokens_to_add; - if (bw_tokens > (long long)io_bwlimit) - bw_tokens = (long long)io_bwlimit; + long long tokens_to_add = (long long)((double)session->bwlimit * elapsed_ns / 1000000000.0); + session->bw_tokens += tokens_to_add; + if (session->bw_tokens > (long long)session->bwlimit) + session->bw_tokens = (long long)session->bwlimit; - bw_tokens -= (long long)bytes_written; + session->bw_tokens -= bytes_written; - if (bw_tokens < 0) { - long long deficit_us = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000.0); + if (session->bw_tokens < 0) { + long long deficit_us = + (long long)((double)(-session->bw_tokens) / session->bwlimit * 1000000.0); if (deficit_us >= 1000) poll(NULL, 0, (int)(deficit_us / 1000)); else usleep((useconds_t)deficit_us); - bw_tokens = 0; - clock_gettime(CLOCK_MONOTONIC, &bw_last_refill); + session->bw_tokens = 0; + session->bw_last_refill_sec = now.tv_sec; + session->bw_last_refill_nsec = now.tv_nsec; } - mtx_unlock(&bw_mutex); } void io_set_ssl(SSL* ssl) { + bound_session = NULL; io_ssl = ssl; } @@ -89,8 +129,30 @@ SSL* io_get_ssl(void) { return io_ssl; } -static int io_fd(int dir_fd, int file_descriptor) { - return (dir_fd != -1) ? dir_fd : file_descriptor; +static ProtocolSession* legacy_session(int read_fd, int write_fd) { + if (bound_session) + return bound_session; + int target_read_fd = io_read_fd != -1 ? io_read_fd : read_fd; + int target_write_fd = io_write_fd != -1 ? io_write_fd : write_fd; + if (legacy_io_session.read_fd != target_read_fd || + legacy_io_session.write_fd != target_write_fd) { + legacy_io_session.read_fd = target_read_fd; + legacy_io_session.write_fd = target_write_fd; + legacy_io_session.total_allocated_bytes = 0; + protocol_session_set_bwlimit(&legacy_io_session, io_bwlimit); + } else if (legacy_io_session.bwlimit != io_bwlimit) { + protocol_session_set_bwlimit(&legacy_io_session, io_bwlimit); + } + legacy_io_session.ssl = io_ssl; + return &legacy_io_session; +} + +bool send_n_data(int file_descriptor, const void* data, size_t data_size) { + return protocol_send_n_data(legacy_session(-1, file_descriptor), data, data_size); +} + +bool receive_n_data(int file_descriptor, void* data, size_t data_size) { + return protocol_receive_n_data(legacy_session(file_descriptor, -1), data, data_size); } static int deadline_remaining_ms(const struct timespec* deadline) { @@ -104,9 +166,11 @@ static int deadline_remaining_ms(const struct timespec* deadline) { return ms > INT_MAX ? INT_MAX : (int)ms; } -bool send_n_data(int file_descriptor, const void* data, size_t data_size) { +bool protocol_send_n_data(ProtocolSession* session, const void* data, size_t data_size) { log_message(LOG_LEVEL_DEBUG, " Sending n Data: %zu", data_size); - int fd = io_fd(io_write_fd, file_descriptor); + if (!session) + return false; + int fd = session->write_fd; struct timespec deadline; clock_gettime(CLOCK_MONOTONIC, &deadline); deadline.tv_sec += SEND_TIMEOUT_SEC; @@ -114,7 +178,7 @@ bool send_n_data(int file_descriptor, const void* data, size_t data_size) { ssize_t total_bytes_send = 0; while ((size_t)total_bytes_send < data_size) { size_t chunk = data_size - total_bytes_send; - if (io_bwlimit > 0 && chunk > 65536) + if (session->bwlimit > 0 && chunk > 65536) chunk = 65536; struct pollfd pfd = {.fd = fd, .events = wait_events}; int poll_result = poll(&pfd, 1, deadline_remaining_ms(&deadline)); @@ -127,13 +191,13 @@ bool send_n_data(int file_descriptor, const void* data, size_t data_size) { if (pfd.revents & (POLLERR | POLLNVAL)) return false; ssize_t bytes_send; - if (io_ssl) - bytes_send = SSL_write(io_ssl, (const char*)data + total_bytes_send, chunk); + if (session->ssl) + bytes_send = SSL_write(session->ssl, (const char*)data + total_bytes_send, chunk); else bytes_send = write(fd, (const char*)data + total_bytes_send, chunk); if (bytes_send <= 0) { - if (io_ssl) { - int ssl_err = SSL_get_error(io_ssl, (int)bytes_send); + if (session->ssl) { + int ssl_err = SSL_get_error(session->ssl, (int)bytes_send); if (ssl_err == SSL_ERROR_WANT_WRITE || ssl_err == SSL_ERROR_WANT_READ) { wait_events = ssl_err == SSL_ERROR_WANT_WRITE ? POLLOUT : POLLIN; continue; @@ -142,16 +206,18 @@ bool send_n_data(int file_descriptor, const void* data, size_t data_size) { log_message(LOG_LEVEL_ERROR, "Could not send data"); return false; } - bw_throttle((size_t)bytes_send); + bw_throttle_session(session, (size_t)bytes_send); total_bytes_send += bytes_send; } log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send); return true; } -bool receive_n_data(int file_descriptor, void* data, size_t data_size) { +bool protocol_receive_n_data(ProtocolSession* session, void* data, size_t data_size) { log_message(LOG_LEVEL_DEBUG, " Receiving n Data: %zu", data_size); - int fd = io_fd(io_read_fd, file_descriptor); + if (!session) + return false; + int fd = session->read_fd; struct timespec deadline; clock_gettime(CLOCK_MONOTONIC, &deadline); @@ -176,15 +242,15 @@ bool receive_n_data(int file_descriptor, void* data, size_t data_size) { return false; ssize_t bytes_received; - if (io_ssl) - bytes_received = - SSL_read(io_ssl, (char*)data + total_bytes_received, data_size - total_bytes_received); + if (session->ssl) + bytes_received = SSL_read(session->ssl, (char*)data + total_bytes_received, + data_size - total_bytes_received); else bytes_received = read(fd, (char*)data + total_bytes_received, data_size - total_bytes_received); if (bytes_received <= 0) { - if (io_ssl) { - int ssl_err = SSL_get_error(io_ssl, (int)bytes_received); + if (session->ssl) { + int ssl_err = SSL_get_error(session->ssl, (int)bytes_received); if (ssl_err == SSL_ERROR_WANT_WRITE || ssl_err == SSL_ERROR_WANT_READ) { wait_events = ssl_err == SSL_ERROR_WANT_WRITE ? POLLOUT : POLLIN; continue; @@ -231,24 +297,24 @@ static const char* status_to_string(Status status) { } } -bool send_str(int file_descriptor, const char* data) { +bool protocol_send_str(ProtocolSession* session, const char* data) { if (data == NULL) return false; size_t size = strlen(data); - if (!send_n_data(file_descriptor, &size, sizeof(size_t))) + if (!protocol_send_n_data(session, &size, sizeof(size_t))) return false; - if (!send_n_data(file_descriptor, data, size)) + if (!protocol_send_n_data(session, data, size)) return false; log_message(LOG_LEVEL_DEBUG, "Send String: %s", data); return true; } -char* receive_str(int file_descriptor) { +char* protocol_receive_str(ProtocolSession* session) { size_t size; - if (!receive_n_data(file_descriptor, &size, sizeof(size_t))) + if (!protocol_receive_n_data(session, &size, sizeof(size_t))) return NULL; if (size > MAX_STRING_SIZE || size > SIZE_MAX - 1 || - size + 1 > MAX_CONNECTION_MEMORY - total_allocated_bytes) { + size + 1 > MAX_CONNECTION_MEMORY - session->total_allocated_bytes) { log_message(LOG_LEVEL_ERROR, "String size %zu exceeds maximum %llu", size, (unsigned long long)MAX_STRING_SIZE); return NULL; @@ -256,29 +322,29 @@ char* receive_str(int file_descriptor) { char* data = (char*)malloc(size + 1); if (data == NULL) return NULL; - if (!receive_n_data(file_descriptor, data, size)) { + if (!protocol_receive_n_data(session, data, size)) { free(data); return NULL; } data[size] = '\0'; - total_allocated_bytes += size + 1; + session->total_allocated_bytes += size + 1; log_message(LOG_LEVEL_DEBUG, "Received String: %s", data); return data; } -bool send_data(int file_descriptor, const Data* data) { +bool protocol_send_data(ProtocolSession* session, const Data* data) { unsigned long long data_size = data->size; - if (!send_n_data(file_descriptor, &data_size, sizeof(unsigned long long))) + if (!protocol_send_n_data(session, &data_size, sizeof(unsigned long long))) return false; - if (!send_n_data(file_descriptor, data->data, data_size)) + if (!protocol_send_n_data(session, data->data, data_size)) return false; log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size); return true; } -Data* receive_data(int file_descriptor) { +Data* protocol_receive_data(ProtocolSession* session) { unsigned long long size = 0; - if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long))) + if (!protocol_receive_n_data(session, &size, sizeof(unsigned long long))) return NULL; if (size > MAX_DATA_PAYLOAD_SIZE) { log_message(LOG_LEVEL_ERROR, "Data size %llu exceeds maximum %llu", size, @@ -286,53 +352,78 @@ Data* receive_data(int file_descriptor) { return NULL; } size_t allocation_size = size == 0 ? 1 : (size_t)size; - if (allocation_size > MAX_CONNECTION_MEMORY - total_allocated_bytes) { + if (allocation_size > MAX_CONNECTION_MEMORY - session->total_allocated_bytes) { log_message(LOG_LEVEL_ERROR, "Per-connection memory limit exceeded (%llu + %llu > %llu)", - (unsigned long long)total_allocated_bytes, size, + (unsigned long long)session->total_allocated_bytes, size, (unsigned long long)MAX_CONNECTION_MEMORY); return NULL; } void* data = malloc(allocation_size); if (data == NULL) return NULL; - if (!receive_n_data(file_descriptor, data, (size_t)size)) { + if (!protocol_receive_n_data(session, data, (size_t)size)) { free(data); return NULL; } - total_allocated_bytes += allocation_size; + session->total_allocated_bytes += allocation_size; log_message(LOG_LEVEL_DEBUG, "Received %lld data", size); Data* result = data_create(data, (size_t)size); if (!result) { free(data); - total_allocated_bytes -= allocation_size; + session->total_allocated_bytes -= allocation_size; } return result; } -bool send_int(int file_descriptor, int data) { - if (!send_n_data(file_descriptor, &data, sizeof(int))) +bool protocol_send_int(ProtocolSession* session, int data) { + if (!protocol_send_n_data(session, &data, sizeof(int))) return false; log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data); return true; } -bool receive_int(int file_descriptor, int* data) { - if (!receive_n_data(file_descriptor, data, sizeof(int))) +bool protocol_receive_int(ProtocolSession* session, int* data) { + if (!protocol_receive_n_data(session, data, sizeof(int))) return false; log_message(LOG_LEVEL_DEBUG, "Received Int: %d", *data); return true; } -bool send_status(int file_descriptor, Status status) { - if (!send_n_data(file_descriptor, &status, sizeof(Status))) +bool protocol_send_status(ProtocolSession* session, Status status) { + if (!protocol_send_n_data(session, &status, sizeof(Status))) return false; log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status)); return true; } -bool receive_status(int file_descriptor, Status* status) { - if (!receive_n_data(file_descriptor, status, sizeof(Status))) +bool protocol_receive_status(ProtocolSession* session, Status* status) { + if (!protocol_receive_n_data(session, status, sizeof(Status))) return false; log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(*status)); return true; } + +bool send_str(int fd, const char* data) { + return protocol_send_str(legacy_session(-1, fd), data); +} +char* receive_str(int fd) { + return protocol_receive_str(legacy_session(fd, -1)); +} +bool send_data(int fd, const Data* data) { + return protocol_send_data(legacy_session(-1, fd), data); +} +Data* receive_data(int fd) { + return protocol_receive_data(legacy_session(fd, -1)); +} +bool send_int(int fd, int data) { + return protocol_send_int(legacy_session(-1, fd), data); +} +bool receive_int(int fd, int* data) { + return protocol_receive_int(legacy_session(fd, -1), data); +} +bool send_status(int fd, Status status) { + return protocol_send_status(legacy_session(-1, fd), status); +} +bool receive_status(int fd, Status* status) { + return protocol_receive_status(legacy_session(fd, -1), status); +} diff --git a/src/shared/protocol.h b/src/shared/protocol.h index ded825f..b5811db 100644 --- a/src/shared/protocol.h +++ b/src/shared/protocol.h @@ -19,6 +19,23 @@ typedef struct ssl_st SSL; +/* + * Explicit owner of protocol I/O. A session does not own the descriptors or + * SSL object; it only describes the transport used by a transfer. This makes + * it safe to pass the transport to a worker without relying on inherited + * thread-local state. + */ +typedef struct ProtocolSession { + int read_fd; + int write_fd; + SSL* ssl; + unsigned long long bwlimit; + long long bw_tokens; + long long bw_last_refill_sec; + long bw_last_refill_nsec; + unsigned long long total_allocated_bytes; +} ProtocolSession; + typedef int Status; enum NET_STATUS { STATUS_OK, @@ -39,6 +56,23 @@ void io_set_fds(int read_fd, int write_fd); void io_set_bwlimit(unsigned long long bytes_per_sec); void io_set_ssl(SSL* ssl); SSL* io_get_ssl(void); + +void protocol_session_init(ProtocolSession* session, int read_fd, int write_fd); +/* Transitional bridge for helpers whose signatures still carry only an fd. */ +void protocol_session_bind(ProtocolSession* session); +void protocol_session_unbind(void); +void protocol_session_set_ssl(ProtocolSession* session, SSL* ssl); +void protocol_session_set_bwlimit(ProtocolSession* session, unsigned long long bytes_per_sec); +bool protocol_send_n_data(ProtocolSession* session, const void* data, size_t data_size); +bool protocol_receive_n_data(ProtocolSession* session, void* data, size_t data_size); +bool protocol_send_str(ProtocolSession* session, const char* data); +char* protocol_receive_str(ProtocolSession* session); +bool protocol_send_data(ProtocolSession* session, const Data* data); +Data* protocol_receive_data(ProtocolSession* session); +bool protocol_send_int(ProtocolSession* session, int data); +bool protocol_receive_int(ProtocolSession* session, int* data); +bool protocol_send_status(ProtocolSession* session, Status status); +bool protocol_receive_status(ProtocolSession* session, Status* status); bool send_n_data(int file_descriptor, const void* data, size_t data_size); bool receive_n_data(int file_descriptor, void* data, size_t data_size); diff --git a/src/shared/transport_tcp.c b/src/shared/transport_tcp.c index 49c2fb9..062d60c 100644 --- a/src/shared/transport_tcp.c +++ b/src/shared/transport_tcp.c @@ -151,6 +151,10 @@ int tcp_get_contimeout_sec(void) { return g_contimeout_sec; } +int tcp_get_timeout_sec(void) { + return g_timeout_sec; +} + static void tcp_apply_socket_timeout(int fd) { struct timeval tv; tv.tv_sec = g_timeout_sec; @@ -173,7 +177,7 @@ Client* client_create() { return client; } -bool client_connect(Client* client, char* host, int port) { +bool tcp_connect_socket(Client* client, char* host, int port) { struct addrinfo hints; struct addrinfo* result; memset(&hints, 0, sizeof(hints)); @@ -222,6 +226,12 @@ bool client_connect(Client* client, char* host, int port) { return false; } + return true; +} + +bool client_connect(Client* client, char* host, int port) { + if (!tcp_connect_socket(client, host, port)) + return false; tcp_apply_socket_timeout(client->file_descriptor); return true; } diff --git a/src/shared/transport_tcp.h b/src/shared/transport_tcp.h index d74d694..156531a 100644 --- a/src/shared/transport_tcp.h +++ b/src/shared/transport_tcp.h @@ -30,9 +30,11 @@ void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* chil void server_delete(Server** server); Client* client_create(); bool client_connect(Client* client, char* host, int port); +bool tcp_connect_socket(Client* client, char* host, int port); void client_disconnect(Client* client); void client_delete(Client* client); void tcp_set_timeouts(int timeout_sec, int contimeout_sec); int tcp_get_contimeout_sec(void); +int tcp_get_timeout_sec(void); #endif diff --git a/src/shared/transport_tls.c b/src/shared/transport_tls.c index 49fc554..2c5a4ff 100644 --- a/src/shared/transport_tls.c +++ b/src/shared/transport_tls.c @@ -3,15 +3,13 @@ #include "protocol.h" #include "transport_tcp.h" #include -#include #include #include #include #include #include #include -#include -#include +#include #include bool tls_global_init(void) { @@ -95,6 +93,7 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server, const char* h } // Retry SSL_accept/SSL_connect on WANT_READ/WANT_WRITE (non-blocking handshake) + time_t deadline = time(NULL) + (is_server ? tcp_get_timeout_sec() : tcp_get_contimeout_sec()); int ret; do { if (is_server) @@ -104,7 +103,8 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server, const char* h if (ret <= 0) { int ssl_err = SSL_get_error(ssl, ret); - if (ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE) + if ((ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE) && + time(NULL) < deadline) continue; log_message(LOG_LEVEL_ERROR, "SSL %s failed", is_server ? "accept" : "connect"); log_ssl_errors(); @@ -149,53 +149,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) diff --git a/src/shared/utils.c b/src/shared/utils.c index 9e11988..2926fca 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -237,6 +237,10 @@ bool has_path_traversal(const char* path) { return false; } +bool utils_valid_batch_path(const char* path) { + return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path); +} + char* path_cat(const char* path1, const char* path2) { if (path1 == NULL || *path1 == '\0') return str_dup(path2); diff --git a/src/shared/utils.h b/src/shared/utils.h index d04daf5..8ab4a64 100644 --- a/src/shared/utils.h +++ b/src/shared/utils.h @@ -11,5 +11,6 @@ bool glob_match(const char* pattern, const char* str); bool delete_extras(const char* dest_root, ArrayList* manifest); void utils_set_authorized_root_fd(int fd); bool has_path_traversal(const char* path); +bool utils_valid_batch_path(const char* path); #endif diff --git a/tests/test_client_cli.c b/tests/test_client_cli.c index c6387f5..014c6d3 100644 --- a/tests/test_client_cli.c +++ b/tests/test_client_cli.c @@ -1,4 +1,5 @@ #include "test_client_cli.h" +#include "client_validation.h" #include "config.h" #include "test_utils.h" #include "utils.h" @@ -9,6 +10,56 @@ /* Declaration of parse_args from client_cli.c */ int parse_args(Config* config, int argc, char* argv[], int* positional_args, int* positional_count); +static Config* valid_client_config() { + Config* cfg = config_create(); + if (!cfg) + return NULL; + cfg->send_directory = str_dup("/src"); + cfg->receive_root_directory = str_dup("/dst"); + return cfg; +} + +static void test_validate_config_required_paths() { + Config* cfg = config_create(); + EXPECT_FALSE(validate_config(cfg)); + cfg->send_directory = str_dup("/src"); + EXPECT_FALSE(validate_config(cfg)); + config_delete(cfg); +} + +static void test_validate_config_incompatible_options() { + Config* cfg = valid_client_config(); + cfg->use_sendfile = true; + cfg->use_compression = true; + EXPECT_FALSE(validate_config(cfg)); + cfg->use_compression = false; + cfg->use_incremental = true; + cfg->use_chunk_serialization = true; + EXPECT_FALSE(validate_config(cfg)); + config_delete(cfg); +} + +static void test_validate_config_tls_requirements() { + Config* cfg = valid_client_config(); + cfg->use_tls = true; + EXPECT_FALSE(validate_config(cfg)); + cfg->tls_cert = str_dup("cert.pem"); + EXPECT_FALSE(validate_config(cfg)); + cfg->tls_key = str_dup("key.pem"); + EXPECT_TRUE(validate_config(cfg)); + config_delete(cfg); +} + +static void test_validate_config_delta_sendfile_constraints() { + Config* cfg = valid_client_config(); + cfg->use_delta = true; + EXPECT_FALSE(validate_config(cfg)); + cfg->use_incremental = true; + cfg->use_sendfile = true; + EXPECT_FALSE(validate_config(cfg)); + config_delete(cfg); +} + /* Test main() with --help flag (early return path, no server connection needed) */ static void test_cli_help() { /* We can't easily call main() because it calls send_files which needs a server. @@ -287,6 +338,10 @@ static void test_parse_args_archive() { } void test_client_cli() { + test_validate_config_required_paths(); + test_validate_config_incompatible_options(); + test_validate_config_tls_requirements(); + test_validate_config_delta_sendfile_constraints(); test_cli_help(); test_cli_archive_flags(); test_cli_dry_run(); diff --git a/tests/test_config.c b/tests/test_config.c index 5fb3b60..817d6df 100644 --- a/tests/test_config.c +++ b/tests/test_config.c @@ -224,6 +224,26 @@ static void test_config_send_receive_version_mismatch() { } } +static void test_config_receive_truncated() { + int p[2]; + EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0); + io_set_fds(p[0], p[0]); + io_set_bwlimit(0); + + /* A valid prefix exercises cleanup after allocated wire strings and a + * partially received scalar field. */ + EXPECT_TRUE(send_str(p[1], PROTOCOL_VERSION)); + EXPECT_TRUE(send_str(p[1], "/src")); + EXPECT_TRUE(send_str(p[1], "/dst")); + EXPECT_TRUE(send_int(p[1], 1)); + shutdown(p[1], SHUT_WR); + + const Config* cfg = config_receive(p[0]); + EXPECT_NULL(cfg); + close(p[0]); + close(p[1]); +} + static void test_is_remote_dest() { /* Valid SSH-style destinations */ EXPECT_TRUE(is_remote_dest("user@host:/path")); @@ -256,6 +276,7 @@ void test_config() { if (!is_running_under_valgrind()) { test_config_send_receive(); test_config_send_receive_version_mismatch(); + test_config_receive_truncated(); } test_is_remote_dest(); } diff --git a/tests/test_protocol.c b/tests/test_protocol.c index af1dc00..4c67fee 100644 --- a/tests/test_protocol.c +++ b/tests/test_protocol.c @@ -38,6 +38,23 @@ static void test_send_receive_n_data_zero() { close(p[1]); } +static void test_explicit_session_context() { + int p[2]; + EXPECT_EQ_INT(pipe(p), 0); + ProtocolSession session; + protocol_session_init(&session, p[0], p[1]); + protocol_session_set_bwlimit(&session, 0); + + const char payload[] = "explicit context"; + char received[sizeof(payload)] = {0}; + EXPECT_TRUE(protocol_send_n_data(&session, payload, sizeof(payload))); + EXPECT_TRUE(protocol_receive_n_data(&session, received, sizeof(received))); + EXPECT_EQ_INT(memcmp(payload, received, sizeof(payload)), 0); + + close(p[0]); + close(p[1]); +} + static void test_send_receive_str() { int p[2]; EXPECT_EQ_INT(pipe(p), 0); @@ -173,6 +190,7 @@ static void test_receive_str_truncated() { void test_protocol() { test_send_receive_n_data(); test_send_receive_n_data_zero(); + test_explicit_session_context(); test_send_receive_str(); test_send_receive_str_normal(); test_send_receive_data(); diff --git a/tests/test_scanner.c b/tests/test_scanner.c index 41609f2..0dbcdc1 100644 --- a/tests/test_scanner.c +++ b/tests/test_scanner.c @@ -385,6 +385,34 @@ static void test_scanner_no_patterns() { rmdir(dir); } +static void test_parallel_scanner_root_chunks_without_workers() { + const char* dir = "test_parallel_scan_root"; + const char* file1 = "test_parallel_scan_root/a.txt"; + const char* file2 = "test_parallel_scan_root/b.txt"; + + EXPECT_EQ_INT(mkdir(dir, 0755), 0); + create_test_file(file1, "a"); + create_test_file(file2, "b"); + + ParallelScanner* scanner = parallel_scanner_create(dir, false, 1, NULL, 0, NULL, 0, 0, 0, 0, 0, + false, false, false, false, false); + EXPECT_NOT_NULL(scanner); + + int total_files = 0; + Chunk* chunk; + while ((chunk = parallel_scanner_next(scanner)) != NULL) { + total_files += chunk->element_count; + chunk_destroy(chunk); + } + EXPECT_EQ_INT(total_files, 2); + EXPECT_FALSE(parallel_scanner_failed(scanner)); + + parallel_scanner_destroy(scanner); + unlink(file1); + unlink(file2); + rmdir(dir); +} + void test_scanner() { test_scanner_single_file(); test_scanner_multiple_files(); @@ -399,4 +427,5 @@ void test_scanner() { test_scanner_size_range(); test_scanner_mixed_patterns(); test_scanner_no_patterns(); + test_parallel_scanner_root_chunks_without_workers(); } diff --git a/tests/test_server.c b/tests/test_server.c index 38c99d9..8e86795 100644 --- a/tests/test_server.c +++ b/tests/test_server.c @@ -11,11 +11,7 @@ #include #include -/* Include server.c but rename main to avoid conflict with test runner's main */ -#define main server_main_ -#define FASTSYNC_SERVER_AS_LIB -#include "server.c" -#undef main +#include "receiver.h" /* Test receive_files with immediate FINISHED status */ static void test_receive_files_finished() { @@ -36,7 +32,7 @@ static void test_receive_files_finished() { /* Child: use p[0] for both read and write */ close(p[1]); io_set_fds(p[0], p[0]); - int ret = receive_files(cfg, p[0]); + int ret = receiver_receive_files(cfg, p[0]); close(p[0]); config_delete(cfg); _exit(ret == 0 ? 0 : 1); @@ -88,7 +84,7 @@ static void test_receive_files_single_file() { /* Child: use p[0] for both read and write */ close(p[1]); io_set_fds(p[0], p[0]); - int ret = receive_files(cfg, p[0]); + int ret = receiver_receive_files(cfg, p[0]); close(p[0]); config_delete(cfg); _exit(ret == 0 ? 0 : 1); @@ -149,7 +145,7 @@ static void test_receive_files_abort() { if (pid == 0) { close(p[1]); io_set_fds(p[0], p[0]); - int ret = receive_files(cfg, p[0]); + int ret = receiver_receive_files(cfg, p[0]); close(p[0]); config_delete(cfg); /* Should return -1 on abort */