fix(security): keep file paths inside the working directory

ga.py

@@ -293,7 +293,15 @@ def _get_tool_maxlen(self, l, args, growth_rate=1.0):
         return int(l * multiplier / args.get('_tool_num', 1))
     def _get_abs_path(self, path):
         if not path: return ""
-        return os.path.abspath(os.path.join(self.cwd, path))   
+        root = os.path.realpath(self.cwd)
+        resolved = os.path.realpath(os.path.join(root, path))
+        try:
+            inside_root = os.path.commonpath((root, resolved)) == root
+        except ValueError:
+            inside_root = False
+        if not inside_root:
+            raise ValueError(f"Path must stay inside working directory: {path}")
+        return resolved
 
     def _extract_code_block(self, response, code_type):
         code_type = {'python':'python|py', 'powershell':'powershell|ps1|pwsh', 'bash':'bash|sh|shell'}.get(code_type, re.escape(code_type))
@@ -353,21 +361,27 @@ def do_web_execute_js(self, args, response):
         '''web情况下的优先使用工具，执行任何js达成对浏览器的*完全*控制。支持将结果保存到文件供后续读取分析。'''
         script = args.get("script", "") or self._extract_code_block(response, "javascript")
         if not script: return StepOutcome("[Error] Script missing. Use ```javascript block or 'script' arg.", next_prompt="\n")
-        abs_path = self._get_abs_path(script.strip())
-        if os.path.isfile(abs_path):
+        script_path = os.path.join(self.cwd, script.strip())
+        if os.path.isfile(script_path):
+            try:
+                abs_path = self._get_abs_path(script.strip())
+            except ValueError as e:
+                return StepOutcome({"status": "error", "msg": str(e)}, next_prompt="\n")
             with open(abs_path, 'r', encoding='utf-8') as f: script = f.read()
         save_to_file = args.get("save_to_file", "")
         switch_tab_id = args.get("switch_tab_id") or args.get("tab_id")
         no_monitor = _arg(args, "no_monitor", False, bool)
         result = web_execute_js(script, switch_tab_id=switch_tab_id, no_monitor=no_monitor)
         if save_to_file and "js_return" in result:
             content = str(result["js_return"] or '')
-            abs_path = self._get_abs_path(save_to_file)
-            result["js_return"] = smart_format(content, max_str_len=170)
             try:
+                abs_path = self._get_abs_path(save_to_file)
+                result["js_return"] = smart_format(content, max_str_len=170)
                 with open(abs_path, 'w', encoding='utf-8') as f: f.write(str(content))
                 result["js_return"] += f"\n\n[已保存完整内容到 {abs_path}]"
-            except: result['js_return'] += f"\n\n[保存失败，无法写入文件 {abs_path}]"
+            except Exception as e:
+                result["js_return"] = smart_format(content, max_str_len=170)
+                result['js_return'] += f"\n\n[保存失败: {e}]"
         show = smart_format(json.dumps(result, ensure_ascii=False, indent=2, default=json_default), max_str_len=300)
         self.print("Web Execute JS Result:", show)
         yield f"JS 执行结果:\n{show}\n"
@@ -377,7 +391,11 @@ def do_web_execute_js(self, args, response):
         return StepOutcome(smart_format(result, max_str_len=maxlen), next_prompt=next_prompt)
 
     def do_file_patch(self, args, response):
-        path = self._get_abs_path(args.get("path", ""))
+        try:
+            path = self._get_abs_path(args.get("path", ""))
+        except ValueError as e:
+            yield f"[Status] ❌ 路径拒绝: {e}\n"
+            return StepOutcome({"status": "error", "msg": str(e)}, next_prompt="\n")
         yield f"[Action] Patching file: {path}\n"
         old_content = args.get("old_content", "")
         new_content = args.get("new_content", "")
@@ -393,7 +411,11 @@ def do_file_patch(self, args, response):
     def do_file_write(self, args, response):
         '''用于对整个文件的大量处理，精细修改要用file_patch。
         需要将要写入的内容放在<file_content>标签内，或者放在代码块中'''
-        path = self._get_abs_path(args.get("path", ""))
+        try:
+            path = self._get_abs_path(args.get("path", ""))
+        except ValueError as e:
+            yield f"[Status] ❌ 路径拒绝: {e}\n"
+            return StepOutcome({"status": "error", "msg": str(e)}, next_prompt="\n")
         mode = args.get("mode", "overwrite")  # overwrite/append/prepend
         action_str = {"prepend": "Prepending to", "append": "Appending to"}.get(mode, "Overwriting")
         yield f"[Action] {action_str} file: {os.path.basename(path)}\n"
@@ -426,7 +448,11 @@ def extract_robust_content(text):
 
     def do_file_read(self, args, response):
         '''读取文件内容。从第start行开始读取。如有keyword则返回第一个keyword(忽略大小写)周边内容'''
-        path = self._get_abs_path(args.get("path", ""))
+        try:
+            path = self._get_abs_path(args.get("path", ""))
+        except ValueError as e:
+            yield f"[Status] ❌ 路径拒绝: {e}\n"
+            return StepOutcome({"status": "error", "msg": str(e)}, next_prompt="\n")
         yield f"\n[Action] Reading file: {path}\n"
         start = _arg(args, "start", 1, int)
         count = _arg(args, "count", 200, int)

tests/test_path_jail.py

@@ -0,0 +1,58 @@
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+from ga import GenericAgentHandler
+
+
+class TestPathJail(unittest.TestCase):
+    def make_handler(self, cwd: Path) -> GenericAgentHandler:
+        handler = GenericAgentHandler.__new__(GenericAgentHandler)
+        handler.cwd = str(cwd)
+        return handler
+
+    def test_relative_and_absolute_paths_inside_cwd(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            handler = self.make_handler(root)
+            self.assertEqual(handler._get_abs_path("notes.txt"), str(root / "notes.txt"))
+            self.assertEqual(handler._get_abs_path(str(root / "notes.txt")), str(root / "notes.txt"))
+
+    def test_parent_path_is_rejected(self):
+        with tempfile.TemporaryDirectory() as directory:
+            handler = self.make_handler(Path(directory) / "temp")
+            with self.assertRaisesRegex(ValueError, "inside working directory"):
+                handler._get_abs_path("../assets/code_run_header.py")
+
+    def test_symlink_escape_is_rejected(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "temp"
+            root.mkdir()
+            outside = Path(directory) / "outside"
+            outside.mkdir()
+            link = root / "link"
+            try:
+                link.symlink_to(outside, target_is_directory=True)
+            except (OSError, NotImplementedError):
+                self.skipTest("symlinks are unavailable on this platform")
+
+            handler = self.make_handler(root)
+            with self.assertRaisesRegex(ValueError, "inside working directory"):
+                handler._get_abs_path("link/poisoned.py")
+
+    def test_file_write_reports_rejected_path(self):
+        with tempfile.TemporaryDirectory() as directory:
+            handler = self.make_handler(Path(directory) / "temp")
+            response = SimpleNamespace(content="<file_content>poison</file_content>")
+            operation = handler.do_file_write({"path": "../assets/code_run_header.py"}, response)
+
+            first_message = next(operation)
+            self.assertIn("路径拒绝", first_message)
+            with self.assertRaises(StopIteration) as stopped:
+                next(operation)
+            self.assertEqual(stopped.exception.value.data["status"], "error")
+
+
+if __name__ == "__main__":
+    unittest.main()