diff --git a/backend/package/yuxi/channel/doctor/__init__.py b/backend/package/yuxi/channel/doctor/__init__.py new file mode 100644 index 00000000..b8cf14fc --- /dev/null +++ b/backend/package/yuxi/channel/doctor/__init__.py @@ -0,0 +1,535 @@ +from __future__ import annotations + +import asyncio +import contextvars +import copy +import json +import logging +import time +from pathlib import Path +from typing import Any + +from yuxi.channel.doctor.models import ( + CheckStatus, + ConfigMutation, + ConnectivityCheckResult, + CredentialCheckResult, + DiagnosisResult, + DiagnosisWarning, + DoctorOptions, + PermissionCheckResult, + RepairResult, + Severity, +) +from yuxi.channel.config.defaults import TIMEOUT +from yuxi.channel.protocols import DoctorConfigMutation, DoctorProtocol +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +from yuxi.channel.security.log_sanitizer import sanitize_config_object + +logger = logging.getLogger(__name__) + +CHECK_TIMEOUT = TIMEOUT.doctor.check + +_NOT_FOUND = object() + + +def _safe_serialize(val: Any) -> str: + if val is None or val is _NOT_FOUND: + return "" + if isinstance(val, (dict, list, tuple)): + return json.dumps(val, ensure_ascii=False, default=str) + if isinstance(val, bool): + return "true" if val else "false" + return str(val) + + +def _values_differ(old: Any, new: Any) -> bool: + if old is _NOT_FOUND and new is _NOT_FOUND: + return False + if old is _NOT_FOUND or new is _NOT_FOUND: + return True + + old_normalized = "" if old is None else old + new_normalized = "" if new is None else new + + if type(old_normalized) is not type(new_normalized): + return True + + if isinstance(old_normalized, (dict, list)): + return json.dumps(old_normalized, sort_keys=True, ensure_ascii=False, default=str) != json.dumps( + new_normalized, sort_keys=True, ensure_ascii=False, default=str + ) + + return old_normalized != new_normalized + + +class ChannelDoctorRunner: + async def diagnose( + self, + channel_type: str, + account_id: str, + config: dict, + options: DoctorOptions | None = None, + ) -> DiagnosisResult: + plugin = self._get_doctor_plugin(channel_type) + opts = options or DoctorOptions() + + config_before = copy.deepcopy(config) + config = copy.deepcopy(config) + + result = DiagnosisResult( + channel_type=channel_type, + account_id=account_id, + overall_status=CheckStatus.PASS, + timestamp=time.time(), + config_before=sanitize_config_object(config_before), + config_after=sanitize_config_object(config), + ) + + self._run_config_normalization(result, plugin, config) + self._run_stale_config_cleanup(result, plugin, config, opts) + + await self._run_checks(result, plugin, config, opts) + + result.config_after = sanitize_config_object(config) + + self._aggregate_and_plan(result, plugin, opts) + + return result + + async def repair( + self, + channel_type: str, + step_id: str, + config: dict, + options: DoctorOptions | None = None, + diagnosis: DiagnosisResult | None = None, + ) -> RepairResult: + plugin = self._get_doctor_plugin(channel_type) + opts = options or DoctorOptions() + + if diagnosis is None: + diagnosis = await self.diagnose(channel_type, "default", config, opts) + + for step in diagnosis.repair_plan: + if step.id == step_id: + return await plugin.execute_repair(step, config) + + raise ValueError(f"Repair step {step_id} not found in diagnosis plan") + + async def repair_all( + self, + channel_type: str, + config: dict, + options: DoctorOptions | None = None, + diagnosis: DiagnosisResult | None = None, + ) -> DiagnosisResult: + plugin = self._get_doctor_plugin(channel_type) + opts = options or DoctorOptions() + + if not opts.should_repair: + raise ValueError( + f"repair_all requires repair_mode in (REPAIR, FORCE, YES), " + f"got {opts.repair_mode.value}" + ) + + if diagnosis is None: + diagnosis = await self.diagnose(channel_type, "default", config, opts) + + diagnosis = await self._execute_repairs(diagnosis, plugin, config, opts) + return diagnosis + + DEFAULT_DIAGNOSE_ALL_MAX_CONCURRENCY = 10 + + async def diagnose_all_channels( + self, + configs: dict[str, dict], + options: DoctorOptions | None = None, + max_concurrency: int | None = None, + ) -> list[DiagnosisResult]: + opts = options or DoctorOptions() + limit = max_concurrency if max_concurrency is not None else self.DEFAULT_DIAGNOSE_ALL_MAX_CONCURRENCY + semaphore = asyncio.Semaphore(limit) + + async def _diagnose_one(channel_type: str, cfg: dict) -> DiagnosisResult | None: + async with semaphore: + try: + return await self.diagnose(channel_type, "default", cfg, opts) + except NotImplementedError: + logger.warning( + "Channel %s skipped: DoctorProtocol not implemented", channel_type + ) + return None + except Exception as e: + logger.exception("Diagnosis failed for channel %s", channel_type) + return DiagnosisResult( + channel_type=channel_type, + account_id="default", + overall_status=CheckStatus.FAIL, + timestamp=time.time(), + warnings=[ + DiagnosisWarning( + severity=Severity.ERROR, + code="diagnose_error", + message=f"Diagnosis failed: {e}", + suggestion="Check plugin implementation and config", + ) + ], + ) + + tasks = [_diagnose_one(ct, cfg) for ct, cfg in configs.items()] + gathered = await asyncio.gather(*tasks) + return [r for r in gathered if r is not None] + + async def write_config( + self, + config_path: str | Path, + config: dict, + backup: bool = True, + ) -> None: + config_path = Path(config_path) + + if backup and config_path.exists(): + backup_path = config_path.with_suffix(config_path.suffix + ".bak") + config_path.rename(backup_path) + + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text( + json.dumps(config, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" + ) + + # ── internal helpers ────────────────────────────────── + + def _get_doctor_plugin(self, channel_type: str) -> DoctorProtocol: + plugin = ChannelPluginRegistry.get(channel_type) + if plugin is None: + raise NotImplementedError(f"Channel {channel_type} is not registered") + if not isinstance(plugin, DoctorProtocol): + raise NotImplementedError( + f"Channel {channel_type} does not implement DoctorProtocol" + ) + return plugin + + def _run_config_normalization( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + ) -> None: + try: + normalized = plugin.normalize_compatibility_config(config) + if normalized != config: + for rule in plugin.legacy_config_rules(): + old_val = self._resolve_nested(config, rule.old_key) + new_val = self._resolve_nested(normalized, rule.new_key) + if _values_differ(old_val, new_val): + result.config_changes.append( + ConfigMutation( + key=rule.new_key, + description=f"Migrated from {rule.old_key} -> {rule.new_key}", + old_value=_safe_serialize(old_val), + new_value=_safe_serialize(new_val), + version=len(result.config_changes) + 1, + reason="config_normalization", + ) + ) + config.clear() + config.update(normalized) + except Exception as e: + logger.exception( + "Config normalization failed for %s", result.channel_type + ) + result.warnings.append( + DiagnosisWarning( + severity=Severity.WARNING, + code="normalization_failed", + message=f"配置规范化失败: {e}", + suggestion="检查插件 normalize_compatibility_config 实现", + ) + ) + + if hasattr(plugin, "run_config_sequence"): + try: + repaired = plugin.run_config_sequence(config, False) + if repaired != config: + config.clear() + config.update(repaired) + except Exception: + logger.exception( + "Config sequence failed for %s", result.channel_type + ) + + def _run_stale_config_cleanup( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + options: DoctorOptions, + ) -> None: + if not options.should_repair: + return + + if hasattr(plugin, "clean_stale_config"): + try: + mutation: DoctorConfigMutation = plugin.clean_stale_config(config) + if mutation.config != config: + config.clear() + config.update(mutation.config) + for change in mutation.changes: + result.config_changes.append( + ConfigMutation( + key="", + description=change, + version=len(result.config_changes) + 1, + reason="stale_config_cleanup", + ) + ) + if mutation.warnings: + result.security_warnings.extend(mutation.warnings) + except Exception: + logger.exception( + "Stale config cleanup failed for %s", result.channel_type + ) + + async def _run_checks( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + options: DoctorOptions, + ) -> None: + async def _check_credential() -> CredentialCheckResult: + try: + return await asyncio.wait_for( + plugin.check_credentials(config), timeout=CHECK_TIMEOUT + ) + except TimeoutError: + logger.warning("Credential check timed out for %s", result.channel_type) + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error=f"凭证检查超时 ({CHECK_TIMEOUT}s)", + ) + except Exception as e: + logger.exception("Credential check failed for %s", result.channel_type) + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error=str(e), + ) + + async def _check_permission() -> PermissionCheckResult: + try: + return await asyncio.wait_for( + plugin.check_permissions(config), timeout=CHECK_TIMEOUT + ) + except TimeoutError: + logger.warning("Permission check timed out for %s", result.channel_type) + return PermissionCheckResult( + status=CheckStatus.FAIL, + error=f"权限检查超时 ({CHECK_TIMEOUT}s)", + ) + except Exception as e: + logger.exception("Permission check failed for %s", result.channel_type) + return PermissionCheckResult( + status=CheckStatus.FAIL, + error=str(e), + ) + + async def _check_connectivity() -> ConnectivityCheckResult: + try: + return await asyncio.wait_for( + plugin.check_connectivity(config), timeout=CHECK_TIMEOUT + ) + except TimeoutError: + logger.warning("Connectivity check timed out for %s", result.channel_type) + return ConnectivityCheckResult( + status=CheckStatus.FAIL, + error=f"连通性检查超时 ({CHECK_TIMEOUT}s)", + ) + except Exception as e: + logger.exception( + "Connectivity check failed for %s", result.channel_type + ) + return ConnectivityCheckResult( + status=CheckStatus.FAIL, + error=str(e), + ) + + result.credential, result.permission, result.connectivity = await asyncio.gather( + _check_credential(), + _check_permission(), + _check_connectivity(), + ) + + self._collect_allowlist_warnings(result, plugin, config) + self._collect_security_warnings(result, plugin, config) + + def _collect_allowlist_warnings( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + ) -> None: + try: + allowlist_cfg = config.get("allowlist", {}) + result.warnings.extend(plugin.collect_allowlist_warnings(allowlist_cfg)) + except Exception: + logger.exception( + "Allowlist warning collection failed for %s", + result.channel_type, + ) + + def _collect_security_warnings( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + ) -> None: + if hasattr(plugin, "collect_preview_warnings"): + try: + warnings = plugin.collect_preview_warnings(config, "doctor --fix") + result.security_warnings.extend(warnings) + except Exception: + logger.exception( + "Preview warning collection failed for %s", + result.channel_type, + ) + + if hasattr(plugin, "collect_mutable_allowlist_warnings"): + try: + warnings = plugin.collect_mutable_allowlist_warnings(config) + result.security_warnings.extend(warnings) + except Exception: + logger.exception( + "Mutable allowlist warning collection failed for %s", + result.channel_type, + ) + + if hasattr(plugin, "collect_empty_allowlist_extra_warnings"): + try: + ctx = {"config": config, "plugin": plugin} + if not plugin.should_skip_default_empty_group_allowlist_warning(ctx): + warnings = plugin.collect_empty_allowlist_extra_warnings(ctx) + result.security_warnings.extend(warnings) + except Exception: + logger.exception( + "Empty allowlist extra warning collection failed for %s", + result.channel_type, + ) + + def _aggregate_and_plan( + self, + result: DiagnosisResult, + plugin: DoctorProtocol, + options: DoctorOptions, + ) -> None: + if self._has_failures(result): + result.repair_plan = plugin.generate_repair_plan(result) + + result.overall_status = self._aggregate_status(result) + + def _has_failures(self, result: DiagnosisResult) -> bool: + checks = [result.credential, result.permission, result.connectivity] + return any(c and c.status != CheckStatus.PASS for c in checks) + + def _aggregate_status(self, result: DiagnosisResult) -> CheckStatus: + checks = [result.credential, result.permission, result.connectivity] + statuses = [c.status for c in checks if c] + if any(s == CheckStatus.FAIL for s in statuses): + return CheckStatus.FAIL + if any(s == CheckStatus.WARN for s in statuses): + return CheckStatus.WARN + if result.security_warnings: + return CheckStatus.WARN + return CheckStatus.PASS + + async def _execute_repairs( + self, + diagnosis: DiagnosisResult, + plugin: DoctorProtocol, + config: dict, + options: DoctorOptions, + ) -> DiagnosisResult: + for step in diagnosis.repair_plan: + if step.requires_interactive and options.is_non_interactive: + diagnosis.repair_results.append( + RepairResult( + step_id=step.id, + success=False, + message="Skipped: requires interactive confirmation", + ) + ) + logger.info( + "Repair step %s skipped: requires interactive confirmation", + step.id, + ) + continue + + if step.is_aggressive and not options.should_force: + diagnosis.repair_results.append( + RepairResult( + step_id=step.id, + success=False, + message="Skipped: requires --force for aggressive repair", + ) + ) + logger.info( + "Repair step %s skipped: requires --force", step.id + ) + continue + + try: + result = await plugin.execute_repair(step, config) + diagnosis.repair_results.append(result) + logger.info("Repair step %s: success=%s", step.id, result.success) + except Exception as e: + logger.exception("Repair step %s failed", step.id) + diagnosis.repair_results.append( + RepairResult( + step_id=step.id, + success=False, + message=str(e), + ) + ) + + return diagnosis + + @staticmethod + def _resolve_nested(d: dict, dotted_key: str) -> object: + keys = dotted_key.split(".") + current: object = d + for k in keys: + if isinstance(current, dict): + current = current.get(k, _NOT_FOUND) + else: + return _NOT_FOUND + return current + + +_doctor_runner_ctx: contextvars.ContextVar[ChannelDoctorRunner | None] = contextvars.ContextVar( + "doctor_runner", default=None +) + + +def get_doctor_runner() -> ChannelDoctorRunner: + runner = _doctor_runner_ctx.get() + if runner is None: + runner = ChannelDoctorRunner() + _doctor_runner_ctx.set(runner) + return runner + + +def set_doctor_runner(runner: ChannelDoctorRunner) -> None: + _doctor_runner_ctx.set(runner) + + +def reset_doctor_runner() -> None: + _doctor_runner_ctx.set(None) + + +def __getattr__(name: str): + if name == "doctor_runner": + return get_doctor_runner() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/backend/package/yuxi/channel/doctor/models.py b/backend/package/yuxi/channel/doctor/models.py new file mode 100644 index 00000000..6ff9de65 --- /dev/null +++ b/backend/package/yuxi/channel/doctor/models.py @@ -0,0 +1,264 @@ +from __future__ import annotations + +import hashlib +import time +from dataclasses import dataclass, field +from enum import StrEnum + + +class CheckStatus(StrEnum): + PASS = "pass" + WARN = "warn" + FAIL = "fail" + SKIP = "skip" + + +class Severity(StrEnum): + INFO = "info" + WARNING = "warning" + ERROR = "error" + + +class DoctorRepairMode(StrEnum): + NONE = "none" + YES = "yes" + REPAIR = "repair" + FORCE = "force" + NON_INTERACTIVE = "non_interactive" + DEEP = "deep" + + +@dataclass +class DoctorOptions: + repair_mode: DoctorRepairMode = DoctorRepairMode.NONE + deep: bool = False + generate_gateway_token: bool = False + workspace_suggestions: bool = True + + @property + def should_repair(self) -> bool: + return self.repair_mode in (DoctorRepairMode.REPAIR, DoctorRepairMode.FORCE, DoctorRepairMode.YES) + + @property + def should_force(self) -> bool: + return self.repair_mode == DoctorRepairMode.FORCE + + @property + def is_non_interactive(self) -> bool: + return self.repair_mode in ( + DoctorRepairMode.NON_INTERACTIVE, + DoctorRepairMode.REPAIR, + DoctorRepairMode.YES, + ) + + +@dataclass +class CredentialCheckResult: + status: CheckStatus + token_obtained: bool + expires_in: int | None = None + error: str | None = None + detail: dict = field(default_factory=dict) + + +@dataclass +class PermissionCheckResult: + status: CheckStatus + can_send_message: bool = False + can_read_message: bool = False + can_manage_group: bool = False + missing_permissions: list[str] = field(default_factory=list) + error: str | None = None + + +@dataclass +class ConnectivityCheckResult: + status: CheckStatus + latency_ms: float | None = None + endpoint: str = "" + error: str | None = None + + +@dataclass +class DiagnosisWarning: + severity: Severity + code: str + message: str + suggestion: str + + +@dataclass +class LegacyConfigRule: + old_key: str + new_key: str + transform: str | None = None + + +@dataclass +class ConfigMutation: + description: str = "" + old_value: str = "" + new_value: str = "" + key: str = "" + version: int = 0 + timestamp: float = field(default_factory=time.time) + operator: str = "system" + reason: str = "" + checksum: str = "" + + def __post_init__(self): + if not self.checksum: + self.checksum = self._compute_checksum() + + def _compute_checksum(self) -> str: + payload = f"{self.key}|{self.description}|{self.old_value}|{self.new_value}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +@dataclass +class ChannelStatusWarning: + channel_type: str + account_id: str + status: CheckStatus + message: str + detail: dict = field(default_factory=dict) + + +@dataclass +class RepairStep: + id: str + description: str + action: str + reversible: bool = True + params: dict = field(default_factory=dict) + requires_interactive: bool = False + is_aggressive: bool = False + + +@dataclass +class RepairResult: + step_id: str + success: bool + message: str + config_before: dict | None = None + config_after: dict | None = None + + +@dataclass +class DiagnosisResult: + channel_type: str + account_id: str + overall_status: CheckStatus + timestamp: float + + credential: CredentialCheckResult | None = None + permission: PermissionCheckResult | None = None + connectivity: ConnectivityCheckResult | None = None + + config_before: dict | None = None + config_after: dict | None = None + config_changes: list[ConfigMutation] = field(default_factory=list) + + warnings: list[DiagnosisWarning] = field(default_factory=list) + security_warnings: list[str] = field(default_factory=list) + channel_status_warnings: list[ChannelStatusWarning] = field(default_factory=list) + + repair_plan: list[RepairStep] = field(default_factory=list) + repair_results: list[RepairResult] = field(default_factory=list) + + def to_dict(self) -> dict: + result: dict = { + "channel_type": self.channel_type, + "account_id": self.account_id, + "overall_status": self.overall_status.value, + "timestamp": self.timestamp, + "credential": self._credential_dict(), + "permission": self._permission_dict(), + "connectivity": self._connectivity_dict(), + "config_before": self.config_before, + "config_after": self.config_after, + "config_changes": [ + { + "key": c.key, + "description": c.description, + "old_value": c.old_value, + "new_value": c.new_value, + "version": c.version, + "timestamp": c.timestamp, + "operator": c.operator, + "reason": c.reason, + "checksum": c.checksum, + } + for c in self.config_changes + ], + "warnings": [ + {"severity": w.severity.value, "code": w.code, "message": w.message, "suggestion": w.suggestion} + for w in self.warnings + ], + "security_warnings": self.security_warnings, + "channel_status_warnings": [ + { + "channel_type": c.channel_type, + "account_id": c.account_id, + "status": c.status.value, + "message": c.message, + "detail": c.detail, + } + for c in self.channel_status_warnings + ], + "repair_plan": [ + { + "id": s.id, + "description": s.description, + "action": s.action, + "reversible": s.reversible, + "params": s.params, + "requires_interactive": s.requires_interactive, + "is_aggressive": s.is_aggressive, + } + for s in self.repair_plan + ], + "repair_results": [ + { + "step_id": r.step_id, + "success": r.success, + "message": r.message, + "config_before": r.config_before, + "config_after": r.config_after, + } + for r in self.repair_results + ], + } + return result + + def _credential_dict(self) -> dict | None: + if not self.credential: + return None + return { + "status": self.credential.status.value, + "token_obtained": self.credential.token_obtained, + "expires_in": self.credential.expires_in, + "error": self.credential.error, + } + + def _permission_dict(self) -> dict | None: + if not self.permission: + return None + return { + "status": self.permission.status.value, + "can_send_message": self.permission.can_send_message, + "can_read_message": self.permission.can_read_message, + "can_manage_group": self.permission.can_manage_group, + "missing_permissions": self.permission.missing_permissions, + "error": self.permission.error, + } + + def _connectivity_dict(self) -> dict | None: + if not self.connectivity: + return None + return { + "status": self.connectivity.status.value, + "latency_ms": self.connectivity.latency_ms, + "endpoint": self.connectivity.endpoint, + "error": self.connectivity.error, + }