96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
"""节点与登录凭据的本机加密存储。"""
|
|
|
|
import ctypes
|
|
import json
|
|
import os
|
|
|
|
|
|
class _DpapiBlob(ctypes.Structure):
|
|
_fields_ = [
|
|
("cbData", ctypes.c_uint32),
|
|
("pbData", ctypes.POINTER(ctypes.c_char)),
|
|
]
|
|
|
|
|
|
class NodeSecretsStore:
|
|
"""使用 Windows DPAPI 保存节点和 SSO 凭据；非 Windows 保持明文兼容。"""
|
|
|
|
def __init__(self, user_dir_getter):
|
|
self._user_dir_getter = user_dir_getter
|
|
|
|
@staticmethod
|
|
def protect_bytes(data: bytes) -> bytes:
|
|
crypt32 = ctypes.WinDLL("crypt32")
|
|
kernel32 = ctypes.WinDLL("kernel32")
|
|
crypt32.CryptProtectData.argtypes = [
|
|
ctypes.POINTER(_DpapiBlob), ctypes.c_wchar_p, ctypes.POINTER(_DpapiBlob),
|
|
ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
|
|
]
|
|
crypt32.CryptProtectData.restype = ctypes.c_bool
|
|
kernel32.LocalFree.argtypes = [ctypes.c_void_p]
|
|
buffer = ctypes.create_string_buffer(data, len(data))
|
|
blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
|
|
blob_out = _DpapiBlob()
|
|
if not crypt32.CryptProtectData(ctypes.byref(blob_in), "RRPP nodes.json", None,
|
|
None, None, 0, ctypes.byref(blob_out)):
|
|
raise OSError("CryptProtectData 调用失败")
|
|
try:
|
|
return ctypes.string_at(blob_out.pbData, blob_out.cbData)
|
|
finally:
|
|
kernel32.LocalFree(blob_out.pbData)
|
|
|
|
@staticmethod
|
|
def unprotect_bytes(data: bytes) -> bytes:
|
|
crypt32 = ctypes.WinDLL("crypt32")
|
|
kernel32 = ctypes.WinDLL("kernel32")
|
|
crypt32.CryptUnprotectData.argtypes = [
|
|
ctypes.POINTER(_DpapiBlob), ctypes.POINTER(ctypes.c_wchar_p), ctypes.POINTER(_DpapiBlob),
|
|
ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
|
|
]
|
|
crypt32.CryptUnprotectData.restype = ctypes.c_bool
|
|
kernel32.LocalFree.argtypes = [ctypes.c_void_p]
|
|
buffer = ctypes.create_string_buffer(data, len(data))
|
|
blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
|
|
description = ctypes.c_wchar_p()
|
|
blob_out = _DpapiBlob()
|
|
if not crypt32.CryptUnprotectData(ctypes.byref(blob_in), ctypes.byref(description), None,
|
|
None, None, 0, ctypes.byref(blob_out)):
|
|
raise OSError("CryptUnprotectData 调用失败")
|
|
try:
|
|
return ctypes.string_at(blob_out.pbData, blob_out.cbData)
|
|
finally:
|
|
kernel32.LocalFree(blob_out.pbData)
|
|
if description.value:
|
|
kernel32.LocalFree(description)
|
|
|
|
def encrypt_text(self, text: str) -> bytes:
|
|
if os.name != "nt":
|
|
return text.encode("utf-8")
|
|
return self.protect_bytes(text.encode("utf-8"))
|
|
|
|
def decrypt_bytes(self, blob: bytes) -> str:
|
|
if os.name != "nt":
|
|
return blob.decode("utf-8")
|
|
try:
|
|
return self.unprotect_bytes(blob).decode("utf-8")
|
|
except Exception:
|
|
return blob.decode("utf-8")
|
|
|
|
def save_auth(self, username, password):
|
|
directory = self._user_dir_getter()
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
text = json.dumps({"username": username, "password": password}, ensure_ascii=False)
|
|
(directory / "auth.json").write_bytes(self.encrypt_text(text))
|
|
|
|
def load_auth(self):
|
|
try:
|
|
path = self._user_dir_getter() / "auth.json"
|
|
if not path.is_file():
|
|
return None
|
|
data = json.loads(self.decrypt_bytes(path.read_bytes()))
|
|
if not (isinstance(data, dict) and data.get("username") and data.get("password")):
|
|
return None
|
|
return data
|
|
except Exception:
|
|
return None
|