feat: 添加仪表板路由和响应数据脱敏功能 - 添加了仪表板模块路由并集成到主路由器中 - 实现了敏感数据响应脱敏配置和功能 - 增加了 Feishu 审批卡片操作处理功能 - 支持通过任务队列异步推送日常简报和项目周报 - 添加了风险事件生成的任务队列支持 - 在 smoke 测试中增加了相关功能验证 refactor: 格式化模型注册模块导入列表 - 将单行导入列表改为多行格式以提高可读性 ```
55 lines
1.4 KiB
Python
55 lines
1.4 KiB
Python
from typing import Any
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
MASKED_VALUE = "[MASKED]"
|
|
SENSITIVE_RESPONSE_KEYS = frozenset(
|
|
{
|
|
"account_number",
|
|
"api_key",
|
|
"bank_account",
|
|
"card_no",
|
|
"direct_llm_api_key",
|
|
"email",
|
|
"feishu_app_secret",
|
|
"feishu_verification_token",
|
|
"hermes_api_key",
|
|
"id_card",
|
|
"mobile",
|
|
"openclaw_gateway_token",
|
|
"password",
|
|
"payment_account",
|
|
"phone",
|
|
"secret",
|
|
"tenant_access_token",
|
|
"token",
|
|
}
|
|
)
|
|
|
|
|
|
def mask_configured(value: Any) -> Any:
|
|
"""Mask sensitive response fields when response masking is enabled."""
|
|
|
|
settings = get_settings()
|
|
if not settings.mask_sensitive_responses:
|
|
return value
|
|
return mask_sensitive(value)
|
|
|
|
|
|
def mask_sensitive(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
masked: dict[str, Any] = {}
|
|
for key, item in value.items():
|
|
key_text = str(key)
|
|
if key_text.lower() in SENSITIVE_RESPONSE_KEYS:
|
|
masked[key_text] = MASKED_VALUE
|
|
else:
|
|
masked[key_text] = mask_sensitive(item)
|
|
return masked
|
|
if isinstance(value, list):
|
|
return [mask_sensitive(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return [mask_sensitive(item) for item in value]
|
|
return value
|