diff --git a/core/harness/command_guard.py b/core/harness/command_guard.py new file mode 100644 index 00000000..17cc7e08 --- /dev/null +++ b/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 diff --git a/tests/test_command_guard.py b/tests/test_command_guard.py new file mode 100644 index 00000000..363f3c5c --- /dev/null +++ b/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 diff --git a/tools/code_implementation_server.py b/tools/code_implementation_server.py index 746d70d6..d16a9cd5 100644 --- a/tools/code_implementation_server.py +++ b/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,20 +784,27 @@ 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) @@ -804,11 +812,10 @@ async def execute_bash(command: str, timeout: int = 30) -> str: 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(