leefer
27a3b4c6a8
- 本地端口转发隧道管理(密码/私钥认证,自动重连,随应用启动) - 主机与隧道可视化配置,DPAPI 加密存储机密,known_hosts 校验 - 自定义无边框窗口,深色界面,托盘驻留(最小化/关闭到托盘) - 内置本地测试 SSH 服务端与端到端自测脚本
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""配置存储:JSON 文件 + DPAPI 加密敏感字段(密码/口令绑定当前 Windows 用户)。"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from ctypes import wintypes
|
|
from pathlib import Path
|
|
|
|
_CRYPTPROTECT_UI_FORBIDDEN = 0x1
|
|
_SENSITIVE = ("password", "key_passphrase")
|
|
|
|
|
|
class _DataBlob(ctypes.Structure):
|
|
_fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.c_void_p)]
|
|
|
|
|
|
def protect(text: str) -> str:
|
|
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),
|
|
)
|
|
if not ok:
|
|
raise OSError("DPAPI 加密失败")
|
|
try:
|
|
return "dpapi:" + 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:])
|
|
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),
|
|
)
|
|
if not ok:
|
|
raise OSError("DPAPI 解密失败:配置可能属于其他 Windows 用户")
|
|
try:
|
|
return ctypes.string_at(dst.pbData, dst.cbData).decode("utf-8")
|
|
finally:
|
|
ctypes.windll.kernel32.LocalFree(ctypes.c_void_p(dst.pbData))
|
|
|
|
|
|
class Store:
|
|
"""hosts/tunnels 配置的读写。内存中敏感字段为明文,落盘前用 DPAPI 加密。"""
|
|
|
|
def __init__(self, data_dir: Path):
|
|
self.data_dir = data_dir
|
|
self.path = data_dir / "config.json"
|
|
self._lock = threading.Lock()
|
|
self.data = self._load()
|
|
|
|
def _load(self) -> dict:
|
|
data = {"hosts": [], "tunnels": []}
|
|
if self.path.exists():
|
|
try:
|
|
loaded = json.loads(self.path.read_text(encoding="utf-8"))
|
|
data["hosts"] = loaded.get("hosts") or []
|
|
data["tunnels"] = loaded.get("tunnels") or []
|
|
except Exception:
|
|
try:
|
|
self.path.rename(self.path.with_suffix(".json.broken"))
|
|
except OSError:
|
|
pass
|
|
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] = ""
|
|
return data
|
|
|
|
def save(self) -> None:
|
|
with self._lock:
|
|
out = json.loads(json.dumps(self.data)) # 深拷贝
|
|
for h in out["hosts"]:
|
|
for k in _SENSITIVE:
|
|
if h.get(k):
|
|
h[k] = protect(h[k])
|
|
tmp = self.path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
os.replace(tmp, self.path)
|
|
|
|
@staticmethod
|
|
def new_id() -> str:
|
|
return uuid.uuid4().hex[:8]
|