diff --git a/src/openjd/sessions/_linux/_sudo.py b/src/openjd/sessions/_linux/_sudo.py index fccfef88..186acf85 100644 --- a/src/openjd/sessions/_linux/_sudo.py +++ b/src/openjd/sessions/_linux/_sudo.py @@ -9,6 +9,7 @@ from .._logging import LoggerAdapter, LogContent, LogExtraInfo from .._os_checker import is_posix, is_linux +from .._system_commands import system_command_path PGREP_NO_MATCH = 1 """``pgrep``'s exit status for "no processes matched". @@ -174,7 +175,7 @@ def find_child_process_id_pgrep( sudo_pid: int, ) -> Optional[int]: pgrep_result = run( - ["pgrep", "-P", str(sudo_pid)], + [system_command_path("pgrep"), "-P", str(sudo_pid)], stdout=PIPE, stderr=STDOUT, stdin=DEVNULL, diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index 9d0929e6..3af0ba1b 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -19,6 +19,7 @@ from ._logging import LoggerAdapter, LogContent, LogExtraInfo from ._os_checker import is_linux, is_macos, is_posix, is_windows from ._session_user import PosixSessionUser, WindowsSessionUser, SessionUser +from ._system_commands import find_system_command, system_command_path from ._action_filter import redact_openjd_redacted_env_requests if is_windows(): # pragma: nocover @@ -594,7 +595,7 @@ def _start_subprocess(self) -> Optional[Popen]: ) command.extend( [ - "sudo", + system_command_path("sudo"), "-u", user.user, "-i", @@ -605,7 +606,16 @@ def _start_subprocess(self) -> Optional[Popen]: ] ) else: - command.extend(["sudo", "-u", user.user, "-i", "setsid", "-w"]) + command.extend( + [ + system_command_path("sudo"), + "-u", + user.user, + "-i", + system_command_path("setsid"), + "-w", + ] + ) elif is_windows(): user = cast(WindowsSessionUser, self._user) # type: ignore @@ -882,13 +892,15 @@ def _posix_signal_subprocess( else: raise NotImplementedError(f"Unsupported signal: {signal_name}") - kill_cmd = list[str]() - - if self._user is not None: - user = cast(PosixSessionUser, self._user) - # Only sudo if the user to run as is not the same as the current user. - if not user.is_process_user(): - kill_cmd = ["sudo", "-u", user.user, "-i"] + # Whether this signal has to go through sudo. The argv itself is not built + # yet: the direct os.killpg() attempt below returns on success, and on a + # host with CAP_KILL that is the only path taken. Resolving sudo here would + # make a host without sudo unable to signal *even when it never needs + # sudo* -- a regression this method did not have when the name was a bare + # string in a list. + needs_sudo = ( + self._user is not None and not cast(PosixSessionUser, self._user).is_process_user() + ) # If we were unable to detect sudo's child process PID after launching the # subprocess, we try again now @@ -941,9 +953,25 @@ def _posix_signal_subprocess( # Uncomment to visualize process tree when debugging tests # self._log_process_tree() + # `kill` is resolved only for the same-user branch, where `run()` really + # does execvp() a bare name against this process's PATH. + # + # In the cross-user branch it must stay a bare name. `sudo -i` simulates an + # initial login: it starts the target user's login shell and hands it the + # command via -c, so `kill` is a *shell builtin* there. Nothing is looked up + # on PATH, and -i has already reset the environment, so a job-controlled + # PATH cannot reach this position. Resolving it would invent a hard dependency + # on kill(1) from procps, which Debian `-slim` images -- a common worker + # base -- do not install, and would break every cross-user SIGKILL fallback + # on such a host. + if needs_sudo: + user = cast(PosixSessionUser, self._user) + kill_cmd = [system_command_path("sudo"), "-u", user.user, "-i", "kill"] + else: + kill_cmd = [system_command_path("kill")] + kill_cmd.extend( [ - "kill", "-s", signal_name, "--", @@ -971,16 +999,22 @@ def _posix_signal_subprocess( def _log_process_tree(self) -> None: """A developer method to visualize the process tree including PIDs and PGIDs when debuging tests""" - pstree_result = run(["pstree", "-pg"], stdout=PIPE, stderr=STDOUT, stdin=DEVNULL, text=True) - self._logger.debug( - f"pstree -pg output: {pstree_result.stdout}", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) - ps_result = run(["ps", "-ejH"], stdout=PIPE, stderr=STDOUT, stdin=DEVNULL, text=True) - self._logger.debug( - f"ps -ejH output:\n{ps_result.stdout}", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) + # find_system_command rather than system_command_path: neither of these is + # required for correct operation, and a debug aid must not raise on a host + # that happens not to install pstree. + for name, args in (("pstree", ["-pg"]), ("ps", ["-ejH"])): + command = find_system_command(name) + if command is None: + self._logger.debug( + f"{name} is not installed; skipping its process tree dump", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), + ) + continue + result = run([command, *args], stdout=PIPE, stderr=STDOUT, stdin=DEVNULL, text=True) + self._logger.debug( + f"{name} {' '.join(args)} output:\n{result.stdout}", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), + ) def _windows_notify_subprocess(self, process: Popen) -> None: """Sends a CTRL_BREAK_EVENT signal to the subprocess. diff --git a/src/openjd/sessions/_system_commands.py b/src/openjd/sessions/_system_commands.py new file mode 100644 index 00000000..987dd333 --- /dev/null +++ b/src/openjd/sessions/_system_commands.py @@ -0,0 +1,184 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +"""Resolution of system command names to absolute paths, without consulting PATH. + +The problem: a session launches its own privileged helpers (``sudo``, ``setsid``, +``kill``) with the environment it also gives the job, and that environment +includes the job's ``PATH``. A bare command name in such an argv is resolved +through that ``PATH``, so a job that puts an executable named ``sudo`` early on it +has that executable run at the session's privilege level rather than the job +user's. + +The solution: never let a command name reach ``execvp``-style resolution. Callers +pass a bare name here and get back an absolute path found by scanning a fixed +list of trusted directories. + +Three properties make that work, and all three are easy to undo by accident: + +* ``PATH`` is never read. Not directly, and not through :func:`shutil.which` or + ``command -v``, which resolve via ``PATH`` and so would restore the original + behaviour while looking like a fix. +* Only paths under :data:`TRUSTED_SYSTEM_DIRECTORIES` are returned. A name + containing a path separator is rejected, because ``os.path.join`` would + otherwise let ``../../tmp/evil`` escape the directory being searched. +* A command that cannot be found raises. Returning the bare name as a fallback + would put resolution back on ``PATH`` while the code still read as though it + did not. + +This module is POSIX-oriented, and :data:`TRUSTED_SYSTEM_DIRECTORIES` lists POSIX +locations. On Windows nothing is found and the lookups raise, which suits this +library because the cross-user paths needing these commands are POSIX-only. +""" + +from __future__ import annotations + +import os +from typing import Optional, Tuple + +__all__ = [ + "SystemCommandNotFoundError", + "clear_command_cache", + "TRUSTED_SYSTEM_DIRECTORIES", + "find_system_command", + "system_command_path", +] + + +TRUSTED_SYSTEM_DIRECTORIES: Tuple[str, ...] = ( + # Ordered, and the order is deliberate. On NixOS the setuid `sudo` wrapper + # lives in /run/wrappers/bin and the /usr/bin copy is either absent or not + # setuid, so the wrapper directory has to be consulted first. Everywhere else + # this directory does not exist and costs one stat(). + "/run/wrappers/bin", + # ...and the two NixOS entries are a pair. /run/wrappers/bin holds *only* the + # setuid/setcap wrappers, so on NixOS it resolves `sudo` and nothing else: + # /usr/bin holds just `env`, /bin just `sh`, and the sbin directories do not + # exist. `setsid` and `pgrep` live in this symlink farm, which nixos-rebuild + # manages and root owns, making it trust-equivalent to /usr/bin there. + # + # Without this entry the ordering above buys nothing: a cross-user session + # would resolve `sudo` and then fail on `setsid` one line later. The pairing + # is asserted in TestTrustedDirectories so it cannot be half-removed. + "/run/current-system/sw/bin", + "/usr/bin", + "/bin", + # sbin entries are last: on non-usr-merged distributions (some Debian + # releases) some system commands exist only under /sbin. + "/usr/sbin", + "/sbin", +) + + +class SystemCommandNotFoundError(FileNotFoundError): + """A required system command was not present in any trusted directory. + + A :class:`FileNotFoundError`, and therefore an :class:`OSError`, on purpose. + + An earlier revision of this module deliberately made it a plain ``Exception``, + reasoning that "a privileged helper is unavailable" must not be absorbed by + handlers that catch ``OSError`` to mean "carry on degraded". That reasoning was + wrong here, because it assumed rather than checked what those handlers do. + ``_runner_base``'s cancel path catches ``OSError`` around + ``notify()``/``terminate()`` precisely so a failure to signal does not unwind an + in-progress cancelation -- its own comment says "a cancel path is the wrong + place to raise". Escaping that handler would lose the cancel's bookkeeping, + which is worse than the warning it logs. + + So the semantics this class wants are exactly ``FileNotFoundError``'s: the thing + we tried to launch is not there. Remaining a distinct type still lets a caller + that cares tell "not in any trusted directory" apart from "``exec`` failed", and + the message says which. + """ + + +def _validate_command_name(name: str) -> None: + """Reject anything that is not a bare command name. + + Without this the resolver would itself become the injection point it exists + to remove: ``os.path.join("/usr/bin", "../../tmp/evil")`` escapes the trusted + directory entirely, so a caller that passed attacker-influenced text would be + no better off than before. + """ + if not name: + raise ValueError("A system command name must not be empty.") + if name in (os.curdir, os.pardir): + raise ValueError(f"{name!r} is not a system command name.") + # Checking both separators regardless of platform. A backslash is a legal + # filename character on POSIX, but no command this module resolves contains + # one, and treating it as suspect keeps the check identical on both + # platforms rather than subtly weaker on one. + # + # The colon is rejected for the same reason, and it is not hypothetical: + # ntpath.join(r"C:\Windows\System32", "D:evil") == "D:evil". A drive-relative + # name discards the trusted prefix entirely while containing no separator at + # all, so a separator-only check lets it through. POSIX joins it harmlessly, + # but the guard belongs here rather than depending on which os.path is loaded. + if "/" in name or "\\" in name or ":" in name: + raise ValueError( + f"A system command name must not contain a path separator or drive " + f"specifier, but got {name!r}." + ) + + +def _is_executable_file(path: str) -> bool: + return os.path.isfile(path) and os.access(path, os.X_OK) + + +_RESOLVED_COMMANDS: dict[str, str] = {} +"""Cache of successful lookups only, keyed by bare command name. + +Successes are safe to keep for the life of the process: a resolved path stays +valid, and these lookups sit on process-launch and signal-delivery paths where a +directory walk per signal is wasteful. + +Failures are not cached, and that asymmetry is the point. A package manager +replacing a binary unlinks and relinks it, so a lookup landing in that window sees +nothing. Caching that answer would make one unlucky moment permanent for the rest +of a long-lived agent's life, turning a transient miss into a session that can +never start a cross-user process again. Re-walking a handful of directories on the +miss path costs nothing worth having. +""" + + +def clear_command_cache() -> None: + """Discard resolved paths. For tests that patch the trusted directory list.""" + _RESOLVED_COMMANDS.clear() + + +def find_system_command(name: str) -> Optional[str]: + """Return the absolute path to ``name``, or ``None`` if it is not installed. + + Searches :data:`TRUSTED_SYSTEM_DIRECTORIES` in order. ``PATH`` is not + consulted. Use this for commands whose absence is tolerable; use + :func:`system_command_path` when the command is required. + + Raises: + ValueError: if ``name`` is not a bare command name. + """ + _validate_command_name(name) + cached = _RESOLVED_COMMANDS.get(name) + if cached is not None: + return cached + for directory in TRUSTED_SYSTEM_DIRECTORIES: + candidate = os.path.join(directory, name) + if _is_executable_file(candidate): + _RESOLVED_COMMANDS[name] = candidate + return candidate + return None + + +def system_command_path(name: str) -> str: + """Return the absolute path to ``name``. + + Raises: + ValueError: if ``name`` is not a bare command name. + SystemCommandNotFoundError: if ``name`` is not in any trusted directory. + """ + path = find_system_command(name) + if path is None: + raise SystemCommandNotFoundError( + f"Could not find the system command {name!r} in any trusted directory " + f"({', '.join(TRUSTED_SYSTEM_DIRECTORIES)}). PATH is deliberately not " + f"searched; see openjd.sessions._system_commands." + ) + return path diff --git a/test/openjd/sessions_v0/test_subprocess.py b/test/openjd/sessions_v0/test_subprocess.py index 11231ed2..20e58366 100644 --- a/test/openjd/sessions_v0/test_subprocess.py +++ b/test/openjd/sessions_v0/test_subprocess.py @@ -1308,6 +1308,15 @@ def test_builds_setsid_shim_command_on_macos(self, queue_handler: QueueHandler) patch.object( subprocess_mod, "_macos_shim_interpreter", return_value="/usr/local/bin/python3" ), + # Stubbed so the assertion below pins the *shape* of the argv rather + # than this host's sudo location. It also makes the assertion prove + # the command went through the trusted-path resolver: a literal + # "sudo" in the source would no longer match. + patch.object( + subprocess_mod, + "system_command_path", + side_effect=lambda name: f"/trusted/bin/{name}", + ), patch.object(subprocess_mod, "Popen") as mock_popen, ): subproc._start_subprocess() @@ -1315,7 +1324,7 @@ def test_builds_setsid_shim_command_on_macos(self, queue_handler: QueueHandler) # THEN built_command = mock_popen.call_args.kwargs["args"] assert built_command == [ - "sudo", + "/trusted/bin/sudo", "-u", "job-user", "-i", diff --git a/test/openjd/sessions_v0/test_subprocess_process_group.py b/test/openjd/sessions_v0/test_subprocess_process_group.py index b088be76..54a4f912 100644 --- a/test/openjd/sessions_v0/test_subprocess_process_group.py +++ b/test/openjd/sessions_v0/test_subprocess_process_group.py @@ -14,7 +14,9 @@ import pytest +from openjd.sessions import _subprocess as subprocess_mod from openjd.sessions._os_checker import is_posix +from openjd.sessions._session_user import PosixSessionUser from openjd.sessions._subprocess import LoggingSubprocess from .conftest import build_logger @@ -87,3 +89,91 @@ def test_sudo_helper_returns_unknown_when_sudo_is_already_gone(self) -> None: # THEN: the established "unknown" value, not an escaping ESRCH. assert result is None + + +@pytest.mark.skipif(not is_posix(), reason="posix signal delivery") +class TestSignalArgvCommandResolution: + """How `kill` is spelled in the signal argv, per branch. + + The two branches deliberately differ, and the difference is easy to "tidy" + into a bug in either direction, so both are pinned here. + + Cross-user goes through `sudo -u -i kill ...`. `sudo -i` simulates an + initial login: it starts the target user's login shell and passes the command + to it via -c, so `kill` is a **shell builtin** in that position. Nothing is + resolved on PATH and -i has already reset the environment, so a job-controlled + PATH cannot reach it. Resolving it to an absolute path would instead invent a hard + dependency on kill(1) from procps, which Debian `-slim` images do not install, + and would break every cross-user SIGKILL fallback on such a host. + + Same-user goes through `run([...])`, which really does execvp() a bare name + against this process's PATH, so there the resolution is load-bearing. + """ + + def _signal_and_capture_argv(self, user, message_queue, queue_handler) -> list[str]: + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=["/path/to/workload.sh"], user=user) + proc._sudo_child_process_group_id = 4321 + + with ( + patch.object(subprocess_mod, "is_posix", return_value=True), + patch.object(subprocess_mod, "is_windows", return_value=False), + patch.object(subprocess_mod, "is_linux", return_value=False), + patch.object( + subprocess_mod, + "system_command_path", + side_effect=lambda name: f"/trusted/{name}", + ), + patch.object(subprocess_mod, "os") as mock_os, + patch.object(subprocess_mod, "run") as mock_run, + ): + mock_os.killpg.side_effect = OSError(1, "not permitted") + mock_run.return_value = MagicMock(returncode=0) + proc._posix_signal_subprocess(MagicMock(pid=999999), signal_name="kill") + + assert mock_run.call_count == 1, "the sudo/run fallback did not execute" + return list(mock_run.call_args.args[0]) + + def test_cross_user_leaves_kill_bare_for_the_login_shell_builtin( + self, message_queue: SimpleQueue, queue_handler: QueueHandler + ) -> None: + # GIVEN + user = MagicMock(spec=PosixSessionUser) + user.user = "job-user" + user.is_process_user.return_value = False + + # WHEN + argv = self._signal_and_capture_argv(user, message_queue, queue_handler) + + # THEN + assert argv == [ + "/trusted/sudo", + "-u", + "job-user", + "-i", + "kill", + "-s", + "kill", + "--", + "-4321", + ] + # Spelled out separately from the equality above, because this is the + # property that matters and the reason for it is not obvious from the list. + assert ( + "kill" in argv and "/trusted/kill" not in argv + ), "kill must stay a bare name so the login shell builtin is used" + + def test_same_user_resolves_kill_because_run_execvps_it( + self, message_queue: SimpleQueue, queue_handler: QueueHandler + ) -> None: + # GIVEN + user = MagicMock(spec=PosixSessionUser) + user.user = "same-user" + user.is_process_user.return_value = True + + # WHEN + argv = self._signal_and_capture_argv(user, message_queue, queue_handler) + + # THEN + assert argv == ["/trusted/kill", "-s", "kill", "--", "-4321"] + assert "sudo" not in " ".join(argv) diff --git a/test/openjd/sessions_v0/test_system_commands.py b/test/openjd/sessions_v0/test_system_commands.py new file mode 100644 index 00000000..a0422fa2 --- /dev/null +++ b/test/openjd/sessions_v0/test_system_commands.py @@ -0,0 +1,347 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +"""Tests for the trusted-path command resolver. + +``_system_commands`` fails silently when it fails at all: a resolver that quietly +falls back to ``PATH`` returns a working path for every command that is installed, +so it behaves identically to a correct one on any normal host. Only a caller with +an attacker-controlled ``PATH`` sees the difference. + +These tests exist so that the difference is observable here instead. Each one +fails if the behaviour it describes is removed, which is what makes the resolver's +guarantees checkable rather than asserted. +""" + +import os +import posixpath +import stat +from pathlib import Path +from unittest.mock import patch + +import pytest + +from openjd.sessions._os_checker import is_posix +from openjd.sessions._system_commands import ( + SystemCommandNotFoundError, + clear_command_cache, + TRUSTED_SYSTEM_DIRECTORIES, + find_system_command, + system_command_path, +) + +_MODULE = "openjd.sessions._system_commands" + + +@pytest.fixture(autouse=True) +def clear_resolver_cache(): + """Successful lookups are cached, so a path resolved under one patched + directory list would otherwise leak into the next test.""" + clear_command_cache() + yield + clear_command_cache() + + +@pytest.fixture +def executable_dir(tmp_path: Path) -> Path: + """A directory containing an executable file named ``target-cmd``.""" + target = tmp_path / "target-cmd" + target.write_text("#!/bin/sh\ntrue\n") + target.chmod(target.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + return tmp_path + + +class TestOnlySearchesTrustedDirectories: + def test_resolves_a_command_in_a_searched_directory(self, executable_dir: Path) -> None: + """The negative control. Without it, the "not found" assertions below would + be indistinguishable from a resolver that never finds anything.""" + # GIVEN + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(executable_dir),)): + # WHEN + result = find_system_command("target-cmd") + + # THEN + assert result == str(executable_dir / "target-cmd") + + def test_does_not_resolve_a_command_outside_searched_directories( + self, executable_dir: Path + ) -> None: + # GIVEN + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", ("/usr/bin", "/bin")): + # WHEN + result = find_system_command("target-cmd") + + # THEN + assert result is None + + def test_ignores_path_even_when_it_contains_a_matching_command( + self, executable_dir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Pins "PATH is never read". + + This is the property a ``shutil.which`` implementation would silently + violate, and the one a PATH fallback for missing commands would undo. It + is the closest thing this suite has to a direct test of the reported + vulnerability: the job-controlled PATH must not influence resolution. + """ + # GIVEN + monkeypatch.setenv("PATH", str(executable_dir)) + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", ("/usr/bin", "/bin")): + # WHEN + result = find_system_command("target-cmd") + + # THEN + assert result is None, "PATH was consulted" + + def test_returns_the_first_matching_directory(self, tmp_path: Path) -> None: + """Order is load-bearing: on NixOS the setuid sudo wrapper must win over a + non-setuid /usr/bin copy.""" + # GIVEN + first, second = tmp_path / "first", tmp_path / "second" + for directory in (first, second): + directory.mkdir() + target = directory / "target-cmd" + target.write_text("#!/bin/sh\ntrue\n") + target.chmod(target.stat().st_mode | stat.S_IXUSR) + + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(first), str(second))): + # WHEN + result = find_system_command("target-cmd") + + # THEN + assert result == str(first / "target-cmd") + + @pytest.mark.skipif( + not is_posix(), + reason="On Windows os.access(X_OK) is true for any existing file, so " + "'not executable' is not expressible there", + ) + def test_ignores_a_non_executable_file(self, tmp_path: Path) -> None: + # GIVEN + (tmp_path / "target-cmd").write_text("not executable") + + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(tmp_path),)): + # WHEN + result = find_system_command("target-cmd") + + # THEN + assert result is None + + +class TestRejectsNonBareNames: + @pytest.mark.parametrize( + "name", + [ + pytest.param("a/b", id="forward-slash"), + pytest.param("a\\b", id="backslash"), + pytest.param("", id="empty"), + pytest.param(".", id="curdir"), + pytest.param("..", id="pardir"), + # ntpath.join(r"C:\Windows\System32", "D:evil") == "D:evil" -- a + # drive-relative name discards the trusted prefix while containing no + # separator, so a separator-only guard lets it through. + pytest.param("D:evil", id="drive-relative"), + pytest.param("a:b", id="colon"), + ], + ) + def test_rejects_name_with_a_path_component(self, name: str) -> None: + with pytest.raises(ValueError): + find_system_command(name) + + def test_rejects_traversal_even_though_the_target_is_reachable(self, tmp_path: Path) -> None: + """The guard must be about the name, not about whether the join happens to + land on a real file -- so prove the target IS reachable by that join before + asserting the name is refused. + + Without this, a test whose searched directory has nothing above it would + pass even with the guard deleted, pinning nothing. That was verified: the + parametrized test above does not catch the mutation on its own. + """ + # GIVEN + target = tmp_path / "target-cmd" + target.write_text("#!/bin/sh\ntrue\n") + target.chmod(target.stat().st_mode | stat.S_IXUSR) + nested = tmp_path / "nested" + nested.mkdir() + assert os.path.isfile( + os.path.join(str(nested), "../target-cmd") + ), "precondition: the traversal target is reachable by this join" + + # WHEN / THEN + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(nested),)): + with pytest.raises(ValueError, match="path separator"): + find_system_command("../target-cmd") + + def test_rejection_is_valueerror_not_notfound(self) -> None: + """A bad name is a caller bug; a missing command is an environment problem. + Conflating them would let a caller mistake one for the other.""" + with pytest.raises(ValueError): + system_command_path("a/b") + + +class TestMissingCommandRaises: + def test_find_returns_none(self) -> None: + assert find_system_command("openjd-definitely-not-installed") is None + + def test_raises_rather_than_returning_the_bare_name(self) -> None: + """The silent-fallback failure mode: returning "sudo" here would look fixed + and behave exactly as the vulnerability did.""" + with pytest.raises(SystemCommandNotFoundError) as excinfo: + system_command_path("openjd-definitely-not-installed") + + message = str(excinfo.value) + assert "openjd-definitely-not-installed" in message + assert "PATH is deliberately not searched" in message + + def test_is_an_oserror_so_the_cancel_path_can_absorb_it(self) -> None: + """This assertion is the inverse of what an earlier revision asserted, and + the reversal is the point. + + That revision made the error a plain ``Exception`` on the theory that an + unavailable privileged helper must not be absorbed by "carry on degraded" + handlers. The theory was never checked against the handlers themselves. + ``_runner_base``'s cancel path catches ``OSError`` around + ``notify()``/``terminate()`` deliberately, so that failing to signal does + not unwind an in-progress cancelation -- its comment says "a cancel path is + the wrong place to raise". A non-OSError escapes that guard and costs the + cancel its bookkeeping. + + Signalling reaches ``system_command_path("sudo")``, so this is a live path, + not a hypothetical one. + """ + assert issubclass(SystemCommandNotFoundError, OSError) + assert issubclass(SystemCommandNotFoundError, FileNotFoundError) + + def test_is_still_distinguishable_from_a_plain_exec_failure(self) -> None: + """Being a FileNotFoundError must not cost a caller the ability to tell + "no trusted directory has it" apart from "exec failed".""" + assert SystemCommandNotFoundError is not FileNotFoundError + with pytest.raises(SystemCommandNotFoundError): + system_command_path("openjd-definitely-not-installed") + + +class TestTrustedDirectories: + def test_all_entries_are_absolute(self) -> None: + """A relative entry would resolve against the process working directory, + which a session changes. + + posixpath rather than os.path: these entries are POSIX paths, and from Python + 3.13 ntpath.isabs() treats a single-slash path as drive-relative rather than + absolute. Using os.path made this a statement about the host running the + tests rather than about the constant, and it failed the windows-latest 3.13 + leg on "/run/wrappers/bin is not absolute". + """ + for directory in TRUSTED_SYSTEM_DIRECTORIES: + assert posixpath.isabs(directory), f"{directory} is not absolute" + + def test_searches_the_setuid_wrapper_directory_before_usr_bin(self) -> None: + assert TRUSTED_SYSTEM_DIRECTORIES.index( + "/run/wrappers/bin" + ) < TRUSTED_SYSTEM_DIRECTORIES.index("/usr/bin") + + def test_searches_both_sbin_locations(self) -> None: + """On non-usr-merged distributions some system commands exist only under + /sbin.""" + assert "/usr/sbin" in TRUSTED_SYSTEM_DIRECTORIES + assert "/sbin" in TRUSTED_SYSTEM_DIRECTORIES + + def test_the_setuid_wrapper_precedes_the_nixos_symlink_farm(self) -> None: + """Order between the two NixOS entries decides which `sudo` wins. + + `/run/current-system/sw/bin` also contains a `sudo`, but a non-setuid one: + nix store paths cannot carry the setuid bit, which is the whole reason the + wrapper directory exists. `_is_executable_file` checks only that the + candidate is a file with an execute bit, so it cannot tell the two apart -- + tuple order is the only thing that does. + + Getting this backwards would resolve a `sudo` that cannot elevate, so + cross-user sessions would fail on a host where the correct binary was + present all along. The wrapper-before-/usr/bin assertion below does not + cover it, because sw/bin sits between them. + """ + directories = TRUSTED_SYSTEM_DIRECTORIES + + assert directories.index("/run/wrappers/bin") < directories.index( + "/run/current-system/sw/bin" + ) + + def test_the_two_nixos_entries_are_present_as_a_pair(self) -> None: + """/run/wrappers/bin alone supports no complete code path. + + It holds only the setuid wrappers, so on NixOS it resolves `sudo` and + nothing else -- /usr/bin has just `env`, /bin just `sh`, and the sbin + directories are absent. `setsid` and `pgrep` are in the sw/bin symlink + farm. Keeping the wrapper entry without that one would resolve `sudo` and + then fail on `setsid` one line later, so the ordering comment would be + describing support that does not exist. + """ + assert "/run/wrappers/bin" in TRUSTED_SYSTEM_DIRECTORIES + assert "/run/current-system/sw/bin" in TRUSTED_SYSTEM_DIRECTORIES + + +@pytest.mark.skipif( + not is_posix(), + reason="TRUSTED_SYSTEM_DIRECTORIES is a POSIX layout; on Windows none of these " + "directories exist and every lookup is expected to raise", +) +class TestRealCommandsResolve: + """Positive controls against the real filesystem. + + The tests above use temporary directories, so they would all pass on a host + where the genuine trusted directories were empty or misspelled. These assert + that the commands the library actually launches are found where it looks. + """ + + @pytest.mark.parametrize("name", ["sh", "ls"]) + def test_a_universally_present_command_resolves(self, name: str) -> None: + resolved = system_command_path(name) + + assert os.path.isabs(resolved) + assert os.path.dirname(resolved) in TRUSTED_SYSTEM_DIRECTORIES + + def test_sudo_resolves_on_this_host(self) -> None: + """sudo is the command the reported vulnerability abused. If this fails, + cross-user sessions cannot start on this host -- which is exactly the + portability regression that hardcoded literals introduced.""" + if find_system_command("sudo") is None: + pytest.skip("sudo is not installed on this host") + + assert os.path.dirname(system_command_path("sudo")) in TRUSTED_SYSTEM_DIRECTORIES + + +class TestCacheDoesNotRememberAbsence: + """Successes are cached; failures are not. + + A package manager replacing a binary unlinks and relinks it, so a lookup + landing in that window finds nothing. Caching that answer would make one + unlucky moment permanent for the rest of a long-lived agent's life: every + later cross-user launch would fail on a command that is sitting on disk. + """ + + def test_a_command_that_appears_later_is_found(self, tmp_path: Path) -> None: + # GIVEN a lookup that misses, as it would mid-upgrade + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(tmp_path),)): + assert find_system_command("target-cmd") is None + + # WHEN the binary appears + target = tmp_path / "target-cmd" + target.write_text("#!/bin/sh\ntrue\n") + target.chmod(target.stat().st_mode | stat.S_IXUSR) + + # THEN the next lookup finds it rather than repeating the cached miss + assert find_system_command("target-cmd") == str(target) + + def test_a_resolved_path_is_cached(self, executable_dir: Path) -> None: + """The other half of the asymmetry. Without this, 'do not cache misses' + could be satisfied by caching nothing at all, and the reason the cache + exists (these lookups sit on signal-delivery paths) would be lost.""" + # GIVEN one successful lookup + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", (str(executable_dir),)): + first = find_system_command("target-cmd") + assert first == str(executable_dir / "target-cmd") + + # WHEN the directory list no longer contains it, and the file is gone + (executable_dir / "target-cmd").unlink() + + # THEN the cached answer is still returned, without touching the filesystem + with patch(f"{_MODULE}.TRUSTED_SYSTEM_DIRECTORIES", ()): + assert find_system_command("target-cmd") == first diff --git a/test/openjd/sessions_v1/scenarios/parameter_types/all_types_template.yaml b/test/openjd/sessions_v1/scenarios/parameter_types/all_types_template.yaml index 48d5a29e..57920844 100644 --- a/test/openjd/sessions_v1/scenarios/parameter_types/all_types_template.yaml +++ b/test/openjd/sessions_v1/scenarios/parameter_types/all_types_template.yaml @@ -56,7 +56,7 @@ steps: echo PathParam={{repr_sh(Param.PathParam)}} echo PathParam.name={{repr_sh(Param.PathParam.name)}} echo BoolParam={{repr_sh(string(Param.BoolParam))}} - echo RangeExprParam={{repr_sh(Param.RangeExprParam)}} + echo RangeExprParam={{repr_sh(string(Param.RangeExprParam))}} echo len-RangeExprParam={{repr_sh(string(len(Param.RangeExprParam)))}} echo ListStringParam={{repr_sh(string(Param.ListStringParam))}} echo ListStringParam-0={{repr_sh(Param.ListStringParam[0])}}