Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 54 additions & 0 deletions gcode/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
import re
import shutil
import subprocess
import sys
from pathlib import Path

from langchain.tools import tool

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
137 changes: 116 additions & 21 deletions tests/test_tools.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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