SenSu/services/web_panel/routes/apikeys.py

#!/usr/bin/env python3
"""API Key 管理路由 — 创建/列表/删除长期有效的 API 密钥，支持权限范围和过期时间"""
import secrets
import time
import logging
from aiohttp import web
from ..utils.auth import panel_auth

logger = logging.getLogger(__name__)

# 预定义权限模板
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] = {}


def _load_keys_from_db(app):
    """从数据库恢复 API keys"""
    try:
        db = app.get("sensu_db")
        if not db:
            sm = app.get("service_manager")
            if sm:
                try:
                    db = sm.get_service("sensu_db")
                except Exception:
                    pass
        if db:
            import json
            raw = db.get_config("api_keys", "{}")
            stored = json.loads(raw) if raw else {}
            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 API_KEYS:
                logger.info(f"🔑 从数据库恢复了 {len(API_KEYS)} 个 API Key")
    except Exception as e:
        logger.warning(f"API Key 恢复失败: {e}")


def _save_keys_to_db(app):
    """持久化 API keys 到数据库"""
    try:
        db = app.get("sensu_db")
        if not db:
            sm = app.get("service_manager")
            if sm:
                try:
                    db = sm.get_service("sensu_db")
                except Exception:
                    pass
        if db:
            import json
            db.set_config("api_keys", json.dumps(API_KEYS))
    except Exception as e:
        logger.debug(f"API Key 持久化失败: {e}")


def validate_api_key(token: str) -> dict | 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=""):
    """注册 API Key 管理路由"""
    _load_keys_from_db(app)
    app["api_keys"] = API_KEYS

    # ── 列表 ──
    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,
                "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, "presets": PERMISSION_PRESETS})

    # ── 创建 ──
    async def create_key(req):
        try:
            data = await req.json()
            name = data.get("name", "").strip()
            if not name:
                return web.json_response({"error": "名称不能为空"}, status=400)

            # 权限范围: 模板名 或 自定义列表
            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]}...) "
                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:
            return web.json_response({"error": str(e)}, status=500)

    # ── 删除 ──
    async def delete_key(req):
        try:
            data = await req.json()
            full_key = data.get("key", "")
            if full_key in API_KEYS:
                name = API_KEYS[full_key]["name"]
                del API_KEYS[full_key]
                _save_keys_to_db(req.app)
                logger.info(f"🔑 API Key 已删除: {name}")
                return web.json_response({"ok": True})
            return web.json_response({"error": "Key 不存在"}, status=404)
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    app.router.add_get(f"{prefix}/api/apikeys", panel_auth(list_keys))
    app.router.add_post(f"{prefix}/api/apikeys", panel_auth(create_key))
    app.router.add_delete(f"{prefix}/api/apikeys", panel_auth(delete_key))
    logger.info(f"🔑 API Key 管理路由已注册 ({prefix}/api/apikeys)")