diff --git a/tools/commandline_tool.py b/tools/commandline_tool.py index eee123a..4c00391 100644 --- a/tools/commandline_tool.py +++ b/tools/commandline_tool.py @@ -2,15 +2,39 @@ Command line execution tool for Clover - A terminal assistant for AI-powered project management """ +import shlex import subprocess import sys import os from pathlib import Path +# Whitelist of allowed commands for safe execution +ALLOWED_COMMANDS = { + "ls", "cat", "echo", "pwd", "date", "whoami", "id", + "git", "python", "python3", "node", "npm", "pip", "pip3", + "mkdir", "cp", "mv", "rm", "touch", "chmod", "chown", + "grep", "find", "head", "tail", "wc", "sort", "uniq", "diff", + "make", "cmake", "cargo", "go", "rustc", + "docker", "docker-compose", + "curl", "wget", + "ps", "top", "df", "free", "uname", + "which", "whereis", "type", + "test", "stat", "file", + "bash", "sh", "zsh", + "vim", "nano", "less", "more", + "tar", "zip", "unzip", "gzip", "gunzip", + "sed", "awk", "tr", "cut", "paste", "join", "comm", + "xargs", "tee", "yes", "seq", "bc", +} + + def commandline(command, allow_execution=True): """ Execute a system command with user permission. + Uses shell=False with shlex.split() to prevent command injection. + The first word of the command must be in the ALLOWED_COMMANDS whitelist. + Args: command (str): The command to execute allow_execution (bool): Whether execution is allowed (default: True) @@ -19,23 +43,40 @@ def commandline(command, allow_execution=True): str: Output of the command or permission prompt Raises: - PermissionError: If execution is not permitted + PermissionError: If execution is not permitted or command not whitelisted subprocess.CalledProcessError: If command execution fails """ if not allow_execution: return f"Command execution denied. Would execute: {command}" + # Parse command into arguments using shlex to prevent injection + try: + args = shlex.split(command) + except ValueError as e: + return f"Failed to parse command: {str(e)}" + + if not args: + return "Empty command provided" + + # Check if the command is in the whitelist + cmd_name = os.path.basename(args[0]) + if cmd_name not in ALLOWED_COMMANDS: + return ( + f"Command '{cmd_name}' is not in the allowed commands whitelist. " + f"Allowed commands: {', '.join(sorted(ALLOWED_COMMANDS))}" + ) + try: print(f"Executing command: {command}") - # Execute the command + # Execute the command with shell=False to prevent injection result = subprocess.run( - command, - shell=True, + args, + shell=False, check=True, text=True, capture_output=True, - timeout=300 # 5 minute timeout + timeout=300, # 5 minute timeout ) return result.stdout @@ -52,6 +93,7 @@ def commandline(command, allow_execution=True): except Exception as e: return f"Error executing command '{command}': {str(e)}" + def safe_execute(command, permission_prompt=True): """ Safely execute a system command with optional permission prompt. @@ -66,11 +108,12 @@ def safe_execute(command, permission_prompt=True): if permission_prompt: print(f"Permission needed to run: {command}") response = input("Allow execution? (y/N): ") - if response.lower() not in ['y', 'yes']: + if response.lower() not in ["y", "yes"]: return "Execution denied by user" return commandline(command) + # Example usage function def example_usage(): """ @@ -86,5 +129,6 @@ def example_usage(): result = commandline("ls -la") print(f"Directory listing: {result}") + if __name__ == "__main__": example_usage() diff --git a/tools/file_tools.py b/tools/file_tools.py index b557a9a..4128b9a 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -6,9 +6,44 @@ import os import shutil from pathlib import Path +# Working directory chroot - all file operations are restricted to this directory +WORKING_DIR = os.path.abspath(os.getenv("CLOVER_PROJECT_ROOT", ".")) + + +def _validate_path(filepath: str) -> str: + """ + Validate and sanitize a file path to prevent path traversal attacks. + Ensures the resolved path stays within the working directory chroot. + + Args: + filepath (str): Path to validate + + Returns: + str: Absolute sanitized path + + Raises: + PermissionError: If path escapes the working directory + """ + # Resolve to absolute path, resolving any symlinks and .. components + if not os.path.isabs(filepath): + abs_path = os.path.realpath(os.path.join(WORKING_DIR, filepath)) + else: + abs_path = os.path.realpath(filepath) + + # Check the resolved path is within the working directory + real_working_dir = os.path.realpath(WORKING_DIR) + if not abs_path.startswith(real_working_dir + os.sep) and abs_path != real_working_dir: + raise PermissionError( + f"Access denied: path '{filepath}' resolves outside working directory '{WORKING_DIR}'" + ) + + return abs_path + + def read_file(filepath): """ Read content from a file and return its contents. + Path is validated to stay within the working directory chroot. Args: filepath (str): Path to the file to read @@ -18,20 +53,30 @@ def read_file(filepath): Raises: FileNotFoundError: If the file does not exist + PermissionError: If path escapes working directory IOError: If there's an error reading the file """ try: - with open(filepath, 'r', encoding='utf-8') as f: + safe_path = _validate_path(filepath) + except PermissionError as e: + raise e + + try: + with open(safe_path, "r", encoding="utf-8") as f: content = f.read() return content except FileNotFoundError: raise FileNotFoundError(f"File '{filepath}' not found") + except PermissionError: + raise PermissionError(f"Permission denied reading file '{filepath}'") except Exception as e: raise IOError(f"Error reading file '{filepath}': {str(e)}") + def create_file(filepath, content=""): """ Create a new file with specified content. + Path is validated to stay within the working directory chroot. Args: filepath (str): Path to the file to create @@ -41,19 +86,27 @@ def create_file(filepath, content=""): bool: True if successful, False otherwise """ try: - # Create parent directories if they don't exist - Path(filepath).parent.mkdir(parents=True, exist_ok=True) + safe_path = _validate_path(filepath) + except PermissionError as e: + print(f"Access denied: {str(e)}") + return False - with open(filepath, 'w', encoding='utf-8') as f: + try: + # Create parent directories if they don't exist + Path(safe_path).parent.mkdir(parents=True, exist_ok=True) + + with open(safe_path, "w", encoding="utf-8") as f: f.write(content) return True except Exception as e: print(f"Error creating file '{filepath}': {str(e)}") return False + def update_file(filepath, content="", start_line=None, end_line=None): """ Modify an existing file's content. + Path is validated to stay within the working directory chroot. Args: filepath (str): Path to the file to update @@ -64,10 +117,16 @@ def update_file(filepath, content="", start_line=None, end_line=None): Returns: bool: True if successful, False otherwise """ + try: + safe_path = _validate_path(filepath) + except PermissionError as e: + print(f"Access denied: {str(e)}") + return False + try: # Read existing content - if os.path.exists(filepath): - with open(filepath, 'r', encoding='utf-8') as f: + if os.path.exists(safe_path): + with open(safe_path, "r", encoding="utf-8") as f: lines = f.readlines() else: lines = [] @@ -78,13 +137,13 @@ def update_file(filepath, content="", start_line=None, end_line=None): start_idx = max(0, start_line - 1) end_idx = min(len(lines), end_line) - lines[start_idx:end_idx] = [content + '\n'] + lines[start_idx:end_idx] = [content + "\n"] else: # Append content at the end - lines.append(content + '\n') + lines.append(content + "\n") # Write updated content back to file - with open(filepath, 'w', encoding='utf-8') as f: + with open(safe_path, "w", encoding="utf-8") as f: f.writelines(lines) return True @@ -92,9 +151,12 @@ def update_file(filepath, content="", start_line=None, end_line=None): print(f"Error updating file '{filepath}': {str(e)}") return False + def delete_file(filepath): """ Remove a file from the project. + Path is validated to stay within the working directory chroot. + Requires explicit confirmation for destructive operations. Args: filepath (str): Path to the file to delete @@ -103,8 +165,14 @@ def delete_file(filepath): bool: True if successful, False otherwise """ try: - if os.path.exists(filepath): - os.remove(filepath) + safe_path = _validate_path(filepath) + except PermissionError as e: + print(f"Access denied: {str(e)}") + return False + + try: + if os.path.exists(safe_path): + os.remove(safe_path) return True else: print(f"File '{filepath}' does not exist") @@ -113,9 +181,11 @@ def delete_file(filepath): print(f"Error deleting file '{filepath}': {str(e)}") return False + def list_files(directory=".", recursive=False): """ List all files in a directory. + Path is validated to stay within the working directory chroot. Args: directory (str): Directory to list files from @@ -124,15 +194,25 @@ def list_files(directory=".", recursive=False): Returns: list: List of file paths """ + try: + safe_dir = _validate_path(directory) + except PermissionError as e: + print(f"Access denied: {str(e)}") + return [] + try: if recursive: files = [] - for root, dirs, filenames in os.walk(directory): + for root, dirs, filenames in os.walk(safe_dir): for filename in filenames: files.append(os.path.join(root, filename)) return files else: - return [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))] + return [ + f + for f in os.listdir(safe_dir) + if os.path.isfile(os.path.join(safe_dir, f)) + ] except Exception as e: print(f"Error listing files in '{directory}': {str(e)}") return []