JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from typing import Any
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
MASKED_VALUE = "[MASKED]"
|
|
SENSITIVE_RESPONSE_KEYS = frozenset(
|
|
{
|
|
"access_token",
|
|
"account_number",
|
|
"api_key",
|
|
"apikey",
|
|
"app_access_token",
|
|
"app_ticket",
|
|
"audit_api_key",
|
|
"authorization",
|
|
"bank_account",
|
|
"card_no",
|
|
"client_secret",
|
|
"cookie",
|
|
"direct_llm_api_key",
|
|
"email",
|
|
"encrypt_key",
|
|
"feishu_app_secret",
|
|
"feishu_app_ticket",
|
|
"feishu_encrypt_key",
|
|
"feishu_verification_token",
|
|
"hermes_api_key",
|
|
"id_card",
|
|
"market_data_token",
|
|
"mobile",
|
|
"openclaw_api_key",
|
|
"openclaw_gateway_token",
|
|
"password",
|
|
"payment_account",
|
|
"phone",
|
|
"private_key",
|
|
"refresh_token",
|
|
"secret",
|
|
"secret_key",
|
|
"set-cookie",
|
|
"set_cookie",
|
|
"tenant_access_token",
|
|
"token",
|
|
"x-api-key",
|
|
"x-audit-api-key",
|
|
"x_api_key",
|
|
"x_audit_api_key",
|
|
}
|
|
)
|
|
NORMALIZED_SENSITIVE_RESPONSE_KEYS = frozenset(
|
|
"".join(character for character in key.casefold() if character.isalnum())
|
|
for key in SENSITIVE_RESPONSE_KEYS
|
|
)
|
|
|
|
|
|
def is_sensitive_key(key: str) -> bool:
|
|
"""Return whether a key matches a built-in sensitive field name."""
|
|
|
|
normalized = "".join(
|
|
character for character in key.casefold() if character.isalnum()
|
|
)
|
|
return normalized in NORMALIZED_SENSITIVE_RESPONSE_KEYS
|
|
|
|
|
|
def mask_configured(value: Any, domain: str | None = None) -> Any:
|
|
"""Mask sensitive response fields when response masking is enabled."""
|
|
|
|
settings = get_settings()
|
|
if not settings.mask_sensitive_responses:
|
|
return value
|
|
configured_fields = {item.lower() for item in settings.masked_response_fields}
|
|
return mask_sensitive(value, domain=domain, configured_fields=configured_fields)
|
|
|
|
|
|
def mask_sensitive(
|
|
value: Any,
|
|
domain: str | None = None,
|
|
configured_fields: set[str] | None = None,
|
|
) -> Any:
|
|
if isinstance(value, dict):
|
|
masked: dict[str, Any] = {}
|
|
for key, item in value.items():
|
|
key_text = str(key)
|
|
if _should_mask(key_text, domain, configured_fields or set()):
|
|
masked[key_text] = MASKED_VALUE
|
|
else:
|
|
masked[key_text] = mask_sensitive(item, domain, configured_fields)
|
|
return masked
|
|
if isinstance(value, list):
|
|
return [mask_sensitive(item, domain, configured_fields) for item in value]
|
|
if isinstance(value, tuple):
|
|
return [mask_sensitive(item, domain, configured_fields) for item in value]
|
|
return value
|
|
|
|
|
|
def _should_mask(key: str, domain: str | None, configured_fields: set[str]) -> bool:
|
|
field = key.lower()
|
|
if is_sensitive_key(key) or field in configured_fields:
|
|
return True
|
|
if f"*.{field}" in configured_fields:
|
|
return True
|
|
return bool(domain and f"{domain.lower()}.{field}" in configured_fields)
|