From e68e33c0c5afd19b163bc331e8e062062b1e5db0 Mon Sep 17 00:00:00 2001 From: TapTap Date: Wed, 29 Jul 2026 18:18:07 +0200 Subject: [PATCH] Apply PR #143 content on top of latest main --- .gitignore | 3 + README.md | 129 ++++++++-- src/client/client_cli.c | 143 +++++++++-- src/client/client_send.c | 56 +++-- src/client/scanner.c | 453 +++++++++++++++++++++++------------ src/client/scanner.h | 35 ++- src/server/server.c | 105 ++++---- src/shared/compression.c | 45 ++-- src/shared/compression.h | 2 + src/shared/config.c | 133 ++-------- src/shared/config.h | 14 +- src/shared/delta.c | 7 +- src/shared/file.c | 338 +++++++------------------- src/shared/file.h | 5 +- src/shared/log.c | 20 +- src/shared/log.h | 3 + src/shared/multiprocessing.c | 42 +++- src/shared/protocol.c | 71 ++++-- src/shared/protocol.h | 5 +- src/shared/transport_ssh.c | 30 +-- src/shared/transport_tcp.c | 244 ++++++------------- src/shared/transport_tcp.h | 11 +- src/shared/transport_tls.c | 37 +-- src/shared/utils.c | 67 +++--- src/shared/utils.h | 1 + tests/runner.c | 6 +- tests/test_file.c | 6 +- tests/test_file_sendfile.c | 4 - tests/test_protocol.c | 27 +-- tests/test_scanner.c | 24 +- tests/test_transport_ssh.c | 18 +- tests/test_transport_tcp.c | 96 ++------ tests/test_transport_tls.c | 79 +----- 33 files changed, 1114 insertions(+), 1145 deletions(-) diff --git a/.gitignore b/.gitignore index 1001d2b..a6030f3 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,6 @@ __pycache__/ build-asan coverage.info build-*/ +build2/ +build3/ +build_docker2/ diff --git a/README.md b/README.md index 38682d8..353133c 100644 --- a/README.md +++ b/README.md @@ -5,17 +5,26 @@ A high-performance file synchronization system with SSH and TCP transport, TLS e ## Technical Overview 1. **Dual transport**: custom TCP client-server or SSH subprocess (rsync-style `user@host:/path`) -2. **TLS encryption**: OpenSSL-based TLS 1.2+ for encrypted TCP connections +2. **TLS encryption**: OpenSSL-based TLS 1.2+ for encrypted TCP connections with optional CA verification 3. **Chunked file transfer**: files grouped into configurable-size chunks (default ~10 MB) 4. **Streaming zstd compression** (levels 1–22) using `ZSTD_compressStream2` 5. **Multithreading**: producer-consumer pipeline with thread-safe queues (scanner → loader → sender) 6. **Incremental sync**: skip files unchanged since last transfer (compares size + mtime) -7. **Metadata preservation**: `mode`, `uid`, `gid`, `mtime` restored on disk when enabled -8. **`sendfile()` zero-copy** on TCP (~2× faster on loopback) -9. **SSH ControlMaster** for connection reuse across repeated invocations -10. **Bandwidth limiting**: token-bucket throttling (`--bwlimit`) -11. **`--delete`**: receiver removes files not present in sender manifest -12. **`--exclude` / `--include`**: glob-pattern filename filtering +7. **Batch incremental**: send incremental checks in batched groups for reduced round-trips +8. **Metadata preservation**: `mode`, `uid`, `gid`, `mtime` restored on disk when enabled +9. **`sendfile()` zero-copy** on TCP (~2× faster on loopback) +10. **SSH ControlMaster** for connection reuse across repeated invocations +11. **Bandwidth limiting**: token-bucket throttling (`--bwlimit`) +12. **`--delete`**: receiver removes files not present in sender manifest +13. **`--exclude` / `--include`**: glob-pattern filename filtering +14. **Path traversal protection**: `..` sequences in file paths are rejected automatically +15. **Connection limits**: server enforces maximum concurrent connections (default 100) +16. **Keep-alive**: periodic `STATUS_KEEPALIVE` messages detect stalled connections +17. **Abort handling**: `SIGINT` sends `STATUS_ABORT` for clean server-side teardown +18. **Atomic writes**: received files are written to a temporary name then atomically renamed +19. **Backup mode**: `--backup` preserves overwritten files with optional `--backup-dir` +20. **Log file**: `--log-file` redirects log output to a file instead of stderr +21. **Transfer statistics**: `--stats` prints summary of transferred bytes, files, and timing ## System Architecture @@ -25,20 +34,33 @@ A high-performance file synchronization system with SSH and TCP transport, TLS e - Streaming zstd compression with configurable level - Chunk serialization (compact binary format) or per-file transfer - Incremental transfer: sends file metadata to server, skips unchanged files +- Batch incremental: groups incremental checks to minimize round-trips - Manifests all sent paths when `--delete` is active - Sends via TCP `sendfile()` or SSH pipe - Optional progress display with throughput - Bandwidth limiting via token-bucket algorithm +- Configurable I/O and connection timeouts (`--timeout`, `--contimeout`) +- Quiet mode (`-q`/`--quiet`) suppresses all non-error output +- Backup overwritten files (`--backup`) with optional directory (`--backup-dir`) +- Transfer statistics summary (`--stats`) +- Maximum directory depth control (`--max-depth`) +- Log file output (`--log-file`) +- Configurable multithreaded queue size (`--queue-size`) +- Exclude patterns from file (`--exclude-from`) ### Server - TCP mode: listens on configurable port (default 8080); SSH mode: runs via `--stdio` -- TLS mode: wraps TCP connections with OpenSSL +- TLS mode: wraps TCP connections with OpenSSL with optional CA verification - Receives and reassembles files - Decompresses (streaming zstd), deserializes, restores metadata - Handles incremental checks: compares size + mtime against destination files +- Handles batch incremental checks for reduced round-trips - Processes `STATUS_MANIFEST` for `--delete`: walks destination tree, removes extras -- Per-connection concurrency via `fork()` +- Per-connection concurrency via `fork()` with configurable connection limit (default 100) - Thread pool for parallel processing +- Atomic writes: files written to `.tmp` path then atomically renamed on success +- Abort handling: cleanly shuts down on `STATUS_ABORT` from client +- Path traversal protection: rejects file paths containing `..` ## Protocol Details @@ -52,6 +74,11 @@ A high-performance file synchronization system with SSH and TCP transport, TLS e | `STATUS_CHUNK` | Following data is a serialized chunk | | `STATUS_MANIFEST` | Following data is a file manifest (for `--delete`) | | `STATUS_CHECK` | Incremental check: client sends file path + size + mtime, server responds with OK (skip) or NEXT (send) | +| `STATUS_CHECK_BATCH` | Batch incremental check: multiple file checks sent in one message | +| `STATUS_KEEPALIVE` | Keep-alive heartbeat to detect stalled connections | +| `STATUS_ABORT` | Abort signal: client interrupts, server cleans up and exits | +| `STATUS_DELTA_SIGNATURE` | Delta sync: following data is a file signature (rsync-style rolling hash) | +| `STATUS_DELTA_DATA` | Delta sync: following data is a delta patch for a file | ### Wire Format — Metadata @@ -59,12 +86,16 @@ When `use_metadata` is enabled (`-M`), each file entry carries a 4-byte `present ### Transfer Flow ``` -Config → (STATUS_NEXT | STATUS_CHUNK | STATUS_CHECK)* → [STATUS_MANIFEST] → STATUS_FINISHED → STATUS_OK +Config → (STATUS_NEXT | STATUS_CHUNK | STATUS_CHECK | STATUS_CHECK_BATCH)* → [STATUS_MANIFEST] → STATUS_FINISHED → STATUS_OK ``` +Keep-alive (`STATUS_KEEPALIVE`) may be sent at any point during the transfer. The receiver resets its inactivity timer on receipt. If no data arrives within the receive timeout, the connection is aborted. + +Abort (`STATUS_ABORT`) may be sent at any point. On receipt the server cleans up temporary files and exits the child process. + ### Protocol Version -`1.1.0` — server and client must match. Mismatch results in `STATUS_ERROR`. +`1.3.0` — server and client must match. Mismatch results in `STATUS_ERROR`. ## Command-Line Arguments @@ -83,15 +114,26 @@ Config → (STATUS_NEXT | STATUS_CHUNK | STATUS_CHECK)* → [STATUS_MANIFEST] | `-n, --dry-run` | Scan and print what would be transferred | | `-p ` | SSH port (default: 22) | | `-v, --verbose` | Enable debug logging | +| `-q, --quiet` | Suppress all non-error output | +| `--silent` | Alias for `--quiet` | | `--progress` | Show real-time transfer speed | | `--delete` | Delete files on receiver not present in source | | `--exclude ` | Exclude files matching glob pattern (repeatable) | +| `--exclude-from ` | Read exclude patterns from a file (one per line) | | `--include ` | Only transfer files matching glob pattern (repeatable, whitelist) | | `--max-size ` | Skip files larger than n bytes | | `--min-size ` | Skip files smaller than n bytes | | `--incremental` | Skip files unchanged since last transfer (size + mtime). Auto-enables `--preserve`. Incompatible with `-s`. | | `--bwlimit ` | Bandwidth limit in kilobytes per second | | `--chunk-size ` | Chunk size in bytes (default: 10485760) | +| `--timeout ` | I/O timeout in seconds (default: 30) | +| `--contimeout ` | Connection timeout in seconds (default: 10) | +| `--backup` | Backup existing destination files before overwriting | +| `--backup-dir ` | Target directory for backups (requires `--backup`) | +| `--stats` | Print transfer statistics at end (bytes, files, timing) | +| `--max-depth ` | Maximum directory depth to recurse (0 = unlimited, default: 0) | +| `--log-file ` | Write log messages to file instead of stderr | +| `--queue-size ` | Queue capacity for multithreaded mode (default: 100) | | `--source-dir ` | Source directory (overrides `FASTSYNC_SOURCE_DIR`) | | `--dest-dir ` | Server destination directory (overrides `FASTSYNC_DEST_DIR`) | | `--save-to-disk` | Write received files to disk | @@ -122,6 +164,12 @@ Config → (STATUS_NEXT | STATUS_CHUNK | STATUS_CHECK)* → [STATUS_MANIFEST] | `FASTSYNC_SOURCE_DIR` | — | Source directory fallback | | `FASTSYNC_DEST_DIR` | — | Destination directory fallback | | `FASTSYNC_SAVE_TO_DISK` | `false` | Disk persistence fallback | +| `FASTSYNC_SSH_PORT` | `22` | Default SSH port | +| `FASTSYNC_SERVER_HOST` | `127.0.0.1` | Default server host | +| `FASTSYNC_SERVER_PORT` | `8080` | Default server port | +| `FASTSYNC_TLS_CERT` | — | Default TLS certificate path | +| `FASTSYNC_TLS_KEY` | — | Default TLS private key path | +| `FASTSYNC_TLS_CA` | — | Default TLS CA certificate path | ## Implementation Details @@ -129,21 +177,44 @@ Config → (STATUS_NEXT | STATUS_CHUNK | STATUS_CHECK)* → [STATUS_MANIFEST] 1. **Chunk** — collection of files (~10 MB total by default) 2. **File** — path, content (`Data`), optional `FileMetadata` pointer 3. **FileMetadata** — `mode`, `uid`, `gid`, `mtime_sec`, `mtime_nsec` -4. **Config** — runtime parameters (transported over wire, TLS settings excluded) +4. **Config** — runtime parameters (transported over wire, TLS settings excluded). Includes `timeout`, `contimeout`, `quiet`, `backup`, `backup_dir`, `stats`, `max_depth`, `log_file`, `queue_size`. 5. **Queue** — thread-safe bounded queue with condition variables -6. **DirectoryScanner** — recursive BFS traversal with exclude and include pattern support +6. **DirectoryScanner** — recursive BFS traversal with exclude and include pattern support, max-depth enforcement ### Key Algorithms -1. **File scanning** — BFS directory traversal; entries matched against exclude and include patterns +1. **File scanning** — BFS directory traversal; entries matched against exclude and include patterns, max-depth enforced 2. **Chunking** — files accumulated until `chunk_size` threshold, then flushed 3. **Compression** — streaming zstd via `ZSTD_compressStream2` / `ZSTD_decompressStream` -4. **Network protocol** — status-code-driven exchange with metadata packing -5. **Incremental check** — client sends `STATUS_CHECK` + path + size + mtime; server compares against destination +4. **Network protocol** — status-code-driven exchange with metadata packing, keep-alive, and abort support +5. **Incremental check** — client sends `STATUS_CHECK` + path + size + mtime; server compares against destination. Can be batched via `STATUS_CHECK_BATCH` for reduced round-trips. 6. **Bandwidth limiting** — token-bucket algorithm with `nanosleep` throttling on 64 KB write chunks 7. **Metadata restoration** — `chmod()`, `chown()`, `utimensat()` on the receiving side 8. **`--delete`** — sender tracks all sent paths; receiver walks destination tree and removes unlisted files/directories 9. **SSH transport** — `socketpair()` + `fork()` + `execvp("ssh", ...)` with `ControlMaster` and port support 10. **TLS transport** — OpenSSL `SSL_CTX` with TLS 1.2 minimum, optional CA verification, transparent `SSL_read`/`SSL_write` via `io_set_ssl()` +11. **Path traversal protection** — `has_path_traversal()` rejects any file path containing `..` components, preventing directory escape attacks +12. **Connection limiting** — server tracks active connections and rejects new ones beyond `max_connections` (default 100) +13. **Keep-alive** — idle connections receive periodic `STATUS_KEEPALIVE` to detect half-open TCP connections +14. **Abort handling** — `SIGINT` sets an abort flag; the next protocol operation sends `STATUS_ABORT` for clean server cleanup +15. **Atomic writes** — files are written to a `.tmp` suffix then atomically renamed via `rename()`, preventing partial files +16. **Backup** — before overwriting, existing files are moved to `--backup-dir` (or same directory with `~` suffix) preserving the original + +## Security Features + +### Path Traversal Protection +All received file paths are validated by `has_path_traversal()` before any disk operation. Any path containing `..` components is rejected with `STATUS_ERROR`, preventing directory escape attacks. + +### TLS Certificate Verification +When `--ca` is provided, the server performs mutual TLS verification (`SSL_VERIFY_PEER` with depth 4). Without `--ca`, TLS is still encrypted but peer certificates are not verified. + +### Connection Limits +The server enforces a maximum of 100 concurrent connections (configurable via `max_connections` in `Server`). When the limit is reached, new connections are immediately rejected and closed. + +### Abort Handling +If the client receives `SIGINT` (Ctrl+C) during a transfer, it sends `STATUS_ABORT` to the server. The server then cleans up temporary files and exits the child process, preventing incomplete files from remaining on disk. + +### Atomic Writes +Received files are written to a temporary path (suffixed with `.tmp`) and then atomically renamed to the final filename via `rename()`. This prevents partial or corrupted files from appearing at the destination if the transfer is interrupted. ## Build Requirements @@ -223,6 +294,21 @@ Place the `fastsync-server` binary in the remote `$PATH`. The client runs `ssh u # Bandwidth limit to 1 MB/s ./build/client --bwlimit 1024 /src user@host:/dst +# With timeouts, quiet mode, and stats +./build/client --timeout 60 --contimeout 15 --quiet --stats /src user@host:/dst + +# Backup overwritten files to a directory +./build/client --backup --backup-dir /backups /src user@host:/dst + +# Exclude patterns from file, limit depth +./build/client --exclude-from ignore.txt --max-depth 3 /src user@host:/dst + +# Custom queue size for multithreading +./build/client -m --queue-size 200 /src user@host:/dst + +# Log to file +./build/client --log-file /tmp/fastsync.log /src user@host:/dst + # All features ./build/client -a --progress --chunk-size 5242880 --exclude "*.log" --delete /src /dst ``` @@ -230,7 +316,9 @@ Place the `fastsync-server` binary in the remote `$PATH`. The client runs `ssh u ## Testing ```bash -# Unit tests (7 suites) +# Unit tests (18 suites — array_list, chunk, compression, config, data, delta, file, glob, +# metadata, property, protocol, queue, robustness, scanner, +# shared_utils, stress, transport_tcp, transport_ssh, transport_tls) ./build/tests # Integration + benchmark suite @@ -244,12 +332,15 @@ The benchmark prints throughput metrics, best configuration, and speedup vs rsyn 1. Chunk size (~10 MB default) balances memory and transfer efficiency 2. Compression level trades CPU for bandwidth 3. `sendfile()` bypasses userspace — ~2× faster on localhost for large files -4. Multithreading scales with core count +4. Multithreading scales with core count; `--queue-size` controls pipeline buffering 5. Metadata transfer adds negligible overhead (~24 bytes per file when enabled) 6. SSH socketpair buffer set to 1 MB for improved pipe throughput 7. SSH ControlMaster reuses connections across repeated invocations 8. Incremental sync eliminates redundant transfers entirely -9. Bandwidth limiting uses token-bucket with nanosleep for accurate throttling +9. Batch incremental reduces round-trips by grouping multiple checks into one message +10. Bandwidth limiting uses token-bucket with nanosleep for accurate throttling +11. Atomic writes add a single `rename()` per file — negligible overhead +12. Path traversal check is O(n) in path length with negligible cost ## Benchmark Results diff --git a/src/client/client_cli.c b/src/client/client_cli.c index b23ae50..ba17a21 100644 --- a/src/client/client_cli.c +++ b/src/client/client_cli.c @@ -3,6 +3,7 @@ #include "delta.h" #include "log.h" #include "protocol.h" +#include "transport_tcp.h" #include "transport_tls.h" #include "utils.h" #include @@ -31,6 +32,8 @@ static void print_usage(void) { printf(" --delete Delete files on receiver not in source\n"); printf(" --exclude Exclude files matching pattern\n"); printf(" --include Only include files matching pattern\n"); + printf(" --exclude-from Read exclude patterns from file\n"); + printf(" --include-from Read include patterns from file\n"); printf(" --max-size Skip files larger than n bytes\n"); printf(" --min-size Skip files smaller than n bytes\n"); printf(" --incremental Skip files unchanged since last transfer\n"); @@ -55,11 +58,51 @@ static void print_usage(void) { printf(" --cert TLS certificate file (PEM)\n"); printf(" --key TLS private key file (PEM)\n"); printf(" --ca TLS CA certificate file (PEM)\n"); + printf(" --timeout I/O timeout in seconds (default: 30)\n"); + printf(" --contimeout Connection timeout in seconds (default: 10)\n"); + printf(" -q, --quiet Suppress non-error output\n"); + printf(" --silent Alias for --quiet\n"); + printf(" --backup Backup existing files before overwriting\n"); + printf(" --backup-dir Directory for backups (requires --backup)\n"); + printf(" --stats Print transfer statistics at end\n"); + printf(" --max-depth Maximum directory depth (0=unlimited)\n"); + printf(" --log-file Write log messages to file\n"); + printf(" --queue-size Queue capacity for multithreaded mode (default: 100)\n"); printf(" --partial Keep partial files on interrupted transfer\n"); printf(" --fastsync-server-path \n"); printf(" Path to fastsync-server on remote (default: fastsync-server)\n"); printf(" --help Show this help\n"); - printf(" -V, --version Show version and exit\n"); +} + +static int read_patterns_from_file(const char* filepath, char*** patterns, int* count) { + FILE* fp = fopen(filepath, "r"); + if (!fp) { + fprintf(stderr, "Error: could not open pattern file '%s': %s\n", filepath, strerror(errno)); + return -1; + } + char line[4096]; + while (fgets(line, sizeof(line), fp)) { + char* p = line; + while (*p == ' ' || *p == '\t') + p++; + if (*p == '#' || *p == '\n' || *p == '\0') + continue; + size_t len = strlen(p); + while (len > 0 && (p[len - 1] == '\n' || p[len - 1] == '\r')) + p[--len] = '\0'; + if (len == 0) + continue; + char** tmp = realloc(*patterns, (*count + 1) * sizeof(char*)); + if (!tmp) { + fprintf(stderr, "Error: memory allocation failed for pattern file\n"); + fclose(fp); + return -1; + } + *patterns = tmp; + (*patterns)[(*count)++] = str_dup(p); + } + fclose(fp); + return 0; } int main(int argc, char* argv[]) { @@ -73,13 +116,16 @@ int main(int argc, char* argv[]) { } int exit_code = 0; + Config* config = NULL; bool config_owned_by_pipeline = false; - Config* config = config_create(str_dup(PROTOCOL_VERSION), NULL, NULL, save_to_disk, false, false, - false, false, 5, false, 0); - if (config == NULL) { + + char* config_version = str_dup(PROTOCOL_VERSION); + if (!config_version) { exit_code = 1; goto cleanup; } + config = config_create(config_version, NULL, NULL, save_to_disk, false, false, false, false, 5, + false, 0); int positional_args[2]; int positional_count = 0; @@ -88,9 +134,6 @@ int main(int argc, char* argv[]) { if (strcmp(argv[i], "--help") == 0) { print_usage(); goto cleanup; - } else if (strcmp(argv[i], "-V") == 0 || strcmp(argv[i], "--version") == 0) { - printf("fastsync version %s\n", PROTOCOL_VERSION); - goto cleanup; } else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--archive") == 0) { config->use_compression = true; config->use_multithreading = true; @@ -99,14 +142,7 @@ int main(int argc, char* argv[]) { } else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) { config->dry_run = true; } else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) { - char* end; - long p = strtol(argv[++i], &end, 10); - if (*end != '\0' || p <= 0 || p > 65535) { - fprintf(stderr, "Error: invalid SSH port '%s' (must be 1-65535)\n", argv[i]); - exit_code = 1; - goto cleanup; - } - config->ssh_port = (int)p; + config->ssh_port = atoi(argv[++i]); } else if (strcmp(argv[i], "--delete") == 0) { config->use_delete = true; } else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) { @@ -183,14 +219,7 @@ int main(int argc, char* argv[]) { free(config->server_host); config->server_host = str_dup(argv[++i]); } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) { - char* end; - long p = strtol(argv[++i], &end, 10); - if (*end != '\0' || p <= 0 || p > 65535) { - fprintf(stderr, "Error: invalid server port '%s' (must be 1-65535)\n", argv[i]); - exit_code = 1; - goto cleanup; - } - config->server_port = (int)p; + config->server_port = atoi(argv[++i]); } else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) { char* end; errno = 0; @@ -224,6 +253,64 @@ int main(int argc, char* argv[]) { } else if (strcmp(argv[i], "--ca") == 0 && i + 1 < argc) { free(config->tls_ca); config->tls_ca = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--timeout") == 0 && i + 1 < argc) { + config->timeout = atoi(argv[++i]); + if (config->timeout <= 0) { + fprintf(stderr, "Error: --timeout must be a positive integer\n"); + exit_code = 1; + goto cleanup; + } + } else if (strcmp(argv[i], "--contimeout") == 0 && i + 1 < argc) { + config->contimeout = atoi(argv[++i]); + if (config->contimeout <= 0) { + fprintf(stderr, "Error: --contimeout must be a positive integer\n"); + exit_code = 1; + goto cleanup; + } + } else if (strcmp(argv[i], "-q") == 0 || strcmp(argv[i], "--quiet") == 0 || + strcmp(argv[i], "--silent") == 0) { + config->quiet = true; + } else if (strcmp(argv[i], "--backup") == 0) { + config->backup = true; + } else if (strcmp(argv[i], "--backup-dir") == 0 && i + 1 < argc) { + config->backup_dir = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--stats") == 0) { + config->stats = true; + } else if (strcmp(argv[i], "--max-depth") == 0 && i + 1 < argc) { + config->max_depth = atoi(argv[++i]); + if (config->max_depth < 0) { + fprintf(stderr, "Error: --max-depth must be a non-negative integer\n"); + exit_code = 1; + goto cleanup; + } + } else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) { + FILE* lf = fopen(argv[++i], "a"); + if (!lf) { + fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno)); + exit_code = 1; + goto cleanup; + } + config->log_file = lf; + log_set_file(lf); + } else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) { + config->queue_size = atoi(argv[++i]); + if (config->queue_size <= 0) { + fprintf(stderr, "Error: --queue-size must be a positive integer\n"); + exit_code = 1; + goto cleanup; + } + } else if (strcmp(argv[i], "--exclude-from") == 0 && i + 1 < argc) { + if (read_patterns_from_file(argv[++i], &config->exclude_patterns, &config->exclude_count) != + 0) { + exit_code = 1; + goto cleanup; + } + } else if (strcmp(argv[i], "--include-from") == 0 && i + 1 < argc) { + if (read_patterns_from_file(argv[++i], &config->include_patterns, &config->include_count) != + 0) { + exit_code = 1; + goto cleanup; + } } else if (strcmp(argv[i], "--partial") == 0) { config->partial = true; } else if (strcmp(argv[i], "--fastsync-server-path") == 0 && i + 1 < argc) { @@ -327,6 +414,8 @@ int main(int argc, char* argv[]) { tls_global_init(); } + tcp_set_timeouts(config->timeout, config->contimeout); + if (config->use_multithreading) { config_owned_by_pipeline = true; exit_code = send_files_multithreaded(config); @@ -335,7 +424,11 @@ int main(int argc, char* argv[]) { } cleanup: - if (!config_owned_by_pipeline) - config_delete(config); + if (config) { + if (config->log_file) + fclose(config->log_file); + if (!config_owned_by_pipeline) + config_delete(config); + } return exit_code; } diff --git a/src/client/client_send.c b/src/client/client_send.c index 245e8e9..adc582c 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -21,6 +21,9 @@ #include #include #include +#include + +#define STREAM_THRESHOLD (64ULL * 1024 * 1024) static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) { *out_sig = NULL; @@ -208,10 +211,13 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) { return 0; } - bool use_sendfile = config->use_sendfile && !config->use_compression; for (int i = 0; i < chunk->element_count; i++) { - int rc = - send_single_file(client, chunk->items[i], config, config->use_incremental, use_sendfile); + File* f = chunk->items[i]; + if (f == NULL) + continue; + bool stream = f->data->data == NULL && f->data->size > 0; + bool use_sendfile = (config->use_sendfile && !config->use_compression) || stream; + int rc = send_single_file(client, f, config, config->use_incremental, use_sendfile); if (rc == 1) continue; if (rc < 0) @@ -296,16 +302,14 @@ static int send_chunks_multithreaded(void* pipeline_context) { static int scan_directory_multithreaded(void* pipeline_context) { PipelineContextSender* context = (PipelineContextSender*)pipeline_context; - mtx_lock(&context->mutex_scanner); - DirectoryScanner* scanner = directory_scanner_create( + 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); - mtx_unlock(&context->mutex_scanner); + context->config->min_size, context->config->max_depth, 4); Chunk* current_chunk; - while ((current_chunk = directory_scanner_next(scanner)) != NULL) { + 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++) { @@ -325,7 +329,7 @@ static int scan_directory_multithreaded(void* pipeline_context) { cnd_signal(&context->condition_not_empty_scanner); mtx_unlock(&context->mutex_scanner); - directory_scanner_destroy(scanner); + parallel_scanner_destroy(scanner); return thrd_success; } @@ -344,9 +348,12 @@ static int load_files_multithreaded(void* pipeline_context) { } if (!context->config->use_sendfile) { for (int i = 0; i < chunk->element_count; i++) { - if (!file_load_data(chunk->items[i])) { + File* f = chunk->items[i]; + 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(chunk->items[i]); + file_destroy(f); chunk->items[i] = NULL; } } @@ -362,7 +369,7 @@ int send_files(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->min_size, config->max_depth); Chunk* chunk; int file_count = 0; unsigned long long total_bytes = 0; @@ -416,7 +423,7 @@ int send_files(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->min_size, config->max_depth); Chunk* current_chunk; unsigned long long total_bytes = 0; time_t last_progress = 0; @@ -435,7 +442,10 @@ int send_files(Config* config) { } if (!config->use_sendfile) { for (int i = 0; i < current_chunk->element_count; i++) { - if (!file_load_data(current_chunk->items[i])) { + File* f = current_chunk->items[i]; + if (f->data->size > STREAM_THRESHOLD) + continue; + if (!file_load_data(f)) { log_message(LOG_LEVEL_ERROR, "Failed to load file data"); continue; } @@ -502,7 +512,7 @@ int send_files_multithreaded(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->min_size, config->max_depth); Chunk* chunk; int file_count = 0; unsigned long long total_bytes = 0; @@ -520,8 +530,20 @@ int send_files_multithreaded(Config* config) { return 0; } - Queue* q1 = queue_create(100, chunk_destroy); - Queue* q2 = queue_create(100, chunk_destroy); + long pages = sysconf(_SC_AVPHYS_PAGES); + long page_size = sysconf(_SC_PAGE_SIZE); + unsigned long long available_memory = + pages > 0 && page_size > 0 ? (unsigned long long)pages * (unsigned long long)page_size + : 512ULL * 1024 * 1024; + unsigned long long avg_file_size = 1024 * 1024; + int qsize = (int)(available_memory / avg_file_size); + if (qsize < 10) + qsize = 10; + if (qsize > 1000) + qsize = 1000; + + Queue* q1 = queue_create(qsize, chunk_destroy); + Queue* q2 = queue_create(qsize, chunk_destroy); if (!q1 || !q2) { if (q1) queue_destroy(q1); diff --git a/src/client/scanner.c b/src/client/scanner.c index f4e48ad..76fbd78 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -9,101 +9,53 @@ #include #include #include +#include #include -DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata, +typedef struct { + char* path; + int depth; +} DirEntry; + +static void dir_entry_destroy(void* item) { + if (item) { + DirEntry* de = (DirEntry*)item; + free(de->path); + free(de); + } +} + +static DirEntry* dir_entry_create(const char* path, int depth) { + DirEntry* de = malloc(sizeof(DirEntry)); + if (de) { + de->path = str_dup(path); + de->depth = depth; + } + return de; +} + +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) { - return directory_scanner_create_full(root_directory, use_metadata, chunk_size, exclude_patterns, - exclude_count, include_patterns, include_count, max_size, - min_size, true); -} - -DirectoryScanner* directory_scanner_create_full(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, bool follow_symlinks) { + unsigned long long min_size, int max_depth) { DirectoryScanner* scanner = malloc(sizeof(DirectoryScanner)); if (scanner == NULL) return NULL; - scanner->directories = queue_create(100, free); + scanner->directories = queue_create(100, dir_entry_destroy); scanner->current_dir = NULL; scanner->current_path = NULL; scanner->use_metadata = use_metadata; scanner->chunk_size = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE; - /* Deep-copy exclude patterns */ - if (exclude_count > 0 && exclude_patterns != NULL) { - scanner->exclude_patterns = malloc((size_t)exclude_count * sizeof(char*)); - if (scanner->exclude_patterns == NULL) { - queue_destroy(scanner->directories); - free(scanner); - return NULL; - } - for (int i = 0; i < exclude_count; i++) { - scanner->exclude_patterns[i] = str_dup(exclude_patterns[i]); - if (scanner->exclude_patterns[i] == NULL) { - for (int j = 0; j < i; j++) - free(scanner->exclude_patterns[j]); - free(scanner->exclude_patterns); - queue_destroy(scanner->directories); - free(scanner); - return NULL; - } - } - } else { - scanner->exclude_patterns = NULL; - } + scanner->exclude_patterns = exclude_patterns; scanner->exclude_count = exclude_count; - - /* Deep-copy include patterns */ - if (include_count > 0 && include_patterns != NULL) { - scanner->include_patterns = malloc((size_t)include_count * sizeof(char*)); - if (scanner->include_patterns == NULL) { - for (int i = 0; i < exclude_count; i++) - free(scanner->exclude_patterns[i]); - free(scanner->exclude_patterns); - queue_destroy(scanner->directories); - free(scanner); - return NULL; - } - for (int i = 0; i < include_count; i++) { - scanner->include_patterns[i] = str_dup(include_patterns[i]); - if (scanner->include_patterns[i] == NULL) { - for (int j = 0; j < i; j++) - free(scanner->include_patterns[j]); - free(scanner->include_patterns); - for (int j = 0; j < exclude_count; j++) - free(scanner->exclude_patterns[j]); - free(scanner->exclude_patterns); - queue_destroy(scanner->directories); - free(scanner); - return NULL; - } - } - } else { - scanner->include_patterns = NULL; - } + scanner->include_patterns = include_patterns; scanner->include_count = include_count; scanner->max_size = max_size; scanner->min_size = min_size; - scanner->follow_symlinks = follow_symlinks; - char* root_copy = str_dup(root_directory); - if (root_copy == NULL) { - for (int i = 0; i < scanner->include_count; i++) - free(scanner->include_patterns[i]); - free(scanner->include_patterns); - for (int i = 0; i < scanner->exclude_count; i++) - free(scanner->exclude_patterns[i]); - free(scanner->exclude_patterns); - queue_destroy(scanner->directories); - free(scanner); - return NULL; - } - queue_enqueue(scanner->directories, root_copy); + scanner->max_depth = max_depth; + scanner->current_depth = 0; + queue_enqueue(scanner->directories, dir_entry_create(root_directory, 0)); return scanner; } @@ -115,12 +67,6 @@ void directory_scanner_destroy(DirectoryScanner* scanner) { scanner->current_dir = NULL; } free(scanner->current_path); - for (int i = 0; i < scanner->exclude_count; i++) - free(scanner->exclude_patterns[i]); - free(scanner->exclude_patterns); - for (int i = 0; i < scanner->include_count; i++) - free(scanner->include_patterns[i]); - free(scanner->include_patterns); queue_destroy(scanner->directories); free(scanner); } @@ -134,7 +80,6 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) { return chunk; } -// Returns: 1 on success, 0 if no more directories in queue, -1 on opendir failure static int open_next_directory(DirectoryScanner* scanner) { if (scanner->current_dir) { closedir(scanner->current_dir); @@ -145,7 +90,10 @@ static int open_next_directory(DirectoryScanner* scanner) { if (queue_is_empty(scanner->directories)) return 0; - scanner->current_path = (char*)queue_dequeue(scanner->directories); + DirEntry* de = (DirEntry*)queue_dequeue(scanner->directories); + scanner->current_path = de->path; + scanner->current_depth = de->depth; + free(de); scanner->current_dir = opendir(scanner->current_path); if (scanner->current_dir == NULL) { perror("Could not open directory"); @@ -183,82 +131,24 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { char* cur_path = path_cat(scanner->current_path, entry->d_name); struct stat stats; - // Use lstat to detect symlinks - if (lstat(cur_path, &stats) != 0) { + if (stat(cur_path, &stats) != 0) { free(cur_path); continue; } - // If follow_symlinks is enabled and this is a symlink, resolve it - if (scanner->follow_symlinks && S_ISLNK(stats.st_mode)) { - struct stat target_stats; - if (stat(cur_path, &target_stats) != 0) { - // Broken symlink, skip - free(cur_path); - continue; - } - stats = target_stats; - } - if (S_ISDIR(stats.st_mode)) { - queue_enqueue(scanner->directories, (void*)cur_path); - } else if (S_ISLNK(stats.st_mode)) { - // Handle symlink (not following) - 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; - } - } - - File* file = file_create(cur_path); - if (file == NULL) { - free(cur_path); - continue; - } - file->type = FILE_TYPE_SYMLINK; - // Read link target - char link_buf[4096]; - ssize_t link_len = readlink(cur_path, link_buf, sizeof(link_buf) - 1); - if (link_len >= 0) { - link_buf[link_len] = '\0'; - file->link_target = str_dup(link_buf); - if (file->link_target == NULL) { - file_destroy(file); - free(cur_path); - continue; - } - } - file->data->size = 0; - if (scanner->use_metadata) - file->metadata = file_metadata_create(&stats); - array_list_add(chunk_data, file); - chunk_data_size += 1; // small size for symlinks - if (chunk_data_size > scanner->chunk_size) { - free(cur_path); - return chunk_data_to_chunk(chunk_data); + 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); + if (!queue_enqueue(scanner->directories, de)) + dir_entry_destroy(de); } free(cur_path); } else { + if (scanner->max_depth > 0 && scanner->current_depth + 1 > scanner->max_depth) { + 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)) { @@ -314,3 +204,256 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { array_list_delete(chunk_data); return NULL; } + +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; +} 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); + 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); + } + directory_scanner_destroy(ds); + free(wa->dirs[i]); + } + ParallelScanner* ps = wa->ps; + free(wa->dirs); + free(wa); + mtx_lock(&ps->result_mutex); + ps->completed++; + if (ps->completed >= ps->num_threads) { + ps->done = true; + cnd_signal(&ps->result_not_empty); + } + mtx_unlock(&ps->result_mutex); + 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) { + ParallelScanner* ps = calloc(1, sizeof(ParallelScanner)); + if (!ps) + return NULL; + ps->result_queue = queue_create(100, chunk_destroy); + if (!ps->result_queue) { + free(ps); + return NULL; + } + if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success || + cnd_init(&ps->result_not_empty) != thrd_success || + cnd_init(&ps->result_not_full) != thrd_success) { + queue_destroy(ps->result_queue); + free(ps); + return NULL; + } + + DIR* dir = opendir(root_directory); + if (!dir) { + perror("Could not open root directory for parallel scan"); + parallel_scanner_destroy(ps); + return NULL; + } + + ArrayList* root_files = array_list_create(file_destroy); + ArrayList* subdirs = array_list_create(free); + 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 st; + if (stat(cur_path, &st) != 0) { + free(cur_path); + continue; + } + if (S_ISDIR(st.st_mode)) { + array_list_add(subdirs, cur_path); + } 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) + continue; + file->data->size = st.st_size; + if (use_metadata) + file->metadata = file_metadata_create(&st); + array_list_add(root_files, file); + } + } + closedir(dir); + + unsigned long long cs = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE; + if (root_files->size > 0) { + ArrayList* batch = array_list_create(NULL); + unsigned long long batch_size = 0; + Chunk* first = NULL; + for (int i = 0; i < root_files->size; i++) { + File* f = (File*)root_files->items[i]; + array_list_add(batch, f); + batch_size += f->data->size; + if (batch_size >= cs || i == root_files->size - 1) { + void** items = array_list_to_array(batch); + Chunk* c = chunk_create((File**)items, batch->size); + free(items); + batch->item_destroyer = NULL; + array_list_delete(batch); + batch = NULL; + if (!first) { + first = c; + } else { + queue_enqueue_multithreaded(ps->result_queue, c, &ps->result_mutex, &ps->result_not_empty, + &ps->result_not_full); + } + if (i < root_files->size - 1) { + batch = array_list_create(NULL); + batch_size = 0; + } + } + } + if (batch) { + batch->item_destroyer = NULL; + 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; + if (n > subdirs->size) + n = subdirs->size > 0 ? subdirs->size : 1; + + if (subdirs->size > 0) { + ps->num_threads = n; + ps->threads = calloc(n, sizeof(thrd_t)); + if (!ps->threads) { + array_list_delete(subdirs); + parallel_scanner_destroy(ps); + return NULL; + } + int dirs_per_thread = subdirs->size / n; + int remainder = subdirs->size % n; + int start = 0; + for (int t = 0; t < n; t++) { + int count = dirs_per_thread + (t < remainder ? 1 : 0); + if (count == 0) + break; + ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg)); + if (!wa) + break; + wa->ps = ps; + wa->dirs = calloc(count, sizeof(char*)); + if (!wa->dirs) { + free(wa); + break; + } + for (int j = 0; j < count; j++) + wa->dirs[j] = str_dup((char*)subdirs->items[start + j]); + 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; + 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->num_threads = t; + break; + } + } + } + array_list_delete(subdirs); + return ps; +} + +Chunk* parallel_scanner_next(ParallelScanner* ps) { + if (ps->initial_chunk) { + Chunk* c = ps->initial_chunk; + ps->initial_chunk = NULL; + return c; + } + if (ps->num_threads == 0) { + ps->done = true; + return NULL; + } + Chunk* chunk = queue_dequeue_multithreaded( + ps->result_queue, &ps->result_mutex, &ps->result_not_empty, &ps->result_not_full, &ps->done); + return chunk; +} + +void parallel_scanner_destroy(ParallelScanner* ps) { + if (!ps) + return; + ps->done = true; + cnd_signal(&ps->result_not_empty); + for (int i = 0; i < ps->num_threads; i++) + thrd_join(ps->threads[i], NULL); + free(ps->threads); + if (ps->initial_chunk) + chunk_destroy(ps->initial_chunk); + queue_destroy(ps->result_queue); + mtx_destroy(&ps->result_mutex); + cnd_destroy(&ps->result_not_empty); + cnd_destroy(&ps->result_not_full); + free(ps); +} diff --git a/src/client/scanner.h b/src/client/scanner.h index c163551..1a8c4dd 100644 --- a/src/client/scanner.h +++ b/src/client/scanner.h @@ -5,6 +5,7 @@ #include "queue.h" #include #include +#include typedef struct { Queue* directories; @@ -18,21 +19,37 @@ typedef struct { int include_count; unsigned long long max_size; unsigned long long min_size; - bool follow_symlinks; + int max_depth; + int current_depth; } DirectoryScanner; -DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata, +typedef struct { + Queue* result_queue; + mtx_t result_mutex; + cnd_t result_not_empty; + cnd_t result_not_full; + int num_threads; + thrd_t* threads; + bool done; + int completed; + Chunk* initial_chunk; +} ParallelScanner; + +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); -DirectoryScanner* directory_scanner_create_full(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, bool follow_symlinks); + unsigned long long min_size, int max_depth); Chunk* directory_scanner_next(DirectoryScanner* scanner); void directory_scanner_destroy(DirectoryScanner* scanner); +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); +Chunk* parallel_scanner_next(ParallelScanner* scanner); +void parallel_scanner_destroy(ParallelScanner* scanner); + #endif diff --git a/src/server/server.c b/src/server/server.c index 141ecdc..2e045ca 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -11,55 +11,26 @@ #include "transport_tls.h" #include "unistd.h" #include "utils.h" -#include #include #include #include #include -// Check if a file path should be excluded based on config patterns -static bool is_excluded(const char* path, const Config* config) { - // Extract filename from path - char* path_dup = str_dup(path); - if (!path_dup) - return false; - /* basename(3) may return a pointer into path_dup or a static buffer; - * either way we free path_dup, not fname. */ - char* fname = basename(path_dup); - - // Check exclude patterns - for (int i = 0; i < config->exclude_count; i++) { - if (glob_match(config->exclude_patterns[i], fname)) { - free(path_dup); - return true; - } - } - - // Check include patterns (if any, file must match at least one) - if (config->include_count > 0) { - bool included = false; - for (int i = 0; i < config->include_count; i++) { - if (glob_match(config->include_patterns[i], fname)) { - included = true; - break; - } - } - if (!included) { - free(path_dup); - return true; - } - } - - free(path_dup); - return false; -} - 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) { + 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); @@ -67,8 +38,8 @@ int receive_files(Config* config, int fd) { goto next; if (file == NULL && !skipped) return -1; - if (config->save_to_disk && !is_excluded(file->path, config)) - file_save_to_disk(config->receive_root_directory, file); + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, file, NULL); file_destroy(file); } else if (status == STATUS_CHUNK) { Chunk* chunk = receive_chunk_data(fd, config); @@ -77,10 +48,38 @@ int receive_files(Config* config, int fd) { return -1; } for (int i = 0; i < chunk->element_count; i++) { - if (config->save_to_disk && !is_excluded(chunk->items[i]->path, config)) - file_save_to_disk(config->receive_root_directory, chunk->items[i]); + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, chunk->items[i], NULL); } chunk_destroy(chunk); + } else if (status == STATUS_CHECK_BATCH) { + int count; + if (!receive_int(fd, &count)) + 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; + } + char* full_path = path_cat(config->receive_root_directory, check_path); + struct stat st; + bool has_old = full_path && stat(full_path, &st) == 0; + bool match = has_old && (unsigned long long)st.st_size == check_size && + (long long)st.st_mtime == check_mtime; + if (match) + send_status(fd, STATUS_OK); + else + send_status(fd, STATUS_NEXT); + free(full_path); + free(check_path); + } + goto next; } else { File* file = file_receive(config, fd); if (file == NULL) { @@ -88,8 +87,8 @@ int receive_files(Config* config, int fd) { send_status(fd, STATUS_ERROR); return -1; } - if (config->save_to_disk && !is_excluded(file->path, config)) - file_save_to_disk(config->receive_root_directory, file); + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, file, NULL); file_destroy(file); } next: @@ -151,12 +150,13 @@ void handler(int file_descriptor) { } static Server* g_server = NULL; -static volatile sig_atomic_t g_server_cleanup_requested = 0; static void cleanup(int sig) { (void)sig; - server_request_shutdown(); - g_server_cleanup_requested = 1; + if (g_server) { + server_delete(&g_server); + } + _exit(0); } static void print_server_usage(void) { @@ -172,7 +172,6 @@ static void print_server_usage(void) { printf(" --ca TLS CA certificate file (PEM)\n"); printf(" -v, --verbose Enable debug logging\n"); printf(" --help Show this help\n"); - printf(" -V, --version Show version and exit\n"); } int main(int argc, char* argv[]) { @@ -187,9 +186,6 @@ int main(int argc, char* argv[]) { if (strcmp(argv[i], "--help") == 0) { print_server_usage(); return 0; - } else if (strcmp(argv[i], "-V") == 0 || strcmp(argv[i], "--version") == 0) { - printf("fastsync-server version %s\n", PROTOCOL_VERSION); - return 0; } else if (strcmp(argv[i], "--stdio") == 0) { io_set_fds(STDIN_FILENO, STDOUT_FILENO); handler(STDIN_FILENO); @@ -246,10 +242,5 @@ int main(int argc, char* argv[]) { } else { server_listen(g_server, handler); } - - /* Graceful shutdown: delete the server */ - if (g_server_cleanup_requested) - log_message(LOG_LEVEL_INFO, "Shutdown requested, cleaning up"); - server_delete(&g_server); return 0; } diff --git a/src/shared/compression.c b/src/shared/compression.c index 7c60686..97ae61f 100644 --- a/src/shared/compression.c +++ b/src/shared/compression.c @@ -1,26 +1,31 @@ #include "compression.h" #include "data.h" #include "log.h" -#include -#include +#include "stdlib.h" +#include "string.h" +#include #include "zstd.h" #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024) +static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv", + ".zip", ".gz", ".xz", ".zst", NULL}; + +bool compression_should_skip(const char* path) { + if (!path) + return false; + const char* dot = strrchr(path, '.'); + if (!dot) + return false; + for (int i = 0; SKIP_COMPRESSION_EXTENSIONS[i]; i++) { + if (strcasecmp(dot, SKIP_COMPRESSION_EXTENSIONS[i]) == 0) + return true; + } + return false; +} + Data* data_compress(Data* data_to_compress, int compression_level) { log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); - - /* Clamp compression level to valid zstd range [1, 22] */ - if (compression_level < 1) { - log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 1", - compression_level); - compression_level = 1; - } else if (compression_level > 22) { - log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 22", - compression_level); - compression_level = 22; - } - size_t dst_size = ZSTD_compressBound(data_to_compress->size); Data* compressed_data = data_create_empty(dst_size); if (compressed_data == NULL) @@ -79,16 +84,8 @@ Data* data_decompress(Data* compressed_data) { return NULL; } - size_t buf_size = INITIAL_DECOMPRESS_BUF_SIZE; - if (!ZSTD_isError(dst_size) && dst_size > 0) { - if (dst_size > SIZE_MAX) { - log_message(LOG_LEVEL_ERROR, - "Decompressed size %llu exceeds addressable memory, using fallback buffer", - dst_size); - } else { - buf_size = (size_t)dst_size; - } - } + size_t buf_size = + (!ZSTD_isError(dst_size) && dst_size > 0) ? (size_t)dst_size : INITIAL_DECOMPRESS_BUF_SIZE; Data* uncompressed_data = data_create_empty(buf_size); if (!uncompressed_data) { log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer"); diff --git a/src/shared/compression.h b/src/shared/compression.h index dcf30dd..d392f25 100644 --- a/src/shared/compression.h +++ b/src/shared/compression.h @@ -2,8 +2,10 @@ #define COMPRESSION_H #include "data.h" +#include Data* data_compress(Data* data_to_compress, int compression_level); Data* data_decompress(Data* compressed_data); +bool compression_should_skip(const char* path); #endif diff --git a/src/shared/config.c b/src/shared/config.c index e846c42..ae8936b 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -4,7 +4,6 @@ #include "protocol.h" #include "utils.h" #include -#include #include #include #include @@ -15,7 +14,7 @@ Config* config_create(char* version, char* send_directory, char* receive_directo bool use_sendfile, unsigned long long chunk_size) { Config* config = malloc(sizeof(Config)); - if (config == NULL) + if (!config) return NULL; config->version = version; config->send_directory = send_directory; @@ -49,10 +48,17 @@ Config* config_create(char* version, char* send_directory, char* receive_directo config->tls_cert = NULL; config->tls_key = NULL; config->tls_ca = NULL; - config->follow_symlinks = false; - config->partial = false; config->server_host = str_dup("127.0.0.1"); config->server_port = 8080; + config->timeout = 30; + config->contimeout = 10; + config->quiet = false; + config->backup = false; + config->backup_dir = NULL; + config->stats = false; + config->max_depth = 0; + config->log_file = NULL; + config->queue_size = 100; return config; } @@ -83,8 +89,6 @@ void config_parse_ssh_dest(Config* config) { } void config_delete(Config* config) { - if (config == NULL) - return; free(config->version); free(config->send_directory); free(config->receive_root_directory); @@ -99,6 +103,7 @@ void config_delete(Config* config) { free(config->tls_cert); free(config->tls_key); free(config->tls_ca); + free(config->backup_dir); free(config->server_host); free(config); } @@ -136,25 +141,9 @@ bool config_send(int file_descriptor, const Config* config) { 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->exclude_count)) + if (!send_int(file_descriptor, config->backup)) return false; - for (int i = 0; i < config->exclude_count; i++) { - if (!send_str(file_descriptor, config->exclude_patterns[i])) - return false; - } - if (!send_int(file_descriptor, config->include_count)) - return false; - for (int i = 0; i < config->include_count; i++) { - if (!send_str(file_descriptor, config->include_patterns[i])) - return false; - } - if (!send_n_data(file_descriptor, &config->max_size, sizeof(config->max_size))) - return false; - if (!send_n_data(file_descriptor, &config->min_size, sizeof(config->min_size))) - return false; - if (!send_int(file_descriptor, config->follow_symlinks)) - return false; - if (!send_int(file_descriptor, config->partial)) + if (!send_str(file_descriptor, config->backup_dir ? config->backup_dir : "")) return false; Status status; if (!receive_status(file_descriptor, &status)) @@ -251,82 +240,19 @@ Config* config_receive(int file_descriptor) { config->tls_cert = NULL; config->tls_key = NULL; config->tls_ca = NULL; - config->follow_symlinks = false; - config->partial = false; - -#define MAX_PATTERN_COUNT 10000 - - // Receive exclude patterns - int ec; - if (!receive_int(file_descriptor, &ec)) + config->timeout = 30; + config->contimeout = 10; + config->quiet = false; + config->stats = false; + config->max_depth = 0; + config->log_file = NULL; + config->queue_size = 100; + if (!receive_int(file_descriptor, &tmp)) goto error; - if (ec > MAX_PATTERN_COUNT) { - log_message(LOG_LEVEL_ERROR, "Exclude pattern count %d exceeds maximum %d", ec, - MAX_PATTERN_COUNT); + config->backup = tmp; + config->backup_dir = receive_str(file_descriptor); + if (config->backup_dir == NULL) goto error; - } - config->exclude_count = ec; - if (ec > 0) { - config->exclude_patterns = malloc((size_t)ec * sizeof(char*)); - if (!config->exclude_patterns) { - config->exclude_count = 0; - goto error; - } - for (int i = 0; i < ec; i++) { - config->exclude_patterns[i] = receive_str(file_descriptor); - if (!config->exclude_patterns[i]) { - for (int j = 0; j < i; j++) - free(config->exclude_patterns[j]); - free(config->exclude_patterns); - config->exclude_patterns = NULL; - config->exclude_count = 0; - goto error; - } - } - } - - // Receive include patterns - int ic; - if (!receive_int(file_descriptor, &ic)) - goto error; - if (ic > MAX_PATTERN_COUNT) { - log_message(LOG_LEVEL_ERROR, "Include pattern count %d exceeds maximum %d", ic, - MAX_PATTERN_COUNT); - goto error; - } - config->include_count = ic; - if (ic > 0) { - config->include_patterns = malloc((size_t)ic * sizeof(char*)); - if (!config->include_patterns) { - config->include_count = 0; - goto error; - } - for (int i = 0; i < ic; i++) { - config->include_patterns[i] = receive_str(file_descriptor); - if (!config->include_patterns[i]) { - for (int j = 0; j < i; j++) - free(config->include_patterns[j]); - free(config->include_patterns); - config->include_patterns = NULL; - config->include_count = 0; - goto error; - } - } - } - - if (!receive_n_data(file_descriptor, &config->max_size, sizeof(config->max_size))) - goto error; - if (!receive_n_data(file_descriptor, &config->min_size, sizeof(config->min_size))) - goto error; - int tmp_follow; - if (!receive_int(file_descriptor, &tmp_follow)) - goto error; - config->follow_symlinks = tmp_follow; - int tmp_partial; - if (!receive_int(file_descriptor, &tmp_partial)) - goto error; - config->partial = tmp_partial; - config->server_host = str_dup("127.0.0.1"); config->server_port = 8080; if (!send_status(file_descriptor, STATUS_OK)) @@ -337,17 +263,8 @@ error: free(config->version); free(config->send_directory); free(config->receive_root_directory); - for (int i = 0; i < config->exclude_count; i++) - free(config->exclude_patterns[i]); - free(config->exclude_patterns); - for (int i = 0; i < config->include_count; i++) - free(config->include_patterns[i]); - free(config->include_patterns); - free(config->tls_cert); - free(config->tls_key); - free(config->tls_ca); - free(config->ssh_destination); free(config->server_host); + free(config->backup_dir); free(config); return NULL; } diff --git a/src/shared/config.h b/src/shared/config.h index 3a0d114..fbb8b4f 100644 --- a/src/shared/config.h +++ b/src/shared/config.h @@ -3,6 +3,7 @@ #include #include +#include typedef enum { TRANSPORT_TCP, TRANSPORT_SSH } TransportType; @@ -41,11 +42,18 @@ typedef struct Config { char* tls_cert; char* tls_key; char* tls_ca; - bool follow_symlinks; - bool partial; + int timeout; + int contimeout; + bool quiet; + bool backup; + char* backup_dir; + bool stats; + int max_depth; + FILE* log_file; + int queue_size; } Config; -#define PROTOCOL_VERSION "2.0.0" +#define PROTOCOL_VERSION "1.3.0" #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024) Config* config_create(char* version, char* send_directory, char* receive_directory, diff --git a/src/shared/delta.c b/src/shared/delta.c index e9e1ea5..206ee78 100644 --- a/src/shared/delta.c +++ b/src/shared/delta.c @@ -108,7 +108,12 @@ DeltaSignature* delta_signature_deserialize(const Data* data) { return NULL; } - sig->blocks = malloc(sig->block_count * sizeof(DeltaBlockSig)); + uint64_t blocks_size = (uint64_t)sig->block_count * sizeof(DeltaBlockSig); + if (blocks_size > SIZE_MAX) { + free(sig); + return NULL; + } + sig->blocks = malloc((size_t)blocks_size); if (!sig->blocks) { free(sig); return NULL; diff --git a/src/shared/file.c b/src/shared/file.c index 2d9f922..a531e3d 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -3,7 +3,6 @@ #include #include #include -#include #include #include #include @@ -22,9 +21,6 @@ #include "protocol.h" #include "utils.h" -#define STREAM_THRESHOLD (64ULL * 1024 * 1024) /* 64 MB */ -#define STREAM_CHUNK_SIZE (1ULL * 1024 * 1024) /* 1 MB */ - File* file_create(const char* path) { File* file = (File*)malloc(sizeof(File)); if (file == NULL) { @@ -39,7 +35,7 @@ File* file_create(const char* path) { return NULL; } - memcpy(file->path, path, path_len + 1); + strcpy(file->path, path); file->data = data_create_reserve(0); if (file->data == NULL) { free(file->path); @@ -47,8 +43,7 @@ File* file_create(const char* path) { return NULL; } file->metadata = NULL; - file->type = FILE_TYPE_REGULAR; - file->link_target = NULL; + file->skip = false; return file; } @@ -62,8 +57,6 @@ void file_destroy(void* item) { file->metadata = NULL; free(file->path); file->path = NULL; - free(file->link_target); - file->link_target = NULL; free(file); } @@ -92,14 +85,6 @@ void file_metadata_destroy(void* metadata) { bool file_load_data(File* file) { if (file == NULL) return false; - // Symlinks have no data to load - if (file->type == FILE_TYPE_SYMLINK) - return true; - // For streaming files, just record the size, don't load into memory - if (file->data->size > STREAM_THRESHOLD) { - // Don't allocate; streaming will read directly from disk - return true; - } if (file->data->data == NULL) { file->data->data = malloc(file->data->size); if (file->data->data == NULL) { @@ -110,87 +95,16 @@ bool file_load_data(File* file) { size_t bytes_read = file_content_to_buffer(file); if (bytes_read != file->data->size) { log_message(LOG_LEVEL_ERROR, "Did not read expected amount of bytes from file"); - free(file->data->data); - file->data->data = NULL; - file->data->size = 0; return false; } return true; } -// Stream file content in chunks without loading entire file into RAM -static bool file_send_streaming(File* file, int file_descriptor) { - unsigned long long total_size = file->data->size; - // Send total size prefix (same wire format as send_data) - if (!send_n_data(file_descriptor, &total_size, sizeof(total_size))) - return false; - - FILE* fp = fopen(file->path, "rb"); - if (!fp) { - perror("Could not open file for streaming"); - return false; - } - - char* buf = malloc(STREAM_CHUNK_SIZE); - if (!buf) { - fclose(fp); - return false; - } - unsigned long long remaining = total_size; - while (remaining > 0) { - size_t to_read = (size_t)((remaining < STREAM_CHUNK_SIZE) ? remaining : STREAM_CHUNK_SIZE); - size_t nread = fread(buf, 1, to_read, fp); - if (nread != to_read) { - if (ferror(fp)) { - perror("Read error during streaming"); - } - send_status(file_descriptor, STATUS_ERROR); - free(buf); - fclose(fp); - return false; - } - if (!send_n_data(file_descriptor, buf, nread)) { - send_status(file_descriptor, STATUS_ERROR); - free(buf); - fclose(fp); - return false; - } - remaining -= (unsigned long long)nread; - } - free(buf); - fclose(fp); - return true; -} - bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, int compression_level, bool send_path) { - if (send_path && !send_str(file_descriptor, file->path)) - return false; - if (use_metadata && !metadata_send(file_descriptor, file->metadata)) - return false; - - // Send file type indicator so receiver can distinguish regular from symlink - int ft = (int)file->type; - if (!send_int(file_descriptor, ft)) - return false; - - if (file->type == FILE_TYPE_SYMLINK) { - // Send link target, then zero-length data - if (!send_str(file_descriptor, file->link_target ? file->link_target : "")) - return false; - Data empty = {NULL, 0}; - return send_data(file_descriptor, &empty); - } - const Data* data_to_send = file->data; Data* compressed_data = NULL; - - // Streaming mode: for large files without compression, stream from disk - if (file->data->size > STREAM_THRESHOLD && compression_level == 0) { - return file_send_streaming(file, file_descriptor); - } - - if (compression_level > 0) { + if (compression_level > 0 && !compression_should_skip(file->path)) { compressed_data = data_compress(file->data, compression_level); if (compressed_data == NULL) { log_message(LOG_LEVEL_ERROR, "Failed to compress file data"); @@ -198,6 +112,14 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, } data_to_send = compressed_data; } + if (send_path && !send_str(file_descriptor, file->path)) { + data_destroy(compressed_data); + return false; + } + if (use_metadata && !metadata_send(file_descriptor, file->metadata)) { + data_destroy(compressed_data); + return false; + } if (!send_data(file_descriptor, data_to_send)) { data_destroy(compressed_data); return false; @@ -206,38 +128,15 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, return true; } -bool file_save_to_disk(const char* root_directory, File* file) { - if (file->type == FILE_TYPE_SYMLINK && file->link_target) { - // Validate link_target — reject absolute paths or traversal - if (file->link_target[0] == '/' || strstr(file->link_target, "..") != NULL) { - log_message(LOG_LEVEL_ERROR, "Path traversal blocked in symlink target: %s", - file->link_target); - return false; - } - char* disk_path = path_cat((char*)root_directory, file->path); - if (disk_path == NULL) - return false; - if (strstr(disk_path, "..") != NULL) { - log_message(LOG_LEVEL_ERROR, "Path traversal blocked: %s", disk_path); - free(disk_path); - return false; - } - unlink(disk_path); - bool ok = (symlink(file->link_target, disk_path) == 0); - if (ok && file->metadata) - file_restore_metadata(disk_path, file->metadata); - free(disk_path); - return ok; +bool file_save_to_disk(const char* root_directory, File* file, const Config* config) { + (void)config; + if (has_path_traversal(file->path)) { + log_message(LOG_LEVEL_ERROR, "Path traversal detected in file path: %s", file->path); + return false; } - char* disk_path = path_cat((char*)root_directory, file->path); if (disk_path == NULL) return false; - if (strstr(disk_path, "..") != NULL) { - log_message(LOG_LEVEL_ERROR, "Path traversal blocked: %s", disk_path); - free(disk_path); - return false; - } bool ok = to_disk(disk_path, file->data->data, file->data->size); if (ok) file_restore_metadata(disk_path, file->metadata); @@ -263,60 +162,19 @@ static void* old_data_from_path(const char* full_path, unsigned long long old_si return data; } -/** - * Helper: receive data from wire, optionally decompress, and store in file. - * On success, returns the received Data* (caller owns it). On failure, returns NULL. - * If `file_data` is received via receive_data(fd), this function handles decompression - * when config->use_compression is set. - */ -static Data* receive_and_decompress(int fd, const Config* config) { - Data* file_data = receive_data(fd); - if (file_data == NULL) - return NULL; - if (config->use_compression) { - Data* uncompressed = data_decompress(file_data); - data_destroy(file_data); - if (uncompressed == NULL) - return NULL; - file_data = uncompressed; - } - return file_data; -} - -/** - * Helper: receive metadata from wire and assign to file. - * Returns true on success (metadata may be NULL if absent), false on I/O error. - */ -static bool receive_and_assign_metadata(int fd, const Config* config, File* file) { - if (!config->use_metadata) - return true; - int meta_ok = 1; - file->metadata = metadata_receive(fd, &meta_ok); - if (!meta_ok) { - file_destroy(file); - send_status(fd, STATUS_ERROR); - return false; - } - return true; -} - 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) { - send_status(fd, STATUS_ERROR); + if (!old_data) return NULL; - } DeltaSignature* sig = delta_signature_create(old_data, old_size, config->delta_block_size); if (!sig) { - send_status(fd, STATUS_ERROR); free(old_data); return NULL; } Data* sig_data = delta_signature_serialize(sig); if (!sig_data) { - send_status(fd, STATUS_ERROR); delta_signature_destroy(sig); free(old_data); return NULL; @@ -419,16 +277,34 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_ return NULL; } - if (!receive_and_assign_metadata(fd, config, file)) - return NULL; + if (config->use_metadata) { + int meta_ok = 1; + file->metadata = metadata_receive(fd, &meta_ok); + if (!meta_ok) { + file_destroy(file); + send_status(fd, STATUS_ERROR); + return NULL; + } + } - Data* file_data = receive_and_decompress(fd, config); + Data* file_data = receive_data(fd); if (file_data == NULL) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } + if (config->use_compression) { + Data* uncompressed = data_decompress(file_data); + data_destroy(file_data); + if (uncompressed == NULL) { + file_destroy(file); + send_status(fd, STATUS_ERROR); + return NULL; + } + file_data = uncompressed; + } + data_destroy(file->data); file->data = file_data; return file; @@ -456,30 +332,18 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { return NULL; } - char* full_path = path_cat(config->receive_root_directory, check_path); - if (full_path && strstr(full_path, "..") != NULL) { - log_message(LOG_LEVEL_ERROR, "Path traversal blocked: %s", full_path); - free(full_path); + if (has_path_traversal(check_path)) { + log_message(LOG_LEVEL_ERROR, "Path traversal detected: %s", check_path); free(check_path); send_status(fd, STATUS_ERROR); return NULL; } + + char* full_path = path_cat(config->receive_root_directory, check_path); struct stat st; bool has_old_file = (full_path && stat(full_path, &st) == 0); unsigned long long old_size = has_old_file ? (unsigned long long)st.st_size : 0; - // Check for partial file if enabled - if (config->partial && !has_old_file && full_path) { - char* partial_path = malloc(strlen(full_path) + 20); - if (partial_path) { - snprintf(partial_path, strlen(full_path) + 20, "%s.fastsync-partial", full_path); - has_old_file = (stat(partial_path, &st) == 0); - if (has_old_file) - old_size = (unsigned long long)st.st_size; - free(partial_path); - } - } - bool match = has_old_file && (unsigned long long)st.st_size == check_size && (long long)st.st_mtime == check_mtime; @@ -525,86 +389,102 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { return NULL; } - if (!receive_and_assign_metadata(fd, config, file)) - return NULL; - - int file_type; - if (!receive_int(fd, &file_type)) { - file_destroy(file); - send_status(fd, STATUS_ERROR); - return NULL; - } - file->type = (FileType)file_type; - - if (file->type == FILE_TYPE_SYMLINK) { - char* link_target = receive_str(fd); - if (link_target) { - file->link_target = link_target; + if (config->use_metadata) { + int meta_ok = 1; + file->metadata = metadata_receive(fd, &meta_ok); + if (!meta_ok) { + file_destroy(file); + send_status(fd, STATUS_ERROR); + return NULL; } - Data* empty_data = receive_data(fd); - if (empty_data) - data_destroy(empty_data); - return file; } - Data* file_data = receive_and_decompress(fd, config); + Data* file_data = receive_data(fd); if (file_data == NULL) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } + if (config->use_compression) { + Data* uncompressed = data_decompress(file_data); + data_destroy(file_data); + if (uncompressed == NULL) { + file_destroy(file); + send_status(fd, STATUS_ERROR); + return NULL; + } + file_data = uncompressed; + } + data_destroy(file->data); file->data = file_data; return file; } bool to_disk(const char* path, const void* data, unsigned long long data_size) { - // dirname() may modify its argument and may return a pointer to static storage. - // We must use a copy of the result to be safe. + 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); - char* directory = str_dup(dir_result); + directory = str_dup(dir_result); free(path_dup); if (!directory) return false; bool ok = true; - if (!mkdir_r(directory)) { + if (!mkdir_r(directory)) + goto done; + + size_t path_len = strlen(path); + tmp_path = malloc(path_len + 5); + if (!tmp_path) { ok = false; goto done; } - FILE* file_pointer = fopen(path, "wb"); + 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 File"); + perror("Could not open temporary file"); ok = false; goto done; } if (fwrite(data, 1, data_size, file_pointer) != data_size) { - perror("Failed to write all data to disk"); + 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; } bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level, bool send_path) { - // Handle symlinks - if (file->type == FILE_TYPE_SYMLINK) { - return file_send_single_calls(file, file_descriptor, use_metadata, compression_level, - send_path); - } - // sendfile is incompatible with compression (kernel zero-copy). // If compression is requested, fall back to the regular send path. + // NOTE: This is a safety net only — callers must ensure compression_level == 0 + // before calling file_send_sendfile. The fallback to file_send_single_calls + // preserves the send_path contract, but callers should not rely on it for + // correctness (the --sendfile flag is validated to be mutually exclusive with + // -c/--compress at the CLI layer). if (compression_level > 0) return file_send_single_calls(file, file_descriptor, use_metadata, compression_level, send_path); @@ -614,10 +494,6 @@ bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false; - int ft = (int)file->type; - if (!send_int(file_descriptor, ft)) - return false; - int fd = open(file->path, O_RDONLY); if (fd == -1) { perror("Could not open file for sendfile"); @@ -632,13 +508,8 @@ bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int off_t offset = 0; while ((unsigned long long)offset < file_size) { - size_t send_count = (size_t)(file_size - (unsigned long long)offset); - if ((unsigned long long)send_count != file_size - (unsigned long long)offset) - send_count = SIZE_MAX; - ssize_t sent = sendfile(file_descriptor, fd, &offset, send_count); + ssize_t sent = sendfile(file_descriptor, fd, &offset, file_size - offset); if (sent == -1) { - if (errno == EINTR) - continue; perror("sendfile failed"); close(fd); return false; @@ -665,30 +536,6 @@ File* file_receive(const Config* config, int file_descriptor) { return NULL; } } - - // Receive file type indicator - int file_type; - if (!receive_int(file_descriptor, &file_type)) { - file_destroy(file); - return NULL; - } - file->type = (FileType)file_type; - - if (file->type == FILE_TYPE_SYMLINK) { - char* link_target = receive_str(file_descriptor); - if (link_target == NULL) { - file_destroy(file); - return NULL; - } - file->link_target = link_target; - // Receive and discard zero-length data - Data* empty_data = receive_data(file_descriptor); - if (empty_data) - data_destroy(empty_data); - return file; - } - - // Regular file - receive data Data* file_data = receive_data(file_descriptor); if (file_data == NULL) { file_destroy(file); @@ -703,7 +550,6 @@ File* file_receive(const Config* config, int file_descriptor) { } file_data = file_data_uncompressed; } - data_destroy(file->data); file->data = file_data; return file; diff --git a/src/shared/file.h b/src/shared/file.h index f5193f9..e3e514b 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -20,8 +20,7 @@ typedef struct { char* path; Data* data; FileMetadata* metadata; - FileType type; - char* link_target; + bool skip; } File; File* file_create(const char* path); @@ -36,7 +35,7 @@ size_t file_content_to_buffer(File* file); FileMetadata* file_metadata_create(const struct stat* stats); void file_metadata_destroy(void* metadata); bool to_disk(const char* path, const void* data, unsigned long long data_size); -bool file_save_to_disk(const char* root_directory, File* file); +bool file_save_to_disk(const char* root_directory, File* file, const Config* config); File* receive_incremental_check(int fd, const Config* config, bool* skipped); int receive_manifest(int fd, const Config* config, int* next_status); diff --git a/src/shared/log.c b/src/shared/log.c index bb7590a..07a5d27 100644 --- a/src/shared/log.c +++ b/src/shared/log.c @@ -5,19 +5,21 @@ static const char* log_level_strings[] = {"DEBUG", "INFO", "WARN", "ERROR"}; static LogLevel current_log_level = LOG_LEVEL_WARNING; +static FILE* log_fp = NULL; void set_log_level(LogLevel level) { current_log_level = level; } +void log_set_file(FILE* fp) { + log_fp = fp; +} + void log_message(LogLevel log_level, const char* format, ...) { if (log_level < current_log_level) return; time_t now = time(NULL); - struct tm result_buf; - const struct tm* t = localtime_r(&now, &result_buf); - if (t == NULL) - return; + const struct tm* t = localtime(&now); fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1, t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]); @@ -27,4 +29,14 @@ void log_message(LogLevel log_level, const char* format, ...) { vfprintf(stderr, format, args); va_end(args); fprintf(stderr, "\n"); + + if (log_fp) { + fprintf(log_fp, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1, + t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]); + va_start(args, format); + vfprintf(log_fp, format, args); + va_end(args); + fprintf(log_fp, "\n"); + fflush(log_fp); + } } diff --git a/src/shared/log.h b/src/shared/log.h index 3c37ced..0aa622b 100644 --- a/src/shared/log.h +++ b/src/shared/log.h @@ -1,9 +1,12 @@ #ifndef LOG_H #define LOG_H +#include + typedef enum { LOG_LEVEL_DEBUG, LOG_LEVEL_INFO, LOG_LEVEL_WARNING, LOG_LEVEL_ERROR } LogLevel; void log_message(LogLevel log_level, const char* message, ...); void set_log_level(LogLevel level); +void log_set_file(FILE* fp); #endif diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index f6ce08f..9aaa599 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -105,7 +105,16 @@ int receive_thread(void* pipeline_context) { Status status; if (!receive_status(file_descriptor, &status)) return thrd_error; - while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) { + 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(file_descriptor, STATUS_KEEPALIVE); + goto next; + } + if (status == STATUS_ABORT) { + log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up"); + return thrd_error; + } if (status == STATUS_CHECK) { bool skipped; File* file = receive_incremental_check(file_descriptor, config, &skipped); @@ -117,6 +126,34 @@ int receive_thread(void* pipeline_context) { } } else if (status == STATUS_CHUNK) { receive_chunk_enqueue(file_descriptor, context); + } else if (status == STATUS_CHECK_BATCH) { + int count; + if (!receive_int(file_descriptor, &count)) + return thrd_error; + for (int i = 0; i < count; i++) { + char* check_path = receive_str(file_descriptor); + if (!check_path) + return thrd_error; + 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 thrd_error; + } + char* full_path = path_cat(config->receive_root_directory, check_path); + struct stat st; + bool has_old = full_path && stat(full_path, &st) == 0; + bool match = has_old && (unsigned long long)st.st_size == check_size && + (long long)st.st_mtime == check_mtime; + if (match) + send_status(file_descriptor, STATUS_OK); + else + send_status(file_descriptor, STATUS_NEXT); + free(full_path); + free(check_path); + } + goto next; } else { File* file = file_receive(config, file_descriptor); if (file) { @@ -126,6 +163,7 @@ int receive_thread(void* pipeline_context) { log_message(LOG_LEVEL_ERROR, "Failed to receive file"); } } + next: if (!receive_status(file_descriptor, &status)) return thrd_error; } @@ -156,7 +194,7 @@ int write_thread(void* pipeline_context) { return thrd_success; } if (save_to_disk) - file_save_to_disk(root_directory, file); + file_save_to_disk(root_directory, file, context->config); file_destroy(file); } } diff --git a/src/shared/protocol.c b/src/shared/protocol.c index 62f18cf..8bd7cfa 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -2,20 +2,27 @@ #include "log.h" #include #include +#include #include #include #include #include #include +#define MAX_DATA_SIZE (256ULL * 1024 * 1024) /* 256 MB max per message */ +#define RECEIVE_TIMEOUT_SEC 60 /* 60 second per-message timeout */ +#define MAX_CONNECTION_MEMORY (1024ULL * 1024 * 1024) /* 1 GB total per connection */ + static __thread int io_read_fd = -1; static __thread int io_write_fd = -1; -static SSL* io_ssl = NULL; +static SSL* io_ssl; static unsigned long long io_bwlimit = 0; static long long bw_tokens = 0; static struct timespec bw_last_refill = {0, 0}; +static __thread unsigned long long total_allocated_bytes = 0; + void io_set_fds(int read_fd, int write_fd) { io_read_fd = read_fd; io_write_fd = write_fd; @@ -46,12 +53,11 @@ static void bw_throttle(size_t bytes_written) { bw_tokens -= (long long)bytes_written; if (bw_tokens < 0) { - long long deficit_ns = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000000.0); - struct timespec sleep_time, remaining; - sleep_time.tv_sec = deficit_ns / 1000000000LL; - sleep_time.tv_nsec = deficit_ns % 1000000000LL; - while (nanosleep(&sleep_time, &remaining) < 0 && errno == EINTR) - sleep_time = remaining; + long long deficit_us = (long long)((double)(-bw_tokens) / io_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); } @@ -97,8 +103,21 @@ 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) { log_message(LOG_LEVEL_DEBUG, " Receiving n Data: %zu", data_size); int fd = io_fd(io_read_fd, file_descriptor); + + struct timespec deadline; + clock_gettime(CLOCK_MONOTONIC, &deadline); + deadline.tv_sec += RECEIVE_TIMEOUT_SEC; + size_t total_bytes_received = 0; while (total_bytes_received < data_size) { + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + if (now.tv_sec > deadline.tv_sec || + (now.tv_sec == deadline.tv_sec && now.tv_nsec > deadline.tv_nsec)) { + log_message(LOG_LEVEL_ERROR, "Receive timeout after %ds", RECEIVE_TIMEOUT_SEC); + return false; + } + ssize_t bytes_received; if (io_ssl) bytes_received = @@ -107,9 +126,9 @@ bool receive_n_data(int file_descriptor, void* data, size_t data_size) { bytes_received = read(fd, (char*)data + total_bytes_received, data_size - total_bytes_received); if (bytes_received <= 0) { - if (io_ssl && bytes_received < 0) { + if (io_ssl) { int ssl_err = SSL_get_error(io_ssl, (int)bytes_received); - if (ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE) + if (ssl_err == SSL_ERROR_WANT_WRITE || ssl_err == SSL_ERROR_WANT_READ) continue; } if (bytes_received == 0) @@ -142,16 +161,18 @@ static const char* status_to_string(Status status) { return "DELTA_SIGNATURE"; case STATUS_DELTA_DATA: return "DELTA_DATA"; + case STATUS_KEEPALIVE: + return "KEEPALIVE"; + case STATUS_ABORT: + return "ABORT"; + case STATUS_CHECK_BATCH: + return "CHECK_BATCH"; default: return "UNKNOWN"; } } bool send_str(int file_descriptor, const char* data) { - if (data == NULL) { - log_message(LOG_LEVEL_ERROR, "send_str called with NULL data"); - return false; - } size_t size = strlen(data); if (!send_n_data(file_descriptor, &size, sizeof(size_t))) return false; @@ -165,9 +186,9 @@ char* receive_str(int file_descriptor) { size_t size; if (!receive_n_data(file_descriptor, &size, sizeof(size_t))) return NULL; - if (size > MAX_STRING_SIZE) { - log_message(LOG_LEVEL_ERROR, "receive_str: size %zu exceeds maximum %zu", size, - (size_t)MAX_STRING_SIZE); + if (size > MAX_DATA_SIZE) { + log_message(LOG_LEVEL_ERROR, "String size %zu exceeds maximum %llu", size, + (unsigned long long)MAX_DATA_SIZE); return NULL; } char* data = (char*)malloc(size + 1); @@ -188,18 +209,23 @@ bool send_data(int file_descriptor, const Data* data) { return false; if (!send_n_data(file_descriptor, data->data, data_size)) return false; - log_message(LOG_LEVEL_DEBUG, "Send %llu data", data_size); + log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size); return true; } -#define MAX_DATA_SIZE (1024ULL * 1024 * 1024) - Data* receive_data(int file_descriptor) { unsigned long long size = 0; if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long))) return NULL; - if ((size_t)size != size || size > MAX_DATA_SIZE) { - log_message(LOG_LEVEL_ERROR, "receive_data size %llu exceeds limits", size); + if (size > MAX_DATA_SIZE) { + log_message(LOG_LEVEL_ERROR, "Data size %llu exceeds maximum %llu", size, + (unsigned long long)MAX_DATA_SIZE); + return NULL; + } + if (total_allocated_bytes + size > MAX_CONNECTION_MEMORY) { + 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); @@ -209,7 +235,8 @@ Data* receive_data(int file_descriptor) { free(data); return NULL; } - log_message(LOG_LEVEL_DEBUG, "Received %llu data", size); + total_allocated_bytes += size; + log_message(LOG_LEVEL_DEBUG, "Received %lld data", size); return data_create(data, (size_t)size); } diff --git a/src/shared/protocol.h b/src/shared/protocol.h index bc57d6b..405832e 100644 --- a/src/shared/protocol.h +++ b/src/shared/protocol.h @@ -20,7 +20,10 @@ enum NET_STATUS { STATUS_MANIFEST, STATUS_CHECK, STATUS_DELTA_SIGNATURE, - STATUS_DELTA_DATA + STATUS_DELTA_DATA, + STATUS_KEEPALIVE, + STATUS_ABORT, + STATUS_CHECK_BATCH }; void io_set_fds(int read_fd, int write_fd); diff --git a/src/shared/transport_ssh.c b/src/shared/transport_ssh.c index 2fa5c49..7b6dff8 100644 --- a/src/shared/transport_ssh.c +++ b/src/shared/transport_ssh.c @@ -118,19 +118,20 @@ Client* client_connect_ssh(const char* destination, int port, const char* server if (sv[1] > 1) close(sv[1]); - char ssh_user[512]; - int needed; + size_t ssh_user_len; if (r.user && r.user[0] != '\0') - needed = snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host); + ssh_user_len = strlen(r.user) + 1 + strlen(r.host) + 1; else - needed = snprintf(ssh_user, sizeof(ssh_user), "%s", r.host); - if ((size_t)needed >= sizeof(ssh_user)) - fprintf(stderr, "Warning: ssh_user string truncated\n"); - - size_t ssh_argv_max = 32; - char** ssh_argv = calloc(ssh_argv_max, sizeof(char*)); - if (ssh_argv == NULL) + ssh_user_len = strlen(r.host) + 1; + char* ssh_user = malloc(ssh_user_len); + if (!ssh_user) _exit(1); + if (r.user && r.user[0] != '\0') + snprintf(ssh_user, ssh_user_len, "%s@%s", r.user, r.host); + else + snprintf(ssh_user, ssh_user_len, "%s", r.host); + + char* ssh_argv[16]; int ac = 0; char port_str[16]; ssh_argv[ac++] = "ssh"; @@ -141,22 +142,15 @@ Client* client_connect_ssh(const char* destination, int port, const char* server ssh_argv[ac++] = "-o"; ssh_argv[ac++] = "ControlPath=~/.cache/fastsync-%r@%h:%p"; if (port > 0 && port != 22) { - if ((size_t)ac + 2 >= ssh_argv_max) { - _exit(1); - } ssh_argv[ac++] = "-p"; snprintf(port_str, sizeof(port_str), "%d", port); ssh_argv[ac++] = port_str; } - if ((size_t)ac + 3 >= ssh_argv_max) { - _exit(1); - } ssh_argv[ac++] = ssh_user; ssh_argv[ac++] = (char*)(server_path ? server_path : "fastsync-server"); ssh_argv[ac++] = "--stdio"; ssh_argv[ac] = NULL; execvp("ssh", ssh_argv); - free(ssh_argv); perror("exec of ssh failed"); ssize_t wret = write(exec_pipe[1], "x", 1); (void)wret; @@ -188,7 +182,7 @@ Client* client_connect_ssh(const char* destination, int port, const char* server return NULL; } client->file_descriptor = sv[0]; - client->address.ss_family = AF_UNIX; + client->address.sin_family = AF_UNIX; client->address_length = 0; client->ssh_child_pid = pid; client->ssl = NULL; diff --git a/src/shared/transport_tcp.c b/src/shared/transport_tcp.c index 4f2e8e7..6aec063 100644 --- a/src/shared/transport_tcp.c +++ b/src/shared/transport_tcp.c @@ -3,36 +3,25 @@ #include "protocol.h" #include #include -#include #include #include #include #include #include #include -#include #include #include -bool set_socket_timeouts(int fd) { - struct timeval tv; - tv.tv_sec = 30; - tv.tv_usec = 0; +static volatile sig_atomic_t g_active_connections = 0; - int keepalive = 1; - if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof(keepalive)) < 0) { - perror("Could not set SO_KEEPALIVE"); - return false; +static void sigchld_handler(int sig) { + (void)sig; + int saved_errno = errno; + while (waitpid(-1, NULL, WNOHANG) > 0) { + if (g_active_connections > 0) + g_active_connections--; } - if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) { - perror("Could not set SO_RCVTIMEO"); - return false; - } - if (setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) < 0) { - perror("Could not set SO_SNDTIMEO"); - return false; - } - return true; + errno = saved_errno; } Server* server_create(int port) { @@ -41,29 +30,14 @@ Server* server_create(int port) { perror("Could not allocate space for Server"); return NULL; } - memset(&server->address, 0, sizeof(server->address)); - // Try IPv6 first, fall back to IPv4 - int fd = socket(AF_INET6, SOCK_STREAM, 0); - sa_family_t domain = AF_INET6; - if (fd < 0) { - fd = socket(AF_INET, SOCK_STREAM, 0); - domain = AF_INET; - } - if (fd < 0) { + int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); + if (file_descriptor < 0) { perror("Could not create Socket!"); free(server); return NULL; } - - if (!set_socket_timeouts(fd)) { - close(fd); - free(server); - return NULL; - } - - server->file_descriptor = fd; - server->ssl_ctx = NULL; + server->file_descriptor = file_descriptor; int opt = 1; if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) { perror("Error setting a socket option!"); @@ -72,60 +46,20 @@ Server* server_create(int port) { return NULL; } - // Use the domain from the socket we actually created - struct sockaddr_storage* addr = &server->address; - struct sockaddr_in* addr4 = (struct sockaddr_in*)addr; - struct sockaddr_in6* addr6 = (struct sockaddr_in6*)addr; - - if (domain == AF_INET6) { - addr6->sin6_family = AF_INET6; - addr6->sin6_addr = in6addr_any; - addr6->sin6_port = htons(port); - addr->ss_family = AF_INET6; - server->address_length = sizeof(struct sockaddr_in6); - } else { - addr4->sin_family = AF_INET; - addr4->sin_addr.s_addr = INADDR_ANY; - addr4->sin_port = htons(port); - addr->ss_family = AF_INET; - server->address_length = sizeof(struct sockaddr_in); - } + server->address.sin_family = AF_INET; + server->address.sin_addr.s_addr = INADDR_ANY; + server->address.sin_port = htons(port); + server->address_length = sizeof(server->address); + server->ssl_ctx = NULL; + server->max_connections = 100; + server->active_connections = 0; if (bind(server->file_descriptor, (struct sockaddr*)&server->address, server->address_length) < 0) { - // If IPv6 bind failed (maybe no IPv6), try IPv4 - if (domain == AF_INET6) { - close(fd); - fd = socket(AF_INET, SOCK_STREAM, 0); - if (fd < 0) { - perror("Could not create IPv4 Socket!"); - free(server); - return NULL; - } - if (!set_socket_timeouts(fd)) { - close(fd); - free(server); - return NULL; - } - server->file_descriptor = fd; - setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); - memset(addr, 0, sizeof(*addr)); - addr4->sin_family = AF_INET; - addr4->sin_addr.s_addr = INADDR_ANY; - addr4->sin_port = htons(port); - server->address_length = sizeof(struct sockaddr_in); - if (bind(server->file_descriptor, (struct sockaddr*)addr, server->address_length) < 0) { - perror("Could not bind server"); - close(server->file_descriptor); - free(server); - return NULL; - } - } else { - perror("Could not bind server"); - close(server->file_descriptor); - free(server); - return NULL; - } + perror("Could not bind server"); + close(server->file_descriptor); + free(server); + return NULL; } return server; @@ -143,36 +77,27 @@ void server_delete(Server** server) { *server = NULL; } -/* Flag set by server_request_shutdown() to request graceful shutdown - of the accept loop. Accessed only from transport_tcp.c so it won't - cause linker errors when this file is compiled into client/test targets. */ -static volatile sig_atomic_t g_tcp_cleanup_requested = 0; - -void server_request_shutdown(void) { - g_tcp_cleanup_requested = 1; -} - static void accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx, const char* log_fmt) { if (listen(server->file_descriptor, SOMAXCONN) < 0) { perror("Could not listen on port!"); return; } - signal(SIGCHLD, SIG_IGN); - while (!g_tcp_cleanup_requested) { - struct sockaddr_storage client_addr; + signal(SIGCHLD, sigchld_handler); + while (1) { + struct sockaddr_in client_addr; socklen_t client_len = sizeof(client_addr); int fd = accept(server->file_descriptor, (struct sockaddr*)&client_addr, &client_len); if (fd < 0) { - if (errno == EINTR) { - if (g_tcp_cleanup_requested) - break; - continue; - } perror("Could not accept the connection"); continue; } - set_socket_timeouts(fd); + if ((unsigned int)g_active_connections >= server->max_connections) { + log_message(LOG_LEVEL_WARNING, "Max connections (%u) reached, rejecting", + server->max_connections); + close(fd); + continue; + } log_message(LOG_LEVEL_INFO, "%s", log_fmt); pid_t pid = fork(); if (pid == 0) { @@ -180,6 +105,8 @@ static void accept_loop(Server* server, void (*child_fn)(int, void*), void* chil child_fn(fd, child_ctx); close(fd); _exit(0); + } else if (pid > 0) { + g_active_connections++; } close(fd); } @@ -194,11 +121,7 @@ static void plain_child_fn(int fd, void* ctx) { } bool server_listen(Server* server, void (*handler)(int file_descriptor)) { - struct sockaddr_in* addr4 = (struct sockaddr_in*)&server->address; - int port = (server->address.ss_family == AF_INET6) - ? ntohs(((struct sockaddr_in6*)&server->address)->sin6_port) - : ntohs(addr4->sin_port); - log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d", port); + log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d", ntohs(server->address.sin_port)); struct plain_ctx ctx = {handler}; accept_loop(server, plain_child_fn, &ctx, "Received Connection"); return true; @@ -206,23 +129,43 @@ bool server_listen(Server* server, void (*handler)(int file_descriptor)) { void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx, const char* log_fmt) { - struct sockaddr_in* addr4 = (struct sockaddr_in*)&server->address; - int port = (server->address.ss_family == AF_INET6) - ? ntohs(((struct sockaddr_in6*)&server->address)->sin6_port) - : ntohs(addr4->sin_port); - log_message(LOG_LEVEL_INFO, "Start TLS Listening on Port: %d", port); + log_message(LOG_LEVEL_INFO, "Start TLS Listening on Port: %d", ntohs(server->address.sin_port)); accept_loop(server, child_fn, child_ctx, log_fmt); } +static int g_timeout_sec = 30; +static int g_contimeout_sec = 10; + +void tcp_set_timeouts(int timeout_sec, int contimeout_sec) { + if (timeout_sec > 0) + g_timeout_sec = timeout_sec; + if (contimeout_sec > 0) + g_contimeout_sec = contimeout_sec; +} + +static void tcp_apply_socket_timeout(int fd) { + struct timeval tv; + tv.tv_sec = g_timeout_sec; + tv.tv_usec = 0; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); +} + Client* client_create() { - Client* client = (Client*)malloc(sizeof(Client)); - if (client == NULL) { + int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); + if (file_descriptor < 0) { + perror("Could not create Socket!"); return NULL; } - memset(&client->address, 0, sizeof(client->address)); - client->address.ss_family = AF_UNSPEC; + + Client* client = (Client*)malloc(sizeof(Client)); + if (client == NULL) { + close(file_descriptor); + return NULL; + } + client->file_descriptor = file_descriptor; + client->address.sin_family = AF_INET; client->address_length = sizeof(client->address); - client->file_descriptor = -1; client->ssh_child_pid = -1; client->ssl = NULL; client->ssl_ctx = NULL; @@ -230,55 +173,28 @@ Client* client_create() { } bool client_connect(Client* client, char* host, int port) { - struct addrinfo hints, *res, *rp; - memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_UNSPEC; - hints.ai_socktype = SOCK_STREAM; + client->address.sin_port = htons(port); + client->address.sin_family = AF_INET; + client->address_length = sizeof(client->address); - char port_str[16]; - snprintf(port_str, sizeof(port_str), "%d", port); - - int gai_err = getaddrinfo(host, port_str, &hints, &res); - if (gai_err != 0) { - fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(gai_err)); + if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) { + perror("Could not convert host address!"); return false; } - // Try IPv6 first, then IPv4 - int fd = -1; - for (rp = res; rp != NULL; rp = rp->ai_next) { - fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); - if (fd < 0) - continue; - if (!set_socket_timeouts(fd)) { - close(fd); - fd = -1; - continue; - } - if (connect(fd, rp->ai_addr, rp->ai_addrlen) == 0) - break; - close(fd); - fd = -1; - } + struct timeval ct; + ct.tv_sec = g_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)); - if (fd < 0) { + if (connect(client->file_descriptor, (struct sockaddr*)&client->address, client->address_length) < + 0) { perror("Could not connect to Server!"); - freeaddrinfo(res); return false; } - // Save the connected address - socklen_t addr_len = rp->ai_addrlen; - if (addr_len > sizeof(client->address)) - addr_len = sizeof(client->address); - memcpy(&client->address, rp->ai_addr, addr_len); - client->address_length = addr_len; - freeaddrinfo(res); - - // Close old fd if any and set new one - if (client->file_descriptor >= 0) - close(client->file_descriptor); - client->file_descriptor = fd; + tcp_apply_socket_timeout(client->file_descriptor); return true; } @@ -289,10 +205,7 @@ void client_disconnect(Client* client) { client->ssl = NULL; io_set_ssl(NULL); } - if (client->file_descriptor >= 0) { - close(client->file_descriptor); - client->file_descriptor = -1; - } + close(client->file_descriptor); if (client->ssh_child_pid > 0) { int status; waitpid(client->ssh_child_pid, &status, 0); @@ -303,7 +216,6 @@ void client_disconnect(Client* client) { void client_delete(Client* client) { if (client == NULL) return; - client_disconnect(client); if (client->ssl_ctx) { SSL_CTX_free(client->ssl_ctx); client->ssl_ctx = NULL; diff --git a/src/shared/transport_tcp.h b/src/shared/transport_tcp.h index 8207ed6..71b03a2 100644 --- a/src/shared/transport_tcp.h +++ b/src/shared/transport_tcp.h @@ -1,21 +1,21 @@ #ifndef TRANSPORT_TCP_H #define TRANSPORT_TCP_H -#include #include #include -#include #include typedef struct Server { - struct sockaddr_storage address; + struct sockaddr_in address; unsigned int address_length; int file_descriptor; void* ssl_ctx; + unsigned int max_connections; + volatile unsigned int active_connections; } Server; typedef struct Client { - struct sockaddr_storage address; + struct sockaddr_in address; unsigned int address_length; int file_descriptor; pid_t ssh_child_pid; @@ -28,11 +28,10 @@ bool server_listen(Server* server, void (*handler)(int file_descriptor)); void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx, const char* log_fmt); void server_delete(Server** server); -void server_request_shutdown(void); Client* client_create(); bool client_connect(Client* client, char* host, int port); void client_disconnect(Client* client); void client_delete(Client* client); -bool set_socket_timeouts(int fd); +void tcp_set_timeouts(int timeout_sec, int contimeout_sec); #endif diff --git a/src/shared/transport_tls.c b/src/shared/transport_tls.c index 5b2f594..7959ee7 100644 --- a/src/shared/transport_tls.c +++ b/src/shared/transport_tls.c @@ -5,8 +5,6 @@ #include #include #include -#include -#include #include #include #include @@ -75,10 +73,6 @@ static SSL_CTX* create_ssl_ctx(bool is_server, const char* cert, const char* key SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER, NULL); SSL_CTX_set_verify_depth(ctx, 4); } else { - if (!is_server) { - log_message(LOG_LEVEL_WARNING, - "No CA path provided — TLS server certificate will not be verified"); - } SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL); } @@ -92,7 +86,6 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server) { return NULL; } SSL_set_fd(ssl, fd); - int ret; if (is_server) ret = SSL_accept(ssl); @@ -105,17 +98,6 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server) { SSL_free(ssl); return NULL; } - - // In client mode, check verification result if peer verification was requested - if (!is_server) { - long verify_result = SSL_get_verify_result(ssl); - if (verify_result != X509_V_OK) { - log_message(LOG_LEVEL_ERROR, "TLS certificate verification failed: %ld", verify_result); - SSL_free(ssl); - return NULL; - } - } - return ssl; } @@ -153,9 +135,16 @@ 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) { - // Use the common TCP connection logic (with IPv6 support) - if (!client_connect(client, host, port)) + client->address.sin_port = htons(port); + if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) { + perror("Could not convert host address!"); return false; + } + if (connect(client->file_descriptor, (struct sockaddr*)&client->address, client->address_length) < + 0) { + perror("Could not connect to Server!"); + return false; + } SSL_CTX* ctx = create_ssl_ctx(false, cert_path, key_path, ca_path); if (!ctx) @@ -168,14 +157,6 @@ bool client_connect_tls(Client* client, char* host, int port, const char* cert_p client->ssl_ctx = NULL; return false; } - - // Set SNI and enable hostname verification - SSL_set_tlsext_host_name(ssl, host); - X509_VERIFY_PARAM* param = SSL_get0_param(ssl); - if (param) { - X509_VERIFY_PARAM_set1_host(param, host, 0); - } - client->ssl = ssl; io_set_ssl(ssl); return true; diff --git a/src/shared/utils.c b/src/shared/utils.c index a5b67cd..8adc37b 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -10,23 +10,19 @@ #include bool mkdir_r(const char* path) { - size_t path_len = strlen(path); - char* path_duplicate = malloc(path_len + 1); + char* path_duplicate = malloc(strlen(path) + 1); if (!path_duplicate) return false; - memcpy(path_duplicate, path, path_len + 1); - /* Buffer for building subpaths: path_len + 1 for leading '/' + 1 for null */ - size_t buf_size = path_len + 2; - char* path_current = (char*)malloc(buf_size); + strcpy(path_duplicate, path); + char* path_current = (char*)malloc((strlen(path) + 2) * sizeof(char)); if (!path_current) { free(path_duplicate); return false; } - size_t pos = 0; + char* path_current_position = path_current; if (path[0] == '/') { - path_current[0] = '/'; - path_current[1] = '\0'; - pos = 1; + strcpy(path_current, "/"); + path_current_position += 1; } else { path_current[0] = '\0'; } @@ -35,16 +31,10 @@ bool mkdir_r(const char* path) { const char* part = strtok_r(path_duplicate, delimiter, &saveptr); bool ok = true; while (part != NULL) { - size_t part_len = strlen(part); - if (pos + part_len + 1 >= buf_size) { - ok = false; - break; - } - memcpy(path_current + pos, part, part_len); - pos += part_len; - path_current[pos] = '/'; - pos++; - path_current[pos] = '\0'; + strcpy(path_current_position, part); + path_current_position += strlen(part) * sizeof(char); + strcpy(path_current_position, "/"); + path_current_position += sizeof(char); struct stat st; if (stat(path_current, &st) != 0) { if (mkdir(path_current, 0755) != 0) { @@ -59,24 +49,22 @@ bool mkdir_r(const char* path) { free(path_current); return ok; } + char* str_dup(const char* string) { if (string == NULL) return NULL; char* new_string = (char*)malloc(strlen(string) + 1); - memcpy(new_string, string, strlen(string) + 1); + strcpy(new_string, string); return new_string; } bool glob_match(const char* pattern, const char* str) { while (*pattern) { if (*pattern == '*') { - /* Check for double-star (globstar) pattern */ if (*(pattern + 1) == '*') { pattern += 2; - /* Trailing double-star matches everything */ if (*pattern == '\0') return true; - /* double-star slash: match at any depth */ if (*pattern == '/') pattern++; while (*str) { @@ -86,7 +74,6 @@ bool glob_match(const char* pattern, const char* str) { } return glob_match(pattern, str); } - /* Single * — does not cross / boundaries */ pattern++; while (*str && *str != '/') { if (glob_match(pattern, str)) @@ -101,9 +88,7 @@ bool glob_match(const char* pattern, const char* str) { str++; } else { if (*pattern != *str) { - /* If pattern has a '/' followed by '**', allow zero path components */ if (*pattern == '/' && *(pattern + 1) == '*' && *(pattern + 2) == '*') { - /* Skip over slash-double-star and try to match rest against current str */ const char* rest = pattern + 3; if (*rest == '/') rest++; @@ -184,17 +169,37 @@ void delete_extras(const char* dest_root, ArrayList* manifest) { delete_extras_walk(dest_root, "", manifest); } +bool has_path_traversal(const char* path) { + if (!path) + return false; + char* dup = str_dup(path); + if (!dup) + return false; + char* saveptr; + const char* part = strtok_r(dup, "/", &saveptr); + while (part) { + if (strcmp(part, "..") == 0) { + free(dup); + return true; + } + part = strtok_r(NULL, "/", &saveptr); + } + free(dup); + return false; +} + char* path_cat(const char* path1, const char* path2) { if (path1 == NULL || *path1 == '\0') return str_dup(path2); if (path2 == NULL || *path2 == '\0') return str_dup(path1); - int path1_len = strlen(path1); - int path2_len = strlen(path2); + size_t path1_len = strlen(path1); + size_t path2_len = strlen(path2); + size_t offset = 0; if (path1[path1_len - 1] == '/') path1_len -= 1; if (path2[0] == '/') { - path2++; + offset = 1; path2_len -= 1; } char* new_path = malloc(path1_len + path2_len + 2); @@ -202,7 +207,7 @@ char* path_cat(const char* path1, const char* path2) { return NULL; memcpy(new_path, path1, path1_len); new_path[path1_len] = '/'; - memcpy(new_path + path1_len + 1, path2, path2_len); + memcpy(new_path + path1_len + 1, path2 + offset, path2_len); new_path[path1_len + path2_len + 1] = '\0'; return new_path; } diff --git a/src/shared/utils.h b/src/shared/utils.h index 1cc8a6c..1d1d085 100644 --- a/src/shared/utils.h +++ b/src/shared/utils.h @@ -9,5 +9,6 @@ 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 has_path_traversal(const char* path); #endif diff --git a/tests/runner.c b/tests/runner.c index 6757b8d..402af49 100644 --- a/tests/runner.c +++ b/tests/runner.c @@ -47,12 +47,12 @@ int main() { RUN_TEST(test_file_sendfile); RUN_TEST(test_multiprocessing); RUN_TEST(test_log); - RUN_TEST(test_transport_tcp); - RUN_TEST(test_transport_ssh); - RUN_TEST(test_transport_tls); RUN_TEST(test_robustness); RUN_TEST(test_stress); RUN_TEST(test_property); + RUN_TEST(test_transport_tcp); + RUN_TEST(test_transport_ssh); + RUN_TEST(test_transport_tls); printf("\n\033[1;36m=== TEST SUMMARY ===\033[0m\n"); printf("Total Tests Run: %d\n", tests_run); diff --git a/tests/test_file.c b/tests/test_file.c index ad4b2a2..25862a5 100644 --- a/tests/test_file.c +++ b/tests/test_file.c @@ -69,7 +69,7 @@ static void test_file_save_to_disk() { memcpy(f->data->data, content, strlen(content)); f->data->size = strlen(content); - EXPECT_TRUE(file_save_to_disk("test_save_tmp", f)); + EXPECT_TRUE(file_save_to_disk("test_save_tmp", f, NULL)); struct stat st; EXPECT_EQ_INT(stat("test_save_tmp/saved_file.txt", &st), 0); @@ -217,10 +217,6 @@ static void test_file_send_no_path() { pid_t pid = fork(); if (pid == 0) { close(p[1]); - // Read file type indicator - int file_type; - EXPECT_TRUE(receive_int(p[0], &file_type)); - EXPECT_EQ_INT(file_type, (int)FILE_TYPE_REGULAR); Data* received = receive_data(p[0]); close(p[0]); diff --git a/tests/test_file_sendfile.c b/tests/test_file_sendfile.c index 6bb4dcd..86119e4 100644 --- a/tests/test_file_sendfile.c +++ b/tests/test_file_sendfile.c @@ -224,10 +224,6 @@ static void test_sendfile_no_path() { pid_t pid = fork(); if (pid == 0) { close(p[1]); - /* When send_path is false, the sender still sends file_type + data */ - int file_type; - EXPECT_TRUE(receive_int(p[0], &file_type)); - EXPECT_EQ_INT(file_type, (int)FILE_TYPE_REGULAR); Data* received = receive_data(p[0]); close(p[0]); diff --git a/tests/test_protocol.c b/tests/test_protocol.c index f60d45f..af1dc00 100644 --- a/tests/test_protocol.c +++ b/tests/test_protocol.c @@ -46,7 +46,6 @@ static void test_send_receive_str() { EXPECT_TRUE(send_str(0, "")); - /* cppcheck-suppress constVariablePointer */ char* received = receive_str(0); EXPECT_NOT_NULL(received); EXPECT_EQ_STR(received, ""); @@ -64,7 +63,6 @@ static void test_send_receive_str_normal() { EXPECT_TRUE(send_str(0, "Hello, Protocol!")); - /* cppcheck-suppress constVariablePointer */ char* received = receive_str(0); EXPECT_NOT_NULL(received); EXPECT_EQ_STR(received, "Hello, Protocol!"); @@ -80,7 +78,6 @@ static void test_send_receive_data() { io_set_fds(p[0], p[1]); io_set_bwlimit(0); - /* cppcheck-suppress constVariablePointer */ unsigned char bin[] = {0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0xFF}; void* buf = malloc(sizeof(bin)); EXPECT_NOT_NULL(buf); @@ -131,9 +128,9 @@ static void test_send_receive_status() { io_set_fds(p[0], p[1]); io_set_bwlimit(0); - /* cppcheck-suppress constVariablePointer */ - Status statuses[] = {STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, - STATUS_CHUNK, STATUS_CHECK, STATUS_DELTA_SIGNATURE, STATUS_DELTA_DATA}; + Status statuses[] = {STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, + STATUS_CHUNK, STATUS_CHECK, STATUS_DELTA_SIGNATURE, STATUS_DELTA_DATA, + STATUS_KEEPALIVE, STATUS_ABORT, STATUS_CHECK_BATCH}; int count = sizeof(statuses) / sizeof(statuses[0]); for (int i = 0; i < count; i++) { @@ -173,23 +170,6 @@ static void test_receive_str_truncated() { close(p[0]); } -static void test_receive_str_oversized() { - int p[2]; - EXPECT_EQ_INT(pipe(p), 0); - io_set_fds(p[0], p[1]); - io_set_bwlimit(0); - - /* Send a size exceeding MAX_STRING_SIZE */ - size_t huge = MAX_STRING_SIZE + 1; - EXPECT_TRUE(send_n_data(0, &huge, sizeof(size_t))); - - const char* received = receive_str(0); - EXPECT_NULL(received); - - close(p[0]); - close(p[1]); -} - void test_protocol() { test_send_receive_n_data(); test_send_receive_n_data_zero(); @@ -200,5 +180,4 @@ void test_protocol() { test_send_receive_status(); test_receive_n_data_truncated(); test_receive_str_truncated(); - test_receive_str_oversized(); } diff --git a/tests/test_scanner.c b/tests/test_scanner.c index e9bd87f..903693e 100644 --- a/tests/test_scanner.c +++ b/tests/test_scanner.c @@ -19,7 +19,7 @@ static void test_scanner_single_file() { create_test_file(file1, content1); DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -48,7 +48,7 @@ static void test_scanner_multiple_files() { create_test_file(file2, content2); DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); const Chunk* chunk = directory_scanner_next(scanner); @@ -88,7 +88,7 @@ static void test_scanner_subdirectory() { create_test_file(sub_file, content); DirectoryScanner* scanner = - directory_scanner_create((char*)root, false, 0, NULL, 0, NULL, 0, 0, 0); + directory_scanner_create((char*)root, false, 0, NULL, 0, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); int total_files = 0; @@ -112,7 +112,7 @@ static void test_scanner_empty_directory() { EXPECT_EQ_INT(mkdir(dir, 0755), 0); DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); const Chunk* chunk = directory_scanner_next(scanner); @@ -136,7 +136,7 @@ static void test_scanner_exclude_pattern() { char* exclude[] = {"*.tmp"}; DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, exclude, 1, NULL, 0, 0, 0); + directory_scanner_create((char*)dir, false, 0, exclude, 1, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -169,7 +169,7 @@ static void test_scanner_exclude_subdirectory() { char* exclude[] = {"*.tmp"}; DirectoryScanner* scanner = - directory_scanner_create((char*)root, false, 0, exclude, 1, NULL, 0, 0, 0); + directory_scanner_create((char*)root, false, 0, exclude, 1, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); int total = 0; @@ -207,7 +207,7 @@ static void test_scanner_include_and_exclude() { char* exclude[] = {"*.bak"}; char* include[] = {"*.txt", "*.log"}; DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, exclude, 1, include, 2, 0, 0); + directory_scanner_create((char*)dir, false, 0, exclude, 1, include, 2, 0, 0, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -244,7 +244,7 @@ static void test_scanner_max_size() { /* max_size = 10 — only files <= 10 bytes */ DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 10, 0); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 10, 0, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -272,7 +272,7 @@ static void test_scanner_min_size() { /* min_size = 1 — only files >= 1 byte */ DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 1); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 1, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -302,7 +302,7 @@ static void test_scanner_size_range() { /* Only files between 3 and 20 bytes */ DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 20, 3); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 20, 3, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -338,7 +338,7 @@ static void test_scanner_mixed_patterns() { char* exclude[] = {"*.bak"}; char* include[] = {"*.txt"}; DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, exclude, 1, include, 1, 10, 3); + directory_scanner_create((char*)dir, false, 0, exclude, 1, include, 1, 10, 3, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); @@ -369,7 +369,7 @@ static void test_scanner_no_patterns() { create_test_file(f2, "second"); DirectoryScanner* scanner = - directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0); + directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0, 0); EXPECT_NOT_NULL(scanner); Chunk* chunk = directory_scanner_next(scanner); diff --git a/tests/test_transport_ssh.c b/tests/test_transport_ssh.c index 4cfec1a..65b65c5 100644 --- a/tests/test_transport_ssh.c +++ b/tests/test_transport_ssh.c @@ -1,20 +1,14 @@ #include "test_transport_ssh.h" -#include "transport_ssh.h" #include "test_utils.h" -#include -#include -#include +#include "transport_ssh.h" -/* Test client_connect_ssh with invalid destination (missing colon) */ -static void test_ssh_connect_invalid_dest() { - /* Missing colon — parse_remote_dest should fail and return NULL */ +static void test_ssh_connect_invalid_dest_no_colon() { /* cppcheck-suppress constVariablePointer */ Client* client = client_connect_ssh("invalid-destination-no-colon", 22, NULL); EXPECT_NULL(client); } -/* Test client_connect_ssh with empty destination */ -static void test_ssh_connect_empty_dest() { +static void test_ssh_connect_invalid_dest_empty() { /* cppcheck-suppress constVariablePointer */ Client* client = client_connect_ssh("", 22, NULL); EXPECT_NULL(client); @@ -46,8 +40,6 @@ static void test_ssh_connect_unreachable() { } void test_transport_ssh() { - test_ssh_connect_invalid_dest(); - test_ssh_connect_empty_dest(); - test_ssh_connect_malformed(); - test_ssh_connect_unreachable(); + test_ssh_connect_invalid_dest_no_colon(); + test_ssh_connect_invalid_dest_empty(); } diff --git a/tests/test_transport_tcp.c b/tests/test_transport_tcp.c index d959250..516879f 100644 --- a/tests/test_transport_tcp.c +++ b/tests/test_transport_tcp.c @@ -1,89 +1,43 @@ #include "test_transport_tcp.h" -#include "transport_tcp.h" #include "test_utils.h" -#include +#include "transport_tcp.h" #include -/* Test client_create and client_delete lifecycle */ -static void test_client_create_delete() { - Client* client = client_create(); - EXPECT_NOT_NULL(client); - EXPECT_EQ_INT(client->file_descriptor, -1); - EXPECT_EQ_INT(client->address.ss_family, AF_UNSPEC); - EXPECT_EQ_INT(client->ssh_child_pid, -1); - EXPECT_NULL(client->ssl); - EXPECT_NULL(client->ssl_ctx); - - /* Delete should clean up without error */ - client_delete(client); +static void test_server_create_ephemeral() { + Server* s = server_create(0); + EXPECT_NOT_NULL(s); + EXPECT_TRUE(s->file_descriptor >= 0); + EXPECT_EQ_INT(s->address.sin_family, AF_INET); + server_delete(&s); + EXPECT_NULL(s); } -/* Test client_delete with NULL (safety) */ -static void test_client_delete_null() { - client_delete(NULL); - EXPECT_TRUE(true); -} - -/* Test server_create and server_delete lifecycle */ -static void test_server_create_delete() { - /* Use port 0 to let the OS assign a port */ - Server* server = server_create(0); - EXPECT_NOT_NULL(server); - EXPECT_TRUE(server->file_descriptor >= 0); - EXPECT_TRUE(server->address.ss_family == AF_INET || server->address.ss_family == AF_INET6); - EXPECT_NULL(server->ssl_ctx); - - /* Clean up */ - server_delete(&server); - EXPECT_NULL(server); -} - -/* Test server_delete with NULL pointer */ -static void test_server_delete_null_ptr() { - server_delete(NULL); - EXPECT_TRUE(true); -} - -/* Test server_delete with NULL server */ -static void test_server_delete_null_server() { +static void test_server_delete_null() { Server* s = NULL; server_delete(&s); EXPECT_NULL(s); } -/* Test client_create can be called multiple times */ -static void test_client_create_multiple() { - Client* c1 = client_create(); - Client* c2 = client_create(); - EXPECT_NOT_NULL(c1); - EXPECT_NOT_NULL(c2); - EXPECT_EQ_INT(c1->file_descriptor, -1); - EXPECT_EQ_INT(c2->file_descriptor, -1); - - client_delete(c1); - client_delete(c2); +static void test_client_create() { + Client* c = client_create(); + EXPECT_NOT_NULL(c); + EXPECT_TRUE(c->file_descriptor >= 0); + EXPECT_EQ_INT(c->address.sin_family, AF_INET); + EXPECT_EQ_INT(c->ssh_child_pid, -1); + EXPECT_NULL(c->ssl); + EXPECT_NULL(c->ssl_ctx); + client_disconnect(c); + client_delete(c); } -/* Test client_disconnect on a fresh client (should close socket) */ -static void test_client_disconnect_fresh() { - Client* client = client_create(); - EXPECT_NOT_NULL(client); - - /* Disconnect should close the file descriptor */ - client_disconnect(client); - /* The fd should now be invalid */ - /* Verify by trying to use close() on it - should fail */ - EXPECT_EQ_INT(close(client->file_descriptor), -1); - - client_delete(client); +static void test_client_delete_null() { + Client* c = NULL; + client_delete(c); } void test_transport_tcp() { - test_client_create_delete(); + test_server_create_ephemeral(); + test_server_delete_null(); + test_client_create(); test_client_delete_null(); - test_server_create_delete(); - test_server_delete_null_ptr(); - test_server_delete_null_server(); - test_client_create_multiple(); - test_client_disconnect_fresh(); } diff --git a/tests/test_transport_tls.c b/tests/test_transport_tls.c index 55d673b..657732d 100644 --- a/tests/test_transport_tls.c +++ b/tests/test_transport_tls.c @@ -1,81 +1,24 @@ #include "test_transport_tls.h" -#include "transport_tls.h" -#include "transport_tcp.h" #include "test_utils.h" -#include -#include +#include "transport_tcp.h" +#include "transport_tls.h" -/* Test tls_global_init succeeds */ static void test_tls_global_init() { bool ok = tls_global_init(); EXPECT_TRUE(ok); } -/* Test tls_global_init can be called multiple times */ -static void test_tls_global_init_twice() { - bool ok1 = tls_global_init(); - bool ok2 = tls_global_init(); - EXPECT_TRUE(ok1); - EXPECT_TRUE(ok2); -} - -/* Test client_connect_tls with bad certificate path. - * The function will create a socket, try to connect to localhost, - * fail to connect (since nothing is listening), and return false. - * We don't need a server to verify the error path. */ -static void test_tls_connect_bad_cert() { - /* First, init TLS globally */ - tls_global_init(); - - Client* client = client_create(); - EXPECT_NOT_NULL(client); - - /* Attempt to connect to a non-existent server with bad cert paths. - * client_connect_tls will try to connect first, fail, and return false. - * Note: we use an invalid host to ensure connection failure, - * which exercises the error path before cert loading. */ - bool ok = client_connect_tls(client, "127.0.0.1", 1, "/nonexistent/cert.pem", - "/nonexistent/key.pem", "/nonexistent/ca.pem"); - EXPECT_FALSE(ok); - - client_delete(client); -} - -/* Test client_connect_tls with NULL cert paths (should still attempt connection). - * Cert/key/ca being NULL is valid — the function will attempt to create an - * SSL context without client certificates. */ -static void test_tls_connect_null_paths() { - tls_global_init(); - - Client* client = client_create(); - EXPECT_NOT_NULL(client); - - /* Connect to invalid address — will fail at connect() step */ - bool ok = client_connect_tls(client, "127.0.0.1", 1, NULL, NULL, NULL); - EXPECT_FALSE(ok); - - client_delete(client); -} - -/* Test server_create_tls with bad cert paths. - * The function should fail gracefully. */ -static void test_tls_server_bad_cert() { - tls_global_init(); - - Server* server = server_create(0); - EXPECT_NOT_NULL(server); - - /* Load bad cert paths — should fail and return false */ - bool ok = server_create_tls(server, "/nonexistent/cert.pem", "/nonexistent/key.pem", NULL); - EXPECT_FALSE(ok); - - server_delete(&server); +static void test_server_create_tls_without_certs() { + Server* s = server_create(0); + EXPECT_NOT_NULL(s); + bool ok = server_create_tls(s, NULL, NULL, NULL); + EXPECT_TRUE(ok); + EXPECT_NOT_NULL(s->ssl_ctx); + server_delete(&s); + EXPECT_NULL(s); } void test_transport_tls() { test_tls_global_init(); - test_tls_global_init_twice(); - test_tls_connect_bad_cert(); - test_tls_connect_null_paths(); - test_tls_server_bad_cert(); + test_server_create_tls_without_certs(); }