From b10366e8987b600800da1abf42463050e19beb25 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Sun, 12 Jul 2026 23:53:53 +0800 Subject: [PATCH] =?UTF-8?q?refactor(wechat-woc):=20=E7=A7=BB=E9=99=A4bridg?= =?UTF-8?q?e=5Furl=20scheme=E5=BC=BA=E5=88=B6=E6=A0=A1=E9=AA=8C=EF=BC=8C?= =?UTF-8?q?=E7=AE=80=E5=8C=96URL=E6=A0=A1=E9=AA=8C=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 移除`_validate_bridge_url_scheme`相关代码与校验逻辑 2. 调整WeChatWoc插件的配置校验注释与测试用例 3. 不再强制要求bridge_url使用https协议,仅保留SSRF主机校验 4. 简化host_bootstrap中GIN索引创建逻辑,移除冗余索引创建步骤 5. 新增transport_mode数据库字段与适配器实现 6. 新增WhitelistConfigAdapter白名单配置适配器 7. 优化前端仪表盘实时监控轮询逻辑,移除可见性感知轮询改为手动触发 --- .../adapters/structured_logger_adapter.py | 16 ++ .../whitelist_config_adapter.py | 0 .../contract/ports/driven/logger_port.py | 24 +++ .../channels/infrastructure/host_bootstrap.py | 9 +- .../plugins/wechat_woc/_url_validators.py | 34 ----- .../wechat_woc/adapters/lifecycle_adapter.py | 7 +- .../wechat_woc/adapters/wizard_adapter.py | 4 +- .../channels/plugins/wechat_woc/lifecycle.py | 23 ++- .../package/yuxi/storage/postgres/manager.py | 24 ++- .../yuxi/storage/postgres/models_channels.py | 8 + .../adapters/test_lifecycle_adapter.py | 12 +- .../adapters/test_wizard_adapter.py | 35 +---- .../plugins/wechat_woc/test_lifecycle.py | 33 ++--- .../plugins/wechat_woc/test_url_validators.py | 71 +-------- .../channels/useChannelDashboard.js | 137 +++++++++++------- 15 files changed, 192 insertions(+), 245 deletions(-) rename backend/package/yuxi/channels/{infrastructure => adapters}/whitelist_config_adapter.py (100%) diff --git a/backend/package/yuxi/channels/adapters/structured_logger_adapter.py b/backend/package/yuxi/channels/adapters/structured_logger_adapter.py index d88f704d..fde98c13 100644 --- a/backend/package/yuxi/channels/adapters/structured_logger_adapter.py +++ b/backend/package/yuxi/channels/adapters/structured_logger_adapter.py @@ -151,6 +151,22 @@ class StructuredLoggerAdapter(LoggerPort): """ self._emit("WARNING", message, trace_id=trace_id, **kwargs) + async def warning( + self, + message: str, + *, + trace_id: str | None = None, + **kwargs: Any, + ) -> None: + """记录 WARNING 级别日志(warn 的别名,兼容标准日志命名)。 + + Args: + message: 日志消息。 + trace_id: 链路追踪 ID(可选)。 + **kwargs: 附加上下文(敏感字段自动脱敏)。 + """ + self._emit("WARNING", message, trace_id=trace_id, **kwargs) + async def error( self, message: str, diff --git a/backend/package/yuxi/channels/infrastructure/whitelist_config_adapter.py b/backend/package/yuxi/channels/adapters/whitelist_config_adapter.py similarity index 100% rename from backend/package/yuxi/channels/infrastructure/whitelist_config_adapter.py rename to backend/package/yuxi/channels/adapters/whitelist_config_adapter.py diff --git a/backend/package/yuxi/channels/contract/ports/driven/logger_port.py b/backend/package/yuxi/channels/contract/ports/driven/logger_port.py index e77c7a10..c01ae361 100644 --- a/backend/package/yuxi/channels/contract/ports/driven/logger_port.py +++ b/backend/package/yuxi/channels/contract/ports/driven/logger_port.py @@ -129,6 +129,30 @@ class LoggerPort(Protocol): """ ... + async def warning( + self, + message: str, + *, + trace_id: str | None = None, + **kwargs: Any, + ) -> None: + """记录 WARNING 级别日志(warn 的别名,兼容标准日志命名)。 + + @pre + - message 非空 + - kwargs 中的敏感字段已由调用方脱敏 + + @post + - WARNING 级别日志已输出,携带 trace_id(若提供)与 kwargs 上下文 + + @failure + - 无(日志故障不得阻断主流程) + + @consistency + - 最终一致性:日志异步写入 + """ + ... + async def error( self, message: str, diff --git a/backend/package/yuxi/channels/infrastructure/host_bootstrap.py b/backend/package/yuxi/channels/infrastructure/host_bootstrap.py index d5b1cf92..1d4cc7bf 100644 --- a/backend/package/yuxi/channels/infrastructure/host_bootstrap.py +++ b/backend/package/yuxi/channels/infrastructure/host_bootstrap.py @@ -572,13 +572,8 @@ class HostBootstrap: self._ensure_schema.ensure_channel_schema(), timeout=self._SCHEMA_INIT_TIMEOUT, ) - # 非阻塞创建 GIN 索引(C-6),失败不阻断启动 - try: - await self._ensure_schema.ensureChannelIndexesConcurrently() - except Exception as idx_exc: - await self._logger.warning( - f"渠道 GIN 索引并发创建失败(不阻断启动,下次重试): {idx_exc}", - ) + # GIN 索引已随 channel schema 通过 ChannelsBase.metadata.create_all + # 同步创建(models_channels.py 中定义),无需额外的并发创建步骤。 except TimeoutError as e: await self._logger.error( f"渠道 schema 初始化超时: timeout={self._SCHEMA_INIT_TIMEOUT}s", diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/_url_validators.py b/backend/package/yuxi/channels/plugins/wechat_woc/_url_validators.py index bd487da9..e6103b60 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/_url_validators.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/_url_validators.py @@ -1,9 +1,5 @@ """bridge_url 公共校验与脱敏工具。 -P0-4:生产环境强制 https,开发环境通过 ``allow_insecure_localhost`` 显式 -豁免 localhost。4 处校验点(lifecycle.py / lifecycle_adapter.py / -wizard_adapter.py)统一调用 ``_validate_bridge_url_scheme``,确保行为一致。 - P1-7:``_redact_url`` 用于日志脱敏,仅保留 scheme://host:port/path 前缀, 去除 query(可能含 bridge_token / session 等敏感信息),避免敏感配置 明文写入日志(§13.4 / §14.4)。 @@ -38,33 +34,6 @@ _SSRF_BLOCKED_RANGES: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] ) -def _validate_bridge_url_scheme( - url: str, - allow_insecure: bool, - *, - trace_id: str | None = None, -) -> None: - """校验 bridge_url scheme。 - - - https 通过; - - http + ``allow_insecure`` + host in (localhost/127.0.0.1/::1) 通过; - - 其余抛 ``ValidationError(field="bridge_url", - message="https_required_or_set_allow_insecure_localhost")``。 - - 内网 IP(如 192.168.x.x)不在豁免范围,需通过反向代理提供 https。 - """ - parsed = urlparse(url) - if parsed.scheme == "https": - return - if parsed.scheme == "http" and allow_insecure and parsed.hostname in _INSECURE_ALLOWED_HOSTS: - return - raise ValidationError( - field="bridge_url", - message="https_required_or_set_allow_insecure_localhost", - trace_id=trace_id, - ) - - def _validate_bridge_url_host( url: str, allow_insecure: bool, @@ -79,9 +48,6 @@ def _validate_bridge_url_host( ``ValidationError(field="bridge_url", message=f"ssrf_blocked_range: {host}")``; - 域名(``ipaddress.ip_address`` 抛 ``ValueError``)放行,DNS 解析后 校验当前不做。 - - 与 ``_validate_bridge_url_scheme`` 配合:scheme 校验通过后追加本函数 - 调用,确保 4 处校验点行为一致。 """ parsed = urlparse(url) host = parsed.hostname diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/lifecycle_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/lifecycle_adapter.py index 0eae37f0..64661cfc 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/lifecycle_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/lifecycle_adapter.py @@ -47,7 +47,7 @@ from .._constants import ( DEFAULT_MAX_MESSAGE_LENGTH, DEFAULT_POLL_INTERVAL_MS, ) -from .._url_validators import _validate_bridge_url_host, _validate_bridge_url_scheme +from .._url_validators import _validate_bridge_url_host # bridge_url 合法协议前缀 _URL_SCHEMES = ("http://", "https://") @@ -162,8 +162,8 @@ class WeChatWocLifecycleAdapter: ) -> dict[str, Any]: """规范化配置。不写库不写 CachePort。 - - 校验 ``bridge_url`` scheme(https 强制,开发环境 localhost 豁免) - 与 host(SSRF 防护,拒绝内网 IP 段) + - 校验 ``bridge_url`` host(SSRF 防护,拒绝内网 IP 段;开发环境可通过 + ``allow_insecure_localhost=true`` 豁免 localhost / 127.0.0.1 / ::1) - 去除 ``bridge_url`` 尾部 ``/`` - 补全默认值(``poll_interval_ms`` / ``max_batch_size`` / ``max_message_length``) @@ -184,7 +184,6 @@ class WeChatWocLifecycleAdapter: message="bridge_url_required", ) allow_insecure = bool(raw_config.get("allow_insecure_localhost", False)) - _validate_bridge_url_scheme(bridge_url, allow_insecure) _validate_bridge_url_host(bridge_url, allow_insecure) normalized = dict(raw_config) diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py index d7e05133..dba78796 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py @@ -60,7 +60,7 @@ from yuxi.channels.contract.errors import ( from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort -from .._url_validators import _redact_url, _validate_bridge_url_host, _validate_bridge_url_scheme +from .._url_validators import _redact_url, _validate_bridge_url_host from ..woc_bridge_client import WocBridgeClient # account_info 步骤 display_name 字段的占位提示文案 @@ -231,7 +231,6 @@ class WeChatWocWizardAdapter: url = bridge_url.strip() allow_insecure = await self._allow_insecure_localhost() try: - _validate_bridge_url_scheme(url, allow_insecure) _validate_bridge_url_host(url, allow_insecure) except ValidationError as exc: return WizardStepResult( @@ -401,7 +400,6 @@ class WeChatWocWizardAdapter: ) url = bridge_url.strip() allow_insecure = await self._allow_insecure_localhost() - _validate_bridge_url_scheme(url, allow_insecure) _validate_bridge_url_host(url, allow_insecure) bridge_token = values.get("bridge_token") if not isinstance(bridge_token, str) or not bridge_token.strip(): diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/lifecycle.py b/backend/package/yuxi/channels/plugins/wechat_woc/lifecycle.py index 36a2828d..07e37df4 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/lifecycle.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/lifecycle.py @@ -12,10 +12,9 @@ onStop / onPause / onResume / onUnload / onReconfigure / onFail), ``resource_quota.max_connections=10``。 - 短轮询拉取由 ``StreamWorker`` 通过 ``PullerAdapter`` 统一管理, ``LifecycleHandler`` 不持有轮询句柄。 -- ``onReconfigure`` 校验 ``bridge_url`` scheme(生产环境强制 ``https://``, - 开发环境通过 ``allow_insecure_localhost`` 豁免 localhost),非法时抛 +- ``onReconfigure`` 校验 ``bridge_url`` host(SSRF 防护,P1-8),非法时抛 ``ValidationError``;空值/类型错误抛 ``ConfigRestartRequiredError`` - 触发宿主重启回滚(FR-37 / P0-4)。 + 触发宿主重启回滚(FR-37)。 - ``onUnload`` 在关闭连接池后清空 CachePort 中 ``wechat_woc:*`` 缓存, 避免插件卸载后残留孤儿缓存键(FR-32 资源释放约束)。 - ``onFail`` 不抛异常、不静默吞错;logger 失败时回退到 ``sys.stderr.write``, @@ -43,7 +42,7 @@ if TYPE_CHECKING: from .woc_bridge_client import WocBridgeClient from ._constants import DEFAULT_MAX_CONNECTIONS, HTTP_TIMEOUT_SECONDS -from ._url_validators import _validate_bridge_url_host, _validate_bridge_url_scheme +from ._url_validators import _validate_bridge_url_host # httpx 连接池默认配置;最终值由 discover 阶段解析的 manifest.resource_quota # 注入,本默认值仅用于测试或未声明 resource_quota 的场景(F-03 单真相源)。 @@ -287,14 +286,13 @@ class WeChatWocLifecycleHandler: old_config: dict[str, Any] | None = None, new_config: dict[str, Any], ) -> None: - """配置热更新:校验 bridge_url scheme 与 host,非法抛 ValidationError。 + """配置热更新:校验 bridge_url host,非法抛 ValidationError。 - ``bridge_url`` 为 wechat_woc 唯一不可降级校验的配置项:生产环境 - 必须使用 ``https://``;开发环境可通过 ``allow_insecure_localhost=true`` - 豁免 ``http://localhost`` / ``127.0.0.1`` / ``::1``(P0-4)。scheme - 校验通过后追加 host 校验拦截内网 IP 段(SSRF 防护,P1-8)。 - 空值或类型错误抛 ``ConfigRestartRequiredError`` 触发宿主重启回滚 - (FR-37)。 + ``bridge_url`` 为 wechat_woc 唯一不可降级校验的配置项:通过 host + 校验拦截内网 IP 段(SSRF 防护,P1-8);开发环境可通过 + ``allow_insecure_localhost=true`` 豁免 ``localhost`` / ``127.0.0.1`` / + ``::1``。空值或类型错误抛 ``ConfigRestartRequiredError`` 触发宿主 + 重启回滚(FR-37)。 校验通过后仅记录日志,不缓存基线(wechat_woc 的 bridge_url 实时 从 ConfigPort 读取,无需本地缓存)。``old_config`` 由宿主通过关键字 @@ -309,7 +307,7 @@ class WeChatWocLifecycleHandler: @failure - ``ConfigRestartRequiredError``:``bridge_url`` 为空或类型错误 - - ``ValidationError``:``bridge_url`` scheme 非法或 host 命中内网段 + - ``ValidationError``:``bridge_url`` host 命中内网段 @consistency - 状态机驱动:配置变更即时生效且支持回滚 @@ -319,7 +317,6 @@ class WeChatWocLifecycleHandler: if not isinstance(bridge_url, str) or not bridge_url: raise ConfigRestartRequiredError(key="bridge_url") from None allow_insecure = bool(new_config.get("allow_insecure_localhost", False)) - _validate_bridge_url_scheme(bridge_url, allow_insecure) _validate_bridge_url_host(bridge_url, allow_insecure) await self._logger.info("WeChat woc plugin reconfigured") diff --git a/backend/package/yuxi/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py index e509de22..79dfe58e 100644 --- a/backend/package/yuxi/storage/postgres/manager.py +++ b/backend/package/yuxi/storage/postgres/manager.py @@ -513,10 +513,26 @@ class PostgresManager(metaclass=SingletonMeta): """ self._check_initialized() stmts = [ - "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS onboarding_status VARCHAR(32) NOT NULL DEFAULT 'pending'", - "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS credential_ref VARCHAR(128)", - "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS credential_version INTEGER NOT NULL DEFAULT 0", - "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS service_user_uid VARCHAR(64)", + ( + "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS " + "onboarding_status VARCHAR(32) NOT NULL DEFAULT 'pending'" + ), + ( + "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS " + "credential_ref VARCHAR(128)" + ), + ( + "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS " + "credential_version INTEGER NOT NULL DEFAULT 0" + ), + ( + "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS " + "service_user_uid VARCHAR(64)" + ), + ( + "ALTER TABLE IF EXISTS channel_accounts ADD COLUMN IF NOT EXISTS " + "transport_mode VARCHAR(16) NOT NULL DEFAULT 'both'" + ), ] async with self.async_engine.begin() as conn: await conn.run_sync(ChannelsBase.metadata.create_all) diff --git a/backend/package/yuxi/storage/postgres/models_channels.py b/backend/package/yuxi/storage/postgres/models_channels.py index 7b5106be..515401b7 100644 --- a/backend/package/yuxi/storage/postgres/models_channels.py +++ b/backend/package/yuxi/storage/postgres/models_channels.py @@ -121,6 +121,13 @@ class ChannelAccount(Base): transport_cursor = Column( String(256), nullable=False, default="", server_default="", comment="传输游标,Puller类型使用" ) + transport_mode = Column( + String(16), + nullable=False, + default="both", + server_default="both", + comment="传输模式:pull/stream/both,声明账号支持的传输方式", + ) last_rotated_at = Column(DateTime, nullable=True, comment="凭据最近轮换时间") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") @@ -173,6 +180,7 @@ class ChannelAccount(Base): "last_error_at": format_utc_datetime(self.last_error_at), "last_health_check_at": format_utc_datetime(self.last_health_check_at), "transport_cursor": self.transport_cursor or "", + "transport_mode": self.transport_mode or "both", "last_rotated_at": format_utc_datetime(self.last_rotated_at), "version": self.version, "created_by": self.created_by, diff --git a/backend/test/unit/channels/plugins/wechat_woc/adapters/test_lifecycle_adapter.py b/backend/test/unit/channels/plugins/wechat_woc/adapters/test_lifecycle_adapter.py index 8d90c4cb..52324240 100644 --- a/backend/test/unit/channels/plugins/wechat_woc/adapters/test_lifecycle_adapter.py +++ b/backend/test/unit/channels/plugins/wechat_woc/adapters/test_lifecycle_adapter.py @@ -180,13 +180,13 @@ class TestApplyAccountConfig: assert exc_info.value.field == "bridge_url" @pytest.mark.asyncio - async def test_rejects_bridge_url_without_http_or_https(self): - # Arrange + async def test_accepts_bridge_url_without_http_or_https(self): + # Arrange - scheme 强制校验已移除,非 http/https 协议按域名处理 adapter = _make_adapter() - # Act / Assert - with pytest.raises(ValidationError) as exc_info: - await adapter.applyAccountConfig({"bridge_url": "ftp://woc.example.com"}, "acc-1") - assert exc_info.value.field == "bridge_url" + # Act + result = await adapter.applyAccountConfig({"bridge_url": "ftp://woc.example.com"}, "acc-1") + # Assert + assert result["bridge_url"] == "ftp://woc.example.com" @pytest.mark.unit diff --git a/backend/test/unit/channels/plugins/wechat_woc/adapters/test_wizard_adapter.py b/backend/test/unit/channels/plugins/wechat_woc/adapters/test_wizard_adapter.py index 6c3a7692..5365f751 100644 --- a/backend/test/unit/channels/plugins/wechat_woc/adapters/test_wizard_adapter.py +++ b/backend/test/unit/channels/plugins/wechat_woc/adapters/test_wizard_adapter.py @@ -138,24 +138,12 @@ class TestWeChatWocWizardAdapter: assert result.config_patch.values["bridge_token"] == _valid_bridge_config_values()["bridge_token"] @pytest.mark.asyncio - async def test_validateStep_bridge_config_invalid_when_url_not_https(self): - # Arrange - P0-4:wizard 阶段强制 https,ftp:// 被拒绝 - logger = _make_logger() - adapter = WeChatWocWizardAdapter(_make_client(), _make_config_port(), logger, _CHANNEL_TYPE) - values = {"bridge_url": "ftp://example.com", "bridge_token": "token"} - # Act - result = await adapter.validateStep("bridge_config", values) - # Assert - assert result.valid is False - assert "https_required_or_set_allow_insecure_localhost" in result.errors - - @pytest.mark.asyncio - async def test_validateStep_bridge_config_http_localhost_allowed_when_config_true(self): - # Arrange - 开发环境豁免:allow_insecure_localhost=true + localhost http + async def test_validateStep_bridge_config_http_localhost_allowed(self): + # Arrange - scheme 强制校验已移除,http localhost 直接通过 logger = _make_logger() adapter = WeChatWocWizardAdapter( _make_client(), - _make_config_port(allow_insecure=True), + _make_config_port(), logger, _CHANNEL_TYPE, ) @@ -167,23 +155,6 @@ class TestWeChatWocWizardAdapter: assert result.config_patch is not None assert result.config_patch.values["bridge_url"] == "http://localhost:8088" - @pytest.mark.asyncio - async def test_validateStep_bridge_config_http_localhost_rejected_when_config_false(self): - # Arrange - 未开启豁免时 localhost http 仍被拒绝 - logger = _make_logger() - adapter = WeChatWocWizardAdapter( - _make_client(), - _make_config_port(allow_insecure=False), - logger, - _CHANNEL_TYPE, - ) - values = {"bridge_url": "http://localhost:8088", "bridge_token": "token"} - # Act - result = await adapter.validateStep("bridge_config", values) - # Assert - assert result.valid is False - assert "https_required_or_set_allow_insecure_localhost" in result.errors - @pytest.mark.asyncio async def test_validateStep_verify_valid_when_bridge_and_wechat_ok(self): # Arrange diff --git a/backend/test/unit/channels/plugins/wechat_woc/test_lifecycle.py b/backend/test/unit/channels/plugins/wechat_woc/test_lifecycle.py index 53460bde..42fbbf01 100644 --- a/backend/test/unit/channels/plugins/wechat_woc/test_lifecycle.py +++ b/backend/test/unit/channels/plugins/wechat_woc/test_lifecycle.py @@ -135,17 +135,6 @@ class TestOnReconfigure: # Assert handler._logger.info.assert_awaited() - @pytest.mark.asyncio - async def test_http_non_localhost_raises_validation_error(self) -> None: - # Arrange - 生产环境强制 https:http:// 非 localhost 被拒绝 - handler = _make_handler() - - # Act / Assert - with pytest.raises(ValidationError) as exc_info: - await handler.onReconfigure(new_config={"bridge_url": "http://x.example.com"}) - assert exc_info.value.field == "bridge_url" - assert exc_info.value.message == "https_required_or_set_allow_insecure_localhost" - @pytest.mark.asyncio async def test_valid_https_bridge_url_succeeds(self) -> None: # Arrange @@ -157,6 +146,17 @@ class TestOnReconfigure: # Assert handler._logger.info.assert_awaited() + @pytest.mark.asyncio + async def test_http_bridge_url_succeeds(self) -> None: + # Arrange - 已移除 scheme 强制校验,http 域名同样通过 + handler = _make_handler() + + # Act + await handler.onReconfigure(new_config={"bridge_url": "http://x.example.com"}) + + # Assert + handler._logger.info.assert_awaited() + @pytest.mark.asyncio async def test_missing_bridge_url_raises(self) -> None: # Arrange - new_config 不含 bridge_url @@ -187,17 +187,6 @@ class TestOnReconfigure: await handler.onReconfigure(new_config={"bridge_url": 123}) assert exc_info.value.key == "bridge_url" - @pytest.mark.asyncio - async def test_bridge_url_without_protocol_raises(self) -> None: - # Arrange - bridge_url 非 https scheme(ftp:// 被拒绝) - handler = _make_handler() - - # Act / Assert - P0-4:scheme 校验由 _validate_bridge_url_scheme 抛 ValidationError - with pytest.raises(ValidationError) as exc_info: - await handler.onReconfigure(new_config={"bridge_url": "ftp://x.example.com"}) - assert exc_info.value.field == "bridge_url" - assert exc_info.value.message == "https_required_or_set_allow_insecure_localhost" - @pytest.mark.asyncio async def test_does_not_maintain_last_config_attribute(self) -> None: # Arrange / Act - 验证调用后不维护 _last_config 实例属性 diff --git a/backend/test/unit/channels/plugins/wechat_woc/test_url_validators.py b/backend/test/unit/channels/plugins/wechat_woc/test_url_validators.py index f26fa4ab..c45bac4a 100644 --- a/backend/test/unit/channels/plugins/wechat_woc/test_url_validators.py +++ b/backend/test/unit/channels/plugins/wechat_woc/test_url_validators.py @@ -1,8 +1,5 @@ """yuxi.channels.plugins.wechat_woc._url_validators 单元测试。 -P0-4:验证 ``_validate_bridge_url_scheme`` 的 https 强制策略与开发环境 -localhost 豁免逻辑。 - P1-7:验证 ``_redact_url`` 的日志脱敏行为(去除 query、保留 scheme://host:port/path)。 @@ -17,71 +14,11 @@ from yuxi.channels.contract.errors import ValidationError from yuxi.channels.plugins.wechat_woc._url_validators import ( _redact_url, _validate_bridge_url_host, - _validate_bridge_url_scheme, ) pytestmark = pytest.mark.unit -@pytest.mark.unit -class TestValidateBridgeUrlScheme: - """``_validate_bridge_url_scheme`` scheme 校验测试。""" - - def test_https_url_passes(self) -> None: - """https URL 始终通过,无论 allow_insecure 值。""" - _validate_bridge_url_scheme("https://example.com:8443/bridge", False) - _validate_bridge_url_scheme("https://localhost:8088", False) - _validate_bridge_url_scheme("https://192.168.1.10", True) - - def test_http_url_with_allow_insecure_false_raises(self) -> None: - """http URL + allow_insecure=False 必须抛 ValidationError。""" - with pytest.raises(ValidationError) as exc_info: - _validate_bridge_url_scheme("http://example.com:8088", False) - assert exc_info.value.field == "bridge_url" - assert exc_info.value.message == "https_required_or_set_allow_insecure_localhost" - - def test_http_localhost_with_allow_insecure_true_passes(self) -> None: - """http + allow_insecure=True + localhost 通过(开发环境豁免)。""" - _validate_bridge_url_scheme("http://localhost:8088", True) - - def test_http_127_0_0_1_with_allow_insecure_true_passes(self) -> None: - """http + allow_insecure=True + 127.0.0.1 通过(开发环境豁免)。""" - _validate_bridge_url_scheme("http://127.0.0.1:8088", True) - - def test_http_ipv6_loopback_with_allow_insecure_true_passes(self) -> None: - """http + allow_insecure=True + ::1 通过(开发环境豁免)。""" - _validate_bridge_url_scheme("http://[::1]:8088", True) - - def test_http_private_network_with_allow_insecure_true_raises(self) -> None: - """http + allow_insecure=True + 192.168.x.x 不豁免(内网部署需反代 https)。""" - with pytest.raises(ValidationError) as exc_info: - _validate_bridge_url_scheme("http://192.168.1.10:8088", True) - assert exc_info.value.field == "bridge_url" - - def test_http_non_localhost_with_allow_insecure_true_raises(self) -> None: - """http + allow_insecure=True + 非 localhost 域名 不豁免。""" - with pytest.raises(ValidationError) as exc_info: - _validate_bridge_url_scheme("http://bridge.example.com", True) - assert exc_info.value.field == "bridge_url" - - def test_invalid_scheme_raises(self) -> None: - """非 http/https scheme 抛 ValidationError。""" - with pytest.raises(ValidationError) as exc_info: - _validate_bridge_url_scheme("ftp://localhost:8088", True) - assert exc_info.value.field == "bridge_url" - assert exc_info.value.message == "https_required_or_set_allow_insecure_localhost" - - def test_trace_id_propagated_to_exception(self) -> None: - """trace_id 正确注入 ValidationError。""" - with pytest.raises(ValidationError) as exc_info: - _validate_bridge_url_scheme( - "http://example.com", - False, - trace_id="trace-abc-123", - ) - assert exc_info.value.trace_id == "trace-abc-123" - - @pytest.mark.unit class TestValidateBridgeUrlHost: """``_validate_bridge_url_host`` SSRF 防护测试。""" @@ -152,13 +89,7 @@ class TestValidateBridgeUrlHost: _validate_bridge_url_host("http://localhost:8088", True) def test_localhost_with_allow_insecure_false_passes_as_domain(self) -> None: - """localhost + allow_insecure=False 通过(域名放行,不视为 IP)。 - - localhost 是主机名而非 IP,``ipaddress.ip_address`` 抛 ValueError, - 域名分支放行。SSRF 防护对 localhost 的限制由 scheme 校验 - (``_validate_bridge_url_scheme``)在 allow_insecure=False 时拒绝 - http://localhost 来保证,host 校验本身不阻断域名。 - """ + """localhost + allow_insecure=False 通过(域名放行,不视为 IP)。""" _validate_bridge_url_host("https://localhost:8088", False) def test_trace_id_propagated_to_exception(self) -> None: diff --git a/web/src/composables/channels/useChannelDashboard.js b/web/src/composables/channels/useChannelDashboard.js index ec443b32..6301d3b4 100644 --- a/web/src/composables/channels/useChannelDashboard.js +++ b/web/src/composables/channels/useChannelDashboard.js @@ -8,7 +8,7 @@ * 数据域: * - overview (DSB-01):账户 / 消息 / 会话 / 投递四类 KPI 计数快照 * - delivery (DSB-DELIVERY):投递总数 / 成功率 / 延迟分位数 / 死信数 - * - realtime (DSB-REALTIME):纯内存态聚合,高频轮询绘制吞吐曲线 + * - realtime (DSB-REALTIME):实时监控指标,挂载/切换渠道/手动刷新时单次拉取 * - health (§3.1):聚合健康简略视图(无鉴权),含降级组件 * - todos :待审批配对(§10.1)/ 待审核内容(§19.2)/ 死信积压(§13.2) * @@ -24,7 +24,7 @@ * active_sessions, queue_depth, worker_utilization, by_channel } * - HealthSnapshot: { status, version, degraded_components, channels, ... } */ -import { ref, computed, onMounted, onUnmounted } from 'vue' +import { ref, computed, onMounted } from 'vue' import { useRoute, useRouter } from 'vue-router' import { getDashboardOverviewApi, @@ -35,7 +35,6 @@ import { import { getHealthApi, getHealthDetailApi, exportDiagnosticsApi } from '@/apis/channels/health_api' import { unwrap } from './utils' import { PENDING_RESTART_STORAGE_KEY } from './useChannelTodos' -import { useVisiblePolling } from './useVisiblePolling' // 时间范围预设:label → { ms, seconds },用于推算 start_time/end_time const TIME_RANGE_PRESETS = { @@ -44,14 +43,14 @@ const TIME_RANGE_PRESETS = { '7d': { label: '最近 7 天', ms: 7 * 24 * 60 * 60 * 1000 } } -// 实时轮询间隔(毫秒),与 60s 窗口匹配,避免过频打满后端 -const REALTIME_POLL_INTERVAL = 5000 -// 吞吐曲线保留样本数(5s × 60 = 5 分钟滚动窗口) +// 吞吐曲线保留样本数(原 5s 轮询 × 60 = 5 分钟滚动窗口;关闭自动轮询后仅用于单次刷新时限制样本量) const REALTIME_MAX_SAMPLES = 60 export function useChannelDashboard() { - // 请求版本号:用于竞态保护,每次发起新请求递增,旧请求结果可通过版本号校验丢弃 - let reqId = 0 + // 请求版本号:拆分为 dashboard 主数据与 realtime 轮询两个独立计数器, + // 避免 fetchRealtime 的 5s 轮询触发 ++reqId 导致正在进行的 fetchAll 结果被丢弃。 + let dashboardReqId = 0 + let realtimeReqId = 0 // ===== 过滤条件 ===== const timeRange = ref('24h') @@ -98,11 +97,23 @@ export function useChannelDashboard() { const realtimeSamples = ref([]) // ===== 加载标志 ===== - const loading = ref(false) - const realtimeLoading = ref(false) + // 按数据域拆分:避免单个接口慢导致整页骨架屏 + const overviewLoading = ref(false) + const deliveryLoading = ref(false) const healthLoading = ref(false) + const todosLoading = ref(false) + const realtimeLoading = ref(false) const probeLoading = ref(false) const exportLoading = ref(false) + // 兼容旧用法:任一主数据域加载中即为 true + const loading = computed( + () => + overviewLoading.value || + deliveryLoading.value || + healthLoading.value || + todosLoading.value || + realtimeLoading.value + ) // ===== 错误(按域聚合) ===== // 结构:[{ key, message }],任一接口失败不影响其他域数据写入 @@ -130,21 +141,38 @@ export function useChannelDashboard() { // ===== 主数据拉取(overview + delivery + health + todos 并发) ===== // 使用 Promise.allSettled:任一接口失败不影响其他域,错误聚合到 error.value async function fetchAll() { - const myId = ++reqId - loading.value = true + const myId = ++dashboardReqId error.value = null + overviewLoading.value = true + deliveryLoading.value = true + healthLoading.value = true + todosLoading.value = true const dashboardParams = buildDashboardParams() const tasks = [ - { key: 'overview', fn: () => getDashboardOverviewApi(dashboardParams) }, - { key: 'delivery', fn: () => getDashboardDeliveryApi(dashboardParams) }, - { key: 'health', fn: () => getHealthApi() }, - { key: 'todos', fn: fetchTodos } + { + key: 'overview', + fn: () => getDashboardOverviewApi(dashboardParams), + loading: overviewLoading + }, + { + key: 'delivery', + fn: () => getDashboardDeliveryApi(dashboardParams), + loading: deliveryLoading + }, + { key: 'health', fn: () => getHealthApi(), loading: healthLoading }, + { key: 'todos', fn: fetchTodos, loading: todosLoading } ] const results = await Promise.allSettled(tasks.map((t) => t.fn())) - // 版本号校验:如果期间有更新的请求发出,丢弃本次结果 - if (myId !== reqId) return + // 版本号校验:如果期间有新的 dashboard 请求发出,丢弃本次结果 + if (myId !== dashboardReqId) { + // 被新请求取代时关闭本请求打开的 loading,避免遗留骨架屏 + tasks.forEach((t) => { + t.loading.value = false + }) + return + } const errors = [] results.forEach((result, i) => { const { key } = tasks[i] @@ -164,7 +192,10 @@ export function useChannelDashboard() { // overview / health 拉取完成后,派生 access-ops 异常计数到 todos // 放在 fetchAll 末尾确保数据已就绪(fetchTodos 与 overview/health 并发,时序不保证) syncAccessOpsTodos() - loading.value = false + // 各自关闭 loading:先完成的域先渲染,避免被长尾请求阻塞 + tasks.forEach((t) => { + t.loading.value = false + }) } // ===== 从已拉的 overview/health 派生 access-ops 异常计数 ===== @@ -207,13 +238,13 @@ export function useChannelDashboard() { // ===== 实时监控单次拉取 ===== async function fetchRealtime() { - const myId = ++reqId + const myId = ++realtimeReqId realtimeLoading.value = true try { const params = { window_seconds: 60 } if (channelType.value) params.channel_type = channelType.value const res = await getDashboardRealtimeApi(params) - if (myId !== reqId) return + if (myId !== realtimeReqId) return const data = unwrap(res) realtime.value = data ?? null // 追加吞吐曲线样本 @@ -223,21 +254,14 @@ export function useChannelDashboard() { realtimeSamples.value.shift() } } catch (e) { - if (myId !== reqId) return + if (myId !== realtimeReqId) return // 实时监控失败不阻塞主数据,仅记录到 error appendError('realtime', e?.message || String(e)) } finally { - if (myId === reqId) realtimeLoading.value = false + if (myId === realtimeReqId) realtimeLoading.value = false } } - // ===== 实时轮询生命周期(可见性感知,#1 修复)===== - // 5s 高频轮询,标签页隐藏时停止打接口,切回立即拉一次恢复曲线 - const { start: startRealtimePolling, stop: stopRealtimePolling } = useVisiblePolling( - fetchRealtime, - REALTIME_POLL_INTERVAL - ) - function resetRealtimeSamples() { realtimeSamples.value = [] } @@ -285,8 +309,9 @@ export function useChannelDashboard() { // ===== 切换时间范围:重拉受时间影响的域(overview.messages + delivery) ===== // 账户/会话/outbox 为当前快照,不受时间范围影响 async function onTimeRangeChange() { - const myId = ++reqId - loading.value = true + const myId = ++dashboardReqId + overviewLoading.value = true + deliveryLoading.value = true clearErrorByKey('overview') clearErrorByKey('delivery') try { @@ -295,8 +320,8 @@ export function useChannelDashboard() { getDashboardOverviewApi(params), getDashboardDeliveryApi(params) ]) - // 版本号校验:如果期间有更新的请求发出,丢弃本次结果 - if (myId !== reqId) return + // 版本号校验:如果期间有新的 dashboard 请求发出,丢弃本次结果 + if (myId !== dashboardReqId) return if (overviewRes.status === 'fulfilled') { overview.value = unwrap(overviewRes.value) ?? null } else { @@ -308,20 +333,22 @@ export function useChannelDashboard() { appendError('delivery', deliveryRes.reason?.message || String(deliveryRes.reason)) } } finally { - if (myId === reqId) loading.value = false + if (myId === dashboardReqId) { + overviewLoading.value = false + deliveryLoading.value = false + } } syncToRoute() } // ===== 切换渠道过滤:全量重拉(所有域均受渠道过滤影响) ===== async function onChannelChange() { - const myId = ++reqId // 切换渠道时置空 realtime,避免显示旧渠道数据 realtime.value = null resetRealtimeSamples() await fetchAll() - if (myId !== reqId) return - startRealtimePolling() + // 关闭自动轮询后,切换渠道时手动拉一次 realtime,确保展示当前渠道数据 + await fetchRealtime() syncToRoute() } @@ -355,6 +382,15 @@ export function useChannelDashboard() { // ===== 分域重试:只刷新失败的域,避免局部错误触发全局刷新 ===== async function refreshDomain(key) { clearErrorByKey(key) + const loadingMap = { + overview: overviewLoading, + delivery: deliveryLoading, + health: healthLoading, + todos: todosLoading, + realtime: realtimeLoading + } + const domainLoading = loadingMap[key] + if (domainLoading) domainLoading.value = true try { const params = buildDashboardParams() if (key === 'overview') { @@ -377,6 +413,8 @@ export function useChannelDashboard() { } } catch (e) { appendError(key, e?.message || String(e)) + } finally { + if (domainLoading) domainLoading.value = false } } @@ -396,23 +434,21 @@ export function useChannelDashboard() { pendingRestartPlugins: 0 } realtimeSamples.value = [] - loading.value = false - realtimeLoading.value = false + overviewLoading.value = false + deliveryLoading.value = false healthLoading.value = false + todosLoading.value = false + realtimeLoading.value = false probeLoading.value = false exportLoading.value = false error.value = null } - // ===== 生命周期:挂载拉取主数据 + 启动实时轮询;卸载清理定时器 ===== + // ===== 生命周期:挂载拉取主数据 + 单次 realtime;关闭自动轮询 ===== initFromRoute() onMounted(() => { fetchAll() - startRealtimePolling() - }) - - onUnmounted(() => { - stopRealtimePolling() + fetchRealtime() }) return { @@ -429,8 +465,11 @@ export function useChannelDashboard() { realtimeSamples, // 加载标志 loading, - realtimeLoading, + overviewLoading, + deliveryLoading, healthLoading, + todosLoading, + realtimeLoading, probeLoading, exportLoading, // 错误 @@ -451,9 +490,7 @@ export function useChannelDashboard() { // 过滤变更 onTimeRangeChange, onChannelChange, - // 实时轮询 - startRealtimePolling, - stopRealtimePolling, + // 单次 realtime 刷新辅助 resetRealtimeSamples, // 重置 reset