diff --git a/README.md b/README.md index 4c22efc..8aba054 100644 --- a/README.md +++ b/README.md @@ -1,53 +1,69 @@ # SSH 隧道管理器 -Windows 桌面小工具,用于可视化管理 SSH 本地端口转发。 +跨平台 SSH 本地端口转发工具:Windows 图形界面,Linux/macOS 命令行。 -典型场景:远程主机上有个服务只监听它自己的 `127.0.0.1:3000`,公网访问不到。 -用本工具建一条隧道后,在本机打开 `http://127.0.0.1:3000` 即可直接访问该服务。 +## 一键安装 -## 运行 +**Windows**(图形界面,自动装好依赖、建快捷方式并启动): -双击 **`启动.bat`**。首次运行会自动创建虚拟环境并安装依赖(pywebview、paramiko), -之后秒开。也可以手动: - -```bat -python -m venv .venv -.venv\Scripts\python -m pip install pywebview paramiko -.venv\Scripts\python main.py +```powershell +powershell -c "irm https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.ps1 | iex" ``` -要求:Windows 10/11(自带 WebView2 运行时,如缺失请安装 -[Edge WebView2 Runtime](https://developer.microsoft.com/microsoft-edge/webview2/))、Python 3.10+。 +**Linux / macOS**(命令行版,安装到 `~/ssh-e2e`,提供 `ssh-tunnel` 命令): -## 使用 +```bash +curl -fsSL https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.sh | bash +``` -1. **主机**:添加一台 SSH 主机(地址、端口、用户名;认证支持密码或私钥,可点「测试连接」验证)。 -2. **隧道**:新建一条隧道,选择主机,填远程地址(一般 `127.0.0.1`)、远程端口和本地端口,点「启动」。 -3. 本机访问 `127.0.0.1:本地端口` 即等于访问远程服务;卡片上的「打开」按钮可直接在浏览器打开。 +## 典型场景 -其他:断线自动重连(间隔递增)、随应用启动、运行时长与累计连接数、每条隧道及全局日志。 -点「最小化」或「关闭」会驻留到**系统托盘**(隧道在后台保持运行):左键托盘图标恢复窗口, -右键菜单可显示或退出;首次最小化会有气泡提示。 -配置保存在 `%APPDATA%\SshTunnelManager\config.json`,密码与私钥口令经 Windows DPAPI -加密后落盘(绑定当前 Windows 用户,明文不落盘);主机公钥记录在同目录 `known_hosts`, -主机密钥变更时会拒绝连接。 +远程主机上有个服务只监听它自己的 `127.0.0.1:3000`,公网访问不到。 +建一条隧道后,在本机打开 `http://127.0.0.1:3000` 即可直接访问该服务。 -## 文件结构 +## 命令行用法(Linux/macOS,Windows 亦可 `python cli.py`) + +```bash +ssh-tunnel host add 我的服务器 root@1.2.3.4 # 添加主机(交互输入密码;私钥加 --key ~/.ssh/id_rsa) +ssh-tunnel test 我的服务器 # 测试连通与认证 +ssh-tunnel tunnel add web --host 我的服务器 --lp 3000 --rp 3000 +ssh-tunnel start web # 前台运行,Ctrl+C 断开 +ssh-tunnel list # 查看全部 +ssh-tunnel remove tunnel web +``` + +## 图形界面(Windows) + +双击仓库里的 **`启动.bat`**,或用上面的一键安装命令。 + +1. **主机**:添加 SSH 主机(密码/私钥认证,可「测试连接」)。 +2. **隧道**:选主机,填远程地址(一般 `127.0.0.1`)、远程端口、本地端口,点「启动」。 +3. 本机访问 `127.0.0.1:本地端口` 即等于远程服务;点「最小化/关闭」驻留系统托盘,隧道后台保持。 + +## 数据与安全 + +配置文件位置:Windows `%APPDATA%\SshTunnelManager\`、Linux `~/ssh-e2e/.config/`、 +macOS `~/ssh-e2e/Library/`。密码与私钥口令:Windows 经 DPAPI 加密;Linux/macOS +在装有 keyring 时经 AES-GCM 加密(keyring 派生密钥),否则明文保存并在运行时提示。 +主机公钥记录在配置目录 `known_hosts`,主机密钥变更时拒绝连接。 + +## 开发 ``` -main.py 入口:pywebview 窗口与 JS API +main.py GUI 入口(pywebview + WebView2,仅 Windows) +cli.py 命令行入口(跨平台) engine.py 隧道引擎:paramiko 转发、保活、自动重连 -storage.py 配置存储:JSON + DPAPI 加密 -ui/ 前端界面(原生 WebView2 渲染) -test_sshd.py 本地测试用 SSH 服务端(自测) -test_e2e.py 端到端自测脚本(需先启动 test_sshd) +storage.py 配置存储:JSON + 平台加密 +ui/ 图形界面(HTML/CSS/JS) +install.sh / install.ps1 一键安装脚本 +test_sshd.py / test_e2e.py 本地自测(见下) ``` -## 自测 +自测: -```bat -:: 窗口 1 -.venv\Scripts\python test_sshd.py -:: 窗口 2(会在本机 3210 起一个模拟"远程服务",经隧道 13000 访问) -.venv\Scripts\python test_e2e.py +```bash +.venv/Scripts/python test_sshd.py # 窗口1:本地测试 SSH 服务端 +.venv/Scripts/python test_e2e.py # 窗口2:端到端隧道验证 +.venv/Scripts/python cli.py list # CLI 冒烟 ``` + diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..9c2c372 --- /dev/null +++ b/cli.py @@ -0,0 +1,271 @@ +#!/usr/bin/env python3 +"""SSH 隧道管理器 — 命令行模式(跨平台,无界面)。 + +用法: + ssh-tunnel host add <名称> [端口=22] [--password] [--key 路径] + ssh-tunnel tunnel add <名称> --host <主机名或ID> --lp 本地端口 --rp 远程端口 [--rh 远程地址=127.0.0.1] + ssh-tunnel start <隧道名或ID> [隧道...] # 前台运行,Ctrl+C 退出 + ssh-tunnel stop-all # (与 start 配套,信号处理自动调用) + ssh-tunnel list # 列出主机与隧道 + ssh-tunnel remove host|tunnel <名称或ID> + ssh-tunnel test <主机名或ID> +""" +from __future__ import annotations + +import argparse +import getpass +import signal +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +from engine import TunnelManager, probe_host # noqa: E402 +from storage import Store, data_root # noqa: E402 + + +def _print(msg: str) -> None: + print(msg, flush=True) + + +class _QuietEmit: + """CLI 模式下把引擎事件打到终端。""" + + def __init__(self): + self._last: dict[str, str] = {} + + def __call__(self, ev: dict) -> None: + if ev.get("type") == "tunnel": + if self._last.get(ev["id"]) == ev["status"]: + return # 状态未变(如连接计数刷新),不重复打印 + self._last[ev["id"]] = ev["status"] + t = ev["status"] + if t == "connected": + _print("[隧道] 已连接") + elif t == "error": + _print(f"[隧道] 连接失败: {ev.get('error', '')}") + elif ev.get("type") == "log": + e = ev["entry"] + lv = e["level"] + mark = {"info": "·", "success": "✔", "warn": "!", "error": "×"}.get(lv, "·") + _print(f"[{e['t']}] {mark} {e['msg']}") + + +def find_host(store: Store, ref: str): + hosts = store.data["hosts"] + return next((h for h in hosts if h["id"] == ref or h["name"] == ref), None) + + +def find_tunnel(store: Store, ref: str): + ts = store.data["tunnels"] + return next((t for t in ts if t["id"] == ref or t["name"] == ref), None) + + +def cmd_host_add(store, args) -> int: + if "@" not in args.target: + _print("错误: 目标格式应为 user@host") + return 1 + user, host = args.target.rsplit("@", 1) + auth = "key" if args.key else "password" + if find_host(store, args.name): + _print(f"错误: 主机「{args.name}」已存在") + return 1 + password = passphrase = "" + if auth == "password" and args.password: + password = getpass.getpass("登录密码: ") + if auth == "key" and args.key: + if not Path(args.key).expanduser().exists(): + _print(f"错误: 私钥文件不存在: {args.key}") + return 1 + passphrase = getpass.getpass("私钥口令(无则直接回车): ") + store.data["hosts"].append({ + "id": Store.new_id(), "name": args.name, "hostname": host, + "port": args.port, "username": user, "auth": auth, + "password": password, "key_path": str(Path(args.key).expanduser()) if args.key else "", + "key_passphrase": passphrase, "note": "", + }) + store.save() + _print(f"已添加主机 {user}@{host}:{args.port}(认证:{'私钥' if auth == 'key' else '密码'})") + return 0 + + +def cmd_tunnel_add(store, args) -> int: + host = find_host(store, args.host) + if not host: + _print(f"错误: 找不到主机「{args.host}」,先执行 host add") + return 1 + for t in store.data["tunnels"]: + if t["local_port"] == args.lp: + _print(f"错误: 本地端口 {args.lp} 已被隧道「{t['name']}」使用") + return 1 + store.data["tunnels"].append({ + "id": Store.new_id(), "name": args.name, "host_id": host["id"], + "local_port": args.lp, "remote_host": args.rh or "127.0.0.1", + "remote_port": args.rp, "autostart": False, "autoreconnect": True, + }) + store.save() + _print(f"已添加隧道 {args.name}: 127.0.0.1:{args.lp} → {args.rh or '127.0.0.1'}:{args.rp}(经 {host['name']})") + _print(f"启动: ssh-tunnel start {args.name}") + return 0 + + +def cmd_list(store, _args) -> int: + if store.warn_plaintext: + _print("提示: 当前平台无可用密钥后端,密码以明文保存于配置文件。") + _print("主机:") + if not store.data["hosts"]: + _print(" (无) 添加: ssh-tunnel host add <名称> user@host") + for h in store.data["hosts"]: + _print(f" {h['name']} {h['username']}@{h['hostname']}:{h['port']} " + f"{'私钥' if h.get('auth') == 'key' else '密码'}") + _print("隧道:") + if not store.data["tunnels"]: + _print(" (无) 添加: ssh-tunnel tunnel add <名称> --host <主机> --lp 本地端口 --rp 远程端口") + for t in store.data["tunnels"]: + h = next((x for x in store.data["hosts"] if x["id"] == t["host_id"]), None) + hn = h["name"] if h else "(主机缺失)" + _print(f" {t['name']} 127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']} 经 {hn}") + return 0 + + +def cmd_test(store, args) -> int: + host = find_host(store, args.host) + if not host: + _print(f"错误: 找不到主机「{args.host}」") + return 1 + _print(f"正在测试 {host['username']}@{host['hostname']}:{host['port']} ...") + ok, msg = probe_host(host, data_root() / "known_hosts") + _print(("✔ " if ok else "× ") + msg) + return 0 if ok else 1 + + +def cmd_remove(store, args) -> int: + if args.kind == "host": + h = find_host(store, args.ref) + if not h: + _print(f"错误: 找不到主机「{args.ref}」") + return 1 + used = [t["name"] for t in store.data["tunnels"] if t["host_id"] == h["id"]] + if used: + _print(f"错误: 主机被隧道使用: {', '.join(used)}") + return 1 + store.data["hosts"].remove(h) + store.save() + _print(f"已删除主机 {h['name']}") + else: + t = find_tunnel(store, args.ref) + if not t: + _print(f"错误: 找不到隧道「{args.ref}」") + return 1 + store.data["tunnels"].remove(t) + store.save() + _print(f"已删除隧道 {t['name']}") + return 0 + + +def cmd_start(store, args) -> int: + refs = args.tunnels + if not refs: # 未指定则启动全部 + refs = [t["name"] for t in store.data["tunnels"]] + if not refs: + _print("没有已配置的隧道。先用 tunnel add 添加。") + return 1 + hosts_by_id = {h["id"]: h for h in store.data["hosts"]} + mgr = TunnelManager(_QuietEmit(), data_root()) chosen = [] + for r in refs: + t = find_tunnel(store, r) + if not t: + _print(f"错误: 找不到隧道「{r}」") + return 1 + chosen.append(t) + mgr.apply_config(chosen, hosts_by_id) + + if store.warn_plaintext: + _print("提示: 当前平台无可用密钥后端,密码以明文保存于配置文件。") + names = ", ".join(t["name"] for t in chosen) + _print(f"启动隧道: {names} — Ctrl+C 退出") + + stopping = False + + def _stop(*_a): + nonlocal stopping + if not stopping: + stopping = True + _print("\n正在断开所有隧道 ...") + mgr.stop_all() + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + for t in mgr.tunnels.values(): + t.start() + try: + while not stopping: + time.sleep(0.3) + alive = [t for t in mgr.tunnels.values() if t.status != "stopped"] + if not alive: + _print("所有隧道已停止。") + return 1 + except KeyboardInterrupt: + _stop() + return 0 + + +def main() -> int: + p = argparse.ArgumentParser(prog="ssh-tunnel", description="SSH 隧道管理器(命令行)") + sub = p.add_subparsers(dest="cmd", required=True) + + sp = sub.add_parser("host", help="主机管理") + hsub = sp.add_subparsers(dest="sub", required=True) + ha = hsub.add_parser("add", help="添加主机") + ha.add_argument("name", help="备注名称") + ha.add_argument("target", help="user@host") + ha.add_argument("port", nargs="?", type=int, default=22) + ha.add_argument("--password", action="store_true", help="交互输入密码(默认)") + ha.add_argument("--key", help="私钥文件路径") + + sp = sub.add_parser("tunnel", help="隧道管理") + tsub = sp.add_subparsers(dest="sub", required=True) + ta = tsub.add_parser("add", help="添加隧道") + ta.add_argument("name") + ta.add_argument("--host", required=True, help="主机名称或 ID") + ta.add_argument("--lp", type=int, required=True, help="本地端口") + ta.add_argument("--rp", type=int, required=True, help="远程端口") + ta.add_argument("--rh", default="127.0.0.1", help="远程地址(默认 127.0.0.1)") + + sub.add_parser("list", help="列出主机与隧道") + sub.add_parser("stop-all", help="预留(前台模式退出时自动断开)") + + st = sub.add_parser("start", help="前台启动隧道") + st.add_argument("tunnels", nargs="*", help="隧道名称或 ID(空=全部)") + + ts = sub.add_parser("test", help="测试主机连通") + ts.add_argument("host") + + rm = sub.add_parser("remove", help="删除主机或隧道") + rm.add_argument("kind", choices=["host", "tunnel"]) + rm.add_argument("ref") + + args = p.parse_args() + store = Store(data_root()) + + if args.cmd == "host": + return cmd_host_add(store, args) + if args.cmd == "tunnel": + return cmd_tunnel_add(store, args) + if args.cmd == "list": + return cmd_list(store, args) + if args.cmd == "test": + return cmd_test(store, args) + if args.cmd == "remove": + return cmd_remove(store, args) + if args.cmd == "start": + return cmd_start(store, args) + if args.cmd == "stop-all": + _print("隧道随前台进程退出自动断开,无需单独停止。") + return 0 + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 0000000..168e6d4 --- /dev/null +++ b/install.ps1 @@ -0,0 +1,70 @@ +# SSH 隧道管理器 一键安装(Windows) +# 用法: powershell -c "irm https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.ps1 | iex" +$ErrorActionPreference = "Stop" +$BaseUrl = "https://gitea.xbay.cc/leefer/ssh-e2e" +$AppDir = Join-Path $env:USERPROFILE "ssh-e2e" + +function Say($m) { Write-Host "===> $m" -ForegroundColor Green } +function Die($m) { Write-Host "错误: $m" -ForegroundColor Red; Read-Host "按回车退出"; exit 1 } + +# ---------- 下载源码 ---------- +Say "下载最新代码到 $AppDir ..." +New-Item -ItemType Directory -Force -Path $AppDir | Out-Null +$gotCode = $false +if (Get-Command git -ErrorAction SilentlyContinue) { + if (Test-Path (Join-Path $AppDir ".git")) { + git -C $AppDir fetch --all --quiet 2>$null + git -C $AppDir reset --hard origin/main --quiet 2>$null + $gotCode = $? + } else { + git clone --depth 1 "$BaseUrl.git" $AppDir 2>$null + if ($?) { $gotCode = $true } + } + if (-not $gotCode -and (Test-Path (Join-Path $AppDir "cli.py"))) { $gotCode = $true } +} +if (-not $gotCode) { + # git 不可用或失败:下载 zip 归档 + $zip = Join-Path $env:TEMP "ssh-e2e-main.zip" + [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12 + Invoke-WebRequest "$BaseUrl/archive/main.zip" -OutFile $zip -UseBasicParsing + $tmp = Join-Path $env:TEMP "ssh-e2e-extract" + if (Test-Path $tmp) { Remove-Item $tmp -Recurse -Force } + Expand-Archive $zip -DestinationPath $tmp + $inner = Get-ChildItem $tmp | Where-Object PSIsContainer | Select-Object -First 1 + Get-ChildItem $inner.FullName | Move-Item -Destination $AppDir -Force + Remove-Item $tmp, $zip -Recurse -Force -ErrorAction SilentlyContinue +} +if (-not (Test-Path (Join-Path $AppDir "main.py"))) { Die "代码下载失败,请检查网络" } + +# ---------- Python 检查 ---------- +$py = Get-Command python -ErrorAction SilentlyContinue +if (-not $py) { Die "未找到 python,请先安装 Python 3.10+(https://www.python.org/downloads/)" } + +# ---------- 虚拟环境 ---------- +if (-not (Test-Path (Join-Path $AppDir ".venv\Scripts\pythonw.exe"))) { + Say "创建虚拟环境并安装依赖(首次约 1-2 分钟)..." + python -m venv (Join-Path $AppDir ".venv") + if ($LASTEXITCODE -ne 0) { Die "创建虚拟环境失败" } + & (Join-Path $AppDir ".venv\Scripts\python.exe") -m pip install --quiet --disable-pip-version-check -r (Join-Path $AppDir "requirements.txt") + if ($LASTEXITCODE -ne 0) { Die "依赖安装失败,请检查网络" } +} elseif (-not (Test-Path (Join-Path $AppDir ".venv\Lib\site-packages\pystray"))) { + & (Join-Path $AppDir ".venv\Scripts\python.exe") -m pip install --quiet --disable-pip-version-check -r (Join-Path $AppDir "requirements.txt") +} + +# ---------- 桌面快捷方式 ---------- +try { + $ws = New-Object -ComObject WScript.Shell + $lnk = $ws.CreateShortcut([IO.Path]::Combine([Environment]::GetFolderPath("Desktop"), "SSH 隧道管理器.lnk")) + $lnk.TargetPath = Join-Path $AppDir ".venv\Scripts\pythonw.exe" + $lnk.Arguments = '"' + (Join-Path $AppDir "main.py") + '"' + $lnk.WorkingDirectory = $AppDir + $lnk.IconLocation = Join-Path $AppDir "ui\app.ico" + $lnk.Save() + Say "已创建桌面快捷方式" +} catch { } + +# ---------- 启动 ---------- +Say "启动 SSH 隧道管理器 ..." +Start-Process -FilePath (Join-Path $AppDir ".venv\Scripts\pythonw.exe") ` + -ArgumentList ('"' + (Join-Path $AppDir "main.py") + '"') -WorkingDirectory $AppDir +Say "完成!窗口即将弹出;点最小化/关闭会驻留系统托盘。" diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..1c6146b --- /dev/null +++ b/install.sh @@ -0,0 +1,89 @@ +#!/usr/bin/env bash +# SSH 隧道管理器 一键安装(Linux / macOS) +# 用法: curl -fsSL https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.sh | bash +set -euo pipefail + +BASE_URL="https://gitea.xbay.cc/leefer/ssh-e2e" +APP_DIR="$HOME/ssh-e2e" +BIN_DIR="$HOME/.local/bin" + +say() { printf '\033[1;32m==>\033[0m %s\n' "$*"; } +die() { printf '\033[1;31m错误:\033[0m %s\n' "$*" >&2; exit 1; } + +[ "$(uname)" = "Linux" ] || [ "$(uname)" = "Darwin" ] || die "此脚本仅支持 Linux / macOS" + +# ---------- 依赖 ---------- +need_py() { + command -v python3 >/dev/null 2>&1 || die "未找到 python3,请先安装 Python 3.10+" + pyver=$(python3 -c 'import sys;print("%d.%d"%sys.version_info[:2])') + case "$pyver" in + 3.1[0-9]|3.[2-9][0-9]|[4-9].*) ;; + *) die "需要 Python 3.10+,当前 $pyver" ;; + esac +} + +# ---------- 下载源码 ---------- +say "下载最新代码到 $APP_DIR ..." +mkdir -p "$APP_DIR" +if command -v git >/dev/null 2>&1; then + if [ -d "$APP_DIR/.git" ]; then + git -C "$APP_DIR" fetch --all --quiet && git -C "$APP_DIR" reset --hard origin/main --quiet + else + git clone --depth 1 "$BASE_URL.git" "$APP_DIR" 2>/dev/null || true + fi +fi +if [ ! -f "$APP_DIR/cli.py" ]; then + # git 不可用或克隆失败:直接拉归档 + archive_url="$BASE_URL/archive/main.tar.gz" + if command -v curl >/dev/null 2>&1; then + curl -fsSL "$archive_url" | tar -xz -C "$APP_DIR" --strip-components 1 + else + wget -qO- "$archive_url" | tar -xz -C "$APP_DIR" --strip-components 1 + fi +fi +[ -f "$APP_DIR/cli.py" ] || die "代码下载失败,请检查网络后重试" + +# ---------- 虚拟环境 ---------- +if [ ! -x "$APP_DIR/.venv/bin/python" ]; then + say "创建虚拟环境并安装依赖(首次约 1-2 分钟)..." + need_py + python3 -m venv "$APP_DIR/.venv" +fi +if [ ! -e "$APP_DIR/.venv/deps-done" ]; then + "$APP_DIR/.venv/bin/pip" install --quiet --disable-pip-version-check \ + paramiko cryptography keyring > /dev/null + touch "$APP_DIR/.venv/deps-done" +fi + +# ---------- ssh-tunnel 命令 ---------- +mkdir -p "$BIN_DIR" +cat > "$BIN_DIR/ssh-tunnel" << EOF +#!/usr/bin/env bash +exec "$APP_DIR/.venv/bin/python" "$APP_DIR/cli.py" "\$@" +EOF +chmod +x "$BIN_DIR/ssh-tunnel" + +# PATH 提示 +case ":$PATH:" in + *":$BIN_DIR:"*) ;; + *) + shell_rc="$HOME/.bashrc" + [ "$(uname)" = "Darwin" ] && [ -f "$HOME/.zshrc" ] && shell_rc="$HOME/.zshrc" + grep -q 'ssh-e2e' "$shell_rc" 2>/dev/null || { + echo 'export PATH="$HOME/.local/bin:$PATH"' >> "$shell_rc" + export PATH="$BIN_DIR:$PATH" + } + ;; +esac + +say "安装完成!" +echo +echo " 配置隧道(示例):" +echo " ssh-tunnel host add 我的服务器 root@1.2.3.4 22" +echo " ssh-tunnel tunnel add web --host 我的服务器 --lp 3000 --rp 3000" +echo " ssh-tunnel start web" +echo +echo " 常用命令: ssh-tunnel list | test <主机> | remove host|tunnel <名称>" +echo " 图形界面(Windows): powershell -c \"irm $BASE_URL/raw/branch/main/install.ps1 | iex\"" +echo +[ -n "${ZSH_VERSION:-}${BASH_VERSION:-}" ] || echo "(注:新开终端或 source ~/.bashrc 后 ssh-tunnel 生效)" diff --git a/main.py b/main.py index 9e2e3ce..f4eee11 100644 --- a/main.py +++ b/main.py @@ -10,7 +10,7 @@ from pathlib import Path import webview from engine import TunnelManager, probe_host -from storage import Store +from storage import Store, data_root APP_DIR_NAME = "SshTunnelManager" APP_TITLE = "SSH 隧道管理器" @@ -21,8 +21,7 @@ _SENSITIVE = ("password", "key_passphrase") def data_dir() -> Path: - base = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming") - d = Path(base) / APP_DIR_NAME + d = data_root() d.mkdir(parents=True, exist_ok=True) return d diff --git a/storage.py b/storage.py index d9cc782..2a3ff17 100644 --- a/storage.py +++ b/storage.py @@ -1,50 +1,74 @@ -"""配置存储:JSON 文件 + DPAPI 加密敏感字段(密码/口令绑定当前 Windows 用户)。""" +"""配置存储:JSON 文件 + 平台相关的机密加密。 +Windows: DPAPI(绑定当前用户) +Linux/macOS: keyring(可用则用,不可用则明文并在加载时告警) +""" from __future__ import annotations -import base64 -import ctypes import json import os +import sys import threading import uuid -from ctypes import wintypes from pathlib import Path -_CRYPTPROTECT_UI_FORBIDDEN = 0x1 _SENSITIVE = ("password", "key_passphrase") +_PREFIX = "dpapi:" # Windows DPAPI +_PREFIX2 = "kr:" # keyring,base64(AES-GCM) -class _DataBlob(ctypes.Structure): - _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.c_void_p)] +def data_root() -> Path: + """配置目录:Windows %APPDATA%\\SshTunnelManager / macOS ~/ssh-e2e/Library / Linux ~/ssh-e2e/.config""" + app = "SshTunnelManager" + if sys.platform == "win32": + base = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming") + return Path(base) / app + if sys.platform == "darwin": + return Path.home() / "ssh-e2e" / "Library" + base = os.environ.get("XDG_CONFIG_HOME") + if base and Path(base).is_absolute(): + return Path(base) + return Path.home() / "ssh-e2e" / ".config" -def protect(text: str) -> str: +# ---------- Windows DPAPI ---------- + +def _dpapi_protect(text: str) -> str: + import base64 + import ctypes + from ctypes import wintypes + + class _DataBlob(ctypes.Structure): + _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.c_void_p)] + raw = text.encode("utf-8") buf = ctypes.create_string_buffer(raw, len(raw)) src = _DataBlob(len(raw), ctypes.cast(buf, ctypes.c_void_p)) dst = _DataBlob() ok = ctypes.windll.crypt32.CryptProtectData( - ctypes.byref(src), None, None, None, None, - _CRYPTPROTECT_UI_FORBIDDEN, ctypes.byref(dst), - ) + ctypes.byref(src), None, None, None, None, 0x1, ctypes.byref(dst)) if not ok: raise OSError("DPAPI 加密失败") try: - return "dpapi:" + base64.b64encode(ctypes.string_at(dst.pbData, dst.cbData)).decode("ascii") + return _PREFIX + base64.b64encode( + ctypes.string_at(dst.pbData, dst.cbData)).decode("ascii") finally: ctypes.windll.kernel32.LocalFree(ctypes.c_void_p(dst.pbData)) -def unprotect(text: str) -> str: - if not text.startswith("dpapi:"): - return text - raw = base64.b64decode(text[6:]) +def _dpapi_unprotect(text: str) -> str: + import base64 + import ctypes + from ctypes import wintypes + + class _DataBlob(ctypes.Structure): + _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.c_void_p)] + + raw = base64.b64decode(text[len(_PREFIX):]) buf = ctypes.create_string_buffer(raw, len(raw)) src = _DataBlob(len(raw), ctypes.cast(buf, ctypes.c_void_p)) dst = _DataBlob() ok = ctypes.windll.crypt32.CryptUnprotectData( - ctypes.byref(src), None, None, None, None, 0, ctypes.byref(dst), - ) + ctypes.byref(src), None, None, None, None, 0, ctypes.byref(dst)) if not ok: raise OSError("DPAPI 解密失败:配置可能属于其他 Windows 用户") try: @@ -53,13 +77,78 @@ def unprotect(text: str) -> str: ctypes.windll.kernel32.LocalFree(ctypes.c_void_p(dst.pbData)) +# ---------- keyring(Linux/macOS) ---------- + +def _keyring(): + try: + import keyring # noqa: PLC0415 + if keyring.get_keyring().priority >= 0: # 有真实后端(非 fail/null) + return keyring + except Exception: + pass + return None + + +def _kr_protect(text: str) -> str: + import base64 + kr = _keyring() + if kr is None: + return text # 无后端:明文(加载时会告警) + import hashlib + key = hashlib.sha256((kr.get_password("ssh-tunnel-manager", "salt") + or "fixed-salt").encode()).digest()[:16] + nonce = os.urandom(12) + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + ct = AESGCM(key).encrypt(nonce, text.encode("utf-8"), None) + return _PREFIX2 + base64.b64encode(nonce + ct).decode("ascii") + + +def _kr_unprotect(text: str) -> str: + if not text.startswith(_PREFIX2): + return text + import base64 + raw = base64.b64decode(text[len(_PREFIX2):]) + kr = _keyring() + if kr is None: + return "" + import hashlib + key = hashlib.sha256((kr.get_password("ssh-tunnel-manager", "salt") + or "fixed-salt").encode()).digest()[:16] + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + return AESGCM(key).decrypt(raw[:12], raw[12:], None).decode("utf-8") + + +# ---------- 统一入口 ---------- + +def protect(text: str) -> str: + if not text: + return text + if sys.platform == "win32": + return _dpapi_protect(text) + return _kr_protect(text) + + +def unprotect(text: str) -> str: + if not isinstance(text, str) or not text: + return text + if text.startswith(_PREFIX): + if sys.platform != "win32": + raise OSError("该配置来自 Windows 的 DPAPI 加密,无法在当前平台解密") + return _dpapi_unprotect(text) + if text.startswith(_PREFIX2): + return _kr_unprotect(text) + return text + + class Store: - """hosts/tunnels 配置的读写。内存中敏感字段为明文,落盘前用 DPAPI 加密。""" + """hosts/tunnels 配置的读写。内存中敏感字段为明文,落盘前加密。""" def __init__(self, data_dir: Path): self.data_dir = data_dir + self.data_dir.mkdir(parents=True, exist_ok=True) self.path = data_dir / "config.json" self._lock = threading.Lock() + self.warn_plaintext = False self.data = self._load() def _load(self) -> dict: @@ -77,11 +166,14 @@ class Store: for h in data["hosts"]: for k in _SENSITIVE: v = h.get(k) - if isinstance(v, str) and v.startswith("dpapi:"): - try: - h[k] = unprotect(v) - except Exception: - h[k] = "" + if isinstance(v, str) and v: + if v.startswith((_PREFIX, _PREFIX2)): + try: + h[k] = unprotect(v) + except Exception: + h[k] = "" + else: + self.warn_plaintext = True # 明文机密(如无 keyring 的 Linux) return data def save(self) -> None: