diff --git a/config/plugins/commands.yaml b/config/plugins/commands.yaml index 6f1eff5..c586a70 100644 --- a/config/plugins/commands.yaml +++ b/config/plugins/commands.yaml @@ -89,6 +89,6 @@ commands: permissions: - framework.command.test source: internal -last_updated: 13895.634814802 +last_updated: 13989.462275548 plugin_commands: {} total_commands: 18 diff --git a/config/services/network_routes.yaml b/config/services/network_routes.yaml index 7149116..22682ca 100644 --- a/config/services/network_routes.yaml +++ b/config/services/network_routes.yaml @@ -1,4 +1,4 @@ http_port: 4200 -last_updated: 13895.658168292 +last_updated: 13989.479860756 plugin_routes: {} websocket_port: 4240 diff --git a/services/web_panel/routes/apikeys.py b/services/web_panel/routes/apikeys.py index 8ed0286..09cff6f 100644 --- a/services/web_panel/routes/apikeys.py +++ b/services/web_panel/routes/apikeys.py @@ -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: diff --git a/services/web_panel/utils/auth.py b/services/web_panel/utils/auth.py index 21e7783..1361a36 100644 --- a/services/web_panel/utils/auth.py +++ b/services/web_panel/utils/auth.py @@ -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. 拦截逻辑