app/auth.py

"""鉴权：access_key 校验 + JWT 临时令牌。

简化点：
- 去掉 XTC_ACCESS_SHORT_KEY / XTC_ACCESS_PIN
- 用 JWT 替代内存 Map，多实例可用
- access_token 持久化到 SQLite（用于吊销/查询）
"""
from __future__ import annotations

import secrets
import time
from typing import Optional

import jwt
from fastapi import Header, Request

from .config import get_settings
from .database import get_conn
from .errors import HttpError

ALGORITHM = "HS256"


# ===== access_key 校验 =====

def _extract_access_key(
    *,
    authorization: Optional[str],
    x_access_key: Optional[str],
) -> Optional[str]:
    if x_access_key:
        return x_access_key.strip()
    if authorization:
        auth = authorization.strip()
        if auth.lower().startswith("bearer "):
            return auth[7:].strip()
        if auth.lower().startswith("basic "):
            return auth[6:].strip()
        return auth
    return None


def verify_access_key(provided: Optional[str]) -> None:
    settings = get_settings()
    if not settings.has_access_key:
        # 未配置 access_key 时放行（开发模式）
        return
    if not provided:
        raise HttpError(
            "Missing access key: provide x-xtc-access-key header or Authorization: Bearer <key>",
            status=401,
            code="unauthorized",
        )
    # 先校验长期 key
    if _consteq(provided, settings.xtc_access_key):
        return
    # 再校验临时 token
    if verify_access_token(provided):
        return
    raise HttpError(
        "Invalid access key or token",
        status=401,
        code="unauthorized",
    )


def verify_access_token(token: str) -> bool:
    """校验临时令牌：JWT 签名 + SQLite 未吊销 + 未过期。"""
    settings = get_settings()
    if not settings.has_jwt_secret:
        return False
    try:
        payload = jwt.decode(token, settings.xtc_jwt_secret, algorithms=[ALGORITHM])
    except jwt.PyJWTError:
        return False
    if payload.get("type") != "access":
        return False
    jti = payload.get("jti")
    if not jti:
        return False
    with get_conn() as conn:
        row = conn.execute(
            "SELECT expires_at, revoked FROM access_tokens WHERE token = ?",
            (jti,),
        ).fetchone()
    if not row:
        return False
    if row["revoked"]:
        return False
    if int(row["expires_at"]) < int(time.time()):
        return False
    return True


# ===== 临时令牌签发 =====

def issue_access_token(*, issued_by: str = "admin") -> dict:
    settings = get_settings()
    if not settings.has_jwt_secret:
        raise HttpError(
            "XTC_JWT_SECRET not configured, cannot issue token",
            status=500,
            code="server_misconfigured",
        )
    ttl = settings.xtc_access_token_ttl_sec
    length = settings.xtc_access_token_length
    jti = secrets.token_urlsafe(length)[:length] if length <= 32 else secrets.token_urlsafe(32)
    now = int(time.time())
    expires_at = now + ttl
    payload = {
        "type": "access",
        "jti": jti,
        "iat": now,
        "exp": expires_at,
    }
    token = jwt.encode(payload, settings.xtc_jwt_secret, algorithm=ALGORITHM)
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO access_tokens(token, issued_at, expires_at, issued_by, revoked) "
            "VALUES(?,?,?,?,0)",
            (jti, now, expires_at, issued_by),
        )
    return {
        "token": token,
        "jti": jti,
        "issued_at": now,
        "expires_at": expires_at,
        "expires_in_sec": ttl,
    }


def revoke_access_token(jti: str) -> bool:
    with get_conn() as conn:
        cur = conn.execute(
            "UPDATE access_tokens SET revoked = 1 WHERE token = ? AND revoked = 0",
            (jti,),
        )
        return cur.rowcount > 0


def list_access_tokens() -> list[dict]:
    now = int(time.time())
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT token, issued_at, expires_at, issued_by, revoked FROM access_tokens "
            "ORDER BY issued_at DESC LIMIT 200"
        ).fetchall()
    out = []
    for r in rows:
        out.append(
            {
                "jti": r["token"],
                "issued_at": int(r["issued_at"]),
                "expires_at": int(r["expires_at"]),
                "issued_by": r["issued_by"],
                "revoked": bool(r["revoked"]),
                "expired": int(r["expires_at"]) < now,
            }
        )
    return out


# ===== admin key 校验 =====

def verify_admin_key(provided: Optional[str]) -> None:
    settings = get_settings()
    if not settings.is_admin_enabled:
        raise HttpError(
            "Admin API disabled (XTC_DISABLE_ADMIN=true)",
            status=403,
            code="forbidden",
        )
    if not settings.has_admin_key:
        raise HttpError(
            "XTC_ADMIN_KEY not configured",
            status=500,
            code="server_misconfigured",
        )
    if not provided:
        raise HttpError(
            "Missing admin key: provide x-xtc-admin-key header or Authorization: Bearer <key>",
            status=401,
            code="unauthorized",
        )
    if not _consteq(provided, settings.xtc_admin_key):
        raise HttpError("Invalid admin key", status=401, code="unauthorized")


# ===== FastAPI 依赖 =====

def require_access_key(
    request: Request,
    authorization: Optional[str] = Header(default=None),
    x_xtc_access_key: Optional[str] = Header(default=None, alias="x-xtc-access-key"),
    x_xtc_access_token: Optional[str] = Header(default=None, alias="x-xtc-access-token"),
) -> str:
    provided = _extract_access_key(
        authorization=authorization,
        x_access_key=x_xtc_access_key or x_xtc_access_token,
    )
    verify_access_key(provided)
    return provided or ""


def require_admin_key(
    request: Request,
    authorization: Optional[str] = Header(default=None),
    x_xtc_admin_key: Optional[str] = Header(default=None, alias="x-xtc-admin-key"),
) -> str:
    provided = _extract_access_key(
        authorization=authorization,
        x_access_key=x_xtc_admin_key,
    )
    verify_admin_key(provided)
    return provided or ""


def _consteq(a: str, b: str) -> bool:
    if not isinstance(a, str) or not isinstance(b, str):
        return False
    if len(a) != len(b):
        return False
    result = 0
    for x, y in zip(a, b):
        result |= ord(x) ^ ord(y)
    return result == 0