feat: add Linux support with cross-platform memory scanning
- Add Linux memory scanner (`find_all_keys_linux.py`) using `/proc/<pid>/mem`, same approach as Windows/macOS — no GDB, no function offsets, no restart needed - Extract Windows-specific code to `find_all_keys_windows.py` - Make `find_all_keys.py` a platform dispatcher (Windows / Linux) - Add `key_utils.py` for cross-platform path matching (`/` vs `\` in all_keys.json) - Update `config.py` with Linux auto-detection of db_storage paths - Update all consumers (decrypt_db, monitor, monitor_web, mcp_server) to use `get_key_info()` for platform-agnostic key lookup Tested on remote Linux container: 15/15 DBs scanned, decrypted, and verified.
This commit is contained in:
PeanutSplash
339
config.py
339
config.py
@@ -1,138 +1,201 @@
|
||||
"""
|
||||
配置加载器 - 从 config.json 读取路径配置
|
||||
首次运行时自动检测微信数据目录，检测失败则提示手动配置
|
||||
"""
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
||||
|
||||
_DEFAULT_TEMPLATE_DIR = r"D:\xwechat_files\your_wxid\db_storage"
|
||||
|
||||
_DEFAULT = {
|
||||
"db_dir": _DEFAULT_TEMPLATE_DIR,
|
||||
"keys_file": "all_keys.json",
|
||||
"decrypted_dir": "decrypted",
|
||||
"decoded_image_dir": "decoded_images",
|
||||
"wechat_process": "Weixin.exe",
|
||||
}
|
||||
|
||||
|
||||
def auto_detect_db_dir():
|
||||
"""从微信本地配置自动检测 db_storage 路径。
|
||||
|
||||
读取 %APPDATA%\\Tencent\\xwechat\\config\\*.ini，
|
||||
找到数据存储根目录，然后匹配 xwechat_files\\*\\db_storage。
|
||||
"""
|
||||
appdata = os.environ.get("APPDATA", "")
|
||||
config_dir = os.path.join(appdata, "Tencent", "xwechat", "config")
|
||||
if not os.path.isdir(config_dir):
|
||||
return None
|
||||
|
||||
# 从 ini 文件中找到有效的目录路径
|
||||
data_roots = []
|
||||
for ini_file in glob.glob(os.path.join(config_dir, "*.ini")):
|
||||
try:
|
||||
# 微信 ini 可能是 utf-8 或 gbk 编码（中文路径）
|
||||
content = None
|
||||
for enc in ("utf-8", "gbk"):
|
||||
try:
|
||||
with open(ini_file, "r", encoding=enc) as f:
|
||||
content = f.read(1024).strip()
|
||||
break
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if not content or any(c in content for c in "\n\r\x00"):
|
||||
continue
|
||||
if os.path.isdir(content):
|
||||
data_roots.append(content)
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
# 在每个根目录下搜索 xwechat_files\*\db_storage
|
||||
seen = set()
|
||||
candidates = []
|
||||
for root in data_roots:
|
||||
pattern = os.path.join(root, "xwechat_files", "*", "db_storage")
|
||||
for match in glob.glob(pattern):
|
||||
normalized = os.path.normcase(os.path.normpath(match))
|
||||
if os.path.isdir(match) and normalized not in seen:
|
||||
seen.add(normalized)
|
||||
candidates.append(match)
|
||||
|
||||
if len(candidates) == 1:
|
||||
return candidates[0]
|
||||
if len(candidates) > 1:
|
||||
# 非交互环境（MCP、无 stdin 管道等）直接取第一个
|
||||
if not sys.stdin.isatty():
|
||||
return candidates[0]
|
||||
print("[!] 检测到多个微信数据目录（请选择当前正在运行的微信账号）:")
|
||||
for i, c in enumerate(candidates, 1):
|
||||
print(f" {i}. {c}")
|
||||
print(" 0. 跳过，稍后手动配置")
|
||||
try:
|
||||
while True:
|
||||
choice = input("请选择 [0-{}]: ".format(len(candidates))).strip()
|
||||
if choice == "0":
|
||||
return None
|
||||
if choice.isdigit() and 1 <= int(choice) <= len(candidates):
|
||||
return candidates[int(choice) - 1]
|
||||
print(" 无效输入，请重新选择")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print()
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def load_config():
|
||||
cfg = {}
|
||||
if os.path.exists(CONFIG_FILE):
|
||||
try:
|
||||
with open(CONFIG_FILE) as f:
|
||||
cfg = json.load(f)
|
||||
except json.JSONDecodeError:
|
||||
print(f"[!] {CONFIG_FILE} 格式损坏，将使用默认配置")
|
||||
cfg = {}
|
||||
|
||||
# db_dir 缺失或仍为模板值时，尝试自动检测
|
||||
db_dir = cfg.get("db_dir", "")
|
||||
if not db_dir or db_dir == _DEFAULT_TEMPLATE_DIR or "your_wxid" in db_dir:
|
||||
detected = auto_detect_db_dir()
|
||||
if detected:
|
||||
print(f"[+] 自动检测到微信数据目录: {detected}")
|
||||
# 合并默认值并保存
|
||||
cfg = {**_DEFAULT, **cfg, "db_dir": detected}
|
||||
with open(CONFIG_FILE, "w") as f:
|
||||
json.dump(cfg, f, indent=4, ensure_ascii=False)
|
||||
print(f"[+] 已保存到: {CONFIG_FILE}")
|
||||
else:
|
||||
if not os.path.exists(CONFIG_FILE):
|
||||
with open(CONFIG_FILE, "w") as f:
|
||||
json.dump(_DEFAULT, f, indent=4)
|
||||
print(f"[!] 未能自动检测微信数据目录")
|
||||
print(f" 请手动编辑 {CONFIG_FILE} 中的 db_dir 字段")
|
||||
print(f" 路径可在 微信设置 → 文件管理 中找到")
|
||||
sys.exit(1)
|
||||
|
||||
# 将相对路径转为绝对路径
|
||||
base = os.path.dirname(os.path.abspath(__file__))
|
||||
for key in ("keys_file", "decrypted_dir", "decoded_image_dir"):
|
||||
if key in cfg and not os.path.isabs(cfg[key]):
|
||||
cfg[key] = os.path.join(base, cfg[key])
|
||||
|
||||
# 自动推导微信数据根目录（db_dir 的上级目录）
|
||||
# db_dir 格式: D:\xwechat_files\<wxid>\db_storage
|
||||
# base_dir 格式: D:\xwechat_files\<wxid>
|
||||
db_dir = cfg.get("db_dir", "")
|
||||
if db_dir and os.path.basename(db_dir) == "db_storage":
|
||||
cfg["wechat_base_dir"] = os.path.dirname(db_dir)
|
||||
else:
|
||||
cfg["wechat_base_dir"] = db_dir
|
||||
|
||||
# decoded_image_dir 默认值
|
||||
if "decoded_image_dir" not in cfg:
|
||||
cfg["decoded_image_dir"] = os.path.join(base, "decoded_images")
|
||||
|
||||
return cfg
|
||||
"""
|
||||
配置加载器 - 从 config.json 读取路径配置
|
||||
首次运行时自动检测微信数据目录，检测失败则提示手动配置
|
||||
"""
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import sys
|
||||
|
||||
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
||||
|
||||
_SYSTEM = platform.system().lower()
|
||||
_DEFAULT_TEMPLATE_DIR = (
|
||||
os.path.expanduser("~/Documents/xwechat_files/your_wxid/db_storage")
|
||||
if _SYSTEM == "linux"
|
||||
else r"D:\xwechat_files\your_wxid\db_storage"
|
||||
)
|
||||
|
||||
_DEFAULT = {
|
||||
"db_dir": _DEFAULT_TEMPLATE_DIR,
|
||||
"keys_file": "all_keys.json",
|
||||
"decrypted_dir": "decrypted",
|
||||
"decoded_image_dir": "decoded_images",
|
||||
"wechat_process": "wechat" if _SYSTEM == "linux" else "Weixin.exe",
|
||||
}
|
||||
|
||||
|
||||
def _choose_candidate(candidates):
|
||||
"""在多个候选目录中选择一个。"""
|
||||
if len(candidates) == 1:
|
||||
return candidates[0]
|
||||
if len(candidates) > 1:
|
||||
if not sys.stdin.isatty():
|
||||
return candidates[0]
|
||||
print("[!] 检测到多个微信数据目录（请选择当前正在运行的微信账号）:")
|
||||
for i, c in enumerate(candidates, 1):
|
||||
print(f" {i}. {c}")
|
||||
print(" 0. 跳过，稍后手动配置")
|
||||
try:
|
||||
while True:
|
||||
choice = input("请选择 [0-{}]: ".format(len(candidates))).strip()
|
||||
if choice == "0":
|
||||
return None
|
||||
if choice.isdigit() and 1 <= int(choice) <= len(candidates):
|
||||
return candidates[int(choice) - 1]
|
||||
print(" 无效输入，请重新选择")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print()
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _auto_detect_db_dir_windows():
|
||||
"""从微信本地配置自动检测 Windows db_storage 路径。
|
||||
|
||||
读取 %APPDATA%\\Tencent\\xwechat\\config\\*.ini，
|
||||
找到数据存储根目录，然后匹配 xwechat_files\\*\\db_storage。
|
||||
"""
|
||||
appdata = os.environ.get("APPDATA", "")
|
||||
config_dir = os.path.join(appdata, "Tencent", "xwechat", "config")
|
||||
if not os.path.isdir(config_dir):
|
||||
return None
|
||||
|
||||
# 从 ini 文件中找到有效的目录路径
|
||||
data_roots = []
|
||||
for ini_file in glob.glob(os.path.join(config_dir, "*.ini")):
|
||||
try:
|
||||
# 微信 ini 可能是 utf-8 或 gbk 编码（中文路径）
|
||||
content = None
|
||||
for enc in ("utf-8", "gbk"):
|
||||
try:
|
||||
with open(ini_file, "r", encoding=enc) as f:
|
||||
content = f.read(1024).strip()
|
||||
break
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if not content or any(c in content for c in "\n\r\x00"):
|
||||
continue
|
||||
if os.path.isdir(content):
|
||||
data_roots.append(content)
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
# 在每个根目录下搜索 xwechat_files\*\db_storage
|
||||
seen = set()
|
||||
candidates = []
|
||||
for root in data_roots:
|
||||
pattern = os.path.join(root, "xwechat_files", "*", "db_storage")
|
||||
for match in glob.glob(pattern):
|
||||
normalized = os.path.normcase(os.path.normpath(match))
|
||||
if os.path.isdir(match) and normalized not in seen:
|
||||
seen.add(normalized)
|
||||
candidates.append(match)
|
||||
|
||||
return _choose_candidate(candidates)
|
||||
|
||||
|
||||
def _auto_detect_db_dir_linux():
|
||||
"""自动检测 Linux 微信 db_storage 路径。"""
|
||||
seen = set()
|
||||
candidates = []
|
||||
search_roots = {
|
||||
os.path.expanduser("~/Documents/xwechat_files"),
|
||||
}
|
||||
|
||||
if os.path.isdir("/home"):
|
||||
for entry in os.listdir("/home"):
|
||||
search_roots.add(os.path.join("/home", entry, "Documents", "xwechat_files"))
|
||||
|
||||
for root in search_roots:
|
||||
if not os.path.isdir(root):
|
||||
continue
|
||||
pattern = os.path.join(root, "*", "db_storage")
|
||||
for match in glob.glob(pattern):
|
||||
normalized = os.path.normcase(os.path.normpath(match))
|
||||
if os.path.isdir(match) and normalized not in seen:
|
||||
seen.add(normalized)
|
||||
candidates.append(match)
|
||||
|
||||
old_path = os.path.expanduser("~/.local/share/weixin/data/db_storage")
|
||||
if os.path.isdir(old_path):
|
||||
normalized = os.path.normcase(os.path.normpath(old_path))
|
||||
if normalized not in seen:
|
||||
candidates.append(old_path)
|
||||
|
||||
# Linux 优先使用最近活跃账号：按 message 目录 mtime 降序
|
||||
def _mtime(path):
|
||||
msg_dir = os.path.join(path, "message")
|
||||
target = msg_dir if os.path.isdir(msg_dir) else path
|
||||
try:
|
||||
return os.path.getmtime(target)
|
||||
except OSError:
|
||||
return 0
|
||||
|
||||
candidates.sort(key=_mtime, reverse=True)
|
||||
return _choose_candidate(candidates)
|
||||
|
||||
|
||||
def auto_detect_db_dir():
|
||||
if _SYSTEM == "windows":
|
||||
return _auto_detect_db_dir_windows()
|
||||
if _SYSTEM == "linux":
|
||||
return _auto_detect_db_dir_linux()
|
||||
return None
|
||||
|
||||
|
||||
def load_config():
|
||||
cfg = {}
|
||||
if os.path.exists(CONFIG_FILE):
|
||||
try:
|
||||
with open(CONFIG_FILE) as f:
|
||||
cfg = json.load(f)
|
||||
except json.JSONDecodeError:
|
||||
print(f"[!] {CONFIG_FILE} 格式损坏，将使用默认配置")
|
||||
cfg = {}
|
||||
cfg = {**_DEFAULT, **cfg}
|
||||
|
||||
# db_dir 缺失或仍为模板值时，尝试自动检测
|
||||
db_dir = cfg.get("db_dir", "")
|
||||
if not db_dir or db_dir == _DEFAULT_TEMPLATE_DIR or "your_wxid" in db_dir:
|
||||
detected = auto_detect_db_dir()
|
||||
if detected:
|
||||
print(f"[+] 自动检测到微信数据目录: {detected}")
|
||||
# 合并默认值并保存
|
||||
cfg = {**_DEFAULT, **cfg, "db_dir": detected}
|
||||
with open(CONFIG_FILE, "w") as f:
|
||||
json.dump(cfg, f, indent=4, ensure_ascii=False)
|
||||
print(f"[+] 已保存到: {CONFIG_FILE}")
|
||||
else:
|
||||
if not os.path.exists(CONFIG_FILE):
|
||||
with open(CONFIG_FILE, "w") as f:
|
||||
json.dump(_DEFAULT, f, indent=4, ensure_ascii=False)
|
||||
print(f"[!] 未能自动检测微信数据目录")
|
||||
print(f" 请手动编辑 {CONFIG_FILE} 中的 db_dir 字段")
|
||||
if _SYSTEM == "linux":
|
||||
print(" Linux 默认路径类似: ~/Documents/xwechat_files/<wxid>/db_storage")
|
||||
else:
|
||||
print(f" 路径可在 微信设置 → 文件管理 中找到")
|
||||
sys.exit(1)
|
||||
|
||||
# 将相对路径转为绝对路径
|
||||
base = os.path.dirname(os.path.abspath(__file__))
|
||||
for key in ("keys_file", "decrypted_dir", "decoded_image_dir"):
|
||||
if key in cfg and not os.path.isabs(cfg[key]):
|
||||
cfg[key] = os.path.join(base, cfg[key])
|
||||
|
||||
# 自动推导微信数据根目录（db_dir 的上级目录）
|
||||
# db_dir 格式: D:\xwechat_files\<wxid>\db_storage
|
||||
# base_dir 格式: D:\xwechat_files\<wxid>
|
||||
db_dir = cfg.get("db_dir", "")
|
||||
if db_dir and os.path.basename(db_dir) == "db_storage":
|
||||
cfg["wechat_base_dir"] = os.path.dirname(db_dir)
|
||||
else:
|
||||
cfg["wechat_base_dir"] = db_dir
|
||||
|
||||
# decoded_image_dir 默认值
|
||||
if "decoded_image_dir" not in cfg:
|
||||
cfg["decoded_image_dir"] = os.path.join(base, "decoded_images")
|
||||
|
||||
return cfg
|
||||
|
||||
Reference in New Issue
Block a user