feat: macOS 图片 AES key 从磁盘 kvcomm 缓存派生（issue #23）

macOS 用户长期无法用 C 版 find_image_key_macos 从微信进程内存提取
V2 图片密钥（issue #23 报告 197K 候选全部失败）。新增
find_image_key_macos.py 走完全不同的路径：从磁盘 kvcomm 缓存
文件名派生密钥，无需扫描内存、无需 root、无需重签名。

派生算法
--------
- 扫 ~/.../app_data/net/kvcomm/key_<code>_*.statistic 文件名
- 对每个 (code, wxid) 候选：
    xor_key = code & 0xFF
    aes_key = MD5(str(code) + cleaned_wxid).hex()[:16]   # ASCII 字符串
- 用 V2 _t.dat 文件 [0xF:0x1F] 16 字节做 AES-128-ECB 模板验证：
  解出来必须是图像 magic（JPEG / PNG / GIF / WebP / wxgf）
- 为防短 magic 偶然命中，要求多个不同模板都通过验证才算成功
- 命中后写回 config.json 的 image_aes_key / image_xor_key，
  monitor_web.py 自动加载

致谢
----
派生算法源自 @hicccc77 在 issue #23 的评论；参考实现见其 WeFlow
项目 (CC BY-NC-SA 4.0)。本模块是独立的 Python clean-room 实现，
未复制其 TypeScript 源码；函数边界与变量命名沿用算法的自然结构
（regex 模式 / MD5 调用顺序 / magic 字节表等不可避免地相同）。

健壮性细节
----------
- 多候选 kvcomm 路径：枚举 5 个不同的 macOS 微信版本路径布局
- 多模板交叉验证：默认收集 3 个不同密文，全部通过才算命中
- 已有 image_aes_key 仍有效时短路返回，不重写 config
- 原子写 config.json：tmp + os.replace + finally 清理 .tmp
- 多 wxid 候选：同时试 raw 和归一化后的 wxid（A_Hare_626a → A_Hare）
- print(flush=True) 逐次显式（与 find_image_key.py 风格一致）

测试
----
新增 tests/test_find_image_key_macos.py，53 个测试覆盖：
派生算法 / wxid 归一化 / kvcomm 路径推算（含多候选）/ 模板收集
（去重 / 子目录 / max_files 边界）/ AES 验证（5 种 magic / 短输入
/ 空 key）/ 多模板交叉验证 / 端到端集成（命中 / 各种失败分支）/
原子写 / main 短路（已有有效 key 不重写 / 已有错 key 落到派生）。
全部通过：python -m unittest discover tests → 88/88。

兼容性
------
- 无新增依赖（pycryptodome 已在 requirements.txt）
- 不改任何现有 Python 文件，零回归风险
- 现有 Windows / Linux 路径 (find_image_key.py / find_image_key_monitor.py) 不受影响

README.md

@@ -205,7 +205,9 @@ claude mcp add wechat -- python C:\Users\你的用户名\wechat-decrypt\mcp_serv
 
 ### 图片解密 (V2 格式)
 
-微信 4.0 (2025-08+) 的 .dat 图片文件使用 AES-128-ECB + XOR 混合加密 (V2 格式)。AES 密钥需要从运行中的微信进程内存中提取：
+微信 4.0 (2025-08+) 的 .dat 图片文件使用 AES-128-ECB + XOR 混合加密 (V2 格式)。AES 密钥的获取方式因平台而异：
+
+**Windows / Linux**（从进程内存扫描）：
 
 ```bash
 # 1. 在微信中打开查看 2-3 张图片（点击看大图）
@@ -216,9 +218,19 @@ python find_image_key_monitor.py
 python find_image_key.py
 ```
 
-密钥会自动保存到 `config.json` 的 `image_aes_key` 字段。之后 `monitor_web.py` 启动时会自动加载密钥，图片消息将显示内联预览。
+> AES 密钥仅在微信查看图片时临时加载到内存中。如果扫描未找到密钥，请先在微信中查看几张图片，然后立即重新运行脚本。
 
-> **注意**: AES 密钥仅在微信查看图片时临时加载到内存中。如果扫描未找到密钥，请先在微信中查看几张图片，然后立即重新运行脚本。
+**macOS**（从磁盘 kvcomm 缓存派生，**无需扫描进程内存**）：
+
+```bash
+python find_image_key_macos.py
+```
+
+无需提前在微信中查看图片，无需 root 权限，无需重签名。脚本会扫描 `~/Library/Containers/com.tencent.xinWeChat/.../app_data/net/kvcomm/key_*.statistic` 文件名提取派生码 `code`，配合 `db_dir` 路径里的 wxid，按 `aes_key = MD5(str(code) + cleaned_wxid)[:16]` / `xor_key = code & 0xFF` 的规则推算密钥，并用一张 V2 `_t.dat` 缩略图做 AES 模板验证。解决 [issue #23](https://github.com/ylytdeng/wechat-decrypt/issues/23)（macOS 内存扫描器 197K 候选全部失败）。
+
+派生算法的发现归功于 [@hicccc77](https://github.com/hicccc77) 在 issue #23 的[评论](https://github.com/ylytdeng/wechat-decrypt/issues/23)，参考实现见其 [WeFlow 项目](https://github.com/hicccc77/WeFlow/blob/dev/electron/services/keyServiceMac.ts)（CC BY-NC-SA 4.0）。本仓库的 `find_image_key_macos.py` 是基于该算法的独立 Python clean-room 实现。
+
+密钥会自动保存到 `config.json` 的 `image_aes_key` / `image_xor_key` 字段。之后 `monitor_web.py` 启动时会自动加载，图片消息将显示内联预览。
 
 ## 文件说明
 
@@ -234,8 +246,9 @@ python find_image_key.py
 | `monitor_web.py` | 实时消息监听 (Web UI + SSE + 图片预览) |
 | `monitor.py` | 实时消息监听 (命令行) |
 | `decode_image.py` | 图片 .dat 文件解密模块 (XOR / V1 / V2) |
-| `find_image_key.py` | 从微信进程内存提取图片 AES 密钥 |
+| `find_image_key.py` | 从微信进程内存提取图片 AES 密钥（Windows / Linux） |
-| `find_image_key_monitor.py` | 持续监控版密钥提取（推荐） |
+| `find_image_key_monitor.py` | 持续监控版密钥提取（Windows / Linux，推荐） |
+| `find_image_key_macos.py` | macOS 版图片密钥派生（从磁盘 kvcomm 缓存推算，无需扫描内存） |
 | `latency_test.py` | 延迟测量诊断工具 |
 | `find_all_keys_macos.c` | macOS 版内存密钥扫描器 (C, Mach VM API) |
 

find_image_key_macos.py

@@ -0,0 +1,362 @@
+"""macOS WeChat 4.x 图片 AES key 派生（无需扫描进程内存）。
+
+通过 macOS 微信 4.x 在磁盘上的 kvcomm 缓存文件命名约定，派生出 V2 .dat
+图片解密所需的 (xor_key, aes_key)。解决 issue #23：macOS 用户无法用
+C 版扫描器从进程内存提取图片密钥（197K 候选全部失败）。
+
+派生算法
+--------
+- 扫 ~/.../app_data/net/kvcomm/key_<code>_*.statistic 文件名
+- 对每个 (code, wxid) 候选：
+    xor_key = code & 0xFF
+    aes_key = MD5(str(code) + cleaned_wxid).hex()[:16]   # ASCII 字符串
+- 用 V2 _t.dat 文件 [0xF:0x1F] 16 字节做模板验证：派生出的 aes_key 把
+  密文 AES-128-ECB 解出图像 magic（JPEG / PNG / GIF / WebP / wxgf）即视为命中
+- 为防短 magic 偶然命中，要求多个不同模板都通过验证才视为成功
+- 命中后写回 config.json 的 image_aes_key / image_xor_key 字段，
+  monitor_web.py 启动时自动加载，图片消息显示内联预览
+
+致谢
+----
+派生算法源自 @hicccc77 在 issue #23 的评论，参考实现位于
+https://github.com/hicccc77/WeFlow （CC BY-NC-SA 4.0）。本模块是独立的
+Python 实现，未复制其 TypeScript 源码；函数边界与变量命名沿用算法的自然
+结构（regex 模式 / MD5 调用顺序 / magic 字节表等不可避免地相同）。
+
+用法
+----
+  python find_image_key_macos.py
+"""
+import glob
+import hashlib
+import json
+import os
+import platform
+import re
+import sys
+
+from Crypto.Cipher import AES
+
+# V2 .dat 文件 magic（与 decode_image.py 中 V2_MAGIC_FULL 一致）
+V2_MAGIC = bytes.fromhex("070856320807")
+
+# kvcomm 文件名格式：key_<code>_<其他段>.statistic
+# code 必须紧跟在 "key_" 之后（不能是 "key_reportnow_..." 这种带前缀的）
+_KVCOMM_FILENAME_RE = re.compile(r"^key_(\d+)_.+\.statistic$", re.IGNORECASE)
+
+# AES 解密结果允许的图像 magic
+_IMAGE_MAGICS = (
+    b"\xff\xd8\xff",      # JPEG
+    b"\x89\x50\x4e\x47",  # PNG
+    b"GIF",               # GIF
+    b"RIFF",              # WebP container（首块只能看前 16B，全检需 [8:12]==b"WEBP"）
+    b"wxgf",              # 微信 HEVC GIF / Live Photo
+)
+
+
+def normalize_wxid(account_id):
+    """归一化账号 ID。
+
+    - wxid_<seg> 形式：保留 wxid_<seg>，丢弃后续下划线分段
+    - <base>_<4 alnum> 形式：丢弃 _<4 alnum> 后缀（macOS 路径目录名常见）
+    - 其他：原样返回
+    """
+    aid = (account_id or "").strip()
+    if not aid:
+        return ""
+    if aid.lower().startswith("wxid_"):
+        m = re.match(r"^(wxid_[^_]+)", aid, re.IGNORECASE)
+        return m.group(1) if m else aid
+    m = re.match(r"^(.+)_([a-zA-Z0-9]{4})$", aid)
+    return m.group(1) if m else aid
+
+
+def derive_image_keys(code, wxid):
+    """从 (code, wxid) 派生 (xor_key, aes_key_ascii)。
+
+    aes_key_ascii 是 16 字符 hex 字符串；调用方按 ASCII 编码取前 16 字节作为
+    AES-128 密钥。本函数不做 wxid 归一化（由调用方枚举原值与归一化值）。
+    """
+    xor_key = int(code) & 0xFF
+    aes_key = hashlib.md5(f"{code}{wxid}".encode("utf-8")).hexdigest()[:16]
+    return xor_key, aes_key
+
+
+def derive_kvcomm_dir_candidates(db_dir):
+    """从 db_dir 推算所有可能的 kvcomm 缓存目录（按优先级排序）。
+
+    微信 4.x 在不同版本 / 安装方式下 kvcomm 路径不固定，需要枚举多个候选。
+    返回的列表里至少有一项被 os.path.isdir 确认存在时才算可用。
+    """
+    parts = db_dir.rstrip(os.sep).split(os.sep)
+    candidates = []
+    if "xwechat_files" in parts:
+        idx = parts.index("xwechat_files")
+        documents_root = os.sep.join(parts[:idx])
+        # 1) 与 xwechat_files 兄弟目录的 app_data
+        candidates.append(os.path.join(documents_root, "app_data", "net", "kvcomm"))
+        # 2) 旧版可能放 xwechat 子目录
+        candidates.append(os.path.join(documents_root, "xwechat", "net", "kvcomm"))
+        # 3) 容器内 Application Support 路径（部分版本）
+        if idx >= 1:
+            container_root = os.sep.join(parts[:idx - 1])  # Documents 之上
+            candidates.append(os.path.join(
+                container_root, "Library", "Application Support",
+                "com.tencent.xinWeChat", "xwechat", "net", "kvcomm"))
+            candidates.append(os.path.join(
+                container_root, "Library", "Application Support",
+                "com.tencent.xinWeChat", "net", "kvcomm"))
+    # 4) 兜底：HOME 下默认沙盒路径
+    home = os.path.expanduser("~")
+    candidates.append(os.path.join(
+        home, "Library", "Containers", "com.tencent.xinWeChat", "Data",
+        "Documents", "app_data", "net", "kvcomm"))
+    # 去重，保留顺序
+    seen = set()
+    deduped = []
+    for c in candidates:
+        if c not in seen:
+            seen.add(c)
+            deduped.append(c)
+    return deduped
+
+
+def find_existing_kvcomm_dir(db_dir):
+    """从候选路径中返回第一个存在的 kvcomm 目录；都不存在返回 None。"""
+    for candidate in derive_kvcomm_dir_candidates(db_dir):
+        if os.path.isdir(candidate):
+            return candidate
+    return None
+
+
+def collect_kvcomm_codes(kvcomm_dir):
+    """扫 kvcomm 目录，返回去重排序的 code 列表。"""
+    if not kvcomm_dir or not os.path.isdir(kvcomm_dir):
+        return []
+    codes = set()
+    try:
+        names = os.listdir(kvcomm_dir)
+    except OSError:
+        return []
+    for name in names:
+        m = _KVCOMM_FILENAME_RE.match(name)
+        if not m:
+            continue
+        try:
+            code = int(m.group(1))
+        except ValueError:
+            continue
+        if 0 < code <= 0xFFFFFFFF:
+            codes.add(code)
+    return sorted(codes)
+
+
+def collect_wxid_candidates(db_dir):
+    """从 db_dir 提取候选 wxid（含原值和归一化值）。"""
+    parts = db_dir.rstrip(os.sep).split(os.sep)
+    if "xwechat_files" not in parts:
+        return []
+    idx = parts.index("xwechat_files")
+    if idx + 1 >= len(parts):
+        return []
+    raw = parts[idx + 1]
+    candidates = [raw]
+    normalized = normalize_wxid(raw)
+    if normalized and normalized != raw:
+        candidates.append(normalized)
+    return candidates
+
+
+def find_v2_template_ciphertexts(attach_dir, max_templates=3, max_files=64):
+    """在 attach_dir 下找 V2 .dat 文件的模板密文（[0xF:0x1F] 16 字节）。
+
+    优先 _t.dat（缩略图小、读得快），找不到再降级用任意 .dat。
+    返回最多 max_templates 个**不同**的密文，用于交叉验证防止短 magic 偶然命中。
+    """
+    if not attach_dir or not os.path.isdir(attach_dir):
+        return []
+
+    def _scan(suffix):
+        # 出口条件只看是否凑够 max_templates 个**不同**密文；不因为
+        # examined 达到 max_files 提前退出 —— 否则若前 64 个文件都是同一
+        # 张图的副本，结果只有 1 个 template，交叉验证就退化成单模板。
+        out, seen = [], set()
+        examined = 0
+        for root, _, files in os.walk(attach_dir):
+            for f in files:
+                if not f.endswith(suffix):
+                    continue
+                examined += 1
+                try:
+                    with open(os.path.join(root, f), "rb") as fp:
+                        data = fp.read(0x20)
+                except OSError:
+                    continue
+                if len(data) >= 0x1F and data[:6] == V2_MAGIC:
+                    ct = data[0xF:0x1F]
+                    if ct not in seen:
+                        seen.add(ct)
+                        out.append(ct)
+                        if len(out) >= max_templates:
+                            return out
+                # 兜底：扫了 max_files 个文件还凑不齐 max_templates 个不同的，
+                # 提前停止以免在巨型 attach 目录里跑很久（只在 out 不空时才能停）
+                if examined >= max_files and out:
+                    return out
+        return out
+
+    return _scan("_t.dat") or _scan(".dat")
+
+
+def verify_aes_key(aes_key_ascii, template_ct):
+    """AES-128-ECB 解 template_ct（16 字节），检查头部是否是图像 magic。"""
+    if not aes_key_ascii or not template_ct or len(template_ct) != 16:
+        return False
+    key_bytes = aes_key_ascii.encode("ascii", errors="ignore")[:16]
+    if len(key_bytes) < 16:
+        return False
+    try:
+        cipher = AES.new(key_bytes, AES.MODE_ECB)
+        decrypted = cipher.decrypt(template_ct)
+    except (ValueError, KeyError):
+        return False
+    return any(decrypted.startswith(m) for m in _IMAGE_MAGICS)
+
+
+def verify_aes_key_against_all(aes_key_ascii, templates):
+    """在多个模板上交叉验证 aes_key。全部通过才算命中（防短 magic 偶然碰撞）。"""
+    if not templates:
+        return False
+    return all(verify_aes_key(aes_key_ascii, ct) for ct in templates)
+
+
+def find_image_key_macos(db_dir):
+    """在 macOS 上派生并交叉验证 V2 图片密钥。
+
+    Returns:
+        (xor_key, aes_key_ascii) on success；失败返回 None 并打印诊断信息。
+    """
+    kvcomm_dir = find_existing_kvcomm_dir(db_dir)
+    if not kvcomm_dir:
+        print(f"[!] 找不到 kvcomm 缓存目录，已尝试以下候选:", flush=True)
+        for c in derive_kvcomm_dir_candidates(db_dir):
+            print(f"      {c}", flush=True)
+        print("    通常意味着微信尚未生成密钥缓存，请先在微信中查看 1-2 张图片",
+              flush=True)
+        return None
+    print(f"[+] 使用 kvcomm 目录: {kvcomm_dir}", flush=True)
+
+    codes = collect_kvcomm_codes(kvcomm_dir)
+    if not codes:
+        print(f"[!] kvcomm 目录无 key_*.statistic 文件: {kvcomm_dir}", flush=True)
+        return None
+    print(f"[+] 找到 {len(codes)} 个 code 候选", flush=True)
+
+    wxid_candidates = collect_wxid_candidates(db_dir)
+    if not wxid_candidates:
+        print(f"[!] 无法从 db_dir 提取 wxid: {db_dir}", flush=True)
+        return None
+    print(f"[+] wxid 候选: {wxid_candidates}", flush=True)
+
+    base_dir = os.path.dirname(db_dir)  # 去掉 db_storage
+    attach_dir = os.path.join(base_dir, "msg", "attach")
+    templates = find_v2_template_ciphertexts(attach_dir)
+    if not templates:
+        print(f"[!] 在 {attach_dir} 下找不到 V2 模板文件", flush=True)
+        print("    请先在微信中查看 1-2 张图片，让微信生成 V2 .dat 文件",
+              flush=True)
+        return None
+    print(f"[+] 找到 {len(templates)} 个不同模板用于交叉验证", flush=True)
+
+    # 穷举顺序：wxid 外、code 内。这样多账号系统下当前账号的所有 code 优先尝试。
+    for wxid in wxid_candidates:
+        for code in codes:
+            xor_key, aes_key = derive_image_keys(code, wxid)
+            if verify_aes_key_against_all(aes_key, templates):
+                print()
+                print("[✓] 验证成功（所有模板均通过）:", flush=True)
+                print(f"    code     = {code}", flush=True)
+                print(f"    wxid     = {wxid}", flush=True)
+                print(f"    xor_key  = 0x{xor_key:02x}", flush=True)
+                print(f"    aes_key  = {aes_key}", flush=True)
+                return xor_key, aes_key
+
+    print()
+    print("[!] 所有 (wxid × code) 组合都未通过交叉验证", flush=True)
+    print("    可能原因：微信版本变更了派生算法 / 缓存已失效 / 模板文件损坏",
+          flush=True)
+    return None
+
+
+def _save_config_atomic(config_path, config):
+    """原子写 config.json：tmp + os.replace 防止中断留下半截文件。
+
+    若 json.dump 或 os.replace 抛错，向上抛出（让 main 给出 stacktrace
+    而不是默默写坏 config）；同时清理可能残留的 .tmp 文件。
+    """
+    tmp_path = config_path + ".tmp"
+    try:
+        with open(tmp_path, "w", encoding="utf-8") as f:
+            json.dump(config, f, indent=2, ensure_ascii=False)
+        os.replace(tmp_path, config_path)
+    finally:
+        # 失败路径上 .tmp 可能残留；成功路径上 os.replace 已经把 tmp 移走了
+        if os.path.exists(tmp_path):
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+
+
+def main(config_path=None):
+    """CLI 入口。`config_path` 默认是脚本同目录下的 config.json，
+    暴露此参数主要为方便单元测试注入隔离的临时配置。"""
+    if platform.system().lower() != "darwin":
+        print("此脚本只在 macOS 上工作。其他平台请用 find_image_key.py（内存扫描）。",
+              file=sys.stderr, flush=True)
+        sys.exit(1)
+
+    if config_path is None:
+        config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                                   "config.json")
+    try:
+        with open(config_path, encoding="utf-8") as f:
+            config = json.load(f)
+    except (OSError, json.JSONDecodeError) as e:
+        print(f"[!] 读取 {config_path} 失败: {e}", file=sys.stderr, flush=True)
+        sys.exit(1)
+
+    db_dir = config.get("db_dir", "")
+    if not db_dir:
+        print("[!] config.json 中未配置 db_dir", file=sys.stderr, flush=True)
+        sys.exit(1)
+    print(f"[*] db_dir = {db_dir}", flush=True)
+
+    # 短路：如果已有 image_aes_key 且仍能在所有模板上验证通过，直接退出
+    # （沿用 find_image_key.py 的 UX 约定，避免无谓重写 config.json）
+    existing_aes = config.get("image_aes_key")
+    if existing_aes:
+        base_dir = os.path.dirname(db_dir)
+        attach_dir = os.path.join(base_dir, "msg", "attach")
+        templates = find_v2_template_ciphertexts(attach_dir)
+        if templates and verify_aes_key_against_all(existing_aes, templates):
+            print(f"[+] 已有 image_aes_key={existing_aes} 在 "
+                  f"{len(templates)} 个模板上仍然有效，无需重新派生", flush=True)
+            return
+
+    result = find_image_key_macos(db_dir)
+    if result is None:
+        sys.exit(1)
+
+    xor_key, aes_key = result
+    config["image_aes_key"] = aes_key
+    config["image_xor_key"] = xor_key
+    _save_config_atomic(config_path, config)
+    print()
+    print(f"[+] 已写入 {config_path}", flush=True)
+    print("    下次启动 monitor_web.py 时会自动加载新密钥，图片消息显示内联预览",
+          flush=True)
+
+
+if __name__ == "__main__":
+    main()

tests/test_find_image_key_macos.py

@@ -0,0 +1,488 @@
+"""单元测试：find_image_key_macos 派生算法 + 端到端 smoke。
+
+不依赖真实微信数据；用 tempdir + 合成密文构造测试。
+"""
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from Crypto.Cipher import AES
+
+import find_image_key_macos as fkm
+
+
+class NormalizeWxidTests(unittest.TestCase):
+    def test_wxid_with_extra_segments_keeps_only_first(self):
+        # wxid_<seg> 形式只保留第一段下划线之内的内容
+        self.assertEqual(fkm.normalize_wxid("wxid_abc123_extra_more"), "wxid_abc123")
+
+    def test_wxid_no_extra_segments(self):
+        self.assertEqual(fkm.normalize_wxid("wxid_abc123"), "wxid_abc123")
+
+    def test_account_with_4char_alnum_suffix_stripped(self):
+        # macOS 路径常见：your_wxid_a1b2c3 → your_wxid
+        self.assertEqual(fkm.normalize_wxid("your_wxid_a1b2c3"), "your_wxid")
+
+    def test_account_without_recognizable_suffix_returned_asis(self):
+        self.assertEqual(fkm.normalize_wxid("simple"), "simple")
+        self.assertEqual(fkm.normalize_wxid("foo_bar_baz"), "foo_bar_baz")  # baz 是 3 char
+
+    def test_empty_or_none_returns_empty(self):
+        self.assertEqual(fkm.normalize_wxid(""), "")
+        self.assertEqual(fkm.normalize_wxid(None), "")
+        self.assertEqual(fkm.normalize_wxid("   "), "")
+
+
+class DeriveImageKeysTests(unittest.TestCase):
+    def test_xor_is_low_byte_of_code(self):
+        xor, _ = fkm.derive_image_keys(0x12345678, "anything")
+        self.assertEqual(xor, 0x78)
+
+    def test_xor_handles_small_codes(self):
+        self.assertEqual(fkm.derive_image_keys(0xFF, "x")[0], 0xFF)
+        self.assertEqual(fkm.derive_image_keys(0x00, "x")[0], 0x00)
+
+    def test_aes_is_md5_hex_truncated_to_16(self):
+        # Golden value：来自 POC 在真实微信数据上的验证（issue #23 解的就是这一对）
+        xor, aes = fkm.derive_image_keys(18709375, "your_wxid")
+        self.assertEqual(xor, 0x7F)
+        self.assertEqual(aes, "b73bd4126969d30f")
+
+    def test_aes_does_not_normalize_wxid_internally(self):
+        # 归一化由调用方负责；不同 wxid 字符串产出不同 key
+        _, aes_full = fkm.derive_image_keys(18709375, "your_wxid_a1b2c3")
+        _, aes_norm = fkm.derive_image_keys(18709375, "your_wxid")
+        self.assertNotEqual(aes_full, aes_norm)
+
+
+class DeriveKvcommDirCandidatesTests(unittest.TestCase):
+    def test_canonical_macos_path_is_first_candidate(self):
+        db_dir = (
+            "/Users/x/Library/Containers/com.tencent.xinWeChat/Data/Documents/"
+            "xwechat_files/wxid_abc/db_storage"
+        )
+        candidates = fkm.derive_kvcomm_dir_candidates(db_dir)
+        self.assertGreater(len(candidates), 0)
+        expected_primary = (
+            "/Users/x/Library/Containers/com.tencent.xinWeChat/Data/Documents/"
+            "app_data/net/kvcomm"
+        )
+        self.assertEqual(candidates[0], expected_primary)
+
+    def test_returns_multiple_candidates(self):
+        # 多候选是 Round 1 review 的关键修复点：跨版本路径覆盖
+        db_dir = (
+            "/Users/x/Library/Containers/com.tencent.xinWeChat/Data/Documents/"
+            "xwechat_files/wxid_abc/db_storage"
+        )
+        candidates = fkm.derive_kvcomm_dir_candidates(db_dir)
+        self.assertGreaterEqual(len(candidates), 3,
+                                "应返回多个候选路径以覆盖不同微信版本布局")
+
+    def test_no_xwechat_files_still_returns_home_fallback(self):
+        # 即使无法从 db_dir 推算，也至少返回 HOME 默认路径作兜底
+        candidates = fkm.derive_kvcomm_dir_candidates("/random/path")
+        self.assertGreaterEqual(len(candidates), 1)
+        self.assertTrue(any("Containers/com.tencent.xinWeChat" in c
+                            for c in candidates))
+
+    def test_candidates_are_unique(self):
+        db_dir = "/x/y/Documents/xwechat_files/wxid_abc/db_storage"
+        candidates = fkm.derive_kvcomm_dir_candidates(db_dir)
+        self.assertEqual(len(candidates), len(set(candidates)))
+
+
+class FindExistingKvcommDirTests(unittest.TestCase):
+    def test_returns_first_existing_candidate(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            # 构造合法 db_dir 路径，在第一个候选位置创建实际目录
+            base = os.path.join(tmp, "Documents", "xwechat_files", "wxid_x")
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+            kvcomm = os.path.join(tmp, "Documents", "app_data", "net", "kvcomm")
+            os.makedirs(kvcomm)
+
+            self.assertEqual(fkm.find_existing_kvcomm_dir(db_dir), kvcomm)
+
+    def test_returns_none_when_no_candidate_exists(self):
+        # 即使 HOME fallback 候选也不存在时，应返回 None。
+        # 隔离测试不能依赖宿主机有/无微信安装；patch expanduser 指向 tmp。
+        with tempfile.TemporaryDirectory() as fake_home:
+            with patch("os.path.expanduser", return_value=fake_home):
+                self.assertIsNone(fkm.find_existing_kvcomm_dir("/nonexistent/x/y/z"))
+
+
+class CollectKvcommCodesTests(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.kvdir = self._tmp.name
+
+    def _touch(self, name):
+        with open(os.path.join(self.kvdir, name), "w") as f:
+            f.write("")
+
+    def test_extracts_code_from_filename(self):
+        self._touch("key_18709375_4066645761_1_1777096531_137785059_3600_input.statistic")
+        self._touch("key_99999999_yyy_zzz.statistic")
+        self.assertEqual(fkm.collect_kvcomm_codes(self.kvdir), [18709375, 99999999])
+
+    def test_ignores_files_with_non_numeric_first_segment(self):
+        self._touch("key_reportnow_18709375_xxx.statistic")
+        self._touch("key_abc_def.statistic")
+        self._touch("config.ini")
+        self._touch("monitordata_x")
+        self.assertEqual(fkm.collect_kvcomm_codes(self.kvdir), [])
+
+    def test_dedupes_same_code_across_files(self):
+        self._touch("key_42_a.statistic")
+        self._touch("key_42_b.statistic")
+        self.assertEqual(fkm.collect_kvcomm_codes(self.kvdir), [42])
+
+    def test_missing_dir_returns_empty(self):
+        self.assertEqual(fkm.collect_kvcomm_codes("/nonexistent/xxx"), [])
+
+    def test_none_dir_returns_empty(self):
+        self.assertEqual(fkm.collect_kvcomm_codes(None), [])
+
+
+class CollectWxidCandidatesTests(unittest.TestCase):
+    def test_returns_raw_and_normalized_when_different(self):
+        db_dir = "/x/Documents/xwechat_files/your_wxid_a1b2c3/db_storage"
+        self.assertEqual(fkm.collect_wxid_candidates(db_dir),
+                         ["your_wxid_a1b2c3", "your_wxid"])
+
+    def test_returns_one_when_normalize_is_identity(self):
+        db_dir = "/x/Documents/xwechat_files/wxid_abc/db_storage"
+        self.assertEqual(fkm.collect_wxid_candidates(db_dir), ["wxid_abc"])
+
+    def test_no_xwechat_files_returns_empty(self):
+        self.assertEqual(fkm.collect_wxid_candidates("/random/path"), [])
+
+    def test_xwechat_files_at_end_returns_empty(self):
+        self.assertEqual(fkm.collect_wxid_candidates("/x/xwechat_files"), [])
+
+
+class VerifyAesKeyTests(unittest.TestCase):
+    KEY = "b73bd4126969d30f"
+
+    def _encrypt(self, plaintext_16):
+        return AES.new(self.KEY.encode("ascii"), AES.MODE_ECB).encrypt(plaintext_16)
+
+    def test_jpeg_magic_passes(self):
+        ct = self._encrypt(b"\xff\xd8\xff\xe0" + b"\x00" * 12)
+        self.assertTrue(fkm.verify_aes_key(self.KEY, ct))
+
+    def test_png_magic_passes(self):
+        ct = self._encrypt(b"\x89PNG\r\n\x1a\n" + b"\x00" * 8)
+        self.assertTrue(fkm.verify_aes_key(self.KEY, ct))
+
+    def test_gif_magic_passes(self):
+        ct = self._encrypt(b"GIF89a" + b"\x00" * 10)
+        self.assertTrue(fkm.verify_aes_key(self.KEY, ct))
+
+    def test_wxgf_magic_passes(self):
+        ct = self._encrypt(b"wxgf" + b"\x00" * 12)
+        self.assertTrue(fkm.verify_aes_key(self.KEY, ct))
+
+    def test_random_data_fails(self):
+        self.assertFalse(fkm.verify_aes_key(self.KEY, bytes(range(16))))
+
+    def test_wrong_length_template_fails(self):
+        self.assertFalse(fkm.verify_aes_key(self.KEY, b"short"))
+        self.assertFalse(fkm.verify_aes_key(self.KEY, b""))
+
+    def test_short_aes_key_fails(self):
+        self.assertFalse(fkm.verify_aes_key("short", b"\x00" * 16))
+
+    def test_empty_aes_key_fails(self):
+        self.assertFalse(fkm.verify_aes_key("", b"\x00" * 16))
+
+
+class VerifyAesKeyAgainstAllTests(unittest.TestCase):
+    """交叉验证：必须所有模板都通过才算命中（防短 magic 偶然碰撞）。"""
+
+    KEY = "b73bd4126969d30f"
+
+    def _encrypt(self, plaintext_16):
+        return AES.new(self.KEY.encode("ascii"), AES.MODE_ECB).encrypt(plaintext_16)
+
+    def test_all_templates_pass(self):
+        ct1 = self._encrypt(b"\xff\xd8\xff\xe0" + b"\x00" * 12)
+        ct2 = self._encrypt(b"\x89PNG\r\n\x1a\n" + b"\x00" * 8)
+        self.assertTrue(fkm.verify_aes_key_against_all(self.KEY, [ct1, ct2]))
+
+    def test_one_template_fails_overall_fails(self):
+        ct1 = self._encrypt(b"\xff\xd8\xff\xe0" + b"\x00" * 12)  # passes
+        ct2 = bytes(range(16))                                    # random, fails
+        self.assertFalse(fkm.verify_aes_key_against_all(self.KEY, [ct1, ct2]))
+
+    def test_empty_template_list_returns_false(self):
+        # 没模板就不能验证；不视为通过（防"零样本=自动通过"陷阱）
+        self.assertFalse(fkm.verify_aes_key_against_all(self.KEY, []))
+
+
+class FindV2TemplateCiphertextsTests(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.dir = self._tmp.name
+
+    def _build_v2_dat(self, name, ciphertext_16, subdir=""):
+        target_dir = os.path.join(self.dir, subdir) if subdir else self.dir
+        os.makedirs(target_dir, exist_ok=True)
+        path = os.path.join(target_dir, name)
+        with open(path, "wb") as f:
+            f.write(fkm.V2_MAGIC + b"\x00" * 9 + ciphertext_16 + b"\x00\x00")
+        return path
+
+    def test_finds_one_template_in_v2_thumb(self):
+        ct = bytes(range(0xF, 0x1F))
+        self._build_v2_dat("abc_t.dat", ct)
+        result = fkm.find_v2_template_ciphertexts(self.dir)
+        self.assertEqual(result, [ct])
+
+    def test_finds_multiple_distinct_templates(self):
+        cts = [bytes([i] * 16) for i in (0x11, 0x22, 0x33)]
+        for i, ct in enumerate(cts):
+            self._build_v2_dat(f"chat{i}_t.dat", ct, subdir=f"chat{i}")
+        result = fkm.find_v2_template_ciphertexts(self.dir, max_templates=3)
+        self.assertEqual(set(result), set(cts))
+
+    def test_dedupes_identical_templates(self):
+        ct = b"\x42" * 16
+        self._build_v2_dat("a_t.dat", ct, subdir="a")
+        self._build_v2_dat("b_t.dat", ct, subdir="b")
+        result = fkm.find_v2_template_ciphertexts(self.dir)
+        self.assertEqual(result, [ct])
+
+    def test_falls_back_to_any_dat_if_no_thumb(self):
+        ct = b"\x33" * 16
+        self._build_v2_dat("only_full.dat", ct)
+        self.assertEqual(fkm.find_v2_template_ciphertexts(self.dir), [ct])
+
+    def test_skips_non_v2_files(self):
+        path = os.path.join(self.dir, "abc_t.dat")
+        with open(path, "wb") as f:
+            f.write(b"\x00" * 100)
+        self.assertEqual(fkm.find_v2_template_ciphertexts(self.dir), [])
+
+    def test_empty_dir_returns_empty(self):
+        self.assertEqual(fkm.find_v2_template_ciphertexts(self.dir), [])
+
+    def test_missing_dir_returns_empty(self):
+        self.assertEqual(fkm.find_v2_template_ciphertexts("/nonexistent"), [])
+
+    def test_walks_into_subdirs(self):
+        ct = b"\x44" * 16
+        self._build_v2_dat("x_t.dat", ct, subdir="sub/deeper")
+        self.assertEqual(fkm.find_v2_template_ciphertexts(self.dir), [ct])
+
+    def test_respects_max_templates(self):
+        cts = [bytes([i] * 16) for i in range(10)]
+        for i, ct in enumerate(cts):
+            self._build_v2_dat(f"x{i}_t.dat", ct, subdir=f"d{i}")
+        result = fkm.find_v2_template_ciphertexts(self.dir, max_templates=2)
+        self.assertEqual(len(result), 2)
+
+
+class FindImageKeyMacosIntegrationTests(unittest.TestCase):
+    """端到端集成：合成 kvcomm 文件 + 合成 V2 模板 → 期望派生出已知 key。"""
+
+    def _build_test_env(self, tmpdir, code, wxid_raw, num_templates=2):
+        """构造测试环境，返回 (db_dir, expected_xor, expected_aes)。"""
+        wxid_norm = fkm.normalize_wxid(wxid_raw)
+        base = os.path.join(tmpdir, "Documents", "xwechat_files", wxid_raw)
+        db_dir = os.path.join(base, "db_storage")
+        os.makedirs(db_dir)
+
+        kvcomm = os.path.join(tmpdir, "Documents", "app_data", "net", "kvcomm")
+        os.makedirs(kvcomm)
+        with open(os.path.join(kvcomm, f"key_{code}_x.statistic"), "w") as f:
+            f.write("")
+
+        xor_expected, aes_expected = fkm.derive_image_keys(code, wxid_norm)
+        # 多个模板用不同的 plaintext 加密（仍是图像 magic 开头但内容不同）
+        plaintexts = [
+            b"\xff\xd8\xff\xe0" + b"\x00" * 12,           # JPEG
+            b"\x89PNG\r\n\x1a\n" + b"\x00" * 8,           # PNG
+            b"GIF89a" + b"\x01\x02" + b"\x00" * 8,        # GIF
+        ]
+        for i in range(num_templates):
+            pt = plaintexts[i % len(plaintexts)]
+            ct = AES.new(aes_expected.encode("ascii"), AES.MODE_ECB).encrypt(pt)
+            attach = os.path.join(base, "msg", "attach", f"chat{i}")
+            os.makedirs(attach)
+            with open(os.path.join(attach, f"img{i}_t.dat"), "wb") as f:
+                f.write(fkm.V2_MAGIC + b"\x00" * 9 + ct + b"\x00\x00")
+        return db_dir, xor_expected, aes_expected
+
+    def test_full_flow_succeeds_with_normalized_wxid(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db_dir, xor_exp, aes_exp = self._build_test_env(
+                tmp, code=18709375, wxid_raw="your_wxid_a1b2c3", num_templates=3)
+            result = fkm.find_image_key_macos(db_dir)
+            self.assertIsNotNone(result, "派生应该成功")
+            self.assertEqual(result, (xor_exp, aes_exp))
+
+    def test_returns_none_when_no_kvcomm_codes(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = os.path.join(tmp, "Documents", "xwechat_files", "wxid_x")
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+            self.assertIsNone(fkm.find_image_key_macos(db_dir))
+
+    def test_returns_none_when_no_v2_template(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = os.path.join(tmp, "Documents", "xwechat_files", "wxid_x")
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+            kvcomm = os.path.join(tmp, "Documents", "app_data", "net", "kvcomm")
+            os.makedirs(kvcomm)
+            with open(os.path.join(kvcomm, "key_42_x.statistic"), "w") as f:
+                f.write("")
+            self.assertIsNone(fkm.find_image_key_macos(db_dir))
+
+    def test_returns_none_when_no_combination_verifies(self):
+        # 有 code 也有 V2 .dat，但密文是随机的，没有任何 key 能解出
+        with tempfile.TemporaryDirectory() as tmp:
+            base = os.path.join(tmp, "Documents", "xwechat_files", "wxid_x")
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+            kvcomm = os.path.join(tmp, "Documents", "app_data", "net", "kvcomm")
+            os.makedirs(kvcomm)
+            with open(os.path.join(kvcomm, "key_42_x.statistic"), "w") as f:
+                f.write("")
+            attach = os.path.join(base, "msg", "attach", "x")
+            os.makedirs(attach)
+            with open(os.path.join(attach, "x_t.dat"), "wb") as f:
+                f.write(fkm.V2_MAGIC + b"\x00" * 9 + b"\xde\xad\xbe\xef" * 4 + b"\x00\x00")
+            self.assertIsNone(fkm.find_image_key_macos(db_dir))
+
+    def test_empty_db_dir_returns_none_without_crash(self):
+        # 防御：空字符串、不合理路径不应抛异常。
+        # patch expanduser 让 HOME fallback 也指向不存在的路径，避免
+        # 测试在装了真实微信的开发机上意外深入到 wxid 缺失分支。
+        with tempfile.TemporaryDirectory() as fake_home:
+            with patch("os.path.expanduser", return_value=fake_home):
+                self.assertIsNone(fkm.find_image_key_macos(""))
+
+
+class MainShortCircuitTests(unittest.TestCase):
+    """main() 短路：已有 image_aes_key 仍然有效时，不应重新派生 / 不应改写 config。"""
+
+    def test_existing_valid_key_skips_derivation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            wxid = "wxid_abc"
+            base = os.path.join(tmp, "Documents", "xwechat_files", wxid)
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+
+            # kvcomm 里放个 code，证明若真去派生也能算出 key
+            kvcomm = os.path.join(tmp, "Documents", "app_data", "net", "kvcomm")
+            os.makedirs(kvcomm)
+            code = 42
+            with open(os.path.join(kvcomm, f"key_{code}_x.statistic"), "w") as f:
+                f.write("")
+
+            # 用真实派生的 key 加密 V2 模板，使现有 key 在该模板上能验证通过
+            xor_exp, aes_exp = fkm.derive_image_keys(code, wxid)
+            jpeg_pt = b"\xff\xd8\xff\xe0" + b"\x00" * 12
+            ct = AES.new(aes_exp.encode("ascii"), AES.MODE_ECB).encrypt(jpeg_pt)
+            attach = os.path.join(base, "msg", "attach", "x")
+            os.makedirs(attach)
+            with open(os.path.join(attach, "test_t.dat"), "wb") as f:
+                f.write(fkm.V2_MAGIC + b"\x00" * 9 + ct + b"\x00\x00")
+
+            # 写入"已有有效 key"的 config
+            cfg_path = os.path.join(tmp, "config.json")
+            cfg_initial = {
+                "db_dir": db_dir,
+                "image_aes_key": aes_exp,
+                "image_xor_key": xor_exp,
+                "extra_field": "must_be_preserved",  # 证明 main 不会重写
+            }
+            with open(cfg_path, "w", encoding="utf-8") as f:
+                json.dump(cfg_initial, f)
+            mtime_before = os.path.getmtime(cfg_path)
+
+            # 关键：patch find_image_key_macos 让它若被误调用立刻可见
+            with patch.object(fkm, "find_image_key_macos") as mock_derive:
+                fkm.main(config_path=cfg_path)
+
+            mock_derive.assert_not_called()  # 短路应直接 return，不进派生
+            # config.json 不应被重写
+            self.assertEqual(os.path.getmtime(cfg_path), mtime_before)
+            with open(cfg_path, encoding="utf-8") as f:
+                self.assertEqual(json.load(f), cfg_initial)
+
+    def test_existing_invalid_key_falls_through_to_derivation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            wxid = "wxid_abc"
+            base = os.path.join(tmp, "Documents", "xwechat_files", wxid)
+            db_dir = os.path.join(base, "db_storage")
+            os.makedirs(db_dir)
+
+            kvcomm = os.path.join(tmp, "Documents", "app_data", "net", "kvcomm")
+            os.makedirs(kvcomm)
+            code = 42
+            with open(os.path.join(kvcomm, f"key_{code}_x.statistic"), "w") as f:
+                f.write("")
+
+            xor_exp, aes_exp = fkm.derive_image_keys(code, wxid)
+            jpeg_pt = b"\xff\xd8\xff\xe0" + b"\x00" * 12
+            ct = AES.new(aes_exp.encode("ascii"), AES.MODE_ECB).encrypt(jpeg_pt)
+            attach = os.path.join(base, "msg", "attach", "x")
+            os.makedirs(attach)
+            with open(os.path.join(attach, "test_t.dat"), "wb") as f:
+                f.write(fkm.V2_MAGIC + b"\x00" * 9 + ct + b"\x00\x00")
+
+            cfg_path = os.path.join(tmp, "config.json")
+            cfg_initial = {
+                "db_dir": db_dir,
+                "image_aes_key": "deadbeefdeadbeef",  # 故意写一个错的
+            }
+            with open(cfg_path, "w", encoding="utf-8") as f:
+                json.dump(cfg_initial, f)
+
+            fkm.main(config_path=cfg_path)
+
+            # 短路应失败，进入派生路径，配置应被改写为正确的 key
+            with open(cfg_path, encoding="utf-8") as f:
+                cfg_after = json.load(f)
+            self.assertEqual(cfg_after["image_aes_key"], aes_exp)
+            self.assertEqual(cfg_after["image_xor_key"], xor_exp)
+
+
+class SaveConfigAtomicTests(unittest.TestCase):
+    """原子写测试：os.replace 保证 config.json 不会被半截覆盖。"""
+
+    def test_roundtrip_writes_pretty_utf8(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            cfg_path = os.path.join(tmp, "config.json")
+            cfg = {"db_dir": "/x", "image_aes_key": "中文测试key"}
+            fkm._save_config_atomic(cfg_path, cfg)
+            with open(cfg_path, encoding="utf-8") as f:
+                self.assertEqual(json.load(f), cfg)
+            # ensure_ascii=False：中文应直接落盘，不被转义
+            with open(cfg_path, "rb") as f:
+                self.assertIn("中文测试key".encode("utf-8"), f.read())
+
+    def test_failed_replace_leaves_original_intact(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            cfg_path = os.path.join(tmp, "config.json")
+            with open(cfg_path, "w", encoding="utf-8") as f:
+                json.dump({"original": True}, f)
+            with patch.object(os, "replace",
+                              side_effect=OSError("disk full during rename")):
+                with self.assertRaises(OSError):
+                    fkm._save_config_atomic(cfg_path, {"new": True})
+            # 原文件应保持不变
+            with open(cfg_path, encoding="utf-8") as f:
+                self.assertEqual(json.load(f), {"original": True})
+
+
+if __name__ == "__main__":
+    unittest.main()