cmautobuy/client/src/windows_credential_store.py

"""使用 Windows 凭据管理器保存在线更新密码。

密码只在调用期间存在于内存，不进入 SQLite、日志或普通错误信息。本模块只使用
Windows 自带 Credential API，不增加第三方依赖。
"""

from __future__ import annotations

import ctypes
import sys
from ctypes import wintypes
from typing import Optional
from urllib.parse import urlsplit


UPDATE_CREDENTIAL_TARGET_PREFIX = "CMAutoBuy/update"
_CRED_TYPE_GENERIC = 1
_CRED_PERSIST_LOCAL_MACHINE = 2
_ERROR_NOT_FOUND = 1168


class CredentialStoreError(RuntimeError):
    """Windows 凭据管理器操作失败。"""


class _CredentialW(ctypes.Structure):
    _fields_ = [
        ("Flags", wintypes.DWORD),
        ("Type", wintypes.DWORD),
        ("TargetName", wintypes.LPWSTR),
        ("Comment", wintypes.LPWSTR),
        ("LastWritten", wintypes.FILETIME),
        ("CredentialBlobSize", wintypes.DWORD),
        ("CredentialBlob", ctypes.POINTER(ctypes.c_ubyte)),
        ("Persist", wintypes.DWORD),
        ("AttributeCount", wintypes.DWORD),
        ("Attributes", wintypes.LPVOID),
        ("TargetAlias", wintypes.LPWSTR),
        ("UserName", wintypes.LPWSTR),
    ]


class WindowsCredentialStore:
    """读写当前 Windows 用户的通用凭据。"""

    def __init__(self, target: str):
        self._target = target
        if sys.platform != "win32":
            self._api = None
            return
        api = ctypes.WinDLL("Advapi32.dll", use_last_error=True)
        api.CredWriteW.argtypes = [ctypes.POINTER(_CredentialW), wintypes.DWORD]
        api.CredWriteW.restype = wintypes.BOOL
        api.CredReadW.argtypes = [
            wintypes.LPCWSTR,
            wintypes.DWORD,
            wintypes.DWORD,
            ctypes.POINTER(ctypes.POINTER(_CredentialW)),
        ]
        api.CredReadW.restype = wintypes.BOOL
        api.CredDeleteW.argtypes = [
            wintypes.LPCWSTR,
            wintypes.DWORD,
            wintypes.DWORD,
        ]
        api.CredDeleteW.restype = wintypes.BOOL
        api.CredFree.argtypes = [wintypes.LPVOID]
        api.CredFree.restype = None
        self._api = api

    @property
    def target(self) -> str:
        """返回不含密码的系统凭据目标名。"""

        return self._target

    @classmethod
    def for_url(cls, url: str) -> "WindowsCredentialStore":
        """按 URL 源创建独立凭据，避免把一个服务器密码发送给另一个服务器。"""

        parsed = urlsplit(url)
        scheme = parsed.scheme.lower()
        host = (parsed.hostname or "").lower()
        if not scheme or not host:
            raise CredentialStoreError("无法为无效更新地址保存密码")
        port = parsed.port or (443 if scheme == "https" else 80)
        return cls(f"{UPDATE_CREDENTIAL_TARGET_PREFIX}/{scheme}/{host}/{port}")

    def save(self, username: str, password: str) -> None:
        """保存账号和密码；密码为空时拒绝覆盖已有凭据。"""

        api = self._require_windows()
        normalized_username = username.strip()
        if not normalized_username:
            raise CredentialStoreError("更新账号不能为空")
        if not password:
            raise CredentialStoreError("更新密码不能为空")

        password_bytes = password.encode("utf-16-le")
        if len(password_bytes) > 2560:
            raise CredentialStoreError("更新密码过长")
        blob = ctypes.create_string_buffer(password_bytes)
        credential = _CredentialW()
        credential.Type = _CRED_TYPE_GENERIC
        credential.TargetName = self._target
        credential.CredentialBlobSize = len(password_bytes)
        credential.CredentialBlob = ctypes.cast(
            blob,
            ctypes.POINTER(ctypes.c_ubyte),
        )
        credential.Persist = _CRED_PERSIST_LOCAL_MACHINE
        credential.UserName = normalized_username
        if not api.CredWriteW(ctypes.byref(credential), 0):
            raise self._system_error("保存更新凭据失败")

    def read(self) -> Optional[tuple[str, str]]:
        """返回保存的 ``(账号, 密码)``；不存在时返回 ``None``。"""

        api = self._require_windows()
        pointer = ctypes.POINTER(_CredentialW)()
        if not api.CredReadW(
            self._target,
            _CRED_TYPE_GENERIC,
            0,
            ctypes.byref(pointer),
        ):
            error_code = ctypes.get_last_error()
            if error_code == _ERROR_NOT_FOUND:
                return None
            raise self._system_error("读取更新凭据失败", error_code)
        try:
            credential = pointer.contents
            password_bytes = ctypes.string_at(
                credential.CredentialBlob,
                credential.CredentialBlobSize,
            )
            password = password_bytes.decode("utf-16-le")
            return credential.UserName or "", password
        except (UnicodeDecodeError, ValueError) as exc:
            raise CredentialStoreError("保存的更新凭据无法读取，请重新保存") from exc
        finally:
            api.CredFree(pointer)

    def exists(self) -> bool:
        """返回是否已经保存更新凭据。"""

        return self.read() is not None

    def delete(self) -> bool:
        """删除更新凭据；原本不存在时返回 ``False``。"""

        api = self._require_windows()
        if api.CredDeleteW(self._target, _CRED_TYPE_GENERIC, 0):
            return True
        error_code = ctypes.get_last_error()
        if error_code == _ERROR_NOT_FOUND:
            return False
        raise self._system_error("删除更新凭据失败", error_code)

    def _require_windows(self):
        if self._api is None:
            raise CredentialStoreError("更新密码只能保存到 Windows 凭据管理器")
        return self._api

    @staticmethod
    def _system_error(message: str, error_code: Optional[int] = None):
        code = ctypes.get_last_error() if error_code is None else error_code
        return CredentialStoreError(f"{message}（Windows 错误 {code}）")