Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 91 additions & 0 deletions docs/demo-database-seed.md
Original file line number Diff line number Diff line change
@@ -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")`),公开无风险;
它只用于打开本脚本自己生成的样本库。
142 changes: 142 additions & 0 deletions tests/test_seed_demo_database.py
Original file line number Diff line number Diff line change
@@ -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()
Loading