快捷導(dǎo)航

使用Python實(shí)現(xiàn)SSH隧道界面功能

更新時(shí)間：2022年02月08日 10:09:37 作者：清水雅然君

這篇文章主要介紹了使用Python實(shí)現(xiàn)一個(gè)SSH隧道界面功能，界面使用tkinter實(shí)現(xiàn)，左邊是輸入隧道的信息，右邊為歷史列表,本文通過示例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧

開發(fā)原因

MobaXterm作為一個(gè)全能型終端神器，功能十分強(qiáng)大，我經(jīng)常使用其中隧道功能，使用內(nèi)部無法直接服務(wù)器，查詢數(shù)據(jù)，一般來說，一個(gè)本地端口對(duì)于一個(gè)隧道，但是MobaXterm，免費(fèi)版本最多只能建立三個(gè)隧道，比如，我需要一次查詢統(tǒng)計(jì)，就會(huì)用到四個(gè)隧道的操作，就非常不方便，需要調(diào)整一個(gè)隧道，于是，便用python寫了多隧道的客戶端

效果圖

界面使用tkinter實(shí)現(xiàn)，左邊是輸入隧道的信息，右邊為歷史列表，

源碼分析

構(gòu)建隧道

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

這段代碼就是整個(gè)功能的核心代碼，使用SSHTunnelForwarder模塊的sshtunnel

構(gòu)建隧道，由于我只需要一個(gè)本地端口訪問遠(yuǎn)程遠(yuǎn)程服務(wù)器的功能，默認(rèn)本地端口固定為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, "未啟動(dòng)")
            tunnel_infos.update({tree_id: tunnel_info})

read_json是讀取歷史記錄，其中使用 cryptocode模版對(duì)明文的json進(jìn)行加密，并且對(duì)ssh_password進(jìn)行再加密

開始服務(wù)

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, "啟動(dòng)")
                    pass
                except Exception  as e:
                    messagebox.showinfo(title='連接異常', message=e.args[0])
    else:
        messagebox.showinfo(title='選擇異常', message="未選擇列表")

tunnel_infos為報(bào)存的隧道信息字典，tunnel_infos_start為報(bào)存的已經(jīng)啟動(dòng)的隧道字典，先獲取點(diǎn)擊到的行的ID，然后查詢是否在已啟動(dòng)的字典中，如果不存在則，啟動(dòng)隧道，同時(shí)更新到tunnel_infos_start中

停止服務(wù)

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, "未啟動(dòng)")
                    pass
                except Exception  as e:
                    messagebox.showinfo(title='連接異常', message=e.args[0])
    else:
        messagebox.showinfo(title='選擇異常', message="未選擇列表")

這段代碼操作和啟動(dòng)相反，則是從停止掉服務(wù)，同時(shí)從tunnel_infos_start中移除掉該隧道

移除服務(wù)

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="未選擇列表")

移除服務(wù)的時(shí)候，會(huì)判斷ID在tunnel_infos_start是否存在，存在則表明當(dāng)前刪除的隧道還在啟動(dòng)中，則停止這個(gè)服務(wù)，同時(shí)從tunnel_infos移除這配置，更新tunnel_data.json文件

不足之處

雖然這個(gè)簡(jiǎn)單的工具可以滿足超過多個(gè)隧道的使用，但是每次報(bào)存的時(shí)候，都要更新tunnel_data.json文件，如果隧道比較多，加載比較費(fèi)時(shí)間；同時(shí)由于設(shè)計(jì)界面的時(shí)候，考慮比較簡(jiǎn)單，并不支持修改的功能，只能刪除錯(cuò)誤的記錄，然后重新報(bào)存