leefer
97d8b7ae89
- cli.py:无界面命令行(host/tunnel/list/test/start/remove),复用隧道引擎 - storage.py:配置目录跨平台(Win %APPDATA%,Linux ~/ssh-e2e/.config,macOS ~/ssh-e2e/Library), 非 Windows 经 keyring+AES-GCM 加密机密,无后端时明文并提示 - install.sh / install.ps1:经 gitea.xbay.cc 一键拉取安装 - 修复:Store 未创建配置目录的崩溃(CLI 路径)
488 lines
17 KiB
Python
488 lines
17 KiB
Python
"""SSH 隧道管理器 — Windows 桌面入口(pywebview + paramiko)。"""
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import webview
|
|
|
|
from engine import TunnelManager, probe_host
|
|
from storage import Store, data_root
|
|
|
|
APP_DIR_NAME = "SshTunnelManager"
|
|
APP_TITLE = "SSH 隧道管理器"
|
|
UI_PATH = Path(__file__).parent / "ui" / "index.html"
|
|
ICON_PATH = Path(__file__).parent / "ui" / "app.ico"
|
|
TRAY_ICON_PATH = Path(__file__).parent / "ui" / "icon-256.png"
|
|
_SENSITIVE = ("password", "key_passphrase")
|
|
|
|
|
|
def data_dir() -> Path:
|
|
d = data_root()
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
def _port_ok(v) -> bool:
|
|
try:
|
|
return 1 <= int(v) <= 65535
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
class TrayController:
|
|
"""系统托盘:窗口最小化/关闭后驻留托盘,可恢复或真正退出。"""
|
|
|
|
def __init__(self, api: "Api"):
|
|
self._api = api
|
|
self._icon = None
|
|
self._hinted = False
|
|
|
|
def start(self) -> None:
|
|
try:
|
|
from PIL import Image
|
|
from pystray import Icon, Menu, MenuItem
|
|
image = Image.open(TRAY_ICON_PATH)
|
|
menu = Menu(
|
|
MenuItem("显示主窗口", self._on_show, default=True),
|
|
Menu.SEPARATOR,
|
|
MenuItem("退出", self._on_exit),
|
|
)
|
|
self._icon = Icon("ssh-tunnel-manager", image, APP_TITLE, menu)
|
|
self._icon.run_detached()
|
|
self._icon.visible = True
|
|
except Exception: # noqa: BLE001 托盘失败不影响主窗口
|
|
self._icon = None
|
|
|
|
def _on_show(self, *_args) -> None:
|
|
self._api.show_main_window()
|
|
|
|
def _on_exit(self, *_args) -> None:
|
|
self._api.exit_application()
|
|
|
|
def notify_minimized_once(self) -> None:
|
|
if self._hinted or not self._icon:
|
|
return
|
|
self._hinted = True
|
|
try:
|
|
self._icon.notify("已最小化到系统托盘,隧道在后台保持运行", APP_TITLE)
|
|
except Exception:
|
|
pass
|
|
|
|
def stop(self) -> None:
|
|
icon, self._icon = self._icon, None
|
|
if icon:
|
|
try:
|
|
icon.stop()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class Api:
|
|
"""暴露给前端 JS 的接口。所有方法返回 {"ok": bool, ...}。"""
|
|
|
|
def __init__(self, store: Store, manager: TunnelManager):
|
|
self._store = store
|
|
self._manager = manager
|
|
self._win: webview.Window | None = None
|
|
self._maximized = False
|
|
self._really_exiting = False
|
|
self.tray = TrayController(self)
|
|
|
|
# ---------- 事件推送 ----------
|
|
|
|
def _emit(self, ev: dict) -> None:
|
|
if not self._win:
|
|
return
|
|
try:
|
|
self._win.evaluate_js("window.__bus && window.__bus(%s)"
|
|
% json.dumps(ev, ensure_ascii=False))
|
|
except Exception:
|
|
pass
|
|
|
|
def _push_state(self) -> None:
|
|
self._emit({"type": "state", "state": self._state()})
|
|
|
|
# ---------- 序列化 ----------
|
|
|
|
def _state(self) -> dict:
|
|
hosts_by_id = {h["id"]: h for h in self._store.data["hosts"]}
|
|
hosts = [self._host_out(h) for h in self._store.data["hosts"]]
|
|
tunnels = []
|
|
for t in self._store.data["tunnels"]:
|
|
out = {k: t.get(k) for k in
|
|
("id", "name", "host_id", "local_port", "remote_host",
|
|
"remote_port", "autostart", "autoreconnect")}
|
|
host = hosts_by_id.get(t.get("host_id"))
|
|
out["host_label"] = (f"{host['username']}@{host['hostname']}:{host.get('port', 22)}"
|
|
if host else "(主机不存在)")
|
|
out["host_auth"] = host.get("auth") if host else None
|
|
tun = self._manager.tunnels.get(t["id"])
|
|
out["runtime"] = tun.snapshot() if tun else {
|
|
"id": t["id"], "status": "stopped", "error": "",
|
|
"started_at": None, "conn_count": 0}
|
|
out["logs"] = list(tun.logs)[-60:] if tun else []
|
|
tunnels.append(out)
|
|
return {
|
|
"hosts": hosts,
|
|
"tunnels": tunnels,
|
|
"feed": self._feed(),
|
|
"config_dir": str(self._store.data_dir),
|
|
}
|
|
|
|
@staticmethod
|
|
def _host_out(h: dict) -> dict:
|
|
return {
|
|
"id": h["id"], "name": h.get("name"), "hostname": h.get("hostname"),
|
|
"port": h.get("port", 22), "username": h.get("username"),
|
|
"auth": h.get("auth", "password"), "key_path": h.get("key_path", ""),
|
|
"note": h.get("note", ""),
|
|
"has_password": bool(h.get("password")),
|
|
"has_passphrase": bool(h.get("key_passphrase")),
|
|
}
|
|
|
|
def _feed(self) -> list:
|
|
rows = []
|
|
for tun in self._manager.tunnels.values():
|
|
for e in tun.logs:
|
|
rows.append({"tunnel_id": tun.cfg["id"],
|
|
"tunnel_name": tun.cfg.get("name", tun.cfg["id"]), **e})
|
|
rows.sort(key=lambda r: r["t"])
|
|
return rows[-300:]
|
|
|
|
def _apply(self) -> None:
|
|
self._manager.apply_config(
|
|
self._store.data["tunnels"],
|
|
{h["id"]: h for h in self._store.data["hosts"]})
|
|
|
|
@staticmethod
|
|
def _fail(error: str) -> dict:
|
|
return {"ok": False, "error": error}
|
|
|
|
# ---------- JS 接口 ----------
|
|
|
|
def get_state(self) -> dict:
|
|
return {"ok": True, "state": self._state()}
|
|
|
|
def save_host(self, p: dict) -> dict:
|
|
try:
|
|
hostname = (p.get("hostname") or "").strip()
|
|
username = (p.get("username") or "").strip()
|
|
port = p.get("port") or 22
|
|
if not hostname:
|
|
return self._fail("请填写主机地址")
|
|
if not username:
|
|
return self._fail("请填写用户名")
|
|
if not _port_ok(port):
|
|
return self._fail("SSH 端口须为 1-65535")
|
|
auth = p.get("auth") or "password"
|
|
if auth == "key" and not (p.get("key_path") or "").strip():
|
|
return self._fail("请选择私钥文件")
|
|
|
|
hosts = self._store.data["hosts"]
|
|
old = next((h for h in hosts if h["id"] == p.get("id")), None)
|
|
rec = {
|
|
"id": old["id"] if old else self._store.new_id(),
|
|
"name": (p.get("name") or "").strip() or hostname,
|
|
"hostname": hostname,
|
|
"port": int(port),
|
|
"username": username,
|
|
"auth": auth,
|
|
"password": (p.get("password") or "").strip(),
|
|
"key_path": (p.get("key_path") or "").strip(),
|
|
"key_passphrase": (p.get("key_passphrase") or ""),
|
|
"note": (p.get("note") or "").strip(),
|
|
}
|
|
if old:
|
|
# 留空 = 保持原值;切换认证方式时清掉另一侧的机密
|
|
for k in _SENSITIVE:
|
|
if rec[k] == "":
|
|
rec[k] = old.get(k, "")
|
|
if auth == "key":
|
|
rec["password"] = ""
|
|
else:
|
|
rec["key_passphrase"] = ""
|
|
hosts[hosts.index(old)] = rec
|
|
else:
|
|
hosts.append(rec)
|
|
self._store.save()
|
|
self._apply()
|
|
return {"ok": True, "id": rec["id"], "state": self._state()}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
def delete_host(self, host_id: str) -> dict:
|
|
used = [t["name"] for t in self._store.data["tunnels"] if t.get("host_id") == host_id]
|
|
if used:
|
|
return self._fail("该主机正被隧道使用:" + "、".join(used))
|
|
self._store.data["hosts"] = [h for h in self._store.data["hosts"] if h["id"] != host_id]
|
|
self._store.save()
|
|
self._apply()
|
|
return {"ok": True, "state": self._state()}
|
|
|
|
def save_tunnel(self, p: dict) -> dict:
|
|
try:
|
|
host_id = p.get("host_id")
|
|
if not host_id or host_id not in {h["id"] for h in self._store.data["hosts"]}:
|
|
return self._fail("请选择主机(可在「主机」页先添加)")
|
|
local_port = p.get("local_port")
|
|
remote_port = p.get("remote_port")
|
|
if not _port_ok(local_port):
|
|
return self._fail("本地端口须为 1-65535")
|
|
if not _port_ok(remote_port):
|
|
return self._fail("远程端口须为 1-65535")
|
|
remote_host = (p.get("remote_host") or "").strip() or "127.0.0.1"
|
|
tid = p.get("id")
|
|
for t in self._store.data["tunnels"]:
|
|
if t["id"] != tid and int(t["local_port"]) == int(local_port):
|
|
return self._fail(f"本地端口 {local_port} 已被隧道「{t['name']}」使用")
|
|
|
|
tunnels = self._store.data["tunnels"]
|
|
old = next((t for t in tunnels if t["id"] == tid), None)
|
|
rec = {
|
|
"id": old["id"] if old else self._store.new_id(),
|
|
"name": (p.get("name") or "").strip() or f"{remote_host}:{remote_port}",
|
|
"host_id": host_id,
|
|
"local_port": int(local_port),
|
|
"remote_host": remote_host,
|
|
"remote_port": int(remote_port),
|
|
"autostart": bool(p.get("autostart")),
|
|
"autoreconnect": bool(p.get("autoreconnect", True)),
|
|
}
|
|
if old:
|
|
tunnels[tunnels.index(old)] = rec
|
|
else:
|
|
tunnels.append(rec)
|
|
self._store.save()
|
|
self._apply()
|
|
return {"ok": True, "id": rec["id"], "state": self._state()}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
def delete_tunnel(self, tunnel_id: str) -> dict:
|
|
tun = self._manager.tunnels.pop(tunnel_id, None)
|
|
if tun:
|
|
tun.stop()
|
|
self._store.data["tunnels"] = [t for t in self._store.data["tunnels"] if t["id"] != tunnel_id]
|
|
self._store.save()
|
|
return {"ok": True, "state": self._state()}
|
|
|
|
def start_tunnel(self, tunnel_id: str) -> dict:
|
|
self._apply()
|
|
tun = self._manager.tunnels.get(tunnel_id)
|
|
if not tun:
|
|
return self._fail("隧道不存在")
|
|
tun.start()
|
|
return {"ok": True}
|
|
|
|
def stop_tunnel(self, tunnel_id: str) -> dict:
|
|
tun = self._manager.tunnels.get(tunnel_id)
|
|
if not tun:
|
|
return {"ok": True}
|
|
tun.stop()
|
|
return {"ok": True}
|
|
|
|
def probe_host(self, p: dict) -> dict:
|
|
"""用表单当前值(空密码沿用已存值)测试连接。"""
|
|
try:
|
|
old = next((h for h in self._store.data["hosts"] if h["id"] == p.get("id")), None)
|
|
host = {
|
|
"hostname": (p.get("hostname") or "").strip(),
|
|
"port": int(p.get("port") or 22),
|
|
"username": (p.get("username") or "").strip(),
|
|
"auth": p.get("auth") or "password",
|
|
"password": (p.get("password") or ""),
|
|
"key_path": (p.get("key_path") or "").strip(),
|
|
"key_passphrase": (p.get("key_passphrase") or ""),
|
|
}
|
|
if old:
|
|
for k in _SENSITIVE:
|
|
if host[k] == "":
|
|
host[k] = old.get(k, "")
|
|
if not host["hostname"]:
|
|
return self._fail("请先填写主机地址")
|
|
ok, msg = probe_host(host, self._manager.known_hosts_path)
|
|
return {"ok": ok, "message": msg}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
def open_url(self, url: str) -> dict:
|
|
if not (url.startswith("http://") or url.startswith("https://")):
|
|
return self._fail("仅允许打开 http(s) 链接")
|
|
try:
|
|
os.startfile(url) # noqa: S606
|
|
return {"ok": True}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
def copy_text(self, text: str) -> dict:
|
|
try:
|
|
CF_UNICODETEXT = 13
|
|
GMEM_MOVEABLE = 0x0002
|
|
u32 = ctypes.WinDLL("user32")
|
|
k32 = ctypes.WinDLL("kernel32")
|
|
if not u32.OpenClipboard(None):
|
|
return self._fail("剪贴板被占用")
|
|
try:
|
|
u32.EmptyClipboard()
|
|
buf = ctypes.create_unicode_buffer(text)
|
|
h = k32.GlobalAlloc(GMEM_MOVEABLE, ctypes.sizeof(buf))
|
|
ptr = k32.GlobalLock(h)
|
|
ctypes.memmove(ptr, buf, ctypes.sizeof(buf))
|
|
k32.GlobalUnlock(h)
|
|
u32.SetClipboardData(CF_UNICODETEXT, h)
|
|
finally:
|
|
u32.CloseClipboard()
|
|
return {"ok": True}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
def pick_key_file(self) -> dict:
|
|
try:
|
|
import tkinter as tk
|
|
from tkinter import filedialog
|
|
root = tk.Tk()
|
|
root.withdraw()
|
|
root.attributes("-topmost", True)
|
|
path = filedialog.askopenfilename(
|
|
parent=root, title="选择私钥文件",
|
|
filetypes=[("私钥 / 所有文件", "*.*")])
|
|
root.destroy()
|
|
return {"ok": True, "path": path}
|
|
except Exception as e: # noqa: BLE001
|
|
return self._fail(str(e))
|
|
|
|
# ---------- 窗口控制 ----------
|
|
|
|
def win_minimize(self) -> dict:
|
|
"""标题栏「最小化」:隐藏到系统托盘。"""
|
|
self.hide_main_window()
|
|
return {"ok": True}
|
|
|
|
def win_maximize(self) -> dict:
|
|
if self._win:
|
|
if self._maximized:
|
|
self._win.restore()
|
|
else:
|
|
self._win.maximize()
|
|
self._maximized = not self._maximized
|
|
return {"ok": True}
|
|
|
|
def win_close(self) -> dict:
|
|
"""标题栏「关闭」:隐藏到系统托盘,隧道保持运行;退出走托盘菜单。"""
|
|
self.hide_main_window()
|
|
return {"ok": True}
|
|
|
|
# ---------- 托盘联动(供 TrayController / 事件调用) ----------
|
|
|
|
def hide_main_window(self) -> None:
|
|
if self._win:
|
|
try:
|
|
self._win.hide()
|
|
except Exception:
|
|
pass
|
|
self.tray.notify_minimized_once()
|
|
|
|
def show_main_window(self) -> None:
|
|
if self._win:
|
|
try:
|
|
self._win.show()
|
|
except Exception:
|
|
pass
|
|
|
|
def exit_application(self) -> None:
|
|
self._really_exiting = True
|
|
self.tray.stop()
|
|
try:
|
|
self._manager.stop_all()
|
|
except Exception:
|
|
pass
|
|
if self._win:
|
|
try:
|
|
self._win.destroy()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main() -> None:
|
|
dd = data_dir()
|
|
store = Store(dd)
|
|
holder: dict = {}
|
|
manager = TunnelManager(lambda ev: holder["api"]._emit(ev), dd)
|
|
api = Api(store, manager)
|
|
holder["api"] = api
|
|
|
|
# 随应用启动的隧道在窗口出现前就拉起
|
|
api._apply()
|
|
|
|
win = webview.create_window(
|
|
"SSH 隧道管理器", str(UI_PATH), js_api=api,
|
|
width=1120, height=740, min_size=(900, 580),
|
|
frameless=True, easy_drag=False,
|
|
background_color="#111418",
|
|
)
|
|
api._win = win
|
|
win.events.closed += api._manager.stop_all
|
|
|
|
def on_closing(*_args):
|
|
"""Alt+F4 / 系统关窗请求也转为隐藏到托盘。"""
|
|
if api._really_exiting:
|
|
return None
|
|
api.hide_main_window()
|
|
return False
|
|
|
|
try:
|
|
win.events.closing += on_closing
|
|
except Exception:
|
|
pass
|
|
|
|
def after_start():
|
|
# 任务栏 / Alt+Tab / 窗口图标
|
|
try:
|
|
import clr
|
|
clr.AddReference("System.Drawing")
|
|
from System.Drawing import Icon
|
|
win.native.Icon = Icon(str(ICON_PATH))
|
|
except Exception:
|
|
pass
|
|
api.tray.start()
|
|
|
|
try:
|
|
webview.start(after_start)
|
|
except Exception as e: # noqa: BLE001
|
|
msg = ("启动失败:未检测到 WebView2 运行时。\n\n"
|
|
"请安装 Microsoft Edge WebView2 Runtime 后重试。\n"
|
|
"下载: https://developer.microsoft.com/microsoft-edge/webview2/\n\n"
|
|
f"详细错误: {e}")
|
|
ctypes.windll.user32.MessageBoxW(None, msg, "SSH 隧道管理器", 0x10)
|
|
raise SystemExit(1) from e
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except SystemExit:
|
|
raise
|
|
except Exception: # noqa: BLE001
|
|
# pythonw 无控制台,任何未捕获异常都必须可见,否则表现为"闪退"
|
|
import traceback
|
|
tb = traceback.format_exc()
|
|
log_path = None
|
|
try:
|
|
log_path = data_dir() / "crash.log"
|
|
log_path.write_text(tb, encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
msg = "程序遇到错误,已退出。"
|
|
if log_path:
|
|
msg += f"\n\n详细日志:{log_path}"
|
|
try:
|
|
ctypes.windll.user32.MessageBoxW(
|
|
None, msg + "\n\n" + tb[-900:], "SSH 隧道管理器", 0x10)
|
|
except Exception:
|
|
pass
|
|
raise SystemExit(1)
|