fix(notification): prevent untyped message broadcasts

This commit is contained in:
jxxghp
2026-09-20 11:32:40 +08:00
parent fa1df511e1
commit 2e578a69ee
9 changed files with 202 additions and 37 deletions

View File

@@ -44,12 +44,12 @@ def get_notification_configs(
)
def get_notification_switch(mtype: MessageType) -> Optional[str]:
"""返回指定通知场景的目标范围。"""
def get_notification_switch(mtype: MessageType) -> str:
"""返回指定通知场景的目标范围,缺少配置时默认仅管理员。"""
for switch in get_service_configs(
SystemConfigKey.NotificationSwitchs,
NotificationSwitchConf,
):
if switch.type == mtype.value:
return switch.action
return None
return switch.action or "admin"
return "admin"

View File

@@ -69,48 +69,100 @@ def _notification_route_steps(
生成同步和异步通知共用的路由状态转换。
生成器只表达用户设置查询、日志和投递决策,调用方分别执行真实同步或
异步 I/O,避免两套外壳各自维护管理员回退和原消息投递规则。
异步 I/O,避免两套外壳各自维护管理员回退和原消息投递规则。没有可用的
用户通知设置时只记录日志,不产出无目标消息,避免渠道默认广播。
"""
admin_sent = False
send_original = not action
for route_action in action.split(",") if action else ():
send_original = action is None
route_actions: tuple[str, ...] = ()
if action is not None:
route_actions = tuple(
route_action.strip() or "admin"
for route_action in action.split(",")
)
for route_action in route_actions:
routed_message = copy.deepcopy(message)
if route_action == "admin" and not admin_sent:
settings = yield _NotificationRouteLookup(
username=superuser,
log_message=f"{routed_message.mtype} 的消息已设置发送给管理员",
)
routed_message.targets = cast(dict[str, Any], settings)
admin_sent = True
elif route_action == "user" and routed_message.username:
username = cast(str, routed_message.username)
settings = yield _NotificationRouteLookup(
username=username,
log_message=(
f"{routed_message.mtype} 的消息已设置发送给用户 {username}"
),
)
routed_message.targets = cast(dict[str, Any], settings)
if settings is None:
if not admin_sent:
settings = yield _NotificationRouteLookup(
username=superuser,
log_message=(
f"用户 {username} 不存在,消息将发送给管理员"
),
)
routed_message.targets = cast(dict[str, Any], settings)
admin_sent = True
else:
yield _NotificationRouteLog(
message="管理员未配置通知渠道,消息不会广播发送"
)
continue
routed_message.targets = cast(dict[str, Any], settings)
elif route_action == "user":
if not routed_message.username:
if admin_sent:
yield _NotificationRouteLog(
message=f"用户 {username} 不存在,消息无法发送到对应用户"
message="消息未关联用户,且管理员已收到消息"
)
continue
elif username == superuser:
settings = yield _NotificationRouteLookup(
username=superuser,
log_message="消息未关联用户,消息将发送给管理员",
)
admin_sent = True
else:
if settings is None:
yield _NotificationRouteLog(
message="管理员未配置通知渠道,消息不会广播发送"
)
continue
routed_message.targets = cast(dict[str, Any], settings)
else:
username = cast(str, routed_message.username)
settings = yield _NotificationRouteLookup(
username=username,
log_message=(
f"{routed_message.mtype} 的消息已设置发送给用户 {username}"
),
)
routed_message.targets = cast(dict[str, Any], settings)
if settings is None:
if not admin_sent:
settings = yield _NotificationRouteLookup(
username=superuser,
log_message=(
f"用户 {username} 不存在,消息将发送给管理员"
),
)
routed_message.targets = cast(dict[str, Any], settings)
admin_sent = True
if settings is None:
yield _NotificationRouteLog(
message="管理员未配置通知渠道,消息不会广播发送"
)
continue
else:
yield _NotificationRouteLog(
message=f"用户 {username} 不存在,消息无法发送到对应用户"
)
continue
elif username == superuser:
admin_sent = True
elif route_action == "all":
send_original = not admin_sent
break
else:
yield _NotificationRouteLog(
message=f"未知通知路由 {route_action},消息将按管理员范围发送"
)
if admin_sent:
continue
settings = yield _NotificationRouteLookup(
username=superuser,
log_message="通知路由无效,消息将发送给管理员",
)
admin_sent = True
if settings is None:
yield _NotificationRouteLog(
message="管理员未配置通知渠道,消息不会广播发送"
)
continue
routed_message.targets = cast(dict[str, Any], settings)
yield _NotificationRouteDelivery(
message=routed_message,
immediately=False,
@@ -145,10 +197,12 @@ def _render_notification_message(
def _notification_route_action(message: Message) -> Optional[str]:
"""仅为未绑定真实用户的业务通知读取隔离路由配置。"""
if message.userid or not message.mtype:
"""为未绑定真实用户的通知确定隔离路由,缺少类型时默认仅管理员。"""
if message.userid:
return None
return get_notification_switch(message.mtype)
if not message.mtype:
return "admin"
return get_notification_switch(message.mtype) or "admin"
class MessageProcessingMixin:

View File

@@ -262,6 +262,9 @@ class _MessageBase(ServiceBase[TService, NotificationConf]):
# 检查消息来源
if message.source and message.source != source:
return False
# 没有类型且没有显式目标时不得进入渠道的默认广播路径。
if not message.userid and not message.mtype:
return bool(message.targets)
# 不是定向发送时,检查消息类型开关
if not message.userid and message.mtype:
conf = self.get_config(source)

View File

@@ -87,9 +87,9 @@ class ServiceConfigHelper:
)
@staticmethod
def get_notification_switch(mtype: MessageType) -> Optional[str]:
"""返回指定通知场景的目标范围。"""
def get_notification_switch(mtype: MessageType) -> str:
"""返回指定通知场景的目标范围,缺少配置时默认仅管理员。"""
for switch in ServiceConfigHelper.get_notification_switches():
if switch.type == mtype.value:
return switch.action
return None
return switch.action or "admin"
return "admin"

View File

@@ -47,3 +47,4 @@ def test_named_service_config_helpers_preserve_enabled_policy(monkeypatch) -> No
item.name for item in get_notification_configs(include_disabled=True)
] == ["enabled-channel", "disabled-channel"]
assert get_notification_switch(MessageType.Download) == "admin"
assert get_notification_switch(MessageType.Agent) == "admin"

View File

@@ -182,6 +182,11 @@ def test_notification_post_message_is_persisted_without_sse_queue(monkeypatch) -
chain = ChainBase()
# messagequeue 是全局单例,用 monkeypatch 避免用例间污染
monkeypatch.setattr(
chain.user_repository,
"get_notification_settings",
lambda _username: {"telegram_userid": "admin-1"},
)
send_message = Mock()
monkeypatch.setattr(chain.messagequeue, "send_message", send_message)
monkeypatch.setattr(chain.eventmanager, "send_event", Mock())
@@ -212,6 +217,11 @@ def test_agent_notification_post_message_is_persisted_without_sse_queue(monkeypa
chain = ChainBase()
# messagequeue 是全局单例,用 monkeypatch 避免用例间污染
monkeypatch.setattr(
chain.user_repository,
"get_notification_settings",
lambda _username: {"telegram_userid": "admin-1"},
)
send_message = Mock()
monkeypatch.setattr(chain.messagequeue, "send_message", send_message)
monkeypatch.setattr(chain.eventmanager, "send_event", Mock())
@@ -240,6 +250,11 @@ def test_transient_notification_post_message_skips_history_but_dispatches(monkey
chain = ChainBase()
# messagequeue 是全局单例,用 monkeypatch 避免用例间污染
monkeypatch.setattr(
chain.user_repository,
"get_notification_settings",
lambda _username: {"telegram_userid": "admin-1"},
)
send_message = Mock()
monkeypatch.setattr(chain.messagequeue, "send_message", send_message)
send_event = Mock()

View File

@@ -6,7 +6,9 @@ from unittest.mock import AsyncMock, Mock
import pytest
from app.chain._messaging import MessageProcessingMixin, NotificationMixin
from app.modules.wechat import WechatModule
from app.schemas.message import Message
from app.schemas.system import NotificationConf
from app.schemas.types import MessageType
@@ -175,3 +177,79 @@ def test_notification_routing_lookup_failure_sync_async_parity(monkeypatch) -> N
async_chain.messagequeue.async_send_message.assert_not_awaited()
assert sync_chain.messageoper.add.call_count == 1
assert async_chain.messageoper.async_add.await_count == 1
def test_untyped_notification_defaults_to_admin_sync_async(monkeypatch) -> None:
"""未设置消息类型的通知应同步、异步都只查询管理员并按其设置投递。"""
monkeypatch.setattr(
"app.chain._messaging.get_notification_switch",
lambda _mtype: None,
)
settings = {"admin": {"wechat_userid": "admin-1"}}
sync_repository = _NotificationSettingsRepository(settings)
async_repository = _NotificationSettingsRepository(settings)
sync_chain = _NotificationHarness(sync_repository)
async_chain = _NotificationHarness(async_repository)
message = Message(title="工作流结果", text="执行完成")
sync_chain.post_message(message.model_copy(deep=True))
asyncio.run(async_chain.async_post_message(message.model_copy(deep=True)))
assert sync_repository.sync_calls == ["admin"]
assert async_repository.async_calls == ["admin"]
expected_targets = [{"wechat_userid": "admin-1"}]
assert [
item[0]["targets"]
for item in _delivery_snapshot(sync_chain.messagequeue.send_message)
] == expected_targets
assert [
item[0]["targets"]
for item in _delivery_snapshot(async_chain.messagequeue.async_send_message)
] == expected_targets
def test_user_route_without_username_falls_back_to_admin(monkeypatch) -> None:
"""用户范围通知缺少责任人时应回退管理员,不得恢复为全体广播。"""
monkeypatch.setattr(
"app.chain._messaging.get_notification_switch",
lambda _mtype: "user,all",
)
repository = _NotificationSettingsRepository(
{"admin": {"telegram_userid": "admin-1"}}
)
chain = _NotificationHarness(repository)
chain.post_message(Message(mtype=MessageType.Download, title="下载完成"))
assert repository.sync_calls == ["admin"]
deliveries = _delivery_snapshot(chain.messagequeue.send_message)
assert [item[0]["targets"] for item in deliveries] == [
{"telegram_userid": "admin-1"}
]
assert [item[1] for item in deliveries] == [False]
def test_wechat_rejects_untyped_broadcast_without_explicit_targets(monkeypatch) -> None:
"""企业微信渠道不得把无类型且无目标的消息交给客户端默认广播。"""
module = WechatModule()
config = NotificationConf(name="wechat-test", type="wechat", enabled=True)
module._configs = {config.name: config}
client = Mock()
monkeypatch.setattr(module, "get_configs", lambda: {config.name: config})
monkeypatch.setattr(module, "get_instance", lambda _name: client)
module.post_message(
Message(source=config.name, title="工作流结果", text="执行完成")
)
client.send_msg.assert_not_called()
module.post_message(
Message(
source=config.name,
title="工作流结果",
text="执行完成",
targets={"wechat_userid": "admin-1"},
)
)
client.send_msg.assert_called_once()
assert client.send_msg.call_args.kwargs["userid"] == "admin-1"

View File

@@ -458,6 +458,11 @@ def test_strict_notification_preserves_legacy_provider_signature(monkeypatch) ->
chain = CommandChain()
received = []
monkeypatch.setattr(
chain.user_repository,
"get_notification_settings",
lambda _username: {"telegram_userid": "admin-1"},
)
def legacy_provider(message) -> None:
"""模拟只接受旧式单参数签名的第三方通知 provider。"""
received.append((message, get_correlation_id()))
@@ -488,6 +493,11 @@ def test_strict_notification_retry_writes_history_once(monkeypatch) -> None:
history_sources = set()
provider_sources = []
monkeypatch.setattr(
chain.user_repository,
"get_notification_settings",
lambda _username: {"telegram_userid": "admin-1"},
)
monkeypatch.setattr(chain.eventmanager, "send_event", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
chain.messageoper,

View File

@@ -42,6 +42,10 @@ class TestSystemNotificationDispatch(unittest.TestCase):
), patch.object(chain.messageoper, "add"), patch.object(
chain.eventmanager, "send_event"
) as send_event, patch.object(
chain.user_repository,
"get_notification_settings",
return_value={"telegram_userid": "admin-1"},
), patch.object(
chain.messagequeue, "send_message"
) as send_message:
chain.post_message(message)