当前位置：首页>python>用 Python 操作 SSH 远程执行命令 | 第四篇

用 Python 操作 SSH 远程执行命令 | 第四篇

  • 2026-09-02 19:34:03
用 Python 操作 SSH 远程执行命令 | 第四篇

第四篇：用 Python 操作 SSH 远程执行命令

---

前面几篇都是本地操作，这一篇开始学习远程操作。SSH 是网络设备管理的标配，学会用 Python 操作 SSH，就能批量管理服务器和网络设备了。

本篇介绍如何使用 paramiko 库连接 SSH 并执行命令。

---

安装依赖

【bash】

pip install paramiko

在虚拟环境中安装：

【bash】

E:\qclawdata\.venv\Scripts\python.exe -m pip install paramiko

---

SSH 基础连接

paramiko 是 Python 最流行的 SSH 库，用它可以远程执行命令、传输文件。

【python】

import paramiko import time def ssh_connect(host, port=22, username='root', password=None, key_file=None):    """    建立 SSH 连接。    参数:        host (str): 主机 IP 或域名        port (int): SSH 端口，默认 22        username (str): 用户名        password (str): 密码（与 key_file 二选一）        key_file (str): 私钥文件路径    返回:        paramiko.SSHClient: SSH 客户端对象    """    # 创建 SSH 客户端    ssh = paramiko.SSHClient()        # 自动添加未知主机密钥（生产环境建议使用 known_hosts）    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())        try:        if key_file:            # 使用密钥连接            key = paramiko.RSAKey.from_private_key_file(key_file)            ssh.connect(                hostname=host,                port=port,                username=username,                pkey=key,                timeout=10            )        else:            # 使用密码连接            ssh.connect(                hostname=host,                port=port,                username=username,                password=password,                timeout=10            )                print(f"✅ 成功连接到 {host}")        return ssh            except Exception as e:        print(f"❌ 连接失败: {e}")        return None def ssh_exec_command(ssh, command, timeout=30):    """    执行 SSH 命令。    参数:        ssh: paramiko.SSHClient 对象        command (str): 要执行的命令        timeout (int): 超时时间（秒）    返回:        tuple: (stdin, stdout, stderr)    """    try:        stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)                # 获取输出        stdout_text = stdout.read().decode('utf-8', errors='ignore')        stderr_text = stderr.read().decode('utf-8', errors='ignore')                return stdout_text, stderr_text            except Exception as e:        return '', str(e) def ssh_disconnect(ssh):    """    关闭 SSH 连接    """    if ssh:        ssh.close()        print("🔌 SSH 连接已关闭") if __name__ == '__main__':    # 测试连接（请修改为你的服务器信息）    # 这是模拟代码，实际使用需要真实的 SSH 服务器        # 方法1：密码连接    # ssh = ssh_connect('192.168.1.100', username='admin', password='password')        # 方法2：密钥连接    # ssh = ssh_connect('192.168.1.100', username='admin', key_file='id_rsa')        print("SSH 连接代码已准备好，请配置实际参数后使用")

---

实际连接测试

由于我们没有真实的 SSH 服务器，这里展示完整的使用示例：

【python】

import paramiko def ssh_command_demo():    """    完整的 SSH 命令执行示例    """    host = '192.168.1.100'   # 替换为实际 IP    port = 22    username = 'root'          # 替换为实际用户名    password = 'your_password' # 替换为实际密码        # 建立连接    ssh = paramiko.SSHClient()    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())        try:        print(f"正在连接 {host}...")        ssh.connect(host, port=port, username=username, password=password)        print(f"✅ 连接成功")                # 执行命令        commands = [            'uname -a',           # 查看系统信息            'df -h',              # 查看磁盘空间            'free -m',            # 查看内存            'whoami',             # 查看当前用户        ]                for cmd in commands:            print(f"\n$ {cmd}")            print("-" * 40)                        stdin, stdout, stderr = ssh.exec_command(cmd)                        stdout_text = stdout.read().decode('utf-8', errors='ignore')            stderr_text = stderr.read().decode('utf-8', errors='ignore')                        if stdout_text:                print(stdout_text, end='')            if stderr_text:                print(f"错误: {stderr_text}", end='')            except paramiko.AuthenticationException:        print("❌ 认证失败，请检查用户名和密码")    except paramiko.SSHException as e:        print(f"❌ SSH 错误: {e}")    except Exception as e:        print(f"❌ 连接错误: {e}")    finally:        ssh.close()        print("\n🔌 连接已关闭") if __name__ == '__main__':    ssh_command_demo()

预期输出：

【代码】

$ uname -a ---------------------------------------- Linux server01 5.4.0-generic #1 SMP x86_64 GNU/Linux $ df -h ---------------------------------------- Filesystem      Size  Used Avail Use% Mounted on /dev/sda1       100G   45G   55G  45% / tmpfs           16G     0   16G   0% /dev/shm $ free -m ----------------------------------------              total        used        free      shared  buff/cache   available Mem:          32000        2048       24000         512        5500       29000 Swap:          8192           0        8192 $ whoami ---------------------------------------- root 🔌 连接已关闭

---

SSH 连接池：复用连接

频繁创建 SSH 连接开销大，可以使用连接池：

【python】

import paramiko from queue import Queue import threading class SSHPool:    """    SSH 连接池，支持多线程复用连接    """        def __init__(self, host, port, username, password, pool_size=5):        self.host = host        self.port = port        self.username = username        self.password = password        self.pool_size = pool_size        self.pool = Queue()        self._init_pool()        def _create_connection(self):        """创建新的 SSH 连接"""        ssh = paramiko.SSHClient()        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())        ssh.connect(            self.host,            port=self.port,            username=self.username,            password=self.password        )        return ssh        def _init_pool(self):        """初始化连接池"""        for _ in range(self.pool_size):            conn = self._create_connection()            self.pool.put(conn)        def get_connection(self):        """获取一个连接"""        return self.pool.get()        def return_connection(self, conn):        """归还连接"""        self.pool.put(conn)        def execute(self, command):        """执行命令（自动获取和归还连接）"""        conn = self.get_connection()        try:            stdin, stdout, stderr = conn.exec_command(command)            result = stdout.read().decode('utf-8', errors='ignore')            return result        finally:            self.return_connection(conn)        def close(self):        """关闭所有连接"""        while not self.pool.empty():            conn = self.pool.get()            conn.close() if __name__ == '__main__':    # 使用连接池    pool = SSHPool(        host='192.168.1.100',        port=22,        username='admin',        password='password',        pool_size=3    )        # 执行多条命令    for i in range(5):        result = pool.execute('echo hello')        print(f"第{i+1}次: {result.strip()}")        pool.close()

---

交互式 SSH 会话

有时候需要执行多个命令并保持会话，可以这样做：

【python】

import paramiko import sys import tty import termios def interactive_ssh(host, port, username, password):    """    创建交互式 SSH 会话    （类似终端操作）    """    ssh = paramiko.SSHClient()    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())    ssh.connect(host, port=port, username=username, password=password)        # 打开交互式 shell    channel = ssh.invoke_shell()        print(f"已连接到 {host}，输入 exit 退出")        # 设置终端    old_settings = termios.tcgetattr(sys.stdin)        try:        tty.setraw(sys.stdin)                while True:            if channel.recv_ready():                data = channel.recv(1024)                sys.stdout.write(data.decode())                sys.stdout.flush()                        if channel.send_ready():                key = sys.stdin.read(1)                if key == '\x03':  # Ctrl+C                    break                channel.send(key)                    finally:        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)        ssh.close() if __name__ == '__main__':    print("交互式 SSH 会话代码已准备好")    print("使用方法: interactive_ssh('IP', 22, '用户名', '密码')")

---

常见错误处理

【python】

import paramiko def ssh_with_error_handling():    """    带错误处理的 SSH 连接示例    """    host = '192.168.1.100'    username = 'admin'    password = 'wrong_password'        ssh = paramiko.SSHClient()    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())        try:        # 连接        ssh.connect(host, username=username, password=password, timeout=10)            except paramiko.AuthenticationException:        print("❌ 认证失败 - 用户名或密码错误")            except paramiko.SSHException as e:        print(f"❌ SSH 协议错误: {e}")            except TimeoutError:        print("❌ 连接超时 - 主机不可达")            except Exception as e:        print(f"❌ 其他错误: {e}")            finally:        ssh.close() if __name__ == '__main__':    ssh_with_error_handling()

---

实际应用场景

1. 批量执行命令 — 同时在多台服务器上执行相同命令

2. 配置文件下发 — 批量修改多台设备的配置

3. 日志收集 — 远程登录收集日志

4. 自动化运维 — 集成到运维系统

---

注意事项

  • 密码 vs 密钥：生产环境推荐使用密钥认证，更安全
  • known_hosts：生产环境应该管理 known_hosts 文件，不要自动添加
  • 超时设置：一定要设置 timeout，避免连接卡死
  • 连接池：频繁操作时使用连接池，减少连接开销
  • 错误处理：网络不稳定，做好重试机制

    • ---

    • 测试环境说明

  • Python 版本： 3.14.3
  • 操作系统： Windows 10
  • 依赖库： paramiko

    • ---

    • 下篇预告

    • 第五篇我们将学习如何批量 SSH 执行命令，实现真正的自动化运维。

    • ---

    • 相关历史文章

    • 用 Python 获取本机网络信息

    • 用 Python ping 检测主机存活

    • 用 Python 扫描局域网存活主机

    • 使用 json 文件作为配置文件

本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：596563534@qq.com 。

Python Tkinter + SQLite:轻量级MES工单系统的数据基础

Python基础学习手册第6章:while循环实战,结合生活场景+经典案例解析练习掌握

最新文章

随机文章