refactor(utils): 统一敏感字段脱敏逻辑，新增trace_id上下文工具

1.  新增请求级trace_id上下文共享模块，通过ContextVar存储trace_id供跨层使用
2.  重构gRPC和SOAP执行器的脱敏逻辑，统一使用crypto模块的敏感头集合
3.  扩展crypto模块的敏感字段匹配规则，新增全量遮蔽的强敏感字段类型
4.  新增mask_full和mask_partial两种脱敏模式，完善脱敏工具链
5.  定义SENSITIVE_HTTP_HEADERS作为HTTP敏感头的单一事实源

backend/package/yuxi/external_systems/framework/adapters/grpc/executor.py

@@ -42,14 +42,12 @@ from yuxi.external_systems.framework.auth_plugins.credentials import (
 )
 from yuxi.external_systems.framework.auth_plugins.registry import resolve_credential
 from yuxi.utils import logger
+from yuxi.utils.crypto import SENSITIVE_HTTP_HEADERS
 from yuxi.utils.net_security import SSRFValidator
 
 if TYPE_CHECKING:
     from yuxi.external_systems.core.contracts import ExecutableTool, RuntimeContext
 
-# 敏感 metadata 键名（大小写不敏感）
-_SENSITIVE_METADATA_KEYS = {"authorization", "cookie", "token", "x-api-key", "api-key"}
-
 
 def _load_grpc():
     """懒加载 grpc，缺失时抛出 DomainValidationError。"""
@@ -163,10 +161,13 @@ def _endpoint_to_url(endpoint: str) -> str:
 
 
 def _sanitize_metadata(metadata: dict[str, str]) -> dict[str, str]:
-    """对敏感 metadata 值进行脱敏。"""
+    """对敏感 metadata 值进行脱敏。
+
+    敏感键名集合统一引用 ``yuxi.utils.crypto.SENSITIVE_HTTP_HEADERS``（单一事实源）。
+    """
     result: dict[str, str] = {}
     for key, value in metadata.items():
-        if key.lower() in _SENSITIVE_METADATA_KEYS:
+        if key.lower() in SENSITIVE_HTTP_HEADERS:
             result[key] = "***"
         else:
             result[key] = value

backend/package/yuxi/external_systems/framework/adapters/soap/executor.py

@@ -39,6 +39,7 @@ from yuxi.external_systems.framework.auth_plugins.credentials import (
 )
 from yuxi.external_systems.framework.auth_plugins.registry import resolve_credential
 from yuxi.utils import logger
+from yuxi.utils.crypto import SENSITIVE_HTTP_HEADERS
 from yuxi.utils.net_security import SSRFValidator
 
 if TYPE_CHECKING:
@@ -46,9 +47,6 @@ if TYPE_CHECKING:
     from zeep.transports import AsyncTransport
     from zeep.wsse.username import UsernameToken
 
-# 敏感 header 键名（大小写不敏感）
-_SENSITIVE_HEADER_KEYS = {"authorization", "cookie", "token", "x-api-key", "api-key", "password"}
-
 
 def _load_zeep():
     """懒加载 zeep 及其子模块，缺失时抛出 DomainValidationError。"""
@@ -63,10 +61,13 @@ def _load_zeep():
 
 
 def _sanitize_headers(headers: dict[str, str]) -> dict[str, str]:
-    """对敏感 header 值进行脱敏。"""
+    """对敏感 header 值进行脱敏。
+
+    敏感键名集合统一引用 ``yuxi.utils.crypto.SENSITIVE_HTTP_HEADERS``（单一事实源）。
+    """
     result: dict[str, str] = {}
     for key, value in headers.items():
-        if key.lower() in _SENSITIVE_HEADER_KEYS:
+        if key.lower() in SENSITIVE_HTTP_HEADERS:
             result[key] = "***"
         else:
             result[key] = value

backend/package/yuxi/utils/crypto.py

@@ -59,8 +59,12 @@ class CryptoHelper:
         "id_token",
         "bearer",
         "authorization",
+        "jwt",
+        "verification_token",
         # 密钥类
         "secret",
+        "app_secret",
+        "encrypt_key",
         "api_key",
         "apikey",
         "api_secret",
@@ -76,6 +80,31 @@ class CryptoHelper:
         # 凭证类
         "credential",
         "credentials",
+        # URL 内嵌 secret 类
+        "webhook_url",
+        "webhook_secret",
+    )
+
+    # 强敏感字段子集：即使 partial 模式也全量遮蔽（不保留前 4 位）。
+    # 依据：password 前缀是常用密码、api_key 前缀可被利用、webhook_url 内嵌
+    # secret（Slack/Discord/Teams），任何片段泄露都等于泄露完整凭证。
+    _FULL_MASK_KEYWORDS = (
+        "password",
+        "passwd",
+        "pwd",
+        "api_key",
+        "apikey",
+        "api_secret",
+        "client_secret",
+        "client_key",
+        "private_key",
+        "webhook_url",
+        "webhook_secret",
+        "ca_cert",
+        "client_cert",
+        "certificate",
+        "credential",
+        "credentials",
     )
 
     def __init__(self, key: str | None = None, fernet: Fernet | None = None) -> None:
@@ -206,9 +235,13 @@ class CryptoHelper:
         return self._transform_config(config, "decrypt")
 
     def mask_sensitive_fields(self, config: dict[str, Any] | None) -> dict[str, Any]:
-        """返回配置字典的脱敏副本，用于 API 响应。"""
-        result = self._transform_config(config, "mask")
-        return result if result is not None else {}
+        """返回配置字典的脱敏副本（全量遮蔽），用于 API 响应。
+
+        等价于 ``mask_full(config)``，None 输入返回 ``{}``（向后兼容）。
+        """
+        if config is None:
+            return {}
+        return self.mask_full(config)
 
     def mask_all_string_values(self, obj: Any) -> Any:
         """递归将对象中的所有字符串值替换为脱敏占位符。"""
@@ -220,6 +253,62 @@ class CryptoHelper:
             return "***"
         return obj
 
+    def mask_full(self, value: Any, *, parent_key: str = "") -> Any:
+        """全量遮蔽脱敏（对外 HTTP 响应、审计日志）。
+
+        敏感字段统一替换为 ``"***"``，不保留任何片段。递归语义：
+        - dict：按子键判定敏感性
+        - list：按父键判定敏感性（修复原 ``_mask_nested`` list 项不脱敏漏洞）
+        - str：按父键判定敏感性
+        - 其他类型：原样返回
+
+        Args:
+            value: 待脱敏的任意值（dict / list / str / 其他）。
+            parent_key: 父级键名，用于 list 项的敏感性判定（顶层调用留空）。
+
+        Returns:
+            脱敏后的值（新对象，不修改原对象）。
+        """
+        if isinstance(value, dict):
+            return {k: self.mask_full(v, parent_key=k) for k, v in value.items()}
+        if isinstance(value, list):
+            return [self.mask_full(item, parent_key=parent_key) for item in value]
+        if isinstance(value, str):
+            if self._is_sensitive_key(parent_key):
+                return "***"
+            return value
+        return value
+
+    def mask_partial(self, value: Any, *, parent_key: str = "") -> Any:
+        """部分遮蔽脱敏（日志、诊断包）。
+
+        - ``_FULL_MASK_KEYWORDS`` 命中 → ``"***"``（强敏感字段不保留前缀）
+        - 其他 ``_SENSITIVE_KEYWORDS`` 命中 → 前 4 位 + ``"***"``（len<=4 时全量）
+        - 非敏感字段原样返回
+
+        递归语义同 ``mask_full``。
+
+        Args:
+            value: 待脱敏的任意值。
+            parent_key: 父级键名，用于 list 项的敏感性判定。
+
+        Returns:
+            脱敏后的值（新对象，不修改原对象）。
+        """
+        if isinstance(value, dict):
+            return {k: self.mask_partial(v, parent_key=k) for k, v in value.items()}
+        if isinstance(value, list):
+            return [self.mask_partial(item, parent_key=parent_key) for item in value]
+        if isinstance(value, str):
+            if not value:
+                return value
+            if self._is_full_mask_key(parent_key):
+                return "***"
+            if self._is_sensitive_key(parent_key):
+                return value[:4] + "***" if len(value) > 4 else "***"
+            return value
+        return value
+
     # ------------------------------------------------------------------
     # 内部实现
     # ------------------------------------------------------------------
@@ -230,6 +319,12 @@ class CryptoHelper:
         normalized = re.sub(r"[^a-z0-9_]+", "_", key.lower())
         return any(keyword in normalized for keyword in cls._SENSITIVE_KEYWORDS)
 
+    @classmethod
+    def _is_full_mask_key(cls, key: str) -> bool:
+        """判断字段名是否属于强敏感字段（partial 模式也全量遮蔽）。"""
+        normalized = re.sub(r"[^a-z0-9_]+", "_", key.lower())
+        return any(keyword in normalized for keyword in cls._FULL_MASK_KEYWORDS)
+
     def _transform_config(
         self,
         config: dict[str, Any] | None,
@@ -244,7 +339,7 @@ class CryptoHelper:
         if mode == "decrypt":
             return self._decrypt_nested(config)
         # mask
-        return self._mask_nested(config)
+        return self.mask_full(config)
 
     def _encrypt_nested(self, obj: Any) -> Any:
         if isinstance(obj, dict):
@@ -266,21 +361,6 @@ class CryptoHelper:
             return [self._decrypt_nested(item) for item in obj]
         return obj
 
-    def _mask_nested(self, obj: Any) -> Any:
-        if isinstance(obj, dict):
-            return {
-                k: self._mask_string(v) if self._is_sensitive_key(k) and isinstance(v, str) else self._mask_nested(v)
-                for k, v in obj.items()
-            }
-        if isinstance(obj, list):
-            return [self._mask_nested(item) for item in obj]
-        return obj
-
-    @staticmethod
-    def _mask_string(value: str) -> str:
-        """对字符串进行脱敏，返回统一占位符。"""
-        return "***"
-
 
 # ----------------------------------------------------------------------
 # 模块级默认实例与便捷函数（向后兼容现有调用）
@@ -339,10 +419,55 @@ def decrypt(value: str) -> str:
 
 
 def mask_sensitive_fields(config: dict[str, Any] | None) -> dict[str, Any]:
-    """返回配置字典的脱敏副本（委托给默认实例）。"""
+    """返回配置字典的脱敏副本（全量遮蔽，委托给默认实例）。"""
     return _default_helper.mask_sensitive_fields(config)
 
 
 def mask_all_string_values(obj: Any) -> Any:
     """递归将对象中的所有字符串值替换为脱敏占位符（委托给默认实例）。"""
     return _default_helper.mask_all_string_values(obj)
+
+
+def mask_full(value: Any, *, parent_key: str = "") -> Any:
+    """全量遮蔽脱敏（委托给默认实例）。
+
+    供对外 HTTP 响应、审计日志使用，敏感字段统一替换为 ``"***"``。
+    """
+    return _default_helper.mask_full(value, parent_key=parent_key)
+
+
+def mask_partial(value: Any, *, parent_key: str = "") -> Any:
+    """部分遮蔽脱敏（委托给默认实例）。
+
+    供日志、诊断包使用：强敏感字段全量遮蔽，普通敏感字段保留前 4 位。
+    """
+    return _default_helper.mask_partial(value, parent_key=parent_key)
+
+
+def is_sensitive_key(key: str) -> bool:
+    """判断字段名是否属于敏感字段（委托给默认实例）。"""
+    return _default_helper._is_sensitive_key(key)
+
+
+# ----------------------------------------------------------------------
+# HTTP header 敏感字段集合（单一事实源）
+# ----------------------------------------------------------------------
+# 用于过滤 / 脱敏日志与诊断包中的 HTTP header。统一收口到 crypto 模块，
+# 避免在 channels/__init__.py / webhook_router.py / grpc/executor.py /
+# soap/executor.py 等多处重复定义发散（统一脱敏重构）。
+# 取值是 4 处历史定义的并集：authorization / cookie / set-cookie /
+# x-api-key / x-auth-token / proxy-authorization / token / api-key /
+# password。调用方使用 ``key.lower() in SENSITIVE_HTTP_HEADERS`` 判定。
+SENSITIVE_HTTP_HEADERS: frozenset[str] = frozenset(
+    {
+        "authorization",
+        "proxy-authorization",
+        "cookie",
+        "set-cookie",
+        "x-api-key",
+        "x-auth-token",
+        "token",
+        "api-key",
+        "password",
+    }
+)

backend/package/yuxi/utils/trace_context.py

@@ -0,0 +1,40 @@
+"""请求级 trace_id 上下文（共享模块）。
+
+通过 ContextVar 存储每个请求的 trace_id，供异常处理器、日志适配器
+等跨层读取。中间件入口生成（或从 X-Request-Id header 读取）。
+
+本模块为 server 层与 channels 层共享的 trace_id 机制，确保两套
+ContextVar（server 层 ``_trace_id_var`` 与 channels 层 ``_current_span``）
+保持同步。
+"""
+from __future__ import annotations
+
+import uuid
+from contextvars import ContextVar
+
+_trace_id_var: ContextVar[str | None] = ContextVar(
+    "yuxi_request_trace_id", default=None
+)
+
+
+def set_trace_id(trace_id: str) -> None:
+    """设置当前请求的 trace_id。"""
+    _trace_id_var.set(trace_id)
+
+
+def get_trace_id() -> str | None:
+    """获取当前请求的 trace_id，未设置时返回 None。"""
+    return _trace_id_var.get()
+
+
+def generate_trace_id() -> str:
+    """生成新的 trace_id（uuid4 十六进制）。"""
+    return uuid.uuid4().hex
+
+
+def reset_trace_id() -> None:
+    """请求结束时重置 trace_id。
+
+    ContextVar 在请求结束后自动回收，显式重置作为保险。
+    """
+    _trace_id_var.set(None)