Python 批量 SSH

摘要：需要在几十上百台机器上执行同一条命令时，串行跑太慢、简单 & 并发又容易失控。本文用 concurrent.futures 线程池 + paramiko 写一个可控并发的批量执行脚本，包含并发度控制、单台超时、失败汇总与安全限制。

适用环境

python3 --version
pip install paramiko
# 准备密钥与主机清单
ls ~/.ssh/id_ed25519
printf '10.0.1.10\n10.0.1.11\n10.0.1.12\n' > hosts.txt

操作步骤

一、先明确安全边界

批量 SSH 是「一键影响所有机器」的能力，必须内建三道闸：

1. 命令白名单或二次确认：脚本只接受预设命令，不接受任意输入。
2. 并发上限：避免瞬间打满网络或让所有机器同时高负载。
3. 分批灰度：先在 1 台跑，确认无误再全量。

危险

不要把「任意命令」作为参数暴露给批量脚本。一个拼错的 rm -rf / data（多了空格）会在所有机器上执行。生产脚本应只暴露预定义的动作名称，如 check_disk、restart_nginx。

二、可控并发的批量执行器

#!/usr/bin/env python3
"""批量 SSH 执行命令：可控并发、单台超时、结果汇总。"""
import argparse
import logging
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import paramiko

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("batch-ssh")

# 动作白名单：只允许执行预定义命令，避免任意命令风险
ACTIONS = {
    "check_disk":  "df -P / | awk 'NR==2{print $5}'",
    "check_load":  "cat /proc/loadavg",
    "uptime":      "uptime",
    "nginx_status":"systemctl is-active nginx",
}


def ssh_run(host, user, key_path, cmd, timeout=30, port=22):
    """在单台机器执行命令，返回字典结果"""
    client = paramiko.SSHClient()
    # 生产环境建议改用 RejectPolicy + 预置 known_hosts
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    client.load_system_host_keys()
    started = time.time()
    try:
        client.connect(hostname=host, username=user, key_filename=key_path,
                       port=port, timeout=timeout, banner_timeout=timeout)
        _, stdout, stderr = client.exec_command(cmd, timeout=timeout)
        rc = stdout.channel.recv_exit_status()
        return {
            "host": host, "ok": rc == 0, "rc": rc,
            "out": stdout.read().decode(errors="replace").strip(),
            "err": stderr.read().decode(errors="replace").strip(),
            "cost": round(time.time() - started, 2),
        }
    except Exception as e:                      # 连接或执行失败
        return {"host": host, "ok": False, "rc": -1,
                "out": "", "err": f"{type(e).__name__}: {e}",
                "cost": round(time.time() - started, 2)}
    finally:
        client.close()


def run_all(hosts, user, key_path, cmd, workers=10, timeout=30):
    results = []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(ssh_run, h, user, key_path, cmd, timeout): h for h in hosts}
        for i, fut in enumerate(as_completed(futures), 1):
            r = fut.result()
            results.append(r)
            log.info("[%d/%d] %s rc=%s %ss", i, len(hosts), r["host"], r["rc"], r["cost"])
    return results


def main():
    ap = argparse.ArgumentParser(description="批量 SSH 执行预定义动作")
    ap.add_argument("--hosts", required=True, help="主机清单文件，每行一个 IP")
    ap.add_argument("--action", required=True, choices=sorted(ACTIONS), help="动作名")
    ap.add_argument("--user", default="ops")
    ap.add_argument("--key", default=os.path.expanduser("~/.ssh/id_ed25519"))
    ap.add_argument("--workers", type=int, default=10, help="并发度")
    ap.add_argument("--timeout", type=int, default=30, help="单台超时秒数")
    ap.add_argument("--limit", type=int, default=0, help="只跑前 N 台（灰度）")
    args = ap.parse_args()

    hosts = [l.strip() for l in open(args.hosts) if l.strip() and not l.startswith("#")]
    if args.limit:
        hosts = hosts[:args.limit]
    cmd = ACTIONS[args.action]
    log.info("将在 %d 台机器上执行: %s", len(hosts), cmd)

    results = run_all(hosts, args.user, args.key, cmd, args.workers, args.timeout)

    print("\n=== 结果 ===")
    for r in sorted(results, key=lambda x: not x["ok"]):
        flag = "OK  " if r["ok"] else "FAIL"
        print(f"{flag} {r['host']:<16} rc={r['rc']:<4} {r['out'] or r['err']}")

    failed = [r["host"] for r in results if not r["ok"]]
    print(f"\n成功 {len(results)-len(failed)}/{len(results)}")
    if failed:
        print("失败主机:", " ".join(failed))
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())

三、灰度执行：先一台，再全量

# 先在 1 台上验证命令正确性
python3 batch_ssh.py --hosts hosts.txt --action check_disk --limit 1
# 确认输出无误后，放开全量，并发控制在 10
python3 batch_ssh.py --hosts hosts.txt --action check_disk --workers 10

四、并发度怎么定

机器数 < 20   ：workers = 5~10
机器数 20~200 ：workers = 10~20
机器数 > 200  ：分批执行，每批 50，批间隔观察

注意

并发度不是越大越好。目标机器若同时执行重 IO 命令（如 du、备份），并发过高会把所有机器的磁盘打满，反而引发线上故障。轻量命令可以并发高些，重命令应降到个位数。

五、把结果落盘便于审计

# 输出重定向留痕，包含执行时间便于回溯
python3 batch_ssh.py --hosts hosts.txt --action nginx_status \
  | tee -a /var/log/ops/batch-$(date +%F).log

六、更进一步的替代方案

批量执行若频繁使用且有状态收敛需求，应转向专业工具：

# Ansible 天然支持并发、幂等与结果汇总
ansible all -i hosts.ini -m shell -a "df -P / | awk 'NR==2{print \$5}'" --forks 20

验证

• [ ] --limit 1 灰度执行结果正确
• [ ] 故意填一台不可达主机，脚本能标记为 FAIL 且不影响其它主机
• [ ] --workers 生效（可通过耗时对比验证：并发 10 明显快于并发 1）
• [ ] 传入非白名单动作时 argparse 直接拒绝
• [ ] 退出码在有失败时为 1，可被 cron/监控识别

常见坑

• 并发过高打爆目标机：尤其执行重命令时，应按批执行并观察。
• 不设单台超时：一台机器网络黑洞会让整个批次卡住。
• 用 AutoAddPolicy：自动接受主机密钥，存在中间人风险，生产应预置 known_hosts。
• 密钥权限不对：私钥权限需为 600，否则 paramiko 会报错或静默失败。
• 结果只打印成功项：失败被忽略等于没执行，必须汇总失败清单并让退出码非零。

参考资料

• Paramiko 官方文档
• Python concurrent.futures 文档
• Ansible 临时命令文档
• OpenSSH ssh_config 文档