feat: add coding tools (edit_file, write_file, bash)
New tools for code modification: - EditFileTool: find-and-replace with safety checks (unique match required) - WriteFileTool: create/overwrite files with path validation - BashTool: full bash with controlled write access Security controls on BashTool: - Allowed: mkdir, touch, cp, mv, rm (single files), git, pip, pytest - Forbidden: sudo, curl, wget, ssh, rm -rf, chmod 777 Includes 39 new tests (78 total now passing). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.5
@@ -8,8 +8,11 @@ from pydantic_ai import Agent, RunContext
|
||||
from src.domains.agents.base import AgentContext
|
||||
from src.domains.tools.file.read import ReadFileTool
|
||||
from src.domains.tools.file.glob import GlobFilesTool
|
||||
from src.domains.tools.file.edit import EditFileTool
|
||||
from src.domains.tools.file.write import WriteFileTool
|
||||
from src.domains.tools.search.grep import GrepContentTool
|
||||
from src.domains.tools.shell.bash import BashReadOnlyTool
|
||||
from src.domains.tools.shell.bash_full import BashTool
|
||||
|
||||
|
||||
def register_explore_tools(agent: Agent[AgentContext, str]) -> None:
|
||||
@@ -161,3 +164,110 @@ def register_explore_tools(agent: Agent[AgentContext, str]) -> None:
|
||||
timeout=min(timeout, ctx.deps.timeout_seconds)
|
||||
)
|
||||
return result.to_string()
|
||||
|
||||
# === Write-capable tools ===
|
||||
|
||||
@agent.tool
|
||||
async def edit_file(
|
||||
ctx: RunContext[AgentContext],
|
||||
file_path: str,
|
||||
old_string: str,
|
||||
new_string: str,
|
||||
replace_all: bool = False
|
||||
) -> str:
|
||||
"""Make targeted edits to a file using find-and-replace.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file to edit
|
||||
old_string: The exact text to find and replace (must exist in file)
|
||||
new_string: The replacement text
|
||||
replace_all: If True, replace all occurrences. If False (default),
|
||||
old_string must be unique (appear exactly once).
|
||||
|
||||
Returns:
|
||||
Success message with diff preview, or error.
|
||||
|
||||
IMPORTANT:
|
||||
- old_string must exactly match file content (including whitespace)
|
||||
- By default, old_string must appear exactly once (for safety)
|
||||
- Always read the file first to verify exact content before editing
|
||||
"""
|
||||
tool = EditFileTool(allowed_paths=ctx.deps.allowed_paths)
|
||||
result = await tool.execute(
|
||||
file_path=file_path,
|
||||
old_string=old_string,
|
||||
new_string=new_string,
|
||||
replace_all=replace_all
|
||||
)
|
||||
return result.to_string()
|
||||
|
||||
@agent.tool
|
||||
async def write_file(
|
||||
ctx: RunContext[AgentContext],
|
||||
file_path: str,
|
||||
content: str
|
||||
) -> str:
|
||||
"""Create a new file or overwrite an existing file.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file to create/write
|
||||
content: The content to write to the file
|
||||
|
||||
Returns:
|
||||
Success message with file path and size.
|
||||
|
||||
IMPORTANT:
|
||||
- Parent directory must exist (use mkdir first if needed)
|
||||
- For editing existing files, prefer edit_file instead
|
||||
- Will overwrite existing files without confirmation
|
||||
"""
|
||||
tool = WriteFileTool(allowed_paths=ctx.deps.allowed_paths)
|
||||
result = await tool.execute(
|
||||
file_path=file_path,
|
||||
content=content
|
||||
)
|
||||
return result.to_string()
|
||||
|
||||
@agent.tool
|
||||
async def bash(
|
||||
ctx: RunContext[AgentContext],
|
||||
command: str,
|
||||
cwd: str | None = None,
|
||||
timeout: int = 60
|
||||
) -> str:
|
||||
"""Execute a bash command with write capabilities.
|
||||
|
||||
ALLOWED:
|
||||
- File operations: ls, find, mkdir, touch, cp, mv, rm (single files)
|
||||
- Git (full): git add, git commit, git checkout, git merge, git pull
|
||||
- Python: python, pip install, pytest, mypy, ruff
|
||||
- Text processing: grep, awk, sed, sort
|
||||
- Command chaining: && and || are allowed
|
||||
|
||||
FORBIDDEN:
|
||||
- sudo, su (privilege escalation)
|
||||
- Network: curl, wget, ssh, scp, rsync
|
||||
- Dangerous: rm -rf, chmod 777, dd, mkfs
|
||||
|
||||
Args:
|
||||
command: The bash command to execute
|
||||
cwd: Working directory (default: agent working directory)
|
||||
timeout: Timeout in seconds (default: 60)
|
||||
|
||||
Returns:
|
||||
Command output or error message.
|
||||
|
||||
Examples:
|
||||
- "mkdir -p src/utils" creates directory
|
||||
- "git add . && git commit -m 'fix: bug'" commits changes
|
||||
- "pytest tests/ -v" runs tests
|
||||
- "rm old_file.py" removes single file
|
||||
"""
|
||||
tool = BashTool(allowed_paths=ctx.deps.allowed_paths)
|
||||
working_dir = cwd or ctx.deps.working_dir
|
||||
result = await tool.execute(
|
||||
command=command,
|
||||
cwd=working_dir,
|
||||
timeout=min(timeout, ctx.deps.timeout_seconds)
|
||||
)
|
||||
return result.to_string()
|
||||
|
||||
@@ -4,15 +4,18 @@ Tool implementations for agent use.
|
||||
All tools inherit from BaseTool and return ToolResult.
|
||||
"""
|
||||
from src.domains.tools.base import BaseTool, ToolResult
|
||||
from src.domains.tools.file import ReadFileTool, GlobFilesTool
|
||||
from src.domains.tools.file import ReadFileTool, GlobFilesTool, EditFileTool, WriteFileTool
|
||||
from src.domains.tools.search import GrepContentTool
|
||||
from src.domains.tools.shell import BashReadOnlyTool
|
||||
from src.domains.tools.shell import BashReadOnlyTool, BashTool
|
||||
|
||||
__all__ = [
|
||||
"BaseTool",
|
||||
"ToolResult",
|
||||
"ReadFileTool",
|
||||
"GlobFilesTool",
|
||||
"EditFileTool",
|
||||
"WriteFileTool",
|
||||
"GrepContentTool",
|
||||
"BashReadOnlyTool",
|
||||
"BashTool",
|
||||
]
|
||||
|
||||
@@ -3,5 +3,7 @@ File operation tools.
|
||||
"""
|
||||
from src.domains.tools.file.read import ReadFileTool
|
||||
from src.domains.tools.file.glob import GlobFilesTool
|
||||
from src.domains.tools.file.edit import EditFileTool
|
||||
from src.domains.tools.file.write import WriteFileTool
|
||||
|
||||
__all__ = ["ReadFileTool", "GlobFilesTool"]
|
||||
__all__ = ["ReadFileTool", "GlobFilesTool", "EditFileTool", "WriteFileTool"]
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
"""
|
||||
File editing tool with find-and-replace functionality.
|
||||
"""
|
||||
import difflib
|
||||
import aiofiles
|
||||
from pathlib import Path
|
||||
|
||||
from src.domains.tools.base import BaseTool, ToolResult
|
||||
from src.shared.logging import logged, get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# Binary file extensions to skip
|
||||
BINARY_EXTENSIONS = {
|
||||
'.pyc', '.pyo', '.so', '.o', '.a', '.lib', '.dll', '.exe',
|
||||
'.bin', '.dat', '.db', '.sqlite', '.sqlite3',
|
||||
'.png', '.jpg', '.jpeg', '.gif', '.ico', '.bmp', '.webp',
|
||||
'.pdf', '.doc', '.docx', '.xls', '.xlsx',
|
||||
'.zip', '.tar', '.gz', '.bz2', '.7z', '.rar',
|
||||
'.mp3', '.mp4', '.avi', '.mov', '.wav',
|
||||
'.woff', '.woff2', '.ttf', '.eot',
|
||||
}
|
||||
|
||||
|
||||
class EditFileTool(BaseTool):
|
||||
"""
|
||||
Edit files using find-and-replace.
|
||||
|
||||
Safely modifies files by finding exact text matches and replacing them.
|
||||
Includes safety checks to prevent accidental edits.
|
||||
"""
|
||||
|
||||
name = "edit_file"
|
||||
description = """Make targeted edits to a file using find-and-replace.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file to edit
|
||||
old_string: The exact text to find and replace (must exist in file)
|
||||
new_string: The replacement text
|
||||
replace_all: If True, replace all occurrences. If False (default),
|
||||
old_string must be unique in the file (appear exactly once).
|
||||
|
||||
Returns:
|
||||
Success message with diff preview showing changes, or error.
|
||||
|
||||
IMPORTANT:
|
||||
- The old_string must exactly match text in the file (including whitespace/indentation)
|
||||
- By default, old_string must appear exactly once in the file (for safety)
|
||||
- Use replace_all=True only when you intentionally want to replace all occurrences
|
||||
- Always read the file first to verify exact content before editing
|
||||
- Cannot edit binary files
|
||||
|
||||
Examples:
|
||||
- Fix a bug: old_string="return x + y", new_string="return x * y"
|
||||
- Rename function: old_string="def old_name(", new_string="def new_name("
|
||||
- Add import: old_string="import os", new_string="import os\\nimport sys"
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
allowed_paths: list[str] | None = None,
|
||||
max_file_size: int = 1_000_000, # 1MB
|
||||
):
|
||||
"""
|
||||
Initialize EditFileTool.
|
||||
|
||||
Args:
|
||||
allowed_paths: List of allowed directory prefixes (empty = no restrictions)
|
||||
max_file_size: Maximum file size to edit in bytes
|
||||
"""
|
||||
self.allowed_paths = allowed_paths or []
|
||||
self.max_file_size = max_file_size
|
||||
|
||||
def _is_binary_file(self, path: Path) -> bool:
|
||||
"""Check if file is likely binary based on extension."""
|
||||
return path.suffix.lower() in BINARY_EXTENSIONS
|
||||
|
||||
def _generate_diff(
|
||||
self,
|
||||
original: str,
|
||||
modified: str,
|
||||
file_path: str
|
||||
) -> str:
|
||||
"""Generate a unified diff between original and modified content."""
|
||||
original_lines = original.splitlines(keepends=True)
|
||||
modified_lines = modified.splitlines(keepends=True)
|
||||
|
||||
diff = difflib.unified_diff(
|
||||
original_lines,
|
||||
modified_lines,
|
||||
fromfile=f"a/{Path(file_path).name}",
|
||||
tofile=f"b/{Path(file_path).name}",
|
||||
lineterm=""
|
||||
)
|
||||
return "".join(diff)
|
||||
|
||||
@logged()
|
||||
async def execute(
|
||||
self,
|
||||
file_path: str,
|
||||
old_string: str,
|
||||
new_string: str,
|
||||
replace_all: bool = False
|
||||
) -> ToolResult:
|
||||
"""
|
||||
Edit a file by replacing old_string with new_string.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file
|
||||
old_string: Text to find (must exist)
|
||||
new_string: Replacement text
|
||||
replace_all: Replace all occurrences (default: False)
|
||||
|
||||
Returns:
|
||||
ToolResult with diff preview or error
|
||||
"""
|
||||
path = Path(file_path)
|
||||
|
||||
# Validate path is allowed
|
||||
if not self._validate_path(path, self.allowed_paths):
|
||||
return self._error(f"Path not in allowed paths: {file_path}")
|
||||
|
||||
# Check file exists
|
||||
if not path.exists():
|
||||
return self._error(f"File not found: {file_path}")
|
||||
|
||||
if not path.is_file():
|
||||
return self._error(f"Not a file: {file_path}")
|
||||
|
||||
# Check for binary files
|
||||
if self._is_binary_file(path):
|
||||
return self._error(f"Cannot edit binary file: {file_path}")
|
||||
|
||||
# Check file size
|
||||
file_size = path.stat().st_size
|
||||
if file_size > self.max_file_size:
|
||||
return self._error(
|
||||
f"File too large ({file_size} bytes). Max: {self.max_file_size} bytes"
|
||||
)
|
||||
|
||||
# Validate inputs
|
||||
if not old_string:
|
||||
return self._error("old_string cannot be empty")
|
||||
|
||||
if old_string == new_string:
|
||||
return self._error("old_string and new_string are identical")
|
||||
|
||||
try:
|
||||
# Read file content
|
||||
async with aiofiles.open(path, 'r', encoding='utf-8', errors='replace') as f:
|
||||
content = await f.read()
|
||||
|
||||
# Check if old_string exists
|
||||
count = content.count(old_string)
|
||||
if count == 0:
|
||||
return self._error(
|
||||
f"old_string not found in file. "
|
||||
f"Make sure to match exact whitespace and indentation."
|
||||
)
|
||||
|
||||
# Check uniqueness if replace_all is False
|
||||
if not replace_all and count > 1:
|
||||
return self._error(
|
||||
f"old_string appears {count} times in file. "
|
||||
f"Use replace_all=True to replace all, or provide a more specific string."
|
||||
)
|
||||
|
||||
# Perform replacement
|
||||
if replace_all:
|
||||
modified = content.replace(old_string, new_string)
|
||||
else:
|
||||
modified = content.replace(old_string, new_string, 1)
|
||||
|
||||
# Generate diff for preview
|
||||
diff = self._generate_diff(content, modified, file_path)
|
||||
|
||||
# Write modified content
|
||||
async with aiofiles.open(path, 'w', encoding='utf-8') as f:
|
||||
await f.write(modified)
|
||||
|
||||
replacements = count if replace_all else 1
|
||||
return self._success(
|
||||
data=f"Successfully edited {file_path}\n\n{diff}",
|
||||
replacements=replacements,
|
||||
file_path=str(path.resolve())
|
||||
)
|
||||
|
||||
except PermissionError:
|
||||
return self._error(f"Permission denied: {file_path}")
|
||||
except UnicodeDecodeError as e:
|
||||
return self._error(f"Unable to decode file (binary?): {e}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error editing file: {file_path}")
|
||||
return self._error(f"Error editing file: {e}")
|
||||
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
File writing tool for creating and overwriting files.
|
||||
"""
|
||||
import aiofiles
|
||||
from pathlib import Path
|
||||
|
||||
from src.domains.tools.base import BaseTool, ToolResult
|
||||
from src.shared.logging import logged, get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class WriteFileTool(BaseTool):
|
||||
"""
|
||||
Create new files or overwrite existing files.
|
||||
|
||||
Validates paths are within allowed directories and enforces size limits.
|
||||
"""
|
||||
|
||||
name = "write_file"
|
||||
description = """Create a new file or overwrite an existing file.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file to create/write
|
||||
content: The content to write to the file
|
||||
|
||||
Returns:
|
||||
Success message with file path and size, or error.
|
||||
Includes a warning if overwriting an existing file.
|
||||
|
||||
IMPORTANT:
|
||||
- Use absolute paths only
|
||||
- Parent directory must exist (will not create directories)
|
||||
- Will overwrite existing files without confirmation
|
||||
- For targeted edits to existing files, use edit_file instead
|
||||
- Maximum content size: 1MB
|
||||
|
||||
Examples:
|
||||
- Create new module: file_path="/project/src/utils.py", content="def helper(): pass"
|
||||
- Create config: file_path="/project/config.json", content='{"key": "value"}'
|
||||
- Create test: file_path="/project/tests/test_new.py", content="def test_example(): assert True"
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
allowed_paths: list[str] | None = None,
|
||||
max_content_size: int = 1_000_000, # 1MB
|
||||
):
|
||||
"""
|
||||
Initialize WriteFileTool.
|
||||
|
||||
Args:
|
||||
allowed_paths: List of allowed directory prefixes (empty = no restrictions)
|
||||
max_content_size: Maximum content size in bytes
|
||||
"""
|
||||
self.allowed_paths = allowed_paths or []
|
||||
self.max_content_size = max_content_size
|
||||
|
||||
@logged()
|
||||
async def execute(
|
||||
self,
|
||||
file_path: str,
|
||||
content: str
|
||||
) -> ToolResult:
|
||||
"""
|
||||
Write content to a file.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to the file
|
||||
content: Content to write
|
||||
|
||||
Returns:
|
||||
ToolResult with success info or error
|
||||
"""
|
||||
path = Path(file_path).resolve()
|
||||
|
||||
# Validate path is allowed
|
||||
if not self._validate_path(path, self.allowed_paths):
|
||||
return self._error(f"Path not in allowed paths: {file_path}")
|
||||
|
||||
# Check content size
|
||||
content_bytes = len(content.encode('utf-8'))
|
||||
if content_bytes > self.max_content_size:
|
||||
return self._error(
|
||||
f"Content too large ({content_bytes} bytes). "
|
||||
f"Max: {self.max_content_size} bytes"
|
||||
)
|
||||
|
||||
# Check parent directory exists
|
||||
if not path.parent.exists():
|
||||
return self._error(
|
||||
f"Parent directory does not exist: {path.parent}. "
|
||||
f"Create it first with mkdir."
|
||||
)
|
||||
|
||||
if not path.parent.is_dir():
|
||||
return self._error(f"Parent path is not a directory: {path.parent}")
|
||||
|
||||
# Check if we're overwriting
|
||||
overwritten = path.exists() and path.is_file()
|
||||
|
||||
try:
|
||||
# Write the file
|
||||
async with aiofiles.open(path, 'w', encoding='utf-8') as f:
|
||||
await f.write(content)
|
||||
|
||||
# Count lines for metadata
|
||||
lines = content.count('\n') + (1 if content and not content.endswith('\n') else 0)
|
||||
|
||||
status = "Overwrote" if overwritten else "Created"
|
||||
return self._success(
|
||||
data=f"{status} {path} ({content_bytes} bytes, {lines} lines)",
|
||||
file_path=str(path),
|
||||
file_size=content_bytes,
|
||||
lines=lines,
|
||||
overwritten=overwritten
|
||||
)
|
||||
|
||||
except PermissionError:
|
||||
return self._error(f"Permission denied: {file_path}")
|
||||
except OSError as e:
|
||||
return self._error(f"OS error writing file: {e}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error writing file: {file_path}")
|
||||
return self._error(f"Error writing file: {e}")
|
||||
@@ -2,5 +2,6 @@
|
||||
Shell execution tools.
|
||||
"""
|
||||
from src.domains.tools.shell.bash import BashReadOnlyTool
|
||||
from src.domains.tools.shell.bash_full import BashTool
|
||||
|
||||
__all__ = ["BashReadOnlyTool"]
|
||||
__all__ = ["BashReadOnlyTool", "BashTool"]
|
||||
|
||||
@@ -0,0 +1,430 @@
|
||||
"""
|
||||
Full bash command execution tool with controlled write capabilities.
|
||||
|
||||
Allows more operations than BashReadOnlyTool but still with safety controls.
|
||||
"""
|
||||
import asyncio
|
||||
import shlex
|
||||
from pathlib import Path
|
||||
|
||||
from src.domains.tools.base import BaseTool, ToolResult
|
||||
from src.shared.logging import logged, get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# Commands allowed (includes write operations)
|
||||
ALLOWED_COMMANDS = {
|
||||
# File inspection (read-only)
|
||||
"ls", "find", "cat", "head", "tail", "wc", "file", "stat",
|
||||
"tree", "du", "df",
|
||||
# Text processing
|
||||
"grep", "awk", "sed", "sort", "uniq", "cut", "tr",
|
||||
# Git (full operations)
|
||||
"git",
|
||||
# System info
|
||||
"pwd", "whoami", "hostname", "uname", "date", "env", "printenv",
|
||||
"which", "type", "echo",
|
||||
# Archive operations
|
||||
"tar", "unzip", "zipinfo",
|
||||
# Write operations (controlled)
|
||||
"mkdir", "touch", "cp", "mv", "rm",
|
||||
"chmod",
|
||||
# Python ecosystem
|
||||
"python", "python3", "pip", "pip3", "pytest", "mypy", "ruff",
|
||||
# Other dev tools
|
||||
"make", "cargo", "npm", "node", "tsc",
|
||||
}
|
||||
|
||||
# Git subcommands (includes write operations)
|
||||
ALLOWED_GIT_SUBCOMMANDS = {
|
||||
# Read-only
|
||||
"status", "log", "diff", "show", "branch", "tag",
|
||||
"remote", "config", "ls-files", "ls-tree",
|
||||
"rev-parse", "describe", "shortlog", "blame",
|
||||
# Write operations
|
||||
"add", "commit", "checkout", "switch", "restore",
|
||||
"merge", "rebase", "cherry-pick",
|
||||
"stash", "pull", "fetch", "init",
|
||||
"reset", "clean", "rm", "mv",
|
||||
}
|
||||
|
||||
# Absolutely forbidden - too dangerous regardless of context
|
||||
ABSOLUTELY_FORBIDDEN = [
|
||||
# Catastrophic deletes
|
||||
"rm -rf /",
|
||||
"rm -rf ~",
|
||||
"rm -rf .",
|
||||
"rm -rf *",
|
||||
# Privilege escalation
|
||||
"sudo ",
|
||||
"su ",
|
||||
"doas ",
|
||||
# Dangerous permissions
|
||||
"chmod 777",
|
||||
"chmod -R 777",
|
||||
"chown ",
|
||||
"chgrp ",
|
||||
# Disk operations
|
||||
"dd if=",
|
||||
"mkfs",
|
||||
"fdisk",
|
||||
"parted",
|
||||
# System control
|
||||
"shutdown",
|
||||
"reboot",
|
||||
"poweroff",
|
||||
"init ",
|
||||
"systemctl",
|
||||
# Fork bomb pattern
|
||||
":(){ :|:& };:",
|
||||
]
|
||||
|
||||
# Network commands are forbidden
|
||||
NETWORK_FORBIDDEN = [
|
||||
"curl", "wget", "ssh", "scp", "rsync", "sftp", "ftp",
|
||||
"nc", "netcat", "telnet", "nmap", "ping",
|
||||
]
|
||||
|
||||
# Dangerous rm flags
|
||||
DANGEROUS_RM_FLAGS = {"-rf", "-fr", "-r -f", "-f -r", "--recursive --force"}
|
||||
|
||||
|
||||
class BashTool(BaseTool):
|
||||
"""
|
||||
Execute bash commands with controlled write capabilities.
|
||||
|
||||
More permissive than BashReadOnlyTool but still with safety controls.
|
||||
"""
|
||||
|
||||
name = "bash"
|
||||
description = """Execute a bash command with write capabilities.
|
||||
|
||||
ALLOWED commands:
|
||||
- File operations: ls, find, cat, head, tail, mkdir, touch, cp, mv
|
||||
- Git (full): git add, git commit, git checkout, git merge, git pull, etc.
|
||||
- Python: python, pip install, pytest, mypy, ruff
|
||||
- Text processing: grep, awk, sed, sort, uniq
|
||||
- System info: pwd, whoami, date, which
|
||||
|
||||
RESTRICTED:
|
||||
- rm: Single files only, no -rf flag, must be in allowed paths
|
||||
- mv/cp: Target must be in allowed paths
|
||||
- chmod: Only safe modes (no 777)
|
||||
|
||||
FORBIDDEN (always blocked):
|
||||
- sudo, su (privilege escalation)
|
||||
- Network: curl, wget, ssh, scp, rsync
|
||||
- Dangerous: chmod 777, rm -rf, dd, mkfs, shutdown
|
||||
|
||||
Args:
|
||||
command: The bash command to execute
|
||||
cwd: Working directory (default: current directory)
|
||||
timeout: Timeout in seconds (default: 60)
|
||||
|
||||
Returns:
|
||||
Command stdout on success, or error message.
|
||||
|
||||
Examples:
|
||||
- "mkdir -p src/new_module" - Create directory
|
||||
- "cp template.py src/new_file.py" - Copy file
|
||||
- "git add . && git commit -m 'feat: add feature'" - Git commit
|
||||
- "pip install requests" - Install package
|
||||
- "pytest tests/ -v" - Run tests
|
||||
- "rm src/old_file.py" - Remove single file
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
allowed_paths: list[str] | None = None,
|
||||
default_timeout: int = 60,
|
||||
max_output_size: int = 50000
|
||||
):
|
||||
"""
|
||||
Initialize BashTool.
|
||||
|
||||
Args:
|
||||
allowed_paths: Allowed working/target directories
|
||||
default_timeout: Default command timeout in seconds
|
||||
max_output_size: Maximum output size in characters
|
||||
"""
|
||||
self.allowed_paths = allowed_paths or []
|
||||
self.default_timeout = default_timeout
|
||||
self.max_output_size = max_output_size
|
||||
|
||||
@logged()
|
||||
async def execute(
|
||||
self,
|
||||
command: str,
|
||||
cwd: str | None = None,
|
||||
timeout: int | None = None
|
||||
) -> ToolResult:
|
||||
"""
|
||||
Execute a bash command.
|
||||
|
||||
Args:
|
||||
command: Command to execute
|
||||
cwd: Working directory
|
||||
timeout: Timeout in seconds
|
||||
|
||||
Returns:
|
||||
ToolResult with command output or error
|
||||
"""
|
||||
timeout = timeout or self.default_timeout
|
||||
working_dir = Path(cwd) if cwd else Path.cwd()
|
||||
|
||||
# Validate working directory
|
||||
if not self._validate_path(working_dir, self.allowed_paths):
|
||||
return self._error(f"Working directory not allowed: {working_dir}")
|
||||
|
||||
if not working_dir.exists():
|
||||
return self._error(f"Working directory not found: {working_dir}")
|
||||
|
||||
# Security validation
|
||||
validation_error = self._validate_command(command, working_dir)
|
||||
if validation_error:
|
||||
return self._error(validation_error)
|
||||
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_shell(
|
||||
command,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
cwd=str(working_dir)
|
||||
)
|
||||
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(),
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
stdout_str = stdout.decode('utf-8', errors='replace')
|
||||
stderr_str = stderr.decode('utf-8', errors='replace')
|
||||
|
||||
# Truncate if necessary
|
||||
truncated = False
|
||||
if len(stdout_str) > self.max_output_size:
|
||||
stdout_str = stdout_str[:self.max_output_size]
|
||||
truncated = True
|
||||
|
||||
if proc.returncode != 0:
|
||||
# Command failed, return stderr
|
||||
error_msg = stderr_str or f"Command exited with code {proc.returncode}"
|
||||
return ToolResult(
|
||||
success=False,
|
||||
data=stdout_str if stdout_str else None,
|
||||
error=error_msg,
|
||||
truncated=truncated
|
||||
)
|
||||
|
||||
# Success - combine stdout and stderr if both present
|
||||
output = stdout_str
|
||||
if stderr_str and not output:
|
||||
output = stderr_str
|
||||
|
||||
return self._success(
|
||||
data=output,
|
||||
truncated=truncated,
|
||||
exit_code=proc.returncode
|
||||
)
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
return self._error(f"Command timed out after {timeout} seconds")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error executing command: {command}")
|
||||
return self._error(f"Error executing command: {e}")
|
||||
|
||||
def _validate_command(self, command: str, working_dir: Path) -> str | None:
|
||||
"""
|
||||
Validate command is safe to execute.
|
||||
|
||||
Returns:
|
||||
Error message if invalid, None if valid
|
||||
"""
|
||||
command_lower = command.lower()
|
||||
|
||||
# Check absolutely forbidden patterns first
|
||||
for pattern in ABSOLUTELY_FORBIDDEN:
|
||||
if pattern.lower() in command_lower:
|
||||
return f"Command contains forbidden pattern: {pattern}"
|
||||
|
||||
# Check network commands
|
||||
for net_cmd in NETWORK_FORBIDDEN:
|
||||
# Check as standalone command or with path
|
||||
if (f" {net_cmd} " in f" {command_lower} " or
|
||||
command_lower.startswith(f"{net_cmd} ") or
|
||||
command_lower == net_cmd or
|
||||
f"/{net_cmd} " in command_lower or
|
||||
f"/{net_cmd}" == command_lower[-len(net_cmd)-1:]):
|
||||
return f"Network command not allowed: {net_cmd}"
|
||||
|
||||
# Split by command separators to validate each sub-command
|
||||
# Handle &&, ||, ; but also pipe |
|
||||
sub_commands = self._split_commands(command)
|
||||
|
||||
for sub_cmd in sub_commands:
|
||||
sub_cmd = sub_cmd.strip()
|
||||
if not sub_cmd:
|
||||
continue
|
||||
|
||||
error = self._validate_single_command(sub_cmd, working_dir)
|
||||
if error:
|
||||
return error
|
||||
|
||||
return None
|
||||
|
||||
def _split_commands(self, command: str) -> list[str]:
|
||||
"""Split command by separators (&&, ||, ;) but not pipes."""
|
||||
# Simple split - could be improved with proper shell parsing
|
||||
result = []
|
||||
current = ""
|
||||
i = 0
|
||||
while i < len(command):
|
||||
if command[i:i+2] in ("&&", "||"):
|
||||
result.append(current)
|
||||
current = ""
|
||||
i += 2
|
||||
elif command[i] == ";":
|
||||
result.append(current)
|
||||
current = ""
|
||||
i += 1
|
||||
else:
|
||||
current += command[i]
|
||||
i += 1
|
||||
if current:
|
||||
result.append(current)
|
||||
return result
|
||||
|
||||
def _validate_single_command(self, command: str, working_dir: Path) -> str | None:
|
||||
"""Validate a single command (no &&, ||, ;)."""
|
||||
# Handle pipes - validate first command in pipe chain
|
||||
if "|" in command:
|
||||
command = command.split("|")[0].strip()
|
||||
|
||||
# Parse command
|
||||
try:
|
||||
tokens = shlex.split(command)
|
||||
if not tokens:
|
||||
return None # Empty is ok (could be whitespace)
|
||||
except ValueError as e:
|
||||
return f"Invalid command syntax: {e}"
|
||||
|
||||
# Get base command
|
||||
base_cmd = Path(tokens[0]).name
|
||||
|
||||
# Check if command is allowed
|
||||
if base_cmd not in ALLOWED_COMMANDS:
|
||||
return f"Command not allowed: {base_cmd}"
|
||||
|
||||
# Special handling for specific commands
|
||||
if base_cmd == "git":
|
||||
return self._validate_git_command(tokens)
|
||||
elif base_cmd == "rm":
|
||||
return self._validate_rm_command(tokens, working_dir)
|
||||
elif base_cmd in ("cp", "mv"):
|
||||
return self._validate_copy_move_command(tokens, working_dir)
|
||||
elif base_cmd == "chmod":
|
||||
return self._validate_chmod_command(tokens)
|
||||
elif base_cmd in ("pip", "pip3"):
|
||||
return self._validate_pip_command(tokens)
|
||||
|
||||
return None
|
||||
|
||||
def _validate_git_command(self, tokens: list[str]) -> str | None:
|
||||
"""Validate git command."""
|
||||
if len(tokens) < 2:
|
||||
return "Git command requires a subcommand"
|
||||
|
||||
git_subcommand = tokens[1]
|
||||
|
||||
# Handle git with flags before subcommand (e.g., git -C path status)
|
||||
if git_subcommand.startswith("-"):
|
||||
# Find the actual subcommand
|
||||
for i, token in enumerate(tokens[2:], 2):
|
||||
if not token.startswith("-"):
|
||||
git_subcommand = token
|
||||
break
|
||||
else:
|
||||
return "Git command requires a subcommand"
|
||||
|
||||
if git_subcommand not in ALLOWED_GIT_SUBCOMMANDS:
|
||||
return f"Git subcommand not allowed: {git_subcommand}"
|
||||
|
||||
# Block git push (could push to remote)
|
||||
if git_subcommand == "push":
|
||||
return "git push is not allowed (use manually)"
|
||||
|
||||
return None
|
||||
|
||||
def _validate_rm_command(self, tokens: list[str], working_dir: Path) -> str | None:
|
||||
"""Validate rm command - only single files, no -rf."""
|
||||
# Check for dangerous flags
|
||||
flags = [t for t in tokens[1:] if t.startswith("-")]
|
||||
for flag in flags:
|
||||
if "r" in flag and "f" in flag:
|
||||
return "rm -rf is not allowed"
|
||||
if flag in DANGEROUS_RM_FLAGS:
|
||||
return f"rm flag not allowed: {flag}"
|
||||
|
||||
# Must have at least one non-flag argument
|
||||
args = [t for t in tokens[1:] if not t.startswith("-")]
|
||||
if not args:
|
||||
return "rm requires a file argument"
|
||||
|
||||
# Validate each path
|
||||
for arg in args:
|
||||
path = Path(arg)
|
||||
if not path.is_absolute():
|
||||
path = working_dir / path
|
||||
|
||||
if not self._validate_path(path, self.allowed_paths):
|
||||
return f"rm target not in allowed paths: {arg}"
|
||||
|
||||
return None
|
||||
|
||||
def _validate_copy_move_command(
|
||||
self, tokens: list[str], working_dir: Path
|
||||
) -> str | None:
|
||||
"""Validate cp/mv command - target must be in allowed paths."""
|
||||
# Get non-flag arguments
|
||||
args = [t for t in tokens[1:] if not t.startswith("-")]
|
||||
|
||||
if len(args) < 2:
|
||||
return None # Let the command fail naturally
|
||||
|
||||
# Last argument is typically the destination
|
||||
dest = args[-1]
|
||||
dest_path = Path(dest)
|
||||
if not dest_path.is_absolute():
|
||||
dest_path = working_dir / dest_path
|
||||
|
||||
if not self._validate_path(dest_path, self.allowed_paths):
|
||||
return f"Copy/move destination not in allowed paths: {dest}"
|
||||
|
||||
return None
|
||||
|
||||
def _validate_chmod_command(self, tokens: list[str]) -> str | None:
|
||||
"""Validate chmod command - block dangerous modes."""
|
||||
for token in tokens[1:]:
|
||||
if token.startswith("-"):
|
||||
continue
|
||||
# Block 777 and similar
|
||||
if "777" in token or "666" in token:
|
||||
return f"chmod mode not allowed: {token}"
|
||||
|
||||
return None
|
||||
|
||||
def _validate_pip_command(self, tokens: list[str]) -> str | None:
|
||||
"""Validate pip command - allow install, block uninstall of system packages."""
|
||||
if len(tokens) < 2:
|
||||
return None
|
||||
|
||||
subcommand = tokens[1]
|
||||
|
||||
# Allow help, list, show, freeze, check
|
||||
allowed_pip = {"install", "list", "show", "freeze", "check", "help", "--help", "-h"}
|
||||
|
||||
if subcommand not in allowed_pip:
|
||||
return f"pip subcommand not allowed: {subcommand}"
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,481 @@
|
||||
"""
|
||||
Tests for coding tools (edit, write, bash full).
|
||||
"""
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.domains.tools.file.edit import EditFileTool
|
||||
from src.domains.tools.file.write import WriteFileTool
|
||||
from src.domains.tools.shell.bash_full import BashTool
|
||||
|
||||
|
||||
class TestEditFileTool:
|
||||
"""Tests for EditFileTool."""
|
||||
|
||||
@pytest.fixture
|
||||
def tool(self):
|
||||
return EditFileTool()
|
||||
|
||||
@pytest.fixture
|
||||
def temp_file(self):
|
||||
"""Create a temporary file with content."""
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
|
||||
f.write("def hello():\n return 'Hello'\n\ndef world():\n return 'World'\n")
|
||||
f.flush()
|
||||
yield Path(f.name)
|
||||
Path(f.name).unlink(missing_ok=True)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_success(self, tool, temp_file):
|
||||
"""Test successful single replacement."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="return 'Hello'",
|
||||
new_string="return 'Hi'"
|
||||
)
|
||||
assert result.success
|
||||
assert "Hi" in Path(temp_file).read_text()
|
||||
assert "Hello" not in Path(temp_file).read_text()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_not_found(self, tool):
|
||||
"""Test editing non-existent file."""
|
||||
result = await tool.execute(
|
||||
file_path="/nonexistent/file.py",
|
||||
old_string="old",
|
||||
new_string="new"
|
||||
)
|
||||
assert not result.success
|
||||
assert "not found" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_old_string_not_found(self, tool, temp_file):
|
||||
"""Test when old_string doesn't exist in file."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="nonexistent text",
|
||||
new_string="replacement"
|
||||
)
|
||||
assert not result.success
|
||||
assert "not found" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_multiple_matches_error(self, tool, temp_file):
|
||||
"""Test error when old_string has multiple matches and replace_all=False."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="return",
|
||||
new_string="yield"
|
||||
)
|
||||
assert not result.success
|
||||
assert "2" in result.error # Should mention count
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_replace_all(self, tool, temp_file):
|
||||
"""Test replace_all=True replaces all occurrences."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="return",
|
||||
new_string="yield",
|
||||
replace_all=True
|
||||
)
|
||||
assert result.success
|
||||
content = Path(temp_file).read_text()
|
||||
assert "return" not in content
|
||||
assert content.count("yield") == 2
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_path_restriction(self, temp_file):
|
||||
"""Test path restriction enforcement."""
|
||||
tool = EditFileTool(allowed_paths=["/some/other/path"])
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="Hello",
|
||||
new_string="Hi"
|
||||
)
|
||||
assert not result.success
|
||||
assert "not in allowed" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_empty_old_string(self, tool, temp_file):
|
||||
"""Test that empty old_string is rejected."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="",
|
||||
new_string="new"
|
||||
)
|
||||
assert not result.success
|
||||
assert "empty" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_edit_file_same_string(self, tool, temp_file):
|
||||
"""Test that identical old/new strings are rejected."""
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_file),
|
||||
old_string="Hello",
|
||||
new_string="Hello"
|
||||
)
|
||||
assert not result.success
|
||||
assert "identical" in result.error.lower()
|
||||
|
||||
|
||||
class TestWriteFileTool:
|
||||
"""Tests for WriteFileTool."""
|
||||
|
||||
@pytest.fixture
|
||||
def tool(self):
|
||||
return WriteFileTool()
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir(self):
|
||||
"""Create a temporary directory."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield Path(tmpdir)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_new_file(self, tool, temp_dir):
|
||||
"""Test creating a new file."""
|
||||
file_path = temp_dir / "new_file.py"
|
||||
result = await tool.execute(
|
||||
file_path=str(file_path),
|
||||
content="# New file\nprint('hello')"
|
||||
)
|
||||
assert result.success
|
||||
assert file_path.exists()
|
||||
assert "hello" in file_path.read_text()
|
||||
assert result.metadata.get("overwritten") is False
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_overwrite_existing(self, tool, temp_dir):
|
||||
"""Test overwriting an existing file."""
|
||||
file_path = temp_dir / "existing.txt"
|
||||
file_path.write_text("old content")
|
||||
|
||||
result = await tool.execute(
|
||||
file_path=str(file_path),
|
||||
content="new content"
|
||||
)
|
||||
assert result.success
|
||||
assert file_path.read_text() == "new content"
|
||||
assert result.metadata.get("overwritten") is True
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_file_path_restriction(self, temp_dir):
|
||||
"""Test path restriction enforcement."""
|
||||
tool = WriteFileTool(allowed_paths=["/some/other/path"])
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_dir / "file.txt"),
|
||||
content="content"
|
||||
)
|
||||
assert not result.success
|
||||
assert "not in allowed" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_file_parent_not_exists(self, tool, temp_dir):
|
||||
"""Test writing to path where parent directory doesn't exist."""
|
||||
file_path = temp_dir / "nonexistent_dir" / "file.txt"
|
||||
result = await tool.execute(
|
||||
file_path=str(file_path),
|
||||
content="content"
|
||||
)
|
||||
assert not result.success
|
||||
assert "parent" in result.error.lower() or "directory" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_file_content_size_limit(self, temp_dir):
|
||||
"""Test content size limit enforcement."""
|
||||
tool = WriteFileTool(max_content_size=100)
|
||||
result = await tool.execute(
|
||||
file_path=str(temp_dir / "large.txt"),
|
||||
content="x" * 200
|
||||
)
|
||||
assert not result.success
|
||||
assert "large" in result.error.lower() or "size" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_write_file_returns_metadata(self, tool, temp_dir):
|
||||
"""Test that metadata is returned correctly."""
|
||||
file_path = temp_dir / "meta.txt"
|
||||
content = "line1\nline2\nline3"
|
||||
result = await tool.execute(
|
||||
file_path=str(file_path),
|
||||
content=content
|
||||
)
|
||||
assert result.success
|
||||
assert result.metadata.get("lines") == 3
|
||||
assert result.metadata.get("file_size") == len(content.encode('utf-8'))
|
||||
|
||||
|
||||
class TestBashTool:
|
||||
"""Tests for BashTool (full write capabilities)."""
|
||||
|
||||
@pytest.fixture
|
||||
def tool(self):
|
||||
return BashTool()
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir(self):
|
||||
"""Create a temporary directory."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield Path(tmpdir)
|
||||
|
||||
# === Allowed commands ===
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_ls_command(self, tool, temp_dir):
|
||||
"""Test ls is allowed."""
|
||||
result = await tool.execute(command="ls", cwd=str(temp_dir))
|
||||
assert result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_mkdir_command(self, tool, temp_dir):
|
||||
"""Test mkdir is allowed."""
|
||||
result = await tool.execute(
|
||||
command=f"mkdir {temp_dir}/new_dir",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert (temp_dir / "new_dir").exists()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_touch_command(self, tool, temp_dir):
|
||||
"""Test touch is allowed."""
|
||||
result = await tool.execute(
|
||||
command=f"touch {temp_dir}/new_file.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert (temp_dir / "new_file.txt").exists()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_cp_command(self, tool, temp_dir):
|
||||
"""Test cp within allowed paths."""
|
||||
(temp_dir / "source.txt").write_text("content")
|
||||
result = await tool.execute(
|
||||
command=f"cp {temp_dir}/source.txt {temp_dir}/dest.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert (temp_dir / "dest.txt").exists()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_mv_command(self, tool, temp_dir):
|
||||
"""Test mv within allowed paths."""
|
||||
(temp_dir / "source.txt").write_text("content")
|
||||
result = await tool.execute(
|
||||
command=f"mv {temp_dir}/source.txt {temp_dir}/moved.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert (temp_dir / "moved.txt").exists()
|
||||
assert not (temp_dir / "source.txt").exists()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_command_chaining_and(self, tool, temp_dir):
|
||||
"""Test && chaining is allowed."""
|
||||
result = await tool.execute(
|
||||
command=f"mkdir {temp_dir}/dir1 && touch {temp_dir}/dir1/file.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert (temp_dir / "dir1" / "file.txt").exists()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_command_chaining_or(self, tool, temp_dir):
|
||||
"""Test || chaining is allowed."""
|
||||
result = await tool.execute(
|
||||
command=f"ls {temp_dir}/nonexistent || echo 'fallback'",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
# Either succeeds or falls back
|
||||
assert result.success or "fallback" in (result.data or "")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_echo_command(self, tool, temp_dir):
|
||||
"""Test echo command."""
|
||||
result = await tool.execute(
|
||||
command="echo 'hello world'",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert "hello world" in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_git_status(self, tool, temp_dir):
|
||||
"""Test git status is allowed."""
|
||||
# Initialize a git repo first
|
||||
await tool.execute(command="git init", cwd=str(temp_dir))
|
||||
result = await tool.execute(command="git status", cwd=str(temp_dir))
|
||||
assert result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_git_add_allowed(self, tool, temp_dir):
|
||||
"""Test git add is allowed."""
|
||||
await tool.execute(command="git init", cwd=str(temp_dir))
|
||||
(temp_dir / "file.txt").write_text("content")
|
||||
result = await tool.execute(command="git add file.txt", cwd=str(temp_dir))
|
||||
assert result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_pip_help(self, tool, temp_dir):
|
||||
"""Test pip help is allowed."""
|
||||
result = await tool.execute(
|
||||
command="pip --help",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_rm_single_file(self, tool, temp_dir):
|
||||
"""Test rm of a single file is allowed."""
|
||||
file_path = temp_dir / "to_delete.txt"
|
||||
file_path.write_text("content")
|
||||
|
||||
result = await tool.execute(
|
||||
command=f"rm {file_path}",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert result.success
|
||||
assert not file_path.exists()
|
||||
|
||||
# === Forbidden commands ===
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_sudo(self, tool, temp_dir):
|
||||
"""Test sudo is blocked."""
|
||||
result = await tool.execute(command="sudo ls", cwd=str(temp_dir))
|
||||
assert not result.success
|
||||
assert "forbidden" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_curl(self, tool, temp_dir):
|
||||
"""Test curl is blocked."""
|
||||
result = await tool.execute(
|
||||
command="curl http://example.com",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_wget(self, tool, temp_dir):
|
||||
"""Test wget is blocked."""
|
||||
result = await tool.execute(
|
||||
command="wget http://example.com",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_ssh(self, tool, temp_dir):
|
||||
"""Test ssh is blocked."""
|
||||
result = await tool.execute(
|
||||
command="ssh user@host",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_rm_rf(self, tool, temp_dir):
|
||||
"""Test rm -rf is blocked."""
|
||||
result = await tool.execute(
|
||||
command="rm -rf /tmp/test",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_rm_rf_dot(self, tool, temp_dir):
|
||||
"""Test rm -rf . is blocked."""
|
||||
result = await tool.execute(
|
||||
command="rm -rf .",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_chmod_777(self, tool, temp_dir):
|
||||
"""Test chmod 777 is blocked."""
|
||||
result = await tool.execute(
|
||||
command="chmod 777 /tmp/file",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_git_push(self, tool, temp_dir):
|
||||
"""Test git push is blocked."""
|
||||
await tool.execute(command="git init", cwd=str(temp_dir))
|
||||
result = await tool.execute(
|
||||
command="git push origin main",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_forbidden_pip_uninstall(self, tool, temp_dir):
|
||||
"""Test pip uninstall is blocked."""
|
||||
result = await tool.execute(
|
||||
command="pip uninstall requests",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower()
|
||||
|
||||
# === Path restrictions ===
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_working_dir_not_allowed(self, temp_dir):
|
||||
"""Test working directory restriction."""
|
||||
tool = BashTool(allowed_paths=["/some/other/path"])
|
||||
result = await tool.execute(
|
||||
command="ls",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "not allowed" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_cp_outside_allowed_paths(self, temp_dir):
|
||||
"""Test cp to path outside allowed_paths fails."""
|
||||
tool = BashTool(allowed_paths=[str(temp_dir)])
|
||||
(temp_dir / "source.txt").write_text("content")
|
||||
|
||||
result = await tool.execute(
|
||||
command=f"cp {temp_dir}/source.txt /tmp/dest.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "allowed" in result.error.lower()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_rm_outside_allowed_paths(self, temp_dir):
|
||||
"""Test rm of path outside allowed_paths fails."""
|
||||
tool = BashTool(allowed_paths=[str(temp_dir)])
|
||||
|
||||
result = await tool.execute(
|
||||
command="rm /tmp/some_file.txt",
|
||||
cwd=str(temp_dir)
|
||||
)
|
||||
assert not result.success
|
||||
assert "allowed" in result.error.lower()
|
||||
|
||||
# === Timeout ===
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_timeout(self, tool, temp_dir):
|
||||
"""Test command timeout using find on root (slow)."""
|
||||
# Use find on a large directory which will be slow
|
||||
result = await tool.execute(
|
||||
command="find / -name '*.nonexistent' 2>/dev/null",
|
||||
cwd=str(temp_dir),
|
||||
timeout=1
|
||||
)
|
||||
# The command should either timeout or fail
|
||||
# (it may complete quickly with errors, which is also acceptable)
|
||||
assert not result.success or result.truncated or "timeout" in str(result.error or "").lower()
|
||||
Reference in New Issue
Block a user