From ecc1cfca64542e19b2f796269c136b2e04c5d124 Mon Sep 17 00:00:00 2001 From: xincheng Date: Sun, 17 May 2026 06:24:08 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E4=BC=81=E4=B8=9A=E5=BE=AE?= =?UTF-8?q?=E4=BF=A1=E7=9A=84=E8=A7=A3=E5=AF=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 9 +- EXE_USAGE.md | 39 +- README.md | 41 +- WeChatDecrypt.spec | 2 +- app_gui.py | 275 ++++++++++++- build.bat | 4 + config.example.json | 2 + config.py | 7 +- decrypt_wxwork_db.py | 176 +++++++++ export_wxwork_messages.py | 745 ++++++++++++++++++++++++++++++++++++ find_wxwork_keys.py | 189 ++++++++- tests/test_wxwork_crypto.py | 48 +++ wxwork_crypto.py | 132 +++++++ 13 files changed, 1635 insertions(+), 34 deletions(-) create mode 100644 decrypt_wxwork_db.py create mode 100644 export_wxwork_messages.py create mode 100644 tests/test_wxwork_crypto.py create mode 100644 wxwork_crypto.py diff --git a/.gitignore b/.gitignore index 3537ab4..088c647 100644 --- a/.gitignore +++ b/.gitignore @@ -384,6 +384,9 @@ data/ export/ build/ dist/ -decrypted/ -all_keys.json -config.json \ No newline at end of file +decrypted/ +all_keys.json +wxwork_decrypted/ +wxwork_export/ +wxwork_keys.json +config.json diff --git a/EXE_USAGE.md b/EXE_USAGE.md index d72f7c0..df4c55e 100644 --- a/EXE_USAGE.md +++ b/EXE_USAGE.md @@ -2,17 +2,21 @@ ## 快速开始 -1. **启动微信**并登录账号 +1. **启动微信**并登录账号；如果要解密企业微信，也请先启动企业微信 2. 双击 `WeChatDecrypt.exe` 打开工具箱 -3. 按顺序点击三个按钮： - - **① 解密数据库** → 从微信进程提取密钥并解密数据库到 `decrypted/` 目录 - - **② 导出消息** → 将聊天记录导出为 CSV / HTML / JSON 到 `export/` 目录 - - **③ 转换音频** → 将语音消息从 SILK 格式转为 MP3 到 `data/` 目录 +3. 根据需要点击按钮： + - **① 微信解密** → 从微信进程提取密钥并解密数据库到 `decrypted/` 目录 + - **② 图片密钥** → 从微信进程提取新版图片 AES 密钥 + - **③ 导出数据** → 将聊天记录导出为 CSV / HTML / JSON 到 `export/` 目录 + - **④ 朋友圈图片** → 解密朋友圈缓存图片 + - **⑤ 企业微信解密** → 从企业微信进程提取密钥并解密数据库到 `wxwork_decrypted/` 目录 + - **⑥ 企业微信导出** → 选择某个人或群，导出 CSV / HTML / JSON 到 `wxwork_export/` 目录 ## 前置要求 - Windows 10 / 11 -- 微信 PC 版已登录（解密时需要微信进程运行） +- 微信 PC 版已登录（解密微信时需要微信进程运行） +- 企业微信 PC 版已登录（解密企业微信时需要企业微信进程运行） - [FFmpeg](https://ffmpeg.org/download.html) 已安装并加入 PATH（转换音频需要） ### 检查 FFmpeg @@ -31,7 +35,14 @@ ffmpeg -version WeChatDecrypt.exe config.json ← 首次运行自动生成的配置文件 decrypted/ ← ① 解密后的数据库文件 -export/ ← ② 导出的聊天记录 +wxwork_decrypted/ ← ⑤ 解密后的企业微信数据库文件 +wxwork_export/ ← ⑥ 导出的企业微信聊天记录 + 群名_R_123/ + .info + messages.csv + messages.html + messages.json +export/ ← ③ 导出的聊天记录 张三/ .info ← 联系人信息（username/alias/remark/nick_name） message_0.db.csv ← CSV 格式（Excel 可直接打开） @@ -39,7 +50,7 @@ export/ ← ② 导出的聊天记录 message_0.db.json← JSON 格式（程序处理用） 李四/ ... -data/ ← ③ 语音 MP3 文件 +data/ ← 导出时选择“同时转换语音为 MP3”后的输出 张三/ .info 20250101_120000_1.mp3 @@ -70,7 +81,11 @@ data/ ← ③ 语音 MP3 文件 "db_dir": "D:\\xwechat_files\\wxid_xxx\\db_storage", "keys_file": "all_keys.json", "decrypted_dir": "decrypted", - "wechat_process": "Weixin.exe" + "wechat_process": "Weixin.exe", + "wxwork_db_dir": "C:\\Users\\<用户>\\Documents\\WXWork\\\\Data", + "wxwork_keys_file": "wxwork_keys.json", + "wxwork_decrypted_dir": "wxwork_decrypted", + "wxwork_export_dir": "wxwork_export" } ``` @@ -85,6 +100,12 @@ A: 请确保微信 PC 版已启动并登录，然后重试。 **Q: 解密失败 / 密钥提取失败** A: 检查 `config.json` 中的 `db_dir` 是否与当前登录的微信账号匹配。切换账号后需要删除 `all_keys.json` 重新提取。 +**Q: 企业微信解密失败 / 找不到企业微信数据目录** +A: 确认企业微信 PC 版已启动并登录。若自动检测失败，请在 `config.json` 中设置 `wxwork_db_dir`，路径通常类似 `C:\Users\<用户>\Documents\WXWork\\Data`。切换企业微信账号后删除 `wxwork_keys.json` 重新提取。 + +**Q: 企业微信导出为空 / 找不到会话** +A: 先执行"⑤ 企业微信解密"，确认 `wxwork_decrypted/message.db` 和 `wxwork_decrypted/session.db` 存在，然后再执行"⑥ 企业微信导出"。 + **Q: 转换音频没有输出** A: 确认已安装 FFmpeg 并加入系统 PATH。确认已先执行"① 解密数据库"。 diff --git a/README.md b/README.md index 81a5961..38567b0 100644 --- a/README.md +++ b/README.md @@ -184,6 +184,9 @@ python find_image_key.py | `find_all_keys_windows.py` | Windows 版内存扫描提 key | | `find_all_keys_linux.py` | Linux 版内存扫描提 key | | `decrypt_db.py` | 全量解密所有数据库 | +| `find_wxwork_keys.py` | 企业微信 Windows 版内存扫描提 key | +| `decrypt_wxwork_db.py` | 企业微信 wxSQLite3 AES-128 数据库解密 | +| `export_wxwork_messages.py` | 企业微信聊天记录导出（按个人/群筛选，CSV / HTML / JSON） | | `mcp_server.py` | MCP Server，让 Claude AI 查询微信数据 | | `monitor_web.py` | 实时消息监听 (Web UI + SSE + 图片预览) | | `monitor.py` | 实时消息监听 (命令行) | @@ -197,11 +200,13 @@ python find_image_key.py ### GUI 工具箱 & 单 exe 打包 -提供 tkinter 图形界面 (`app_gui.py`)，集成三个核心功能： +提供 tkinter 图形界面 (`app_gui.py`)，集成核心功能： 1. **解密数据库** — 调用 `main.py decrypt` 2. **导出消息** — 调用 `export_messages.py`，输出 CSV / HTML / JSON 3. **转换音频** — 调用 `voice_to_mp3.py`，SILK_V3 → MP3 +4. **企业微信解密** — 调用 `find_wxwork_keys.py` + `decrypt_wxwork_db.py` +5. **企业微信导出** — 调用 `export_wxwork_messages.py`，按个人/群导出 CSV / HTML / JSON #### 直接运行 @@ -241,6 +246,40 @@ build.bat V2 文件结构: `[6B signature] [4B aes_size LE] [4B xor_size LE] [1B padding]` + `[AES-ECB encrypted] [raw unencrypted] [XOR encrypted]` +### 企业微信数据库解密 (实验) + +企业微信 Windows 5.x 的本地数据库不是普通微信 SQLCipher 4 格式，而是 wxSQLite3 AES-128-CBC： + +- 16 字节 raw key +- 每页按 page index + `sAlT` 派生 AES key +- 每页 IV 由 page index 派生 +- 无 SQLCipher HMAC / reserve 区 + +提取并解密： + +```bash +python find_wxwork_keys.py +python decrypt_wxwork_db.py +python export_wxwork_messages.py +``` + +如果自动提取失败但你已有 raw key，也可以直接传入 32 位 hex key： + +```bash +python decrypt_wxwork_db.py --key 00112233445566778899aabbccddeeff +``` + +配置项： + +```json +{ + "wxwork_db_dir": "C:\\Users\\<用户>\\Documents\\WXWork\\\\Data", + "wxwork_keys_file": "wxwork_keys.json", + "wxwork_decrypted_dir": "wxwork_decrypted", + "wxwork_export_dir": "wxwork_export" +} +``` + ### 数据库结构 解密后包含约 26 个数据库： diff --git a/WeChatDecrypt.spec b/WeChatDecrypt.spec index 96ab499..30fa867 100644 --- a/WeChatDecrypt.spec +++ b/WeChatDecrypt.spec @@ -1,7 +1,7 @@ # -*- mode: python ; coding: utf-8 -*- from PyInstaller.utils.hooks import collect_all -datas = [('main.py', '.'), ('config.py', '.'), ('decrypt_db.py', '.'), ('export_messages.py', '.'), ('voice_to_mp3.py', '.'), ('find_all_keys.py', '.'), ('find_all_keys_windows.py', '.'), ('find_all_keys_linux.py', '.'), ('key_scan_common.py', '.'), ('key_utils.py', '.'), ('decode_image.py', '.'), ('find_image_key.py', '.'), ('find_image_key_monitor.py', '.'), ('decrypt_sns.py', '.'), ('export_sns.py', '.'), ('monitor.py', '.'), ('monitor_web.py', '.'), ('mcp_server.py', '.'), ('config.example.json', '.')] +datas = [('main.py', '.'), ('config.py', '.'), ('decrypt_db.py', '.'), ('export_messages.py', '.'), ('voice_to_mp3.py', '.'), ('find_all_keys.py', '.'), ('find_all_keys_windows.py', '.'), ('find_all_keys_linux.py', '.'), ('find_wxwork_keys.py', '.'), ('decrypt_wxwork_db.py', '.'), ('export_wxwork_messages.py', '.'), ('wxwork_crypto.py', '.'), ('key_scan_common.py', '.'), ('key_utils.py', '.'), ('decode_image.py', '.'), ('find_image_key.py', '.'), ('find_image_key_monitor.py', '.'), ('decrypt_sns.py', '.'), ('export_sns.py', '.'), ('monitor.py', '.'), ('monitor_web.py', '.'), ('mcp_server.py', '.'), ('config.example.json', '.')] binaries = [] hiddenimports = [] tmp_ret = collect_all('pilk') diff --git a/app_gui.py b/app_gui.py index 4b00fe2..3a0a961 100644 --- a/app_gui.py +++ b/app_gui.py @@ -30,6 +30,8 @@ if False: # noqa: never executed, only for PyInstaller dependency detection import zstandard # noqa: F401 import pilk # noqa: F401 import Crypto, Crypto.Cipher, Crypto.Cipher.AES, Crypto.Util.Padding # noqa: F401 + import wxwork_crypto # noqa: F401 + import export_wxwork_messages # noqa: F401 def _run_subtask(task: str): @@ -59,6 +61,9 @@ def _run_subtask(task: str): "find_image_key": "find_image_key.py", "decrypt_sns": "decrypt_sns.py", "export_sns": "export_sns.py", + "find_wxwork_keys": "find_wxwork_keys.py", + "decrypt_wxwork": "decrypt_wxwork_db.py", + "export_wxwork": "export_wxwork_messages.py", } script = mapping.get(task) if not script: @@ -78,6 +83,12 @@ def _run_subtask(task: str): sys.argv = ["main.py", "decrypt"] elif task == "find_image_key": sys.argv = ["find_image_key.py"] + elif task == "find_wxwork_keys": + sys.argv = ["find_wxwork_keys.py"] + elif task == "decrypt_wxwork": + sys.argv = ["decrypt_wxwork_db.py"] + elif task == "export_wxwork": + sys.argv = ["export_wxwork_messages.py"] else: sys.argv = [script] @@ -360,11 +371,140 @@ class ExportOptionsDialog(tk.Toplevel): self.destroy() +class WxworkExportOptionsDialog(tk.Toplevel): + def __init__(self, parent, conversations): + """conversations: [{conversation_id, display_name, kind, message_count, last_time}, ...]""" + super().__init__(parent) + self.title("企业微信导出选项") + self.geometry("560x620") + self.transient(parent) + self.grab_set() + self.result = None + self.configure(bg="#f0f0f0") + self._conversations = conversations + self._vars = {} + + fmt_frame = ttk.LabelFrame(self, text="导出格式", padding=6) + fmt_frame.pack(fill="x", padx=12, pady=(10, 4)) + + self._fmt_csv = tk.BooleanVar(value=True) + self._fmt_html = tk.BooleanVar(value=False) + self._fmt_json = tk.BooleanVar(value=False) + + ttk.Checkbutton(fmt_frame, text="CSV（默认）", variable=self._fmt_csv).pack(side="left", padx=10) + ttk.Checkbutton(fmt_frame, text="HTML", variable=self._fmt_html).pack(side="left", padx=10) + ttk.Checkbutton(fmt_frame, text="JSON", variable=self._fmt_json).pack(side="left", padx=10) + + top = ttk.Frame(self) + top.pack(fill="x", padx=12, pady=(4, 4)) + ttk.Label(top, text=f"共 {len(conversations)} 个企业微信会话", + font=("Microsoft YaHei UI", 10)).pack(side="left") + + self._all_selected = True + self._toggle_btn = ttk.Button(top, text="取消全选", command=self._toggle_all) + self._toggle_btn.pack(side="right") + + search_frame = ttk.Frame(self) + search_frame.pack(fill="x", padx=12, pady=(0, 4)) + self._search_var = tk.StringVar() + self._search_var.trace_add("write", lambda *_: self._filter_list()) + ttk.Entry(search_frame, textvariable=self._search_var, + font=("Microsoft YaHei UI", 10)).pack(fill="x") + + container = ttk.Frame(self) + container.pack(fill="both", expand=True, padx=12, pady=4) + + self._canvas = tk.Canvas(container, bg="#ffffff", highlightthickness=0) + scrollbar = ttk.Scrollbar(container, orient="vertical", command=self._canvas.yview) + self._inner = ttk.Frame(self._canvas) + self._inner.bind("", + lambda e: self._canvas.configure(scrollregion=self._canvas.bbox("all"))) + self._canvas.create_window((0, 0), window=self._inner, anchor="nw") + self._canvas.configure(yscrollcommand=scrollbar.set) + self._canvas.pack(side="left", fill="both", expand=True) + scrollbar.pack(side="right", fill="y") + + self._canvas.bind("", lambda e: self._bind_mousewheel()) + self._canvas.bind("", lambda e: self._unbind_mousewheel()) + + self._cb_widgets = [] + for conv in conversations: + cid = conv["conversation_id"] + var = tk.BooleanVar(value=True) + self._vars[cid] = var + last_time = self._format_time(conv.get("last_time")) + suffix = f" · {last_time}" if last_time else "" + label = ( + f"[{conv.get('kind', '会话')}] {conv.get('display_name') or cid}" + f" · {conv.get('message_count', 0)} 条{suffix}" + ) + cb = ttk.Checkbutton(self._inner, text=label, variable=var) + cb.pack(anchor="w", padx=6, pady=1) + self._cb_widgets.append((cid, label.lower(), cb)) + + bottom = ttk.Frame(self) + bottom.pack(fill="x", padx=12, pady=(4, 10)) + ttk.Button(bottom, text="确定", command=self._on_ok).pack(side="right", padx=4) + ttk.Button(bottom, text="取消", command=self._on_cancel).pack(side="right", padx=4) + + def _format_time(self, value): + if not value: + return "" + try: + from datetime import datetime + return datetime.fromtimestamp(int(value)).strftime("%Y-%m-%d") + except Exception: + return "" + + def _bind_mousewheel(self): + self._canvas.bind_all("", + lambda e: self._canvas.yview_scroll(-1 * (e.delta // 120), "units")) + + def _unbind_mousewheel(self): + self._canvas.unbind_all("") + + def _toggle_all(self): + self._all_selected = not self._all_selected + for var in self._vars.values(): + var.set(self._all_selected) + self._toggle_btn.configure(text="取消全选" if self._all_selected else "全选") + + def _filter_list(self): + keyword = self._search_var.get().strip().lower() + for _cid, label, cb in self._cb_widgets: + if not keyword or keyword in label: + cb.pack(anchor="w", padx=6, pady=1) + else: + cb.pack_forget() + + def _on_ok(self): + formats = [] + if self._fmt_csv.get(): + formats.append("csv") + if self._fmt_html.get(): + formats.append("html") + if self._fmt_json.get(): + formats.append("json") + if not formats: + from tkinter import messagebox + messagebox.showwarning("提示", "请至少选择一种导出格式", parent=self) + return + self.result = { + "conversations": [cid for cid, var in self._vars.items() if var.get()], + "formats": formats, + } + self.destroy() + + def _on_cancel(self): + self.result = None + self.destroy() + + class App(tk.Tk): def __init__(self): super().__init__() self.title("WeChat Decrypt 工具箱") - self.geometry("750x520") + self.geometry("820x600") self.resizable(True, True) self.configure(bg="#f0f0f0") self._running = False @@ -375,6 +515,8 @@ class App(tk.Tk): self._include_images = True self._include_sns = False self._include_sns_media = False + self._selected_wxwork_conversations = None + self._wxwork_export_formats = None self._build_ui() @@ -394,13 +536,13 @@ class App(tk.Tk): btn_frame.pack(fill="x", padx=20, pady=(4, 4)) self.btn_decrypt = ttk.Button( - btn_frame, text="① 解密数据库", style="Big.TButton", + btn_frame, text="① 微信解密", style="Big.TButton", command=lambda: self._run_task("decrypt") ) self.btn_decrypt.pack(side="left", expand=True, fill="x", padx=4) self.btn_imgkey = ttk.Button( - btn_frame, text="② 查找图片密钥", style="Big.TButton", + btn_frame, text="② 图片密钥", style="Big.TButton", command=lambda: self._run_task("find_image_key") ) self.btn_imgkey.pack(side="left", expand=True, fill="x", padx=4) @@ -417,17 +559,34 @@ class App(tk.Tk): ) self.btn_sns.pack(side="left", expand=True, fill="x", padx=4) + wxwork_frame = ttk.Frame(self) + wxwork_frame.pack(fill="x", padx=20, pady=(0, 6)) + + self.btn_wxwork = ttk.Button( + wxwork_frame, text="⑤ 企业微信解密", style="Big.TButton", + command=lambda: self._run_task("wxwork_decrypt") + ) + self.btn_wxwork.pack(side="left", expand=True, fill="x", padx=4) + + self.btn_wxwork_export = ttk.Button( + wxwork_frame, text="⑥ 企业微信导出", style="Big.TButton", + command=lambda: self._run_task("wxwork_export") + ) + self.btn_wxwork_export.pack(side="left", expand=True, fill="x", padx=4) + # 提示信息 tips_frame = ttk.LabelFrame(self, text="使用提示", padding=6) tips_frame.pack(fill="x", padx=20, pady=(0, 4)) tips_text = ( - "• 解密数据库：需要微信正在运行中，会自动提取密钥并解密\n" + "• 微信解密：需要微信正在运行中，会自动提取密钥并解密\n" "• 查找图片密钥：先在微信中打开 2-3 张图片查看，然后立即运行\n" "• 导出数据：选择联系人和格式，可同时导出消息/图片/语音\n" - "• 朋友圈图片：解密朋友圈缓存图片（_t缩略图自动跳过）" + "• 朋友圈图片：解密朋友圈缓存图片（_t缩略图自动跳过）\n" + "• 企业微信解密：需要企业微信正在运行中，输出到 wxwork_decrypted/\n" + "• 企业微信导出：选择某个人或群，输出 CSV / HTML / JSON 到 wxwork_export/" ) ttk.Label(tips_frame, text=tips_text, font=("Microsoft YaHei UI", 9), - wraplength=680, justify="left").pack(anchor="w") + wraplength=760, justify="left").pack(anchor="w") # 进度条 self.progress = ttk.Progressbar(self, mode="indeterminate") @@ -468,6 +627,8 @@ class App(tk.Tk): self.btn_imgkey.configure(state=state) self.btn_export.configure(state=state) self.btn_sns.configure(state=state) + self.btn_wxwork.configure(state=state) + self.btn_wxwork_export.configure(state=state) # ── 任务调度 ─────────────────────────────────────────────────────────── def _run_task(self, task: str): @@ -479,6 +640,8 @@ class App(tk.Tk): self._include_voice = False self._include_sns = False self._include_sns_media = False + self._selected_wxwork_conversations = None + self._wxwork_export_formats = None self._clear_log() self._set_buttons(False) @@ -496,6 +659,14 @@ class App(tk.Tk): self.progress.start(15) self.status_var.set("正在解密朋友圈图片...") threading.Thread(target=self._exec_task, args=(task,), daemon=True).start() + elif task == "wxwork_decrypt": + self.progress.start(15) + self.status_var.set("正在解密企业微信数据库...") + threading.Thread(target=self._exec_wxwork_decrypt, daemon=True).start() + elif task == "wxwork_export": + self.progress.start(15) + self.status_var.set("正在扫描企业微信会话...") + threading.Thread(target=self._discover_wxwork_and_select, daemon=True).start() else: self.progress.start(15) labels = {"decrypt": "解密数据库"} @@ -555,6 +726,51 @@ class App(tk.Tk): self.status_var.set(f"正在{action}...（{n_sel}/{len(contacts)} 个联系人）") threading.Thread(target=self._exec_combined, daemon=True).start() + def _discover_wxwork_and_select(self): + """后台扫描企业微信会话，然后在主线程弹出选择对话框""" + try: + from export_wxwork_messages import discover_conversations + conversations = discover_conversations() + except Exception as e: + self.after(0, self._log, f"扫描企业微信会话失败: {e}\n") + self.after(0, self._on_task_done) + return + + if not conversations: + self.after(0, self._log, "未找到任何企业微信会话，请先运行「企业微信解密」\n") + self.after(0, self._on_task_done) + return + + self.after(0, self._show_wxwork_dialog, conversations) + + def _show_wxwork_dialog(self, conversations): + self.progress.stop() + self.status_var.set(f"请选择企业微信导出选项 ({len(conversations)} 个会话)") + + dlg = WxworkExportOptionsDialog(self, conversations) + self.wait_window(dlg) + + if dlg.result is None: + self._on_task_done() + return + + if not dlg.result["conversations"]: + self._log("未选择任何企业微信会话\n") + self._on_task_done() + return + + self._selected_wxwork_conversations = dlg.result["conversations"] + self._wxwork_export_formats = dlg.result["formats"] + + self._clear_log() + self.progress.start(15) + n_sel = len(dlg.result["conversations"]) + self.status_var.set( + f"正在导出企业微信 {'/'.join(f.upper() for f in self._wxwork_export_formats)}..." + f"（{n_sel}/{len(conversations)} 个会话）" + ) + threading.Thread(target=self._exec_wxwork_export, daemon=True).start() + # ── 子进程执行 ───────────────────────────────────────────────────────── def _run_subprocess(self, task: str) -> int: """运行子进程，返回退出码""" @@ -572,6 +788,10 @@ class App(tk.Tk): env["WECHAT_EXPORT_IMAGES"] = "1" if getattr(self, '_include_images', True) else "0" if getattr(self, '_include_sns_media', False): env["WECHAT_SNS_DOWNLOAD_MEDIA"] = "1" + if self._selected_wxwork_conversations: + env["WXWORK_EXPORT_CONVERSATIONS"] = ",".join(self._selected_wxwork_conversations) + if self._wxwork_export_formats: + env["WXWORK_EXPORT_FORMATS"] = ",".join(self._wxwork_export_formats) proc = subprocess.Popen( cmd, @@ -631,6 +851,49 @@ class App(tk.Tk): self._include_sns_media = False self.after(0, self._on_task_done) + def _exec_wxwork_decrypt(self): + """执行企业微信 key 提取 + 数据库解密。""" + try: + self.after(0, self._log, "━━━ 开始提取企业微信密钥 ━━━\n\n") + rc = self._run_subprocess("find_wxwork_keys") + if rc != 0: + self.after(0, self._log, f"\n❌ 企业微信密钥提取失败 (返回码 {rc})\n") + self.after(0, self.status_var.set, f"企业微信密钥提取失败 (返回码 {rc})") + return + + self.after(0, self._log, "\n\n━━━ 开始解密企业微信数据库 ━━━\n\n") + rc = self._run_subprocess("decrypt_wxwork") + if rc != 0: + self.after(0, self._log, f"\n❌ 企业微信数据库解密失败 (返回码 {rc})\n") + self.after(0, self.status_var.set, f"企业微信解密失败 (返回码 {rc})") + return + + self.after(0, self._log, "\n✅ 企业微信解密完成！输出目录: wxwork_decrypted\n") + self.after(0, self.status_var.set, "企业微信解密完成") + except Exception as e: + self.after(0, self._log, f"\n❌ 异常: {e}\n") + self.after(0, self.status_var.set, "异常") + finally: + self.after(0, self._on_task_done) + + def _exec_wxwork_export(self): + """执行企业微信消息导出。""" + try: + rc = self._run_subprocess("export_wxwork") + if rc != 0: + self.after(0, self._log, f"\n❌ 企业微信导出失败 (返回码 {rc})\n") + self.after(0, self.status_var.set, f"企业微信导出失败 (返回码 {rc})") + return + self.after(0, self._log, "\n✅ 企业微信导出完成！输出目录: wxwork_export\n") + self.after(0, self.status_var.set, "企业微信导出完成") + except Exception as e: + self.after(0, self._log, f"\n❌ 异常: {e}\n") + self.after(0, self.status_var.set, "异常") + finally: + self._selected_wxwork_conversations = None + self._wxwork_export_formats = None + self.after(0, self._on_task_done) + def _exec_task(self, task: str): """执行单一任务（解密）""" try: diff --git a/build.bat b/build.bat index d0b43eb..25a3c07 100644 --- a/build.bat +++ b/build.bat @@ -24,6 +24,10 @@ pyinstaller --noconfirm --onefile --console --name "WeChatDecrypt" ^ --add-data "find_all_keys.py;." ^ --add-data "find_all_keys_windows.py;." ^ --add-data "find_all_keys_linux.py;." ^ + --add-data "find_wxwork_keys.py;." ^ + --add-data "decrypt_wxwork_db.py;." ^ + --add-data "export_wxwork_messages.py;." ^ + --add-data "wxwork_crypto.py;." ^ --add-data "key_scan_common.py;." ^ --add-data "key_utils.py;." ^ --add-data "decode_image.py;." ^ diff --git a/config.example.json b/config.example.json index df34721..6a9e592 100644 --- a/config.example.json +++ b/config.example.json @@ -5,5 +5,7 @@ "wechat_process": "Weixin.exe", "wxwork_db_dir": "", "wxwork_keys_file": "wxwork_keys.json", + "wxwork_decrypted_dir": "wxwork_decrypted", + "wxwork_export_dir": "wxwork_export", "wxwork_process": "WXWork.exe" } diff --git a/config.py b/config.py index 7398c2f..c738ee2 100644 --- a/config.py +++ b/config.py @@ -43,6 +43,8 @@ _DEFAULT = { "wechat_process": _DEFAULT_PROCESS, "wxwork_db_dir": "", "wxwork_keys_file": "wxwork_keys.json", + "wxwork_decrypted_dir": "wxwork_decrypted", + "wxwork_export_dir": "wxwork_export", "wxwork_process": "WXWork.exe", } @@ -216,7 +218,10 @@ def load_config(): # 将相对路径转为绝对路径 base = _app_base_dir() - for key in ("keys_file", "decrypted_dir", "decoded_image_dir", "wxwork_keys_file"): + for key in ( + "keys_file", "decrypted_dir", "decoded_image_dir", + "wxwork_keys_file", "wxwork_decrypted_dir", "wxwork_export_dir", + ): if key in cfg and cfg[key] and not os.path.isabs(cfg[key]): cfg[key] = os.path.join(base, cfg[key]) diff --git a/decrypt_wxwork_db.py b/decrypt_wxwork_db.py new file mode 100644 index 0000000..3894d4b --- /dev/null +++ b/decrypt_wxwork_db.py @@ -0,0 +1,176 @@ +""" +Decrypt WXWork databases encrypted with wxSQLite3 AES-128-CBC. + +This handles the database page format. A 16-byte raw key is still required, +either from wxwork_keys.json or via --key. +""" +import argparse +import json +import os +import shutil +import sys + +from key_utils import get_key_info, strip_key_metadata +from wxwork_crypto import ( + decrypt_wxwork_database, + is_plain_sqlite_page, + is_wxsqlite3_aes128_page1, + verify_sqlite_file, + verify_wxsqlite3_aes128_key, +) + + +def _app_paths(): + from config import _app_base_dir, _config_file_path + + return _app_base_dir(), _config_file_path() + + +def _load_config(): + base, config_file = _app_paths() + cfg = {} + if os.path.exists(config_file): + with open(config_file, encoding="utf-8") as f: + cfg = json.load(f) + + db_dir = cfg.get("wxwork_db_dir", "") + if not db_dir or not os.path.isdir(db_dir): + from find_wxwork_keys import auto_detect_wxwork_db_dir + + detected = auto_detect_wxwork_db_dir() + if detected: + db_dir = detected + else: + raise RuntimeError("wxwork_db_dir is not configured") + + keys_file = cfg.get("wxwork_keys_file", "wxwork_keys.json") + if not os.path.isabs(keys_file): + keys_file = os.path.join(base, keys_file) + + out_dir = cfg.get("wxwork_decrypted_dir", "wxwork_decrypted") + if not os.path.isabs(out_dir): + out_dir = os.path.join(base, out_dir) + + return { + "db_dir": db_dir, + "keys_file": keys_file, + "out_dir": out_dir, + "global_key": cfg.get("wxwork_db_key", ""), + } + + +def _parse_key_hex(value): + value = (value or "").strip() + if value.startswith("x'") and value.endswith("'"): + value = value[2:-1] + if len(value) != 32: + raise ValueError("WXWork wxSQLite3 AES-128 key must be 32 hex chars") + return bytes.fromhex(value) + + +def _load_keys(keys_file): + if not os.path.exists(keys_file): + return {} + with open(keys_file, encoding="utf-8") as f: + return strip_key_metadata(json.load(f)) + + +def _iter_db_files(db_dir): + for root, dirs, files in os.walk(db_dir): + dirs[:] = [d for d in dirs if d not in ("-journal",)] + for name in files: + if not name.endswith(".db") or name.endswith("-wal") or name.endswith("-shm"): + continue + path = os.path.join(root, name) + rel = os.path.relpath(path, db_dir) + yield rel, path + + +def main(argv=None): + parser = argparse.ArgumentParser(description="Decrypt WXWork wxSQLite3 AES-128 databases") + parser.add_argument("--key", help="16-byte raw key as 32 hex chars") + args = parser.parse_args(argv) + + cfg = _load_config() + db_dir = cfg["db_dir"] + out_dir = cfg["out_dir"] + keys_file = cfg["keys_file"] + keys = _load_keys(keys_file) + + global_key = None + key_arg = args.key or cfg.get("global_key") + if key_arg: + global_key = _parse_key_hex(key_arg) + + print("=" * 60) + print(" WXWork Database Decryptor") + print("=" * 60) + print(f"DB dir: {db_dir}") + print(f"Output: {out_dir}") + if keys: + print(f"Loaded {len(keys)} per-DB keys from {keys_file}") + elif global_key: + print("Using global key from argument/config") + else: + print(f"No key available. Run find_wxwork_keys.py or pass --key.") + return 1 + + os.makedirs(out_dir, exist_ok=True) + + success = 0 + copied = 0 + failed = 0 + for rel, path in sorted(_iter_db_files(db_dir)): + out_path = os.path.join(out_dir, rel) + with open(path, "rb") as f: + page1 = f.read(4096) + + if is_plain_sqlite_page(page1): + os.makedirs(os.path.dirname(out_path), exist_ok=True) + shutil.copy2(path, out_path) + copied += 1 + print(f"COPY: {rel} (plain SQLite)") + continue + + if not is_wxsqlite3_aes128_page1(page1): + failed += 1 + print(f"SKIP: {rel} (unknown encrypted format)") + continue + + key = global_key + key_info = get_key_info(keys, rel) if keys else None + if key_info: + try: + key = _parse_key_hex(key_info["enc_key"]) + except (KeyError, ValueError) as exc: + failed += 1 + print(f"FAIL: {rel} (bad key entry: {exc})") + continue + + if key is None: + failed += 1 + print(f"SKIP: {rel} (no key)") + continue + + if not verify_wxsqlite3_aes128_key(key, page1): + failed += 1 + print(f"FAIL: {rel} (key validation failed)") + continue + + try: + decrypt_wxwork_database(path, out_path, key) + tables = verify_sqlite_file(out_path) + success += 1 + table_preview = ", ".join(tables[:5]) + suffix = f" tables: {table_preview}" if table_preview else " no tables" + print(f"OK: {rel} ({suffix})") + except Exception as exc: + failed += 1 + print(f"FAIL: {rel} ({exc})") + + print(f"\nResult: {success} decrypted, {copied} copied, {failed} failed") + return 0 if failed == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/export_wxwork_messages.py b/export_wxwork_messages.py new file mode 100644 index 0000000..3065f43 --- /dev/null +++ b/export_wxwork_messages.py @@ -0,0 +1,745 @@ +"""导出企业微信消息记录到 CSV / HTML / JSON。 + +输入目录默认来自 wxwork_decrypted_dir，输出到 wxwork_export_dir。 +可用环境变量: + WXWORK_EXPORT_CONVERSATIONS=conversation_id1,conversation_id2 + WXWORK_EXPORT_FORMATS=csv,html,json +""" +import argparse +import csv +import json +import os +import re +import sqlite3 +import sys +from collections import defaultdict +from datetime import datetime +from html import escape + + +if sys.platform == "win32" and hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + + +MSG_TYPES = { + 0: "文本/混合", + 2: "文本", + 4: "图片", + 7: "语音", + 15: "图片/文件", + 38: "应用消息", + 40: "通话/音视频", + 503: "状态", + 1011: "会议通知", +} + +_MESSAGE_TABLES = ("message_table", "message_small_table", "kf_message_tableV1") + + +def _app_paths(): + from config import _app_base_dir, _config_file_path + + return _app_base_dir(), _config_file_path() + + +def _load_config(): + base, config_file = _app_paths() + cfg = {} + if os.path.exists(config_file): + with open(config_file, encoding="utf-8") as f: + cfg = json.load(f) + + decrypted_dir = cfg.get("wxwork_decrypted_dir", "wxwork_decrypted") + if not os.path.isabs(decrypted_dir): + decrypted_dir = os.path.join(base, decrypted_dir) + + output_dir = cfg.get("wxwork_export_dir", "wxwork_export") + if not os.path.isabs(output_dir): + output_dir = os.path.join(base, output_dir) + + db_dir = cfg.get("wxwork_db_dir", "") + return { + "base": base, + "decrypted_dir": decrypted_dir, + "output_dir": output_dir, + "self_id": _infer_self_id(db_dir), + } + + +def _infer_self_id(db_dir): + if not db_dir: + return None + parts = os.path.normpath(db_dir).split(os.sep) + for part in reversed(parts): + if part.isdigit() and len(part) >= 10: + return int(part) + return None + + +def _safe_dirname(name): + name = re.sub(r'[\\/:*?"<>|\r\n\t]+', "_", str(name)) + name = re.sub(r"\s+", " ", name).strip(" .") + return (name or "unknown")[:120] + + +def _table_exists(conn, table): + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", + (table,), + ).fetchone() + return row is not None + + +def _open_db(path): + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + return conn + + +def _load_user_map(decrypted_dir): + user_db = os.path.join(decrypted_dir, "user.db") + users = {} + if not os.path.exists(user_db): + return users + + conn = _open_db(user_db) + try: + if _table_exists(conn, "user_table"): + for row in conn.execute( + "SELECT id, name, real_name, account, external_corp_name, external_job " + "FROM user_table" + ): + name = row["real_name"] or row["name"] or row["account"] or "" + if row["external_corp_name"] and row["external_corp_name"] not in name: + name = f"{name} ({row['external_corp_name']})" if name else row["external_corp_name"] + if name: + users[int(row["id"])] = name + + if _table_exists(conn, "external_user_relation_v3"): + for row in conn.execute( + "SELECT user_id, remarks, real_remarks, corp_remark FROM external_user_relation_v3" + ): + name = row["real_remarks"] or row["remarks"] or row["corp_remark"] or "" + if name: + users[int(row["user_id"])] = name + finally: + conn.close() + return users + + +def _load_group_member_names(decrypted_dir): + session_db = os.path.join(decrypted_dir, "session.db") + members = defaultdict(dict) + if not os.path.exists(session_db): + return members + + conn = _open_db(session_db) + try: + if _table_exists(conn, "conversation_user_table"): + for row in conn.execute( + "SELECT conversation_id, user_id, nick_name FROM conversation_user_table" + ): + if row["nick_name"]: + members[row["conversation_id"]][int(row["user_id"])] = row["nick_name"] + + if _table_exists(conn, "conversation_member_nickname_table"): + # 该表使用 room_id，需要用 conversation_table.con_numeric_id 转成会话 ID。 + room_map = {} + if _table_exists(conn, "conversation_table"): + for row in conn.execute("SELECT con_numeric_id, id FROM conversation_table"): + room_map[int(row["con_numeric_id"])] = row["id"] + for row in conn.execute( + "SELECT room_id, userid, nickname FROM conversation_member_nickname_table" + ): + cid = room_map.get(int(row["room_id"])) + if cid and row["nickname"]: + members[cid][int(row["userid"])] = row["nickname"] + finally: + conn.close() + return members + + +def _conversation_kind(conversation_id): + if conversation_id.startswith("R:"): + return "群聊" + if conversation_id.startswith("S:"): + return "单聊" + if conversation_id.startswith("M:"): + return "微信联系人" + if conversation_id.startswith("O:"): + return "应用/公众号" + if conversation_id.startswith("Y:"): + return "系统会话" + return "其他" + + +def _name_from_conversation_id(conversation_id, user_map, self_id): + if conversation_id.startswith("S:"): + ids = [] + for value in conversation_id[2:].split("_"): + if value.isdigit(): + ids.append(int(value)) + other_ids = [uid for uid in ids if self_id is None or uid != self_id] + for uid in other_ids or ids: + if uid in user_map: + return user_map[uid] + + if ":" in conversation_id: + tail = conversation_id.split(":", 1)[1] + if tail.isdigit() and int(tail) in user_map: + return user_map[int(tail)] + + return conversation_id + + +def _load_message_counts(decrypted_dir): + msg_db = os.path.join(decrypted_dir, "message.db") + counts = defaultdict(int) + last_times = defaultdict(int) + if not os.path.exists(msg_db): + return counts, last_times + + conn = _open_db(msg_db) + try: + for table in _MESSAGE_TABLES: + if not _table_exists(conn, table): + continue + for row in conn.execute( + f'SELECT conversation_id, COUNT(*) AS c, MAX(send_time) AS t ' + f'FROM "{table}" GROUP BY conversation_id' + ): + cid = row["conversation_id"] + if not cid: + continue + counts[cid] += int(row["c"] or 0) + last_times[cid] = max(last_times[cid], int(row["t"] or 0)) + finally: + conn.close() + return counts, last_times + + +def discover_conversations(decrypted_dir=None): + cfg = _load_config() + if decrypted_dir is None: + decrypted_dir = cfg["decrypted_dir"] + if not os.path.isdir(decrypted_dir): + raise FileNotFoundError(f"企业微信解密目录不存在: {decrypted_dir}") + + user_map = _load_user_map(decrypted_dir) + counts, message_last_times = _load_message_counts(decrypted_dir) + session_db = os.path.join(decrypted_dir, "session.db") + conversations = {} + + if os.path.exists(session_db): + conn = _open_db(session_db) + try: + if _table_exists(conn, "conversation_table"): + for row in conn.execute( + "SELECT id, name, roomname_remark, last_message_time, last_message_id " + "FROM conversation_table" + ): + cid = row["id"] + if not cid: + continue + raw_name = row["roomname_remark"] or row["name"] or "" + display = raw_name or _name_from_conversation_id( + cid, user_map, cfg["self_id"] + ) + last_time = max( + int(row["last_message_time"] or 0), + message_last_times.get(cid, 0), + ) + conversations[cid] = { + "conversation_id": cid, + "display_name": display, + "kind": _conversation_kind(cid), + "message_count": counts.get(cid, 0), + "last_time": last_time, + "last_message_id": int(row["last_message_id"] or 0), + } + finally: + conn.close() + + for cid, count in counts.items(): + if cid in conversations: + conversations[cid]["message_count"] = count + conversations[cid]["last_time"] = max( + conversations[cid]["last_time"], message_last_times.get(cid, 0) + ) + continue + conversations[cid] = { + "conversation_id": cid, + "display_name": _name_from_conversation_id(cid, user_map, cfg["self_id"]), + "kind": _conversation_kind(cid), + "message_count": count, + "last_time": message_last_times.get(cid, 0), + "last_message_id": 0, + } + + result = [c for c in conversations.values() if c["message_count"] > 0] + result.sort(key=lambda c: (c["last_time"], c["message_count"]), reverse=True) + return result + + +def _read_varint(data, pos): + value = 0 + shift = 0 + while pos < len(data) and shift < 64: + b = data[pos] + pos += 1 + value |= (b & 0x7F) << shift + if not (b & 0x80): + return value, pos + shift += 7 + raise ValueError("bad varint") + + +def _clean_text(text): + text = "".join( + ch if ch in "\n\t" or (ch.isprintable() and ch not in "\x0b\x0c") else " " + for ch in text + ) + text = re.sub(r"[ \t]+", " ", text) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +def _looks_like_plain_text(data, text): + if not text: + return False + control = sum(1 for b in data if b < 32 and b not in (9, 10, 13)) + if control / max(len(data), 1) > 0.08: + return False + printable = sum(1 for ch in text if ch.isprintable() or ch in "\n\t") + return printable / max(len(text), 1) > 0.9 + + +def _decode_text_segment(segment): + if not segment or b"\x00" in segment: + return None + try: + text = segment.decode("utf-8") + except UnicodeDecodeError: + return None + text = _clean_text(text) + if len(text) < 2: + return None + if re.fullmatch(r"[0-9a-fA-F]{32,}", text): + return None + printable = sum(1 for ch in text if ch.isprintable() or ch in "\n\t") + if printable / max(len(text), 1) < 0.9: + return None + return text + + +def _parse_protobuf_strings(data, depth=0): + if depth > 4 or not data: + return [] + pos = 0 + out = [] + fields = 0 + try: + while pos < len(data): + tag, pos = _read_varint(data, pos) + if tag == 0: + return [] + wire = tag & 7 + fields += 1 + if wire == 0: + _, pos = _read_varint(data, pos) + elif wire == 1: + pos += 8 + elif wire == 5: + pos += 4 + elif wire == 2: + length, pos = _read_varint(data, pos) + if length < 0 or pos + length > len(data): + return [] + segment = data[pos:pos + length] + pos += length + text = _decode_text_segment(segment) + if text: + out.append(text) + else: + out.extend(_parse_protobuf_strings(segment, depth + 1)) + else: + return [] + if pos > len(data): + return [] + except Exception: + return [] + return out if fields else [] + + +def _dedupe_texts(values): + seen = set() + out = [] + for value in values: + value = _clean_text(value) + if not value or value in seen: + continue + seen.add(value) + out.append(value) + return out + + +def decode_content(raw): + if raw is None: + return "" + if isinstance(raw, str): + return _clean_text(raw) + data = bytes(raw) + if not data: + return "" + + try: + plain = data.decode("utf-8") + if _looks_like_plain_text(data, plain): + return _clean_text(plain) + except UnicodeDecodeError: + pass + + texts = _dedupe_texts(_parse_protobuf_strings(data)) + if texts: + return "\n".join(texts[:12]) + + for enc in ("utf-8", "gbk", "utf-16le"): + try: + text = _clean_text(data.decode(enc, errors="replace")) + if text and "\ufffd" not in text[:20]: + return text[:2000] + except Exception: + continue + return f"[二进制内容 {len(data)} 字节]" + + +def _format_time(ts): + try: + ts = int(ts or 0) + except (TypeError, ValueError): + ts = 0 + if ts <= 0: + return "" + if ts > 20_000_000_000: + ts = ts / 1000 + return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") + + +def _message_type_name(content_type): + return MSG_TYPES.get(int(content_type or 0), f"未知({content_type})") + + +def _display_message_content(content_type, content, extra_content, local_extra_content): + text = content or extra_content or local_extra_content + if text: + return text + return f"[{_message_type_name(content_type)}]" + + +def _build_message(row, conv_map, user_map, member_names, self_id): + cid = row["conversation_id"] + sender_id = int(row["sender_id"] or 0) + sender = member_names.get(cid, {}).get(sender_id) or user_map.get(sender_id) + if self_id is not None and sender_id == self_id: + sender = "我" + if not sender: + sender = str(sender_id) if sender_id else "系统" + + content = decode_content(row["content"]) + extra_content = decode_content(row["extra_content"]) + local_extra_content = decode_content(row["local_extra_content"]) + content_type = int(row["content_type"] or 0) + conv = conv_map.get(cid, {}) + + return { + "source_table": row["source_table"], + "message_id": int(row["message_id"] or 0), + "server_id": int(row["server_id"] or 0), + "sequence": int(row["sequence"] or 0), + "conversation_id": cid, + "conversation": conv.get("display_name") or cid, + "conversation_kind": conv.get("kind") or _conversation_kind(cid), + "sender_id": sender_id, + "sender": sender, + "content_type": content_type, + "type_name": _message_type_name(content_type), + "send_time": int(row["send_time"] or 0), + "time": _format_time(row["send_time"]), + "flag": int(row["flag"] or 0), + "content": content, + "extra_content": extra_content, + "local_extra_content": local_extra_content, + "display_content": _display_message_content( + content_type, content, extra_content, local_extra_content + ), + "is_sent": self_id is not None and sender_id == self_id, + } + + +def _iter_message_rows(message_db, selected_ids=None): + selected_ids = set(selected_ids or []) + conn = _open_db(message_db) + try: + for table in _MESSAGE_TABLES: + if not _table_exists(conn, table): + continue + where = "" + params = [] + if selected_ids: + placeholders = ",".join("?" for _ in selected_ids) + where = f"WHERE conversation_id IN ({placeholders})" + params = list(selected_ids) + sql = ( + f'SELECT "{table}" AS source_table, message_id, server_id, sequence, ' + f"sender_id, conversation_id, content_type, send_time, flag, " + f"content, extra_content, local_extra_content " + f'FROM "{table}" {where} ' + f"ORDER BY send_time, sequence, message_id" + ) + yield from conn.execute(sql, params) + finally: + conn.close() + + +HTML_TEMPLATE = """ + + + + +{title} + + + +
{title}
{meta}
+
+{body} +
+ + +""" + + +def _write_html(path, conv, messages): + parts = [] + last_day = None + is_group = conv.get("kind") == "群聊" + for msg in messages: + day = msg["time"][:10] if msg["time"] else "" + if day and day != last_day: + parts.append(f'
{escape(day)}
') + last_day = day + + side = "sent" if msg["is_sent"] else "received" + sender_label = "" + if is_group or not msg["is_sent"]: + sender_label = f'
{escape(msg["sender"])}
' + initial = escape((msg["sender"] or "?")[0].upper()) + content = escape(msg["display_content"] or "") + type_line = escape(f'{msg["type_name"]} · {msg["time"]}') + parts.append( + f'
' + f'
{initial}
' + f'
{sender_label}' + f'
{content}
' + f'
{type_line}
' + f'
' + ) + + meta = f'{conv.get("kind", "")} · {len(messages)} 条消息 · {conv["conversation_id"]}' + with open(path, "w", encoding="utf-8") as f: + f.write( + HTML_TEMPLATE.format( + title=escape(conv["display_name"]), + meta=escape(meta), + body="\n".join(parts), + ) + ) + + +def _write_csv(path, messages): + with open(path, "w", newline="", encoding="utf-8-sig") as f: + writer = csv.writer(f) + writer.writerow([ + "时间", "会话", "会话ID", "发送者", "发送者ID", "消息类型", + "内容", "message_id", "server_id", "sequence", "flag", + ]) + for msg in messages: + writer.writerow([ + msg["time"], + msg["conversation"], + msg["conversation_id"], + msg["sender"], + msg["sender_id"], + msg["type_name"], + msg["display_content"], + msg["message_id"], + msg["server_id"], + msg["sequence"], + msg["flag"], + ]) + + +def _write_json(path, conv, messages): + with open(path, "w", encoding="utf-8") as f: + json.dump( + { + "conversation": conv, + "message_count": len(messages), + "messages": messages, + }, + f, + ensure_ascii=False, + indent=2, + ) + + +def _selected_from_env(): + raw = os.environ.get("WXWORK_EXPORT_CONVERSATIONS", "").strip() + if not raw: + return None + return {item.strip() for item in raw.split(",") if item.strip()} + + +def _formats_from_env(): + raw = os.environ.get("WXWORK_EXPORT_FORMATS", "").strip() + if not raw: + return {"csv"} + formats = {item.strip().lower() for item in raw.split(",") if item.strip()} + valid = {"csv", "html", "json"} + return formats & valid or {"csv"} + + +def export_messages(selected_ids=None, formats=None): + cfg = _load_config() + decrypted_dir = cfg["decrypted_dir"] + output_dir = cfg["output_dir"] + message_db = os.path.join(decrypted_dir, "message.db") + + if not os.path.isdir(decrypted_dir): + raise FileNotFoundError(f"企业微信解密目录不存在: {decrypted_dir}") + if not os.path.exists(message_db): + raise FileNotFoundError(f"企业微信消息库不存在: {message_db}") + + formats = formats or _formats_from_env() + selected_ids = selected_ids if selected_ids is not None else _selected_from_env() + conversations = discover_conversations(decrypted_dir) + conv_map = {conv["conversation_id"]: conv for conv in conversations} + if selected_ids: + missing = sorted(selected_ids - set(conv_map)) + if missing: + print(f"提示: {len(missing)} 个选择的会话没有消息或不存在") + + user_map = _load_user_map(decrypted_dir) + member_names = _load_group_member_names(decrypted_dir) + grouped = defaultdict(list) + seen = set() + + for row in _iter_message_rows(message_db, selected_ids): + key = (row["conversation_id"], row["message_id"], row["server_id"], row["sequence"]) + if key in seen: + continue + seen.add(key) + msg = _build_message(row, conv_map, user_map, member_names, cfg["self_id"]) + grouped[msg["conversation_id"]].append(msg) + + os.makedirs(output_dir, exist_ok=True) + + total_conversations = 0 + total_messages = 0 + for cid, messages in sorted( + grouped.items(), + key=lambda item: (conv_map.get(item[0], {}).get("last_time", 0), len(item[1])), + reverse=True, + ): + conv = conv_map.get(cid) or { + "conversation_id": cid, + "display_name": cid, + "kind": _conversation_kind(cid), + "message_count": len(messages), + "last_time": messages[-1]["send_time"] if messages else 0, + } + folder = _safe_dirname(f'{conv["display_name"]}_{cid}') + out_dir = os.path.join(output_dir, folder) + os.makedirs(out_dir, exist_ok=True) + + info_path = os.path.join(out_dir, ".info") + with open(info_path, "w", encoding="utf-8") as f: + f.write(f"conversation_id: {cid}\n") + f.write(f"display_name: {conv['display_name']}\n") + f.write(f"kind: {conv.get('kind', '')}\n") + f.write(f"message_count: {len(messages)}\n") + + if "csv" in formats: + _write_csv(os.path.join(out_dir, "messages.csv"), messages) + if "html" in formats: + _write_html(os.path.join(out_dir, "messages.html"), conv, messages) + if "json" in formats: + _write_json(os.path.join(out_dir, "messages.json"), conv, messages) + + total_conversations += 1 + total_messages += len(messages) + print(f" {conv['display_name']} ({cid}): {len(messages)} 条") + + print(f"\n完成: {total_conversations} 个会话, 共 {total_messages} 条消息") + print(f"输出目录: {os.path.abspath(output_dir)}") + return { + "conversation_count": total_conversations, + "message_count": total_messages, + "output_dir": os.path.abspath(output_dir), + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description="Export WXWork messages") + parser.add_argument("--list", action="store_true", help="list conversations and exit") + parser.add_argument( + "--conversation", + action="append", + help="conversation ID to export; can be passed multiple times", + ) + parser.add_argument("--formats", help="comma separated formats: csv,html,json") + args = parser.parse_args(argv) + + if args.list: + conversations = discover_conversations() + print(f"发现 {len(conversations)} 个有消息的企业微信会话") + for conv in conversations: + last_time = _format_time(conv["last_time"]) + print( + f"{conv['conversation_id']}\t{conv['message_count']}\t" + f"{last_time}\t{conv['kind']}\t{conv['display_name']}" + ) + return 0 + + selected = set(args.conversation) if args.conversation else None + formats = None + if args.formats: + formats = {item.strip().lower() for item in args.formats.split(",") if item.strip()} + export_messages(selected_ids=selected, formats=formats) + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Exception as exc: + print(f"导出失败: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/find_wxwork_keys.py b/find_wxwork_keys.py index c99c4db..78d3a81 100644 --- a/find_wxwork_keys.py +++ b/find_wxwork_keys.py @@ -1,14 +1,13 @@ """ 从企业微信(WXWork)进程内存中提取所有数据库的缓存raw key -企业微信使用 WCDB/SQLCipher 加密，但与个人微信参数不同： -- 个人微信: AES-256-CBC, 32字节密钥, HMAC-SHA512 -- 企业微信: AES-128-CBC, 16字节密钥, 每页根据page index重新派生IV/key - -密钥在内存中的缓存格式仍为 x''，但 key 为16字节(32 hex chars) +企业微信的本地数据库与个人微信不同。实测 Windows 版使用 wxSQLite3 +AES-128-CBC 页面加密：16 字节 raw key，每页按 page index 派生 AES key +和 IV；页面没有 SQLCipher HMAC/reserve 区。 """ import ctypes import ctypes.wintypes as wt +import bisect import functools import hashlib import hmac as hmac_mod @@ -21,6 +20,11 @@ import sys import time from key_scan_common import collect_db_files +from wxwork_crypto import ( + is_plain_sqlite_page, + is_wxsqlite3_aes128_page1, + verify_wxsqlite3_aes128_key, +) print = functools.partial(print, flush=True) @@ -67,12 +71,11 @@ def enum_regions(h): # ── 常量 ───────────────────────────────────────────────────────────── WXWORK_PROCESS = "WXWork.exe" -SQLITE_HEADER_HEX = b"SQLite format 3\x00".hex() PAGE_SZ = 4096 SALT_SZ = 16 -# 企业微信可能的加密参数组合 (按可能性排序) +# 旧版本/其他平台可能回落到 SQLCipher 参数，保留作兼容验证。 # (key_sz, hmac_hash_name, hmac_sz, pbkdf2_iter, reserve_sz) VERIFY_CONFIGS = [ # WCDB optimized cipher with AES-128, HMAC-SHA512 (最可能) @@ -89,6 +92,9 @@ VERIFY_CONFIGS = [ def verify_enc_key_wxwork(enc_key, db_page1): """尝试多种参数组合验证密钥，返回 (成功?, 使用的配置描述)""" + if len(enc_key) == 16 and verify_wxsqlite3_aes128_key(enc_key, db_page1): + return True, "wxSQLite3 AES-128-CBC, per-page MD5 key/IV, no HMAC" + key_sz = len(enc_key) for cfg_key_sz, hmac_hash, hmac_sz, iterations, reserve_sz in VERIFY_CONFIGS: if key_sz != cfg_key_sz: @@ -185,16 +191,20 @@ def auto_detect_wxwork_db_dir(): def filter_encrypted_dbs(db_files, salt_to_dbs): - """过滤掉未加密的数据库（salt 等于 SQLite header 的）""" + """过滤掉未加密的数据库。""" filtered_files = [ - entry for entry in db_files if entry[3] != SQLITE_HEADER_HEX + entry for entry in db_files if not is_plain_sqlite_page(entry[4]) ] filtered_salts = { - s: dbs for s, dbs in salt_to_dbs.items() if s != SQLITE_HEADER_HEX + s: dbs for s, dbs in salt_to_dbs.items() + if any(entry[3] == s and not is_plain_sqlite_page(entry[4]) for entry in db_files) } removed = len(db_files) - len(filtered_files) if removed: print(f"[*] 跳过 {removed} 个未加密数据库") + wxsqlite3_count = sum(1 for entry in filtered_files if is_wxsqlite3_aes128_page1(entry[4])) + if wxsqlite3_count: + print(f"[*] 检测到 {wxsqlite3_count} 个 wxSQLite3 AES-128 格式数据库") return filtered_files, filtered_salts @@ -282,6 +292,137 @@ def scan_memory_for_wxwork_keys(data, hex_re, db_files, salt_to_dbs, key_map, return matches +def _find_region(memory_regions, starts, addr, length=4): + idx = bisect.bisect_right(starts, addr) - 1 + if idx < 0: + return None + base, end, data = memory_regions[idx] + if base <= addr and addr + length <= end: + return base, end, data + return None + + +def _read_u32(memory_regions, starts, addr): + region = _find_region(memory_regions, starts, addr, 4) + if not region: + return None + base, _end, data = region + return struct.unpack_from("= 0 and off < max_off: + if time.time() - t0 > max_seconds: + print_fn( + f"[WARN] 结构体扫描超时: checked={checked}, " + f"ptr_hits={ptr_hits}, chain_hits={chain_hits}, key_tests={key_tests}" + ) + return key_tests + + # The AES-128 decrypt branch checks two non-zero flags at +0 and +4. + flag0, flag4 = struct.unpack_from("= 6: + key_tests += 1 + if _record_candidate_key( + enc_key, db_files, salt_to_dbs, key_map, + remaining_salts, pid, cipher_addr, + f"wxSQLite3 AES-128-CBC, page_size={page_size}", + print_fn, + ): + if not remaining_salts: + return key_tests + + checked += 1 + off += 4 + + print_fn( + f"[*] 结构体扫描完成: checked={checked}, ptr_hits={ptr_hits}, " + f"chain_hits={chain_hits}, key_tests={key_tests}" + ) + return key_tests + + def cross_verify_wxwork_keys(db_files, salt_to_dbs, key_map, print_fn): """用已找到的 key 交叉验证未匹配的 salt。""" missing_salts = set(salt_to_dbs.keys()) - set(key_map.keys()) @@ -395,11 +536,17 @@ def main(): # 2. 打开所有企业微信进程 pids = get_wxwork_pids() - # 宽松正则：匹配 32+ hex chars (16字节key起) + # Some versions do not keep the key as SQL literal x'...'. Bare ASCII + # hex scanning is much slower, so keep it behind an explicit switch. hex_re = re.compile(b"x'([0-9a-fA-F]{32,192})'") + scan_bare_hex = "--scan-bare-hex" in sys.argv + bare_hex_re = re.compile(b"(?= 512 + and page_size <= 65536 + and (page_size & (page_size - 1)) == 0 + and header[5] == 0x40 + and header[6] == 0x20 + and header[7] == 0x20 + ) + + +def is_wxsqlite3_aes128_page1(page): + return not is_plain_sqlite_page(page) and has_wxsqlite3_plain_header_fragment(page) + + +def _decrypt_aes128_cbc(raw_key, page_no, data): + page_key = derive_wxsqlite3_aes128_page_key(raw_key, page_no) + iv = generate_initial_vector(page_no) + return AES.new(page_key, AES.MODE_CBC, iv).decrypt(data) + + +def decrypt_wxsqlite3_aes128_page(raw_key, page_data, page_no): + """Decrypt one wxSQLite3 AES-128-CBC page to a normal SQLite page.""" + if len(page_data) != PAGE_SZ: + raise ValueError(f"page must be exactly {PAGE_SZ} bytes") + + data = bytearray(page_data) + if page_no == 1 and has_wxsqlite3_plain_header_fragment(data): + db_header_fragment = bytes(data[16:24]) + data[16:24] = data[8:16] + decrypted_tail = _decrypt_aes128_cbc(raw_key, page_no, bytes(data[16:])) + data[16:] = decrypted_tail + if bytes(data[16:24]) != db_header_fragment: + raise ValueError("wxSQLite3 AES-128 key validation failed") + data[:16] = SQLITE_HDR + return bytes(data) + + return _decrypt_aes128_cbc(raw_key, page_no, bytes(data)) + + +def looks_like_sqlite_page1(page): + if page[: len(SQLITE_HDR)] != SQLITE_HDR: + return False + if len(page) < 108: + return False + btree_page_type = page[100] + return btree_page_type in (0x02, 0x05, 0x0A, 0x0D) + + +def verify_wxsqlite3_aes128_key(raw_key, page1): + if len(raw_key) != 16 or len(page1) < PAGE_SZ: + return False + try: + decrypted = decrypt_wxsqlite3_aes128_page(raw_key, page1[:PAGE_SZ], 1) + except (ValueError, KeyError): + return False + return looks_like_sqlite_page1(decrypted) + + +def decrypt_wxwork_database(db_path, out_path, raw_key): + size = os.path.getsize(db_path) + total_pages = (size + PAGE_SZ - 1) // PAGE_SZ + os.makedirs(os.path.dirname(out_path), exist_ok=True) + + with open(db_path, "rb") as fin, open(out_path, "wb") as fout: + for page_no in range(1, total_pages + 1): + page = fin.read(PAGE_SZ) + if not page: + break + if len(page) < PAGE_SZ: + page += b"\x00" * (PAGE_SZ - len(page)) + fout.write(decrypt_wxsqlite3_aes128_page(raw_key, page, page_no)) + + +def verify_sqlite_file(path): + conn = sqlite3.connect(path) + try: + return [row[0] for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" + ).fetchall()] + finally: + conn.close()