Python使用paramiko远程执行Linux命令的完整示例

准备工作与安全提醒

在编写代码前，请确保目标Linux主机已开启SSH服务（通常端口为22）且网络可达。远程执行命令涉及敏感权限，切勿在正式代码中硬编码明文密码。推荐遵循以下规范：

• 将账号凭证存入环境变量或配置文件（如.env）动态加载。
• 生产环境优先使用SSH私钥进行无密码认证，移除明文密码参数。
• 对高危命令（如清理磁盘、重启服务）增加二次确认环节，并在非生产环境充分验证。

基础版：基于SSHClient的标准执行

该方案适用于无需交互、一次性获取输出结果的场景。核心依赖paramiko.SSHClient()，能够自动管理TCP连接生命周期，并直接返回标准输出流对象。

第1步：初始化连接与策略配置

创建客户端实例后，需设定主机密钥策略。当目标设备首次连接且不在信任列表中时，默认会阻断连接。配置为AutoAddPolicy可自动记录公钥，也可改用WarningPolicy打印警告日志。随后调用connect()传入地址、用户名与密码。示例代码如下：

# coding: utf-8
import paramiko

def simple_ssh_exec():
    client = paramiko.SSHClient()
    # 自动接受新主机密钥，避免首次连接阻塞
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 替换为目标服务器实际信息
        client.connect(hostname='172.16.1.5', port=22, username='linyouyi', password='123456')
        
        stdin, stdout, stderr = client.exec_command('df -hl')
        # exec_command返回的是文件类对象，必须立即读取并缓存
        out_msg = stdout.read().decode('utf-8').strip()
        err_msg = stderr.read().decode('utf-8').strip()
        
        print(out_msg)
        if err_msg:
            print(f"执行出现警告或错误: {err_msg}")
            
    except paramiko.AuthenticationException:
        print("认证失败：请检查用户名或密码")
    except Exception as e:
        print(f"连接异常: {e}")
    finally:
        client.close()

if __name__ == '__main__':
    simple_ssh_exec()

第2步：理解输出流特性

代码中stdout和stderr对象仅能读取一次。若未保存至变量而重复调用read()，第二次将返回空字节串。对于超长命令输出，建议在连接建立后调用client.get_transport().set_keepalive()防止空闲断开，或使用分块读取逻辑。

进阶版：封装交互式会话客户端

部分Linux工具或脚本依赖标准输入终端（PTY）。此时需绕过简单的exec_command，转而构建底层传输通道，模拟真实的Shell环境以实现持续交互。

第3步：构建可重试的Linux通信类

该类内部使用Transport建立加密管道，通过open_session()创建会话，并调用get_pty()分配伪终端。针对网络波动内置了重试机制，修正了原稿中循环缩进缺失的问题，使状态机逻辑完整闭合：

# coding: utf-8
import paramiko
import re
from time import sleep

class Linux(object):
    # 通过IP, 用户名，密码，超时时间初始化一个远程Linux主机
    def __init__(self, ip, username, password, timeout=30):
        self.ip = ip
        self.username = username
        self.password = password
        self.timeout = timeout
        self.t = None
        self.chan = None
        self.try_times = 3

    def connect(self):
        while True:
            try:
                self.t = paramiko.Transport(sock=(self.ip, 22))
                self.t.connect(username=self.username, password=self.password)
                self.chan = self.t.open_session()
                self.chan.settimeout(self.timeout)
                self.chan.get_pty()
                self.chan.invoke_shell()
                print(f'连接{self.ip}成功')
                # 接收初始欢迎信息并清空缓冲区
                print(self.chan.recv(65535).decode('utf-8', errors='ignore'))
                return
            except Exception as e1:
                if self.try_times != 0:
                    print(f'连接{self.ip}失败，进行重试')
                    self.try_times -= 1
                else:
                    print('重试3次失败，结束程序')
                    exit(1)

    def close(self):
        if self.chan: self.chan.close()
        if self.t: self.t.close()

    def send(self, cmd):
        cmd += '\r\n'
        p = re.compile(r'[#$%>]\s*$|^\r\n')
        result = ''
        self.chan.send(cmd)
        
        # 修复原稿缩进：补充循环体以分批次取回长回显
        while True:
            try:
                ret = self.chan.recv(65535)
                ret = ret.decode('utf-8', errors='ignore')
                result += ret
                if p.search(ret):
                    break
            except Exception:
                break
        print(result)
        return result

if __name__ == '__main__':
    host = Linux('172.16.1.5', 'linyouyi', '123456')
    host.connect()
    host.send('ll')
    host.close()

第4步：处理长时间运行的回显数据

由于底层使用的是原始字节流，执行耗时较长或带有滚动输出的命令时，recv()会分批返回数据。上述封装采用了轮询接收配合提示符正则匹配的逻辑。注意：实际使用时请确保已导入re与time模块。对于极复杂的交互流程，建议评估是否切换至pexpect库或结合tmux持久化终端会话。

常见问题与排错指南

1. 返回值乱码或抛出UnicodeDecodeError：目标系统可能使用了GBK或特定编码。请将.decode('utf-8')替换为对应编码，或在接收层使用errors='ignore'/'replace'容错。
2. 程序卡在recv或read处无响应：交互式Shell若未正确触发提示符，接收线程会永久阻塞。务必设置合理的settimeout()时长，或在业务逻辑中增加最大轮询阈值。
3. 提权失败(sudo: a password is required)：exec_command默认不分配TTY，无法弹出交互式密码输入框。可在命令前拼接echo 'your_pass' | sudo -S ...（注意脱敏），或将当前用户加入/etc/sudoers免密白名单。

以上就是Python远程执行Linux命令的代码详解与paramiko实操指南，更多关于Python自动化运维与SSH通信的资料请关注本站其它相关文章！