command_safety.py

"""
============================================================================
 COMMAND SAFETY — 4-Tier Terminal Command Classification
============================================================================
 Ported from agent-swarm's command_executor.py.
 Classifies shell commands by danger level before execution.

 Tiers:
   ✅ SAFE       — Auto-approve (ls, cat, git status)
   🟡 MODERATE   — Auto-approve + log (npm install, git commit)
   🔴 DANGEROUS  — Require user approval (rm, deploy, git push)
   🚫 BLOCKED    — Never allowed (rm -rf /, shutdown, fork bomb)
============================================================================
"""

import json
import subprocess
import time
from pathlib import Path
from datetime import datetime
from typing import Optional


# ==========================================================================
# SAFETY CLASSIFICATION LISTS
# ==========================================================================

SAFE_COMMANDS = [
    # Read-only operations
    "ls", "cat", "head", "tail", "less", "more", "wc", "file",
    "find", "grep", "rg", "ag", "ack", "which", "whereis", "type",
    "pwd", "whoami", "date", "echo", "printf",
    # Git read-only
    "git status", "git log", "git diff", "git show", "git branch",
    "git remote", "git stash list", "git reflog",
    # Node/npm read-only
    "npm list", "npm ls", "npm outdated", "npm audit",
    "node --version", "npm --version", "npx --version",
    # Python read-only
    "python --version", "python3 --version", "pip list", "pip show",
    "pip freeze",
    # Docker read-only
    "docker ps", "docker images", "docker logs", "docker stats",
    # System info
    "uname", "df", "du", "free", "uptime", "top", "ps",
    # File content
    "jq", "sort", "uniq", "cut", "awk", "sed", "tr",
]

MODERATE_COMMANDS = [
    # Git write operations
    "git add", "git commit", "git checkout", "git switch",
    "git merge", "git rebase", "git stash", "git tag",
    "git fetch", "git pull",
    # Package management
    "npm install", "npm i", "npm ci", "npm update",
    "npm uninstall", "npm run", "npm test", "npm build",
    "pip install", "pip uninstall", "pip upgrade",
    "yarn", "pnpm",
    # File creation/modification
    "touch", "mkdir", "cp", "mv",
    # Docker operations
    "docker build", "docker run", "docker stop", "docker start",
    "docker-compose up", "docker-compose down",
    # Make
    "make", "cmake",
    # Build tools
    "tsc", "vite", "webpack", "rollup", "esbuild",
    "cargo build", "cargo test",
    "go build", "go test",
]

DANGEROUS_COMMANDS = [
    # Destructive file operations
    "rm ", "rm -", "rmdir",
    # Git dangerous operations
    "git push", "git push --force", "git reset --hard",
    "git clean -fd", "git branch -D",
    # Deployment
    "fly deploy", "vercel deploy", "netlify deploy",
    "aws deploy", "gcloud deploy", "kubectl apply",
    # System modifications
    "chmod", "chown", "sudo",
    "systemctl", "service",
    # Network
    "curl -X POST", "curl -X PUT", "curl -X DELETE",
    "wget",
]

BLOCKED_COMMANDS = [
    # Absolutely never
    "rm -rf /", "rm -rf /*",
    "dd if=", "mkfs",
    "shutdown", "reboot", "halt", "poweroff",
    "passwd", "userdel", "useradd",
    "> /dev/sda", "format",
    ":(){:|:&};:",  # fork bomb
]


# ==========================================================================
# SAFETY LEVELS
# ==========================================================================

class SafetyLevel:
    SAFE = "safe"
    MODERATE = "moderate"
    DANGEROUS = "dangerous"
    BLOCKED = "blocked"


# ==========================================================================
# COMMAND RESULT
# ==========================================================================

class CommandResult:
    """Result of a command execution with safety metadata."""

    def __init__(self, command: str, stdout: str, stderr: str,
                 returncode: int, safety: str, approved: bool, duration: float):
        self.command = command
        self.stdout = stdout
        self.stderr = stderr
        self.returncode = returncode
        self.safety = safety
        self.approved = approved
        self.duration = duration
        self.timestamp = datetime.now().isoformat()

    @property
    def success(self) -> bool:
        return self.returncode == 0

    def to_dict(self) -> dict:
        return {
            "command": self.command,
            "stdout": self.stdout[:2000],
            "stderr": self.stderr[:500],
            "returncode": self.returncode,
            "safety": self.safety,
            "approved": self.approved,
            "duration": round(self.duration, 2),
            "timestamp": self.timestamp,
            "success": self.success,
        }


# ==========================================================================
# COMMAND EXECUTOR
# ==========================================================================

class CommandExecutor:
    """
    Execute CLI commands with 4-tier safety classification.

    In swarm mode, agents may need to run shell commands to build projects.
    This executor classifies each command and blocks dangerous ones unless
    explicitly approved.
    """

    def __init__(self, cwd: str = ".", auto_approve: list = None,
                 approval_callback=None, log_dir: str = None):
        self.cwd = Path(cwd)
        self.auto_approve = auto_approve or [SafetyLevel.SAFE, SafetyLevel.MODERATE]
        self.approval_callback = approval_callback
        self.command_log: list[dict] = []
        self._log_dir = Path(log_dir) if log_dir else None

    def classify(self, command: str) -> str:
        """Classify a command's safety level."""
        cmd_lower = command.lower().strip()

        # Check blocked first (highest priority)
        for blocked in BLOCKED_COMMANDS:
            if blocked.lower() in cmd_lower:
                return SafetyLevel.BLOCKED

        # Check dangerous
        for dangerous in DANGEROUS_COMMANDS:
            if cmd_lower.startswith(dangerous.lower()):
                return SafetyLevel.DANGEROUS

        # Check moderate
        for moderate in MODERATE_COMMANDS:
            if cmd_lower.startswith(moderate.lower()):
                return SafetyLevel.MODERATE

        # Check safe
        for safe in SAFE_COMMANDS:
            if cmd_lower.startswith(safe.lower()):
                return SafetyLevel.SAFE

        # Unknown commands default to moderate
        return SafetyLevel.MODERATE

    def execute(self, command: str, timeout: int = 120,
                cwd: Optional[str] = None) -> CommandResult:
        """Execute a command with safety checks."""
        safety = self.classify(command)
        working_dir = Path(cwd) if cwd else self.cwd

        # Blocked — never execute
        if safety == SafetyLevel.BLOCKED:
            result = CommandResult(
                command=command, stdout="", returncode=-1,
                stderr=f"🚫 BLOCKED: Command not allowed: {command}",
                safety=safety, approved=False, duration=0,
            )
            self._log(result)
            return result

        # Check approval
        approved = True
        if safety not in self.auto_approve:
            if self.approval_callback:
                approved = self.approval_callback(command, safety)
            elif safety == SafetyLevel.DANGEROUS:
                result = CommandResult(
                    command=command, stdout="", returncode=-1,
                    stderr=f"🔴 APPROVAL REQUIRED: {command}",
                    safety=safety, approved=False, duration=0,
                )
                self._log(result)
                return result

        if not approved:
            result = CommandResult(
                command=command, stdout="", returncode=-1,
                stderr="Command rejected by approval callback",
                safety=safety, approved=False, duration=0,
            )
            self._log(result)
            return result

        # Execute
        start_time = time.time()
        try:
            proc = subprocess.run(
                command, shell=True, capture_output=True, text=True,
                timeout=timeout, cwd=str(working_dir),
            )
            duration = time.time() - start_time
            result = CommandResult(
                command=command, stdout=proc.stdout, stderr=proc.stderr,
                returncode=proc.returncode, safety=safety,
                approved=approved, duration=duration,
            )
        except subprocess.TimeoutExpired:
            duration = time.time() - start_time
            result = CommandResult(
                command=command, stdout="", returncode=124,
                stderr=f"Command timed out after {timeout}s",
                safety=safety, approved=approved, duration=duration,
            )
        except Exception as e:
            duration = time.time() - start_time
            result = CommandResult(
                command=command, stdout="", returncode=1,
                stderr=str(e), safety=safety, approved=approved, duration=duration,
            )

        self._log(result)
        return result

    def _log(self, result: CommandResult) -> None:
        """Log command execution."""
        self.command_log.append(result.to_dict())

        if self._log_dir:
            log_file = self._log_dir / "command_log.json"
            log_file.parent.mkdir(parents=True, exist_ok=True)

            existing = []
            if log_file.exists():
                try:
                    existing = json.loads(log_file.read_text())
                except Exception:
                    existing = []

            existing.append(result.to_dict())
            # Keep last 500 entries
            if len(existing) > 500:
                existing = existing[-500:]

            log_file.write_text(json.dumps(existing, indent=2))