安全验证功能

Author: lzy98276

app/Language/modules/safety_settings.py

@@ -38,19 +38,70 @@
             "name": "设置/修改密码",
             "description": "设置或修改安全验证密码",
         },
+        "password_rules": {
+            "name": "密码要求",
+            "description": "长度≥8，且至少包含字母、数字、特殊字符中的任意两类（推荐三类）",
+        },
+        "current_password": {"name": "当前密码"},
+        "new_password": {"name": "新密码"},
+        "confirm_password": {"name": "确认新密码"},
+        "password_strength_title": {"name": "密码强度"},
+        "strength_weak": {"name": "弱"},
+        "strength_medium": {"name": "中"},
+        "strength_strong": {"name": "强"},
+        "save_button": {"name": "保存"},
+        "cancel_button": {"name": "取消"},
+        "error_current_password": {"name": "当前密码不正确"},
+        "error_mismatch": {"name": "新密码与确认不一致"},
+        "error_strength_insufficient": {"name": "密码强度不足"},
+        "success_updated": {"name": "密码已更新"},
+        "remove_password": {"name": "移除密码", "description": "取消当前安全验证密码"},
+        "remove_password_confirm_title": {"name": "确认移除密码"},
+        "remove_password_confirm_content": {"name": "移除密码后将禁用安全开关，是否继续？"},
+        "remove_password_success": {"name": "已移除密码并关闭安全开关"},
+        "error_title": {"name": "错误"},
+        "dialog_yes_text": {"name": "来啦老弟"},
+        "dialog_cancel_text": {"name": "但是我拒绝"},
         "totp_switch": {
             "name": "TOTP验证",
             "description": "启用后可在安全操作中使用TOTP动态口令",
             "switchbutton_name": {"enable": "启用", "disable": "禁用"},
         },
         "set_totp": {"name": "设置TOTP", "description": "配置TOTP动态口令验证"},
+        "generate_totp_secret": {"name": "生成密钥"},
+        "verify_totp_code": {"name": "校验验证码"},
+        "totp_input_placeholder": {"name": "输入TOTP验证码进行校验"},
+        "totp_secret_prefix": {"name": "密钥"},
+        "totp_uri_prefix": {"name": "URI"},
+        "totp_generated_saved": {"name": "已生成并保存TOTP密钥"},
+        "totp_generated_error": {"name": "生成TOTP失败"},
+        "totp_code_valid": {"name": "验证码有效"},
+        "totp_code_invalid": {"name": "验证码无效"},
+        "totp_save_success": {"name": "设置已保存"},
+        "totp_verify_before_save": {"name": "请先校验验证码后再保存"},
+        "totp_qr_unavailable": {"name": "未能显示二维码，请安装二维码库"},
         "usb_switch": {
             "name": "U盘验证",
             "description": "启用后可在安全操作中使用U盘验证",
             "switchbutton_name": {"enable": "启用", "disable": "禁用"},
         },
         "bind_usb": {"name": "绑定U盘", "description": "绑定用于验证的U盘设备"},
         "unbind_usb": {"name": "解绑U盘", "description": "解除U盘设备绑定"},
+        "usb_refresh": {"name": "刷新"},
+        "usb_bind": {"name": "绑定"},
+        "usb_unbind_all": {"name": "解绑全部"},
+        "usb_no_removable": {"name": "未检测到可移动盘"},
+        "usb_bind_success": {"name": "已绑定 U盘"},
+        "usb_unbind_all_success": {"name": "已解绑全部 U盘"},
+        "error_set_password_first": {"name": "请先设置密码"},
+        "error_set_totp_first": {"name": "请先设置TOTP"},
+        "error_bind_usb_first": {"name": "请先绑定U盘"},
+        "usb_unbind_selected": {"name": "解绑选中"},
+        "usb_unbind_selected_success": {"name": "已解绑选中 U盘"},
+        "usb_select_bound_hint": {"name": "请选择一个已绑定设备"},
+        "usb_bound_devices": {"name": "已绑定设备"},
+        "usb_status_connected": {"name": "U盘已连接"},
+        "usb_status_disconnected": {"name": "U盘未连接"},
         "show_hide_floating_window_switch": {
             "name": "显示/隐藏浮窗验证",
             "description": "启用后显示或隐藏浮窗时需要安全验证",

app/common/safety/password.py

@@ -0,0 +1,68 @@
+import json
+import os
+import base64
+import hmac
+import hashlib
+
+from app.tools.path_utils import get_config_path, get_settings_path, ensure_dir, open_file, file_exists
+
+def _secrets_path():
+    return get_settings_path("secrets.json")
+
+def _legacy_path():
+    return get_config_path("security", "secrets.json")
+
+def _read():
+    p = _secrets_path()
+    if file_exists(p):
+        with open_file(p, "r", encoding="utf-8") as f:
+            return json.load(f)
+    lp = _legacy_path()
+    if file_exists(lp):
+        with open_file(lp, "r", encoding="utf-8") as f:
+            return json.load(f)
+    return {}
+
+def _write(d):
+    ensure_dir(_secrets_path().parent)
+    with open_file(_secrets_path(), "w", encoding="utf-8") as f:
+        json.dump(d, f, ensure_ascii=False, indent=4)
+
+def is_configured():
+    d = _read()
+    rec = d.get("password")
+    return isinstance(rec, dict) and bool(rec.get("hash")) and bool(rec.get("salt"))
+
+def set_password(plain: str):
+    salt = os.urandom(16)
+    iterations = 200000
+    dk = hashlib.pbkdf2_hmac("sha256", plain.encode("utf-8"), salt, iterations)
+    rec = {
+        "algorithm": "pbkdf2_sha256",
+        "iterations": iterations,
+        "salt": base64.b64encode(salt).decode("ascii"),
+        "hash": base64.b64encode(dk).decode("ascii"),
+    }
+    d = _read()
+    d["password"] = rec
+    _write(d)
+
+def verify_password(plain: str) -> bool:
+    d = _read()
+    rec = d.get("password")
+    if not rec:
+        return False
+    try:
+        salt = base64.b64decode(rec.get("salt", ""))
+        iterations = int(rec.get("iterations", 200000))
+        expected = base64.b64decode(rec.get("hash", ""))
+    except Exception:
+        return False
+    dk = hashlib.pbkdf2_hmac("sha256", plain.encode("utf-8"), salt, iterations)
+    return hmac.compare_digest(dk, expected)
+
+def clear_password():
+    d = _read()
+    if "password" in d:
+        del d["password"]
+    _write(d)

app/common/safety/totp.py

@@ -0,0 +1,57 @@
+import json
+import pyotp
+
+from app.tools.path_utils import get_config_path, get_settings_path, ensure_dir, open_file, file_exists
+
+def _secrets_path():
+    return get_settings_path("secrets.json")
+
+def _legacy_path():
+    return get_config_path("security", "secrets.json")
+
+def _read():
+    p = _secrets_path()
+    if file_exists(p):
+        with open_file(p, "r", encoding="utf-8") as f:
+            return json.load(f)
+    lp = _legacy_path()
+    if file_exists(lp):
+        with open_file(lp, "r", encoding="utf-8") as f:
+            return json.load(f)
+    return {}
+
+def _write(d):
+    ensure_dir(_secrets_path().parent)
+    with open_file(_secrets_path(), "w", encoding="utf-8") as f:
+        json.dump(d, f, ensure_ascii=False, indent=4)
+
+def is_configured():
+    d = _read()
+    rec = d.get("totp")
+    return isinstance(rec, dict) and bool(rec.get("secret"))
+
+def generate_secret():
+    return pyotp.random_base32()
+
+def set_totp(secret: str | None, issuer: str = "SecRandom", account: str = "user") -> str:
+    if not secret:
+        secret = generate_secret()
+    d = _read()
+    d["totp"] = {"secret": secret, "issuer": issuer, "account": account}
+    _write(d)
+    totp = pyotp.TOTP(secret)
+    return totp.provisioning_uri(name=account, issuer_name=issuer)
+
+def verify(code: str, window: int = 1) -> bool:
+    d = _read()
+    rec = d.get("totp")
+    if not rec:
+        return False
+    secret = rec.get("secret")
+    if not secret:
+        return False
+    totp = pyotp.TOTP(secret)
+    try:
+        return bool(totp.verify(code, valid_window=window))
+    except Exception:
+        return False

app/common/safety/usb.py

@@ -0,0 +1,139 @@
+import json
+import string
+import ctypes
+from pathlib import Path
+import os
+import platform
+
+from app.tools.path_utils import get_config_path, get_settings_path, ensure_dir, open_file, file_exists
+
+def _secrets_path():
+    return get_settings_path("secrets.json")
+
+def _legacy_path():
+    return get_config_path("security", "secrets.json")
+
+def _read():
+    p = _secrets_path()
+    if file_exists(p):
+        with open_file(p, "r", encoding="utf-8") as f:
+            return json.load(f)
+    lp = _legacy_path()
+    if file_exists(lp):
+        with open_file(lp, "r", encoding="utf-8") as f:
+            return json.load(f)
+    return {}
+
+def _write(d):
+    ensure_dir(_secrets_path().parent)
+    with open_file(_secrets_path(), "w", encoding="utf-8") as f:
+        json.dump(d, f, ensure_ascii=False, indent=4)
+
+def list_removable_drives():
+    if platform.system() == "Windows":
+        letters = []
+        GetDriveTypeW = ctypes.windll.kernel32.GetDriveTypeW
+        for ch in string.ascii_uppercase:
+            root = f"{ch}:\\"
+            p = Path(root)
+            if p.exists():
+                t = GetDriveTypeW(root)
+                if t == 2:
+                    letters.append(ch)
+        return letters
+    return []
+
+def get_volume_serial(letter: str) -> str:
+    buf_name = ctypes.create_unicode_buffer(256)
+    vol_serial = ctypes.c_uint()
+    max_comp_len = ctypes.c_uint()
+    fs_flags = ctypes.c_uint()
+    fs_name = ctypes.create_unicode_buffer(256)
+    ctypes.windll.kernel32.GetVolumeInformationW(
+        f"{letter}:\\",
+        buf_name,
+        ctypes.sizeof(buf_name),
+        ctypes.byref(vol_serial),
+        ctypes.byref(max_comp_len),
+        ctypes.byref(fs_flags),
+        fs_name,
+        ctypes.sizeof(fs_name),
+    )
+    return f"{vol_serial.value:08X}"
+
+def bind(volume_serial: str):
+    d = _read()
+    rec = d.get("usb") or {}
+    arr = rec.get("volume_serials") or []
+    if volume_serial not in arr:
+        arr.append(volume_serial)
+    rec["volume_serials"] = arr
+    d["usb"] = rec
+    _write(d)
+
+def unbind(volume_serial: str | None = None):
+    d = _read()
+    rec = d.get("usb") or {}
+    arr = rec.get("volume_serials") or []
+    if volume_serial is None:
+        arr = []
+    else:
+        arr = [s for s in arr if s != volume_serial]
+    rec["volume_serials"] = arr
+    d["usb"] = rec
+    _write(d)
+
+def is_bound_connected() -> bool:
+    d = _read()
+    rec = d.get("usb") or {}
+    arr = rec.get("volume_serials") or []
+    if not arr:
+        return False
+    if platform.system() == "Windows":
+        for lt in list_removable_drives():
+            try:
+                if get_volume_serial(lt) in arr:
+                    return True
+            except Exception:
+                pass
+        return False
+    ids = _linux_usb_ids()
+    for s in arr:
+        if s in ids:
+            return True
+    return False
+
+def get_bound_serials() -> list:
+    d = _read()
+    rec = d.get("usb") or {}
+    arr = rec.get("volume_serials") or []
+    return list(arr)
+
+def is_serial_connected(serial: str) -> bool:
+    if platform.system() == "Windows":
+        try:
+            for lt in list_removable_drives():
+                try:
+                    if get_volume_serial(lt) == serial:
+                        return True
+                except Exception:
+                    pass
+        except Exception:
+            pass
+        return False
+    try:
+        return serial in _linux_usb_ids()
+    except Exception:
+        return False
+
+def _linux_usb_ids() -> set:
+    ids = set()
+    base = "/dev/disk/by-id"
+    try:
+        if os.path.isdir(base):
+            for name in os.listdir(base):
+                if name.startswith("usb-"):
+                    ids.add(name)
+    except Exception:
+        pass
+    return ids

app/common/safety/verify_ops.py

@@ -0,0 +1,27 @@
+from app.tools.settings_access import readme_settings_async
+from app.common.safety.password import is_configured as password_is_configured
+from app.page_building.security_window import create_verify_password_window
+
+
+def should_require_password(op: str) -> bool:
+    if not password_is_configured():
+        return False
+    if not readme_settings_async("basic_safety_settings", "safety_switch"):
+        return False
+    key_map = {
+        "show_hide_floating_window": "show_hide_floating_window_switch",
+        "restart": "restart_switch",
+        "exit": "exit_switch",
+    }
+    k = key_map.get(op)
+    if not k:
+        return False
+    return bool(readme_settings_async("basic_safety_settings", k))
+
+
+def require_and_run(op: str, parent, func):
+    if not should_require_password(op):
+        func()
+        return
+    create_verify_password_window(on_verified=func)
+

app/config/security/secrets.json

@@ -0,0 +1,5 @@
+{
+    "usb": {
+        "volume_serials": []
+    }
+}

app/page_building/page_template.py

@@ -33,8 +33,7 @@ def __init__(self, content_widget_class=None, parent: QFrame = None):
         self.content_widget_class = content_widget_class
 
         self.__connectSignalToSlot()
-
-        QTimer.singleShot(0, self.create_ui_components)
+        self.create_ui_components()
 
     def __connectSignalToSlot(self):
         qconfig.themeChanged.connect(setTheme)
@@ -75,7 +74,7 @@ def create_ui_components(self):
         self.ui_created = True
 
         if self.content_widget_class:
-            QTimer.singleShot(0, self.create_content)
+            self.create_content()
 
     def create_content(self):
         """后台创建内容组件，避免堵塞进程"""