Victor Phan
107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
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 -s 10.160.138.2:5555 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:
|
|
logging.error(f"Lệnh ADB thất bại: {full_cmd}")
|
|
logging.error(f"Mã lỗi: {e.returncode}")
|
|
logging.error(f"Chi tiết lỗi: {e.stderr}")
|
|
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()
|