From a561e7396793866f79d299816ef3d14ea2c0ece7 Mon Sep 17 00:00:00 2001 From: shauryagangrade <288927048+shauryagangrade@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:15:43 +0530 Subject: [PATCH] fix: enforce workspace boundary for file tools: lint + non-tty fixes, workspace-aware tests --- gcode/tools.py | 54 +++++++++++++++++ tests/test_tools.py | 137 +++++++++++++++++++++++++++++++++++++------- 2 files changed, 170 insertions(+), 21 deletions(-) diff --git a/gcode/tools.py b/gcode/tools.py index e919f37..f0a90fb 100644 --- a/gcode/tools.py +++ b/gcode/tools.py @@ -10,6 +10,8 @@ import re import shutil import subprocess +import sys +from pathlib import Path from langchain.tools import tool @@ -40,6 +42,46 @@ def set_bash_timeout(value: int) -> None: BASH_TIMEOUT = value +def _workspace_root() -> Path: + """Return the current workspace root (session cwd).""" + return Path(os.getcwd()).resolve() + + +def _is_within_workspace(path: str) -> bool: + """Return True if *path* resolves inside the workspace root. + + Handles `..`, absolute paths, and symlinks via :meth:`Path.resolve`. + An empty path is treated as outside. + """ + if not path: + return False + try: + return Path(path).resolve().is_relative_to(_workspace_root()) + except (ValueError, OSError): + return False + + +def _check_workspace_boundary(path: str) -> str | None: + """Enforce workspace boundary for file tools. + + Returns an error string if the path is outside the workspace and the + user does not confirm, or ``None`` if the operation may proceed. + In non-interactive mode (no tty) the operation is rejected unless + auto-approve is enabled, mirroring :func:`execute_bash`. + """ + if _is_within_workspace(path) or AUTO_APPROVE: + return None + root = _workspace_root() + refuse = f"Refusing to access '{path}': outside workspace '{root}'." + if not sys.stdin.isatty(): + return refuse + " No terminal available; run with --yes to allow." + try: + ans = input(f"Path '{path}' is outside workspace '{root}'. Allow? (y/n): ") + except (EOFError, KeyboardInterrupt, OSError): + return refuse + " No terminal available; run with --yes to allow." + return None if ans.strip().lower() in ("y", "yes") else refuse + + @tool def execute_bash(command: str) -> str: """Execute a bash command on the local machine and return its output. @@ -90,6 +132,9 @@ def read_file(path: str, max_lines: int = 2000) -> str: max_lines: Maximum number of lines to return (default 2000); longer files are truncated with a note. """ + err = _check_workspace_boundary(path) + if err is not None: + return err if not os.path.isfile(path): return f"File not found: {path}" try: @@ -120,6 +165,9 @@ def write_file(path: str, content: str, force: bool = False) -> str: content: Full text content to write. force: If True, overwrite an existing file. """ + err = _check_workspace_boundary(path) + if err is not None: + return err if os.path.exists(path) and not force: return f"Refusing to overwrite existing file {path} (pass force=True to overwrite)." try: @@ -146,6 +194,9 @@ def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = F new_string: Replacement text. replace_all: If True, replace every occurrence. """ + err = _check_workspace_boundary(path) + if err is not None: + return err if not os.path.isfile(path): return f"File not found: {path}" try: @@ -183,6 +234,9 @@ def list_dir(path: str = ".") -> str: Args: path: Directory to list (default current directory). """ + err = _check_workspace_boundary(path) + if err is not None: + return err if not os.path.isdir(path): return f"Not a directory: {path}" try: diff --git a/tests/test_tools.py b/tests/test_tools.py index 739ad43..7d491f9 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -1,40 +1,39 @@ import os +import sys from unittest.mock import patch -from gcode.tools import _grep_python, edit_file, grep, list_dir +import pytest +from gcode.tools import _grep_python, edit_file, grep, list_dir, read_file, write_file -def test_edit_file_unique(): - d = "/tmp/gcode_test_edit_unique" - os.makedirs(d, exist_ok=True) - p = os.path.join(d, "f.txt") - with open(p, "w") as f: - f.write("hello world\n") - out = edit_file.invoke({"path": p, "old_string": "world", "new_string": "there"}) +def test_edit_file_unique(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + p = tmp_path / "f.txt" + p.write_text("hello world\n") + out = edit_file.invoke({"path": str(p), "old_string": "world", "new_string": "there"}) assert "Edited" in out - with open(p) as f: - assert f.read() == "hello there\n" + assert p.read_text() == "hello there\n" -def test_edit_file_ambiguous(): - d = "/tmp/gcode_test_edit_ambiguous" - os.makedirs(d, exist_ok=True) - p = os.path.join(d, "f.txt") - with open(p, "w") as f: - f.write("a a a\n") - out = edit_file.invoke({"path": p, "old_string": "a", "new_string": "b"}) +def test_edit_file_ambiguous(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + p = tmp_path / "f.txt" + p.write_text("a a a\n") + out = edit_file.invoke({"path": str(p), "old_string": "a", "new_string": "b"}) assert "found 3 times" in out -def test_edit_file_not_found(): - out = edit_file.invoke({"path": "/no/such/file.txt", "old_string": "x", "new_string": "y"}) +def test_edit_file_not_found(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + out = edit_file.invoke({"path": "no_such_file.txt", "old_string": "x", "new_string": "y"}) assert "File not found" in out -def test_list_dir(tmp_path): +def test_list_dir(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) (tmp_path / "a.txt").write_text("x") (tmp_path / "sub").mkdir() - out = list_dir.invoke({"path": str(tmp_path)}) + out = list_dir.invoke({"path": "."}) assert "a.txt" in out assert "sub/" in out @@ -245,3 +244,99 @@ def test_grep_filters_by_glob(tmp_path): assert "needle in python" in out assert "needle in text" not in out + + +# -- workspace boundary ------------------------------------------------------- + + +class _FakeStdIn: + """Stand-in for sys.stdin to control tty detection.""" + + def __init__(self, interactive: bool): + self._interactive = interactive + + def isatty(self) -> bool: + return self._interactive + + +def _outside_file(tmp_path): + """A file that resolves outside the workspace (cwd == tmp_path).""" + outside = tmp_path.parent / "outside.txt" + outside.write_text("secret content\n") + return outside + + +def test_file_tools_reject_outside_workspace_non_tty(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=False)) + + out = read_file.invoke({"path": str(_outside_file(tmp_path))}) + assert "Refusing to access" in out + assert "outside workspace" in out + + +def test_file_tools_reject_outside_workspace_interactive(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=True)) + + with patch("builtins.input", return_value="n"): + out = read_file.invoke({"path": str(_outside_file(tmp_path))}) + assert "Refusing to access" in out + + +def test_file_tools_allow_confirmed_outside_workspace(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=True)) + + with patch("builtins.input", return_value="y"): + out = read_file.invoke({"path": str(_outside_file(tmp_path))}) + assert "secret content" in out + + +def test_write_file_outside_workspace_rejected_non_tty(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=False)) + + outside = tmp_path.parent / "outside_write.txt" + out = write_file.invoke({"path": str(outside), "content": "boom"}) + assert "Refusing to access" in out + assert not outside.exists() + + +def test_auto_approve_allows_outside_workspace(tmp_path, monkeypatch): + from gcode.tools import AUTO_APPROVE, set_auto_approve + + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=False)) + + set_auto_approve(True) + try: + out = read_file.invoke({"path": str(_outside_file(tmp_path))}) + finally: + set_auto_approve(AUTO_APPROVE) + assert "secret content" in out + + +def test_workspace_boundary_resolves_inside_and_dotdot(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=False)) + + (tmp_path / "inside.txt").write_text("ok\n") + assert "ok" in read_file.invoke({"path": "inside.txt"}) + + out = read_file.invoke({"path": "../escaped.txt"}) + assert "Refusing to access" in out + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks not supported") +def test_workspace_boundary_blocks_symlink_escape(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "stdin", _FakeStdIn(interactive=False)) + + target = tmp_path.parent / "target.txt" + target.write_text("secret\n") + link = tmp_path / "link.txt" + link.symlink_to(target) + + out = read_file.invoke({"path": str(link)}) + assert "Refusing to access" in out