""" core/config_validator.py - 配置文件安全验证模块 ================================================ 当 "配置助手" agent 修改配置文件时，确保不会把配置改坏。 核心流程: 备份 → 验证 → 原子写入（三步安全保存）。 使用示例:: from core.config_validator import ConfigValidator validator = ConfigValidator(config_file_path="~/.myagent/config.json") # 安全保存 result = validator.safe_save(new_config) if not result["ok"]: print(f"验证失败: {result['errors']}") else: print(f"保存成功，备份: {result['backup_path']}") # 查看差异 diff = validator.get_diff(new_config) print(f"变更的键: {diff['changed_keys']}") # 恢复备份 result = validator.restore_latest() """ from __future__ import annotations import json import os import re import shutil import tempfile from datetime import datetime from pathlib import Path from typing import Any, Optional # ============================================================================== # 合法值常量 # ============================================================================== # LLM provider 白名单 VALID_LLM_PROVIDERS = {"openai", "anthic", "ollama", "zhipu", "custom"} # 修正拼写：anthropic VALID_LLM_PROVIDERS = {"openai", "anthropic", "ollama", "zhipu", "custom"} # executor 执行模式 VALID_EXECUTION_MODES = {"local", "sandbox"} # 沙盒内存限制正则 SANDBOX_MEMORY_RE = re.compile(r"^\d+[bkmg]$", re.IGNORECASE) # 端口范围 MIN_PORT = 1 MAX_PORT = 65535 # JSON 序列化后最大尺寸（10 MB） MAX_JSON_SIZE_BYTES = 10 * 1024 * 1024 # 最多保留备份数量 MAX_BACKUPS = 20 # 备份文件命名格式 BACKUP_NAME_PATTERN = "config_{datetime}.json" # ── 可疑注入模式 ── # 匹配配置值中可能存在的命令注入 / 路径穿越等危险内容 SUSPICIOUS_PATTERNS: list[re.Pattern] = [ re.compile(r";\s*(rm|sudo|chmod|chown|sh|bash|curl|wget|nc|python|perl|ruby|php)\b", re.IGNORECASE), re.compile(r"\|\s*(rm|sudo|chmod|sh|bash|curl|wget|nc)\b", re.IGNORECASE), re.compile(r"`[^`]*(rm|sudo|sh|bash)\b", re.IGNORECASE), re.compile(r"\$\([^)]*\)"), # $(command) 形式 re.compile(r"\.\.[\\/]"), # 路径穿越 ../ re.compile(r"/etc/(passwd|shadow|hosts)"), # 敏感系统文件 re.compile(r"eval\s*\("), # eval() 调用 re.compile(r"exec\s*\("), # exec() 调用 re.compile(r"__import__"), # Python 动态导入 re.compile(r"os\.(system|popen|exec)", re.IGNORECASE), re.compile(r"subprocess\.(call|run|Popen)", re.IGNORECASE), ] # Agent 名称合法字符（字母、数字、下划线、连字符、中文） AGENT_NAME_RE = re.compile(r"^[\w\u4e00-\u9fff\-]+$") class ConfigValidator: """配置文件安全验证器。 负责： 1. 备份当前配置文件（带时间戳，最多保留 20 份） 2. 多维度验证新配置数据的合法性 3. 原子性安全保存（备份 → 验证 → 临时文件 → os.replace） 4. 从备份恢复配置 5. 比较新旧配置差异 """ def __init__(self, config_file_path: str, backup_dir: str = ""): """初始化验证器。 Args: config_file_path: 配置文件路径（如 ~/.myagent/config.json）。 支持 ``~`` 展开。 backup_dir: 备份目录路径。为空时默认使用配置文件所在目录下的 ``backups/`` 子目录。支持 ``~`` 展开。 """ self._config_file = Path(config_file_path).expanduser().resolve() if backup_dir: self._backup_dir = Path(backup_dir).expanduser().resolve() else: self._backup_dir = self._config_file.parent / "backups" # ========================================================================== # 公共属性 # ========================================================================== @property def config_file(self) -> Path: """配置文件路径 (Path 对象)。""" return self._config_file @property def backup_dir(self) -> Path: """备份目录路径 (Path 对象)。""" return self._backup_dir # ========================================================================== # 备份管理 # ========================================================================== def backup(self) -> str: """备份当前配置文件。 创建带时间戳的备份文件，格式为 ``config_20260408_153045.json``。 最多保留最近 ``MAX_BACKUPS``（20）个备份，超出部分自动删除最旧的。 Returns: str: 备份文件的绝对路径。 Raises: FileNotFoundError: 当配置文件不存在时。 OSError: 当备份创建失败时。 """ if not self._config_file.exists(): raise FileNotFoundError(f"配置文件不存在: {self._config_file}") # 确保备份目录存在 self._backup_dir.mkdir(parents=True, exist_ok=True) # 生成带时间戳的文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_name = BACKUP_NAME_PATTERN.format(datetime=timestamp) backup_path = self._backup_dir / backup_name # 如果同名备份已存在（极低概率），追加序号 if backup_path.exists(): for i in range(1, 100): backup_path = self._backup_dir / f"config_{timestamp}_{i}.json" if not backup_path.exists(): break shutil.copy2(str(self._config_file), str(backup_path)) # 清理超出数量的旧备份 self._cleanup_old_backups() return str(backup_path) def _cleanup_old_backups(self) -> None: """删除最旧的备份，保留最多 MAX_BACKUPS 份。""" backups = self._get_sorted_backups() if len(backups) > MAX_BACKUPS: for old_backup in backups[MAX_BACKUPS:]: try: old_backup.unlink() except OSError: pass # 忽略删除失败的个别文件 def _get_sorted_backups(self) -> list[Path]: """获取按修改时间排序（最新在前）的备份文件列表。""" if not self._backup_dir.exists(): return [] backups = [ p for p in self._backup_dir.iterdir() if p.is_file() and p.name.startswith("config_") and p.suffix == ".json" ] # 按修改时间降序排列 backups.sort(key=lambda p: p.stat().st_mtime, reverse=True) return backups def list_backups(self) -> list[dict]: """列出所有备份文件。 Returns: list[dict]: 每个元素包含 ``filename``, ``path``, ``size_bytes``, ``modified_at`` 四个字段，按时间降序排列。 """ backups = self._get_sorted_backups() result = [] for bp in backups: stat = bp.stat() result.append({ "filename": bp.name, "path": str(bp), "size_bytes": stat.st_size, "modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat(), }) return result # ========================================================================== # 配置验证 # ========================================================================== def validate(self, config_data: dict) -> tuple[bool, list[str]]: """验证配置数据的合法性。 执行以下检查： 1. 顶层必须是有效的 ``dict`` 2. ``llm`` 部分字段类型与范围检查 3. ``executor`` 部分字段检查 4. ``agent`` 部分字段检查 5. ``models_library`` 中模型 ID 唯一性 6. JSON 序列化后大小不超过 10 MB 7. 配置值中不包含可疑的系统命令或路径（注入检测） 8. Agent 名称合法性 9. 端口号合法范围（1-65535） Args: config_data: 待验证的配置字典。 Returns: tuple[bool, list[str]]: - 第一项 ``True`` 表示通过，``False`` 表示存在错误 - 第二项为错误描述列表（通过时为空列表） """ errors: list[str] = [] # ── 1. 顶层类型 ── if not isinstance(config_data, dict): return (False, ["配置数据必须是 dict 类型"]) # ── 2. LLM 部分 ── self._validate_llm(config_data.get("llm"), errors) # ── 3. Executor 部分 ── self._validate_executor(config_data.get("executor"), errors) # ── 4. Agent 部分 ── self._validate_agent(config_data.get("agent"), errors) # ── 5. models_library 唯一性 ── self._validate_models_library(config_data.get("models_library"), errors) # ── 6. JSON 大小 ── self._validate_json_size(config_data, errors) # ── 7. 可疑注入检测 ── self._validate_no_injection(config_data, errors) # ── 8. Agent 名称特殊字符 ── self._validate_agent_names(config_data, errors) # ── 9. 端口号范围 ── self._validate_ports(config_data, errors) return (len(errors) == 0, errors) # ------------------------------------------------------------------ # LLM 验证 # ------------------------------------------------------------------ def _validate_llm(self, llm: Any, errors: list[str]) -> None: """验证 llm 配置节。""" if llm is None: return # llm 为可选节，不强制存在 if not isinstance(llm, dict): errors.append("llm 必须是 dict 类型") return # provider provider = llm.get("provider") if provider is not None and not isinstance(provider, str): errors.append("llm.provider 必须是字符串") elif isinstance(provider, str) and provider and provider not in VALID_LLM_PROVIDERS: errors.append( f"llm.provider 值 '{provider}' 无效，" f"必须是以下之一: {', '.join(sorted(VALID_LLM_PROVIDERS))}" ) # temperature temperature = llm.get("temperature") if temperature is not None: if not isinstance(temperature, (int, float)): errors.append("llm.temperature 必须是数字") elif not (0.0 <= float(temperature) <= 2.0): errors.append("llm.temperature 必须在 0.0 ~ 2.0 之间") # max_tokens max_tokens = llm.get("max_tokens") if max_tokens is not None: if not isinstance(max_tokens, int) or isinstance(max_tokens, bool): errors.append("llm.max_tokens 必须是整数") elif max_tokens <= 0: errors.append("llm.max_tokens 必须是正整数") elif max_tokens > 128000: errors.append("llm.max_tokens 不能超过 128000") # timeout timeout = llm.get("timeout") if timeout is not None: if not isinstance(timeout, int) or isinstance(timeout, bool): errors.append("llm.timeout 必须是整数") elif timeout <= 0: errors.append("llm.timeout 必须是正整数") elif timeout > 600: errors.append("llm.timeout 不能超过 600") # max_retries max_retries = llm.get("max_retries") if max_retries is not None: if not isinstance(max_retries, int) or isinstance(max_retries, bool): errors.append("llm.max_retries 必须是整数") elif not (0 <= max_retries <= 10): errors.append("llm.max_retries 必须在 0 ~ 10 之间") # ------------------------------------------------------------------ # Executor 验证 # ------------------------------------------------------------------ def _validate_executor(self, executor: Any, errors: list[str]) -> None: """验证 executor 配置节。""" if executor is None: return if not isinstance(executor, dict): errors.append("executor 必须是 dict 类型") return # timeout timeout = executor.get("timeout") if timeout is not None: if not isinstance(timeout, int) or isinstance(timeout, bool): errors.append("executor.timeout 必须是整数") elif timeout <= 0: errors.append("executor.timeout 必须是正整数") elif timeout > 3600: errors.append("executor.timeout 不能超过 3600") # sandbox_memory sandbox_memory = executor.get("sandbox_memory") if sandbox_memory is not None: if not isinstance(sandbox_memory, str): errors.append("executor.sandbox_memory 必须是字符串") elif not SANDBOX_MEMORY_RE.match(sandbox_memory): errors.append( f"executor.sandbox_memory 格式错误 '{sandbox_memory}'，" f"必须匹配 ^\\d+[bkmg]$（如 512m, 2g）" ) # execution_mode execution_mode = executor.get("execution_mode") if execution_mode is not None: if not isinstance(execution_mode, str): errors.append("executor.execution_mode 必须是字符串") elif execution_mode not in VALID_EXECUTION_MODES: errors.append( f"executor.execution_mode 值 '{execution_mode}' 无效，" f"必须是: {', '.join(sorted(VALID_EXECUTION_MODES))}" ) # ------------------------------------------------------------------ # Agent 验证 # ------------------------------------------------------------------ def _validate_agent(self, agent: Any, errors: list[str]) -> None: """验证 agent 配置节。""" if agent is None: return if not isinstance(agent, dict): errors.append("agent 必须是 dict 类型") return # max_iterations max_iterations = agent.get("max_iterations") if max_iterations is not None: if not isinstance(max_iterations, int) or isinstance(max_iterations, bool): errors.append("agent.max_iterations 必须是整数") elif max_iterations <= 0: errors.append("agent.max_iterations 必须是正整数") elif max_iterations > 200: errors.append("agent.max_iterations 不能超过 200") # max_parallel max_parallel = agent.get("max_parallel") if max_parallel is not None: if not isinstance(max_parallel, int) or isinstance(max_parallel, bool): errors.append("agent.max_parallel 必须是整数") elif max_parallel <= 0: errors.append("agent.max_parallel 必须是正整数") elif max_parallel > 20: errors.append("agent.max_parallel 不能超过 20") # ------------------------------------------------------------------ # models_library 唯一性 # ------------------------------------------------------------------ def _validate_models_library(self, models: Any, errors: list[str]) -> None: """验证 models_library 中模型 ID 不能重复。""" if models is None: return if not isinstance(models, list): errors.append("models_library 必须是 list 类型") return seen_ids: dict[str, int] = {} for idx, entry in enumerate(models): if not isinstance(entry, dict): errors.append(f"models_library[{idx}] 必须是 dict 类型") continue model_id = entry.get("id") if model_id is not None and isinstance(model_id, str): if model_id in seen_ids: errors.append( f"models_library 中模型 ID '{model_id}' 重复" f"（索引 {seen_ids[model_id]} 和 {idx}）" ) else: seen_ids[model_id] = idx # ------------------------------------------------------------------ # JSON 大小 # ------------------------------------------------------------------ def _validate_json_size(self, config_data: dict, errors: list[str]) -> None: """验证 JSON 序列化后不超过 10 MB。""" try: serialized = json.dumps(config_data, ensure_ascii=False) size = len(serialized.encode("utf-8")) if size > MAX_JSON_SIZE_BYTES: size_mb = size / (1024 * 1024) errors.append( f"配置文件 JSON 序列化后大小 {size_mb:.2f} MB，" f"超过限制 {MAX_JSON_SIZE_BYTES // (1024 * 1024)} MB" ) except (TypeError, ValueError) as e: errors.append(f"配置数据无法序列化为 JSON: {e}") # ------------------------------------------------------------------ # 注入检测 # ------------------------------------------------------------------ def _validate_no_injection(self, config_data: dict, errors: list[str]) -> None: """检测配置值中是否包含可疑的系统命令或路径（防止注入）。""" found_issues = self._scan_suspicious(config_data, path="root") for issue in found_issues: errors.append(f"安全风险 - {issue}") def _scan_suspicious(self, obj: Any, path: str) -> list[str]: """递归扫描配置值中的可疑模式。""" issues: list[str] = [] if isinstance(obj, str): for pattern in SUSPICIOUS_PATTERNS: match = pattern.search(obj) if match: # 提取上下文：匹配位置前后各 20 字符 start = max(0, match.start() - 20) end = min(len(obj), match.end() + 20) context = obj[start:end] issues.append( f"[{path}] 检测到可疑内容: ...{context}..." ) break # 每个字段只报告一次 elif isinstance(obj, dict): for key, value in obj.items(): child_path = f"{path}.{key}" if path != "root" else key issues.extend(self._scan_suspicious(value, child_path)) elif isinstance(obj, list): for idx, item in enumerate(obj): child_path = f"{path}[{idx}]" issues.extend(self._scan_suspicious(item, child_path)) return issues # ------------------------------------------------------------------ # Agent 名称 # ------------------------------------------------------------------ def _validate_agent_names(self, config_data: dict, errors: list[str]) -> None: """验证 chat_platforms 等节中的名称字段不含特殊字符。""" # chat_platforms 中的 platform 字段 platforms = config_data.get("chat_platforms") if isinstance(platforms, list): for idx, p in enumerate(platforms): if not isinstance(p, dict): continue platform_name = p.get("platform") if platform_name and isinstance(platform_name, str): if not AGENT_NAME_RE.match(platform_name): errors.append( f"chat_platforms[{idx}].platform '{platform_name}' " f"包含非法字符（仅允许字母、数字、下划线、连字符、中文）" ) # organization.knowledge_admin org = config_data.get("organization") if isinstance(org, dict): admin = org.get("knowledge_admin") if admin and isinstance(admin, str) and not AGENT_NAME_RE.match(admin): errors.append( f"organization.knowledge_admin '{admin}' " f"包含非法字符（仅允许字母、数字、下划线、连字符、中文）" ) # ------------------------------------------------------------------ # 端口号 # ------------------------------------------------------------------ def _validate_ports(self, config_data: dict, errors: list[str]) -> None: """验证端口号在 1-65535 合法范围内。""" # 检查常见的端口字段 port_fields = [ ("web", "port"), ("web", "ssl_port"), ("api_server", "port"), ("server", "port"), ] for section, field in port_fields: section_data = config_data.get(section) if not isinstance(section_data, dict): continue port_val = section_data.get(field) if port_val is not None: if not isinstance(port_val, int) or isinstance(port_val, bool): errors.append(f"{section}.{field} 必须是整数") elif not (MIN_PORT <= port_val <= MAX_PORT): errors.append( f"{section}.{field} 值 {port_val} 超出合法端口范围 " f"({MIN_PORT}-{MAX_PORT})" ) # 遍历所有嵌套字段，查找名称包含 "port" 的整数配置项 self._scan_ports_recursive(config_data, "root", port_fields, errors) def _scan_ports_recursive( self, obj: Any, path: str, known_fields: list[tuple[str, str]], errors: list[str], ) -> None: """递归扫描所有包含 'port' 的字段。""" if not isinstance(obj, dict): return known_set = {f"{s}.{f}" for s, f in known_fields} for key, value in obj.items(): child_path = f"{path}.{key}" if path != "root" else key if isinstance(value, (int, float)) and not isinstance(value, bool): # 检测字段名中包含 port if "port" in key.lower() and child_path not in known_set: port_val = int(value) if not (MIN_PORT <= port_val <= MAX_PORT): errors.append( f"{child_path} 值 {port_val} 超出合法端口范围 " f"({MIN_PORT}-{MAX_PORT})" ) elif isinstance(value, dict): self._scan_ports_recursive(value, child_path, known_fields, errors) elif isinstance(value, list): for idx, item in enumerate(value): if isinstance(item, dict): self._scan_ports_recursive( item, f"{child_path}[{idx}]", known_fields, errors ) # ========================================================================== # 安全保存 # ========================================================================== def safe_save(self, new_config_data: dict) -> dict: """安全保存配置：备份 → 验证 → 原子写入。 流程： 1. 自动备份当前配置文件（如果存在） 2. 验证新配置数据 3. 验证通过：写入临时文件 → ``os.replace`` 原子替换 4. 验证失败：不保存，返回错误 Args: new_config_data: 新的配置字典。 Returns: dict: 成功时 ``{"ok": True, "backup_path": "..."}``， 失败时 ``{"ok": False, "errors": [...]}``。 """ # ── Step 1: 备份 ── backup_path = "" if self._config_file.exists(): try: backup_path = self.backup() except (OSError, FileNotFoundError) as e: return {"ok": False, "errors": [f"备份失败: {e}"]} # ── Step 2: 验证 ── is_valid, validation_errors = self.validate(new_config_data) if not is_valid: return {"ok": False, "errors": validation_errors} # ── Step 3: 原子写入 ── try: self._atomic_write(new_config_data) except (OSError, TypeError, ValueError) as e: return {"ok": False, "errors": [f"写入失败: {e}"]} return {"ok": True, "backup_path": backup_path} def _atomic_write(self, config_data: dict) -> None: """原子性写入：先写临时文件，再 os.replace。 Args: config_data: 要写入的配置字典。 Raises: OSError: 文件操作失败。 TypeError / ValueError: 序列化失败。 """ # 确保目标目录存在 self._config_file.parent.mkdir(parents=True, exist_ok=True) # 写入临时文件（与目标文件同目录，确保同一文件系统，os.replace 才能原子操作） fd, tmp_path = tempfile.mkstemp( suffix=".tmp", prefix="config_", dir=str(self._config_file.parent), ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(config_data, f, ensure_ascii=False, indent=2) f.flush() os.fsync(f.fileno()) # 原子替换 os.replace(tmp_path, str(self._config_file)) except BaseException: # 写入失败时清理临时文件 try: os.unlink(tmp_path) except OSError: pass raise # ========================================================================== # 恢复备份 # ========================================================================== def restore_latest(self) -> dict: """从最近的备份恢复配置。 Returns: dict: 成功时 ``{"ok": True, "restored_from": "..."}``， 失败时 ``{"ok": False, "error": "..."}``。 """ backups = self._get_sorted_backups() if not backups: return {"ok": False, "error": "没有可用的备份文件"} latest_backup = backups[0] try: shutil.copy2(str(latest_backup), str(self._config_file)) except OSError as e: return {"ok": False, "error": f"恢复失败: {e}"} return {"ok": True, "restored_from": str(latest_backup)} # ========================================================================== # 差异比较 # ========================================================================== def get_diff(self, new_config_data: dict) -> dict: """比较当前配置和新配置的差异。 读取当前配置文件内容并与 ``new_config_data`` 进行深度比较， 报告变更、新增、删除的键。 Args: new_config_data: 新的配置字典。 Returns: dict: 包含三个列表的字典： - ``changed_keys``: 值发生变化的键路径列表 - ``added_keys``: 新配置中新增的键路径列表 - ``removed_keys``: 当前配置中存在但新配置中删除的键路径列表 """ # 读取当前配置 current_config: dict = {} if self._config_file.exists(): try: with open(self._config_file, "r", encoding="utf-8") as f: current_config = json.load(f) except (json.JSONDecodeError, OSError): current_config = {} changed: list[str] = [] added: list[str] = [] removed: list[str] = [] self._diff_recursive(current_config, new_config_data, "", changed, added, removed) return { "changed_keys": sorted(changed), "added_keys": sorted(added), "removed_keys": sorted(removed), } def _diff_recursive( self, old: Any, new: Any, path: str, changed: list[str], added: list[str], removed: list[str], ) -> None: """递归比较两个值，记录差异。""" if isinstance(old, dict) and isinstance(new, dict): all_keys = set(old.keys()) | set(new.keys()) for key in sorted(all_keys): child_path = f"{path}.{key}" if path else key if key in old and key in new: self._diff_recursive(old[key], new[key], child_path, changed, added, removed) elif key in new: added.append(child_path) else: removed.append(child_path) elif isinstance(old, list) and isinstance(new, list): # 列表比较：长度变化视为 changed if len(old) != len(new): changed.append(path if path else "root") else: for idx in range(len(old)): child_path = f"{path}[{idx}]" if path else f"[{idx}]" self._diff_recursive(old[idx], new[idx], child_path, changed, added, removed) elif old != new: changed.append(path if path else "root")