feat(identity): implement --copy-as USER[:GROUP] safe subset (P7 Wave E)

Force the receiver to apply the requested owner/group to every written
entry through the confined fd-relative identity path instead of switching
the process credentials (unsafe for the multithreaded receiver).  An
unprivileged receiver refuses the transfer up front in server_module_gate,
before STATUS_OK, so no data is written with the wrong ownership.

- new Config fields copy_as_set/copy_as_uid/copy_as_gid + defaults
- identity_parse_copy_as (name/@N/* resolution, primary-gid default,
  gid==uid fallback for numeric ids with no passwd entry); implies -M
- identity snapshot + highest-priority forcing in identity_resolve_targets
- identity_copy_as_refused() helper
- trailing config-frame block (presence int + two int32 ids, >=0 checked)
- PROTOCOL_VERSION 2.17.0 -> 2.18.0; version-sensitive tests updated
- unit tests for parse + wire round-trip/negative-id rejection
- integration TestCopyAs: unprivileged refusal + root chown assertion
- RSYNC_COMPAT.md --copy-as row updated (safe subset + divergence); README
  protocol version refreshed

tests/integration/test_features.py

@@ -5098,3 +5098,85 @@ class TestDirectoryAndSymlinkTimes:
         with open(blocker, "rb") as fh:
             assert fh.read() == b"pre-existing blocker\n", "the blocker file was clobbered"
         assert os.path.isfile(os.path.join(received, "keep.txt")), "regular file missing"
+
+
+class TestCopyAs:
+    """P7 Wave E: --copy-as=USER[:GROUP] safe subset.
+
+    FastSync never switches the receiver's process credentials; the receiver
+    forces the ownership of every entry it writes to the requested ids through
+    the confined fd-relative identity path, which REQUIRES a privileged (root)
+    receiver.  An unprivileged receiver refuses the whole transfer up front at
+    the config handshake, before any file data moves.
+    """
+
+    @pytest.mark.ci
+    def test_unprivileged_receiver_refuses_copy_as(self, shared_server):
+        """The key assertable behavior: an unprivileged receiver REFUSES a
+        --copy-as transfer cleanly (non-zero exit, no data written) instead of
+        silently writing the wrong ownership."""
+        source = os.path.join(TEST_DATA_DIR, "copyas_refuse_src")
+        dest = os.path.join(TEST_DATA_DIR, "copyas_refuse_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        with open(os.path.join(source, "secret.txt"), "wb") as fh:
+            fh.write(b"must not be written\n")
+
+        captured = None
+        if os.geteuid() == 0:
+            if shutil.which("setpriv") is None:
+                pytest.skip("root runner without setpriv cannot start an unprivileged receiver")
+            os.chmod(dest, 0o777)
+            proc, port = _start_captured_server(
+                prefix=["setpriv", "--reuid=65534", "--regid=65534", "--clear-groups"])
+            captured = proc
+        else:
+            # The session server already runs unprivileged.
+            port = shared_server.port
+
+        try:
+            result, _ = run_client(source, dest,
+                                   flags=["--copy-as=@65534:@65534"], port=port)
+        finally:
+            if captured is not None:
+                out, err = _stop_captured_server(captured)
+            else:
+                out, err = "", ""
+
+        assert result.returncode != 0, (
+            f"an unprivileged receiver must refuse --copy-as: rc={result.returncode} "
+            f"out={result.stdout[:200]!r} err={result.stderr[:200]!r}"
+        )
+        received = get_dest_received_dir(dest, source)
+        assert not os.path.exists(os.path.join(received, "secret.txt")), (
+            "--copy-as refusal leaked file data into the destination"
+        )
+        if captured is not None:
+            assert "copy-as requires a privileged receiver" in (out + err), (
+                f"refusal reason was not logged: out={out!r} err={err!r}"
+            )
+
+    @pytest.mark.ci
+    @pytest.mark.skipif(os.geteuid() != 0, reason="requires a root receiver to chown")
+    def test_root_copy_as_chowns_transferred_file(self, shared_server):
+        """Root-gated: --copy-as=USER:GROUP forces the transferred file's
+        ownership to exactly that uid/gid (numeric form for determinism)."""
+        source = os.path.join(TEST_DATA_DIR, "copyas_root_src")
+        dest = os.path.join(TEST_DATA_DIR, "copyas_root_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        with open(os.path.join(source, "owned.txt"), "wb") as fh:
+            fh.write(b"owned by nobody\n")
+
+        result, _ = run_client(source, dest,
+                               flags=["--copy-as=@65534:@65534"], port=shared_server.port)
+        assert result.returncode == 0, (
+            f"--copy-as root transfer failed: {(result.stderr or result.stdout)[:400]}"
+        )
+        received = get_dest_received_dir(dest, source)
+        target = os.path.join(received, "owned.txt")
+        assert os.path.isfile(target), f"transferred file missing at {target}"
+        st = os.lstat(target)
+        assert (st.st_uid, st.st_gid) == (65534, 65534), (
+            f"--copy-as did not force ownership: uid={st.st_uid} gid={st.st_gid}"
+        )

tests/integration/test_preflight.py

@@ -94,14 +94,14 @@ def _seed_protocol_source(source):
 class TestProtocol:
     @pytest.mark.ci
     def test_protocol_current_version_accepted(self, shared_server):
-        """--protocol=2.17.0 (the current PROTOCOL_VERSION) is accepted and the
+        """--protocol=2.18.0 (the current PROTOCOL_VERSION) is accepted and the
         transfer completes normally."""
         source = os.path.join(TEST_DATA_DIR, "proto_ok_src")
         dest = os.path.join(TEST_DATA_DIR, "proto_ok_dst")
         shutil.rmtree(dest, ignore_errors=True)
         os.makedirs(dest)
         _seed_protocol_source(source)
-        result, _ = run_client(source, dest, flags=["--protocol=2.17.0"],
+        result, _ = run_client(source, dest, flags=["--protocol=2.18.0"],
                                port=shared_server.port)
         assert result.returncode == 0, \
             f"--protocol current run failed: {(result.stderr or result.stdout)[:400]}"
@@ -118,7 +118,7 @@ class TestProtocol:
         shutil.rmtree(dest, ignore_errors=True)
         os.makedirs(dest)
         _seed_protocol_source(source)
-        for bad in ("2.15.0", "2.16.0", "216", "31"):
+        for bad in ("2.15.0", "2.16.0", "2.17.0", "216", "31"):
             result, _ = run_client(source, dest, flags=[f"--protocol={bad}"],
                                    port=shared_server.port)
             assert result.returncode != 0, f"--protocol={bad} should be rejected"

tests/test_client_cli.c

@@ -223,7 +223,7 @@ static void test_parse_args_protocol_accept_current() {
   Config* cfg = valid_client_config();
   EXPECT_NOT_NULL(cfg);
   char* argv_equals[] = {"fastsync",   "--source-dir", "/src",
-                         "--dest-dir", "/dst",         "--protocol=2.17.0"};
+                         "--dest-dir", "/dst",         "--protocol=2.18.0"};
   int positional_args[2];
   int positional_count = 0;
   EXPECT_EQ_INT(parse_args(cfg, 6, argv_equals, positional_args, &positional_count), 0);
@@ -233,7 +233,7 @@ static void test_parse_args_protocol_accept_current() {
   cfg = valid_client_config();
   EXPECT_NOT_NULL(cfg);
   char* argv_space[] = {"fastsync", "--source-dir", "/src",  "--dest-dir",
-                        "/dst",     "--protocol",   "2.17.0"};
+                        "/dst",     "--protocol",   "2.18.0"};
   positional_count = 0;
   EXPECT_EQ_INT(parse_args(cfg, 7, argv_space, positional_args, &positional_count), 0);
   EXPECT_EQ_STR(cfg->version, PROTOCOL_VERSION);
@@ -243,7 +243,8 @@ static void test_parse_args_protocol_accept_current() {
 /* Any --protocol value other than the current PROTOCOL_VERSION must end in
  * failure (parse_args simply stores it; validate_config rejects it up front). */
 static void test_parse_args_protocol_rejects_other_versions() {
-  static const char* const bad_versions[] = {"2.16", "2.15.0", "2.16.0", "216", "31", "abc", ""};
+  static const char* const bad_versions[] = {"2.16", "2.15.0", "2.16.0", "2.17.0",
+                                             "216",  "31",     "abc",    ""};
   for (size_t i = 0; i < sizeof(bad_versions) / sizeof(bad_versions[0]); i++) {
     Config* cfg = valid_client_config();
     EXPECT_NOT_NULL(cfg);
@@ -2539,6 +2540,64 @@ static void test_parse_args_chown() {
   config_delete(cfg);
 }
 
+/* --copy-as=USER[:GROUP] (P7 Wave E): resolve the user/group against the local
+ * databases, imply metadata, and apply the documented group-default rule. */
+static void test_parse_args_copy_as() {
+  /* Explicit numeric user and group. */
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--copy-as=@1000:@1001", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->copy_as_set);
+  EXPECT_TRUE(cfg->use_metadata);
+  EXPECT_EQ_INT(cfg->copy_as_uid, 1000);
+  EXPECT_EQ_INT(cfg->copy_as_gid, 1001);
+  config_delete(cfg);
+
+  /* Space form. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv2[] = {"fastsync", "--copy-as", "@2000:3000", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv2, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->copy_as_uid, 2000);
+  EXPECT_EQ_INT(cfg->copy_as_gid, 3000);
+  config_delete(cfg);
+
+  /* Group omitted: a resolvable user uses its primary gid. */
+  struct passwd* self = getpwuid(geteuid());
+  if (self) {
+    cfg = config_create();
+    positional_count = 0;
+    char* argv3[] = {"fastsync", (char*)"--copy-as", (char*)self->pw_name, "/src", "/dst"};
+    EXPECT_EQ_INT(parse_args(cfg, 5, argv3, positional_args, &positional_count), 0);
+    EXPECT_EQ_INT(cfg->copy_as_uid, (int32_t)self->pw_uid);
+    EXPECT_EQ_INT(cfg->copy_as_gid, (int32_t)self->pw_gid);
+    config_delete(cfg);
+  }
+
+  /* Group omitted with a numeric id that has no passwd entry: gid falls back
+   * to uid (documented divergence). */
+  if (!getpwuid((uid_t)4242)) {
+    cfg = config_create();
+    positional_count = 0;
+    char* argv4[] = {"fastsync", "--copy-as=@4242", "/src", "/dst"};
+    EXPECT_EQ_INT(parse_args(cfg, 4, argv4, positional_args, &positional_count), 0);
+    EXPECT_EQ_INT(cfg->copy_as_uid, 4242);
+    EXPECT_EQ_INT(cfg->copy_as_gid, 4242);
+    config_delete(cfg);
+  }
+
+  /* '*' means the client's current euid/egid. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv5[] = {"fastsync", "--copy-as=*:*", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv5, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->copy_as_uid, (int32_t)geteuid());
+  EXPECT_EQ_INT(cfg->copy_as_gid, (int32_t)getegid());
+  config_delete(cfg);
+}
+
 /* Malformed identity specs are rejected, never silently ignored. */
 static void test_parse_args_rejects_malformed_identity() {
   struct {
@@ -2552,6 +2611,12 @@ static void test_parse_args_rejects_malformed_identity() {
       {"--groupmap", "no_such_group_qqq:x"},
       {"--chown", "a:b:c"},
       {"--chown", "no_such_user_zzz:"},
+      {"--copy-as", ""},
+      {"--copy-as", ":"},
+      {"--copy-as", "a:b:c"},
+      {"--copy-as", "@1000:"},
+      {"--copy-as", "definitely_not_a_real_user_zzz"},
+      {"--copy-as", "no_such_group_qqq_group"},
   };
   for (size_t i = 0; i < sizeof(bad) / sizeof(bad[0]); i++) {
     Config* cfg = config_create();
@@ -2569,6 +2634,12 @@ static void test_parse_args_rejects_malformed_identity() {
   int positional_count = 0;
   EXPECT_EQ_INT(parse_args(cfg, 2, argv, positional_args, &positional_count), -1);
   config_delete(cfg);
+
+  cfg = config_create();
+  positional_count = 0;
+  char* argv2[] = {"fastsync", "--copy-as"};
+  EXPECT_EQ_INT(parse_args(cfg, 2, argv2, positional_args, &positional_count), -1);
+  config_delete(cfg);
 }
 
 /* --preallocate parses as a boolean flag and validates cleanly. */
@@ -2972,6 +3043,7 @@ void test_client_cli() {
   test_parse_args_groupmap();
   test_parse_args_usermap_name_resolution();
   test_parse_args_chown();
+  test_parse_args_copy_as();
   test_parse_args_rejects_malformed_identity();
   test_parse_args_preallocate();
   test_parse_args_metadata_times();

tests/test_config.c

@@ -1669,6 +1669,90 @@ static void test_config_receive_rejects_invalid_iconv_spec() {
   }
 }
 
+/* --copy-as (P7 Wave E, protocol 2.18.0) travels as a trailing config-frame
+   block: a presence int, then the two int32 ids when set. */
+static void test_config_copy_as_wire_roundtrip() {
+  struct {
+    bool set;
+    int32_t uid;
+    int32_t gid;
+  } cases[] = {{false, 0, 0}, {true, 1000, 1001}};
+  if (is_running_under_valgrind())
+    return;
+  for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
+    int p[2];
+    EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+    pid_t pid = fork();
+    if (pid == 0) {
+      close(p[1]);
+      io_set_fds(p[0], p[0]);
+      Config* recv = config_receive(p[0]);
+      bool ok = recv != NULL && recv->copy_as_set == cases[i].set &&
+                (!cases[i].set ||
+                 (recv->copy_as_uid == cases[i].uid && recv->copy_as_gid == cases[i].gid));
+      config_delete(recv);
+      close(p[0]);
+      _exit(ok ? 0 : 1);
+    } else {
+      close(p[0]);
+      io_set_fds(p[1], p[1]);
+      Config* send_cfg = config_create();
+      EXPECT_NOT_NULL(send_cfg);
+      send_cfg->send_directory = str_dup("/src");
+      send_cfg->receive_root_directory = str_dup("/dst");
+      send_cfg->copy_as_set = cases[i].set;
+      send_cfg->copy_as_uid = cases[i].uid;
+      send_cfg->copy_as_gid = cases[i].gid;
+      bool sent = config_send(p[1], send_cfg);
+      int status;
+      waitpid(pid, &status, 0);
+      close(p[1]);
+      config_delete(send_cfg);
+      EXPECT_TRUE(sent);
+      EXPECT_TRUE(WIFEXITED(status) && WEXITSTATUS(status) == 0);
+    }
+  }
+}
+
+/* A hostile peer must not smuggle a negative (sentinel) copy-as id into the
+   ownership path: the receive side rejects it and the run fails the handshake. */
+static void test_config_receive_rejects_negative_copy_as() {
+  if (is_running_under_valgrind())
+    return;
+  Config* send_cfg = config_create();
+  EXPECT_NOT_NULL(send_cfg);
+  send_cfg->send_directory = str_dup("/src");
+  send_cfg->receive_root_directory = str_dup("/dst");
+  send_cfg->copy_as_set = true;
+  send_cfg->copy_as_uid = -1;
+  send_cfg->copy_as_gid = 0;
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  pid_t pid = fork();
+  if (pid == 0) {
+    close(p[1]);
+    io_set_fds(p[0], p[0]);
+    Config* recv_cfg = config_receive(p[0]);
+    config_delete(recv_cfg);
+    close(p[0]);
+    _exit(recv_cfg ? 1 : 0);
+  } else {
+    close(p[0]);
+    io_set_fds(p[1], p[1]);
+    bool sent = config_send(p[1], send_cfg);
+    int status;
+    waitpid(pid, &status, 0);
+    close(p[1]);
+    config_delete(send_cfg);
+    EXPECT_FALSE(sent);
+    EXPECT_TRUE(WIFEXITED(status) && WEXITSTATUS(status) == 0);
+  }
+}
+
 void test_config() {
   test_config_lifecycle();
   test_config_ssh_dest();
@@ -1715,6 +1799,8 @@ void test_config() {
     test_config_iconv_spec_wire_roundtrip();
     test_config_iconv_spec_empty_canonicalizes_to_null();
     test_config_receive_rejects_invalid_iconv_spec();
+    test_config_copy_as_wire_roundtrip();
+    test_config_receive_rejects_negative_copy_as();
     test_config_receive_with_validate_rejects();
   }
   test_config_delete_timing_early_helper();