Python paramiko库实现SSH远程操作的实战指南_Python

1、引言

paramiko 是 python 生态中最流行的 ssh（secure shell）协议实现库之一，它基于纯 python 实现，底层依赖 cryptography 库完成加密与密钥交换。借助 paramiko，开发者可以在不依赖系统 ssh 命令的情况下，直接在 python 代码中完成远程主机连接、命令执行、文件传输（sftp）、端口转发等操作，是自动化运维、批量部署、远程巡检等场景的核心工具。

2、环境准备与安装

2.1、版本说明

paramiko 3.x 是当前的主线版本，相比 2.x 有以下重要变化：

最低要求 python 3.6+（3.4 及以上版本建议 python 3.8+）；
底层加密库从 cryptography 的旧接口迁移到新接口；
移除了部分已废弃的 api，例如 sshclient.load_system_host_keys 的默认行为调整；
对 openssh 新格式私钥（-----begin openssh private key-----）的支持更加完善。

提示：本文示例基于 paramiko 3.4.x 编写，向下兼容 3.0 及以上版本。

2.2、安装 paramiko

推荐使用 pip 安装：

pip install paramiko

如需安装到指定 python 环境：

python3 -m pip install paramiko

安装完成后，验证版本：

import paramiko
print(paramiko.__version__)

输出示例：

3.4.0

paramiko 会自动安装依赖库 cryptography 和 bcrypt（用于解析 openssh 私钥），无需手动额外安装。

3、核心组件与架构

paramiko 的核心组件包括：

组件	说明
`sshclient`	高层封装，提供连接、执行命令、sftp 等便捷接口，日常使用最频繁
`transport`	底层传输层，负责 ssh 协议握手、加密通道建立，`sshclient` 内部基于它实现
`sftpclient`	基于 ssh 通道的文件传输客户端，支持上传、下载、目录操作
`channel`	表示一条 ssh 通道，用于执行命令、传输数据
`authhandler`	处理认证流程，支持密码、公钥、键盘交互等多种方式
`hostkeys`	管理已知主机密钥，用于防止中间人攻击

它们之间的关系如下：

4、基础连接与命令执行

4.1 、密码方式连接

最简单的连接方式是使用用户名和密码：

import paramiko

ssh = paramiko.sshclient()
# 自动加载系统 known_hosts（可选）
ssh.load_system_host_keys()

# 自动添加未知主机密钥（测试环境常用，生产环境慎用）
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())

try:
    ssh.connect(
        hostname="192.168.1.100",
        port=22,
        username="root",
        password="your_password",
        timeout=10
    )
    print("连接成功")
except paramiko.authenticationexception:
    print("认证失败")
except paramiko.sshexception as e:
    print(f"ssh 错误: {e}")
finally:
    ssh.close()

4.2 、执行远程命令

连接建立后，使用 exec_command 执行命令：

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

stdin, stdout, stderr = ssh.exec_command("ls -la /opt")
# 读取标准输出
out = stdout.read().decode("utf-8")
# 读取错误输出
err = stderr.read().decode("utf-8")

print("stdout:", out)
print("stderr:", err)

# 获取退出状态码
exit_code = stdout.channel.recv_exit_status()
print("exit code:", exit_code)

ssh.close()

4.3、带交互的命令执行

某些命令需要交互式输入（如 sudo 密码），可以通过 stdin 写入：

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

stdin, stdout, stderr = ssh.exec_command("sudo -s apt-get update")
# 通过 stdin 发送密码
stdin.write("your_password\n")
stdin.flush()

# 等待命令执行完成
exit_code = stdout.channel.recv_exit_status()
print(stdout.read().decode("utf-8"))
print(stderr.read().decode("utf-8"))
print("exit code:", exit_code)

ssh.close()

4.4、实时读取输出

对于耗时较长的命令，可以逐行读取输出：

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

stdin, stdout, stderr = ssh.exec_command("tail -f /var/log/syslog", timeout=5)

# 逐行读取（注意：tail -f 不会自动结束，需要配合超时或手动终止）
for line in iter(stdout.readline, ""):
    print(line.strip())

ssh.close()

5、密钥认证

5.1 、使用 rsa/ed25519 私钥

生产环境推荐使用密钥认证，避免密码明文传输：

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())

# 方式一：直接指定私钥文件路径
ssh.connect(
    hostname="192.168.1.100",
    port=22,
    username="root",
    key_filename="/home/user/.ssh/id_rsa",
    timeout=10
)

# 方式二：使用 pkey 对象
private_key = paramiko.rsakey.from_private_key_file("/home/user/.ssh/id_rsa")
# 如果私钥有密码保护：
# private_key = paramiko.rsakey.from_private_key_file("/home/user/.ssh/id_rsa", password="key_password")

ssh.connect(
    hostname="192.168.1.100",
    port=22,
    username="root",
    pkey=private_key,
    timeout=10
)

stdin, stdout, stderr = ssh.exec_command("whoami")
print(stdout.read().decode("utf-8"))

ssh.close()

5.2 、支持 openssh 新格式私钥

paramiko 3.x 原生支持 -----begin openssh private key----- 格式：

import paramiko

# 自动识别私钥类型（rsa / ed25519 / ecdsa）
key = paramiko.ed25519key.from_private_key_file("/home/user/.ssh/id_ed25519")

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", pkey=key)
print("ed25519 密钥认证成功")
ssh.close()

5.3 、多密钥自动尝试

paramiko 支持在连接时传入多个密钥，自动逐个尝试：

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())

keys = []
for path in ["/home/user/.ssh/id_rsa", "/home/user/.ssh/id_ed25519"]:
    try:
        keys.append(paramiko.rsakey.from_private_key_file(path))
    except exception:
        try:
            keys.append(paramiko.ed25519key.from_private_key_file(path))
        except exception:
            pass

ssh.connect(
    hostname="192.168.1.100",
    username="root",
    pkey=keys[0] if keys else none,
    key_filename=["/home/user/.ssh/id_rsa", "/home/user/.ssh/id_ed25519"],
    timeout=10
)
print("连接成功")
ssh.close()

6、sftp 文件传输

6.1 、上传文件

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

# 打开 sftp 通道
sftp = ssh.open_sftp()

# 上传本地文件到远程
sftp.put("/local/path/file.txt", "/remote/path/file.txt")

# 使用文件对象上传
with sftp.open("/remote/path/upload.txt", "w") as remote_file:
    remote_file.write("hello, paramiko!\n")

sftp.close()
ssh.close()

6.2、下载文件

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

sftp = ssh.open_sftp()

# 下载远程文件到本地
sftp.get("/remote/path/file.txt", "/local/path/file.txt")

# 使用文件对象读取
with sftp.open("/remote/path/readme.md", "r") as remote_file:
    content = remote_file.read().decode("utf-8")
    print(content)

sftp.close()
ssh.close()

6.3 、目录操作

import paramiko

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

sftp = ssh.open_sftp()

# 列出目录
files = sftp.listdir("/opt")
for f in files:
    print(f)

# 带属性的目录列表
for attr in sftp.listdir_attr("/opt"):
    print(f"{attr.filename}  size={attr.st_size}  mode={oct(attr.st_mode)}")

# 创建目录
sftp.mkdir("/opt/new_dir")

# 删除文件 / 目录
sftp.remove("/opt/old_file.txt")
sftp.rmdir("/opt/empty_dir")

# 重命名
sftp.rename("/opt/old_name.txt", "/opt/new_name.txt")

# 获取文件状态
stat = sftp.stat("/opt/file.txt")
print(f"大小: {stat.st_size} 字节")

sftp.close()
ssh.close()

6.4、递归上传目录

paramiko 本身不提供递归上传，需要自行实现：

import os
import paramiko

def upload_dir(sftp, local_dir, remote_dir):
    for root, dirs, files in os.walk(local_dir):
        # 计算相对路径
        rel_path = os.path.relpath(root, local_dir)
        target_dir = remote_dir if rel_path == "." else f"{remote_dir}/{rel_path}"
        try:
            sftp.mkdir(target_dir)
        except ioerror:
            pass  # 目录已存在
        for file in files:
            local_file = os.path.join(root, file)
            remote_file = f"{target_dir}/{file}"
            sftp.put(local_file, remote_file)
            print(f"上传: {local_file} -> {remote_file}")

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")
sftp = ssh.open_sftp()

upload_dir(sftp, "/local/project", "/opt/project")

sftp.close()
ssh.close()

7、高级特性

7.1 、端口转发（隧道）

paramiko 支持本地端口转发，将本地端口映射到远程主机：

import paramiko
import threading
import socket

def local_port_forward(ssh, local_port, remote_host, remote_port):
    """将本地 local_port 转发到 remote_host:remote_port"""
    transport = ssh.get_transport()
    
    def handler():
        # 监听本地端口
        server = socket.socket(socket.af_inet, socket.sock_stream)
        server.setsockopt(socket.sol_socket, socket.so_reuseaddr, 1)
        server.bind(("127.0.0.1", local_port))
        server.listen(100)
        print(f"本地端口 {local_port} 已转发到 {remote_host}:{remote_port}")
        
        while true:
            client_sock, addr = server.accept()
            # 建立远程通道
            channel = transport.open_channel(
                "direct-tcpip",
                (remote_host, remote_port),
                addr
            )
            # 双向转发数据
            def forward(src, dst):
                while true:
                    data = src.recv(4096)
                    if not data:
                        break
                    dst.send(data)
                dst.close()
            
            t1 = threading.thread(target=forward, args=(client_sock, channel))
            t2 = threading.thread(target=forward, args=(channel, client_sock))
            t1.daemon = true
            t2.daemon = true
            t1.start()
            t2.start()
    
    thread = threading.thread(target=handler)
    thread.daemon = true
    thread.start()
    return thread

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")

# 将本地 3307 端口转发到远程 127.0.0.1:3306（mysql）
local_port_forward(ssh, 3307, "127.0.0.1", 3306)

# 保持程序运行
import time
time.sleep(60)
ssh.close()

7.2 、使用代理（proxycommand）

通过跳板机连接目标主机：

import paramiko

class proxycommand:
    """模拟 openssh 的 proxycommand"""
    def __init__(self, proxy_cmd):
        import subprocess
        self.process = subprocess.popen(
            proxy_cmd,
            shell=true,
            stdin=subprocess.pipe,
            stdout=subprocess.pipe,
            stderr=subprocess.pipe
        )
    
    def send(self, data):
        self.process.stdin.write(data)
        self.process.stdin.flush()
    
    def recv(self, size):
        return self.process.stdout.read(size)
    
    def close(self):
        self.process.terminate()

# 通过跳板机连接
proxy = proxycommand("ssh -w %h:%p jump_user@jump_host")

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect(
    hostname="10.0.0.5",
    port=22,
    username="target_user",
    password="target_password",
    sock=proxy
)

stdin, stdout, stderr = ssh.exec_command("hostname")
print(stdout.read().decode("utf-8"))
ssh.close()

7.3 、键盘交互认证

某些服务器使用键盘交互方式认证（如双因素认证）：

import paramiko

class myserverhandler(paramiko.serverinterface):
    pass

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())

# 使用键盘交互认证
ssh.connect(
    hostname="192.168.1.100",
    username="root",
    password="your_password",
    allow_agent=false,
    look_for_keys=false
)

# 或者自定义交互回调
def interactive_handler(title, instructions, prompt_list):
    answers = []
    for prompt in prompt_list:
        if "password" in prompt[0].lower():
            answers.append("your_password")
        else:
            answers.append("")
    return answers

ssh.connect(
    hostname="192.168.1.100",
    username="root",
    password="your_password",
    allow_agent=false,
    look_for_keys=false
)
print("连接成功")
ssh.close()

7.4、连接池与复用

对于批量操作，复用连接可以显著提升性能：

import paramiko
import threading

class sshconnectionpool:
    """简单的 ssh 连接池"""
    def __init__(self, hostname, username, password, pool_size=5):
        self.hostname = hostname
        self.username = username
        self.password = password
        self.pool_size = pool_size
        self._pool = []
        self._lock = threading.lock()
    
    def _create_connection(self):
        ssh = paramiko.sshclient()
        ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
        ssh.connect(
            hostname=self.hostname,
            username=self.username,
            password=self.password,
            timeout=10
        )
        return ssh
    
    def get_connection(self):
        with self._lock:
            if self._pool:
                return self._pool.pop()
            return self._create_connection()
    
    def return_connection(self, ssh):
        with self._lock:
            if len(self._pool) < self.pool_size:
                self._pool.append(ssh)
            else:
                ssh.close()
    
    def close_all(self):
        with self._lock:
            for ssh in self._pool:
                ssh.close()
            self._pool.clear()

# 使用连接池
pool = sshconnectionpool("192.168.1.100", "root", "your_password", pool_size=3)

def run_command(host, cmd):
    ssh = pool.get_connection()
    try:
        stdin, stdout, stderr = ssh.exec_command(cmd)
        return stdout.read().decode("utf-8")
    finally:
        pool.return_connection(ssh)

# 批量执行命令
results = []
for cmd in ["hostname", "uptime", "df -h"]:
    results.append(run_command("192.168.1.100", cmd))

for r in results:
    print(r)

# 程序结束时关闭所有连接
pool.close_all()

8、异常处理与调试

8.1 、常见异常类型

paramiko 在连接和操作过程中会抛出多种异常，了解它们有助于编写健壮的代码：

异常类	触发场景
`paramiko.authenticationexception`	用户名、密码或密钥认证失败
`paramiko.sshexception`	通用的 ssh 协议错误，如握手失败、通道异常
`paramiko.badhostkeyexception`	主机密钥与 known_hosts 中记录的不一致，可能存在中间人攻击
`paramiko.channelexception`	通道打开失败，如请求的远程端口不可达
`paramiko.sftp.sftperror`	sftp 操作失败，如文件不存在、权限不足
`socket.timeout`	连接或读取超时
`oserror`	网络层错误，如主机不可达、连接被拒绝

8.2 、统一异常处理示例

import paramiko
import socket

def safe_connect(hostname, username, password=none, key_filename=none, timeout=10):
    """带完整异常处理的连接函数"""
    ssh = paramiko.sshclient()
    ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
    try:
        ssh.connect(
            hostname=hostname,
            port=22,
            username=username,
            password=password,
            key_filename=key_filename,
            timeout=timeout
        )
        return ssh
    except paramiko.authenticationexception:
        print(f"[{hostname}] 认证失败，请检查用户名、密码或密钥")
    except paramiko.badhostkeyexception:
        print(f"[{hostname}] 主机密钥校验失败，可能存在中间人攻击")
    except paramiko.sshexception as e:
        print(f"[{hostname}] ssh 协议错误: {e}")
    except socket.timeout:
        print(f"[{hostname}] 连接超时")
    except oserror as e:
        print(f"[{hostname}] 网络错误: {e}")
    return none

# 使用示例
ssh = safe_connect("192.168.1.100", "root", password="your_password")
if ssh:
    try:
        stdin, stdout, stderr = ssh.exec_command("uptime")
        print(stdout.read().decode("utf-8"))
    finally:
        ssh.close()

8.3 、开启调试日志

排查问题时，可以开启 paramiko 的调试日志：

import paramiko
import logging

# 开启 paramiko 调试日志
logging.basicconfig(level=logging.debug)
paramiko.util.log_to_file("/tmp/paramiko.log", level=logging.debug)

ssh = paramiko.sshclient()
ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
ssh.connect("192.168.1.100", username="root", password="your_password")
print("连接成功")
ssh.close()

日志会记录完整的握手过程、密钥交换、认证流程和通道操作，便于定位问题。

9、最佳实践与性能优化

9.1、连接管理

复用连接：避免频繁创建和销毁连接，使用连接池（见 7.4 节）可以显著降低握手开销；
设置超时：始终为 connect 和 exec_command 设置合理的 timeout，防止程序卡死；
及时关闭：使用 with 语句或 try/finally 确保连接和 sftp 通道被正确关闭；
使用上下文管理器：paramiko 的 sshclient 支持上下文管理器协议：

import paramiko

with paramiko.sshclient() as ssh:
    ssh.set_missing_host_key_policy(paramiko.autoaddpolicy())
    ssh.connect("192.168.1.100", username="root", password="your_password")
    stdin, stdout, stderr = ssh.exec_command("hostname")
    print(stdout.read().decode("utf-8"))
# 退出 with 块时自动关闭连接

9.2、安全建议

优先使用密钥认证：避免密码在网络上明文传输，推荐使用 ed25519 密钥；
限制 known_hosts 策略：生产环境不要使用 autoaddpolicy，应使用 rejectpolicy（默认）并预先加载 known_hosts，防止中间人攻击；
最小权限原则：使用专用账号而非 root，并限制其可执行命令范围；
敏感信息保护：不要将密码硬编码在代码中，建议使用环境变量或密钥管理服务（如 vault）。

9.3、性能优化

批量命令合并：将多条命令用 && 或 ; 合并为一条执行，减少往返次数；
使用 sftp 批量传输：传输大量小文件时，优先使用 sftp 而非逐条执行 scp 命令；
调整窗口大小：大文件传输时，可通过 transport 的 window_size 参数调整 ssh 窗口大小，提升吞吐量。

10、总结

paramiko 是 python 生态中功能强大且稳定的 ssh 库，本文基于 3.x 最新版本，系统讲解了：

安装与版本特性：3.x 相比 2.x 的改进与兼容性；
核心组件：sshclient、transport、sftpclient、channel 等的关系与职责；
连接与命令执行：密码认证、密钥认证、交互式命令、实时输出读取；
sftp 文件传输：上传、下载、目录操作与递归传输；
高级特性：端口转发、代理连接、键盘交互认证、连接池复用；
异常处理与最佳实践：常见异常、调试日志、安全与性能优化建议。

掌握这些内容后，你可以用 paramiko 构建自动化运维脚本、批量部署工具、远程监控系统等。建议在实际项目中结合 asyncio 或线程池实现并发操作，进一步提升效率。

代码示例

import paramiko

def ssh_run_cmd(host: str, port: int, username: str, password: str, cmd: str):
    # 创建ssh客户端
    ssh_client = paramiko.sshclient()
    # 自动添加主机密钥，不校验known_hosts（测试用，生产不推荐）
    ssh_client.set_missing_host_key_policy(paramiko.autoaddpolicy())
    try:
        # 连接服务器
        ssh_client.connect(hostname=host, port=port, username=username, password=password, timeout=10)
        # 执行命令
        stdin, stdout, stderr = ssh_client.exec_command(cmd)
        # 读取输出，decode转字符串
        out = stdout.read().decode("utf-8")
        err = stderr.read().decode("utf-8")
        print("==== stdout ====")
        print(out)
        if err:
            print("==== stderr ====")
            print(err)
        return out, err
    except exception as e:
        print(f"ssh连接异常: {e}")
    finally:
        ssh_client.close()

if __name__ == "__main__":
    # 修改为你的服务器信息
    ssh_run_cmd(
        host="192.168.1.100",
        port=22,
        username="root",
        password="your_password",
        cmd="df -h"
    )

以上就是python paramiko库实现ssh远程操作的实战指南的详细内容，更多关于python paramiko库ssh远程操作的资料请关注代码网其它相关文章！

Python paramiko库实现SSH远程操作的实战指南

2026年10月11日 • Python •我要评论

1、引言

2、环境准备与安装

2.1、版本说明

2.2、安装 paramiko

3、核心组件与架构

4、基础连接与命令执行

4.1 、密码方式连接

4.2 、执行远程命令

4.3、带交互的命令执行

4.4、实时读取输出

5、密钥认证

5.1 、使用 rsa/ed25519 私钥

5.2 、支持 openssh 新格式私钥

5.3 、多密钥自动尝试

6、sftp 文件传输

6.1 、上传文件

6.2、下载文件

6.3 、目录操作

6.4、递归上传目录

7、高级特性

7.1 、端口转发（隧道）

7.2 、使用代理（proxycommand）

7.3 、键盘交互认证

7.4、连接池与复用

8、异常处理与调试

8.1 、常见异常类型

8.2 、统一异常处理示例

8.3 、开启调试日志

9、最佳实践与性能优化

9.1、连接管理

9.2、安全建议

9.3、性能优化

10、总结

代码示例

相关文章:

发表评论


验证码：

Python paramiko库实现SSH远程操作的实战指南

2026年10月11日 • Python •我要评论

1、 引言

2、环境准备与安装

2.1、 版本说明

2.2、 安装 paramiko

3、核心组件与架构

4、 基础连接与命令执行

4.1 、密码方式连接

4.2 、执行远程命令

4.3、 带交互的命令执行

4.4、 实时读取输出

5、 密钥认证

5.1 、使用 rsa/ed25519 私钥

5.2 、支持 openssh 新格式私钥

5.3 、多密钥自动尝试

6、sftp 文件传输

6.1 、上传文件

6.2、 下载文件

6.3 、目录操作

6.4、 递归上传目录

7、高级特性

7.1 、端口转发（隧道）

7.2 、使用代理（proxycommand）

7.3 、键盘交互认证

7.4、 连接池与复用

8、 异常处理与调试

8.1 、常见异常类型

8.2 、统一异常处理示例

8.3 、开启调试日志

9、 最佳实践与性能优化

9.1、 连接管理

9.2、 安全建议

9.3、 性能优化

10、 总结

代码示例

相关文章:

发表评论

1、引言

2.1、版本说明

2.2、安装 paramiko

4、基础连接与命令执行

4.3、带交互的命令执行

4.4、实时读取输出

5、密钥认证

6.2、下载文件

6.4、递归上传目录

7.4、连接池与复用

8、异常处理与调试

9、最佳实践与性能优化

9.1、连接管理

9.2、安全建议

9.3、性能优化

10、总结