feat: 更稳定的密钥获取

- 不再维护偏移量/特征码 - 不再hook图片密钥
2026-06-18 15:54:08 +08:00 · 2026-04-12 03:49:48 +08:00
parent 1a67d37fa3
commit f598289939
4 changed files with 101 additions and 124 deletions
@@ -58,7 +58,7 @@
                  <svg v-else class="w-4 h-4 mr-2" fill="none" stroke="currentColor" viewBox="0 0 24 24">
                    <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M4 4v5h.582m15.356 2A8.001 8.001 0 004.582 9m0 0H9m11 11v-5h-.581m0 0a8.003 8.003 0 01-15.357-2m15.357 2H15" />
                  </svg>
-                  {{ isGettingDbKey ? '获取中...' : '一键获取全部密钥' }}
+                  {{ isGettingDbKey ? '获取中...' : '一键获取数据库密钥' }}
                </button>
              </div>
              <p v-if="formErrors.key" class="mt-1 text-sm text-red-600 flex items-center">
@@ -71,7 +71,7 @@
                <svg class="w-4 h-4 mr-1 text-[#10AEEF]" fill="none" stroke="currentColor" viewBox="0 0 24 24">
                  <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M13 16h-1v-4h-1m1-4h.01M21 12a9 9 0 11-18 0 9 9 0 0118 0z"/>
                </svg>
-                点击按钮将自动获取【数据库】与【图片】双重密钥。您也可以手动输入已知的64位密钥（使用<a href="https://github.com/ycccccccy/wx_key" target="_blank" class="text-[#07C160] hover:text-[#06AD56]">wx_key</a>等工具获取）。
+                点击按钮将自动获取【数据库解密密钥】。您也可以手动输入已知的64位密钥。
              </p>
            </div>
            
@@ -189,7 +189,7 @@
                <svg class="w-4 h-4 mr-1 text-[#10AEEF]" fill="none" stroke="currentColor" viewBox="0 0 24 24">
                  <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M13 16h-1v-4h-1m1-4h.01M21 12a9 9 0 11-18 0 9 9 0 0118 0z"/>
                </svg>
-                如果您在第一步使用了“一键获取”或触发了云端解析，下方输入框已被自动填充。您也可可以使用<a href="https://github.com/ycccccccy/wx_key" target="_blank" class="text-[#07C160] hover:text-[#06AD56]">wx_key</a>等工具手动获取。
+                系统已为您尝试通过【本地算法】或【云端解析】自动获取图片密钥。如果输入框为空，请手动填写。
              </p>

              <div class="grid grid-cols-1 md:grid-cols-2 gap-4">
@@ -547,14 +547,7 @@ const handleGetDbKey = async () => {
      if (res.data?.db_key) {
        formData.key = res.data.db_key
      }
-      // 直接把图片密钥也存好
-      if (res.data?.xor_key) {
-        manualKeys.xor_key = res.data.xor_key
-      }
-      if (res.data?.aes_key) {
-        manualKeys.aes_key = res.data.aes_key
-      }
-      warning.value = '🎉 数据库与图片密钥均已获取成功！'
+      warning.value = '🎉 数据库解密密钥已获取成功！'
      // 3秒后清除成功提示，保持 UI 干净
      setTimeout(() => { if(warning.value.includes('获取成功')) warning.value = '' }, 3000)
    } else {
@@ -32,81 +32,11 @@ logger = logging.getLogger(__name__)

 # ======================  以下是hook逻辑  ======================================

-@dataclass
-class HookConfig:
-    min_version: str
-    pattern: str
-    mask: str
-    offset: int
-    md5_pattern: str = ""
-    md5_mask: str = ""
-    md5_offset: int = 0
-
 class WeChatKeyFetcher:
    def __init__(self):
        self.process_name = "Weixin.exe"
        self.timeout_seconds = 60

-    @staticmethod
-    def _hex_array_to_str(hex_array: List[int]) -> str:
-        return " ".join([f"{b:02X}" for b in hex_array])
-
-    def _get_hook_config(self, version_str: str) -> Optional[HookConfig]:
-        try:
-            v_curr = pkg_version.parse(version_str)
-        except Exception as e:
-            logger.error(f"版本号解析失败: {version_str} || {e}")
-            return None
-
-
-        if v_curr > pkg_version.parse("4.1.6.14"):
-            return HookConfig(
-                min_version=">4.1.6.14",
-                pattern=self._hex_array_to_str([
-                    0x24, 0x50, 0x48, 0xC7, 0x45, 0x00, 0xFE, 0xFF, 0xFF, 0xFF,
-                    0x44, 0x89, 0xCF, 0x44, 0x89, 0xC3, 0x49, 0x89, 0xD6, 0x48,
-                    0x89, 0xCE, 0x48, 0x89
-                ]),
-                mask="xxxxxxxxxxxxxxxxxxxxxxxx",
-                offset=-3,
-                md5_pattern="48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9",
-                md5_mask="xxx?xxxxxxxxxxx?xxxxxxx",
-                md5_offset=4
-            )
-
-        if pkg_version.parse("4.1.4") <= v_curr <= pkg_version.parse("4.1.6.14"):
-            return HookConfig(
-                min_version="4.1.4-4.1.6.14",
-                pattern=self._hex_array_to_str([
-                    0x24, 0x08, 0x48, 0x89, 0x6c, 0x24, 0x10, 0x48, 0x89, 0x74,
-                    0x00, 0x18, 0x48, 0x89, 0x7c, 0x00, 0x20, 0x41, 0x56, 0x48,
-                    0x83, 0xec, 0x50, 0x41
-                ]),
-                mask="xxxxxxxxxx?xxxx?xxxxxxxx",
-                offset=-3,
-                md5_pattern="48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9",
-                md5_mask="xxx?xxxxxxxxxxx?xxxxxxx",
-                md5_offset=4
-            )
-
-        if v_curr < pkg_version.parse("4.1.4"):
-            """图片密钥可能是错的，版本过低没有测试"""
-            return HookConfig(
-                min_version="<4.1.4",
-                pattern=self._hex_array_to_str([
-                    0x24, 0x50, 0x48, 0xc7, 0x45, 0x00, 0xfe, 0xff, 0xff, 0xff,
-                    0x44, 0x89, 0xcf, 0x44, 0x89, 0xc3, 0x49, 0x89, 0xd6, 0x48,
-                    0x89, 0xce, 0x48, 0x89
-                ]),
-                mask="xxxxxxxxxxxxxxxxxxxxxxxx",
-                offset=-15,  # -0xf
-                md5_pattern="48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9",
-                md5_mask="xxx?xxxxxxxxxxx?xxxxxxx",
-                md5_offset=4
-            )
-
-        return None
-
    def kill_wechat(self):
        """检测并查杀微信进程"""
        killed = False
@@ -125,9 +55,7 @@ class WeChatKeyFetcher:
    def launch_wechat(self, exe_path: str) -> int:
        """启动微信并返回 PID"""
        try:
-
            process = subprocess.Popen(exe_path)
-
            time.sleep(2)
            candidates = []
            for proc in psutil.process_iter(['pid', 'name', 'create_time']):
@@ -135,7 +63,6 @@ class WeChatKeyFetcher:
                    candidates.append(proc)

            if candidates:
-
                candidates.sort(key=lambda x: x.info['create_time'], reverse=True)
                target_pid = candidates[0].info['pid']
                return target_pid
@@ -146,8 +73,8 @@ class WeChatKeyFetcher:
            logger.error(f"启动微信失败: {e}")
            raise RuntimeError(f"无法启动微信: {e}")

-    def fetch_key(self) -> dict:
-        """调用 wx_key 获取双密钥"""
+    def fetch_db_key(self) -> dict:
+        """调用 wx_key 仅获取数据库密钥 (Hook 模式)"""
        if wx_key is None:
            raise RuntimeError("wx_key 模块未安装或加载失败")

@@ -160,36 +87,26 @@ class WeChatKeyFetcher:

        logger.info(f"Detect WeChat: {version} at {exe_path}")

-        config = self._get_hook_config(version)
-        if not config:
-            raise RuntimeError(f"原生获取失败：当前微信版本 ({version}) 过低，为保证稳定性，仅支持 4.1.5 及以上版本使用原生获取。")
-
        self.kill_wechat()
        pid = self.launch_wechat(exe_path)
        logger.info(f"WeChat launched, PID: {pid}")

-        if not wx_key.initialize_hook(pid, "", config.pattern, config.mask, config.offset,
-                                      config.md5_pattern, config.md5_mask, config.md5_offset):
+        # 仅传入 PID，触发数据库密钥自动 Hook
+        if not wx_key.initialize_hook(pid):
            err = wx_key.get_last_error_msg()
-            raise RuntimeError(f"Hook初始化失败: {err}")
+            raise RuntimeError(f"数据库 Hook 初始化失败: {err}")

        start_time = time.time()
        found_db_key = None
-        found_md5_data = None

        try:
            while True:
                if time.time() - start_time > self.timeout_seconds:
-                    raise TimeoutError("获取密钥超时 (60s)，请确保在弹出的微信中完成登录。")
+                    raise TimeoutError("获取数据库密钥超时 (60s)，请确保在弹出的微信中完成登录。")

                key_data = wx_key.poll_key_data()
-                if key_data:
-                    if 'key' in key_data:
-                        found_db_key = key_data['key']
-                    if 'md5' in key_data:
-                        found_md5_data = key_data['md5']
-
-                if found_db_key and found_md5_data:
+                if key_data and 'key' in key_data:
+                    found_db_key = key_data['key']
                    break

                while True:
@@ -204,22 +121,13 @@ class WeChatKeyFetcher:
            logger.info("Cleaning up hook...")
            wx_key.cleanup_hook()

-        aes_key = None  # gemini !!! ???
-        xor_key = None
-
-        if found_md5_data and "|" in found_md5_data:
-            aes_key, xor_key_dec = found_md5_data.split("|")
-            xor_key = f"0x{int(xor_key_dec):02X}"
-
        return {
-            "db_key": found_db_key,
-            "aes_key": aes_key,
-            "xor_key": xor_key
+            "db_key": found_db_key
        }

 def get_db_key_workflow():
    fetcher = WeChatKeyFetcher()
-    return fetcher.fetch_key()
+    return fetcher.fetch_db_key()


 # ==============================   以下是图片密钥逻辑  =====================================
@@ -232,6 +140,82 @@ def get_wechat_internal_global_config(wx_dir: Path, file_name1) -> bytes:
    return Path(target_path).read_bytes()


+def try_get_local_image_keys() -> List[Dict[str, Any]]:
+    """尝试通过本地算法提取图片密钥 (无需 Hook)"""
+    if wx_key is None or not hasattr(wx_key, 'get_image_key'):
+        return []
+    
+    try:
+        res_json = wx_key.get_image_key()
+        if not res_json:
+            return []
+        
+        data = json.loads(res_json)
+        accounts = data.get('accounts', [])
+        results = []
+        for acc in accounts:
+            wxid = acc.get('wxid')
+            keys = acc.get('keys', [])
+            for k in keys:
+                xor_key = k.get('xorKey')
+                aes_key = k.get('aesKey')
+                if xor_key is not None:
+                    results.append({
+                        "wxid": wxid,
+                        "xor_key": f"0x{int(xor_key):02X}",
+                        "aes_key": aes_key
+                    })
+        return results
+    except Exception as e:
+        logger.error(f"本地提取图片密钥失败: {e}")
+        return []
+
+
+async def get_image_key_integrated_workflow(account: Optional[str] = None) -> Dict[str, Any]:
+    """
+    集成图片密钥获取流程：
+    1. 优先尝试本地算法提取
+    2. 如果本地提取失败或未匹配到指定账号，尝试远程 API 解析
+    """
+    # 1. 尝试本地提取
+    local_keys = try_get_local_image_keys()
+    
+    target_account_wxid = None
+    if account:
+        try:
+            account_dir = _resolve_account_dir(account)
+            wx_id_dir = _resolve_account_wxid_dir(account_dir)
+            target_account_wxid = wx_id_dir.name
+        except:
+            target_account_wxid = account
+
+    if local_keys:
+        # 如果指定了账号，尝试在本地结果中找匹配的
+        if target_account_wxid:
+            for k in local_keys:
+                if k['wxid'] == target_account_wxid:
+                    logger.info(f"成功通过本地算法匹配到账号 {target_account_wxid} 的图片密钥")
+                    upsert_account_keys_in_store(
+                        account=k['wxid'],
+                        image_xor_key=k['xor_key'],
+                        image_aes_key=k['aes_key']
+                    )
+                    return k
+        else:
+            # 如果没指定账号，返回第一个发现的并存入 store (如果有的话)
+            k = local_keys[0]
+            logger.info(f"本地算法提取成功 (未指定账号，返回首个): {k['wxid']}")
+            upsert_account_keys_in_store(
+                account=k['wxid'],
+                image_xor_key=k['xor_key'],
+                image_aes_key=k['aes_key']
+            )
+            return k
+
+    # 2. 本地提取失败或不匹配，尝试远程解析
+    logger.info("本地算法提取未命中，尝试远程 API 解析...")
+    return await fetch_and_save_remote_keys(account)
+

 async def fetch_and_save_remote_keys(account: Optional[str] = None) -> Dict[str, Any]:
    account_dir = _resolve_account_dir(account)
@@ -3,7 +3,7 @@ from typing import Optional
 from fastapi import APIRouter

 from ..key_store import get_account_keys_from_store
-from ..key_service import get_db_key_workflow, fetch_and_save_remote_keys
+from ..key_service import get_db_key_workflow, get_image_key_integrated_workflow
 from ..media_helpers import _load_media_keys, _resolve_account_dir
 from ..path_fix import PathFixRoute

@@ -97,7 +97,7 @@ async def get_image_key(account: Optional[str] = None):
    4. 解析返回流，自动存入本地数据库
    """
    try:
-        result = await fetch_and_save_remote_keys(account)
+        result = await get_image_key_integrated_workflow(account)

        return {
            "status": 0,
@@ -105,8 +105,8 @@ async def get_image_key(account: Optional[str] = None):
            "data": {
                "xor_key": result["xor_key"],
                "aes_key": result["aes_key"],
-                "nick_name": result.get("nick_name"),
-                "account": result["wxid"]
+                "nick_name": result.get("nick_name", ""),
+                "account": result.get("wxid", "")
            }
        }
    except FileNotFoundError as e:
@@ -1,5 +1,5 @@
 version = 1
-revision = 2
+revision = 3
 requires-python = ">=3.11"

 [[package]]
@@ -919,7 +919,7 @@ requires-dist = [
    { name = "requests", specifier = ">=2.32.4" },
    { name = "typing-extensions", specifier = ">=4.8.0" },
    { name = "uvicorn", extras = ["standard"], specifier = ">=0.24.0" },
-    { name = "wx-key", specifier = ">=1.1.0" },
+    { name = "wx-key", specifier = ">=2.0.0" },
    { name = "zstandard", specifier = ">=0.23.0" },
 ]
 provides-extras = ["build"]
@@ -935,13 +935,13 @@ wheels = [

 [[package]]
 name = "wx-key"
-version = "1.1.0"
+version = "2.0.0"
 source = { registry = "tools/key_wheels" }
 wheels = [
-    { path = "wx_key-1.1.0-cp311-cp311-win_amd64.whl" },
-    { path = "wx_key-1.1.0-cp312-cp312-win_amd64.whl" },
-    { path = "wx_key-1.1.0-cp313-cp313-win_amd64.whl" },
-    { path = "wx_key-1.1.0-cp314-cp314-win_amd64.whl" },
+    { path = "wx_key-2.0.0-cp311-cp311-win_amd64.whl" },
+    { path = "wx_key-2.0.0-cp312-cp312-win_amd64.whl" },
+    { path = "wx_key-2.0.0-cp313-cp313-win_amd64.whl" },
+    { path = "wx_key-2.0.0-cp314-cp314-win_amd64.whl" },
 ]

 [[package]]