AskaEth/SenSu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

security: API Key 权限范围 + 过期时间 + 用量追踪

Browse Source 
增强:
- 权限模板: readonly / monitor / full
- 每 key 独立权限列表, 注入 user['perms']
- 可选过期时间 (ttl 参数)
- 过期 key 自动清理 (加载时 + 验证时)
- request_count 用量追踪
- 标记 is_api_key = True

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
qinglong
2026-06-13 14:27:07 +08:00
parent 9f6eb29820
commit b92b1cc4fc
4 changed files with 70 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/plugins/commands.yaml
+1 -1
View File
@@ -89,6 +89,6 @@ commands:
permissions:
- framework.command.test
source: internal
last_updated: 13895.634814802
last_updated: 13989.462275548
plugin_commands: {}
total_commands: 18
config/services/network_routes.yaml
+1 -1
View File
@@ -1,4 +1,4 @@
http_port: 4200
last_updated: 13895.658168292
last_updated: 13989.479860756
plugin_routes: {}
websocket_port: 4240
services/web_panel/routes/apikeys.py
+65 -16
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""API Key 管理路由 — 创建/列表/删除长期有效的 API 密钥"""
"""API Key 管理路由 — 创建/列表/删除长期有效的 API 密钥，支持权限范围和过期时间"""
import secrets
import time
import logging
@@ -8,8 +8,21 @@ from ..utils.auth import panel_auth
logger = logging.getLogger(__name__)
# 持久化 key: 内存 + DB 双写
# 格式: {key_string: {name, created_at, last_used}}
# 预定义权限模板
PERMISSION_PRESETS = {
"readonly": [
"framework.status.read",
"plugin.info.read",
],
"monitor": [
"framework.status.read",
"plugin.info.read",
"framework.event.subscribe",
],
"full": ["admin"],
}
# 持久化 key: {key_string: {name, created_at, expires_at, permissions, last_used, request_count}}
API_KEYS: dict[str, dict] = {}
@@ -28,10 +41,15 @@ def _load_keys_from_db(app):
import json
raw = db.get_config("api_keys", "{}")
stored = json.loads(raw) if raw else {}
for k, v in stored.items():
now = time.time()
for k, v in list(stored.items()):
# 清理过期 key
if v.get("expires_at") and now > v["expires_at"]:
logger.info(f"🔑 过期 API Key 已清理: {v.get('name', '?')}")
continue
API_KEYS[k] = v
if stored:
logger.info(f"🔑 从数据库恢复了 {len(stored)} 个 API Key")
if API_KEYS:
logger.info(f"🔑 从数据库恢复了 {len(API_KEYS)} 个 API Key")
except Exception as e:
logger.warning(f"API Key 恢复失败: {e}")
@@ -55,11 +73,17 @@ def _save_keys_to_db(app):
def validate_api_key(token: str) -> dict | None:
"""验证 API Key，返回 key_info 或 None"""
if token in API_KEYS:
API_KEYS[token]["last_used"] = time.time()
return API_KEYS[token]
return None
"""验证 API Key — 检查过期，更新使用统计，返回 key_info 或 None"""
if token not in API_KEYS:
return None
info = API_KEYS[token]
# 检查过期
if info.get("expires_at") and time.time() > info["expires_at"]:
del API_KEYS[token]
return None
info["last_used"] = time.time()
info["request_count"] = info.get("request_count", 0) + 1
return info
def setup_routes(app, prefix=""):
@@ -69,16 +93,22 @@ def setup_routes(app, prefix=""):
# ── 列表 ──
async def list_keys(req):
now = time.time()
keys = []
for k, v in API_KEYS.items():
expired = v.get("expires_at") and now > v["expires_at"]
keys.append({
"key": k[:8] + "..." + k[-4:], # 脱敏显示
"full_key": k, # 仅创建时返回完整 key
"key": k[:8] + "..." + k[-4:],
"full_key": k,
"name": v.get("name", ""),
"permissions": v.get("permissions", []),
"created_at": v.get("created_at", 0),
"expires_at": v.get("expires_at", 0),
"last_used": v.get("last_used", 0),
"request_count": v.get("request_count", 0),
"expired": expired,
})
return web.json_response({"keys": keys})
return web.json_response({"keys": keys, "presets": PERMISSION_PRESETS})
# ── 创建 ──
async def create_key(req):
@@ -88,19 +118,38 @@ def setup_routes(app, prefix=""):
if not name:
return web.json_response({"error": "名称不能为空"}, status=400)
key = "sk-" + secrets.token_hex(24) # sk- + 48 hex = 51 chars
# 权限范围: 模板名 或 自定义列表
preset = data.get("preset", "monitor")
if preset in PERMISSION_PRESETS:
permissions = list(PERMISSION_PRESETS[preset])
else:
permissions = data.get("permissions", ["framework.status.read"])
# 过期时间 (秒), 0 = 永不过期
ttl = int(data.get("ttl", 0))
expires_at = (time.time() + ttl) if ttl > 0 else 0
key = "sk-" + secrets.token_hex(24)
API_KEYS[key] = {
"name": name,
"permissions": permissions,
"created_at": time.time(),
"expires_at": expires_at,
"last_used": 0,
"request_count": 0,
}
_save_keys_to_db(req.app)
logger.info(f"🔑 API Key 已创建: {name} ({key[:12]}...)")
logger.info(
f"🔑 API Key 已创建: {name} ({key[:12]}...) "
f"perms={permissions} ttl={ttl}s"
)
return web.json_response({
"ok": True,
"key": key,
"name": name,
"permissions": permissions,
"expires_at": expires_at,
"created_at": API_KEYS[key]["created_at"],
})
except Exception as e:
services/web_panel/utils/auth.py
+3 -2
View File
@@ -23,7 +23,7 @@ def panel_auth(handler, csrf_protect: bool = False):
is_valid = True
request['user'] = session_store[token]
# 2b. 从 API Key Store 验证 (服务器间调用)
# 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)
@@ -31,8 +31,9 @@ def panel_auth(handler, csrf_protect: bool = False):
is_valid = True
request['user'] = {
"username": f"apikey:{key_info['name']}",
"perms": ["admin"],
"perms": key_info.get("permissions", ["framework.status.read"]),
"login_time": key_info.get("created_at", 0),
"is_api_key": True,
}
# 3. 拦截逻辑