From 9e7a9c79d27dd231a643fa4cdc67ed746b3965d0 Mon Sep 17 00:00:00 2001 From: Solaris-star <820622658@qq.com> Date: Wed, 5 Aug 2026 08:28:25 +0800 Subject: [PATCH] fix(security): keep file paths inside the working directory --- ga.py | 44 ++++++++++++++++++++++++------- tests/test_path_jail.py | 58 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 9 deletions(-) create mode 100644 tests/test_path_jail.py diff --git a/ga.py b/ga.py index c56db03f1..cfbb82221 100644 --- a/ga.py +++ b/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,8 +361,12 @@ 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") @@ -362,12 +374,14 @@ def do_web_execute_js(self, args, response): 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。 需要将要写入的内容放在标签内，或者放在代码块中''' - 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) diff --git a/tests/test_path_jail.py b/tests/test_path_jail.py new file mode 100644 index 000000000..5a6c3b5f0 --- /dev/null +++ b/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="poison") + 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()