728dbb72df
两个核心增强: 1. **CSV 计划导出工作流** (`--write-plan-csv` / `--from-plan-csv`) - 支持 blacklist (export=0 跳过) / whitelist (export=1 导出) 两种模式 - `--size-mode estimate|scan` 控制是否扫本地附件 - UTF-8 BOM 编码, Excel/WPS 直接打开 - 解决了之前"动不动全量导出"的痛点 2. **`_export_index.json` 稳定导出索引** - 用 username 追踪当前 JSON 文件 - 联系人备注/群名变化时自动重命名旧文件 (而不是产生孤儿) - 同名联系人冲突时自动追加 `__<username>` 后缀 - atomic write (tmp + os.replace), bootstrap from existing files 3. **JSON metadata 扩展** - 新增 `date_first_msg / date_last_msg / contact_remark / contact_nick_name / contact_tags / contact_memo` 测试: 717 行, 20 tests pass, 覆盖 CRUD 索引、黑白名单、命名冲突、incremental rename、UTF-8 BOM。
243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
"""
|
||
WeChat 4.0 数据库解密器
|
||
|
||
使用从进程内存提取的per-DB enc_key解密SQLCipher 4加密的数据库
|
||
参数: SQLCipher 4, AES-256-CBC, HMAC-SHA512, reserve=80, page_size=4096
|
||
密钥来源: all_keys.json (由find_all_keys.py从内存提取)
|
||
"""
|
||
import hashlib, struct, os, sys, json
|
||
import hmac as hmac_mod
|
||
from Crypto.Cipher import AES
|
||
|
||
import argparse
|
||
import functools
|
||
print = functools.partial(print, flush=True)
|
||
|
||
PAGE_SZ = 4096
|
||
KEY_SZ = 32
|
||
SALT_SZ = 16
|
||
IV_SZ = 16
|
||
HMAC_SZ = 64
|
||
RESERVE_SZ = 80 # IV(16) + HMAC(64)
|
||
SQLITE_HDR = b'SQLite format 3\x00'
|
||
|
||
from config import load_config
|
||
from key_utils import get_key_info, strip_key_metadata
|
||
_cfg = load_config()
|
||
DB_DIR = _cfg["db_dir"]
|
||
OUT_DIR = _cfg["decrypted_dir"]
|
||
KEYS_FILE = _cfg["keys_file"]
|
||
|
||
|
||
def derive_mac_key(enc_key, salt):
|
||
"""从enc_key派生HMAC密钥"""
|
||
mac_salt = bytes(b ^ 0x3a for b in salt)
|
||
return hashlib.pbkdf2_hmac("sha512", enc_key, mac_salt, 2, dklen=KEY_SZ)
|
||
|
||
|
||
def decrypt_page(enc_key, page_data, pgno):
|
||
"""解密单个页面，输出4096字节的标准SQLite页面"""
|
||
iv = page_data[PAGE_SZ - RESERVE_SZ : PAGE_SZ - RESERVE_SZ + IV_SZ]
|
||
|
||
if pgno == 1:
|
||
encrypted = page_data[SALT_SZ : PAGE_SZ - RESERVE_SZ]
|
||
cipher = AES.new(enc_key, AES.MODE_CBC, iv)
|
||
decrypted = cipher.decrypt(encrypted)
|
||
page = bytearray(SQLITE_HDR + decrypted + b'\x00' * RESERVE_SZ)
|
||
# 保留 reserve=80, B-tree 基于 usable_size=4016 构建
|
||
return bytes(page)
|
||
else:
|
||
encrypted = page_data[:PAGE_SZ - RESERVE_SZ]
|
||
cipher = AES.new(enc_key, AES.MODE_CBC, iv)
|
||
decrypted = cipher.decrypt(encrypted)
|
||
return decrypted + b'\x00' * RESERVE_SZ
|
||
|
||
|
||
def decrypt_database(db_path, out_path, enc_key):
|
||
"""解密整个数据库文件"""
|
||
file_size = os.path.getsize(db_path)
|
||
total_pages = file_size // PAGE_SZ
|
||
|
||
if file_size % PAGE_SZ != 0:
|
||
print(f" [WARN] 文件大小 {file_size} 不是 {PAGE_SZ} 的倍数")
|
||
total_pages += 1
|
||
|
||
with open(db_path, 'rb') as fin:
|
||
page1 = fin.read(PAGE_SZ)
|
||
|
||
if len(page1) < PAGE_SZ:
|
||
print(f" [ERROR] 文件太小")
|
||
return False
|
||
|
||
# 提取salt并派生mac_key, 验证page 1
|
||
salt = page1[:SALT_SZ]
|
||
mac_key = derive_mac_key(enc_key, salt)
|
||
p1_hmac_data = page1[SALT_SZ : PAGE_SZ - RESERVE_SZ + IV_SZ]
|
||
p1_stored_hmac = page1[PAGE_SZ - HMAC_SZ : PAGE_SZ]
|
||
hm = hmac_mod.new(mac_key, p1_hmac_data, hashlib.sha512)
|
||
hm.update(struct.pack('<I', 1))
|
||
if hm.digest() != p1_stored_hmac:
|
||
print(f" [ERROR] Page 1 HMAC验证失败! salt: {salt.hex()}")
|
||
return False
|
||
|
||
print(f" HMAC OK, {total_pages} pages")
|
||
|
||
# 解密所有页面
|
||
os.makedirs(os.path.dirname(out_path), exist_ok=True)
|
||
with open(db_path, 'rb') as fin, open(out_path, 'wb') as fout:
|
||
for pgno in range(1, total_pages + 1):
|
||
page = fin.read(PAGE_SZ)
|
||
if len(page) < PAGE_SZ:
|
||
if len(page) > 0:
|
||
page = page + b'\x00' * (PAGE_SZ - len(page))
|
||
else:
|
||
break
|
||
|
||
decrypted = decrypt_page(enc_key, page, pgno)
|
||
fout.write(decrypted)
|
||
|
||
if pgno == 1:
|
||
if decrypted[:16] != SQLITE_HDR:
|
||
print(f" [WARN] 解密后header不匹配!")
|
||
|
||
if pgno % 10000 == 0:
|
||
print(f" 进度: {pgno}/{total_pages} ({100*pgno/total_pages:.1f}%)")
|
||
|
||
return True
|
||
|
||
|
||
def main(argv=None):
|
||
parser = argparse.ArgumentParser(
|
||
description="WeChat 4.0 数据库解密器"
|
||
)
|
||
parser.add_argument(
|
||
"-i", "--incremental",
|
||
action="store_true",
|
||
help="增量模式：仅当源 .db 更新于已解密文件时才重新解密",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="预览模式：显示将要解密的数据库列表",
|
||
)
|
||
args = parser.parse_args(argv)
|
||
|
||
print("=" * 60)
|
||
print(" WeChat 4.0 数据库解密器")
|
||
print("=" * 60)
|
||
|
||
# 加载密钥
|
||
if not os.path.exists(KEYS_FILE):
|
||
print(f"[ERROR] 密钥文件不存在: {KEYS_FILE}")
|
||
print("请先运行 python main.py decrypt 提取密钥并解密")
|
||
sys.exit(1)
|
||
|
||
|
||
with open(KEYS_FILE, encoding="utf-8") as f:
|
||
keys = json.load(f)
|
||
|
||
keys = strip_key_metadata(keys)
|
||
print(f"\n加载 {len(keys)} 个数据库密钥")
|
||
print(f"输出目录: {OUT_DIR}")
|
||
if args.incremental:
|
||
print(f"模式: 增量 (跳过未变更的数据库)")
|
||
os.makedirs(OUT_DIR, exist_ok=True)
|
||
|
||
# 收集所有DB文件
|
||
db_files = []
|
||
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)
|
||
sz = os.path.getsize(path)
|
||
db_files.append((rel, path, sz))
|
||
|
||
db_files.sort(key=lambda x: x[2]) # 从小到大
|
||
|
||
print(f"找到 {len(db_files)} 个数据库文件\n")
|
||
|
||
success = 0
|
||
failed = 0
|
||
skipped = 0
|
||
skipped_unmodified = 0
|
||
total_bytes = 0
|
||
|
||
for rel, path, sz in db_files:
|
||
key_info = get_key_info(keys, rel)
|
||
if not key_info:
|
||
print(f"SKIP: {rel} (无密钥，如已安装微信补丁可能需要重新运行密钥提取)")
|
||
skipped += 1
|
||
continue
|
||
|
||
out_path = os.path.join(OUT_DIR, rel)
|
||
|
||
# 增量模式：检查 mtime
|
||
if args.incremental and os.path.exists(out_path):
|
||
src_mtime = os.path.getmtime(path)
|
||
dst_mtime = os.path.getmtime(out_path)
|
||
if src_mtime <= dst_mtime:
|
||
skipped_unmodified += 1
|
||
if args.dry_run:
|
||
print(f"SKIP: {rel} (未修改)")
|
||
continue
|
||
elif args.dry_run:
|
||
print(f"NEW: {rel} (源较新)")
|
||
elif not args.dry_run:
|
||
print(f"更新: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ")
|
||
elif args.dry_run:
|
||
print(f"NEW: {rel} ({sz/1024/1024:.1f}MB)")
|
||
else:
|
||
print(f"解密: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ")
|
||
|
||
if args.dry_run:
|
||
skipped_unmodified += 1
|
||
continue
|
||
|
||
enc_key = bytes.fromhex(key_info["enc_key"])
|
||
ok = decrypt_database(path, out_path, enc_key)
|
||
if ok:
|
||
# SQLite验证
|
||
try:
|
||
import sqlite3
|
||
conn = sqlite3.connect(out_path)
|
||
tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
|
||
conn.close()
|
||
table_names = [t[0] for t in tables]
|
||
print(f" OK! 表: {', '.join(table_names[:5])}", end="")
|
||
if len(table_names) > 5:
|
||
print(f" ...共{len(table_names)}个", end="")
|
||
print()
|
||
success += 1
|
||
total_bytes += sz
|
||
except Exception as e:
|
||
print(f" [WARN] SQLite验证失败: {e}")
|
||
failed += 1
|
||
else:
|
||
failed += 1
|
||
|
||
# 清理 sqlite3.connect() 验证遗留的 -shm/-wal 空文件
|
||
# 避免后续工具打开 .db 时优先读旧 WAL 报 "database disk image is malformed"
|
||
for suffix in ("-shm", "-wal"):
|
||
residual = out_path + suffix
|
||
if os.path.exists(residual):
|
||
try:
|
||
os.remove(residual)
|
||
except OSError:
|
||
pass
|
||
|
||
if args.dry_run:
|
||
print(f"\n{'='*60}")
|
||
print(f"预览: 需要解密 {skipped_unmodified} 个数据库")
|
||
return
|
||
|
||
print(f"\n{'='*60}")
|
||
inc_note = f" (跳过 {skipped_unmodified} 个未变更)" if skipped_unmodified else ""
|
||
print(f"结果: {success} 成功, {failed} 失败, {skipped} 跳过(无密钥){inc_note}, 共 {len(db_files)} 个")
|
||
print(f"解密数据量: {total_bytes/1024/1024/1024:.1f}GB")
|
||
print(f"解密文件在: {OUT_DIR}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|