Bläddra i källkod

feat(workspace): P1 工作文件夹硬沙箱 — WriteFile/EditFile 越界写直接拒绝

诊断:工作区对话的工作目录虽一致(F),但 agent 实测把 5 次 WriteFile 全写到了
F 之外的另一个 run 目录(run_20260606/cycle_2),并跑偏成"续跑别人的研究"。
P0 的提示约束("就地在 F 操作")挡不住模型——需工具层硬拦。

实现:
- 新增 lambdagent/builtin_tools/_sandbox.py:模块级 root(+锁,同 shell _set_cwd)、
  check_write_allowed(realpath 比对、防 symlink/.. 逃逸)
- file_tools.write_file/edit_file:写前校验,越界返回 [SANDBOX_DENIED] 给模型纠正
- _execute_agent:工作区模式(inplace_dir 有值)设 root=F,finally 清空(防泄漏到
  下个 run);仅工作区模式启用,专项智能体 root=None 不受影响
- 只拦写,不拦读(读不破坏、常需交叉引用);Bash 靠 CWD=F + dangerousCommandBlock

效果:agent 再想写 F 之外会被当场拒绝并被要求改回 F 内,杜绝"写错目录+跑偏"。
回归 568(含 test_sandbox_blocks_writes_outside_root)+ e2e/run 41 全绿。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
kenny67nju 2 månader sedan
förälder
incheckning
3bae21d

+ 15 - 0
agentpaas/src/agentpaas/api/v1/agents.py

@@ -2046,6 +2046,15 @@ def _execute_agent(config: dict, input_text: str, on_step=None,
     _cwd_to_set = (
     _cwd_to_set = (
         work_dir if (work_dir and os.path.isdir(work_dir)) else workspace_path
         work_dir if (work_dir and os.path.isdir(work_dir)) else workspace_path
     )
     )
+    # P1 工作文件夹硬沙箱:工作区模式(inplace_dir 有值)下,把 WriteFile/EditFile
+    # 的写限定在所选文件夹 F 内,越界写直接拒绝。仅工作区模式启用——专项智能体
+    # 写各自 run 工作区,root=None 不受影响。finally(下方)清空,防泄漏到下个 run。
+    try:
+        from lambdagent.builtin_tools._sandbox import set_sandbox_root
+        set_sandbox_root(inplace_dir if (inplace_dir and os.path.isdir(inplace_dir)) else None)
+    except Exception as _sbe:
+        logger.warning("set sandbox root failed: %s", _sbe)
+
     if _cwd_to_set:
     if _cwd_to_set:
         try:
         try:
             from lambdagent.builtin_tools.shell_tools import _set_cwd as _shell_set_cwd
             from lambdagent.builtin_tools.shell_tools import _set_cwd as _shell_set_cwd
@@ -2221,6 +2230,12 @@ def _execute_agent(config: dict, input_text: str, on_step=None,
 
 
         return result, trace_info
         return result, trace_info
     finally:
     finally:
+        # P1 沙箱:清空写沙箱 root,防泄漏到下一个 run(模块级全局状态)。
+        try:
+            from lambdagent.builtin_tools._sandbox import set_sandbox_root
+            set_sandbox_root(None)
+        except Exception:
+            pass
         # 只删除 /tmp 临时文件,workspace 里的保留
         # 只删除 /tmp 临时文件,workspace 里的保留
         if not workspace_path:
         if not workspace_path:
             try:
             try:

+ 56 - 0
lambdagent/src/lambdagent/builtin_tools/_sandbox.py

@@ -0,0 +1,56 @@
+"""工作文件夹硬沙箱(P1)。
+
+工作区对话模式下,把「写」限定在用户选的工作文件夹 F 内:WriteFile / EditFile
+的目标路径若落在 F 之外,直接拒绝并把原因返回给模型,让它改到 F 内。
+
+为什么需要:仅靠 systemPrompt 提示"就地在 F 操作"挡不住模型——实测过工作区助手
+把产物写到了 F 之外的另一个 run 目录。工具层硬拦是唯一可靠手段。
+
+作用范围:只拦【写】(write_file/edit_file)。读不拦(读不破坏,且常需交叉引用
+F 之外的文件)。Bash 是真正的 shell、无法在工具层可靠沙箱,靠 CWD=F + guard
+(dangerousCommandBlock) 兜。
+
+线程模型:与 shell_tools._session_cwd 一致——模块级 + 锁。单用户桌面 run 通常串行。
+未设 root(None)= 不限制(专项智能体写各自 run 目录,不受影响)。
+"""
+from __future__ import annotations
+
+import os
+import threading
+from typing import Optional, Tuple
+
+_root: Optional[str] = None
+_lock = threading.Lock()
+
+
+def set_sandbox_root(path: str | None) -> None:
+    """设置写沙箱根目录;None / 空 = 解除限制。"""
+    global _root
+    with _lock:
+        _root = os.path.realpath(path) if path else None
+
+
+def get_sandbox_root() -> Optional[str]:
+    with _lock:
+        return _root
+
+
+def check_write_allowed(target_path: str) -> Tuple[bool, str]:
+    """目标路径是否允许写。未设 root → 总允许;设了 → 必须落在 root 内。
+
+    返回 (ok, error_msg)。realpath 解析符号链接,防 symlink 逃逸。
+    """
+    root = get_sandbox_root()
+    if not root:
+        return True, ""
+    try:
+        rp = os.path.realpath(target_path)
+    except Exception:
+        rp = os.path.abspath(target_path)
+    if rp == root or rp.startswith(root + os.sep):
+        return True, ""
+    return False, (
+        f"[SANDBOX_DENIED] 拒绝写到工作文件夹之外:{target_path}\n"
+        f"本次会话只允许在工作文件夹内读写:{root}\n"
+        f"请把目标路径改到该文件夹内(用其下的绝对路径)后重试。"
+    )

+ 12 - 0
lambdagent/src/lambdagent/builtin_tools/file_tools.py

@@ -13,6 +13,8 @@ import json
 import os
 import os
 import re
 import re
 import subprocess
 import subprocess
+
+from lambdagent.builtin_tools._sandbox import check_write_allowed
 from typing import Any, Dict, Optional
 from typing import Any, Dict, Optional
 
 
 
 
@@ -190,6 +192,11 @@ def edit_file(input_val: Any) -> str:
     new = params["new_string"]
     new = params["new_string"]
     replace_all = params.get("replace_all", False)
     replace_all = params.get("replace_all", False)
 
 
+    # P1 工作文件夹硬沙箱:工作区模式下禁止改写所选文件夹之外的文件。
+    ok, err = check_write_allowed(path)
+    if not ok:
+        return err
+
     if not os.path.exists(path):
     if not os.path.exists(path):
         return f"[ERROR] File not found: {path}"
         return f"[ERROR] File not found: {path}"
 
 
@@ -246,6 +253,11 @@ def write_file(input_val: Any) -> str:
     path = params["file_path"]
     path = params["file_path"]
     content = params["content"]
     content = params["content"]
 
 
+    # P1 工作文件夹硬沙箱:工作区模式下禁止写到所选文件夹之外。
+    ok, err = check_write_allowed(path)
+    if not ok:
+        return err
+
     os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
     os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
 
 
     existed = os.path.exists(path)
     existed = os.path.exists(path)

+ 26 - 0
lambdagent/tests/test_builtin_tools.py

@@ -675,3 +675,29 @@ class TestProjectConfig:
 
 
 if __name__ == "__main__":
 if __name__ == "__main__":
     pytest.main([__file__, "-v", "--tb=short"])
     pytest.main([__file__, "-v", "--tb=short"])
+
+
+def test_sandbox_blocks_writes_outside_root(tmp_path):
+    """P1 工作文件夹硬沙箱:设了 root 后,WriteFile/EditFile 只能写 root 内。"""
+    from lambdagent.builtin_tools._sandbox import set_sandbox_root
+    from lambdagent.builtin_tools.file_tools import write_file
+    inside = tmp_path / "F"; inside.mkdir()
+    outside = tmp_path / "OTHER"; outside.mkdir()
+    try:
+        set_sandbox_root(str(inside))
+        # 根内 → 允许
+        r_ok = write_file({"file_path": str(inside / "a.md"), "content": "hi"})
+        assert "[OK]" in r_ok, r_ok
+        assert (inside / "a.md").exists()
+        # 根外 → 拒绝,且文件不被创建
+        r_deny = write_file({"file_path": str(outside / "evil.md"), "content": "x"})
+        assert "SANDBOX_DENIED" in r_deny, r_deny
+        assert not (outside / "evil.md").exists()
+        # 用 .. 逃逸也拒
+        r_esc = write_file({"file_path": str(inside / ".." / "OTHER" / "esc.md"), "content": "x"})
+        assert "SANDBOX_DENIED" in r_esc, r_esc
+    finally:
+        set_sandbox_root(None)
+    # 解除后:任意路径都允许(专项智能体不受影响)
+    r_free = write_file({"file_path": str(outside / "free.md"), "content": "ok"})
+    assert "[OK]" in r_free