Fix a TOCTOU in remote harvest zipapp

Promote uploaded zipapps into a private root-owned directory before
verification and execution. Copy and hash through the same pinned file
descriptor, reject unsafe file types and metadata, publish atomically,
and ensure sudo executes only the verified root-owned copy.

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.8.1
+
+ * Security: fix a TOCTOU in remote harvest.
+
 # 0.8.0
 
  * Security: keep sudo-created remote harvest bundles root-owned while root packages and hashes them, expose only the archive to the authenticated SSH uid, and verify the root-computed digest after download. This removes the post-harvest tampering window created by recursively chowning the bundle before packaging without making the plaintext archive world-readable.

debian/changelog

@@ -1,3 +1,9 @@
+enroll (0.8.1) unstable; urgency=medium
+
+  * Security: fix a TOCTOU in remote harvest.
+
+ -- Miguel Jacq <mig@mig5.net>  Mon, 03 Aug 2026 15:20:00 +1000
+
 enroll (0.8.0) unstable; urgency=medium
 
   * Security: keep sudo-created remote harvest bundles root-owned while root packages and hashes them, expose only the archive to the authenticated SSH uid, and verify the root-computed digest after download. This removes the post-harvest tampering window created by recursively chowning the bundle before packaging without making the plaintext archive world-readable.

enroll/remote.py

@@ -300,8 +300,9 @@ def _build_enroll_pyz(tmpdir: Path) -> tuple[Path, str]:
 
     Returns ``(pyz_path, sha256_hex)``. The digest is computed on the exact
     bytes written locally so the caller can verify, on the remote side, that the
-    file that is about to be executed as root is byte-for-byte the one we built
+    file that is executed is byte-for-byte the one we built (see
-    (see ``_remote_verify_pyz_sha256``). This is transport/staging integrity
+    ``_remote_promote_verified_pyz`` and ``_remote_verify_pyz_sha256``). This is
+    transport/staging integrity
     defence-in-depth: it detects a swap of the staged file between upload and
     execution by anyone who gained write access to the staging directory. It is
     NOT a defence against a remote host that is already root-compromised -- such
@@ -461,6 +462,243 @@ def _remote_file_sha256_sudo(
     return digest
 
 
+_REMOTE_PROMOTE_PYZ_SCRIPT = r"""import hashlib
+import os
+import stat
+import sys
+
+def _promote(argv):
+    (
+        stage_dir,
+        source_name,
+        root_dir,
+        destination_name,
+        expected,
+        size_text,
+        owner_uid_text,
+    ) = argv
+    expected_size = int(size_text)
+    expected_owner_uid = int(owner_uid_text)
+
+    for required_flag in ("O_DIRECTORY", "O_NOFOLLOW", "O_NONBLOCK"):
+        if not hasattr(os, required_flag):
+            raise RuntimeError(
+                "remote platform lacks required safe-open flag " + required_flag
+            )
+
+    if os.geteuid() != expected_owner_uid:
+        raise RuntimeError("promotion helper is not running as the expected user")
+    if not source_name or source_name in {".", ".."} or "/" in source_name:
+        raise RuntimeError("uploaded enroll.pyz name is not a single path component")
+
+    no_follow = os.O_NOFOLLOW
+    cloexec = getattr(os, "O_CLOEXEC", 0)
+    dir_flags = os.O_RDONLY | os.O_DIRECTORY | no_follow | cloexec
+    stage_fd = None
+    root_fd = None
+    source_fd = None
+    destination_fd = None
+    temporary_name = ".enroll.pyz.tmp"
+    temporary_created = False
+    destination_published = False
+    promotion_complete = False
+
+    try:
+        stage_fd = os.open(stage_dir, dir_flags)
+        root_fd = os.open(root_dir, dir_flags)
+
+        root_stat = os.fstat(root_fd)
+        if not stat.S_ISDIR(root_stat.st_mode):
+            raise RuntimeError("destination path is not a directory")
+        if root_stat.st_uid != expected_owner_uid:
+            raise RuntimeError("destination directory has an unexpected owner")
+        if stat.S_IMODE(root_stat.st_mode) != 0o700:
+            raise RuntimeError("destination directory is not mode 0700")
+
+        source_fd = os.open(
+            source_name,
+            os.O_RDONLY | no_follow | os.O_NONBLOCK | cloexec,
+            dir_fd=stage_fd,
+        )
+        source_stat = os.fstat(source_fd)
+        if not stat.S_ISREG(source_stat.st_mode):
+            raise RuntimeError("uploaded enroll.pyz is not a regular file")
+        if source_stat.st_nlink != 1:
+            raise RuntimeError("uploaded enroll.pyz must not be hard-linked")
+        if source_stat.st_size != expected_size:
+            raise RuntimeError("uploaded enroll.pyz has an unexpected size")
+
+        # The directory was freshly created by root, so neither name should exist.
+        # O_EXCL protects the temporary name. Publishing with link() instead of
+        # rename() also fails rather than replacing an unexpected final path.
+        destination_fd = os.open(
+            temporary_name,
+            os.O_WRONLY | os.O_CREAT | os.O_EXCL | no_follow | cloexec,
+            0o500,
+            dir_fd=root_fd,
+        )
+        temporary_created = True
+
+        digest = hashlib.sha256()
+        copied = 0
+        while True:
+            # Read one byte beyond the expected length so concurrent growth cannot
+            # be hidden by stopping exactly at expected_size.
+            chunk = os.read(source_fd, min(1048576, expected_size - copied + 1))
+            if not chunk:
+                break
+            copied += len(chunk)
+            if copied > expected_size:
+                raise RuntimeError("uploaded enroll.pyz grew while being copied")
+            digest.update(chunk)
+
+            view = memoryview(chunk)
+            while view:
+                written = os.write(destination_fd, view)
+                if written <= 0:
+                    raise RuntimeError("short write while promoting enroll.pyz")
+                view = view[written:]
+
+        if copied != expected_size:
+            raise RuntimeError("uploaded enroll.pyz changed size while being copied")
+
+        actual = digest.hexdigest()
+        if actual != expected:
+            raise RuntimeError(
+                "uploaded enroll.pyz SHA-256 mismatch: expected "
+                + expected
+                + ", received "
+                + actual
+            )
+
+        os.fchmod(destination_fd, 0o500)
+        os.fsync(destination_fd)
+        os.close(destination_fd)
+        destination_fd = None
+
+        os.link(
+            temporary_name,
+            destination_name,
+            src_dir_fd=root_fd,
+            dst_dir_fd=root_fd,
+            follow_symlinks=False,
+        )
+        destination_published = True
+        os.unlink(temporary_name, dir_fd=root_fd)
+        temporary_created = False
+        os.fsync(root_fd)
+        promotion_complete = True
+    finally:
+        for fd in (destination_fd, source_fd, stage_fd):
+            if fd is not None:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+
+        # On every failed path, remove anything this invocation created before the
+        # directory descriptor is closed. This leaves no executable partial copy.
+        if root_fd is not None and not promotion_complete:
+            if destination_published:
+                try:
+                    os.unlink(destination_name, dir_fd=root_fd)
+                except OSError:
+                    pass
+            if temporary_created:
+                try:
+                    os.unlink(temporary_name, dir_fd=root_fd)
+                except OSError:
+                    pass
+
+        if root_fd is not None:
+            try:
+                os.close(root_fd)
+            except OSError:
+                pass
+
+
+if __name__ == "__main__":
+    _promote(sys.argv[1:])
+"""
+
+
+def _remote_promote_verified_pyz(
+    ssh,
+    uploaded_pyz_path: str,
+    root_tmp_dir: str,
+    expected_sha256: str,
+    expected_size: int,
+    *,
+    remote_python: str,
+    sudo_password: Optional[str],
+) -> str:
+    """Copy an uploaded zipapp into root-private storage and verify it there.
+
+    The SFTP upload initially lives in a directory controlled by the
+    authenticated SSH account. Hashing that path and later executing it with
+    sudo leaves a verify/use race: the account can replace the path after the
+    hash check. Instead, one privileged process opens the source safely, copies
+    and hashes the bytes into a private root-owned directory, and atomically
+    publishes the destination only after the digest and size match.
+
+    The returned root-owned path is the only path the caller may execute with
+    sudo. Racing or replacing the source can therefore cause only a verified
+    copy of the locally built payload to be published, or a closed failure.
+    """
+
+    expected = expected_sha256.strip().lower()
+    if len(expected) != 64 or any(c not in "0123456789abcdef" for c in expected):
+        raise ValueError("expected_sha256 must be a valid SHA-256 digest")
+    if expected_size < 0:
+        raise ValueError("expected_size must not be negative")
+
+    uploaded = PurePosixPath(uploaded_pyz_path)
+    root_dir = PurePosixPath(root_tmp_dir)
+    if not uploaded.is_absolute() or not root_dir.is_absolute():
+        raise ValueError("remote promotion paths must be absolute")
+
+    stage_dir = str(uploaded.parent)
+    source_name = uploaded.name
+    if stage_dir in {"", ".", "/"} or source_name in {"", ".", ".."}:
+        raise ValueError("uploaded_pyz_path must name a file inside a directory")
+
+    destination_name = "enroll.pyz"
+    destination_path = str(root_dir / destination_name)
+
+    cmd = " ".join(
+        shlex.quote(tok)
+        for tok in (
+            remote_python,
+            "-I",
+            "-c",
+            _REMOTE_PROMOTE_PYZ_SCRIPT,
+            stage_dir,
+            source_name,
+            str(root_dir),
+            destination_name,
+            expected,
+            str(expected_size),
+            "0",
+        )
+    )
+    rc, out, err = _ssh_run_sudo(
+        ssh,
+        cmd,
+        sudo_password=sudo_password,
+        get_pty=True,
+    )
+    if rc != 0:
+        raise RuntimeError(
+            "Unable to promote and verify the uploaded enroll.pyz into "
+            "root-private storage. Refusing to execute it.\n"
+            f"Command: sudo {cmd}\n"
+            f"Exit code: {rc}\n"
+            f"Stdout: {out.strip()}\n"
+            f"Stderr: {err.strip()}"
+        )
+    return destination_path
+
+
 def _remote_verify_pyz_sha256(
     ssh,
     remote_pyz_path: str,
@@ -468,23 +706,13 @@ def _remote_verify_pyz_sha256(
     *,
     remote_python: str,
 ) -> None:
-    """Verify the uploaded zipapp's SHA-256 on the remote before executing it.
+    """Verify the uploaded zipapp before same-user ``--no-sudo`` execution.
 
-    This is transport/staging integrity defence-in-depth. The check runs on the
+    The privileged path uses :func:`_remote_promote_verified_pyz` instead,
-    remote, immediately before the (root) execution of the zipapp, and fails
+    because hashing a user-writable pathname and later executing that pathname
-    closed if the digest does not match the bytes we built locally. It shrinks
+    as root would leave a verify/use race. This simpler check remains useful in
-    the window in which a *non-root* tamperer who somehow gained write access to
+    ``--no-sudo`` mode for detecting transfer corruption; there is no privilege
-    the staging directory could swap the file between upload and execution.
+    transition in that mode.
-
-    It deliberately does NOT establish trust in a root-compromised remote: a
-    host that is already root can forge any check it runs about itself. Per
-    SECURITY.md, such a host is outside Enroll's threat model. The value here is
-    catching accidental corruption and unprivileged-local-user staging races,
-    not defeating a compromised root.
-
-    The hashing is done with Python's hashlib (already required to run the
-    zipapp) rather than a ``sha256sum`` binary, so it does not depend on
-    coreutils being present or on PATH resolution of a hashing tool.
     """
 
     # Hash the staged file using the same interpreter that will execute it.
@@ -850,16 +1078,6 @@ def _remote_harvest(
             rapp = f"{rtmp}/enroll.pyz"
             sftp.put(str(pyz), rapp)
 
-            # Before executing the uploaded zipapp (as root, under sudo), verify
-            # on the remote that the staged bytes match what we built locally.
-            # This is staging/transport integrity defence-in-depth: it fails
-            # closed if the file was swapped or corrupted between upload and
-            # execution. It does not (and cannot) defend against a remote that
-            # is already root-compromised; see _remote_verify_pyz_sha256.
-            _remote_verify_pyz_sha256(
-                ssh, rapp, pyz_sha256, remote_python=remote_python
-            )
-
             if not no_sudo:
                 # The remote zipapp is staged as the SSH user, but the harvest
                 # itself runs as root.  Root must not write its bundle under the
@@ -883,8 +1101,26 @@ def _remote_harvest(
                 )
                 if rc != 0:
                     raise RuntimeError(f"Remote sudo chmod failed: {err.strip()}")
+                # Promote the uploaded payload into the root-private directory
+                # and verify the bytes during that privileged copy. From this point
+                # onward, execute only the immutable root-owned copy; never return
+                # to the SSH-user-controlled staging path.
+                rapp = _remote_promote_verified_pyz(
+                    ssh,
+                    rapp,
+                    remote_root_tmp,
+                    pyz_sha256,
+                    pyz.stat().st_size,
+                    remote_python=remote_python,
+                    sudo_password=sudo_password,
+                )
                 rbundle = f"{remote_root_tmp}/bundle"
             else:
+                # There is no privilege transition in --no-sudo mode, but retain
+                # the transport/corruption check before executing as the SSH user.
+                _remote_verify_pyz_sha256(
+                    ssh, rapp, pyz_sha256, remote_python=remote_python
+                )
                 rbundle = f"{rtmp}/bundle"
 
             # Run remote harvest.

rpm/enroll.spec

@@ -1,4 +1,4 @@
-%global upstream_version 0.8.0
+%global upstream_version 0.8.1
 
 Name:           enroll
 Version:        %{upstream_version}
@@ -43,6 +43,8 @@ Enroll a server's running state retrospectively into Ansible.
 %{_bindir}/enroll
 
 %changelog
+* Mon Aug 03 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
+- Security: fix a TOCTOU in remote harvest.
 * Mon Jul 13 2026 Miguel Jacq <mig@mig5.net> - %{version}-%{release}
 - Security: keep sudo-created remote harvest bundles root-owned while root packages and hashes them, expose only the archive to the authenticated SSH uid, and verify the root-computed digest after download. This removes the post-harvest tampering window created by recursively chowning the bundle before packaging without making the plaintext archive world-readable.
 - Security: enforce tar member limits while lazily parsing untrusted archives rather than after `TarFile.getmembers()` has already indexed the entire archive; count repeated `.` entries and cap remote compressed downloads as well.

tests/test_remote.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import hashlib
 import io
+import shlex
 import tarfile
 import warnings
 from pathlib import Path
@@ -9,10 +10,9 @@ from pathlib import Path
 import pytest
 
 
-# The remote harvest now SHA-256-verifies the uploaded zipapp on the remote
+# Sudo harvests promote and verify the upload into a root-private directory;
-# before executing it. Tests mock _build_enroll_pyz to write these fixed bytes
+# no-sudo harvests retain the direct remote digest check. Tests mock the local
-# and return (path, _FAKE_PYZ_SHA256); the fake SSH routers below answer the
+# builder with these fixed bytes and digest so both paths can be exercised.
-# verification command with the same digest so the happy paths proceed.
 _FAKE_PYZ_BYTES = b"PYZ"
 _FAKE_PYZ_SHA256 = "d6f4e1dbf7ba69af6c798c6f6f67383c978e68f4201bf31902275ef37e6263e1"
 
@@ -28,9 +28,16 @@ def _fake_build_enroll_pyz(td) -> tuple[Path, str]:
     return p, _FAKE_PYZ_SHA256
 
 
+def _is_pyz_promote_cmd(cmd: str) -> bool:
+    """True if *cmd* is the privileged copy-and-verify operation."""
+    return ".enroll.pyz.tmp" in cmd and "os.link" in cmd
+
+
 def _is_pyz_verify_cmd(cmd: str) -> bool:
-    """True if *cmd* is the remote pyz SHA-256 integrity check."""
+    """True if *cmd* is the no-sudo remote pyz SHA-256 check."""
-    return "hashlib.sha256" in cmd and "enroll.pyz" in cmd
+    return (
+        "hashlib.sha256" in cmd and "enroll.pyz" in cmd and not _is_pyz_promote_cmd(cmd)
+    )
 
 
 def _is_remote_uid_cmd(cmd: str) -> bool:
@@ -193,7 +200,9 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
             return self._sftp
 
         def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
-            # Integrity check of the uploaded pyz (added with SHA-256 verify).
+            if _is_pyz_promote_cmd(cmd):
+                calls.append((cmd, bool(get_pty)))
+                return (None, _Stdout(b""), _Stderr())
             if _is_pyz_verify_cmd(cmd):
                 return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
             calls.append((cmd, bool(get_pty)))
@@ -286,6 +295,26 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
     assert "chown -- 1000 /tmp/enroll-root-123/bundle.tgz" in joined
     assert "chmod 0711 -- /tmp/enroll-root-123" in joined
 
+    # The upload is promoted and verified before harvest, and sudo executes
+    # only the root-owned copy. A later replacement of the SSH-user path can no
+    # longer alter the bytes that Python opens as root.
+    promote_i, promote_cmd = next(
+        (i, c) for i, (c, _pty) in enumerate(calls) if _is_pyz_promote_cmd(c)
+    )
+    harvest_i, harvest_cmd = next(
+        (i, c)
+        for i, (c, _pty) in enumerate(calls)
+        if c.startswith("sudo -n") and " harvest " in c
+    )
+    assert promote_i < harvest_i
+    promote_argv = shlex.split(promote_cmd)
+    assert "/tmp/enroll-remote-123" in promote_argv
+    assert "/tmp/enroll-root-123" in promote_argv
+    assert promote_argv.count("enroll.pyz") >= 2
+    assert "/tmp/enroll-root-123/enroll.pyz" in harvest_cmd
+    assert "/tmp/enroll-remote-123/enroll.pyz" not in harvest_cmd
+    assert not any(_is_pyz_verify_cmd(c) for c, _pty in calls)
+
     # The trusted digest must be obtained while the archive is still private,
     # before ownership/read access is handed to the SSH account.
     archive_hash_i = next(
@@ -410,7 +439,9 @@ def test_remote_harvest_no_sudo_does_not_request_pty_or_chown(
             return self._sftp
 
         def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
-            # Integrity check of the uploaded pyz (added with SHA-256 verify).
+            if _is_pyz_promote_cmd(cmd):
+                calls.append((cmd, bool(get_pty)))
+                return (None, _Stdout(b""), _Stderr())
             if _is_pyz_verify_cmd(cmd):
                 return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
             calls.append((cmd, bool(get_pty)))
@@ -570,7 +601,9 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
             return self._sftp
 
         def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
-            # Integrity check of the uploaded pyz (added with SHA-256 verify).
+            if _is_pyz_promote_cmd(cmd):
+                calls.append((cmd, bool(get_pty)))
+                return (None, _Stdout(b""), _Stderr())
             if _is_pyz_verify_cmd(cmd):
                 return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
             calls.append((cmd, bool(get_pty)))
@@ -960,7 +993,8 @@ def test_remote_harvest_ssh_key_passphrase_retry(monkeypatch, tmp_path: Path):
             return self._sftp
 
         def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
-            # Integrity check of the uploaded pyz (added with SHA-256 verify).
+            if _is_pyz_promote_cmd(cmd):
+                return (None, _Stdout(b""), _Stderr())
             if _is_pyz_verify_cmd(cmd):
                 return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
             if cmd.startswith("tar -cz -C"):
@@ -1084,7 +1118,8 @@ def test_remote_harvest_ssh_key_passphrase_raises_when_not_interactive(
             return self._sftp
 
         def exec_command(self, cmd: str, **_kwargs):
-            # Integrity check of the uploaded pyz (added with SHA-256 verify).
+            if _is_pyz_promote_cmd(cmd):
+                return (None, _Stdout(b""), _Stderr())
             if _is_pyz_verify_cmd(cmd):
                 return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
             return (_Stdout(), _Stdout(), _Stderr())

tests/test_remote_pyz_promotion.py

@@ -0,0 +1,508 @@
+from __future__ import annotations
+
+import hashlib
+import os
+import shlex
+import stat
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+import enroll.remote as remote
+
+
+_TRUSTED_PYZ = b"trusted enroll zipapp bytes\n"
+_TRUSTED_SHA256 = hashlib.sha256(_TRUSTED_PYZ).hexdigest()
+
+
+def _prepare_dirs(tmp_path: Path) -> tuple[Path, Path]:
+    stage_dir = tmp_path / "user-stage"
+    root_dir = tmp_path / "root-stage"
+    stage_dir.mkdir(mode=0o700)
+    root_dir.mkdir(mode=0o700)
+    # mkdir is affected by umask; the production flow explicitly chmods 0700.
+    root_dir.chmod(0o700)
+    return stage_dir, root_dir
+
+
+def _load_promotion_function():
+    namespace = {"__name__": "enroll_promotion_test"}
+    exec(remote._REMOTE_PROMOTE_PYZ_SCRIPT, namespace)
+    return namespace["_promote"]
+
+
+def _promotion_args(
+    stage_dir: Path,
+    root_dir: Path,
+    *,
+    expected_sha256: str = _TRUSTED_SHA256,
+    expected_size: int = len(_TRUSTED_PYZ),
+    source_name: str = "enroll.pyz",
+    destination_name: str = "enroll.pyz",
+    expected_owner_uid: int | None = None,
+) -> list[str]:
+    if expected_owner_uid is None:
+        expected_owner_uid = os.geteuid()
+    return [
+        str(stage_dir),
+        source_name,
+        str(root_dir),
+        destination_name,
+        expected_sha256,
+        str(expected_size),
+        str(expected_owner_uid),
+    ]
+
+
+def _promote_direct(
+    stage_dir: Path,
+    root_dir: Path,
+    **kwargs,
+) -> None:
+    _load_promotion_function()(_promotion_args(stage_dir, root_dir, **kwargs))
+
+
+def _run_promotion_subprocess(
+    stage_dir: Path,
+    root_dir: Path,
+    *,
+    timeout: float = 5,
+    **kwargs,
+) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [
+            sys.executable,
+            "-I",
+            "-c",
+            remote._REMOTE_PROMOTE_PYZ_SCRIPT,
+            *_promotion_args(stage_dir, root_dir, **kwargs),
+        ],
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+
+
+def _assert_no_promoted_copy(root_dir: Path) -> None:
+    assert not (root_dir / "enroll.pyz").exists()
+    assert not (root_dir / ".enroll.pyz.tmp").exists()
+
+
+def test_promotion_script_entrypoint_publishes_exact_private_copy(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+
+    result = _run_promotion_subprocess(stage_dir, root_dir)
+
+    assert result.returncode == 0, result.stderr
+    promoted = root_dir / "enroll.pyz"
+    assert promoted.read_bytes() == _TRUSTED_PYZ
+    promoted_stat = promoted.stat()
+    assert stat.S_IMODE(promoted_stat.st_mode) == 0o500
+    assert promoted_stat.st_nlink == 1
+    assert not (root_dir / ".enroll.pyz.tmp").exists()
+
+
+def test_replacing_user_path_after_promotion_cannot_change_root_copy(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    source = stage_dir / "enroll.pyz"
+    source.write_bytes(_TRUSTED_PYZ)
+
+    _promote_direct(stage_dir, root_dir)
+
+    replacement = stage_dir / "replacement.pyz"
+    replacement.write_bytes(b"print('attacker code')\n")
+    os.replace(replacement, source)
+
+    assert source.read_bytes() != _TRUSTED_PYZ
+    assert (root_dir / "enroll.pyz").read_bytes() == _TRUSTED_PYZ
+
+
+def test_digest_mismatch_fails_closed_and_removes_partial_copy(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    payload = b"attacker-controlled bytes"
+    (stage_dir / "enroll.pyz").write_bytes(payload)
+
+    with pytest.raises(RuntimeError, match="SHA-256 mismatch"):
+        _promote_direct(stage_dir, root_dir, expected_size=len(payload))
+
+    _assert_no_promoted_copy(root_dir)
+
+
+@pytest.mark.parametrize(
+    "reported_size", [0, len(_TRUSTED_PYZ) - 1, len(_TRUSTED_PYZ) + 1]
+)
+def test_unexpected_source_size_fails_before_publication(
+    tmp_path: Path, reported_size: int
+):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+
+    with pytest.raises(RuntimeError, match="unexpected size"):
+        _promote_direct(stage_dir, root_dir, expected_size=reported_size)
+
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_source_symlink_is_rejected_without_following_it(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    target = stage_dir / "payload"
+    target.write_bytes(_TRUSTED_PYZ)
+    (stage_dir / "enroll.pyz").symlink_to(target.name)
+
+    with pytest.raises(OSError):
+        _promote_direct(stage_dir, root_dir)
+
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_hardlinked_source_is_rejected(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    source = stage_dir / "enroll.pyz"
+    source.write_bytes(_TRUSTED_PYZ)
+    os.link(source, stage_dir / "second-name")
+
+    with pytest.raises(RuntimeError, match="must not be hard-linked"):
+        _promote_direct(stage_dir, root_dir)
+
+    _assert_no_promoted_copy(root_dir)
+
+
+@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFO test requires POSIX")
+def test_fifo_source_is_rejected_without_blocking(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    os.mkfifo(stage_dir / "enroll.pyz", mode=0o600)
+
+    result = _run_promotion_subprocess(
+        stage_dir,
+        root_dir,
+        expected_sha256=hashlib.sha256(b"").hexdigest(),
+        expected_size=0,
+        timeout=2,
+    )
+
+    assert result.returncode != 0
+    assert "not a regular file" in result.stderr
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_directory_source_is_rejected(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").mkdir()
+
+    with pytest.raises(RuntimeError, match="not a regular file"):
+        _promote_direct(stage_dir, root_dir)
+
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_symlink_destination_directory_is_rejected(tmp_path: Path):
+    stage_dir = tmp_path / "user-stage"
+    stage_dir.mkdir()
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+    real_root = tmp_path / "real-root"
+    real_root.mkdir(mode=0o700)
+    real_root.chmod(0o700)
+    root_link = tmp_path / "root-link"
+    root_link.symlink_to(real_root, target_is_directory=True)
+
+    with pytest.raises(OSError):
+        _promote_direct(stage_dir, root_link)
+
+    _assert_no_promoted_copy(real_root)
+
+
+def test_non_private_destination_directory_is_rejected(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+    root_dir.chmod(0o755)
+
+    with pytest.raises(RuntimeError, match="not mode 0700"):
+        _promote_direct(stage_dir, root_dir)
+
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_promotion_requires_expected_effective_uid(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+
+    with pytest.raises(RuntimeError, match="not running as the expected user"):
+        _promote_direct(
+            stage_dir,
+            root_dir,
+            expected_owner_uid=os.geteuid() + 1,
+        )
+
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_existing_temporary_name_is_not_overwritten(tmp_path: Path):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+    temporary = root_dir / ".enroll.pyz.tmp"
+    temporary.write_bytes(b"pre-existing")
+
+    with pytest.raises(FileExistsError):
+        _promote_direct(stage_dir, root_dir)
+
+    assert temporary.read_bytes() == b"pre-existing"
+    assert not (root_dir / "enroll.pyz").exists()
+
+
+def test_existing_final_name_is_not_replaced_and_partial_copy_is_removed(
+    tmp_path: Path,
+):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    (stage_dir / "enroll.pyz").write_bytes(_TRUSTED_PYZ)
+    final = root_dir / "enroll.pyz"
+    final.write_bytes(b"pre-existing")
+
+    with pytest.raises(FileExistsError):
+        _promote_direct(stage_dir, root_dir)
+
+    assert final.read_bytes() == b"pre-existing"
+    assert not (root_dir / ".enroll.pyz.tmp").exists()
+
+
+def test_path_replacement_after_open_cannot_change_promoted_inode(
+    tmp_path: Path, monkeypatch
+):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    source = stage_dir / "enroll.pyz"
+    source.write_bytes(_TRUSTED_PYZ)
+    replacement = stage_dir / "attacker.pyz"
+    replacement.write_bytes(b"print('attacker code')\n")
+
+    promote = _load_promotion_function()
+    real_read = os.read
+    replaced = False
+
+    def replace_path_before_first_read(fd: int, count: int) -> bytes:
+        nonlocal replaced
+        if not replaced:
+            replaced = True
+            os.replace(replacement, source)
+        return real_read(fd, count)
+
+    monkeypatch.setattr(os, "read", replace_path_before_first_read)
+    promote(_promotion_args(stage_dir, root_dir))
+
+    assert replaced is True
+    assert source.read_bytes() != _TRUSTED_PYZ
+    assert (root_dir / "enroll.pyz").read_bytes() == _TRUSTED_PYZ
+
+
+def test_in_place_mutation_after_fstat_fails_digest_and_cleans_output(
+    tmp_path: Path, monkeypatch
+):
+    stage_dir, root_dir = _prepare_dirs(tmp_path)
+    source = stage_dir / "enroll.pyz"
+    source.write_bytes(_TRUSTED_PYZ)
+
+    promote = _load_promotion_function()
+    real_read = os.read
+    attacker_fd = os.open(source, os.O_WRONLY)
+    mutated = False
+
+    def mutate_inode_before_first_read(fd: int, count: int) -> bytes:
+        nonlocal mutated
+        if not mutated:
+            mutated = True
+            os.lseek(attacker_fd, 0, os.SEEK_SET)
+            os.write(attacker_fd, b"X" * len(_TRUSTED_PYZ))
+            os.fsync(attacker_fd)
+        return real_read(fd, count)
+
+    monkeypatch.setattr(os, "read", mutate_inode_before_first_read)
+    try:
+        with pytest.raises(RuntimeError, match="SHA-256 mismatch"):
+            promote(_promotion_args(stage_dir, root_dir))
+    finally:
+        os.close(attacker_fd)
+
+    assert mutated is True
+    _assert_no_promoted_copy(root_dir)
+
+
+def test_remote_promote_builds_one_privileged_atomic_command(monkeypatch):
+    captured: dict[str, object] = {}
+
+    def fake_ssh_run_sudo(ssh, cmd, *, sudo_password, get_pty):
+        captured.update(
+            ssh=ssh,
+            cmd=cmd,
+            sudo_password=sudo_password,
+            get_pty=get_pty,
+        )
+        return 0, "", ""
+
+    monkeypatch.setattr(remote, "_ssh_run_sudo", fake_ssh_run_sudo)
+    marker = object()
+
+    result = remote._remote_promote_verified_pyz(
+        marker,
+        "/tmp/user-stage/enroll.pyz",
+        "/tmp/root-stage",
+        _TRUSTED_SHA256,
+        len(_TRUSTED_PYZ),
+        remote_python="/usr/bin/python3",
+        sudo_password="secret",
+    )
+
+    assert result == "/tmp/root-stage/enroll.pyz"
+    assert captured["ssh"] is marker
+    assert captured["sudo_password"] == "secret"
+    assert captured["get_pty"] is True
+    argv = shlex.split(str(captured["cmd"]))
+    assert argv == [
+        "/usr/bin/python3",
+        "-I",
+        "-c",
+        remote._REMOTE_PROMOTE_PYZ_SCRIPT,
+        "/tmp/user-stage",
+        "enroll.pyz",
+        "/tmp/root-stage",
+        "enroll.pyz",
+        _TRUSTED_SHA256,
+        str(len(_TRUSTED_PYZ)),
+        "0",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("uploaded", "root_dir", "digest", "size"),
+    [
+        ("relative/enroll.pyz", "/tmp/root", _TRUSTED_SHA256, len(_TRUSTED_PYZ)),
+        ("/tmp/stage/enroll.pyz", "relative/root", _TRUSTED_SHA256, len(_TRUSTED_PYZ)),
+        ("/tmp/stage/enroll.pyz", "/tmp/root", "not-a-digest", len(_TRUSTED_PYZ)),
+        ("/tmp/stage/enroll.pyz", "/tmp/root", _TRUSTED_SHA256, -1),
+    ],
+)
+def test_remote_promote_rejects_invalid_local_parameters_before_sudo(
+    monkeypatch, uploaded: str, root_dir: str, digest: str, size: int
+):
+    called = False
+
+    def unexpected_sudo(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        raise AssertionError("sudo must not run")
+
+    monkeypatch.setattr(remote, "_ssh_run_sudo", unexpected_sudo)
+
+    with pytest.raises(ValueError):
+        remote._remote_promote_verified_pyz(
+            object(),
+            uploaded,
+            root_dir,
+            digest,
+            size,
+            remote_python="python3",
+            sudo_password=None,
+        )
+    assert called is False
+
+
+def test_remote_promote_reports_privileged_failure(monkeypatch):
+    monkeypatch.setattr(
+        remote,
+        "_ssh_run_sudo",
+        lambda *_args, **_kwargs: (1, "", "digest mismatch"),
+    )
+
+    with pytest.raises(RuntimeError, match="Refusing to execute"):
+        remote._remote_promote_verified_pyz(
+            object(),
+            "/tmp/stage/enroll.pyz",
+            "/tmp/root",
+            _TRUSTED_SHA256,
+            len(_TRUSTED_PYZ),
+            remote_python="python3",
+            sudo_password=None,
+        )
+
+
+def test_sudo_harvest_stops_before_execution_when_promotion_fails(
+    tmp_path: Path, monkeypatch
+):
+    import types
+
+    # The fake builder must create the file because _remote_harvest reads its size.
+    def build_pyz(td):
+        pyz = Path(td) / "enroll.pyz"
+        pyz.write_bytes(_TRUSTED_PYZ)
+        return pyz, _TRUSTED_SHA256
+
+    monkeypatch.setattr(remote, "_build_enroll_pyz", build_pyz)
+    monkeypatch.setattr(
+        remote,
+        "_remote_promote_verified_pyz",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            RuntimeError("promotion rejected attacker bytes")
+        ),
+    )
+
+    user_commands: list[str] = []
+    sudo_commands: list[str] = []
+
+    def fake_run(_ssh, cmd, **_kwargs):
+        user_commands.append(cmd)
+        if cmd == "mktemp -d":
+            return 0, "/tmp/user-stage\n", ""
+        return 0, "", ""
+
+    def fake_sudo(_ssh, cmd, **_kwargs):
+        sudo_commands.append(cmd)
+        if cmd == "mktemp -d":
+            return 0, "/tmp/root-stage\n", ""
+        return 0, "", ""
+
+    monkeypatch.setattr(remote, "_ssh_run", fake_run)
+    monkeypatch.setattr(remote, "_ssh_run_sudo", fake_sudo)
+
+    class FakeSFTP:
+        def put(self, _local, _remote):
+            return None
+
+        def close(self):
+            return None
+
+    class FakeSSH:
+        def load_system_host_keys(self):
+            return None
+
+        def set_missing_host_key_policy(self, _policy):
+            return None
+
+        def connect(self, **_kwargs):
+            return None
+
+        def open_sftp(self):
+            return FakeSFTP()
+
+        def close(self):
+            return None
+
+    monkeypatch.setitem(
+        sys.modules,
+        "paramiko",
+        types.SimpleNamespace(
+            SSHClient=FakeSSH,
+            RejectPolicy=type("RejectPolicy", (), {}),
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="promotion rejected attacker bytes"):
+        remote._remote_harvest(
+            local_out_dir=tmp_path / "out",
+            remote_host="example.com",
+            no_sudo=False,
+        )
+
+    assert not any(" harvest " in command for command in sudo_commands)
+    assert any(command == "rm -rf -- /tmp/root-stage" for command in sudo_commands)
+    assert any(command == "rm -rf -- /tmp/user-stage" for command in user_commands)