From 2c190b2a5c8661ef2ebc69a4ed4db860852d9b31 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 1 Jul 2026 19:48:25 +0800 Subject: [PATCH] =?UTF-8?q?refactor(utils):=20=E7=BB=9F=E4=B8=80=E6=95=8F?= =?UTF-8?q?=E6=84=9F=E5=AD=97=E6=AE=B5=E8=84=B1=E6=95=8F=E9=80=BB=E8=BE=91?= =?UTF-8?q?=EF=BC=8C=E6=96=B0=E5=A2=9Etrace=5Fid=E4=B8=8A=E4=B8=8B?= =?UTF-8?q?=E6=96=87=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 新增请求级trace_id上下文共享模块，通过ContextVar存储trace_id供跨层使用 2. 重构gRPC和SOAP执行器的脱敏逻辑，统一使用crypto模块的敏感头集合 3. 扩展crypto模块的敏感字段匹配规则，新增全量遮蔽的强敏感字段类型 4. 新增mask_full和mask_partial两种脱敏模式，完善脱敏工具链 5. 定义SENSITIVE_HTTP_HEADERS作为HTTP敏感头的单一事实源 --- .../framework/adapters/grpc/executor.py | 11 +- .../framework/adapters/soap/executor.py | 11 +- backend/package/yuxi/utils/crypto.py | 165 +++++++++++++++--- backend/package/yuxi/utils/trace_context.py | 40 +++++ 4 files changed, 197 insertions(+), 30 deletions(-) create mode 100644 backend/package/yuxi/utils/trace_context.py diff --git a/backend/package/yuxi/external_systems/framework/adapters/grpc/executor.py b/backend/package/yuxi/external_systems/framework/adapters/grpc/executor.py index ff0dd3e9..6bb75012 100644 --- a/backend/package/yuxi/external_systems/framework/adapters/grpc/executor.py +++ b/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 diff --git a/backend/package/yuxi/external_systems/framework/adapters/soap/executor.py b/backend/package/yuxi/external_systems/framework/adapters/soap/executor.py index ccdf3ee8..647eed9c 100644 --- a/backend/package/yuxi/external_systems/framework/adapters/soap/executor.py +++ b/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 diff --git a/backend/package/yuxi/utils/crypto.py b/backend/package/yuxi/utils/crypto.py index 6c895b1f..66b79b8d 100644 --- a/backend/package/yuxi/utils/crypto.py +++ b/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", + } +) diff --git a/backend/package/yuxi/utils/trace_context.py b/backend/package/yuxi/utils/trace_context.py new file mode 100644 index 00000000..3687ef81 --- /dev/null +++ b/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)