diff --git a/.gitignore b/.gitignore index c31ae07d..63db1c17 100644 --- a/.gitignore +++ b/.gitignore @@ -50,6 +50,7 @@ template/compare/model_config_template.toml (临时版)麦麦开始学习.bat src/plugins/utils/statistic.py CLAUDE.md +MaiBot-Dashboard/ # Byte-compiled / optimized / DLL files __pycache__/ diff --git a/src/common/server.py b/src/common/server.py index 87760b89..140d86d1 100644 --- a/src/common/server.py +++ b/src/common/server.py @@ -18,8 +18,8 @@ class Server: # 配置 CORS origins = [ - "http://localhost:3000", # 允许的前端源 - "http://127.0.0.1:3000", + "http://localhost:7999", # 允许的前端源 + "http://127.0.0.1:7999", # 在生产环境中,您应该添加实际的前端域名 ] diff --git a/src/main.py b/src/main.py index 28e9f137..a75d4d26 100644 --- a/src/main.py +++ b/src/main.py @@ -36,6 +36,37 @@ class MainSystem: # 使用消息API替代直接的FastAPI实例 self.app: MessageServer = get_global_api() self.server: Server = get_global_server() + + # 注册 WebUI API 路由 + self._register_webui_routes() + + # 设置 WebUI(开发/生产模式) + self._setup_webui() + + def _register_webui_routes(self): + """注册 WebUI API 路由""" + try: + from src.webui.routes import router as webui_router + self.server.register_router(webui_router) + logger.info("WebUI API 路由已注册") + except Exception as e: + logger.warning(f"注册 WebUI API 路由失败: {e}") + + def _setup_webui(self): + """设置 WebUI(根据环境变量决定模式)""" + import os + webui_enabled = os.getenv("WEBUI_ENABLED", "false").lower() == "true" + if not webui_enabled: + logger.info("WebUI 已禁用") + return + + webui_mode = os.getenv("WEBUI_MODE", "production").lower() + + try: + from src.webui.manager import setup_webui + setup_webui(mode=webui_mode) + except Exception as e: + logger.error(f"设置 WebUI 失败: {e}") async def initialize(self): """初始化系统组件""" diff --git a/src/webui/__init__.py b/src/webui/__init__.py new file mode 100644 index 00000000..713145a3 --- /dev/null +++ b/src/webui/__init__.py @@ -0,0 +1 @@ +"""WebUI 模块""" diff --git a/src/webui/manager.py b/src/webui/manager.py new file mode 100644 index 00000000..f4302c50 --- /dev/null +++ b/src/webui/manager.py @@ -0,0 +1,93 @@ +"""WebUI 管理器 - 处理开发/生产环境的 WebUI 启动""" +import os +from pathlib import Path +from src.common.logger import get_logger +from .token_manager import get_token_manager + +logger = get_logger("webui") + + +def setup_webui(mode: str = "production") -> bool: + """ + 设置 WebUI + + Args: + mode: 运行模式,"development" 或 "production" + + Returns: + bool: 是否成功设置 + """ + # 初始化 Token 管理器(确保 token 文件存在) + token_manager = get_token_manager() + current_token = token_manager.get_token() + logger.info(f"🔑 WebUI Access Token: {current_token}") + logger.info("💡 请使用此 Token 登录 WebUI") + + if mode == "development": + return setup_dev_mode() + else: + return setup_production_mode() + + +def setup_dev_mode() -> bool: + """设置开发模式 - 仅启用 CORS,前端自行启动""" + logger.info("📝 WebUI 开发模式已启用") + logger.info("🌐 请手动启动前端开发服务器: cd webui && npm run dev") + logger.info("💡 前端将运行在 http://localhost:7999") + return True + + +def setup_production_mode() -> bool: + """设置生产模式 - 挂载静态文件""" + try: + from src.common.server import get_global_server + from fastapi.staticfiles import StaticFiles + from fastapi.responses import FileResponse + + server = get_global_server() + base_dir = Path(__file__).parent.parent.parent + static_path = base_dir / "webui" / "dist" + + if not static_path.exists(): + logger.warning(f"❌ WebUI 静态文件目录不存在: {static_path}") + logger.warning("💡 请先构建前端: cd webui && npm run build") + return False + + if not (static_path / "index.html").exists(): + logger.warning(f"❌ 未找到 index.html: {static_path / 'index.html'}") + logger.warning("💡 请确认前端已正确构建") + return False + + # 挂载静态资源 + if (static_path / "assets").exists(): + server.app.mount( + "/assets", + StaticFiles(directory=str(static_path / "assets")), + name="assets" + ) + + # 处理 SPA 路由 + @server.app.get("/{full_path:path}") + async def serve_spa(full_path: str): + """服务单页应用""" + # API 路由不处理 + if full_path.startswith("api/"): + return None + + # 检查文件是否存在 + file_path = static_path / full_path + if file_path.is_file(): + return FileResponse(file_path) + + # 返回 index.html(SPA 路由) + return FileResponse(static_path / "index.html") + + host = os.getenv("HOST", "127.0.0.1") + port = os.getenv("PORT", "8000") + logger.info("✅ WebUI 生产模式已挂载") + logger.info(f"🌐 访问 http://{host}:{port} 查看 WebUI") + return True + + except Exception as e: + logger.error(f"挂载 WebUI 静态文件失败: {e}") + return False diff --git a/src/webui/routes.py b/src/webui/routes.py new file mode 100644 index 00000000..37f82e5d --- /dev/null +++ b/src/webui/routes.py @@ -0,0 +1,154 @@ +"""WebUI API 路由""" +from fastapi import APIRouter, HTTPException, Header +from pydantic import BaseModel, Field +from typing import Optional +from src.common.logger import get_logger +from .token_manager import get_token_manager + +logger = get_logger("webui.api") + +# 创建路由器 +router = APIRouter(prefix="/api/webui", tags=["WebUI"]) + + +class TokenVerifyRequest(BaseModel): + """Token 验证请求""" + token: str = Field(..., description="访问令牌") + + +class TokenVerifyResponse(BaseModel): + """Token 验证响应""" + valid: bool = Field(..., description="Token 是否有效") + message: str = Field(..., description="验证结果消息") + + +class TokenUpdateRequest(BaseModel): + """Token 更新请求""" + new_token: str = Field(..., description="新的访问令牌", min_length=10) + + +class TokenUpdateResponse(BaseModel): + """Token 更新响应""" + success: bool = Field(..., description="是否更新成功") + message: str = Field(..., description="更新结果消息") + + +class TokenRegenerateResponse(BaseModel): + """Token 重新生成响应""" + success: bool = Field(..., description="是否生成成功") + token: str = Field(..., description="新生成的令牌") + message: str = Field(..., description="生成结果消息") + + +@router.get("/health") +async def health_check(): + """健康检查""" + return {"status": "healthy", "service": "MaiBot WebUI"} + + +@router.post("/auth/verify", response_model=TokenVerifyResponse) +async def verify_token(request: TokenVerifyRequest): + """ + 验证访问令牌 + + Args: + request: 包含 token 的验证请求 + + Returns: + 验证结果 + """ + try: + token_manager = get_token_manager() + is_valid = token_manager.verify_token(request.token) + + if is_valid: + return TokenVerifyResponse( + valid=True, + message="Token 验证成功" + ) + else: + return TokenVerifyResponse( + valid=False, + message="Token 无效或已过期" + ) + except Exception as e: + logger.error(f"Token 验证失败: {e}") + raise HTTPException(status_code=500, detail="Token 验证失败") from e + + +@router.post("/auth/update", response_model=TokenUpdateResponse) +async def update_token( + request: TokenUpdateRequest, + authorization: Optional[str] = Header(None) +): + """ + 更新访问令牌(需要当前有效的 token) + + Args: + request: 包含新 token 的更新请求 + authorization: Authorization header (Bearer token) + + Returns: + 更新结果 + """ + try: + # 验证当前 token + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(status_code=401, detail="未提供有效的认证信息") + + current_token = authorization.replace("Bearer ", "") + token_manager = get_token_manager() + + if not token_manager.verify_token(current_token): + raise HTTPException(status_code=401, detail="当前 Token 无效") + + # 更新 token + success, message = token_manager.update_token(request.new_token) + + return TokenUpdateResponse( + success=success, + message=message + ) + except HTTPException: + raise + except Exception as e: + logger.error(f"Token 更新失败: {e}") + raise HTTPException(status_code=500, detail="Token 更新失败") from e + + +@router.post("/auth/regenerate", response_model=TokenRegenerateResponse) +async def regenerate_token(authorization: Optional[str] = Header(None)): + """ + 重新生成访问令牌(需要当前有效的 token) + + Args: + authorization: Authorization header (Bearer token) + + Returns: + 新生成的 token + """ + try: + # 验证当前 token + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(status_code=401, detail="未提供有效的认证信息") + + current_token = authorization.replace("Bearer ", "") + token_manager = get_token_manager() + + if not token_manager.verify_token(current_token): + raise HTTPException(status_code=401, detail="当前 Token 无效") + + # 重新生成 token + new_token = token_manager.regenerate_token() + + return TokenRegenerateResponse( + success=True, + token=new_token, + message="Token 已重新生成" + ) + except HTTPException: + raise + except Exception as e: + logger.error(f"Token 重新生成失败: {e}") + raise HTTPException(status_code=500, detail="Token 重新生成失败") from e + diff --git a/src/webui/token_manager.py b/src/webui/token_manager.py new file mode 100644 index 00000000..2b606b84 --- /dev/null +++ b/src/webui/token_manager.py @@ -0,0 +1,244 @@ +""" +WebUI Token 管理模块 +负责生成、保存、验证和更新访问令牌 +""" + +import json +import secrets +from pathlib import Path +from typing import Optional + +from src.common.logger import get_logger + +logger = get_logger("webui") + + +class TokenManager: + """Token 管理器""" + + def __init__(self, config_path: Optional[Path] = None): + """ + 初始化 Token 管理器 + + Args: + config_path: 配置文件路径,默认为项目根目录的 data/webui.json + """ + if config_path is None: + # 获取项目根目录 (src/webui -> src -> 根目录) + project_root = Path(__file__).parent.parent.parent + config_path = project_root / "data" / "webui.json" + + self.config_path = config_path + self.config_path.parent.mkdir(parents=True, exist_ok=True) + + # 确保配置文件存在并包含有效的 token + self._ensure_config() + + def _ensure_config(self): + """确保配置文件存在且包含有效的 token""" + if not self.config_path.exists(): + logger.info(f"WebUI 配置文件不存在,正在创建: {self.config_path}") + self._create_new_token() + else: + # 验证配置文件格式 + try: + config = self._load_config() + if not config.get("access_token"): + logger.warning("WebUI 配置文件中缺少 access_token,正在重新生成") + self._create_new_token() + else: + logger.info(f"WebUI Token 已加载: {config['access_token'][:8]}...") + except Exception as e: + logger.error(f"读取 WebUI 配置文件失败: {e},正在重新创建") + self._create_new_token() + + def _load_config(self) -> dict: + """加载配置文件""" + try: + with open(self.config_path, "r", encoding="utf-8") as f: + return json.load(f) + except Exception as e: + logger.error(f"加载 WebUI 配置失败: {e}") + return {} + + def _save_config(self, config: dict): + """保存配置文件""" + try: + with open(self.config_path, "w", encoding="utf-8") as f: + json.dump(config, f, ensure_ascii=False, indent=2) + logger.info(f"WebUI 配置已保存到: {self.config_path}") + except Exception as e: + logger.error(f"保存 WebUI 配置失败: {e}") + raise + + def _create_new_token(self) -> str: + """生成新的 64 位随机 token""" + # 生成 64 位十六进制字符串 (32 字节 = 64 hex 字符) + token = secrets.token_hex(32) + + config = { + "access_token": token, + "created_at": self._get_current_timestamp(), + "updated_at": self._get_current_timestamp() + } + + self._save_config(config) + logger.info(f"新的 WebUI Token 已生成: {token[:8]}...") + + return token + + def _get_current_timestamp(self) -> str: + """获取当前时间戳字符串""" + from datetime import datetime + return datetime.now().isoformat() + + def get_token(self) -> str: + """获取当前有效的 token""" + config = self._load_config() + return config.get("access_token", "") + + def verify_token(self, token: str) -> bool: + """ + 验证 token 是否有效 + + Args: + token: 待验证的 token + + Returns: + bool: token 是否有效 + """ + if not token: + return False + + current_token = self.get_token() + if not current_token: + logger.error("系统中没有有效的 token") + return False + + # 使用 secrets.compare_digest 防止时序攻击 + is_valid = secrets.compare_digest(token, current_token) + + if is_valid: + logger.debug("Token 验证成功") + else: + logger.warning("Token 验证失败") + + return is_valid + + def update_token(self, new_token: str) -> tuple[bool, str]: + """ + 更新 token + + Args: + new_token: 新的 token (最少 10 位,必须包含大小写字母和特殊符号) + + Returns: + tuple[bool, str]: (是否更新成功, 错误消息) + """ + # 验证新 token 格式 + is_valid, error_msg = self._validate_custom_token(new_token) + if not is_valid: + logger.error(f"Token 格式无效: {error_msg}") + return False, error_msg + + try: + config = self._load_config() + old_token = config.get("access_token", "")[:8] + + config["access_token"] = new_token + config["updated_at"] = self._get_current_timestamp() + + self._save_config(config) + logger.info(f"Token 已更新: {old_token}... -> {new_token[:8]}...") + + return True, "Token 更新成功" + except Exception as e: + logger.error(f"更新 Token 失败: {e}") + return False, f"更新失败: {str(e)}" + + def regenerate_token(self) -> str: + """ + 重新生成 token + + Returns: + str: 新生成的 token + """ + logger.info("正在重新生成 WebUI Token...") + return self._create_new_token() + + def _validate_token_format(self, token: str) -> bool: + """ + 验证 token 格式是否正确(旧的 64 位十六进制验证,保留用于系统生成的 token) + + Args: + token: 待验证的 token + + Returns: + bool: 格式是否正确 + """ + if not token or not isinstance(token, str): + return False + + # 必须是 64 位十六进制字符串 + if len(token) != 64: + return False + + # 验证是否为有效的十六进制字符串 + try: + int(token, 16) + return True + except ValueError: + return False + + def _validate_custom_token(self, token: str) -> tuple[bool, str]: + """ + 验证自定义 token 格式 + + 要求: + - 最少 10 位 + - 包含大写字母 + - 包含小写字母 + - 包含特殊符号 + + Args: + token: 待验证的 token + + Returns: + tuple[bool, str]: (是否有效, 错误消息) + """ + if not token or not isinstance(token, str): + return False, "Token 不能为空" + + # 检查长度 + if len(token) < 10: + return False, "Token 长度至少为 10 位" + + # 检查是否包含大写字母 + has_upper = any(c.isupper() for c in token) + if not has_upper: + return False, "Token 必须包含大写字母" + + # 检查是否包含小写字母 + has_lower = any(c.islower() for c in token) + if not has_lower: + return False, "Token 必须包含小写字母" + + # 检查是否包含特殊符号 + special_chars = "!@#$%^&*()_+-=[]{}|;:,.<>?/" + has_special = any(c in special_chars for c in token) + if not has_special: + return False, f"Token 必须包含特殊符号 ({special_chars})" + + return True, "Token 格式正确" + + +# 全局单例 +_token_manager_instance: Optional[TokenManager] = None + + +def get_token_manager() -> TokenManager: + """获取 TokenManager 单例""" + global _token_manager_instance + if _token_manager_instance is None: + _token_manager_instance = TokenManager() + return _token_manager_instance diff --git a/template/template.env b/template/template.env index d9b6e2bd..b9d612af 100644 --- a/template/template.env +++ b/template/template.env @@ -1,2 +1,7 @@ HOST=127.0.0.1 -PORT=8000 \ No newline at end of file +PORT=8000 + +# WebUI 配置 +# WEBUI_ENABLED=true +# WEBUI_MODE=development # 开发模式(需手动启动前端: cd webui && npm run dev,端口 7999) +# WEBUI_MODE=production # 生产模式(需先构建前端: cd webui && npm run build) \ No newline at end of file