From 7776a63d3d8d485de7f6febce4de2099b120b41b Mon Sep 17 00:00:00 2001 From: TapTap Date: Mon, 10 Aug 2026 17:26:00 +0200 Subject: [PATCH] fix: close remaining PR 208 review gaps --- src/client/client_send.c | 75 +++++++++++++------- src/client/scanner.c | 60 ++++++++++++++-- src/client/scanner.h | 6 ++ src/shared/file.c | 134 +++++++++++++++++++++++++++-------- src/shared/metadata.c | 16 ++--- src/shared/metadata.h | 2 +- src/shared/multiprocessing.c | 2 + src/shared/protocol.c | 18 +++-- src/shared/utils.c | 39 +++++----- src/shared/utils.h | 2 +- tests/runner.c | 2 + 11 files changed, 262 insertions(+), 94 deletions(-) diff --git a/src/client/client_send.c b/src/client/client_send.c index 21b32b8..ac2bc09 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -28,6 +28,20 @@ /* Forward declaration for progress-reporting thread used in multithreaded send. */ static int progress_thread_fn(void* arg); +static void pipeline_cancel(PipelineContextSender* context) { + mtx_lock(&context->mutex_scanner); + mtx_lock(&context->mutex_loader); + atomic_store(&context->cancelled, true); + context->scanner_done = true; + context->loader_done = true; + cnd_broadcast(&context->condition_not_full_scanner); + cnd_broadcast(&context->condition_not_empty_scanner); + cnd_broadcast(&context->condition_not_full_loader); + cnd_broadcast(&context->condition_not_empty_loader); + mtx_unlock(&context->mutex_loader); + mtx_unlock(&context->mutex_scanner); +} + /* 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( @@ -344,6 +358,7 @@ static int send_chunks_multithreaded(void* pipeline_context) { return ok ? thrd_success : thrd_error; send_fail: + pipeline_cancel(context); client_disconnect(client); client_delete(client); mtx_lock(&context->mutex_progress); @@ -354,6 +369,7 @@ static int send_chunks_multithreaded(void* pipeline_context) { 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); @@ -388,15 +404,7 @@ static int scan_directory_multithreaded(void* pipeline_context) { Chunk* current_chunk; if (scanner == NULL) { log_message(LOG_LEVEL_ERROR, "Failed to create parallel scanner"); - atomic_store(&context->cancelled, true); - cnd_broadcast(&context->condition_not_full_scanner); - cnd_broadcast(&context->condition_not_empty_scanner); - cnd_broadcast(&context->condition_not_full_loader); - cnd_broadcast(&context->condition_not_empty_loader); - mtx_lock(&context->mutex_scanner); - context->scanner_done = true; - cnd_broadcast(&context->condition_not_empty_scanner); - mtx_unlock(&context->mutex_scanner); + pipeline_cancel(context); return thrd_error; } while ((current_chunk = parallel_scanner_next(scanner)) != NULL) { @@ -410,18 +418,14 @@ static int scan_directory_multithreaded(void* pipeline_context) { if (!manifest_entry) { log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); mtx_unlock(&context->mutex_scanner); - atomic_store(&context->cancelled, true); - cnd_broadcast(&context->condition_not_full_scanner); - cnd_broadcast(&context->condition_not_empty_scanner); + pipeline_cancel(context); parallel_scanner_destroy(scanner); return thrd_error; } if (!array_list_add(context->manifest, manifest_entry)) { free(manifest_entry); - atomic_store(&context->cancelled, true); - cnd_broadcast(&context->condition_not_full_scanner); - cnd_broadcast(&context->condition_not_empty_scanner); mtx_unlock(&context->mutex_scanner); + pipeline_cancel(context); chunk_destroy(current_chunk); parallel_scanner_destroy(scanner); return thrd_error; @@ -434,13 +438,21 @@ static int scan_directory_multithreaded(void* pipeline_context) { &context->condition_not_empty_scanner, &context->condition_not_full_scanner, &context->cancelled)) { chunk_destroy(current_chunk); - atomic_store(&context->cancelled, true); - cnd_broadcast(&context->condition_not_full_scanner); - cnd_broadcast(&context->condition_not_empty_scanner); + pipeline_cancel(context); parallel_scanner_destroy(scanner); return thrd_error; } } + if (parallel_scanner_failed(scanner)) { + parallel_scanner_destroy(scanner); + mtx_lock(&context->mutex_scanner); + context->scanner_done = true; + cnd_broadcast(&context->condition_not_empty_scanner); + cnd_broadcast(&context->condition_not_full_scanner); + mtx_unlock(&context->mutex_scanner); + pipeline_cancel(context); + return thrd_error; + } mtx_lock(&context->mutex_scanner); context->scanner_done = true; cnd_signal(&context->condition_not_empty_scanner); @@ -576,6 +588,15 @@ int send_files(Config* config) { time_t last_progress = 0; time_t start = time(NULL); ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL; + if (!scanner || (config->use_delete && !manifest)) { + if (scanner) + directory_scanner_destroy(scanner); + if (manifest) + array_list_delete(manifest); + client_disconnect(client); + client_delete(client); + return 1; + } while ((current_chunk = directory_scanner_next(scanner)) != NULL) { unsigned long long chunk_bytes = 0; for (int i = 0; i < current_chunk->element_count; i++) { @@ -620,6 +641,9 @@ int send_files(Config* config) { if (send_chunk(client, current_chunk, config) != 0) { log_message(LOG_LEVEL_ERROR, "Failed to send chunk"); chunk_destroy(current_chunk); + if (manifest) + array_list_delete(manifest); + manifest = NULL; break; } if (config->show_progress) { @@ -635,12 +659,15 @@ int send_files(Config* config) { } chunk_destroy(current_chunk); } + if (directory_scanner_failed(scanner) || (config->use_delete && manifest == NULL)) + goto send_fail; if (config->use_delete) { if (send_delete_manifest(client->file_descriptor, manifest) != 0) { array_list_delete(manifest); goto send_fail; } array_list_delete(manifest); + manifest = NULL; } if (!send_status(client->file_descriptor, STATUS_FINISHED)) goto send_fail; @@ -662,6 +689,8 @@ int send_files(Config* config) { return ok ? 0 : 1; send_fail: + if (manifest) + array_list_delete(manifest); directory_scanner_destroy(scanner); client_disconnect(client); client_delete(client); @@ -715,14 +744,10 @@ int send_files_multithreaded(Config* config) { if (!scanner_created || !loader_created || !sender_created) { perror("Error creating threads.\n"); - atomic_store(&context->cancelled, true); - context->scanner_done = true; - context->loader_done = true; + pipeline_cancel(context); + mtx_lock(&context->mutex_progress); context->sender_done = true; - cnd_broadcast(&context->condition_not_full_scanner); - cnd_broadcast(&context->condition_not_empty_scanner); - cnd_broadcast(&context->condition_not_full_loader); - cnd_broadcast(&context->condition_not_empty_loader); + mtx_unlock(&context->mutex_progress); if (sender_created) thrd_join(sender, NULL); if (loader_created) diff --git a/src/client/scanner.c b/src/client/scanner.c index 848f66d..aef3fef 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -121,6 +121,7 @@ static int open_next_directory(DirectoryScanner* scanner) { perror("Could not open directory"); free(scanner->current_path); scanner->current_path = NULL; + scanner->failed = true; return -1; } return 1; @@ -128,6 +129,10 @@ static int open_next_directory(DirectoryScanner* scanner) { Chunk* directory_scanner_next(DirectoryScanner* scanner) { ArrayList* chunk_data = array_list_create(file_destroy); + if (!chunk_data) { + scanner->failed = true; + return NULL; + } unsigned long long chunk_data_size = 0; while (1) { @@ -136,7 +141,7 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { if (ret == 0) break; if (ret < 0) - continue; + break; } struct dirent* entry = readdir(scanner->current_dir); @@ -259,7 +264,11 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { file->data->size = stats.st_size; if (scanner->use_metadata) file->metadata = file_metadata_create(&stats); - array_list_add(chunk_data, file); + if (!array_list_add(chunk_data, file)) { + file_destroy(file); + scanner->failed = true; + break; + } chunk_data_size += file->data->size; if (chunk_data_size > scanner->chunk_size) { free(cur_path); @@ -275,6 +284,10 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { return NULL; } +bool directory_scanner_failed(const DirectoryScanner* scanner) { + return scanner == NULL || scanner->failed; +} + typedef struct { ParallelScanner* ps; char** dirs; @@ -302,10 +315,31 @@ static int parallel_worker_thread(void* arg) { 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); + if (!ds) { + mtx_lock(&wa->ps->result_mutex); + wa->ps->failed = true; + atomic_store(&wa->ps->cancelled, true); + cnd_broadcast(&wa->ps->result_not_empty); + cnd_broadcast(&wa->ps->result_not_full); + mtx_unlock(&wa->ps->result_mutex); + break; + } Chunk* chunk; while ((chunk = directory_scanner_next(ds)) != NULL) { - queue_enqueue_multithreaded(wa->ps->result_queue, chunk, &wa->ps->result_mutex, - &wa->ps->result_not_empty, &wa->ps->result_not_full); + if (!queue_enqueue_multithreaded_cancel(wa->ps->result_queue, chunk, &wa->ps->result_mutex, + &wa->ps->result_not_empty, &wa->ps->result_not_full, + &wa->ps->cancelled)) { + chunk_destroy(chunk); + break; + } + } + if (directory_scanner_failed(ds)) { + mtx_lock(&wa->ps->result_mutex); + wa->ps->failed = true; + atomic_store(&wa->ps->cancelled, true); + cnd_broadcast(&wa->ps->result_not_empty); + cnd_broadcast(&wa->ps->result_not_full); + mtx_unlock(&wa->ps->result_mutex); } directory_scanner_destroy(ds); free(wa->dirs[i]); @@ -338,6 +372,7 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata free(ps); return NULL; } + atomic_init(&ps->cancelled, false); int init = 0; bool ok = true; if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success) @@ -500,8 +535,10 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata if (!first) { first = c; } else { - queue_enqueue_multithreaded(ps->result_queue, c, &ps->result_mutex, &ps->result_not_empty, - &ps->result_not_full); + if (!queue_enqueue(ps->result_queue, c)) { + chunk_destroy(c); + ps->failed = true; + } } if (i < root_files->size - 1) { batch = array_list_create(NULL); @@ -585,7 +622,9 @@ Chunk* parallel_scanner_next(ParallelScanner* ps) { return c; } if (ps->num_threads == 0) { + mtx_lock(&ps->result_mutex); ps->done = true; + mtx_unlock(&ps->result_mutex); return NULL; } Chunk* chunk = queue_dequeue_multithreaded( @@ -593,12 +632,19 @@ Chunk* parallel_scanner_next(ParallelScanner* ps) { return chunk; } +bool parallel_scanner_failed(const ParallelScanner* ps) { + return ps == NULL || ps->failed; +} + void parallel_scanner_destroy(ParallelScanner* ps) { if (!ps) return; + mtx_lock(&ps->result_mutex); ps->done = true; - cnd_signal(&ps->result_not_empty); + atomic_store(&ps->cancelled, true); + cnd_broadcast(&ps->result_not_empty); cnd_broadcast(&ps->result_not_full); + mtx_unlock(&ps->result_mutex); for (int i = 0; i < ps->num_threads; i++) thrd_join(ps->threads[i], NULL); free(ps->threads); diff --git a/src/client/scanner.h b/src/client/scanner.h index e4538e4..982b3d2 100644 --- a/src/client/scanner.h +++ b/src/client/scanner.h @@ -6,6 +6,7 @@ #include #include #include +#include typedef struct { Queue* directories; @@ -26,6 +27,7 @@ typedef struct { bool safe_links; bool copy_unsafe_links; bool checksum; + bool failed; } DirectoryScanner; typedef struct { @@ -36,6 +38,8 @@ typedef struct { int num_threads; thrd_t* threads; bool done; + bool failed; + atomic_bool cancelled; int completed; Chunk* initial_chunk; } ParallelScanner; @@ -48,6 +52,7 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_ bool follow_symlinks, bool copy_links, bool safe_links, bool copy_unsafe_links, bool checksum); 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, @@ -58,6 +63,7 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata int num_threads, bool follow_symlinks, bool copy_links, bool safe_links, bool copy_unsafe_links, bool checksum); Chunk* parallel_scanner_next(ParallelScanner* scanner); +bool parallel_scanner_failed(const ParallelScanner* scanner); void parallel_scanner_destroy(ParallelScanner* scanner); #endif diff --git a/src/shared/file.c b/src/shared/file.c index c4419da..063d150 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -10,6 +11,7 @@ #include #include #include +#include #include "compression.h" #include "delta.h" @@ -147,6 +149,8 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, static bool to_disk_secure(const char* path, const void* data, unsigned long long data_size, bool inplace, bool sparse, 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 bool path_is_within_root(const char* root, const char* path) { size_t n = strlen(root); @@ -162,7 +166,10 @@ bool file_save_to_disk(const char* root_directory, File* file, const Config* con const char* partial_dir = (config && config->partial_dir) ? config->partial_dir : NULL; char *confined_backup = NULL, *confined_partial = NULL; - if (has_path_traversal(file->path)) { + if (!file || !file->path || !file->data || has_path_traversal(file->path) || + (backup_enabled && + (!backup_suffix || backup_suffix[0] == '\0' || strchr(backup_suffix, '/') != NULL || + strcmp(backup_suffix, ".") == 0 || strcmp(backup_suffix, "..") == 0))) { log_message(LOG_LEVEL_ERROR, "Path traversal detected in file path: %s", file->path); return false; } @@ -259,7 +266,14 @@ bool file_save_to_disk(const char* root_directory, File* file, const Config* con mkdir_r(bdir); free(backup_dir_path); } - rename(disk_path, backup_path); + if (!rename_secure(disk_path, backup_path)) { + free(backup_path); + free(resolved_root); + free(confined_backup); + free(confined_partial); + free(disk_path); + return false; + } free(backup_path); } } @@ -315,24 +329,6 @@ bool file_save_to_disk(const char* root_directory, File* file, const Config* con return ok; } -static void* old_data_from_path(const char* full_path, unsigned long long old_size) { - void* data = malloc((size_t)old_size); - if (!data) - return NULL; - FILE* fp = fopen(full_path, "rb"); - if (!fp) { - free(data); - return NULL; - } - size_t nread = fread(data, 1, (size_t)old_size, fp); - fclose(fp); - if (nread != (size_t)old_size) { - free(data); - return NULL; - } - return data; -} - static File* receive_delta_file(int fd, const Config* config, const char* check_path, void* old_data, unsigned long long old_size) { if (!old_data) @@ -518,22 +514,54 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { char* full_path = path_cat(config->receive_root_directory, check_path); struct stat st; - bool has_old_file = (full_path && lstat(full_path, &st) == 0); + bool has_old_file = false; + int old_fd = -1; + if (full_path) { + char* leaf = NULL; + int parent_fd = open_secure_parent(full_path, &leaf); + if (parent_fd >= 0) { + old_fd = openat(parent_fd, leaf, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + free(leaf); + close(parent_fd); + has_old_file = old_fd >= 0 && fstat(old_fd, &st) == 0 && S_ISREG(st.st_mode); + } + } unsigned long long old_size = has_old_file ? (unsigned long long)st.st_size : 0; + void* old_data = NULL; + if (has_old_file && old_size > 0) { + old_data = malloc((size_t)old_size); + if (old_data) { + size_t got = 0; + while (got < (size_t)old_size) { + ssize_t n = read(old_fd, (char*)old_data + got, (size_t)old_size - got); + if (n <= 0) { + free(old_data); + old_data = NULL; + break; + } + got += (size_t)n; + } + } + } + if (old_fd >= 0) { + close(old_fd); + old_fd = -1; + } bool match = has_old_file && (unsigned long long)st.st_size == check_size; if (match && config->checksum) { - void* old_data = old_size > 0 ? old_data_from_path(full_path, old_size) : NULL; uint64_t old_checksum = old_size == 0 ? delta_xxhash64("", 0) : 0; if (old_data) old_checksum = delta_xxhash64(old_data, (size_t)old_size); match = (old_size == 0 || old_data) && old_checksum == check_checksum; free(old_data); + old_data = NULL; } else if (match) { match = (long long)st.st_mtime == check_mtime; } if (match) { + free(old_data); if (!send_status(fd, STATUS_OK)) { free(full_path); free(check_path); @@ -549,13 +577,15 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { delta_should_attempt(old_size, check_size, config->delta_max_file_size); if (try_delta) { - void* old_data = old_data_from_path(full_path, old_size); File* delta_file = receive_delta_file(fd, config, check_path, old_data, old_size); + old_data = NULL; /* receive_delta_file consumes the snapshot on every path */ if (delta_file) { free(full_path); free(check_path); return delta_file; } + free(old_data); + old_data = NULL; try_delta = false; } @@ -649,6 +679,21 @@ static int open_secure_parent(const char* path, char** leaf_out) { 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; @@ -677,7 +722,7 @@ static bool to_disk_secure(const char* path, const void* data, unsigned long lon if (!sparse || data_size == 0 || ftruncate(fd, (off_t)data_size) == 0) ok = write_all(fd, data, data_size); if (ok && metadata) - file_restore_metadata_fd(fd, metadata); + ok = file_restore_metadata_fd(fd, metadata); } } else { char tmp[NAME_MAX]; @@ -691,7 +736,7 @@ static bool to_disk_secure(const char* path, const void* data, unsigned long lon if (ok || (!sparse || data_size == 0)) ok = write_all(fd, data, data_size); if (ok && metadata) - file_restore_metadata_fd(fd, metadata); + ok = file_restore_metadata_fd(fd, metadata); if (close(fd) != 0) ok = false; fd = -1; @@ -838,8 +883,35 @@ bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int return false; } + /* sendfile cannot encrypt TLS records. Keep the framing identical but + route encrypted transfers through the deadline-aware IO layer. */ + if (io_get_ssl() != NULL) { + bool loaded = file->data->data != NULL || file_load_data(file); + bool ok = loaded && send_n_data(file_descriptor, file->data->data, (size_t)file_size); + close(fd); + return ok; + } + off_t offset = 0; + struct timespec deadline; + clock_gettime(CLOCK_MONOTONIC, &deadline); + deadline.tv_sec += 60; while ((unsigned long long)offset < file_size) { + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + long long remaining = (long long)(deadline.tv_sec - now.tv_sec) * 1000LL + + (deadline.tv_nsec - now.tv_nsec) / 1000000LL; + if (remaining <= 0) { + close(fd); + return false; + } + struct pollfd pfd = {.fd = file_descriptor, .events = POLLOUT}; + int timeout = remaining > INT_MAX ? INT_MAX : (int)remaining; + int polled = poll(&pfd, 1, timeout); + if (polled <= 0 || (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))) { + close(fd); + return false; + } ssize_t sent = sendfile(file_descriptor, fd, &offset, file_size - offset); if (sent == -1) { if (errno == EAGAIN || errno == EINTR) @@ -911,6 +983,8 @@ size_t file_content_to_buffer(File* file) { } int receive_manifest(int fd, const Config* config, int* next_status) { + int received_status = STATUS_ERROR; + int* status_out = next_status ? next_status : &received_status; int count; if (!receive_int(fd, &count)) return -1; @@ -931,18 +1005,18 @@ int receive_manifest(int fd, const Config* config, int* next_status) { return -1; } } - if (!receive_status(fd, next_status)) { + if (!receive_status(fd, status_out)) { array_list_delete(manifest); return -1; } /* Deletion is a commit operation: never perform it until the sender has completed the manifest frame successfully. */ - if (*next_status != STATUS_FINISHED || !config->use_delete) { + if (*status_out != STATUS_FINISHED || !config->use_delete) { array_list_delete(manifest); - return *next_status == STATUS_FINISHED ? 0 : -1; + return *status_out == STATUS_FINISHED ? 0 : -1; } fprintf(stderr, "Deleting files not in manifest...\n"); - delete_extras(config->receive_root_directory, manifest); + bool deletion_ok = delete_extras(config->receive_root_directory, manifest); array_list_delete(manifest); - return 0; + return deletion_ok ? 0 : -1; } diff --git a/src/shared/metadata.c b/src/shared/metadata.c index 6502e37..c18830c 100644 --- a/src/shared/metadata.c +++ b/src/shared/metadata.c @@ -188,16 +188,16 @@ void file_restore_metadata(const char* path, FileMetadata* metadata) { log_message(LOG_LEVEL_WARNING, "Failed to set timestamps on %s: %s", path, strerror(errno)); } -void file_restore_metadata_fd(int fd, FileMetadata* metadata) { +bool file_restore_metadata_fd(int fd, FileMetadata* metadata) { if (fd < 0 || metadata == NULL) - return; + return metadata == NULL; + bool ok = true; if (fchmod(fd, metadata->mode & 07777 & ~(S_ISUID | S_ISGID)) != 0) - log_message(LOG_LEVEL_WARNING, "Failed to fchmod received file: %s", strerror(errno)); - if (fchown(fd, metadata->uid, metadata->gid) != 0 && errno != EPERM) - log_message(LOG_LEVEL_WARNING, "Failed to fchown received file: %s", strerror(errno)); - struct timespec times[2] = {{.tv_sec = metadata->mtime_sec, .tv_nsec = metadata->mtime_nsec}, + ok = false; + /* Client uid/gid values are deliberately not authoritative. */ + struct timespec times[2] = {{.tv_sec = 0, .tv_nsec = UTIME_OMIT}, {.tv_sec = metadata->mtime_sec, .tv_nsec = metadata->mtime_nsec}}; if (futimens(fd, times) != 0) - log_message(LOG_LEVEL_WARNING, "Failed to restore received file timestamps: %s", - strerror(errno)); + ok = false; + return ok; } diff --git a/src/shared/metadata.h b/src/shared/metadata.h index 7654dfc..0ab6859 100644 --- a/src/shared/metadata.h +++ b/src/shared/metadata.h @@ -30,6 +30,6 @@ FileMetadata* metadata_from_buf(char** buf); bool metadata_send(int file_descriptor, FileMetadata* m); FileMetadata* metadata_receive(int file_descriptor, int* ok); void file_restore_metadata(const char* path, FileMetadata* metadata); -void file_restore_metadata_fd(int fd, FileMetadata* metadata); +bool file_restore_metadata_fd(int fd, FileMetadata* metadata); #endif diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 162add0..1a5b0d4 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -239,6 +239,7 @@ int receive_thread(void* pipeline_context) { 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 { @@ -285,6 +286,7 @@ int write_thread(void* pipeline_context) { file_destroy(file); 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); diff --git a/src/shared/protocol.c b/src/shared/protocol.c index 7b6681d..a3a3aed 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -33,6 +33,7 @@ void io_set_fds(int read_fd, int 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; } static void bw_mutex_init(void) { @@ -247,7 +248,7 @@ char* receive_str(int file_descriptor) { if (!receive_n_data(file_descriptor, &size, sizeof(size_t))) return NULL; if (size > MAX_STRING_SIZE || size > SIZE_MAX - 1 || - total_allocated_bytes > MAX_CONNECTION_MEMORY - (size + 1)) { + size + 1 > MAX_CONNECTION_MEMORY - total_allocated_bytes) { log_message(LOG_LEVEL_ERROR, "String size %zu exceeds maximum %llu", size, (unsigned long long)MAX_STRING_SIZE); return NULL; @@ -260,6 +261,7 @@ char* receive_str(int file_descriptor) { return NULL; } data[size] = '\0'; + total_allocated_bytes += size + 1; log_message(LOG_LEVEL_DEBUG, "Received String: %s", data); return data; } @@ -283,22 +285,28 @@ Data* receive_data(int file_descriptor) { (unsigned long long)MAX_DATA_PAYLOAD_SIZE); return NULL; } - if (total_allocated_bytes + size > MAX_CONNECTION_MEMORY) { + size_t allocation_size = size == 0 ? 1 : (size_t)size; + if (allocation_size > MAX_CONNECTION_MEMORY - 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)MAX_CONNECTION_MEMORY); return NULL; } - void* data = malloc((size_t)size); + void* data = malloc(allocation_size); if (data == NULL) return NULL; if (!receive_n_data(file_descriptor, data, (size_t)size)) { free(data); return NULL; } - total_allocated_bytes += size + 1; + total_allocated_bytes += allocation_size; log_message(LOG_LEVEL_DEBUG, "Received %lld data", size); - return data_create(data, (size_t)size); + Data* result = data_create(data, (size_t)size); + if (!result) { + free(data); + total_allocated_bytes -= allocation_size; + } + return result; } bool send_int(int file_descriptor, int data) { diff --git a/src/shared/utils.c b/src/shared/utils.c index e866a45..c1cedd0 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -137,16 +137,17 @@ static bool is_dir_in_manifest(const char* rel_path, ArrayList* manifest) { return false; } -static void delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifest) { +static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifest) { int scanfd = dup(dirfd); if (scanfd < 0) - return; + return false; DIR* dir = fdopendir(scanfd); if (!dir) { close(scanfd); - return; + return false; } bool all_removed = true; + bool operation_ok = true; const struct dirent* entry; while ((entry = readdir(dir)) != NULL) { if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) @@ -164,9 +165,15 @@ static void delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes } if (S_ISDIR(st.st_mode)) { int childfd = openat(dirfd, entry->d_name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (childfd >= 0) - delete_extras_fd(childfd, child_rel, manifest); - if (unlinkat(dirfd, entry->d_name, AT_REMOVEDIR) != 0 && errno != ENOENT) { + bool child_removed = false; + if (childfd >= 0) { + child_removed = delete_extras_fd(childfd, child_rel, manifest); + close(childfd); + } + if (child_removed && !is_dir_in_manifest(child_rel, manifest) && + unlinkat(dirfd, entry->d_name, AT_REMOVEDIR) != 0 && errno != ENOENT) { + operation_ok = false; + } else if (!child_removed) { all_removed = false; } } else { @@ -180,7 +187,7 @@ static void delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes } if (!found) { if (unlinkat(dirfd, entry->d_name, 0) != 0 && errno != ENOENT) - all_removed = false; + operation_ok = false; fprintf(stderr, " Deleted: %s\n", child_rel); } else { all_removed = false; @@ -189,20 +196,18 @@ static void delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes free(child_rel); } closedir(dir); - // Only remove the directory itself if it is not in the manifest - // and contained no kept entries. - if (all_removed && rel_path[0] != '\0' && !is_dir_in_manifest(rel_path, manifest)) { - /* The caller owns the directory fd; removing this name is done by its - parent, so recursive callers perform it in their own frame. */ - } + (void)all_removed; + return operation_ok; } -void delete_extras(const char* dest_root, ArrayList* manifest) { +bool delete_extras(const char* dest_root, ArrayList* manifest) { int rootfd = open(dest_root, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); if (rootfd < 0) - return; - delete_extras_fd(rootfd, "", manifest); - close(rootfd); + return false; + bool ok = delete_extras_fd(rootfd, "", manifest); + if (close(rootfd) != 0) + ok = false; + return ok; } bool has_path_traversal(const char* path) { diff --git a/src/shared/utils.h b/src/shared/utils.h index 1d1d085..ce01d33 100644 --- a/src/shared/utils.h +++ b/src/shared/utils.h @@ -8,7 +8,7 @@ bool mkdir_r(const char* path); char* str_dup(const char* string); char* path_cat(const char* path1, const char* path2); bool glob_match(const char* pattern, const char* str); -void delete_extras(const char* dest_root, ArrayList* manifest); +bool delete_extras(const char* dest_root, ArrayList* manifest); bool has_path_traversal(const char* path); #endif diff --git a/tests/runner.c b/tests/runner.c index 0bba0f1..ad30e09 100644 --- a/tests/runner.c +++ b/tests/runner.c @@ -25,6 +25,7 @@ #include "test_transport_tls.h" #include "test_utils.h" #include +#include // Define global test state variables int tests_run = 0; @@ -32,6 +33,7 @@ int tests_failed = 0; bool current_test_failed = false; int main() { + signal(SIGPIPE, SIG_IGN); printf("\033[1;36m=== RUNNING UNIT TESTS ===\033[0m\n\n"); RUN_TEST(test_queue);