File size: 2,943 Bytes
fa1140b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """网关 API Key 管理：多 key 库（JSON 文件存储）+ 生成/校验。
- 文件：``<dir>/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)
|