Files
kefu/wechat_rpa/probe_wxwork_db.py
2026-08-27 14:04:28 +08:00

61 lines
1.9 KiB
Python

# -*- coding: utf-8 -*-
"""
probe_wxwork_db.py - 解密 message.db 并探查表结构(Phase 2 开发辅助工具)
用法: python probe_wxwork_db.py <db_path> <key_hex>
"""
from __future__ import annotations
import os
import sqlite3
import sys
from wxwork_crypto import decrypt_db_open, verify_key
def main():
if len(sys.argv) < 3:
print("用法: python probe_wxwork_db.py <db_path> <key_hex>")
sys.exit(1)
db_path = sys.argv[1]
key_hex = sys.argv[2]
key = bytes.fromhex(key_hex)
if len(key) != 16:
print("key 必须为 16 字节 (32 hex)")
sys.exit(1)
if not verify_key(key, db_path):
print("[-] 密钥验证失败: 无法解密该 db")
sys.exit(1)
print(f"[+] 密钥验证通过: {key_hex}")
conn, tmp = decrypt_db_open(db_path, key)
if conn is None:
print("[-] 解密打开失败")
sys.exit(1)
try:
cur = conn.cursor()
cur.execute("SELECT name, type FROM sqlite_master WHERE type IN ('table','view') ORDER BY name")
tables = cur.fetchall()
print(f"\n{len(tables)} 个表/视图:")
for name, typ in tables:
print(f" [{typ}] {name}")
# 对消息相关的表打印 schema
for name, typ in tables:
if any(k in name.lower() for k in ("msg", "message", "chat", "session", "contact")):
try:
cur.execute(f'PRAGMA table_info("{name}")')
cols = cur.fetchall()
print(f"\n=== {name} schema ===")
for c in cols:
print(f" {c[1]} {c[2]}")
cur.execute(f'SELECT COUNT(*) FROM "{name}"')
print(f" 行数: {cur.fetchone()[0]}")
except Exception as e:
print(f" (读取失败: {e})")
finally:
conn.close()
if os.path.exists(tmp):
os.remove(tmp)
if __name__ == "__main__":
main()