"""PikPak 云下载：假传输上的登录、刷新轮换、人机验证、离线提交与对账。 断言落在请求形状与凭据去向上：refresh token 每次刷新都写回，密码只在勾选时保存， 要人机验证时把验证页交出来而不是吞掉。 """ from __future__ import annotations import json import unittest from urllib.parse import parse_qs, urlsplit import httpx from peach import downloads as dl from peach import downloads_pikpak as pk HASH = "c9e15763f722f23e98a29decdfae341b98d53056" MAGNET = dl.parse_magnet(f"magnet:?xt=urn:btih:{HASH}&dn=ABC-123") TARGET = "/云下载" class FakePikPak: def __init__(self): self.requests: list[httpx.Request] = [] self.captcha_url = "" self.refresh_count = 0 self.expired_once = False self.tasks: list[dict] = [] self.signin_error = None def __call__(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) url = urlsplit(str(request.url)) body = json.loads(request.content) if request.content else {} route = (request.method, url.netloc, url.path) if route[2] == "/v1/shield/captcha/init": if self.captcha_url: return httpx.Response(200, json={"url": self.captcha_url}) return httpx.Response(200, json={"captcha_token": f"ct-{body['action']}"}) if route[2] == "/v1/auth/signin": if self.signin_error: return httpx.Response(400, json=self.signin_error) return httpx.Response(200, json=self._tokens()) if route[2] == "/v1/auth/token": if body.get("refresh_token") != f"rt{self.refresh_count}": return httpx.Response(400, json={"error": "invalid_grant", "error_code": 4126}) return httpx.Response(200, json=self._tokens()) if self.expired_once: self.expired_once = False return httpx.Response(401, json={"error": "unauthenticated", "error_code": 16}) if route == ("GET", "api-drive.mypikpak.com", "/drive/v1/files"): parent = parse_qs(url.query).get("parent_id", [""])[0] files = {"": [{"id": "f1", "name": "云下载", "kind": "drive#folder"}], "f1": [{"id": "x", "name": "ABC-123", "kind": "drive#folder"}]} return httpx.Response(200, json={"files": files.get(parent, []), "next_page_token": ""}) if route == ("POST", "api-drive.mypikpak.com", "/drive/v1/files"): task = {"id": "t1", "file_name": "ABC-123", "phase": "PHASE_TYPE_RUNNING", "progress": 0, "params": {"url": body["url"]["url"]}} self.tasks.append(task) return httpx.Response(200, json={"task": task}) if route == ("GET", "api-drive.mypikpak.com", "/drive/v1/tasks"): return httpx.Response(200, json={"tasks": self.tasks, "next_page_token": ""}) if route == ("DELETE", "api-drive.mypikpak.com", "/drive/v1/tasks"): ids = parse_qs(url.query)["task_ids"] self.tasks = [task for task in self.tasks if task["id"] not in ids] return httpx.Response(200, json={}) return httpx.Response(404, json={"error": "not_found"}) def _tokens(self): self.refresh_count += 1 return {"access_token": f"at{self.refresh_count}", "refresh_token": f"rt{self.refresh_count}", "expires_in": 7200, "sub": "user-1"} def paths(self) -> list[str]: return [f"{request.method} {request.url.path}" for request in self.requests] class _Case(unittest.TestCase): def setUp(self): self.fake = FakePikPak() self.transport = httpx.MockTransport(self.fake) self.stored: dict = {} def persist(self, values): self.stored = dict(values) def login(self, remember=False): return pk.login(lambda: self.stored, "me@example.com", "pw", remember=remember, persist=self.persist, transport=self.transport) def provider(self): return pk.PikPakProvider(lambda: dict(self.stored), persist=self.persist, transport=self.transport) class LoginTests(_Case): def test_signing_in_keeps_tokens_and_drops_the_password_unless_asked(self): self.assertEqual(self.login(), {"username": "me@example.com", "remember": False}) self.assertEqual(self.stored["refresh_token"], "rt1") self.assertNotIn("password", self.stored) self.assertEqual(len(self.stored["device_id"]), 32) signin = json.loads(self.fake.requests[1].content) self.assertEqual((signin["username"], signin["captcha_token"]), ("me@example.com", "ct-POST:https://user.mypikpak.com/v1/auth/signin")) self.login(remember=True) self.assertEqual(self.stored["password"], "pw") self.assertEqual(pk.account(self.stored), {"logged_in": True, "username": "me@example.com", "remember": True, "method": "password"}) def test_a_captcha_hands_back_the_page_and_keeps_the_device(self): self.fake.captcha_url = "https://user.mypikpak.com/captcha?x=1" with self.assertRaises(pk.CaptchaRequired) as caught: self.login() self.assertEqual(caught.exception.url, self.fake.captcha_url) device = self.stored["device_id"] self.assertNotIn("refresh_token", self.stored) self.fake.captcha_url = "" self.login() self.assertEqual(self.stored["device_id"], device) def test_a_wrong_password_is_a_configuration_failure(self): self.fake.signin_error = {"error": "invalid_account_or_password", "error_code": 4022, "error_description": "账号或密码错误"} with self.assertRaises(dl.DownloadError) as caught: self.login() self.assertEqual(caught.exception.failure, "config") self.assertIn("登录失败：账号或密码错误", caught.exception.detail) def test_the_captcha_signature_matches_the_reference_algorithm(self): self.assertEqual(pk.captcha_sign("abc", "123"), "1.3b6c3d5e08b41cc77d291c082ba26e7d") class DriveTests(_Case): def setUp(self): super().setUp() self.login() def task(self, **overrides) -> dl.DownloadTask: return dl.DownloadTask(1, HASH, "pikpak", MAGNET.uri, "ABC-123", TARGET, dl.SUBMITTED, **overrides) def test_a_submission_resolves_the_folder_and_carries_a_captcha_token(self): status = self.provider().submit(MAGNET, TARGET) self.assertEqual((status.state, status.remote_id, status.adopted), (dl.RUNNING, "t1", False)) post = [r for r in self.fake.requests if r.method == "POST" and r.url.path == "/drive/v1/files"][0] self.assertEqual(post.headers["X-Captcha-Token"], "ct-POST:/drive/v1/files") body = json.loads(post.content) self.assertEqual((body["parent_id"], body["url"]["url"], body["upload_type"]), ("f1", MAGNET.uri, "UPLOAD_TYPE_URL")) def test_the_same_magnet_is_adopted_from_the_task_list(self): self.provider().submit(MAGNET, TARGET) status = self.provider().submit(MAGNET, TARGET) self.assertTrue(status.adopted) self.assertEqual(len(self.fake.tasks), 1) def test_a_missing_folder_is_reported(self): with self.assertRaises(dl.DownloadError) as caught: self.provider().submit(MAGNET, "/没有") self.assertEqual(caught.exception.failure, "config") def test_an_expired_access_token_is_refreshed_and_the_new_refresh_token_saved(self): self.fake.expired_once = True self.assertEqual(self.provider().status(self.task()).state, dl.MISSING) self.assertEqual(self.stored["refresh_token"], "rt2") self.assertIn("POST /v1/auth/token", self.fake.paths()) def test_a_dead_refresh_token_without_a_saved_password_asks_to_sign_in_again(self): self.stored["refresh_token"] = "stale" self.stored["expires_at"] = "0" with self.assertRaises(dl.DownloadError) as caught: self.provider().status(self.task()) self.assertEqual(caught.exception.failure, "config") self.assertIn("重新登录", caught.exception.detail) def test_a_dead_refresh_token_with_a_saved_password_signs_in_again(self): self.login(remember=True) self.stored["refresh_token"] = "stale" self.stored["expires_at"] = "0" self.provider().status(self.task()) self.assertTrue(self.stored["refresh_token"].startswith("rt")) self.assertEqual(self.stored["password"], "pw") def test_phases_map_onto_remote_states(self): self.provider().submit(MAGNET, TARGET) self.fake.tasks[0].update(phase="PHASE_TYPE_ERROR", message="no seeds") status = self.provider().status(self.task(remote_id="t1")) self.assertEqual((status.state, status.message), (dl.ERROR, "no seeds")) self.fake.tasks[0].update(phase="PHASE_TYPE_COMPLETE") self.assertEqual(self.provider().status(self.task()).state, dl.DONE) self.fake.tasks[0].update(phase="SOMETHING_NEW") with self.assertRaises(dl.DownloadError) as caught: self.provider().status(self.task()) self.assertIn("接口可能已经变了", caught.exception.detail) def test_landed_lists_the_target_folder_and_cancel_keeps_files(self): self.assertTrue(self.provider().landed(self.task())) self.provider().submit(MAGNET, TARGET) self.provider().cancel(self.task(remote_id="t1")) delete = [r for r in self.fake.requests if r.method == "DELETE"][0] self.assertEqual(parse_qs(delete.url.query.decode())["delete_files"], ["false"]) self.assertEqual(self.fake.tasks, []) def test_without_a_login_nothing_is_sent(self): self.stored = {} sent = len(self.fake.requests) with self.assertRaises(dl.DownloadError): self.provider().submit(MAGNET, TARGET) self.assertEqual(len(self.fake.requests), sent) class WebClientTests(_Case): """浏览器登录读到的 token 由网页端 client 签发：刷新、captcha/init 与签名都按网页端常量走。""" def setUp(self): super().setUp() self.fake.refresh_count = 1 self.stored = {"refresh_token": "rt1", "access_token": "at1", "expires_at": "0", "user_id": "user-1", "device_id": "webdevice", "client_id": pk.WEB.client_id} def task(self) -> dl.DownloadTask: return dl.DownloadTask(1, HASH, "pikpak", MAGNET.uri, "ABC-123", TARGET, dl.SUBMITTED) def bodies(self, path: str) -> list[dict]: return [json.loads(r.content) for r in self.fake.requests if r.url.path == path] def test_the_web_signature_matches_the_page_script(self): """期望值用网页脚本里那段 `reduce` 的写法在 Node 里算出：md5 链、固定时间戳、网页端 15 条盐。""" self.assertEqual(pk.captcha_sign("abc", pk.WEB.sign_timestamp, pk.WEB), "1.e07351bd60bb92ce14b49dcf7273008f") def test_web_tokens_refresh_with_the_web_client_and_no_secret(self): self.assertEqual(self.provider().status(self.task()).state, dl.MISSING) refresh = [r for r in self.fake.requests if r.url.path == "/v1/auth/token"][0] body = json.loads(refresh.content) self.assertEqual(body["client_id"], "YUMx5nI8ZU8Ap8pm") self.assertNotIn("client_secret", body) self.assertEqual((refresh.headers["x-client-id"], refresh.headers["x-device-id"]), ("YUMx5nI8ZU8Ap8pm", "webdevice")) self.assertEqual((self.stored["refresh_token"], self.stored["client_id"]), ("rt2", pk.WEB.client_id)) def test_a_web_submission_signs_the_captcha_with_web_constants(self): self.stored["expires_at"] = "9999999999" self.provider().submit(MAGNET, TARGET) captcha = self.bodies("/v1/shield/captcha/init")[0] self.assertEqual(captcha["client_id"], pk.WEB.client_id) self.assertEqual(captcha["meta"]["timestamp"], "1790733736477") self.assertEqual((captcha["meta"]["client_version"], captcha["meta"]["package_name"]), ("2.0.0", "mypikpak.com")) self.assertEqual(captcha["meta"]["captcha_sign"], pk.captcha_sign("webdevice", "1790733736477", pk.WEB)) def test_a_rejected_web_refresh_names_the_reason_and_asks_for_the_browser(self): self.stored["refresh_token"] = "stale" with self.assertRaises(dl.DownloadError) as caught: self.provider().status(self.task()) self.assertEqual(caught.exception.failure, "config") self.assertIn("用浏览器登录", caught.exception.detail) self.assertIn("invalid_grant", caught.exception.detail) self.assertNotIn("stale", caught.exception.detail) def test_a_rejected_web_signature_says_the_constants_may_have_changed(self): self.stored["expires_at"] = "9999999999" def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/v1/shield/captcha/init": return httpx.Response(400, json={"error": "invalid_argument", "error_description": "captcha_sign"}) return self.fake(request) provider = pk.PikPakProvider(lambda: dict(self.stored), persist=self.persist, transport=httpx.MockTransport(handler)) with self.assertRaises(dl.DownloadError) as caught: provider.submit(MAGNET, TARGET) self.assertIn("网页端的签名常量可能已经更新", caught.exception.detail) def test_an_unknown_client_refreshes_but_refuses_to_sign(self): self.stored.update(client_id="SomeNewClient", expires_at="9999999999") with self.assertRaises(dl.DownloadError) as caught: self.provider().submit(MAGNET, TARGET) self.assertIn("SomeNewClient", caught.exception.detail) self.assertNotIn("/v1/shield/captcha/init", [r.url.path for r in self.fake.requests]) self.stored["expires_at"] = "0" self.provider().status(self.task()) self.assertEqual(self.bodies("/v1/auth/token")[0]["client_id"], "SomeNewClient") def test_the_account_says_how_it_was_signed_in(self): self.assertEqual(pk.account(self.stored)["method"], "browser") self.assertEqual(pk.account({"refresh_token": "x"})["method"], "password") self.login() self.assertEqual(self.stored["client_id"], pk.ANDROID.client_id) self.assertEqual(pk.account(self.stored)["method"], "password") if __name__ == "__main__": unittest.main()