Merge pull request #740 from MervinPraison/claude/issue-737-20250707_104724

fix: critical security vulnerabilities in tools

Author: MervinPraison

src/praisonai-agents/praisonaiagents/tools/duckdb_tools.py

@@ -13,6 +13,7 @@
 from typing import List, Dict, Any, Optional, Union, TYPE_CHECKING
 from importlib import util
 import json
+import re
 
 if TYPE_CHECKING:
     import duckdb
@@ -29,6 +30,25 @@ def __init__(self, database: str = ':memory:'):
         """
         self.database = database
         self._conn = None
+    
+    def _validate_identifier(self, identifier: str) -> str:
+        """Validate and quote a SQL identifier to prevent injection.
+        
+        Args:
+            identifier: Table or column name to validate
+            
+        Returns:
+            Quoted identifier safe for SQL
+            
+        Raises:
+            ValueError: If identifier contains invalid characters
+        """
+        # Only allow alphanumeric characters, underscores, and dots
+        if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*(\.[a-zA-Z_][a-zA-Z0-9_]*)?$', identifier):
+            raise ValueError(f"Invalid identifier: {identifier}")
+        
+        # Quote the identifier to handle reserved words
+        return f'"{identifier}"'
 
     def _get_duckdb(self) -> Optional['duckdb']:
         """Get duckdb module, installing if needed"""
@@ -125,39 +145,50 @@ def load_csv(
             if conn is None:
                 return False
 
-            # Check if table exists
-            exists = conn.execute(f"""
+            # Check if table exists using parameterized query
+            exists = conn.execute("""
                 SELECT name FROM sqlite_master 
-                WHERE type='table' AND name='{table_name}'
-            """).fetchone() is not None
+                WHERE type='table' AND name=?
+            """, [table_name]).fetchone() is not None
 
             if exists:
                 if if_exists == 'fail':
                     raise ValueError(f"Table {table_name} already exists")
                 elif if_exists == 'replace':
-                    conn.execute(f"DROP TABLE IF EXISTS {table_name}")
+                    # Validate and quote table name to prevent injection
+                    safe_table = self._validate_identifier(table_name)
+                    conn.execute(f"DROP TABLE IF EXISTS {safe_table}")
                 elif if_exists != 'append':
                     raise ValueError("if_exists must be 'fail', 'replace', or 'append'")
 
             # Create table if needed
             if not exists or if_exists == 'replace':
+                safe_table = self._validate_identifier(table_name)
                 if schema:
-                    # Create table with schema
-                    columns = ', '.join(f"{k} {v}" for k, v in schema.items())
-                    conn.execute(f"CREATE TABLE {table_name} ({columns})")
+                    # Create table with schema - validate column names
+                    column_defs = []
+                    for col_name, col_type in schema.items():
+                        safe_col = self._validate_identifier(col_name)
+                        # Validate column type to prevent injection
+                        if not re.match(r'^[A-Z][A-Z0-9_]*(\([0-9,]+\))?$', col_type.upper()):
+                            raise ValueError(f"Invalid column type: {col_type}")
+                        column_defs.append(f"{safe_col} {col_type}")
+                    columns = ', '.join(column_defs)
+                    conn.execute(f"CREATE TABLE {safe_table} ({columns})")
                 else:
-                    # Infer schema from CSV
+                    # Infer schema from CSV - use parameterized query for filepath
                     conn.execute(f"""
-                        CREATE TABLE {table_name} AS 
-                        SELECT * FROM read_csv_auto('{filepath}')
+                        CREATE TABLE {safe_table} AS 
+                        SELECT * FROM read_csv_auto(?)
                         WHERE 1=0
-                    """)
+                    """, [filepath])
 
-            # Load data
+            # Load data - use validated table name and parameterized filepath
+            safe_table = self._validate_identifier(table_name)
             conn.execute(f"""
-                INSERT INTO {table_name}
-                SELECT * FROM read_csv_auto('{filepath}')
-            """)
+                INSERT INTO {safe_table}
+                SELECT * FROM read_csv_auto(?)
+            """, [filepath])
 
             return True
 

src/praisonai-agents/praisonaiagents/tools/file_tools.py

@@ -21,6 +21,32 @@
 class FileTools:
     """Tools for file operations including read, write, list, and information."""
 
+    @staticmethod
+    def _validate_path(filepath: str) -> str:
+        """
+        Validate and normalize a file path to prevent path traversal attacks.
+        
+        Args:
+            filepath: Path to validate
+            
+        Returns:
+            str: Normalized absolute path
+            
+        Raises:
+            ValueError: If path contains suspicious patterns
+        """
+        # Normalize the path
+        normalized = os.path.normpath(filepath)
+        absolute = os.path.abspath(normalized)
+        
+        # Check for suspicious patterns
+        if '..' in filepath or filepath.startswith('~'):
+            raise ValueError(f"Suspicious path pattern detected: {filepath}")
+        
+        # Additional check: ensure the resolved path doesn't escape expected boundaries
+        # This is a basic check - in production, you'd want to define allowed directories
+        return absolute
+    
     @staticmethod
     def read_file(filepath: str, encoding: str = 'utf-8') -> str:
         """
@@ -34,7 +60,9 @@ def read_file(filepath: str, encoding: str = 'utf-8') -> str:
             str: Content of the file
         """
         try:
-            with open(filepath, 'r', encoding=encoding) as f:
+            # Validate path to prevent traversal attacks
+            safe_path = FileTools._validate_path(filepath)
+            with open(safe_path, 'r', encoding=encoding) as f:
                 return f.read()
         except Exception as e:
             error_msg = f"Error reading file {filepath}: {str(e)}"
@@ -56,9 +84,11 @@ def write_file(filepath: str, content: str, encoding: str = 'utf-8') -> bool:
             bool: True if successful, False otherwise
         """
         try:
+            # Validate path to prevent traversal attacks
+            safe_path = FileTools._validate_path(filepath)
             # Create directory if it doesn't exist
-            os.makedirs(os.path.dirname(filepath), exist_ok=True)
-            with open(filepath, 'w', encoding=encoding) as f:
+            os.makedirs(os.path.dirname(safe_path), exist_ok=True)
+            with open(safe_path, 'w', encoding=encoding) as f:
                 f.write(content)
             return True
         except Exception as e:
@@ -79,7 +109,9 @@ def list_files(directory: str, pattern: Optional[str] = None) -> List[Dict[str,
             List[Dict]: List of file information dictionaries
         """
         try:
-            path = Path(directory)
+            # Validate directory path
+            safe_dir = FileTools._validate_path(directory)
+            path = Path(safe_dir)
             if pattern:
                 files = path.glob(pattern)
             else:
@@ -114,7 +146,9 @@ def get_file_info(filepath: str) -> Dict[str, Union[str, int]]:
             Dict: File information including size, dates, etc.
         """
         try:
-            path = Path(filepath)
+            # Validate file path
+            safe_path = FileTools._validate_path(filepath)
+            path = Path(safe_path)
             if not path.exists():
                 return {'error': f'File not found: {filepath}'}
 
@@ -149,9 +183,12 @@ def copy_file(src: str, dst: str) -> bool:
             bool: True if successful, False otherwise
         """
         try:
+            # Validate paths to prevent traversal attacks
+            safe_src = FileTools._validate_path(src)
+            safe_dst = FileTools._validate_path(dst)
             # Create destination directory if it doesn't exist
-            os.makedirs(os.path.dirname(dst), exist_ok=True)
-            shutil.copy2(src, dst)
+            os.makedirs(os.path.dirname(safe_dst), exist_ok=True)
+            shutil.copy2(safe_src, safe_dst)
             return True
         except Exception as e:
             error_msg = f"Error copying file from {src} to {dst}: {str(e)}"
@@ -172,9 +209,12 @@ def move_file(src: str, dst: str) -> bool:
             bool: True if successful, False otherwise
         """
         try:
+            # Validate paths to prevent traversal attacks
+            safe_src = FileTools._validate_path(src)
+            safe_dst = FileTools._validate_path(dst)
             # Create destination directory if it doesn't exist
-            os.makedirs(os.path.dirname(dst), exist_ok=True)
-            shutil.move(src, dst)
+            os.makedirs(os.path.dirname(safe_dst), exist_ok=True)
+            shutil.move(safe_src, safe_dst)
             return True
         except Exception as e:
             error_msg = f"Error moving file from {src} to {dst}: {str(e)}"
@@ -194,7 +234,9 @@ def delete_file(filepath: str) -> bool:
             bool: True if successful, False otherwise
         """
         try:
-            os.remove(filepath)
+            # Validate path to prevent traversal attacks
+            safe_path = FileTools._validate_path(filepath)
+            os.remove(safe_path)
             return True
         except Exception as e:
             error_msg = f"Error deleting file {filepath}: {str(e)}"

src/praisonai-agents/praisonaiagents/tools/python_tools.py

@@ -46,20 +46,100 @@ def execute_code(
         timeout: int = 30,
         max_output_size: int = 10000
     ) -> Dict[str, Any]:
-        """Execute Python code safely."""
+        """Execute Python code safely with restricted builtins."""
         try:
-            # Set up execution environment
+            # Create safe builtins - restricted set of functions
+            safe_builtins = {
+                # Basic functions
+                'print': print,
+                'len': len,
+                'range': range,
+                'enumerate': enumerate,
+                'zip': zip,
+                'map': map,
+                'filter': filter,
+                'sum': sum,
+                'min': min,
+                'max': max,
+                'abs': abs,
+                'round': round,
+                'sorted': sorted,
+                'reversed': reversed,
+                'any': any,
+                'all': all,
+                # Type constructors
+                'int': int,
+                'float': float,
+                'str': str,
+                'bool': bool,
+                'list': list,
+                'tuple': tuple,
+                'dict': dict,
+                'set': set,
+                # Math functions
+                'pow': pow,
+                'divmod': divmod,
+                # Exceptions
+                'Exception': Exception,
+                'ValueError': ValueError,
+                'TypeError': TypeError,
+                'KeyError': KeyError,
+                'IndexError': IndexError,
+                'RuntimeError': RuntimeError,
+                # Other safe functions
+                'isinstance': isinstance,
+                'type': type,
+                'hasattr': hasattr,
+                'getattr': getattr,
+                'setattr': setattr,
+                'dir': dir,
+                'help': help,
+                # Disable dangerous functions
+                '__import__': None,
+                'eval': None,
+                'exec': None,
+                'compile': None,
+                'open': None,
+                'input': None,
+                'globals': None,
+                'locals': None,
+                'vars': None,
+            }
+            
+            # Set up execution environment with safe builtins
             if globals_dict is None:
-                globals_dict = {'__builtins__': __builtins__}
+                globals_dict = {'__builtins__': safe_builtins}
+            else:
+                # Override builtins in provided globals
+                globals_dict['__builtins__'] = safe_builtins
+                
             if locals_dict is None:
                 locals_dict = {}
 
+            # Security check: validate code doesn't contain dangerous patterns
+            dangerous_patterns = [
+                '__import__', 'import ', 'from ', 'exec', 'eval', 
+                'compile', 'open(', 'file(', 'input(', 'raw_input',
+                '__subclasses__', '__bases__', '__globals__', '__code__',
+                '__class__', 'globals(', 'locals(', 'vars('
+            ]
+            
+            code_lower = code.lower()
+            for pattern in dangerous_patterns:
+                if pattern.lower() in code_lower:
+                    return {
+                        'result': None,
+                        'stdout': '',
+                        'stderr': f'Security Error: Code contains restricted pattern: {pattern}',
+                        'success': False
+                    }
+            
             # Capture output
             stdout_buffer = io.StringIO()
             stderr_buffer = io.StringIO()
 
             try:
-                # Compile code
+                # Compile code with restricted mode
                 compiled_code = compile(code, '<string>', 'exec')
 
                 # Execute with output capture

src/praisonai-agents/praisonaiagents/tools/shell_tools.py

@@ -40,7 +40,6 @@ def execute_command(
         command: str,
         cwd: Optional[str] = None,
         timeout: int = 30,
-        shell: bool = False,
         env: Optional[Dict[str, str]] = None,
         max_output_size: int = 10000
     ) -> Dict[str, Union[str, int, bool]]:
@@ -50,22 +49,20 @@ def execute_command(
             command: Command to execute
             cwd: Working directory
             timeout: Maximum execution time in seconds
-            shell: Whether to run command in shell
             env: Environment variables
             max_output_size: Maximum output size in bytes
             
         Returns:
             Dictionary with execution results
         """
         try:
-            # Split command if not using shell
-            if not shell:
-                # Use shlex.split with appropriate posix flag
-                if platform.system() == 'Windows':
-                    # Use shlex with posix=False for Windows to handle quotes properly
-                    command = shlex.split(command, posix=False)
-                else:
-                    command = shlex.split(command)
+            # Always split command for safety (no shell execution)
+            # Use shlex.split with appropriate posix flag
+            if platform.system() == 'Windows':
+                # Use shlex with posix=False for Windows to handle quotes properly
+                command = shlex.split(command, posix=False)
+            else:
+                command = shlex.split(command)
 
             # Set up process environment
             process_env = os.environ.copy()
@@ -79,7 +76,7 @@ def execute_command(
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 cwd=cwd,
-                shell=shell,
+                shell=False,  # Always use shell=False for security
                 env=process_env,
                 text=True
             )