feat: implement Zalo database extraction and schema analysis script via ADB

2026-07-23 22:27:26 +07:00
parent 27337702b5
commit 1a684ddcc9
1 changed files with 103 additions and 0 deletions
@@ -0,0 +1,103 @@
+import os
+import subprocess
+import logging
+import sqlite3
+import shutil
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def run_adb_command(command: str):
+    """Chạy lệnh ADB shell và trả về kết quả"""
+    full_cmd = f"adb shell {command}"
+    try:
+        result = subprocess.run(full_cmd, shell=True, check=True, capture_output=True, text=True)
+        return result.stdout.strip()
+    except subprocess.CalledProcessError as e:
+        return None
+
+def analyze_zalo():
+    logging.info("BẮT ĐẦU TRÍCH XUẤT DATABASE ZALO (PULL VỀ SERVER)")
+    # Đảm bảo kết nối
+    subprocess.run("adb connect 10.160.138.2:5555", shell=True, capture_output=True)
+    
+    # Tạo thư mục chứa db
+    os.makedirs("zalo_dbs", exist_ok=True)
+    
+    # Lấy danh sách các UID (thư mục) trong databases
+    zalo_db_path = "/data/data/com.zing.zalo/databases"
+    ls_output = run_adb_command(f"su -c 'ls -l {zalo_db_path}'")
+    
+    if not ls_output:
+        logging.error("Không thể truy cập thư mục Zalo. Kiểm tra lại quyền Root.")
+        return
+        
+    # Tìm các thư mục có tên là số (UID)
+    uid_folders = []
+    for line in ls_output.split('\n'):
+        if line.startswith('d'): # là thư mục
+            parts = line.split()
+            folder_name = parts[-1]
+            if folder_name.isdigit():
+                uid_folders.append(folder_name)
+                
+    if not uid_folders:
+        logging.error("Không tìm thấy thư mục UID nào của Zalo!")
+        return
+        
+    logging.info(f"Tìm thấy các thư mục tài khoản Zalo: {uid_folders}")
+    
+    with open("zalo_schema_dump.txt", "w", encoding="utf-8") as f:
+        f.write("=== DUMP CẤU TRÚC DATABASE ZALO ===\n")
+    
+    for uid in uid_folders:
+        logging.info(f"Đang xử lý tài khoản: {uid}")
+        # Liệt kê các file trong thư mục UID
+        files_in_uid = run_adb_command(f"su -c 'ls {zalo_db_path}/{uid}/'")
+        if not files_in_uid: continue
+        
+        for db_file in files_in_uid.split('\n'):
+            db_file = db_file.strip()
+            if not db_file or not db_file.endswith('.db'): continue
+            
+            logging.info(f"Phát hiện file DB: {db_file}")
+            
+            # Copy file ra sdcard để có thể pull về
+            run_adb_command(f"su -c 'cp {zalo_db_path}/{uid}/{db_file} /sdcard/temp_zalo_{db_file}'")
+            run_adb_command(f"su -c 'chmod 666 /sdcard/temp_zalo_{db_file}'")
+            
+            # Pull file về server
+            local_path = os.path.join("zalo_dbs", f"{uid}_{db_file}")
+            subprocess.run(f"adb pull /sdcard/temp_zalo_{db_file} {local_path}", shell=True, capture_output=True)
+            run_adb_command(f"su -c 'rm /sdcard/temp_zalo_{db_file}'") # Xóa file tạm
+            
+            if os.path.exists(local_path):
+                logging.info(f"Đã kéo file về thành công: {local_path}. Đang phân tích...")
+                try:
+                    conn = sqlite3.connect(local_path)
+                    cursor = conn.cursor()
+                    
+                    # Lấy danh sách các bảng
+                    cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
+                    tables = cursor.fetchall()
+                    
+                    if tables:
+                        with open("zalo_schema_dump.txt", "a", encoding="utf-8") as f:
+                            f.write(f"\n\n--- Cấu trúc file: {db_file} (Tài khoản {uid}) ---\n")
+                            f.write(f"Danh sách bảng: {[t[0] for t in tables]}\n")
+                            
+                            for table_tuple in tables:
+                                table_name = table_tuple[0]
+                                if "msg" in table_name.lower() or "chat" in table_name.lower() or "group" in table_name.lower():
+                                    cursor.execute(f"PRAGMA table_info({table_name});")
+                                    columns = cursor.fetchall()
+                                    f.write(f"\nBảng '{table_name}':\n")
+                                    for col in columns:
+                                        f.write(f"  - {col[1]} ({col[2]})\n")
+                    conn.close()
+                except Exception as e:
+                    logging.warning(f"Lỗi khi đọc file SQLite {local_path}: {str(e)}")
+
+    logging.info("Hoàn tất! Hãy gửi cho tôi nội dung file zalo_schema_dump.txt để phân tích.")
+
+if __name__ == "__main__":
+    analyze_zalo()