diff --git a/webber-api/src/domains/agents/explore/tools.py b/webber-api/src/domains/agents/explore/tools.py index 2e4f6a8..d236b32 100644 --- a/webber-api/src/domains/agents/explore/tools.py +++ b/webber-api/src/domains/agents/explore/tools.py @@ -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() diff --git a/webber-api/src/domains/tools/__init__.py b/webber-api/src/domains/tools/__init__.py index 7ca5c3c..9b9d847 100644 --- a/webber-api/src/domains/tools/__init__.py +++ b/webber-api/src/domains/tools/__init__.py @@ -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", ] diff --git a/webber-api/src/domains/tools/file/__init__.py b/webber-api/src/domains/tools/file/__init__.py index 4cfd384..fdeec41 100644 --- a/webber-api/src/domains/tools/file/__init__.py +++ b/webber-api/src/domains/tools/file/__init__.py @@ -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"] diff --git a/webber-api/src/domains/tools/file/edit.py b/webber-api/src/domains/tools/file/edit.py new file mode 100644 index 0000000..176d2fb --- /dev/null +++ b/webber-api/src/domains/tools/file/edit.py @@ -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}") diff --git a/webber-api/src/domains/tools/file/write.py b/webber-api/src/domains/tools/file/write.py new file mode 100644 index 0000000..2cba895 --- /dev/null +++ b/webber-api/src/domains/tools/file/write.py @@ -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}") diff --git a/webber-api/src/domains/tools/shell/__init__.py b/webber-api/src/domains/tools/shell/__init__.py index 0b2118f..2955706 100644 --- a/webber-api/src/domains/tools/shell/__init__.py +++ b/webber-api/src/domains/tools/shell/__init__.py @@ -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"] diff --git a/webber-api/src/domains/tools/shell/bash_full.py b/webber-api/src/domains/tools/shell/bash_full.py new file mode 100644 index 0000000..5686b1c --- /dev/null +++ b/webber-api/src/domains/tools/shell/bash_full.py @@ -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 diff --git a/webber-api/tests/test_coding_tools.py b/webber-api/tests/test_coding_tools.py new file mode 100644 index 0000000..5f93b17 --- /dev/null +++ b/webber-api/tests/test_coding_tools.py @@ -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()