From 591a1d420dd06233c8e69a9ae331c53e9bca9a89 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Thu, 13 Aug 2026 08:20:14 +0800 Subject: [PATCH] fix(agent): close secret confirmation delivery gaps (#6285) --- app/agent/__init__.py | 136 +++++--- app/agent/middleware/policy.py | 40 ++- app/agent/prompt/System Core Prompt.txt | 1 + app/agent/tools/impl/query_system_settings.py | 5 +- app/api/endpoints/agent.py | 14 +- app/chain/message.py | 1 - app/modules/feishu/__init__.py | 18 + app/modules/telegram/__init__.py | 1 + app/modules/telegram/telegram.py | 48 ++- app/schemas/message.py | 4 +- tests/test_agent_prompt_secrets.py | 17 + tests/test_agent_secret_confirmation.py | 328 +++++++++++++++++- tests/test_agent_tool_policy.py | 34 +- tests/test_feishu.py | 53 +++ tests/test_telegram.py | 33 ++ tests/test_web_agent_stream.py | 20 +- 16 files changed, 672 insertions(+), 81 deletions(-) diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 05aaf8770..bfc13eae5 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -317,7 +317,6 @@ class _PendingSecretConfirmation: user_id: str channel: str source: str - original_chat_id: str class MoviePilotAgent: @@ -339,7 +338,7 @@ class MoviePilotAgent: replay_mode: ReplyMode = ReplyMode.DISPATCH, allow_message_tools: bool = True, output_callback: Optional[Callable[[str], None]] = None, - protected_output_callback: Optional[Callable[[str], None]] = None, + protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, ): self.session_id = session_id self.user_id = user_id @@ -771,7 +770,7 @@ class MoviePilotAgent: def set_protected_output_callback( self, - protected_output_callback: Optional[Callable[[str], None]], + protected_output_callback: Optional[Callable[[str], Optional[bool]]], ) -> None: """更新仅供当前请求接收的受保护文本输出回调。""" self.protected_output_callback = protected_output_callback @@ -790,7 +789,7 @@ class MoviePilotAgent: """判断当前渠道能否把密钥结果直接交付给原用户。""" if self.channel == MessageChannel.WebAgent.value: return callable(self.protected_output_callback) - return self.channel in { + return bool(self.user_id and self.source) and self.channel in { MessageChannel.Telegram.value, MessageChannel.Feishu.value, } @@ -831,15 +830,6 @@ class MoviePilotAgent: self._pending_secret_confirmation = None return permission_result - self._pending_secret_confirmation = _PendingSecretConfirmation( - tool=tool, - arguments=validated_arguments, - created_at=datetime.now(), - user_id=str(self.user_id or ""), - channel=str(self.channel or ""), - source=str(self.source or ""), - original_chat_id=str(self.original_chat_id or ""), - ) target = validated_arguments.get("setting_key") or ( validated_arguments.get("group") or "all" ) @@ -849,34 +839,73 @@ class MoviePilotAgent: "请在 5 分钟内回复“确认”继续，或回复“取消”放弃。" ) if self.channel == MessageChannel.WebAgent.value: + self._pending_secret_confirmation = _PendingSecretConfirmation( + tool=tool, + arguments=validated_arguments, + created_at=datetime.now(), + user_id=str(self.user_id or ""), + channel=str(self.channel or ""), + source=str(self.source or ""), + ) self._emit_output(confirmation_message) else: - await self.send_agent_message(confirmation_message) + delivered = await self._deliver_private_channel_message( + confirmation_message + ) + if not delivered: + self._pending_secret_confirmation = None + return "无法向当前用户建立私聊，未执行敏感设置读取。" + self._pending_secret_confirmation = _PendingSecretConfirmation( + tool=tool, + arguments=validated_arguments, + created_at=datetime.now(), + user_id=str(self.user_id or ""), + channel=str(self.channel or ""), + source=str(self.source or ""), + ) self._tool_context["user_reply_sent"] = True return confirmation_message - async def _deliver_protected_output(self, content: str) -> None: + async def _deliver_private_channel_message(self, content: str) -> bool: + """按渠道用户身份私聊投递，禁止回退群聊或广播。""" + if self.channel not in { + MessageChannel.Telegram.value, + MessageChannel.Feishu.value, + }: + return False + try: + response = await run_in_threadpool( + AgentChain().send_direct_message, + Notification( + channel=self.channel, + source=self.source, + mtype=NotificationType.Agent, + userid=self.user_id, + username=self.username, + text=content, + private_delivery=True, + parse_mode="plain", + save_history=False, + ), + ) + except Exception as error: + logger.error( + f"Agent私聊投递失败: channel={self.channel}, " + f"error_type={type(error).__name__}" + ) + return False + return bool(response and response.success) + + async def _deliver_protected_output(self, content: str) -> bool: """绕过模型与会话历史，把敏感结果直接交付给当前用户。""" if callable(self.protected_output_callback): try: - self.protected_output_callback(content) + delivered = self.protected_output_callback(content) except Exception as e: logger.error(f"受保护输出回调失败: {e}") - return - - await AgentChain().async_post_message( - Notification( - channel=self.channel, - source=self.source, - mtype=NotificationType.Agent, - userid=self.user_id, - username=self.username, - original_message_id=self.original_message_id, - original_chat_id=self.original_chat_id, - text=content, - save_history=False, - ) - ) + return False + return delivered is not False + return await self._deliver_private_channel_message(content) async def _handle_secret_confirmation_control( self, @@ -899,7 +928,6 @@ class MoviePilotAgent: pending.user_id != str(self.user_id or "") or pending.channel != str(self.channel or "") or pending.source != str(self.source or "") - or pending.original_chat_id != str(self.original_chat_id or "") ): return None if datetime.now() - pending.created_at > SECRET_CONFIRMATION_TTL: @@ -913,16 +941,37 @@ class MoviePilotAgent: await self._deliver_protected_output(message_text) return message_text - permission_result = await pending.tool._check_permission() - if permission_result: - await self._deliver_protected_output(permission_result) - return permission_result - if not self._can_confirm_secret_read(): return "当前入口不支持安全交付敏感设置，未执行读取。" - result = await pending.tool._run_confirmed(**pending.arguments) - await self._deliver_protected_output(result) + async def _execute_confirmed() -> str: + permission_result = await pending.tool._check_permission() + if permission_result: + return permission_result + return await pending.tool._run_confirmed(**pending.arguments) + + policy = AgentPolicyMiddleware( + context=self._build_policy_context(), + tools=[pending.tool], + ) + try: + _, result = await policy.execute_tool_call( + tool=pending.tool, + arguments=pending.arguments, + invocation_id=f"secret-confirmation-{uuid.uuid4().hex}", + handler=_execute_confirmed, + ) + except Exception: + message_text = "敏感设置读取失败，请稍后重试。" + await self._deliver_protected_output(message_text) + return message_text + delivered = await self._deliver_protected_output(result) + if not delivered: + message_text = "敏感设置读取已完成，但结果投递失败，请重新发起。" + self._emit_output(message_text) + if self.should_dispatch_reply: + await self.send_agent_message(message_text) + return message_text return "敏感设置确认已处理。" def _build_policy_context(self) -> ToolPolicyContext: @@ -2166,7 +2215,7 @@ class _MessageTask: reply_mode: ReplyMode = ReplyMode.DISPATCH allow_message_tools: bool = True output_callback: Optional[Callable[[str], None]] = None - protected_output_callback: Optional[Callable[[str], None]] = None + protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None notification_callback: Optional[Callable[[Any], None]] = None agent_factory: Optional[Callable[..., MoviePilotAgent]] = None completion_future: Optional[asyncio.Future] = None @@ -2227,7 +2276,6 @@ class AgentManager: user_id: str, channel: Optional[str] = None, source: Optional[str] = None, - original_chat_id: Optional[str] = None, ) -> bool: """判断指定用户是否可继续当前会话的敏感设置确认。""" agent = self.active_agents.get(session_id) @@ -2238,10 +2286,6 @@ class AgentManager: and str(agent.user_id) == str(user_id) and (channel is None or pending.channel == str(channel)) and (source is None or pending.source == str(source)) - and ( - original_chat_id is None - or pending.original_chat_id == str(original_chat_id) - ) ) async def initialize(self): @@ -2345,7 +2389,7 @@ class AgentManager: reply_mode: ReplyMode = ReplyMode.DISPATCH, allow_message_tools: bool = True, output_callback: Optional[Callable[[str], None]] = None, - protected_output_callback: Optional[Callable[[str], None]] = None, + protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, notification_callback: Optional[Callable[[Any], None]] = None, agent_factory: Optional[Callable[..., MoviePilotAgent]] = None, wait_for_completion: bool = False, diff --git a/app/agent/middleware/policy.py b/app/agent/middleware/policy.py index f66d2af09..c74c587ea 100644 --- a/app/agent/middleware/policy.py +++ b/app/agent/middleware/policy.py @@ -17,6 +17,9 @@ from app.agent.tools.catalog import ToolCatalogSnapshot from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool +POLICY_DENIED_MESSAGE = "当前宿主策略不允许执行该工具。" + + class AgentPolicyMiddleware(AgentMiddleware): """观测进入本地 ToolNode 的 client-side 工具调用和结果。 @@ -103,16 +106,43 @@ class AgentPolicyMiddleware(AgentMiddleware): arguments = tool_call.get("args") or {} if not isinstance(arguments, dict): arguments = {} + _, result = await self.execute_tool_call( + tool=request.tool, + arguments=arguments, + invocation_id=tool_call.get("id"), + handler=lambda: handler(request), + enforce_decision=False, + ) + # 普通 ToolNode 在严格策略接管前保持 shadow 观测语义。 + # 已确认的受保护调用会使用默认的强制决策语义。 + return result + + async def execute_tool_call( + self, + *, + tool: Any, + arguments: dict[str, Any], + handler: Callable[[], Awaitable[Any]], + invocation_id: str | None = None, + enforce_decision: bool = True, + ) -> tuple[bool, Any]: + """执行一次本地工具调用，并复用 ToolNode 的策略生命周期。""" observation = call_policy_hook( "start", self.orchestrator.start, context=self.context, - tool=request.tool, + tool=tool, arguments=arguments, - invocation_id=tool_call.get("id"), + invocation_id=invocation_id, ) + if ( + enforce_decision + and observation is not None + and observation.decision.allowed is False + ): + return False, POLICY_DENIED_MESSAGE try: - result = await handler(request) + result = await handler() except Exception as error: if observation is not None: call_policy_hook( @@ -129,7 +159,7 @@ class AgentPolicyMiddleware(AgentMiddleware): observation, result, ) - return result + return True, result -__all__ = ["AgentPolicyMiddleware"] +__all__ = ["AgentPolicyMiddleware", "POLICY_DENIED_MESSAGE"] diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt index 11ebe1c2d..7f78ce1b4 100644 --- a/app/agent/prompt/System Core Prompt.txt +++ b/app/agent/prompt/System Core Prompt.txt @@ -21,6 +21,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel - Do not stop for approval on read-only operations. +- Raw secret reads are protected operations rather than ordinary read-only queries. When a user explicitly asks for a raw credential or another unredacted sensitive setting, call `query_system_settings` with `show_secrets=true`; do not refuse the request solely because the value is sensitive. The host verifies administrator authority, obtains any required confirmation, and delivers the result through a protected channel. Never expose or repeat the secret in an ordinary assistant response, tool narration, or follow-up model context. - If the user has not explicitly requested an operation that changes system behavior, ask for confirmation before proceeding. This includes modifying system settings, updating plugin configuration, reloading plugins, running restart/stop/start commands, or triggering slash commands such as `/restart`. - Always get explicit consent before destructive or high-impact actions such as starting downloads, deleting subscriptions, deleting download tasks or files, removing history, installing/uninstalling plugins, changing site authentication, changing scheduler or workflow execution state, restarting services, or stopping services. - When the user explicitly asks for delayed, recurring, reminder, or monitoring work, use `create_agent_task` instead of promising to remember it or writing a JOB.md file. Use a `date` trigger with `delay_minutes` for requests such as "in 30 minutes", an exact `date` trigger for other single future runs, and a five-field `cron` trigger for recurring work. Manage existing autonomous tasks with `query_agent_tasks`, `update_agent_task`, `run_agent_task`, and `delete_agent_task`; these tools use integer `task_id` values. Use `query_schedulers` and `run_scheduler` only for MoviePilot system, plugin, or workflow runtime services, whose string `job_id` values must never be passed to autonomous-task tools. diff --git a/app/agent/tools/impl/query_system_settings.py b/app/agent/tools/impl/query_system_settings.py index 19e2609ff..01a35bc0c 100644 --- a/app/agent/tools/impl/query_system_settings.py +++ b/app/agent/tools/impl/query_system_settings.py @@ -58,7 +58,10 @@ class QuerySystemSettingsInput(BaseModel): False, description=( "Whether to return raw secret values such as API keys, tokens, cookies, and passwords. " - "Defaults to false; secret-like fields are redacted in returned values and previews." + "Defaults to false; secret-like fields are redacted in returned values and previews. " + "Set this to true when the user explicitly asks for an unredacted secret; the host verifies " + "administrator authority, requests confirmation, and delivers the result outside the ordinary " + "model response. Do not refuse the tool call solely because the requested value is sensitive." ), ) diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index 364ea6f7a..c5d693998 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -84,10 +84,10 @@ class _WebAgentEventPublisher: """返回本轮发布器观测到的最大积压深度。""" return self._max_depth - def publish(self, event: dict) -> None: - """发布事件；相邻文本会按时间或长度边界合并。""" + def publish(self, event: dict) -> bool: + """发布事件；返回关闭状态以便受保护投递能准确报告失败。""" if self._disposed: - return + return False if event.get("type") == "delta": self._pending_delta += str(event.get("content") or "") if len(self._pending_delta) >= WEB_AGENT_STREAM_COALESCE_MAX_CHARS: @@ -98,10 +98,11 @@ class _WebAgentEventPublisher: WEB_AGENT_STREAM_COALESCE_SECONDS, self._flush_delta, ) - return + return True self._flush_delta() self._append_event(event) + return True async def get(self) -> dict: """等待并返回下一条已排序事件。""" @@ -1955,7 +1956,6 @@ async def web_agent_stream( str(current_user.id), channel=MessageChannel.WebAgent.value, source=WEB_AGENT_SOURCE, - original_chat_id=str(payload.original_chat_id or ""), ) ) protected_transport_supported = ( @@ -2177,9 +2177,9 @@ async def web_agent_stream( _apply_web_agent_display_event(item, assistant_display_message) event_publisher.publish(item) - def protected_output_callback(content: str) -> None: + def protected_output_callback(content: str) -> bool: """将敏感文本封装为不进入普通展示快照的命名 SSE 事件。""" - event_publisher.publish( + return event_publisher.publish( { "type": "interaction-protected", "content": content, diff --git a/app/chain/message.py b/app/chain/message.py index 3627262c4..cfe6c97d9 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -323,7 +323,6 @@ class MessageChain(ChainBase): str(userid), channel=channel.value, source=source, - original_chat_id=original_chat_id, ): return False return self._handle_ai_message( diff --git a/app/modules/feishu/__init__.py b/app/modules/feishu/__init__.py index 9448e4fad..bc670d0ff 100644 --- a/app/modules/feishu/__init__.py +++ b/app/modules/feishu/__init__.py @@ -59,6 +59,10 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]): chat_id = None receive_id_type = "open_id" if userid else None + # 私聊投递只能按用户身份寻址，原会话 ID 可能属于群聊。 + if message.private_delivery and userid: + return userid, None, None + # 回复类消息携带原会话 ID 时，必须发回原会话， # 否则群聊 @ 机器人的回复会错误地发送到机器人与用户的私聊窗口。 original_chat_id = str(message.original_chat_id or "").strip() or None @@ -246,6 +250,20 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]): receive_id_type=receive_id_type, original_message_id=str(message.original_message_id) if message.original_message_id else None, ) + elif str(message.parse_mode or "").strip().lower() == "plain": + # 受保护结果必须绕过 Markdown 卡片，避免密钥字符被解释或改写。 + plain_text = "\n".join( + part + for part in (message.title, message.text, message.link) + if part + ) + result = client.send_text( + text=plain_text, + userid=userid, + chat_id=chat_id, + receive_id_type=receive_id_type, + original_message_id=str(message.original_message_id) if message.original_message_id else None, + ) else: result = client.send_notification( message=message, diff --git a/app/modules/telegram/__init__.py b/app/modules/telegram/__init__.py index 13b701132..73c34a03b 100644 --- a/app/modules/telegram/__init__.py +++ b/app/modules/telegram/__init__.py @@ -761,6 +761,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): original_chat_id=original_chat_id, disable_web_page_preview=message.disable_web_page_preview, parse_mode=message.parse_mode, + private_delivery=message.private_delivery, ) if result and result.get("success"): return MessageResponse( diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 650fb2147..816a42116 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -47,10 +47,13 @@ from app.utils.string import StringUtils # noqa: E402 TELEGRAM_PARSE_MODE_MARKDOWN = "MarkdownV2" TELEGRAM_PARSE_MODE_HTML = "HTML" +TELEGRAM_PARSE_MODE_PLAIN = "" TELEGRAM_PARSE_MODE_ALIASES = { "markdownv2": TELEGRAM_PARSE_MODE_MARKDOWN, "mdv2": TELEGRAM_PARSE_MODE_MARKDOWN, "html": TELEGRAM_PARSE_MODE_HTML, + "plain": TELEGRAM_PARSE_MODE_PLAIN, + "text": TELEGRAM_PARSE_MODE_PLAIN, } @@ -295,10 +298,13 @@ class Telegram: @staticmethod def _normalize_parse_mode(parse_mode: Optional[str] = None) -> str: """规范化 Telegram 消息格式类型。""" - if not parse_mode: + if parse_mode is None: return TELEGRAM_PARSE_MODE_MARKDOWN + normalized = str(parse_mode).strip() + if not normalized: + return TELEGRAM_PARSE_MODE_PLAIN return TELEGRAM_PARSE_MODE_ALIASES.get( - str(parse_mode).strip().lower(), TELEGRAM_PARSE_MODE_MARKDOWN + normalized.lower(), TELEGRAM_PARSE_MODE_MARKDOWN ) @staticmethod @@ -306,11 +312,18 @@ class Telegram: """判断本次发送是否使用 Telegram HTML 格式。""" return Telegram._normalize_parse_mode(parse_mode) == TELEGRAM_PARSE_MODE_HTML + @staticmethod + def _is_plain_parse_mode(parse_mode: Optional[str] = None) -> bool: + """判断本次发送是否禁用 Telegram 文本格式解析。""" + return Telegram._normalize_parse_mode(parse_mode) == TELEGRAM_PARSE_MODE_PLAIN + @staticmethod def _format_title(title: Optional[str], parse_mode: Optional[str] = None) -> Optional[str]: """按 parse_mode 生成 Telegram 标题文本。""" if not title: return None + if Telegram._is_plain_parse_mode(parse_mode): + return title.removesuffix("\n") if Telegram._is_html_parse_mode(parse_mode): return f"{html_utils.escape(title).removesuffix(chr(10))}" return f"**{standardize(title).removesuffix(chr(10))}**" @@ -318,6 +331,8 @@ class Telegram: @staticmethod def _format_link(label: str, link: str, parse_mode: Optional[str] = None) -> str: """按 parse_mode 生成 Telegram 链接文本。""" + if Telegram._is_plain_parse_mode(parse_mode): + return f"{label}: {link}" if Telegram._is_html_parse_mode(parse_mode): return ( f'' @@ -328,6 +343,8 @@ class Telegram: @staticmethod def _format_italic(text: str, parse_mode: Optional[str] = None) -> str: """按 parse_mode 生成 Telegram 斜体文本。""" + if Telegram._is_plain_parse_mode(parse_mode): + return text if Telegram._is_html_parse_mode(parse_mode): return f"{html_utils.escape(text)}" return f"_{text}_" @@ -342,7 +359,10 @@ class Telegram: """按 parse_mode 生成 Telegram 可发送文本。""" if not text: return None - if Telegram._is_html_parse_mode(parse_mode): + if ( + Telegram._is_plain_parse_mode(parse_mode) + or Telegram._is_html_parse_mode(parse_mode) + ): return text return standardize(text) @@ -595,6 +615,7 @@ class Telegram: disable_web_page_preview: Optional[bool] = None, stop_typing: bool = False, parse_mode: Optional[str] = None, + private_delivery: bool = False, ) -> Optional[dict]: """ 发送Telegram消息 @@ -610,6 +631,7 @@ class Telegram: :param disable_web_page_preview: 是否禁用链接预览 :param stop_typing: 发送完成后是否立即停止 typing :param parse_mode: Telegram 消息格式类型，默认 MarkdownV2，可传 HTML + :param private_delivery: 是否绕过最近会话映射，直接以用户 ID 作为私聊目标 :return: 包含 message_id, chat_id, success 的字典 """ if not self._telegram_token or not self._telegram_chat_id: @@ -617,7 +639,11 @@ class Telegram: parse_mode = self._normalize_parse_mode(parse_mode) # Determine target chat_id with improved logic using user mapping - chat_id = self._determine_target_chat_id(userid, original_chat_id) + chat_id = self._determine_target_chat_id( + userid, + original_chat_id, + private_delivery=private_delivery, + ) if not title and not text: logger.warn("标题和内容不能同时为空") self._stop_typing_if_needed(chat_id, stop_typing) @@ -836,7 +862,10 @@ class Telegram: return {"success": False} def _determine_target_chat_id( - self, userid: Optional[str] = None, original_chat_id: Optional[str] = None + self, + userid: Optional[str] = None, + original_chat_id: Optional[str] = None, + private_delivery: bool = False, ) -> str: """ 确定目标聊天ID，使用用户映射确保回复到正确的聊天 @@ -844,6 +873,10 @@ class Telegram: :param original_chat_id: 原消息的聊天ID :return: 目标聊天ID """ + # 私聊投递以渠道用户 ID 为目标，最近会话映射可能指向群聊，不能参与解析。 + if private_delivery and userid: + return str(userid) + # 1. 优先使用原消息的聊天ID (编辑消息场景) if original_chat_id: return original_chat_id @@ -1354,7 +1387,10 @@ class Telegram: ret = self.__send_short_message(image, caption, disable_web_page_preview=disable_web_page_preview, **kwargs) - elif self._is_html_parse_mode(parse_mode): + elif ( + self._is_plain_parse_mode(parse_mode) + or self._is_html_parse_mode(parse_mode) + ): ret = self.__send_long_plain_message( image, caption, diff --git a/app/schemas/message.py b/app/schemas/message.py index ca48a9d2a..1110003b6 100644 --- a/app/schemas/message.py +++ b/app/schemas/message.py @@ -262,9 +262,11 @@ class Notification(BaseModel): original_message_id: Optional[Union[str, int]] = None # 原消息的聊天ID，用于编辑消息 original_chat_id: Optional[str] = None + # 是否必须按用户身份投递到私聊，禁止回退原会话或最近会话映射 + private_delivery: bool = False # 是否禁用链接预览（仅Telegram支持） disable_web_page_preview: Optional[bool] = None - # Telegram 消息格式类型，默认 MarkdownV2，可传 HTML + # 消息文本格式；Telegram 支持 MarkdownV2、HTML、plain，飞书直发支持 plain parse_mode: Optional[str] = None # 是否写入消息历史 save_history: bool = True diff --git a/tests/test_agent_prompt_secrets.py b/tests/test_agent_prompt_secrets.py index d57afab63..ad81a19bd 100644 --- a/tests/test_agent_prompt_secrets.py +++ b/tests/test_agent_prompt_secrets.py @@ -1,4 +1,5 @@ from app.agent.prompt import PromptManager +from app.agent.tools.impl.query_system_settings import QuerySystemSettingsInput from app.core.config import settings @@ -64,3 +65,19 @@ def test_moviepilot_info_lists_command_names_without_paths(monkeypatch) -> None: assert "/opt/homebrew/bin/rg" not in moviepilot_info assert "/usr/local/bin/ffmpeg" not in moviepilot_info assert "rg --files" in moviepilot_info + + +def test_agent_prompt_delegates_explicit_secret_reads_to_host_confirmation() -> None: + """管理员明确索取密钥时，模型应发起工具调用并把授权交给宿主。""" + prompt = PromptManager().get_agent_prompt(channel="webagent") + + assert "query_system_settings" in prompt + assert "show_secrets=true" in prompt + assert "do not refuse" in prompt + assert "host verifies administrator authority" in prompt + assert "Never expose or repeat the secret" in prompt + + field_description = QuerySystemSettingsInput.model_fields["show_secrets"].description or "" + assert "user explicitly asks" in field_description + assert "host verifies administrator authority" in field_description + assert "Do not refuse the tool call" in field_description diff --git a/tests/test_agent_secret_confirmation.py b/tests/test_agent_secret_confirmation.py index 945fd0588..7b670eaa9 100644 --- a/tests/test_agent_secret_confirmation.py +++ b/tests/test_agent_secret_confirmation.py @@ -2,7 +2,8 @@ import asyncio from datetime import datetime, timedelta -from unittest.mock import AsyncMock, patch +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch from langchain.agents import create_agent from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel @@ -288,7 +289,6 @@ def test_expired_confirmation_reaches_agent_expiry_receipt() -> None: "1", channel=MessageChannel.WebAgent.value, source="web-agent", - original_chat_id="", ) return await agent.process("确认") finally: @@ -324,7 +324,7 @@ def test_background_agent_refuses_secret_read_without_pending() -> None: def test_message_channel_receives_confirmation_prompt_once() -> None: - """TG/飞书应由宿主直接发送确认提示，不依赖图状态转成渠道输出。""" + """TG/飞书应先向用户私聊发送提示，再登记待确认操作。""" agent = MoviePilotAgent( session_id="session-secret", user_id="1", @@ -347,17 +347,79 @@ def test_message_channel_receives_confirmation_prompt_once() -> None: new=AsyncMock(return_value=True), ), patch.object( agent, - "send_agent_message", - new=AsyncMock(), + "_deliver_private_channel_message", + new=AsyncMock(return_value=True), ) as send_message: prompt = asyncio.run(scenario()) send_message.assert_awaited_once_with(prompt) assert agent._tool_context["user_reply_sent"] is True + assert agent.has_pending_secret_confirmation() is True -def test_pending_secret_read_keeps_original_owner_and_action() -> None: - """新请求不得覆盖 pending，错误交付目标也不得消费它。""" +def test_message_channel_does_not_register_pending_when_private_delivery_fails() -> None: + """无法建立私聊时不得等待确认，更不能回退群聊投递结果。""" + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.Feishu.value, + source="feishu-main", + username="admin", + original_chat_id="group-1", + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch.object( + agent, + "_deliver_private_channel_message", + new=AsyncMock(return_value=False), + ) as deliver, + ): + result = asyncio.run( + agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + ) + + assert result == "无法向当前用户建立私聊，未执行敏感设置读取。" + deliver.assert_awaited_once() + assert agent.has_pending_secret_confirmation() is False + + +def test_private_delivery_requests_literal_plain_text() -> None: + """敏感提示与结果均须请求渠道按纯文本私聊投递。""" + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.Telegram.value, + source="telegram-main", + username="admin", + original_chat_id="group-1", + ) + response = SimpleNamespace(success=True) + + with patch( + "app.agent.AgentChain.send_direct_message", + return_value=response, + ) as send_direct: + delivered = asyncio.run( + agent._deliver_private_channel_message( + "G2A1_PROTECTED_MARKER_20260812\n**literal markdown**\n" + ) + ) + + assert delivered is True + notification = send_direct.call_args.args[0] + assert notification.private_delivery is True + assert notification.parse_mode == "plain" + assert notification.original_chat_id is None + + +def test_pending_secret_read_keeps_actor_and_action_across_chat_targets() -> None: + """新请求不得覆盖 pending，同一用户可从私聊消费群聊发起的确认。""" agent = MoviePilotAgent( session_id="session-secret", user_id="1", @@ -367,6 +429,8 @@ def test_pending_secret_read_keeps_original_owner_and_action() -> None: original_chat_id="chat-1", ) tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + delivered = [] async def scenario() -> tuple[str, str, str]: first = await agent._register_secret_confirmation( @@ -384,12 +448,254 @@ def test_pending_secret_read_keeps_original_owner_and_action() -> None: with ( patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), patch.object(agent, "_execute_agent", new=AsyncMock(return_value="普通回复")), - patch.object(QuerySystemSettingsTool, "_load_setting_value") as load_value, + patch.object( + agent, + "_deliver_private_channel_message", + new=AsyncMock(side_effect=lambda content: delivered.append(content) or True), + ), + patch.object( + QuerySystemSettingsTool, + "_load_setting_value", + return_value="secret-marker", + ) as load_value, ): first, second, result = asyncio.run(scenario()) assert "TMDB_API_KEY" in first assert "已有待确认" in second - assert result == "普通回复" - load_value.assert_not_called() - assert agent.has_pending_secret_confirmation() is True + assert result == "敏感设置确认已处理。" + load_value.assert_called_once() + assert "secret-marker" in delivered[-1] + assert agent.has_pending_secret_confirmation() is False + + +def test_confirm_reports_result_delivery_failure_without_secret() -> None: + """工具已读取但受保护结果未送达时，只能通过普通渠道报告非敏感失败。""" + secret_marker = "confirmed-secret-marker" + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.Telegram.value, + source="telegram-main", + username="admin", + original_chat_id="group-1", + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + + async def scenario() -> str: + await agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + return await agent.process("确认") + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch.object( + agent, + "_deliver_private_channel_message", + new=AsyncMock(side_effect=[True, False]), + ) as deliver, + patch.object(agent, "send_agent_message", new=AsyncMock()) as send_notice, + patch.object( + QuerySystemSettingsTool, + "_load_setting_value", + return_value=secret_marker, + ), + ): + result = asyncio.run(scenario()) + + assert result == "敏感设置读取已完成，但结果投递失败，请重新发起。" + assert deliver.await_count == 2 + assert secret_marker in deliver.await_args_list[-1].args[0] + send_notice.assert_awaited_once_with(result) + assert secret_marker not in send_notice.await_args.args[0] + + +def test_web_protected_callback_failure_returns_ordinary_safe_notice() -> None: + """Web 受保护回调失败时，普通流只能收到不含敏感结果的提示。""" + secret_marker = "confirmed-secret-marker" + ordinary_output = [] + + def broken_protected_callback(_content: str) -> None: + raise RuntimeError("delivery unavailable") + + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.WebAgent.value, + source="web-agent", + username="admin", + replay_mode=ReplyMode.CAPTURE_ONLY, + output_callback=ordinary_output.append, + protected_output_callback=broken_protected_callback, + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + + async def scenario() -> str: + await agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + ordinary_output.clear() + return await agent.process("确认") + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch.object( + QuerySystemSettingsTool, + "_load_setting_value", + return_value=secret_marker, + ), + ): + result = asyncio.run(scenario()) + + assert result == "敏感设置读取已完成，但结果投递失败，请重新发起。" + assert ordinary_output == [result] + assert secret_marker not in ordinary_output[0] + + +def test_confirm_reuses_policy_lifecycle() -> None: + """确认后的冻结调用必须生成与 ToolNode 相同的 start/finish 生命周期。""" + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.WebAgent.value, + source="web-agent", + username="admin", + replay_mode=ReplyMode.CAPTURE_ONLY, + protected_output_callback=lambda _content: None, + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + + async def scenario() -> str: + await agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + return await agent.process("确认") + + from app.agent.policy import DEFAULT_TOOL_POLICY_ORCHESTRATOR + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch.object(QuerySystemSettingsTool, "_load_setting_value", return_value="secret"), + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "start", + wraps=DEFAULT_TOOL_POLICY_ORCHESTRATOR.start, + ) as start, + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "finish", + wraps=DEFAULT_TOOL_POLICY_ORCHESTRATOR.finish, + ) as finish, + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "fail", + wraps=DEFAULT_TOOL_POLICY_ORCHESTRATOR.fail, + ) as fail, + ): + result = asyncio.run(scenario()) + + assert result == "敏感设置确认已处理。" + start.assert_called_once() + finish.assert_called_once() + fail.assert_not_called() + + +def test_confirm_respects_policy_denial_without_running_tool() -> None: + """确认不能覆盖宿主策略的拒绝决定。""" + protected_output = [] + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.WebAgent.value, + source="web-agent", + username="admin", + replay_mode=ReplyMode.CAPTURE_ONLY, + protected_output_callback=protected_output.append, + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + denied = SimpleNamespace(decision=SimpleNamespace(allowed=False)) + + async def scenario() -> str: + await agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + return await agent.process("确认") + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch("app.agent.middleware.policy.DEFAULT_TOOL_POLICY_ORCHESTRATOR.start", return_value=denied), + patch.object(QuerySystemSettingsTool, "_run_confirmed", new=AsyncMock()) as run_tool, + ): + result = asyncio.run(scenario()) + + assert result == "敏感设置确认已处理。" + assert protected_output[-1] == "当前宿主策略不允许执行该工具。" + run_tool.assert_not_awaited() + + +def test_confirm_records_policy_failure_and_returns_protected_error() -> None: + """确认执行异常必须闭合 fail 生命周期，且不把异常交给普通对话。""" + protected_output = [] + agent = MoviePilotAgent( + session_id="session-secret", + user_id="1", + channel=MessageChannel.WebAgent.value, + source="web-agent", + username="admin", + replay_mode=ReplyMode.CAPTURE_ONLY, + protected_output_callback=protected_output.append, + ) + tool = QuerySystemSettingsTool(session_id="session-secret", user_id="1") + tool.set_agent_context(agent._tool_context) + orchestrator = MagicMock() + orchestrator.start.return_value = SimpleNamespace( + decision=SimpleNamespace(allowed=True) + ) + + async def scenario() -> str: + await agent._register_secret_confirmation( + tool, + {"setting_key": "TMDB_API_KEY", "show_secrets": True}, + ) + return await agent.process("确认") + + from app.agent.policy import DEFAULT_TOOL_POLICY_ORCHESTRATOR + + with ( + patch.object(agent, "_is_system_admin_context", new=AsyncMock(return_value=True)), + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "start", + return_value=orchestrator.start.return_value, + ), + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "fail", + side_effect=orchestrator.fail, + ) as fail, + patch.object( + DEFAULT_TOOL_POLICY_ORCHESTRATOR, + "finish", + side_effect=orchestrator.finish, + ) as finish, + patch.object( + QuerySystemSettingsTool, + "_run_confirmed", + new=AsyncMock(side_effect=RuntimeError("secret-bearing-error")), + ), + ): + result = asyncio.run(scenario()) + + assert result == "敏感设置读取失败，请稍后重试。" + assert protected_output[-1] == result + fail.assert_called_once() + finish.assert_not_called() diff --git a/tests/test_agent_tool_policy.py b/tests/test_agent_tool_policy.py index 774625561..afc2fd597 100644 --- a/tests/test_agent_tool_policy.py +++ b/tests/test_agent_tool_policy.py @@ -373,7 +373,9 @@ def test_middleware_observation_failure_does_not_replace_success( ) -> None: """shadow start/finish 故障不能阻止 handler 或替换成功结果。""" orchestrator = MagicMock() - orchestrator.start.return_value = object() + orchestrator.start.return_value = SimpleNamespace( + decision=SimpleNamespace(allowed=True) + ) getattr(orchestrator, failed_phase).side_effect = RuntimeError( f"policy-{failed_phase}-failure" ) @@ -402,7 +404,9 @@ def test_middleware_observation_failure_does_not_replace_success( def test_middleware_fail_observation_does_not_mask_tool_error() -> None: """shadow fail hook 故障后仍必须抛出原始工具异常。""" orchestrator = MagicMock() - orchestrator.start.return_value = object() + orchestrator.start.return_value = SimpleNamespace( + decision=SimpleNamespace(allowed=True) + ) orchestrator.fail.side_effect = RuntimeError("policy-fail-hook-failure") middleware = AgentPolicyMiddleware( context=_interactive_context(), @@ -423,6 +427,32 @@ def test_middleware_fail_observation_does_not_mask_tool_error() -> None: assert error_info.value is tool_error +def test_middleware_keeps_shadow_observation_until_strict_runtime_takeover() -> None: + """普通 ToolNode 在严格策略接管前不得因观测决策改变既有行为。""" + orchestrator = MagicMock() + orchestrator.start.return_value = SimpleNamespace( + decision=SimpleNamespace(allowed=False) + ) + middleware = AgentPolicyMiddleware( + context=_interactive_context(), + orchestrator=orchestrator, + ) + request = SimpleNamespace( + tool=_EchoTool(session_id="session-1", user_id="user-1"), + tool_call={"id": "call-1", "name": "policy_echo", "args": {"query": "same"}}, + ) + handler = AsyncMock( + return_value=ToolMessage(content="same", tool_call_id="call-1") + ) + + result = asyncio.run(middleware.awrap_tool_call(request, handler)) + + assert result.content == "same" + handler.assert_awaited_once_with(request) + orchestrator.finish.assert_called_once() + orchestrator.fail.assert_not_called() + + def test_policy_hook_failure_logs_only_stable_type_information() -> None: """fail-open 诊断只记录阶段和异常类型，不读取可能含凭据的异常文本。""" mock_logger = MagicMock() diff --git a/tests/test_feishu.py b/tests/test_feishu.py index c2b9bd470..5fe6b0d68 100644 --- a/tests/test_feishu.py +++ b/tests/test_feishu.py @@ -1155,6 +1155,45 @@ class TestFeishu(unittest.TestCase): self.assertEqual(response.message_id, "om_789") self.assertEqual(response.chat_id, "oc_789") + def test_module_plain_direct_message_uses_literal_text_transport(self): + """纯文本直发不得进入会解释密钥字符的 Markdown 卡片路径。""" + module = FeishuModule() + module._channel = MessageChannel.Feishu + conf = SimpleNamespace(name="feishu-main") + client = MagicMock() + client.send_text.return_value = { + "success": True, + "message_id": "om_plain", + "chat_id": "oc_plain", + } + literal_text = "G2A1_PROTECTED_MARKER_20260812\n**literal markdown**\n" + + with ( + patch.object(module, "get_configs", return_value={"feishu-main": conf}), + patch.object(module, "check_message", return_value=True), + patch.object(module, "get_instance", return_value=client), + ): + response = module.send_direct_message( + Notification( + channel=MessageChannel.Feishu, + source="feishu-main", + userid="ou_target", + text=literal_text, + private_delivery=True, + parse_mode="plain", + ) + ) + + client.send_text.assert_called_once_with( + text=literal_text, + userid="ou_target", + chat_id=None, + receive_id_type=None, + original_message_id=None, + ) + client.send_notification.assert_not_called() + self.assertTrue(response.success) + def test_run_ws_client_binds_thread_local_event_loop(self): client = self._build_client() original_loop = object() @@ -1515,6 +1554,20 @@ class TestFeishu(unittest.TestCase): self.assertIsNone(userid) self.assertEqual(chat_id, "oc_config") + def test_module_private_delivery_ignores_original_group_chat(self): + """私聊投递只保留用户身份，并让客户端按已记录 ID 类型发送。""" + userid, chat_id, receive_id_type = FeishuModule._resolve_message_target( + Notification( + userid="user_target", + original_chat_id="oc_group", + private_delivery=True, + ) + ) + + self.assertEqual(userid, "user_target") + self.assertIsNone(chat_id) + self.assertIsNone(receive_id_type) + def test_module_post_message_replies_to_original_chat_for_group_message(self): """携带原会话上下文的回复应定向到原会话（群聊）。""" module = FeishuModule() diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 6b9cecb6c..81eae088b 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -578,6 +578,39 @@ def test_telegram_module_plain_direct_message_keeps_userid_target(): assert response.message_id == 456 +def test_telegram_private_delivery_bypasses_group_chat_mapping(telegram): + """私聊投递必须直接使用用户 ID，不能沿用该用户最近发言的群聊。""" + telegram._user_chat_mapping["10001"] = "group-1" + + result = telegram.send_msg( + title="", + text="受保护消息", + userid="10001", + private_delivery=True, + ) + + assert result and result.get("success") + assert telegram.bot.send_message.call_args.kwargs["chat_id"] == "10001" + + +def test_telegram_plain_private_delivery_keeps_literal_text(telegram): + """纯文本私聊不得解释或改写密钥中可能出现的 Markdown/HTML 字符。""" + literal_text = "G2A1_PROTECTED_MARKER_20260812\n**literal markdown**\n" + + result = telegram.send_msg( + title="", + text=literal_text, + userid="10001", + private_delivery=True, + parse_mode="plain", + ) + + assert result and result.get("success") + send_kwargs = telegram.bot.send_message.call_args.kwargs + assert send_kwargs["text"] == literal_text + assert send_kwargs["parse_mode"] == "" + + def test_send_msg_with_force_reply_uses_force_reply_when_no_buttons(telegram): """无按钮时force_reply应生成Telegram ForceReply标记""" result = telegram.send_msg( diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 752700e55..336ff0fb0 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -106,6 +106,19 @@ def test_web_agent_event_publisher_coalesces_text_before_semantic_events(): assert max_depth == 2 +def test_web_agent_event_publisher_rejects_events_after_close(): + """连接关闭后必须显式拒绝事件，避免把敏感结果误报为已交付。""" + + async def scenario(): + publisher = _WebAgentEventPublisher() + await publisher.aclose() + return publisher.publish( + {"type": "interaction-protected", "content": "secret"} + ) + + assert asyncio.run(scenario()) is False + + def test_build_web_agent_session_id_is_stable_per_user_and_seed(): """同一用户和前端会话标识应生成稳定的服务端会话 ID。""" user = SimpleNamespace(id=1, name="admin") @@ -969,11 +982,15 @@ def test_web_agent_stream_drops_secret_result_after_disconnect(): client_session_id=payload.session_id, ) + delivery_results = [] + async def finish_after_disconnect(**kwargs): """断线后继续完成只读任务，并尝试向已关闭发布器投递。""" agent_started.set() await release_agent.wait() - kwargs["protected_output_callback"]("DISCONNECTED_SECRET_MARKER") + delivery_results.append( + kwargs["protected_output_callback"]("DISCONNECTED_SECRET_MARKER") + ) agent_completed.set() async def scenario(): @@ -999,6 +1016,7 @@ def test_web_agent_stream_drops_secret_result_after_disconnect(): body = asyncio.run(scenario()) assert '"type": "start"' in body + assert delivery_results == [False] assert "cancel_on_waiter_cancel" not in process.await_args.kwargs save_snapshot.assert_not_called() preserved_chat = AgentChatOper().get(session_id=session_id, user_id="1")