Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 43 additions & 18 deletions app/models/ConfigBase.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -296,6 +301,11 @@ def correct(self, value):
)


# 密文读不出来时对外给出的占位值。界面据此提示用户重新设置,
# 存量密文本身不会被它覆盖。
UNREADABLE_SECRET_PLACEHOLDER = "数据损坏, 请重新设置"


class EncryptValidator(ValidatorBase):
"""加密数据验证器"""

Expand All @@ -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):
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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

Expand Down
2 changes: 2 additions & 0 deletions app/utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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",
Expand Down
32 changes: 32 additions & 0 deletions app/utils/platform/common/secret.py
Original file line number Diff line number Diff line change
@@ -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:
"""当前平台是否提供配置层所需的密文存储能力。"""
Expand Down
23 changes: 22 additions & 1 deletion app/utils/platform/windows/secret.py
Original file line number Diff line number Diff line change
@@ -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 跑在运行池的隔离
Expand All @@ -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")

Expand Down
7 changes: 6 additions & 1 deletion app/utils/security.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
]


Expand Down
2 changes: 2 additions & 0 deletions changelog.d/fix-dpapi-machine-scope-20260913.fix.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
project: scheduler
修复多 Windows 账户共用安装时密码被反复改写成乱码的问题
128 changes: 128 additions & 0 deletions tests/models/test_config_item_secret.py
Original file line number Diff line number Diff line change
@@ -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) == ""
Loading
Loading