ssh_client.py

"""
ssh_client.py — 通过 paramiko 在远程 Linux 编译机执行命令
"""
import logging
from typing import Tuple

log = logging.getLogger("ssh")


def run_ssh_command(
    host: str,
    port: int,
    username: str,
    key_path: str,
    command: str,
    timeout: int = 600,
) -> Tuple[int, str, str]:
    """
    在远程主机执行命令
    返回 (exit_code, stdout, stderr)
    """
    try:
        import paramiko
    except ImportError:
        return (-1, "", "paramiko not installed. pip install paramiko")

    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        if key_path:
            ssh.connect(
                hostname=host,
                port=port,
                username=username,
                key_filename=key_path,
                timeout=15,
            )
        else:
            ssh.connect(
                hostname=host,
                port=port,
                username=username,
                look_for_keys=True,
                timeout=15,
            )

        stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)
        out = stdout.read().decode("utf-8", errors="replace")
        err = stderr.read().decode("utf-8", errors="replace")
        rc = stdout.channel.recv_exit_status()
        ssh.close()

        log.info(f"SSH {username}@{host}: rc={rc}, out={len(out)}B, err={len(err)}B")
        return (rc, out, err)

    except Exception as e:
        log.error(f"SSH FAILED {username}@{host}: {e}")
        return (-1, "", str(e))