Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了
Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了
写给谁看:运维兄弟、刚学 Python 想做自动化的同学、被重复手工操作折磨的打工人
需要什么基础:会装 Python、会 pip 安装包即可
目标:看完就能上手,把日常手动操作变成脚本一键搞定
文章目录
一、先认识一下 paramiko 是什么
paramiko 是一个 Python 库,专门用来做 SSH 相关的操作。
平时我们用 XShell、PuTTY、FinalShell 连服务器,底层走的就是 SSH 协议。paramiko 相当于把这些操作搬到了 Python 代码里——你不用打开那些软件,写几行脚本就能完成登录、敲命令、拿结果、传文件这些事。
安装方法:
pip install paramiko
如果网络慢,加个国内源:
pip install paramiko -i https://pypi.tuna.tsinghua.edu.cn/simple
装完就可以用了。
二、最基础的:SSH 登录 + 执行一条命令
先写一个能跑起来的 demo,连上服务器、执行一条命令、看输出:
import paramiko
# 创建 SSH 客户端
client = paramiko.SSHClient()
# 自动接受陌生服务器的密钥(不弹确认框)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连上服务器
client.connect(
hostname='192.168.1.100', # 改成你的服务器 IP
port=22, # SSH 端口,默认 22
username='root', # 登录用户名
password='你的密码' # 登录密码
)
# 执行一条命令
stdin, stdout, stderr = client.exec_command('free -h')
# 拿到命令输出
output = stdout.read().decode()
print('=== 命令执行结果 ===')
print(output)
# 看看有没有报错
err_msg = stderr.read().decode()
if err_msg:
print('=== 错误信息 ===')
print(err_msg)
# 关掉连接
client.close()
把上面的 IP、用户名、密码换成你自己的,保存为 ssh_demo.py,然后运行:
python ssh_demo.py
正常的话,你会看到远程服务器的内存使用情况。
💡
stdout、stderr是啥?
stdout.read().decode()→ 命令正常执行后返回的结果stderr.read().decode()→ 命令报错时的信息- 为什么要
.decode()?因为网络传回来的是字节(bytes),得转成我们能看懂的字符串
三、进阶:处理需要交互的场景(比如连路由器/交换机)
上面的方式,适合执行一条命令就拿结果走人。但有些设备不走寻常路——比如思科、华为的路由器/交换机,或者某些需要 sudo 提权后输密码的场景。这时候 exec_command 就不大够用了。
得换成 invoke_shell(),模拟一个真实的终端对话:
import paramiko
import time
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname='192.168.1.1', # 你的路由器 IP
port=22,
username='admin',
password='admin123'
)
# 打开一个交互式 Shell 通道
shell = client.invoke_shell()
# 等设备初始化完毕
time.sleep(1)
# 敲命令(每条命令后面要加 \n,模拟回车)
shell.send('display version\n')
time.sleep(2) # 等设备返回结果
shell.send('display ip interface brief\n')
time.sleep(2)
# 把屏幕上的所有内容读出来
result = shell.recv(65535).decode()
print(result)
client.close()
那什么时候用哪个?这里给个判断方法:
| 场景 | 用什么 | 原因 |
|---|---|---|
| Linux 服务器执行一条命令 | exec_command |
简单直接,一次一结果 |
| 连续执行多条命令 | exec_command 也行,但每条单独调 |
每次都是独立连接 |
| 路由器/交换机登录 | invoke_shell |
需要模拟终端交互 |
| 需要 sudo 提权后输密码 | invoke_shell |
要等人机交互 |
| 登录后需要先进入某个模式 | invoke_shell |
类似 putty 手动操作 |
一句话总结:exec_command 是发一条命令拿一次结果,invoke_shell 是模拟你在终端里从头操作到尾。
四、实用的:通过 SFTP 上传/下载文件
光敲命令肯定不够,传文件也是高频需求——备份配置、上传更新包、下载日志。
paramiko 自带了 SFTP 功能,用法也挺简单:
上传文件到服务器
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname='192.168.1.100',
port=22,
username='root',
password='你的密码'
)
# 开启 SFTP 通道
sftp = client.open_sftp()
# 上传本地文件到服务器
sftp.put(
localpath='C:/backup/config.tar.gz', # 本地文件路径
remotepath='/home/backup/config.tar.gz' # 服务器上的目标路径
)
print('上传完成 ✅')
sftp.close()
client.close()
从服务器下载文件
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname='192.168.1.100',
port=22,
username='root',
password='你的密码'
)
sftp = client.open_sftp()
# 从服务器下载文件到本地
sftp.get(
remotepath='/var/log/nginx/access.log', # 服务器上的文件
localpath='D:/logs/access.log' # 保存到本地的路径
)
print('下载完成 ✅')
sftp.close()
client.close()
⚠️ 注意两点:
- 本地路径的目录要提前建好,sftp 不会自动帮你创建文件夹
- 如果目标文件已存在,默认会直接覆盖
再加个小功能:列出远程目录的文件
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname='192.168.1.100',
port=22,
username='root',
password='你的密码'
)
sftp = client.open_sftp()
# 列出远程目录
files = sftp.listdir('/home')
for f in files:
print(f)
sftp.close()
client.close()
五、批量管理几十台服务器
上班最烦什么?几十台服务器,每台都要上去敲同样的命令。比如检查磁盘空间、更新配置文件、重启服务。
手动一台台连效率太低。用多线程做批量执行,几秒钟就跑完所有机器。
先搞清楚思路
不要一台台顺序执行(那样太慢),而是同时启动多个线程,每个线程负责连一台机器,并发执行。相当于同时开好几个终端窗口。
批量执行脚本
import paramiko
import threading
import sys
def run_command(ip, username, password, command, port=22):
"""在单台服务器上执行命令并打印结果"""
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(
hostname=ip,
port=port,
username=username,
password=password,
timeout=10 # 超时控制,防止卡死
)
stdin, stdout, stderr = ssh.exec_command(command)
result = stdout.read().decode()
error = stderr.read().decode()
print(f'\n{"="*50}')
print(f'▶ 服务器: {ip}')
print(f'▶ 执行: {command}')
if result:
print(f'▶ 结果:\n{result}')
if error:
print(f'▶ 错误:\n{error}')
print(f'{"="*50}')
ssh.close()
except Exception as e:
print(f'\n❌ 服务器 {ip} 连接失败: {e}')
def batch_run(hosts, command):
"""批量在多台服务器上执行命令(多线程并发)"""
threads = []
for ip, username, password in hosts:
t = threading.Thread(
target=run_command,
args=(ip, username, password, command)
)
threads.append(t)
t.start() # 启动线程
# 等待所有线程执行完毕
for t in threads:
t.join()
print('\n全部执行完毕 ✅')
if __name__ == '__main__':
# 服务器列表:每行是 (IP, 用户名, 密码)
server_list = [
('192.168.1.10', 'root', 'pass123'),
('192.168.1.11', 'root', 'pass456'),
('192.168.1.12', 'root', 'pass789'),
]
# 从命令行参数获取要执行的命令
# 用法:python batch_ssh.py "df -h"
cmd = sys.argv[1] if len(sys.argv) > 1 else 'uptime'
batch_run(server_list, cmd)
怎么用?
# 检查所有服务器的磁盘空间
python batch_ssh.py "df -h"
# 查看所有服务器运行了多久
python batch_ssh.py "uptime"
# 重启某个服务
python batch_ssh.py "systemctl restart nginx"
把服务器列表单独放一个文件(更规范)
服务器信息写死在代码里,既不灵活也不安全。建议单独写个配置文件:
servers.txt
192.168.1.10 root pass123
192.168.1.11 root pass456
192.168.1.12 root pass789
然后在代码里读取这个文件:
def load_servers(file_path):
"""从配置文件加载服务器列表"""
hosts = []
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'): # 跳过空行和注释
parts = line.split()
if len(parts) >= 3:
hosts.append((parts[0], parts[1], parts[2]))
return hosts
if __name__ == '__main__':
server_list = load_servers('servers.txt')
cmd = sys.argv[1] if len(sys.argv) > 1 else 'uptime'
batch_run(server_list, cmd)
以后加机器直接改 servers.txt 就行,不用动代码。
六、一个完整的实用脚本:每天定时备份服务器配置
把上面所有功能串起来,写一个实际能用的小工具——每天自动登录服务器,打包备份关键配置文件,下载到本地:
import paramiko
import os
from datetime import datetime
def backup_server(ip, username, password, backup_dir='./backups'):
"""备份单台服务器的关键配置"""
today = datetime.now().strftime('%Y%m%d')
# 在服务器上创建备份目录
remote_backup = f'/tmp/backup_{today}'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=ip, username=username, password=password)
# 执行备份命令
commands = [
f'mkdir -p {remote_backup}',
f'cp /etc/nginx/nginx.conf {remote_backup}/ 2>/dev/null; echo "done"',
f'cp /etc/ssh/sshd_config {remote_backup}/ 2>/dev/null; echo "done"',
f'crontab -l > {remote_backup}/crontab.txt 2>/dev/null; echo "done"',
f'tar -czf /tmp/backup_{today}.tar.gz -C /tmp backup_{today} 2>/dev/null'
]
for cmd in commands:
ssh.exec_command(cmd)
# 用 SFTP 把打包好的文件下载到本地
sftp = ssh.open_sftp()
remote_file = f'/tmp/backup_{today}.tar.gz'
local_file = f'{backup_dir}/{ip}_backup_{today}.tar.gz'
os.makedirs(backup_dir, exist_ok=True)
try:
sftp.get(remote_file, local_file)
print(f'✅ {ip} 备份完成 → {local_file}')
except FileNotFoundError:
print(f'⚠️ {ip} 备份文件未生成,可能是没有对应配置')
# 清理远程临时文件
ssh.exec_command(f'rm -rf /tmp/backup_{today}*')
sftp.close()
ssh.close()
def batch_backup(servers):
"""批量备份所有服务器"""
threads = []
for ip, user, pwd in servers:
t = threading.Thread(target=backup_server, args=(ip, user, pwd))
threads.append(t)
t.start()
for t in threads:
t.join()
print('\n所有服务器备份完成 ✅')
if __name__ == '__main__':
# 这里改成你自己的服务器列表
my_servers = [
('192.168.1.10', 'root', 'pass123'),
('192.168.1.11', 'root', 'pass456'),
]
batch_backup(my_servers)
把这个脚本加到 crontab(Linux)或者任务计划程序(Windows)里,每天自动跑一次就行。这样万一哪天手抖改坏了配置,还能找回之前的版本。
七、避坑指南(新手最容易踩的雷)
玩 paramiko 的过程中,这几个坑几乎每个人都会遇到,提前知道能省不少时间:
❌ 坑1:连接报错 Authentication failed
- 原因:密码不对,或者服务器只支持密钥登录
- 解决:先手动 SSH 一遍确认密码是对的;如果用密钥登录,参考下面的写法:
# 用密钥登录的方式
key = paramiko.RSAKey.from_private_key_file('C:/Users/me/.ssh/id_rsa')
client.connect(
hostname='192.168.1.100',
port=22,
username='root',
pkey=key
)
❌ 坑2:执行命令没输出
- 原因:命令还没执行完就去读结果了
- 解决:加个
time.sleep(1)等一会儿,或者用stdout.channel.recv_exit_status()等命令真正结束再读
❌ 坑3:中文乱码
- 原因:服务器返回的编码不是 UTF-8
- 解决:试一下
decode('gbk')或decode('utf-8', errors='ignore')
output = stdout.read().decode('gbk', errors='ignore')
❌ 坑4:连接多了报 socket.error
- 原因:服务器限制了并发连接数,或者代码里没及时关闭连接
- 解决:每次用完记得调用
ssh.close(),并发数别太大(建议不超过 20)
八、总结一下
| 功能 | 关键代码 | 适用场景 |
|---|---|---|
| SSH 登录执行命令 | exec_command('命令') |
Linux 服务器日常运维 |
| 交互式终端 | invoke_shell() + send()/recv() |
路由器/交换机、交互式程序 |
| 上传文件 | sftp.put(本地, 远程) |
部署代码、上传安装包 |
| 下载文件 | sftp.get(远程, 本地) |
拉取日志、备份配置文件 |
| 批量管理 | threading.Thread + paramiko |
几十台服务器同时操作 |
核心流程:
SSH 操作 = 创建客户端 → 设置策略 → 连接 → 执行操作 → 关闭
掌握了这些,日常大部分运维操作基本都能写成脚本自动化。剩下的无非是加一些错误处理、日志记录之类的完善工作。
如果你工作中有什么手动操作觉得很烦,不妨想想能不能用 paramiko 写成脚本。写完之后你会发现:以前那些重复劳动,其实都可以交给代码去做。
📝 说明:文中所有代码在 Python 3.6+ 环境下实测可用。
🐛 遇到问题先自查:① paramiko 装了吗 ② IP 和密码对吗 ③ 防火墙 22 端口开了吗
更多推荐


所有评论(0)