ohmu/third/server/openwebui/mcp-servers/terminal_server.py

import os
import subprocess
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("terminal", host="0.0.0.0", port=8011)

# Allowed directories for file operations
ALLOWED_DIRS = ["/projects", "/data", "/scripts", "/backups", "/tmp"]

ALLOWED_PREFIXES = [
    "ls", "cat", "head", "tail", "wc", "find", "grep", "rg", "which",
    "pwd", "whoami", "id", "date", "uname", "df", "du", "ps",
    "free", "uptime", "env", "echo", "sort", "uniq", "cut", "tr",
    "diff", "cmp", "stat", "file", "tree",
    "mkdir", "touch", "npm", "npx", "node", "python3", "ng",
    "docker ps", "docker images", "docker compose ps",
    "docker inspect", "docker logs",
    "curl", "ping", "traceroute", "nslookup", "dig", "host",
]

BLOCKED = [
    "rm ", "dd ", "mkfs.", "shutdown", "reboot", "halt",
    "sudo", "su", "chmod", "chown", "passwd", "kill ",
    "|", "&&", "||", "`", "$(",
    "docker rm", "docker rmi", "docker stop", "docker kill",
    "docker compose down", "docker compose rm",
]


def _is_path_allowed(path: str) -> bool:
    resolved = Path(path).resolve()
    for allowed in ALLOWED_DIRS:
        allowed_path = Path(allowed).resolve()
        if allowed_path in resolved.parents or allowed_path == resolved:
            return True
    return False


@mcp.tool()
def run_command(command: str, timeout: int = 60) -> str:
    """Execute a shell command.

    Allowed: development commands (npm, npx, node, python3, ng, mkdir, touch),
    read-only commands (ls, cat, head, tail, grep, find, etc.), and docker inspect.
    File redirect '>' is allowed only when writing to /projects/ paths.
    """
    if not command or len(command) > 4000:
        return "Error: command is empty or too long"
    lowered = command.strip().lower()
    # Check for blocked patterns first
    for b in BLOCKED:
        if b in lowered:
            return f"Error: blocked pattern: {b}"
    # For redirects, check allowed path
    if ">" in lowered or ">>" in lowered:
        # Extract the redirect target path (last argument after >)
        parts = command.split()
        for i, p in enumerate(parts):
            if p in (">", ">>"):
                target = parts[i + 1] if i + 1 < len(parts) else ""
                if target and not _is_path_allowed(target):
                    return f"Error: redirect target not in allowed directories: {target}"
        # Allow redirect commands without checking prefix
    else:
        ok = any(lowered.startswith(p) for p in ALLOWED_PREFIXES)
        if not ok:
            return "Error: command not in allowed list"
    try:
        r = subprocess.run(
            ["sh", "-c", command],
            capture_output=True, text=True, timeout=min(timeout, 120),
        )
        out = r.stdout or ""
        if r.stderr:
            out += "\n--- stderr ---\n" + r.stderr
        if r.returncode != 0:
            out += f"\n(exit code: {r.returncode})"
        # Truncate output
        if len(out) > 50000:
            out = out[:50000] + "\n... (output truncated)"
        return out or "(no output)"
    except subprocess.TimeoutExpired:
        return f"Error: timed out after {timeout}s"
    except Exception as e:
        return f"Error: {e}"


@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write or overwrite a file with the given content. Creates parent directories.
    Only allowed within: /projects, /data, /scripts, /backups, /tmp
    """
    if not _is_path_allowed(path):
        return f"Error: path not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        p = Path(path).resolve()
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content)
        return f"OK - wrote {len(content)} bytes to {path}"
    except Exception as e:
        return f"Error writing file: {e}"


@mcp.tool()
def create_directory(path: str) -> str:
    """Create a directory and all parent directories if they do not exist.
    Only allowed within: /projects, /data, /scripts, /backups, /tmp
    """
    if not _is_path_allowed(path):
        return f"Error: path not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        p = Path(path).resolve()
        p.mkdir(parents=True, exist_ok=True)
        return f"OK - created directory {path}"
    except Exception as e:
        return f"Error creating directory: {e}"


if __name__ == "__main__":
    mcp.run(transport="stdio")