fix(p7-privilege): harden copy-as/super gates, own dirs/specials

- fake-super owner replay honors --no-super and an active --copy-as
- copy-as/identity ownership now applied to directories and special nodes
- reject copy_as_set && !use_metadata (receiver + client --no-preserve)
- daemon refuses --copy-as; add server-side --no-super operator veto
- implement identity_copy_as_refused/identity_copy_as_active
- reject copy-as ids that overflow int32; escape spec in log errors
- copy-as chown EPERM/EACCES logged at ERROR (still non-fatal)
- identity_wire_valid copy-as bounds; CLI help and RSYNC_COMPAT docs
- add unit tests and root-gated integration coverage

tests/integration/test_daemon.py

@@ -320,6 +320,27 @@ class TestDaemonRejection:
         assert result.returncode != 0
         assert _tree_file_count(AUTH_MODULE) == 0
 
+    def test_copy_as_refused_by_daemon(self, daemon):
+        """P7 Wave E: a daemon refuses client-chosen ownership (--copy-as)
+        outright.  There is no per-module opt-in, so even a root daemon must not
+        honor an arbitrary client-selected owner.  The refusal happens at the
+        config handshake, before any data lands."""
+        log_path = os.path.join(TEST_DATA_DIR, "fastsyncd.log")
+        before = os.path.getsize(log_path) if os.path.exists(log_path) else 0
+        before_files = self._tree_files()
+        result, _ = run_client(SOURCE_DIR, "127.0.0.1::files", port=daemon.port,
+                               flags=["--copy-as=@65534:@65534"])
+        assert result.returncode != 0, "the daemon must refuse --copy-as"
+        assert self._tree_files() == before_files, \
+            "--copy-as refusal wrote under the module root"
+        time.sleep(0.3)
+        with open(log_path, "rb") as f:
+            f.seek(before)
+            tail = f.read().decode("utf-8", "replace")
+        assert "copy-as is refused by the daemon" in tail, (
+            f"daemon did not log the copy-as refusal: {tail[-400:]!r}"
+        )
+
     @pytest.mark.daemon_detach
     def test_real_detach_path(self):
         """--daemon WITHOUT --no-detach double-forks a real background daemon;

tests/integration/test_features.py

@@ -4181,6 +4181,24 @@ class TestSuperPrivilege:
         assert (st.st_uid, st.st_gid) == (12345, 12346), \
             f"--super should apply raw ids: uid={st.st_uid} gid={st.st_gid}"
 
+    @pytest.mark.ci
+    @pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
+    def test_fake_super_no_super_does_not_change_owner(self, shared_server):
+        """--fake-super records the source owner, but --no-super must suppress the
+        live chown even for root: the destination keeps the receiver's owner
+        instead of the recorded source owner."""
+        source, dest = self._seed("fakesuper_nosuper")
+        os.chown(os.path.join(source, "f.txt"), 12345, 12346)
+        result, _ = run_client(source, dest,
+                               flags=["--fake-super", "--preserve", "--no-super"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"exit {result.returncode}: {(result.stderr or '')[:300]}"
+        received = get_dest_received_dir(dest, source)
+        st = os.lstat(os.path.join(received, "f.txt"))
+        assert (st.st_uid, st.st_gid) != (12345, 12346), \
+            f"--no-super must suppress fake-super's owner replay: uid={st.st_uid} gid={st.st_gid}"
+
 
 class TestHardLinks:
     """-H/--hard-links: source files sharing an inode are re-created as hard
@@ -5243,3 +5261,83 @@ class TestCopyAs:
         assert (st.st_uid, st.st_gid) == (65534, 65534), (
             f"--copy-as did not force ownership: uid={st.st_uid} gid={st.st_gid}"
         )
+
+    @pytest.mark.ci
+    @pytest.mark.skipif(os.geteuid() != 0, reason="requires a root receiver to chown")
+    def test_root_copy_as_owns_directory(self, shared_server):
+        """--copy-as must own an explicitly-created directory entry, not just the
+        files inside it.  A listed directory (--files-from + --dirs -R) is sent
+        as a STATUS_MKDIR entry, exercising the directory ownership path."""
+        source = os.path.join(TEST_DATA_DIR, "copyas_dir_src")
+        dest = os.path.join(TEST_DATA_DIR, "copyas_dir_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        os.makedirs(os.path.join(source, "owned_dir"), exist_ok=True)
+        lst = os.path.join(TEST_DATA_DIR, "copyas_dir_list.txt")
+        with open(lst, "wb") as fh:
+            fh.write(b"owned_dir\n")
+
+        result, _ = run_client(
+            source, dest,
+            flags=["--copy-as=@65534:@65534", "--files-from", lst, "--dirs", "-R"],
+            port=shared_server.port)
+        assert result.returncode == 0, (
+            f"--copy-as directory transfer failed: {(result.stderr or result.stdout)[:400]}"
+        )
+        target = os.path.join(dest, "owned_dir")
+        assert os.path.isdir(target), f"explicit directory missing at {target}"
+        st = os.stat(target)
+        assert (st.st_uid, st.st_gid) == (65534, 65534), (
+            f"--copy-as did not own the directory: uid={st.st_uid} gid={st.st_gid}"
+        )
+
+    @pytest.mark.ci
+    @pytest.mark.skipif(os.geteuid() != 0, reason="requires a root receiver to chown")
+    def test_root_copy_as_owns_fifo(self, shared_server):
+        """--copy-as must own a recreated FIFO special node."""
+        source = os.path.join(TEST_DATA_DIR, "copyas_fifo_src")
+        dest = os.path.join(TEST_DATA_DIR, "copyas_fifo_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        os.mkfifo(os.path.join(source, "pipe.fifo"))
+
+        result, _ = run_client(source, dest,
+                               flags=["--copy-as=@65534:@65534", "--specials"],
+                               port=shared_server.port)
+        assert result.returncode == 0, (
+            f"--copy-as FIFO transfer failed: {(result.stderr or result.stdout)[:400]}"
+        )
+        received = get_dest_received_dir(dest, source)
+        target = os.path.join(received, "pipe.fifo")
+        assert stat.S_ISFIFO(os.lstat(target).st_mode), f"FIFO missing at {target}"
+        st = os.lstat(target)
+        assert (st.st_uid, st.st_gid) == (65534, 65534), (
+            f"--copy-as did not own the FIFO: uid={st.st_uid} gid={st.st_gid}"
+        )
+
+    @pytest.mark.ci
+    @pytest.mark.skipif(os.geteuid() != 0, reason="requires a root receiver to chown")
+    def test_root_copy_as_with_fake_super_keeps_target_owner(self, shared_server):
+        """--fake-super must not let the recorded source owner override the
+        --copy-as forced owner (copy-as is authoritative)."""
+        source = os.path.join(TEST_DATA_DIR, "copyas_fakesuper_src")
+        dest = os.path.join(TEST_DATA_DIR, "copyas_fakesuper_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        src_file = os.path.join(source, "mixed.txt")
+        with open(src_file, "wb") as fh:
+            fh.write(b"copy-as wins over fake-super\n")
+        os.chown(src_file, 12345, 12346)
+
+        result, _ = run_client(source, dest,
+                               flags=["--copy-as=@65534:@65534", "--fake-super"],
+                               port=shared_server.port)
+        assert result.returncode == 0, (
+            f"--copy-as --fake-super transfer failed: "
+            f"{(result.stderr or result.stdout)[:400]}"
+        )
+        received = get_dest_received_dir(dest, source)
+        st = os.lstat(os.path.join(received, "mixed.txt"))
+        assert (st.st_uid, st.st_gid) == (65534, 65534), (
+            f"--fake-super overrode --copy-as: uid={st.st_uid} gid={st.st_gid}"
+        )

tests/test_client_cli.c

@@ -243,8 +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.17", "2.16",   "2.15.0", "2.16.0", "2.17.0",
-                                             "216",  "31",     "abc",    ""};
+  static const char* const bad_versions[] = {"2.17", "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);

tests/test_config.c

@@ -1741,6 +1741,10 @@ static void test_config_copy_as_wire_roundtrip() {
       send_cfg->copy_as_set = cases[i].set;
       send_cfg->copy_as_uid = cases[i].uid;
       send_cfg->copy_as_gid = cases[i].gid;
+      /* --copy-as requires the metadata path (the receiver chowns from the
+         transmitted source ids); a raw frame with copy_as_set but no metadata
+         is now rejected by validate_received_config. */
+      send_cfg->use_metadata = cases[i].set;
       bool sent = config_send(p[1], send_cfg);
       int status;
       waitpid(pid, &status, 0);
@@ -1814,6 +1818,63 @@ static void test_config_receive_rejects_negative_copy_as() {
   }
 }
 
+/* --copy-as forces ownership through the metadata path.  A frame that sets
+   copy_as_set but not use_metadata would pass the receiver's privilege gate
+   while chowning nothing, so validate_received_config must reject it (and the
+   sender observes the rejection as a failed config_send). */
+static void test_config_receive_rejects_copy_as_without_metadata() {
+  if (is_running_under_valgrind())
+    return;
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->copy_as_set = true;
+  c->copy_as_uid = 1000;
+  c->copy_as_gid = 1000;
+  c->use_metadata = false;
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+
+  /* With metadata enabled the same block is accepted. */
+  c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->copy_as_set = true;
+  c->copy_as_uid = 1000;
+  c->copy_as_gid = 1000;
+  c->use_metadata = true;
+  EXPECT_TRUE(roundtrip_config_ok(c));
+  config_delete(c);
+}
+
+/* identity_copy_as_refused() is the pure, pre-snapshot refusal predicate: a
+   --copy-as is refused when the receiver is not root OR the effective super
+   mode is OFF (an operator veto), and never when --copy-as is unset. */
+static void test_identity_copy_as_refused() {
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  EXPECT_FALSE(identity_copy_as_refused(c));
+  EXPECT_FALSE(identity_copy_as_refused(NULL));
+
+  c->copy_as_set = true;
+  c->super_mode = SUPER_MODE_AUTO;
+  if (geteuid() == 0) {
+    EXPECT_FALSE(identity_copy_as_refused(c)); /* AUTO permits as root */
+    c->super_mode = SUPER_MODE_ON;
+    EXPECT_FALSE(identity_copy_as_refused(c));
+    c->super_mode = SUPER_MODE_OFF;
+    EXPECT_TRUE(identity_copy_as_refused(c));
+  } else {
+    /* Unprivileged: refused regardless of the mode. */
+    EXPECT_TRUE(identity_copy_as_refused(c));
+    c->super_mode = SUPER_MODE_OFF;
+    EXPECT_TRUE(identity_copy_as_refused(c));
+  }
+  config_delete(c);
+}
+
 /* P7 Wave E: privilege_super_permitted() maps the super_mode tri-state.  OFF
    forbids super-user activities even for root; ON and AUTO permit the confined
    attempt (matching FastSync's historical best-effort behavior, where the kernel
@@ -1888,8 +1949,10 @@ void test_config() {
     test_config_receive_rejects_invalid_super_mode();
     test_config_copy_as_wire_roundtrip();
     test_config_receive_rejects_negative_copy_as();
+    test_config_receive_rejects_copy_as_without_metadata();
     test_config_receive_with_validate_rejects();
   }
+  test_identity_copy_as_refused();
   test_privilege_super_permitted_modes();
   test_config_delete_timing_early_helper();
   test_config_is_remote_dest();

tests/test_server_cli.c

@@ -31,9 +31,28 @@ static void test_server_cli_defaults() {
   EXPECT_EQ_INT(opts.bind_family, AF_UNSPEC);
   EXPECT_FALSE(opts.allow_delete);
   EXPECT_FALSE(opts.allow_unauthenticated);
+  EXPECT_FALSE(opts.no_super);
   server_cli_options_free(&opts);
 }
 
+/* --no-super is a standalone/SSH operator veto (does not require --daemon):
+   it forces SUPER_MODE_OFF for every connection and refuses client --copy-as. */
+static void test_server_cli_no_super() {
+  const char* args[] = {"fastsync-server", "--no-super", "--destination-root", "/srv"};
+  ServerCliOptions opts;
+  EXPECT_EQ_INT(parse_ok(args, 4, &opts), 0);
+  EXPECT_TRUE(opts.no_super);
+  EXPECT_EQ_STR(opts.destination_root, "/srv");
+  server_cli_options_free(&opts);
+
+  const char* args2[] = {"fastsync-server", "--daemon", "--config=/tmp/x.conf", "--no-super"};
+  ServerCliOptions opts2;
+  EXPECT_EQ_INT(parse_ok(args2, 4, &opts2), 0);
+  EXPECT_TRUE(opts2.no_super);
+  EXPECT_TRUE(opts2.daemon_mode);
+  server_cli_options_free(&opts2);
+}
+
 static void test_server_cli_daemon_flags() {
   const char* args[] = {"fastsync-server", "--daemon", "--no-detach", "--allow-unauthenticated"};
   ServerCliOptions opts;
@@ -197,5 +216,6 @@ void test_server_cli() {
   test_server_cli_invalid();
   test_server_cli_password_and_early_input();
   test_server_cli_password_requires_daemon();
+  test_server_cli_no_super();
   test_server_cli_help();
 }

tests/test_xattr.c

@@ -1,6 +1,8 @@
 #include "test_xattr.h"
 #include "xattr.h"
+#include "config.h"
 #include "file.h"
+#include "identity.h"
 #include "protocol.h"
 #include "test_utils.h"
 #include <fcntl.h>
@@ -273,6 +275,72 @@ static void test_fake_super_restore() {
   unlink(path);
 }
 
+/* --fake-super owner replay must honor the super gate and copy-as authority:
+   --no-super suppresses the recorded-source-owner chown even for root, and an
+   active --copy-as keeps its forced owner (the recorded source owner must never
+   override it).  Root-gated: only root can observe a chown actually landing. */
+static void test_fake_super_owner_gate() {
+  if (geteuid() != 0)
+    return; /* non-root cannot observe ownership changes; skip silently */
+  const char* path = "test_fake_super_owner_gate.txt";
+  unlink(path);
+  int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+  if (fd < 0)
+    return;
+  bool has_xattr = setxattr(path, "user.fastsync.xprobe", "p", 1, 0) == 0;
+  if (has_xattr)
+    removexattr(path, "user.fastsync.xprobe");
+  if (!has_xattr) {
+    close(fd);
+    unlink(path);
+    return; /* filesystem without xattr support */
+  }
+  if (fchown(fd, 0, 0) != 0) {
+    close(fd);
+    unlink(path);
+    return;
+  }
+  fake_super_store_fd(fd, 12345, 12346, 0755, 1700000000, 0);
+
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+
+  /* --no-super: the owner leg is skipped even as root. */
+  c->super_mode = SUPER_MODE_OFF;
+  identity_set_active(c);
+  EXPECT_TRUE(fake_super_restore_fd(fd));
+  struct stat st;
+  EXPECT_EQ_INT(fstat(fd, &st), 0);
+  EXPECT_EQ_INT((int)st.st_uid, 0);
+  EXPECT_EQ_INT((int)st.st_gid, 0);
+
+  /* AUTO: the recorded source owner is applied. */
+  c->super_mode = SUPER_MODE_AUTO;
+  identity_set_active(c);
+  EXPECT_TRUE(fake_super_restore_fd(fd));
+  EXPECT_EQ_INT(fstat(fd, &st), 0);
+  EXPECT_EQ_INT((int)st.st_uid, 12345);
+  EXPECT_EQ_INT((int)st.st_gid, 12346);
+
+  /* Active --copy-as is authoritative: the recorded source owner must not
+     override it, even with AUTO/ON. */
+  EXPECT_EQ_INT(fchown(fd, 0, 0), 0);
+  c->super_mode = SUPER_MODE_ON;
+  c->copy_as_set = true;
+  c->copy_as_uid = 777;
+  c->copy_as_gid = 778;
+  identity_set_active(c);
+  EXPECT_TRUE(fake_super_restore_fd(fd));
+  EXPECT_EQ_INT(fstat(fd, &st), 0);
+  EXPECT_EQ_INT((int)st.st_uid, 0);
+  EXPECT_EQ_INT((int)st.st_gid, 0);
+
+  identity_clear_active();
+  config_delete(c);
+  close(fd);
+  unlink(path);
+}
+
 void test_xattr() {
   test_xattr_wire_roundtrip();
   test_xattr_reject_privileged_namespace();
@@ -281,4 +349,5 @@ void test_xattr() {
   test_xattr_capture_and_appliable();
   test_link_copy_fallback_preserves_xattrs();
   test_fake_super_restore();
+  test_fake_super_owner_gate();
 }