fix(security): tokenize the execute_bash destructive-command screen (#128)

core/harness/command_guard.py

@@ -0,0 +1,151 @@
+"""Destructive-command screening — cheap defense-in-depth, not the boundary.
+
+The sandbox in :mod:`core.harness.sandbox` is what actually enforces the
+execution boundary (writes fenced to the workspace on seatbelt/bwrap). This
+module is the *shallow first pass* that sits in front of it: a fast, best-effort
+check that catches obviously destructive commands before they ever reach the
+shell.
+
+Why this exists as its own module. The original check in ``execute_bash`` was::
+
+    dangerous = ["rm -rf", "sudo", "chmod 777", "mkfs", "dd if="]
+    if any(d in command.lower() for d in dangerous):
+        block
+
+That is a raw substring match, and it is trivially bypassable — which the
+project already acknowledged (issue #128). ``rm  -rf`` (two spaces), ``rm -r -f``
+(split flags), ``rm  --recursive --force``, or ``chmod  0777`` all sail straight
+through, while a *benign* path like ``touch rm-rf-notes.txt`` is falsely blocked.
+A blocklist can never be a security boundary; the sandbox is. But if we keep a
+blocklist at all, it should honestly catch what it *claims* to, rather than
+offering a false sense of coverage.
+
+So this module splits the command on the shell control operators and then
+*tokenises* each segment with :func:`shlex.split`, matching on the resulting
+argv — the command name and its flags — instead of substrings of the raw
+string. That closes the whitespace / flag-order / flag-spelling gaps without
+pretending to be exhaustive.
+
+:func:`screen_command` returns a human-readable reason string when a command
+looks destructive, else ``None``. It never raises: a command it cannot parse is
+passed through (``None``) and left to the sandbox, exactly as before — this
+layer only ever *adds* friction to clearly dangerous commands, never removes the
+real protection underneath.
+"""
+
+from __future__ import annotations
+
+import re
+import shlex
+
+__all__ = ["screen_command"]
+
+# Shell control operators that separate one simple command from the next.
+# We split the raw string on these *before* tokenising, because shlex.split is
+# a word splitter, not a shell parser — it would keep "x;" or "/tmp&&" as a
+# single token and hide the following command.
+_OPERATOR_SPLIT = re.compile(r"(?:\|\||\||&&|&|;|\n)")
+
+
+def _has_flag(flag_tokens: list[str], *letters: str) -> bool:
+    """Whether any short-flag cluster contains all of ``letters``.
+
+    ``-rf``, ``-fr`` and ``-r -f`` all count as having both ``r`` and ``f``,
+    because short flags may be combined in any order or split apart.
+    """
+    joined = "".join(t.lstrip("-") for t in flag_tokens)
+    return all(letter in joined for letter in letters)
+
+
+def _has_long_flag(tokens: list[str], *names: str) -> bool:
+    """Whether every long flag in ``names`` (e.g. ``recursive``) is present."""
+    present = {t.lstrip("-") for t in tokens if t.startswith("--")}
+    return all(name in present for name in names)
+
+
+def _is_recursive_force_rm(cmd: str, args: list[str]) -> bool:
+    if cmd != "rm":
+        return False
+    flags = [a for a in args if a.startswith("-")]
+    # rm treats both -r and -R as recursive.
+    recursive = (
+        _has_flag(flags, "r")
+        or _has_flag(flags, "R")
+        or _has_long_flag(args, "recursive")
+    )
+    force = _has_flag(flags, "f") or _has_long_flag(args, "force")
+    return recursive and force
+
+
+def _is_reckless_chmod(cmd: str, args: list[str]) -> bool:
+    """Permissive chmod granting full rwx to everyone (the classic ``777``).
+
+    Catches the numeric ``777``/``0777`` form and the symbolic ``a+rwx`` /
+    ``a=rwx`` form; ignores harmless modes like ``755`` or ``+x``.
+    """
+    if cmd != "chmod":
+        return False
+    for a in args:
+        if a.startswith("-"):
+            continue
+        mode = a
+        if mode.isdigit() and mode[-3:] == "777":
+            return True
+        if mode in {"a+rwx", "a=rwx", "+rwx", "=rwx", "ugo+rwx", "ugo=rwx"}:
+            return True
+    return False
+
+
+def _is_disk_write(cmd: str, args: list[str]) -> bool:
+    if cmd == "mkfs" or cmd.startswith("mkfs."):
+        return True
+    if cmd == "dd":
+        return any(a.startswith("of=") for a in args)
+    return False
+
+
+def _classify(cmd: str, args: list[str]) -> str | None:
+    if _is_recursive_force_rm(cmd, args):
+        return "recursive force remove (rm -rf)"
+    if _is_reckless_chmod(cmd, args):
+        return "world-writable permissions (chmod 777)"
+    if _is_disk_write(cmd, args):
+        return "raw disk/filesystem write (dd of= / mkfs)"
+    if cmd == "sudo":
+        return "privilege escalation (sudo)"
+    if cmd in {"shutdown", "reboot", "halt", "poweroff"}:
+        return f"host power control ({cmd})"
+    return None
+
+
+def screen_command(command: str) -> str | None:
+    """Return a reason string if ``command`` looks destructive, else ``None``.
+
+    The raw string is first split on the shell control operators
+    (``;`` ``&&`` ``||`` ``|`` ``&`` and newlines) so that a destructive stage
+    hidden in a pipeline or sequence is still caught, e.g. ``echo hi && rm -rf /``
+    or ``cd /tmp; rm -rf x``. Each segment is then tokenised with
+    :func:`shlex.split` and classified on its argv. A segment that cannot be
+    tokenised (unbalanced quotes, etc.) is skipped rather than guessed at — this
+    layer never blocks what it cannot understand, and the sandbox remains the
+    boundary.
+    """
+    if not command or not command.strip():
+        return None
+
+    for raw_segment in _OPERATOR_SPLIT.split(command):
+        segment = raw_segment.strip()
+        if not segment:
+            continue
+        try:
+            tokens = shlex.split(segment, comments=False, posix=True)
+        except ValueError:
+            # Unbalanced quotes etc. — don't guess; defer to the sandbox.
+            continue
+        if not tokens:
+            continue
+        reason = _classify(tokens[0].lower(), tokens[1:])
+        if reason is not None:
+            return reason
+
+    return None

tests/test_command_guard.py

@@ -0,0 +1,85 @@
+"""Tests for the destructive-command screen (core.harness.command_guard).
+
+The screen is cheap defense-in-depth in front of the real sandbox boundary
+(see tests/test_harness_sandbox.py). These tests pin down two things:
+
+* it catches the destructive commands the old substring blocklist *claimed* to
+  catch but didn't (whitespace / split flags / flag spelling / numeric forms);
+* it does not fire on benign commands that merely *contain* a scary substring.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.harness.command_guard import screen_command
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "rm -rf /",
+        "rm -rf /tmp/x",
+        "rm  -rf /tmp/x",  # extra whitespace — old substring bypass
+        "RM -rf /tmp/x",  # uppercase command — old screen was case-insensitive
+        "rm -r -f /tmp/x",  # split flags — old substring bypass
+        "rm -fr /tmp/x",  # reversed flag order
+        "rm -Rf /tmp/x",  # capital R
+        "rm --recursive --force /tmp/x",  # long flags — old substring bypass
+        "chmod 777 file",
+        "chmod 0777 file",  # leading zero — old substring bypass
+        "chmod a+rwx file",  # symbolic form — old substring bypass
+        "sudo rm file",
+        "dd if=/dev/zero of=/dev/sda",
+        "mkfs /dev/sda1",
+        "mkfs.ext4 /dev/sda1",  # mkfs variant — old substring bypass
+        "reboot",
+        "shutdown -h now",
+        "echo hi && rm -rf /",  # destructive stage inside a sequence
+        "ls | rm -rf /",  # destructive stage inside a pipeline
+        "cd /tmp; rm -rf build; ls",  # destructive stage in the middle
+    ],
+)
+def test_blocks_destructive_commands(command):
+    assert screen_command(command) is not None
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "ls -la",
+        "rm file.txt",  # plain remove is allowed
+        "rm -r builddir",  # recursive but not forced
+        "rm -f stale.lock",  # forced but not recursive
+        "chmod 755 script.sh",
+        "chmod +x script.sh",
+        "touch rm-rf-notes.txt",  # scary substring, harmless command
+        'echo "run without sudo"',  # 'sudo' only inside a string literal
+        "dd if=input.bin count=1",  # dd reading only, no of=
+        "python train.py",
+        'git commit -m "drop the -rf flag from docs"',
+        "ls && echo done",  # benign sequence
+        "make && make install",  # benign sequence
+        "echo a | grep b | wc -l",  # benign pipeline
+    ],
+)
+def test_allows_benign_commands(command):
+    assert screen_command(command) is None
+
+
+def test_empty_command_is_allowed():
+    assert screen_command("") is None
+    assert screen_command("   ") is None
+
+
+def test_unparseable_command_defers_to_sandbox():
+    # Unbalanced quotes can't be tokenised; the screen must not raise and must
+    # pass the command through (None) so the sandbox stays the boundary.
+    assert screen_command('echo "unterminated') is None

tools/code_implementation_server.py

@@ -30,6 +30,7 @@
     subprocess_text_kwargs,
 )
 from core.harness.sandbox import build_exec_command, describe_backend, fences_writes
+from core.harness.command_guard import screen_command
 
 configure_utf8_stdio()
 
@@ -783,32 +784,38 @@ async def execute_bash(command: str, timeout: int = 30) -> str:
         JSON string of execution result
     """
     try:
-        # 安全检查：禁止危险命令
-        dangerous_commands = ["rm -rf", "sudo", "chmod 777", "mkfs", "dd if="]
-        # Normalize case and whitespace runs before matching: "RM  -rf" or
-        # "rm\t-rf" must not slip past a plain substring check. Shallow
-        # defense-in-depth only — the sandbox below is the real boundary.
-        normalized_command = re.sub(r"\s+", " ", command).lower()
-        if any(dangerous in normalized_command for dangerous in dangerous_commands):
+        # Cheap defense-in-depth: screen out obviously destructive commands
+        # before they reach the shell. This is NOT the security boundary — the
+        # workspace sandbox below is (see core.harness.sandbox).
+        #
+        # This started as a substring match ("rm -rf" in command.lower()) and
+        # grew a whitespace-normalization pass, but a normalized substring check
+        # still misses split/long flag spellings ("rm -r -f",
+        # "rm --recursive --force"), numeric permission variants ("chmod 0777")
+        # and falsely trips on benign commands that merely contain the text
+        # ("touch rm-rf-notes.txt"). screen_command tokenises the command and
+        # matches on argv instead, so it honestly catches what it claims to.
+        # Anything it cannot parse is passed through and left to the sandbox.
+        blocked_reason = screen_command(command)
+        if blocked_reason is not None:
             result = {
                 "status": "error",
-                "message": f"Dangerous command execution prohibited: {command}",
+                "message": f"Dangerous command prohibited ({blocked_reason}): {command}",
             }
             log_operation(
                 "execute_bash_blocked",
-                {"command": command, "reason": "dangerous_command"},
+                {"command": command, "reason": blocked_reason},
             )
             return json.dumps(result, ensure_ascii=False, indent=2)
 
         # Ensure workspace directory exists
         ensure_workspace_exists()
 
         # Execute inside the workspace write-fence sandbox (P1 security base).
-        # The pre-existing dangerous-command blocklist above stays as cheap
-        # defense-in-depth; the sandbox additionally prevents any write
-        # outside the workspace (e.g. tampering with the repo or $HOME) even
-        # if a command slips past the blocklist. Degrades to a bare run when
-        # no sandbox backend is available.
+        # The screen_command check above stays as cheap defense-in-depth; the
+        # sandbox additionally prevents any write outside the workspace (e.g.
+        # tampering with the repo or $HOME) even if a command slips past the
+        # screen. Degrades to a bare run when no sandbox backend is available.
         wrapped = build_exec_command(command=command, workspace=str(WORKSPACE_DIR))
         try:
             result = subprocess.run(