阿杰 👍 我来给你整理一份 Python 库 Paramiko 的详解笔记,带上用法示例,方便你快速上手。
📌 Python 库 Paramiko
1️⃣ 简介
- Paramiko 是 Python 实现的 SSHv2 协议库
- 功能:支持远程服务器 SSH 登录、命令执行、文件传输 (SFTP)
- 常见应用场景:
- 自动化运维
- 批量执行命令
- 上传/下载文件
- 远程管理 Linux 服务器
安装:
pip install paramiko
2️⃣ 基本用法
(1)远程执行命令
import paramiko
# 创建 SSH 客户端
ssh = paramiko.SSHClient()
# 自动添加主机密钥(避免手动确认)
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname="192.168.1.10", port=22, username="root", password="your_password")
# 执行命令
stdin, stdout, stderr = ssh.exec_command("ls -l /home")
# 获取输出
print(stdout.read().decode())
print(stderr.read().decode())
# 关闭连接
ssh.close()
(2)使用私钥登录
private_key = paramiko.RSAKey.from_private_key_file("/home/user/.ssh/id_rsa")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname="192.168.1.10", port=22, username="root", pkey=private_key)
stdin, stdout, stderr = ssh.exec_command("whoami")
print(stdout.read().decode())
ssh.close()
(3)SFTP 文件传输
import paramiko
transport = paramiko.Transport(("192.168.1.10", 22))
transport.connect(username="root", password="your_password")
sftp = paramiko.SFTPClient.from_transport(transport)
# 上传文件
sftp.put("local.txt", "/home/root/remote.txt")
# 下载文件
sftp.get("/home/root/remote.txt", "local_copy.txt")
sftp.close()
transport.close()
(4)保持长连接 & 交互式 Shell
import paramiko
import time
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("192.168.1.10", 22, "root", "your_password")
# 启动交互式 Shell
channel = ssh.invoke_shell()
channel.send("cd /var/log\n")
time.sleep(1)
channel.send("ls -l\n")
time.sleep(2)
output = channel.recv(65535).decode()
print(output)
ssh.close()
3️⃣ 常见问题
- 连接超时 / 密钥不匹配
- 检查
known_hosts
文件 - 或者设置:
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- 检查
- 执行命令卡住
- 可能是命令需要交互式输入
- 解决:用
invoke_shell()
而不是exec_command()
- 大文件传输慢
- 考虑分块传输,或者用
scp
模块(基于 paramiko 实现)
- 考虑分块传输,或者用
4️⃣ 高级用法
- 批量服务器执行命令(循环
ssh.connect
) - 多线程并发执行(结合
concurrent.futures.ThreadPoolExecutor
) - 安全控制(限制 host key 策略、防止 MITM 攻击)
- 异常捕获:
paramiko.ssh_exception.AuthenticationException
等
好的,阿杰 👍 我给你写一个 批量远程服务器执行命令的小脚本,支持 密码登录 / 私钥登录,适合做自动化运维。
📌 Python 脚本:批量执行远程命令(Paramiko)
import paramiko
from concurrent.futures import ThreadPoolExecutor
# ===== 配置区 =====
# 服务器列表(支持密码或私钥)
servers = [
{"host": "192.168.1.10", "port": 22, "user": "root", "password": "your_password"},
{"host": "192.168.1.11", "port": 22, "user": "root", "key": "/home/user/.ssh/id_rsa"},
]
# 要执行的命令
command = "hostname && uptime"
def run_command(server, command):
host = server["host"]
port = server.get("port", 22)
user = server.get("user", "root")
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if "password" in server:
ssh.connect(host, port, user, server["password"], timeout=10)
elif "key" in server:
private_key = paramiko.RSAKey.from_private_key_file(server["key"])
ssh.connect(host, port, user, pkey=private_key, timeout=10)
else:
raise ValueError("必须提供 password 或 key")
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read().decode()
error = stderr.read().decode()
print(f"\n[{host}] 执行结果:")
if result:
print(result.strip())
if error:
print("错误:", error.strip())
ssh.close()
except Exception as e:
print(f"[{host}] 连接失败: {e}")
# ===== 主程序入口 =====
if __name__ == "__main__":
# 使用多线程并发执行
with ThreadPoolExecutor(max_workers=5) as executor:
for server in servers:
executor.submit(run_command, server, command)
🔹 用法
- 修改
servers
列表,把你的服务器 IP、用户名、密码/密钥路径写进去 - 修改
command
变量(要执行的远程命令) - 运行脚本:
python batch_ssh.py
🔹 输出示例
[192.168.1.10] 执行结果:
myserver1
15:20:01 up 5 days, 4:13, 2 users, load average: 0.05, 0.10, 0.12
[192.168.1.11] 执行结果:
myserver2
15:20:02 up 3 days, 7:42, 1 user, load average: 0.00, 0.01, 0.05
发表回复