qinglong
b92b1cc4fc
增强: - 权限模板: readonly / monitor / full - 每 key 独立权限列表, 注入 user['perms'] - 可选过期时间 (ttl 参数) - 过期 key 自动清理 (加载时 + 验证时) - request_count 用量追踪 - 标记 is_api_key = True Co-Authored-By: Claude <noreply@anthropic.com>
57 lines
2.1 KiB
Python
57 lines
2.1 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 验证 (服务器间调用 — 使用 key 自身权限)
|
|
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": key_info.get("permissions", ["framework.status.read"]),
|
|
"login_time": key_info.get("created_at", 0),
|
|
"is_api_key": True,
|
|
}
|
|
|
|
# 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
|