From 67968dd0cbc1b592474a513abbfbc6c6107186b3 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 15 Aug 2026 20:44:19 +0800 Subject: [PATCH] =?UTF-8?q?fix(trimemedia):=20=E9=A3=9E=E7=89=9B=E5=BD=B1?= =?UTF-8?q?=E8=A7=86=E7=99=BB=E5=BD=95=E9=80=82=E9=85=8D=E6=96=B0=E7=89=88?= =?UTF-8?q?=20v2=20=E5=8D=8F=E8=AE=AE=EF=BC=8C=E5=AF=86=E7=A0=81=E6=94=B9?= =?UTF-8?q?=E7=94=A8=20SHA256=20=E6=91=98=E8=A6=81=20(#6328)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/modules/trimemedia/api.py | 30 ++++++++- tests/test_trimemedia_login.py | 110 +++++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 2 deletions(-) create mode 100644 tests/test_trimemedia_login.py diff --git a/app/modules/trimemedia/api.py b/app/modules/trimemedia/api.py index 4905041ca..5218cfb79 100644 --- a/app/modules/trimemedia/api.py +++ b/app/modules/trimemedia/api.py @@ -194,11 +194,34 @@ class Api: """ 登录飞牛影视 + 新版服务端已废弃 v1 明文登录接口，优先使用 v2 协议登录（密码传输 SHA256 摘要）， + v2 接口不存在时回退旧版 v1 明文登录 + :return: 成功返回token 否则返回None """ # 开启访问码后需先通过访问码校验，否则无法访问登录接口 if not self.verify_access_code(): return None + # v2 协议要求密码为明文密码的 SHA256 十六进制小写摘要 + password_hash = hashlib.sha256(password.encode()).hexdigest() + res = self.request( + "/user/loginByPassword", + data={ + "username": username, + "password": password_hash, + "app_name": "trimemedia-web", + }, + base_path="/api/v2", + suppress_log=True, + ) + if res and res.success: + self._token = res.data.get("token") + return self._token + if res: + # v2 接口存在但登录失败（如账号密码错误），回退 v1 也无法成功 + logger.error(f"飞牛影视登录失败，错误码：{res.code} {res.msg}") + return None + # v2 接口不可用（旧版服务端），回退 v1 明文登录 if ( res := self.request( "/login", @@ -511,11 +534,13 @@ class Api: method: Optional[str] = None, params: Optional[dict] = None, data: Optional[dict] = None, + base_path: Optional[str] = None, suppress_log=False, ): """ 请求飞牛影视API + :param base_path: 接口路径前缀（如 /api/v2），默认使用 v1 路径 :param suppress_log: 是否禁止日志 """ @@ -537,10 +562,11 @@ class Api: if not self._host or not api: return None + prefix = base_path if base_path is not None else self._api_path if not api.startswith("/"): - api_path = f"{self._api_path}/{api}" + api_path = f"{prefix}/{api}" else: - api_path = self._api_path + api + api_path = prefix + api url = self._host + api_path if method is None: method = "get" if data is None else "post" diff --git a/tests/test_trimemedia_login.py b/tests/test_trimemedia_login.py new file mode 100644 index 000000000..0c3c79173 --- /dev/null +++ b/tests/test_trimemedia_login.py @@ -0,0 +1,110 @@ +""" +飞牛影视（trimemedia）v2 登录协议回归测试（Issue #6328） + +新版服务端废弃 v1 明文登录接口，login() 需优先使用 +POST /api/v2/user/loginByPassword（密码为 SHA256 十六进制摘要）， +并在 v2 接口不可用时回退旧版 v1 明文登录。 +""" + +import hashlib +from unittest.mock import patch + +from app.modules.trimemedia.api import Api + + +class _FakeResponse: + def __init__(self, payload: dict): + self._payload = payload + + def json(self): + return self._payload + + +def _create_api() -> Api: + """ + 构造未配置访问码的 Api 实例，跳过访问码校验流程 + """ + return Api(host="http://fn.local/v", apikey="test-api-key") + + +def test_login_uses_v2_login_by_password_with_sha256(): + """ + 新版服务端：走 v2 接口，密码为明文密码的 SHA256 十六进制小写摘要 + """ + api = _create_api() + with patch.object(api._request_utils, "request") as mock_request: + mock_request.return_value = _FakeResponse( + {"code": 0, "data": {"token": "v2-token"}} + ) + token = api.login("admin", "secret") + + assert token == "v2-token" + assert api.token == "v2-token" + assert mock_request.call_count == 1 + call = mock_request.call_args + assert call.kwargs["url"] == "http://fn.local/v/api/v2/user/loginByPassword" + body = call.kwargs["data"] + expected_hash = hashlib.sha256("secret".encode()).hexdigest() + assert f'"password": "{expected_hash}"' in body + assert '"password": "secret"' not in body + + +def test_login_falls_back_to_v1_when_v2_unavailable(): + """ + 旧版服务端：v2 接口不可用（HTTP 失败返回 None）时回退 v1 明文登录 + """ + api = _create_api() + with patch.object(api._request_utils, "request") as mock_request: + mock_request.side_effect = [ + None, + _FakeResponse({"code": 0, "data": {"token": "v1-token"}}), + ] + token = api.login("admin", "secret") + + assert token == "v1-token" + assert mock_request.call_count == 2 + urls = [call.kwargs["url"] for call in mock_request.call_args_list] + assert urls[0] == "http://fn.local/v/api/v2/user/loginByPassword" + assert urls[1] == "http://fn.local/v/api/v1/login" + # v1 回退仍传输明文密码 + assert '"password": "secret"' in mock_request.call_args_list[1].kwargs["data"] + + +def test_login_does_not_fallback_when_v2_rejects_credentials(): + """ + v2 接口存在但登录失败（账号密码错误）时直接返回 None，不回退 v1 + """ + api = _create_api() + with patch.object(api._request_utils, "request") as mock_request: + mock_request.return_value = _FakeResponse( + {"code": -15, "msg": "Password Incorrect"} + ) + token = api.login("admin", "wrong-password") + + assert token is None + assert api.token is None + # 新版服务端 v1 登录一律返回 -15，回退无意义，仅请求一次 + assert mock_request.call_count == 1 + assert ( + mock_request.call_args.kwargs["url"] + == "http://fn.local/v/api/v2/user/loginByPassword" + ) + + +def test_request_base_path_override_does_not_affect_default_path(): + """ + base_path 参数仅影响单次请求，其余接口仍使用默认 v1 路径 + """ + api = _create_api() + with patch.object(api._request_utils, "request") as mock_request: + mock_request.return_value = _FakeResponse({"code": 0, "data": {}}) + api.request("/user/info") + api.request("/user/loginByPassword", data={}, base_path="/api/v2") + api.request("/mediadb/sum") + + urls = [call.kwargs["url"] for call in mock_request.call_args_list] + assert urls == [ + "http://fn.local/v/api/v1/user/info", + "http://fn.local/v/api/v2/user/loginByPassword", + "http://fn.local/v/api/v1/mediadb/sum", + ]