fix(agent): 浏览器与验证码工具的私网访问参数仅限管理员

browse_webpage、recognize_captcha 的 allow_private_network 是模型可以直接
填写的工具参数,此前对任何调用方都生效。现与同工具的 evaluate、
get_cookies 保持一致:仅管理员可设置为 true,其他调用方返回
admin_required,并提示去掉该参数后只访问公网地址。

browse_webpage 的三项管理员检查合并到 _admin_required_error,run() 的
圈复杂度维持在门禁阈值以内。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
LIN student
2026-10-05 15:14:53 +08:00
parent 4ca70b478b
commit aa78bc9d8e
4 changed files with 121 additions and 16 deletions

View File

@@ -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,

View File

@@ -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",

View File

@@ -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):

View File

@@ -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