diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b8757e..ee0be80 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/debian/changelog b/debian/changelog index 7afa14d..dbbc03c 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +enroll (0.8.1) unstable; urgency=medium + + * Security: fix a TOCTOU in remote harvest. + + -- Miguel Jacq 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. diff --git a/enroll/remote.py b/enroll/remote.py index 4b936da..83006d9 100644 --- a/enroll/remote.py +++ b/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 - (see ``_remote_verify_pyz_sha256``). This is transport/staging integrity + file that is executed is byte-for-byte the one we built (see + ``_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 - remote, immediately before the (root) execution of the zipapp, and fails - closed if the digest does not match the bytes we built locally. It shrinks - the window in which a *non-root* tamperer who somehow gained write access to - the staging directory could swap the file between upload and execution. - - 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. + The privileged path uses :func:`_remote_promote_verified_pyz` instead, + because hashing a user-writable pathname and later executing that pathname + as root would leave a verify/use race. This simpler check remains useful in + ``--no-sudo`` mode for detecting transfer corruption; there is no privilege + transition in that mode. """ # 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. diff --git a/rpm/enroll.spec b/rpm/enroll.spec index b026fee..5253c80 100644 --- a/rpm/enroll.spec +++ b/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 - %{version}-%{release} +- Security: fix a TOCTOU in remote harvest. * Mon Jul 13 2026 Miguel Jacq - %{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. diff --git a/tests/test_remote.py b/tests/test_remote.py index 255b00f..1e5ffcc 100644 --- a/tests/test_remote.py +++ b/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 -# before executing it. Tests mock _build_enroll_pyz to write these fixed bytes -# and return (path, _FAKE_PYZ_SHA256); the fake SSH routers below answer the -# verification command with the same digest so the happy paths proceed. +# Sudo harvests promote and verify the upload into a root-private directory; +# no-sudo harvests retain the direct remote digest check. Tests mock the local +# builder with these fixed bytes and digest so both paths can be exercised. _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.""" - return "hashlib.sha256" in cmd and "enroll.pyz" in cmd + """True if *cmd* is the no-sudo remote pyz SHA-256 check.""" + 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()) diff --git a/tests/test_remote_pyz_promotion.py b/tests/test_remote_pyz_promotion.py new file mode 100644 index 0000000..4ab6f38 --- /dev/null +++ b/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)