Python Paramiko如何实现SSH远程连接与管理？-杰瑞科技汇

Paramiko 是一个强大的、纯 Python 实现的 SSHv2 协议库，它允许你用 Python 代码来连接和操作远程服务器，就像你在终端里使用 ssh 命令一样，你可以执行命令、上传下载文件、设置端口转发等。

（图片来源网络，侵删）

核心概念
环境准备
核心功能详解
- 1 基础 SSH 连接
- 2 执行命令
- 3 交互式 Shell
- 4 文件传输 (SFTP)
- 5 SSH 端口转发
高级主题
- 1 SSH Agent 和密钥认证
- 2 密钥处理
- 3 日志调试
- 4 异常处理
完整代码示例
总结与最佳实践

核心概念

在深入代码之前，理解 Paramiko 的几个核心类非常重要：

SSHClient: 这是你最常打交道的类，它提供了高级别的 API 来建立 SSH 连接，它封装了连接、认证、执行命令等复杂操作，你可以把它看作是 ssh 命令本身。
AutoAddPolicy: 一个策略类，用于处理服务器的主机密钥（host key），当你第一次连接一个新服务器时，ssh 客户端会保存该服务器的公钥（known_hosts），以确保你下次连接时不会被“中间人攻击”。AutoAddPolicy 会自动接受并保存服务器的密钥，方便但不安全（在生产环境中应避免使用）。
SSHConfig: 用于解析 ~/.ssh/config 文件,让你可以在代码中复用配置文件中的主机别名和设置。
SFTPClient: 一个 SFTP (SSH File Transfer Protocol) 客户端，用于在本地和远程服务器之间安全地传输文件，它提供了类似 os 模块的文件操作接口（如 put, get, mkdir, listdir 等）。
Channel: 代表一个 SSH 连接中的一个“通道”，你可以把它想象成一个独立的会话，当你执行命令或打开交互式 Shell 时，Paramiko 都会在 SSH 连接上创建一个 Channel 来传输数据。
Transport: 这是 Paramiko 的底层核心，它负责建立和管理 SSH 连接本身，处理加密、认证等。SSHClient 内部就是使用一个 Transport 对象来完成工作的,通常你不需要直接操作它。

环境准备

你需要安装 paramiko 库，如果还没有安装，可以通过 pip 安装：

pip install paramiko

核心功能详解

1 基础 SSH 连接

这是所有操作的前提，连接方式主要有两种：密码认证和密钥认证。

密码认证

（图片来源网络，侵删）

import paramiko
# 创建一个SSHClient实例
client = paramiko.SSHClient()
# 设置自动添加主机密钥策略 (生产环境不推荐)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
hostname = "your_server_ip"
port = 22
username = "your_username"
password = "your_password"
try:
    client.connect(hostname, port=port, username=username, password=password)
    print("连接成功！")
    # 在这里执行其他操作...
except paramiko.AuthenticationException:
    print("认证失败，请检查用户名或密码")
except paramiko.SSHException as e:
    print(f"SSH连接失败: {e}")
except Exception as e:
    print(f"发生错误: {e}")
finally:
    # 无论成功与否，最后都要关闭连接
    client.close()
    print("连接已关闭")

重要提示: AutoAddPolicy 虽然方便，但它会接受任何服务器的密钥，这使得你容易受到中间人攻击，在生产环境中，更安全的做法是使用 SSHClient.load_system_host_keys() 来加载系统已知的 known_hosts 文件,或者手动指定服务器的主机密钥。

# 更安全的方式
client = paramiko.SSHClient()
# 加载系统自带的known_hosts文件
client.load_system_host_keys() 
# 如果服务器不在known_hosts中，连接会抛出SSHException，这是更安全的行为
# 如果你确定要连接，并且信任该服务器，可以手动添加
# client.get_host_keys().add(hostname, 'ssh-rsa', server_public_key)

密钥认证

这是比密码认证更安全、更推荐的方式,尤其是在自动化脚本中。

import paramiko
private_key_path = "/path/to/your/private_key" # ~/.ssh/id_rsa
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
    # 使用密钥文件进行连接
    # 如果你的密钥有密码，可以使用 passphrase 参数
    private_key = paramiko.RSAKey.from_private_key_file(private_key_path)
    client.connect(hostname, port=port, username=username, pkey=private_key)
    print("密钥认证连接成功！")
    # ...
except paramiko.AuthenticationException:
    print("密钥认证失败")
except paramiko.PasswordRequiredException:
    print("密钥文件受密码保护，请提供passphrase")
except paramiko.SSHException as e:
    print(f"SSH连接失败: {e}")
finally:
    client.close()

2 执行命令

这是最常见的用法。exec_command() 方法会在远程服务器上执行一个命令，并返回三个对象：stdin, stdout, stderr。

（图片来源网络，侵删）

stdin: 标准输入流,通常用于向命令传递参数。
stdout: 标准输出流,用于获取命令的执行结果。
stderr: 标准错误流,用于获取命令的错误信息。

示例：

import paramiko
# ... (连接代码，假设 client 已连接)
try:
    # 执行一个简单的命令
    stdin, stdout, stderr = client.exec_command('ls -l /tmp')
    # stdout.read() 会读取所有输出，返回 bytes 类型
    # 需要解码成字符串
    output = stdout.read().decode('utf-8')
    error = stderr.read().decode('utf-8')
    if error:
        print(f"命令执行出错:\n{error}")
    else:
        print("命令执行结果:\n")
        print(output)
    # 检查命令的退出状态码 (0表示成功)
    exit_status = stdout.channel.recv_exit_status()
    print(f"\n命令退出状态码: {exit_status}")
finally:
    client.close()

注意: exec_command 是一个阻塞调用，它会等待命令在远程服务器上完全执行完毕后才会返回，对于长时间运行的命令,你需要考虑超时设置。

# 设置超时
stdin, stdout, stderr = client.exec_command('long_running_command', timeout=10)

3 交互式 Shell

如果你需要执行需要交互的命令（如 sudo, vim, top 等），exec_command 就不够用了，这时你需要创建一个交互式的 Channel。

这比执行简单命令要复杂一些，因为你需要手动处理输入、输出和错误流的读取,以避免死锁。

示例：

import paramiko
import time
# ... (连接代码，假设 client 已连接)
channel = client.invoke_shell()
# 发送命令
channel.send("sudo ls /root\n") # 注意，命令要以换行符结尾
# 等待提示符出现 (需要根据实际情况调整)
# sudo 命令通常会提示输入密码
time.sleep(1) 
if channel.recv_ready():
    output = channel.recv(4096).decode('utf-8')
    print(output)
# 输入 sudo 密码
channel.send("your_sudo_password\n")
time.sleep(1)
# 获取最终结果
if channel.recv_ready():
    final_output = channel.recv(4096).decode('utf-8')
    print("最终输出:\n", final_output)
# 关闭 channel
channel.close()
client.close()

更健壮的交互式 Shell 实现: 上面的例子很脆弱，因为它使用 time.sleep 来等待输出，一个更健壮的方法是使用循环来持续检查 channel 的状态,直到它关闭。

# 更健壮的交互式示例 (伪代码)
channel = client.invoke_shell()
channel.send("command\n")
while True:
    if channel.recv_ready():
        print(channel.recv(1024).decode('utf-8'), end='')
    if channel.exit_status_ready():
        break
    time.sleep(0.1)
channel.close()

4 文件传输 (SFTP)

paramiko 通过 SFTPClient 类提供了强大的文件传输功能。

上传文件

import paramiko
# ... (连接代码，假设 client 已连接)
try:
    # 从SSHClient获取SFTP客户端
    sftp = client.open_sftp()
    # 本地文件路径
    local_path = "/path/to/local_file.txt"
    # 远程文件路径
    remote_path = "/path/to/remote_file.txt"
    # 上传文件
    print(f"正在上传 {local_path} 到 {remote_path}...")
    sftp.put(local_path, remote_path)
    print("上传完成！")
finally:
    sftp.close()
    client.close()

下载文件

# ... (连接代码，假设 client 已连接)
try:
    sftp = client.open_sftp()
    local_path = "/path/to/downloaded_file.txt"
    remote_path = "/path/to/remote_file.txt"
    print(f"正在下载 {remote_path} 到 {local_path}...")
    sftp.get(remote_path, local_path)
    print("下载完成！")
finally:
    sftp.close()
    client.close()

其他 SFTP 操作

SFTPClient 还提供了许多类似 os 模块的功能：

sftp = client.open_sftp()
# 列出目录
print(sftp.listdir('/tmp'))
# 创建目录
sftp.mkdir('/tmp/new_dir')
# 删除文件
sftp.remove('/tmp/old_file.txt')
# 重命名文件
sftp.rename('/tmp/file1.txt', '/tmp/file2.txt')
# 获取文件属性
stat = sftp.stat('/tmp/remote_file.txt')
print(f"文件大小: {stat.st_size} bytes")
print(f"文件修改时间: {time.ctime(stat.st_mtime)}")
sftp.close()

5 SSH 端口转发

Paramiko 也支持 SSH 的端口转发功能，包括本地转发、远程转发和动态转发（SOCKS 代理），这是一个非常高级的功能,但非常强大。

示例：本地转发 将本地机器的 8080 端口转发到远程服务器的 80 端口，这相当于 ssh -L 8080:localhost:80 user@remote_host。

import paramiko
import socket
# ... (连接代码，假设 client 已连接)
# transport 对象是端口转发的基础
transport = client.get_transport()
# 请求端口转发
# 参数: 本地端口, 远程主机, 远程端口
transport.request_port_forward('', 8080) # '' 表示绑定到所有接口
# 创建一个 socket 来监听本地 8080 端口
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(('', 8080))
server_socket.listen(1)
print("本地端口转发已建立: localhost:8080 -> remote_host:80")
print("等待本地连接...")
try:
    while True:
        # 等待本地客户端连接
        local_client, addr = server_socket.accept()
        print(f"收到来自 {addr} 的连接")
        # 在 SSH 通道上创建一个端口转发
        channel = transport.open_channel("direct-tcpip", ('localhost', 80), addr)
        # 将本地客户端的数据转发到 SSH 通道，反之亦然
        paramiko.util.channel.copy(local_client, channel)
finally:
    server_socket.close()
    client.close()

注意：端口转发的实现相对复杂，上面的例子是一个简化版，在实际应用中，你可能需要用多线程或 select 来处理数据流的转发。

高级主题

1 SSH Agent 和密钥认证

SSH Agent (如 ssh-agent) 是一个在后台运行的程序，用于缓存你的解密后的私钥,这样你就不需要在每次连接时都输入密钥的密码。

你可以让 Paramiko 连接到本地的 SSH Agent：

import paramiko
# 从环境变量或默认路径加载SSH Agent的socket
agent = paramiko.Agent()
# 获取Agent中所有可用的密钥
keys = agent.get_keys()
if keys:
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        # 使用Agent中的第一个密钥进行连接
        client.connect(hostname, username=username, pkey=keys[0])
        print("通过SSH Agent连接成功！")
        # ...
    finally:
        client.close()
else:
    print("SSH Agent中没有找到可用的密钥。")

2 密钥处理

除了从文件加载,你还可以直接在内存中创建密钥对象。

from paramiko import RSAKey
# 生成一个新的RSA密钥对
key = RSAKey.generate(2048) # 2048是密钥长度
# 获取公钥字符串，可以添加到远程服务器的~/.ssh/authorized_keys文件中
public_key = key.get_name() + b' ' + key.asbytes()
print("公钥内容:\n", public_key.decode('utf-8'))
# 使用内存中的密钥对象进行连接
# client.connect(hostname, username='user', pkey=key)

3 日志调试

当连接或认证出现问题时，Paramiko 的日志非常有用，你可以使用 Python 的 logging 模块来获取底层的调试信息。

import logging
import paramiko
# 设置日志级别为DEBUG
paramiko.util.log_to_file('paramiko.log', level=logging.DEBUG)
# ... 你的连接和操作代码 ...
# 之后查看 paramiko.log 文件，会有非常详细的输出，包括握手过程、密钥交换等。

4 异常处理

Paramiko 提供了多种异常类,帮助你精确定位问题。

paramiko.AuthenticationException: 认证失败（用户名/密码错误，或密钥不被接受）。
paramiko.SSHException: SSH 连接或协议层面的错误，例如服务器拒绝连接、主机密钥不匹配等。
paramiko.BadHostKeyException: 服务器的主机密钥与 known_hosts 文件中的不匹配。
paramiko.PasswordRequiredException: 需要提供密码才能加载私钥。
socket.error: 底层的网络错误，如主机无法解析、连接超时等。

始终使用 try...except 块来捕获这些异常,使你的脚本更加健壮。

完整代码示例

下面是一个结合了密码认证、命令执行和文件上传的完整脚本。

import paramiko
import os
import sys
def remote_server_operation():
    # --- 配置信息 ---
    HOSTNAME = "your_server_ip"
    PORT = 22
    USERNAME = "your_username"
    PASSWORD = "your_password"
    REMOTE_DIR = "/tmp/upload_test"
    LOCAL_FILE = "example.txt"
    REMOTE_FILE = os.path.join(REMOTE_DIR, "uploaded_example.txt")
    # 创建SSHClient实例
    ssh_client = paramiko.SSHClient()
    # --- 连接 ---
    try:
        # 设置自动添加主机密钥策略 (仅用于演示)
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        print(f"正在连接到 {HOSTNAME}...")
        ssh_client.connect(HOSTNAME, port=PORT, username=USERNAME, password=PASSWORD)
        print("连接成功！")
        # --- 执行命令 ---
        print("\n--- 执行远程命令 ---")
        command = f"mkdir -p {REMOTE_DIR} && echo 'Hello from Paramiko!' > {LOCAL_FILE}"
        stdin, stdout, stderr = ssh_client.exec_command(command)
        # 检查错误
        error = stderr.read().decode()
        if error:
            print(f"命令执行出错: {error}")
        else:
            print("命令执行成功: '创建目录并生成本地示例文件'")
        # 检查退出状态
        exit_status = stdout.channel.recv_exit_status()
        if exit_status == 0:
            print(f"命令退出状态码: {exit_status} (成功)")
        else:
            print(f"命令退出状态码: {exit_status} (失败)")
            return # 如果命令失败，则后续操作可能无意义
        # --- 文件上传 ---
        print("\n--- 执行文件上传 ---")
        with ssh_client.open_sftp() as sftp:
            print(f"正在上传本地文件 '{LOCAL_FILE}' 到远程 '{REMOTE_FILE}'...")
            sftp.put(LOCAL_FILE, REMOTE_FILE)
            print("文件上传成功！")
            # 验证上传
            print("\n--- 验证上传 ---")
            stdin, stdout, stderr = ssh_client.exec_command(f"ls -l {REMOTE_FILE}")
            output = stdout.read().decode()
            error = stderr.read().decode()
            if not error and output:
                print("远程文件列表:")
                print(output)
            else:
                print("验证失败，文件可能未正确上传。")
    except paramiko.AuthenticationException:
        print("错误: 认证失败，请检查用户名和密码。")
    except paramiko.SSHException as e:
        print(f"SSH错误: {e}")
    except FileNotFoundError:
        print(f"错误: 本地文件 '{LOCAL_FILE}' 不存在。")
    except Exception as e:
        print(f"发生未知错误: {e}")
    finally:
        # --- 关闭连接 ---
        if ssh_client:
            ssh_client.close()
            print("\n连接已关闭。")
if __name__ == "__main__":
    remote_server_operation()

总结与最佳实践

优先使用密钥认证: 密钥认证比密码认证更安全,也更适合自动化。
避免 AutoAddPolicy: 在生产环境中，始终使用 load_system_host_keys() 或手动验证主机密钥,以防止中间人攻击。
资源管理: 使用 try...finally 或 with 语句来确保 SSHClient 和 SFTPClient 对象被正确关闭,防止资源泄漏。
处理输出: stdout.read() 返回的是 bytes，通常需要用 .decode('utf-8') 转换成字符串。
检查退出状态码: 使用 stdout.channel.recv_exit_status() 来检查命令是否成功执行,这是判断命令执行结果的标准方式。
利用日志: 遇到问题时，启用 paramiko.util.log_to_file 是排查问题的利器。
异常处理: 对可能发生的异常（网络、认证、权限等）进行捕获和处理,让你的脚本更健壮。
考虑高级库: 对于非常复杂的场景，可以考虑使用封装了 Paramiko 的高级库，如 fabric 或 ansible，它们提供了更简洁的 API 和更丰富的功能集，但对于学习和轻量级任务，直接使用 Paramiko 是最佳选择。

Python Paramiko如何实现SSH远程连接与管理？

目录

核心概念

环境准备

核心功能详解

1 基础 SSH 连接

2 执行命令

3 交互式 Shell

4 文件传输 (SFTP)

5 SSH 端口转发

高级主题

1 SSH Agent 和密钥认证

2 密钥处理

3 日志调试

4 异常处理

完整代码示例

总结与最佳实践

99ANYc3cd6

Apple Watch怎么用？新手入门指南在此！

Pokemon Go新手怎么快速入门？

Discuz! DIY教程如何快速上手？

Python 中如何实现 lastIndexOf 功能？

Java Base64加密解密怎么用？

Python数据分析视频教程怎么学更高效？

Java webservice调用方法有哪些？

Cool Edit Pro 2.0教程从哪里开始学？

ASP如何调用Java WebService？

Autodesk Inventor教程从哪里开始学？

Java文件名/目录名语法错误，如何正确命名？

Photoshop CS4中文版教程怎么学？

Java CXF如何调用WebService？

Python ImageFont API 如何正确使用？

Python如何用Orange实现Markov模型？

MySQL Workbench使用教程，如何快速上手？

Python Paramiko如何实现SSH远程连接与管理？

目录

核心概念

环境准备

核心功能详解

1 基础 SSH 连接

2 执行命令

3 交互式 Shell

4 文件传输 (SFTP)

5 SSH 端口转发

高级主题

1 SSH Agent 和密钥认证

2 密钥处理

3 日志调试

4 异常处理

完整代码示例

总结与最佳实践

相关推荐

Python数据分析视频教程怎么学更高效？