基于Python脚本实现跳板机跳转及主机服务检测与命令执行方案咨询

需求1：SSH连接跳板机检查服务并执行命令 #

直接通过SSH远程命令完成，无需交互式登录，步骤如下：

• 免密配置（可选）：提前用ssh-copy-id user@跳板机IP配置密钥对，避免每次输入密码
• 远程执行逻辑：将服务检查和命令执行逻辑写在一条SSH命令中，利用shell条件判断实现：

  ssh user@跳板机IP "if systemctl is-active --quiet 目标服务名; then 要执行的命令; fi"
  

  说明：
  • systemctl is-active --quiet 服务名：systemd系统下检查服务状态，退出码0表示运行中
  • 若为非systemd系统，可替换为service 服务名 status >/dev/null 2>&1，同样通过退出码判断

需求2：Python脚本实现跨跳板机批量运维操作 #

使用paramiko库实现SSH代理与批量操作，核心逻辑是通过跳板机作为中转，完成ping检测、目标主机SSH登录、服务检查与命令执行。

核心依赖 #

先安装依赖库：

pip install paramiko

实现步骤与代码示例 #

import paramiko
from paramiko.ssh_exception import SSHException

# 配置参数（建议移至配置文件，避免硬编码）
JUMP_CONFIG = {
    "host": "跳板机IP",
    "user": "跳板机用户名",
    "password": "跳板机密码"  # 密钥登录可替换为 "key_path": "/path/to/private/key"
}
TARGET_CONFIG = {
    "user": "目标主机用户名",
    "password": "目标主机密码"
}
TARGET_IPS = ["192.168.1.10", "192.168.1.11"]
CHECK_SERVICE = "nginx"
EXEC_COMMAND = "/usr/bin/echo '执行运维操作' && /path/to/your/script.sh"

def get_jump_client():
    """建立跳板机SSH连接"""
    jump_client = paramiko.SSHClient()
    jump_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        # 密码登录
        jump_client.connect(
            JUMP_CONFIG["host"],
            username=JUMP_CONFIG["user"],
            password=JUMP_CONFIG["password"],
            timeout=10
        )
        # 密钥登录替换为以下代码
        # private_key = paramiko.RSAKey.from_private_key_file(JUMP_CONFIG["key_path"])
        # jump_client.connect(JUMP_CONFIG["host"], username=JUMP_CONFIG["user"], pkey=private_key, timeout=10)
        return jump_client
    except SSHException as e:
        print(f"跳板机连接失败: {str(e)}")
        return None

def ping_target_on_jump(jump_client, target_ip):
    """在跳板机上执行ping检测目标IP"""
    stdin, stdout, stderr = jump_client.exec_command(f"ping -c 2 -W 2 {target_ip}")
    # 退出码0表示ping通
    return stdout.channel.recv_exit_status() == 0

def get_target_client(jump_client, target_ip):
    """通过跳板机代理连接目标主机"""
    try:
        transport = jump_client.get_transport()
        # 建立端口转发通道
        dest_addr = (target_ip, 22)
        local_addr = (JUMP_CONFIG["host"], 22)
        channel = transport.open_channel("direct-tcpip", dest_addr, local_addr)
        
        target_client = paramiko.SSHClient()
        target_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        target_client.connect(
            target_ip,
            username=TARGET_CONFIG["user"],
            password=TARGET_CONFIG["password"],
            sock=channel,
            timeout=10
        )
        return target_client
    except SSHException as e:
        print(f"目标主机{target_ip}连接失败: {str(e)}")
        return None

def check_service_exec_cmd(target_client):
    """检查目标主机服务状态并执行命令"""
    # 检查服务运行状态
    stdin, stdout, stderr = target_client.exec_command(f"systemctl is-active --quiet {CHECK_SERVICE}")
    if stdout.channel.recv_exit_status() == 0:
        print(f"[{target_client.get_transport().getpeername()[0]}] 服务{CHECK_SERVICE}运行中，执行命令")
        stdin, stdout, stderr = target_client.exec_command(EXEC_COMMAND)
        # 打印命令输出
        print(stdout.read().strip().decode())
        if stderr.read():
            print(f"命令执行警告: {stderr.read().strip().decode()}")
    else:
        print(f"[{target_client.get_transport().getpeername()[0]}] 服务{CHECK_SERVICE}未运行，跳过执行")

if __name__ == "__main__":
    jump_client = get_jump_client()
    if not jump_client:
        exit(1)
    
    for ip in TARGET_IPS:
        print(f"===== 处理目标IP: {ip} =====")
        if ping_target_on_jump(jump_client, ip):
            target_client = get_target_client(jump_client, ip)
            if target_client:
                check_service_exec_cmd(target_client)
                target_client.close()
        else:
            print(f"IP {ip} ping不通，跳过")
    
    jump_client.close()

关键细节说明 #

• 端口转发：通过跳板机的SSH通道转发目标主机的22端口，实现间接登录
• 状态判断：所有操作均通过命令退出码判断结果（0为成功），避免解析复杂的输出文本
• 异常处理：添加SSH连接异常捕获，避免脚本因单台主机失败而终止
• 扩展性：可将配置参数移至YAML/JSON文件，或添加多线程实现批量操作加速

内容的提问来源于stack exchange，提问作者usman