#!/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())