feat: add new router for db key

src/wechat_decrypt_tool/key_service.py

@@ -0,0 +1,187 @@
+# import sys
+
+try:
+    import wx_key
+except ImportError:
+    print('[!] 环境中未安装wx_key依赖，可能无法自动获取数据库密钥')
+    wx_key = None
+    # sys.exit(1)
+
+import time
+import psutil
+import subprocess
+import logging
+from typing import Optional, List
+from dataclasses import dataclass
+from packaging import version as pkg_version  # 建议使用 packaging 库处理版本比较
+from wechat_detection import detect_wechat_installation
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class HookConfig:
+    min_version: str
+    pattern: str  # 用 00 不要用 ?  !!!!  否则C++内存会炸
+    mask: str
+    offset: int
+
+
+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]:
+        """搬运自wx_key代码，未来用ida脚本直接获取即可"""
+        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
+            )
+
+        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
+            )
+
+        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
+            )
+
+        return None
+
+    def kill_wechat(self):
+        """检测并查杀微信进程"""
+        killed = False
+        for proc in psutil.process_iter(['pid', 'name']):
+            try:
+                if proc.info['name'] == self.process_name:
+                    logger.info(f"Killing WeChat process: {proc.info['pid']}")
+                    proc.terminate()
+                    killed = True
+            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+                pass
+
+        if killed:
+            time.sleep(1)  # 等待完全退出
+
+    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']):
+                if proc.info['name'] == self.process_name:
+                    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
+
+            return process.pid
+
+        except Exception as e:
+            logger.error(f"启动微信失败: {e}")
+            raise RuntimeError(f"无法启动微信: {e}")
+
+    def fetch_key(self) -> str:
+        """没有wx_key模块无法自动获取密钥"""
+        if wx_key is None:
+            raise RuntimeError("wx_key 模块未安装或加载失败")
+
+        install_info = detect_wechat_installation()
+
+        exe_path = install_info.get('wechat_exe_path')
+        version = install_info.get('wechat_version')
+
+        if not exe_path or not version:
+            raise RuntimeError("无法自动定位微信安装路径或版本")
+
+        logger.info(f"Detect WeChat: {version} at {exe_path}")
+
+        config = self._get_hook_config(version)
+        if not config:
+            raise RuntimeError(f"不支持的微信版本: {version}")
+
+        self.kill_wechat()
+
+        pid = self.launch_wechat(exe_path)
+        logger.info(f"WeChat launched, PID: {pid}")
+
+        logger.info(f"Initializing Hook with pattern: {config.pattern[:20]}... Offset: {config.offset}")
+
+        if not wx_key.initialize_hook(pid, "", config.pattern, config.mask, config.offset):
+            err = wx_key.get_last_error_msg()
+            raise RuntimeError(f"Hook初始化失败: {err}")
+
+        start_time = time.time()
+
+
+        try:
+            while True:
+                if time.time() - start_time > self.timeout_seconds:
+                    raise TimeoutError("获取密钥超时 (60s)")
+
+                key = wx_key.poll_key_data()
+                if key:
+                    found_key = key
+                    break
+
+                while True:
+                    msg, level = wx_key.get_status_message()
+                    if msg is None:
+                        break
+                    if level == 2:
+                        logger.error(f"[Hook Error] {msg}")
+
+                time.sleep(0.1)
+
+        finally:
+            logger.info("Cleaning up hook...")
+            wx_key.cleanup_hook()
+
+        if found_key:
+            return found_key
+        else:
+            raise RuntimeError("未知错误，未获取到密钥")
+
+def get_db_key_workflow():
+    fetcher = WeChatKeyFetcher()
+    return fetcher.fetch_key()

src/wechat_decrypt_tool/routers/keys.py

@@ -51,3 +51,37 @@ async def get_saved_keys(account: Optional[str] = None):
         "keys": result,
     }
 
+
+@router.get("/api/get_db_key", summary="自动获取微信数据库密钥")
+async def get_wechat_db_key():
+    """
+    自动流程：
+    1. 结束微信进程
+    2. 启动微信
+    3. 根据版本注入 Hook
+    4. 抓取密钥并返回
+    """
+    try:
+        # 不需要async吧，我相信fastapi的线程池
+        db_key = get_db_key_workflow()
+
+        return {
+            "status": 0,
+            "errmsg": "ok",
+            "data": {
+                "db_key": db_key
+            }
+        }
+
+    except TimeoutError:
+        return {
+            "status": -1,
+            "errmsg": "获取超时，请确保微信没有开启自动登录 或者 加快手速",
+            "data": {}
+        }
+    except Exception as e:
+        return {
+            "status": -1,
+            "errmsg": f"获取失败: {str(e)}",
+            "data": {}
+        }