From 496ac5ec7c98b31fdbdceb16253d9f4e6456d95d Mon Sep 17 00:00:00 2001 From: opencode Date: Sun, 5 Jul 2026 07:00:29 +0000 Subject: [PATCH 1/2] fix: prevent command injection via shell=True (issue #1, #3) - Replace shell=True with shell=False + shlex.split() - Add command whitelist to restrict allowed executables - Parse commands safely to prevent metacharacter injection (;, &&, |, backticks) --- tools/commandline_tool.py | 56 ++++++++++++++++++++++++++++++++++----- 1 file changed, 50 insertions(+), 6 deletions(-) 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() From 7680fea98eee8f37054a17551a18ab0684653080 Mon Sep 17 00:00:00 2001 From: opencode Date: Sun, 5 Jul 2026 07:01:13 +0000 Subject: [PATCH 2/2] fix: prevent path traversal in file operations (issue #2, #4) - Add _validate_path() using os.path.realpath() to resolve symlinks - Enforce working directory chroot (CLOVER_PROJECT_ROOT env var) - All file ops (read, create, update, delete, list) now validated - Block access to files outside project directory --- tools/file_tools.py | 106 ++++++++++++++++++++++++++++++++++++++------ 1 file changed, 93 insertions(+), 13 deletions(-) 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 []