| """网关 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 |
|
|
|
|
| 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): |
| 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) |
|
|
