fix(p8-security): enforce daemon ownership policy, gate fake-super replay, drop implicit numeric-ids, make copy-as failures per-entry

A1: daemon refuses every client-chosen ownership/super-user request
(--numeric-ids/--chown/--usermap/--groupmap/--fake-super/--copy-as/--super)
unless the selected module opts in with 'client owner = yes'.
A2: fake-super owner replay requires an explicit ownership identity policy.
A3: --super no longer implies --numeric-ids (ownership stays opt-in).
A5: a failed --copy-as chown marks the entry failed instead of reporting
success with the wrong owner.

tests/integration/test_daemon.py

@@ -41,6 +41,7 @@ FILES_MODULE = os.path.join(MODULE_ROOT, "files")
 READONLY_MODULE = os.path.join(MODULE_ROOT, "readonly")
 AUTH_MODULE = os.path.join(MODULE_ROOT, "auth")
 TEAM_MODULE = os.path.join(MODULE_ROOT, "team")
+OWNER_MODULE = os.path.join(MODULE_ROOT, "owner")
 CONF_FILE = os.path.join(TEST_DATA_DIR, "fastsyncd.conf")
 CRED_FILE = os.path.join(TEST_DATA_DIR, "fastsyncd.passwd")
 STARTFAIL_CONF = os.path.join(TEST_DATA_DIR, "fastsyncd_startfail.conf")
@@ -149,7 +150,8 @@ def _config_port(config_path):
 
 @pytest.fixture(scope="module", autouse=True)
 def daemon_env():
-    for d in (MODULE_ROOT, FILES_MODULE, READONLY_MODULE, AUTH_MODULE, TEAM_MODULE, DETACH_MODULE):
+    for d in (MODULE_ROOT, FILES_MODULE, READONLY_MODULE, AUTH_MODULE, TEAM_MODULE, OWNER_MODULE,
+              DETACH_MODULE):
         shutil.rmtree(d, ignore_errors=True)
         os.makedirs(d, exist_ok=True)
     generate_test_files(SOURCE_DIR, full=False)
@@ -184,7 +186,11 @@ def daemon_env():
             "[team]\n"
             "path = %s\n"
             "auth users = alice,bob\n"
-            % (config_port, FILES_MODULE, READONLY_MODULE, AUTH_MODULE, TEAM_MODULE))
+            "\n"
+            "[owner]\n"
+            "path = %s\n"
+            "client owner = yes\n"
+            % (config_port, FILES_MODULE, READONLY_MODULE, AUTH_MODULE, TEAM_MODULE, OWNER_MODULE))
 
     # A dedicated config for the fail-closed startup check: an auth-required
     # module with no credential store must refuse to start.  Its own free port
@@ -320,49 +326,62 @@ 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}"
-        )
-
-    def test_super_refused_by_daemon(self, daemon):
-        """P7 Wave E: --super (SUPER_MODE_ON) implies raw numeric-id ownership
-        with no explicit identity flag, so a daemon refuses it for the same
-        reason it refuses --copy-as: there is no per-module opt-in for
-        client-chosen ownership.  The refusal happens at the config handshake,
+    def _assert_ownership_refused(self, daemon, module, flags):
+        """A daemon module without `client owner = yes` refuses every
+        client-chosen ownership / super-user request 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=["--super", "--preserve"])
-        assert result.returncode != 0, "the daemon must refuse --super"
+        result, _ = run_client(SOURCE_DIR, f"127.0.0.1::{module}", port=daemon.port, flags=flags)
+        assert result.returncode != 0, f"the daemon must refuse {flags}"
         assert self._tree_files() == before_files, \
-            "--super refusal wrote under the module root"
+            f"{flags} 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 "super is refused by the daemon" in tail, (
-            f"daemon did not log the --super refusal: {tail[-400:]!r}"
+        assert "client-chosen ownership" in tail, (
+            f"daemon did not log the ownership refusal: {tail[-400:]!r}"
         )
 
+    def test_copy_as_refused_by_daemon(self, daemon):
+        """P7 Wave E hardening: a daemon refuses client-chosen ownership
+        (--copy-as) outright unless the module opts in with `client owner = yes`,
+        so even a root daemon must not honor an arbitrary client-selected owner
+        by default.  The refusal happens at the config handshake, before any data
+        lands."""
+        self._assert_ownership_refused(daemon, "files", ["--copy-as=@65534:@65534"])
+
+    def test_super_refused_by_daemon(self, daemon):
+        """An explicit --super is a super-user activity request, so a daemon
+        module refuses it unless it opts in with `client owner = yes`.  The
+        refusal happens at the config handshake, before any data lands."""
+        self._assert_ownership_refused(daemon, "files", ["--super", "--preserve"])
+
+    def test_numeric_ids_refused_by_daemon(self, daemon):
+        """P7 Wave E hardening (A1): the daemon ownership gate must cover the
+        pre-existing identity flags too, not only --copy-as/--super.  A module
+        without `client owner = yes` refuses --numeric-ids at the handshake."""
+        self._assert_ownership_refused(daemon, "files", ["--numeric-ids", "--preserve"])
+
+    def test_chown_refused_by_daemon(self, daemon):
+        """--chown is client-chosen ownership too and must be refused by a
+        non-opted-in module."""
+        self._assert_ownership_refused(daemon, "files", ["--chown=@65534:@65534", "--preserve"])
+
+    def test_owner_opt_in_allows_numeric_ids(self, daemon):
+        """A module that opts in with `client owner = yes` accepts the
+        client-chosen ownership flags (here --numeric-ids); the transfer
+        succeeds and lands inside that module root."""
+        result, _ = run_client(SOURCE_DIR, "127.0.0.1::owner", port=daemon.port,
+                               flags=["--numeric-ids", "--preserve"])
+        assert result.returncode == 0, result.stderr or result.stdout
+        received = get_dest_received_dir(OWNER_MODULE, SOURCE_DIR)
+        mismatches, missing = verify_transfer(SOURCE_DIR, received)
+        assert not missing, f"missing: {missing[:5]}"
+        assert not mismatches, f"mismatch: {mismatches[:5]}"
+
     @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

@@ -4165,11 +4165,11 @@ class TestSuperPrivilege:
             f"--no-super must not apply ownership (uid={st.st_uid} gid={st.st_gid})"
 
     @pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
-    def test_super_applies_ownership_as_root(self, shared_server):
-        """Control/proof the flag is not inert for root: --super with no explicit
-        identity policy treats ownership as raw numeric ids (as --numeric-ids),
-        applying the very ownership --no-super suppressed."""
-        source, dest = self._seed("super")
+    def test_super_alone_does_not_apply_ownership_as_root(self, shared_server):
+        """A3: --super no longer implies --numeric-ids, so --super alone must NOT
+        apply client-chosen ownership even for root; the destination keeps the
+        receiver's owner (the exact ownership --no-super would also suppress)."""
+        source, dest = self._seed("superonly")
         os.chown(os.path.join(source, "f.txt"), 12345, 12346)
         result, _ = run_client(source, dest,
                                flags=["--preserve", "--super"],
@@ -4178,8 +4178,25 @@ class TestSuperPrivilege:
             f"exit {result.returncode}: {(result.stderr or '')[:300]}"
         received = get_dest_received_dir(dest, source)
         st = os.stat(os.path.join(received, "f.txt"))
+        assert (st.st_uid, st.st_gid) != (12345, 12346), \
+            f"--super alone must not apply ownership (uid={st.st_uid} gid={st.st_gid})"
+
+    @pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
+    def test_super_with_numeric_ids_applies_ownership_as_root(self, shared_server):
+        """Control: an explicit identity policy is what enables ownership, so
+        --numeric-ids --super still applies the raw ids as root (the very
+        ownership --no-super suppresses)."""
+        source, dest = self._seed("supernumeric")
+        os.chown(os.path.join(source, "f.txt"), 12345, 12346)
+        result, _ = run_client(source, dest,
+                               flags=["--preserve", "--numeric-ids", "--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.stat(os.path.join(received, "f.txt"))
         assert (st.st_uid, st.st_gid) == (12345, 12346), \
-            f"--super should apply raw ids: uid={st.st_uid} gid={st.st_gid}"
+            f"--numeric-ids --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")

tests/test_config.c

@@ -1899,6 +1899,68 @@ static void test_privilege_super_permitted_modes() {
   EXPECT_TRUE(privilege_super_permitted());
 }
 
+/* P7 Wave E hardening (A1): identity_ownership_requested() is the pure,
+   config-only predicate the daemon module gate uses.  It must fire for every
+   client-chosen ownership / super-user request and stay false for a plain
+   transfer and for SUPER_MODE_AUTO (the default) alone. */
+static void test_identity_ownership_requested() {
+  EXPECT_FALSE(identity_ownership_requested(NULL));
+
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  EXPECT_FALSE(identity_ownership_requested(c));
+  c->super_mode = SUPER_MODE_AUTO;
+  EXPECT_FALSE(identity_ownership_requested(c)); /* AUTO alone is not ownership */
+  c->super_mode = SUPER_MODE_ON;
+  EXPECT_TRUE(identity_ownership_requested(c)); /* explicit --super is */
+  c->super_mode = SUPER_MODE_AUTO;
+
+  c->numeric_ids = true;
+  EXPECT_TRUE(identity_ownership_requested(c));
+  c->numeric_ids = false;
+  c->chown_uid_set = true;
+  EXPECT_TRUE(identity_ownership_requested(c));
+  c->chown_uid_set = false;
+  c->chown_gid_set = true;
+  EXPECT_TRUE(identity_ownership_requested(c));
+  c->chown_gid_set = false;
+  c->copy_as_set = true;
+  EXPECT_TRUE(identity_ownership_requested(c));
+  c->copy_as_set = false;
+  c->fake_super = true;
+  EXPECT_TRUE(identity_ownership_requested(c));
+  config_delete(c);
+
+  Config* um = config_create();
+  EXPECT_NOT_NULL(um);
+  EXPECT_EQ_INT(identity_parse_map(um, "@1:@2", false), 0);
+  EXPECT_TRUE(identity_ownership_requested(um));
+  config_delete(um);
+
+  Config* gm = config_create();
+  EXPECT_NOT_NULL(gm);
+  EXPECT_EQ_INT(identity_parse_map(gm, "@1:@2", true), 0);
+  EXPECT_TRUE(identity_ownership_requested(gm));
+  config_delete(gm);
+}
+
+/* P7 Wave E hardening (A3): --super no longer implies raw numeric-id
+   preservation, so it must never enable ownership application on its own; an
+   explicit identity flag is required. */
+static void test_super_does_not_imply_numeric() {
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->super_mode = SUPER_MODE_ON;
+  c->use_metadata = true;
+  identity_set_active(c);
+  EXPECT_FALSE(identity_active_enabled());
+  c->numeric_ids = true;
+  identity_set_active(c);
+  EXPECT_TRUE(identity_active_enabled());
+  identity_clear_active();
+  config_delete(c);
+}
+
 void test_config() {
   test_config_lifecycle();
   test_config_ssh_dest();
@@ -1953,6 +2015,8 @@ void test_config() {
     test_config_receive_with_validate_rejects();
   }
   test_identity_copy_as_refused();
+  test_identity_ownership_requested();
+  test_super_does_not_imply_numeric();
   test_privilege_super_permitted_modes();
   test_config_delete_timing_early_helper();
   test_config_is_remote_dest();

tests/test_daemon_conf.c

@@ -43,6 +43,7 @@ static void test_daemon_conf_full_parse() {
                            "[backup]\n"
                            "path = /srv/backup\n"
                            "read only = yes\n"
+                           "client owner = yes\n"
                            "auth users = alice, bob\n",
                            &path),
                 0);
@@ -57,6 +58,7 @@ static void test_daemon_conf_full_parse() {
   EXPECT_EQ_STR(conf->modules[0].name, "backup");
   EXPECT_EQ_STR(conf->modules[0].path, "/srv/backup");
   EXPECT_TRUE(conf->modules[0].read_only);
+  EXPECT_TRUE(conf->modules[0].client_owner);
   EXPECT_EQ_INT(conf->modules[0].auth_user_count, 2);
   EXPECT_EQ_STR(conf->modules[0].auth_users[0], "alice");
   EXPECT_EQ_STR(conf->modules[0].auth_users[1], "bob");
@@ -84,6 +86,10 @@ static void test_daemon_conf_comments_and_blank_lines() {
   EXPECT_EQ_INT(conf->module_count, 2);
   EXPECT_EQ_STR(conf->modules[0].name, "alpha");
   EXPECT_EQ_STR(conf->modules[1].name, "beta");
+  /* `client owner` defaults to off: a module must opt in to client-chosen
+     ownership. */
+  EXPECT_FALSE(conf->modules[0].client_owner);
+  EXPECT_FALSE(conf->modules[1].client_owner);
   daemon_conf_free(conf);
 }
 
@@ -207,6 +213,12 @@ static void test_daemon_conf_malformed_rejected() {
   EXPECT_NULL(conf);
   EXPECT_TRUE(strstr(err, "read only") != NULL);
 
+  EXPECT_EQ_INT(write_conf("[m]\npath = /x\nclient owner = maybe\n", &path), 0);
+  conf = daemon_conf_load(path, err, sizeof(err));
+  free(path);
+  EXPECT_NULL(conf);
+  EXPECT_TRUE(strstr(err, "client owner") != NULL);
+
   EXPECT_EQ_INT(write_conf("= value\n", &path), 0);
   conf = daemon_conf_load(path, err, sizeof(err));
   free(path);

tests/test_xattr.c

@@ -305,6 +305,10 @@ static void test_fake_super_owner_gate() {
   Config* c = config_create();
   EXPECT_NOT_NULL(c);
 
+  /* An explicit ownership policy is required before fake-super replay may
+     chown; --fake-super alone only records the source owner (A2). */
+  c->numeric_ids = true;
+
   /* --no-super: the owner leg is skipped even as root. */
   c->super_mode = SUPER_MODE_OFF;
   identity_set_active(c);
@@ -314,7 +318,7 @@ static void test_fake_super_owner_gate() {
   EXPECT_EQ_INT((int)st.st_uid, 0);
   EXPECT_EQ_INT((int)st.st_gid, 0);
 
-  /* AUTO: the recorded source owner is applied. */
+  /* AUTO with an identity policy: the recorded source owner is applied. */
   c->super_mode = SUPER_MODE_AUTO;
   identity_set_active(c);
   EXPECT_TRUE(fake_super_restore_fd(fd));
@@ -322,10 +326,19 @@ static void test_fake_super_owner_gate() {
   EXPECT_EQ_INT((int)st.st_uid, 12345);
   EXPECT_EQ_INT((int)st.st_gid, 12346);
 
+  /* --super / --fake-super with NO explicit identity flag must NOT apply a
+     client-chosen owner: super_mode alone never enables ownership. */
+  EXPECT_EQ_INT(fchown(fd, 0, 0), 0);
+  c->numeric_ids = false;
+  c->super_mode = SUPER_MODE_ON;
+  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);
+
   /* 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;