GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-31 13:07:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(site): follow tracker login page for cookie refresh (#6473)

Browse Source
This commit is contained in:
jxxghp
2026-08-27 13:52:37 +08:00
parent f9fb909c93
commit ff355cbe5d
4 changed files with 199 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/adapters/network/browser.py
+12 -4
View File
@@ -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)
app/application/security/cookie.py
+43 -5
View File
@@ -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:
tests/test_browser_helper.py
+29 -2
View File
@@ -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 == "<html>ok</html>"
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")
tests/test_cookie_helper.py
+115
View File
@@ -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 '<html><body><a href="logout.php">退出</a></body></html>'
if self.url.endswith("/login.php"):
return (
'<html><body><form action="login.php">'
'<input name="username">'
'<input name="password" type="password">'
'<input type="submit" value="登录">'
"</form></body></html>"
)
return '<html><body><a href="login.php">登录</a></body></html>'
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(
'<html><body><a href="https://other.example/login.php">登录</a></body></html>'
),
"https://dicmusic.example/",
)
assert login_url is None