Kris
e9b57546ea
新增大量WhatsApp适配器相关代码，包括账号管理、会话处理、消息收发、验证授权、媒体处理、互动命令、审批流程、健康检测等完整功能模块，搭建基础的Baileys协议WhatsApp接入能力
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
|
|
from yuxi.utils.logging_config import logger
|
|
|
|
|
|
class PathOwnership(StrEnum):
|
|
OWNED = "owned"
|
|
UNSAFE_OWNED = "unsafe_owned"
|
|
EXTERNAL = "external"
|
|
|
|
|
|
def _classify_path(auth_dir: Path, target: Path) -> PathOwnership:
|
|
try:
|
|
resolved_auth = auth_dir.resolve()
|
|
resolved_target = target.resolve()
|
|
except OSError:
|
|
return PathOwnership.EXTERNAL
|
|
|
|
try:
|
|
resolved_target.relative_to(resolved_auth)
|
|
except ValueError:
|
|
return PathOwnership.EXTERNAL
|
|
|
|
if target.is_symlink():
|
|
return PathOwnership.UNSAFE_OWNED
|
|
|
|
try:
|
|
if not target.exists():
|
|
return PathOwnership.OWNED
|
|
abs_resolved = target.resolve(strict=False)
|
|
abs_resolved.relative_to(resolved_auth)
|
|
except ValueError:
|
|
return PathOwnership.UNSAFE_OWNED
|
|
|
|
return PathOwnership.OWNED
|
|
|
|
|
|
def _is_safe_path(auth_dir: Path, target: Path) -> bool:
|
|
return _classify_path(auth_dir, target) == PathOwnership.OWNED
|
|
|
|
|
|
def perform_logout_cleanup(auth_dir: Path) -> dict[str, int]:
|
|
auth_dir = Path(auth_dir)
|
|
classification = _classify_path(auth_dir, auth_dir)
|
|
|
|
if classification == PathOwnership.EXTERNAL:
|
|
logger.error(f"LogoutSecurity: auth_dir '{auth_dir}' is outside expected path tree")
|
|
return {"owned": 0, "unsafe_owned": 0, "external": 1}
|
|
if classification == PathOwnership.UNSAFE_OWNED:
|
|
logger.error(f"LogoutSecurity: auth_dir '{auth_dir}' is a symlink — refusing cleanup")
|
|
return {"owned": 0, "unsafe_owned": 1, "external": 0}
|
|
|
|
counts = {"owned": 0, "unsafe_owned": 0, "external": 0}
|
|
|
|
files_to_check = {
|
|
"creds": "creds.json",
|
|
"creds_bak": "creds.json.bak",
|
|
"app_state": "app-state-sync-key.json",
|
|
"prekeys": "pre-keys.json",
|
|
"sender_key": "sender-key-store.json",
|
|
}
|
|
|
|
for label, filename in files_to_check.items():
|
|
file_path = auth_dir / filename
|
|
if not file_path.exists():
|
|
continue
|
|
|
|
ownership = _classify_path(auth_dir, file_path)
|
|
if ownership == PathOwnership.EXTERNAL:
|
|
logger.warning(f"LogoutSecurity: skipping external path '{file_path}'")
|
|
counts["external"] += 1
|
|
continue
|
|
if ownership == PathOwnership.UNSAFE_OWNED:
|
|
logger.warning(f"LogoutSecurity: skipping symlink/unresolvable '{file_path}'")
|
|
counts["unsafe_owned"] += 1
|
|
continue
|
|
|
|
try:
|
|
file_path.unlink()
|
|
logger.info(f"LogoutSecurity: removed '{file_path}' ({label})")
|
|
counts["owned"] += 1
|
|
except OSError as e:
|
|
logger.error(f"LogoutSecurity: failed to remove '{file_path}': {e}")
|
|
|
|
logger.info(
|
|
f"LogoutSecurity: cleanup complete — owned={counts['owned']}, "
|
|
f"unsafe_owned={counts['unsafe_owned']}, external={counts['external']}"
|
|
)
|
|
return counts
|
|
|
|
|
|
def validate_credential_freshness(auth_dir: Path, max_age_hours: int = 168) -> bool:
|
|
creds = auth_dir / "creds.json"
|
|
if not creds.exists():
|
|
return False
|
|
try:
|
|
stat = creds.stat()
|
|
age_hours = (creds.stat().st_mtime - stat.st_ctime) / 3600
|
|
return abs(age_hours) < max_age_hours
|
|
except OSError:
|
|
return False
|