improvement(media): 移除进程提取密钥并优化媒体解密

- 移除 pymem/yara-python 依赖，明确仅使用 wx_key 获取密钥

- 删除 media_key_finder.py，简化媒体密钥与资源定位逻辑

- 更新媒体接口/脚本与导出说明，避免误导进程提取能力

src/wechat_decrypt_tool/chat_export_service.py

@@ -1461,7 +1461,7 @@ def _attach_offline_media(
     lock: threading.Lock,
     job: ExportJob,
 ) -> None:
-    # allow_process_key_extract is reserved for future: try to decrypt missing media using process memory keys
+    # allow_process_key_extract is reserved; this project does not extract keys from process (use wx_key instead).
     _ = allow_process_key_extract
 
     rt = str(msg.get("renderType") or "")

src/wechat_decrypt_tool/media_helpers.py

@@ -8,13 +8,9 @@ import os
 import re
 import sqlite3
 import struct
-import threading
-from collections import Counter
-from concurrent.futures import ThreadPoolExecutor
 from functools import lru_cache
 from pathlib import Path
 from typing import Any, Optional
-from ctypes import wintypes
 
 from fastapi import HTTPException
 
@@ -22,11 +18,6 @@ from .logging_config import get_logger
 
 logger = get_logger(__name__)
 
-try:
-    import psutil  # type: ignore
-except Exception:
-    psutil = None
-
 
 # 仓库根目录（用于定位 output/databases）
 _REPO_ROOT = Path(__file__).resolve().parents[2]
@@ -1031,376 +1022,6 @@ def _detect_wechat_dat_version(data: bytes) -> int:
         return 2
     return 0
 
-
-def _extract_yyyymm_for_sort(p: Path) -> str:
-    m = re.search(r"(\d{4}-\d{2})", str(p))
-    return m.group(1) if m else "0000-00"
-
-
-@lru_cache(maxsize=16)
-def _get_wechat_template_most_common_last2(weixin_root_str: str) -> Optional[bytes]:
-    try:
-        root = Path(weixin_root_str)
-        if not root.exists() or not root.is_dir():
-            return None
-    except Exception:
-        return None
-
-    try:
-        template_files = list(root.rglob("*_t.dat"))
-    except Exception:
-        template_files = []
-
-    if not template_files:
-        return None
-
-    template_files.sort(key=_extract_yyyymm_for_sort, reverse=True)
-    last_bytes_list: list[bytes] = []
-    for file in template_files[:16]:
-        try:
-            with open(file, "rb") as f:
-                f.seek(-2, 2)
-                b2 = f.read(2)
-                if b2 and len(b2) == 2:
-                    last_bytes_list.append(b2)
-        except Exception:
-            continue
-
-    if not last_bytes_list:
-        return None
-    return Counter(last_bytes_list).most_common(1)[0][0]
-
-
-@lru_cache(maxsize=16)
-def _find_wechat_xor_key(weixin_root_str: str) -> Optional[int]:
-    try:
-        root = Path(weixin_root_str)
-        if not root.exists() or not root.is_dir():
-            return None
-    except Exception:
-        return None
-
-    most_common = _get_wechat_template_most_common_last2(weixin_root_str)
-    if not most_common or len(most_common) != 2:
-        return None
-    x, y = most_common[0], most_common[1]
-    xor_key = x ^ 0xFF
-    if xor_key != (y ^ 0xD9):
-        return None
-    return xor_key
-
-
-def _get_wechat_v2_ciphertext(weixin_root: Path, most_common_last2: bytes) -> Optional[bytes]:
-    try:
-        template_files = list(weixin_root.rglob("*_t.dat"))
-    except Exception:
-        return None
-    if not template_files:
-        return None
-
-    template_files.sort(key=_extract_yyyymm_for_sort, reverse=True)
-    sig = b"\x07\x08V2\x08\x07"
-
-    def try_read_ct(file: Path, require_last2: bool) -> Optional[bytes]:
-        try:
-            with open(file, "rb") as f:
-                if f.read(6) != sig:
-                    return None
-                if require_last2 and most_common_last2 and len(most_common_last2) == 2:
-                    try:
-                        f.seek(-2, 2)
-                        if f.read(2) != most_common_last2:
-                            return None
-                    except Exception:
-                        return None
-                f.seek(0xF)
-                ct = f.read(16)
-                if ct and len(ct) == 16:
-                    return ct
-        except Exception:
-            return None
-        return None
-
-    # Prefer matching last2 bytes (older heuristic), but fall back to any V2 template like wx_key.
-    if most_common_last2 and len(most_common_last2) == 2:
-        for file in template_files:
-            ct = try_read_ct(file, require_last2=True)
-            if ct:
-                return ct
-
-    for file in template_files:
-        ct = try_read_ct(file, require_last2=False)
-        if ct:
-            return ct
-    return None
-
-
-def _verify_wechat_aes_key(ciphertext: bytes, key16: bytes) -> bool:
-    try:
-        from Crypto.Cipher import AES
-
-        cipher = AES.new(key16[:16], AES.MODE_ECB)
-        plain = cipher.decrypt(ciphertext)
-        if plain.startswith(b"\xff\xd8\xff"):
-            return True
-        if plain.startswith(b"\x89PNG\r\n\x1a\n"):
-            return True
-        if plain.startswith(b"GIF87a") or plain.startswith(b"GIF89a"):
-            return True
-        if plain.startswith(b"wxgf"):
-            return True
-        if len(plain) >= 12 and plain.startswith(b"RIFF") and plain[8:12] == b"WEBP":
-            return True
-        if len(plain) >= 8 and plain[4:8] == b"ftyp":
-            return True
-        return False
-    except Exception:
-        return False
-
-
-class _MEMORY_BASIC_INFORMATION(ctypes.Structure):
-    _fields_ = [
-        ("BaseAddress", ctypes.c_void_p),
-        ("AllocationBase", ctypes.c_void_p),
-        ("AllocationProtect", ctypes.c_ulong),
-        ("RegionSize", ctypes.c_size_t),
-        ("State", ctypes.c_ulong),
-        ("Protect", ctypes.c_ulong),
-        ("Type", ctypes.c_ulong),
-    ]
-
-
-def _find_weixin_pids() -> list[int]:
-    if psutil is None:
-        return []
-
-    preferred = ["weixin.exe", "wechat.exe", "wechatappex.exe", "wechatapp.exe"]
-    preferred_set = set(preferred)
-    pids_by_name: dict[str, list[int]] = {n: [] for n in preferred}
-    extra: list[int] = []
-
-    for p in psutil.process_iter(["pid", "name"]):
-        try:
-            name = (p.info.get("name") or "").lower()
-            pid = int(p.info.get("pid") or 0)
-        except Exception:
-            continue
-        if pid <= 0:
-            continue
-
-        if name in preferred_set:
-            pids_by_name[name].append(pid)
-            continue
-
-        if name.startswith("wechat") or name.startswith("weixin"):
-            extra.append(pid)
-
-    ordered: list[int] = []
-    for n in preferred:
-        ordered.extend(pids_by_name.get(n, []))
-    ordered.extend(extra)
-
-    seen: set[int] = set()
-    out: list[int] = []
-    for pid in ordered:
-        if pid in seen:
-            continue
-        seen.add(pid)
-        out.append(pid)
-    return out
-
-
-def _try_enable_windows_debug_privilege() -> None:
-    if os.name != "nt":
-        return
-
-    try:
-        advapi32 = ctypes.windll.advapi32
-        kernel32 = ctypes.windll.kernel32
-
-        TOKEN_ADJUST_PRIVILEGES = 0x0020
-        TOKEN_QUERY = 0x0008
-        SE_PRIVILEGE_ENABLED = 0x0002
-
-        class _LUID(ctypes.Structure):
-            _fields_ = [("LowPart", wintypes.DWORD), ("HighPart", wintypes.LONG)]
-
-        class _LUID_AND_ATTRIBUTES(ctypes.Structure):
-            _fields_ = [("Luid", _LUID), ("Attributes", wintypes.DWORD)]
-
-        class _TOKEN_PRIVILEGES(ctypes.Structure):
-            _fields_ = [("PrivilegeCount", wintypes.DWORD), ("Privileges", _LUID_AND_ATTRIBUTES * 1)]
-
-        token = wintypes.HANDLE()
-        if not advapi32.OpenProcessToken(
-            kernel32.GetCurrentProcess(),
-            TOKEN_ADJUST_PRIVILEGES | TOKEN_QUERY,
-            ctypes.byref(token),
-        ):
-            return
-
-        try:
-            luid = _LUID()
-            if not advapi32.LookupPrivilegeValueW(None, "SeDebugPrivilege", ctypes.byref(luid)):
-                return
-
-            tp = _TOKEN_PRIVILEGES()
-            tp.PrivilegeCount = 1
-            tp.Privileges[0].Luid = luid
-            tp.Privileges[0].Attributes = SE_PRIVILEGE_ENABLED
-            advapi32.AdjustTokenPrivileges(token, False, ctypes.byref(tp), 0, None, None)
-        finally:
-            kernel32.CloseHandle(token)
-    except Exception:
-        return
-
-
-def _extract_wechat_aes_key_from_process(ciphertext: bytes) -> Optional[bytes]:
-    _try_enable_windows_debug_privilege()
-    pids = _find_weixin_pids()
-    if not pids:
-        return None
-
-    PROCESS_VM_READ = 0x0010
-    PROCESS_QUERY_INFORMATION = 0x0400
-    MEM_COMMIT = 0x1000
-    MEM_PRIVATE = 0x20000
-
-    PAGE_NOACCESS = 0x01
-    PAGE_READONLY = 0x02
-    PAGE_READWRITE = 0x04
-    PAGE_WRITECOPY = 0x08
-    PAGE_EXECUTE_READ = 0x20
-    PAGE_EXECUTE_READWRITE = 0x40
-    PAGE_EXECUTE_WRITECOPY = 0x80
-    PAGE_GUARD = 0x100
-
-    kernel32 = ctypes.windll.kernel32
-
-    OpenProcess = kernel32.OpenProcess
-    OpenProcess.argtypes = [ctypes.c_ulong, ctypes.c_bool, ctypes.c_ulong]
-    OpenProcess.restype = ctypes.c_void_p
-
-    ReadProcessMemory = kernel32.ReadProcessMemory
-    ReadProcessMemory.argtypes = [
-        ctypes.c_void_p,
-        ctypes.c_void_p,
-        ctypes.c_void_p,
-        ctypes.c_size_t,
-        ctypes.POINTER(ctypes.c_size_t),
-    ]
-    ReadProcessMemory.restype = ctypes.c_bool
-
-    VirtualQueryEx = kernel32.VirtualQueryEx
-    VirtualQueryEx.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t]
-    VirtualQueryEx.restype = ctypes.c_size_t
-
-    CloseHandle = kernel32.CloseHandle
-    CloseHandle.argtypes = [ctypes.c_void_p]
-    CloseHandle.restype = ctypes.c_bool
-
-    readable_mask = (
-        PAGE_READONLY
-        | PAGE_READWRITE
-        | PAGE_WRITECOPY
-        | PAGE_EXECUTE_READ
-        | PAGE_EXECUTE_READWRITE
-        | PAGE_EXECUTE_WRITECOPY
-    )
-
-    def is_readable(protect: int) -> bool:
-        if protect & PAGE_GUARD:
-            return False
-        if protect & PAGE_NOACCESS:
-            return False
-        return bool(protect & readable_mask)
-
-    # Keep pattern consistent with wx_key: search for 16/32 lower/upper alpha-num strings with word-boundary-like guards.
-    # (Using 32 first reduces false positives in some builds.)
-    pattern = re.compile(rb"(?i)(?<![0-9a-z])([0-9a-z]{32}|[0-9a-z]{16})(?![0-9a-z])")
-
-    def scan_pid(pid: int) -> Optional[bytes]:
-        handle = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
-        if not handle:
-            return None
-
-        stop = threading.Event()
-        result: list[Optional[bytes]] = [None]
-
-        def read_mem(addr: int, size: int) -> Optional[bytes]:
-            buf = ctypes.create_string_buffer(size)
-            read = ctypes.c_size_t(0)
-            ok = ReadProcessMemory(handle, ctypes.c_void_p(addr), buf, size, ctypes.byref(read))
-            if not ok or read.value <= 0:
-                return None
-            return buf.raw[: read.value]
-
-        def scan_region(base: int, region_size: int) -> Optional[bytes]:
-            chunk = 4 * 1024 * 1024
-            offset = 0
-            tail = b""
-            while offset < region_size and not stop.is_set():
-                to_read = min(chunk, region_size - offset)
-                b = read_mem(base + offset, int(to_read))
-                if not b:
-                    # Don't abort the whole region on a single read failure (wx_key keeps scanning).
-                    offset += to_read
-                    tail = b""
-                    continue
-                data = tail + b
-                for m in pattern.finditer(data):
-                    cand = m.group(1)
-                    if len(cand) == 32:
-                        # wx_key uses key[:16] to validate; keep that but also try the second half for compatibility.
-                        candidates = [cand[:16], cand[16:]]
-                    else:
-                        candidates = [cand]
-                    for cand16 in candidates:
-                        if _verify_wechat_aes_key(ciphertext, cand16):
-                            return cand16
-                tail = data[-64:] if len(data) > 64 else data
-                offset += to_read
-            return None
-
-        regions: list[tuple[int, int]] = []
-        mbi = _MEMORY_BASIC_INFORMATION()
-        addr = 0
-        try:
-            while VirtualQueryEx(handle, ctypes.c_void_p(addr), ctypes.byref(mbi), ctypes.sizeof(mbi)):
-                try:
-                    if int(mbi.State) == MEM_COMMIT and int(mbi.Type) == MEM_PRIVATE:
-                        protect = int(mbi.Protect)
-                        if is_readable(protect):
-                            base = int(mbi.BaseAddress)
-                            size = int(mbi.RegionSize)
-                            if size > 0:
-                                # Skip extremely large regions to keep runtime bounded (same idea as wx_key).
-                                if size <= 100 * 1024 * 1024:
-                                    regions.append((base, size))
-                    addr = int(mbi.BaseAddress) + int(mbi.RegionSize)
-                except Exception:
-                    addr += 0x1000
-                if addr <= 0:
-                    break
-
-            with ThreadPoolExecutor(max_workers=min(32, max(1, len(regions)))) as ex:
-                for found in ex.map(lambda r: scan_region(r[0], r[1]), regions):
-                    if found:
-                        result[0] = found
-                        stop.set()
-                        break
-        finally:
-            CloseHandle(handle)
-
-        return result[0]
-
-    for pid in pids:
-        found = scan_pid(pid)
-        if found:
-            return found
-    return None
-
-
 @lru_cache(maxsize=4096)
 def _fallback_search_media_by_file_id(
     weixin_root_str: str,
@@ -1495,11 +1116,17 @@ def _fallback_search_media_by_file_id(
     return None
 
 
-def _save_media_keys(account_dir: Path, xor_key: int, aes_key16: bytes) -> None:
+def _save_media_keys(account_dir: Path, xor_key: int, aes_key16: Optional[bytes] = None) -> None:
     try:
+        aes_str = ""
+        if aes_key16:
+            try:
+                aes_str = aes_key16.decode("ascii", errors="ignore")[:16]
+            except Exception:
+                aes_str = ""
         payload = {
             "xor": int(xor_key),
-            "aes": aes_key16.decode("ascii", errors="ignore"),
+            "aes": aes_str,
         }
         (account_dir / "_media_keys.json").write_text(
             json.dumps(payload, ensure_ascii=False, indent=2),
@@ -1650,17 +1277,13 @@ def _read_and_maybe_decrypt_media(
     # Try WeChat .dat v1/v2 decrypt.
     version = _detect_wechat_dat_version(data)
     if version in (0, 1, 2):
-        root = weixin_root
-        if root is None and account_dir is not None:
-            root = _resolve_account_wxid_dir(account_dir)
-        if root is None and account_dir is not None:
-            ds = _resolve_account_db_storage_dir(account_dir)
-            root = ds.parent if ds else None
-
-        xor_key = _find_wechat_xor_key(str(root)) if root else None
-        if xor_key is None and account_dir is not None:
+        # 不在本项目内做任何密钥提取；仅使用用户保存的密钥（_media_keys.json）。
+        xor_key: Optional[int] = None
+        aes_key16 = b""
+        if account_dir is not None:
             try:
                 keys2 = _load_media_keys(account_dir)
+
                 x2 = keys2.get("xor")
                 if x2 is not None:
                     xor_key = int(x2)
@@ -1668,8 +1291,13 @@ def _read_and_maybe_decrypt_media(
                         xor_key = None
                     else:
                         logger.debug("使用 _media_keys.json 中保存的 xor key")
+
+                aes_str = str(keys2.get("aes") or "").strip()
+                if len(aes_str) >= 16:
+                    aes_key16 = aes_str[:16].encode("ascii", errors="ignore")
             except Exception:
                 xor_key = None
+                aes_key16 = b""
         try:
             if version == 0 and xor_key is not None:
                 out = _decrypt_wechat_dat_v3(data, xor_key)
@@ -1707,41 +1335,24 @@ def _read_and_maybe_decrypt_media(
                 mt1 = _detect_image_media_type(out[:32])
                 if mt1 != "application/octet-stream":
                     return out, mt1
-            elif version == 2 and xor_key is not None and account_dir is not None and root is not None:
-                keys = _load_media_keys(account_dir)
-                aes_str = str(keys.get("aes") or "").strip()
-                aes_key16 = aes_str.encode("ascii", errors="ignore")[:16] if aes_str else b""
-
-                if not aes_key16:
-                    most_common = _get_wechat_template_most_common_last2(str(root))
-                    if most_common:
-                        ct = _get_wechat_v2_ciphertext(Path(root), most_common)
+            elif version == 2 and xor_key is not None and aes_key16:
+                out = _decrypt_wechat_dat_v4(data, xor_key, aes_key16)
+                try:
+                    out2, mtp2 = _try_strip_media_prefix(out)
+                    if mtp2 != "application/octet-stream":
+                        return out2, mtp2
+                except Exception:
+                    pass
+                if out.startswith(b"wxgf"):
+                    converted = _wxgf_to_image_bytes(out)
+                    if converted:
+                        out = converted
+                        logger.info(f"wxgf->image: {path} -> {len(out)} bytes")
                     else:
-                        ct = None
-
-                    if ct:
-                        aes_key16 = _extract_wechat_aes_key_from_process(ct) or b""
-                        if aes_key16:
-                            _save_media_keys(account_dir, xor_key, aes_key16)
-
-                if aes_key16:
-                    out = _decrypt_wechat_dat_v4(data, xor_key, aes_key16)
-                    try:
-                        out2, mtp2 = _try_strip_media_prefix(out)
-                        if mtp2 != "application/octet-stream":
-                            return out2, mtp2
-                    except Exception:
-                        pass
-                    if out.startswith(b"wxgf"):
-                        converted = _wxgf_to_image_bytes(out)
-                        if converted:
-                            out = converted
-                            logger.info(f"wxgf->image: {path} -> {len(out)} bytes")
-                        else:
-                            logger.info(f"wxgf->image failed: {path}")
-                    mt2b = _detect_image_media_type(out[:32])
-                    if mt2b != "application/octet-stream":
-                        return out, mt2b
+                        logger.info(f"wxgf->image failed: {path}")
+                mt2b = _detect_image_media_type(out[:32])
+                if mt2b != "application/octet-stream":
+                    return out, mt2b
         except Exception:
             pass
 

src/wechat_decrypt_tool/media_key_finder.py

@@ -1,250 +0,0 @@
-import ctypes
-import json
-import os
-import re
-import threading
-from collections import Counter
-from concurrent.futures import ThreadPoolExecutor
-from ctypes import wintypes
-from functools import lru_cache
-from pathlib import Path
-from typing import Any
-
-import pymem
-import yara
-from Crypto.Cipher import AES
-
-PROCESS_ALL_ACCESS = 0x1F0FFF
-MEM_COMMIT = 0x1000
-MEM_PRIVATE = 0x20000
-
-kernel32 = ctypes.windll.kernel32
-
-
-class MEMORY_BASIC_INFORMATION(ctypes.Structure):
-    _fields_ = [
-        ("BaseAddress", ctypes.c_void_p),
-        ("AllocationBase", ctypes.c_void_p),
-        ("AllocationProtect", ctypes.c_ulong),
-        ("RegionSize", ctypes.c_size_t),
-        ("State", ctypes.c_ulong),
-        ("Protect", ctypes.c_ulong),
-        ("Type", ctypes.c_ulong),
-    ]
-
-
-def _open_process(pid: int):
-    return kernel32.OpenProcess(PROCESS_ALL_ACCESS, False, pid)
-
-
-def _read_process_memory(process_handle, address: int, size: int) -> bytes | None:
-    buffer = ctypes.create_string_buffer(size)
-    bytes_read = ctypes.c_size_t(0)
-    success = kernel32.ReadProcessMemory(
-        process_handle,
-        ctypes.c_void_p(address),
-        buffer,
-        size,
-        ctypes.byref(bytes_read),
-    )
-    if not success:
-        return None
-    return buffer.raw
-
-
-def _get_memory_regions(process_handle) -> list[tuple[int, int]]:
-    regions: list[tuple[int, int]] = []
-    mbi = MEMORY_BASIC_INFORMATION()
-    address = 0
-    while kernel32.VirtualQueryEx(
-        process_handle,
-        ctypes.c_void_p(address),
-        ctypes.byref(mbi),
-        ctypes.sizeof(mbi),
-    ):
-        if mbi.State == MEM_COMMIT and mbi.Type == MEM_PRIVATE:
-            regions.append((int(mbi.BaseAddress), int(mbi.RegionSize)))
-        address += int(mbi.RegionSize)
-    return regions
-
-
-@lru_cache
-def _verify(encrypted: bytes, key: bytes) -> bool:
-    aes_key = key[:16]
-    cipher = AES.new(aes_key, AES.MODE_ECB)
-    text = cipher.decrypt(encrypted)
-    if text.startswith(b"\xff\xd8\xff"):
-        return True
-    if text.startswith(b"\x89PNG\r\n\x1a\n"):
-        return True
-    if text.startswith(b"GIF87a") or text.startswith(b"GIF89a"):
-        return True
-    if text.startswith(b"wxgf"):
-        return True
-    if len(text) >= 12 and text.startswith(b"RIFF") and text[8:12] == b"WEBP":
-        return True
-    if len(text) >= 8 and text[4:8] == b"ftyp":
-        return True
-    return False
-
-
-def _search_memory_chunk(process_handle, base_address: int, region_size: int, encrypted: bytes, rules):
-    memory = _read_process_memory(process_handle, base_address, region_size)
-    if not memory:
-        return None
-
-    matches = rules.match(data=memory)
-    if matches:
-        for match in matches:
-            if match.rule == "AesKey":
-                for string in match.strings:
-                    for instance in string.instances:
-                        content = instance.matched_data[1:-1]
-                        if _verify(encrypted, content):
-                            return content[:16]
-    return None
-
-
-def _get_aes_key(encrypted: bytes, pid: int) -> Any:
-    process_handle = _open_process(pid)
-    if not process_handle:
-        raise RuntimeError(f"无法打开进程 {pid}")
-
-    rules_key = r"""
-    rule AesKey {
-        strings:
-            $pattern = /[^0-9a-z]([0-9a-z]{16}|[0-9a-z]{32})[^0-9a-z]/ nocase
-        condition:
-            $pattern
-    }
-    """
-    rules = yara.compile(source=rules_key)
-
-    process_infos = _get_memory_regions(process_handle)
-
-    found_result = threading.Event()
-    result = [None]
-
-    def process_chunk(args):
-        if found_result.is_set():
-            return None
-        base_address, region_size = args
-        res = _search_memory_chunk(process_handle, base_address, region_size, encrypted, rules)
-        if res:
-            result[0] = res
-            found_result.set()
-        return res
-
-    with ThreadPoolExecutor(max_workers=min(32, len(process_infos) or 1)) as executor:
-        executor.map(process_chunk, process_infos)
-
-    kernel32.CloseHandle(process_handle)
-    return result[0]
-
-
-def _dump_wechat_info_v4(encrypted: bytes, pid: int) -> bytes:
-    result = _get_aes_key(encrypted, pid)
-    if isinstance(result, bytes):
-        return result[:16]
-    raise RuntimeError("未找到 AES 密钥")
-
-
-def _sort_template_files_by_date(template_files: list[Path]) -> list[Path]:
-    def get_date_from_path(filepath: Path) -> str:
-        match = re.search(r"(\d{4}-\d{2})", str(filepath))
-        if match:
-            return match.group(1)
-        return "0000-00"
-
-    return sorted(template_files, key=get_date_from_path, reverse=True)
-
-
-def find_key(
-    weixin_dir: Path,
-    version: int = 4,
-    xor_key_: int | None = None,
-    aes_key_: bytes | None = None,
-) -> tuple[int, bytes]:
-    if os.name != "nt":
-        raise RuntimeError("仅支持 Windows")
-    if version not in (3, 4):
-        raise RuntimeError("version must be 3 or 4")
-
-    template_files = _sort_template_files_by_date(list(weixin_dir.rglob("*_t.dat")))
-    if not template_files:
-        raise RuntimeError("未找到模板文件")
-
-    last_bytes_list: list[bytes] = []
-    for file in template_files[:16]:
-        try:
-            with open(file, "rb") as f:
-                f.seek(-2, 2)
-                last_bytes_list.append(f.read(2))
-        except Exception:
-            continue
-
-    if not last_bytes_list:
-        raise RuntimeError("对于 XOR, 未能成功读取任何模板文件")
-
-    counter = Counter(last_bytes_list)
-    most_common = counter.most_common(1)[0][0]
-
-    x, y = most_common
-    xor_key = x ^ 0xFF
-    if xor_key != (y ^ 0xD9):
-        raise RuntimeError("未能找到 XOR 密钥")
-
-    if xor_key_ is not None:
-        if xor_key_ != xor_key:
-            raise RuntimeError("XOR 密钥校验失败")
-        return xor_key_, aes_key_ or b""
-
-    if version == 3:
-        return xor_key, b"cfcd208495d565ef"
-
-    ciphertext: bytes | None = None
-    for file in template_files:
-        with open(file, "rb") as f:
-            if f.read(6) != b"\x07\x08V2\x08\x07":
-                continue
-            f.seek(-2, 2)
-            if f.read(2) != most_common:
-                continue
-            f.seek(0xF)
-            ciphertext = f.read(16)
-            break
-
-    if not ciphertext:
-        raise RuntimeError("对于 AES, 未能成功读取任何模板文件")
-
-    try:
-        pm = pymem.Pymem("Weixin.exe")
-        pid = pm.process_id
-        if not isinstance(pid, int):
-            raise RuntimeError("找不到微信进程")
-    except Exception:
-        raise RuntimeError("找不到微信进程")
-
-    aes_key = _dump_wechat_info_v4(ciphertext, pid)
-    return xor_key, aes_key
-
-
-CONFIG_FILE = "config.json"
-
-
-def read_key_from_config() -> tuple[int, bytes]:
-    if os.path.exists(CONFIG_FILE):
-        with open(CONFIG_FILE, "r", encoding="utf-8") as f:
-            key_dict = json.loads(f.read())
-        x, y = key_dict["xor"], key_dict["aes"]
-        return x, y.encode()[:16]
-    return 0, b""
-
-
-def store_key(xor_k: int, aes_k: bytes) -> None:
-    key_dict = {
-        "xor": xor_k,
-        "aes": aes_k.decode(),
-    }
-    with open(CONFIG_FILE, "w", encoding="utf-8") as f:
-        f.write(json.dumps(key_dict))

src/wechat_decrypt_tool/routers/chat_export.py

@@ -33,7 +33,7 @@ class ChatExportCreateRequest(BaseModel):
     )
     allow_process_key_extract: bool = Field(
         False,
-        description="是否允许尝试从微信进程提取媒体密钥（预留；当前仅使用已存在的本地文件）",
+        description="预留字段：本项目不从微信进程提取媒体密钥，请使用 wx_key 获取并保存/批量解密",
     )
     privacy_mode: bool = Field(
         False,

src/wechat_decrypt_tool/routers/media.py

@@ -11,11 +11,7 @@ from ..media_helpers import (
     _collect_all_dat_files,
     _decrypt_and_save_resource,
     _detect_image_media_type,
-    _extract_wechat_aes_key_from_process,
-    _find_wechat_xor_key,
     _get_resource_dir,
-    _get_wechat_template_most_common_last2,
-    _get_wechat_v2_ciphertext,
     _load_media_keys,
     _resolve_account_dir,
     _resolve_account_wxid_dir,
@@ -29,11 +25,12 @@ logger = get_logger(__name__)
 router = APIRouter(route_class=PathFixRoute)
 
 
-class MediaKeysRequest(BaseModel):
-    """媒体密钥请求模型"""
+class MediaKeysSaveRequest(BaseModel):
+    """媒体密钥保存请求模型（用户手动提供）"""
 
     account: Optional[str] = Field(None, description="账号目录名（可选，默认使用第一个）")
-    force_extract: bool = Field(False, description="是否强制从微信进程重新提取密钥")
+    xor_key: str = Field(..., description="XOR密钥（十六进制格式，如 0xA5 或 A5）")
+    aes_key: Optional[str] = Field(None, description="AES密钥（可选，至少16字符，V4-V2需要）")
 
 
 class MediaDecryptRequest(BaseModel):
@@ -44,105 +41,37 @@ class MediaDecryptRequest(BaseModel):
     aes_key: Optional[str] = Field(None, description="AES密钥（16字符ASCII字符串）")
 
 
-@router.get("/api/media/keys", summary="获取图片解密密钥")
-async def get_media_keys(account: Optional[str] = None, force_extract: bool = False):
-    """获取图片解密密钥（XOR和AES）
-
-    如果已缓存密钥且不强制提取，直接返回缓存的密钥。
-    否则尝试从微信进程中提取密钥。
-
-    注意：提取AES密钥需要微信进程正在运行。
-    """
-    account_dir = _resolve_account_dir(account)
-    wxid_dir = _resolve_account_wxid_dir(account_dir)
-
-    # 尝试加载已缓存的密钥
-    cached_keys = _load_media_keys(account_dir)
-    if cached_keys and not force_extract:
-        xor_key = cached_keys.get("xor")
-        aes_key = cached_keys.get("aes")
-        if xor_key is not None and aes_key:
-            return {
-                "status": "success",
-                "source": "cache",
-                "xor_key": f"0x{int(xor_key):02X}",
-                "aes_key": str(aes_key)[:16] if aes_key else "",
-                "message": "已从缓存加载密钥",
-            }
-
-    if not wxid_dir:
-        return {
-            "status": "error",
-            "message": "未找到微信数据目录，请确保已正确配置 db_storage_path",
-        }
-
-    # 尝试提取XOR密钥
-    xor_key = _find_wechat_xor_key(str(wxid_dir))
-    if xor_key is None:
-        return {
-            "status": "error",
-            "message": "无法提取XOR密钥，请确保微信数据目录中存在 _t.dat 模板文件",
-        }
-
-    # 尝试提取AES密钥（需要微信进程运行）
-    aes_key16: Optional[bytes] = None
-    aes_message = ""
-
-    most_common = _get_wechat_template_most_common_last2(str(wxid_dir))
-    if most_common:
-        ct = _get_wechat_v2_ciphertext(wxid_dir, most_common)
-        if ct:
-            aes_key16 = _extract_wechat_aes_key_from_process(ct)
-            if aes_key16:
-                aes_message = "已从微信进程提取AES密钥"
-                # 保存密钥到缓存
-                _save_media_keys(account_dir, xor_key, aes_key16)
-            else:
-                aes_message = "无法从微信进程提取AES密钥（请确认微信正在运行，并尝试以管理员身份运行后端；可尝试打开朋友圈图片并点开大图 2-3 次后再提取）"
-        else:
-            aes_message = "未找到V2加密模板文件"
-    else:
-        aes_message = "未找到足够的模板文件用于提取AES密钥"
-
-    return {
-        "status": "success",
-        "source": "extracted",
-        "xor_key": f"0x{xor_key:02X}",
-        "aes_key": aes_key16.decode("ascii", errors="ignore") if aes_key16 else "",
-        "message": f"XOR密钥提取成功。{aes_message}",
-    }
-
-
 @router.post("/api/media/keys", summary="保存图片解密密钥")
-async def save_media_keys_api(request: MediaKeysRequest, xor_key: str, aes_key: str):
+async def save_media_keys_api(request: MediaKeysSaveRequest):
     """手动保存图片解密密钥
 
     参数:
     - xor_key: XOR密钥（十六进制格式，如 0xA5 或 A5）
-    - aes_key: AES密钥（16字符ASCII字符串）
+    - aes_key: AES密钥（可选，至少16个字符；V4-V2需要）
     """
     account_dir = _resolve_account_dir(request.account)
 
     # 解析XOR密钥
     try:
-        xor_hex = xor_key.strip().lower().replace("0x", "")
+        xor_hex = request.xor_key.strip().lower().replace("0x", "")
         xor_int = int(xor_hex, 16)
     except Exception:
         raise HTTPException(status_code=400, detail="XOR密钥格式无效，请使用十六进制格式如 0xA5")
 
-    # 验证AES密钥
-    aes_str = aes_key.strip()
-    if len(aes_str) < 16:
+    # 验证AES密钥（可选）
+    aes_str = str(request.aes_key or "").strip()
+    if aes_str and len(aes_str) < 16:
         raise HTTPException(status_code=400, detail="AES密钥长度不足，需要至少16个字符")
 
     # 保存密钥
-    _save_media_keys(account_dir, xor_int, aes_str[:16].encode("ascii", errors="ignore"))
+    aes_key16 = aes_str[:16].encode("ascii", errors="ignore") if aes_str else None
+    _save_media_keys(account_dir, xor_int, aes_key16)
 
     return {
         "status": "success",
         "message": "密钥已保存",
         "xor_key": f"0x{xor_int:02X}",
-        "aes_key": aes_str[:16],
+        "aes_key": aes_str[:16] if aes_str else "",
     }
 
 
@@ -193,14 +122,10 @@ async def decrypt_all_media(request: MediaDecryptRequest):
             if len(aes_str) >= 16:
                 aes_key16 = aes_str[:16].encode("ascii", errors="ignore")
 
-    # 如果仍然没有XOR密钥，尝试自动提取
-    if xor_key_int is None:
-        xor_key_int = _find_wechat_xor_key(str(wxid_dir))
-
     if xor_key_int is None:
         raise HTTPException(
             status_code=400,
-            detail="未找到XOR密钥，请先调用 /api/media/keys 获取密钥或手动提供",
+            detail="未找到XOR密钥，请先使用 wx_key 获取并通过前端填写（或调用 /api/media/keys 保存）",
         )
 
     # 收集所有.dat文件
@@ -353,12 +278,8 @@ async def decrypt_all_media_stream(
                     if len(aes_str) >= 16:
                         aes_key16 = aes_str[:16].encode("ascii", errors="ignore")
 
-            # 如果仍然没有XOR密钥，尝试自动提取
             if xor_key_int is None:
-                xor_key_int = _find_wechat_xor_key(str(wxid_dir))
-
-            if xor_key_int is None:
-                yield f"data: {json.dumps({'type': 'error', 'message': '未找到XOR密钥，请先获取密钥'})}\n\n"
+                yield f"data: {json.dumps({'type': 'error', 'message': '未找到XOR密钥，请先使用 wx_key 获取并保存/填写'})}\n\n"
                 return
 
             # 收集所有.dat文件