81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""鼎捷云 IAM token 鉴权服务
|
||
|
||
通过调用 IAM `/api/iam/v2/identity/token/analyze` 校验请求头中的
|
||
digi-middleware-auth-user / digi-middleware-auth-app，解析出用户信息。
|
||
|
||
进程内 LRU 缓存（userToken -> userInfo），TTL 由 IAM_CACHE_TTL 控制，
|
||
减少对 IAM 的重复调用。
|
||
"""
|
||
|
||
import time
|
||
|
||
import httpx
|
||
|
||
from .config import settings
|
||
|
||
# httpx 异步客户端（进程级单例，复用连接池）
|
||
_http_client: httpx.AsyncClient | None = None
|
||
|
||
# 缓存：userToken -> (user_info, fetched_at)
|
||
_cache: dict[str, tuple[dict | None, float]] = {}
|
||
|
||
|
||
def _get_http_client() -> httpx.AsyncClient:
|
||
global _http_client
|
||
if _http_client is None:
|
||
_http_client = httpx.AsyncClient(timeout=5.0)
|
||
return _http_client
|
||
|
||
|
||
async def close_iam_client() -> None:
|
||
"""关闭 httpx 客户端（进程退出时调用）。"""
|
||
global _http_client
|
||
if _http_client is not None:
|
||
await _http_client.aclose()
|
||
_http_client = None
|
||
|
||
|
||
async def analyze_token(user_token: str, app_token: str | None = None) -> dict | None:
|
||
"""校验 IAM userToken，返回用户信息 dict 或 None。
|
||
|
||
调用 POST {IAM_BASE_URL}/api/iam/v2/identity/token/analyze，
|
||
请求头携带 digi-middleware-auth-app 与 digi-middleware-auth-user。
|
||
成功时返回包含 id/name/tenantId/tenantName 等字段的 dict。
|
||
"""
|
||
# 1. 查缓存
|
||
now = time.time()
|
||
cached = _cache.get(user_token)
|
||
if cached is not None and (now - cached[1]) < settings.IAM_CACHE_TTL:
|
||
return cached[0]
|
||
|
||
# 2. 调用 IAM analyze
|
||
headers = {
|
||
"digi-middleware-auth-user": user_token,
|
||
"digi-middleware-auth-app": app_token or settings.IAM_APP_TOKEN,
|
||
}
|
||
try:
|
||
client = _get_http_client()
|
||
resp = await client.post(
|
||
f"{settings.IAM_BASE_URL}/api/iam/v2/identity/token/analyze",
|
||
headers=headers,
|
||
)
|
||
except Exception as ex:
|
||
# IAM 不可达，缓存短时间避免雪崩
|
||
_cache[user_token] = (None, now)
|
||
print(f"[IAM] analyze 请求异常: {ex}")
|
||
return None
|
||
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
# 兼容字段：id（用户标识）/ name（姓名）
|
||
if data.get("id") or data.get("name"):
|
||
_cache[user_token] = (data, now)
|
||
return data
|
||
# 返回体无用户标识，视为无效
|
||
_cache[user_token] = (None, now)
|
||
return None
|
||
|
||
# 非 200（token 无效/过期），缓存避免雪崩
|
||
_cache[user_token] = (None, now)
|
||
return None
|