diff --git a/app/models/ConfigBase.py b/app/models/ConfigBase.py index d65ad72db..1fb61305f 100644 --- a/app/models/ConfigBase.py +++ b/app/models/ConfigBase.py @@ -38,7 +38,12 @@ from typing import Any, Callable, Coroutine, Generic, Type, TypeVar from urllib.parse import urlparse -from app.utils import dpapi_decrypt, dpapi_encrypt, get_logger +from app.utils import ( + dpapi_decrypt, + dpapi_encrypt, + get_logger, + looks_like_dpapi_blob, +) from app.utils.constants import ( DEFAULT_DATETIME, EMULATOR_PATH_BOOK, @@ -296,6 +301,11 @@ def correct(self, value): ) +# 密文读不出来时对外给出的占位值。界面据此提示用户重新设置， +# 存量密文本身不会被它覆盖。 +UNREADABLE_SECRET_PLACEHOLDER = "数据损坏, 请重新设置" + + class EncryptValidator(ValidatorBase): """加密数据验证器""" @@ -312,7 +322,7 @@ def correct(self, value: Any) -> Any: if self.validate(value): return value logger.warning("加密配置项无法解密, 已替换为占位值, 请重新设置") - return dpapi_encrypt("数据损坏, 请重新设置") + return dpapi_encrypt(UNREADABLE_SECRET_PLACEHOLDER) class VirtualConfigValidator(ValidatorBase): @@ -750,11 +760,18 @@ def setValue(self, value: Any) -> bool: 值是否真正发生了变化 """ - if ( - dpapi_decrypt(self.value) - if isinstance(self.validator, EncryptValidator) - else self.value - ) == value: + if isinstance(self.validator, EncryptValidator): + try: + is_unchanged = dpapi_decrypt(self.value) == value + except Exception: + # 当前密文本机解不开（多为同机另一个 Windows 账户所写）， + # 无从比较。此处一律按「有变化」处理放行，否则用户想重新 + # 填一遍账号密码时会卡在这一行抛错，连覆盖都做不到。 + is_unchanged = False + else: + is_unchanged = self.value == value + + if is_unchanged: return False if self.is_locked: @@ -769,12 +786,13 @@ def setValue(self, value: Any) -> bool: self.value = value if isinstance(self.validator, EncryptValidator): - if self.validator.validate(self.value): - self.value = self.value - else: + # 传进来的既可能是用户新填的明文，也可能是 load() 从配置文件读回的 + # 密文，只按结构区分：已经是密文就原样存下，本机解不开也不例外。 + # 那多半是同一台机器上另一个 Windows 账户写的，再加密一层会把它变成 + # 永远恢复不了的乱码，并随 load() 的脏标记写回配置文件。 + if not looks_like_dpapi_blob(self.value): self.value = dpapi_encrypt(self.value) - - if not self.validator.validate(self.value): + elif not self.validator.validate(self.value): try: self.value = self.validator.correct(self.value) except Exception: @@ -791,16 +809,23 @@ def getValue(self, if_decrypt: bool = True) -> Any: 获取配置项值 """ + is_encrypted_item = isinstance(self.validator, EncryptValidator) + try: - v = ( - self.value - if self.validator.validate(self.value) - else self.validator.correct(self.value) - ) + if self.validator.validate(self.value): + v = self.value + elif is_encrypted_item and looks_like_dpapi_blob(self.value): + # 结构完好但本机解不开的密文，多半属于同一台机器上的另一个 + # Windows 账户。落盘口径原样返回，否则 toDict 会比对出差异、 + # 把存量密文当脏数据覆盖掉；读取口径给占位提示，让界面提示 + # 重新设置。 + return UNREADABLE_SECRET_PLACEHOLDER if if_decrypt else self.value + else: + v = self.validator.correct(self.value) except Exception: v = "" - if isinstance(self.validator, EncryptValidator) and if_decrypt: + if is_encrypted_item and if_decrypt: return dpapi_decrypt(v) return v diff --git a/app/utils/__init__.py b/app/utils/__init__.py index dec7399da..c1b7587a5 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -31,6 +31,7 @@ dpapi_decrypt, dpapi_encrypt, format_exception_reason, + looks_like_dpapi_blob, sanitize_log_message, ) from .supervision import is_backend_dev_mode, is_supervised @@ -133,6 +134,7 @@ def __getattribute__(self, name: str): "get_logger", "dpapi_encrypt", "dpapi_decrypt", + "looks_like_dpapi_blob", "format_exception_reason", "sanitize_log_message", "is_backend_dev_mode", diff --git a/app/utils/platform/common/secret.py b/app/utils/platform/common/secret.py index 28f410c67..e8fc47f4a 100644 --- a/app/utils/platform/common/secret.py +++ b/app/utils/platform/common/secret.py @@ -1,5 +1,37 @@ +import base64 + from app.utils.platform.common.errors import UnsupportedPlatformError +__all__ = [ + "looks_like_dpapi_blob", + "dpapi_encrypt", + "dpapi_decrypt", + "supports_secret_storage", + "is_secret_storage_error", +] + +# DPAPI 密文固定以 4 字节版本号加 16 字节 provider GUID 开头， +# 用户态与机器态密文的这 20 字节完全一致。 +_DPAPI_BLOB_PREFIX = b"\x01\x00\x00\x00" + bytes.fromhex( + "d08c9ddf0115d1118c7a00c04fc297eb" +) + + +def looks_like_dpapi_blob(value: object) -> bool: + """按结构判断一个值是否为 DPAPI 密文，不关心本机能否解开。 + + 同一台机器上另一个 Windows 账户写的密文，本账户解不开，但它依然是密文， + 必须原样保留；只有真正的明文才需要加密。 + """ + + if not isinstance(value, str) or not value: + return False + try: + blob = base64.b64decode(value, validate=True) + except ValueError: + return False + return blob.startswith(_DPAPI_BLOB_PREFIX) + def supports_secret_storage() -> bool: """当前平台是否提供配置层所需的密文存储能力。""" diff --git a/app/utils/platform/windows/secret.py b/app/utils/platform/windows/secret.py index 8ed3af931..17b963270 100644 --- a/app/utils/platform/windows/secret.py +++ b/app/utils/platform/windows/secret.py @@ -1,9 +1,25 @@ import base64 from app.utils.platform.common.errors import UnsupportedPlatformError +from app.utils.platform.common.secret import looks_like_dpapi_blob + +__all__ = [ + "looks_like_dpapi_blob", + "dpapi_encrypt", + "dpapi_decrypt", + "supports_secret_storage", + "is_secret_storage_error", +] _SECRET_STORAGE_PROBE = "AUTO-MAS secret storage probe" +# 默认的 DPAPI 作用域绑定当前 Windows 账户，同一台机器上的另一个账户解不开。 +# 一份安装被多个本地账户共用时（配置目录只有一份），换账户启动就会读不出已存的 +# 账号密码。改用机器作用域后同机各账户都能解密；解密侧不需要对应改动，作用域写在 +# 密文里，CryptUnprotectData 与 .NET ProtectedData 都会忽略调用方传入的作用域， +# 因此既能读回旧的用户态密文，SRA 也能照常解开 MAS 写给它的密文。 +CRYPTPROTECT_LOCAL_MACHINE = 0x4 + def _win32crypt(): # pywin32 只装在宿主进程的环境里；MaaFW 内置 runner worker 跑在运行池的隔离 @@ -23,7 +39,12 @@ def dpapi_encrypt( return "" encrypted = _win32crypt().CryptProtectData( - note.encode("utf-8"), description, entropy, None, None, 0 + note.encode("utf-8"), + description, + entropy, + None, + None, + CRYPTPROTECT_LOCAL_MACHINE, ) return base64.b64encode(encrypted).decode("utf-8") diff --git a/app/utils/security.py b/app/utils/security.py index a7ed142cf..d1af34ffe 100644 --- a/app/utils/security.py +++ b/app/utils/security.py @@ -22,13 +22,18 @@ import re -from app.utils.platform.secret import dpapi_decrypt, dpapi_encrypt +from app.utils.platform.secret import ( + dpapi_decrypt, + dpapi_encrypt, + looks_like_dpapi_blob, +) __all__ = [ "sanitize_log_message", "format_exception_reason", "dpapi_encrypt", "dpapi_decrypt", + "looks_like_dpapi_blob", ] diff --git a/changelog.d/fix-dpapi-machine-scope-20260913.fix.md b/changelog.d/fix-dpapi-machine-scope-20260913.fix.md new file mode 100644 index 000000000..6f9824331 --- /dev/null +++ b/changelog.d/fix-dpapi-machine-scope-20260913.fix.md @@ -0,0 +1,2 @@ +project: scheduler +修复多 Windows 账户共用安装时密码被反复改写成乱码的问题 diff --git a/tests/models/test_config_item_secret.py b/tests/models/test_config_item_secret.py new file mode 100644 index 000000000..273c40db9 --- /dev/null +++ b/tests/models/test_config_item_secret.py @@ -0,0 +1,128 @@ +"""`EncryptValidator` 配置项对存量密文的处理口径。 + +一份安装的 `config/` 只有一份，同一台机器上换个 Windows 账户启动，读到的就是本账户 +解不开的密文。这类值必须原样保留：它既不是明文，也不是坏数据，只是钥匙不在手上。 + +回归的是「每次更新后账号密码变成一串密文」——换账户启动时 `load()` 把解不开的密文当 +成明文又加密一层，再顺着脏标记写回配置文件，原密码从此不可恢复。 +""" + +from __future__ import annotations + +import base64 + +import pytest + +from app.models.ConfigBase import ( + UNREADABLE_SECRET_PLACEHOLDER, + ConfigItem, + EncryptValidator, +) +from app.utils.platform import IS_WINDOWS + +pytestmark = pytest.mark.skipif(not IS_WINDOWS, reason="仅 Windows 走 DPAPI 实现") + + +@pytest.fixture +def foreign_ciphertext() -> str: + """构造一个结构合法、但本账户解不开的密文。 + + 取一份真实密文，把其中的 master key GUID 抹掉：从本账户看过去，另一个 + Windows 账户写的密文就是这个样子——头部合法，钥匙找不到。 + """ + + import win32crypt + + real = win32crypt.CryptProtectData( + "another-account-password".encode("utf-8"), None, None, None, None, 0 + ) + broken = real[:20] + bytes(16) + real[36:] + return base64.b64encode(broken).decode("utf-8") + + +@pytest.fixture +def password_item() -> ConfigItem: + return ConfigItem("Info", "Password", "", EncryptValidator()) + + +def _load(item: ConfigItem, disk_value: str) -> None: + """模拟 ConfigBase.load()：把配置文件里的原始值喂给配置项。""" + + item.setValue(disk_value) + + +def test_foreign_ciphertext_is_never_re_encrypted( + password_item: ConfigItem, foreign_ciphertext: str +) -> None: + _load(password_item, foreign_ciphertext) + + # 落盘口径必须与配置文件里的原值逐字相同，否则 load() 会判定脏数据并写回， + # 把另一个账户的密码永久覆盖掉。 + assert password_item.getValue(if_decrypt=False) == foreign_ciphertext + + +def test_foreign_ciphertext_reads_back_as_placeholder( + password_item: ConfigItem, foreign_ciphertext: str +) -> None: + _load(password_item, foreign_ciphertext) + + assert password_item.getValue(if_decrypt=True) == UNREADABLE_SECRET_PLACEHOLDER + + +def test_password_can_be_re_entered_over_foreign_ciphertext( + password_item: ConfigItem, foreign_ciphertext: str +) -> None: + """密文解不开时，用户至少还能重新填一遍，不能连覆盖都做不到。""" + + _load(password_item, foreign_ciphertext) + + password_item.setValue("brand-new-password") + + assert password_item.getValue(if_decrypt=True) == "brand-new-password" + + +def test_own_ciphertext_survives_load_unchanged(password_item: ConfigItem) -> None: + from app.utils.platform.windows.secret import dpapi_encrypt + + stored = dpapi_encrypt("my-real-password") + + _load(password_item, stored) + + assert password_item.getValue(if_decrypt=False) == stored + assert password_item.getValue(if_decrypt=True) == "my-real-password" + + +def test_legacy_user_scope_ciphertext_survives_load_unchanged( + password_item: ConfigItem, +) -> None: + """机器作用域上线前写下的密文，加载后不得被改写。""" + + import win32crypt + + legacy_blob = win32crypt.CryptProtectData( + "legacy-password".encode("utf-8"), None, None, None, None, 0 + ) + legacy = base64.b64encode(legacy_blob).decode("utf-8") + + _load(password_item, legacy) + + assert password_item.getValue(if_decrypt=False) == legacy + assert password_item.getValue(if_decrypt=True) == "legacy-password" + + +def test_plaintext_input_is_encrypted(password_item: ConfigItem) -> None: + from app.utils.platform.common.secret import looks_like_dpapi_blob + + password_item.setValue("typed-by-the-user") + + stored = password_item.getValue(if_decrypt=False) + assert looks_like_dpapi_blob(stored) is True + assert stored != "typed-by-the-user" + assert password_item.getValue(if_decrypt=True) == "typed-by-the-user" + + +def test_empty_value_stays_empty(password_item: ConfigItem) -> None: + password_item.setValue("") + + assert password_item.getValue(if_decrypt=False) == "" + assert password_item.getValue(if_decrypt=True) == "" diff --git a/tests/platform/test_secret_machine_scope.py b/tests/platform/test_secret_machine_scope.py new file mode 100644 index 000000000..1a034a7f8 --- /dev/null +++ b/tests/platform/test_secret_machine_scope.py @@ -0,0 +1,104 @@ +"""配置层密文的作用域与结构判别。 + +一份安装的 `config/` 只有一份，但同一台机器上可能有多个 Windows 账户轮流启动它。 +DPAPI 默认把密钥绑到当前账户，换账户就解不开已存的账号密码，因此加密固定走机器 +作用域；解密侧不需要对应改动，作用域写在密文里。 + +`looks_like_dpapi_blob` 是配置层区分「已加密的存量密文」与「用户新填的明文」的唯一 +依据，必须只看结构、不看本机能否解开。 +""" + +from __future__ import annotations + +import base64 + +import pytest + +from app.utils.platform import IS_WINDOWS +from app.utils.platform.common.secret import looks_like_dpapi_blob + +# 真实 DPAPI 密文固定的开头：4 字节版本号 + 16 字节 provider GUID。 +# 用户作用域与机器作用域的这 20 字节完全一致。 +_REAL_BLOB_HEADER = bytes.fromhex("01000000") + bytes.fromhex( + "d08c9ddf0115d1118c7a00c04fc297eb" +) + + +def _blob(payload: bytes = b"\x00" * 64) -> str: + return base64.b64encode(_REAL_BLOB_HEADER + payload).decode("utf-8") + + +def test_real_blob_header_is_recognized() -> None: + assert looks_like_dpapi_blob(_blob()) is True + + +@pytest.mark.parametrize( + "value", + [ + "", + "my-plain-password", + "不是 base64 的中文明文", + # 合法 base64，但开头不是 DPAPI 的 provider GUID + base64.b64encode(b"\x01\x00\x00\x00" + b"\xab" * 32).decode("utf-8"), + # 长度不足以容纳 20 字节头部 + base64.b64encode(b"\x01\x00\x00\x00").decode("utf-8"), + ], +) +def test_non_ciphertext_values_are_treated_as_plaintext(value: str) -> None: + assert looks_like_dpapi_blob(value) is False + + +@pytest.mark.parametrize("value", [None, 123, b"bytes", ["list"]]) +def test_non_string_values_are_rejected(value: object) -> None: + assert looks_like_dpapi_blob(value) is False + + +@pytest.mark.skipif(not IS_WINDOWS, reason="仅 Windows 走 DPAPI 实现") +def test_encrypt_uses_machine_scope() -> None: + """加密必须带机器作用域标志，否则换个 Windows 账户就读不出密码。""" + + from app.utils.platform.windows import secret as windows_secret + + captured: dict[str, object] = {} + + class _FakeWin32Crypt: + @staticmethod + def CryptProtectData(data, description, entropy, reserved, prompt, flags): + captured["flags"] = flags + return b"blob" + + original = windows_secret._win32crypt + windows_secret._win32crypt = lambda: _FakeWin32Crypt() + try: + windows_secret.dpapi_encrypt("secret") + finally: + windows_secret._win32crypt = original + + assert captured["flags"] == windows_secret.CRYPTPROTECT_LOCAL_MACHINE + + +@pytest.mark.skipif(not IS_WINDOWS, reason="仅 Windows 走 DPAPI 实现") +def test_machine_scope_ciphertext_round_trips() -> None: + from app.utils.platform.windows.secret import dpapi_decrypt, dpapi_encrypt + + ciphertext = dpapi_encrypt("账号密码 with ascii") + + assert looks_like_dpapi_blob(ciphertext) is True + assert dpapi_decrypt(ciphertext) == "账号密码 with ascii" + + +@pytest.mark.skipif(not IS_WINDOWS, reason="仅 Windows 走 DPAPI 实现") +def test_legacy_user_scope_ciphertext_still_decrypts() -> None: + """改用机器作用域后，此前写下的用户作用域密文必须照常读回。""" + + import win32crypt + + from app.utils.platform.windows.secret import dpapi_decrypt + + legacy_blob = win32crypt.CryptProtectData( + "legacy-password".encode("utf-8"), None, None, None, None, 0 + ) + legacy = base64.b64encode(legacy_blob).decode("utf-8") + + assert looks_like_dpapi_blob(legacy) is True + assert dpapi_decrypt(legacy) == "legacy-password"