"""网关 API Key 管理:多 key 库(JSON 文件存储)+ 生成/校验。 - 文件:``/api_keys.json``(dir 通常 = settings.log_dir,HF 持久化在 /data/logs)。 - 兼容:config 的 ``[gateway].api_key``(gateway_api_key)自动并入校验集合, 面板生成的 key 与它任一匹配即放行 —— 老客户端不用换 key。 - 线程安全:写盘用 flock 原子替换;内存缓存由 app.state.api_keys 持有, admin 修改后同步(见 app/admin.py 的 /admin/apikeys 端点)。 """ from __future__ import annotations import json import secrets import time from pathlib import Path from typing import Any from app._compat import flock_ex KEY_LEN = 32 # token_urlsafe 字节数 → 43 字符 def generate_key() -> str: """生成随机 API key(url-safe,无需转义)。""" return "an-" + secrets.token_urlsafe(KEY_LEN) def _path(dir_path: Path) -> Path: return dir_path / "api_keys.json" def load_keys(dir_path: Path) -> list[dict[str, Any]]: """读 api_keys.json;文件不存在/损坏 → 空列表。""" p = _path(dir_path) if not p.is_file(): return [] try: data = json.loads(p.read_text(encoding="utf-8")) if isinstance(data, list): return [k for k in data if isinstance(k, dict) and k.get("key")] except (ValueError, OSError): # noqa: BLE001 pass return [] def save_keys(dir_path: Path, keys: list[dict[str, Any]]) -> None: """原子写 api_keys.json(flock + tmp 替换)。""" dir_path.mkdir(parents=True, exist_ok=True) p = _path(dir_path) tmp = p.with_suffix(".json.tmp") payload = json.dumps(keys, ensure_ascii=False, indent=2) with tmp.open("w", encoding="utf-8") as f: f.write(payload) f.write("\n") flock_ex(f.fileno()) tmp.replace(p) def add_key(dir_path: Path, name: str) -> dict[str, Any]: """生成并追加一个新 key,写盘。name 为空给默认名。""" entry = { "name": (name or "").strip() or f"key-{int(time.time())}", "key": generate_key(), "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "last_used": 0.0, } keys = load_keys(dir_path) keys.append(entry) save_keys(dir_path, keys) return entry def delete_key(dir_path: Path, key: str) -> bool: """删除指定 key(按完整 key 值匹配);删除成功返回 True。""" keys = load_keys(dir_path) keep = [k for k in keys if k.get("key") != key] if len(keep) == len(keys): return False save_keys(dir_path, keep) return True def is_valid_key(provided: str | None, keys: list[dict[str, Any]], legacy: str = "") -> bool: """校验:面板 key 库任一项匹配,或等于 config 兼容 key。""" if not provided: return False if legacy and provided == legacy: return True return any(k.get("key") == provided for k in keys)