From df0bb4c2ad5abf6ebeee7340158f780e2cdda583 Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Mon, 31 Aug 2026 07:16:51 +0800
Subject: [PATCH] fix(agent): avoid compound command safety false positives

---
 app/agent/tools/impl/_command_safety.py | 63 +++++++++++++++++--------
 tests/test_command_safety.py            | 15 ++++++
 2 files changed, 58 insertions(+), 20 deletions(-)
 create mode 100644 tests/test_command_safety.py

diff --git a/app/agent/tools/impl/_command_safety.py b/app/agent/tools/impl/_command_safety.py
index 45e6cfc24..e4b2c140c 100644
--- a/app/agent/tools/impl/_command_safety.py
+++ b/app/agent/tools/impl/_command_safety.py
@@ -7,6 +7,9 @@ import re
 import shlex
 
 
+_SHELL_PUNCTUATION = ";&|<>"
+
+
 COMMAND_FORBIDDEN_KEYWORDS = (
     ":(){ :|:& };:",
     "dd if=/dev/zero",
@@ -27,33 +30,53 @@ COMMAND_DANGEROUS_PATTERNS = (
 def _command_tokens(command: str) -> list[str]:
     """尽力解析 shell 命令 token，解析失败时退回空白分割。"""
     try:
-        return shlex.split(command, posix=True)
+        lexer = shlex.shlex(command, posix=True, punctuation_chars=_SHELL_PUNCTUATION)
+        lexer.whitespace_split = True
+        return list(lexer)
     except ValueError:
         return re.split(r"\s+", command.strip())
 
 
+def _command_segments(command: str) -> list[list[str]]:
+    """按 shell 控制运算符拆分命令，避免跨命令误关联参数。"""
+    segments: list[list[str]] = []
+    current: list[str] = []
+    for token in _command_tokens(command):
+        if token and all(char in _SHELL_PUNCTUATION for char in token):
+            if current:
+                segments.append(current)
+                current = []
+            continue
+        current.append(token)
+    if current:
+        segments.append(current)
+    return segments
+
+
 def _contains_recursive_root_delete(command: str) -> bool:
     """识别递归删除根目录或一级目录的 rm 命令。"""
-    tokens = _command_tokens(command)
-    if not any(token == "rm" or token.endswith("/rm") for token in tokens):
-        return False
-    has_recursive = any(
-        token.startswith("-") and ("r" in token or "R" in token)
-        for token in tokens
-    )
-    if not has_recursive:
-        return False
+    for tokens in _command_segments(command):
+        rm_indexes = [
+            index
+            for index, token in enumerate(tokens)
+            if token == "rm" or token.endswith("/rm")
+        ]
+        for rm_index in rm_indexes:
+            rm_arguments = tokens[rm_index + 1 :]
+            has_recursive = any(
+                token.startswith("-") and ("r" in token or "R" in token)
+                for token in rm_arguments
+            )
+            if not has_recursive:
+                continue
 
-    for token in tokens:
-        clean_token = re.match(r"^([^;|&><]+)", token)
-        if not clean_token:
-            continue
-        path_value = clean_token.group(1).strip("\"'")
-        if not path_value.startswith("/"):
-            continue
-        norm_path = os.path.normpath(path_value)
-        if norm_path == "/" or re.match(r"^/[^/]+$", norm_path):
-            return True
+            for token in rm_arguments:
+                path_value = token.strip("\"'")
+                if not path_value.startswith("/"):
+                    continue
+                norm_path = os.path.normpath(path_value)
+                if norm_path == "/" or re.match(r"^/[^/]+$", norm_path):
+                    return True
     return False
 
 
diff --git a/tests/test_command_safety.py b/tests/test_command_safety.py
new file mode 100644
index 000000000..b4063a155
--- /dev/null
+++ b/tests/test_command_safety.py
@@ -0,0 +1,15 @@
+"""Agent 命令安全校验测试。"""
+
+from app.agent.tools.impl._command_safety import detect_dangerous_command
+
+
+def test_compound_command_does_not_associate_jq_recursive_flag_with_rm():
+    """复合命令中其他子命令的 -r 不应触发 rm 递归删除误报。"""
+    command = "cd /tmp && rm -f qbcookiejar; printf '{}' | jq -r ."
+
+    assert detect_dangerous_command(command) == ""
+
+
+def test_recursive_rm_of_root_level_directory_remains_blocked():
+    """递归删除根目录下一级目录仍需显式确认。"""
+    assert detect_dangerous_command("cd /tmp && rm -rf /tmp")