From ff355cbe5d3704067b57ee019f8ec4af271e707f Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 27 Aug 2026 13:52:37 +0800 Subject: [PATCH] fix(site): follow tracker login page for cookie refresh (#6473) --- app/adapters/network/browser.py | 16 +++- app/application/security/cookie.py | 48 ++++++++++-- tests/test_browser_helper.py | 31 +++++++- tests/test_cookie_helper.py | 115 +++++++++++++++++++++++++++++ 4 files changed, 199 insertions(+), 11 deletions(-) create mode 100644 tests/test_cookie_helper.py diff --git a/app/adapters/network/browser.py b/app/adapters/network/browser.py index ba0b9752c..8d9e473df 100644 --- a/app/adapters/network/browser.py +++ b/app/adapters/network/browser.py @@ -8,13 +8,13 @@ from dataclasses import dataclass, field from typing import Any, Callable, Optional, Protocol from urllib.parse import urlparse +from app.adapters.network.http import RequestUtils, cookie_parse from app.runtime.log import logger -from app.runtime.settings import get_runtime_setting from app.runtime.managed_resources import ( acquire_managed_resource, acquire_managed_resource_async, ) -from app.adapters.network.http import RequestUtils, cookie_parse +from app.runtime.settings import get_runtime_setting class BrowserElement(Protocol): @@ -1070,6 +1070,7 @@ class PlaywrightHelper: :param headless: 是否无头模式 :param timeout: 超时时间 """ + timeout = timeout or 60 result = None try: context = None @@ -1095,8 +1096,15 @@ class PlaywrightHelper: if merged_cookie: page.set_extra_http_headers({"cookie": merged_cookie}) - page.goto(url) - page.wait_for_load_state("networkidle", timeout=timeout * 1000) + page.goto(url, wait_until="domcontentloaded", timeout=timeout * 1000) + try: + # 登录页的统计与长连接请求可能持续存在，不应阻断已就绪表单的处理。 + page.wait_for_load_state( + "networkidle", + timeout=min(timeout, 15) * 1000, + ) + except Exception: + pass # 回调函数 result = callback(page) diff --git a/app/application/security/cookie.py b/app/application/security/cookie.py index ff75e4efc..2ffb28495 100644 --- a/app/application/security/cookie.py +++ b/app/application/security/cookie.py @@ -1,16 +1,17 @@ import base64 import time -from typing import Tuple, Optional +from typing import Optional, Tuple +from urllib.parse import urljoin, urlparse from lxml import etree -from app.adapters.network.browser import BrowserPage, PlaywrightHelper from app.adapters.external.ocr import OcrHelper -from app.application.security.twofactor import TwoFactorAuth -from app.runtime.log import logger +from app.adapters.network.browser import BrowserPage, PlaywrightHelper from app.adapters.network.http import RequestUtils +from app.application.security.twofactor import TwoFactorAuth from app.domain.site import SiteUtils from app.foundation import url as url_tools +from app.runtime.log import logger class CookieHelper: @@ -107,6 +108,24 @@ class CookieHelper: cookie_str += f"{cookie['name']}={cookie['value']}; " return cookie_str + @staticmethod + def _find_login_page_url(html, current_url: str) -> Optional[str]: + """从首页查找同源登录入口，避免把账号密码提交到跨域页面。""" + login_hrefs = html.xpath( + "//a[" + "contains(translate(@href, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'login')" + " or contains(translate(@href, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'signin')" + "]/@href" + ) + current = urlparse(current_url) + for href in login_hrefs: + login_url = urljoin(current_url, href) + target = urlparse(login_url) + if target.scheme in ("http", "https") and \ + target.scheme == current.scheme and target.netloc == current.netloc: + return login_url + return None + def get_site_cookie_ua(self, url: str, username: str, @@ -161,6 +180,25 @@ class CookieHelper: if html.xpath(xpath): username_xpath = xpath break + if not username_xpath: + login_url = self._find_login_page_url(html, page.url or url) + if login_url: + try: + page.goto( + login_url, + wait_until="domcontentloaded", + timeout=(timeout or 60) * 1000, + ) + except Exception as e: + return None, None, f"打开登录页面失败：{str(e)}" + html_text = self.get_page_content(page) + html = etree.HTML(html_text) if html_text else None + if html is None: + return None, None, "解析网页源码失败" + for xpath in self._SITE_LOGIN_XPATH.get("username"): + if html.xpath(xpath): + username_xpath = xpath + break if not username_xpath: # 登录页可能为JS动态渲染（如SPA），等待用户名输入框出现后重试 try: @@ -335,7 +373,7 @@ class CookieHelper: error_msg = html.xpath(error_xpath)[0] return None, None, error_msg finally: - if html: + if html is not None: del html if not url or not username or not password: diff --git a/tests/test_browser_helper.py b/tests/test_browser_helper.py index f01988471..b1d67b783 100644 --- a/tests/test_browser_helper.py +++ b/tests/test_browser_helper.py @@ -1,7 +1,7 @@ from __future__ import annotations -import json import asyncio +import json import sys import threading from concurrent.futures import ThreadPoolExecutor @@ -11,13 +11,13 @@ from unittest.mock import AsyncMock, MagicMock, call, patch import pytest -from app.agent.tools.impl.browse_webpage import BrowserAction, BrowseWebpageTool from app.adapters.network.browser import ( BrowserSessionHelper, PlaywrightHelper, launch_browser_context, launch_browser_context_async, ) +from app.agent.tools.impl.browse_webpage import BrowserAction, BrowseWebpageTool from app.runtime.correlation import correlation_scope, get_correlation_id @@ -239,6 +239,33 @@ def test_legacy_browser_type_constructor_is_accepted(): assert source == "ok" +def test_browser_action_runs_callback_when_network_never_becomes_idle(): + """页面 DOM 已就绪时，持续后台请求不得阻断登录等页面回调。""" + page = _FakePage() + page.wait_for_load_state = MagicMock(side_effect=TimeoutError("still busy")) + context = _FakeContext([page]) + + with patch( + "app.adapters.network.browser.get_runtime_setting", + return_value="cloakbrowser", + ), patch.object( + PlaywrightHelper, + "_PlaywrightHelper__launch_cloakbrowser_context", + return_value=context, + ): + result = PlaywrightHelper().action( + url="https://example.com", + callback=lambda current_page: current_page.loaded_url, + timeout=30, + ) + + assert result == "https://example.com" + assert page.loaded_url == "https://example.com" + assert page.wait_for_load_state.call_args == call("networkidle", timeout=15000) + assert page.closed + assert context.closed + + def test_sync_browser_facade_activates_display_only_for_headed_mode(monkeypatch): """同步启动仅在明确有界面模式获取 host.display，参数原样交给浏览器。""" provider = ModuleType("cloakbrowser") diff --git a/tests/test_cookie_helper.py b/tests/test_cookie_helper.py new file mode 100644 index 000000000..7c2bf4a6e --- /dev/null +++ b/tests/test_cookie_helper.py @@ -0,0 +1,115 @@ +from unittest.mock import patch + +from lxml import etree + +from app.application.security.cookie import CookieHelper + + +class _CookieContext: + """提供登录测试所需的最小浏览器上下文。""" + + @staticmethod + def cookies() -> list[dict[str, str]]: + """返回登录后的会话 Cookie。""" + return [{"name": "session", "value": "authenticated"}] + + +class _CookiePage: + """模拟首页跳转登录页并提交表单的浏览器页面。""" + + def __init__(self) -> None: + self.url = "https://dicmusic.example/" + self.context = _CookieContext() + self.fills: list[tuple[str, str]] = [] + self.goto_calls: list[tuple[str, dict]] = [] + self.submitted = False + + def content(self) -> str: + """按当前页面阶段返回对应 HTML。""" + if self.submitted: + return '退出' + if self.url.endswith("/login.php"): + return ( + '
' + '' + '' + '' + "
" + ) + return '登录' + + def goto(self, url: str, **kwargs) -> None: + """记录并切换页面地址。""" + self.url = url + self.goto_calls.append((url, kwargs)) + + @staticmethod + def wait_for_load_state(_state: str, timeout: int) -> None: + """模拟页面已完成所需加载。""" + + @staticmethod + def wait_for_selector(_selector: str, *args, **kwargs) -> None: + """模拟表单元素已经可用。""" + + @staticmethod + def query_selector(_selector: str): + """当前页面没有保持登录复选框或验证码。""" + return None + + def fill(self, selector: str, value: str) -> None: + """记录表单填充值。""" + self.fills.append((selector, value)) + + def click(self, _selector: str) -> None: + """模拟提交登录表单。""" + self.submitted = True + + @staticmethod + def evaluate(_expression: str) -> str: + """返回浏览器 User-Agent。""" + return "Browser UA" + + +def test_cookie_login_follows_same_origin_login_link(): + """首页仅提供登录链接时应进入同源登录页后完成 Cookie 获取。""" + page = _CookiePage() + + def run_action(**kwargs): + return kwargs["callback"](page) + + with patch( + "app.application.security.cookie.PlaywrightHelper.action", + side_effect=run_action, + ): + cookie, ua, message = CookieHelper().get_site_cookie_ua( + url="https://dicmusic.example/", + username="moviepilot", + password="secret-password", + timeout=30, + ) + + assert cookie == "session=authenticated; " + assert ua == "Browser UA" + assert message == "" + assert page.goto_calls == [ + ( + "https://dicmusic.example/login.php", + {"wait_until": "domcontentloaded", "timeout": 30000}, + ) + ] + assert page.fills == [ + ('//input[@name="username"]', "moviepilot"), + ('//input[@name="password"]', "secret-password"), + ] + + +def test_cookie_login_rejects_cross_origin_login_link(): + """跨域登录链接不得成为账号密码填充目标。""" + login_url = CookieHelper._find_login_page_url( + etree.HTML( + '登录' + ), + "https://dicmusic.example/", + ) + + assert login_url is None