Merge branch 'main' into fix/issue-6
This commit is contained in:
commit
29bec257fa
@ -29,7 +29,9 @@ def load_config():
|
|||||||
"threads": int(os.getenv("CLOVER_THREADS", "5")),
|
"threads": int(os.getenv("CLOVER_THREADS", "5")),
|
||||||
"model": os.getenv("CLOVER_MODEL", "qwen3-coder:30b"),
|
"model": os.getenv("CLOVER_MODEL", "qwen3-coder:30b"),
|
||||||
"api_key": os.getenv("CLOVER_API_KEY"),
|
"api_key": os.getenv("CLOVER_API_KEY"),
|
||||||
"base_url": os.getenv("CLOVER_BASE_URL", "http://192.168.8.223:11434"),
|
"base_url": os.getenv(
|
||||||
|
"CLOVER_BASE_URL", "error-set-CLOVER_BASE_URL-in-.env-file"
|
||||||
|
),
|
||||||
"debug": os.getenv("CLOVER_DEBUG", "false").lower() == "true",
|
"debug": os.getenv("CLOVER_DEBUG", "false").lower() == "true",
|
||||||
"verbose": os.getenv("CLOVER_VERBOSE", "false").lower() == "true",
|
"verbose": os.getenv("CLOVER_VERBOSE", "false").lower() == "true",
|
||||||
"cache_enabled": os.getenv("CLOVER_CACHE_ENABLED", "true").lower() == "true",
|
"cache_enabled": os.getenv("CLOVER_CACHE_ENABLED", "true").lower() == "true",
|
||||||
@ -206,8 +208,12 @@ def validate_config():
|
|||||||
issues = []
|
issues = []
|
||||||
|
|
||||||
# Check required settings
|
# Check required settings
|
||||||
if not config.get("base_url"):
|
base_url = config.get("base_url", "")
|
||||||
issues.append("base_url is required")
|
if not base_url or base_url.startswith("error-set-"):
|
||||||
|
issues.append(
|
||||||
|
"CLOVER_BASE_URL is not set. Set it in your .env file or environment. "
|
||||||
|
"Example: CLOVER_BASE_URL=http://localhost:11434"
|
||||||
|
)
|
||||||
|
|
||||||
# Check numeric values
|
# Check numeric values
|
||||||
try:
|
try:
|
||||||
|
|||||||
@ -2,15 +2,39 @@
|
|||||||
Command line execution tool for Clover - A terminal assistant for AI-powered project management
|
Command line execution tool for Clover - A terminal assistant for AI-powered project management
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import shlex
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
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):
|
def commandline(command, allow_execution=True):
|
||||||
"""
|
"""
|
||||||
Execute a system command with user permission.
|
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:
|
Args:
|
||||||
command (str): The command to execute
|
command (str): The command to execute
|
||||||
allow_execution (bool): Whether execution is allowed (default: True)
|
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
|
str: Output of the command or permission prompt
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
PermissionError: If execution is not permitted
|
PermissionError: If execution is not permitted or command not whitelisted
|
||||||
subprocess.CalledProcessError: If command execution fails
|
subprocess.CalledProcessError: If command execution fails
|
||||||
"""
|
"""
|
||||||
if not allow_execution:
|
if not allow_execution:
|
||||||
return f"Command execution denied. Would execute: {command}"
|
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:
|
try:
|
||||||
print(f"Executing command: {command}")
|
print(f"Executing command: {command}")
|
||||||
|
|
||||||
# Execute the command
|
# Execute the command with shell=False to prevent injection
|
||||||
result = subprocess.run(
|
result = subprocess.run(
|
||||||
command,
|
args,
|
||||||
shell=True,
|
shell=False,
|
||||||
check=True,
|
check=True,
|
||||||
text=True,
|
text=True,
|
||||||
capture_output=True,
|
capture_output=True,
|
||||||
timeout=300 # 5 minute timeout
|
timeout=300, # 5 minute timeout
|
||||||
)
|
)
|
||||||
|
|
||||||
return result.stdout
|
return result.stdout
|
||||||
@ -52,6 +93,7 @@ def commandline(command, allow_execution=True):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
return f"Error executing command '{command}': {str(e)}"
|
return f"Error executing command '{command}': {str(e)}"
|
||||||
|
|
||||||
|
|
||||||
def safe_execute(command, permission_prompt=True):
|
def safe_execute(command, permission_prompt=True):
|
||||||
"""
|
"""
|
||||||
Safely execute a system command with optional permission prompt.
|
Safely execute a system command with optional permission prompt.
|
||||||
@ -66,11 +108,12 @@ def safe_execute(command, permission_prompt=True):
|
|||||||
if permission_prompt:
|
if permission_prompt:
|
||||||
print(f"Permission needed to run: {command}")
|
print(f"Permission needed to run: {command}")
|
||||||
response = input("Allow execution? (y/N): ")
|
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 "Execution denied by user"
|
||||||
|
|
||||||
return commandline(command)
|
return commandline(command)
|
||||||
|
|
||||||
|
|
||||||
# Example usage function
|
# Example usage function
|
||||||
def example_usage():
|
def example_usage():
|
||||||
"""
|
"""
|
||||||
@ -86,5 +129,6 @@ def example_usage():
|
|||||||
result = commandline("ls -la")
|
result = commandline("ls -la")
|
||||||
print(f"Directory listing: {result}")
|
print(f"Directory listing: {result}")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
example_usage()
|
example_usage()
|
||||||
|
|||||||
@ -6,9 +6,44 @@ import os
|
|||||||
import shutil
|
import shutil
|
||||||
from pathlib import Path
|
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):
|
def read_file(filepath):
|
||||||
"""
|
"""
|
||||||
Read content from a file and return its contents.
|
Read content from a file and return its contents.
|
||||||
|
Path is validated to stay within the working directory chroot.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
filepath (str): Path to the file to read
|
filepath (str): Path to the file to read
|
||||||
@ -18,20 +53,30 @@ def read_file(filepath):
|
|||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
FileNotFoundError: If the file does not exist
|
FileNotFoundError: If the file does not exist
|
||||||
|
PermissionError: If path escapes working directory
|
||||||
IOError: If there's an error reading the file
|
IOError: If there's an error reading the file
|
||||||
"""
|
"""
|
||||||
try:
|
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()
|
content = f.read()
|
||||||
return content
|
return content
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
raise FileNotFoundError(f"File '{filepath}' not found")
|
raise FileNotFoundError(f"File '{filepath}' not found")
|
||||||
|
except PermissionError:
|
||||||
|
raise PermissionError(f"Permission denied reading file '{filepath}'")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise IOError(f"Error reading file '{filepath}': {str(e)}")
|
raise IOError(f"Error reading file '{filepath}': {str(e)}")
|
||||||
|
|
||||||
|
|
||||||
def create_file(filepath, content=""):
|
def create_file(filepath, content=""):
|
||||||
"""
|
"""
|
||||||
Create a new file with specified content.
|
Create a new file with specified content.
|
||||||
|
Path is validated to stay within the working directory chroot.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
filepath (str): Path to the file to create
|
filepath (str): Path to the file to create
|
||||||
@ -41,19 +86,27 @@ def create_file(filepath, content=""):
|
|||||||
bool: True if successful, False otherwise
|
bool: True if successful, False otherwise
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
# Create parent directories if they don't exist
|
safe_path = _validate_path(filepath)
|
||||||
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
|
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)
|
f.write(content)
|
||||||
return True
|
return True
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"Error creating file '{filepath}': {str(e)}")
|
print(f"Error creating file '{filepath}': {str(e)}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def update_file(filepath, content="", start_line=None, end_line=None):
|
def update_file(filepath, content="", start_line=None, end_line=None):
|
||||||
"""
|
"""
|
||||||
Modify an existing file's content.
|
Modify an existing file's content.
|
||||||
|
Path is validated to stay within the working directory chroot.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
filepath (str): Path to the file to update
|
filepath (str): Path to the file to update
|
||||||
@ -64,10 +117,16 @@ def update_file(filepath, content="", start_line=None, end_line=None):
|
|||||||
Returns:
|
Returns:
|
||||||
bool: True if successful, False otherwise
|
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:
|
try:
|
||||||
# Read existing content
|
# Read existing content
|
||||||
if os.path.exists(filepath):
|
if os.path.exists(safe_path):
|
||||||
with open(filepath, 'r', encoding='utf-8') as f:
|
with open(safe_path, "r", encoding="utf-8") as f:
|
||||||
lines = f.readlines()
|
lines = f.readlines()
|
||||||
else:
|
else:
|
||||||
lines = []
|
lines = []
|
||||||
@ -78,13 +137,13 @@ def update_file(filepath, content="", start_line=None, end_line=None):
|
|||||||
start_idx = max(0, start_line - 1)
|
start_idx = max(0, start_line - 1)
|
||||||
end_idx = min(len(lines), end_line)
|
end_idx = min(len(lines), end_line)
|
||||||
|
|
||||||
lines[start_idx:end_idx] = [content + '\n']
|
lines[start_idx:end_idx] = [content + "\n"]
|
||||||
else:
|
else:
|
||||||
# Append content at the end
|
# Append content at the end
|
||||||
lines.append(content + '\n')
|
lines.append(content + "\n")
|
||||||
|
|
||||||
# Write updated content back to file
|
# 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)
|
f.writelines(lines)
|
||||||
|
|
||||||
return True
|
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)}")
|
print(f"Error updating file '{filepath}': {str(e)}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def delete_file(filepath):
|
def delete_file(filepath):
|
||||||
"""
|
"""
|
||||||
Remove a file from the project.
|
Remove a file from the project.
|
||||||
|
Path is validated to stay within the working directory chroot.
|
||||||
|
Requires explicit confirmation for destructive operations.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
filepath (str): Path to the file to delete
|
filepath (str): Path to the file to delete
|
||||||
@ -103,8 +165,14 @@ def delete_file(filepath):
|
|||||||
bool: True if successful, False otherwise
|
bool: True if successful, False otherwise
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
if os.path.exists(filepath):
|
safe_path = _validate_path(filepath)
|
||||||
os.remove(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
|
return True
|
||||||
else:
|
else:
|
||||||
print(f"File '{filepath}' does not exist")
|
print(f"File '{filepath}' does not exist")
|
||||||
@ -113,9 +181,11 @@ def delete_file(filepath):
|
|||||||
print(f"Error deleting file '{filepath}': {str(e)}")
|
print(f"Error deleting file '{filepath}': {str(e)}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def list_files(directory=".", recursive=False):
|
def list_files(directory=".", recursive=False):
|
||||||
"""
|
"""
|
||||||
List all files in a directory.
|
List all files in a directory.
|
||||||
|
Path is validated to stay within the working directory chroot.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
directory (str): Directory to list files from
|
directory (str): Directory to list files from
|
||||||
@ -124,15 +194,25 @@ def list_files(directory=".", recursive=False):
|
|||||||
Returns:
|
Returns:
|
||||||
list: List of file paths
|
list: List of file paths
|
||||||
"""
|
"""
|
||||||
|
try:
|
||||||
|
safe_dir = _validate_path(directory)
|
||||||
|
except PermissionError as e:
|
||||||
|
print(f"Access denied: {str(e)}")
|
||||||
|
return []
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if recursive:
|
if recursive:
|
||||||
files = []
|
files = []
|
||||||
for root, dirs, filenames in os.walk(directory):
|
for root, dirs, filenames in os.walk(safe_dir):
|
||||||
for filename in filenames:
|
for filename in filenames:
|
||||||
files.append(os.path.join(root, filename))
|
files.append(os.path.join(root, filename))
|
||||||
return files
|
return files
|
||||||
else:
|
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:
|
except Exception as e:
|
||||||
print(f"Error listing files in '{directory}': {str(e)}")
|
print(f"Error listing files in '{directory}': {str(e)}")
|
||||||
return []
|
return []
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user