leefer/ssh-e2e
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

跨平台支持:CLI 模式 + 三平台一键安装命令

Browse Source 
- 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 路径)
This commit is contained in:
leefer
2026-09-27 19:59:25 +08:00
parent 27a3b4c6a8
commit 97d8b7ae89
6 changed files with 599 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+51 -35
View File
@@ -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 冒烟
```
cli.py
+271
View File
@@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""SSH 隧道管理器 — 命令行模式(跨平台,无界面)。
用法:
ssh-tunnel host add <名称> <user@host> [端口=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())
install.ps1
+70
View File
@@ -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 "完成!窗口即将弹出;点最小化/关闭会驻留系统托盘。"
install.sh
+89
View File
@@ -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 生效)"
main.py
+2 -3
View File
@@ -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
storage.py
+116 -24
View File
@@ -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: