Python 通过跳板机连接远程服务器：Paramiko 实战指南-北京尧图网络

1. 引言在实际的企业级开发和运维场景中出于安全考虑我们通常无法直接访问生产环境服务器而是需要通过一个跳板机Bastion Host作为中转。本文将详细介绍如何使用 Python 的 Paramiko 库通过跳板机安全地连接远程服务器并实现命令执行和文件传输功能。2. 环境准备2.1 安装 Paramikopipinstallparamiko2.2 基础概念跳板机位于公网和内网之间的安全网关所有对内网服务器的访问都必须经过它SSH 隧道通过跳板机建立的加密通道用于安全地访问内网服务器ParamikoPython 的 SSHv2 协议实现库支持 SSH 连接和 SFTP 文件传输3. 核心实现代码以下是完整的实现类支持通过跳板机连接远程服务器#!/usr/bin/env python3# -*- coding: utf-8 -*-importosimportparamikofromtypingimportOptional,Tuple,UnionclassSSHJumpHostConnector: 通过跳板机连接远程服务器的 SSH/SFTP 客户端功能 1. 通过跳板机建立 SSH 连接 2. 执行远程命令 3. 上传/下载文件 4. 安全关闭连接 def__init__(self,jump_host:str,jump_user:str,jump_password:str,target_host:str,target_user:str,target_password:str,jump_port:int22,target_port:int22): 初始化跳板机连接 Args: jump_host: 跳板机 IP 地址 jump_user: 跳板机用户名 jump_password: 跳板机密码 target_host: 目标服务器 IP 地址 target_user: 目标服务器用户名 target_password: 目标服务器密码 jump_port: 跳板机 SSH 端口默认 22 target_port: 目标服务器 SSH 端口默认 22 self.jump_hostjump_host self.target_hosttarget_host# 第一步连接跳板机self.jump_sshparamiko.SSHClient()self.jump_ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())self.jump_ssh.connect(hostnamejump_host,portjump_port,usernamejump_user,passwordjump_password)# 第二步通过跳板机建立到目标服务器的通道jump_transportself.jump_ssh.get_transport()src_addr(jump_host,jump_port)dest_addr(target_host,target_port)jump_channeljump_transport.open_channel(kinddirect-tcpip,dest_addrdest_addr,src_addrsrc_addr)# 第三步连接目标服务器self.target_sshparamiko.SSHClient()self.target_ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())self.target_ssh.connect(hostnametarget_host,porttarget_port,usernametarget_user,passwordtarget_password,sockjump_channel)# 第四步创建 SFTP 客户端target_transportself.target_ssh.get_transport()self.sftpparamiko.SFTPClient.from_transport(target_transport)print(f✅ 成功连接到目标服务器{target_host}通过跳板机{jump_host})defexecute_command(self,command:str,timeout:int30)-Tuple[str,str,int]: 在目标服务器上执行命令 Args: command: 要执行的命令 timeout: 命令执行超时时间秒 Returns: tuple: (stdout, stderr, return_code) stdin,stdout,stderrself.target_ssh.exec_command(command,timeouttimeout)stdout_strstdout.read().decode(utf-8).strip()stderr_strstderr.read().decode(utf-8).strip()return_codestdout.channel.recv_exit_status()print(f 执行命令:{command})print(f 输出:{stdout_str})ifstderr_str:print(f⚠️ 错误:{stderr_str})print(f 返回码:{return_code})returnstdout_str,stderr_str,return_codedefupload_file(self,local_path:str,remote_path:str)-None: 上传本地文件到远程服务器 Args: local_path: 本地文件路径 remote_path: 远程文件路径 ifnotos.path.exists(local_path):raiseFileNotFoundError(f本地文件不存在:{local_path})self.sftp.put(local_path,remote_path)print(f 上传成功:{local_path}→{remote_path})defdownload_file(self,remote_path:str,local_path:str)-None: 从远程服务器下载文件到本地 Args: remote_path: 远程文件路径 local_path: 本地文件路径 # 确保本地目录存在local_diros.path.dirname(local_path)iflocal_dirandnotos.path.exists(local_dir):os.makedirs(local_dir)self.sftp.get(remote_path,local_path)print(f 下载成功:{remote_path}→{local_path})defclose(self)-None:安全关闭所有连接ifhasattr(self,sftp):self.sftp.close()print( 已关闭 SFTP 连接)ifhasattr(self,target_ssh):self.target_ssh.close()print( 已关闭目标服务器 SSH 连接)ifhasattr(self,jump_ssh):self.jump_ssh.close()print( 已关闭跳板机 SSH 连接)print(✅ 所有连接已安全关闭)# 使用示例if__name____main__:# 配置连接信息实际使用时请替换为真实信息JUMP_HOST10.110.19.100# 跳板机 IPJUMP_USERjump_user# 跳板机用户名JUMP_PASSWORDjump_password# 跳板机密码TARGET_HOST172.20.10.50# 目标服务器 IPTARGET_USERremote_user# 目标服务器用户名TARGET_PASSWORDremote_pass# 目标服务器密码try:# 创建连接connectorSSHJumpHostConnector(jump_hostJUMP_HOST,jump_userJUMP_USER,jump_passwordJUMP_PASSWORD,target_hostTARGET_HOST,target_userTARGET_USER,target_passwordTARGET_PASSWORD)# 示例 1执行命令print(\n 执行命令示例 )output,error,codeconnector.execute_command(pwd)print(f当前工作目录:{output})# 示例 2查看系统信息output,error,codeconnector.execute_command(uname -a)print(f系统信息:{output})# 示例 3上传文件print(\n 文件传输示例 )# connector.upload_file(local_file.txt, /tmp/remote_file.txt)# 示例 4下载文件# connector.download_file(/tmp/remote_file.txt, downloaded_file.txt)# 示例 5批量执行命令commands[whoami,date,df -h | head -5]print(\n 批量执行命令 )forcmdincommands:output,error,codeconnector.execute_command(cmd)print(f命令 {cmd} 输出:{output})exceptExceptionase:print(f❌ 连接失败:{e})finally:# 确保连接被关闭ifconnectorinlocals():connector.close()4. 工作原理详解4.1 连接建立流程本地客户端连接跳板机 SSHClient.connect建立SSH隧道 open_channel连接目标服务器 connect sockchannel创建SFTP客户端 from_transport执行命令或传输文件4.2 关键参数说明AutoAddPolicy()自动添加未知主机到 known_hosts生产环境建议使用更安全的策略direct-tcpip建立直接的 TCP/IP 通道用于 SSH 隧道sock参数指定使用已有的 socket 连接实现通过跳板机连接5. 高级用法与最佳实践5.1 使用密钥认证推荐# 使用密钥文件代替密码private_keyparamiko.RSAKey.from_private_key_file(/path/to/private_key.pem)jumpbox_ssh.connect(hostnamejump_host,usernamejump_user,pkeyprivate_key# 使用密钥认证)5.2 连接池管理对于频繁的连接需求可以实现连接池classSSHConnectionPool:def__init__(self,max_connections10):self.pool[]self.max_connectionsmax_connectionsdefget_connection(self,config):# 从池中获取或创建新连接passdefrelease_connection(self,connector):# 释放连接回池中pass5.3 错误处理与重试importtimefromparamiko.ssh_exceptionimportSSHException,AuthenticationExceptiondefconnect_with_retry(connector_class,config,max_retries3):forattemptinrange(max_retries):try:connectorconnector_class(**config)returnconnectorexcept(SSHException,AuthenticationException)ase:ifattemptmax_retries-1:raiseprint(f连接失败{attempt1}秒后重试...)time.sleep(attempt1)5.4 安全建议不要硬编码密码使用环境变量或配置文件使用密钥认证比密码更安全限制权限使用最小权限原则记录日志记录所有连接和操作定期更换密钥定期更新 SSH 密钥6. 常见问题与解决方案Q1: 连接超时怎么办检查网络连通性ping 跳板机IP检查防火墙设置增加超时时间connect(timeout60)Q2: 认证失败怎么办确认用户名/密码正确检查密钥文件权限chmod 600 private_key.pem验证跳板机是否允许该用户连接Q3: SFTP 传输失败怎么办检查文件路径是否存在确认用户有读写权限检查磁盘空间是否充足Q4: 如何提高传输速度使用压缩sftp.get(remotepath, localpath, prefetchFalse)批量传输减少连接开销使用更快的加密算法7. 总结关键点包括安全连接通过跳板机建立加密隧道完整功能支持命令执行和文件传输健壮性完善的错误处理和资源管理可扩展性易于扩展为连接池或集成到自动化工具中附录完整配置示例创建config.yaml配置文件# config.yamljump_host:host:10.110.19.100user:jump_user# 使用密码或密钥文件password:your_password# key_file: /path/to/private_key.pemtarget_hosts:web_server:host:172.20.10.50user:web_userpassword:web_passworddb_server:host:172.20.10.51user:db_userpassword:db_password使用配置文件连接importyamlwithopen(config.yaml,r)asf:configyaml.safe_load(f)# 连接多个服务器forserver_name,server_configinconfig[target_hosts].items():connectorSSHJumpHostConnector(jump_hostconfig[jump_host][host],jump_userconfig[jump_host][user],jump_passwordconfig[jump_host][password],target_hostserver_config[host],target_userserver_config[user],target_passwordserver_config[password])# 执行操作...connector.close()

Python 通过跳板机连接远程服务器：Paramiko 实战指南

相关推荐

125.BGP路由最常用的属性

35+技术人的年中复盘：在大模型浪潮中重新定义自己的技术价值

AI数据库产品的商业化思考：从内部工具到对外服务的关键跨越

更多资讯

低压电工作业实训总结：上岗必备，证书过期了不知道怎么复审延期？

安塞县物业电工证怎么考？2026河南最新政策别踩坑

特种工证网上办理在哪考最靠谱？怕考不过白交培训费的看过来

许昌电工证代审全国通用怎么操作？别再被中介骗了

以前电工证图片培训机构怎么选通过率揭秘

工地太忙根本没时间复习考试，吉林电工证哪家便宜又靠谱？

今日更新

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本周精选

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

相关推荐

125.BGP路由最常用的属性

35+技术人的年中复盘：在大模型浪潮中重新定义自己的技术价值

AI数据库产品的商业化思考：从内部工具到对外服务的关键跨越

更多资讯

低压电工作业实训总结：上岗必备，证书过期了不知道怎么复审延期？

安塞县物业电工证怎么考？2026河南最新政策别踩坑

特种工证网上办理在哪考最靠谱？怕考不过白交培训费的看过来

许昌电工证代审全国通用怎么操作？别再被中介骗了

以前电工证图片培训机构怎么选 通过率揭秘

工地太忙根本没时间复习考试，吉林电工证哪家便宜又靠谱？

今日更新

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本周精选

普宁电工证在哪报名官方回应怕考不过白交培训费？这些避坑干货你必须知道

高压电工证登高练习题在哪考合格率高？怕考不过白交培训费

电工作业登记表通过率揭秘：怕考不过白交培训费？洛阳学员亲测经验全公开

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

以前电工证图片培训机构怎么选通过率揭秘