app/services/log_api_key_store.py

"""日志查询 API Key 管理。

用途：生成可吊销的随机 API Key，凭 key 调用 /api/logs 查询请求日志
（不需要 admin key，方便外部工具/脚本排查线上问题）。

- key 随机生成（xtclog_ 前缀 + 32 字节 urlsafe 随机数），只在建时完整返回一次
- 可随时吊销（revoked_at > 0 即失效，立即生效）
- 记录 last_used_at 便于审计
"""
from __future__ import annotations

import secrets
import time
from typing import Optional

from ..database import get_conn

KEY_PREFIX = "xtclog_"


def _now() -> int:
    return int(time.time())


def generate_key(name: str = "") -> dict:
    """生成新 key。返回 {key, name, createdAt}（key 仅此一次完整返回）。"""
    key = KEY_PREFIX + secrets.token_urlsafe(24)
    now = _now()
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO log_api_keys(key, name, created_at, revoked_at, last_used_at) "
            "VALUES(?, ?, ?, 0, 0)",
            (key, str(name or "")[:100], now),
        )
    return {"key": key, "name": str(name or "")[:100], "createdAt": now}


def list_keys() -> list[dict]:
    """列出全部 key（脱敏：只显示前缀 + 前 6 位）。"""
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT key, name, created_at, revoked_at, last_used_at "
            "FROM log_api_keys ORDER BY created_at DESC"
        ).fetchall()
    out = []
    for r in rows:
        k = str(r["key"])
        out.append({
            "keyMasked": k[:len(KEY_PREFIX) + 6] + "..." if len(k) > len(KEY_PREFIX) + 6 else k,
            "name": r["name"] or "",
            "createdAt": int(r["created_at"]),
            "revoked": int(r["revoked_at"] or 0) > 0,
            "revokedAt": int(r["revoked_at"]) if int(r["revoked_at"] or 0) > 0 else None,
            "lastUsedAt": int(r["last_used_at"] or 0) or None,
        })
    return out


def revoke_key(key: str) -> bool:
    """吊销 key（支持完整 key 或脱敏前缀匹配的完整 key）。返回是否吊销成功。"""
    k = str(key or "").strip()
    if not k:
        return False
    now = _now()
    with get_conn() as conn:
        cur = conn.execute(
            "UPDATE log_api_keys SET revoked_at = ? "
            "WHERE key = ? AND revoked_at = 0",
            (now, k),
        )
        return cur.rowcount > 0


def verify_key(key: str) -> bool:
    """校验 key 有效（存在且未吊销），成功时更新 last_used_at。"""
    k = str(key or "").strip()
    if not k.startswith(KEY_PREFIX):
        return False
    now = _now()
    with get_conn() as conn:
        row = conn.execute(
            "SELECT revoked_at FROM log_api_keys WHERE key = ?", (k,)
        ).fetchone()
        if not row or int(row["revoked_at"] or 0) > 0:
            return False
        try:
            conn.execute(
                "UPDATE log_api_keys SET last_used_at = ? WHERE key = ?", (now, k)
            )
        except Exception:
            pass
    return True