Kris
068cf70fe9
新增 iMessage 通道适配器完整实现，包含： 1. 核心适配器与工具工厂导出 2. 运行时存储、反射防护、会话路由等基础组件 3. 消息信封、线程管理、回复上下文格式化 4. Tapback 表情反应处理、自定义异常体系 5. 审批按钮、联系人解析、速率限制功能 6. 文本净化、目标解析、缓存管理模块 7. 配置 schema、多账户支持、安装向导等配置模块 8. 审计日志、媒体AI处理等扩展功能
85 lines
2.2 KiB
Python
85 lines
2.2 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from yuxi.utils.logging_config import logger
|
||
|
||
DEFAULT_IMESSAGE_ATTACHMENT_ROOTS = [
|
||
"/Users/*/Library/Messages/Attachments",
|
||
]
|
||
|
||
|
||
def resolve_attachment_roots(
|
||
configured_roots: list[str] | None = None,
|
||
) -> list[str]:
|
||
if configured_roots:
|
||
return configured_roots
|
||
expanded: list[str] = []
|
||
for pattern in DEFAULT_IMESSAGE_ATTACHMENT_ROOTS:
|
||
import glob as _glob_module
|
||
|
||
matches = _glob_module.glob(pattern)
|
||
if matches:
|
||
expanded.extend(matches)
|
||
return expanded or DEFAULT_IMESSAGE_ATTACHMENT_ROOTS
|
||
|
||
|
||
def validate_attachment_path(
|
||
file_path: str,
|
||
allowed_roots: list[str] | None = None,
|
||
) -> bool:
|
||
"""校验附件路径是否在允许的根路径内。
|
||
|
||
阻止路径遍历攻击（如 ../../etc/passwd）。
|
||
"""
|
||
if not file_path:
|
||
return False
|
||
|
||
real_path = os.path.realpath(file_path) if os.path.exists(file_path) else os.path.abspath(file_path)
|
||
|
||
roots = allowed_roots or []
|
||
if not roots:
|
||
return True
|
||
|
||
for root in roots:
|
||
root_real = os.path.realpath(root) if os.path.exists(root) else os.path.abspath(root)
|
||
try:
|
||
Path(real_path).relative_to(root_real)
|
||
return True
|
||
except ValueError:
|
||
continue
|
||
|
||
logger.warning(f"[iMessage/Security] Attachment path rejected: {file_path} not in allowed roots")
|
||
return False
|
||
|
||
|
||
def validate_remote_attachment_url(
|
||
url: str,
|
||
allowed_roots: list[str] | None = None,
|
||
server_url: str = "",
|
||
) -> bool:
|
||
"""校验远程附件 URL 是否安全。"""
|
||
if not url:
|
||
return False
|
||
|
||
if allowed_roots:
|
||
for root in allowed_roots:
|
||
if url.startswith(root):
|
||
return True
|
||
return False
|
||
|
||
if server_url and url.startswith(server_url):
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def sanitize_attachment_filename(filename: str) -> str:
|
||
"""净化附件文件名，移除路径分隔符。"""
|
||
if not filename:
|
||
return "attachment"
|
||
basename = os.path.basename(filename)
|
||
safe = "".join(c for c in basename if c.isalnum() or c in "._-() ")
|
||
return safe or "attachment"
|