server-manager/core/claude_setup.py

"""
Claude Code integration setup.
Installs ssh.py, encryption.py, /ssh skill, SSH key — everything needed
for Claude Code to manage servers via the shared servers.json.
"""

import os
import sys
import shutil
from core.logger import log

SHARED_DIR = os.path.expanduser("~/.server-connections")

# PyInstaller: bundled data is in sys._MEIPASS; otherwise use project dir
if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"):
    _BASE_DIR = sys._MEIPASS
else:
    _BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

SSH_SCRIPT_SRC = os.path.join(_BASE_DIR, "tools", "ssh.py")
ENCRYPTION_SRC = os.path.join(_BASE_DIR, "core", "encryption.py")
SKILL_SRC = os.path.join(_BASE_DIR, "tools", "skill-ssh.md")

SKILL_DST_DIR = os.path.expanduser("~/.claude/commands")
SKILL_DST = os.path.join(SKILL_DST_DIR, "ssh.md")
SSH_KEY_PATH = os.path.expanduser("~/.ssh/id_ed25519")


def check_status() -> dict:
    """Check what's installed and what's missing."""
    return {
        "shared_dir": os.path.exists(SHARED_DIR),
        "servers_json": os.path.exists(os.path.join(SHARED_DIR, "servers.json")),
        "ssh_script": os.path.exists(os.path.join(SHARED_DIR, "ssh.py")),
        "encryption": os.path.exists(os.path.join(SHARED_DIR, "encryption.py")),
        "skill_installed": os.path.exists(SKILL_DST),
        "ssh_key_exists": os.path.exists(SSH_KEY_PATH),
        "ssh_key_pub": os.path.exists(SSH_KEY_PATH + ".pub"),
    }


def install_ssh_script() -> str:
    """Copy ssh.py and encryption.py to shared dir."""
    os.makedirs(SHARED_DIR, exist_ok=True)
    results = []

    # Copy ssh.py
    dst = os.path.join(SHARED_DIR, "ssh.py")
    if os.path.exists(SSH_SCRIPT_SRC):
        shutil.copy2(SSH_SCRIPT_SRC, dst)
        log.info(f"ssh.py installed: {dst}")
        results.append(f"ssh.py installed: {dst}")
    elif os.path.exists(dst):
        results.append(f"ssh.py already exists: {dst}")
    else:
        results.append("ERROR: ssh.py source not found")

    # Copy encryption.py
    enc_dst = os.path.join(SHARED_DIR, "encryption.py")
    if os.path.exists(ENCRYPTION_SRC):
        shutil.copy2(ENCRYPTION_SRC, enc_dst)
        log.info(f"encryption.py installed: {enc_dst}")
        results.append(f"encryption.py installed: {enc_dst}")
    elif os.path.exists(enc_dst):
        results.append(f"encryption.py already exists: {enc_dst}")
    else:
        results.append("ERROR: encryption.py source not found")

    return "\n".join(results)


def install_skill() -> str:
    """Install /ssh skill for Claude Code."""
    os.makedirs(SKILL_DST_DIR, exist_ok=True)
    if os.path.exists(SKILL_SRC):
        shutil.copy2(SKILL_SRC, SKILL_DST)
        log.info(f"Skill installed: {SKILL_DST}")
        return f"Skill installed: {SKILL_DST}"
    # Fallback: check existing
    if os.path.exists(SKILL_DST):
        return f"Skill already exists: {SKILL_DST}"
    # Generate minimal skill
    skill_content = _generate_skill_content()
    with open(SKILL_DST, "w", encoding="utf-8") as f:
        f.write(skill_content)
    log.info(f"Skill generated: {SKILL_DST}")
    return f"Skill generated: {SKILL_DST}"


def generate_ssh_key() -> str:
    """Generate ed25519 SSH key if not exists."""
    if os.path.exists(SSH_KEY_PATH):
        return f"Key already exists: {SSH_KEY_PATH}"

    os.makedirs(os.path.dirname(SSH_KEY_PATH), exist_ok=True)

    try:
        import paramiko
    except ImportError:
        msg = "ERROR: paramiko is not installed — cannot generate SSH key"
        log.error(msg)
        return msg

    key = paramiko.Ed25519Key.generate()
    key.write_private_key_file(SSH_KEY_PATH)

    pub_key = f"ssh-ed25519 {key.get_base64()} server-manager"
    with open(SSH_KEY_PATH + ".pub", "w") as f:
        f.write(pub_key + "\n")

    log.info(f"SSH key generated: {SSH_KEY_PATH}")
    return f"Key generated: {SSH_KEY_PATH}"


def install_all() -> list[str]:
    """Full setup — install everything."""
    results = []

    steps = [
        ("ssh_script", install_ssh_script),
        ("skill", install_skill),
        ("ssh_key", generate_ssh_key),
    ]

    for name, func in steps:
        try:
            log.info(f"install_all: running {name}")
            result = func()
            results.append(result)
        except Exception as e:
            msg = f"ERROR ({name}): {e}"
            log.error(msg)
            results.append(msg)

    return results


def _generate_skill_content() -> str:
    """Generate /ssh skill markdown."""
    ssh_py_path = os.path.join(SHARED_DIR, "ssh.py").replace("\\", "/")
    return f"""SSH skill for Claude Code.

RULES:
- NEVER read servers.json directly
- ONLY use ssh.py for all server operations
- Maximum 1 connection attempt per command
- If connection fails — report error, do NOT retry automatically

Usage:
  python "{ssh_py_path}" ALIAS "command"
  python "{ssh_py_path}" ALIAS --no-sudo "command"
  python "{ssh_py_path}" ALIAS --upload LOCAL REMOTE
  python "{ssh_py_path}" ALIAS --download REMOTE LOCAL
  python "{ssh_py_path}" ALIAS --install-key
  python "{ssh_py_path}" ALIAS --ping
  python "{ssh_py_path}" --list
  python "{ssh_py_path}" --status

The user's arguments: $ARGUMENTS
"""