Harden read-only shell commands and filesystem sandboxes

plugins/home/src/reme_home/auto_bash.py

@@ -15,17 +15,28 @@ _READ_COMMANDS = {
     "diff": {"denied": ["--output"]},
     "echo": {},
     "false": {},
-    "file": {},
+    "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"],
+        "denied": [
+            "--output",
+            "--ext-diff",
+            "--textconv",
+            "--paginate",
+            "--submodule",
+            "--ignore-submodules",
+            "--show-signature",
+            "--remerge-diff",
+            "--diff-merges",
+        ],
     },
     "grep": {},
     "head": {},
     "ls": {},
     "pwd": {},
-    "rg": {"denied": ["--pre", "--pre-glob", "--replace"]},
     "sort": {"denied": ["-o", "--output", "--compress-program", "--temporary-directory"]},
     "stat": {},
     "tail": {},
@@ -35,9 +46,88 @@ _READ_COMMANDS = {
 }
 _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 GIT_TRACE GIT_TRACE_SETUP GIT_TRACE_PERFORMANCE
+unset GIT_TRACE_PACK_ACCESS GIT_TRACE_PACKET GIT_TRACE_CURL GIT_TRACE2 GIT_TRACE2_EVENT GIT_TRACE2_PERF
+export GIT_OPTIONAL_LOCKS=0 GIT_PAGER=/bin/cat GIT_NO_LAZY_FETCH=1 GIT_TERMINAL_PROMPT=0
+git() {
+    if 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 trace2.normalTarget=0 -c trace2.eventTarget=0 -c trace2.perfTarget=0 "$@"
+}
+"""
+
+
+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 read-only execution guard."""
+    """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."
@@ -67,26 +157,24 @@ class AutoBash(Bash):
         # remaining shell syntax and nested commands.
         if "$" in command or "`" in command:
             return False
-        from safecmd import CmdSpec, validate  # pylint: disable=import-outside-toplevel
-        from safecmd.bashxtract import extract_commands  # pylint: disable=import-outside-toplevel
-        from safecmd.core import normalize_dest  # pylint: disable=import-outside-toplevel
-
-        specs = {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()}
         try:
+            from safecmd import CmdSpec, validate  # pylint: disable=import-outside-toplevel
+            from safecmd.bashxtract import extract_commands, parse_bash  # pylint: disable=import-outside-toplevel
+            from safecmd.core import normalize_dest  # pylint: disable=import-outside-toplevel
+
+            if not _literal_shell(parse_bash(command), command):
+                return False
+            specs = {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()}
             validate(command, cmds=specs, dests=_SAFE_DESTINATIONS)
             commands, operators, redirects = extract_commands(command)
-            for tokens in commands:
-                options = _READ_COMMANDS[tokens[0]]
-                # An allow list constrains the first argument (for git, its subcommand).
-                # CmdSpec independently rejects every denied argument.
-                if "allow" in options and (len(tokens) < 2 or tokens[1] not in options["allow"]):
-                    return False
-            if "=" in operators or any(normalize_dest(destination) != "/dev/null" for _, destination in redirects):
+            if any(not _allowed_arguments(tokens) for tokens in commands):
                 return False
+            return "=" not in operators and all(
+                normalize_dest(destination) == "/dev/null" for _, destination in redirects
+            )
         except Exception:  # pylint: disable=broad-exception-caught
             # Parser failures and missing shfmt must never allow execution.
             return False
-        return True
 
     async def check_read_only(self, tool_input: dict) -> bool:
         """Report the same policy used by the execution guard."""
@@ -113,10 +201,8 @@ class AutoBash(Bash):
                 is_last=True,
             )
             return
-        # Git status may otherwise refresh the index, and Git's pager may run
-        # a configured program. These exports apply only in read-only mode.
         if self.is_read_only:
-            command = "export GIT_OPTIONAL_LOCKS=0 GIT_PAGER=/bin/cat; " + command
+            command = _READ_ONLY_SETUP + command
         async for chunk in super().call(command=command, description=description, timeout=timeout):
             yield chunk
 

plugins/home/tests/test_auto_bash.py

@@ -1,6 +1,7 @@
 """Shell job and its optional read-only execution guard."""
 
 import inspect
+import subprocess
 from importlib.metadata import EntryPoint
 from pathlib import Path
 
@@ -64,6 +65,27 @@ async def test_auto_bash_runs_in_workspace(tmp_path):
         "GIT_PAGER=sh git log",
         "nop",
         "rg --pre='touch denied.txt' pattern .",
+        "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 show --remerge-diff",
+        "rg --hostname-bin=./hook --hyperlink-format=default hello .",
+        "rg -z hello .",
+        "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):
@@ -248,7 +270,6 @@ def test_home_plugin_registers_job_and_step(monkeypatch):
         "true",
         "false",
         "grep hello note.txt",
-        "rg hello note.txt",
         "find . -name note.txt",
         "sort note.txt",
         "diff note.txt note.txt",
@@ -277,6 +298,46 @@ async def test_safe_redirection_is_allowed(tmp_path):
 
     assert await tool.check_read_only({"command": "echo hello > /dev/null"})
     assert await tool.check_read_only({"command": "cat < note.txt"})
+    assert await tool.check_read_only({"command": "echo hello >| /dev/null"})
+    assert await tool.check_read_only({"command": "echo hello 2>&1"})
+    assert await tool.check_read_only({"command": "find . -name '*.txt'"})
+    assert await tool.check_read_only({"command": "(cat note.txt) | head -n 1"})
+
+
+@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

reme/utils/path_bubble_backend.py

@@ -10,6 +10,16 @@ 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."""
@@ -65,23 +75,14 @@ class PathBubbleBackend(FileSandboxBackend):
         argv = self._bwrap_argv(command, cwd)
         group = self._create_cgroup() if self._memory_limit_bytes else None
         try:
-
-            def join_cgroup() -> None:
-                """Move the forked child before it executes bubblewrap."""
-                assert group is not None
-                descriptor = os.open(group / "cgroup.procs", os.O_WRONLY)
-                try:
-                    os.write(descriptor, str(os.getpid()).encode("ascii"))
-                finally:
-                    os.close(descriptor)
-
+            if group is not None:
+                argv = [sys.executable, "-I", "-S", "-c", _CGROUP_EXEC, str(group / "cgroup.procs"), *argv]
             process = await asyncio.create_subprocess_exec(
                 *argv,
                 stdin=stdin,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
                 start_new_session=True,
-                preexec_fn=join_cgroup if group is not None else None,
             )
         except BaseException:
             if group is not None:

reme/utils/path_seatbelt_backend.py

@@ -8,7 +8,7 @@ import sys
 import tempfile
 import warnings
 
-from .file_sandbox_backend import FileSandboxBackend
+from .file_sandbox_backend import FileSandboxBackend, contains
 
 
 class PathSeatbeltBackend(FileSandboxBackend):
@@ -43,9 +43,35 @@ class PathSeatbeltBackend(FileSandboxBackend):
         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)"]
+        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)
-        lines.extend(f"(allow file-write* {self._path_rule(path)})" for path in self._writable_paths + [tempdir])
+        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})")
         return "\n".join(lines)
 
     async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):

tests/integration/test_seatbelt_backend.py

@@ -0,0 +1,71 @@
+"""Exercise macOS sandbox startup and file isolation using real subprocesses."""
+
+import sys
+
+import pytest
+
+from reme.utils.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_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"

tests/unit/test_file_sandbox_backend.py

@@ -60,6 +60,24 @@ def test_seatbelt_profile_limits_file_access(tmp_path):
     assert "network*" 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
 async def test_seatbelt_launch_failure_cleans_up_without_host_fallback(tmp_path, monkeypatch):
     """A failed sandbox launcher never starts the requested command directly."""

tests/unit/test_path_bubble_backend.py

@@ -67,6 +67,7 @@ async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch)
     backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_bytes=1024)
     groups = []
     subprocess_options = []
+    subprocess_commands = []
 
     def create_group():
         group = tmp_path / f"group-{len(groups)}"
@@ -74,7 +75,8 @@ async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch)
         groups.append(group)
         return group
 
-    async def create_process(*_argv, **kwargs):
+    async def create_process(*argv, **kwargs):
+        subprocess_commands.append(argv)
         subprocess_options.append(kwargs)
         return SimpleNamespace(pid=100 + len(subprocess_options))
 
@@ -86,7 +88,10 @@ async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch)
     second = await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
 
     assert len(groups) == 2 and groups[0] != groups[1]
-    assert all(options["preexec_fn"] is not None for options in subprocess_options)
+    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_cgroup(first)  # pylint: disable=protected-access
     backend._release_cgroup(second)  # pylint: disable=protected-access
     assert not any(group.exists() for group in groups)
@@ -106,7 +111,7 @@ async def test_zero_memory_limit_skips_cgroup(tmp_path, monkeypatch):
     monkeypatch.setattr("reme.utils.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap")
     monkeypatch.setattr("reme.utils.path_bubble_backend.asyncio.create_subprocess_exec", create_process)
     await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
-    assert options["preexec_fn"] is None
+    assert "preexec_fn" not in options
 
 
 @pytest.mark.parametrize("path", ["relative", "/", "/does-not-exist"])