mirror of https://github.com/jackwener/wx-cli.git
fix(linux): address code review feedback
- SUDO_USER: skip fallback entirely when user is invalid (KeyError) - load_config: move default merge after db_dir check to avoid dead code - _is_wechat_process: prefer exact comm match, use exe substring as fallbackfeat/daemon-cli
parent
3d58b6508c
commit
30112b9a10
11
config.py
11
config.py
|
|
@ -119,12 +119,10 @@ def _auto_detect_db_dir_linux():
|
|||
# 验证 SUDO_USER 是合法系统用户,防止路径注入
|
||||
import pwd
|
||||
try:
|
||||
pw = pwd.getpwnam(sudo_user)
|
||||
sudo_home = pw.pw_dir
|
||||
sudo_home = pwd.getpwnam(sudo_user).pw_dir
|
||||
except KeyError:
|
||||
sudo_home = None
|
||||
if not sudo_home:
|
||||
sudo_home = os.path.expanduser(f"~{sudo_user}")
|
||||
if sudo_home:
|
||||
fallback = os.path.join(sudo_home, "Documents", "xwechat_files")
|
||||
if fallback not in search_roots:
|
||||
search_roots.append(fallback)
|
||||
|
|
@ -176,15 +174,12 @@ def load_config():
|
|||
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)
|
||||
|
|
@ -200,6 +195,8 @@ def load_config():
|
|||
else:
|
||||
print(f" 路径可在 微信设置 → 文件管理 中找到")
|
||||
sys.exit(1)
|
||||
else:
|
||||
cfg = {**_DEFAULT, **cfg}
|
||||
|
||||
# 将相对路径转为绝对路径
|
||||
base = os.path.dirname(os.path.abspath(__file__))
|
||||
|
|
|
|||
|
|
@ -28,27 +28,31 @@ def _safe_readlink(path):
|
|||
return ""
|
||||
|
||||
|
||||
_KNOWN_COMMS = {"wechat", "wechatappex", "weixin"}
|
||||
_INTERPRETER_PREFIXES = ("python", "bash", "sh", "zsh", "node", "perl", "ruby")
|
||||
|
||||
|
||||
def _is_wechat_process(pid):
|
||||
"""检查 pid 是否为微信进程。
|
||||
|
||||
使用子串匹配以覆盖 wechat 主进程和 WeChatAppEx 子进程,
|
||||
同时排除自身和解释器进程(如 python3 find_all_keys.py)。
|
||||
优先精确匹配 comm 名称(wechat、WeChatAppEx 等),
|
||||
再用 exe 路径子串匹配作为 fallback,同时排除解释器进程。
|
||||
"""
|
||||
if pid == os.getpid():
|
||||
return False
|
||||
try:
|
||||
with open(f"/proc/{pid}/comm") as f:
|
||||
comm = f.read().strip()
|
||||
# 优先精确匹配 comm(最可靠)
|
||||
if comm.lower() in _KNOWN_COMMS:
|
||||
return True
|
||||
exe_path = _safe_readlink(f"/proc/{pid}/exe")
|
||||
exe_name = os.path.basename(exe_path)
|
||||
# 排除脚本解释器进程(避免匹配 python3.11 wechat-decrypt 等)
|
||||
if any(exe_name.lower().startswith(p) for p in _INTERPRETER_PREFIXES):
|
||||
return False
|
||||
haystack = f"{comm} {exe_name}".lower()
|
||||
return "wechat" in haystack or "weixin" in haystack
|
||||
# fallback: exe 名称子串匹配
|
||||
return "wechat" in exe_name.lower() or "weixin" in exe_name.lower()
|
||||
except (PermissionError, FileNotFoundError, ProcessLookupError):
|
||||
return False
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue