欢迎光临
我们一直在努力

Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了

Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了

写给谁看:运维兄弟、刚学 Python 想做自动化的同学、被重复手工操作折磨的打工人 需要什么基础:会装 Python、会 pip 安装包即可 目标:看完就能上手,把日常手动操作变成脚本一键搞定


文章目录

  • Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了
    • 一、先认识一下 paramiko 是什么
    • 二、最基础的:SSH 登录 + 执行一条命令
    • 三、进阶:处理需要交互的场景(比如连路由器/交换机)
    • 四、实用的:通过 SFTP 上传/下载文件
      • 上传文件到服务器
      • 从服务器下载文件
      • 再加个小功能:列出远程目录的文件
    • 五、批量管理几十台服务器
      • 先搞清楚思路
      • 批量执行脚本
      • 把服务器列表单独放一个文件(更规范)
    • 六、一个完整的实用脚本:每天定时备份服务器配置
    • 七、避坑指南(新手最容易踩的雷)
      • ❌ 坑1:连接报错 `Authentication failed`
      • ❌ 坑2:执行命令没输出
      • ❌ 坑3:中文乱码
      • ❌ 坑4:连接多了报 `socket.error`
    • 八、总结一下

一、先认识一下 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)
    未经允许不得转载:171主机测评 » Python 用 paramiko 操作远程服务器:SSH 登录、传文件、批量执行,看这一篇就够了
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址