vent_deep_agent/api/auth.py

# -*- coding: utf-8 -*-
"""
认证模块 —— 通过请求头 x-access-token 验证用户登录状态

调用 VENT_SYSTEM_BASE_URL + "/sys/user/getUserInfo"，
将客户端请求头中的 x-access-token 转发到通风系统进行身份验证。
"""

import os

import httpx
from fastapi import HTTPException, Request

# ── 通风系统认证服务地址 ──
VENT_SYSTEM_BASE_URL = os.getenv("VENT_SYSTEM_BASE_URL", "").strip()


async def get_current_user(request: Request) -> dict:
    """FastAPI 依赖项：从请求头 x-access-token 验证用户登录状态。

    验证流程：
    1. 从 HTTP 请求头中提取 x-access-token
    2. 转发到通风系统 GET /sys/user/getUserInfo
    3. 若响应 success=false 或无 userInfo 则抛出 401

    Returns:
        dict: 用户信息（取自 result.userInfo）

    Raises:
        HTTPException 401: 未登录或令牌无效
        HTTPException 502: 认证服务不可用
        HTTPException 504: 认证服务超时
    """
    # ── 1. 提取令牌 ──
    token = request.headers.get("x-access-token")
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证令牌")

    # ── 2. 检查后端地址配置 ──
    if not VENT_SYSTEM_BASE_URL:
        raise HTTPException(status_code=500, detail="认证服务地址未配置 (VENT_SYSTEM_BASE_URL)")

    url = f"{VENT_SYSTEM_BASE_URL}/sys/user/getUserInfo"

    # ── 3. 转发令牌到通风系统验证 ──
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url, headers={"x-access-token": token})
            data = resp.json()
    except httpx.ConnectError as e:
        raise HTTPException(status_code=502, detail=f"认证服务连接失败: {e}")
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="认证服务响应超时")
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=502, detail=f"认证服务返回错误: {e.response.status_code}")
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"认证服务异常: {e}")

    # ── 4. 校验响应 ──
    if not isinstance(data, dict) or not data.get("success"):
        raise HTTPException(status_code=401, detail="未登录或令牌已过期")

    user_info = data.get("result", {}).get("userInfo")

    if not user_info:
        raise HTTPException(status_code=401, detail="未登录或令牌已过期")

    print(f"[认证] {user_info.get('realname') or user_info.get('username')} | {user_info.get('username')} | {user_info.get('orgCode','?')}")

    return user_info