#!/usr/bin/env python3 # -*- coding: utf-8 -*- import secrets import logging from aiohttp import web from ..utils.auth import panel_auth logger = logging.getLogger(__name__) # 全局 Session 存储 (内存型) # 格式: { "token_string": { "username": "...", "perms": [...] } } PANEL_SESSION_STORE = {} def setup_routes(app, prefix=''): """注册面板认证路由""" # 🟢 关键:将 Session Store 挂载到 app,供拦截器读取 app['panel_session_store'] = PANEL_SESSION_STORE # 路由注册 app.router.add_post(f'{prefix}/api/login', handle_login) # 退出和状态检查都需要拦截 app.router.add_post(f'{prefix}/api/logout', panel_auth(handle_logout)) app.router.add_get(f'{prefix}/api/auth/status', panel_auth(handle_auth_status)) async def handle_login(req): """处理面板登录""" try: data = await req.json() username = data.get('username') password = data.get('password') cfg = req.app.get('panel_config', {}) cfg_user = cfg.get('username', 'admin') cfg_pass = cfg.get('password', 'admin') # 校验配置中的账号密码 if username == cfg_user and password == cfg_pass: # 登录成功:生成 Token token = secrets.token_hex(16) # 写入 Session Store user_info = { "username": username, "perms": ["admin"], "login_time": __import__('time').time() } PANEL_SESSION_STORE[token] = user_info logger.info(f"✅ 面板登录成功: {username} (Session: {token[:4]}...)") resp = web.json_response({"success": True, "username": username}) # 设置 Cookie resp.set_cookie("panel_token", token, max_age=259200, httponly=True, samesite="Lax") return resp else: logger.warning(f"❌ 面板登录失败: 用户 {username} 密码错误") return web.json_response({"success": False, "msg": "用户名或密码错误"}, status=401) except Exception as e: logger.error(f"登录异常: {e}") return web.json_response({"error": str(e)}, status=500) async def handle_logout(req): """处理退出登录""" token = req.cookies.get("panel_token") if token and token in PANEL_SESSION_STORE: del PANEL_SESSION_STORE[token] logger.info(f"👋 用户退出登录") resp = web.json_response({"success": True}) resp.del_cookie("panel_token") return resp async def handle_auth_status(req): """获取当前认证状态 (被 panel_auth 拦截,能进来说明已认证)""" user = req.get('user', {}) return web.json_response({ "authenticated": True, "username": user.get("username", "Unknown"), "perms": user.get("perms", []) })