From 0d8d9f111ece96d3457deae592c01ccd9113d067 Mon Sep 17 00:00:00 2001 From: Magicapple <1540796514@qq.com> Date: Tue, 22 Sep 2026 12:25:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E6=96=B0=E5=A2=9E=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=20WCDB=20=E5=8A=A0=E5=AF=86=E7=9A=84=E6=BC=94?= =?UTF-8?q?=E7=A4=BA=E6=95=B0=E6=8D=AE=E5=BA=93=E7=A7=8D=E5=AD=90=E8=84=9A?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 贡献者本地通常没有微信环境，也不应拿真实聊天记录做调试；已有的 tools/seed_ai_acceptance.py 生成的是明文 SQLite，只覆盖 AI 验收， 无法验证「密钥 -> 逐页 HMAC 校验 -> 解密 -> 解析」这条主链路。 新增 tools/seed_demo_database.py，生成完全虚构但真实加密的演示账号： - 逐页 AES-CBC + 逐页 HMAC-SHA512，密钥派生复用上游 wechat_decrypt 的常量与函数（PAGE_SIZE / RESERVE_SIZE / _derive_sqlcipher_enc_key / _derive_mac_key），上游调整页格式时脚本随之变化； - 5 个会话 29 条消息，覆盖文字、链接卡片 / 转账卡片（含 refermsg 引用）、 系统消息、表情与语音占位； - Name2Id 不含账号本人，与真实微信结构一致，避免会话列表多出一条自己； - 昵称统一带「示例」标记、域名 example.com，任何导出物都能一眼识别为虚构数据。 同时提供 --check 自校验：调用 scan_account_databases_from_path 与 decrypt_wechat_databases 走完整解密链路，逐页校验 HMAC，并检查解密后 PRAGMA integrity_check 与关键表是否存在。 回归测试 tests/test_seed_demo_database.py 守护「演示数据与真实解密链路一致」 这一契约，逐页比对 HMAC 与 _compute_page_hmac 的结果。 实测：--check 输出「成功解密 2/2 个」，上游解密器日志打印 Page 1 HMAC verification passed: mode=sqlcipher_passphrase， successful_pages=7 failed_pages=0，reserved_space=80。 默认输出到 output/demo，已被 /output/ 规则忽略，不进入版本库； 脚本只写 --output 指定目录，不读取也不修改任何真实微信目录。 --- docs/demo-database-seed.md | 91 ++++++ tests/test_seed_demo_database.py | 142 ++++++++++ tools/seed_demo_database.py | 465 +++++++++++++++++++++++++++++++ 3 files changed, 698 insertions(+) create mode 100644 docs/demo-database-seed.md create mode 100644 tests/test_seed_demo_database.py create mode 100644 tools/seed_demo_database.py diff --git a/docs/demo-database-seed.md b/docs/demo-database-seed.md new file mode 100644 index 00000000..668c1fcf --- /dev/null +++ b/docs/demo-database-seed.md @@ -0,0 +1,91 @@ +# 演示数据库种子脚本 + +`tools/seed_demo_database.py` 生成一个**完全虚构、但真实经过微信 4.x WCDB 加密**的演示账号， +用于开发、回归测试、文档截图与演示视频。 + +## 为什么需要它 + +项目里已有的 `tools/seed_ai_acceptance.py` 生成的是**明文 SQLite**，只覆盖 AI 验收这一条链路。 +但在以下场景里，明文库不够用： + +- 验证「密钥 -> 逐页 HMAC 校验 -> 解密 -> 解析 -> 展示」的完整链路； +- 排查解密失败时，需要一个**格式正确、内容已知**的对照样本； +- 贡献者本地没有微信环境，或不愿意用真实聊天记录做调试； +- 文档、官网、演示视频需要可以公开的素材。 + +真实聊天记录包含隐私，不适合进入任何公开产物；明文库又无法覆盖解密链路。 +本脚本填补的正是这个空档：**格式与真实库一致，内容 100% 虚构**。 + +## 生成的内容 + +``` +output/demo/ +├── demo_keys.json # 演示密钥（account -> key/display_name/alias） +└── databases/ + └── wxid_demo_2026/ + └── db_storage/ + ├── message/message_0.db # 加密，5 个会话共 29 条消息 + └── contact/contact.db # 加密，6 个联系人 / 1 个群 +``` + +覆盖的消息类型：文字（`1`）、链接卡片与转账卡片（`49`，含 `refermsg` 引用结构）、 +系统消息（`10000`）、表情（`47`）、语音占位（`34`）。 + +结构上的两点刻意设计： + +1. **`Name2Id` 不包含账号本人** —— 真实微信如此。账号本人由「出现在全部单聊表中的 + `real_sender_id`」判定；如果把自己的行写进 `Name2Id`，会话列表会多出一条自己和自己聊天的记录。 +2. **昵称全部带「示例」标记，域名统一 `example.com`，群号使用 `10000000001@chatroom`** —— + 确保任何截图或导出产物都能被一眼识别为虚构数据。 + +## 用法 + +```bash +# 生成到 output/demo（已被 /output/ 规则忽略，不会进入版本库） +python tools/seed_demo_database.py + +# 生成后用项目自身的扫描与解密器做端到端自校验 +python tools/seed_demo_database.py --check + +# 指定输出目录 +python tools/seed_demo_database.py --output /tmp/demo-account +``` + +`--check` 会调用 `scan_account_databases_from_path` 与 `decrypt_wechat_databases`， +逐页验证 HMAC、检查解密后 `PRAGMA integrity_check` 是否为 `ok`、以及 `Name2Id` / `Contact` +表是否存在。自校验通过时输出类似： + +``` +自校验通过：扫描到 5 个数据库，成功解密 2/2 个 +``` + +## 加密格式 + +脚本实现的是 `wechat_decrypt._decrypt_page` 的**逆过程**，参数直接从上游模块导入而非硬编码， +因此上游调整页格式时脚本会跟着一起变： + +| 项 | 值 | +| --- | --- | +| 页大小 | 4096 字节 | +| 密钥派生 | PBKDF2-HMAC-SHA512，256000 轮，32 字节 | +| mac 密钥 | PBKDF2-HMAC-SHA512(enc_key, salt ^ 0x3A, 2 轮) | +| 每页布局 | `ciphertext + iv(16) + hmac(64)`，第 1 页前加 16 字节 salt | +| HMAC 覆盖 | 密文 + IV + 小端页码（SHA-512） | +| 每页预留 | 80 字节（= IV 16 + HMAC 64） | + +明文库在加密前需要「每页预留 80 字节」的页面布局，这一步容易出错：仅靠 +`PRAGMA user_version` 让 SQLite 落盘文件头是不够的，还必须把 page1 btree 头里 +「cell 内容区起始」（文件偏移 `100 + 5`）从 4096 改成 4016，否则 SQLite 会按原始 +页头计算空闲空间并报 `database integrity check: database disk image is malformed`。 + +## 回归保护 + +`tests/test_seed_demo_database.py` 守护「演示数据与真实解密链路一致」这一契约： +逐页比对 HMAC 与 `_compute_page_hmac` 的结果、验证解密后是合法 SQLite、 +校验 `Name2Id` 不含本人、并确认昵称都带虚构标记。 + +## 安全边界 + +- 脚本只写 `--output` 指定的目录，**不读取、不修改**任何真实微信目录。 +- 演示密钥是固定常量（`sha256("wechat-data-analysis-demo-key")`），公开无风险； + 它只用于打开本脚本自己生成的样本库。 diff --git a/tests/test_seed_demo_database.py b/tests/test_seed_demo_database.py new file mode 100644 index 00000000..137af4a5 --- /dev/null +++ b/tests/test_seed_demo_database.py @@ -0,0 +1,142 @@ +"""回归：``tools/seed_demo_database.py`` 生成的演示库必须能被项目解密器读取。 + +这个测试守护的是「演示数据与真实解密链路一致」这一契约：如果 +``wechat_decrypt`` 的页面格式、HMAC 覆盖范围或预留字节数发生变化，而种子脚本 +没有同步，测试会立刻失败——避免演示环境悄悄偏离真实格式后又被人当成基准。 +""" + +from __future__ import annotations + +import sqlite3 +import sys +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "tools")) + +from seed_demo_database import ( # noqa: E402 + DEMO_ACCOUNT, + DEMO_GROUP, + DEMO_KEY, + encrypt_wcdb, + seed, +) + +from wechat_decrypt_tool.wechat_decrypt import ( # noqa: E402 + PAGE_SIZE, + RESERVE_SIZE, + SQLITE_HEADER, + _compute_page_hmac, + _derive_mac_key, + _derive_sqlcipher_enc_key, +) + + +def _decrypt_to_plain(encrypted: bytes, key_hex: str) -> bytes: + """按 ``wechat_decrypt._decrypt_page`` 的规则还原整库明文。""" + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + if len(encrypted) % PAGE_SIZE != 0: + raise AssertionError("加密库长度不是 4096 的整数倍") + + salt = encrypted[:16] + enc_key = _derive_sqlcipher_enc_key(bytes.fromhex(key_hex), salt) + plain = bytearray() + for page_num in range(1, len(encrypted) // PAGE_SIZE + 1): + start = (page_num - 1) * PAGE_SIZE + page = encrypted[start : start + PAGE_SIZE] + iv = page[PAGE_SIZE - RESERVE_SIZE : PAGE_SIZE - RESERVE_SIZE + 16] + offset = 16 if page_num == 1 else 0 + body = page[offset : PAGE_SIZE - RESERVE_SIZE] + cipher = Cipher( + algorithms.AES(enc_key), modes.CBC(iv), backend=default_backend() + ) + decryptor = cipher.decryptor() + decrypted = decryptor.update(body) + decryptor.finalize() + if page_num == 1: + plain += SQLITE_HEADER + decrypted + b"\x00" * RESERVE_SIZE + else: + plain += decrypted + b"\x00" * RESERVE_SIZE + return bytes(plain) + + +class TestSeedDemoDatabase(unittest.TestCase): + def setUp(self) -> None: + self._tmp = TemporaryDirectory() + self.output = Path(self._tmp.name) + self.paths = seed(self.output) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_generated_page_hmac_matches_decryptor(self): + """逐页 HMAC 必须与解密器的校验算法一致，否则真实解密会拒绝该库。""" + raw = self.paths["message_db"].read_bytes() + self.assertEqual(len(raw) % PAGE_SIZE, 0) + + salt = raw[:16] + enc_key = _derive_sqlcipher_enc_key(bytes.fromhex(DEMO_KEY), salt) + mac_key = _derive_mac_key(enc_key, salt) + + for page_num in range(1, len(raw) // PAGE_SIZE + 1): + start = (page_num - 1) * PAGE_SIZE + page = raw[start : start + PAGE_SIZE] + expected = _compute_page_hmac(mac_key, page, page_num) + stored = page[PAGE_SIZE - 64 :] + self.assertEqual( + stored, expected, f"第 {page_num} 页 HMAC 与解密器算法不一致" + ) + + def test_generated_databases_decrypt_to_valid_sqlite(self): + for key_name in ("message_db", "contact_db"): + path = self.paths[key_name] + plain = _decrypt_to_plain(path.read_bytes(), DEMO_KEY) + self.assertTrue(plain.startswith(SQLITE_HEADER)) + + decrypted_path = self.output / f"{key_name}.decrypted.db" + decrypted_path.write_bytes(plain) + conn = sqlite3.connect(decrypted_path) + try: + status = conn.execute("PRAGMA integrity_check").fetchone()[0] + finally: + conn.close() + self.assertEqual(status, "ok", f"{key_name} 解密后完整性检查失败") + + def test_demo_account_excludes_self_from_name2id(self): + """真实微信的 Name2Id 不含账号本人，演示库必须保持同样结构。""" + plain = _decrypt_to_plain(self.paths["message_db"].read_bytes(), DEMO_KEY) + decrypted_path = self.output / "name2id.decrypted.db" + decrypted_path.write_bytes(plain) + conn = sqlite3.connect(decrypted_path) + try: + names = [row[0] for row in conn.execute("SELECT user_name FROM Name2Id")] + finally: + conn.close() + self.assertNotIn(DEMO_ACCOUNT, names) + self.assertIn(DEMO_GROUP, names) + + def test_demo_content_is_marked_fictional(self): + """所有昵称都带「示例」标记，避免演示数据被误认为真实数据。""" + plain = _decrypt_to_plain(self.paths["contact_db"].read_bytes(), DEMO_KEY) + decrypted_path = self.output / "contact_content.decrypted.db" + decrypted_path.write_bytes(plain) + conn = sqlite3.connect(decrypted_path) + try: + rows = conn.execute("SELECT nick_name FROM Contact").fetchall() + finally: + conn.close() + self.assertTrue(rows) + for (nick_name,) in rows: + self.assertIn("示例", str(nick_name)) + + def test_encrypt_wcdb_rejects_unaligned_input(self): + with self.assertRaises(ValueError): + encrypt_wcdb(b"not-a-full-page", DEMO_KEY) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/seed_demo_database.py b/tools/seed_demo_database.py new file mode 100644 index 00000000..50f887f0 --- /dev/null +++ b/tools/seed_demo_database.py @@ -0,0 +1,465 @@ +#!/usr/bin/env python3 +"""生成完全虚构、且真实经过 WCDB 加密的演示账号，用于开发、测试与截图。 + +与 ``tools/seed_ai_acceptance.py`` 的区别：后者生成的是**明文** SQLite，只覆盖 +AI 验收场景；本脚本生成的是**逐页 AES-CBC + 逐页 HMAC 的微信 4.x 加密库**， +因此可以驱动完整的「密钥 -> 解密 -> 解析 -> 展示」链路，也不需要任何真实微信数据。 + +用途 +---- +- 贡献者在没有真实微信环境时跑通解密与读取流程； +- 截图、文档、演示视频使用可公开的虚构数据； +- 回归测试需要「真实加密格式 + 真实表结构」的样本。 + +所有姓名、群名、正文都带「示例」标记；域名统一 example.com；电话 13800000000。 + +用法:: + + python tools/seed_demo_database.py # 输出到 output/demo + python tools/seed_demo_database.py --check # 解密器自校验 + python tools/seed_demo_database.py --output /tmp/demo-account # 指定输出目录 + +默认输出到 ``output/demo``，该目录已被 ``/output/`` 规则忽略；脚本只写 +``--output`` 指定的目录，不读取也不修改任何真实微信目录。 +""" + +from __future__ import annotations + +import argparse +import hashlib +import hmac +import json +import os +import sqlite3 +import struct +import sys +import time +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +SRC = REPO_ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from wechat_decrypt_tool.wechat_decrypt import ( # noqa: E402 + HMAC_SIZE, + IV_SIZE, + KEY_SIZE, + PAGE_SIZE, + RESERVE_SIZE, + SALT_SIZE, + _derive_mac_key, + _derive_sqlcipher_enc_key, +) + +DEMO_ACCOUNT = "wxid_demo_2026" +DEMO_DISPLAY_NAME = "示例账号" +# 固定密钥：公开样本库不需要保密，固定值让同一提交产出可复现的库结构。 +DEMO_KEY = hashlib.sha256(b"wechat-data-analysis-demo-key").hexdigest() +DEMO_ALIAS = "demo_user" + +EXAMPLE_URL = "https://example.com" + +# 群聊房间号固定为示例值，避免与任何真实群聊撞号。 +DEMO_GROUP = "10000000001@chatroom" + + +def _derive_keys(key_hex: str, salt: bytes) -> tuple[bytes, bytes]: + key_material = bytes.fromhex(key_hex) + enc_key = _derive_sqlcipher_enc_key(key_material, salt) + mac_key = _derive_mac_key(enc_key, salt) + return enc_key, mac_key + + +def encrypt_wcdb(plain: bytes, key_hex: str) -> bytes: + """把明文 SQLite 整库加密为微信 4.x 页面格式（``wechat_decrypt`` 的逆过程）。 + + 每页布局：``[salt(仅第 1 页)] + ciphertext + iv + hmac``，其中密文长度为 + ``PAGE_SIZE - RESERVE_SIZE - salt_offset``；HMAC 覆盖密文与 IV，再拼小端页码。 + """ + if len(plain) % PAGE_SIZE != 0: + raise ValueError("待加密数据库必须严格按 4096 字节分页") + + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + salt = os.urandom(SALT_SIZE) + enc_key, mac_key = _derive_keys(key_hex, salt) + + out = bytearray() + for page_num in range(1, len(plain) // PAGE_SIZE + 1): + start = (page_num - 1) * PAGE_SIZE + chunk = plain[start : start + PAGE_SIZE] + offset = SALT_SIZE if page_num == 1 else 0 + payload = chunk[offset : PAGE_SIZE - RESERVE_SIZE] + iv = os.urandom(IV_SIZE) + cipher = Cipher( + algorithms.AES(enc_key), modes.CBC(iv), backend=default_backend() + ) + encryptor = cipher.encryptor() + ciphertext = encryptor.update(payload) + encryptor.finalize() + digest = hmac.new(mac_key, digestmod=hashlib.sha512) + digest.update(ciphertext) + digest.update(iv) + digest.update(page_num.to_bytes(4, "little")) + page = (salt if page_num == 1 else b"") + ciphertext + iv + digest.digest() + if len(page) != PAGE_SIZE: + raise AssertionError(f"第 {page_num} 页长度异常: {len(page)}") + out += page + return bytes(out) + + +def _patch_reserved_space(path: Path) -> None: + """把明文库改成每页预留 80 字节（与解密还原后的页面布局一致）。 + + ``PRAGMA user_version`` 只写文件头、不产生任何 cell，因此页尾预留区保持空闲； + 随后必须把 page1 btree 头的「cell 内容区起始」从 4096 改成 4096-80，否则 + SQLite 会按原页头判定空闲空间，报 ``database disk image is malformed``。 + """ + with path.open("r+b") as handle: + handle.seek(20) + handle.write(bytes([RESERVE_SIZE])) + handle.seek(100 + 5) + handle.write(struct.pack(">H", PAGE_SIZE - RESERVE_SIZE)) + + +def create_plain_database(path: Path) -> sqlite3.Connection: + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + path.unlink() + bootstrap = sqlite3.connect(path) + bootstrap.execute("PRAGMA user_version=1") + bootstrap.commit() + bootstrap.close() + _patch_reserved_space(path) + conn = sqlite3.connect(path) + conn.execute("PRAGMA journal_mode=DELETE") + return conn + + +def _timestamp(days_ago: float, hour: int, minute: int) -> int: + base = time.time() - days_ago * 86400 + local = time.localtime(base) + return int( + time.mktime( + (local.tm_year, local.tm_mon, local.tm_mday, hour, minute, 0, 0, 0, -1) + ) + ) + + +def _message( + local_id: int, + sender_rowid: int, + local_type: int, + created_at: int, + content: str, + compress_flag: int = 0, +) -> tuple: + server_id = 880000000 + local_id * 13 + return ( + local_id, + server_id, + local_type, + sender_rowid, + created_at, + content, + b"", + b"", + compress_flag, + ) + + +def build_message_database(path: Path) -> None: + conn = create_plain_database(path) + cur = conn.cursor() + + # 真实微信的 Name2Id 不包含账号本人；本人由「出现在全部单聊表中的 + # real_sender_id」判定，因此这里保持同样的结构，聊天列表才不会多出一条。 + name2id = [ + (2, "wxid_demo_zhangwei"), + (3, "wxid_demo_liting"), + (4, "wxid_demo_wangfang"), + (5, "wxid_demo_liuyang"), + (6, DEMO_GROUP), + ] + cur.execute("CREATE TABLE Name2Id (user_name TEXT)") + cur.executemany("INSERT INTO Name2Id (rowid, user_name) VALUES (?, ?)", name2id) + + chats: dict[str, list[tuple]] = {} + + # --- 群聊：覆盖文字、链接卡片、系统消息、表情、引用 --- + rows: list[tuple] = [] + ts = _timestamp(2, 9, 15) + rows.append(_message(1, 4, 1, ts, "各位早上好，今天十点对齐一下示例项目的排期")) + rows.append(_message(2, 2, 1, ts + 180, "收到，上周的示例数据我已经整理好了")) + rows.append(_message(3, 3, 1, ts + 240, "示例模块的联调环境已经部署到测试服")) + rows.append(_message(4, 1, 1, ts + 400, "辛苦大家，会议链接稍后发到群里")) + rows.append( + _message( + 5, + 5, + 49, + ts + 420, + "示例项目排期表（9 月）" + "包含里程碑与分工的示例文档5" + f"{EXAMPLE_URL}/docs/demo-plan", + ) + ) + rows.append( + _message(6, 1, 10000, ts + 500, "示例-李婷 邀请 示例-王芳 加入了群聊") + ) + rows.append(_message(7, 4, 1, ts + 700, "欢迎新同事，示例团队欢迎你")) + rows.append( + _message( + 8, + 2, + 49, + ts + 900, + "示例接口文档 v2" + "更新了示例接口的返回结构57" + f"{EXAMPLE_URL}/docs/demo-api" + "18800000014" + f"{DEMO_GROUP}wxid_demo_liuyang" + "示例-刘洋" + "示例接口的返回结构谁能同步一下？", + ) + ) + rows.append(_message(9, 3, 47, ts + 960, "[示例表情]")) + rows.append(_message(10, 1, 1, _timestamp(0, 9, 41), "上午的示例评审结论我整理好了，下午发出来")) + rows.append(_message(11, 5, 1, _timestamp(0, 9, 45), "收到，示例版本的发布时间定在周五")) + chats[DEMO_GROUP] = rows + + # --- 单聊：日常闲聊 --- + rows = [] + ts = _timestamp(1, 20, 10) + rows.append(_message(1, 2, 1, ts, "周末的示例聚会你来吗？")) + rows.append(_message(2, 1, 1, ts + 120, "来，示例场地我已经订好了")) + rows.append(_message(3, 2, 1, ts + 200, "那我把示例名单统计一下")) + rows.append( + _message( + 4, + 2, + 49, + ts + 300, + "示例餐厅订座确认" + "周六 18:00，8 人桌5" + f"{EXAMPLE_URL}/booking/demo123", + ) + ) + rows.append(_message(5, 1, 1, _timestamp(0, 8, 30), "早，通勤路上听了你推荐的示例播客，不错")) + rows.append(_message(6, 2, 1, _timestamp(0, 8, 42), "示例播客是我最近的下饭神器")) + chats["wxid_demo_zhangwei"] = rows + + # --- 单聊：出行计划（含转账卡片）--- + rows = [] + ts = _timestamp(3, 15, 20) + rows.append(_message(1, 3, 1, ts, "示例城市这周末天气不错，适合去拍外景")) + rows.append(_message(2, 1, 1, ts + 200, "好，示例相机的电池我充上")) + rows.append( + _message( + 3, + 3, + 49, + ts + 600, + "示例民宿预订" + "两晚套房，含示例早餐5" + f"{EXAMPLE_URL}/stay/demo-room", + ) + ) + rows.append( + _message( + 4, + 1, + 49, + ts + 800, + "微信转账2000" + "¥288.00" + "1", + ) + ) + rows.append(_message(5, 3, 1, ts + 900, "示例民宿的定金我收到了，周五出发")) + chats["wxid_demo_liting"] = rows + + # --- 单聊：工作对接（含语音消息占位）--- + rows = [] + ts = _timestamp(0, 9, 2) + rows.append(_message(1, 4, 1, ts, "示例报告初稿在附件里，麻烦帮忙看看第二部分")) + rows.append(_message(2, 1, 1, ts + 300, "收到，我十一点前把批注发你")) + rows.append(_message(3, 4, 34, ts + 600, "")) + rows.append(_message(4, 4, 1, ts + 700, "语音里说的示例口径以文档为准")) + chats["wxid_demo_wangfang"] = rows + + # --- 单聊：发布检查 --- + rows = [] + ts = _timestamp(1, 16, 40) + rows.append(_message(1, 5, 1, ts, "示例环境的冒烟测试我跑完了，全部通过")) + rows.append(_message(2, 1, 1, ts + 400, "效率很高，示例包我来打")) + rows.append(_message(3, 5, 1, ts + 800, "好，打完包我在群里同步")) + chats["wxid_demo_liuyang"] = rows + + for talker, chat_rows in chats.items(): + table = "Msg_" + hashlib.md5(talker.encode("utf-8")).hexdigest() + cur.execute( + f"""CREATE TABLE [{table}] ( + local_id INTEGER, server_id INTEGER, local_type INTEGER, + real_sender_id INTEGER, create_time INTEGER, + message_content BLOB, source BLOB, packed_info_data BLOB, + WCDB_CT_message_content INTEGER)""" + ) + cur.executemany( + f"INSERT INTO [{table}] VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", chat_rows + ) + + conn.commit() + status = conn.execute("PRAGMA integrity_check").fetchone()[0] + if status != "ok": + raise AssertionError(f"message 库完整性检查失败: {status}") + conn.close() + + +def build_contact_database(path: Path) -> None: + conn = create_plain_database(path) + cur = conn.cursor() + cur.execute( + """CREATE TABLE Contact ( + user_name TEXT, nick_name TEXT, remark TEXT, alias TEXT, + description TEXT, local_type INTEGER)""" + ) + rows = [ + (DEMO_ACCOUNT, DEMO_DISPLAY_NAME, "", DEMO_ALIAS, "示例账号自我介绍", 1), + ("wxid_demo_zhangwei", "示例-张伟", "示例-张伟", "demo_zw", "示例好友：产品同事", 1), + ("wxid_demo_liting", "示例-李婷", "示例-李婷", "demo_lt", "示例好友：摄影搭子", 1), + ("wxid_demo_wangfang", "示例-王芳", "示例-王芳", "demo_wf", "示例好友：报告对接", 1), + ("wxid_demo_liuyang", "示例-刘洋", "示例-刘洋", "demo_ly", "示例好友：测试同事", 1), + (DEMO_GROUP, "示例-产品讨论群", "", "", "示例群聊", 2), + ] + cur.executemany("INSERT INTO Contact VALUES (?, ?, ?, ?, ?, ?)", rows) + conn.commit() + status = conn.execute("PRAGMA integrity_check").fetchone()[0] + if status != "ok": + raise AssertionError(f"contact 库完整性检查失败: {status}") + conn.close() + + +def _write_encrypted(plain_db: Path, target: Path) -> None: + plain = plain_db.read_bytes() + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(encrypt_wcdb(plain, DEMO_KEY)) + + +def seed(output: Path) -> dict[str, Path]: + account_root = output / "databases" / DEMO_ACCOUNT + message_dir = account_root / "db_storage" / "message" + contact_dir = account_root / "db_storage" / "contact" + + work_dir = output / "_plain" + work_dir.mkdir(parents=True, exist_ok=True) + plain_message = work_dir / "message_0.plain.db" + plain_contact = work_dir / "contact.plain.db" + + build_message_database(plain_message) + build_contact_database(plain_contact) + _write_encrypted(plain_message, message_dir / "message_0.db") + _write_encrypted(plain_contact, contact_dir / "contact.db") + + keys_path = output / "demo_keys.json" + keys_path.write_text( + json.dumps( + { + DEMO_ACCOUNT: { + "key": DEMO_KEY, + "display_name": DEMO_DISPLAY_NAME, + "alias": DEMO_ALIAS, + } + }, + ensure_ascii=False, + indent=2, + ), + encoding="utf-8", + ) + return { + "account_root": account_root, + "message_db": message_dir / "message_0.db", + "contact_db": contact_dir / "contact.db", + "keys": keys_path, + } + + +def _self_check(paths: dict[str, Path]) -> None: + """用项目自身的扫描与解密链路验证样本可被真实读取。""" + from wechat_decrypt_tool.wechat_decrypt import ( + decrypt_wechat_databases, + scan_account_databases_from_path, + ) + + db_storage = paths["account_root"] / "db_storage" + scanned = scan_account_databases_from_path(str(db_storage)) + if not scanned: + raise AssertionError("扫描演示账号时没有发现任何数据库") + + result = decrypt_wechat_databases(str(db_storage), DEMO_KEY) + if result.get("status") != "success" or not result.get("successful_count"): + raise AssertionError(f"演示账号解密失败: {result}") + + total = result.get("total_databases") + success = result.get("successful_count") + if result.get("failed_count"): + raise AssertionError(f"演示账号存在解密失败的库: {result.get('failed_files')}") + + # 解密产物必须真的是可读 SQLite，且包含我们写入的表。 + output_dir = Path(str(result.get("output_directory") or "")) + for label, expected_table in (("message", "Name2Id"), ("contact", "Contact")): + candidates = sorted(output_dir.rglob(f"{label}*.db")) + sorted( + output_dir.rglob(f"{label}.db") + ) + if not candidates: + raise AssertionError(f"解密输出中找不到 {label} 库: {output_dir}") + conn = sqlite3.connect(candidates[0]) + try: + integrity = conn.execute("PRAGMA integrity_check").fetchone()[0] + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + finally: + conn.close() + if integrity != "ok": + raise AssertionError(f"{label} 解密后完整性检查失败: {integrity}") + if expected_table not in tables: + raise AssertionError(f"{label} 解密后缺少表 {expected_table}: {tables}") + + print(f"自校验通过：扫描到 {len(scanned)} 个数据库，成功解密 {success}/{total} 个") + + +def main() -> int: + parser = argparse.ArgumentParser(description="生成虚构的微信 4.x 加密演示账号") + parser.add_argument( + "--output", + type=Path, + default=REPO_ROOT / "output" / "demo", + help="输出目录（默认 output/demo，已随 /output/ 一起被忽略）", + ) + parser.add_argument( + "--check", action="store_true", help="生成后用项目解密器自校验" + ) + args = parser.parse_args() + + output = args.output.resolve() + paths = seed(output) + + print(f"演示账号: {DEMO_ACCOUNT}（显示名 {DEMO_DISPLAY_NAME}）") + print(f"账号目录: {paths['account_root']}") + print(f"消息库: {paths['message_db']}") + print(f"联系人库: {paths['contact_db']}") + print(f"密钥文件: {paths['keys']}") + print(f"演示密钥: {DEMO_KEY}") + + if args.check: + _self_check(paths) + return 0 + + +if __name__ == "__main__": + sys.exit(main())