Files
SenSu/services/web_panel/utils/auth.py
T
qinglong 9f6eb29820 feat: API Key 系统 — CRUD + panel_auth 集成 + 持久化
新增:
- services/web_panel/routes/apikeys.py — 创建/列表/删除 API Key
- panel_auth() 双重验证: Session Store → API Key fallback
- _check_plugin_auth() 支持 API Key
- API Key 持久化到 SenSuDB (config_kv 表)
- 格式: sk- + 48 hex chars
- 脱敏显示 (前8后4)
- 删除后立即失效 (401)

WebPanelManager 注册 apikeys 路由

Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-13 14:25:55 +08:00

56 lines
2.0 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import functools
from aiohttp import web
def panel_auth(handler, csrf_protect: bool = False):
"""面板专用鉴权装饰器 — Session Store + API Key 双重验证
csrf_protect=True 时额外检查 X-CSRF-Token 头 (用于写操作)
"""
@functools.wraps(handler)
async def wrapper(request, *args, **kwargs):
# 1. 获取 Token (Cookie / Authorization header)
token = request.cookies.get("panel_token")
if not token and request.headers.get("Authorization", "").startswith("Bearer "):
token = request.headers["Authorization"].split(" ", 1)[1]
is_valid = False
# 2a. 从面板 Session Store 验证 (浏览器登录)
session_store = request.app.get('panel_session_store', {})
if token and token in session_store:
is_valid = True
request['user'] = session_store[token]
# 2b. 从 API Key Store 验证 (服务器间调用)
if not is_valid and token:
from services.web_panel.routes.apikeys import validate_api_key
key_info = validate_api_key(token)
if key_info:
is_valid = True
request['user'] = {
"username": f"apikey:{key_info['name']}",
"perms": ["admin"],
"login_time": key_info.get("created_at", 0),
}
# 3. 拦截逻辑
if not is_valid:
return web.json_response({
"error": "未认证或会话已过期",
"status": 401
}, status=401)
# 4. CSRF 检查 — 仅对 session token (API key 免 CSRF)
if csrf_protect and token in session_store:
csrf = request.headers.get("X-CSRF-Token", "")
if csrf != token:
return web.json_response({
"error": "CSRF 验证失败",
"status": 403,
}, status=403)
return await handler(request, *args, **kwargs)
return wrapper