Nash MCP Server

import subprocess import traceback import logging import os import signal import sys import atexit # Store active subprocesses active_processes = [] # Cleanup function def cleanup_subprocesses(): for proc in active_processes: try: if proc.poll() is None: # If process is still running os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except Exception: pass # Register cleanup on exit atexit.register(cleanup_subprocesses) # For signal handling def signal_handler(sig, frame): cleanup_subprocesses() sys.exit(0) # Register signal handlers signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) def execute_command(cmd: str) -> str: """ Execute an arbitrary shell command and return its output. ⚠️ MANDATORY PRE-EXECUTION CHECKLIST: ⚠️ STOP! Before running ANY command, have you completed these REQUIRED checks? 1. Check available packages: list_installed_packages() - Know what tools are available in the environment 2. Check available secrets: nash_secrets() - See what API keys and credentials are available - Don't run commands requiring credentials you don't have 3. Check existing files: list_session_files() - See what code already exists that might help These steps are MANDATORY. Skipping them can lead to wasted effort. This function runs shell commands with full access to your system. Use with caution and follow security best practices. Examples: execute_command("ls -la ~") execute_command("find /usr/local -name 'python*'") execute_command("grep -r 'TODO' ./src") Security considerations: - Uses shell=True which means shell metacharacters are interpreted - Never run commands that could damage your system (rm -rf, etc.) - Avoid commands that might expose sensitive information - Consider using safer alternatives for file operations WHEN TO USE: - For quick system information gathering (file listings, processes, etc.) - When specific shell utilities provide the most efficient solution - For running installed command-line tools not accessible via Python - When file/directory operations are simpler with shell commands Consider using execute_python() instead when: - Complex data processing is required - Error handling needs to be more robust - The operation involves multiple steps or conditional logic Args: cmd: Shell command to execute (string) Returns: Command output (stdout) if successful Detailed error information with exit code and stderr if command fails Exception details with traceback if execution fails """ # Log the full command but still sanitize potentially sensitive parts for security # Split the command on spaces to identify likely arguments cmd_parts = cmd.split() # For logging, replace potentially sensitive arguments (after flags) # This is a simple sanitization that focuses on common sensitivity patterns safe_parts = [] for i, part in enumerate(cmd_parts): # Check for sensitive patterns # If it might be a sensitive argument (like an API key or password) after a flag if (i > 0 and cmd_parts[i-1].startswith('-') and ('key' in cmd_parts[i-1].lower() or 'token' in cmd_parts[i-1].lower() or 'pass' in cmd_parts[i-1].lower() or 'secret' in cmd_parts[i-1].lower())): # Replace with [REDACTED] to avoid logging sensitive info safe_parts.append('[REDACTED]') # Otherwise keep the original part else: safe_parts.append(part) safe_cmd = ' '.join(safe_parts) # Log the full command (but with sensitive parts redacted) logging.info(f"Executing command: {safe_cmd}") try: # Always use shell=True for simplicity and tilde expansion # Use preexec_fn=os.setpgrp to make this process the group leader proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=True, preexec_fn=os.setpgrp # Make this process the group leader ) # Add to active processes list for cleanup active_processes.append(proc) try: stdout, stderr = proc.communicate() if proc.returncode == 0: logging.info(f"Command executed successfully (exit code 0)") return stdout if stdout.strip() else "Command executed (no output)." else: logging.warning(f"Command failed with exit code {proc.returncode}: {safe_cmd}") logging.debug(f"Command stderr: {stderr}") return (f"Command failed (exit code {proc.returncode}).\n" f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}") finally: # Remove from active processes list if proc in active_processes: active_processes.remove(proc) except Exception as e: logging.error(f"Exception while executing command: {str(e)}") logging.error(traceback.format_exc()) return f"Exception: {e}\nTraceback:\n{traceback.format_exc()}"