# File 7: tools/shell_tools.py with open(f"{base_dir}/tools/shell_tools.py", "w") as f: f.write('''"""Shell and Python Execution Tools""" import os import sys import subprocess import tempfile import json from typing import Dict, List, Optional from config import Config from utils import UI, Logger, truncate class ShellTools: """Shell command and Python code execution with safety controls""" def __init__(self): self.logger = Logger() self.blocked_patterns = [ "rm -rf /", "rm -rf /*", ":(){ :|:& };:", "> /dev/sda", "mkfs", "dd if=/dev/zero", "chmod -R 777 /", "wget -O- | sh", "curl | sh", "curl | bash" ] def _is_safe(self, command: str) -> tuple: """Check if command is safe to run""" cmd_lower = command.lower().strip() for pattern in self.blocked_patterns: if pattern.lower() in cmd_lower: return False, f"Blocked dangerous pattern: {pattern}" return True, "OK" def bash_command(self, command: str, cwd: str = "", timeout: int = 30, env_vars: Dict[str, str] = None, capture_output: bool = True, confirm: bool = True) -> Dict: """Execute shell command with safety controls""" try: safe, reason = self._is_safe(command) if not safe: return {"success": False, "error": reason} if cwd: work_dir = os.path.join(Config.SANDBOX_PATH, cwd) else: work_dir = Config.SANDBOX_PATH os.makedirs(work_dir, exist_ok=True) destructive_keywords = ["rm", "del", "remove", "delete", "drop", "truncate", "format", "mkfs", "dd", ">/dev"] needs_confirm = any(kw in command.lower() for kw in destructive_keywords) if needs_confirm and confirm: if not UI.confirm(f"Run destructive command: {command}?"): return {"success": False, "error": "User cancelled"} env = os.environ.copy() if env_vars: env.update(env_vars) result = subprocess.run( command, shell=True, cwd=work_dir, capture_output=capture_output, text=True, timeout=timeout, env=env ) self.logger.log("bash_command", { "command": command, "returncode": result.returncode, "cwd": cwd }) return { "success": result.returncode == 0, "command": command, "returncode": result.returncode, "stdout": truncate(result.stdout) if capture_output else "", "stderr": truncate(result.stderr) if capture_output else "", "cwd": cwd } except subprocess.TimeoutExpired: return {"success": False, "error": f"Command timed out after {timeout}s"} except Exception as e: return {"success": False, "error": str(e)} def python_execute(self, code: str, timeout: int = 30, packages: List[str] = None, save_output: bool = False) -> Dict: """Execute Python code in isolated environment""" try: if packages: for pkg in packages: subprocess.run( [sys.executable, "-m", "pip", "install", "-q", pkg], capture_output=True, timeout=60 ) with tempfile.NamedTemporaryFile(mode="w", suffix=".py", dir=Config.SANDBOX_PATH, delete=False) as f: f.write(code) temp_file = f.name env = os.environ.copy() env["PYTHONPATH"] = Config.SANDBOX_PATH result = subprocess.run( [sys.executable, temp_file], capture_output=True, text=True, timeout=timeout, cwd=Config.SANDBOX_PATH, env=env ) try: os.remove(temp_file) except: pass output = result.stdout if result.stderr: output += "\n[STDERR]\n" + result.stderr output_file = "" if save_output and output: output_file = f"python_output_{os.path.basename(temp_file)}.txt" output_path = os.path.join(Config.SANDBOX_PATH, output_file) with open(output_path, "w") as f: f.write(output) self.logger.log("python_execute", { "code_length": len(code), "returncode": result.returncode }) return { "success": result.returncode == 0, "output": truncate(output, 2000), "returncode": result.returncode, "output_file": output_file if save_output else "", "code_executed": code[:200] + "..." if len(code) > 200 else code } except subprocess.TimeoutExpired: return {"success": False, "error": f"Code execution timed out after {timeout}s"} except Exception as e: return {"success": False, "error": str(e)} def process_manager(self, action: str = "list", pid: int = None, command: str = "", name_filter: str = "") -> Dict: """Manage system processes""" try: if action == "list": if os.name == "nt": cmd = "tasklist /fo csv" else: cmd = "ps aux" result = subprocess.run(cmd, shell=True, capture_output=True, text=True) lines = result.stdout.strip().split("\n") processes = [] for line in lines[1:]: parts = line.split() if len(parts) >= 2: proc_name = " ".join(parts[10:]) if len(parts) > 10 else parts[-1] if name_filter and name_filter.lower() not in proc_name.lower(): continue processes.append({ "pid": parts[1] if len(parts) > 1 else "", "cpu": parts[2] if len(parts) > 2 else "", "mem": parts[3] if len(parts) > 3 else "", "command": proc_name[:50] }) return { "success": True, "action": "list", "processes": processes[:50], "total": len(processes) } elif action == "kill": if not pid: return {"success": False, "error": "PID required for kill action"} if not UI.confirm(f"Kill process {pid}?"): return {"success": False, "error": "User cancelled"} if os.name == "nt": result = subprocess.run(f"taskkill /PID {pid} /F", shell=True, capture_output=True, text=True) else: result = subprocess.run(f"kill -9 {pid}", shell=True, capture_output=True, text=True) return { "success": result.returncode == 0, "action": "kill", "pid": pid, "output": result.stdout or result.stderr } elif action == "start": if not command: return {"success": False, "error": "Command required for start action"} process = subprocess.Popen( command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=Config.SANDBOX_PATH ) return { "success": True, "action": "start", "pid": process.pid, "command": command } else: return {"success": False, "error": f"Unknown action: {action}"} except Exception as e: return {"success": False, "error": str(e)} def system_info(self) -> Dict: """Get comprehensive system information""" try: info = { "success": True, "platform": sys.platform, "python_version": sys.version, "cpu_count": os.cpu_count(), "cwd": os.getcwd(), "sandbox": Config.SANDBOX_PATH } if hasattr(os, "statvfs"): stat = os.statvfs(Config.SANDBOX_PATH) info["disk_total"] = f"{stat.f_frsize * stat.f_blocks / (1024**3):.2f} GB" info["disk_free"] = f"{stat.f_frsize * stat.f_bavail / (1024**3):.2f} GB" try: with open("/proc/meminfo", "r") as f: meminfo = f.read() for line in meminfo.split("\n"): if "MemTotal" in line: info["memory_total"] = line.split(":")[1].strip() elif "MemAvailable" in line: info["memory_available"] = line.split(":")[1].strip() except: pass self.logger.log("system_info", info) return info except Exception as e: return {"success": False, "error": str(e)} def environment_vars(self, action: str = "list", key: str = "", value: str = "") -> Dict: """Manage environment variables""" try: if action == "list": vars_dict = dict(os.environ) sensitive = ["key", "token", "secret", "password", "api"] filtered = {k: "***" if any(s in k.lower() for s in sensitive) else v for k, v in vars_dict.items()} return { "success": True, "action": "list", "variables": filtered, "count": len(filtered) } elif action == "get": if not key: return {"success": False, "error": "Key required"} return { "success": True, "action": "get", "key": key, "value": os.environ.get(key, "(not set)") } elif action == "set": if not key: return {"success": False, "error": "Key required"} os.environ[key] = value return { "success": True, "action": "set", "key": key, "value": value } elif action == "delete": if not key: return {"success": False, "error": "Key required"} if key in os.environ: del os.environ[key] return { "success": True, "action": "delete", "key": key } else: return {"success": False, "error": f"Unknown action: {action}"} except Exception as e: return {"success": False, "error": str(e)} def package_manager(self, action: str = "list", package: str = "", manager: str = "pip") -> Dict: """Manage packages""" try: commands = { "pip": { "list": "pip list --format=json", "install": f"pip install {package}", "uninstall": f"pip uninstall -y {package}", "update": f"pip install --upgrade {package}", "search": f"pip search {package} 2>/dev/null || pip index versions {package} 2>/dev/null || echo \'Search not available\'" }, "npm": { "list": "npm list --depth=0", "install": f"npm install {package}", "uninstall": f"npm uninstall {package}", "update": f"npm update {package}", "search": f"npm search {package}" } } if manager not in commands: return {"success": False, "error": f"Unsupported package manager: {manager}"} cmd = commands[manager].get(action) if not cmd: return {"success": False, "error": f"Unknown action: {action}"} if action in ["install", "uninstall"] and not UI.confirm(f"{action} {package} via {manager}?"): return {"success": False, "error": "User cancelled"} result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=120 ) output = result.stdout if action == "list" and manager == "pip": try: packages = json.loads(output) output = packages except: pass return { "success": result.returncode == 0, "action": action, "manager": manager, "package": package, "output": output[:2000] if isinstance(output, str) else output, "stderr": result.stderr[:500] if result.stderr else "" } except Exception as e: return {"success": False, "error": str(e)} ''') print("tools/shell_tools.py done")