diff --git a/app/agent/tools/impl/browse_webpage.py b/app/agent/tools/impl/browse_webpage.py index 72cfc6339..580aa419e 100644 --- a/app/agent/tools/impl/browse_webpage.py +++ b/app/agent/tools/impl/browse_webpage.py @@ -135,7 +135,10 @@ class BrowseWebpageInput(BaseModel): ) allow_private_network: bool = Field( False, - description="Allow browser navigation to localhost, loopback, private, or link-local addresses.", + description=( + "Allow browser navigation to localhost, loopback, private, or link-local addresses " + "(administrator only; other callers receive an admin_required error)." + ), ) @@ -159,7 +162,8 @@ class BrowseWebpageTool(MoviePilotTool): "first call 'goto' to open a page, inspect 'interactive_elements', then use *_ref actions when possible. " "For a session-bound captcha or failed OCR, use 'screenshot' with the observed image selector " "to inspect the currently rendered captcha before refreshing it. " - "For safety, localhost and private network URLs are blocked by default unless allow_private_network is true." + "Localhost and private network URLs are blocked by default. " + "Only administrators may set allow_private_network to true." ) args_schema: Type[BaseModel] = BrowseWebpageInput @@ -313,16 +317,9 @@ class BrowseWebpageTool(MoviePilotTool): return self._error_response("missing_value", "'fill_ref' 操作需要提供 value 参数", "补充 value 后重试。") if browser_action == BrowserAction.EVALUATE and not script: return self._error_response("missing_script", "'evaluate' 操作需要提供 script 参数", "补充 script 后重试。") - if ( - browser_action == BrowserAction.EVALUATE - and not await self.is_admin_user() - ): - return self._error_response("admin_required", "'evaluate' 操作仅允许管理员使用", "改用只读浏览器 action 或请求管理员授权。") - if ( - browser_action == BrowserAction.GET_COOKIES - and not await self.is_admin_user() - ): - return self._error_response("admin_required", "'get_cookies' 操作仅允许管理员使用", "改用非敏感浏览器 action 或请求管理员授权。") + admin_error = await self._admin_required_error(browser_action, allow_private_network) + if admin_error: + return admin_error if ( browser_action in (BrowserAction.FOCUS_TAB, BrowserAction.CLOSE_TAB) and tab_index is None @@ -357,6 +354,25 @@ class BrowseWebpageTool(MoviePilotTool): return self._screenshot_failure("screenshot_failed", "浏览器截图执行失败") return self._error_response("browser_operation_failed", f"浏览器操作失败: {error_summary}", "检查当前会话和页面状态后再重试。") + async def _admin_required_error( + self, browser_action: BrowserAction, allow_private_network: bool + ) -> Optional[str]: + """执行脚本、读取 Cookie 和访问本机或私网地址仅限管理员，其他调用方返回明确原因。""" + if browser_action == BrowserAction.EVALUATE: + message, recovery = "'evaluate' 操作仅允许管理员使用", "改用只读浏览器 action 或请求管理员授权。" + elif browser_action == BrowserAction.GET_COOKIES: + message, recovery = "'get_cookies' 操作仅允许管理员使用", "改用非敏感浏览器 action 或请求管理员授权。" + elif allow_private_network: + message, recovery = ( + "allow_private_network 仅允许管理员使用", + "去掉 allow_private_network 后只访问公网地址，或请求管理员授权。", + ) + else: + return None + if await self.is_admin_user(): + return None + return self._error_response("admin_required", message, recovery) + def _execute_browser_action( self, browser_action: BrowserAction, diff --git a/app/agent/tools/impl/recognize_captcha.py b/app/agent/tools/impl/recognize_captcha.py index f8ce5fdac..c99e78578 100644 --- a/app/agent/tools/impl/recognize_captcha.py +++ b/app/agent/tools/impl/recognize_captcha.py @@ -46,7 +46,10 @@ class RecognizeCaptchaInput(BaseModel): ) allow_private_network: bool = Field( False, - description="Allow captcha image URLs on localhost, loopback, private, or link-local addresses.", + description=( + "Allow captcha image URLs on localhost, loopback, private, or link-local addresses " + "(administrator only; other callers receive an admin_required error)." + ), ) @model_validator(mode="after") # type: ignore[misc] @@ -77,8 +80,8 @@ class RecognizeCaptchaTool(MoviePilotTool): "This tool uses the configured OCR service, not the multimodal model. If OCR fails, " "inspect the current browser captcha with browse_webpage(action='screenshot', selector=...) " "before refreshing it or asking the user to read it. " - "For safety, localhost and private network URLs are blocked by default unless " - "allow_private_network is true." + "Localhost and private network URLs are blocked by default. " + "Only administrators may set allow_private_network to true." ) args_schema: Type[BaseModel] = RecognizeCaptchaInput @@ -191,6 +194,18 @@ class RecognizeCaptchaTool(MoviePilotTool): f"image_data={'%s bytes' % len(image_data) if image_data else 'none'}" ) + if allow_private_network and not await self.is_admin_user(): + return json.dumps( + { + "success": False, + "captcha_text": "", + "error": "admin_required", + "message": "allow_private_network 仅允许管理员使用", + "recovery": "去掉 allow_private_network 后只识别公网地址或 data:image 图片，或请求管理员授权。", + }, + ensure_ascii=False, + ) + try: captcha_text = await self.run_blocking( "web", diff --git a/tests/test_agent_recognize_captcha_tool.py b/tests/test_agent_recognize_captcha_tool.py index 927a24d32..38726b207 100644 --- a/tests/test_agent_recognize_captcha_tool.py +++ b/tests/test_agent_recognize_captcha_tool.py @@ -1,7 +1,7 @@ import asyncio import base64 import json -from unittest.mock import call, patch +from unittest.mock import AsyncMock, call, patch import pytest @@ -261,6 +261,49 @@ def test_recognize_captcha_tool_blocks_private_network_by_default(): recognize_mock.assert_not_called() +def test_recognize_captcha_tool_private_network_is_admin_only(monkeypatch): + """普通调用方设置 allow_private_network 时收到明确原因，且不会下载图片。""" + tool = RecognizeCaptchaTool(session_id="captcha-session", user_id="10001") + monkeypatch.setattr(RecognizeCaptchaTool, "is_admin_user", AsyncMock(return_value=False)) + + with patch( + "app.agent.tools.impl.recognize_captcha.OcrHelper.get_captcha_text", + return_value="x7p9", + ) as recognize_mock: + result = asyncio.run( + tool.run(image_url="http://qbittorrent:8080/captcha.png", allow_private_network=True) + ) + + payload = json.loads(result) + + assert payload["success"] is False + assert payload["captcha_text"] == "" + assert payload["error"] == "admin_required" + assert "allow_private_network" in payload["message"] + recognize_mock.assert_not_called() + + +def test_recognize_captcha_tool_admin_can_allow_private_network(monkeypatch): + """管理员仍可显式识别本机或私网地址上的验证码图片。""" + tool = RecognizeCaptchaTool(session_id="captcha-session", user_id="10001") + monkeypatch.setattr(RecognizeCaptchaTool, "is_admin_user", AsyncMock(return_value=True)) + + with patch( + "app.agent.tools.impl.recognize_captcha.OcrHelper.get_captcha_text", + return_value="x7p9", + ) as recognize_mock: + result = asyncio.run( + tool.run(image_url="http://qbittorrent:8080/captcha.png", allow_private_network=True) + ) + + assert json.loads(result)["captcha_text"] == "x7p9" + recognize_mock.assert_called_once_with( + image_url="http://qbittorrent:8080/captcha.png", + cookie=None, + ua=None, + ) + + @pytest.mark.parametrize("error", [None, RuntimeError("offline OCR"), ValueError("invalid OCR JSON")]) @pytest.mark.asyncio async def test_ocr_failure_directs_agent_to_current_browser_image(error): diff --git a/tests/test_browser_helper.py b/tests/test_browser_helper.py index e995aa18c..3f85b2fd2 100644 --- a/tests/test_browser_helper.py +++ b/tests/test_browser_helper.py @@ -638,3 +638,34 @@ async def test_browse_webpage_get_cookies_is_admin_only(monkeypatch: pytest.Monk result = await tool.run(action="get_cookies") assert "仅允许管理员" in result + + +@pytest.mark.asyncio +async def test_browse_webpage_private_network_is_admin_only(monkeypatch: pytest.MonkeyPatch) -> None: + """普通调用方设置 allow_private_network 时收到明确原因，且不会执行浏览器动作。""" + tool = BrowseWebpageTool(session_id="session-1", user_id="10001") + monkeypatch.setattr(BrowseWebpageTool, "is_admin_user", AsyncMock(return_value=False)) + run_blocking = AsyncMock(return_value="{}") + monkeypatch.setattr(BrowseWebpageTool, "run_blocking", run_blocking) + + payload = json.loads( + await tool.run(action="goto", url="http://qbittorrent:8080/", allow_private_network=True) + ) + + assert payload["success"] is False + assert payload["error"] == "admin_required" + assert "allow_private_network" in payload["message"] + run_blocking.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_browse_webpage_admin_can_allow_private_network(monkeypatch: pytest.MonkeyPatch) -> None: + """管理员仍可显式允许访问本机或私网地址。""" + tool = BrowseWebpageTool(session_id="session-1", user_id="10001") + monkeypatch.setattr(BrowseWebpageTool, "is_admin_user", AsyncMock(return_value=True)) + run_blocking = AsyncMock(return_value=json.dumps({"success": True})) + monkeypatch.setattr(BrowseWebpageTool, "run_blocking", run_blocking) + + await tool.run(action="goto", url="http://qbittorrent:8080/", allow_private_network=True) + + assert run_blocking.await_args.kwargs["allow_private_network"] is True