Python SSH执行命令如何实现？

Python通过SSH执行命令是自动化运维、远程服务器管理和分布式系统开发中的常见需求，Python提供了多种库来实现SSH功能，其中最常用的是paramiko和fabric，本文将详细介绍如何使用paramiko库实现SSH命令执行，包括连接管理、命令执行、文件传输等核心功能，并辅以代码示例和注意事项说明。

安装与基础连接

首先需要安装paramiko库，可通过pip命令完成安装：pip install paramiko，建立SSH连接的核心步骤包括创建SSH客户端、加载系统主机密钥（或手动指定）、连接服务器并认证，以下为基础连接代码示例：

import paramiko
# 创建SSH客户端对象
client = paramiko.SSHClient()
# 自动添加主机密钥（不推荐生产环境使用）
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器（需替换为实际IP、用户名和密码）
client.connect(hostname='192.168.1.100', port=22, 
               username='root', password='your_password')

连接参数说明：
| 参数 | 类型 | 说明 |
|——|——|——|
| hostname | str | 服务器IP地址或域名 |
| port | int | SSH端口号，默认为22 |
| username | str | 登录用户名 |
| password | str | 登录密码（与pkey二选一） |
| pkey | paramiko.PKey | SSH私钥对象（推荐使用密钥认证） |

执行命令并获取结果

连接成功后,可通过exec_command()方法执行命令，该方法返回三个值：标准输入、标准输出、标准错误流，示例代码如下：

stdin, stdout, stderr = client.exec_command('ls -l /tmp')
# 获取命令执行状态码
exit_status = stdout.channel.recv_exit_status()
# 读取标准输出输出
output = stdout.read().decode('utf-8')
error = stderr.read().decode('utf-8')
if exit_status == 0:
    print("命令执行成功:\n", output)
else:
    print("命令执行失败:\n", error)

注意事项：

1. 命令输出为字节流,需通过decode()方法解码为字符串
2. recv_exit_status()必须在读取输出后调用，否则可能导致阻塞
3. 长时间运行的命令建议结合channel设置超时时间

交互式命令执行

对于需要交互输入的命令（如sudo输入密码），可通过invoke_shell()方法创建交互式shell会话：

channel = client.invoke_shell()
channel.send('sudo ls /root\n')  # 发送命令
channel.send('your_sudo_password\n')  # 发送密码
time.sleep(2)  # 等待命令执行
output = channel.recv(4096).decode('utf-8')
print(output)
channel.close()

使用交互式shell时需注意：

• 需模拟人工输入的延迟时间
• 通过recv()循环读取完整输出
• 手动关闭channel释放资源

文件传输功能

paramiko支持SFTP协议进行文件传输，通过open_sftp()方法创建SFTP客户端：

sftp = client.open_sftp()
# 上传文件
sftp.put('/local/path/file.txt', '/remote/path/file.txt')
# 下载文件
sftp.get('/remote/path/file.txt', '/local/path/file.txt')
sftp.close()

文件传输操作对比：
| 操作 | 方法 | 说明 |
|——|——|——|
| 上传 | sftp.put(local, remote) | 支持目录路径 |
| 下载 | sftp.get(remote, local) | 支持断点续传 |
| 创建目录 | sftp.mkdir(path) | 递归创建需自行实现 |

安全最佳实践

1. 密钥认证：优先使用SSH密钥而非密码认证，可通过paramiko.RSAKey.from_private_key_file()加载私钥
2. 连接池：频繁操作时建议复用SSH连接对象，避免重复建立连接
3. 异常处理：捕获paramiko.SSHException等异常，确保程序健壮性
4. 资源释放：使用with语句或手动调用client.close()释放连接

完整示例代码

import paramiko
import time
def execute_ssh_command(hostname, username, command, password=None, key_path=None):
    try:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        # 密钥认证优先
        if key_path:
            private_key = paramiko.RSAKey.from_private_key_file(key_path)
            client.connect(hostname, username=username, pkey=private_key)
        else:
            client.connect(hostname, username=username, password=password)
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        exit_code = stdout.channel.recv_exit_status()
        return {
            'status': 'success' if exit_code == 0 else 'failed',
            'exit_code': exit_code,
            'output': output,
            'error': error
        }
    except Exception as e:
        return {'status': 'error', 'message': str(e)}
    finally:
        if 'client' in locals():
            client.close()
# 使用示例
result = execute_ssh_command(
    hostname='192.168.1.100',
    username='admin',
    command='df -h',
    password='your_password'
)
print(result)

相关问答FAQs

Q1: 如何解决SSH连接时的“UnknownHostException”？
A: 该错误通常由主机名解析失败导致，建议检查：① 确保服务器IP/域名正确无误；② 测试网络连通性（如ping命令）；③ 若使用域名，检查DNS配置或尝试使用IP地址替代。

Q2: 执行长时间运行的命令时如何避免程序阻塞？
A: 可通过以下方式优化：① 设置channel的超时时间（channel.settimeout(60)）；② 使用exec_command()的get_pty参数分配伪终端（get_pty=True）；③ 对于交互式命令，改用invoke_shell()并配合多线程或异步IO（如asyncssh库）处理输出流。