From efc68ae70142fc3be939a1f689b6a2973068ce98 Mon Sep 17 00:00:00 2001 From: doumao Date: Sat, 28 Feb 2026 22:55:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BB=BF=E8=81=94=E6=8E=A5=E5=8F=A3?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E5=8F=AF=E9=85=8D=E7=BD=AESSL=E8=AF=81?= =?UTF-8?q?=E4=B9=A6=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/modules/ugreen/api.py | 12 ++-- app/modules/ugreen/ugreen.py | 23 ++++++- tests/test_ugreen_api.py | 113 +++++++++++++++++++++++++++++++ tests/test_ugreen_mediaserver.py | 12 ++++ 4 files changed, 153 insertions(+), 7 deletions(-) create mode 100644 tests/test_ugreen_api.py diff --git a/app/modules/ugreen/api.py b/app/modules/ugreen/api.py index 2f682934..6bf669ca 100644 --- a/app/modules/ugreen/api.py +++ b/app/modules/ugreen/api.py @@ -48,6 +48,7 @@ class Api: "_language", "_ug_agent", "_timeout", + "_verify_ssl", ) def __init__( @@ -57,6 +58,7 @@ class Api: language: str = "zh-CN", ug_agent: str = "PC/WEB", timeout: int = 20, + verify_ssl: bool = True, ): self._host = self._normalize_base_url(host) self._session = Session() @@ -73,6 +75,8 @@ class Api: self._language = language self._ug_agent = ug_agent self._timeout = timeout + # 是否校验证书，默认开启；仅在用户明确配置时才应关闭。 + self._verify_ssl = bool(verify_ssl) @property def host(self) -> str: @@ -167,7 +171,7 @@ class Api: params=params, json=json_data, timeout=self._timeout, - verify=False, + verify=self._verify_ssl, ) else: resp = self._session.get( @@ -175,7 +179,7 @@ class Api: headers=headers, params=params, timeout=self._timeout, - verify=False, + verify=self._verify_ssl, ) return resp.json() except Exception as err: @@ -219,7 +223,7 @@ class Api: headers=headers, json={"username": username}, timeout=self._timeout, - verify=False, + verify=self._verify_ssl, ) check_json = check_resp.json() except Exception as err: @@ -378,7 +382,7 @@ class Api: headers=req.headers, params=req.params, timeout=self._timeout, - verify=False, + verify=self._verify_ssl, ) except Exception: pass diff --git a/app/modules/ugreen/ugreen.py b/app/modules/ugreen/ugreen.py index ccc0da97..27166c0e 100644 --- a/app/modules/ugreen/ugreen.py +++ b/app/modules/ugreen/ugreen.py @@ -26,6 +26,7 @@ class Ugreen: _library_paths: dict[str, str] = {} _sync_libraries: List[str] = [] _scan_type: int = 2 + _verify_ssl: bool = True _api: Optional[Api] = None @@ -38,6 +39,7 @@ class Ugreen: sync_libraries: Optional[list] = None, scan_mode: Optional[Union[str, int]] = None, scan_type: Optional[Union[str, int]] = None, + verify_ssl: Optional[Union[bool, str, int]] = True, **kwargs, ): if not host or not username or not password: @@ -51,6 +53,8 @@ class Ugreen: # 绿联媒体库扫描模式： # 1 新添加和修改、2 补充缺失、3 覆盖扫描 self._scan_type = self.__resolve_scan_type(scan_mode=scan_mode, scan_type=scan_type) + # HTTPS 证书校验开关：默认开启，仅兼容自签证书等场景下可关闭。 + self._verify_ssl = self.__resolve_verify_ssl(verify_ssl) if play_host: self._playhost = UrlUtils.standardize_base_url(play_host).rstrip("/") @@ -144,7 +148,7 @@ class Ugreen: self.__remove_persisted_session() return False - api = Api(host=self._host) + api = Api(host=self._host, verify_ssl=self._verify_ssl) if not api.import_session_state(cached): api.close() self.__remove_persisted_session() @@ -174,7 +178,7 @@ class Ugreen: self.get_librarys() return True - self._api = Api(host=self._host) + self._api = Api(host=self._host, verify_ssl=self._verify_ssl) if self._api.login(self._username, self._password) is None: self.__remove_persisted_session() return False @@ -454,6 +458,19 @@ class Ugreen: } return mode_map.get(mode, 2) + @staticmethod + def __resolve_verify_ssl(verify_ssl: Optional[Union[bool, str, int]]) -> bool: + if isinstance(verify_ssl, bool): + return verify_ssl + if verify_ssl is None: + return True + value = str(verify_ssl).strip().lower() + if value in {"1", "true", "yes", "on"}: + return True + if value in {"0", "false", "no", "off"}: + return False + return True + def __scan_library(self, library_id: str, scan_type: Optional[int] = None) -> bool: if not self._api: return False @@ -561,7 +578,7 @@ class Ugreen: if not username or not password or not self._host: return None - api = Api(self._host) + api = Api(self._host, verify_ssl=self._verify_ssl) try: return api.login(username, password) finally: diff --git a/tests/test_ugreen_api.py b/tests/test_ugreen_api.py new file mode 100644 index 00000000..6a01e641 --- /dev/null +++ b/tests/test_ugreen_api.py @@ -0,0 +1,113 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from app.modules.ugreen.api import Api + + +class _FakeResponse: + def __init__(self, payload: dict, headers: dict | None = None): + self._payload = payload + self.headers = headers or {} + + def json(self): + return self._payload + + +class _FakeSession: + def __init__(self, get_responses=None, post_responses=None): + self._get_responses = list(get_responses or []) + self._post_responses = list(post_responses or []) + self.calls: list[tuple[str, dict]] = [] + self.cookies = SimpleNamespace( + get_dict=lambda: {}, + update=lambda *_args, **_kwargs: None, + ) + + def get(self, *args, **kwargs): + if args: + kwargs = {"url": args[0], **kwargs} + self.calls.append(("GET", kwargs)) + return self._get_responses.pop(0) if self._get_responses else _FakeResponse({}) + + def post(self, *args, **kwargs): + if args: + kwargs = {"url": args[0], **kwargs} + self.calls.append(("POST", kwargs)) + return self._post_responses.pop(0) if self._post_responses else _FakeResponse({}) + + @staticmethod + def close(): + return None + + +class _FakeCrypto: + def __init__(self, *args, **kwargs): + pass + + @staticmethod + def rsa_encrypt_long(raw: str) -> str: + return f"enc:{raw}" + + @staticmethod + def build_encrypted_request(url: str, method: str = "GET", params=None, **kwargs): + return SimpleNamespace(url=url, headers={}, params=params or {}, json=None, aes_key="k") + + @staticmethod + def decrypt_response(payload, aes_key): + return payload + + +class UgreenApiVerifySslTest(unittest.TestCase): + def test_request_json_default_verify_ssl_true(self): + api = Api(host="https://example.com") + fake_session = _FakeSession( + get_responses=[_FakeResponse({"code": 200})], + post_responses=[_FakeResponse({"code": 200})], + ) + api._session = fake_session + + api._request_json(url="https://example.com/a", method="GET") + api._request_json(url="https://example.com/b", method="POST", json_data={"x": 1}) + + self.assertEqual(fake_session.calls[0][1].get("verify"), True) + self.assertEqual(fake_session.calls[1][1].get("verify"), True) + + def test_login_logout_follow_verify_ssl_flag(self): + api = Api(host="https://example.com", verify_ssl=False) + fake_session = _FakeSession( + get_responses=[_FakeResponse({})], + post_responses=[ + _FakeResponse({"code": 200, "msg": "ok", "data": {}}, headers={"x-rsa-token": "BEGIN TEST"}), + _FakeResponse( + { + "code": 200, + "msg": "ok", + "data": { + "token": "token-value", + "public_key": "BEGIN LOGIN KEY", + "static_token": "static-token", + "is_ugk": False, + }, + } + ), + ], + ) + api._session = fake_session + + with patch("app.modules.ugreen.api.UgreenCrypto", _FakeCrypto): + token = api.login("tester", "pwd") + self.assertEqual(token, "token-value") + api.logout() + + self.assertEqual(len(fake_session.calls), 3) + self.assertEqual(fake_session.calls[0][0], "POST") + self.assertEqual(fake_session.calls[1][0], "POST") + self.assertEqual(fake_session.calls[2][0], "GET") + self.assertEqual(fake_session.calls[0][1].get("verify"), False) + self.assertEqual(fake_session.calls[1][1].get("verify"), False) + self.assertEqual(fake_session.calls[2][1].get("verify"), False) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ugreen_mediaserver.py b/tests/test_ugreen_mediaserver.py index d0631a77..5bed79de 100644 --- a/tests/test_ugreen_mediaserver.py +++ b/tests/test_ugreen_mediaserver.py @@ -121,6 +121,18 @@ class UgreenScanModeTest(unittest.TestCase): self.assertEqual(resolve(), 2) +class UgreenVerifySslTest(unittest.TestCase): + def test_resolve_verify_ssl(self): + resolve = Ugreen._Ugreen__resolve_verify_ssl + self.assertEqual(resolve(True), True) + self.assertEqual(resolve(False), False) + self.assertEqual(resolve("true"), True) + self.assertEqual(resolve("1"), True) + self.assertEqual(resolve("false"), False) + self.assertEqual(resolve("0"), False) + self.assertEqual(resolve(None), True) + + class UgreenStatisticTest(unittest.TestCase): def test_get_medias_count_episode_is_none(self): ugreen = Ugreen.__new__(Ugreen)