34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
"""FastAPI 依赖：IAM token 校验，提取当前管理员
|
|
|
|
从请求头 digi-middleware-auth-user / digi-middleware-auth-app 读取凭证，
|
|
调用 IAM /api/iam/v2/identity/token/analyze 校验并解析用户信息。
|
|
"""
|
|
|
|
from fastapi import Header, HTTPException, status
|
|
|
|
from . import iam
|
|
|
|
|
|
async def current_admin(
|
|
user_token: str | None = Header(None, alias="digi-middleware-auth-user"),
|
|
app_token: str | None = Header(None, alias="digi-middleware-auth-app"),
|
|
) -> dict:
|
|
if not user_token:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "未提供 IAM userToken")
|
|
|
|
info = await iam.analyze_token(user_token, app_token)
|
|
if info is None:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "IAM token 无效或已过期")
|
|
|
|
# 返回统一的管理员信息（兼容原有 admin.get("username") 调用）
|
|
return {
|
|
"username": info.get("name") or info.get("id") or "unknown",
|
|
"userId": info.get("id"),
|
|
"name": info.get("name"),
|
|
"email": info.get("email"),
|
|
"telephone": info.get("telephone"),
|
|
"tenantId": info.get("tenantId"),
|
|
"tenantName": info.get("tenantName"),
|
|
"raw": info,
|
|
}
|