mirror
/
wx-cli
mirror of https://github.com/jackwener/wx-cli.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge PR #18: feat: Linux 数据库解密支持 

- 新增 find_all_keys_linux.py (通过 /proc/pid/mem 扫描密钥)
- 新增 key_utils.py (跨平台路径兼容)
- 新增 key_scan_common.py (公共扫描逻辑)
- 拆分 find_all_keys.py 为平台分发入口
- 所有下游模块统一使用 get_key_info() 查找密钥

Fixes #12 (部分: Linux 支持)
Co-authored-by: PeanutSplash <b1300658700@outlook.com>
feat/daemon-cli
ylytdeng 2026-03-07 21:35:37 +08:00
parent 5879b58239 30112b9a10
commit a5a347f69e
13 changed files with 1188 additions and 721 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

481
README.md
View File

@ -1,238 +1,243 @@
# WeChat 4.0 Database Decryptor # WeChat 4.x Database Decryptor
微信 4.0 (Windows) 本地数据库解密工具。从运行中的微信进程内存提取加密密钥，解密所有 SQLCipher 4 加密数据库，并提供实时消息监听。 微信 4.0 (Windows、MacOS、Linux) 本地数据库解密工具。从运行中的微信进程内存提取加密密钥，解密所有 SQLCipher 4 加密数据库，并提供实时消息监听。
## 更新日志 ## 更新日志
### 2025-03-03 — 富媒体内容 & 组合消息修复 ### 2025-03-03 — 富媒体内容 & 组合消息修复
- **表情包内联显示**: 自动从 emoticon.db 构建 MD5→CDN 映射，支持自定义表情（NonStore）和商店表情（Store），CDN 下载后本地缓存 - **表情包内联显示**: 自动从 emoticon.db 构建 MD5→CDN 映射，支持自定义表情（NonStore）和商店表情（Store），CDN 下载后本地缓存
- **富媒体内容解析**: 链接卡片（type 49）、文件、视频号、小程序、引用回复、位置分享等在 Web UI 中完整渲染 - **富媒体内容解析**: 链接卡片（type 49）、文件、视频号、小程序、引用回复、位置分享等在 Web UI 中完整渲染
- **文字+图片组合消息不再丢失**: 修复同时发送文字和图片时只显示最后一条的问题（前端去重 key 增加消息类型） - **文字+图片组合消息不再丢失**: 修复同时发送文字和图片时只显示最后一条的问题（前端去重 key 增加消息类型）
- **隐藏消息检测**: 新增 `_check_hidden_messages` 机制，session.db 只保存最后一条消息摘要，现在会异步查 message DB 找回同一秒内的其他消息 - **隐藏消息检测**: 新增 `_check_hidden_messages` 机制，session.db 只保存最后一条消息摘要，现在会异步查 message DB 找回同一秒内的其他消息
- **MonitorDBCache 线程安全**: 引入 per-key 锁，防止多线程并发解密同一数据库导致文件损坏 - **MonitorDBCache 线程安全**: 引入 per-key 锁，防止多线程并发解密同一数据库导致文件损坏
- **Web UI 改进**: 消息气泡样式优化、群聊发送者显示、图片缩略图点击放大 - **Web UI 改进**: 消息气泡样式优化、群聊发送者显示、图片缩略图点击放大
## 原理 ## 原理
微信 4.0 使用 SQLCipher 4 加密本地数据库： 微信 4.0 使用 SQLCipher 4 加密本地数据库：
- **加密算法**: AES-256-CBC + HMAC-SHA512 - **加密算法**: AES-256-CBC + HMAC-SHA512
- **KDF**: PBKDF2-HMAC-SHA512, 256,000 iterations - **KDF**: PBKDF2-HMAC-SHA512, 256,000 iterations
- **页面大小**: 4096 bytes, reserve = 80 (IV 16 + HMAC 64) - **页面大小**: 4096 bytes, reserve = 80 (IV 16 + HMAC 64)
- **每个数据库有独立的 salt 和 enc_key** - **每个数据库有独立的 salt 和 enc_key**
WCDB (微信的 SQLCipher 封装) 会在进程内存中缓存派生后的 raw key，格式为 `x'<64hex_enc_key><32hex_salt>'`。本工具通过扫描进程内存中的这种模式，匹配数据库文件的 salt，并通过 HMAC 验证来提取正确的密钥。 WCDB (微信的 SQLCipher 封装) 会在进程内存中缓存派生后的 raw key，格式为 `x'<64hex_enc_key><32hex_salt>'`。三个平台（Windows / Linux / macOS）均可通过扫描进程内存匹配此模式，再通过 HMAC 校验 page 1 确认密钥正确性。
## 使用方法 ## 使用方法
### 环境要求 ### 环境要求
- Windows 10/11 - Python 3.10+
- Python 3.10+ - 微信 4.x
- 微信 4.0 (正在运行) - `pip install pycryptodome`
- 需要管理员权限 (读取进程内存)
Windows：
### 安装依赖
- Windows 10/11
```bash - 微信正在运行
pip install pycryptodome - 需要管理员权限（读取进程内存）
```
Linux：
### 快速开始
- 64-bit Linux
确保微信正在运行，以**管理员权限**执行： - 需要 root 权限或 `CAP_SYS_PTRACE`（读取 `/proc/<pid>/mem`）
- `db_dir` 默认类似 `~/Documents/xwechat_files/<wxid>/db_storage`
```bash
python main.py # 实时消息监听 (Web UI) ### 安装依赖
python main.py decrypt # 解密全部数据库到 decrypted/
``` ```bash
pip install pycryptodome
程序会自动完成：配置检测 → 密钥提取 → 启动。首次运行会自动检测微信数据目录并生成 `config.json`。 ```
如果自动检测失败（例如微信安装在非默认位置），手动创建 `config.json`： ### 快速开始
```json
{ Windows：
"db_dir": "D:\\xwechat_files\\你的微信ID\\db_storage",
"keys_file": "all_keys.json", ```bash
"decrypted_dir": "decrypted", python main.py
"wechat_process": "Weixin.exe" python main.py decrypt
} ```
```
Linux：
`db_dir` 路径可以在 微信设置 → 文件管理 中找到。
```bash
### Web UI 说明 python3 main.py decrypt
```
`python main.py` 启动后打开 http://localhost:5678 查看实时消息流。
程序会自动完成：配置检测 → 内存扫描提取密钥 → 解密。首次运行会自动检测微信数据目录并生成 `config.json`。微信只要在运行中即可，无需重启或重新登录。
- 30ms 轮询 WAL 文件变化 (mtime)
- 检测到变化后全量解密 + WAL patch (~70ms) 如果自动检测失败（例如微信安装在非默认位置），手动创建 `config.json`：
- SSE 实时推送到浏览器 ```json
- 总延迟约 100ms {
- **图片消息内联预览**（支持旧 XOR / V1 / V2 三种 .dat 加密格式） "db_dir": "D:\\xwechat_files\\你的微信ID\\db_storage",
"keys_file": "all_keys.json",
### MCP Server (Claude AI 集成) "decrypted_dir": "decrypted",
"wechat_process": "Weixin.exe"
将微信数据查询能力接入 [Claude Code](https://claude.ai/claude-code)，让 AI 直接读取你的微信消息。 }
```
```bash
pip install mcp Linux 版 `config.json` 示例：
```
```json
注册到 Claude Code： {
"db_dir": "/home/yourname/Documents/xwechat_files/your_wxid/db_storage",
```bash "keys_file": "all_keys.json",
claude mcp add wechat -- python C:\Users\你的用户名\wechat-decrypt\mcp_server.py "decrypted_dir": "decrypted",
``` "wechat_process": "wechat"
}
或手动编辑 `~/.claude.json`： ```
```json `db_dir` 路径：Windows 可在微信设置 → 文件管理中找到；Linux 默认在 `~/Documents/xwechat_files/<wxid>/db_storage`。
{
"mcpServers": { ### Web UI 说明
"wechat": {
"type": "stdio", `python main.py` 启动后打开 http://localhost:5678 查看实时消息流。
"command": "python",
"args": ["C:\\Users\\你的用户名\\wechat-decrypt\\mcp_server.py"] - 30ms 轮询 WAL 文件变化 (mtime)
} - 检测到变化后全量解密 + WAL patch (~70ms)
} - SSE 实时推送到浏览器
} - 总延迟约 100ms
``` - **图片消息内联预览**（支持旧 XOR / V1 / V2 三种 .dat 加密格式）
注册后在 Claude Code 中即可使用以下工具： ### MCP Server (Claude AI 集成)
| Tool | 功能 | 将微信数据查询能力接入 [Claude Code](https://claude.ai/claude-code)，让 AI 直接读取你的微信消息。
|------|------|
| `get_recent_sessions(limit)` | 最近会话列表（含消息摘要、未读数） | ```bash
| `get_chat_history(chat_name, limit)` | 指定聊天的消息记录（支持模糊匹配名字） | pip install mcp
| `search_messages(keyword, limit)` | 全库搜索消息内容 | ```
| `get_contacts(query, limit)` | 搜索/列出联系人 |
| `get_new_messages()` | 获取自上次调用以来的新消息 | 注册到 Claude Code：
前置条件：需要先运行 `python main.py` 或 `python find_all_keys.py` 完成密钥提取。 ```bash
claude mcp add wechat -- python C:\Users\你的用户名\wechat-decrypt\mcp_server.py
**[查看使用案例 →](USAGE.md)** ```
### 图片解密 (V2 格式) 或手动编辑 `~/.claude.json`：
微信 4.0 (2025-08+) 的 .dat 图片文件使用 AES-128-ECB + XOR 混合加密 (V2 格式)。AES 密钥需要从运行中的微信进程内存中提取： ```json
{
```bash "mcpServers": {
# 1. 在微信中打开查看 2-3 张图片（点击看大图） "wechat": {
# 2. 立即运行密钥提取（持续监控版）： "type": "stdio",
python find_image_key_monitor.py "command": "python",
"args": ["C:\\Users\\你的用户名\\wechat-decrypt\\mcp_server.py"]
# 或单次扫描版： }
python find_image_key.py }
``` }
```
密钥会自动保存到 `config.json` 的 `image_aes_key` 字段。之后 `monitor_web.py` 启动时会自动加载密钥，图片消息将显示内联预览。
注册后在 Claude Code 中即可使用以下工具：
> **注意**: AES 密钥仅在微信查看图片时临时加载到内存中。如果扫描未找到密钥，请先在微信中查看几张图片，然后立即重新运行脚本。
| Tool | 功能 |
#### macOS 图片解密 |------|------|
| `get_recent_sessions(limit)` | 最近会话列表（含消息摘要、未读数） |
macOS 上使用 C 版工具（通过 Mach VM API + CommonCrypto，性能比 Python 高 100 倍）： | `get_chat_history(chat_name, limit)` | 指定聊天的消息记录（支持模糊匹配名字） |
| `search_messages(keyword, limit)` | 全库搜索消息内容 |
**前置条件：** | `get_contacts(query, limit)` | 搜索/列出联系人 |
- Xcode Command Line Tools: `xcode-select --install` | `get_new_messages()` | 获取自上次调用以来的新消息 |
- 微信需要 ad-hoc 签名：`sudo codesign --force --deep --sign - /Applications/WeChat.app`
- 开发者模式：系统设置 → 隐私与安全 → 开发者模式 → 开启 前置条件：需要先运行 `python main.py` 或 `python find_all_keys.py` 完成密钥提取。
```bash **[查看使用案例 →](USAGE.md)**
# 编译
cc -O3 -o find_image_key find_image_key.c -framework Security ### 图片解密 (V2 格式)
cc -O3 -o decrypt_images decrypt_images.c -framework Security
微信 4.0 (2025-08+) 的 .dat 图片文件使用 AES-128-ECB + XOR 混合加密 (V2 格式)。AES 密钥需要从运行中的微信进程内存中提取：
# 1. 持续扫描图片密钥（在微信中浏览图片，扫描器自动捕获密钥）
sudo ./find_image_key ```bash
# 1. 在微信中打开查看 2-3 张图片（点击看大图）
# 2. 批量解密所有 V2 图片 # 2. 立即运行密钥提取（持续监控版）：
./decrypt_images python find_image_key_monitor.py
```
# 或单次扫描版：
`find_image_key` 会自动发现所有未解密的 V2 图片 pattern，持续扫描微信进程内存。当用户在微信中浏览图片时捕获密钥，保存到 `image_keys.json`。支持 `--deep` 模式进行逐字节深度扫描。 python find_image_key.py
```
## 文件说明
密钥会自动保存到 `config.json` 的 `image_aes_key` 字段。之后 `monitor_web.py` 启动时会自动加载密钥，图片消息将显示内联预览。
| 文件 | 说明 |
|------|------| > **注意**: AES 密钥仅在微信查看图片时临时加载到内存中。如果扫描未找到密钥，请先在微信中查看几张图片，然后立即重新运行脚本。
| `main.py` | **一键启动入口** — 自动配置、提取密钥、启动服务 |
| `config.py` | 配置加载器（自动检测微信数据目录） | ## 文件说明
| `find_all_keys.py` | 从微信进程内存提取所有数据库密钥 |
| `decrypt_db.py` | 全量解密所有数据库 | | 文件 | 说明 |
| `mcp_server.py` | MCP Server，让 Claude AI 查询微信数据 | |------|------|
| `monitor_web.py` | 实时消息监听 (Web UI + SSE + 图片预览) | | `main.py` | **一键启动入口** — 自动配置、提取密钥、启动服务 |
| `monitor.py` | 实时消息监听 (命令行) | | `config.py` | 配置加载器（自动检测微信数据目录） |
| `decode_image.py` | 图片 .dat 文件解密模块 (XOR / V1 / V2) | | `find_all_keys.py` | 平台分发入口（Windows / Linux） |
| `find_image_key.py` | 从微信进程内存提取图片 AES 密钥 | | `find_all_keys_windows.py` | Windows 版内存扫描提 key |
| `find_image_key_monitor.py` | 持续监控版密钥提取（推荐） | | `find_all_keys_linux.py` | Linux 版内存扫描提 key |
| `latency_test.py` | 延迟测量诊断工具 | | `decrypt_db.py` | 全量解密所有数据库 |
| `find_all_keys_macos.c` | macOS 版内存密钥扫描器 (C, Mach VM API) | | `mcp_server.py` | MCP Server，让 Claude AI 查询微信数据 |
| `find_image_key.c` | macOS 版图片密钥扫描器 (C, 持续监控模式) | | `monitor_web.py` | 实时消息监听 (Web UI + SSE + 图片预览) |
| `decrypt_images.c` | macOS 版批量图片解密器 (C, 多密钥支持) | | `monitor.py` | 实时消息监听 (命令行) |
| `decode_image.py` | 图片 .dat 文件解密模块 (XOR / V1 / V2) |
## 技术细节 | `find_image_key.py` | 从微信进程内存提取图片 AES 密钥 |
| `find_image_key_monitor.py` | 持续监控版密钥提取（推荐） |
### WAL 处理 | `latency_test.py` | 延迟测量诊断工具 |
| `find_all_keys_macos.c` | macOS 版内存密钥扫描器 (C, Mach VM API) |
微信使用 SQLite WAL 模式，WAL 文件是**预分配固定大小** (4MB)。检测变化时：
- 不能用文件大小 (永远不变) ## 技术细节
- 使用 mtime 检测写入
- 解密 WAL frame 时需校验 salt 值，跳过旧周期遗留的 frame ### WAL 处理
### 图片 .dat 加密格式 微信使用 SQLite WAL 模式，WAL 文件是**预分配固定大小** (4MB)。检测变化时：
- 不能用文件大小 (永远不变)
微信本地图片 (.dat) 有三种加密格式： - 使用 mtime 检测写入
- 解密 WAL frame 时需校验 salt 值，跳过旧周期遗留的 frame
| 格式 | 时期 | Magic | 加密方式 | 密钥来源 |
|------|------|-------|---------|---------| ### 图片 .dat 加密格式
| 旧 XOR | ~2025-07 | 无 | 单字节 XOR | 自动检测 (对比 magic bytes) |
| V1 | 过渡期 | `07 08 V1 08 07` | AES-ECB + XOR | 固定 key: `cfcd208495d565ef` | 微信本地图片 (.dat) 有三种加密格式：
| V2 | 2025-08+ | `07 08 V2 08 07` | AES-128-ECB + XOR | 从进程内存提取 |
| 格式 | 时期 | Magic | 加密方式 | 密钥来源 |
V2 文件结构: `[6B signature] [4B aes_size LE] [4B xor_size LE] [1B padding]` + `[AES-ECB encrypted] [raw unencrypted] [XOR encrypted]` |------|------|-------|---------|---------|
| 旧 XOR | ~2025-07 | 无 | 单字节 XOR | 自动检测 (对比 magic bytes) |
### 数据库结构 | V1 | 过渡期 | `07 08 V1 08 07` | AES-ECB + XOR | 固定 key: `cfcd208495d565ef` |
| V2 | 2025-08+ | `07 08 V2 08 07` | AES-128-ECB + XOR | 从进程内存提取 |
解密后包含约 26 个数据库：
- `session/session.db` - 会话列表 (最新消息摘要) V2 文件结构: `[6B signature] [4B aes_size LE] [4B xor_size LE] [1B padding]` + `[AES-ECB encrypted] [raw unencrypted] [XOR encrypted]`
- `message/message_*.db` - 聊天记录
- `contact/contact.db` - 联系人 ### 数据库结构
- `media_*/media_*.db` - 媒体文件索引
- 其他: head_image, favorite, sns, emoticon 等 解密后包含约 26 个数据库：
- `session/session.db` - 会话列表 (最新消息摘要)
## macOS 数据库密钥扫描 (WeChat 4.x) - `message/message_*.db` - 聊天记录
- `contact/contact.db` - 联系人
macOS 版微信 4.x 使用 SQLCipher 4 加密本地数据库，密钥格式为 `x'<64hex_key><32hex_salt>'`。C 版扫描器通过 Mach VM API 扫描微信进程内存提取密钥。 - `media_*/media_*.db` - 媒体文件索引
- 其他: head_image, favorite, sns, emoticon 等
### 前置条件
## macOS 数据库密钥扫描 (WeChat 4.x)
- macOS (Apple Silicon / Intel)
- WeChat 4.x (macOS 版) macOS 版微信 4.x 使用 SQLCipher 4 加密本地数据库，密钥格式为 `x'<64hex_key><32hex_salt>'`。C 版扫描器通过 Mach VM API 扫描微信进程内存提取密钥。
- Xcode Command Line Tools: `xcode-select --install`
- 微信需要 ad-hoc 签名（或安装了防撤回补丁）： ### 前置条件
`sudo codesign --force --deep --sign - /Applications/WeChat.app`
- macOS (Apple Silicon / Intel)
### 编译和使用 - WeChat 4.x (macOS 版)
- Xcode Command Line Tools: `xcode-select --install`
```bash - 微信需要 ad-hoc 签名（或安装了防撤回补丁）：
# 编译 `sudo codesign --force --deep --sign - /Applications/WeChat.app`
cc -O2 -o find_all_keys_macos find_all_keys_macos.c -framework Foundation
### 编译和使用
# 运行（自动查找微信进程、扫描内存、匹配 DB salt）
sudo ./find_all_keys_macos ```bash
# 编译
# 或指定 PID cc -O2 -o find_all_keys_macos find_all_keys_macos.c -framework Foundation
sudo ./find_all_keys_macos <pid>
``` # 运行（自动查找微信进程、扫描内存、匹配 DB salt）
sudo ./find_all_keys_macos
输出 `all_keys.json`，格式兼容 `decrypt_db.py`，可直接用于解密：
# 或指定 PID
```bash sudo ./find_all_keys_macos <pid>
python3 decrypt_db.py ```
```
输出 `all_keys.json`，格式兼容 `decrypt_db.py`，可直接用于解密：
## 免责声明
```bash
本工具仅用于学习和研究目的，用于解密**自己的**微信数据。请遵守相关法律法规，不要用于未经授权的数据访问。 python3 decrypt_db.py
```
## 免责声明
本工具仅用于学习和研究目的，用于解密**自己的**微信数据。请遵守相关法律法规，不要用于未经授权的数据访问。

12
config.example.json
View File

@ -1,6 +1,6 @@
{ {
"db_dir": "D:\\xwechat_files\\your_wxid\\db_storage", "db_dir": "D:\\xwechat_files\\your_wxid\\db_storage",
"keys_file": "all_keys.json", "keys_file": "all_keys.json",
"decrypted_dir": "decrypted", "decrypted_dir": "decrypted",
"wechat_process": "Weixin.exe" "wechat_process": "Weixin.exe"
} }

358
config.py
View File

@ -1,138 +1,220 @@
""" """
配置加载器 - 从 config.json 读取路径配置 配置加载器 - 从 config.json 读取路径配置
首次运行时自动检测微信数据目录，检测失败则提示手动配置 首次运行时自动检测微信数据目录，检测失败则提示手动配置
""" """
import glob import glob
import json import json
import os import os
import sys import platform
import sys
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
_DEFAULT_TEMPLATE_DIR = r"D:\xwechat_files\your_wxid\db_storage"
_SYSTEM = platform.system().lower()
_DEFAULT = {
"db_dir": _DEFAULT_TEMPLATE_DIR, if _SYSTEM == "linux":
"keys_file": "all_keys.json", _DEFAULT_TEMPLATE_DIR = os.path.expanduser("~/Documents/xwechat_files/your_wxid/db_storage")
"decrypted_dir": "decrypted", _DEFAULT_PROCESS = "wechat"
"decoded_image_dir": "decoded_images", elif _SYSTEM == "darwin":
"wechat_process": "Weixin.exe", # macOS 使用独立的 C 扫描器 (find_all_keys_macos.c)，此处仅提供 config 默认值
} _DEFAULT_TEMPLATE_DIR = os.path.expanduser("~/Documents/xwechat_files/your_wxid/db_storage")
_DEFAULT_PROCESS = "WeChat"
else:
def auto_detect_db_dir(): _DEFAULT_TEMPLATE_DIR = r"D:\xwechat_files\your_wxid\db_storage"
"""从微信本地配置自动检测 db_storage 路径。 _DEFAULT_PROCESS = "Weixin.exe"
读取 %APPDATA%\\Tencent\\xwechat\\config\\*.ini， _DEFAULT = {
找到数据存储根目录，然后匹配 xwechat_files\\*\\db_storage。 "db_dir": _DEFAULT_TEMPLATE_DIR,
""" "keys_file": "all_keys.json",
appdata = os.environ.get("APPDATA", "") "decrypted_dir": "decrypted",
config_dir = os.path.join(appdata, "Tencent", "xwechat", "config") "decoded_image_dir": "decoded_images",
if not os.path.isdir(config_dir): "wechat_process": _DEFAULT_PROCESS,
return None }
# 从 ini 文件中找到有效的目录路径
data_roots = [] def _choose_candidate(candidates):
for ini_file in glob.glob(os.path.join(config_dir, "*.ini")): """在多个候选目录中选择一个。"""
try: if len(candidates) == 1:
# 微信 ini 可能是 utf-8 或 gbk 编码（中文路径） return candidates[0]
content = None if len(candidates) > 1:
for enc in ("utf-8", "gbk"): if not sys.stdin.isatty():
try: return candidates[0]
with open(ini_file, "r", encoding=enc) as f: print("[!] 检测到多个微信数据目录（请选择当前正在运行的微信账号）:")
content = f.read(1024).strip() for i, c in enumerate(candidates, 1):
break print(f" {i}. {c}")
except UnicodeDecodeError: print(" 0. 跳过，稍后手动配置")
continue try:
if not content or any(c in content for c in "\n\r\x00"): while True:
continue choice = input("请选择 [0-{}]: ".format(len(candidates))).strip()
if os.path.isdir(content): if choice == "0":
data_roots.append(content) return None
except OSError: if choice.isdigit() and 1 <= int(choice) <= len(candidates):
continue return candidates[int(choice) - 1]
print(" 无效输入，请重新选择")
# 在每个根目录下搜索 xwechat_files\*\db_storage except (EOFError, KeyboardInterrupt):
seen = set() print()
candidates = [] return None
for root in data_roots: return None
pattern = os.path.join(root, "xwechat_files", "*", "db_storage")
for match in glob.glob(pattern):
normalized = os.path.normcase(os.path.normpath(match)) def _auto_detect_db_dir_windows():
if os.path.isdir(match) and normalized not in seen: """从微信本地配置自动检测 Windows db_storage 路径。
seen.add(normalized)
candidates.append(match) 读取 %APPDATA%\\Tencent\\xwechat\\config\\*.ini，
找到数据存储根目录，然后匹配 xwechat_files\\*\\db_storage。
if len(candidates) == 1: """
return candidates[0] appdata = os.environ.get("APPDATA", "")
if len(candidates) > 1: config_dir = os.path.join(appdata, "Tencent", "xwechat", "config")
# 非交互环境（MCP、无 stdin 管道等）直接取第一个 if not os.path.isdir(config_dir):
if not sys.stdin.isatty(): return None
return candidates[0]
print("[!] 检测到多个微信数据目录（请选择当前正在运行的微信账号）:") # 从 ini 文件中找到有效的目录路径
for i, c in enumerate(candidates, 1): data_roots = []
print(f" {i}. {c}") for ini_file in glob.glob(os.path.join(config_dir, "*.ini")):
print(" 0. 跳过，稍后手动配置") try:
try: # 微信 ini 可能是 utf-8 或 gbk 编码（中文路径）
while True: content = None
choice = input("请选择 [0-{}]: ".format(len(candidates))).strip() for enc in ("utf-8", "gbk"):
if choice == "0": try:
return None with open(ini_file, "r", encoding=enc) as f:
if choice.isdigit() and 1 <= int(choice) <= len(candidates): content = f.read(1024).strip()
return candidates[int(choice) - 1] break
print(" 无效输入，请重新选择") except UnicodeDecodeError:
except (EOFError, KeyboardInterrupt): continue
print() if not content or any(c in content for c in "\n\r\x00"):
return None continue
return None if os.path.isdir(content):
data_roots.append(content)
except OSError:
def load_config(): continue
cfg = {}
if os.path.exists(CONFIG_FILE): # 在每个根目录下搜索 xwechat_files\*\db_storage
try: seen = set()
with open(CONFIG_FILE) as f: candidates = []
cfg = json.load(f) for root in data_roots:
except json.JSONDecodeError: pattern = os.path.join(root, "xwechat_files", "*", "db_storage")
print(f"[!] {CONFIG_FILE} 格式损坏，将使用默认配置") for match in glob.glob(pattern):
cfg = {} normalized = os.path.normcase(os.path.normpath(match))
if os.path.isdir(match) and normalized not in seen:
# db_dir 缺失或仍为模板值时，尝试自动检测 seen.add(normalized)
db_dir = cfg.get("db_dir", "") candidates.append(match)
if not db_dir or db_dir == _DEFAULT_TEMPLATE_DIR or "your_wxid" in db_dir:
detected = auto_detect_db_dir() return _choose_candidate(candidates)
if detected:
print(f"[+] 自动检测到微信数据目录: {detected}")
# 合并默认值并保存 def _auto_detect_db_dir_linux():
cfg = {**_DEFAULT, **cfg, "db_dir": detected} """自动检测 Linux 微信 db_storage 路径。
with open(CONFIG_FILE, "w") as f:
json.dump(cfg, f, indent=4, ensure_ascii=False) 优先搜索当前用户的 home 目录。以 sudo 运行时通过 SUDO_USER 回退到
print(f"[+] 已保存到: {CONFIG_FILE}") 实际用户的 home，避免只搜索 /root 而遗漏真实数据目录。
else: """
if not os.path.exists(CONFIG_FILE): seen = set()
with open(CONFIG_FILE, "w") as f: candidates = []
json.dump(_DEFAULT, f, indent=4) search_roots = [
print(f"[!] 未能自动检测微信数据目录") os.path.expanduser("~/Documents/xwechat_files"),
print(f" 请手动编辑 {CONFIG_FILE} 中的 db_dir 字段") ]
print(f" 路径可在 微信设置 → 文件管理 中找到") # sudo 运行时，~ 展开为 /root；回退到实际用户的 home
sys.exit(1) sudo_user = os.environ.get("SUDO_USER")
if sudo_user:
# 将相对路径转为绝对路径 # 验证 SUDO_USER 是合法系统用户，防止路径注入
base = os.path.dirname(os.path.abspath(__file__)) import pwd
for key in ("keys_file", "decrypted_dir", "decoded_image_dir"): try:
if key in cfg and not os.path.isabs(cfg[key]): sudo_home = pwd.getpwnam(sudo_user).pw_dir
cfg[key] = os.path.join(base, cfg[key]) except KeyError:
sudo_home = None
# 自动推导微信数据根目录（db_dir 的上级目录） if sudo_home:
# db_dir 格式: D:\xwechat_files\<wxid>\db_storage fallback = os.path.join(sudo_home, "Documents", "xwechat_files")
# base_dir 格式: D:\xwechat_files\<wxid> if fallback not in search_roots:
db_dir = cfg.get("db_dir", "") search_roots.append(fallback)
if db_dir and os.path.basename(db_dir) == "db_storage":
cfg["wechat_base_dir"] = os.path.dirname(db_dir) for root in search_roots:
else: if not os.path.isdir(root):
cfg["wechat_base_dir"] = db_dir continue
pattern = os.path.join(root, "*", "db_storage")
# decoded_image_dir 默认值 for match in glob.glob(pattern):
if "decoded_image_dir" not in cfg: normalized = os.path.normcase(os.path.normpath(match))
cfg["decoded_image_dir"] = os.path.join(base, "decoded_images") if os.path.isdir(match) and normalized not in seen:
seen.add(normalized)
return cfg candidates.append(match)
# 早期 Linux 微信版本（wine/容器方案）使用的数据路径
old_path = os.path.expanduser("~/.local/share/weixin/data/db_storage")
if os.path.isdir(old_path):
normalized = os.path.normcase(os.path.normpath(old_path))
if normalized not in seen:
candidates.append(old_path)
# 优先使用最近活跃账号：按 message 目录 mtime 降序（近似排序，best-effort）
def _mtime(path):
msg_dir = os.path.join(path, "message")
target = msg_dir if os.path.isdir(msg_dir) else path
try:
return os.path.getmtime(target)
except OSError:
return 0
candidates.sort(key=_mtime, reverse=True)
return _choose_candidate(candidates)
def auto_detect_db_dir():
if _SYSTEM == "windows":
return _auto_detect_db_dir_windows()
if _SYSTEM == "linux":
return _auto_detect_db_dir_linux()
return None
def load_config():
cfg = {}
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE) as f:
cfg = json.load(f)
except json.JSONDecodeError:
print(f"[!] {CONFIG_FILE} 格式损坏，将使用默认配置")
cfg = {}
# db_dir 缺失或仍为模板值时，尝试自动检测
db_dir = cfg.get("db_dir", "")
if not db_dir or db_dir == _DEFAULT_TEMPLATE_DIR or "your_wxid" in db_dir:
detected = auto_detect_db_dir()
if detected:
print(f"[+] 自动检测到微信数据目录: {detected}")
cfg = {**_DEFAULT, **cfg, "db_dir": detected}
with open(CONFIG_FILE, "w") as f:
json.dump(cfg, f, indent=4, ensure_ascii=False)
print(f"[+] 已保存到: {CONFIG_FILE}")
else:
if not os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, "w") as f:
json.dump(_DEFAULT, f, indent=4, ensure_ascii=False)
print(f"[!] 未能自动检测微信数据目录")
print(f" 请手动编辑 {CONFIG_FILE} 中的 db_dir 字段")
if _SYSTEM == "linux":
print(" Linux 默认路径类似: ~/Documents/xwechat_files/<wxid>/db_storage")
else:
print(f" 路径可在 微信设置 → 文件管理 中找到")
sys.exit(1)
else:
cfg = {**_DEFAULT, **cfg}
# 将相对路径转为绝对路径
base = os.path.dirname(os.path.abspath(__file__))
for key in ("keys_file", "decrypted_dir", "decoded_image_dir"):
if key in cfg and not os.path.isabs(cfg[key]):
cfg[key] = os.path.join(base, cfg[key])
# 自动推导微信数据根目录（db_dir 的上级目录）
# db_dir 格式: D:\xwechat_files\<wxid>\db_storage
# base_dir 格式: D:\xwechat_files\<wxid>
db_dir = cfg.get("db_dir", "")
if db_dir and os.path.basename(db_dir) == "db_storage":
cfg["wechat_base_dir"] = os.path.dirname(db_dir)
else:
cfg["wechat_base_dir"] = db_dir
# decoded_image_dir 默认值
if "decoded_image_dir" not in cfg:
cfg["decoded_image_dir"] = os.path.join(base, "decoded_images")
return cfg

48
decrypt_db.py
View File

@ -7,7 +7,7 @@ WeChat 4.0 数据库解密器
""" """
import hashlib, struct, os, sys, json import hashlib, struct, os, sys, json
import hmac as hmac_mod import hmac as hmac_mod
from Crypto.Cipher import AES from Crypto.Cipher import AES
import functools import functools
print = functools.partial(print, flush=True) print = functools.partial(print, flush=True)
@ -20,11 +20,12 @@ HMAC_SZ = 64
RESERVE_SZ = 80 # IV(16) + HMAC(64) RESERVE_SZ = 80 # IV(16) + HMAC(64)
SQLITE_HDR = b'SQLite format 3\x00' SQLITE_HDR = b'SQLite format 3\x00'
from config import load_config from config import load_config
_cfg = load_config() from key_utils import get_key_info, strip_key_metadata
DB_DIR = _cfg["db_dir"] _cfg = load_config()
OUT_DIR = _cfg["decrypted_dir"] DB_DIR = _cfg["db_dir"]
KEYS_FILE = _cfg["keys_file"] OUT_DIR = _cfg["decrypted_dir"]
KEYS_FILE = _cfg["keys_file"]
def derive_mac_key(enc_key, salt): def derive_mac_key(enc_key, salt):
@ -115,13 +116,13 @@ def main():
print("请先运行 find_all_keys.py") print("请先运行 find_all_keys.py")
sys.exit(1) sys.exit(1)
with open(KEYS_FILE) as f: with open(KEYS_FILE) as f:
keys = json.load(f) keys = json.load(f)
keys.pop("_db_dir", None) keys = strip_key_metadata(keys)
print(f"\n加载 {len(keys)} 个数据库密钥") print(f"\n加载 {len(keys)} 个数据库密钥")
print(f"输出目录: {OUT_DIR}") print(f"输出目录: {OUT_DIR}")
os.makedirs(OUT_DIR, exist_ok=True) os.makedirs(OUT_DIR, exist_ok=True)
# 收集所有DB文件 # 收集所有DB文件
db_files = [] db_files = []
@ -129,7 +130,7 @@ def main():
for f in files: for f in files:
if f.endswith('.db') and not f.endswith('-wal') and not f.endswith('-shm'): if f.endswith('.db') and not f.endswith('-wal') and not f.endswith('-shm'):
path = os.path.join(root, f) path = os.path.join(root, f)
rel = os.path.relpath(path, DB_DIR).replace('\\', '/') rel = os.path.relpath(path, DB_DIR)
sz = os.path.getsize(path) sz = os.path.getsize(path)
db_files.append((rel, path, sz)) db_files.append((rel, path, sz))
@ -141,16 +142,15 @@ def main():
failed = 0 failed = 0
total_bytes = 0 total_bytes = 0
for rel, path, sz in db_files: for rel, path, sz in db_files:
# 统一用正斜杠查找key key_info = get_key_info(keys, rel)
rel_key = rel.replace('\\', '/') if not key_info:
if rel_key not in keys: print(f"SKIP: {rel} (无密钥)")
print(f"SKIP: {rel} (无密钥)") failed += 1
failed += 1 continue
continue
enc_key = bytes.fromhex(key_info["enc_key"])
enc_key = bytes.fromhex(keys[rel_key]["enc_key"]) out_path = os.path.join(OUT_DIR, rel)
out_path = os.path.join(OUT_DIR, rel)
print(f"解密: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ") print(f"解密: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ")

309
find_all_keys.py
View File

@ -1,275 +1,34 @@
""" import functools
从微信进程内存中提取所有数据库的缓存raw key import platform
import sys
WCDB为每个DB缓存: x'<64hex_enc_key><32hex_salt>'
salt嵌在hex字符串中，可以直接匹配DB文件的salt
""" @functools.lru_cache(maxsize=1)
import ctypes def _load_impl():
import ctypes.wintypes as wt system = platform.system().lower()
import struct, os, sys, hashlib, time, re, json if system == "windows":
import hmac as hmac_mod import find_all_keys_windows as impl
from Crypto.Cipher import AES return impl
if system == "linux":
import functools import find_all_keys_linux as impl
print = functools.partial(print, flush=True) return impl
raise RuntimeError(
kernel32 = ctypes.windll.kernel32 f"当前平台暂不支持通过 find_all_keys.py 提取密钥: {platform.system()}\n"
MEM_COMMIT = 0x1000 f"macOS 请使用 find_all_keys_macos.c (C 版扫描器)"
READABLE = {0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80} )
PAGE_SZ = 4096
KEY_SZ = 32
SALT_SZ = 16 def get_pids():
return _load_impl().get_pids()
from config import load_config
_cfg = load_config()
DB_DIR = _cfg["db_dir"] def main():
OUT_FILE = _cfg["keys_file"] return _load_impl().main()
class MBI(ctypes.Structure):
_fields_ = [ if __name__ == "__main__":
("BaseAddress", ctypes.c_uint64), ("AllocationBase", ctypes.c_uint64), try:
("AllocationProtect", wt.DWORD), ("_pad1", wt.DWORD), main()
("RegionSize", ctypes.c_uint64), ("State", wt.DWORD), except RuntimeError as exc:
("Protect", wt.DWORD), ("Type", wt.DWORD), ("_pad2", wt.DWORD), print(f"\n[ERROR] {exc}")
] sys.exit(1)
def get_pids():
"""返回所有 Weixin.exe 进程的 (pid, mem_kb) 列表，按内存降序"""
import subprocess
r = subprocess.run(["tasklist","/FI","IMAGENAME eq Weixin.exe","/FO","CSV","/NH"],
capture_output=True, text=True)
pids = []
for line in r.stdout.strip().split('\n'):
if not line.strip(): continue
p = line.strip('"').split('","')
if len(p)>=5:
pid=int(p[1]); mem=int(p[4].replace(',','').replace(' K','').strip() or '0')
pids.append((pid, mem))
if not pids: raise RuntimeError("Weixin.exe 未运行")
pids.sort(key=lambda x: x[1], reverse=True)
for pid, mem in pids:
print(f"[+] Weixin.exe PID={pid} ({mem//1024}MB)")
return pids
def read_mem(h, addr, sz):
buf = ctypes.create_string_buffer(sz)
n = ctypes.c_size_t(0)
if kernel32.ReadProcessMemory(h, ctypes.c_uint64(addr), buf, sz, ctypes.byref(n)):
return buf.raw[:n.value]
return None
def enum_regions(h):
regs = []
addr = 0
mbi = MBI()
while addr < 0x7FFFFFFFFFFF:
if kernel32.VirtualQueryEx(h, ctypes.c_uint64(addr), ctypes.byref(mbi), ctypes.sizeof(mbi))==0: break
if mbi.State==MEM_COMMIT and mbi.Protect in READABLE and 0<mbi.RegionSize<500*1024*1024:
regs.append((mbi.BaseAddress, mbi.RegionSize))
nxt = mbi.BaseAddress + mbi.RegionSize
if nxt<=addr: break
addr = nxt
return regs
def verify_key_for_db(enc_key, db_page1):
"""验证enc_key是否能解密这个DB的page 1"""
salt = db_page1[:SALT_SZ]
iv = db_page1[PAGE_SZ - 80 : PAGE_SZ - 64]
encrypted = db_page1[SALT_SZ : PAGE_SZ - 80]
# HMAC验证 (最可靠)
mac_salt = bytes(b ^ 0x3a for b in salt)
mac_key = hashlib.pbkdf2_hmac("sha512", enc_key, mac_salt, 2, dklen=KEY_SZ)
hmac_data = db_page1[SALT_SZ : PAGE_SZ - 80 + 16]
stored_hmac = db_page1[PAGE_SZ - 64 : PAGE_SZ]
h = hmac_mod.new(mac_key, hmac_data, hashlib.sha512)
h.update(struct.pack('<I', 1))
return h.digest() == stored_hmac
def main():
print("=" * 60)
print(" 提取所有微信数据库密钥")
print("=" * 60)
# 1. 收集所有DB文件及其salt
db_files = []
salt_to_dbs = {} # salt_hex -> [(rel_path, db_page1), ...]
for root, dirs, files in os.walk(DB_DIR):
for f in files:
if f.endswith('.db') and not f.endswith('-wal') and not f.endswith('-shm'):
path = os.path.join(root, f)
rel = os.path.relpath(path, DB_DIR).replace('\\', '/')
sz = os.path.getsize(path)
if sz < PAGE_SZ:
continue
with open(path, 'rb') as fh:
page1 = fh.read(PAGE_SZ)
salt = page1[:SALT_SZ].hex()
db_files.append((rel, path, sz, salt, page1))
if salt not in salt_to_dbs:
salt_to_dbs[salt] = []
salt_to_dbs[salt].append(rel)
print(f"\n找到 {len(db_files)} 个数据库, {len(salt_to_dbs)} 个不同的salt")
for salt_hex, dbs in sorted(salt_to_dbs.items(), key=lambda x: len(x[1]), reverse=True):
print(f" salt {salt_hex}: {', '.join(dbs)}")
# 2. 打开所有微信进程
pids = get_pids()
hex_re = re.compile(b"x'([0-9a-fA-F]{64,192})'")
key_map = {} # salt_hex -> enc_key_hex
remaining_salts = set(salt_to_dbs.keys())
all_hex_matches = 0
t0 = time.time()
for proc_idx, (pid, mem) in enumerate(pids):
h = kernel32.OpenProcess(0x0010 | 0x0400, False, pid)
if not h:
print(f"[WARN] 无法打开进程 PID={pid}，跳过")
continue
try:
regions = enum_regions(h)
total_bytes = sum(s for _,s in regions)
total_mb = total_bytes/1024/1024
print(f"\n[*] 扫描 PID={pid} ({total_mb:.0f}MB, {len(regions)} 区域)")
scanned_bytes = 0
for reg_idx, (base, size) in enumerate(regions):
data = read_mem(h, base, size)
scanned_bytes += size
if not data: continue
for m in hex_re.finditer(data):
hex_str = m.group(1).decode()
addr = base + m.start()
all_hex_matches += 1
hex_len = len(hex_str)
if hex_len == 96:
enc_key_hex = hex_str[:64]
salt_hex = hex_str[64:]
if salt_hex in remaining_salts:
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, s, page1 in db_files:
if s == salt_hex:
if verify_key_for_db(enc_key, page1):
key_map[salt_hex] = enc_key_hex
remaining_salts.discard(salt_hex)
dbs = salt_to_dbs[salt_hex]
print(f"\n [FOUND] salt={salt_hex}")
print(f" enc_key={enc_key_hex}")
print(f" PID={pid} 地址: 0x{addr:016X}")
print(f" 数据库: {', '.join(dbs)}")
break
elif hex_len == 64:
if not remaining_salts:
continue
enc_key_hex = hex_str
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, salt_hex_db, page1 in db_files:
if salt_hex_db in remaining_salts:
if verify_key_for_db(enc_key, page1):
key_map[salt_hex_db] = enc_key_hex
remaining_salts.discard(salt_hex_db)
dbs = salt_to_dbs[salt_hex_db]
print(f"\n [FOUND] salt={salt_hex_db}")
print(f" enc_key={enc_key_hex}")
print(f" PID={pid} 地址: 0x{addr:016X}")
print(f" 数据库: {', '.join(dbs)}")
break
elif hex_len > 96 and hex_len % 2 == 0:
enc_key_hex = hex_str[:64]
salt_hex = hex_str[-32:]
if salt_hex in remaining_salts:
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, s, page1 in db_files:
if s == salt_hex:
if verify_key_for_db(enc_key, page1):
key_map[salt_hex] = enc_key_hex
remaining_salts.discard(salt_hex)
dbs = salt_to_dbs[salt_hex]
print(f"\n [FOUND] salt={salt_hex} (long hex {hex_len})")
print(f" enc_key={enc_key_hex}")
print(f" PID={pid} 地址: 0x{addr:016X}")
print(f" 数据库: {', '.join(dbs)}")
break
if (reg_idx + 1) % 200 == 0:
elapsed = time.time() - t0
progress = scanned_bytes / total_bytes * 100 if total_bytes else 100
print(f" [{progress:.1f}%] {len(key_map)}/{len(salt_to_dbs)} salts matched, "
f"{all_hex_matches} hex patterns, {elapsed:.1f}s")
finally:
kernel32.CloseHandle(h)
# 所有 salt 都找到了就提前退出
if not remaining_salts:
print(f"\n[+] 所有密钥已找到，跳过剩余进程")
break
elapsed = time.time() - t0
print(f"\n扫描完成: {elapsed:.1f}s, {len(pids)} 个进程, {all_hex_matches} hex模式")
# 4. 如果有未找到的salt，用已找到的key做交叉验证
# (WCDB有时对同一passphrase的不同DB用同一enc_key，如果salt相同)
missing_salts = set(salt_to_dbs.keys()) - set(key_map.keys())
if missing_salts and key_map:
print(f"\n还有 {len(missing_salts)} 个salt未匹配，尝试交叉验证...")
for salt_hex in list(missing_salts):
for rel, path, sz, s, page1 in db_files:
if s == salt_hex:
for known_salt, known_key_hex in key_map.items():
enc_key = bytes.fromhex(known_key_hex)
if verify_key_for_db(enc_key, page1):
key_map[salt_hex] = known_key_hex
print(f" [CROSS] salt={salt_hex} 可用 key from salt={known_salt}")
missing_salts.discard(salt_hex)
break
# 5. 输出结果
print(f"\n{'='*60}")
print(f"结果: {len(key_map)}/{len(salt_to_dbs)} salts 找到密钥")
result = {}
for rel, path, sz, salt_hex, page1 in db_files:
if salt_hex in key_map:
result[rel] = {
"enc_key": key_map[salt_hex],
"salt": salt_hex,
"size_mb": round(sz/1024/1024, 1)
}
print(f" OK: {rel} ({sz/1024/1024:.1f}MB)")
else:
print(f" MISSING: {rel} (salt={salt_hex})")
if not result:
print(f"\n[!] 未提取到任何密钥，保留已有的 {OUT_FILE}（如存在）")
raise RuntimeError("未能从任何微信进程中提取到密钥")
# 写入密钥并记录对应的 db_dir，防止切换账号后误复用
result["_db_dir"] = DB_DIR
with open(OUT_FILE, 'w') as f:
json.dump(result, f, indent=2)
print(f"\n密钥保存到: {OUT_FILE}")
missing = [rel for rel, path, sz, salt_hex, page1 in db_files if salt_hex not in key_map]
if missing:
print(f"\n未找到密钥的数据库:")
for rel in missing:
print(f" {rel}")
if __name__ == '__main__':
try:
main()
except RuntimeError as e:
print(f"\n[ERROR] {e}")
sys.exit(1)

246
find_all_keys_linux.py 100644
View File

@ -0,0 +1,246 @@
"""
Linux 版微信数据库密钥提取
原理: 与 Windows/macOS 相同 — 扫描微信进程内存，查找
WCDB 缓存的 x'<64hex_enc_key><32hex_salt>' 模式，
通过匹配数据库 salt + HMAC 校验确认密钥。
读取方式: /proc/<pid>/maps + /proc/<pid>/mem
权限要求: root 或 CAP_SYS_PTRACE
"""
import functools
import os
import re
import sys
import time
from key_scan_common import (
collect_db_files, scan_memory_for_keys, cross_verify_keys, save_results,
)
print = functools.partial(print, flush=True)
def _safe_readlink(path):
try:
return os.path.realpath(os.readlink(path))
except OSError:
return ""
_KNOWN_COMMS = {"wechat", "wechatappex", "weixin"}
_INTERPRETER_PREFIXES = ("python", "bash", "sh", "zsh", "node", "perl", "ruby")
def _is_wechat_process(pid):
"""检查 pid 是否为微信进程。
优先精确匹配 comm 名称（wechat、WeChatAppEx 等），
再用 exe 路径子串匹配作为 fallback，同时排除解释器进程。
"""
if pid == os.getpid():
return False
try:
with open(f"/proc/{pid}/comm") as f:
comm = f.read().strip()
# 优先精确匹配 comm（最可靠）
if comm.lower() in _KNOWN_COMMS:
return True
exe_path = _safe_readlink(f"/proc/{pid}/exe")
exe_name = os.path.basename(exe_path)
# 排除脚本解释器进程（避免匹配 python3.11 wechat-decrypt 等）
if any(exe_name.lower().startswith(p) for p in _INTERPRETER_PREFIXES):
return False
# fallback: exe 名称子串匹配
return "wechat" in exe_name.lower() or "weixin" in exe_name.lower()
except (PermissionError, FileNotFoundError, ProcessLookupError):
return False
def get_pids():
"""返回所有疑似微信主进程的 (pid, rss_kb) 列表，按内存降序。"""
pids = []
for pid_str in os.listdir("/proc"):
if not pid_str.isdigit():
continue
pid = int(pid_str)
try:
if not _is_wechat_process(pid):
continue
with open(f"/proc/{pid}/statm") as f:
rss_pages = int(f.read().split()[1])
rss_kb = rss_pages * 4
pids.append((pid, rss_kb))
except (PermissionError, FileNotFoundError, ProcessLookupError):
continue
if not pids:
raise RuntimeError("未检测到 Linux 微信进程")
pids.sort(key=lambda item: item[1], reverse=True)
for pid, rss_kb in pids:
exe_path = _safe_readlink(f"/proc/{pid}/exe")
print(f"[+] WeChat PID={pid} ({rss_kb // 1024}MB) {exe_path}")
return pids
_SKIP_MAPPINGS = {"[vdso]", "[vsyscall]", "[vvar]"}
_SKIP_PATH_PREFIXES = ("/usr/lib/", "/lib/", "/usr/share/")
def _get_readable_regions(pid):
"""解析 /proc/<pid>/maps，返回可读内存区域列表。
跳过 [vdso]、[vsyscall] 等特殊映射和系统库映射，
聚焦匿名映射和堆区（WCDB 密钥缓存所在位置）。
"""
regions = []
with open(f"/proc/{pid}/maps") as f:
for line in f:
parts = line.split()
if len(parts) < 2:
continue
if "r" not in parts[1]:
continue
# 跳过特殊映射和无关系统库，但保留 wcdb/wechat 相关库
if len(parts) >= 6:
mapping_name = parts[5]
if mapping_name in _SKIP_MAPPINGS:
continue
mapping_lower = mapping_name.lower()
if (any(mapping_name.startswith(p) for p in _SKIP_PATH_PREFIXES)
and "wcdb" not in mapping_lower
and "wechat" not in mapping_lower
and "weixin" not in mapping_lower):
continue
start_s, end_s = parts[0].split("-")
start = int(start_s, 16)
size = int(end_s, 16) - start
if 0 < size < 500 * 1024 * 1024:
regions.append((start, size))
return regions
def _check_permissions():
"""检查是否有读取进程内存的权限（root 或 CAP_SYS_PTRACE）。"""
if os.geteuid() == 0:
return
# 检查 CAP_SYS_PTRACE: 读取 /proc/self/status 中的 CapEff
try:
with open("/proc/self/status") as f:
for line in f:
if line.startswith("CapEff:"):
cap_eff = int(line.split(":")[1].strip(), 16)
CAP_SYS_PTRACE = 1 << 19
if cap_eff & CAP_SYS_PTRACE:
return
break
except (OSError, ValueError):
pass
print("[!] 需要 root 权限或 CAP_SYS_PTRACE 才能读取进程内存")
print(" 请使用: sudo python3 find_all_keys.py")
print(" 或授予 capability: sudo setcap cap_sys_ptrace=ep $(which python3)")
sys.exit(1)
def main():
from config import load_config
_cfg = load_config()
db_dir = _cfg["db_dir"]
out_file = _cfg["keys_file"]
_check_permissions()
print("=" * 60)
print(" 提取 Linux 微信数据库密钥（内存扫描）")
print("=" * 60)
# 1. 收集 DB 文件和 salt
db_files, salt_to_dbs = collect_db_files(db_dir)
if not db_files:
raise RuntimeError(f"在 {db_dir} 未找到可解密的 .db 文件")
print(f"\n找到 {len(db_files)} 个数据库, {len(salt_to_dbs)} 个不同的 salt")
for salt_hex, dbs in sorted(salt_to_dbs.items(), key=lambda x: len(x[1]), reverse=True):
print(f" salt {salt_hex}: {', '.join(dbs)}")
# 2. 找到微信进程
pids = get_pids()
hex_re = re.compile(rb"x'([0-9a-fA-F]{64,192})'")
key_map = {} # salt_hex -> enc_key_hex
remaining_salts = set(salt_to_dbs.keys())
all_hex_matches = 0
t0 = time.time()
for pid, rss_kb in pids:
try:
regions = _get_readable_regions(pid)
except PermissionError:
print(f"[WARN] 无法读取 /proc/{pid}/maps，权限不足，跳过")
continue
except (FileNotFoundError, ProcessLookupError):
print(f"[WARN] PID {pid} 已退出，跳过")
continue
total_bytes = sum(s for _, s in regions)
total_mb = total_bytes / 1024 / 1024
print(f"\n[*] 扫描 PID={pid} ({total_mb:.0f}MB, {len(regions)} 区域)")
scanned_bytes = 0
try:
mem = open(f"/proc/{pid}/mem", "rb")
except PermissionError:
print(f"[WARN] 无法打开 /proc/{pid}/mem，权限不足，跳过")
continue
except (FileNotFoundError, ProcessLookupError):
print(f"[WARN] PID {pid} 已退出，跳过")
continue
# 防御 TOCTOU: 打开 mem 后再次确认仍为微信进程
if not _is_wechat_process(pid):
print(f"[WARN] PID {pid} 已不是微信进程，跳过")
mem.close()
continue
try:
for reg_idx, (base, size) in enumerate(regions):
try:
mem.seek(base)
data = mem.read(size)
except (OSError, ValueError):
continue
scanned_bytes += len(data)
all_hex_matches += scan_memory_for_keys(
data, hex_re, db_files, salt_to_dbs,
key_map, remaining_salts, base, pid, print,
)
if (reg_idx + 1) % 200 == 0:
elapsed = time.time() - t0
progress = scanned_bytes / total_bytes * 100 if total_bytes else 100
print(
f" [{progress:.1f}%] {len(key_map)}/{len(salt_to_dbs)} salts matched, "
f"{all_hex_matches} hex patterns, {elapsed:.1f}s"
)
finally:
mem.close()
if not remaining_salts:
print(f"\n[+] 所有密钥已找到，跳过剩余进程")
break
elapsed = time.time() - t0
print(f"\n扫描完成: {elapsed:.1f}s, {len(pids)} 个进程, {all_hex_matches} hex 模式")
cross_verify_keys(db_files, salt_to_dbs, key_map, print)
save_results(db_files, salt_to_dbs, key_map, db_dir, out_file, print)
if __name__ == "__main__":
try:
main()
except RuntimeError as exc:
print(f"\n[ERROR] {exc}")
sys.exit(1)

154
find_all_keys_windows.py 100644
View File

@ -0,0 +1,154 @@
"""
从微信进程内存中提取所有数据库的缓存raw key
WCDB为每个DB缓存: x'<64hex_enc_key><32hex_salt>'
salt嵌在hex字符串中，可以直接匹配DB文件的salt
"""
import ctypes
import ctypes.wintypes as wt
import os, sys, time, re
import functools
print = functools.partial(print, flush=True)
from key_scan_common import (
collect_db_files, scan_memory_for_keys, cross_verify_keys, save_results,
)
kernel32 = ctypes.windll.kernel32
MEM_COMMIT = 0x1000
READABLE = {0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80}
class MBI(ctypes.Structure):
_fields_ = [
("BaseAddress", ctypes.c_uint64), ("AllocationBase", ctypes.c_uint64),
("AllocationProtect", wt.DWORD), ("_pad1", wt.DWORD),
("RegionSize", ctypes.c_uint64), ("State", wt.DWORD),
("Protect", wt.DWORD), ("Type", wt.DWORD), ("_pad2", wt.DWORD),
]
def get_pids():
"""返回所有 Weixin.exe 进程的 (pid, mem_kb) 列表，按内存降序"""
import subprocess
r = subprocess.run(["tasklist", "/FI", "IMAGENAME eq Weixin.exe", "/FO", "CSV", "/NH"],
capture_output=True, text=True)
pids = []
for line in r.stdout.strip().split('\n'):
if not line.strip():
continue
p = line.strip('"').split('","')
if len(p) >= 5:
pid = int(p[1])
mem = int(p[4].replace(',', '').replace(' K', '').strip() or '0')
pids.append((pid, mem))
if not pids:
raise RuntimeError("Weixin.exe 未运行")
pids.sort(key=lambda x: x[1], reverse=True)
for pid, mem in pids:
print(f"[+] Weixin.exe PID={pid} ({mem // 1024}MB)")
return pids
def read_mem(h, addr, sz):
buf = ctypes.create_string_buffer(sz)
n = ctypes.c_size_t(0)
if kernel32.ReadProcessMemory(h, ctypes.c_uint64(addr), buf, sz, ctypes.byref(n)):
return buf.raw[:n.value]
return None
def enum_regions(h):
regs = []
addr = 0
mbi = MBI()
while addr < 0x7FFFFFFFFFFF:
if kernel32.VirtualQueryEx(h, ctypes.c_uint64(addr), ctypes.byref(mbi), ctypes.sizeof(mbi)) == 0:
break
if mbi.State == MEM_COMMIT and mbi.Protect in READABLE and 0 < mbi.RegionSize < 500 * 1024 * 1024:
regs.append((mbi.BaseAddress, mbi.RegionSize))
nxt = mbi.BaseAddress + mbi.RegionSize
if nxt <= addr:
break
addr = nxt
return regs
def main():
from config import load_config
_cfg = load_config()
db_dir = _cfg["db_dir"]
out_file = _cfg["keys_file"]
print("=" * 60)
print(" 提取所有微信数据库密钥")
print("=" * 60)
# 1. 收集所有DB文件及其salt
db_files, salt_to_dbs = collect_db_files(db_dir)
print(f"\n找到 {len(db_files)} 个数据库, {len(salt_to_dbs)} 个不同的salt")
for salt_hex, dbs in sorted(salt_to_dbs.items(), key=lambda x: len(x[1]), reverse=True):
print(f" salt {salt_hex}: {', '.join(dbs)}")
# 2. 打开所有微信进程
pids = get_pids()
hex_re = re.compile(b"x'([0-9a-fA-F]{64,192})'")
key_map = {}
remaining_salts = set(salt_to_dbs.keys())
all_hex_matches = 0
t0 = time.time()
for pid, mem_kb in pids:
h = kernel32.OpenProcess(0x0010 | 0x0400, False, pid)
if not h:
print(f"[WARN] 无法打开进程 PID={pid}，跳过")
continue
try:
regions = enum_regions(h)
total_bytes = sum(s for _, s in regions)
total_mb = total_bytes / 1024 / 1024
print(f"\n[*] 扫描 PID={pid} ({total_mb:.0f}MB, {len(regions)} 区域)")
scanned_bytes = 0
for reg_idx, (base, size) in enumerate(regions):
data = read_mem(h, base, size)
scanned_bytes += size
if not data:
continue
all_hex_matches += scan_memory_for_keys(
data, hex_re, db_files, salt_to_dbs,
key_map, remaining_salts, base, pid, print,
)
if (reg_idx + 1) % 200 == 0:
elapsed = time.time() - t0
progress = scanned_bytes / total_bytes * 100 if total_bytes else 100
print(
f" [{progress:.1f}%] {len(key_map)}/{len(salt_to_dbs)} salts matched, "
f"{all_hex_matches} hex patterns, {elapsed:.1f}s"
)
finally:
kernel32.CloseHandle(h)
if not remaining_salts:
print(f"\n[+] 所有密钥已找到，跳过剩余进程")
break
elapsed = time.time() - t0
print(f"\n扫描完成: {elapsed:.1f}s, {len(pids)} 个进程, {all_hex_matches} hex模式")
cross_verify_keys(db_files, salt_to_dbs, key_map, print)
save_results(db_files, salt_to_dbs, key_map, db_dir, out_file, print)
if __name__ == '__main__':
try:
main()
except RuntimeError as e:
print(f"\n[ERROR] {e}")
sys.exit(1)

169
key_scan_common.py 100644
View File

@ -0,0 +1,169 @@
"""
跨平台共享的内存扫描逻辑：HMAC 验证、DB 收集、hex 模式匹配与结果输出。
Windows / Linux 版分别实现进程发现和内存读取，共用此模块的核心算法。
"""
import hashlib
import hmac as hmac_mod
import json
import os
import re
import struct
import time
PAGE_SZ = 4096
KEY_SZ = 32
SALT_SZ = 16
def verify_enc_key(enc_key, db_page1):
"""通过 HMAC-SHA512 校验 page 1 验证 enc_key 是否正确。"""
salt = db_page1[:SALT_SZ]
mac_salt = bytes(b ^ 0x3A for b in salt)
mac_key = hashlib.pbkdf2_hmac("sha512", enc_key, mac_salt, 2, dklen=KEY_SZ)
hmac_data = db_page1[SALT_SZ: PAGE_SZ - 80 + 16]
stored_hmac = db_page1[PAGE_SZ - 64: PAGE_SZ]
hm = hmac_mod.new(mac_key, hmac_data, hashlib.sha512)
hm.update(struct.pack("<I", 1))
return hm.digest() == stored_hmac
def collect_db_files(db_dir):
"""遍历 db_dir 收集所有 .db 文件及其 salt。
返回 (db_files, salt_to_dbs):
db_files: [(rel_path, abs_path, size, salt_hex, page1_bytes), ...]
salt_to_dbs: {salt_hex: [rel_path, ...]}
"""
db_files = []
salt_to_dbs = {}
for root, dirs, files in os.walk(db_dir):
for name in files:
if not name.endswith(".db") or name.endswith("-wal") or name.endswith("-shm"):
continue
path = os.path.join(root, name)
size = os.path.getsize(path)
if size < PAGE_SZ:
continue
with open(path, "rb") as f:
page1 = f.read(PAGE_SZ)
rel = os.path.relpath(path, db_dir)
salt = page1[:SALT_SZ].hex()
db_files.append((rel, path, size, salt, page1))
salt_to_dbs.setdefault(salt, []).append(rel)
return db_files, salt_to_dbs
def scan_memory_for_keys(data, hex_re, db_files, salt_to_dbs, key_map,
remaining_salts, base_addr, pid, print_fn):
"""扫描一段内存数据，匹配 hex 模式并验证密钥。
返回本次扫描匹配到的 hex 模式数量。
"""
matches = 0
for m in hex_re.finditer(data):
hex_str = m.group(1).decode()
addr = base_addr + m.start()
matches += 1
hex_len = len(hex_str)
if hex_len == 96:
enc_key_hex = hex_str[:64]
salt_hex = hex_str[64:]
if salt_hex in remaining_salts:
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, s, page1 in db_files:
if s == salt_hex and verify_enc_key(enc_key, page1):
key_map[salt_hex] = enc_key_hex
remaining_salts.discard(salt_hex)
dbs = salt_to_dbs[salt_hex]
print_fn(f"\n [FOUND] salt={salt_hex}")
print_fn(f" enc_key={enc_key_hex}")
print_fn(f" PID={pid} 地址: 0x{addr:016X}")
print_fn(f" 数据库: {', '.join(dbs)}")
break
elif hex_len == 64:
if not remaining_salts:
continue
enc_key_hex = hex_str
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, salt_hex_db, page1 in db_files:
if salt_hex_db in remaining_salts and verify_enc_key(enc_key, page1):
key_map[salt_hex_db] = enc_key_hex
remaining_salts.discard(salt_hex_db)
dbs = salt_to_dbs[salt_hex_db]
print_fn(f"\n [FOUND] salt={salt_hex_db}")
print_fn(f" enc_key={enc_key_hex}")
print_fn(f" PID={pid} 地址: 0x{addr:016X}")
print_fn(f" 数据库: {', '.join(dbs)}")
break
elif hex_len > 96 and hex_len % 2 == 0:
enc_key_hex = hex_str[:64]
salt_hex = hex_str[-32:]
if salt_hex in remaining_salts:
enc_key = bytes.fromhex(enc_key_hex)
for rel, path, sz, s, page1 in db_files:
if s == salt_hex and verify_enc_key(enc_key, page1):
key_map[salt_hex] = enc_key_hex
remaining_salts.discard(salt_hex)
dbs = salt_to_dbs[salt_hex]
print_fn(f"\n [FOUND] salt={salt_hex} (long hex {hex_len})")
print_fn(f" enc_key={enc_key_hex}")
print_fn(f" PID={pid} 地址: 0x{addr:016X}")
print_fn(f" 数据库: {', '.join(dbs)}")
break
return matches
def cross_verify_keys(db_files, salt_to_dbs, key_map, print_fn):
"""用已找到的 key 交叉验证未匹配的 salt。"""
missing_salts = set(salt_to_dbs.keys()) - set(key_map.keys())
if not missing_salts or not key_map:
return
print_fn(f"\n还有 {len(missing_salts)} 个 salt 未匹配，尝试交叉验证...")
for salt_hex in list(missing_salts):
for rel, path, sz, s, page1 in db_files:
if s == salt_hex:
for known_salt, known_key_hex in key_map.items():
enc_key = bytes.fromhex(known_key_hex)
if verify_enc_key(enc_key, page1):
key_map[salt_hex] = known_key_hex
print_fn(f" [CROSS] salt={salt_hex} 可用 key from salt={known_salt}")
missing_salts.discard(salt_hex)
break
def save_results(db_files, salt_to_dbs, key_map, db_dir, out_file, print_fn):
"""输出扫描结果并保存 JSON。"""
print_fn(f"\n{'=' * 60}")
print_fn(f"结果: {len(key_map)}/{len(salt_to_dbs)} salts 找到密钥")
result = {}
for rel, path, sz, salt_hex, page1 in db_files:
if salt_hex in key_map:
result[rel] = {
"enc_key": key_map[salt_hex],
"salt": salt_hex,
"size_mb": round(sz / 1024 / 1024, 1)
}
print_fn(f" OK: {rel} ({sz / 1024 / 1024:.1f}MB)")
else:
print_fn(f" MISSING: {rel} (salt={salt_hex})")
if not result:
print_fn(f"\n[!] 未提取到任何密钥，保留已有的 {out_file}（如存在）")
raise RuntimeError("未能从任何微信进程中提取到密钥")
result["_db_dir"] = db_dir
with open(out_file, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2, ensure_ascii=False)
print_fn(f"\n密钥保存到: {out_file}")
missing = [rel for rel, path, sz, salt_hex, page1 in db_files if salt_hex not in key_map]
if missing:
print_fn(f"\n未找到密钥的数据库:")
for rel in missing:
print_fn(f" {rel}")

38
key_utils.py 100644
View File

@ -0,0 +1,38 @@
import os
import posixpath
def strip_key_metadata(keys):
"""移除 all_keys.json 中以下划线开头的元数据字段，返回新 dict。"""
return {k: v for k, v in keys.items() if not k.startswith("_")}
def _is_safe_rel_path(path):
"""检查路径不包含 .. 等遍历组件。"""
normalized = path.replace("\\", "/")
return ".." not in posixpath.normpath(normalized).split("/")
def key_path_variants(rel_path):
"""生成同一路径的多种分隔符表示，兼容 Windows/Linux JSON key。"""
normalized = rel_path.replace("\\", "/")
variants = []
for candidate in (
rel_path,
normalized,
normalized.replace("/", "\\"),
normalized.replace("/", os.sep),
):
if candidate not in variants:
variants.append(candidate)
return variants
def get_key_info(keys, rel_path):
"""按相对路径查找数据库密钥，自动兼容不同平台分隔符。"""
if not _is_safe_rel_path(rel_path):
return None
for candidate in key_path_variants(rel_path):
if candidate in keys and not candidate.startswith("_"):
return keys[candidate]
return None

7
main.py
View File

@ -11,6 +11,8 @@ import sys
import functools import functools
print = functools.partial(print, flush=True) print = functools.partial(print, flush=True)
from key_utils import strip_key_metadata
def check_wechat_running(): def check_wechat_running():
"""检查微信是否在运行，返回 True/False""" """检查微信是否在运行，返回 True/False"""
@ -37,6 +39,7 @@ def ensure_keys(keys_file, db_dir):
print(f" 旧: {saved_dir}") print(f" 旧: {saved_dir}")
print(f" 新: {db_dir}") print(f" 新: {db_dir}")
keys = {} keys = {}
keys = strip_key_metadata(keys)
if keys: if keys:
print(f"[+] 已有 {len(keys)} 个数据库密钥") print(f"[+] 已有 {len(keys)} 个数据库密钥")
return return
@ -60,7 +63,7 @@ def ensure_keys(keys_file, db_dir):
keys = json.load(f) keys = json.load(f)
except (json.JSONDecodeError, ValueError): except (json.JSONDecodeError, ValueError):
keys = {} keys = {}
if not keys: if not strip_key_metadata(keys):
print("[!] 未能提取到任何密钥") print("[!] 未能提取到任何密钥")
print(" 可能原因：选择了错误的微信数据目录，或微信需要重启") print(" 可能原因：选择了错误的微信数据目录，或微信需要重启")
print(" 请检查 config.json 中的 db_dir 是否与当前登录的微信账号匹配") print(" 请检查 config.json 中的 db_dir 是否与当前登录的微信账号匹配")
@ -79,7 +82,7 @@ def main():
# 2. 检查微信进程 # 2. 检查微信进程
if not check_wechat_running(): if not check_wechat_running():
print("[!] 未检测到微信进程 (Weixin.exe)") print(f"[!] 未检测到微信进程 ({cfg.get('wechat_process', 'WeChat')})")
print(" 请先启动微信并登录，然后重新运行") print(" 请先启动微信并登录，然后重新运行")
sys.exit(1) sys.exit(1)
print("[+] 微信进程运行中") print("[+] 微信进程运行中")

27
mcp_server.py
View File

@ -5,13 +5,14 @@ Based on FastMCP (stdio transport), reuses existing decryption.
Runs on Windows Python (needs access to D:\ WeChat databases). Runs on Windows Python (needs access to D:\ WeChat databases).
""" """
import os, sys, json, time, sqlite3, tempfile, struct, hashlib, atexit import os, sys, json, time, sqlite3, tempfile, struct, hashlib, atexit, re
import hmac as hmac_mod import hmac as hmac_mod
from datetime import datetime from datetime import datetime
from Crypto.Cipher import AES from Crypto.Cipher import AES
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
import zstandard as zstd import zstandard as zstd
from decode_image import ImageResolver from decode_image import ImageResolver
from key_utils import get_key_info, key_path_variants, strip_key_metadata
# ============ 加密常量 ============ # ============ 加密常量 ============
PAGE_SZ = 4096 PAGE_SZ = 4096
@ -50,7 +51,7 @@ elif not os.path.isabs(DECODED_IMAGE_DIR):
DECODED_IMAGE_DIR = os.path.join(SCRIPT_DIR, DECODED_IMAGE_DIR) DECODED_IMAGE_DIR = os.path.join(SCRIPT_DIR, DECODED_IMAGE_DIR)
with open(KEYS_FILE) as f: with open(KEYS_FILE) as f:
ALL_KEYS = json.load(f) ALL_KEYS = strip_key_metadata(json.load(f))
# ============ 解密函数 ============ # ============ 解密函数 ============
@ -149,7 +150,7 @@ class DBCache:
tmp_path = info["path"] tmp_path = info["path"]
if not os.path.exists(tmp_path): if not os.path.exists(tmp_path):
continue continue
rel_path = rel_key.replace('/', os.sep) rel_path = rel_key.replace('\\', os.sep)
db_path = os.path.join(DB_DIR, rel_path) db_path = os.path.join(DB_DIR, rel_path)
wal_path = db_path + "-wal" wal_path = db_path + "-wal"
try: try:
@ -175,9 +176,10 @@ class DBCache:
pass pass
def get(self, rel_key): def get(self, rel_key):
if rel_key not in ALL_KEYS: key_info = get_key_info(ALL_KEYS, rel_key)
if not key_info:
return None return None
rel_path = rel_key.replace('/', os.sep) rel_path = rel_key.replace('\\', '/').replace('/', os.sep)
db_path = os.path.join(DB_DIR, rel_path) db_path = os.path.join(DB_DIR, rel_path)
wal_path = db_path + "-wal" wal_path = db_path + "-wal"
if not os.path.exists(db_path): if not os.path.exists(db_path):
@ -195,7 +197,7 @@ class DBCache:
return c_path return c_path
tmp_path = self._cache_path(rel_key) tmp_path = self._cache_path(rel_key)
enc_key = bytes.fromhex(ALL_KEYS[rel_key]["enc_key"]) enc_key = bytes.fromhex(key_info["enc_key"])
full_decrypt(db_path, tmp_path, enc_key) full_decrypt(db_path, tmp_path, enc_key)
if os.path.exists(wal_path): if os.path.exists(wal_path):
decrypt_wal(wal_path, tmp_path, enc_key) decrypt_wal(wal_path, tmp_path, enc_key)
@ -248,7 +250,7 @@ def get_contact_names():
pass pass
# 实时解密 # 实时解密
path = _cache.get("contact/contact.db") path = _cache.get(os.path.join("contact", "contact.db"))
if path: if path:
try: try:
_contact_names, _contact_full = _load_contacts_from(path) _contact_names, _contact_full = _load_contacts_from(path)
@ -329,11 +331,12 @@ def _parse_message_content(content, local_type, is_group):
return sender, text return sender, text
# 消息 DB 的 rel_keys（排除 fts/resource/media/biz） # 消息 DB 的 rel_keys
# 用 message_\d+\.db$ 匹配，自然排除 message_resource.db / message_fts_*.db
MSG_DB_KEYS = sorted([ MSG_DB_KEYS = sorted([
k for k in ALL_KEYS k for k in ALL_KEYS
if k.startswith("message/message_") and k.endswith(".db") if any(v.startswith("message/") for v in key_path_variants(k))
and "fts" not in k and "resource" not in k and any(re.search(r"message_\d+\.db$", v) for v in key_path_variants(k))
]) ])
@ -379,7 +382,7 @@ def get_recent_sessions(limit: int = 20) -> str:
Args: Args:
limit: 返回的会话数量，默认20 limit: 返回的会话数量，默认20
""" """
path = _cache.get("session/session.db") path = _cache.get(os.path.join("session", "session.db"))
if not path: if not path:
return "错误: 无法解密 session.db" return "错误: 无法解密 session.db"
@ -635,7 +638,7 @@ def get_new_messages() -> str:
"""获取自上次调用以来的新消息。首次调用返回最近的会话状态。""" """获取自上次调用以来的新消息。首次调用返回最近的会话状态。"""
global _last_check_state global _last_check_state
path = _cache.get("session/session.db") path = _cache.get(os.path.join("session", "session.db"))
if not path: if not path:
return "错误: 无法解密 session.db" return "错误: 无法解密 session.db"

19
monitor.py
View File

@ -7,8 +7,9 @@ session.db 包含每个聊天的最新消息摘要、发送者、时间戳
import hashlib, struct, os, sys, json, time, sqlite3, io import hashlib, struct, os, sys, json, time, sqlite3, io
import hmac as hmac_mod import hmac as hmac_mod
from datetime import datetime from datetime import datetime
from Crypto.Cipher import AES from Crypto.Cipher import AES
import zstandard as zstd import zstandard as zstd
from key_utils import get_key_info, strip_key_metadata
_zstd_dctx = zstd.ZstdDecompressor() _zstd_dctx = zstd.ZstdDecompressor()
@ -148,13 +149,13 @@ def main():
print("=" * 60) print("=" * 60)
# 加载密钥 # 加载密钥
with open(KEYS_FILE) as f: with open(KEYS_FILE) as f:
keys = json.load(f) keys = strip_key_metadata(json.load(f))
session_key_info = keys.get("session/session.db") session_key_info = get_key_info(keys, os.path.join("session", "session.db"))
if not session_key_info: if not session_key_info:
print("[ERROR] 找不到session.db的密钥") print("[ERROR] 找不到session.db的密钥")
sys.exit(1) sys.exit(1)
enc_key = bytes.fromhex(session_key_info["enc_key"]) enc_key = bytes.fromhex(session_key_info["enc_key"])
session_db = os.path.join(DB_DIR, "session", "session.db") session_db = os.path.join(DB_DIR, "session", "session.db")

41
monitor_web.py
View File

@ -17,6 +17,7 @@ import urllib.parse
import glob as glob_mod import glob as glob_mod
import zstandard as zstd import zstandard as zstd
from decode_image import extract_md5_from_packed_info, decrypt_dat_file, is_v2_format from decode_image import extract_md5_from_packed_info, decrypt_dat_file, is_v2_format
from key_utils import get_key_info, strip_key_metadata
_zstd_dctx = zstd.ZstdDecompressor() _zstd_dctx = zstd.ZstdDecompressor()
@ -62,7 +63,7 @@ def _build_emoji_lookup(keys_dict):
"""从 emoticon.db 构建 emoji md5 → URL 映射（直接解密，不走 cache）""" """从 emoticon.db 构建 emoji md5 → URL 映射（直接解密，不走 cache）"""
global _emoji_lookup, _emoji_keys_dict, _emoji_last_refresh global _emoji_lookup, _emoji_keys_dict, _emoji_last_refresh
_emoji_keys_dict = keys_dict _emoji_keys_dict = keys_dict
key_info = keys_dict.get("emoticon/emoticon.db") key_info = get_key_info(keys_dict, os.path.join("emoticon", "emoticon.db"))
if not key_info: if not key_info:
print("[emoji] 无 emoticon.db key，跳过", flush=True) print("[emoji] 无 emoticon.db key，跳过", flush=True)
return return
@ -254,13 +255,14 @@ class MonitorDBCache:
def get(self, rel_key): def get(self, rel_key):
"""返回解密后的临时文件路径，mtime 变化时自动重新解密""" """返回解密后的临时文件路径，mtime 变化时自动重新解密"""
if rel_key not in self.keys: key_info = get_key_info(self.keys, rel_key)
if not key_info:
return None return None
lock = self._get_lock(rel_key) lock = self._get_lock(rel_key)
with lock: with lock:
enc_key = bytes.fromhex(self.keys[rel_key]["enc_key"]) enc_key = bytes.fromhex(key_info["enc_key"])
rel_path = rel_key.replace('/', os.sep) rel_path = rel_key.replace('\\', '/').replace('/', os.sep)
db_path = os.path.join(DB_DIR, rel_path) db_path = os.path.join(DB_DIR, rel_path)
wal_path = db_path + "-wal" wal_path = db_path + "-wal"
@ -273,7 +275,7 @@ class MonitorDBCache:
except OSError: except OSError:
return None return None
out_name = rel_key.replace('/', '_') out_name = rel_key.replace('\\', '_').replace('/', '_')
out_path = os.path.join(self.tmp_dir, out_name) out_path = os.path.join(self.tmp_dir, out_name)
prev = self._state.get(rel_key) prev = self._state.get(rel_key)
@ -313,7 +315,7 @@ def build_username_db_map():
# 先获取每个 DB 的 mtime 用于排序 # 先获取每个 DB 的 mtime 用于排序
db_mtimes = {} db_mtimes = {}
for i in range(5): for i in range(5):
rel_key = f"message/message_{i}.db" rel_key = os.path.join("message", f"message_{i}.db")
db_path = os.path.join(DB_DIR, "message", f"message_{i}.db") db_path = os.path.join(DB_DIR, "message", f"message_{i}.db")
try: try:
db_mtimes[rel_key] = os.path.getmtime(db_path) db_mtimes[rel_key] = os.path.getmtime(db_path)
@ -326,7 +328,7 @@ def build_username_db_map():
db_path = os.path.join(decrypted_msg_dir, f"message_{i}.db") db_path = os.path.join(decrypted_msg_dir, f"message_{i}.db")
if not os.path.exists(db_path): if not os.path.exists(db_path):
continue continue
rel_key = f"message/message_{i}.db" rel_key = os.path.join("message", f"message_{i}.db")
try: try:
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
for row in conn.execute("SELECT user_name FROM Name2Id").fetchall(): for row in conn.execute("SELECT user_name FROM Name2Id").fetchall():
@ -595,7 +597,7 @@ class SessionMonitor:
# local_id 不全局唯一，需要同时匹配 create_time # local_id 不全局唯一，需要同时匹配 create_time
file_md5 = None file_md5 = None
for _try in range(2): for _try in range(2):
res_path = self.db_cache.get("message/message_resource.db") res_path = self.db_cache.get(os.path.join("message", "message_resource.db"))
if not res_path: if not res_path:
return None return None
try: try:
@ -620,7 +622,7 @@ class SessionMonitor:
except Exception as e: except Exception as e:
if 'malformed' in str(e) and _try == 0: if 'malformed' in str(e) and _try == 0:
print(f" [img] resource DB malformed, 强制刷新...", flush=True) print(f" [img] resource DB malformed, 强制刷新...", flush=True)
self.db_cache.invalidate("message/message_resource.db") self.db_cache.invalidate(os.path.join("message", "message_resource.db"))
continue continue
print(f" [img] 查询 message_resource 失败: {e}", flush=True) print(f" [img] 查询 message_resource 失败: {e}", flush=True)
return None return None
@ -753,10 +755,11 @@ class SessionMonitor:
def _fresh_decrypt_query(self, db_key, table_name, prev_ts, curr_ts): def _fresh_decrypt_query(self, db_key, table_name, prev_ts, curr_ts):
"""独立解密 message DB 到临时文件并查询，避免共享缓存竞态""" """独立解密 message DB 到临时文件并查询，避免共享缓存竞态"""
if db_key not in self.db_cache.keys: key_info = get_key_info(self.db_cache.keys, db_key)
if not key_info:
return [] return []
enc_key = bytes.fromhex(self.db_cache.keys[db_key]["enc_key"]) enc_key = bytes.fromhex(key_info["enc_key"])
rel_path = db_key.replace('/', os.sep) rel_path = db_key.replace('\\', '/').replace('/', os.sep)
db_path = os.path.join(DB_DIR, rel_path) db_path = os.path.join(DB_DIR, rel_path)
wal_path = db_path + "-wal" wal_path = db_path + "-wal"
if not os.path.exists(db_path): if not os.path.exists(db_path):
@ -1875,9 +1878,13 @@ def main():
print("=" * 60, flush=True) print("=" * 60, flush=True)
with open(KEYS_FILE) as f: with open(KEYS_FILE) as f:
keys = json.load(f) keys = strip_key_metadata(json.load(f))
enc_key = bytes.fromhex(keys["session/session.db"]["enc_key"]) session_key_info = get_key_info(keys, os.path.join("session", "session.db"))
if not session_key_info:
print("[ERROR] 找不到 session.db 的密钥", flush=True)
sys.exit(1)
enc_key = bytes.fromhex(session_key_info["enc_key"])
session_db = os.path.join(DB_DIR, "session", "session.db") session_db = os.path.join(DB_DIR, "session", "session.db")
print("加载联系人...", flush=True) print("加载联系人...", flush=True)
@ -1910,10 +1917,10 @@ def main():
def _warmup(): def _warmup():
try: try:
t0 = time.perf_counter() t0 = time.perf_counter()
warmup_keys = ["message/message_resource.db"] warmup_keys = [os.path.join("message", "message_resource.db")]
for i in range(5): for i in range(5):
k = f"message/message_{i}.db" k = os.path.join("message", f"message_{i}.db")
if k in keys: if get_key_info(keys, k):
warmup_keys.append(k) warmup_keys.append(k)
for k in warmup_keys: for k in warmup_keys:
t1 = time.perf_counter() t1 = time.perf_counter()