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

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 代码里——你不用打开那些软件写几行脚本就能完成登录、敲命令、拿结果、传文件这些事。安装方法pipinstallparamiko如果网络慢加个国内源pipinstallparamiko-ihttps://pypi.tuna.tsinghua.edu.cn/simple装完就可以用了。二、最基础的SSH 登录 执行一条命令先写一个能跑起来的 demo连上服务器、执行一条命令、看输出importparamiko# 创建 SSH 客户端clientparamiko.SSHClient()# 自动接受陌生服务器的密钥不弹确认框client.set_missing_host_key_policy(paramiko.AutoAddPolicy())# 连上服务器client.connect(hostname192.168.1.100,# 改成你的服务器 IPport22,# SSH 端口默认 22usernameroot,# 登录用户名password你的密码# 登录密码)# 执行一条命令stdin,stdout,stderrclient.exec_command(free -h)# 拿到命令输出outputstdout.read().decode()print( 命令执行结果 )print(output)# 看看有没有报错err_msgstderr.read().decode()iferr_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()模拟一个真实的终端对话importparamikoimporttime clientparamiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname192.168.1.1,# 你的路由器 IPport22,usernameadmin,passwordadmin123)# 打开一个交互式 Shell 通道shellclient.invoke_shell()# 等设备初始化完毕time.sleep(1)# 敲命令每条命令后面要加 \n模拟回车shell.send(display version\n)time.sleep(2)# 等设备返回结果shell.send(display ip interface brief\n)time.sleep(2)# 把屏幕上的所有内容读出来resultshell.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 功能用法也挺简单上传文件到服务器importparamiko clientparamiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname192.168.1.100,port22,usernameroot,password你的密码)# 开启 SFTP 通道sftpclient.open_sftp()# 上传本地文件到服务器sftp.put(localpathC:/backup/config.tar.gz,# 本地文件路径remotepath/home/backup/config.tar.gz# 服务器上的目标路径)print(上传完成 ✅)sftp.close()client.close()从服务器下载文件importparamiko clientparamiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname192.168.1.100,port22,usernameroot,password你的密码)sftpclient.open_sftp()# 从服务器下载文件到本地sftp.get(remotepath/var/log/nginx/access.log,# 服务器上的文件localpathD:/logs/access.log# 保存到本地的路径)print(下载完成 ✅)sftp.close()client.close()⚠️ 注意两点本地路径的目录要提前建好sftp 不会自动帮你创建文件夹如果目标文件已存在默认会直接覆盖再加个小功能列出远程目录的文件importparamiko clientparamiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname192.168.1.100,port22,usernameroot,password你的密码)sftpclient.open_sftp()# 列出远程目录filessftp.listdir(/home)forfinfiles:print(f)sftp.close()client.close()五、批量管理几十台服务器上班最烦什么几十台服务器每台都要上去敲同样的命令。比如检查磁盘空间、更新配置文件、重启服务。手动一台台连效率太低。用多线程做批量执行几秒钟就跑完所有机器。先搞清楚思路不要一台台顺序执行那样太慢而是同时启动多个线程每个线程负责连一台机器并发执行。相当于同时开好几个终端窗口。批量执行脚本importparamikoimportthreadingimportsysdefrun_command(ip,username,password,command,port22):在单台服务器上执行命令并打印结果try:sshparamiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(hostnameip,portport,usernameusername,passwordpassword,timeout10# 超时控制防止卡死)stdin,stdout,stderrssh.exec_command(command)resultstdout.read().decode()errorstderr.read().decode()print(f\n{*50})print(f▶ 服务器:{ip})print(f▶ 执行:{command})ifresult:print(f▶ 结果:\n{result})iferror:print(f▶ 错误:\n{error})print(f{*50})ssh.close()exceptExceptionase:print(f\n❌ 服务器{ip}连接失败:{e})defbatch_run(hosts,command):批量在多台服务器上执行命令多线程并发threads[]forip,username,passwordinhosts:tthreading.Thread(targetrun_command,args(ip,username,password,command))threads.append(t)t.start()# 启动线程# 等待所有线程执行完毕fortinthreads: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 -hcmdsys.argv[1]iflen(sys.argv)1elseuptimebatch_run(server_list,cmd)怎么用# 检查所有服务器的磁盘空间python batch_ssh.pydf -h# 查看所有服务器运行了多久python batch_ssh.pyuptime# 重启某个服务python batch_ssh.pysystemctl restart nginx把服务器列表单独放一个文件更规范服务器信息写死在代码里既不灵活也不安全。建议单独写个配置文件servers.txt192.168.1.10 root pass123 192.168.1.11 root pass456 192.168.1.12 root pass789然后在代码里读取这个文件defload_servers(file_path):从配置文件加载服务器列表hosts[]withopen(file_path,r,encodingutf-8)asf:forlineinf:lineline.strip()iflineandnotline.startswith(#):# 跳过空行和注释partsline.split()iflen(parts)3:hosts.append((parts[0],parts[1],parts[2]))returnhostsif__name____main__:server_listload_servers(servers.txt)cmdsys.argv[1]iflen(sys.argv)1elseuptimebatch_run(server_list,cmd)以后加机器直接改servers.txt就行不用动代码。六、一个完整的实用脚本每天定时备份服务器配置把上面所有功能串起来写一个实际能用的小工具——每天自动登录服务器打包备份关键配置文件下载到本地importparamikoimportosfromdatetimeimportdatetimedefbackup_server(ip,username,password,backup_dir./backups):备份单台服务器的关键配置todaydatetime.now().strftime(%Y%m%d)# 在服务器上创建备份目录remote_backupf/tmp/backup_{today}sshparamiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(hostnameip,usernameusername,passwordpassword)# 执行备份命令commands[fmkdir -p{remote_backup},fcp /etc/nginx/nginx.conf{remote_backup}/ 2/dev/null; echo done,fcp /etc/ssh/sshd_config{remote_backup}/ 2/dev/null; echo done,fcrontab -l {remote_backup}/crontab.txt 2/dev/null; echo done,ftar -czf /tmp/backup_{today}.tar.gz -C /tmp backup_{today}2/dev/null]forcmdincommands:ssh.exec_command(cmd)# 用 SFTP 把打包好的文件下载到本地sftpssh.open_sftp()remote_filef/tmp/backup_{today}.tar.gzlocal_filef{backup_dir}/{ip}_backup_{today}.tar.gzos.makedirs(backup_dir,exist_okTrue)try:sftp.get(remote_file,local_file)print(f✅{ip}备份完成 →{local_file})exceptFileNotFoundError:print(f⚠️{ip}备份文件未生成可能是没有对应配置)# 清理远程临时文件ssh.exec_command(frm -rf /tmp/backup_{today}*)sftp.close()ssh.close()defbatch_backup(servers):批量备份所有服务器threads[]forip,user,pwdinservers:tthreading.Thread(targetbackup_server,args(ip,user,pwd))threads.append(t)t.start()fortinthreads: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)把这个脚本加到 crontabLinux或者任务计划程序Windows里每天自动跑一次就行。这样万一哪天手抖改坏了配置还能找回之前的版本。七、避坑指南新手最容易踩的雷玩 paramiko 的过程中这几个坑几乎每个人都会遇到提前知道能省不少时间❌ 坑1连接报错Authentication failed原因密码不对或者服务器只支持密钥登录解决先手动 SSH 一遍确认密码是对的如果用密钥登录参考下面的写法# 用密钥登录的方式keyparamiko.RSAKey.from_private_key_file(C:/Users/me/.ssh/id_rsa)client.connect(hostname192.168.1.100,port22,usernameroot,pkeykey)❌ 坑2执行命令没输出原因命令还没执行完就去读结果了解决加个time.sleep(1)等一会儿或者用stdout.channel.recv_exit_status()等命令真正结束再读❌ 坑3中文乱码原因服务器返回的编码不是 UTF-8解决试一下decode(gbk)或decode(utf-8, errorsignore)outputstdout.read().decode(gbk,errorsignore)❌ 坑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 端口开了吗