Create initial Linux MCP server

app.py

@@ -0,0 +1,526 @@
+import os
+import asyncio
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.responses import StreamingResponse
+import json
+from mcp.server import Server
+from mcp.types import Tool, TextContent
+
+# Create FastAPI app
+app = FastAPI(title="Linux MCP Server")
+
+# Create MCP server instance
+mcp_app = Server("linux-mcp")
+
+# ============================================================================
+# LINUX MCP HANDLER FUNCTIONS
+# ============================================================================
+
+def handle_user_management(args):
+    """Generate Linux user management commands"""
+    action = args.get("action")
+    username = args.get("username", "")
+    groupname = args.get("groupname", "")
+    use_sudo = args.get("sudo", True)
+    sudo_prefix = "sudo " if use_sudo else ""
+    
+    commands = []
+    
+    if action == "add_user_to_group":
+        if not username or not groupname:
+            return [TextContent(type="text", text="❌ Username and groupname required for add_user_to_group")]
+        
+        commands = [
+            f"# Add user '{username}' to group '{groupname}'",
+            f"{sudo_prefix}usermod -a -G {groupname} {username}",
+            "",
+            "# Alternative methods:",
+            f"{sudo_prefix}gpasswd -a {username} {groupname}",
+            f"{sudo_prefix}adduser {username} {groupname}  # Ubuntu/Debian",
+            "",
+            "# Verify group membership:",
+            f"groups {username}",
+            f"id {username}",
+            f"getent group {groupname}"
+        ]
+    
+    elif action == "create_user":
+        if not username:
+            return [TextContent(type="text", text="❌ Username required for create_user")]
+        
+        commands = [
+            f"# Create new user '{username}'",
+            f"{sudo_prefix}useradd -m -s /bin/bash {username}",
+            f"{sudo_prefix}passwd {username}",
+            "",
+            "# Create user with specific group:",
+            f"{sudo_prefix}useradd -m -g {groupname if groupname else 'users'} -s /bin/bash {username}",
+            "",
+            "# Add to sudo group (Ubuntu/Debian):",
+            f"{sudo_prefix}usermod -a -G sudo {username}",
+            "",
+            "# Add to wheel group (RHEL/CentOS):",
+            f"{sudo_prefix}usermod -a -G wheel {username}",
+            "",
+            "# Verify user creation:",
+            f"id {username}",
+            f"getent passwd {username}"
+        ]
+    
+    elif action == "delete_user":
+        if not username:
+            return [TextContent(type="text", text="❌ Username required for delete_user")]
+        
+        commands = [
+            f"# Delete user '{username}'",
+            f"{sudo_prefix}userdel {username}",
+            "",
+            "# Delete user and home directory:",
+            f"{sudo_prefix}userdel -r {username}",
+            "",
+            "# Remove user from all groups:",
+            f"{sudo_prefix}gpasswd -d {username} groupname",
+            "",
+            "# Verify user deletion:",
+            f"getent passwd {username}",
+            "# Should return no results if deleted successfully"
+        ]
+    
+    elif action == "list_groups":
+        commands = [
+            "# List all groups:",
+            "cat /etc/group",
+            "getent group",
+            "",
+            "# List groups for specific user:",
+            f"groups {username}" if username else "groups $USER",
+            f"id {username}" if username else "id $USER",
+            "",
+            "# List users in specific group:",
+            f"getent group {groupname}" if groupname else "getent group sudo",
+            "",
+            "# List all users:",
+            "cut -d: -f1 /etc/passwd",
+            "getent passwd"
+        ]
+    
+    result = "\n".join(commands)
+    return [TextContent(type="text", text=result)]
+
+def handle_file_permissions(args):
+    """Generate file permission and ownership commands"""
+    action = args.get("action")
+    path = args.get("path")
+    permissions = args.get("permissions", "755")
+    owner = args.get("owner", "")
+    recursive = args.get("recursive", False)
+    
+    if not path:
+        return [TextContent(type="text", text="❌ Path is required")]
+    
+    commands = []
+    recursive_flag = "-R " if recursive else ""
+    
+    if action == "chmod":
+        commands = [
+            f"# Change permissions for {path}",
+            f"chmod {recursive_flag}{permissions} {path}",
+            "",
+            "# Common permission examples:",
+            "# 755 = rwxr-xr-x (executable for owner, readable for others)",
+            "# 644 = rw-r--r-- (readable/writable for owner, readable for others)", 
+            "# 600 = rw------- (readable/writable for owner only)",
+            "# 777 = rwxrwxrwx (full permissions for all - use with caution!)",
+            "",
+            "# Symbolic notation examples:",
+            f"chmod {recursive_flag}u+x {path}    # Add execute for owner",
+            f"chmod {recursive_flag}g-w {path}    # Remove write for group",
+            f"chmod {recursive_flag}o-r {path}    # Remove read for others",
+            "",
+            "# Verify permissions:",
+            f"ls -la {path}",
+            f"stat {path}"
+        ]
+    
+    elif action == "chown":
+        if not owner:
+            return [TextContent(type="text", text="❌ Owner is required for chown")]
+        
+        commands = [
+            f"# Change ownership for {path}",
+            f"sudo chown {recursive_flag}{owner} {path}",
+            "",
+            "# Change owner and group:",
+            f"sudo chown {recursive_flag}{owner}:{owner} {path}",
+            f"sudo chown {recursive_flag}{owner}:users {path}",
+            "",
+            "# Change only group:",
+            f"sudo chgrp {recursive_flag}{owner} {path}",
+            "",
+            "# Verify ownership:",
+            f"ls -la {path}",
+            f"stat {path}"
+        ]
+    
+    elif action == "find_permissions":
+        commands = [
+            f"# Find files with specific permissions in {path}",
+            f"find {path} -type f -perm {permissions}",
+            f"find {path} -type f -perm -{permissions}  # At least these permissions",
+            "",
+            "# Find files by owner:",
+            f"find {path} -user {owner}" if owner else f"find {path} -user $USER",
+            f"find {path} -group {owner}" if owner else f"find {path} -group users",
+            "",
+            "# Find files with dangerous permissions:",
+            f"find {path} -type f -perm -002  # World-writable files",
+            f"find {path} -type f -perm -004  # World-readable files",
+            f"find {path} -type f -perm -006  # World-readable and writable",
+            "",
+            "# Find SUID/SGID files:",
+            f"find {path} -type f -perm -4000  # SUID files",
+            f"find {path} -type f -perm -2000  # SGID files",
+            "",
+            "# Find directories with specific permissions:",
+            f"find {path} -type d -perm 777   # World-writable directories"
+        ]
+    
+    result = "\n".join(commands)
+    return [TextContent(type="text", text=result)]
+
+def handle_system_commands(args):
+    """Generate Linux system administration commands"""
+    category = args.get("category")
+    action = args.get("action", "")
+    target = args.get("target", "")
+    
+    commands = []
+    
+    if category == "process":
+        commands = [
+            "# Process Management Commands",
+            "ps aux                    # List all processes with details",
+            "ps -ef                    # Alternative process listing",
+            "pstree                    # Show process tree",
+            "",
+            "# Find and manage specific processes:",
+            f"ps aux | grep {target if target else 'nginx'}       # Find specific process",
+            f"pgrep -f {target if target else 'nginx'}           # Get PID by name",
+            f"pkill -f {target if target else 'nginx'}           # Kill process by name",
+            f"killall {target if target else 'nginx'}            # Kill all processes by name",
+            "",
+            "# Process control:",
+            "kill -15 PID              # Graceful termination (SIGTERM)",
+            "kill -9 PID               # Force kill (SIGKILL)",
+            "kill -HUP PID             # Reload configuration (SIGHUP)",
+            "",
+            "# Background processes:",
+            "nohup command &           # Run command in background",
+            "screen -S session_name    # Start screen session",
+            "tmux new -s session_name  # Start tmux session",
+            "",
+            "# Process monitoring:",
+            "top                       # Real-time process monitor",
+            "htop                      # Enhanced process monitor",
+            "jobs                      # List background jobs",
+            "bg                        # Resume job in background",
+            "fg                        # Bring job to foreground"
+        ]
+    
+    elif category == "disk":
+        commands = [
+            "# Disk Usage and Management Commands",
+            "df -h                     # Show disk usage (human readable)",
+            "df -i                     # Show inode usage",
+            "du -sh /*                 # Show directory sizes in root",
+            f"du -sh {target if target else '/var/log'}/* | sort -hr      # Sort by size",
+            "",
+            "# Disk analysis:",
+            "lsblk                     # List block devices",
+            "lsblk -f                  # Show filesystems",
+            "blkid                     # Show UUIDs and labels",
+            "fdisk -l                  # List disk partitions",
+            "parted -l                 # Alternative partition listing",
+            "",
+            "# Mount management:",
+            "mount                     # Show mounted filesystems",
+            "findmnt                   # Tree view of mounts",
+            "findmnt -D                # Show duplicate mounts",
+            f"mount /dev/sdb1 /mnt/{target if target else 'backup'}  # Mount filesystem",
+            f"umount /mnt/{target if target else 'backup'}           # Unmount filesystem",
+            "",
+            "# Cleanup commands:",
+            "apt autoremove            # Remove unused packages (Debian/Ubuntu)",
+            "yum autoremove            # Remove unused packages (RHEL/CentOS)",
+            "journalctl --disk-usage   # Check journal disk usage",
+            "journalctl --vacuum-time=3d  # Clean old journal entries"
+        ]
+    
+    elif category == "network":
+        commands = [
+            "# Network Configuration and Diagnostics",
+            "ip addr show              # Show IP addresses",
+            "ip link show              # Show network interfaces",
+            "ip route show             # Show routing table",
+            "ip neigh show             # Show ARP table",
+            "",
+            "# Port and connection analysis:",
+            "ss -tuln                  # Show listening ports",
+            "ss -tulpn                 # Show ports with process names",
+            "netstat -tuln             # Alternative port listing",
+            "netstat -rn               # Show routing table",
+            "",
+            "# Connectivity testing:",
+            f"ping -c 4 {target if target else 'google.com'}     # Test connectivity",
+            f"traceroute {target if target else 'google.com'}    # Trace network path",
+            f"mtr {target if target else 'google.com'}           # Network diagnostic tool",
+            f"nslookup {target if target else 'google.com'}      # DNS lookup",
+            f"dig {target if target else 'google.com'}           # DNS information",
+            "",
+            "# Network configuration:",
+            "wget -O- ifconfig.me      # Get public IP",
+            "curl -s ifconfig.me       # Alternative public IP",
+            "curl -s ipinfo.io         # Detailed IP information",
+            "",
+            "# Interface management:",
+            "sudo ip link set eth0 up    # Bring interface up",
+            "sudo ip link set eth0 down  # Bring interface down",
+            "sudo dhclient eth0          # Request DHCP lease"
+        ]
+    
+    elif category == "service":
+        commands = [
+            "# Service Management (systemd)",
+            f"systemctl status {target if target else 'nginx'}   # Check service status",
+            f"systemctl start {target if target else 'nginx'}    # Start service",
+            f"systemctl stop {target if target else 'nginx'}     # Stop service",
+            f"systemctl restart {target if target else 'nginx'}  # Restart service",
+            f"systemctl reload {target if target else 'nginx'}   # Reload configuration",
+            "",
+            "# Service persistence:",
+            f"systemctl enable {target if target else 'nginx'}   # Enable at boot",
+            f"systemctl disable {target if target else 'nginx'}  # Disable at boot",
+            f"systemctl is-enabled {target if target else 'nginx'}  # Check if enabled",
+            "",
+            "# Service discovery:",
+            "systemctl list-units      # List all active services",
+            "systemctl list-units --failed  # List failed services",
+            "systemctl list-unit-files     # List all unit files",
+            "",
+            "# Logs and troubleshooting:",
+            f"journalctl -u {target if target else 'nginx'}      # View service logs",
+            f"journalctl -u {target if target else 'nginx'} -f   # Follow service logs",
+            f"journalctl -u {target if target else 'nginx'} --since today  # Today's logs",
+            "",
+            "# Legacy service management (SysV):",
+            f"service {target if target else 'nginx'} status     # Check status (legacy)",
+            f"service {target if target else 'nginx'} start      # Start service (legacy)",
+            f"/etc/init.d/{target if target else 'nginx'} status # Direct init script"
+        ]
+    
+    elif category == "firewall":
+        commands = [
+            "# Firewall Management",
+            "sudo ufw status verbose   # Check UFW status (detailed)",
+            "sudo ufw --version        # Check UFW version",
+            "",
+            "# UFW basic operations:",
+            "sudo ufw enable           # Enable UFW",
+            "sudo ufw disable          # Disable UFW",
+            "sudo ufw reset            # Reset UFW rules",
+            "",
+            "# UFW rule management:",
+            f"sudo ufw allow {target if target else '22'}        # Allow port/service",
+            f"sudo ufw deny {target if target else '23'}         # Block port/service",
+            "sudo ufw allow ssh        # Allow SSH service",
+            "sudo ufw allow 80/tcp     # Allow HTTP",
+            "sudo ufw allow 443/tcp    # Allow HTTPS",
+            "sudo ufw allow from 192.168.1.0/24  # Allow from subnet",
+            "",
+            "# UFW advanced rules:",
+            "sudo ufw delete allow 80  # Remove rule",
+            "sudo ufw insert 1 allow 22  # Insert rule at position",
+            "sudo ufw logging on       # Enable logging",
+            "",
+            "# iptables (advanced):",
+            "sudo iptables -L          # List iptables rules",
+            "sudo iptables -L -n       # List with numeric output",
+            "sudo iptables -A INPUT -p tcp --dport 80 -j ACCEPT  # Allow HTTP",
+            "sudo iptables -A INPUT -p tcp --dport 22 -j DROP    # Block SSH",
+            "",
+            "# Save iptables rules:",
+            "sudo iptables-save > /etc/iptables/rules.v4  # Save IPv4 rules",
+            "sudo ip6tables-save > /etc/iptables/rules.v6 # Save IPv6 rules"
+        ]
+    
+    result = "\n".join(commands)
+    return [TextContent(type="text", text=result)]
+
+# ============================================================================
+# MCP SERVER CONFIGURATION
+# ============================================================================
+
+@mcp_app.list_tools()
+async def list_tools():
+    """List all available Linux system administration tools"""
+    return [
+        Tool(
+            name="user_management",
+            description="Generate Linux user management commands for creating, deleting, and managing users and groups",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string", 
+                        "enum": ["add_user_to_group", "create_user", "delete_user", "list_groups"],
+                        "description": "User management action to perform"
+                    },
+                    "username": {
+                        "type": "string",
+                        "description": "Username for the operation"
+                    },
+                    "groupname": {
+                        "type": "string",
+                        "description": "Group name for the operation"
+                    },
+                    "sudo": {
+                        "type": "boolean", 
+                        "default": True,
+                        "description": "Whether to use sudo for privileged operations"
+                    }
+                },
+                "required": ["action"]
+            }
+        ),
+        Tool(
+            name="file_permissions",
+            description="Generate file permission and ownership commands for chmod, chown, and permission discovery",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string", 
+                        "enum": ["chmod", "chown", "find_permissions"],
+                        "description": "File permission action to perform"
+                    },
+                    "path": {
+                        "type": "string",
+                        "description": "File or directory path"
+                    },
+                    "permissions": {
+                        "type": "string",
+                        "description": "Permission mode (e.g., 755, 644, u+x)"
+                    },
+                    "owner": {
+                        "type": "string",
+                        "description": "Owner username for chown operations"
+                    },
+                    "recursive": {
+                        "type": "boolean", 
+                        "default": False,
+                        "description": "Apply operation recursively"
+                    }
+                },
+                "required": ["action", "path"]
+            }
+        ),
+        Tool(
+            name="system_commands",
+            description="Generate common Linux system administration commands for processes, disk, network, services, and firewall",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "category": {
+                        "type": "string", 
+                        "enum": ["process", "disk", "network", "service", "firewall"],
+                        "description": "Category of system commands"
+                    },
+                    "action": {
+                        "type": "string",
+                        "description": "Specific action within the category"
+                    },
+                    "target": {
+                        "type": "string",
+                        "description": "Target for the operation (service name, hostname, etc.)"
+                    }
+                },
+                "required": ["category"]
+            }
+        )
+    ]
+
+@mcp_app.call_tool()
+async def call_tool(name: str, arguments: dict):
+    """Handle tool calls"""
+    try:
+        if name == "user_management":
+            return handle_user_management(arguments)
+        elif name == "file_permissions":
+            return handle_file_permissions(arguments)
+        elif name == "system_commands":
+            return handle_system_commands(arguments)
+        else:
+            return [TextContent(type="text", text=f"Unknown tool: {name}")]
+    except Exception as e:
+        return [TextContent(type="text", text=f"Error: {str(e)}")]
+
+# ============================================================================
+# FASTAPI WEB ENDPOINTS
+# ============================================================================
+
+@app.get("/")
+async def root():
+    """Status endpoint"""
+    return {
+        "service": "linux-mcp-server", 
+        "status": "running",
+        "mcp_endpoint": "/mcp/sse",
+        "tools": 3,
+        "categories": ["user_management", "file_permissions", "system_commands"]
+    }
+
+@app.get("/health")
+async def health():
+    """Health check"""
+    return {"status": "healthy", "service": "linux-mcp-server"}
+
+@app.get("/mcp/sse")
+async def mcp_sse_endpoint(request: Request):
+    """MCP SSE endpoint for agent connections"""
+    
+    async def event_stream():
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                    
+                yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
+                await asyncio.sleep(30)
+                
+        except asyncio.CancelledError:
+            pass
+    
+    return StreamingResponse(
+        event_stream(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "Access-Control-Allow-Origin": "*",
+        }
+    )
+
+def main():
+    """Main entry point"""
+    port = int(os.getenv("PORT", 7860))
+    
+    print(f"🚀 Starting Linux MCP Server on port {port}")
+    print(f"📊 Status: http://localhost:{port}/")
+    print(f"🔗 MCP SSE: http://localhost:{port}/mcp/sse")
+    print(f"🛠️  Tools: User Management, File Permissions, System Commands")
+    
+    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
+
+if __name__ == "__main__":
+    main()