test_db_poll.py
6.88 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
#!/usr/bin/env python3
"""
DB 轮询测试 — 验证本地复制+解密+增量读取全流程。
在 Ubuntu proot 容器内执行。
"""
import subprocess, os, sys, json, time, sqlite3
CONFIG_PATH = "config.json"
REPORT = {"test": "db_poll", "steps": {}, "overall": None}
def log_step(name, ok, detail=""):
status = "[OK]" if ok else "[FAIL]"
print(f" {status} {name}")
if detail:
print(f" {detail}")
REPORT["steps"][name] = {"ok": ok, "detail": detail}
return ok
def find_enmicromsg_db():
"""Auto-discover EnMicroMsg.db path on phone."""
result = subprocess.run(
["su", "-c", "find /data/data/com.tencent.mm/MicroMsg -name 'EnMicroMsg.db' -type f 2>/dev/null"],
capture_output=True, text=True
)
paths = [p.strip() for p in result.stdout.strip().split("\n") if p.strip()]
return paths[0] if paths else None
def copy_db(src, dst_dir):
"""Copy DB + WAL + SHM using su.
Fallback: if proot su cannot access /data/data/, use pre-copied files in dst_dir."""
os.makedirs(dst_dir, exist_ok=True)
files = [src, src + "-wal", src + "-shm"]
copied = []
for f in files:
dst = os.path.join(dst_dir, os.path.basename(f))
r = subprocess.run(["su", "-c", f"cp '{f}' '{dst}' 2>/dev/null || true"], capture_output=True)
if os.path.exists(dst):
copied.append(dst)
# Fallback: if no files copied but EnMicroMsg.db already exists in dst_dir,
# it may have been pre-copied at system level
if not copied:
fallback = os.path.join(dst_dir, "EnMicroMsg.db")
if os.path.exists(fallback):
copied.append(fallback)
# Also check for WAL
wal = fallback + "-wal"
if os.path.exists(wal):
copied.append(wal)
return copied
def decrypt_db(encrypted_path, output_path, key, page_size=1024, compat=1):
"""Decrypt SQLCipher DB using sqlcipher CLI."""
if os.path.exists(output_path):
os.remove(output_path)
sql = f"""
PRAGMA key = "{key}";
PRAGMA cipher_compatibility = {compat};
PRAGMA cipher_page_size = {page_size};
ATTACH DATABASE '{output_path}' AS plain KEY '';
SELECT sqlcipher_export('plain');
DETACH DATABASE plain;
.quit
"""
sqlcipher_path = "/data/data/com.termux/files/usr/bin/sqlcipher"
if not os.path.exists(sqlcipher_path):
sqlcipher_path = "sqlcipher"
r = subprocess.run(
[sqlcipher_path, encrypted_path],
input=sql, capture_output=True, text=True, timeout=60
)
ok = r.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 4096
return ok, r.stdout, r.stderr
def query_messages(db_path, last_msg_id=0, limit=10):
"""Query messages with msgId > last_msg_id."""
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT msgId, talker, content, type, isSend, createTime FROM message WHERE msgId > ? ORDER BY msgId LIMIT ?",
(last_msg_id, limit)
).fetchall()
conn.close()
return [dict(r) for r in rows]
def main():
print("=" * 60)
print("DB 轮询测试")
print("=" * 60)
# Load config
config = {}
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH) as f:
config = json.load(f)
all_ok = True
# Step 1: Discover DB path
print("\n[Step 1] 发现 EnMicroMsg.db 路径")
db_src = config.get("db_src_path", "")
if not db_src:
db_src = find_enmicromsg_db()
if db_src:
log_step("db_discover", True, db_src)
# Save back to config for future runs
config["db_src_path"] = db_src
with open(CONFIG_PATH, "w") as f:
json.dump(config, f, indent=2, ensure_ascii=False)
else:
all_ok &= log_step("db_discover", False, "未找到 EnMicroMsg.db，确认微信已登录")
save_report()
return 1
# Step 2: Copy DB
print("\n[Step 2] 复制 DB 文件")
copy_dir = config.get("db_copy_dir", "/sdcard/hermes/dbs")
copied = copy_db(db_src, copy_dir)
copied_names = [os.path.basename(c) for c in copied]
all_ok &= log_step("db_copy", len(copied) >= 1, f"复制了: {', '.join(copied_names)}")
# Step 3: Decrypt
print("\n[Step 3] SQLCipher 解密")
key = config.get("sqlcipher_key", "e03ce81")
page_size = config.get("cipher_page_size", 1024)
compat = config.get("cipher_compatibility", 1)
enc_path = os.path.join(copy_dir, "EnMicroMsg.db")
plain_path = enc_path + ".decrypted.db"
ok, stdout, stderr = decrypt_db(enc_path, plain_path, key, page_size, compat)
detail = f"输出: {plain_path}"
if not ok:
detail += f" | stderr: {stderr[:200]}"
all_ok &= log_step("db_decrypt", ok, detail)
if not ok:
save_report()
return 1
# Step 4: Query schema
print("\n[Step 4] 验证表结构")
try:
conn = sqlite3.connect(plain_path)
tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('message', 'rcontact', 'conversation')")]
conn.close()
has_msg = "message" in tables
all_ok &= log_step("schema_check", has_msg, f"找到的表: {', '.join(tables)}")
except Exception as e:
all_ok &= log_step("schema_check", False, str(e))
# Step 5: Query latest messages
print("\n[Step 5] 查询最新消息")
try:
msgs = query_messages(plain_path, last_msg_id=0, limit=5)
if msgs:
for m in msgs:
direction = "发送" if m.get("isSend") == 1 else "接收"
print(f" msgId={m['msgId']} {direction} talker={m['talker']} content={m['content'][:40]}")
all_ok &= log_step("msg_query", True, f"共 {len(msgs)} 条")
else:
all_ok &= log_step("msg_query", True, "message 表为空（可能是新账号）")
except Exception as e:
all_ok &= log_step("msg_query", False, str(e))
# Step 6: Incremental read simulation
print("\n[Step 6] 增量读取模拟")
try:
if msgs:
max_id = max(m["msgId"] for m in msgs)
new_msgs = query_messages(plain_path, last_msg_id=max_id, limit=10)
all_ok &= log_step("incremental", True, f"上次 msgId={max_id}, 新消息={len(new_msgs)} 条")
else:
all_ok &= log_step("incremental", True, "跳过（无历史消息）")
except Exception as e:
all_ok &= log_step("incremental", False, str(e))
# Summary
REPORT["overall"] = "PASS" if all_ok else "FAIL"
save_report()
print("\n" + "=" * 60)
print(f"结果: {REPORT['overall']}")
print("=" * 60)
return 0 if all_ok else 1
def save_report():
REPORT["timestamp"] = subprocess.check_output(["date", "-Iseconds"]).decode().strip()
with open("test_report_db_poll.json", "w") as f:
json.dump(REPORT, f, indent=2, ensure_ascii=False)
print("\n详细报告: test_report_db_poll.json")
if __name__ == "__main__":
sys.exit(main())