webber/webber-api/src/domains/tools/shell/bash_full.py

"""
Full bash command execution tool with controlled write capabilities.

Allows more operations than BashReadOnlyTool but still with safety controls.
"""
import asyncio
import shlex
from pathlib import Path

from src.domains.tools.base import BaseTool, ToolResult
from src.shared.logging import get_logger, logged

logger = get_logger(__name__)


# Commands allowed (includes write operations)
ALLOWED_COMMANDS = {
    # File inspection (read-only)
    "ls", "find", "cat", "head", "tail", "wc", "file", "stat",
    "tree", "du", "df",
    # Text processing
    "grep", "awk", "sed", "sort", "uniq", "cut", "tr",
    # Git (full operations)
    "git",
    # System info
    "pwd", "whoami", "hostname", "uname", "date", "env", "printenv",
    "which", "type", "echo",
    # Archive operations
    "tar", "unzip", "zipinfo",
    # Write operations (controlled)
    "mkdir", "touch", "cp", "mv", "rm",
    "chmod",
    # Python ecosystem
    "python", "python3", "pip", "pip3", "pytest", "mypy", "ruff",
    # Other dev tools
    "make", "cargo", "npm", "node", "tsc",
}

# Git subcommands (includes write operations)
ALLOWED_GIT_SUBCOMMANDS = {
    # Read-only
    "status", "log", "diff", "show", "branch", "tag",
    "remote", "config", "ls-files", "ls-tree",
    "rev-parse", "describe", "shortlog", "blame",
    # Write operations
    "add", "commit", "checkout", "switch", "restore",
    "merge", "rebase", "cherry-pick",
    "stash", "pull", "fetch", "init",
    "reset", "clean", "rm", "mv",
}

# Absolutely forbidden - too dangerous regardless of context
ABSOLUTELY_FORBIDDEN = [
    # Catastrophic deletes
    "rm -rf /",
    "rm -rf ~",
    "rm -rf .",
    "rm -rf *",
    # Privilege escalation
    "sudo ",
    "su ",
    "doas ",
    # Dangerous permissions
    "chmod 777",
    "chmod -R 777",
    "chown ",
    "chgrp ",
    # Disk operations
    "dd if=",
    "mkfs",
    "fdisk",
    "parted",
    # System control
    "shutdown",
    "reboot",
    "poweroff",
    "init ",
    "systemctl",
    # Fork bomb pattern
    ":(){ :|:& };:",
]

# Network commands are forbidden
NETWORK_FORBIDDEN = [
    "curl", "wget", "ssh", "scp", "rsync", "sftp", "ftp",
    "nc", "netcat", "telnet", "nmap", "ping",
]

# Dangerous rm flags
DANGEROUS_RM_FLAGS = {"-rf", "-fr", "-r -f", "-f -r", "--recursive --force"}


class BashTool(BaseTool):
    """
    Execute bash commands with controlled write capabilities.

    More permissive than BashReadOnlyTool but still with safety controls.
    """

    name = "bash"
    description = """Execute a bash command with write capabilities.

ALLOWED commands:
- File operations: ls, find, cat, head, tail, mkdir, touch, cp, mv
- Git (full): git add, git commit, git checkout, git merge, git pull, etc.
- Python: python, pip install, pytest, mypy, ruff
- Text processing: grep, awk, sed, sort, uniq
- System info: pwd, whoami, date, which

RESTRICTED:
- rm: Single files only, no -rf flag, must be in allowed paths
- mv/cp: Target must be in allowed paths
- chmod: Only safe modes (no 777)

FORBIDDEN (always blocked):
- sudo, su (privilege escalation)
- Network: curl, wget, ssh, scp, rsync
- Dangerous: chmod 777, rm -rf, dd, mkfs, shutdown

Args:
    command: The bash command to execute
    cwd: Working directory (default: current directory)
    timeout: Timeout in seconds (default: 60)

Returns:
    Command stdout on success, or error message.

Examples:
    - "mkdir -p src/new_module" - Create directory
    - "cp template.py src/new_file.py" - Copy file
    - "git add . && git commit -m 'feat: add feature'" - Git commit
    - "pip install requests" - Install package
    - "pytest tests/ -v" - Run tests
    - "rm src/old_file.py" - Remove single file
"""

    def __init__(
        self,
        allowed_paths: list[str] | None = None,
        default_timeout: int = 60,
        max_output_size: int = 50000
    ):
        """
        Initialize BashTool.

        Args:
            allowed_paths: Allowed working/target directories
            default_timeout: Default command timeout in seconds
            max_output_size: Maximum output size in characters
        """
        self.allowed_paths = allowed_paths or []
        self.default_timeout = default_timeout
        self.max_output_size = max_output_size

    @logged()
    async def execute(  # type: ignore[override]  # see BaseTool.execute
        self,
        command: str,
        cwd: str | None = None,
        timeout: int | None = None
    ) -> ToolResult:
        """
        Execute a bash command.

        Args:
            command: Command to execute
            cwd: Working directory
            timeout: Timeout in seconds

        Returns:
            ToolResult with command output or error
        """
        timeout = timeout or self.default_timeout
        working_dir = Path(cwd) if cwd else Path.cwd()

        # Validate working directory
        if not self._validate_path(working_dir, self.allowed_paths):
            return self._error(f"Working directory not allowed: {working_dir}")

        if not working_dir.exists():
            return self._error(f"Working directory not found: {working_dir}")

        # Security validation
        validation_error = self._validate_command(command, working_dir)
        if validation_error:
            return self._error(validation_error)

        try:
            proc = await asyncio.create_subprocess_shell(
                command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=str(working_dir)
            )

            stdout, stderr = await asyncio.wait_for(
                proc.communicate(),
                timeout=timeout
            )

            stdout_str = stdout.decode('utf-8', errors='replace')
            stderr_str = stderr.decode('utf-8', errors='replace')

            # Truncate if necessary
            truncated = False
            if len(stdout_str) > self.max_output_size:
                stdout_str = stdout_str[:self.max_output_size]
                truncated = True

            if proc.returncode != 0:
                # Command failed, return stderr
                error_msg = stderr_str or f"Command exited with code {proc.returncode}"
                return ToolResult(
                    success=False,
                    data=stdout_str if stdout_str else None,
                    error=error_msg,
                    truncated=truncated
                )

            # Success - combine stdout and stderr if both present
            output = stdout_str
            if stderr_str and not output:
                output = stderr_str

            return self._success(
                data=output,
                truncated=truncated,
                exit_code=proc.returncode
            )

        except TimeoutError:
            return self._error(f"Command timed out after {timeout} seconds")
        except Exception as e:
            logger.exception(f"Error executing command: {command}")
            return self._error(f"Error executing command: {e}")

    def _validate_command(self, command: str, working_dir: Path) -> str | None:
        """
        Validate command is safe to execute.

        Returns:
            Error message if invalid, None if valid
        """
        command_lower = command.lower()

        # Check absolutely forbidden patterns first
        for pattern in ABSOLUTELY_FORBIDDEN:
            if pattern.lower() in command_lower:
                return f"Command contains forbidden pattern: {pattern}"

        # Check network commands
        for net_cmd in NETWORK_FORBIDDEN:
            # Check as standalone command or with path
            if (f" {net_cmd} " in f" {command_lower} " or
                    command_lower.startswith(f"{net_cmd} ") or
                    command_lower == net_cmd or
                    f"/{net_cmd} " in command_lower or
                    f"/{net_cmd}" == command_lower[-len(net_cmd)-1:]):
                return f"Network command not allowed: {net_cmd}"

        # Split by command separators to validate each sub-command
        # Handle &&, ||, ; but also pipe |
        sub_commands = self._split_commands(command)

        for sub_cmd in sub_commands:
            sub_cmd = sub_cmd.strip()
            if not sub_cmd:
                continue

            error = self._validate_single_command(sub_cmd, working_dir)
            if error:
                return error

        return None

    def _split_commands(self, command: str) -> list[str]:
        """Split command by separators (&&, ||, ;) but not pipes."""
        # Simple split - could be improved with proper shell parsing
        result = []
        current = ""
        i = 0
        while i < len(command):
            if command[i:i+2] in ("&&", "||"):
                result.append(current)
                current = ""
                i += 2
            elif command[i] == ";":
                result.append(current)
                current = ""
                i += 1
            else:
                current += command[i]
                i += 1
        if current:
            result.append(current)
        return result

    def _validate_single_command(self, command: str, working_dir: Path) -> str | None:
        """Validate a single command (no &&, ||, ;)."""
        # Handle pipes - validate first command in pipe chain
        if "|" in command:
            command = command.split("|")[0].strip()

        # Parse command
        try:
            tokens = shlex.split(command)
            if not tokens:
                return None  # Empty is ok (could be whitespace)
        except ValueError as e:
            return f"Invalid command syntax: {e}"

        # Get base command
        base_cmd = Path(tokens[0]).name

        # Check if command is allowed
        if base_cmd not in ALLOWED_COMMANDS:
            return f"Command not allowed: {base_cmd}"

        # Special handling for specific commands
        if base_cmd == "git":
            return self._validate_git_command(tokens)
        elif base_cmd == "rm":
            return self._validate_rm_command(tokens, working_dir)
        elif base_cmd in ("cp", "mv"):
            return self._validate_copy_move_command(tokens, working_dir)
        elif base_cmd == "chmod":
            return self._validate_chmod_command(tokens)
        elif base_cmd in ("pip", "pip3"):
            return self._validate_pip_command(tokens)

        return None

    def _validate_git_command(self, tokens: list[str]) -> str | None:
        """Validate git command."""
        if len(tokens) < 2:
            return "Git command requires a subcommand"

        git_subcommand = tokens[1]

        # Handle git with flags before subcommand (e.g., git -C path status)
        if git_subcommand.startswith("-"):
            # Find the actual subcommand
            for _i, token in enumerate(tokens[2:], 2):
                if not token.startswith("-"):
                    git_subcommand = token
                    break
            else:
                return "Git command requires a subcommand"

        if git_subcommand not in ALLOWED_GIT_SUBCOMMANDS:
            return f"Git subcommand not allowed: {git_subcommand}"

        # Block git push (could push to remote)
        if git_subcommand == "push":
            return "git push is not allowed (use manually)"

        return None

    def _validate_rm_command(self, tokens: list[str], working_dir: Path) -> str | None:
        """Validate rm command - only single files, no -rf."""
        # Check for dangerous flags
        flags = [t for t in tokens[1:] if t.startswith("-")]
        for flag in flags:
            if "r" in flag and "f" in flag:
                return "rm -rf is not allowed"
            if flag in DANGEROUS_RM_FLAGS:
                return f"rm flag not allowed: {flag}"

        # Must have at least one non-flag argument
        args = [t for t in tokens[1:] if not t.startswith("-")]
        if not args:
            return "rm requires a file argument"

        # Validate each path
        for arg in args:
            path = Path(arg)
            if not path.is_absolute():
                path = working_dir / path

            if not self._validate_path(path, self.allowed_paths):
                return f"rm target not in allowed paths: {arg}"

        return None

    def _validate_copy_move_command(
        self, tokens: list[str], working_dir: Path
    ) -> str | None:
        """Validate cp/mv command - target must be in allowed paths."""
        # Get non-flag arguments
        args = [t for t in tokens[1:] if not t.startswith("-")]

        if len(args) < 2:
            return None  # Let the command fail naturally

        # Last argument is typically the destination
        dest = args[-1]
        dest_path = Path(dest)
        if not dest_path.is_absolute():
            dest_path = working_dir / dest_path

        if not self._validate_path(dest_path, self.allowed_paths):
            return f"Copy/move destination not in allowed paths: {dest}"

        return None

    def _validate_chmod_command(self, tokens: list[str]) -> str | None:
        """Validate chmod command - block dangerous modes."""
        for token in tokens[1:]:
            if token.startswith("-"):
                continue
            # Block 777 and similar
            if "777" in token or "666" in token:
                return f"chmod mode not allowed: {token}"

        return None

    def _validate_pip_command(self, tokens: list[str]) -> str | None:
        """Validate pip command - allow install, block uninstall of system packages."""
        if len(tokens) < 2:
            return None

        subcommand = tokens[1]

        # Allow help, list, show, freeze, check
        allowed_pip = {"install", "list", "show", "freeze", "check", "help", "--help", "-h"}

        if subcommand not in allowed_pip:
            return f"pip subcommand not allowed: {subcommand}"

        return None