Paramiko SSH 自动化详解

核心概念 Paramiko 是 Python SSHv2 协议库，提供远程连接和文件传输能力。

1. 前置基础

1.1. 架构分层

Paramiko 三层结构，自上而下：

SSHClient（应用层封装，日常使用）
封装 Transport、Channel、主机密钥校验，屏蔽底层复杂逻辑，适合简单运维场景。
Transport（传输层核心通道）
对应 SSH 协议的 TCP 加密连接，一个 TCP 连接 = 一个 Transport；
单个 Transport 可以多路复用多个 Channel（同时执行命令、SFTP、端口转发，不重复建TCP）。
Channel（逻辑会话通道）
一条加密TCP连接内的独立虚拟管道：

exec_command 执行单条命令 = 新建一个 Channel
invoke_shell 交互式终端 = 一个长期 Channel
open_sftp 文件传输 = 独立SFTP Channel 关键知识点：
1 台机器1个Transport，可同时开N个Channel，复用TCP握手，性能远优于反复创建SSHClient；
所有 SSHClient 底层都是创建 Transport。

1.2. 安装依赖说明


# paramiko核心

pip install paramiko

# cryptography：加密底层，加速RSA/AES，必须装，否则加解密极慢

pip install cryptography

# 额外：ecdsa 兼容ECDSA私钥（云厂商常用ecdsa密钥）

pip install ecdsa

一、SSHClient 全参数拆解 + 底层逻辑

2. set_missing_host_key_policy 三种策略详细区别

SSH 首次连接会获取服务端主机公钥指纹，用来防止中间人攻击，三种策略安全等级不同：

paramiko.AutoAddPolicy()

行为：首次连接自动保存主机指纹到本地 ~/.ssh/known_hosts，后续自动校验；
适用：自动化脚本、批量巡检、内网可信服务器；
风险：公网未知机器存在中间人劫持风险。

paramiko.RejectPolicy()（默认策略）

行为：陌生主机直接抛出 NoKnownHostKeys 异常，拒绝连接；
适用：严格安全生产环境，手动预录入服务器指纹。

paramiko.WarningPolicy()

行为：打印警告日志，但允许连接，不保存指纹；
适用：测试环境，不推荐线上。

3. connect() 全部参数完整释义


ssh.connect(

    hostname,           # 主机IP/域名

    port=22,

    username=None,

    password=None,      # 密码登录

    pkey=None,          # RSAKey/ECDSAKey 对象，私钥登录

    key_filename=None,  # 私钥文件路径字符串，内部自动加载成pkey

    timeout=None,       # TCP握手超时（socket层），单位秒，卡死必设

    auth_timeout=None,  # 认证超时（TCP连上后账号密码校验超时）

    banner_timeout=None,# 等待服务器SSH banner信息超时

    allow_agent=False,  # 是否使用本地ssh-agent私钥

    look_for_keys=True, # 是否自动去~/.ssh下寻找id_rsa等私钥

    compress=False,     # 是否开启传输压缩，带宽低的机器开启

)

3.1. 登录优先级（从高到低）

pkey参数对象 > key_filename文件 > password密码 > 本地ssh-agent自动私钥

4. exec_command 深度详解（通道、缓冲区、阻塞问题）

4.1. 完整函数签名


stdin, stdout, stderr = ssh.exec_command(command, bufsize=-1, timeout=None, get_pty=False)

4.2. 参数细分

bufsize：缓冲区大小，默认-1系统默认；大输出建议设4096，分次读取防止缓冲区塞满阻塞。
timeout：命令执行超时（区别于connect的TCP超时），命令运行超过时间直接关闭通道，避免死锁。
get_pty=True：分配伪终端，作用：

执行sudo、需要交互式密码输入的命令；
部分Linux命令（top、vmstat）无pty不会输出彩色/完整日志；
弊端：stdout和stderr会合并到同一个输出流，无法区分标准输出与错误。

4.3. 返回值 ChannelFile 对象

stdin / stdout / stderr 不是普通字符串，是封装了通道的文件类对象，底层共享同一个SSH Channel：

stdout.read()：一次性读取通道全部输出，阻塞直到命令执行完毕；
stdout.readline()：逐行读取，有数据立刻返回，无数据阻塞；
stdout.channel：底层原始Channel对象，可以操作超时、判断是否有可读数据。

4.4. 高频大坑：缓冲区满导致进程卡死

现象：执行输出几万行日志的命令，脚本永久卡住不动。

原理：SSH Channel存在收发缓冲区，缓冲区填满后服务端进程会阻塞写入，等待客户端读取；

解决方案两种：

循环readline边读边消费缓冲区（推荐）；
开启多线程分别读取stdout、stderr，防止单边缓冲区占满。示例：循环读取避免阻塞


def safe_exec(ssh, cmd, timeout=30):

    stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)

    out_lines = []

    err_lines = []

    # 循环读取stdout，释放缓冲区

    while not stdout.channel.exit_status_ready():

        line = stdout.readline()

        if not line:

            break

        out_lines.append(line)

    # 读取剩余错误输出

    err = stderr.read().decode("utf8", errors="ignore")

    code = stdout.channel.recv_exit_status()

    out = "".join(out_lines).strip()

    return code, out, err

4.5. recv_exit_status() 底层逻辑

stdout.channel.recv_exit_status() 会阻塞等待命令执行结束，才能拿到退出码；

如果不调用该方法直接关闭连接，服务器进程可能变成僵尸进程。

5. invoke_shell 交互式终端完整细节

exec_command 是一次性会话，执行完销毁通道；invoke_shell 创建长期交互式shell，等价手动ssh登录后的终端。

5.1. 函数签名


stdin, stdout, stderr = ssh.invoke_shell(term='vt100', width=80, height=24)

term：终端类型，影响命令行颜色、光标；
width/height：终端行列，调整ls、top显示排版。

5.2. 核心API（实时流式输出关键）

stdout.channel.recv_ready()
非阻塞判断：缓冲区是否有新数据，返回True/False，配合循环实现实时打印，不会卡死脚本。
stdout.channel.setblocking(0)
设置通道非阻塞模式，readline无数据时直接返回空字符串，不用等待。
stdin.write("指令\n")发送命令必须加换行 \n，等价回车；不换行服务器不会执行。
stdin.flush() 刷新发送缓冲区，确保指令立刻下发。完整实时流式示例（tail -f实时跟踪日志）


import time

ssh = paramiko.SSHClient()

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

ssh.connect("192.168.1.100", username="root", password="xxx")

stdin, stdout, stderr = ssh.invoke_shell(width=120, height=30)

# 发送跟踪日志命令

stdin.write("tail -f /var/log/messages\n")

stdin.flush()

# 循环实时读取输出

while True:

    # 判断是否有新数据

    if stdout.channel.recv_ready():

        data = stdout.read().decode("utf8", errors="ignore")

        print(data, end="")

    time.sleep(0.2)

5.3. 缺点

stdout/stderr混合输出，无法区分正常日志与错误日志；适合堡垒机展示终端，不适合自动化巡检。

6. 私钥加载全场景细分（RSA/ECDSA/加密私钥）

6.1. 场景1：未加密RSA私钥（最常用）


from paramiko import RSAKey

pri_key = RSAKey.from_private_key_file("/root/.ssh/id_rsa")

ssh.connect(host, username="root", pkey=pri_key)

6.2. 场景2：带密码保护的私钥（私钥加密）


pri_key = RSAKey.from_private_key_file("/root/id_rsa", password="私钥解密密码")

6.3. 场景3：ECDSA椭圆加密私钥（阿里云/腾讯云默认密钥）


from paramiko import ECDSAKey

pri_key = ECDSAKey.from_private_key_file("/root/id_ecdsa")

6.4. 场景4：直接读取私钥字符串（数据库存储密钥，不落地文件）


key_text = """-----BEGIN RSA PRIVATE KEY-----

xxxx密钥内容

-----END RSA PRIVATE KEY-----

"""

from io import StringIO

file_obj = StringIO(key_text)

pri_key = RSAKey.from_private_key(file_obj)

二、SFTPClient 文件传输完整高级详解

7. 获取SFTP通道两种方式

简易：sftp = ssh.open_sftp()
底层Transport手动创建：


transport = ssh.get_transport()

sftp = transport.open_sftp_client()

8. 所有方法细分+边界场景

8.1. 上传下载

put(local, remote, callback=None)
callback 回调函数，可实时获取上传进度，做大文件传输进度条；
get(remote, local, callback=None) 下载进度回调同理。

8.2. 目录操作

listdir(path)：仅返回文件名列表
listdir_attr(path)：返回完整文件对象，包含大小、修改时间、权限（生产巡检必备）


sftp = ssh.open_sftp()

files = sftp.listdir_attr("/tmp")

for f in files:

    print(f.filename, f.st_size, f.st_mode)

mkdir(path, mode=0o755) 创建远程目录，mode必须八进制 0o 开头
rmdir(path) 仅删除空目录，非空目录会抛异常，需要递归删除自行遍历实现

8.3. 文件权限、软硬链接

chmod(path, 0o644) 修改权限
chown(path, uid, gid) 修改属主属组
symlink(source, dest) 创建软链接
lstat(path) 不跟随软链接获取文件信息；stat(path) 跟随软链接

8.4. 大文件断点续传原生不支持

paramiko sftp无内置断点续传，两种解决方案：

结合文件大小判断，分块上传；
使用rclone替代大文件传输，paramiko仅处理小配置文件。

9. SFTP 常见异常

FileNotFoundError：远程路径不存在
IOError: Permission denied 读写权限不足
SFTPError：服务端sftp-server异常、磁盘满

三、底层 Transport、跳板机代理完整实现（堡垒机核心难点）

10. Transport 手动创建连接（脱离SSHClient）


import paramiko

transport = paramiko.Transport(("192.168.1.100", 22))

transport.connect(username="root", password="xxx")

# 手动创建命令通道

channel = transport.open_session()

channel.exec_command("df -h")

out = channel.recv(4096).decode()

print(out)

transport.close()

优势：一条Transport复用多个Channel，多命令执行不用重复TCP握手。

11. 跳板机（代理跳转）完整代码（最常用内网场景）

需求：公网跳板机10.0.0.1，内网机器172.16.0.10无法直连，通过跳板机转发通道。

原理：先建立跳板Transport，通过跳板建立TCP转发通道，再在内网机器新建Transport。


import paramiko

# 1. 连接跳板机

jump_trans = paramiko.Transport(("跳板IP", 22))

jump_trans.connect(username="jumpuser", password="jumppwd")

# 2. 在跳板机上打开转发通道，连接内网机器22端口

channel = jump_trans.open_channel("direct-tcpip", ("内网机器IP",22), ("",0))

# 3. 通过转发通道建立内网SSH Transport

inner_trans = paramiko.Transport(channel)

inner_trans.connect(username="root", password="innerpwd")

# 4. 内网机器执行命令

inner_channel = inner_trans.open_session()

inner_channel.exec_command("hostname")

print(inner_channel.recv(1024).decode())

# 逐层关闭

inner_trans.close()

jump_trans.close()

12. 端口转发（本地端口映射远程内网服务）


# 本地8080端口，转发到远程127.0.0.1:80网页服务

transport = paramiko.Transport((host,22))

transport.connect(...)

transport.request_port_forward("", 8080, ("127.0.0.1", 80))

四、并发/多线程使用规范（生产批量巡检必看）

13. 核心硬性规则

一个SSHClient/Transport 只能单线程使用，不支持多线程共享连接；
多台机器并发：每个线程独立创建SSHClient，线程内自有连接；
线程池批量执行，控制并发数，避免一次性创建上千TCP连接打满文件句柄。线程池示例：


from concurrent.futures import ThreadPoolExecutor

def task(host):

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(host, username="root", password="xxx", timeout=8)

    code, out, err = ssh.exec_command("free -h")

    ssh.close()

    return host, out

# 最多并发10台机器

with ThreadPoolExecutor(max_workers=10) as pool:

    hosts = ["192.168.1.10", "192.168.1.11"]

    res = pool.map(task, hosts)

14. 句柄泄露坑

循环批量创建SSHClient，不执行 ssh.close() / transport.close() 会持续占用TCP句柄，最终触发 Too many open files；

标准写法：try ... finally 强制关闭连接。

五、完整异常体系（所有可捕获异常分类）

15. 认证相关

paramiko.AuthenticationException：账号密码错误、私钥无效、无权限登录

16. 主机密钥

paramiko.NoKnownHostsError：陌生主机指纹拒绝连接（RejectPolicy）

17. 网络连接

socket.timeout：TCP握手超时，服务器22不通/防火墙拦截
paramiko.NoConnectionsException：连接已关闭，通道失效
paramiko.ssh_exception.SSHException：SSH协议握手失败（服务器不是SSH服务）

18. SFTP 文件操作

FileNotFoundError 文件不存在
IOError 权限不足、磁盘满

19. 通道超时

命令执行超时抛出 socket.timeout

标准全捕获模板


try:

    ssh.connect(...)

    code, out, err = ssh.exec_command("df -h", timeout=15)

except paramiko.AuthenticationException:

    print("认证失败")

except socket.timeout:

    print("连接/命令超时")

except paramiko.NoKnownHostsError:

    print("未知主机，拒绝连接")

except paramiko.SSHException as e:

    print("SSH协议异常", e)

except Exception as e:

    print("未知异常", repr(e))

finally:

    ssh.close()

六、生产高频踩坑深度解析

20. 中文乱码分层解决方案

绝大多数Linux服务器输出UTF-8：


out = stdout.read().decode("utf-8", errors="ignore")

老旧CentOS6、国产系统GBK编码：


out = stdout.read().decode("gbk", errors="replace")

errors参数说明：

ignore：直接丢弃无法解析的字符；
replace：乱码字符替换为�，推荐日志场景。

21. sudo 需要交互式密码（get_pty=True 必须开启）

普通exec_command无法输入sudo密码，分配伪终端后才能stdin写入密码：


stdin, stdout, stderr = ssh.exec_command("sudo df -h", get_pty=True)

stdin.write("sudo密码\n")

stdin.flush()

res = stdout.read().decode()

22. 命令注入边界说明

paramiko传输命令是协议层发送字符串，不会本地解析shell；

但远程服务器的shell会解析特殊符号 ; | && > $，例如：


# 危险，远程会分两条命令执行

cmd = f"echo {user_input}; rm -rf /tmp/*"

ssh.exec_command(cmd)

解决方案：高危场景使用远程脚本传参，或过滤输入特殊字符。

23. 长时间运行通道僵尸进程

exec_command执行长时间任务，脚本提前关闭连接，服务器进程不会自动终止，留在后台占用资源；

解决：执行结束务必调用 channel.recv_exit_status() 等待进程退出，再关闭连接。

24. 压缩传输开启场景

跨公网、低带宽机器开启压缩，降低流量：


ssh.connect(..., compress=True)

七、Paramiko vs subprocess ssh 深度对比（补足之前简略内容）

| | | |

|---|---|---|

|对比项|Paramiko|subprocess调用系统ssh|

|依赖|纯Python库，无系统ssh客户端|必须操作系统openssh二进制|

|TCP复用|Transport多路复用Channel，多命令单TCP|每条命令新建独立TCP进程，开销大|

|注入风险|仅远程shell解析特殊符号，可控|shell=True存在严重本地命令注入漏洞|

|SFTP文件传输|原生内置一体化|需要单独调用scp，代码割裂、多子进程|

|交互式终端|原生invoke_shell实时流式输出|实现复杂，需要管道实时读取，易卡死|

|跳板机/代理转发|底层Transport原生支持，代码可控|依赖~/.ssh/config配置，动态IP很难处理|

|网页后端/堡垒机|无本地进程逃逸风险，适合Web后台|会生成本地子进程，存在权限逃逸安全隐患|

|并发批量|线程独立连接，资源可控|大量子进程，容易产生僵尸进程、耗尽句柄|

|Windows兼容|跨平台统一代码|Windows无ssh，必须安装OpenSSH客户端|

八、面向AIOps/自研堡垒机的生产级完整封装类（整合通道复用、跳板、SFTP、超时、异常、自动关闭）


import paramiko

import socket

from paramiko.ssh_exception import AuthenticationException, NoKnownHostsError, SSHException

class ParamikoSSH:

    def __init__(self, host, port=22, username="root", password=None, pri_key_path=None, timeout=10):

        self.host = host

        self.port = port

        self.user = username

        self.pwd = password

        self.key_path = pri_key_path

        self.timeout = timeout

        self.transport = None

        self.sftp = None

    def connect(self):

        """建立Transport底层通道，复用逻辑"""

        self.transport = paramiko.Transport((self.host, self.port))

        self.transport.set_keepalive(30) # 30s保活，防止空闲断开

        try:

            if self.key_path:

                if self.key_path.endswith(".ecdsa"):

                    pri_key = paramiko.ECDSAKey.from_private_key_file(self.key_path)

                else:

                    pri_key = paramiko.RSAKey.from_private_key_file(self.key_path)

                self.transport.connect(username=self.user, pkey=pri_key, auth_timeout=self.timeout)

            else:

                self.transport.connect(username=self.user, password=self.pwd, auth_timeout=self.timeout)

        except AuthenticationException:

            raise Exception("账号/私钥认证失败")

        except socket.timeout:

            raise Exception("TCP连接超时，端口不通")

        except NoKnownHostsError:

            raise Exception("未知主机指纹，拒绝连接")

        except SSHException as e:

            raise Exception(f"SSH握手失败: {str(e)}")

    def exec_cmd(self, cmd, cmd_timeout=20, get_pty=False):

        """执行远程命令，单Transport多路复用Channel"""

        if not self.transport or not self.transport.is_active():

            self.connect()

        channel = self.transport.open_session()

        channel.settimeout(cmd_timeout)

        if get_pty:

            channel.get_pty(width=120, height=30)

        channel.exec_command(cmd)

        # 分次读取缓冲区，防止阻塞

        out_buf = []

        err_buf = []

        while not channel.exit_status_ready():

            if channel.recv_ready():

                out_buf.append(channel.recv(4096))

            if channel.recv_stderr_ready():

                err_buf.append(channel.recv_stderr(4096))

        # 读取剩余数据

        out_buf.append(channel.recv(4096))

        err_buf.append(channel.recv_stderr(4096))

        code = channel.recv_exit_status()

        out = b"".join(out_buf).decode("utf-8", errors="ignore").strip()

        err = b"".join(err_buf).decode("utf-8", errors="ignore").strip()

        channel.close()

        return code, out, err

    def get_sftp(self):

        if not self.sftp and self.transport.is_active():

            self.sftp = self.transport.open_sftp_client()

        return self.sftp

    def close(self):

        """逐层释放资源，防止句柄泄露"""

        if self.sftp:

            self.sftp.close()

            self.sftp = None

        if self.transport:

            self.transport.close()

            self.transport = None

关联文档

Netmiko 网络自动化详解

Python 基础