irpas技术客

项目工具:使用PYTHON实现一个SSH隧道界面_清水雅然君_python ssh隧道

大大的周 3241

开发原因

MobaXterm作为一个全能型终端神器,功能十分强大,我经常使用其中隧道功能,使用内部无法直接服务器,查询数据,一般来说,一个本地端口对于一个隧道,但是MobaXterm,免费版本最多只能建立三个隧道,比如,我需要一次查询统计,就会用到四个隧道的操作,就非常不方便,需要调整一个隧道,于是,便用python写了多隧道的客户端

效果图

界面使用tkinter实现,左边是输入隧道的信息,右边为历史列表,

源码分析 构建隧道 def operate_sshtunnel(tunnel_info): try: tunnel = SSHTunnelForwarder( (tunnel_info.ssh_ip, int(tunnel_info.ssh_port)), ssh_username=tunnel_info.ssh_username, ssh_password=tunnel_info.ssh_password, remote_bind_address=(tunnel_info.remote_ip, int(tunnel_info.remote_port)), local_bind_address=('127.0.0.1', int(tunnel_info.localhost_port)) ) return tunnel except Exception as e: print(e.args[0]) messagebox.showinfo(title='连接异常', message=e.args[0]) return

这段代码就是整个功能的核心代码,使用SSHTunnelForwarder模块的sshtunnel

构建隧道,由于我只需要一个本地端口访问远程远程服务器的功能,默认本地端口固定为127.0.0.1

初始化加载 def read_json(): if os.path.exists('tunnel_data.json'): with open('tunnel_data.json', 'r', encoding='utf-8') as load_f: data = load_f.read() if len(data) > 0: json_str = cryptocode.decrypt(data, "EjdeB55cvQMN2WHf") return json.loads(json_str) else: return def load_config(): load_arr = read_json() if load_arr is not None: for tunnel_info_json in load_arr: tunnel_info = tunnel_info_class() tunnel_info.localhost_port = tunnel_info_json['localhost_port'] tunnel_info.ssh_ip = tunnel_info_json['ssh_ip'] tunnel_info.ssh_port = tunnel_info_json['ssh_port'] tunnel_info.ssh_username = tunnel_info_json['ssh_username'] tunnel_info.ssh_password = cryptocode.decrypt(tunnel_info_json['ssh_password'], "F1jgEg1arVyxmUqC") tunnel_info.remote_ip = tunnel_info_json['remote_ip'] tunnel_info.remote_port = tunnel_info_json['remote_port'] tunnel_info.tunnel_name = tunnel_info_json['tunnel_name'] tree_id = insert_tree_view(tunnel_info, "未启动") tunnel_infos.update({tree_id: tunnel_info})

read_json是读取历史记录,其中使用 cryptocode模版对明文的json进行加密,并且对ssh_password进行再加密

开始服务 def start_tunnel(): iid = treeview.selection() if len(iid) > 0: if iid not in tunnel_infos_start.keys(): tunnel_info = tunnel_infos[iid[0]] tunnel = ssl_tunnel.operate_sshtunnel(tunnel_info) if tunnel is not None: try: tunnel.start() tunnel_infos_start.update({iid[0]: tunnel}) update_tree_view(iid[0], tunnel_info, "启动") pass except Exception as e: messagebox.showinfo(title='连接异常', message=e.args[0]) else: messagebox.showinfo(title='选择异常', message="未选择列表")

tunnel_infos为报存的隧道信息字典,tunnel_infos_start为报存的已经启动的隧道字典,先获取点击到的行的ID,然后查询是否在已启动的字典中,如果不存在则,启动隧道,同时更新到tunnel_infos_start中

停止服务 def stop_tunnel(): iid = treeview.selection() if len(iid) > 0: if iid[0] in tunnel_infos_start.keys(): tunnel_info = tunnel_infos[iid[0]] tunnel = tunnel_infos_start[iid[0]] if tunnel is not None: try: tunnel.stop() tunnel_infos_start.pop(iid[0]) update_tree_view(iid[0], tunnel_info, "未启动") pass except Exception as e: messagebox.showinfo(title='连接异常', message=e.args[0]) else: messagebox.showinfo(title='选择异常', message="未选择列表")

这段代码操作和启动相反,则是从停止掉服务,同时从tunnel_infos_start中移除掉该隧道

移除服务 def remove_tunnel(): iid = treeview.selection() if len(iid) > 0: if iid[0] in tunnel_infos_start.keys(): stop_tunnel() ## 从列表删除 treeview.delete(iid) tunnel_infos.pop(iid[0]) write_json() else: messagebox.showinfo(title='选择异常', message="未选择列表")

移除服务的时候,会判断ID在tunnel_infos_start是否存在,存在则表明当前删除的隧道还在启动中,则停止这个服务,同时从tunnel_infos移除这配置,更新tunnel_data.json文件

不足之处

虽然这个简单的工具可以满足超过多个隧道的使用,但是每次报存的时候,都要更新tunnel_data.json文件,如果隧道比较多,加载比较费时间;同时由于设计界面的时候,考虑比较简单,并不支持修改的功能,只能删除错误的记录,然后重新报存

源码地址

https://github.com/liuhao192/ssh_tunnel


1.本站遵循行业规范,任何转载的稿件都会明确标注作者和来源;2.本站的原创文章,会注明原创字样,如未注明都非原创,如有侵权请联系删除!;3.作者投稿可能会经我们编辑修改或补充;4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。

标签: #Python #ssh隧道