openclaw/skills/server-management/server-status

#!/usr/bin/env python3
"""
Server Status - Check status of managed servers
"""
import subprocess
import sys
import json
from typing import Dict, List, Optional

SERVERS = {
    "192.168.1.220": {"name": "localhost", "user": "alex", "ssh_key": None},
    "192.168.1.23": {"name": "nginx-server", "user": "alex", "ssh_key": "~/.ssh/id_ed25519_local"},
    "192.168.1.24": {"name": "raspberry-pi", "user": "alex", "ssh_key": None},
    "192.168.1.113": {"name": "synology-ds414", "user": "alex", "ssh_key": "~/.ssh/id_ed25519_local"},
}

def ssh_run(ip: str, command: str) -> tuple[int, str]:
    """Run command via SSH"""
    server = SERVERS.get(ip)
    if not server:
        return 1, f"Unknown server: {ip}"
    
    if ip == "192.168.1.220":
        # Local command
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        return result.returncode, result.stdout + result.stderr
    
    ssh_cmd = ["ssh"]
    if server["ssh_key"]:
        ssh_cmd.extend(["-i", server["ssh_key"].replace("~", "/home/alex")])
    ssh_cmd.extend(["-o", "StrictHostKeyChecking=no", f"{server['user']}@{ip}", command])
    
    result = subprocess.run(ssh_cmd, capture_output=True, text=True)
    return result.returncode, result.stdout + result.stderr

def check_updates(ip: str) -> int:
    """Count available updates"""
    code, output = ssh_run(ip, "apt list --upgradable 2>/dev/null | grep -c upgradable || echo 0")
    try:
        return int(output.strip())
    except:
        return 0

def check_reboot_required(ip: str) -> bool:
    """Check if reboot is required"""
    code, output = ssh_run(ip, "[ -f /var/run/reboot-required ] && echo YES || echo NO")
    return "YES" in output

def check_disk_usage(ip: str) -> str:
    """Get disk usage percentage"""
    code, output = ssh_run(ip, "df -h / | tail -1 | awk '{print $5}'")
    return output.strip()

def check_services(ip: str) -> List[str]:
    """Check custom services"""
    code, output = ssh_run(ip, "systemctl list-units --type=service --state=running | grep -E 'kamoer|kanban|openclaw' | awk '{print $1}' || echo ''")
    return [s for s in output.strip().split('\n') if s]

def check_fail2ban(ip: str) -> Optional[Dict]:
    """Check fail2ban status"""
    code, output = ssh_run(ip, "command -v fail2ban-client >/dev/null && sudo fail2ban-client status 2>/dev/null || echo NOTINSTALLED")
    if "NOTINSTALLED" in output:
        return None
    
    # Get banned count
    code2, banned = ssh_run(ip, "sudo fail2ban-client status sshd 2>/dev/null | grep 'Total banned' | awk '{print $4}' || echo 0")
    return {"installed": True, "banned": int(banned.strip() or 0)}

def print_server_status(ip: str, short: bool = False):
    """Print status for one server"""
    server = SERVERS[ip]
    
    print(f"\n{'='*60}")
    print(f"🖥️  {server['name'].upper()} ({ip})")
    print(f"{'='*60}")
    
    updates = check_updates(ip)
    reboot = check_reboot_required(ip)
    disk = check_disk_usage(ip)
    services = check_services(ip)
    fail2ban_info = check_fail2ban(ip)
    
    # Status indicators
    update_icon = "⚠️" if updates > 0 else "✅"
    reboot_icon = "⚠️" if reboot else "✅"
    disk_pct = int(disk.rstrip('%')) if disk else 0
    disk_icon = "⚠️" if disk_pct > 80 else "✅"
    
    print(f"{update_icon} Updates: {updates} packages")
    print(f"{reboot_icon} Reboot: {'Required' if reboot else 'Not needed'}")
    print(f"{disk_icon} Disk: {disk} used")
    
    if fail2ban_info:
        print(f"✅ fail2ban: Active ({fail2ban_info['banned']} IPs banned)")
    else:
        print(f"❌ fail2ban: Not installed")
    
    if services:
        print(f"🐳 Services: {', '.join(services)}")
    
    if not short:
        print(f"\n📊 Quick Actions:")
        if updates > 0:
            print(f"  • Update: server-update --server {ip}")
        if reboot:
            print(f"  • Restart: server-restart {ip}")
        if not fail2ban_info:
            print(f"  • Secure: server-secure --server {ip}")

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Check server status")
    parser.add_argument("--all", action="store_true", help="Check all servers")
    parser.add_argument("--server", help="Check specific server (IP)")
    parser.add_argument("--short", action="store_true", help="Brief output")
    
    args = parser.parse_args()
    
    if args.all:
        for ip in SERVERS.keys():
            print_server_status(ip, args.short)
    elif args.server:
        if args.server not in SERVERS:
            print(f"❌ Unknown server: {args.server}")
            print(f"Available: {', '.join(SERVERS.keys())}")
            sys.exit(1)
        print_server_status(args.server, args.short)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()