Python 运维自动化系列（07）：一条命令管 N 台机器——paramiko 批量远程执行

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 演示从简；生产见下节
client.connect("10.0.0.11", port=22, username="ops",
               key_filename="/home/ops/.ssh/id_rsa", timeout=5)
_, stdout, stderr = client.exec_command("uptime", timeout=10)
print(stdout.read().decode())
client.close()

import socket
import subprocess
import sys

HOST, PORT, USER = "127.0.0.1", 22, "ops"


def port_open(host: str, port: int, timeout: float = 2.0) -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(timeout)
        return s.connect_ex((host, port)) == 0


def run_remote(host: str, port: int, user: str, password: str, command: str) -> str:
    """真实部署里用的函数：连上、执行、只取 stdout。"""
    import paramiko

    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 演示从简
    try:
        client.connect(
            host, port=port, username=user, password=password,
            timeout=5, banner_timeout=5, auth_timeout=5,
        )
        _, stdout, stderr = client.exec_command(command, timeout=10)
        out = stdout.read().decode("utf-8", errors="replace")
        err = stderr.read().decode("utf-8", errors="replace")
        code = stdout.channel.recv_exit_status()
        if code != 0:
            raise RuntimeError(f"远端命令退出码 {code}: {err.strip()}")
        return out
    finally:
        client.close()


if __name__ == "__main__":
    print(f"探测 {HOST}:{PORT} ...")
    if port_open(HOST, PORT):
        print("端口开放，走真实 paramiko 连接：")
        print(run_remote(HOST, PORT, USER, "密码请用密钥或环境变量", "hostname && python --version"))
    else:
        print(f"[本机 {PORT} 端口未开放——没有 SSH 服务，属 Windows 默认状态]")
        print("\n(1) 真实执行一次连接，看错误处理路径拿到的异常：")
        try:
            run_remote(HOST, PORT, USER, "not-a-real-password", "hostname")
        except Exception as exc:
            print(f"    连接失败（符合预期）: {type(exc).__name__}: {exc}")

        print("\n(2) 连接成功时你会看到的输出，用本地子进程模拟：")
        for cmd in ("hostname", "python --version"):
            r = subprocess.run(cmd, capture_output=True, text=True, shell=True)
            print(f"    $ {cmd}\n    {r.stdout.strip()}")

探测 127.0.0.1:22 ...
[本机 22 端口未开放——没有 SSH 服务，属 Windows 默认状态]

(1) 真实执行一次连接，看错误处理路径拿到的异常：
    连接失败（符合预期）: NoValidConnectionsError: [Errno None] Unable to connect to port 22 on 127.0.0.1

(2) 连接成功时你会看到的输出，用本地子进程模拟：
    $ hostname
    LH
    $ python --version
    Python 3.13.15

import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

import paramiko

logger = logging.getLogger("batch_ssh")


def run_on_host(host_cfg: dict, command: str, retries: int = 1) -> dict:
    """在单台主机上执行命令；失败按 retries 重试。返回结构化结果，不抛异常。"""
    last_err = None
    for attempt in range(retries + 1):
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 演示从简
        try:
            client.connect(
                host_cfg["host"], port=host_cfg.get("port", 22),
                username=host_cfg["username"],
                key_filename=host_cfg.get("key"),          # 密钥优先
                timeout=5, banner_timeout=5, auth_timeout=5,
            )
            _, stdout, stderr = client.exec_command(command, timeout=30)
            code = stdout.channel.recv_exit_status()
            return {
                "host": host_cfg["name"],
                "ok": code == 0,
                "stdout": stdout.read().decode().strip(),
                "stderr": stderr.read().decode().strip(),
                "exit_code": code,
            }
        except Exception as exc:
            last_err = exc
            logger.warning("%s 第 %d 次失败: %s", host_cfg["name"], attempt + 1, exc)
        finally:
            client.close()
    return {"host": host_cfg["name"], "ok": False, "error": str(last_err)}


def batch_run(hosts: list[dict], command: str, workers: int = 8) -> list[dict]:
    results = []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(run_on_host, h, command): h["name"] for h in hosts}
        for fut in as_completed(futures):
            results.append(fut.result())
    return results

sftp = client.open_sftp()
sftp.put("configs/app.yaml", "/etc/myapp/app.yaml")   # 推
sftp.get("/var/log/myapp/error.log", "pull/error.log")  # 拉
sftp.close()