feat(p7-super): implement --super/--no-super safe-subset privilege gate (protocol 2.18.0)

Add the receiver-side --super / --no-super tri-state (Config->super_mode)
under the safe-subset + clear-refusal privilege model: FastSync never
elevates privileges, it only permits super-user attempts that are already
confined fd-relative below the authorized receive root.

- identity: privilege_super_permitted() gate (OFF=false, ON=true, AUTO follows
  geteuid()==0); identity_apply_ownership/_link become no-ops when not
  permitted; --super with no explicit identity policy implies raw numeric-id
  preservation (explicit usermap/groupmap/chown/numeric-ids still win); warn
  exactly once when --super is requested by a non-root receiver.
- file_receive: gate char/block device-node creation on the gate; FIFO/socket
  handling is unchanged.
- wire: trailing super_mode int after the --iconv spec, validated 0..2 in
  receive_privilege_options and validate_received_config; PROTOCOL_VERSION
  2.17.0 -> 2.18.0; version-sensitive tests and docs updated.
- CLI: --super/--no-super parsed explicitly before the generic --no-* branch
  (malformed --super=x rejected); usage text added.
- tests: config wire round-trip + invalid-value rejection, privilege-gate mode
  unit test, CLI parse test, integration transfer + root-gated ownership
  suppression/appliance tests.
- docs: RSYNC_COMPAT --super row + Wave E note, protocol mentions, README.
This commit is contained in:
2026-09-12 12:01:19 +02:00
parent f64d252faf
commit a785ec13c4
13 changed files with 366 additions and 26 deletions
+65 -2
View File
@@ -200,8 +200,9 @@ class TestDeviceSpecial:
assert not os.path.lexists(os.path.join(received, "chardev")), (
"a receiver without CAP_MKNOD must skip the device node, not create it"
)
assert "cannot create device node" in (out + err), (
f"receiver did not log the documented CAP_MKNOD skip: out={out!r} err={err!r}"
assert ("cannot create device node" in (out + err)
or "device-node creation is not permitted" in (out + err)), (
f"receiver did not log the documented device skip: out={out!r} err={err!r}"
)
@pytest.mark.skipif(os.geteuid() != 0, reason="requires root to create device nodes")
@@ -4119,6 +4120,68 @@ class TestIdentityMapping:
f"--chown not applied: uid={st.st_uid} gid={st.st_gid}"
class TestSuperPrivilege:
"""P7 Wave E: --super / --no-super control the receiver's already-confined
super-user activities (ownership application, char/block device nodes).
FastSync never elevates, so on an unprivileged receiver --super only
permits a confined attempt (which then skips); --no-super forbids the
activity even for root."""
def _seed(self, tag):
source = os.path.join(TEST_DATA_DIR, f"super_{tag}_source")
dest = os.path.join(TEST_DATA_DIR, f"super_{tag}_dest")
clean_dir(source)
clean_dir(dest)
with open(os.path.join(source, "f.txt"), "wb") as f:
f.write(b"super privilege\n")
return source, dest
def test_super_and_no_super_transfer_successfully(self, shared_server):
"""Both flags parse and the transfer completes normally regardless of
the receiver's privilege level."""
for flag in ("--super", "--no-super"):
source, dest = self._seed(flag.strip("-"))
result, _ = run_client(source, dest, flags=[flag], port=shared_server.port)
assert result.returncode == 0, \
f"{flag} exit {result.returncode}: {(result.stderr or '')[:300]}"
received = get_dest_received_dir(dest, source)
with open(os.path.join(received, "f.txt"), "rb") as f:
assert f.read() == b"super privilege\n"
@pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
def test_no_super_suppresses_ownership_as_root(self, shared_server):
"""As root the default gate would apply a raw numeric id; --no-super
must suppress that ownership application entirely."""
source, dest = self._seed("nosuper")
os.chown(os.path.join(source, "f.txt"), 12345, 12346)
result, _ = run_client(source, dest,
flags=["--preserve", "--numeric-ids", "--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.stat(os.path.join(received, "f.txt"))
assert (st.st_uid, st.st_gid) != (12345, 12346), \
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")
os.chown(os.path.join(source, "f.txt"), 12345, 12346)
result, _ = run_client(source, dest,
flags=["--preserve", "--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}"
class TestHardLinks:
"""-H/--hard-links: source files sharing an inode are re-created as hard
links to one another on the destination (dedup preserved, first copy
+3 -3
View File
@@ -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.17.0", "2.15.0", "2.16.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"