AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
from aiohttp import web
|
|
from .utils.response import json_res
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 白名单 (相对于子应用的路径)
|
|
WHITE_LIST = {
|
|
"/api/login",
|
|
"/api/auth/status",
|
|
"/",
|
|
"/static/"
|
|
}
|
|
|
|
async def auth_middleware(app, handler):
|
|
async def mid(req):
|
|
path = req.path
|
|
|
|
# 检查白名单
|
|
if any(path.startswith(w) for w in WHITE_LIST):
|
|
return await handler(req)
|
|
|
|
# 提取 Token
|
|
token = req.cookies.get("panel_token")
|
|
if not token and req.headers.get("Authorization", "").startswith("Bearer "):
|
|
token = req.headers["Authorization"].split(" ", 1)[1]
|
|
|
|
valid, info = False, {}
|
|
|
|
# 验证 Token (简单内存验证，后期可接 Redis/DB)
|
|
session_store = app.get('session_store', {})
|
|
if token and token in session_store:
|
|
valid, info = True, session_store[token]
|
|
|
|
if valid:
|
|
req['user'] = info
|
|
return await handler(req)
|
|
return json_res({"error": "未认证"}, 401)
|
|
return mid
|