Refactor movie pilot config and test coverage

app/agent/tools/impl/_command_safety.py

@@ -0,0 +1,88 @@
+"""Agent 命令工具的安全校验逻辑。"""
+
+from __future__ import annotations
+
+import os.path
+import re
+import shlex
+
+
+COMMAND_FORBIDDEN_KEYWORDS = (
+    ":(){ :|:& };:",
+    "dd if=/dev/zero",
+    "mkfs",
+    "reboot",
+    "shutdown",
+)
+
+COMMAND_DANGEROUS_PATTERNS = (
+    re.compile(r"\brm\s+[^;&|]*-[^\s;&|]*[rR][fF]?[^\s;&|]*\s+/(?:\s|$|[;&|])"),
+    re.compile(r"\bdd\s+[^;&|]*(?:of=/dev/(?:sd[a-z]\d*|nvme\d+n\d+p?\d*|disk\d+)|if=/dev/zero)"),
+    re.compile(r"\b(?:mkfs|fdisk|parted|diskutil)\b"),
+    re.compile(r"\b(?:chmod|chown)\s+[^;&|]*-R[^;&|]*\s+/(?:\s|$|[;&|])"),
+    re.compile(r"\b(?:reboot|shutdown|halt|poweroff)\b"),
+)
+
+
+def _command_tokens(command: str) -> list[str]:
+    """尽力解析 shell 命令 token，解析失败时退回空白分割。"""
+    try:
+        return shlex.split(command, posix=True)
+    except ValueError:
+        return re.split(r"\s+", command.strip())
+
+
+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 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
+    return False
+
+
+def detect_dangerous_command(command: str) -> str:
+    """返回危险命令原因，安全时返回空字符串。"""
+    normalized = str(command or "").strip()
+    if not normalized:
+        return "命令不能为空"
+    for keyword in COMMAND_FORBIDDEN_KEYWORDS:
+        if keyword in normalized:
+            return f"命令包含禁止使用的关键字 '{keyword}'"
+    if _contains_recursive_root_delete(normalized):
+        return "命令疑似递归删除根目录或一级目录"
+    for pattern in COMMAND_DANGEROUS_PATTERNS:
+        if pattern.search(normalized):
+            return "命令匹配高危系统操作模式"
+    return ""
+
+
+def validate_command_safety(command: str, *, confirmed: bool = False) -> None:
+    """
+    校验 shell 命令安全性。
+
+    :param command: 待执行命令
+    :param confirmed: 是否已经通过显式参数确认高危操作
+    """
+    reason = detect_dangerous_command(command)
+    if not reason:
+        return
+    if confirmed and reason != "命令不能为空":
+        return
+    raise ValueError(f"{reason}。如确认需要执行，请设置 confirm_dangerous=true")