tiku/tiku-server/app/routers/auth.py

"""认证路由"""

import hashlib
import hmac
import time

from fastapi import APIRouter, Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer(auto_error=False)

router = APIRouter()


def verify_token(request: Request, credentials: HTTPAuthorizationCredentials | None = Depends(security)):
    """验证 Bearer Token + 客户端签名（防非客户端请求）"""
    config = request.app.state.config
    if not credentials:
        raise HTTPException(status_code=401, detail={"error": {"code": "unauthorized", "message": "缺少认证 token"}})
    if credentials.credentials != config.token:
        raise HTTPException(status_code=401, detail={"error": {"code": "unauthorized", "message": "Token 无效"}})

    # 客户端请求签名验证（所有认证请求必须带签名）
    req_ts = request.headers.get("X-Timestamp", "")
    req_sig = request.headers.get("X-Signature", "")
    if not req_ts or not req_sig:
        raise HTTPException(status_code=403, detail={
            "error": {"code": "missing_signature", "message": "缺少客户端签名，请使用 Flutter 客户端访问"}
        })
    try:
        ts = int(req_ts)
        if abs(time.time() - ts) > 300:
            raise HTTPException(status_code=401, detail={
                "error": {"code": "timestamp_expired", "message": "请求时间戳过期"}
            })
        path = request.url.path
        expected = hmac.new(
            config.token.encode(), f"{request.method}{path}{req_ts}".encode(), hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(expected, req_sig):
            raise HTTPException(status_code=403, detail={
                "error": {"code": "invalid_signature", "message": "客户端签名无效"}
            })
    except (ValueError, TypeError):
        raise HTTPException(status_code=401, detail={
            "error": {"code": "bad_timestamp", "message": "时间戳格式错误"}
        })

    return True


@router.get("/health")
def health(request: Request):
    """健康检查（无需认证）"""
    from app.routers.import_ import _import_queue
    return {"status": "ok", "version": "1.0.0", "queue_depth": _import_queue.qsize()}


@router.post("/auth/verify")
def verify(request: Request, body: dict):
    """验证 token 有效性"""
    config = request.app.state.config
    token = body.get("token", "")
    if token == config.token:
        return {"valid": True, "token": token}
    raise HTTPException(status_code=401, detail={"error": {"code": "unauthorized", "message": "Token 无效"}})