做运维或后端开发，迟早会碰到这类需求：批量登录几十台服务器执行命令、上传下发文件、自动跑部署脚本。用 ssh 命令当然可以，但要在程序里精细控制、循环执行、统一收集结果时就很难受——要么靠 subprocess 拼 shell 字符串，要么依赖系统装了 sshpass，既不优雅也不跨平台。
paramiko 就是为此而生的：它用纯 Python 实现了 SSH2 协议，让你无需依赖系统的 ssh 客户端，就能在代码里完成远程连接、命令执行、文件传输（SFTP）等操作。它是很多运维自动化工具（如早期的 Ansible、Fabric）的底层依赖，是 Python 世界里最主流的 SSH 库。
它的核心能力：
- • SSH 客户端：连接远程主机，执行命令、拿到标准输出/错误与退出码。
- • SFTP 客户端：上传、下载、遍历、重命名、删除远程文件。
- • 密钥认证：支持 RSA / ECDSA / Ed25519 私钥，比密码更安全。
- • 底层可控：
Transport 与 Channel 让你能复用连接、开交互式会话。
一、安装
pip install paramiko
paramiko 的加密运算依赖 cryptography，pip 会自动一并装上。
最小示例
安装好，我们创建一个 main.py，用密码连上一台远程主机并执行命令：
import paramiko
client = paramiko.SSHClient()
# 自动接受未知主机密钥（仅演示用，生产环境见「注意事项」）
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 换成你自己的 IP、用户名、密码
client.connect("192.168.1.100"
, username="root", password="你的密码")
stdin, stdout, stderr = client.exec_command("whoami")
print("远程返回:", stdout.read().decode().strip()) # 远程返回: root
client.close()
运行 python main.py，终端会打印出远程主机上 whoami 的结果（如 root）。
二、具体执行
本节把 paramiko 的常用能力拆开讲：建立连接与主机密钥策略、执行命令、密钥认证、SFTP 文件传输。
2.1 建立连接：SSHClient.connect()
SSHClient 是最高层的封装，代表一条 SSH 连接。常用参数：
client.connect(
hostname="192.168.1.100", # 主机名或 IP
port=22, # 端口，默认 22
username="root", # 用户名
password="你的密码", # 密码认证（与 pkey 二选一）
pkey=None, # 私钥对象（见 2.3）
key_filename=None, # 或直接给私钥文件路径
timeout=10, # 连接超时（秒）
allow_agent=True, # 允许使用 ssh-agent
look_for_keys=True, # 自动查找 ~/.ssh 下的密钥
)
只填 hostname 和 username 时，paramiko 会自动尝试 ssh-agent 和 ~/.ssh/ 里的密钥（look_for_keys 会依次找 ~/.ssh/id_rsa、id_ecdsa、id_ed25519，Windows 上还会多找一条 ~/ssh/）。想强制只用密码，就关掉这两个开关：
client.connect("192.168.1.100", username="root", password="你的密码",
allow_agent=False, look_for_keys=False)2.2 主机密钥策略：防中间人攻击
SSH 靠"主机密钥"确认对面是不是真的那台机器。paramiko 有三种策略：
# 1. 最安全：只信任 known_hosts 里已有的主机，不认识就拒绝
client.load_system_host_keys() # 载入 ~/.ssh/known_hosts
client.set_missing_host_key_policy(paramiko.RejectPolicy())
# 2. 自动添加未知主机（方便，但有中间人风险，仅适合测试）
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 3. 遇到未知主机只发警告，仍继续连接
client.set_missing_host_key_policy(paramiko.WarningPolicy())
生产环境务必用 RejectPolicy + load_system_host_keys()，或手动校验指纹，不要图省事一律 AutoAddPolicy。
2.3 密钥认证
比密码更安全、也更适合自动化的方式是用私钥，加载后通过 pkey 传入：
import paramiko
# RSA 私钥
key = paramiko.RSAKey.from_private_key_file("/home/user/.ssh/id_rsa")
# Ed25519 私钥（更现代、更推荐）
key = paramiko.Ed25519Key.from_private_key_file("/home/user/.ssh/id_ed25519")
# 带口令（passphrase）的加密私钥
key = paramiko.Ed25519Key.from_private_key_file(
"/home/user/.ssh/id_ed25519", password="私钥口令"
)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect("192.168.1.100", username="root", pkey=key)
也可以直接把路径交给 connect()，省去手动加载：
client.connect("192.168.1.100", username="root",
key_filename="/home/user/.ssh/id_ed25519")
私钥在哪个目录？ Linux 是 /home//.ssh/，Windows 是 C:\Users\\.ssh\（即 %USERPROFILE%\.ssh\），文件名通常是 id_rsa 或 id_ed25519。别把用户名写死，用 Path.home() 自动定位：
from pathlib import Path
import paramiko
key = paramiko.RSAKey.from_private_key_file(
str(Path.home() / ".ssh" / "id_rsa") # Windows 上即 C:\Users\\.ssh\id_rsa
)
⚠️ Windows 路径里的反斜杠是坑："C:\Users\\.ssh\id_rsa" 里的 \U 会被 Python 当成 Unicode 转义而报 SyntaxError。请用原始字符串 r"C:\Users\..."，或统一写正斜杠 "C:/Users/..."。
不想自己判断密钥类型，可以用 PKey.from_path() 让它自动识别：
key = paramiko.PKey.from_path(str(Path.home() / ".ssh" / "id_ed25519"))
密钥对象的常用属性与方法：
print(key.get_name()) # 'ssh-ed25519'（协议名）
print(key.algorithm_name) # 'ED25519'（纯算法名，大写）
print(key.get_bits()) # 密钥长度（位）
print(key.fingerprint) # 指纹
print(key.can_sign()) # 是否含私钥部分（能否用于认证）
PKey没有.name 属性，要取名字请用 get_name() 或 algorithm_name。
2.4 执行命令：exec_command()
exec_command() 会新开一个通道执行命令，返回 (stdin, stdout, stderr) 三个类文件对象：
stdin, stdout, stderr = client.exec_command("ls -l /tmp")
out = stdout.read().decode() # 标准输出（bytes → str）
err = stderr.read().decode() # 标准错误
code = stdout.channel.recv_exit_status() # 退出码，0 表示成功
print
("输出:\n", out)
if err:
print("错误:\n", err)
print("退出码:", code)
签名如下：
exec_command(command, bufsize=-1, timeout=None, get_pty=False, environment=None)
- •
get_pty=True：分配伪终端，适合那些「检测到非交互环境就改变行为」的命令。 - •
environment：传给远程进程的环境变量字典。
如果命令需要交互（比如 sudo 要输密码），往返回的 stdin 里写即可：
stdin, stdout, stderr = client.exec_command("sudo -S whoami")
stdin.write("你的密码\n") # -S 让 sudo 从 stdin 读密码
stdin.flush()
print(stdout.read().decode())
记住一个顺序：先读完 stdout/stderr，再取退出码。输出很大又不及时读取，缓冲区写满会导致命令卡死。
2.5 SFTP 文件传输
用 open_sftp() 打开一条 SFTP 会话，之后就能像操作本地文件一样操作远程：
sftp = client.open_sftp()
# 上传 / 下载
sftp.put("local.txt", "/remote/path/local.txt") # 本地 → 远程
sftp.get("/remote/file.txt", "local_copy.txt") # 远程 → 本地
# 列出目录
print(sftp.listdir("/remote/dir")) # 只返回文件名列表
for attr in sftp.listdir_attr("/remote/dir"): # 返回带属性的对象
print(f"{attr.filename:30s} {attr.st_size:>10d} 字节")
# 文件信息 / 重命名 / 删除 / 建目录
stat = sftp.stat("/remote/file.txt")
sftp.rename("/remote/a.txt", "/remote/b.txt")
sftp.remove("/remote/tmp.txt")
sftp.mkdir("/remote/newdir")
sftp.close()
client.close()
三、优化处理
基础用法掌握后，来看几个真实场景里的关键点。
3.1 用上下文管理器 + 异常处理
SSHClient 支持 with，退出时自动关闭，配合异常处理更稳健：
import paramiko
try:
with paramiko.SSHClient() as client:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect("192.168.1.100", username="root",
password="你的密码", timeout=10
)
_, stdout, _ = client.exec_command("uptime")
print(stdout.read().decode())
except paramiko.AuthenticationException:
print("认证失败：用户名/密码/密钥不正确")
except paramiko.SSHException as e:
print(f"SSH 协议错误: {e}")
3.2 递归上传整个目录
put / get 只传单个文件，递归目录需要自己遍历：
import os
def upload_dir(sftp, local_dir, remote_dir):
try:
sftp.mkdir(remote_dir)
except IOError:
pass # 目录已存在，忽略
for name in os.listdir(local_dir):
local_path = os.path.join(local_dir, name)
remote_path = f"{remote_dir}/{name}"
if os.path.isdir(local_path):
upload_dir(sftp, local_path, remote_path) # 递归子目录
else:
sftp.put(local_path, remote_path)
upload_dir(client.open_sftp(), "./build", "/var/www/html")
3.3 交互式会话：invoke_shell()
有些设备或工具必须要有真正的终端才能工作（网络设备 CLI、需要 TTY 的交互程序）。这时用 invoke_shell()：
import time
channel = client.invoke_shell()
channel.send("cd /tmp\n")
time.sleep(0.5)
channel.send("ls -l\n")
time.sleep(1)
print(channel.recv(65535).decode())
3.4 实战：并发在多台主机上执行命令
paramiko 的 SSHClient不是线程安全的，所以每台主机各建一条独立连接，用线程池并发：
import paramiko
from concurrent.futures import ThreadPoolExecutor
HOSTS = ["192.168.1.101", "192.168.1.102", "192.168.1.103"]
def run_on_host(host):
try:
with paramiko.SSHClient() as client:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(host, username="root", password="你的密码",timeout=10)
_, stdout, _ = client.exec_command("hostname && uptime")
return host, stdout.read().decode().strip()
except Exception as e:
return host, f"失败: {e}"
with ThreadPoolExecutor(max_workers=10) as pool:
for host, result in pool.map(run_on_host, HOSTS):
print(f"=====
{host} =====")
print(result)
3.5 底层玩法：复用 Transport
要在同一台主机上开很多通道（既跑命令又传文件），用 Transport 复用一条底层连接更高效：
import paramiko
transport = paramiko.Transport(("192.168.1.100", 22))
transport.connect(username="root", password="你的密码")
sftp = paramiko.SFTPClient.from_transport(transport) # 从同一个 transport 开 SFTP
print(sftp.listdir("/tmp"))
channel = transport.open_session() # 也可以直接开通道跑命令
channel.exec_command("df -h")
print(channel.recv(65535).decode())
sftp.close()
transport.close()
四、「常见问题」
1、报 AuthenticationException 认证失败：检查用户名、密码或私钥是否正确；若主机只允许密钥登录，密码认证也会失败。
2、报 PasswordRequiredException：私钥是加密的，加载时没提供口令。给 from_private_key_file(path, password="口令") 传入 passphrase。
3、报 Server 'x.x.x.x' not found in known_hosts：用了 RejectPolicy 但主机不在 known_hosts 里。先 ssh user@host 手动记录一次，或执行 ssh-keyscan host >> ~/.ssh/known_hosts。
4、报 Private key file is accessible to others：私钥权限太开放，执行 chmod 600 私钥文件（Windows 上则是检查文件 ACL）。
5、exec_command 很慢或卡住：多为输出量大未及时读取，或命令需要 TTY。可改用循环读取、设置 timeout，或加 get_pty=True。
6、连接超时 / timed out：目标端口不通、防火墙拦截或 IP 有误。先用 ssh 命令确认能连通，再排查 timeout 设置。
7、SFTP 上传报 Permission denied：远程目标目录当前用户不可写。换有写权限的目录，或先调整权限。
8、怎么做到免密登录？客户端生成密钥对（
ssh-keygen -t ed25519），把公钥追加到远程 ~/.ssh/authorized_keys，脚本里用 pkey= 或 key_filename= 指定私钥即可。
9、pysftp 还要用吗？它早已停止维护，直接用 paramiko 自身的 open_sftp() 即可。
从连接一台主机、执行一条命令，到密钥认证、SFTP 传文件，再到并发批量运维与 Transport 复用——paramiko 用纯 Python 把 SSH 完整搬进了代码，不依赖系统 ssh 客户端，因此在任何 Python 环境里都能跑。收紧主机密钥策略、别把密码写进代码，剩下的，就是把你手上那些重复的运维动作一段段换成脚本。
