75 lines
3.5 KiB
Python
75 lines
3.5 KiB
Python
# -*- 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()
|