#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio import uuid import json from typing import Dict, List, Set, Optional from pathlib import Path import yaml logger = logging.getLogger(__name__) class PermissionService: """权限服务""" def __init__(self, config: Dict, tui_service, core_bridge): self.config = config self.tui_service = tui_service self.core_bridge = core_bridge self.permission_rules: Dict = {} self.granted_permissions: Dict[str, Set[str]] = {} self.pending_requests: Dict[str, Dict] = {} self.plugin_status: Dict[str, str] = {} # 插件状态跟踪 self.is_running = False # 配置文件路径 self.config_dir = Path("config") / "permissions" self.granted_file = self.config_dir / "granted_permissions.json" self.pending_file = self.config_dir / "pending_requests.json" self.plugin_status_file = self.config_dir / "plugin_status.json" # 确保配置目录存在 self.config_dir.mkdir(parents=True, exist_ok=True) logger.debug("PermissionService初始化开始") async def start(self): """启动权限服务""" try: logger.info("启动权限服务") # 检查核心桥接服务是否可用 if not self.core_bridge: logger.error("核心桥接服务不可用") return False # 加载权限规则 await self._load_permission_rules() # 加载持久化数据 await self._load_persisted_data() # 订阅权限相关事件 self.core_bridge.subscribe("permission.request", self._handle_permission_request) self.core_bridge.subscribe("permission.grant", self._handle_permission_grant) self.core_bridge.subscribe("permission.deny", self._handle_permission_deny) self.core_bridge.subscribe("permission.ignore", self._handle_permission_ignore) # 检查TUI服务连接状态 if self.tui_service: logger.debug("TUI服务已连接") else: logger.warning("TUI服务未连接，权限请求将显示在控制台") self.is_running = True logger.info("权限服务启动完成") return True except Exception as e: logger.error(f"启动权限服务时出错: {str(e)}", exc_info=True) return False async def _load_permission_rules(self): """加载权限规则""" try: rules_path = Path("config") / "permissions" / "permission_rules.yaml" if rules_path.exists(): with open(rules_path, 'r', encoding='utf-8') as f: self.permission_rules = yaml.safe_load(f) logger.debug(f"加载权限规则: {len(self.permission_rules.get('rules', {}))} 条规则") else: logger.warning("权限规则文件不存在，使用默认规则") self.permission_rules = { "rules": {}, "default_policy": "ask" } except Exception as e: logger.error(f"加载权限规则时出错: {str(e)}", exc_info=True) self.permission_rules = { "rules": {}, "default_policy": "ask" } async def _load_persisted_data(self): """加载持久化的权限数据""" try: # 如果配置文件不存在，创建空的配置文件 if not self.granted_file.exists(): await self._save_granted_permissions() logger.info("创建空的已授予权限文件") if not self.pending_file.exists(): await self._save_pending_requests() logger.info("创建空的待处理请求文件") if not self.plugin_status_file.exists(): await self._save_plugin_status() logger.info("创建空的插件状态文件") # 加载已授予权限 if self.granted_file.exists(): with open(self.granted_file, 'r', encoding='utf-8') as f: granted_data = json.load(f) # 将列表转换回集合 for plugin, permissions in granted_data.items(): self.granted_permissions[plugin] = set(permissions) logger.debug(f"加载已授予权限: {len(self.granted_permissions)} 个插件") # 加载待处理请求 if self.pending_file.exists(): with open(self.pending_file, 'r', encoding='utf-8') as f: self.pending_requests = json.load(f) logger.debug(f"加载待处理请求: {len(self.pending_requests)} 个") # 加载插件状态 if self.plugin_status_file.exists(): with open(self.plugin_status_file, 'r', encoding='utf-8') as f: self.plugin_status = json.load(f) logger.debug(f"加载插件状态: {len(self.plugin_status)} 个插件") logger.info("权限持久化数据加载完成") except Exception as e: logger.error(f"加载持久化权限数据时出错: {str(e)}", exc_info=True) async def _save_granted_permissions(self): """保存已授予权限到文件""" try: # 将集合转换为列表以便JSON序列化 granted_data = {} for plugin, permissions in self.granted_permissions.items(): granted_data[plugin] = list(permissions) with open(self.granted_file, 'w', encoding='utf-8') as f: json.dump(granted_data, f, ensure_ascii=False, indent=2) logger.debug(f"已授予权限已保存: {len(granted_data)} 个插件") except Exception as e: logger.error(f"保存已授予权限时出错: {str(e)}", exc_info=True) async def _save_pending_requests(self): """保存待处理请求到文件""" try: with open(self.pending_file, 'w', encoding='utf-8') as f: json.dump(self.pending_requests, f, ensure_ascii=False, indent=2) logger.debug(f"待处理请求已保存: {len(self.pending_requests)} 个") except Exception as e: logger.error(f"保存待处理请求时出错: {str(e)}", exc_info=True) async def _save_plugin_status(self): """保存插件状态到文件""" try: with open(self.plugin_status_file, 'w', encoding='utf-8') as f: json.dump(self.plugin_status, f, ensure_ascii=False, indent=2) logger.debug(f"插件状态已保存: {len(self.plugin_status)} 个插件") except Exception as e: logger.error(f"保存插件状态时出错: {str(e)}", exc_info=True) async def _save_all_data(self): """保存所有权限数据""" try: await asyncio.gather( self._save_granted_permissions(), self._save_pending_requests(), self._save_plugin_status() ) logger.debug("所有权限数据已保存") except Exception as e: logger.error(f"保存权限数据时出错: {str(e)}", exc_info=True) def _handle_permission_request(self, message: Dict): """处理权限请求""" try: # 从 message 的 data 字段中获取插件名称 data = message.get('data', {}) plugin_name = data.get('plugin_name') logger.debug(f"传递的消息原文：{message}") logger.debug(f"传递的插件名：{plugin_name}") requested_permissions = data.get('permissions', []) request_id = str(uuid.uuid4())[:8] # 简短的请求ID # 加强插件名称验证 if not plugin_name or plugin_name == 'None' or plugin_name.strip() == '': logger.error(f"无效的插件名称: {repr(plugin_name)}") logger.debug(f"完整权限请求消息: {message}") return # 验证插件名称格式 if not self._is_valid_plugin_name(plugin_name): logger.error(f"插件名称格式无效: {plugin_name}") return # 验证权限列表 if not requested_permissions or not isinstance(requested_permissions, list): logger.warning(f"插件 {plugin_name} 请求的权限列表为空或格式错误") requested_permissions = [] # 确保是列表 # 存储待处理请求 self.pending_requests[request_id] = { 'plugin_name': plugin_name, 'permissions': requested_permissions, 'timestamp': asyncio.get_event_loop().time() } # 更新插件状态 self.plugin_status[plugin_name] = "pending" logger.debug(f"处理权限请求: {plugin_name} -> {len(requested_permissions)} 个权限") # 保存数据 asyncio.create_task(self._save_pending_requests()) asyncio.create_task(self._save_plugin_status()) # 显示用户友好的权限请求界面 asyncio.create_task(self._delayed_permission_ui(request_id, plugin_name, requested_permissions)) except Exception as e: logger.error(f"处理权限请求时出错: {str(e)}", exc_info=True) def _is_valid_plugin_name(self, plugin_name: str) -> bool: """验证插件名称是否有效""" try: if not plugin_name or not isinstance(plugin_name, str): return False # 基本格式检查 if plugin_name.strip() == '': return False # 检查常见无效值 invalid_values = ['None', 'null', 'undefined', ''] if plugin_name in invalid_values: return False # 检查长度限制 if len(plugin_name) > 100: return False # 检查字符有效性（允许字母、数字、下划线、点、连字符） import re if not re.match(r'^[a-zA-Z0-9_\.\-]+$', plugin_name): return False return True except Exception as e: logger.error(f"验证插件名称时出错: {str(e)}") return False async def _delayed_permission_ui(self, request_id: str, plugin_name: str, permissions: List[str]): """延迟显示权限请求UI，等待TUI就绪""" try: # 等待TUI服务就绪 tui_ready = await self.wait_for_tui_ready() if tui_ready: logger.debug(f"TUI已就绪，显示权限请求: {plugin_name}") await self._show_permission_request_ui(request_id, plugin_name, permissions) else: # TUI未就绪，使用回退显示 logger.warning(f"TUI未就绪，使用控制台显示权限请求: {plugin_name}") message = f"🔐 插件 {plugin_name} 请求 {len(permissions)} 个权限 (请求ID: {request_id})" self._fallback_permission_display(request_id, plugin_name, message) except Exception as e: logger.error(f"延迟显示权限UI时出错: {str(e)}", exc_info=True) async def _show_permission_request_ui(self, request_id: str, plugin_name: str, permissions: List[str]): """显示权限请求用户界面 - 优化显示""" try: # 创建更友好的权限描述 permission_descriptions = { "plugin.example.read": "📖 读取示例插件数据", "plugin.example.write": "✏️ 写入示例插件数据", "plugin.example.execute": "⚡ 执行示例插件操作", "framework.event.subscribe": "📡 订阅框架事件", "framework.command.execute": "⌨️ 执行框架命令" } # 构建权限列表显示 permission_list = [] for perm in permissions: desc = permission_descriptions.get(perm, f"🔧 {perm}") permission_list.append(f" ✅ {desc}") permission_display = "\n".join(permission_list) if permission_list else " 无具体权限请求" # 显示权限请求界面 - 使用更简洁的格式 messages = [ f"🔐 **插件权限请求**", f"", f"**插件**: {plugin_name}", f"**请求权限**:", permission_display, f"", f"**操作选项**:", f" 🟢 pmallow {request_id} - 同意所有权限", f" 🟡 pmallow {request_id} read,write - 仅同意部分权限", f" 🔴 pmdeny {request_id} - 拒绝所有权限", f" ⏸️ ignore {request_id} - 暂时忽略", f"", f"**快捷命令**:", f" pmallow all - 同意所有待处理请求", f" pmdeny all - 拒绝所有待处理请求" f" pmignore all - 忽略所有待处理请求" ] # 逐行发送消息，确保每行都能正确显示 if self.tui_service and hasattr(self.tui_service, 'show_message'): for line in messages: if line.strip(): # 忽略空行 self.tui_service.show_message(line, "warning", persistent=True) await asyncio.sleep(0.1) # 小延迟确保消息顺序 else: # TUI服务不可用，使用控制台输出 self._fallback_permission_display(request_id, plugin_name, message) except Exception as e: logger.error(f"显示权限请求界面时出错: {str(e)}", exc_info=True) def _fallback_permission_display(self, request_id: str, plugin_name: str, message: str): """回退到控制台显示权限请求""" try: print("\n" + "="*60) print(message) print("="*60) print("🐱 请输入命令处理权限请求:") logger.info(f"权限请求已显示在控制台: {plugin_name} -> {request_id}") except Exception as e: logger.error(f"回退显示权限请求时出错: {str(e)}") def _handle_permission_grant(self, message: Dict): """处理权限授予""" try: request_id = message.get('request_id') granted_permissions = message.get('permissions', []) if request_id in self.pending_requests: request = self.pending_requests[request_id] plugin_name = request['plugin_name'] # 如果未指定具体权限，授予所有请求的权限 if not granted_permissions: granted_permissions = request['permissions'] # 授予权限 asyncio.create_task(self.grant_permissions(plugin_name, granted_permissions)) # 更新插件状态 self.plugin_status[plugin_name] = "granted" # 移除待处理请求 del self.pending_requests[request_id] # 显示成功消息 success_msg = f"✅ 已为插件 '{plugin_name}' 授予 {len(granted_permissions)} 个权限" if self.tui_service and hasattr(self.tui_service, 'show_message'): self.tui_service.show_message(success_msg, "success") else: print(f"🐱 {success_msg}") logger.info(f"权限授予完成: {plugin_name} -> {granted_permissions}") except Exception as e: logger.error(f"处理权限授予时出错: {str(e)}", exc_info=True) def _handle_permission_deny(self, message: Dict): """处理权限拒绝""" try: request_id = message.get('request_id') if request_id in self.pending_requests: request = self.pending_requests[request_id] plugin_name = request['plugin_name'] # 更新插件状态 self.plugin_status[plugin_name] = "denied" # 移除待处理请求 del self.pending_requests[request_id] # 保存数据 asyncio.create_task(self._save_pending_requests()) asyncio.create_task(self._save_plugin_status()) # 显示拒绝消息 deny_msg = f"❌ 已拒绝插件 '{plugin_name}' 的权限请求" if self.tui_service and hasattr(self.tui_service, 'show_message'): self.tui_service.show_message(deny_msg, "error") else: print(f"🐱 {deny_msg}") logger.info(f"权限拒绝完成: {plugin_name}") except Exception as e: logger.error(f"处理权限拒绝时出错: {str(e)}", exc_info=True) def _handle_permission_ignore(self, message: Dict): """处理权限忽略""" try: request_id = message.get('request_id') if request_id in self.pending_requests: request = self.pending_requests[request_id] plugin_name = request['plugin_name'] # 更新插件状态 self.plugin_status[plugin_name] = "ignored" # 移除待处理请求 del self.pending_requests[request_id] # 保存数据 asyncio.create_task(self._save_pending_requests()) asyncio.create_task(self._save_plugin_status()) # 显示忽略消息 ignore_msg = f"⏸️ 已暂时忽略插件 '{plugin_name}' 的权限请求" if self.tui_service and hasattr(self.tui_service, 'show_message'): self.tui_service.show_message(ignore_msg, "info") else: print(f"🐱 {ignore_msg}") logger.info(f"权限请求被忽略: {plugin_name}") except Exception as e: logger.error(f"处理权限忽略时出错: {str(e)}", exc_info=True) async def grant_permissions(self, plugin_name: str, permissions: List[str]): """授予权限""" try: if plugin_name not in self.granted_permissions: self.granted_permissions[plugin_name] = set() for permission in permissions: self.granted_permissions[plugin_name].add(permission) # 更新插件状态 self.plugin_status[plugin_name] = "granted" logger.debug(f"授予权限: {plugin_name} -> {permissions}") # 保存数据 await asyncio.gather( self._save_granted_permissions(), self._save_plugin_status() ) # 通知插件权限已授予 await self.core_bridge.publish("permission.granted", { 'plugin_name': plugin_name, 'permissions': permissions }) except Exception as e: logger.error(f"授予权限时出错: {str(e)}", exc_info=True) raise def has_permission(self, plugin_name: str, permission: str) -> bool: """检查是否具有权限""" try: # 检查显式授予的权限 if plugin_name in self.granted_permissions: if permission in self.granted_permissions[plugin_name]: return True # 检查权限规则 rule_key = f"{plugin_name}.{permission}" if rule_key in self.permission_rules.get('rules', {}): return self.permission_rules['rules'][rule_key] == 'allow' # 默认策略 default_policy = self.permission_rules.get('default_policy', 'ask') return default_policy == 'allow' except Exception as e: logger.error(f"检查权限时出错: {str(e)}", exc_info=True) return False async def request_permissions(self, plugin_name: str, permissions: List[str]) -> bool: """请求权限 - 非阻塞版本""" try: logger.debug(f"权限请求: {plugin_name} -> {permissions}") # 首先检查是否已经有所有权限 if all(self.has_permission(plugin_name, perm) for perm in permissions): logger.debug(f"插件 {plugin_name} 已有所有请求的权限") return True # 标记插件为等待权限状态 self.plugin_status[plugin_name] = "pending" # 发布权限请求事件（非阻塞） await self.core_bridge.publish("permission.request", { 'plugin_name': plugin_name, 'permissions': permissions }) # 立即返回，不等待用户响应 # 插件将在权限被授予后通过事件机制得到通知 logger.debug(f"权限请求已发送，等待用户响应: {plugin_name}") return True # 立即返回True，让插件继续加载 except Exception as e: logger.error(f"请求权限时出错: {str(e)}", exc_info=True) return True # 出错时也返回True，避免阻塞插件加载 def get_pending_requests(self) -> Dict[str, Dict]: """获取待处理请求""" return self.pending_requests.copy() async def process_permission_command(self, command: str, args: List[str]) -> str: """处理权限相关命令""" try: if command == "pmallow": if not args: return "❌ 请指定请求ID，如: pmallow abc123" request_id = args[0] if request_id == "all": # 同意所有待处理请求 count = len(self.pending_requests) for rid in list(self.pending_requests.keys()): self._handle_permission_grant({'request_id': rid}) return f"✅ 已同意所有 {count} 个待处理权限请求" # 检查特定权限 specific_permissions = [] if len(args) > 1: specific_permissions = args[1].split(',') self._handle_permission_grant({ 'request_id': request_id, 'permissions': specific_permissions }) return f"✅ 已处理权限请求 {request_id}" elif command == "pmdeny": if not args: return "❌ 请指定请求ID，如: pmdeny abc123" request_id = args[0] if request_id == "all": # 拒绝所有待处理请求 count = len(self.pending_requests) for rid in list(self.pending_requests.keys()): await self._handle_permission_deny({'request_id': rid}) return f"❌ 已拒绝所有 {count} 个待处理权限请求" await self._handle_permission_deny({'request_id': request_id}) return f"❌ 已拒绝权限请求 {request_id}" elif command == "pmignore": if not args: return "❌ 请指定请求ID，如: pmignore abc123" request_id = args[0] if request_id == "all": # 忽略所有待处理请求 count = len(self.pending_requests) for rid in list(self.pending_requests.keys()): await self._handle_permission_ignore({'request_id': rid}) return f"⏸️ 已忽略所有 {count} 个待处理权限请求" await self._handle_permission_ignore({'request_id': request_id}) return f"⏸️ 已忽略权限请求 {request_id}" elif command == "permissions": # 显示当前权限状态 return await self._show_permission_status() elif command == "pm_plugin_status": # 显示插件状态 return await self._show_plugin_status() elif command == "pmpending" or command == "pmrequests": # 查询待授权权限请求列表 return await self._show_pending_requests() elif command == "pmhelp": # 显示权限命令帮助 return self._show_permission_help() elif command == "pmfix": # 修复权限状态 return await self._fix_permission_status() elif command == "pmclean": # 清理权限数据 return await self._clean_permission_data(args) elif command == "pmbackup": # 备份权限数据 return await self._backup_permission_data() elif command == "pmtest": # 测试权限配置文件 return await self._test_permission_config() else: return f"❌ 未知权限命令: {command}\n💡 输入 'pmhelp' 查看可用命令" except Exception as e: logger.error(f"处理权限命令时出错: {str(e)}", exc_info=True) return f"❌ 处理命令时出错: {str(e)}" async def _test_permission_config(self) -> str: """测试权限配置文件""" try: logger.debug("开始权限配置文件测试") result = ["🔧 **权限配置文件测试**"] result.append("=" * 50) # 测试配置文件路径 result.append("📁 **配置文件路径**:") result.append(f" 配置目录: {self.config_dir}") result.append(f" 已授予权限: {self.granted_file}") result.append(f" 待处理请求: {self.pending_file}") result.append(f" 插件状态: {self.plugin_status_file}") # 测试文件存在性 result.append("\n✅ **文件存在性检查**:") config_files = [ ("配置目录", self.config_dir, self.config_dir.exists()), ("已授予权限", self.granted_file, self.granted_file.exists()), ("待处理请求", self.pending_file, self.pending_file.exists()), ("插件状态", self.plugin_status_file, self.plugin_status_file.exists()) ] for name, path, exists in config_files: status = "✅ 存在" if exists else "❌ 不存在" result.append(f" {name}: {status}") # 测试写入权限 result.append("\n✏️ **写入权限测试**:") try: test_data = {"test": "test_data", "timestamp": asyncio.get_event_loop().time()} with open(self.config_dir / "test_write.json", 'w', encoding='utf-8') as f: json.dump(test_data, f, ensure_ascii=False, indent=2) # 读取测试 with open(self.config_dir / "test_write.json", 'r', encoding='utf-8') as f: read_data = json.load(f) # 清理测试文件 (self.config_dir / "test_write.json").unlink(missing_ok=True) result.append(" ✅ 读写测试: 成功") except Exception as e: result.append(f" ❌ 读写测试: 失败 - {str(e)}") # 显示当前数据状态 result.append("\n📊 **当前数据状态**:") result.append(f" 已授予权限: {len(self.granted_permissions)} 个插件") result.append(f" 待处理请求: {len(self.pending_requests)} 个") result.append(f" 插件状态: {len(self.plugin_status)} 个") return "\n".join(result) except Exception as e: logger.error(f"测试权限配置时出错: {str(e)}", exc_info=True) return f"❌ 测试权限配置时出错: {str(e)}" async def _clean_permission_data(self, args: List[str]) -> str: """清理权限数据""" try: if not args: return "❌ 请指定清理类型\n💡 可用选项: expired, all, plugin <插件名>" clean_type = args[0].lower() result = [] if clean_type == "expired": # 清理过期请求（超过24小时） current_time = asyncio.get_event_loop().time() expired_count = 0 for request_id, request in list(self.pending_requests.items()): if current_time - request.get('timestamp', 0) > 86400: # 24小时 plugin_name = request['plugin_name'] del self.pending_requests[request_id] expired_count += 1 result.append(f"🗑️ 清理过期请求: {request_id} ({plugin_name})") if expired_count > 0: await self._save_pending_requests() result.insert(0, f"✅ 已清理 {expired_count} 个过期权限请求") else: result.append("✅ 没有发现过期权限请求") elif clean_type == "all": # 清理所有数据 pending_count = len(self.pending_requests) granted_count = len(self.granted_permissions) status_count = len(self.plugin_status) self.pending_requests.clear() self.granted_permissions.clear() self.plugin_status.clear() await self._save_all_data() result = [ f"✅ 已清理所有权限数据:", f" 🗑️ 待处理请求: {pending_count} 个", f" 🗑️ 已授予权限: {granted_count} 个插件", f" 🗑️ 插件状态: {status_count} 个" ] elif clean_type == "plugin" and len(args) > 1: # 清理特定插件的数据 plugin_name = args[1] cleaned_items = [] # 清理待处理请求 for request_id, request in list(self.pending_requests.items()): if request['plugin_name'] == plugin_name: del self.pending_requests[request_id] cleaned_items.append(f"待处理请求: {request_id}") # 清理已授予权限 if plugin_name in self.granted_permissions: del self.granted_permissions[plugin_name] cleaned_items.append("已授予权限") # 清理插件状态 if plugin_name in self.plugin_status: del self.plugin_status[plugin_name] cleaned_items.append("插件状态") if cleaned_items: await self._save_all_data() result = [f"✅ 已清理插件 '{plugin_name}' 的权限数据:"] + cleaned_items else: result = [f"ℹ️ 未找到插件 '{plugin_name}' 的权限数据"] else: return "❌ 无效的清理类型\n💡 可用选项: expired, all, plugin <插件名>" return "\n".join(result) except Exception as e: logger.error(f"清理权限数据时出错: {str(e)}", exc_info=True) return f"❌ 清理权限数据时出错: {str(e)}" async def _show_permission_status(self) -> str: """显示当前权限状态""" try: if not self.pending_requests and not self.granted_permissions: return "📋 暂无权限请求和授予记录" result = ["📋 **权限状态**"] if self.pending_requests: result.append("\n🟡 **待处理请求**:") for rid, req in self.pending_requests.items(): result.append(f" {rid}: {req['plugin_name']} -> {len(req['permissions'])} 个权限") if self.granted_permissions: result.append("\n🟢 **已授予权限**:") for plugin, perms in self.granted_permissions.items(): result.append(f" {plugin}: {len(perms)} 个权限") if self.plugin_status: result.append("\n🔵 **插件状态**:") for plugin, status in self.plugin_status.items(): status_icon = { "granted": "✅", "pending": "🟡", "denied": "❌", "ignored": "⏸️", "error": "⚠️" }.get(status, "🔵") result.append(f" {status_icon} {plugin}: {status}") # 添加配置文件状态 result.append("\n📁 **配置文件状态**:") config_files = [ ("已授予权限", self.granted_file), ("待处理请求", self.pending_file), ("插件状态", self.plugin_status_file) ] for name, file_path in config_files: if file_path.exists(): result.append(f" ✅ {name}: 存在") else: result.append(f" ❌ {name}: 不存在") return "\n".join(result) except Exception as e: logger.error(f"显示权限状态时出错: {str(e)}", exc_info=True) return f"❌ 显示权限状态时出错: {str(e)}" async def _show_plugin_status(self) -> str: """显示插件状态""" try: if not self.plugin_status: return "📊 暂无插件状态信息" result = ["📊 **插件状态**"] for plugin, status in self.plugin_status.items(): if status == "granted": result.append(f" ✅ {plugin}: 权限已授予") elif status == "pending": result.append(f" 🟡 {plugin}: 等待权限授予") elif status == "denied": result.append(f" ❌ {plugin}: 权限被拒绝") elif status == "ignored": result.append(f" ⏸️ {plugin}: 权限请求被忽略") elif status == "error": result.append(f" ⚠️ {plugin}: 权限错误") else: result.append(f" 🔵 {plugin}: {status}") return "\n".join(result) except Exception as e: logger.error(f"显示插件状态时出错: {str(e)}", exc_info=True) return f"❌ 显示插件状态时出错: {str(e)}" async def _fix_permission_status(self) -> str: """修复权限状态不一致问题""" try: fixes_applied = [] # 检查插件状态与待处理请求的一致性 for plugin, status in list(self.plugin_status.items()): # 如果插件状态是pending但没有对应的待处理请求 if status == "pending": has_pending_request = False for request in self.pending_requests.values(): if request['plugin_name'] == plugin: has_pending_request = True break if not has_pending_request: # 修复：将状态改为error self.plugin_status[plugin] = "error" fixes_applied.append(f"🟡 {plugin}: pending → error (无权限请求)") # 清理过期的待处理请求 current_time = asyncio.get_event_loop().time() expired_requests = [] for request_id, request in list(self.pending_requests.items()): # 假设请求超过1小时为过期 if current_time - request.get('timestamp', 0) > 3600: expired_requests.append(request_id) for request_id in expired_requests: plugin_name = self.pending_requests[request_id]['plugin_name'] del self.pending_requests[request_id] fixes_applied.append(f"🗑️ 清理过期请求: {request_id} ({plugin_name})") # 保存修复后的数据 if fixes_applied: await self._save_all_data() result = ["🔧 **权限状态修复完成**"] result.extend(fixes_applied) result.append(f"\n✅ 共应用 {len(fixes_applied)} 个修复") else: result = ["✅ **权限状态正常**", "未发现需要修复的问题"] return "\n".join(result) except Exception as e: logger.error(f"修复权限状态时出错: {str(e)}", exc_info=True) return f"❌ 修复权限状态时出错: {str(e)}" async def _backup_permission_data(self) -> str: """备份权限数据""" try: backup_dir = self.config_dir / "backups" backup_dir.mkdir(exist_ok=True) import datetime timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") # 备份文件路径 granted_backup = backup_dir / f"granted_permissions_{timestamp}.json" pending_backup = backup_dir / f"pending_requests_{timestamp}.json" status_backup = backup_dir / f"plugin_status_{timestamp}.json" # 复制文件 import shutil if self.granted_file.exists(): shutil.copy2(self.granted_file, granted_backup) if self.pending_file.exists(): shutil.copy2(self.pending_file, pending_backup) if self.plugin_status_file.exists(): shutil.copy2(self.plugin_status_file, status_backup) return f"✅ 权限数据备份完成\n📁 备份位置: {backup_dir}\n⏰ 时间戳: {timestamp}" except Exception as e: logger.error(f"备份权限数据时出错: {str(e)}", exc_info=True) return f"❌ 备份权限数据时出错: {str(e)}" async def _show_pending_requests(self) -> str: """显示待授权权限请求列表 - 更新为pm前缀""" try: if not self.pending_requests: return "📭 暂无待处理的权限请求" result = ["🟡 **待授权权限请求列表**"] result.append("=" * 50) for request_id, request in self.pending_requests.items(): plugin_name = request['plugin_name'] permissions = request['permissions'] # 创建友好的权限描述 permission_descriptions = { "plugin.example.read": "📖 读取示例插件数据", "plugin.example.write": "✏️ 写入示例插件数据", "plugin.example.execute": "⚡ 执行示例插件操作", "framework.event.subscribe": "📡 订阅框架事件", "framework.command.execute": "⌨️ 执行框架命令" } # 构建权限列表 permission_list = [] for perm in permissions: desc = permission_descriptions.get(perm, f"🔧 {perm}") permission_list.append(f" • {desc}") permission_display = "\n".join(permission_list) # 添加请求信息 result.append(f"\n📦 **插件**: {plugin_name}") result.append(f"🆔 **请求ID**: {request_id}") result.append(f"🔐 **请求权限** ({len(permissions)} 个):") result.append(permission_display) # 添加交互指令 - 更新为pm前缀 result.append(f"\n💡 **交互指令**:") result.append(f" 🟢 同意所有权限: pmallow {request_id}") result.append(f" 🟡 同意部分权限: pmallow {request_id} read,write") result.append(f" 🔴 拒绝所有权限: pmdeny {request_id}") result.append(f" ⏸️ 暂时忽略: pmignore {request_id}") result.append("-" * 50) # 添加快捷指令 - 更新为pm前缀 result.append("\n🚀 **快捷指令**:") result.append(" 🟢 同意所有请求: pmallow pmall") result.append(" 🔴 拒绝所有请求: pmdeny all") result.append(" ⏸️ 忽略所有请求: pmignore all") result.append(" 📋 查看权限状态: permissions") result.append(" 📊 查看插件状态: pm_plugin_status") result.append(" ❓ 查看帮助: pmhelp") return "\n".join(result) except Exception as e: logger.error(f"显示待处理请求时出错: {str(e)}", exc_info=True) return f"❌ 显示待处理请求时出错: {str(e)}" def _show_permission_help(self) -> str: """显示权限命令帮助 - 更新为pm前缀""" help_text = """ 🔐 **权限管理命令帮助 (pm前缀)** 📋 **查询命令**: permissions - 查看权限状态 pmpending 或 pmrequests - 查看待授权请求列表 pm_plugin_status - 查看插件权限状态 🛠️ **操作命令**: pmallow <请求ID> - 同意指定请求的所有权限 pmallow <请求ID> <权限列表> - 同意指定请求的部分权限 pmdeny <请求ID> - 拒绝指定请求的所有权限 。pmignore <请求ID> - 暂时忽略指定请求 🚀 **快捷命令**: pmallow pmall - 同意所有待处理请求 pmdeny all - 拒绝所有待处理请求 pmignore all - 忽略所有待处理请求 🔧 **维护命令**: pmfix - 修复权限状态不一致问题 pmtest - 测试权限配置文件 📖 **示例**: pmallow abc123 - 同意请求ID为abc123的所有权限 pmallow abc123 read,write - 只同意abc123的读取和写入权限 pmdeny abc123 - 拒绝abc123的所有权限 pmignore abc123 - 暂时忽略abc123的请求 pmpending - 查看所有待处理的权限请求 💡 **提示**: • 权限请求ID是自动生成的8位字符串 • 使用 pmpending 命令查看所有待处理请求及其ID • 插件在获得权限前可能以受限模式运行 • 权限管理命令都以 `pm` 为前缀，避免与其他命令冲突 """ return help_text.strip() def is_tui_ready(self) -> bool: """检查TUI服务是否就绪""" try: return (self.tui_service is not None and hasattr(self.tui_service, 'show_message') and hasattr(self.tui_service, 'tui_app') and self.tui_service.tui_app is not None) except Exception as e: logger.debug(f"检查TUI状态时出错: {e}") return False async def wait_for_tui_ready(self, timeout: float = 10.0) -> bool: """等待TUI服务就绪""" try: start_time = asyncio.get_event_loop().time() while asyncio.get_event_loop().time() - start_time < timeout: if self.is_tui_ready(): logger.debug("TUI服务已就绪") return True await asyncio.sleep(0.5) logger.warning(f"等待TUI服务就绪超时 ({timeout}秒)") return False except Exception as e: logger.error(f"等待TUI就绪时出错: {e}") return False def shutdown(self): """关闭权限服务""" try: logger.info("关闭权限服务") self.is_running = False self.pending_requests.clear() logger.debug("权限服务关闭完成") except Exception as e: logger.error(f"关闭权限服务时出错: {str(e)}", exc_info=True)