3. 批量服务器远程命令执行

任务目标

编写一个Python脚本，通过SSH协议连接多台服务器，批量执行指定命令，并将每台服务器的输出结果集中收集、展示。脚本需支持服务器列表配置、命令参数化、连接超时与错误处理。

环境依赖

需要安装 paramiko 库，安装命令：

pip install paramiko

完整实现代码

import paramiko
import sys
from datetime import datetime

# 服务器配置列表：host, port, username, password
SERVERS = [
    {"host": "192.168.1.101", "port": 22, "username": "root", "password": "pass123"},
    {"host": "192.168.1.102", "port": 22, "username": "root", "password": "pass123"},
    {"host": "192.168.1.103", "port": 22, "username": "root", "password": "pass123"},
]

def exec_on_server(server, command, timeout=10):
    """在单台服务器上执行命令，返回 (是否成功, 输出内容或错误信息)"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(
            hostname=server["host"],
            port=server["port"],
            username=server["username"],
            password=server["password"],
            timeout=timeout,
            banner_timeout=timeout,
        )
        stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
        out = stdout.read().decode("utf-8", errors="replace")
        err = stderr.read().decode("utf-8", errors="replace")
        if err.strip():
            return False, f"STDERR: {err}"
        return True, out
    except Exception as e:
        return False, f"连接/执行异常: {str(e)}"
    finally:
        client.close()

def batch_execute(servers, command):
    """批量执行命令，返回结果列表"""
    results = []
    for s in servers:
        ok, output = exec_on_server(s, command)
        results.append({
            "host": s["host"],
            "success": ok,
            "output": output,
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        })
    return results

def print_results(results):
    """格式化打印结果"""
    for r in results:
        status = "✓" if r["success"] else "✗"
        print(f"[{status}] {r['host']} @ {r['time']}")
        print("-" * 60)
        print(r["output"] if r["output"].strip() else "(无输出)")
        print("=" * 60)

if __name__ == "__main__":
    # 从命令行参数获取要执行的命令，默认使用 uptime
    cmd = sys.argv[1] if len(sys.argv) > 1 else "uptime"
    print(f"批量执行命令: {cmd}\n")
    results = batch_execute(SERVERS, cmd)
    print_results(results)

运行方式

# 默认执行 uptime 命令
python batch_exec.py

# 指定其他命令（注意引号包裹）
python batch_exec.py "df -h"
python batch_exec.py "free -m"

验证方法与判定标准

将 SERVERS 列表中的IP、用户名、密码替换为实际可用的服务器信息后运行脚本。满足以下条件即判定通过：

• 脚本能依次连接所有服务器，无卡死或长时间无响应；
• 每台服务器均输出执行结果，成功条目显示命令输出内容，失败条目显示具体错误原因（如认证失败、网络不通、命令不存在）；
• 单台服务器连接失败不影响其他服务器的执行，最终结果中每台服务器都有对应记录。

常见问题

• 连接超时：检查目标服务器是否开启SSH服务、防火墙是否放行22端口、网络是否可达。
• 认证失败：确认用户名和密码正确；若使用密钥认证，可改用 paramiko.RSAKey.from_private_key_file 方式加载私钥。
• 命令输出乱码：部分系统默认编码非UTF-8，可将 decode("utf-8", errors="replace") 中的编码改为 "gbk" 或 "latin-1" 尝试。