3. 批量服务器SSH命令执行

任务目标

编写一个Python脚本，使用paramiko库对多台Linux服务器批量执行SSH命令，并统一收集、展示每台服务器的命令输出结果。脚本需支持从外部文件读取服务器列表，便于运维人员直接复用。

环境依赖

• Python 3.6+
• paramiko库：pip install paramiko

实现思路

脚本核心分为三部分：读取服务器清单（主机、端口、用户名、密码）、定义单台服务器的SSH命令执行函数、循环调用并打印结果。使用paramiko的SSHClient建立连接，执行命令后读取输出流，最后统一关闭连接。密码认证方式直接满足多数内网场景，如需免密可改用密钥认证（代码中已预留接口）。

完整代码

import paramiko
import sys
from datetime import datetime

def load_servers(file_path):
    """从文件读取服务器列表，格式: host port username password"""
    servers = []
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                parts = line.split()
                if len(parts) != 4:
                    print(f"[警告] 跳过无效行: {line}")
                    continue
                servers.append({
                    'host': parts[0],
                    'port': int(parts[1]),
                    'username': parts[2],
                    'password': parts[3]
                })
    except FileNotFoundError:
        print(f"[错误] 文件不存在: {file_path}")
        sys.exit(1)
    return servers

def exec_remote_command(server, command, timeout=10):
    """在单台服务器上执行命令，返回输出内容"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    output = {'host': server['host'], 'success': False, 'stdout': '', 'stderr': ''}
    try:
        client.connect(
            hostname=server['host'],
            port=server['port'],
            username=server['username'],
            password=server['password'],
            timeout=timeout
        )
        stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
        output['stdout'] = stdout.read().decode('utf-8', errors='replace')
        output['stderr'] = stderr.read().decode('utf-8', errors='replace')
        output['success'] = True
    except Exception as e:
        output['stderr'] = str(e)
    finally:
        client.close()
    return output

def main():
    if len(sys.argv) != 3:
        print("用法: python batch_ssh.py <服务器列表文件> <待执行命令>")
        print("示例: python batch_ssh.py servers.txt 'uptime && df -h'")
        sys.exit(1)

    server_file = sys.argv[1]
    command = sys.argv[2]
    servers = load_servers(server_file)

    if not servers:
        print("[错误] 未读取到任何有效服务器配置")
        sys.exit(1)

    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始批量执行，共 {len(servers)} 台服务器")
    print(f"执行命令: {command}\n" + "="*60)

    for server in servers:
        result = exec_remote_command(server, command)
        status = "✓ 成功" if result['success'] else "✗ 失败"
        print(f"\n[{status}] {result['host']}")
        print("-" * 40)
        if result['stdout']:
            print(result['stdout'].rstrip())
        if result['stderr']:
            print(f"[错误信息] {result['stderr'].rstrip()}")

    print("\n" + "="*60)
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 批量执行结束")

if __name__ == '__main__':
    main()

运行方式

1. 创建服务器清单文件 servers.txt，每行一台服务器，格式为：IP地址 端口 用户名 密码，支持#注释：

# 格式: host port username password
192.168.1.10 22 root password123
192.168.1.11 22 root password456
192.168.1.12 22 admin admin@789

2. 执行脚本（命令需用引号包裹）：

python batch_ssh.py servers.txt "uptime && df -h | tail -1"

验证方法

运行脚本后，观察输出是否满足以下条件即可判定通过：

• 每台服务器均显示执行状态（成功/失败），成功时输出命令的stdout内容，失败时输出错误原因；
• 输出顺序与servers.txt中的服务器顺序一致；
• 对不可达或认证失败的服务器，脚本不会中断，而是标记失败并继续执行后续服务器；
• 命令执行结果与手动SSH登录该服务器执行相同命令的输出一致。

常见问题

• 连接超时：检查目标服务器是否开启SSH服务、防火墙是否放行端口，或增大timeout参数；
• 认证失败：确认用户名密码正确，或改用密钥认证（将connect()中的password替换为key_filename='/path/to/key'）；
• 中文乱码：脚本已使用errors='replace'处理非UTF-8字符，若服务器输出GBK编码，可将decode('utf-8')改为decode('gbk')。