Merge branch 'fix/w6-tests' into fix/w6-integration

tests/fuzz/fuzz_manifest.c

@@ -0,0 +1,126 @@
+/*
+ * Fuzz the delete-manifest parser: receive_manifest_entries(int fd).
+ *
+ * The parser reads three length-delimited sections (keeps, protected prefixes,
+ * missing-args paths) from the connection.  Feeding raw bytes alone exercises
+ * the "reject the first malformed count/string" fast paths, but because each
+ * section is self-delimiting a single bad value hides every later section.
+ *
+ * To reach the protected-prefix and missing-args parsers (the paths that drive
+ * actual destination deletion) we build one canonical, fully-valid manifest
+ * with hand-written wire framing and then feed the receiver several shapes:
+ *
+ *   1. raw    : the raw fuzz bytes as the whole manifest.
+ *   2. keeps  : the valid keep count only + the fuzz bytes, so the fuzzer
+ *               drives the keep count and entries directly.
+ *   3. prot   : the valid keeps section + the fuzz bytes, so the fuzzer drives
+ *               the protected count and prefixes.
+ *   4. missing: the valid keeps+protected sections + the fuzz bytes, so the
+ *               fuzzer drives the trailing missing-args section, including the
+ *               aggregate MAX_MANIFEST_BYTES budget.
+ *
+ * The wire encoding matches receive_int (native int) and receive_wire_str
+ * (native size_t length prefix + body); no charset conversion is configured in
+ * the fuzz process, so receive_wire_str is receive_str.
+ */
+#include "file_receive.h"
+#include "protocol.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+static unsigned char g_manifest[512];
+static size_t g_len_after_count;     /* offset of the first keep entry */
+static size_t g_len_after_keeps;     /* offset of the protected count */
+static size_t g_len_after_protected; /* offset of the missing count */
+static int g_manifest_ready;
+
+static void append_int32(unsigned char* buf, size_t* off, int32_t value) {
+  memcpy(buf + *off, &value, sizeof(value));
+  *off += sizeof(value);
+}
+
+static void append_wire_str(unsigned char* buf, size_t* off, const char* s) {
+  size_t n = strlen(s);
+  memcpy(buf + *off, &n, sizeof(n));
+  *off += sizeof(n);
+  memcpy(buf + *off, s, n);
+  *off += n;
+}
+
+static void build_canonical_manifest(void) {
+  g_manifest_ready = 1;
+  size_t off = 0;
+  append_int32(g_manifest, &off, 2);
+  g_len_after_count = off;
+  append_wire_str(g_manifest, &off, "keep/a");
+  append_wire_str(g_manifest, &off, "keep/b");
+  g_len_after_keeps = off;
+  append_int32(g_manifest, &off, 1);
+  append_wire_str(g_manifest, &off, "excluded/prefix");
+  g_len_after_protected = off;
+  append_int32(g_manifest, &off, 1);
+  append_wire_str(g_manifest, &off, "missing/path");
+}
+
+/* Best-effort non-blocking write: an oversized fuzz input is truncated rather
+ * than stalling the harness. */
+static void write_best_effort(int fd, const void* data, size_t size) {
+  const unsigned char* p = data;
+  size_t off = 0;
+  while (off < size) {
+    ssize_t n = write(fd, p + off, size - off);
+    if (n > 0) {
+      off += (size_t)n;
+      continue;
+    }
+    if (n < 0 && errno == EINTR)
+      continue;
+    break;
+  }
+}
+
+/* Build prefix ++ data as a stream and drive receive_manifest_entries over it.
+ * The write half is shut down first so the parser always sees EOF instead of
+ * blocking on a missing frame tail. */
+static void receive_stream(const unsigned char* prefix, size_t prefix_len, const uint8_t* data,
+                           size_t size) {
+  int sv[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) != 0)
+    return;
+
+  int flags = fcntl(sv[0], F_GETFL, 0);
+  if (flags != -1)
+    (void)fcntl(sv[0], F_SETFL, flags | O_NONBLOCK);
+
+  if (prefix_len > 0)
+    write_best_effort(sv[0], prefix, prefix_len);
+  if (size > 0)
+    write_best_effort(sv[0], data, size);
+  shutdown(sv[0], SHUT_WR);
+
+  DeleteManifest* manifest = receive_manifest_entries(sv[1]);
+  delete_manifest_free(manifest);
+
+  close(sv[0]);
+  close(sv[1]);
+}
+
+int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
+  if (!g_manifest_ready)
+    build_canonical_manifest();
+
+  /* Raw bytes as the whole manifest. */
+  receive_stream(NULL, 0, data, size);
+
+  /* Keep the valid framing so the fuzzer reaches each later section. */
+  receive_stream(g_manifest, g_len_after_protected, data, size);
+  receive_stream(g_manifest, g_len_after_keeps, data, size);
+  receive_stream(g_manifest, g_len_after_count, data, size);
+
+  return 0;
+}

tests/fuzz/fuzz_protocol_framing.c

@@ -0,0 +1,127 @@
+/*
+ * Fuzz the base protocol framing: receive_str / receive_data / receive_status
+ * (plus the redacted string, the size-limited data and the timed-status
+ * variants) fed arbitrary bytes over an in-memory socketpair.
+ *
+ * Every receive primitive reads a fixed-width header (a size_t string length,
+ * an unsigned long long data length, an int status/int value) and then a body.
+ * The fuzzer attacks:
+ *   - oversized length headers (the MAX_STRING_SIZE / MAX_DATA_PAYLOAD_SIZE
+ *     gates must reject before allocating),
+ *   - truncated bodies (a declared body larger than the stream must fail
+ *     cleanly at EOF, never read uninitialised memory or leak),
+ *   - embedded NUL bytes in strings (must be refused),
+ *   - out-of-range status enum values (status_to_string must stay in bounds).
+ *
+ * Each entry point gets its own socketpair because a single receive consumes a
+ * variable number of bytes from the stream; reusing one would make the later
+ * calls meaningless.  The write half is shut down first so a truncated frame
+ * always terminates at EOF instead of blocking.
+ */
+#include "data.h"
+#include "protocol.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+static void write_best_effort(int fd, const void* data, size_t size) {
+  const unsigned char* p = data;
+  size_t off = 0;
+  while (off < size) {
+    ssize_t n = write(fd, p + off, size - off);
+    if (n > 0) {
+      off += (size_t)n;
+      continue;
+    }
+    if (n < 0 && errno == EINTR)
+      continue;
+    break;
+  }
+}
+
+/* Create a socketpair pre-loaded with `data`, shut down the write half and
+ * return the read end (which the receiver reads from).  `*write_end` is also
+ * returned so the caller can close it. */
+static int make_stream(const uint8_t* data, size_t size, int* write_end) {
+  int sv[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) != 0) {
+    *write_end = -1;
+    return -1;
+  }
+  int flags = fcntl(sv[0], F_GETFL, 0);
+  if (flags != -1)
+    (void)fcntl(sv[0], F_SETFL, flags | O_NONBLOCK);
+  if (size > 0)
+    write_best_effort(sv[0], data, size);
+  shutdown(sv[0], SHUT_WR);
+  *write_end = sv[0];
+  return sv[1];
+}
+
+int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
+  int w;
+
+  int rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    char* s = receive_str(rd);
+    free(s);
+    close(rd);
+    close(w);
+  }
+
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    char* s = receive_str_redacted(rd);
+    free(s);
+    close(rd);
+    close(w);
+  }
+
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    Data* d = receive_data(rd);
+    data_destroy(d);
+    close(rd);
+    close(w);
+  }
+
+  /* The size-limited variant must reject anything beyond its explicit bound
+   * before allocating the body buffer. */
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    Data* d = receive_data_limited(rd, 256);
+    data_destroy(d);
+    close(rd);
+    close(w);
+  }
+
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    Status status = STATUS_OK;
+    (void)receive_status(rd, &status);
+    close(rd);
+    close(w);
+  }
+
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    Status status = STATUS_OK;
+    (void)receive_status_timed(rd, &status, 1);
+    close(rd);
+    close(w);
+  }
+
+  rd = make_stream(data, size, &w);
+  if (rd >= 0) {
+    int value = 0;
+    (void)receive_int(rd, &value);
+    close(rd);
+    close(w);
+  }
+
+  return 0;
+}

tests/fuzz/fuzz_xattr_block.c

@@ -0,0 +1,115 @@
+/*
+ * Fuzz the xattr wire block parser: xattr_receive(int fd, int* ok).
+ *
+ * The block is a count followed by that many (name_len, name, value_len, value)
+ * records.  The receiver must reject an invalid count, an out-of-range or
+ * negative name/value length, an embedded NUL or non-whitelisted namespace in
+ * the name, an oversized value, and an aggregate payload beyond
+ * XATTR_TOTAL_MAX -- all without over-allocating or leaking.
+ *
+ * Raw bytes mostly stop at the first invalid count/length, so we also build a
+ * canonical, fully-valid two-entry block by hand and feed the receiver valid
+ * prefixes of it followed by the fuzz bytes.  That drives the deep value-
+ * parsing and per-entry namespace/budget checks with attacker-controlled input.
+ */
+#include "protocol.h"
+#include "xattr.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+static unsigned char g_block[512];
+static size_t g_off_after_count;  /* start of entry 0 */
+static size_t g_off_after_entry0; /* start of entry 1 */
+static size_t g_off_value0;       /* start of the first value length */
+static int g_block_ready;
+
+static void append_int32(unsigned char* buf, size_t* off, int32_t value) {
+  memcpy(buf + *off, &value, sizeof(value));
+  *off += sizeof(value);
+}
+
+static void append_bytes(unsigned char* buf, size_t* off, const void* p, size_t n) {
+  if (n > 0)
+    memcpy(buf + *off, p, n);
+  *off += n;
+}
+
+static void build_canonical_block(void) {
+  g_block_ready = 1;
+  size_t off = 0;
+  append_int32(g_block, &off, 2);
+  g_off_after_count = off;
+
+  int32_t name0_len = (int32_t)strlen("user.foo");
+  append_int32(g_block, &off, name0_len);
+  append_bytes(g_block, &off, "user.foo", (size_t)name0_len);
+  g_off_value0 = off;
+  append_int32(g_block, &off, 3);
+  append_bytes(g_block, &off, "bar", 3);
+
+  g_off_after_entry0 = off;
+  int32_t name1_len = (int32_t)strlen("user.empty");
+  append_int32(g_block, &off, name1_len);
+  append_bytes(g_block, &off, "user.empty", (size_t)name1_len);
+  append_int32(g_block, &off, 0);
+}
+
+static void write_best_effort(int fd, const void* data, size_t size) {
+  const unsigned char* p = data;
+  size_t off = 0;
+  while (off < size) {
+    ssize_t n = write(fd, p + off, size - off);
+    if (n > 0) {
+      off += (size_t)n;
+      continue;
+    }
+    if (n < 0 && errno == EINTR)
+      continue;
+    break;
+  }
+}
+
+static void receive_stream(const unsigned char* prefix, size_t prefix_len, const uint8_t* data,
+                           size_t size) {
+  int sv[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) != 0)
+    return;
+
+  int flags = fcntl(sv[0], F_GETFL, 0);
+  if (flags != -1)
+    (void)fcntl(sv[0], F_SETFL, flags | O_NONBLOCK);
+
+  if (prefix_len > 0)
+    write_best_effort(sv[0], prefix, prefix_len);
+  if (size > 0)
+    write_best_effort(sv[0], data, size);
+  shutdown(sv[0], SHUT_WR);
+
+  int ok = 0;
+  FileXattrList* list = xattr_receive(sv[1], &ok);
+  xattr_list_free(list);
+
+  close(sv[0]);
+  close(sv[1]);
+}
+
+int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
+  if (!g_block_ready)
+    build_canonical_block();
+
+  /* Raw bytes as the whole block. */
+  receive_stream(NULL, 0, data, size);
+
+  /* Valid framing so the fuzzer mutates the entry list, the first value and
+   * the second entry respectively instead of stopping at the count. */
+  receive_stream(g_block, g_off_after_entry0, data, size);
+  receive_stream(g_block, g_off_value0, data, size);
+  receive_stream(g_block, g_off_after_count, data, size);
+
+  return 0;
+}

tests/integration/test_fault_injection.py

@@ -0,0 +1,367 @@
+"""Fault injection: the server must survive truncated / corrupted protocol
+frames and abrupt mid-frame disconnects, and keep serving later connections.
+
+These tests deliberately speak raw bytes to a real server process:
+
+  * malformed frames before/inside the config handshake (oversized length
+    headers, truncated string bodies, outright garbage),
+  * a captured *valid* config frame replayed so the connection reaches the
+    operation loop, followed by a partial ``STATUS_MANIFEST`` frame that is cut
+    mid-body and dropped, and
+  * a real client run relayed through a proxy that truncates the stream at a
+    range of byte offsets and resets both ends.
+
+After every fault the server process is asserted alive and a subsequent
+ordinary transfer must complete and verify, proving the accept loop and
+per-connection children recovered cleanly.  All interactions are bounded by
+short socket timeouts (no sleeps).
+"""
+import os
+import select
+import shutil
+import socket
+import struct
+import subprocess
+import sys
+import threading
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (  # noqa: E402
+    ServerManager,
+    TEST_DATA_DIR,
+    get_dest_received_dir,
+    run_client,
+    verify_transfer,
+)
+
+PROTOCOL_VERSION = b"2.20.0"
+STATUS_MANIFEST = 5
+STATUS_OK = 0
+
+SOURCE_DIR = os.path.join(TEST_DATA_DIR, "fault_src")
+DEST_DIR = os.path.join(TEST_DATA_DIR, "fault_dst")
+
+
+@pytest.fixture(scope="module")
+def fault_server():
+    """A dedicated server so the aliveness assertions observe exactly the
+    process these faults were sent to."""
+    server = ServerManager()
+    server.start()
+    yield server
+    server.stop()
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _seed_source():
+    if os.path.exists(SOURCE_DIR):
+        shutil.rmtree(SOURCE_DIR)
+    os.makedirs(os.path.join(SOURCE_DIR, "nested"))
+    with open(os.path.join(SOURCE_DIR, "hello.txt"), "wb") as fh:
+        fh.write(b"fault injection payload\n" * 64)
+    with open(os.path.join(SOURCE_DIR, "nested", "deep.bin"), "wb") as fh:
+        fh.write(bytes(range(256)) * 16)
+    yield
+    shutil.rmtree(SOURCE_DIR, ignore_errors=True)
+    shutil.rmtree(DEST_DIR, ignore_errors=True)
+
+
+def _assert_alive(server):
+    assert server._proc is not None, "server process missing"
+    assert server._proc.poll() is None, (
+        f"server exited with {server._proc.returncode} after fault injection"
+    )
+
+
+def _recover(server, label):
+    """Run one ordinary transfer and verify it end-to-end."""
+    shutil.rmtree(DEST_DIR, ignore_errors=True)
+    os.makedirs(DEST_DIR)
+    result, _ = run_client(SOURCE_DIR, DEST_DIR, flags=["--preserve"], port=server.port)
+    assert result.returncode == 0, (
+        f"{label}: recovery transfer failed rc={result.returncode}: "
+        f"{(result.stderr or result.stdout)[:200]}"
+    )
+    received = get_dest_received_dir(DEST_DIR, SOURCE_DIR)
+    mismatches, missing = verify_transfer(SOURCE_DIR, received)
+    assert not missing, f"{label}: recovery missing {missing}"
+    assert not mismatches, f"{label}: recovery mismatch {mismatches}"
+
+
+def _abrupt_close(sock):
+    """Force an RST instead of a graceful FIN, the nastier mid-frame drop."""
+    try:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+    except OSError:
+        pass
+    try:
+        sock.close()
+    except OSError:
+        pass
+
+
+def _raw_connect(server):
+    sock = socket.create_connection(("127.0.0.1", server.port), timeout=5)
+    sock.settimeout(5)
+    return sock
+
+
+def _recv_exact(sock, n):
+    buf = b""
+    while len(buf) < n:
+        chunk = sock.recv(n - len(buf))
+        if not chunk:
+            return None
+        buf += chunk
+    return buf
+
+
+# --- faults before/inside the config handshake -----------------------------
+
+CONFIG_HANDSHAKE_FAULTS = {
+    "empty": b"",
+    # Length header claims a 1 EiB string body that never arrives.
+    "oversized_length": struct.pack("<Q", 1 << 60),
+    # A truncated 8-byte length header (only 3 bytes of it are sent).
+    "truncated_length_header": b"\x10\x00\x00",
+    # A valid version string followed by a string-length header whose body is
+    # deliberately truncated (mid-config-frame disconnect).
+    "truncated_config_body": struct.pack("<Q", len(PROTOCOL_VERSION)) + PROTOCOL_VERSION
+    + struct.pack("<Q", 4096)
+    + b"partial",
+    # Pure garbage that is not a valid frame at any offset.
+    "garbage": b"\xff" * 32,
+}
+
+
+class TestConfigHandshakeFaults:
+    def test_truncated_and_corrupt_config_frames(self, fault_server):
+        for name, payload in CONFIG_HANDSHAKE_FAULTS.items():
+            sock = _raw_connect(fault_server)
+            if payload:
+                sock.sendall(payload)
+            _abrupt_close(sock)
+            _assert_alive(fault_server)
+        _recover(fault_server, "config handshake faults")
+
+
+# --- capture a valid config frame, then truncate a STATUS_MANIFEST ----------
+
+
+class _CaptureProxy:
+    """Relay one client<->server connection and record the client's config
+    frame (all client bytes forwarded before the server's first reply)."""
+
+    def __init__(self, target_port):
+        self.target = ("127.0.0.1", target_port)
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.listener.settimeout(20)
+        self.port = self.listener.getsockname()[1]
+        self.config_frame = None
+
+    def run(self, cmd):
+        def serve():
+            try:
+                client, _ = self.listener.accept()
+            except OSError:
+                return
+            try:
+                backend = socket.create_connection(self.target, timeout=10)
+            except OSError:
+                client.close()
+                return
+            client.settimeout(20)
+            backend.settimeout(20)
+            buf_c = bytearray()
+            seen_server = False
+            try:
+                while True:
+                    ready, _, _ = select.select([client, backend], [], [], 20)
+                    if not ready:
+                        break
+                    done = False
+                    for sock in ready:
+                        data = sock.recv(65536)
+                        if not data:
+                            done = True
+                            continue
+                        if sock is client:
+                            buf_c += data
+                            backend.sendall(data)
+                        else:
+                            if not seen_server:
+                                seen_server = True
+                                self.config_frame = bytes(buf_c)
+                            client.sendall(data)
+                    if done:
+                        break
+            except OSError:
+                pass
+            finally:
+                client.close()
+                backend.close()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
+        thread.join(20)
+        return result
+
+    def close(self):
+        try:
+            self.listener.close()
+        except OSError:
+            pass
+
+
+@pytest.fixture(scope="module")
+def captured_config(fault_server):
+    """Capture the config frame of one real client run through a relay."""
+    proxy = _CaptureProxy(fault_server.port)
+    cmd = [
+        os.path.join(os.path.dirname(__file__), "..", "..", "build", "client"),
+        "--source-dir",
+        SOURCE_DIR,
+        "--dest-dir",
+        DEST_DIR,
+        "--save-to-disk",
+        "--server-port",
+        str(proxy.port),
+    ]
+    try:
+        result = proxy.run(cmd)
+        assert result.returncode == 0, (
+            f"capture run failed rc={result.returncode}: "
+            f"{(result.stderr or result.stdout)[:200]}"
+        )
+        assert proxy.config_frame, "failed to capture the client config frame"
+        yield proxy.config_frame
+    finally:
+        proxy.close()
+
+
+class TestTruncatedStatusFrame:
+    def test_partial_manifest_frame_then_drop(self, fault_server, captured_config):
+        sock = _raw_connect(fault_server)
+        sock.sendall(captured_config)
+        ack = _recv_exact(sock, 4)
+        assert ack is not None, "server closed before the config ack"
+        (status,) = struct.unpack("<i", ack)
+        assert status == STATUS_OK, f"expected STATUS_OK, got {status}"
+
+        # STATUS_MANIFEST, then only half of the keep-count int, then an RST.
+        sock.sendall(struct.pack("<i", STATUS_MANIFEST) + b"\x02\x00")
+        _abrupt_close(sock)
+
+        _assert_alive(fault_server)
+        _recover(fault_server, "truncated manifest frame")
+
+    def test_manifest_count_without_sections(self, fault_server, captured_config):
+        """A syntactically valid STATUS_MANIFEST whose bodies never arrive."""
+        sock = _raw_connect(fault_server)
+        sock.sendall(captured_config)
+        assert _recv_exact(sock, 4) is not None
+
+        sock.sendall(struct.pack("<i", STATUS_MANIFEST) + struct.pack("<i", 3))
+        # Announce three keeps but send none; then drop.
+        _abrupt_close(sock)
+
+        _assert_alive(fault_server)
+        _recover(fault_server, "manifest body truncation")
+
+
+# --- abrupt truncation of a real transfer ----------------------------------
+
+
+class _TruncatingProxy:
+    """Forward at most ``max_client_bytes`` from client to server, then reset
+    both ends mid-stream.  Runs one client command (which is expected to fail)."""
+
+    def __init__(self, target_port, max_client_bytes):
+        self.target = ("127.0.0.1", target_port)
+        self.max_client_bytes = max_client_bytes
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.listener.settimeout(20)
+        self.port = self.listener.getsockname()[1]
+
+    def run(self, cmd):
+        def serve():
+            try:
+                client, _ = self.listener.accept()
+            except OSError:
+                return
+            try:
+                backend = socket.create_connection(self.target, timeout=10)
+            except OSError:
+                client.close()
+                return
+            # A short receive timeout bounds the case where the client has
+            # nothing left to send and is waiting on the server: the proxy then
+            # cuts the stream anyway instead of stalling the test.
+            client.settimeout(2)
+            backend.settimeout(20)
+            forwarded = 0
+            try:
+                while forwarded < self.max_client_bytes:
+                    data = client.recv(65536)
+                    if not data:
+                        break
+                    room = self.max_client_bytes - forwarded
+                    take = data[:room]
+                    backend.sendall(take)
+                    forwarded += len(take)
+                    if forwarded >= self.max_client_bytes:
+                        break
+            except (OSError, socket.timeout):
+                pass
+            for sock in (client, backend):
+                try:
+                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+                except OSError:
+                    pass
+                try:
+                    sock.close()
+                except OSError:
+                    pass
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        try:
+            subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+        finally:
+            thread.join(20)
+        self.listener.close()
+
+
+class TestAbruptMidTransferDisconnect:
+    def test_client_stream_cut_at_offsets(self, fault_server, captured_config):
+        """Cut the real client stream at offsets anchored to the config frame's
+        actual size: mid-config, right after the config, and into the operation
+        stream -- each followed by an RST of both ends."""
+        config_len = len(captured_config)
+        cuts = sorted({max(1, config_len // 2), max(1, config_len - 1), config_len + 8,
+                       config_len + 256})
+        for cut in cuts:
+            proxy = _TruncatingProxy(fault_server.port, cut)
+            cmd = [
+                os.path.join(os.path.dirname(__file__), "..", "..", "build", "client"),
+                "--source-dir",
+                SOURCE_DIR,
+                "--dest-dir",
+                DEST_DIR,
+                "--save-to-disk",
+                "--server-port",
+                str(proxy.port),
+            ]
+            # The client is expected to fail; what matters is the server survives.
+            proxy.run(cmd)
+            _assert_alive(fault_server)
+        _recover(fault_server, "abrupt mid-transfer disconnects")

tests/runner.c

@@ -16,6 +16,7 @@
 #include "test_file_sendfile.h"
 #include "test_fuzz_smoke.h"
 #include "test_glob.h"
+#include "test_hardlink.h"
 #include "test_iconv.h"
 #include "test_log.h"
 #include "test_metadata.h"
@@ -88,6 +89,7 @@ int main() {
   RUN_TEST(test_server_cli);
   RUN_TEST(test_fuzz_smoke);
   RUN_TEST(test_xattr);
+  RUN_TEST(test_hardlink);
 
   printf("\n\033[1;36m=== TEST SUMMARY ===\033[0m\n");
   printf("Total Tests Run: %d\n", tests_run);

tests/test_hardlink.c

@@ -0,0 +1,203 @@
+#include "test_hardlink.h"
+#include "hardlink.h"
+#include "test_utils.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/types.h>
+
+/* A fresh table starts empty and destroy accepts NULL / a fresh table. */
+static void test_hardlink_create_destroy() {
+  hardlink_table_destroy(NULL);
+
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+  EXPECT_EQ_INT((int)table->count, 0);
+  EXPECT_EQ_INT((int)table->capacity, 0);
+  EXPECT_EQ_INT(table->next_gid, 1);
+  hardlink_table_destroy(table);
+}
+
+/* The first member of an (dev, ino) group is data-carrying and owns the group;
+ * every later member gets the SAME gid, is not first, and points back at the
+ * first member's wire path. */
+static void test_hardlink_grouping() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  int gid_a = -1, gid_b = -1;
+  bool first_a = false, first_b = false;
+  char* first_path_a = NULL;
+  char* first_path_b = NULL;
+
+  EXPECT_TRUE(
+      hardlink_table_assign(table, "dir/first.txt", 7, 42, &gid_a, &first_a, &first_path_a));
+  EXPECT_TRUE(first_a);
+  EXPECT_EQ_INT(gid_a, 1);
+  EXPECT_NOT_NULL(first_path_a);
+  EXPECT_EQ_STR(first_path_a, "dir/first.txt");
+
+  EXPECT_TRUE(
+      hardlink_table_assign(table, "dir/second.txt", 7, 42, &gid_b, &first_b, &first_path_b));
+  EXPECT_FALSE(first_b);
+  EXPECT_EQ_INT(gid_b, gid_a);
+  EXPECT_NOT_NULL(first_path_b);
+  EXPECT_EQ_STR(first_path_b, "dir/first.txt");
+
+  /* Two members map onto a single stored group. */
+  EXPECT_EQ_INT((int)table->count, 1);
+  EXPECT_EQ_INT(table->next_gid, 2);
+
+  free(first_path_a);
+  free(first_path_b);
+  hardlink_table_destroy(table);
+}
+
+/* A different inode on the same device is a distinct group with a fresh gid. */
+static void test_hardlink_distinct_inode() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  int gid1 = -1, gid2 = -1;
+  bool first1 = false, first2 = false;
+  char* path1 = NULL;
+  char* path2 = NULL;
+
+  EXPECT_TRUE(hardlink_table_assign(table, "a", 7, 100, &gid1, &first1, &path1));
+  EXPECT_TRUE(first1);
+  EXPECT_TRUE(hardlink_table_assign(table, "b", 7, 101, &gid2, &first2, &path2));
+  EXPECT_TRUE(first2);
+  EXPECT_TRUE(gid1 != gid2);
+  EXPECT_EQ_INT(gid1, 1);
+  EXPECT_EQ_INT(gid2, 2);
+  EXPECT_EQ_STR(path1, "a");
+  EXPECT_EQ_STR(path2, "b");
+
+  free(path1);
+  free(path2);
+  hardlink_table_destroy(table);
+}
+
+/* Identical (dev, ino) on a DIFFERENT device must never be conflated: inode
+ * numbers are only unique per filesystem, so grouping is scoped by st_dev. */
+static void test_hardlink_distinct_device() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  int gid1 = -1, gid2 = -1;
+  bool first1 = false, first2 = false;
+  char* path1 = NULL;
+  char* path2 = NULL;
+
+  EXPECT_TRUE(hardlink_table_assign(table, "dev_a/one", 1, 55, &gid1, &first1, &path1));
+  EXPECT_TRUE(hardlink_table_assign(table, "dev_b/one", 2, 55, &gid2, &first2, &path2));
+  EXPECT_TRUE(first1);
+  EXPECT_TRUE(first2);
+  EXPECT_TRUE(gid1 != gid2);
+  EXPECT_EQ_INT((int)table->count, 2);
+
+  free(path1);
+  free(path2);
+  hardlink_table_destroy(table);
+}
+
+/* The table owns deep copies of every path: mutating (or freeing) the caller's
+ * buffer after assign must not affect the stored / returned paths. */
+static void test_hardlink_path_ownership() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  char caller[] = "owned/path";
+  int gid = -1;
+  bool is_first = false;
+  char* out = NULL;
+
+  EXPECT_TRUE(hardlink_table_assign(table, caller, 3, 9, &gid, &is_first, &out));
+  EXPECT_TRUE(is_first);
+  /* The returned pointer is a distinct allocation, not the caller's buffer. */
+  EXPECT_TRUE(out != caller);
+  EXPECT_TRUE(table->items[0].first_path != caller);
+
+  memset(caller, 'X', sizeof(caller) - 1);
+  EXPECT_EQ_STR(out, "owned/path");
+  EXPECT_EQ_STR(table->items[0].first_path, "owned/path");
+
+  /* Later members get their own independent copy of the first path. */
+  char second_caller[] = "owned/second";
+  int gid2 = -1;
+  bool first2 = true;
+  char* out2 = NULL;
+  EXPECT_TRUE(hardlink_table_assign(table, second_caller, 3, 9, &gid2, &first2, &out2));
+  EXPECT_FALSE(first2);
+  EXPECT_EQ_STR(out2, "owned/path");
+  EXPECT_TRUE(out2 != table->items[0].first_path);
+  EXPECT_TRUE(out2 != out);
+
+  free(out);
+  free(out2);
+  hardlink_table_destroy(table);
+}
+
+/* Bad arguments must be rejected without touching the table. */
+static void test_hardlink_reject_bad_args() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  int gid = 0;
+  bool is_first = false;
+  char* out = NULL;
+
+  EXPECT_FALSE(hardlink_table_assign(NULL, "x", 1, 1, &gid, &is_first, &out));
+  EXPECT_FALSE(hardlink_table_assign(table, NULL, 1, 1, &gid, &is_first, &out));
+  EXPECT_FALSE(hardlink_table_assign(table, "x", 1, 1, NULL, &is_first, &out));
+  EXPECT_FALSE(hardlink_table_assign(table, "x", 1, 1, &gid, NULL, &out));
+  EXPECT_FALSE(hardlink_table_assign(table, "x", 1, 1, &gid, &is_first, NULL));
+  EXPECT_EQ_INT((int)table->count, 0);
+
+  hardlink_table_destroy(table);
+}
+
+/* Many distinct groups grow the item array through its realloc path and keep
+ * gid assignment stable and monotonic. */
+static void test_hardlink_many_groups() {
+  HardLinkTable* table = hardlink_table_create();
+  EXPECT_NOT_NULL(table);
+
+  const int n = 200;
+  for (int i = 0; i < n; i++) {
+    int gid = -1;
+    bool is_first = false;
+    char* out = NULL;
+    char path[32];
+    snprintf(path, sizeof(path), "file_%d", i);
+    EXPECT_TRUE(hardlink_table_assign(table, path, 1, (ino_t)(1000 + i), &gid, &is_first, &out));
+    EXPECT_TRUE(is_first);
+    EXPECT_EQ_INT(gid, i + 1);
+    EXPECT_EQ_STR(out, path);
+    free(out);
+  }
+  EXPECT_EQ_INT((int)table->count, n);
+  EXPECT_EQ_INT(table->next_gid, n + 1);
+
+  /* Re-querying an existing inode still reports the original gid. */
+  int gid = -1;
+  bool is_first = true;
+  char* out = NULL;
+  EXPECT_TRUE(hardlink_table_assign(table, "file_7_again", 1, 1007, &gid, &is_first, &out));
+  EXPECT_FALSE(is_first);
+  EXPECT_EQ_INT(gid, 8);
+  EXPECT_EQ_STR(out, "file_7");
+  free(out);
+
+  hardlink_table_destroy(table);
+}
+
+void test_hardlink() {
+  test_hardlink_create_destroy();
+  test_hardlink_grouping();
+  test_hardlink_distinct_inode();
+  test_hardlink_distinct_device();
+  test_hardlink_path_ownership();
+  test_hardlink_reject_bad_args();
+  test_hardlink_many_groups();
+}

tests/test_hardlink.h

@@ -0,0 +1,6 @@
+#ifndef TEST_HARDLINK_H
+#define TEST_HARDLINK_H
+
+void test_hardlink(void);
+
+#endif