diff --git a/main.py b/main.py index a9a2948..70c26da 100644 --- a/main.py +++ b/main.py @@ -30,10 +30,12 @@ from core.h5_sig3 import h5_sig3_crc32_from_sign_input, h5_sig3_from_fields from core.h5_jsbridge import H5JsBridgeEncoder, build_h5_sign_input, h5_sig3_for_request from core.h5_kww import H5KwwGenerator, PureH5KwwGenerator from core.h5_kws import KWS_PRODUCT_NAME, build_h5_kws_default_ticket, build_h5_kws_script_ticket +from core.http_transport import create_http_session from core.reward_request import build_reward_body, build_reward_str_e from core.sig import build_sig_plaintext, sig as calc_sig from core.sig3 import Kwsg10418State, load_kwsg_10418_tables from core.tokensig import tokensig as calc_tokensig +from core.weapon_kas import WeaponProofProvider from core.xfalcon import xfalcon_value_from_input_bytes from tools.new_device import run_online_bootstrap @@ -805,6 +807,8 @@ class KsNebulaClient: self.h5_kww_node = os.environ.get("NODE_BIN") or None self.h5_kww_generator: H5KwwGenerator | PureH5KwwGenerator | None = None self._h5_kww_warned = False + self._api_http_session: Any | None = None + self._api_weapon_header_provider = WeaponProofProvider(API_HEADERS["kaw"]) self.sign_in_event_tracking = SIGN_IN_EVENT_TRACKING self.treasure_open_event_tracking = _compact_json(TREASURE_OPEN_EVENT_TRACKING) if self.memory_device: @@ -1117,19 +1121,50 @@ class KsNebulaClient: backoff = max(0.0, float(os.environ.get("KS_RATE_LIMIT_BACKOFF", "8"))) for attempt in range(attempts): attempt_headers = dict(headers) - if urllib.parse.urlsplit(url).netloc == API_HOST: + is_api_request = urllib.parse.urlsplit(url).netloc == API_HOST + if is_api_request: attempt_headers["X-REQUESTID"] = generate_request_id() - req = urllib.request.Request( - url, - data=body if method not in {"GET", "HEAD"} else None, - headers=attempt_headers, - method=method, - ) + attempt_headers.update(self._api_weapon_header_provider(url)) try: - with urllib.request.urlopen(req, timeout=self.timeout) as resp: - text = decode_response(resp.read(), resp.headers.get("Content-Encoding", "")) + if is_api_request: + if self._api_http_session is None: + self._api_http_session = create_http_session("okhttp4-android10") + resp = self._api_http_session.request( + method, + url, + data=body if method not in {"GET", "HEAD"} else None, + headers=attempt_headers, + timeout=self.timeout, + ) + status = int(resp.status_code) + text = decode_response( + bytes(resp.content), + resp.headers.get("Content-Encoding", ""), + ) data = parse_json(text) - record = ResponseRecord(label, title, request_id, resp.status, is_business_ok(resp.status, data), url, data, text) + error = "" if 200 <= status < 300 else f"http_status_{status}" + record = ResponseRecord( + label, + title, + request_id, + status, + is_business_ok(status, data), + url, + data, + text, + error, + ) + else: + req = urllib.request.Request( + url, + data=body if method not in {"GET", "HEAD"} else None, + headers=attempt_headers, + method=method, + ) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + text = decode_response(resp.read(), resp.headers.get("Content-Encoding", "")) + data = parse_json(text) + record = ResponseRecord(label, title, request_id, resp.status, is_business_ok(resp.status, data), url, data, text) except urllib.error.HTTPError as exc: text = decode_response(exc.read(), exc.headers.get("Content-Encoding", "")) data = parse_json(text) diff --git a/tests/test_main_device_profile.py b/tests/test_main_device_profile.py index bc860df..bcaf826 100644 --- a/tests/test_main_device_profile.py +++ b/tests/test_main_device_profile.py @@ -10,8 +10,10 @@ from unittest.mock import patch from core.device_profile import DeviceProfileGenerator, save_device_profile from core.h5_jsbridge import build_h5_sign_input from core.h5_sig3 import h5_sig3_crc32_from_sign_input, parse_h5_sig3 +from core.weapon_kas import WeaponProofProvider from main import ( AD_FETCH_PAYLOADS, + API_HEADERS, DEFAULT_OAID, KsNebulaClient, ResponseRecord, @@ -466,6 +468,78 @@ class MainDeviceProfileTests(unittest.TestCase): self.assertFalse(is_business_ok(200, {"result": 1, "errorMsg": "ANTISPAM_REQUEST"})) self.assertFalse(is_business_ok(200, {"result": 1, "errorMsg": "INVALID_REQUEST"})) + def test_api_requests_use_one_reused_okhttp_session(self): + client = KsNebulaClient( + full_cookie="userId=1579452490; did=ANDROID_device; kuaishou.api_st=API_ST", + client_salt="salt", + cookie_dict={ + "userId": "1579452490", + "did": "ANDROID_device", + "kuaishou.api_st": "API_ST", + }, + timeout=30, + dry_run=False, + ) + calls = [] + + class FakeApiSession: + def request(self, method, url, **kwargs): + calls.append((method, url, kwargs)) + return SimpleNamespace( + status_code=200, + content=b'{"result":1}', + headers={}, + ) + + with patch( + "main.create_http_session", + return_value=FakeApiSession(), + create=True, + ) as create_session, patch( + "main.urllib.request.urlopen", + side_effect=AssertionError("API request must not use urllib"), + ): + first = client._request( + "first", + "first", + 1, + "POST", + "https://api.e.kuaishou.com/rest/e/reward/mixed/ad?a=1", + {"Content-Type": "application/x-www-form-urlencoded"}, + b"x=1", + ) + second = client._request( + "second", + "second", + 2, + "POST", + "https://api.e.kuaishou.com/rest/r/ad/task/report?a=2", + {"Content-Type": "application/x-www-form-urlencoded"}, + b"x=2", + ) + + self.assertTrue(first.ok) + self.assertTrue(second.ok) + create_session.assert_called_once_with("okhttp4-android10") + self.assertEqual([item[0] for item in calls], ["POST", "POST"]) + self.assertEqual([item[2]["data"] for item in calls], [b"x=1", b"x=2"]) + expected_headers = [ + WeaponProofProvider(API_HEADERS["kaw"])(item[1]) + for item in calls + ] + self.assertEqual( + [item[2]["headers"]["kaw"] for item in calls], + [item["kaw"] for item in expected_headers], + ) + self.assertEqual( + [item[2]["headers"]["kas"] for item in calls], + [item["kas"] for item in expected_headers], + ) + self.assertNotEqual( + calls[0][2]["headers"]["kas"], + calls[1][2]["headers"]["kas"], + ) + def test_fetch_ad_builds_dynamic_reward_body(self): client = KsNebulaClient( full_cookie="userId=1579452490; did=ANDROID_oldoldoldold; token=TOKEN; kuaishou.api_st=API_ST", diff --git a/tests/test_sms_login_cli.py b/tests/test_sms_login_cli.py index dbcecd8..b5c3c13 100644 --- a/tests/test_sms_login_cli.py +++ b/tests/test_sms_login_cli.py @@ -108,6 +108,134 @@ class SmsLoginCliTests(unittest.TestCase): output.getvalue(), ) + def test_http_captcha_retries_with_fresh_sn_after_landing_error(self): + solver_module = ModuleType("tools.captcha_solver_http") + results = iter( + ( + { + "ok": False, + "stage": "verify", + "result": 350002, + "captcha_token": "", + "raw": {"result": 350002, "desc": "verify err.try again"}, + }, + { + "ok": True, + "stage": "verify", + "result": 1, + "captcha_token": "CAPTCHA_TOKEN", + "raw": {"result": 1, "captchaToken": "CAPTCHA_TOKEN"}, + }, + ) + ) + calls = [] + + def solve_captcha(**kwargs): + calls.append(kwargs) + return next(results) + + solver_module.solve_captcha = solve_captcha + with ( + patch.dict( + sms_login_cli.sys.modules, + {"tools.captcha_solver_http": solver_module}, + ), + redirect_stdout(StringIO()) as output, + ): + token = sms_login_cli._solve_captcha_http( + "https://app.m.kuaishou.com/verify/captcha.html?key=1&type=7", + session=object(), + ) + + self.assertEqual(token, "CAPTCHA_TOKEN") + self.assertEqual(len(calls), 2) + self.assertIn("重新获取 captchaSn", output.getvalue()) + + def test_main_runs_send_then_verify_without_checker(self): + profile = DeviceProfileGenerator(seed=20260730).new_profile() + http_session = MagicMock() + anonymous_region = RegionTicket(uid="0", ticket="RT_ANON", status=200) + account_region = RegionTicket(uid="101", ticket="RT_ACCOUNT", status=200) + login_session = LoginSession( + api_st="API_ST", + h5_st="H5_ST", + client_salt="SALT", + user_id="101", + mobile="13800000000", + ) + requests = [] + + def record_send(_profile, _mobile, **kwargs): + requests.append(("send", kwargs["encrypted_mobile"])) + return {"status": 200, "body": {"result": 1}} + + def record_login(_profile, _mobile, _code, **kwargs): + requests.append(("login", kwargs["encrypted_mobile"])) + return login_session, [("mobileVerifyCode", login_session)] + + with ( + patch.object(sms_login_cli, "DeviceProfileGenerator") as generator, + patch.object(sms_login_cli, "run_online_bootstrap"), + patch.object( + sms_login_cli, + "generate_profile_passport_account_image", + return_value="PASSPORT", + ), + patch.object(sms_login_cli, "WeaponProofProvider"), + patch.object( + sms_login_cli, + "_new_sig3_state", + return_value=Kwsg10418State(session_seed=1, counter=0x5F), + ), + patch.object( + sms_login_cli, + "create_http_session", + return_value=http_session, + ), + patch.object( + sms_login_cli, + "refresh_region_ticket", + side_effect=(anonymous_region, account_region), + ), + patch.object( + sms_login_cli, + "loginhelper_encrypt_mobile", + side_effect=("SEND_ENCRYPTED", "LOGIN_ENCRYPTED"), + ), + patch.object( + sms_login_cli, + "request_mobile_code", + side_effect=record_send, + ), + patch.object( + sms_login_cli, + "build_account_security_fields", + return_value={"raw": "RAW", "publicKey": "PUB", "secret": "SEC"}, + ), + patch.object( + sms_login_cli, + "login_by_code_try_paths", + side_effect=record_login, + ), + patch.object(sms_login_cli, "input", return_value="123456", create=True), + patch.object( + sms_login_cli.sys, + "argv", + ["sms_login_cli.py", "--mobile", "13800000000"], + ), + redirect_stdout(StringIO()), + ): + generator.return_value.new_profile.return_value = profile + self.assertEqual(sms_login_cli.main(), 0) + + self.assertEqual( + requests, + [ + ("send", "SEND_ENCRYPTED"), + ("login", "LOGIN_ENCRYPTED"), + ], + ) + def test_each_main_call_generates_and_registers_a_fresh_device(self): profiles = [ DeviceProfileGenerator(seed=20260730).new_profile(), diff --git a/tools/sms_login_cli.py b/tools/sms_login_cli.py index 2a5fb3d..1253859 100644 --- a/tools/sms_login_cli.py +++ b/tools/sms_login_cli.py @@ -56,6 +56,7 @@ _CAPTCHA_RESULT_HINTS = { 350017: "验证参数已失效", 350029: "服务端已切换验证码类型", } +_CAPTCHA_RETRYABLE_RESULTS = {350002, 350009, 350014, 350017} _ACCOUNT_COOKIE_ORDER = ( "kpn", @@ -303,24 +304,30 @@ def _solve_captcha_http( except ImportError as exc: print(f"[验证] HTTP solver 导入失败: {exc}") return "" - try: - result = solve_captcha(session=session, error_url=error_url, verbose=True) - except Exception as exc: # noqa: BLE001 - print(f"[验证] HTTP solver 异常: {exc.__class__.__name__}: {exc}") - return "" - raw_result = result.get("raw") if isinstance(result.get("raw"), dict) else {} - result_code = result.get("result") - if result_code is None: - result_code = raw_result.get("result") - server_desc = result.get("desc") or raw_result.get("desc") or "-" - token = str(result.get("captcha_token") or "") - if token: - shown_token = token if show_secrets else _mask_secret(token, head=10, tail=8) - print( - f"[验证] HTTP 求解成功 result={result_code} " - f"captchaToken={shown_token}" - ) - else: + for attempt in range(CAPTCHA_RETRIES + 1): + try: + result = solve_captcha( + session=session, + error_url=error_url, + verbose=True, + ) + except Exception as exc: # noqa: BLE001 + print(f"[验证] HTTP solver 异常: {exc.__class__.__name__}: {exc}") + return "" + raw_result = result.get("raw") if isinstance(result.get("raw"), dict) else {} + result_code = result.get("result") + if result_code is None: + result_code = raw_result.get("result") + server_desc = result.get("desc") or raw_result.get("desc") or "-" + token = str(result.get("captcha_token") or "") + if token: + shown_token = token if show_secrets else _mask_secret(token, head=10, tail=8) + print( + f"[验证] HTTP 求解成功 result={result_code} " + f"captchaToken={shown_token}" + ) + return token + hint = _CAPTCHA_RESULT_HINTS.get(result_code, "未分类响应") print( f"[验证] HTTP 求解未通过 result={result_code} " @@ -333,7 +340,13 @@ def _solve_captcha_http( f"config_path=" f"{result.get('configPath') or raw_result.get('configPath') or '-'}" ) - return token + if result_code not in _CAPTCHA_RETRYABLE_RESULTS or attempt >= CAPTCHA_RETRIES: + return "" + print( + f"[验证] 重新获取 captchaSn 后重试 " + f"({attempt + 1}/{CAPTCHA_RETRIES})" + ) + return "" def _complete_captcha_handoff(