From 5a94ac454ed4e670075f51e97bfcb921421f5093 Mon Sep 17 00:00:00 2001 From: youfak Date: Thu, 27 Aug 2026 14:58:30 +0800 Subject: [PATCH] 1 --- .env.example | 7 +--- main.py | 53 ++++++++++++++++++++++++-- tests/test_api_sig3_state.py | 74 ++++++++++++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 9 deletions(-) create mode 100644 tests/test_api_sig3_state.py diff --git a/.env.example b/.env.example index 7214e12..84cfee1 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,2 @@ -KS_ACCOUNT=大号#kpn=NEBULA; kpf=ANDROID_PHONE; userId=...; did=...; kuaishou.api_st=...; token=...; client_key=2ac2a76d#client_salt -KS_AD_COUNT=1 -KS_TREASURE_AD_COUNT=1 -KS_DELAY_SECONDS=1 -KS_TIMEOUT=30 +ksck=大号#kpn=NEBULA; kpf=ANDROID_PHONE; userId=...; did=...; kuaishou.api_st=...; token=...; client_key=2ac2a76d#client_salt + diff --git a/main.py b/main.py index 70c26da..a848d64 100644 --- a/main.py +++ b/main.py @@ -427,6 +427,22 @@ def replace_pair_values(pairs: list[tuple[str, str]], updates: dict[str, str]) - return result +def new_api_sig3_state() -> Kwsg10418State | None: + """一次 run 的一次冷启动：时间驱动 session_seed + 进程起始 counter。 + + 真机每次冷启动 srand(time) 生成新 seed；counter 从进程已发生的 + 请求数附近起步（0x5F 为 sms_login 实测值），之后随请求递增。 + 设 ``KS_API_SIG3_FRESH_STATE=0`` 时返回 None，由调用方回退旧的 + ``from_digest(HAR_SEED)`` 行为，用于 A/B 对比验证服务端是否按签名降级。 + """ + from core.sig3 import kwsg_10418_session_seed_from_time + + if os.environ.get("KS_API_SIG3_FRESH_STATE", "1").strip().lower() in {"0", "false", "no"}: + return None # 由调用方回退 from_digest(HAR_SEED) + seed = kwsg_10418_session_seed_from_time(int(time.time())) + return Kwsg10418State(session_seed=seed, counter=0x5F) + + def signed_api_url( path: str, params: dict[str, str], @@ -809,6 +825,7 @@ class KsNebulaClient: self._h5_kww_warned = False self._api_http_session: Any | None = None self._api_weapon_header_provider = WeaponProofProvider(API_HEADERS["kaw"]) + self.api_sig3_state: Kwsg10418State | None = None # 进程级共享 10418 state，run_all_once 里初始化 self.sign_in_event_tracking = SIGN_IN_EVENT_TRACKING self.treasure_open_event_tracking = _compact_json(TREASURE_OPEN_EVENT_TRACKING) if self.memory_device: @@ -1332,10 +1349,31 @@ class KsNebulaClient: ) return record + def _resolve_api_sig3_state( + self, + state: Kwsg10418State | None, + seed: str, + ) -> Kwsg10418State: + """解析 10418 state：显式参数 > 进程级共享 > 新 seed > HAR 回退。 + + 真机一次冷启动 = 一个共享 state，counter 随请求递增。这里在 + ``run_all_once`` 里初始化一次 ``self.api_sig3_state`` 后，所有 + fetch/report/sign 共用它；显式 ``state`` 优先（测试用）；新 seed + 禁用（``KS_API_SIG3_FRESH_STATE=0``）时回退旧的 ``from_digest``。 + """ + if state is not None: + return state + if self.api_sig3_state is None: + self.api_sig3_state = new_api_sig3_state() + if self.api_sig3_state is not None: + return self.api_sig3_state + return Kwsg10418State.from_digest(API_SIG3_SEED[seed]) + def sign_in_api(self) -> ResponseRecord: path = "/rest/n/encourage/unionTask/signIn/report" body_pairs = [("eventTrackingLogInfo", "")] + self._api_body_common() - url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED["sign_in_report"]) + state = self._resolve_api_sig3_state(None, "sign_in_report") + url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED["sign_in_report"], state=state) body = urllib.parse.urlencode(body_pairs).encode("utf-8") headers = dict(API_HEADERS) headers["Cookie"] = self._api_cookie_header() @@ -1375,7 +1413,8 @@ class KsNebulaClient: def open_treasure_box_api(self) -> ResponseRecord: path = "/rest/n/encourage/unionTask/treasureBox/report" body_pairs = [("layoutPage", "NEBULA_ACTIVITY"), ("eventTrackingLogInfo", "")] + self._api_body_common() - url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED["treasure_report"]) + state = self._resolve_api_sig3_state(None, "treasure_report") + url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED["treasure_report"], state=state) body = urllib.parse.urlencode(body_pairs).encode("utf-8") headers = dict(API_HEADERS) headers["Cookie"] = self._api_cookie_header() @@ -1398,7 +1437,7 @@ class KsNebulaClient: network_ip=os.environ.get("KS_AD_NETWORK_IP", "172.31.230.147"), ipdx_ip=os.environ.get("KS_AD_IPDX_IP", "106.178.190.155"), ) - state = Kwsg10418State.from_digest(API_SIG3_SEED[f"{kind}_fetch"]) + state = self._resolve_api_sig3_state(None, f"{kind}_fetch") t1, t2 = enc_tables() reward_body = build_reward_body(str_e, state, t1=t1, t2=t2) body_pairs = [("encData", reward_body.enc_data), ("sign", reward_body.sign)] + self._api_body_common() @@ -1427,7 +1466,8 @@ class KsNebulaClient: first_pair = ("bizStr", biz_str) if mode == "bizstr" else ("encData", build_report_enc_data(biz_str.encode("utf-8"))) body_pairs = [first_pair] + self._api_body_common() path = "/rest/r/ad/task/report" - url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED[f"{kind}_report"]) + state = self._resolve_api_sig3_state(None, f"{kind}_report") + url = signed_api_url(path, self._api_params(), body_pairs, self.client_salt, API_SIG3_SEED[f"{kind}_report"], state=state) body = urllib.parse.urlencode(body_pairs).encode("utf-8") headers = dict(API_HEADERS) headers["Cookie"] = self._api_cookie_header() @@ -1492,6 +1532,11 @@ class KsNebulaClient: out_dir.mkdir(parents=True, exist_ok=True) record_path = out_dir / f"request_replay_{time.strftime('%Y%m%d_%H%M%S')}.jsonl" + # 一次 run = 一次冷启动 = 一个进程级共享 10418 state， + # counter 随整条广告/任务链路单调递增（真机行为）。 + if self.api_sig3_state is None: + self.api_sig3_state = new_api_sig3_state() + def run_ad_action(kind: str) -> list[ResponseRecord]: if skip_ad_reward_leq: return self.run_ad_sequence(kind, ad_wait, skip_reward_leq=skip_ad_reward_leq) diff --git a/tests/test_api_sig3_state.py b/tests/test_api_sig3_state.py new file mode 100644 index 0000000..8ce6d75 --- /dev/null +++ b/tests/test_api_sig3_state.py @@ -0,0 +1,74 @@ +# -*- coding: utf-8 -*- +"""main.py 广告/任务链路进程级 10418 state 的行为锁定。 + +真机一次冷启动 = 一个 Kwsg10418State（时间驱动 session_seed + counter +随请求单调递增）。这里锁定 main.py 的 ``new_api_sig3_state`` / +``_resolve_api_sig3_state`` 复刻该行为，防止未来回退成"每次 fetch 新建 +state、counter 恒定 + 旧 HAR seed 重放"的可检测指纹。 +""" + +import os +import unittest +from unittest.mock import patch + +from core.sig3 import Kwsg10418State, kwsg_10418_digest24_unmix +from main import API_SIG3_SEED, KsNebulaClient, new_api_sig3_state + + +class NewApiSig3StateTests(unittest.TestCase): + def test_signing_increments_counter_with_same_seed(self): + # 连签两次：counter 0x60→0x61，session_seed 不变（同一次冷启动） + with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "1"}, clear=False): + state = new_api_sig3_state() + self.assertIsNotNone(state) + d1 = kwsg_10418_digest24_unmix(state.sig3_hex(b"req1")) + d2 = kwsg_10418_digest24_unmix(state.sig3_hex(b"req2")) + self.assertEqual(d1["session_seed"], d2["session_seed"]) + self.assertEqual([d1["counter"], d2["counter"]], [0x60, 0x61]) + + def test_counter_starts_at_sms_login_value(self): + with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "1"}, clear=False): + state = new_api_sig3_state() + self.assertEqual(state.counter, 0x5F) # 同 tools/sms_login_cli.py:_new_sig3_state + + def test_disabled_returns_none(self): + # KS_API_SIG3_FRESH_STATE=0 → None，由调用方回退旧 from_digest(HAR_SEED) + with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "0"}, clear=False): + self.assertIsNone(new_api_sig3_state()) + + +class ResolveApiSig3StateTests(unittest.TestCase): + def _bare_client(self): + # 裸实例避免走重量级 __init__；_resolve 只读写 self.api_sig3_state + client = object.__new__(KsNebulaClient) + client.api_sig3_state = None + return client + + def test_shares_process_state_across_calls(self): + client = self._bare_client() + s1 = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch") + s2 = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch") + self.assertIs(s1, s2) # 同一实例 → counter 持续递增而非恒定重放 + d1 = kwsg_10418_digest24_unmix(s1.sig3_hex(b"req1")) + d2 = kwsg_10418_digest24_unmix(s1.sig3_hex(b"req2")) + self.assertEqual(d1["session_seed"], d2["session_seed"]) + self.assertEqual(d2["counter"], d1["counter"] + 1) + + def test_explicit_state_wins_and_does_not_pollute_shared(self): + client = self._bare_client() + explicit = Kwsg10418State(session_seed=0x12345678, counter=0x10) + resolved = KsNebulaClient._resolve_api_sig3_state(client, explicit, "normal_fetch") + self.assertIs(resolved, explicit) + self.assertIsNone(client.api_sig3_state) # 显式优先时不写回共享 state + + def test_falls_back_to_har_seed_when_disabled(self): + client = self._bare_client() + with patch.dict(os.environ, {"KS_API_SIG3_FRESH_STATE": "0"}, clear=False): + resolved = KsNebulaClient._resolve_api_sig3_state(client, None, "normal_fetch") + har = Kwsg10418State.from_digest(API_SIG3_SEED["normal_fetch"]) + self.assertEqual(resolved.session_seed, har.session_seed) + self.assertEqual(resolved.counter, har.counter) + + +if __name__ == "__main__": + unittest.main()