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数据库产品的商业化思考：从内部工具到对外服务的关键跨越

更多资讯

superpowers实战：为AI编程助手注入工程化协作能力

Jeepay开源聚合支付系统：Java四方支付底座实战指南

蓝牙双模配对总翻车？CTKD跨传输密钥派生原理与实测解析

Substrate区块链开发框架详解：从核心概念到自定义链实操

Kubernetes、Ray、vLLM 三层调度分工与协同优化

Kubernetes 上构建 Agentic 工作负载的运行时调度层：ax 调度设计与实践

今日更新

济南做网站多少钱：3个案例拆解，防黑源码下载全攻略

婚恋网站实战案例：避开3个高价坑，省钱50%还能跑赢流量

制作网页比较方便的软件怎么选?一文搞懂避坑指南

本周精选

如何彻底离线使用 Laya-CoreML：Hugging Face 下载、缓存机制与 local_files_only 详解

基于1Panel 部署MySQL MCP 服务：TaoToken 统一 Key 接入与 config.toml 配置实战

崇州园区营销网站建设避坑指南：5大注意事项保备案安全

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

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