diff --git a/plugins/home/README.md b/plugins/home/README.md new file mode 100644 index 00000000..c3470af3 --- /dev/null +++ b/plugins/home/README.md @@ -0,0 +1,25 @@ +# Home plugin + +Provides the `auto_bash` job through an independently enabled ReMe plugin. + +Install it into the same Python environment as ReMe: + +```bash +reme plugins install plugins/home +``` + +Enable it in an application config with `plugins: [home]`. The plugin then +contributes the `auto_bash` job, which can be run with +`reme start config= job=auto_bash command=pwd`. + +The job defaults to read-only command validation and a 60-second timeout. Set +`is_read_only=false` to disable command validation. Supplying `read_only_paths` +or `writable_paths` enables a file sandbox (Bubblewrap on Linux, Seatbelt on +macOS), with sandbox paths identical to their host paths. On Linux, the +`memory_limit_mib` cgroup limit defaults to 4096 MiB; 0 disables it. Supplying +the limit requires a path policy. macOS has no memory limit and warns when a +nonzero value is supplied. + +Standard output and standard error each have a 1 MiB (1,048,576-byte) limit on +all backends. Exceeding either limit terminates the command and returns a +failure with bounded partial output and a diagnostic in standard error. diff --git a/plugins/home/pyproject.toml b/plugins/home/pyproject.toml new file mode 100644 index 00000000..82e54aed --- /dev/null +++ b/plugins/home/pyproject.toml @@ -0,0 +1,29 @@ +[project] +name = "reme-home" +version = "0.1.0" +description = "Local command execution jobs for ReMe." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "reme-ai[as]>=0.4.1.14", + "safecmd>=0.1.16,<0.2", +] + +[project.entry-points."reme.plugins"] +home = "reme_home" + +[tool.setuptools.packages.find] +where = ["src"] +include = ["reme_home*"] + +[tool.setuptools.package-data] +reme_home = ["plugin.yaml"] + +[tool.pytest.ini_options] +asyncio_default_fixture_loop_scope = "function" +pythonpath = ["src", "../.."] +testpaths = ["tests"] + +[build-system] +requires = ["setuptools>=77", "wheel"] +build-backend = "setuptools.build_meta" diff --git a/plugins/home/src/reme_home/__init__.py b/plugins/home/src/reme_home/__init__.py new file mode 100644 index 00000000..d3dd9b20 --- /dev/null +++ b/plugins/home/src/reme_home/__init__.py @@ -0,0 +1,5 @@ +"""Home plugin: local command execution jobs for ReMe.""" + +from .auto_bash import AutoBash, AutoBashStep + +__all__ = ["AutoBash", "AutoBashStep"] diff --git a/plugins/home/src/reme_home/auto_bash.py b/plugins/home/src/reme_home/auto_bash.py new file mode 100644 index 00000000..a464103d --- /dev/null +++ b/plugins/home/src/reme_home/auto_bash.py @@ -0,0 +1,281 @@ +"""Run allowlisted shell commands without an approval prompt.""" + +import asyncio +import os +import shlex +from functools import cache + +from agentscope.message import TextBlock, ToolResultState +from agentscope.permission import PermissionBehavior, PermissionContext, PermissionDecision +from agentscope.tool import Bash, ToolChunk + +from reme.steps.base_step import BaseStep +from reme.steps.common.file_sandbox import file_sandbox_for_step + +from reme.utils.local_backend import LocalProcessBackend + +_READ_COMMANDS = { + ":": {}, + "cat": {}, + "cd": {}, + "cmp": {}, + "diff": {"denied": ["--output"]}, + "echo": {}, + "false": {}, + "file": { + "denied": ["-C", "--compile", "-p", "--preserve-date", "-z", "-Z", "--uncompress", "--uncompress-noreport"], + }, + "find": {"denied": ["-delete", "-exec", "-execdir", "-fls", "-fprint", "-fprint0", "-fprintf", "-ok", "-okdir"]}, + "git": { + "allow": ["status", "diff", "log", "show", "rev-parse", "ls-files"], + "denied": [ + "--output", + "--ext-diff", + "--textconv", + "--paginate", + "--submodule", + "--ignore-submodules", + "--show-signature", + "--remerge-diff", + "--diff-merges", + "--pretty", + "--format", + ], + }, + "grep": {}, + "head": {}, + "ls": {}, + "pwd": {}, + "sort": {"denied": ["-o", "-T", "--output", "--compress-program", "--temporary-directory"]}, + "stat": {}, + "tail": {}, + "true": {}, + "wc": {}, + "which": {}, +} +_SAFE_DESTINATIONS = {"/dev/null"} + +# Trusted shell setup, applied only after validating the user's command. +# Git's configuration can launch helpers even for otherwise read-only commands. +_READ_ONLY_SETUP = r""" +unset GIT_EXTERNAL_DIFF +export GIT_OPTIONAL_LOCKS=0 GIT_PAGER=/bin/cat GIT_NO_LAZY_FETCH=1 GIT_TERMINAL_PROMPT=0 +sort() { + # A large sort otherwise writes temporary files, even without -o. + command sort -T /dev/null "$@" +} +git() { + # These metadata queries do not read file contents through Git filters. + if [ "$1" != rev-parse ] && [ "$1" != ls-files ] && \ + command git -c trace2.normalTarget=0 -c trace2.eventTarget=0 -c trace2.perfTarget=0 \ + config --get-regexp '^filter\..*\.(clean|smudge|process)$' > /dev/null; then + echo 'Permission denied: Git repositories with external filters require unrestricted mode.' >&2 + return 1 + fi + case "$1" in + diff|log|show) + reme_git_subcommand="$1" + shift + set -- "$reme_git_subcommand" --no-ext-diff --no-textconv --ignore-submodules=all "$@" + ;; + status) shift; set -- status --ignore-submodules=all "$@" ;; + esac + command git --no-pager -c core.fsmonitor=false -c diff.submodule=short \ + -c log.showSignature=false -c log.diffMerges=off -c format.pretty=medium \ + -c trace2.normalTarget=0 -c trace2.eventTarget=0 -c trace2.perfTarget=0 "$@" +} +""" + + +@cache +def _read_specs(): + """Build the fixed safecmd policy once, when read-only mode is used.""" + from safecmd import CmdSpec # pylint: disable=import-outside-toplevel + + return {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()} + + +def _literal_shell(node: dict, command: str, quoted: bool = False) -> bool: + """Reject expansions and syntax whose effects safecmd does not fully check.""" + from safecmd.bashxtract import word_text # pylint: disable=import-outside-toplevel + + kind = node.get("Type") + if kind in {"FuncDecl", "ForClause", "DeclClause", "ArithmExp", "ArithmCmd", "ProcSubst", "CmdSubst"}: + return False + if kind == "SglQuoted": + return True + quoted = quoted or kind == "DblQuoted" + if kind == "Lit": + value = node.get("Value", "") + if "\\" in value or (not quoted and any(char in value for char in "*?[]{}~")): + return False + for redirect in node.get("Redirs", []): + op = redirect.get("Op") + destination = word_text(redirect.get("Word", {}), command) + if op in {"<", "<<", "<<-"}: + continue + if op in {">&", "<&"} and (destination == "-" or destination.rstrip("-").isdigit()): + continue + if op not in {">", ">>", ">|", "<>", "&>", "&>>", ">&"} or destination != "/dev/null": + return False + for value in node.values(): + children = value if isinstance(value, list) else [value] + if any(isinstance(child, dict) and not _literal_shell(child, command, quoted) for child in children): + return False + return True + + +def _allowed_arguments(tokens: list[str]) -> bool: + """Supplement CmdSpec with subcommands, abbreviated flags and Git formats.""" + options = _READ_COMMANDS[tokens[0]] + if "allow" in options and (len(tokens) < 2 or tokens[1] not in options["allow"]): + return False + # Signature placeholders invoke Git's configured GPG program. + if tokens[0] == "git" and any("%G" in token for token in tokens[1:]): + return False + # getopt and Git also accept unambiguous long-option abbreviations. + for token in tokens[1:]: + flag = token.split("=", 1)[0] + if flag.startswith("--") and len(flag) > 2: + if any(denied.startswith(flag) for denied in options.get("denied", []) if denied.startswith("--")): + return False + return True + + +class AutoBash(Bash): + """AgentScope Bash with an optional conservative read-only command guard. + + Shell expansions, function definitions and for loops are rejected because + their runtime arguments or environment can differ from the parsed tokens. + Git external filters require unrestricted mode; other Git helpers are + disabled. Use path injection for OS file isolation. + """ + + name = "AutoBash" + description = "Execute shell commands; read-only mode denies commands that may write or run other commands." + input_schema = { + **Bash.input_schema, + "properties": { + **Bash.input_schema["properties"], + "timeout": { + **Bash.input_schema["properties"]["timeout"], + "description": "Optional timeout in milliseconds (default: 60000, max: 600000)", + "default": 60000, + }, + }, + } + + def __init__(self, *, is_read_only: bool = True, **kwargs): + if kwargs.get("backend") is None: + kwargs["backend"] = LocalProcessBackend() + super().__init__(**kwargs) + self.is_read_only = is_read_only + + @staticmethod + def _is_read_only(command: str) -> bool: + """Use safecmd's Bash parser and a strict command policy.""" + if not isinstance(command, str) or not command.strip(): + return False + # Variable and command expansion can hide a second operation in an + # argument. Keep this first version literal while safecmd checks the + # remaining shell syntax and nested commands. + if "$" in command or "`" in command: + return False + try: + from safecmd.bashxtract import ( # pylint: disable=import-outside-toplevel + check_types, + collect_ops, + collect_redirects, + parse_bash, + visit_stmts, + ) + from safecmd.core import normalize_dest, validate_cmd # pylint: disable=import-outside-toplevel + + ast = parse_bash(command) + check_types(ast) + if not _literal_shell(ast, command): + return False + commands = visit_stmts(ast.get("Stmts", []), command) + specs = _read_specs() + if any(not validate_cmd(tokens, specs) or not _allowed_arguments(tokens) for tokens in commands): + return False + return "=" not in collect_ops(ast) and all( + normalize_dest(destination) in _SAFE_DESTINATIONS for _, destination in collect_redirects(ast, command) + ) + except Exception: # pylint: disable=broad-exception-caught + # Parser failures and missing shfmt must never allow execution. + return False + + async def check_read_only(self, tool_input: dict) -> bool: + """Report the same policy used by the execution guard.""" + return self.is_read_only and await asyncio.to_thread(self._is_read_only, tool_input.get("command", "")) + + async def check_permissions(self, tool_input: dict, _context: PermissionContext) -> PermissionDecision: + """Resolve permission automatically, without an ASK decision.""" + if not self.is_read_only: + return PermissionDecision(behavior=PermissionBehavior.ALLOW, message="Command allowed") + allowed = await self.check_read_only(tool_input) + return PermissionDecision( + behavior=PermissionBehavior.ALLOW if allowed else PermissionBehavior.DENY, + message=( + "Read-only command allowed" if allowed else "Permission denied: only read-only commands are allowed" + ), + ) + + async def call(self, command: str, description: str = "", timeout: int = 60000): + """Apply the read-only guard when configured, then run the command.""" + if self.is_read_only and not await asyncio.to_thread(self._is_read_only, command): + yield ToolChunk( + content=[TextBlock(text="Permission denied: only read-only shell commands are allowed.")], + state=ToolResultState.ERROR, + is_last=True, + ) + return + if self.is_read_only: + # Trace destinations can write files even for metadata-only Git + # commands. Clear the entire inherited family on every invocation, + # including new Git trace variables, without changing os.environ. + trace_variables = " ".join(shlex.quote(name) for name in os.environ if name.startswith("GIT_TRACE")) + command = f"unset {trace_variables}\n" + _READ_ONLY_SETUP + command + async for chunk in super().call(command=command, description=description, timeout=timeout): + yield chunk + + +class AutoBashStep(BaseStep): + """Expose AgentScope Bash with configurable read-only mode as a ReMe job.""" + + async def execute(self): + """Run the command with the selected permission policy and backend.""" + assert self.context is not None + command = self.context.get("command", "") + timeout = self.context.get("timeout", self.kwargs.get("timeout", 60000)) + is_read_only = self.context.get("is_read_only", self.kwargs.get("is_read_only", True)) + if not isinstance(command, str) or not command.strip(): + self.context.response.success = False + self.context.response.answer = "command is required" + return self.context.response + if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0: + self.context.response.success = False + self.context.response.answer = "timeout must be a positive integer in milliseconds" + return self.context.response + if not isinstance(is_read_only, bool): + self.context.response.success = False + self.context.response.answer = "is_read_only must be a boolean" + return self.context.response + + try: + backend = file_sandbox_for_step(self) + except (TypeError, ValueError, RuntimeError) as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + return self.context.response + + tool = AutoBash(cwd=self.workspace_path, is_read_only=is_read_only, backend=backend) + success = True + texts = [] + async for chunk in tool.call(command=command, timeout=timeout): + success = success and chunk.state != ToolResultState.ERROR + texts.extend(block.text for block in chunk.content if isinstance(block, TextBlock)) + self.context.response.success = success + self.context.response.answer = "".join(texts) + return self.context.response diff --git a/plugins/home/src/reme_home/plugin.yaml b/plugins/home/src/reme_home/plugin.yaml new file mode 100644 index 00000000..77ad1584 --- /dev/null +++ b/plugins/home/src/reme_home/plugin.yaml @@ -0,0 +1,40 @@ +backends: + auto_bash_step: reme_home.auto_bash:AutoBashStep + +application_defaults: + jobs: + auto_bash: + backend: base + description: "Run a shell command in the workspace; read-only mode is enabled by default" + parameters: + type: object + properties: + command: + type: string + description: "shell command to execute" + is_read_only: + type: boolean + description: "restrict execution to read-only commands" + default: true + timeout: + type: integer + description: "maximum execution time in milliseconds (default 60000, max 600000)" + default: 60000 + minimum: 1 + maximum: 600000 + read_only_paths: + type: array + items: { type: string } + description: "Read-only host paths; providing either path list enables the platform file sandbox" + writable_paths: + type: array + items: { type: string } + description: "Writable host paths; sandbox paths match host paths" + memory_limit_mib: + type: integer + minimum: 0 + description: "Linux cgroup memory limit in MiB (default 4096; 0 disables it); requires a path policy; macOS warns and ignores nonzero values" + required: + - command + steps: + - backend: auto_bash_step diff --git a/plugins/home/tests/test_auto_bash.py b/plugins/home/tests/test_auto_bash.py new file mode 100644 index 00000000..9ad9f466 --- /dev/null +++ b/plugins/home/tests/test_auto_bash.py @@ -0,0 +1,606 @@ +"""Shell job and its optional read-only execution guard.""" + +import asyncio +import inspect +import os +import subprocess +import threading +from importlib.metadata import EntryPoint +from pathlib import Path + +import pytest +import psutil +import yaml + +from agentscope.message import ToolResultState +from agentscope.permission import PermissionBehavior +from agentscope.tool import ExecResult +from reme_home.auto_bash import AutoBash, AutoBashStep +from reme.utils.local_backend import LocalProcessBackend + +from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend +from reme.components.application_context import ApplicationContext +from reme.config.config_parser import _load_config +from reme.enumeration import ComponentEnum +from reme.plugin import resolve_plugin_runtime + + +@pytest.fixture(autouse=True) +def linux_backend_selection(monkeypatch): + """Keep existing backend integration assertions on the Linux implementation.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux") + + +@pytest.fixture(autouse=True) +def isolated_git_config(tmp_path, monkeypatch): + """Keep Git tests independent of the developer's global and system config.""" + monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(tmp_path / "global.gitconfig")) + monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1") + + +@pytest.mark.asyncio +async def test_auto_bash_runs_in_workspace(tmp_path): + """Approved commands run relative to the configured workspace.""" + (tmp_path / "note.txt").write_text("hello", encoding="utf-8") + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + response = await step(command="cat note.txt") + + assert response.success is True + assert response.answer == "hello" + + +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="POSIX shell pipelines") +@pytest.mark.parametrize("background", [False, True]) +@pytest.mark.parametrize("cancel", [False, True]) +async def test_local_pipeline_timeout_and_cancellation(tmp_path, monkeypatch, background, cancel): + """Timeouts and cancellation clean up pipelines even after the shell exits.""" + (tmp_path / "note.txt").write_text("hello\n", encoding="utf-8") + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + processes = [] + descendants = [] + original_start = asyncio.create_subprocess_exec + original_stop = LocalProcessBackend._stop_process # pylint: disable=protected-access + started = asyncio.Event() + + def group_children(pid): + children = [] + for child in psutil.process_iter(): + try: + if child.pid != pid and os.getpgid(child.pid) == pid: + children.append(child) + except (ProcessLookupError, PermissionError): + pass + return children + + async def start(*args, **kwargs): + process = await original_start(*args, **kwargs) + processes.append(process) + started.set() + return process + + async def stop(self, process, communication): + descendants.extend(group_children(process.pid)) + await original_stop(self, process, communication) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", start) + monkeypatch.setattr(LocalProcessBackend, "_stop_process", stop) + command = "tail -f note.txt | cat" + (" &" if background else "") + assert AutoBash._is_read_only(command) # pylint: disable=protected-access + task = asyncio.create_task(step(command=command, timeout=60000 if cancel else 100)) + try: + await asyncio.wait_for(started.wait(), timeout=2) + if cancel: + # Wait until pipeline children exist before cancelling the invocation. + async def pipeline_started(): + while len(group_children(processes[0].pid)) < 2: + await asyncio.sleep(0.01) + + await asyncio.wait_for(pipeline_started(), timeout=2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=0.8) + else: + response = await asyncio.wait_for(task, timeout=0.8) + assert response.success is False + assert "timed out after 100ms" in response.answer + assert descendants + assert processes[0].returncode is not None + assert all(not child.is_running() or child.status() == psutil.STATUS_ZOMBIE for child in descendants) + finally: + if not task.done(): + task.cancel() + for process in processes: + try: + os.killpg(process.pid, 9) + except ProcessLookupError: + pass + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "command", + [ + "touch denied.txt", + "cat note.txt > denied.txt", + "echo hello > /dev/null-extra", + "cat $(touch denied.txt)", + "echo ${X:-$(touch denied.txt)}", + "cat note.txt; touch denied.txt", + "cat note.txt\ntouch denied.txt", + "tee denied.txt", + "find . -delete", + "find . -exec touch denied.txt {} \\;", + "find . -fprint denied.txt", + "sort -o denied.txt note.txt", + "sort -T denied.txt note.txt", + "sort -rTdenied.txt note.txt", + "sort --compress-program='touch denied.txt' note.txt", + "diff --output=denied.txt note.txt note.txt", + "git add note.txt", + "git branch new-branch", + "git diff --output=denied.txt", + "git log --ext-diff", + "git -c alias.status=add status", + "GIT_OPTIONAL_LOCKS=1 git status", + "GIT_PAGER=sh git log", + "nop", + "rg hello note.txt", + "file -C -m magic", + "file -bC -m magic", + "file --comp -m magic", + r"file -\C -m magic", + "file -z note.txt", + "file -p note.txt", + "echo hello >| denied.txt", + ": <> denied.txt", + "sort --out=denied.txt note.txt", + "git diff --out=denied.txt", + "git diff --ext-di", + "git show --textcon HEAD", + "git log --show-signature", + 'git log --format="%G?"', + "git log --pretty=verify", + "git log --pre=verify", + "git show --format=verify", + "git show --remerge-diff", + "for PATH in .; do git status; done", + "git() { echo overridden; }; git status", + "file -* magic", + ], +) +async def test_auto_bash_denies_mutation_before_execution(tmp_path, command): + """Writing and dynamic shell forms fail before a child process starts.""" + (tmp_path / "note.txt").write_text("hello", encoding="utf-8") + tool = AutoBash(cwd=tmp_path) + + chunks = [chunk async for chunk in tool.call(command)] + + assert len(chunks) == 1 + assert chunks[0].state == ToolResultState.ERROR + assert "Permission denied" in chunks[0].content[0].text + assert not (tmp_path / "denied.txt").exists() + + +@pytest.mark.asyncio +async def test_step_returns_permission_error(tmp_path): + """The job exposes a denied command as an unsuccessful response.""" + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + response = await step(command="touch denied.txt") + + assert response.success is False + assert "Permission denied" in response.answer + assert not (tmp_path / "denied.txt").exists() + + +@pytest.mark.asyncio +async def test_step_can_run_unrestricted_commands(tmp_path): + """Job input can disable the guard for commands that write files.""" + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + response = await step(command="echo hello > written.txt", is_read_only=False) + + assert response.success is True + assert (tmp_path / "written.txt").read_text(encoding="utf-8") == "hello\n" + + +@pytest.mark.asyncio +async def test_unsupported_platform_runs_without_paths_but_rejects_path_policy(tmp_path, monkeypatch): + """AutoBash never drops an explicitly requested sandbox on unsupported systems.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "unsupported") + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + plain = await step(command="echo host", is_read_only=False) + assert plain.success is True + assert plain.answer.strip() == "host" + + restricted = await step(command="echo blocked", is_read_only=False, read_only_paths=[]) + assert restricted.success is False + assert restricted.answer == "No file sandbox backend for unsupported" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("limit", [0, 1, -1]) +async def test_step_rejects_memory_limit_without_path_policy(tmp_path, limit): + """An explicitly supplied limit cannot be ignored by host execution.""" + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(command="echo should-not-run", is_read_only=False, memory_limit_mib=limit) + assert response.success is False + assert response.answer == "memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox" + + +@pytest.mark.asyncio +async def test_step_configuration_can_disable_read_only(tmp_path): + """Step configuration applies when the job does not supply the flag.""" + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)), is_read_only=False) + + response = await step(command="touch written.txt") + + assert response.success is True + assert (tmp_path / "written.txt").exists() + + +@pytest.mark.asyncio +async def test_step_rejects_non_boolean_mode(tmp_path): + """A string false value must not silently enable unrestricted execution.""" + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + response = await step(command="touch denied.txt", is_read_only="false") + + assert response.success is False + assert response.answer == "is_read_only must be a boolean" + assert not (tmp_path / "denied.txt").exists() + + +@pytest.mark.asyncio +async def test_permission_decisions_never_ask(tmp_path): + """Direct AgentScope use makes allow/deny decisions without ASK.""" + tool = AutoBash(cwd=tmp_path) + + allowed = await tool.check_permissions({"command": "ls"}, None) + denied = await tool.check_permissions({"command": "touch denied.txt"}, None) + + assert allowed.behavior == PermissionBehavior.ALLOW + assert denied.behavior == PermissionBehavior.DENY + + +@pytest.mark.asyncio +async def test_read_only_parsing_does_not_block_event_loop(tmp_path, monkeypatch): + """A slow synchronous parser runs in a worker thread during permission checks.""" + started = threading.Event() + release = threading.Event() + + def slow_parse(_command): + started.set() + release.wait(timeout=2) + return True + + tool = AutoBash(cwd=tmp_path) + monkeypatch.setattr(tool, "_is_read_only", slow_parse) + task = asyncio.create_task(tool.check_read_only({"command": "echo hello"})) + try: + assert await asyncio.wait_for(asyncio.to_thread(started.wait), timeout=1) + assert not task.done() + finally: + release.set() + assert await asyncio.wait_for(task, timeout=1) + + +@pytest.mark.asyncio +async def test_unrestricted_tool_allows_commands_without_read_only_check(tmp_path, monkeypatch): + """The unrestricted tool never invokes the read-only parser.""" + tool = AutoBash(cwd=tmp_path, is_read_only=False) + monkeypatch.setattr(tool, "_is_read_only", lambda _command: pytest.fail("read-only parser was called")) + + decision = await tool.check_permissions({"command": "touch written.txt"}, None) + chunks = [chunk async for chunk in tool.call("touch written.txt")] + + assert decision.behavior == PermissionBehavior.ALLOW + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert (tmp_path / "written.txt").exists() + + +@pytest.mark.asyncio +async def test_auto_bash_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch): + """AutoBash routes an injected path policy through the backend.""" + seen = {} + + async def fake_exec(self, command, *, cwd=None, timeout=None): + seen.update(command=command, cwd=cwd, timeout=timeout, backend=self) + return ExecResult(exit_code=0, stdout=b"ok", stderr=b"") + + monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec) + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(command="printf ok", is_read_only=False, read_only_paths=[], writable_paths=[str(tmp_path)]) + + assert response.success is True + assert response.answer == "ok" + assert seen["command"] == ["/bin/sh", "-c", "printf ok"] + assert seen["cwd"] == str(tmp_path) + assert seen["timeout"] == 60 + + +@pytest.mark.asyncio +async def test_sandbox_start_failure_does_not_run_on_host(tmp_path, monkeypatch): + """The plugin reports a sandbox failure instead of executing on the host.""" + + async def unavailable(*_args, **_kwargs): + raise RuntimeError("bwrap unavailable") + + monkeypatch.setattr(PathBubbleBackend, "_start_process", unavailable) + step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + result = await step(command="printf host", is_read_only=False, read_only_paths=[]) + + assert result.success is False + assert "bwrap unavailable" in result.answer + + +def test_home_manifest_declares_step_and_job(): + """The installed manifest declares the job, Step, and tool defaults.""" + manifest = Path(__file__).parents[1] / "src" / "reme_home" / "plugin.yaml" + config = yaml.safe_load(manifest.read_text(encoding="utf-8")) + assert config["backends"]["auto_bash_step"] == "reme_home.auto_bash:AutoBashStep" + job = config["application_defaults"]["jobs"]["auto_bash"] + assert job["steps"] == [{"backend": "auto_bash_step"}] + assert inspect.signature(AutoBash.call).parameters["timeout"].default == 60000 + assert AutoBash.input_schema["properties"]["timeout"]["default"] == 60000 + assert job["parameters"]["properties"]["timeout"]["default"] == 60000 + assert job["parameters"]["properties"]["is_read_only"]["default"] is True + assert job["parameters"]["properties"]["read_only_paths"]["type"] == "array" + assert job["parameters"]["properties"]["writable_paths"]["type"] == "array" + assert job["parameters"]["properties"]["memory_limit_mib"]["minimum"] == 0 + + +def test_home_plugin_registers_job_and_step(monkeypatch): + """Enabling home adds its job and Step to one application runtime.""" + entry = EntryPoint(name="home", value="reme_home", group="reme.plugins") + monkeypatch.setattr("reme.plugin.find_entry_points", lambda group, name: [entry] if name == "home" else []) + config = _load_config("default") + assert "auto_bash" not in config["jobs"] + config["plugins"] = ["home"] + + runtime = resolve_plugin_runtime(config) + + assert runtime.registry.get(ComponentEnum.STEP, "auto_bash_step") is AutoBashStep + assert runtime.config["jobs"]["auto_bash"]["steps"] == [{"backend": "auto_bash_step"}] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "command", + [ + ":", + "cd . && pwd", + "echo hello", + "true", + "false", + "grep hello note.txt", + "find . -name note.txt", + "diff note.txt note.txt", + "cmp note.txt note.txt", + "git status --short", + "git diff --stat", + "git rev-parse HEAD", + "git ls-files", + "echo hello > /dev/null", + "cat < note.txt", + "echo hello >| /dev/null", + "echo hello 2>&1", + "find . -name '*.txt'", + "(cat note.txt) | head -n 1", + ], +) +async def test_safe_command_forms_are_allowed(tmp_path, command): + """Common non-writing commands pass the policy regardless of exit code.""" + (tmp_path / "note.txt").write_text("hello\n", encoding="utf-8") + tool = AutoBash(cwd=tmp_path) + + assert await tool.check_read_only({"command": command}) + + +@pytest.mark.asyncio +async def test_sort_does_not_create_temporary_files(tmp_path): + """Read-only sorting works in memory and fails closed when a spill is needed.""" + (tmp_path / "small.txt").write_text("b\na\n", encoding="utf-8") + tool = AutoBash(cwd=tmp_path) + + small = [chunk async for chunk in tool.call("sort small.txt")] + assert "".join(block.text for chunk in small for block in chunk.content) == "a\nb\n" + + with (tmp_path / "large.txt").open("w", encoding="utf-8") as output: + for number in range(100000, 0, -1): + output.write(f"{number:012d} abcdefghijklmnopqrstuvwxyz\n") + large = [chunk async for chunk in tool.call("sort -S 1K large.txt > /dev/null")] + assert "sort:" in "".join(block.text for chunk in large for block in chunk.content) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("trace_variable", ["GIT_TRACE_REFS", "GIT_TRACE", "GIT_TRACE2_EVENT"]) +async def test_read_only_git_cannot_write_inherited_traces(tmp_path, monkeypatch, trace_variable): + """Allowed metadata queries must not write Git traces from the parent environment.""" + + def git(*args): + return subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True) + + git("init", "-q") + git( + "-c", + "user.name=Test", + "-c", + "user.email=test@example.org", + "-c", + "commit.gpgsign=false", + "commit", + "--allow-empty", + "-qm", + "test", + ) + expected = git("rev-parse", "HEAD").stdout.decode().strip() + marker = tmp_path / "trace output.txt" + tool = AutoBash(cwd=tmp_path) + # Set after construction to check that each invocation uses the current environment. + monkeypatch.setenv(trace_variable, str(marker)) + + chunks = [chunk async for chunk in tool.call("git rev-parse HEAD")] + + assert chunks + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert "".join(block.text for chunk in chunks for block in chunk.content).strip() == expected + assert not marker.exists() + + # Control: the same environment still enables tracing for unrestricted execution. + unrestricted = AutoBash(cwd=tmp_path, is_read_only=False) + chunks = [chunk async for chunk in unrestricted.call("git rev-parse HEAD")] + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert marker.stat().st_size > 0 + + +@pytest.mark.asyncio +async def test_read_only_git_clears_future_trace_variables(tmp_path, monkeypatch): + """Trace isolation uses a prefix rule instead of an enumeration of known names.""" + helper = tmp_path / "git" + helper.write_text('#!/bin/sh\nprintf "%s" "${GIT_TRACE_FUTURE-unset}"\n', encoding="utf-8") + helper.chmod(0o700) + monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ["PATH"]) + monkeypatch.setenv("GIT_TRACE_FUTURE", "inherited") + tool = AutoBash(cwd=tmp_path) + + chunks = [chunk async for chunk in tool.call("git rev-parse HEAD")] + + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert "".join(block.text for chunk in chunks for block in chunk.content) == "unset" + assert os.environ["GIT_TRACE_FUTURE"] == "inherited" + + +@pytest.mark.asyncio +async def test_git_pretty_config_cannot_launch_gpg(tmp_path): + """Git format aliases and defaults cannot smuggle signature verification.""" + repo = tmp_path / "repo" + repo.mkdir() + + def git(*args, input_data=None): + return subprocess.run(["git", *args], cwd=repo, input=input_data, check=True, capture_output=True) + + git("init", "-q") + git("config", "user.name", "Test") + git("config", "user.email", "test@example.org") + (repo / "note.txt").write_text("hello\n", encoding="utf-8") + git("add", "note.txt") + git("commit", "-qm", "test") + commit = git("cat-file", "commit", "HEAD").stdout + signed = commit.replace(b"\n\n", b"\ngpgsig -----BEGIN PGP SIGNATURE-----\n \n -----END PGP SIGNATURE-----\n\n", 1) + signed_hash = git("hash-object", "-t", "commit", "-w", "--stdin", input_data=signed).stdout.strip().decode() + git("update-ref", "HEAD", signed_hash) + + marker = tmp_path / "gpg-ran" + helper = tmp_path / "fake-gpg" + helper.write_text(f'#!/bin/sh\nprintf called > "{marker}"\nexit 1\n', encoding="utf-8") + helper.chmod(0o700) + git("config", "pretty.verify", "format:%G?") + git("config", "pretty.medium", "format:%G?") + git("config", "pretty.oneline", "format:%G?") + git("config", "format.pretty", "format:%G?") + git("config", "gpg.program", str(helper)) + tool = AutoBash(cwd=repo) + + for command in ("git log", "git show HEAD", "git log --oneline"): + chunks = [chunk async for chunk in tool.call(command)] + assert chunks + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert not marker.exists() + + for command in ("git log --pretty=verify", "git show --format=verify"): + chunks = [chunk async for chunk in tool.call(command)] + assert chunks[-1].state == ToolResultState.ERROR + assert not marker.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("helper", ["fsmonitor", "diff", "textconv", "clean"]) +async def test_git_configured_helpers_cannot_write(tmp_path, helper): + """Repository configuration cannot turn an allowed read into helper execution.""" + + def git(*args): + return subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True) + + git("init", "-q") + (tmp_path / "note.txt").write_text("before\n", encoding="utf-8") + (tmp_path / ".gitattributes").write_text("note.txt diff=test filter=test\n", encoding="utf-8") + git("add", "note.txt", ".gitattributes") + hook = tmp_path / "helper" + hook.write_text('#!/bin/sh\ntouch "' + str(tmp_path / "denied.txt") + '"\ncat\n', encoding="utf-8") + hook.chmod(0o700) + key = { + "fsmonitor": "core.fsmonitor", + "diff": "diff.external", + "textconv": "diff.test.textconv", + "clean": "filter.test.clean", + }[helper] + git("config", key, str(hook)) + (tmp_path / "note.txt").write_text("after\n", encoding="utf-8") + command = "git status --short" if helper == "fsmonitor" else "git diff" + tool = AutoBash(cwd=tmp_path) + + chunks = [chunk async for chunk in tool.call(command)] + + assert not (tmp_path / "denied.txt").exists() + if helper == "clean": + assert chunks[-1].state == ToolResultState.ERROR + assert "external filters" in chunks[-1].content[0].text + else: + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + + +@pytest.mark.asyncio +async def test_global_git_lfs_config_allows_metadata_queries(tmp_path, monkeypatch): + """A global filter alone must not block Git queries that never use it.""" + repo = tmp_path / "repo" + repo.mkdir() + subprocess.run(["git", "init", "-q", str(repo)], check=True, capture_output=True) + (repo / "note.txt").write_text("hello\n", encoding="utf-8") + subprocess.run(["git", "add", "note.txt"], cwd=repo, check=True, capture_output=True) + + global_config = tmp_path / "global.gitconfig" + subprocess.run( + ["git", "config", "--file", str(global_config), "filter.lfs.process", "/bin/false"], + check=True, + capture_output=True, + ) + monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(global_config)) + tool = AutoBash(cwd=repo) + + for command, expected in (("git rev-parse --git-dir", ".git"), ("git ls-files", "note.txt")): + chunks = [chunk async for chunk in tool.call(command)] + assert chunks + assert all(chunk.state != ToolResultState.ERROR for chunk in chunks) + assert expected in "".join(block.text for chunk in chunks for block in chunk.content) + + chunks = [chunk async for chunk in tool.call("git diff")] + assert chunks[-1].state == ToolResultState.ERROR + assert "external filters" in chunks[-1].content[0].text + + +@pytest.mark.asyncio +async def test_allow_and_denied_command_options_are_combined(tmp_path, monkeypatch): + """Allow constrains the first argument while denied still checks later flags.""" + from reme_home import auto_bash + + commands = auto_bash._READ_COMMANDS # pylint: disable=protected-access + monkeypatch.setitem(commands, "echo", {"allow": ["hello"]}) + tool = AutoBash(cwd=tmp_path) + + assert await tool.check_read_only({"command": "echo hello"}) + assert not await tool.check_read_only({"command": "echo goodbye"}) + + monkeypatch.setitem(commands, "echo", {"allow": ["hello"], "denied": ["-n"]}) + from safecmd import CmdSpec + + specs = {spec for spec in auto_bash._read_specs() if "echo" not in spec.name} # pylint: disable=protected-access + specs.add(CmdSpec("echo", denied=["-n"])) + monkeypatch.setattr(auto_bash, "_read_specs", lambda: specs) + assert not await tool.check_read_only({"command": "echo hello -n"}) + assert await tool.check_read_only({"command": "find . -name note.txt"}) + assert not await tool.check_read_only({"command": "find . -delete"}) diff --git a/reme/config/benchmark.yaml b/reme/config/benchmark.yaml index 9212092c..46d61455 100644 --- a/reme/config/benchmark.yaml +++ b/reme/config/benchmark.yaml @@ -142,6 +142,10 @@ jobs: python_timeout: type: number description: "Execution timeout in seconds; defaults to 60." + memory_limit_mib: + type: integer + minimum: 0 + description: "Linux memory limit in MiB (default 4096; 0 disables it); requires a path policy; macOS ignores nonzero values" required: - code steps: diff --git a/reme/steps/common/file_sandbox.py b/reme/steps/common/file_sandbox.py new file mode 100644 index 00000000..fd516899 --- /dev/null +++ b/reme/steps/common/file_sandbox.py @@ -0,0 +1,23 @@ +"""Resolve an invocation's optional file sandbox from injected job parameters.""" + +from ...utils.file_sandbox import create_file_sandbox_backend +from ...utils.file_sandbox.file_sandbox_backend import FileSandboxBackend +from ..base_step import BaseStep + + +def file_sandbox_for_step(step: BaseStep) -> FileSandboxBackend | None: + """Keep path and memory policy selection identical across executor steps.""" + assert step.context is not None + read_only_paths = step.context.get("read_only_paths", step.kwargs.get("read_only_paths")) + writable_paths = step.context.get("writable_paths", step.kwargs.get("writable_paths")) + memory_limit_mib = step.context.get("memory_limit_mib", step.kwargs.get("memory_limit_mib")) + if read_only_paths is None and writable_paths is None: + if memory_limit_mib is not None: + raise ValueError("memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox") + return None + return create_file_sandbox_backend( + cwd=step.workspace_path, + read_only_paths=[] if read_only_paths is None else read_only_paths, + writable_paths=[] if writable_paths is None else writable_paths, + memory_limit_mib=memory_limit_mib, + ) diff --git a/reme/steps/common/python_execute.py b/reme/steps/common/python_execute.py index 6da5d441..14ac974c 100644 --- a/reme/steps/common/python_execute.py +++ b/reme/steps/common/python_execute.py @@ -1,27 +1,19 @@ """Execute Python code and return printed stdout.""" -import asyncio import sys -from dataclasses import dataclass from typing import Any +from ...utils.local_backend import LocalProcessBackend, ProcessTimeout from ..base_step import BaseStep from ...components import R +from .file_sandbox import file_sandbox_for_step DEFAULT_TIMEOUT = 60.0 -@dataclass(frozen=True) -class _PythonResult: - stdout: str - stderr: str - returncode: int | None - timed_out: bool = False - - @R.register("python_execute_step") class PythonExecuteStep(BaseStep): - """Run Python code in a subprocess and return stdout as the response answer.""" + """Run Python code and return stdout, limiting each output stream to 1 MiB.""" async def execute(self): assert self.context is not None @@ -39,58 +31,39 @@ class PythonExecuteStep(BaseStep): self.context.response.answer = timeout_error return self.context.response - result = await self._run_python(code, timeout) - if result.timed_out: + timed_out = False + try: + backend = file_sandbox_for_step(self) + result = await (backend or LocalProcessBackend()).exec_shell( + [sys.executable, "-c", code], + cwd=str(self.workspace_path), + timeout=timeout, + **({"raise_errors": True} if backend is None else {}), + ) + except ProcessTimeout as exc: + result, timed_out = exc.result, True + except (TypeError, ValueError, OSError, RuntimeError) as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + return self.context.response + timed_out = timed_out or (result.exit_code == -1 and result.stderr == b"timed out") + stdout = result.stdout.decode(errors="replace") + stderr = result.stderr.decode(errors="replace") + if timed_out: self.context.response.success = False self.context.response.answer = f"Python execution timed out after {timeout:g}s" - self.context.response.metadata.update( - { - "returncode": result.returncode, - "stderr": result.stderr, - "python_timeout": timeout, - }, - ) - return self.context.response - - stdout = result.stdout or "" - stderr = result.stderr or "" - self.context.response.success = result.returncode == 0 - self.context.response.answer = stdout if stdout or result.returncode == 0 else stderr + else: + self.context.response.success = result.exit_code == 0 + self.context.response.answer = stdout if stdout or result.exit_code == 0 else stderr self.context.response.metadata.update( { - "returncode": result.returncode, + "returncode": result.exit_code, "stderr": stderr, "python_timeout": timeout, }, ) return self.context.response - async def _run_python(self, code: str, timeout: float) -> _PythonResult: - process = await asyncio.create_subprocess_exec( - sys.executable, - "-c", - code, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=self.workspace_path, - ) - try: - stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout) - return _PythonResult( - stdout=stdout.decode(), - stderr=stderr.decode(), - returncode=process.returncode, - ) - except TimeoutError: - process.kill() - stdout, stderr = await process.communicate() - return _PythonResult( - stdout=stdout.decode(), - stderr=stderr.decode(), - returncode=process.returncode, - timed_out=True, - ) - @staticmethod def _parse_timeout(raw: Any) -> tuple[float, str]: try: diff --git a/reme/utils/file_sandbox/__init__.py b/reme/utils/file_sandbox/__init__.py new file mode 100644 index 00000000..898d31aa --- /dev/null +++ b/reme/utils/file_sandbox/__init__.py @@ -0,0 +1,18 @@ +"""Select a local file sandbox for supported operating systems.""" + +import sys + +from .file_sandbox_backend import FileSandboxBackend +from .path_bubble_backend import PathBubbleBackend +from .path_seatbelt_backend import PathSeatbeltBackend + + +def create_file_sandbox_backend(*, memory_limit_mib: int | None = None, **kwargs) -> FileSandboxBackend: + """Construct the platform backend; never silently execute without isolation.""" + if sys.platform.startswith("linux"): + limit = 4096 if memory_limit_mib is None else memory_limit_mib + return PathBubbleBackend(memory_limit_mib=limit, **kwargs) + if sys.platform == "darwin": + limit = 0 if memory_limit_mib is None else memory_limit_mib + return PathSeatbeltBackend(memory_limit_mib=limit, **kwargs) + raise RuntimeError(f"No file sandbox backend for {sys.platform}") diff --git a/reme/utils/file_sandbox/file_sandbox_backend.py b/reme/utils/file_sandbox/file_sandbox_backend.py new file mode 100644 index 00000000..220d0e4c --- /dev/null +++ b/reme/utils/file_sandbox/file_sandbox_backend.py @@ -0,0 +1,107 @@ +"""Common contract and path validation for subprocess file sandboxes.""" + +import asyncio +import os +import sys +from abc import abstractmethod + +from agentscope.tool import ExecResult + +from ..local_backend import LocalProcessBackend + + +def contains(root: str, path: str) -> bool: + """Whether a path is inside a configured root.""" + return os.path.commonpath((root, path)) == root + + +class FileSandboxBackend(LocalProcessBackend): + """Execute subprocesses with explicit host-path read and write access.""" + + def __init__(self, *, cwd: str | os.PathLike[str], read_only_paths: list[str], writable_paths: list[str]) -> None: + self._cwd = self._validate_path(os.fspath(cwd)) + if not os.path.isdir(self._cwd): + raise ValueError("sandbox working directory must be a directory") + self._read_only_paths = self._validate_paths(read_only_paths, "read_only_paths") + self._writable_paths = self._validate_paths(writable_paths, "writable_paths") + if not any(contains(path, self._cwd) for path in self._read_only_paths + self._writable_paths): + self._read_only_paths.append(self._cwd) + # A virtual environment can point outside sys.prefix to its base + # interpreter and standard library (notably with Homebrew Python). + for runtime in (sys.prefix, sys.base_prefix, sys.executable): + path = os.path.realpath(runtime) + if path in ("/", "/usr", "/usr/local") or path.startswith("/usr/"): + continue + if not any(contains(root, path) for root in self._read_only_paths + self._writable_paths): + self._read_only_paths.append(self._validate_path(path)) + paths = self._read_only_paths + self._writable_paths + if len(set(paths)) != len(paths): + raise ValueError("sandbox paths must not contain duplicates") + self._path_identities = {path: self._identity(path) for path in paths} + + @staticmethod + def _identity(path: str) -> tuple[int, int]: + info = os.stat(path, follow_symlinks=False) + return info.st_dev, info.st_ino + + @staticmethod + def _validate_path(raw: str) -> str: + if not isinstance(raw, str) or not raw or not os.path.isabs(raw): + raise ValueError("sandbox paths must be non-empty absolute paths") + path = os.path.normpath(raw) + if path == "/" or not os.path.exists(path) or os.path.realpath(path) != path: + raise ValueError(f"sandbox path must exist without symlinks and cannot be root: {raw}") + return path + + @classmethod + def _validate_paths(cls, raw: list[str], name: str) -> list[str]: + if not isinstance(raw, list) or any(not isinstance(item, str) for item in raw): + raise ValueError(f"{name} must be a list of absolute paths") + return [cls._validate_path(path) for path in raw] + + def _check_paths(self, cwd: str) -> None: + if not any(contains(path, cwd) for path in self._read_only_paths + self._writable_paths): + raise ValueError("command working directory must be inside a sandbox path") + for path, identity in self._path_identities.items(): + if not os.path.exists(path) or os.path.realpath(path) != path or self._identity(path) != identity: + raise RuntimeError(f"sandbox path changed before execution: {path}") + + async def getcwd(self) -> str: + """Return the host-identical sandbox working directory.""" + return self._cwd + + @abstractmethod + async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL): + """Start one sandboxed process, never falling back to host execution.""" + + async def exec_shell( + self, + command: list[str], + *, + cwd: str | None = None, + timeout: float | None = None, + ) -> ExecResult: + """Execute an argv list with a timeout inside the platform sandbox.""" + process = await self._start_process(command, cwd or self._cwd) + return await self._communicate(process, timeout=timeout) + + async def read_file(self, path: str) -> bytes: + """Read through a sandboxed command so file rules still apply.""" + result = await self.exec_shell(["cat", path]) + if not result.ok(): + raise OSError(result.stderr.decode(errors="replace")) + return result.stdout + + async def write_file(self, path: str, data: bytes) -> None: + """Write through a sandboxed command so file rules still apply.""" + if not isinstance(data, bytes): + raise TypeError("data must be bytes") + script = 'mkdir -p -- "$(dirname -- "$1")" && cat > "$1"' + process = await self._start_process( + ["/bin/sh", "-c", script, "sh", path], + self._cwd, + stdin=asyncio.subprocess.PIPE, + ) + result = await self._communicate(process, data=data) + if not result.ok(): + raise OSError(result.stderr.decode(errors="replace")) diff --git a/reme/utils/file_sandbox/path_bubble_backend.py b/reme/utils/file_sandbox/path_bubble_backend.py new file mode 100644 index 00000000..cbca8a79 --- /dev/null +++ b/reme/utils/file_sandbox/path_bubble_backend.py @@ -0,0 +1,124 @@ +"""Bubblewrap execution with host-identical, explicitly mounted paths.""" + +import asyncio +import os +import shutil +import uuid +from pathlib import Path + +import sys + +from .file_sandbox_backend import FileSandboxBackend + +# Run after exec of a fresh interpreter: Python preexec_fn can deadlock when +# the parent has threads. The requested command starts only after enrollment. +_CGROUP_EXEC = """ +import os +import sys +with open(sys.argv[1], 'w', encoding='ascii') as group: + group.write(str(os.getpid())) +os.execvp(sys.argv[2], sys.argv[2:]) +""" + + +class PathBubbleBackend(FileSandboxBackend): + """Run commands with selected host paths at the same sandbox paths.""" + + def __init__( + self, + *, + cwd: str | os.PathLike[str], + read_only_paths: list[str], + writable_paths: list[str], + memory_limit_mib: int = 4096, + ) -> None: + if isinstance(memory_limit_mib, bool) or not isinstance(memory_limit_mib, int) or memory_limit_mib < 0: + raise ValueError("memory_limit_mib must be a non-negative integer") + self._memory_limit_mib = memory_limit_mib + self._process_cgroups: dict[int, Path] = {} + super().__init__(cwd=cwd, read_only_paths=read_only_paths, writable_paths=writable_paths) + + def _bwrap_argv(self, command: list[str], cwd: str) -> list[str]: + """Build one isolated execution with host-identical bind targets.""" + if not command: + raise ValueError("command must not be empty") + self._check_paths(cwd) + args = ["bwrap", "--die-with-parent", "--new-session", "--unshare-all", "--proc", "/proc", "--dev", "/dev"] + for path in ("/usr", "/usr/local"): + if os.path.isdir(path): + args.extend(("--ro-bind", path, path)) + for path in ("/bin", "/sbin", "/lib", "/lib64"): + if os.path.islink(path): + args.extend(("--symlink", os.readlink(path), path)) + elif os.path.exists(path): + args.extend(("--ro-bind", path, path)) + args.extend(("--dir", "/etc")) + for path in ("/etc/passwd", "/etc/group", "/etc/nsswitch.conf", "/etc/hosts", "/etc/resolv.conf", "/etc/ssl"): + if os.path.exists(path): + args.extend(("--ro-bind", path, path)) + args.extend(("--tmpfs", "/tmp", "--tmpfs", "/run")) + mounts = [(path, "--ro-bind") for path in self._read_only_paths] + mounts += [(path, "--bind") for path in self._writable_paths] + for path, mode in sorted(mounts, key=lambda item: (item[0].count("/"), item[0])): + args.extend((mode, path, path)) + binary_paths = [str(Path(sys.prefix) / "bin"), "/usr/local/bin", "/usr/bin", "/bin"] + args.extend(("--clearenv", "--setenv", "HOME", self._cwd, "--setenv", "TMPDIR", "/tmp")) + args.extend(("--setenv", "PATH", ":".join(binary_paths), "--chdir", cwd, "--")) + return args + command + + async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL): + """Start bwrap without falling back to host execution.""" + if not sys.platform.startswith("linux"): + raise RuntimeError("PathBubbleBackend requires Linux") + if shutil.which("bwrap") is None: + raise RuntimeError("PathBubbleBackend requires bwrap on PATH") + argv = self._bwrap_argv(command, cwd) + group = self._create_cgroup() if self._memory_limit_mib else None + try: + if group is not None: + argv = [sys.executable, "-I", "-S", "-c", _CGROUP_EXEC, str(group / "cgroup.procs"), *argv] + process = await self._spawn_process( + argv, + stdin=stdin, + ) + except BaseException: + if group is not None: + group.rmdir() + raise + if group is not None: + self._process_cgroups[process.pid] = group + return process + + def _create_cgroup(self) -> Path: + """Create a separate cgroup v2 for one execution, failing closed if unavailable.""" + try: + entries = Path("/proc/self/cgroup").read_text(encoding="ascii").splitlines() + relative = next(line.split("::", 1)[1] for line in entries if line.startswith("0::")) + root = Path("/sys/fs/cgroup") + parent = root / relative.lstrip("/") + if ".." in Path(relative).parts or not parent.resolve().is_relative_to(root): + raise RuntimeError("invalid cgroup v2 path") + group = parent / f"reme-{uuid.uuid4().hex}" + group.mkdir() + try: + (group / "memory.max").write_text(str(self._memory_limit_mib * 1024**2), encoding="ascii") + except BaseException: + group.rmdir() + raise + return group + except (OSError, StopIteration) as exc: + raise RuntimeError("PathBubbleBackend requires a writable delegated cgroup v2 for memory limits") from exc + + def _release_process(self, process: asyncio.subprocess.Process) -> None: + """Release the per-execution group after its process has been reaped.""" + group = self._process_cgroups.pop(process.pid, None) + if group is None: + return + try: + group.rmdir() + except OSError as exc: + try: + (group / "cgroup.kill").write_text("1", encoding="ascii") + group.rmdir() + except OSError as cleanup_error: + raise RuntimeError(f"could not remove sandbox cgroup {group}: {cleanup_error}") from exc diff --git a/reme/utils/file_sandbox/path_seatbelt_backend.py b/reme/utils/file_sandbox/path_seatbelt_backend.py new file mode 100644 index 00000000..ee6a5c67 --- /dev/null +++ b/reme/utils/file_sandbox/path_seatbelt_backend.py @@ -0,0 +1,329 @@ +"""macOS Seatbelt file sandbox with host-identical paths.""" + +import asyncio +import errno +import json +import os +import plistlib +import shutil +import sys +import tempfile +import uuid +import warnings +from pathlib import Path +from types import SimpleNamespace + +import psutil +from agentscope.tool import ExecResult + +from .file_sandbox_backend import FileSandboxBackend, contains +from ..local_backend import ( + LocalProcessBackend, + OutputLimitExceeded, + capture_output, + finish_cleanup, + output_limit_result, +) + + +class LaunchctlError(RuntimeError): + """Preserve launchctl's error code so missing services can be handled narrowly.""" + + def __init__(self, message: str, exit_code: int): + super().__init__(message) + self.exit_code = exit_code + + +class PathSeatbeltBackend(FileSandboxBackend): + """Run each command under a deny-by-default Seatbelt profile.""" + + def __init__( + self, + *, + cwd: str | os.PathLike[str], + read_only_paths: list[str], + writable_paths: list[str], + memory_limit_mib: int = 0, + ) -> None: + if isinstance(memory_limit_mib, bool) or not isinstance(memory_limit_mib, int) or memory_limit_mib < 0: + raise ValueError("memory_limit_mib must be a non-negative integer") + if memory_limit_mib: + warnings.warn( + "PathSeatbeltBackend does not support memory_limit_mib; the limit is ignored on macOS", + RuntimeWarning, + stacklevel=2, + ) + super().__init__(cwd=cwd, read_only_paths=read_only_paths, writable_paths=writable_paths) + self._process_tempdirs: dict[int, str] = {} + + @staticmethod + def _path_rule(path: str) -> str: + """Quote a path as one Seatbelt profile string, never as profile syntax.""" + return f"(subpath {json.dumps(path)})" if os.path.isdir(path) else f"(literal {json.dumps(path)})" + + def _profile(self, tempdir: str) -> str: + """Allow only configured files plus the system runtime needed to launch tools.""" + system_roots = ("/System", "/usr", "/bin", "/sbin", "/private/etc", "/dev") + read_paths = [path for path in system_roots if os.path.exists(path)] + read_paths += self._read_only_paths + self._writable_paths + [tempdir] + lines = [ + "(version 1)", + "(deny default)", + "(allow process*)", + "(allow sysctl-read)", + "(allow mach-lookup)", + # dyld opens the root directory; /bin/sh reads the system shell selector. + '(allow file-read* (literal "/") (literal "/private/var/select/sh"))', + ] + ancestors = set() + for path in read_paths: + parent = os.path.dirname(path) + while parent != "/": + ancestors.add(parent) + parent = os.path.dirname(parent) + lines.extend(f"(allow file-read-metadata (literal {json.dumps(path)}))" for path in sorted(ancestors)) + lines.extend(f"(allow file-read* {self._path_rule(path)})" for path in read_paths) + # Shell redirection to the null device is safe and is accepted by + # AutoBash's read-only policy; do not grant writes to the rest of /dev. + lines.append('(allow file-write* (literal "/dev/null"))') + for path in self._writable_paths + [tempdir]: + # Seatbelt combines allow rules with OR, unlike ordered bwrap mounts. + # Exclude read-only descendants from each writable parent's grant. + exclusions = [ + f"(require-not {self._path_rule(read_only)})" + for read_only in self._read_only_paths + if contains(path, read_only) + ] + rule = self._path_rule(path) + if exclusions: + rule = f"(require-all {rule} {' '.join(exclusions)})" + lines.append(f"(allow file-write* {rule})") + # A writable ancestor can otherwise be renamed, moving a protected + # subtree out of the path-based exclusion above. Pin the protected + # path and every ancestor that falls within a writable root. + pinned = set() + for read_only in self._read_only_paths: + current = read_only + while current != "/": + if any(contains(writable, current) for writable in self._writable_paths): + pinned.add(current) + current = os.path.dirname(current) + for path in sorted(pinned): + rule = f"(literal {json.dumps(path)})" + lines.append(f"(deny file-write-unlink {rule})") + lines.append(f"(deny file-write-create {rule})") + return "\n".join(lines) + + async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL): + """Start Seatbelt directly when the GUI launchd domain is unavailable.""" + tempdir = tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir())) + try: + process = await self._spawn_process( + ["/usr/bin/sandbox-exec", "-p", self._profile(tempdir), *command], + cwd=cwd, + env={**os.environ, "HOME": self._cwd, "TMPDIR": tempdir}, + stdin=stdin, + ) + except BaseException: + shutil.rmtree(tempdir) + raise + self._process_tempdirs[process.pid] = tempdir + return process + + def _release_process(self, process: asyncio.subprocess.Process) -> None: + tempdir = self._process_tempdirs.pop(process.pid, None) + if tempdir is not None: + shutil.rmtree(tempdir) + + @staticmethod + async def _has_gui_launchd(domain: str) -> bool: + result = await LocalProcessBackend().exec_shell(["/bin/launchctl", "print", domain], timeout=5) + return result.ok() + + @staticmethod + async def _launchctl(*arguments: str) -> str: + result = await LocalProcessBackend().exec_shell(["/bin/launchctl", *arguments], timeout=5) + if not result.ok(): + raise LaunchctlError( + f"launchctl {arguments[0]} failed: {result.stderr.decode(errors='replace').strip()}", + result.exit_code, + ) + return result.stdout.decode(errors="replace").strip() + + async def _bootstrap(self, domain: str, plist: Path, service: str) -> None: + """Resolve registration before rolling it back, even when its reply is interrupted.""" + registration = asyncio.create_task(self._launchctl("bootstrap", domain, str(plist))) + try: + await asyncio.shield(registration) + except BaseException: + + async def rollback(): + # Killing the client cannot undo a registration accepted by launchd. + # Wait for the bounded launchctl operation before issuing bootout. + await asyncio.gather(registration, return_exceptions=True) + try: + await self._launchctl("bootout", service) + except LaunchctlError as exc: + if exc.exit_code != errno.ESRCH: + raise + + await finish_cleanup(rollback()) + raise + + @staticmethod + async def _wait_status(status: Path, pid: int) -> int: + while True: + if status.exists(): + return int(status.read_text(encoding="ascii")) + if not psutil.pid_exists(pid): + raise RuntimeError("launchd sandbox job exited without reporting a status") + await asyncio.sleep(0.02) + + @staticmethod + async def _read_fifo(fd: int, buffer: bytearray, name: str, exceeded: asyncio.Future, finished: asyncio.Event): + """Drain launchd output through a bounded pipe instead of an unbounded disk file.""" + while True: + try: + chunk = os.read(fd, 65536) + except BlockingIOError: + if finished.is_set(): + return + await asyncio.sleep(0.005) + continue + capture_output(buffer, chunk, name, exceeded) + await asyncio.sleep(0) + + async def _run_job(self, command: list[str], cwd: str, *, data: bytes | None, timeout: float | None) -> ExecResult: + if sys.platform != "darwin": + raise RuntimeError("PathSeatbeltBackend requires macOS") + if shutil.which("sandbox-exec") is None: + raise RuntimeError("PathSeatbeltBackend requires sandbox-exec on PATH") + if not command: + raise ValueError("command must not be empty") + self._check_paths(cwd) + domain = f"gui/{os.getuid()}" + if not await self._has_gui_launchd(domain): + warnings.warn( + "No macOS GUI launchd domain; process-group cleanup is unavailable, running directly under Seatbelt", + RuntimeWarning, + stacklevel=2, + ) + process = await self._start_process( + command, + cwd, + stdin=asyncio.subprocess.PIPE if data is not None else asyncio.subprocess.DEVNULL, + ) + return await self._communicate(process, data=data, timeout=timeout) + control = Path(tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir()))) + sandbox_tmp = control / "sandbox-tmp" + sandbox_tmp.mkdir() + stdout_path = control / "stdout" + stderr_path = control / "stderr" + status_path = control / "status" + stdin_path = control / "stdin" + label = f"ai.agentscope.reme.sandbox.{uuid.uuid4().hex}" + service = f"{domain}/{label}" + registered = False + pid = None + stdout, stderr = bytearray(), bytearray() + exceeded = asyncio.get_running_loop().create_future() + finished = asyncio.Event() + output_fds = [] + readers = [] + + async def cleanup(completed): + try: + if registered: + if not completed and pid is not None: + self._kill_process(SimpleNamespace(pid=pid)) + await self._launchctl("bootout", service) + finally: + finished.set() + try: + await asyncio.wait_for(asyncio.gather(*readers), timeout=0.25) + except TimeoutError: + pass + + try: + for path, buffer, name in ((stdout_path, stdout, "stdout"), (stderr_path, stderr, "stderr")): + os.mkfifo(path, 0o600) + # Holding both ends open prevents a transient EOF before launchd + # opens its writer. Kernel pipe capacity bounds pending output. + fd = os.open(path, os.O_RDWR | os.O_NONBLOCK) + output_fds.append(fd) + readers.append(asyncio.create_task(self._read_fifo(fd, buffer, name, exceeded, finished))) + if data is not None: + stdin_path.write_bytes(data) + # The shell reports the command's status in a control file. Its + # child inherits the launchd job's process group, so launchd can + # reap same-group background processes when the shell exits. + script = '"$@"; result=$?; printf "%s\\n" "$result" > "$0.tmp"; mv "$0.tmp" "$0"' + job = { + "Label": label, + "ProgramArguments": [ + "/bin/sh", + "-c", + script, + str(status_path), + "/usr/bin/sandbox-exec", + "-p", + self._profile(str(sandbox_tmp)), + *command, + ], + "WorkingDirectory": cwd, + "EnvironmentVariables": {**os.environ, "HOME": self._cwd, "TMPDIR": str(sandbox_tmp)}, + "StandardInPath": str(stdin_path) if data is not None else "/dev/null", + "StandardOutPath": str(stdout_path), + "StandardErrorPath": str(stderr_path), + "RunAtLoad": False, + "KeepAlive": False, + "AbandonProcessGroup": False, + } + plist_path = control / "job.plist" + with os.fdopen(os.open(plist_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as stream: + plistlib.dump(job, stream) + await self._bootstrap(domain, plist_path, service) + registered = True + + async def run(): + nonlocal pid + pid = int(await self._launchctl("kickstart", "-p", service)) + return await self._watch_output(self._wait_status(status_path, pid), exceeded) + + try: + exit_code = await self._execute(run(), timeout=timeout, cleanup=cleanup) + except TimeoutError: + return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out") + except OutputLimitExceeded as exc: + return output_limit_result(bytes(stdout), bytes(stderr), str(exc)) + if exceeded.done(): + return output_limit_result(bytes(stdout), bytes(stderr), exceeded.result()) + return ExecResult( + exit_code=exit_code, + stdout=bytes(stdout), + stderr=bytes(stderr), + ) + finally: + for reader in readers: + reader.cancel() + await asyncio.gather(*readers, return_exceptions=True) + for fd in output_fds: + os.close(fd) + shutil.rmtree(control) + + async def exec_shell( + self, + command: list[str], + *, + cwd: str | None = None, + timeout: float | None = None, + ) -> ExecResult: + return await self._run_job(command, cwd or self._cwd, data=None, timeout=timeout) + + async def write_file(self, path: str, data: bytes) -> None: + if not isinstance(data, bytes): + raise TypeError("data must be bytes") + script = 'mkdir -p -- "$(dirname -- "$1")" && cat > "$1"' + result = await self._run_job(["/bin/sh", "-c", script, "sh", path], self._cwd, data=data, timeout=None) + if not result.ok(): + raise OSError(result.stderr.decode(errors="replace")) diff --git a/reme/utils/local_backend.py b/reme/utils/local_backend.py new file mode 100644 index 00000000..d8b48cb3 --- /dev/null +++ b/reme/utils/local_backend.py @@ -0,0 +1,247 @@ +"""Shared subprocess execution with bounded timeout and cancellation cleanup.""" + +import asyncio +import os +import signal +import sys +from collections.abc import Awaitable, Callable + +import psutil +from agentscope.tool import ExecResult, LocalBackend + +_CLEANUP_TIMEOUT = 0.25 +OUTPUT_LIMIT_BYTES = 1024 * 1024 + + +async def finish_cleanup(operation: Awaitable): + """Finish rollback after cancellation, deferring additional cancellation requests.""" + task = asyncio.ensure_future(operation) + while True: + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + if task.done(): + return task.result() + + +class OutputLimitExceeded(RuntimeError): + """Identify the output stream that exceeded its byte budget.""" + + +def capture_output(buffer: bytearray, chunk: bytes, stream: str, exceeded: asyncio.Future) -> None: + """Retain at most one stream's budget, signalling overflow without blocking draining.""" + remaining = OUTPUT_LIMIT_BYTES - len(buffer) + buffer.extend(chunk[:remaining]) + if len(chunk) > remaining and not exceeded.done(): + exceeded.set_result(stream) + + +def output_limit_result(stdout: bytes, stderr: bytes, stream: str) -> ExecResult: + """Return bounded partial output and an explicit failure diagnostic.""" + message = f"\nOutput limit exceeded: {stream} exceeds {OUTPUT_LIMIT_BYTES} bytes; command terminated.\n".encode() + diagnostic = (stderr[: max(0, OUTPUT_LIMIT_BYTES - len(message))] + message)[:OUTPUT_LIMIT_BYTES] + return ExecResult(exit_code=1, stdout=stdout[:OUTPUT_LIMIT_BYTES], stderr=diagnostic) + + +class ProcessTimeout(TimeoutError): + """Carry the reaped process output for callers with their own timeout protocol.""" + + def __init__(self, result: ExecResult): + super().__init__("timed out") + self.result = result + + +class LocalProcessBackend(LocalBackend): + """Keep local file helpers; terminate commands exceeding 1 MiB on either output stream.""" + + async def exec_shell( + self, + command: list[str], + *, + cwd: str | None = None, + timeout: float | None = None, + raise_errors: bool = False, + ) -> ExecResult: + """Run an argv in its own session and collect output within the deadline.""" + try: + process = await self._start_process(command, cwd) + except OSError as exc: + if raise_errors: + raise + return ExecResult(exit_code=127, stdout=b"", stderr=str(exc).encode("utf-8")) + return await self._communicate(process, timeout=timeout, raise_errors=raise_errors) + + async def _start_process(self, command: list[str], cwd: str | None, *, stdin=None): + """Start a host process; sandboxes override platform-specific setup.""" + return await self._spawn_process(command, cwd=cwd, stdin=stdin) + + async def _spawn_process(self, command: list[str], **kwargs): + """Capture output in a separate process group for every execution backend.""" + spawning = asyncio.create_task( + asyncio.create_subprocess_exec( + *command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + start_new_session=os.name == "posix", + **kwargs, + ), + ) + try: + return await asyncio.shield(spawning) + except asyncio.CancelledError: + + async def rollback(): + outcome = (await asyncio.gather(spawning, return_exceptions=True))[0] + if isinstance(outcome, BaseException): + return + exceeded = asyncio.get_running_loop().create_future() + communication = asyncio.create_task( + self._collect_output(outcome, None, bytearray(), bytearray(), exceeded), + ) + await self._stop_process(outcome, communication) + + await finish_cleanup(rollback()) + raise + + def _release_process(self, process: asyncio.subprocess.Process) -> None: + """Release any platform resources owned by a completed execution.""" + + @staticmethod + async def _watch_output(operation: Awaitable, exceeded: asyncio.Future): + """Stop waiting as soon as either captured stream exceeds its budget.""" + task = asyncio.ensure_future(operation) + try: + await asyncio.wait((task, exceeded), return_when=asyncio.FIRST_COMPLETED) + if exceeded.done(): + raise OutputLimitExceeded(exceeded.result()) + return await task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + @staticmethod + async def _collect_output(process, data, stdout, stderr, exceeded): + """Drain both pipes concurrently while retaining bounded byte buffers.""" + + async def read(reader, buffer, name): + while chunk := await reader.read(65536): + capture_output(buffer, chunk, name, exceeded) + + async def write(): + if process.stdin is not None: + try: + if data is not None: + process.stdin.write(data) + await process.stdin.drain() + except (BrokenPipeError, ConnectionResetError): + pass + finally: + process.stdin.close() + + tasks = [ + asyncio.create_task(read(process.stdout, stdout, "stdout")), + asyncio.create_task(read(process.stderr, stderr, "stderr")), + asyncio.create_task(write()), + ] + try: + await asyncio.gather(*tasks) + await process.wait() + return bytes(stdout), bytes(stderr) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + @staticmethod + async def _execute(operation: Awaitable, *, timeout: float | None, cleanup: Callable[[bool], Awaitable[None]]): + """Apply one deadline and always clean up success, failure, or cancellation.""" + task = asyncio.ensure_future(operation) + completed = False + try: + result = await asyncio.wait_for(asyncio.shield(task), timeout=timeout) + completed = True + return result + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await cleanup(completed) + + async def _communicate( + self, + process: asyncio.subprocess.Process, + *, + data: bytes | None = None, + timeout: float | None = None, + raise_errors: bool = False, + ) -> ExecResult: + """Collect pipes and release the process, preserving each caller's timeout protocol.""" + stdout_buffer, stderr_buffer = bytearray(), bytearray() + exceeded = asyncio.get_running_loop().create_future() + communication = asyncio.create_task( + self._collect_output(process, data, stdout_buffer, stderr_buffer, exceeded), + ) + + async def cleanup(completed): + try: + if not completed: + await self._stop_process(process, communication) + finally: + self._release_process(process) + + try: + stdout, stderr = await self._execute( + self._watch_output(asyncio.shield(communication), exceeded), + timeout=timeout, + cleanup=cleanup, + ) + except OutputLimitExceeded as exc: + return output_limit_result(bytes(stdout_buffer), bytes(stderr_buffer), str(exc)) + except TimeoutError: + if raise_errors: + stdout, stderr = communication.result() if not communication.cancelled() else (b"", b"") + raise ProcessTimeout(ExecResult(exit_code=process.returncode, stdout=stdout, stderr=stderr)) from None + return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out") + return ExecResult(exit_code=process.returncode, stdout=stdout, stderr=stderr) + + @staticmethod + def _kill_process(process: asyncio.subprocess.Process) -> None: + """Kill the group and known detached children on macOS and Windows.""" + if os.name != "posix" or sys.platform == "darwin": + if os.name == "posix": + try: + os.killpg(process.pid, signal.SIGSTOP) + except ProcessLookupError: + pass + try: + parent = psutil.Process(process.pid) + for child in reversed(parent.children(recursive=True)): + try: + child.kill() + except (psutil.Error, OSError): + pass + if os.name != "posix": + parent.kill() + except (psutil.Error, OSError): + pass + if os.name == "posix": + # The launcher may have exited while a pipeline still holds stdout. + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + async def _stop_process(self, process: asyncio.subprocess.Process, communication: asyncio.Task) -> None: + """Kill descendants and bound draining even if a child escaped the group.""" + self._kill_process(process) + try: + await asyncio.wait_for(asyncio.shield(communication), timeout=_CLEANUP_TIMEOUT) + except TimeoutError: + communication.cancel() + await asyncio.gather(communication, return_exceptions=True) + # asyncio has no public API to close a Process's captured pipes. + # An escaped descendant must not keep these transports alive. + process._transport.close() # pylint: disable=protected-access + try: + await asyncio.wait_for(process.wait(), timeout=_CLEANUP_TIMEOUT) + except TimeoutError: + pass diff --git a/tests/integration/test_seatbelt_backend.py b/tests/integration/test_seatbelt_backend.py new file mode 100644 index 00000000..d51146eb --- /dev/null +++ b/tests/integration/test_seatbelt_backend.py @@ -0,0 +1,231 @@ +"""Exercise macOS sandbox startup and file isolation using real subprocesses.""" + +import asyncio +import sys +import venv + +import psutil + +import pytest + +from reme.utils.file_sandbox.path_seatbelt_backend import PathSeatbeltBackend + +pytestmark = pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt requires macOS") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "command", + [ + ["/bin/sh", "-c", "pwd"], + [sys.executable, "-c", "import os; print(os.getcwd())"], + ], +) +async def test_seatbelt_starts_shell_and_python(tmp_path, command): + """System runtime access permits startup and resolution of the working directory.""" + workspace = tmp_path.resolve() + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[]) + + result = await backend.exec_shell(command, timeout=10) + + assert result.exit_code == 0, result.stderr.decode(errors="replace") + assert result.stdout.decode().strip() == str(workspace) + assert result.stderr == b"" + + +@pytest.mark.asyncio +async def test_seatbelt_starts_virtual_environment_python(tmp_path): + """A symlinked virtual environment interpreter can load its base runtime.""" + workspace = tmp_path.resolve() + environment = workspace / "venv" + venv.EnvBuilder(with_pip=False, symlinks=True).create(environment) + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[]) + + result = await backend.exec_shell( + [str(environment / "bin" / "python"), "-c", "import os; print(os.getcwd())"], + timeout=10, + ) + + assert result.exit_code == 0, result.stderr.decode(errors="replace") + assert result.stdout.decode().strip() == str(workspace) + + +@pytest.mark.asyncio +async def test_seatbelt_allows_null_device_redirection(tmp_path): + """The shell can discard output through /dev/null inside a file sandbox.""" + backend = PathSeatbeltBackend(cwd=tmp_path.resolve(), read_only_paths=[], writable_paths=[]) + + result = await backend.exec_shell(["/bin/sh", "-c", "echo hi > /dev/null"], timeout=5) + + assert result.exit_code == 0, result.stderr.decode(errors="replace") + assert result.stdout == b"" + + +@pytest.mark.asyncio +async def test_seatbelt_write_file_with_launchd_stdin(tmp_path): + """The launchd job delivers input bytes through its configured stdin file.""" + workspace = tmp_path.resolve() + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)]) + target = workspace / "nested" / "note.txt" + + await backend.write_file(str(target), b"hello\n") + + assert target.read_bytes() == b"hello\n" + + +@pytest.mark.asyncio +async def test_seatbelt_without_gui_launchd_uses_direct_sandbox(tmp_path, monkeypatch): + """A missing GUI domain warns while Seatbelt still enforces the path policy.""" + root = tmp_path.resolve() + workspace = root / "workspace" + workspace.mkdir() + outside = root / "outside.txt" + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)]) + + async def missing(_domain): + return False + + monkeypatch.setattr(backend, "_has_gui_launchd", missing) + code = ( + "from pathlib import Path;" + "Path('inside.txt').write_text('allowed');" + f"Path({str(outside)!r}).write_text('denied')" + ) + with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"): + result = await backend.exec_shell([sys.executable, "-c", code], timeout=5) + assert result.exit_code != 0 + assert (workspace / "inside.txt").read_text() == "allowed" + assert not outside.exists() + + with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"): + await backend.write_file(str(workspace / "input.txt"), b"hello") + assert (workspace / "input.txt").read_bytes() == b"hello" + + +@pytest.mark.asyncio +async def test_seatbelt_runtime_permissions_preserve_file_isolation(tmp_path): + """Startup grants do not expose sibling files or writable access to protected children.""" + root = tmp_path.resolve() + workspace = root / "workspace" + protected = workspace / "protected" + output = protected / "output" + output.mkdir(parents=True) + (protected / "note.txt").write_text("readable", encoding="utf-8") + (root / "outside.txt").write_text("private", encoding="utf-8") + backend = PathSeatbeltBackend( + cwd=workspace, + read_only_paths=[str(protected)], + writable_paths=[str(workspace), str(output)], + ) + code = """ +from pathlib import Path +assert Path('protected/note.txt').read_text() == 'readable' +Path('allowed.txt').write_text('ok') +Path('protected/output/allowed.txt').write_text('ok') +for path, operation in [ + (Path('protected/denied.txt'), 'write'), + (Path('../outside.txt'), 'read'), + (Path('../outside.txt'), 'write'), +]: + try: + path.read_text() if operation == 'read' else path.write_text('denied') + except PermissionError: + pass + else: + raise AssertionError(f'{operation} unexpectedly allowed: {path}') +""" + + result = await backend.exec_shell([sys.executable, "-c", code], timeout=10) + + assert result.exit_code == 0, result.stderr.decode(errors="replace") + assert (workspace / "allowed.txt").read_text() == "ok" + assert (output / "allowed.txt").read_text() == "ok" + assert not (protected / "denied.txt").exists() + assert (root / "outside.txt").read_text() == "private" + + +@pytest.mark.asyncio +async def test_seatbelt_cannot_move_read_only_ancestor(tmp_path): + """Moving a writable ancestor must not change a protected file's effective path.""" + workspace = tmp_path.resolve() + protected = workspace / "parent" / "protected" + protected.mkdir(parents=True) + note = protected / "note.txt" + note.write_text("original", encoding="utf-8") + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[str(protected)], writable_paths=[str(workspace)]) + + result = await backend.exec_shell( + ["/bin/sh", "-c", "mv parent renamed && echo changed > renamed/protected/note.txt"], + timeout=5, + ) + assert result.exit_code != 0 + assert note.read_text(encoding="utf-8") == "original" + assert not (workspace / "renamed").exists() + + sibling = await backend.exec_shell(["/bin/sh", "-c", "echo allowed > sibling.txt"], timeout=5) + assert sibling.exit_code == 0, sibling.stderr.decode(errors="replace") + assert (workspace / "sibling.txt").read_text(encoding="utf-8").strip() == "allowed" + + +@pytest.mark.asyncio +async def test_seatbelt_reaps_same_group_child_after_parent_exits(tmp_path): + """launchd clears a background child after its foreground command returns.""" + workspace = tmp_path.resolve() + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)]) + marker = workspace / "late.txt" + child_code = f"import time;from pathlib import Path;time.sleep(1);Path({str(marker)!r}).write_text('late')" + parent_code = ( + "import subprocess,sys;" + f"subprocess.Popen([sys.executable,'-c',{child_code!r}]," + "stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)" + ) + + result = await backend.exec_shell([sys.executable, "-c", parent_code], timeout=5) + + assert result.exit_code == 0, result.stderr.decode(errors="replace") + await asyncio.sleep(1.2) + assert not marker.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("cancel", [False, True], ids=["timeout", "cancel"]) +async def test_seatbelt_stops_detached_child(tmp_path, cancel): + """A child that left the launcher's process group should be stopped when observed.""" + workspace = tmp_path.resolve() + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)]) + pidfile = workspace / "child.pid" + marker = workspace / "late.txt" + output = workspace / "child.out" + child_code = ( + f"import os,time;os.setsid();open({str(pidfile)!r},'w').write(str(os.getpid()));" + f"time.sleep(1.2);open({str(marker)!r},'w').write('late')" + ) + parent_code = ( + f"import subprocess,sys,time;f=open({str(output)!r},'w');" + f"subprocess.Popen([sys.executable,'-c',{child_code!r}],stdin=f,stdout=f,stderr=f);" + "f.close();time.sleep(10)" + ) + task = asyncio.create_task(backend.exec_shell([sys.executable, "-c", parent_code], timeout=10 if cancel else 0.5)) + try: + for _ in range(100): + if pidfile.exists(): + break + await asyncio.sleep(0.02) + assert pidfile.exists(), output.read_text() if output.exists() else "child did not start" + if cancel: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=4) + else: + result = await asyncio.wait_for(task, timeout=4) + assert result.exit_code == -1 and result.stderr == b"timed out" + await asyncio.sleep(1.2) + assert not marker.exists() + finally: + if not task.done(): + task.cancel() + if pidfile.exists(): + try: + psutil.Process(int(pidfile.read_text(encoding="utf-8"))).kill() + except psutil.Error: + pass diff --git a/tests/unit/test_file_sandbox_backend.py b/tests/unit/test_file_sandbox_backend.py new file mode 100644 index 00000000..ad09c0ad --- /dev/null +++ b/tests/unit/test_file_sandbox_backend.py @@ -0,0 +1,297 @@ +"""Shared file sandbox selection and macOS policy construction.""" + +import asyncio +import os +import plistlib +import sys +from pathlib import Path + +import pytest +from agentscope.tool import ExecResult + +from reme.components.application_context import ApplicationContext +from reme.steps.common.python_execute import PythonExecuteStep +from reme.utils.local_backend import LocalProcessBackend +from reme.utils.file_sandbox import create_file_sandbox_backend +from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend +from reme.utils.file_sandbox.path_seatbelt_backend import PathSeatbeltBackend + + +def test_platform_selection_and_memory_defaults(tmp_path, monkeypatch): + """Linux retains cgroups while macOS defaults to no unsupported memory cap.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux") + linux = create_file_sandbox_backend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + assert isinstance(linux, PathBubbleBackend) + assert linux._memory_limit_mib == 4096 # pylint: disable=protected-access + + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "darwin") + mac = create_file_sandbox_backend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + assert isinstance(mac, PathSeatbeltBackend) + with pytest.warns(RuntimeWarning, match="limit is ignored on macOS"): + limited = create_file_sandbox_backend( + cwd=tmp_path, + read_only_paths=[], + writable_paths=[], + memory_limit_mib=1024, + ) + assert isinstance(limited, PathSeatbeltBackend) + + +@pytest.mark.asyncio +async def test_unsupported_platform_requires_backend_only_for_path_policy(tmp_path, monkeypatch): + """An unsupported platform runs ordinary Python but rejects a requested sandbox.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "unsupported") + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + plain = await step(code="print('host')") + assert plain.success is True + assert plain.answer == "host\n" + + restricted = await step(code="print('blocked')", read_only_paths=[]) + assert restricted.success is False + assert restricted.answer == "No file sandbox backend for unsupported" + + +def test_seatbelt_profile_limits_file_access(tmp_path): + """The profile grants configured paths and a separate per-run temporary directory.""" + output = tmp_path / "output" + output.mkdir() + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(output)]) + profile = backend._profile(str(tmp_path / "temporary")) # pylint: disable=protected-access + assert "(deny default)" in profile + assert f'(allow file-read* (subpath "{tmp_path}"))' in profile + assert f'(allow file-write* (subpath "{output}"))' in profile + assert f'(allow file-write* (subpath "{tmp_path}"))' not in profile + assert '(allow file-write* (literal "/dev/null"))' in profile + assert '(allow file-write* (subpath "/dev"))' not in profile + assert "network*" not in profile + + +def test_sandbox_reads_virtual_environment_base_runtime(tmp_path, monkeypatch): + """A virtual environment grants its real base interpreter without widening writes.""" + workspace = tmp_path / "workspace" + venv = tmp_path / "venv" + runtime = tmp_path / "runtime" + interpreter = tmp_path / "interpreter" / "python" + workspace.mkdir() + (venv / "bin").mkdir(parents=True) + runtime.mkdir() + interpreter.parent.mkdir(parents=True) + interpreter.touch() + (venv / "bin" / "python").symlink_to(interpreter) + monkeypatch.setattr(sys, "prefix", str(venv)) + monkeypatch.setattr(sys, "base_prefix", str(runtime)) + monkeypatch.setattr(sys, "executable", str(venv / "bin" / "python")) + + backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)]) + profile = backend._profile(str(workspace / "temporary")) # pylint: disable=protected-access + + assert f'(allow file-read* (subpath "{runtime}"))' in profile + assert f'(allow file-read* (literal "{interpreter}"))' in profile + assert f'(allow file-write* (subpath "{runtime}"))' not in profile + assert f'(allow file-write* (literal "{interpreter}"))' not in profile + + +def test_seatbelt_excludes_read_only_child_from_writable_parent(tmp_path): + """A parent write grant must not override a nested read-only restriction.""" + protected = tmp_path / "protected" + protected.mkdir() + output = protected / "output" + output.mkdir() + backend = PathSeatbeltBackend( + cwd=tmp_path, + read_only_paths=[str(protected)], + writable_paths=[str(tmp_path), str(output)], + ) + profile = backend._profile(str(tmp_path / "temporary")) # pylint: disable=protected-access + assert ( + f'(allow file-write* (require-all (subpath "{tmp_path}") ' f'(require-not (subpath "{protected}"))))' + ) in profile + assert f'(allow file-write* (subpath "{output}"))' in profile + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sandbox", [False, True]) +@pytest.mark.parametrize("cancel", [False, True]) +async def test_inherited_pipe_cleanup_is_bounded(tmp_path, monkeypatch, sandbox, cancel): + """A surviving child with an output pipe must not hold the timeout response open.""" + from unittest.mock import Mock + + backend = ( + PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=0) + if sandbox + else LocalProcessBackend() + ) + started = asyncio.Event() + cancelled = asyncio.Event() + + class StuckPipe: + """Model an inherited pipe held open after termination.""" + + async def read(self, _size): + """Wait forever until the caller cancels pipe draining.""" + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + class StuckProcess: + """Model a child that keeps its output pipe open after termination.""" + + pid = 12345 + _transport = Mock() + stdout = StuckPipe() + stderr = StuckPipe() + stdin = None + + async def wait(self): + """A stale process transport must also have a bounded wait.""" + await asyncio.Event().wait() + + async def start(*_args, **_kwargs): + return StuckProcess() + + monkeypatch.setattr(backend, "_start_process", start) + monkeypatch.setattr(backend, "_kill_process", lambda _process: None) + monkeypatch.setattr("reme.utils.local_backend._CLEANUP_TIMEOUT", 0.01) + + task = asyncio.create_task(backend.exec_shell(["true"], timeout=None if cancel else 0.01)) + await asyncio.wait_for(started.wait(), timeout=1) + if cancel: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + else: + result = await asyncio.wait_for(task, timeout=1) + assert result.exit_code == -1 and result.stderr == b"timed out" + assert cancelled.is_set() + StuckProcess._transport.close.assert_called_once() # pylint: disable=protected-access + + +@pytest.mark.asyncio +async def test_seatbelt_launch_failure_cleans_up_without_host_fallback(tmp_path, monkeypatch): + """A failed sandbox launcher never starts the requested command directly.""" + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "/usr/bin/sandbox-exec") + seen = {} + + async def fail(*arguments): + if arguments[0] == "bootout": + import errno + from reme.utils.file_sandbox.path_seatbelt_backend import LaunchctlError + + raise LaunchctlError("service not found", errno.ESRCH) + seen["arguments"] = arguments + with open(arguments[-1], "rb") as stream: + seen["job"] = plistlib.load(stream) + raise RuntimeError("launchd unavailable") + + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + monkeypatch.setattr(backend, "_launchctl", fail) + + async def available(_domain): + return True + + monkeypatch.setattr(backend, "_has_gui_launchd", available) + with pytest.raises(RuntimeError, match="launchd unavailable"): + await backend.exec_shell(["/bin/echo", "unsafe"]) + assert seen["arguments"][0] == "bootstrap" + assert seen["job"]["AbandonProcessGroup"] is False + assert seen["job"]["ProgramArguments"][-2:] == ["/bin/echo", "unsafe"] + assert not Path(seen["arguments"][-1]).exists() + + +@pytest.mark.asyncio +async def test_python_job_selects_seatbelt_on_macos(tmp_path, monkeypatch): + """Path injection reaches the macOS backend with no implicit memory cap.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "darwin") + seen = {} + + async def fake_exec(self, command, *, cwd=None, timeout=None): + seen.update(backend=self, command=command, cwd=cwd, timeout=timeout) + return ExecResult(exit_code=0, stdout=b"ok\n", stderr=b"") + + monkeypatch.setattr(PathSeatbeltBackend, "exec_shell", fake_exec) + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code="print('ok')", read_only_paths=[]) + assert response.success is True + assert isinstance(seen["backend"], PathSeatbeltBackend) + assert seen["command"] == [sys.executable, "-c", "print('ok')"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["success", "failure", "timeout", "cancel", "error", "kickstart_error"]) +async def test_seatbelt_launchd_lifecycle(tmp_path, monkeypatch, mode): + """Every launchd outcome unregisters the job and removes its control files.""" + from unittest.mock import AsyncMock, Mock + + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "sandbox-exec") + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=True)) + killed = Mock() + monkeypatch.setattr(backend, "_kill_process", killed) + calls = [] + control = None + started = asyncio.Event() + + async def launchctl(*args): + nonlocal control + calls.append(args[0]) + if args[0] == "bootstrap": + control = Path(args[-1]).parent + if args[0] == "kickstart" and mode == "kickstart_error": + raise RuntimeError("kickstart failed") + if args[0] == "bootout": + # Output is collected after launchd has reaped the process group. + (control / "stdout").write_bytes(b"output") + (control / "stderr").write_bytes(b"diagnostic") + return "12345" + + async def wait_status(*_args): + started.set() + if mode == "error": + raise RuntimeError("status failed") + if mode in ("timeout", "cancel"): + await asyncio.Event().wait() + return 7 if mode == "failure" else 0 + + monkeypatch.setattr(backend, "_launchctl", launchctl) + monkeypatch.setattr(backend, "_wait_status", wait_status) + task = asyncio.create_task(backend.exec_shell(["true"], timeout=0.01 if mode == "timeout" else None)) + if mode == "cancel": + await asyncio.wait_for(started.wait(), timeout=1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + elif mode in ("error", "kickstart_error"): + with pytest.raises(RuntimeError, match="failed"): + await asyncio.wait_for(task, timeout=1) + else: + result = await asyncio.wait_for(task, timeout=1) + if mode == "timeout": + assert result == ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out") + else: + assert result.exit_code == (7 if mode == "failure" else 0) + assert result.stdout == b"output" and result.stderr == b"diagnostic" + assert calls == ["bootstrap", "kickstart", "bootout"] + assert not control.exists() + assert killed.call_count == int(mode in ("timeout", "cancel", "error")) + + +@pytest.mark.asyncio +async def test_python_timeout_preserves_exit_status_and_stderr(tmp_path): + """The shared executor retains the existing unsandboxed Python metadata.""" + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step( + code="import sys, time; print('diagnostic', file=sys.stderr, flush=True); time.sleep(60)", + python_timeout=0.3, + ) + assert response.success is False + assert response.answer == "Python execution timed out after 0.3s" + assert response.metadata == { + "returncode": -9 if os.name == "posix" else 1, + "stderr": "diagnostic\n", + "python_timeout": 0.3, + } diff --git a/tests/unit/test_path_bubble_backend.py b/tests/unit/test_path_bubble_backend.py new file mode 100644 index 00000000..d7fd495d --- /dev/null +++ b/tests/unit/test_path_bubble_backend.py @@ -0,0 +1,248 @@ +"""Path-preserving Bubblewrap mounts and job integration.""" + +import sys +from types import SimpleNamespace +from pathlib import Path + +import pytest +import yaml + +from agentscope.tool import ExecResult + +from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend +from reme.components.application_context import ApplicationContext +from reme.steps.common.python_execute import PythonExecuteStep + + +@pytest.fixture(autouse=True) +def linux_backend_selection(monkeypatch): + """Keep existing integration assertions on the Linux backend.""" + monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux") + + +def _mounts(argv): + return [ + (argv[index], argv[index + 1], argv[index + 2]) + for index in range(len(argv) - 2) + if argv[index] in ("--bind", "--ro-bind") + ] + + +def test_path_bubble_backend_preserves_host_paths_and_mount_order(tmp_path): + """Nested writable mounts override a read-only parent at the same path.""" + output = tmp_path / "output" + output.mkdir() + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(output)]) + + argv = backend._bwrap_argv(["/bin/sh", "-c", "pwd"], str(tmp_path)) # pylint: disable=protected-access + mounts = _mounts(argv) + + assert ("--ro-bind", str(tmp_path), str(tmp_path)) in mounts + assert ("--bind", str(output), str(output)) in mounts + assert mounts.index(("--ro-bind", str(tmp_path), str(tmp_path))) < mounts.index( + ("--bind", str(output), str(output)), + ) + assert argv[-3:] == ["/bin/sh", "-c", "pwd"] + assert argv[argv.index("--chdir") + 1] == str(tmp_path) + assert "--share-net" not in argv + + +def test_path_bubble_backend_adds_workspace_read_only(tmp_path): + """The working directory is always accessible inside the sandbox.""" + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + argv = backend._bwrap_argv(["true"], str(tmp_path)) # pylint: disable=protected-access + assert ("--ro-bind", str(tmp_path), str(tmp_path)) in _mounts(argv) + + +@pytest.mark.parametrize("limit", [-1, True, 1.5, "4096"]) +def test_path_bubble_backend_rejects_invalid_memory_limit(tmp_path, limit): + """Memory limits must be non-negative whole MiB.""" + with pytest.raises(ValueError, match="memory_limit_mib"): + PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=limit) + + +def test_one_mib_is_written_as_bytes_to_cgroup(tmp_path, monkeypatch): + """The public MiB limit is converted only at the Linux cgroup boundary.""" + proc_file = tmp_path / "proc-self-cgroup" + proc_file.write_text("0::/\n", encoding="ascii") + cgroup_root = tmp_path / "cgroup" + cgroup_root.mkdir() + original_path = Path + + def sandbox_path(value): + if value == "/proc/self/cgroup": + return proc_file + if value == "/sys/fs/cgroup": + return cgroup_root + return original_path(value) + + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.Path", sandbox_path) + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=1) + group = backend._create_cgroup() # pylint: disable=protected-access + assert (group / "memory.max").read_text(encoding="ascii") == str(1024**2) + + +@pytest.mark.asyncio +async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch): + """Each process receives its own cgroup before bubblewrap is executed.""" + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=1024) + groups = [] + subprocess_options = [] + subprocess_commands = [] + + def create_group(): + group = tmp_path / f"group-{len(groups)}" + group.mkdir() + groups.append(group) + return group + + async def create_process(*argv, **kwargs): + subprocess_commands.append(argv) + subprocess_options.append(kwargs) + return SimpleNamespace(pid=100 + len(subprocess_options)) + + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap") + monkeypatch.setattr(backend, "_create_cgroup", create_group) + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.asyncio.create_subprocess_exec", create_process) + first = await backend._start_process(["true"], str(tmp_path)) # pylint: disable=protected-access + second = await backend._start_process(["true"], str(tmp_path)) # pylint: disable=protected-access + + assert len(groups) == 2 and groups[0] != groups[1] + assert all("preexec_fn" not in options for options in subprocess_options) + for argv, group in zip(subprocess_commands, groups): + assert argv[:4] == (sys.executable, "-I", "-S", "-c") + assert argv[5:7] == (str(group / "cgroup.procs"), "bwrap") + backend._release_process(first) # pylint: disable=protected-access + backend._release_process(second) # pylint: disable=protected-access + assert not any(group.exists() for group in groups) + + +@pytest.mark.asyncio +async def test_zero_memory_limit_skips_cgroup(tmp_path, monkeypatch): + """A zero limit bypasses cgroup creation and child enrollment.""" + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=0) + options = {} + + async def create_process(*_argv, **kwargs): + options.update(kwargs) + return SimpleNamespace(pid=123) + + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap") + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.asyncio.create_subprocess_exec", create_process) + await backend._start_process(["true"], str(tmp_path)) # pylint: disable=protected-access + assert "preexec_fn" not in options + + +@pytest.mark.parametrize("path", ["relative", "/", "/does-not-exist"]) +def test_path_bubble_backend_rejects_invalid_mounts(tmp_path, path): + """Only existing absolute non-root paths can be mounted.""" + with pytest.raises(ValueError): + PathBubbleBackend(cwd=tmp_path, read_only_paths=[path], writable_paths=[]) + + +def test_path_bubble_backend_rejects_symlink_and_duplicates(tmp_path): + """Ambiguous mount sources are rejected before a process starts.""" + alias = tmp_path.parent / "alias" + alias.symlink_to(tmp_path) + try: + with pytest.raises(ValueError): + PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(alias)], writable_paths=[]) + with pytest.raises(ValueError): + PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(tmp_path)]) + finally: + alias.unlink() + + +def test_path_bubble_backend_rejects_replaced_mount_source(tmp_path): + """A path swapped after validation cannot become a new mount source.""" + source = tmp_path / "source" + source.mkdir() + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(source)], writable_paths=[]) + source.rename(tmp_path / "old-source") + source.mkdir() + + with pytest.raises(RuntimeError, match="sandbox path changed"): + backend._bwrap_argv(["true"], str(tmp_path)) # pylint: disable=protected-access + + +@pytest.mark.asyncio +async def test_python_execute_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch): + """Python execution uses the same backend and timeout units.""" + seen = {} + + async def fake_exec(self, command, *, cwd=None, timeout=None): + seen.update(command=command, cwd=cwd, timeout=timeout, backend=self) + return ExecResult(exit_code=0, stdout=b"ok\n", stderr=b"") + + monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec) + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code="print('ok')", read_only_paths=[], writable_paths=[str(tmp_path)]) + + assert response.success is True + assert response.answer == "ok\n" + assert seen["command"] == [sys.executable, "-c", "print('ok')"] + assert seen["cwd"] == str(tmp_path) + assert seen["timeout"] == 60 + assert seen["backend"]._memory_limit_mib == 4096 # pylint: disable=protected-access + + +@pytest.mark.asyncio +async def test_python_execute_passes_custom_memory_limit(tmp_path, monkeypatch): + """Job parameters reach the backend without changing path selection.""" + seen = {} + + async def fake_exec(self, _command, *, cwd=None, timeout=None): # pylint: disable=unused-argument + seen["limit"] = self._memory_limit_mib # pylint: disable=protected-access + return ExecResult(exit_code=0, stdout=b"", stderr=b"") + + monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec) + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code="pass", read_only_paths=[], memory_limit_mib=0) + assert response.success is True + assert seen["limit"] == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("limit", [0, 1, -1]) +async def test_python_execute_rejects_memory_limit_without_path_policy(tmp_path, limit): + """A requested memory policy must never be silently ignored.""" + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code="print('should not run')", memory_limit_mib=limit) + assert response.success is False + assert response.answer == "memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox" + + +@pytest.mark.asyncio +async def test_step_rejects_invalid_mount_parameters(tmp_path): + """Malformed path injection cannot silently disable sandboxing.""" + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code="print('unsafe')", writable_paths="/tmp") + assert response.success is False + assert "writable_paths must be a list" in response.answer + + +@pytest.mark.asyncio +async def test_sandbox_start_failure_does_not_run_on_host(tmp_path, monkeypatch): + """Python execution reports a sandbox failure instead of running on the host.""" + + async def unavailable(*_args, **_kwargs): + raise RuntimeError("bwrap unavailable") + + monkeypatch.setattr(PathBubbleBackend, "_start_process", unavailable) + app_context = ApplicationContext(workspace_dir=str(tmp_path)) + python_step = PythonExecuteStep(app_context=app_context) + python_result = await python_step(code="print('host')", read_only_paths=[]) + + assert python_result.success is False + assert "bwrap unavailable" in python_result.answer + + +def test_job_hides_injected_path_policy(): + """Path policy reaches the Step through context without becoming a tool argument.""" + config_path = Path(__file__).parents[2] / "reme" / "config" / "benchmark.yaml" + parameters = yaml.safe_load(config_path.read_text(encoding="utf-8"))["jobs"]["python_execute"]["parameters"] + assert "read_only_paths" not in parameters["properties"] + assert "writable_paths" not in parameters["properties"] + assert "default" not in parameters["properties"]["memory_limit_mib"] diff --git a/tests/unit/test_process_output_limits.py b/tests/unit/test_process_output_limits.py new file mode 100644 index 00000000..55acb81c --- /dev/null +++ b/tests/unit/test_process_output_limits.py @@ -0,0 +1,156 @@ +"""Execution output stays bounded across pipe and launchd backends.""" + +# pylint: disable=protected-access + +import asyncio +import os +import plistlib +import stat +import sys +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +from reme.components.application_context import ApplicationContext +from reme.steps.common.python_execute import PythonExecuteStep +from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend +from reme.utils.file_sandbox.path_seatbelt_backend import PathSeatbeltBackend +from reme.utils.local_backend import LocalProcessBackend, OUTPUT_LIMIT_BYTES + + +@pytest.mark.asyncio +@pytest.mark.parametrize("backend_name", ["local", "bubble", "seatbelt"]) +@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"]) +@pytest.mark.parametrize("extra", [0, 1]) +async def test_pipe_output_boundary(tmp_path, monkeypatch, backend_name, stream, extra): + """Each stream has its own budget; crossing either budget fails execution.""" + local = LocalProcessBackend() + backend = local + if backend_name != "local": + backend_class = PathBubbleBackend if backend_name == "bubble" else PathSeatbeltBackend + backend = backend_class(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + # Exercise the shared collector with a real process, without requiring + # kernel sandbox support or changing the host launchd domain. + monkeypatch.setattr(backend, "_start_process", local._start_process) + if backend_name == "seatbelt": + monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=False)) + size = OUTPUT_LIMIT_BYTES + extra + code = "import os;" + ";".join( + f"os.write({fd}, b'x' * {size})" for fd in (1, 2) if stream in ("both", "stdout" if fd == 1 else "stderr") + ) + if backend_name == "seatbelt": + with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"): + result = await backend.exec_shell([sys.executable, "-c", code], timeout=5) + else: + result = await backend.exec_shell([sys.executable, "-c", code], timeout=5) + assert len(result.stdout) <= OUTPUT_LIMIT_BYTES + assert len(result.stderr) <= OUTPUT_LIMIT_BYTES + if extra: + assert not result.ok() + assert b"Output limit exceeded:" in result.stderr + else: + assert result.ok(), result.stderr + assert result.stdout == (b"x" * size if stream in ("stdout", "both") else b"") + assert result.stderr == (b"x" * size if stream in ("stderr", "both") else b"") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fd", [1, 2]) +async def test_continuous_output_stops_process(tmp_path, fd): + """Overflow terminates a live writer before it can perform a later write.""" + marker = tmp_path / "late.txt" + code = ( + "import os,time;from pathlib import Path;" + f"os.write({fd}, b'x' * {OUTPUT_LIMIT_BYTES + 1});" + f"time.sleep(1);Path({str(marker)!r}).write_text('late')" + ) + backend = LocalProcessBackend() + process = await backend._start_process([sys.executable, "-c", code], str(tmp_path)) + result = await asyncio.wait_for(backend._communicate(process), timeout=3) + assert not result.ok() + assert b"Output limit exceeded:" in result.stderr + assert process.returncode is not None + assert not marker.exists() + + +@pytest.mark.asyncio +async def test_python_step_reports_output_overflow(tmp_path): + """Python callers receive bounded partial output and failure metadata.""" + step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(code=f"import os;os.write(1, b'x' * {OUTPUT_LIMIT_BYTES + 1})") + assert response.success is False + assert len(response.answer) == OUTPUT_LIMIT_BYTES + assert "Output limit exceeded: stdout" in response.metadata["stderr"] + assert response.metadata["returncode"] != 0 + + +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="launchd uses POSIX FIFOs") +@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"]) +@pytest.mark.parametrize("extra", [0, 1]) +async def test_launchd_output_boundary(tmp_path, monkeypatch, stream, extra): + """Real FIFO writers obey each stream's budget and are reaped on overflow.""" + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.sys.platform", "darwin") + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "sandbox-exec") + monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=True)) + size = OUTPUT_LIMIT_BYTES + extra + code = "import os,time;" + ";".join( + f"os.write({fd}, b'x' * {size})" for fd in (1, 2) if stream in ("both", "stdout" if fd == 1 else "stderr") + ) + if extra: + code += ";time.sleep(60)" + job = {} + process = None + control = None + calls = [] + + async def launchctl(*args): + nonlocal job, process, control + calls.append(args[0]) + if args[0] == "bootstrap": + control = Path(args[-1]).parent + with open(args[-1], "rb") as source: + job = plistlib.load(source) + for key in ("StandardOutPath", "StandardErrorPath"): + assert stat.S_ISFIFO(os.stat(job[key]).st_mode) + elif args[0] == "kickstart": + with open(job["StandardOutPath"], "wb") as stdout, open(job["StandardErrorPath"], "wb") as stderr: + process = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + code, + stdout=stdout, + stderr=stderr, + start_new_session=True, + ) + return str(process.pid) + elif args[0] == "bootout": + if process.returncode is None: + process.kill() + await process.wait() + return "" + + async def wait_status(*_args): + return await process.wait() + + monkeypatch.setattr(backend, "_launchctl", launchctl) + monkeypatch.setattr(backend, "_wait_status", wait_status) + try: + result = await asyncio.wait_for(backend.exec_shell(["test-command"], timeout=5), timeout=7) + assert len(result.stdout) <= OUTPUT_LIMIT_BYTES + assert len(result.stderr) <= OUTPUT_LIMIT_BYTES + assert result.ok() is (extra == 0) + if extra: + assert b"Output limit exceeded:" in result.stderr + else: + assert result.stdout == (b"x" * size if stream in ("stdout", "both") else b"") + assert result.stderr == (b"x" * size if stream in ("stderr", "both") else b"") + assert calls == ["bootstrap", "kickstart", "bootout"] + assert process.returncode is not None + assert not control.exists() + finally: + if process is not None and process.returncode is None: + process.kill() + await process.wait() diff --git a/tests/unit/test_sandbox_startup_cancellation.py b/tests/unit/test_sandbox_startup_cancellation.py new file mode 100644 index 00000000..88783383 --- /dev/null +++ b/tests/unit/test_sandbox_startup_cancellation.py @@ -0,0 +1,164 @@ +"""Cancellation during startup must not abandon OS-owned sandbox resources.""" + +# pylint: disable=protected-access + +import asyncio +import errno +import sys +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend +from reme.utils.file_sandbox.path_seatbelt_backend import LaunchctlError, PathSeatbeltBackend + + +@pytest.mark.asyncio +@pytest.mark.parametrize("late_registration", [False, True]) +@pytest.mark.parametrize("cancel_twice", [False, True]) +async def test_cancelled_bootstrap_waits_and_unregisters(tmp_path, monkeypatch, late_registration, cancel_twice): + """Unregister after a pending bootstrap settles, including a late successful reply.""" + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.sys.platform", "darwin") + monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "sandbox-exec") + monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=True)) + entered = asyncio.Event() + reply = asyncio.Event() + bootout_entered = asyncio.Event() + bootout_reply = asyncio.Event() + services = set() + calls = [] + control = None + + async def launchctl(*args): + nonlocal control + calls.append(args[0]) + if args[0] == "bootstrap": + control = Path(args[-1]).parent + if not late_registration: + services.add("sandbox") + entered.set() + await reply.wait() + services.add("sandbox") + elif args[0] == "bootout": + assert reply.is_set(), "bootout must not race an unfinished registration" + assert control.exists(), "keep control files until the service is removed" + bootout_entered.set() + await bootout_reply.wait() + services.clear() + else: + pytest.fail(f"Unexpected launchctl call: {args[0]}") + return "" + + monkeypatch.setattr(backend, "_launchctl", launchctl) + task = asyncio.create_task(backend.exec_shell(["true"])) + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + reply.set() + if cancel_twice: + await asyncio.wait_for(bootout_entered.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + bootout_reply.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + assert calls == ["bootstrap", "bootout"] + assert not services + assert not control.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("accepted", [False, True]) +async def test_failed_bootstrap_attempts_unregister(tmp_path, monkeypatch, accepted): + """A failed reply may still hide an accepted registration; absence alone is benign.""" + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + plist = tmp_path / "job.plist" + calls = [] + services = set() + + async def launchctl(*args): + calls.append(args[0]) + if args[0] == "bootstrap": + if accepted: + services.add("sandbox") + raise LaunchctlError("registration reply failed", 5) + if not services: + raise LaunchctlError("service not found", errno.ESRCH) + services.clear() + return "" + + monkeypatch.setattr(backend, "_launchctl", launchctl) + with pytest.raises(LaunchctlError, match="registration reply failed"): + await backend._bootstrap("gui/501", plist, "gui/501/sandbox") + assert calls == ["bootstrap", "bootout"] + assert not services + + +@pytest.mark.asyncio +async def test_bootstrap_rollback_reports_unregistration_failure(tmp_path, monkeypatch): + """Permission and other cleanup failures must not be mistaken for a missing service.""" + backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[]) + launchctl = AsyncMock(side_effect=[LaunchctlError("bootstrap failed", 5), LaunchctlError("bootout denied", 1)]) + monkeypatch.setattr(backend, "_launchctl", launchctl) + with pytest.raises(LaunchctlError, match="bootout denied"): + await backend._bootstrap("gui/501", tmp_path / "job.plist", "gui/501/sandbox") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("memory_limit", [0, 1]) +@pytest.mark.parametrize("spawn_fails", [False, True]) +async def test_cancelled_bubble_start_reaps_before_removing_cgroup(tmp_path, monkeypatch, memory_limit, spawn_fails): + """A delayed process handle is reclaimed before its cgroup is removed.""" + backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=memory_limit) + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.sys.platform", "linux") + monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "bwrap") + group = tmp_path / "cgroup" + + def create_group(): + group.mkdir() + return group + + monkeypatch.setattr(backend, "_create_cgroup", create_group) + entered = asyncio.Event() + reply = asyncio.Event() + processes = [] + create_process = asyncio.create_subprocess_exec + + async def delayed_spawn(*_args, **kwargs): + if not spawn_fails: + processes.append(await create_process(sys.executable, "-c", "import time;time.sleep(60)", **kwargs)) + entered.set() + await reply.wait() + if spawn_fails: + raise FileNotFoundError("spawn failed after cancellation") + return processes[0] + + monkeypatch.setattr("reme.utils.local_backend.asyncio.create_subprocess_exec", delayed_spawn) + task = asyncio.create_task(backend._start_process(["true"], str(tmp_path))) + try: + await asyncio.wait_for(entered.wait(), timeout=2) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert group.exists() is bool(memory_limit) + reply.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=2) + assert all(process.returncode is not None for process in processes) + assert not group.exists() + assert not backend._process_cgroups + finally: + reply.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + for process in processes: + if process.returncode is None: + process.kill() + await process.communicate()