diff --git a/src/agents/sandbox/sandboxes/unix_local.py b/src/agents/sandbox/sandboxes/unix_local.py index ea7f83e9d8..48cc9e1a39 100644 --- a/src/agents/sandbox/sandboxes/unix_local.py +++ b/src/agents/sandbox/sandboxes/unix_local.py @@ -21,12 +21,12 @@ import time import uuid from collections import deque -from collections.abc import Collection, Mapping, Sequence +from collections.abc import Awaitable, Collection, Mapping, Sequence from contextlib import suppress from dataclasses import dataclass, field from functools import partial from pathlib import Path -from typing import Literal, cast +from typing import Literal, TypeVar, cast from ...logger import log_tool_action_warning from .._mount_security import redact_mount_error_data @@ -101,15 +101,566 @@ ) logger = logging.getLogger(__name__) +_PROCESS_CLEANUP_TIMEOUT_SECONDS = 1.0 +_PROCESS_GROUP_WRAPPER_SCRIPT = """ +import os +import select +import subprocess +import sys + +control_fd = int(sys.argv[1]) +terminate_fd = int(sys.argv[2]) +status_fd = int(sys.argv[3]) +info_fd = int(sys.argv[4]) +target_user = sys.argv[5] or None +command = sys.argv[6:] + + +def close_fd(fd): + try: + os.close(fd) + except OSError: + pass + + +def write_all(fd, payload): + while payload: + try: + written = os.write(fd, payload) + except InterruptedError: + continue + except OSError: + return + if written <= 0: + return + payload = payload[written:] + + +def wait_for_eof(fd): + while True: + try: + if not os.read(fd, 4096): + return + except InterruptedError: + continue + except OSError: + return + + +def kill_as_user(user, args): + try: + subprocess.run( + ["sudo", "-n", "-u", user, "--", "kill", *args], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + timeout=0.25, + ) + except (OSError, subprocess.SubprocessError): + pass + + +def terminate_target(target_pid, group_id, *, can_signal_target): + # The direct target may be a sudo monitor while the actual command runs as + # another UID. Ask that UID to terminate its process group before falling + # back to the supervisor's direct signal. + if target_user: + kill_as_user(target_user, ["-KILL", f"-{group_id}"]) + if can_signal_target: + kill_as_user(target_user, ["-KILL", str(target_pid)]) + if can_signal_target: + try: + os.kill(target_pid, 9) + except OSError: + pass + + +def wait_for_termination_request(fd): + try: + ready, _, _ = select.select([fd], [], [], 0.05) + except (InterruptedError, OSError): + return False, False + if not ready: + return False, False + try: + return bool(os.read(fd, 4096)), True + except (InterruptedError, OSError): + return False, True + + +try: + keeper_ready_read_fd, keeper_ready_write_fd = os.pipe() + keeper_pid = os.fork() +except OSError: + os._exit(127) + +if keeper_pid == 0: + for fd in ( + terminate_fd, + status_fd, + info_fd, + keeper_ready_read_fd, + ): + close_fd(fd) + try: + os.setpgid(0, 0) + write_all(keeper_ready_write_fd, b"1") + except BaseException: + close_fd(keeper_ready_write_fd) + os._exit(127) + close_fd(keeper_ready_write_fd) + # Keep only the control fd open while the supervisor waits for the target + # and later reaps it. The target owns the output pipe descriptors. + close_fd(1) + close_fd(2) + wait_for_eof(control_fd) + os._exit(0) + +close_fd(keeper_ready_write_fd) +try: + keeper_ready = os.read(keeper_ready_read_fd, 1) +except OSError: + keeper_ready = b"" +close_fd(keeper_ready_read_fd) +if keeper_ready != b"1": + try: + os.waitpid(keeper_pid, 0) + except BaseException: + pass + os._exit(127) + +try: + target_pid = os.fork() +except OSError: + close_fd(control_fd) + close_fd(terminate_fd) + try: + os.kill(keeper_pid, 9) + os.waitpid(keeper_pid, 0) + except BaseException: + pass + os._exit(127) + +if target_pid == 0: + for fd in ( + control_fd, + terminate_fd, + status_fd, + info_fd, + ): + close_fd(fd) + try: + os.setpgid(0, keeper_pid) + os.execvpe(command[0], command, os.environ) + except BaseException: + os._exit(127) + +write_all(info_fd, f"{keeper_pid} {target_pid}".encode("ascii")) +close_fd(info_fd) + +target_exit_code = 127 +termination_requested = False +while True: + try: + waited_pid, wait_status = os.waitpid(target_pid, os.WNOHANG) + except InterruptedError: + continue + except ChildProcessError: + waited_pid = target_pid + wait_status = 127 << 8 + except OSError: + waited_pid = target_pid + wait_status = 127 << 8 + + if waited_pid == target_pid: + target_exit_code = os.waitstatus_to_exitcode(wait_status) + break + + requested, request_fd_ready = wait_for_termination_request(terminate_fd) + if requested and not termination_requested: + terminate_target(target_pid, keeper_pid, can_signal_target=True) + termination_requested = True + if request_fd_ready and not requested: + # EOF without a byte is the normal completion path. The parent closes + # this descriptor after collecting the status and output. + close_fd(terminate_fd) + terminate_fd = -1 + +write_all(status_fd, str(target_exit_code).encode("ascii")) +close_fd(status_fd) + +# Do not keep the parent's output pipes open while waiting for the cancellation +# control pipe. Any target descendants that inherited them still keep output +# draining observable to the parent. +close_fd(1) +close_fd(2) + +# A cancellation can race with the target's exit. Continue watching the +# termination pipe after reaping the direct child so a sudo-owned descendant +# can still be killed by process group, but never reuse the reaped target PID. +while True: + fds = [control_fd] + if terminate_fd >= 0: + fds.append(terminate_fd) + try: + ready, _, _ = select.select(fds, [], [], 0.05) + except (InterruptedError, OSError): + continue + if terminate_fd >= 0 and terminate_fd in ready: + try: + requested = bool(os.read(terminate_fd, 4096)) + except (InterruptedError, OSError): + requested = False + if requested and not termination_requested: + terminate_target(target_pid, keeper_pid, can_signal_target=False) + termination_requested = True + if not requested: + close_fd(terminate_fd) + terminate_fd = -1 + if control_fd in ready: + try: + control_data = os.read(control_fd, 4096) + except (InterruptedError, OSError): + control_data = b"" + if not control_data: + # The parent writes the termination request before closing either + # control descriptor. Drain it if both readiness notifications were + # delivered in the opposite order. + if terminate_fd >= 0: + try: + requested = bool(os.read(terminate_fd, 4096)) + except (InterruptedError, OSError): + requested = False + if requested and not termination_requested: + terminate_target(target_pid, keeper_pid, can_signal_target=False) + termination_requested = True + close_fd(terminate_fd) + terminate_fd = -1 + break + +close_fd(control_fd) +try: + os.waitpid(keeper_pid, 0) +except InterruptedError: + while True: + try: + os.waitpid(keeper_pid, 0) + break + except InterruptedError: + continue +except ChildProcessError: + pass + +# The parent uses the status pipe for the target's exact exit status. Use a +# conventional shell-style status for the supervisor's own process status. +wrapper_exit_code = target_exit_code if target_exit_code >= 0 else 128 - target_exit_code +os._exit(min(wrapper_exit_code, 255)) +""" +_SubprocessResultT = TypeVar("_SubprocessResultT") + + +async def _read_process_exit_code(fd: int) -> int: + loop = asyncio.get_running_loop() + result: asyncio.Future[int] = loop.create_future() + payload = bytearray() + + def _read_status() -> None: + if result.done(): + return + try: + chunk = os.read(fd, 4096) + except BlockingIOError: + return + except OSError as error: + result.set_exception(error) + return + + if chunk: + payload.extend(chunk) + return + + try: + exit_code = int(payload.decode("ascii")) + except ValueError as error: + invalid_status = RuntimeError( + "process-group supervisor returned an invalid exit status" + ) + invalid_status.__cause__ = error + result.set_exception(invalid_status) + else: + result.set_result(exit_code) + + os.set_blocking(fd, False) + loop.add_reader(fd, _read_status) + try: + _read_status() + return await result + finally: + loop.remove_reader(fd) + + +async def _read_process_group_info(fd: int) -> tuple[int, int]: + loop = asyncio.get_running_loop() + result: asyncio.Future[tuple[int, int]] = loop.create_future() + payload = bytearray() + + def _read_info() -> None: + if result.done(): + return + try: + chunk = os.read(fd, 4096) + except BlockingIOError: + return + except OSError as error: + result.set_exception(error) + return + + if chunk: + payload.extend(chunk) + return + + try: + values = tuple(int(value) for value in payload.decode("ascii").split()) + if len(values) != 2 or any(value <= 0 for value in values): + raise ValueError("expected positive keeper and target PIDs") + except ValueError as error: + invalid_info = RuntimeError("process-group supervisor returned invalid process info") + invalid_info.__cause__ = error + result.set_exception(invalid_info) + else: + result.set_result(cast(tuple[int, int], values)) + + os.set_blocking(fd, False) + loop.add_reader(fd, _read_info) + try: + _read_info() + return await result + finally: + loop.remove_reader(fd) + + +def _sudo_user_from_command(command: Sequence[str]) -> str | None: + if len(command) >= 3 and Path(command[0]).name == "sudo" and command[1] == "-u": + user = command[2] + if user and user != "--": + return user + return None + + +async def _read_process_stream(stream: asyncio.StreamReader | None) -> bytes: + if stream is None: + return b"" + return await stream.read() + + +async def _read_process_output( + proc: asyncio.subprocess.Process, +) -> tuple[bytes, bytes]: + stdout, stderr = await asyncio.gather( + _read_process_stream(proc.stdout), + _read_process_stream(proc.stderr), + ) + return stdout, stderr + + +async def _settle_subprocess_awaitable( + awaitable: Awaitable[_SubprocessResultT], +) -> tuple[_SubprocessResultT, bool]: + task = asyncio.ensure_future(awaitable) + completion = asyncio.create_task(asyncio.wait((task,))) + caller_cancelled = False + while not completion.done(): + try: + await asyncio.shield(completion) + except asyncio.CancelledError: + caller_cancelled = True + + completion.result() + try: + result = task.result() + except BaseException: + if caller_cancelled: + raise asyncio.CancelledError() from None + raise + return result, caller_cancelled + + +async def _kill_process_as_user(user: str, args: Sequence[str]) -> None: + try: + process = await asyncio.create_subprocess_exec( + "sudo", + "-n", + "-u", + user, + "--", + "kill", + *args, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await asyncio.wait_for(process.wait(), timeout=_PROCESS_CLEANUP_TIMEOUT_SECONDS) + except (OSError, asyncio.TimeoutError): + return + + +async def _terminate_process_group_and_reap_process( + proc: asyncio.subprocess.Process, + *, + process_group_id: int | None = None, + target_pid: int | None = None, + target_user: str | None = None, + process_info_task: asyncio.Task[tuple[int, int]] | None = None, + communication_task: asyncio.Task[tuple[bytes, bytes]] | None = None, + status_task: asyncio.Task[int] | None = None, + control_write_fd: int | None = None, + terminate_write_fd: int | None = None, +) -> None: + cleanup_deadline = asyncio.get_running_loop().time() + _PROCESS_CLEANUP_TIMEOUT_SECONDS + + def _remaining_cleanup_time() -> float: + return max(0.0, cleanup_deadline - asyncio.get_running_loop().time()) + + if process_info_task is not None: + try: + if process_info_task.done(): + discovered_group_id, discovered_target_pid = process_info_task.result() + else: + discovered_group_id, discovered_target_pid = await asyncio.wait_for( + asyncio.shield(process_info_task), + timeout=_remaining_cleanup_time(), + ) + except asyncio.CancelledError: + pass + except Exception: + pass + else: + if process_group_id is None: + process_group_id = discovered_group_id + if target_pid is None: + target_pid = discovered_target_pid + + group_id = process_group_id + if group_id is None and proc.returncode is None: + # The supervisor's PID is only a safe fallback while that direct child + # is still known to be alive; after it exits, its process group ID may + # already belong to an unrelated process. + group_id = proc.pid + # Keep the keeper alive until after the group signal. Closing control first + # would release the only stable owner of this group and allow its ID to be + # reused before killpg runs. + if group_id is not None: + with suppress(OSError): + os.killpg(group_id, signal.SIGKILL) + # Ask the supervisor to terminate the direct child and, when applicable, + # the target-user process group. It remains outside the keeper group so it + # can reap the target after the group signal. + if terminate_write_fd is not None: + with suppress(OSError): + os.write(terminate_write_fd, b"1") + _close_fd_quietly(control_write_fd) + _close_fd_quietly(terminate_write_fd) + try: + # Let the supervisor reap its target before using the direct-child kill + # fallback. This avoids orphaning a zombie in hosts whose PID 1 does + # not reap adopted children. + await asyncio.wait_for(proc.wait(), timeout=_remaining_cleanup_time()) + except asyncio.TimeoutError: + # This is only a fallback for a supervisor that could not finish its + # own reaping protocol. The target PID is still safe to use while the + # supervisor remains unreaped; ask its requested UID as well when the + # command was launched through sudo. + if target_user is not None and group_id is not None: + await _kill_process_as_user(target_user, ("-KILL", f"-{group_id}")) + if target_pid is not None and status_task is not None and not status_task.done(): + with suppress(OSError): + os.kill(target_pid, signal.SIGKILL) + if proc.returncode is None: + with suppress(OSError): + proc.kill() + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(proc.wait(), timeout=_remaining_cleanup_time()) + finally: + remaining_cleanup_time = _remaining_cleanup_time() + if remaining_cleanup_time > 0: + try: + if communication_task is None: + await asyncio.wait_for(proc.communicate(), timeout=remaining_cleanup_time) + else: + await asyncio.wait_for( + asyncio.shield(communication_task), + timeout=remaining_cleanup_time, + ) + except (asyncio.TimeoutError, ValueError): + # A descendant can escape the process group and keep the output + # pipes open. Cancellation must remain bounded in that case. + pass + for task in (communication_task, status_task, process_info_task): + if task is None: + continue + if not task.done(): + task.cancel() + with suppress(BaseException): + await task + # asyncio does not expose a public Process method for closing pipe + # transports after communicate() is cancelled. + transport = getattr(proc, "_transport", None) + if transport is not None: + transport.close() + + +async def _terminate_process_group_and_reap( + proc: asyncio.subprocess.Process, + *, + process_group_id: int | None = None, + target_pid: int | None = None, + target_user: str | None = None, + process_info_task: asyncio.Task[tuple[int, int]] | None = None, + communication_task: asyncio.Task[tuple[bytes, bytes]] | None = None, + status_task: asyncio.Task[int] | None = None, + control_write_fd: int | None = None, + terminate_write_fd: int | None = None, +) -> None: + # Keep process cleanup in an owned task so a second cancellation cannot + # interrupt the drain, direct-child reap, or transport close. + cleanup_task = asyncio.create_task( + _terminate_process_group_and_reap_process( + proc, + process_group_id=process_group_id, + target_pid=target_pid, + target_user=target_user, + process_info_task=process_info_task, + communication_task=communication_task, + status_task=status_task, + control_write_fd=control_write_fd, + terminate_write_fd=terminate_write_fd, + ) + ) + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + continue + except Exception: + break + try: + cleanup_task.result() + except Exception: + logger.debug( + "UnixLocal process cleanup failed after cancellation", + exc_info=True, + ) def _mount_path_diagnostic_extra(mount_path: Path) -> dict[str, object]: return {"mount_path": str(mount_path)} -def _close_fd_quietly(fd: int) -> None: +def _close_fd_quietly(fd: int | None) -> None: with suppress(OSError): - os.close(fd) + if fd is not None: + os.close(fd) def _restore_pty_child_signal_defaults() -> None: @@ -274,39 +825,204 @@ async def _exec_internal( workspace_root=workspace_root, cwd=cwd, ) + target_user = _sudo_user_from_command(command_parts) exec_command = self._confined_exec_command( command_parts=command_parts, workspace_root=workspace_root, env=env, ) + control_read_fd: int | None = None + control_write_fd: int | None = None + terminate_read_fd: int | None = None + terminate_write_fd: int | None = None + status_read_fd: int | None = None + status_write_fd: int | None = None + info_read_fd: int | None = None + info_write_fd: int | None = None + proc: asyncio.subprocess.Process | None = None + communication_task: asyncio.Task[tuple[bytes, bytes]] | None = None + status_task: asyncio.Task[int] | None = None + process_info_task: asyncio.Task[tuple[int, int]] | None = None + process_group_id: int | None = None + target_pid: int | None = None + target_exit_code: int | None = None try: - proc = await asyncio.create_subprocess_exec( - *exec_command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=process_cwd, - env=env, - start_new_session=True, + control_read_fd, control_write_fd = os.pipe() + terminate_read_fd, terminate_write_fd = os.pipe() + status_read_fd, status_write_fd = os.pipe() + info_read_fd, info_write_fd = os.pipe() + proc, proc_cancelled = await _settle_subprocess_awaitable( + asyncio.create_subprocess_exec( + sys.executable, + "-c", + _PROCESS_GROUP_WRAPPER_SCRIPT, + str(control_read_fd), + str(terminate_read_fd), + str(status_write_fd), + str(info_write_fd), + target_user or "", + *exec_command, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=process_cwd, + env=env, + pass_fds=( + control_read_fd, + terminate_read_fd, + status_write_fd, + info_write_fd, + ), + start_new_session=True, + ) ) - + _close_fd_quietly(control_read_fd) + control_read_fd = None + _close_fd_quietly(terminate_read_fd) + terminate_read_fd = None + _close_fd_quietly(status_write_fd) + status_write_fd = None + _close_fd_quietly(info_write_fd) + info_write_fd = None + + assert proc is not None + if proc_cancelled: + cleanup_control_fd, control_write_fd = control_write_fd, None + cleanup_terminate_fd, terminate_write_fd = terminate_write_fd, None + await _terminate_process_group_and_reap( + proc, + control_write_fd=cleanup_control_fd, + terminate_write_fd=cleanup_terminate_fd, + ) + raise asyncio.CancelledError() + + assert control_write_fd is not None + assert terminate_write_fd is not None + assert status_read_fd is not None + assert info_read_fd is not None + communication_task = asyncio.create_task(_read_process_output(proc)) + status_task = asyncio.create_task(_read_process_exit_code(status_read_fd)) + process_info_task = asyncio.create_task(_read_process_group_info(info_read_fd)) + deadline = None if timeout is None else asyncio.get_running_loop().time() + timeout try: - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + info_timeout = ( + None + if deadline is None + else max(0.0, deadline - asyncio.get_running_loop().time()) + ) + process_group_id, target_pid = await ( + process_info_task + if info_timeout is None + else asyncio.wait_for(asyncio.shield(process_info_task), info_timeout) + ) + process_info_task = None + status_timeout = ( + None + if deadline is None + else max(0.0, deadline - asyncio.get_running_loop().time()) + ) + target_exit_code = await ( + status_task + if status_timeout is None + else asyncio.wait_for(asyncio.shield(status_task), status_timeout) + ) + status_task = None + + communicate_timeout = ( + None + if deadline is None + else max(0.0, deadline - asyncio.get_running_loop().time()) + ) + stdout, stderr = await ( + communication_task + if communicate_timeout is None + else asyncio.wait_for(asyncio.shield(communication_task), communicate_timeout) + ) + communication_task = None + _close_fd_quietly(control_write_fd) + control_write_fd = None + _close_fd_quietly(terminate_write_fd) + terminate_write_fd = None + wait_timeout = ( + None + if deadline is None + else max(0.0, deadline - asyncio.get_running_loop().time()) + ) + _, proc_cancelled = await _settle_subprocess_awaitable( + proc.wait() + if wait_timeout is None + else asyncio.wait_for(proc.wait(), wait_timeout) + ) + if proc_cancelled: + raise asyncio.CancelledError() except asyncio.TimeoutError as e: - try: - # process tree cleanup - os.killpg(proc.pid, signal.SIGKILL) - except Exception: - pass + cleanup_control_fd, control_write_fd = control_write_fd, None + cleanup_terminate_fd, terminate_write_fd = terminate_write_fd, None + await _terminate_process_group_and_reap( + proc, + process_group_id=process_group_id, + target_pid=target_pid, + target_user=target_user, + process_info_task=process_info_task, + communication_task=communication_task, + status_task=status_task, + control_write_fd=cleanup_control_fd, + terminate_write_fd=cleanup_terminate_fd, + ) raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e + except asyncio.CancelledError: + cleanup_control_fd, control_write_fd = control_write_fd, None + cleanup_terminate_fd, terminate_write_fd = terminate_write_fd, None + await _terminate_process_group_and_reap( + proc, + process_group_id=process_group_id, + target_pid=target_pid, + target_user=target_user, + process_info_task=process_info_task, + communication_task=communication_task, + status_task=status_task, + control_write_fd=cleanup_control_fd, + terminate_write_fd=cleanup_terminate_fd, + ) + raise + except Exception: + cleanup_control_fd, control_write_fd = control_write_fd, None + cleanup_terminate_fd, terminate_write_fd = terminate_write_fd, None + await _terminate_process_group_and_reap( + proc, + process_group_id=process_group_id, + target_pid=target_pid, + target_user=target_user, + process_info_task=process_info_task, + communication_task=communication_task, + status_task=status_task, + control_write_fd=cleanup_control_fd, + terminate_write_fd=cleanup_terminate_fd, + ) + raise except ExecTimeoutError: raise except Exception as e: raise ExecTransportError(command=command, cause=e) from e - - return ExecResult( - stdout=stdout or b"", stderr=stderr or b"", exit_code=proc.returncode or 0 - ) + finally: + _close_fd_quietly(control_read_fd) + control_read_fd = None + _close_fd_quietly(control_write_fd) + control_write_fd = None + _close_fd_quietly(terminate_write_fd) + terminate_write_fd = None + _close_fd_quietly(status_read_fd) + status_read_fd = None + _close_fd_quietly(status_write_fd) + status_write_fd = None + _close_fd_quietly(info_read_fd) + info_read_fd = None + _close_fd_quietly(info_write_fd) + info_write_fd = None + + assert target_exit_code is not None + return ExecResult(stdout=stdout or b"", stderr=stderr or b"", exit_code=target_exit_code) async def pty_exec_start( self, diff --git a/tests/sandbox/test_unix_local.py b/tests/sandbox/test_unix_local.py index fb13be3c51..f839e12117 100644 --- a/tests/sandbox/test_unix_local.py +++ b/tests/sandbox/test_unix_local.py @@ -2,10 +2,14 @@ import asyncio import io +import os +import shlex import signal +import sys import tarfile import threading import time +from contextlib import suppress from pathlib import Path from types import SimpleNamespace from typing import cast @@ -46,6 +50,658 @@ async def _exec_internal( return ExecResult(stdout=b"", stderr=b"", exit_code=0) +async def _wait_for_integer_file(path: Path, count: int) -> tuple[int, ...]: + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + try: + values = tuple(int(value) for value in path.read_text().split()) + except (FileNotFoundError, ValueError): + values = () + if len(values) == count: + return values + await asyncio.sleep(0.01) + raise AssertionError(f"did not publish {count} integers to {path}") + + +@pytest.mark.parametrize( + "leader_waits", + [pytest.param(True, id="leader-waits"), pytest.param(False, id="leader-exits")], +) +@pytest.mark.asyncio +async def test_unix_local_exec_cancellation_terminates_process_group( + tmp_path: Path, leader_waits: bool +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + pid_file = workspace / "pids" + final_command = 'wait "$child"' if leader_waits else "exit 0" + command = ( + f'sleep 30 & child=$!; printf \'%s %s\' "$$" "$child" > {shlex.quote(str(pid_file))}; ' + f"{final_command}" + ) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + task = asyncio.create_task(session._exec_internal("sh", "-c", command)) + shell_pid: int | None = None + child_pid: int | None = None + process_group_id: int | None = None + try: + shell_pid, child_pid = await _wait_for_integer_file(pid_file, 2) + process_group_id = os.getpgid(child_pid) + + def is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + if not leader_waits: + # Give the shell time to exit while the background child keeps the + # subprocess pipes open. This exercises cleanup after the leader's + # return code has already been set. + await asyncio.sleep(0.1) + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + + deadline = time.monotonic() + 1 + while time.monotonic() < deadline and (is_alive(shell_pid) or is_alive(child_pid)): + await asyncio.sleep(0.01) + assert not is_alive(shell_pid) + assert not is_alive(child_pid) + finally: + if process_group_id is not None: + with suppress(OSError): + os.killpg(process_group_id, signal.SIGKILL) + + +@pytest.mark.asyncio +async def test_unix_local_exec_cancellation_does_not_wait_for_escaped_descendant( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The regression needs the test runner's Python to call setsid. Darwin's + # filesystem profile cannot necessarily read that interpreter, while the + # cancellation behavior under test is platform-independent Unix logic. + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + pid_file = workspace / "pids" + escaped_pid_file = workspace / "escaped-pid" + escaped_code = ( + "import os, time; " + f"os.setsid(); open({str(escaped_pid_file)!r}, 'w').write(str(os.getpid())); " + "time.sleep(30)" + ) + escaped_command = f"{shlex.quote(sys.executable)} -c {shlex.quote(escaped_code)}" + command = ( + f'{escaped_command} & child=$!; printf \'%s %s\' "$$" "$child" > ' + f'{shlex.quote(str(pid_file))}; wait "$child"' + ) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + task = asyncio.create_task(session._exec_internal("sh", "-c", command)) + shell_pid: int | None = None + escaped_pid: int | None = None + process_group_id: int | None = None + + def is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + try: + shell_pid, _background_pid = await _wait_for_integer_file(pid_file, 2) + process_group_id = os.getpgid(shell_pid) + + (escaped_pid,) = await _wait_for_integer_file(escaped_pid_file, 1) + + task.cancel() + done, _ = await asyncio.wait({task}, timeout=2) + assert task in done, "cancellation waited for a descendant outside the process group" + with pytest.raises(asyncio.CancelledError): + task.result() + + assert not is_alive(shell_pid) + assert is_alive(escaped_pid) + finally: + if process_group_id is not None: + with suppress(OSError): + os.killpg(process_group_id, signal.SIGKILL) + if escaped_pid is not None: + with suppress(OSError): + os.kill(escaped_pid, signal.SIGKILL) + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_unix_local_exec_cancellation_retains_process_group_after_leader_exits( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The process-group wrapper must prevent the original group ID from being + # reused while cancellation is still able to clean up the group. + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + pid_file = workspace / "pid" + process_group_file = workspace / "pgid" + escaped_pid_file = workspace / "escaped-pid" + escaped_code = ( + "import os, time; " + f"open({str(process_group_file)!r}, 'w').write(str(os.getpgid(0))); " + f"os.setsid(); open({str(escaped_pid_file)!r}, 'w').write(str(os.getpid())); " + "time.sleep(30)" + ) + escaped_command = f"{shlex.quote(sys.executable)} -c {shlex.quote(escaped_code)}" + command = f'{escaped_command} & printf "%s" "$$" > {shlex.quote(str(pid_file))}; exit 0' + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + task = asyncio.create_task(session._exec_internal("sh", "-c", command)) + shell_pid: int | None = None + escaped_pid: int | None = None + process_group_id: int | None = None + + def is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + try: + (shell_pid,) = await _wait_for_integer_file(pid_file, 1) + (process_group_id,) = await _wait_for_integer_file(process_group_file, 1) + (escaped_pid,) = await _wait_for_integer_file(escaped_pid_file, 1) + await asyncio.sleep(0.1) + + try: + assert process_group_id is not None + os.killpg(process_group_id, 0) + except ProcessLookupError as exc: + raise AssertionError("process group disappeared before cancellation") from exc + + task.cancel() + done, _ = await asyncio.wait({task}, timeout=2) + assert task in done, "cancellation waited for a descendant outside the process group" + with pytest.raises(asyncio.CancelledError): + task.result() + assert is_alive(escaped_pid) + finally: + if process_group_id is not None: + with suppress(OSError): + os.killpg(process_group_id, signal.SIGKILL) + if escaped_pid is not None: + with suppress(OSError): + os.kill(escaped_pid, signal.SIGKILL) + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_unix_local_exec_does_not_expose_wrapper_to_child_waitpid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The wrapper must not remain a child of the command. Otherwise a command + # that waits for all of its own children can wait forever for the wrapper's + # cancellation control pipe to close. + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + command = ( + "import os\n" + "child = os.fork()\n" + "if child == 0:\n" + " os._exit(0)\n" + "while True:\n" + " try:\n" + " os.waitpid(-1, 0)\n" + " except ChildProcessError:\n" + " break\n" + "os.write(1, b'done\\n')\n" + ) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + result = await asyncio.wait_for( + session._exec_internal(sys.executable, "-c", command), + timeout=1, + ) + + assert result.stdout == b"done\n" + assert result.stderr == b"" + + +@pytest.mark.asyncio +async def test_unix_local_exec_uses_spawn_safe_process_group_wrapper( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + subprocess_calls: list[tuple[object, ...]] = [] + closed_fds: list[int] = [] + + class _Process: + pid = 123 + returncode = 0 + + async def communicate(self) -> tuple[bytes, bytes]: + return b"output", b"" + + async def wait(self) -> int: + return self.returncode + + def kill(self) -> None: + raise AssertionError("normal execution should not kill the process group") + + process = _Process() + + async def create_subprocess(*args: object, **kwargs: object) -> _Process: + subprocess_calls.append(args) + assert kwargs["start_new_session"] is True + assert "preexec_fn" not in kwargs + return process + + async def read_status(_fd: int) -> int: + return 0 + + async def read_info(_fd: int) -> tuple[int, int]: + return 123, 456 + + async def read_output(_process: object) -> tuple[bytes, bytes]: + return b"output", b"" + + close_fd = unix_local_module._close_fd_quietly + + def record_close(fd: int | None) -> None: + if fd is not None: + closed_fds.append(fd) + close_fd(fd) + + monkeypatch.setattr(unix_local_module.asyncio, "create_subprocess_exec", create_subprocess) + monkeypatch.setattr(unix_local_module, "_read_process_exit_code", read_status) + monkeypatch.setattr(unix_local_module, "_read_process_group_info", read_info) + monkeypatch.setattr(unix_local_module, "_read_process_output", read_output) + monkeypatch.setattr(unix_local_module, "_close_fd_quietly", record_close) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + result = await session._exec_internal("echo", "hello") + + assert result.stdout == b"output" + assert result.stderr == b"" + assert result.exit_code == 0 + assert len(subprocess_calls) == 1 + args = subprocess_calls[0] + assert args[:3] == ( + sys.executable, + "-c", + unix_local_module._PROCESS_GROUP_WRAPPER_SCRIPT, + ) + assert args[8:] == ("echo", "hello") + assert len(closed_fds) == len(set(closed_fds)) + + +@pytest.mark.parametrize( + ("command", "expected_user"), + [ + (("sudo", "-u", "sandbox-user", "--", "echo"), "sandbox-user"), + (("/usr/bin/sudo", "-u", "sandbox-user", "--", "echo"), "sandbox-user"), + (("echo", "sudo", "-u", "sandbox-user"), None), + ], +) +def test_unix_local_exec_extracts_target_user_only_from_command_prefix( + command: tuple[str, ...], expected_user: str | None +) -> None: + assert unix_local_module._sudo_user_from_command(command) == expected_user + + +@pytest.mark.asyncio +async def test_unix_local_exec_keeps_command_out_of_host_session( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + parent_session_id = os.getsid(0) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + result = await session._exec_internal( + sys.executable, + "-c", + "import os; print(os.getsid(0), os.getpgrp())", + ) + + command_session_id, command_process_group_id = (int(value) for value in result.stdout.split()) + assert command_session_id != parent_session_id + assert command_process_group_id != parent_session_id + assert command_process_group_id != command_session_id + + +@pytest.mark.asyncio +async def test_unix_local_exec_preserves_command_exit_code( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(unix_local_module.sys, "platform", "linux") + workspace = tmp_path / "workspace" + workspace.mkdir() + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + result = await session._exec_internal(sys.executable, "-c", "raise SystemExit(17)") + + assert result.exit_code == 17 + + +@pytest.mark.asyncio +async def test_unix_local_exec_cleanup_survives_repeated_cancellation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + communicate_started = asyncio.Event() + wait_called = asyncio.Event() + transport_closed = False + + class _Transport: + def close(self) -> None: + nonlocal transport_closed + transport_closed = True + + class _Process: + pid = 123 + returncode = None + _transport = _Transport() + + async def communicate(self) -> tuple[bytes, bytes]: + communicate_started.set() + await asyncio.Event().wait() + return b"", b"" + + def kill(self) -> None: + pass + + async def wait(self) -> int: + wait_called.set() + return -signal.SIGKILL + + monkeypatch.setattr(unix_local_module.os, "killpg", lambda *_args: None) + monkeypatch.setattr(unix_local_module, "_PROCESS_CLEANUP_TIMEOUT_SECONDS", 0.01) + + process = cast(asyncio.subprocess.Process, _Process()) + task = asyncio.create_task(unix_local_module._terminate_process_group_and_reap(process)) + await communicate_started.wait() + + task.cancel() + await asyncio.sleep(0) + task.cancel() + + await asyncio.wait_for(task, timeout=1) + assert wait_called.is_set() + assert transport_closed + + +@pytest.mark.asyncio +async def test_unix_local_exec_cancellation_survives_unreapable_output_cleanup( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + communicate_started = asyncio.Event() + wait_called = asyncio.Event() + transport_closed = False + communicate_calls = 0 + + class _Transport: + def close(self) -> None: + nonlocal transport_closed + transport_closed = True + + class _Process: + pid = 123 + returncode = None + _transport = _Transport() + + async def communicate(self) -> tuple[bytes, bytes]: + nonlocal communicate_calls + communicate_calls += 1 + communicate_started.set() + await asyncio.Event().wait() + return b"", b"" + + def kill(self) -> None: + pass + + async def wait(self) -> int: + wait_called.set() + return -signal.SIGKILL + + process = _Process() + + async def _create_process(*_args: object, **_kwargs: object) -> _Process: + return process + + async def _read_status(_fd: int) -> int: + await asyncio.Event().wait() + return 0 + + async def _read_info(_fd: int) -> tuple[int, int]: + return 123, 456 + + async def _read_output(_process: object) -> tuple[bytes, bytes]: + return await process.communicate() + + monkeypatch.setattr( + unix_local_module.asyncio, + "create_subprocess_exec", + _create_process, + ) + monkeypatch.setattr(unix_local_module, "_read_process_exit_code", _read_status) + monkeypatch.setattr(unix_local_module, "_read_process_group_info", _read_info) + monkeypatch.setattr(unix_local_module, "_read_process_output", _read_output) + monkeypatch.setattr(unix_local_module.os, "killpg", lambda *_args: None) + monkeypatch.setattr(unix_local_module, "_PROCESS_CLEANUP_TIMEOUT_SECONDS", 0.01) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest(root=str(workspace)), + snapshot=NoopSnapshot(id="noop"), + ) + ) + task = asyncio.create_task(session._exec_internal("sh", "-c", "sleep 30")) + try: + await asyncio.wait_for(communicate_started.wait(), timeout=1) + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + + assert communicate_calls == 1 + assert wait_called.is_set() + assert transport_closed + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_unix_local_exec_cleanup_kills_direct_child_when_group_signal_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + communicate_started = asyncio.Event() + wait_called = asyncio.Event() + direct_kill_called = False + transport_closed = False + + class _Transport: + def close(self) -> None: + nonlocal transport_closed + transport_closed = True + + class _Process: + pid = 123 + returncode = None + _transport = _Transport() + + async def communicate(self) -> tuple[bytes, bytes]: + communicate_started.set() + await asyncio.Event().wait() + return b"", b"" + + def kill(self) -> None: + nonlocal direct_kill_called + direct_kill_called = True + + async def wait(self) -> int: + wait_called.set() + return -signal.SIGKILL + + def _deny_group_signal(*_args: object) -> None: + raise PermissionError("not allowed to signal a group member") + + monkeypatch.setattr(unix_local_module.os, "killpg", _deny_group_signal) + monkeypatch.setattr(unix_local_module, "_PROCESS_CLEANUP_TIMEOUT_SECONDS", 0.01) + + process = cast(asyncio.subprocess.Process, _Process()) + task = asyncio.create_task(unix_local_module._terminate_process_group_and_reap(process)) + await communicate_started.wait() + + await asyncio.wait_for(task, timeout=1) + assert not direct_kill_called + assert wait_called.is_set() + assert transport_closed + + +@pytest.mark.asyncio +async def test_unix_local_exec_cleanup_does_not_kill_exited_direct_child( + monkeypatch: pytest.MonkeyPatch, +) -> None: + communicate_started = asyncio.Event() + wait_called = asyncio.Event() + direct_kill_called = False + transport_closed = False + + class _Transport: + def close(self) -> None: + nonlocal transport_closed + transport_closed = True + + class _Process: + pid = 123 + returncode = 0 + _transport = _Transport() + + async def communicate(self) -> tuple[bytes, bytes]: + communicate_started.set() + await asyncio.Event().wait() + return b"", b"" + + def kill(self) -> None: + nonlocal direct_kill_called + direct_kill_called = True + + async def wait(self) -> int: + wait_called.set() + return self.returncode + + def _deny_group_signal(*_args: object) -> None: + raise PermissionError("not allowed to signal a group member") + + monkeypatch.setattr(unix_local_module.os, "killpg", _deny_group_signal) + monkeypatch.setattr(unix_local_module, "_PROCESS_CLEANUP_TIMEOUT_SECONDS", 0.01) + + process = cast(asyncio.subprocess.Process, _Process()) + task = asyncio.create_task(unix_local_module._terminate_process_group_and_reap(process)) + await communicate_started.wait() + + await asyncio.wait_for(task, timeout=1) + assert not direct_kill_called + assert wait_called.is_set() + assert transport_closed + + +@pytest.mark.asyncio +async def test_unix_local_exec_cleanup_bounds_unreapable_direct_child( + monkeypatch: pytest.MonkeyPatch, +) -> None: + communicate_started = asyncio.Event() + wait_started = asyncio.Event() + transport_closed = False + + class _Transport: + def close(self) -> None: + nonlocal transport_closed + transport_closed = True + + class _Process: + pid = 123 + returncode = None + _transport = _Transport() + + async def communicate(self) -> tuple[bytes, bytes]: + communicate_started.set() + await asyncio.Event().wait() + return b"", b"" + + def kill(self) -> None: + raise PermissionError("not allowed to signal the direct child") + + async def wait(self) -> int: + wait_started.set() + await asyncio.Event().wait() + return -signal.SIGKILL + + def _deny_group_signal(*_args: object) -> None: + raise PermissionError("not allowed to signal a group member") + + monkeypatch.setattr(unix_local_module.os, "killpg", _deny_group_signal) + monkeypatch.setattr(unix_local_module, "_PROCESS_CLEANUP_TIMEOUT_SECONDS", 0.01) + + process = cast(asyncio.subprocess.Process, _Process()) + task = asyncio.create_task(unix_local_module._terminate_process_group_and_reap(process)) + await wait_started.wait() + + await asyncio.wait_for(task, timeout=1) + assert wait_started.is_set() + assert transport_closed + + @pytest.mark.asyncio async def test_unix_local_inherits_host_environment_by_default( tmp_path: Path,