# -*- coding: utf-8 -*- """ksjsb.py 单文件脚本的行为锁定测试。 只 import ksjsb 一个项目文件(证明单文件自洽), 黄金值以字面量固化, 来自: - 2026-08-27 五份抓包(签到/签到看视频/开宝箱/开宝箱看广告/看广告得金币): URL query 键序编码、JSON body、真实 sig3 内嵌摘要字段(CRC 交叉验证)。 - core/ 算法模块的已知输出(10418/10400/xfalcon/H5 sig3/kww/kas 黄金值)。 ksjsb.py 运行时不读取 HAR, 也不依赖 main/core; 本测试同样只依赖 ksjsb。 """ import io import contextlib import os import unittest import urllib.parse from ksjsb import ( KsJsB, build_h5_sign_input, compact_json, h5_sig3_digest_field, parse_h5_sig3, with_source, ) # 抓包 cookie 中参与 sign_input 的子集(H5_COOKIE_KEYS, 非敏感设备字段)。 # 真实 WebView 对 document.cookie 的编码值(%28/+ 等)签名, 这里保持原样。 CAPTURED_COOKIE = ( "kpn=NEBULA; kpf=ANDROID_PHONE; userId=5402308233; did=ANDROID_f05497e9cef09a7f; " "c=OPPO; appver=14.6.50.11735; language=zh-cn; mod=OnePlus%28PJZ110%29; did_tag=0; " "egid=DFP280DA456E3BBF93A99727CB5A97D727E442E38694913A197D1C201C5CF982; " "oDid=ANDROID_46a032e0a2af8184; androidApiLevel=36; newOc=OPPO; browseType=3; " "socName=Qualcomm+Snapdragon+8750; ftt=; abi=arm64; userRecoBit=0; device_abi=arm64; " "grant_browse_type=AUTHORIZED; iuid=; rdid=ANDROID_741de4351c44850d; " "oaid=700ED68650EF49ECB686B2AAA9994C1Dca31c779ee5acb7d4c3b03337599a591" ) # ---- 真实请求 query(去掉 __NS_sig3, 保留原始顺序) ---- SHOW_POPUP_QUERY = ( "sigCatVer=1&source=bottom_guide_first&ladderRedPackGuide=false" "&bizParams=%7B%22source%22%3A%22bottom_guide_first%22%2C%22ladderRedPackGuide%22%3Afalse%7D" ) REPORT_QUERY = ( "sigCatVer=1&eventTrackingLogInfo=%7B%22deliverOrderId%22%3A%225118%22%2C%22eventTrackingTaskId%22" "%3A20022%2C%22resourceId%22%3A%22externalFeed_externalSignPopup%22%2C%22extParams%22%3A%7B%22signIn" "ThresholdTaskType%22%3A0%2C%22autoSignInSuccess%22%3Afalse%2C%22source%22%3A%22bottom_guide_first%22" "%7D%2C%22resourceFirstLevelName%22%3A%22EXTERNAL_FEED%22%2C%22resourceSecondLevelName%22%3A" "%22EXTERNAL_SIGN_POPUP%22%7D" ) TREASURE_INFO_QUERY = "sigCatVer=1&source=bottom_guide_first" # ---- 真实 sig3 样本(摘要字段交叉验证) ---- CAPTURED_SIG3 = { "show_popup": "4151162679c8e853bd1d741e19181571a0bed5eefa41a30358df0e0e08080b0a3515", "sign_in_report": "9d8dcafaa514348f61c1a9c2c5c474fbd628113d269d7fdf1771d2d2d4d4d7d6e9c9", "treasure_info": "5b4b0c3c63d2f249a70780040302289a4b2effd11e58b9193b6f1414121211102f0f", "treasure_report": "8797d0e0bf0e2e957bdb5ed8dfde440eb392cd11c28465c56be7c8c8cececdccf3d3", "matrix_treasure": "9181c6f6a91838836dcd4fcec9c883c4a4d83813d49273d389d3deded8d8dbdae5c5", "matrix_task": "d0c087b7e85979c22c8c2c8f8889dcd94f6e33ba97d332924b7d9f9f99999a9ba484", } TREASURE_TRACKING = { "eventTrackingTaskId": 20035, "resourceId": "externalFeed_externalTreasureChestWidget", "extParams": {"isServerRecordClickAction": True}, "deliverOrderId": "5128", "resourceFirstLevelName": "EXTERNAL_FEED", "resourceSecondLevelName": "EXTERNAL_TREASURE_CHEST_WIDGET", } TASKLIST_TREASURE_SLOT = { "eventTrackingTaskId": 20022, "resourceId": "externalFeed_externalTreasureChestWidget", "extParams": {"isServerRecordClickAction": True}, "deliverOrderId": "5128", "resourceFirstLevelName": "EXTERNAL_FEED", "resourceSecondLevelName": "EXTERNAL_TREASURE_CHEST_WIDGET", } TASKLIST_AD_SLOT = { "deliverOrderId": "422", "materialKey": "TASK_LIST_17_672_CD_PROGRESSING", "eventTrackingTaskId": 17, "resourceId": "earnPage_taskList_1", "extParams": { "isServerRecordClickAction": True, "businessPriceUniqueId": "EPuAgIDgi5CfChiuBCDBxpyQhDQo2O7EtgM=", }, "resourceFirstLevelName": "EARN_PAGE", "resourceSecondLevelName": "TASK_LIST", } CAPTURED_OAID = "700ED68650EF49ECB686B2AAA9994C1Dca31c779ee5acb7d4c3b03337599a591" GOLDEN_TREASURE_BODY = ( '{"imei":"","idfa":"","oaid":"' + CAPTURED_OAID + '","eventTrackingLogInfo":"{\\"eventTrackingTaskId\\":20035,' '\\"resourceId\\":\\"externalFeed_externalTreasureChestWidget\\",\\"extParams\\":{\\"isServerRecordClickAction\\":true,' '\\"source\\":\\"bottom_guide_first\\"},\\"deliverOrderId\\":\\"5128\\",\\"resourceFirstLevelName\\":\\"EXTERNAL_FEED\\",' '\\"resourceSecondLevelName\\":\\"EXTERNAL_TREASURE_CHEST_WIDGET\\"}"}' ) GOLDEN_MATRIX_TREASURE_BODY = ( '{"actionType":1,"resourceSlotInfo":{"eventTrackingTaskId":20022,' '"resourceId":"externalFeed_externalTreasureChestWidget","extParams":{"isServerRecordClickAction":true,' '"threshold":0,"thresholdTaskId":0,"source":"bottom_guide_first"},"deliverOrderId":"5128",' '"resourceFirstLevelName":"EXTERNAL_FEED","resourceSecondLevelName":"EXTERNAL_TREASURE_CHEST_WIDGET"}}' ) GOLDEN_MATRIX_AD_BODY = ( '{"actionType":1,"resourceSlotInfo":{"deliverOrderId":"422","materialKey":"TASK_LIST_17_672_CD_PROGRESSING",' '"eventTrackingTaskId":17,"resourceId":"earnPage_taskList_1","extParams":{"isServerRecordClickAction":true,' '"businessPriceUniqueId":"EPuAgIDgi5CfChiuBCDBxpyQhDQo2O7EtgM=","source":"bottom_guide_first"},' '"resourceFirstLevelName":"EARN_PAGE","resourceSecondLevelName":"TASK_LIST"}}' ) def make_client() -> KsJsB: return KsJsB(CAPTURED_COOKIE, "", timeout=5, dry_run=True) def query_without_sig3(url: str) -> str: return "&".join(p for p in url.split("?", 1)[1].split("&") if not p.startswith("__NS_sig3=")) class UrlAndBodyTests(unittest.TestCase): def test_show_popup_query_matches_capture(self): self.assertEqual(query_without_sig3(make_client().sign_in_show_popup().url), SHOW_POPUP_QUERY) def test_sign_in_report_query_matches_capture(self): tracking = { "deliverOrderId": "5118", "eventTrackingTaskId": 20022, "resourceId": "externalFeed_externalSignPopup", "extParams": {"signInThresholdTaskType": 0, "autoSignInSuccess": False}, "resourceFirstLevelName": "EXTERNAL_FEED", "resourceSecondLevelName": "EXTERNAL_SIGN_POPUP", } self.assertEqual(query_without_sig3(make_client().sign_in_report(tracking).url), REPORT_QUERY) def test_treasure_info_query_matches_capture(self): self.assertEqual(query_without_sig3(make_client().treasure_box_info().url), TREASURE_INFO_QUERY) def test_treasure_report_body_matches_capture(self): self.assertEqual( compact_json({ "imei": "", "idfa": "", "oaid": CAPTURED_OAID, "eventTrackingLogInfo": compact_json(with_source(TREASURE_TRACKING)), }), GOLDEN_TREASURE_BODY, ) record = make_client().treasure_box_report(TREASURE_TRACKING) self.assertEqual(query_without_sig3(record.url), "sigCatVer=1") def test_matrix_treasure_body_matches_capture(self): resource = dict(TASKLIST_TREASURE_SLOT) ext = dict(resource["extParams"]) ext.setdefault("threshold", 0) ext.setdefault("thresholdTaskId", 0) resource["extParams"] = ext self.assertEqual(compact_json({"actionType": 1, "resourceSlotInfo": with_source(resource)}), GOLDEN_MATRIX_TREASURE_BODY) def test_matrix_ad_body_matches_capture(self): self.assertEqual(compact_json({"actionType": 1, "resourceSlotInfo": with_source(TASKLIST_AD_SLOT)}), GOLDEN_MATRIX_AD_BODY) class Sig3CrossValidationTests(unittest.TestCase): """抓包 query/body + cookie 重建 sign_input, 摘要字段与真实 sig3 内嵌值一致。""" def test_captured_digest_fields(self): cases = { "show_popup": (dict(urllib.parse.parse_qsl(SHOW_POPUP_QUERY, keep_blank_values=True)), None, "GET"), "sign_in_report": (dict(urllib.parse.parse_qsl(REPORT_QUERY, keep_blank_values=True)), None, "GET"), "treasure_info": (dict(urllib.parse.parse_qsl(TREASURE_INFO_QUERY, keep_blank_values=True)), None, "GET"), "treasure_report": ({"sigCatVer": "1"}, GOLDEN_TREASURE_BODY, "POST"), "matrix_treasure": ({"sigCatVer": "1"}, GOLDEN_MATRIX_TREASURE_BODY, "POST"), "matrix_task": ({"sigCatVer": "1"}, GOLDEN_MATRIX_AD_BODY, "POST"), } cookie = dict(part.split("=", 1) for part in CAPTURED_COOKIE.split("; ")) for label, (query, body, method) in cases.items(): with self.subTest(label=label): sign_input = build_h5_sign_input(cookie, query=query, body=body, method=method, request_type="json") calc = h5_sig3_digest_field(sign_input) captured = parse_h5_sig3(CAPTURED_SIG3[label])["digest_field"] self.assertEqual(calc, captured) def test_generated_sig3_roundtrip_and_counter_increments(self): client = make_client() query = dict(urllib.parse.parse_qsl(SHOW_POPUP_QUERY, keep_blank_values=True)) s1 = client._h5_sig3(query=query) s2 = client._h5_sig3(query=query) f1, f2 = parse_h5_sig3(s1), parse_h5_sig3(s2) cookie = dict(part.split("=", 1) for part in CAPTURED_COOKIE.split("; ")) expected = h5_sig3_digest_field(build_h5_sign_input(cookie, query=query)) self.assertEqual(f1["digest_field"], expected) self.assertEqual(f2["digest_field"], expected) # 2026-08-27 抓包: counter 进程级单调递增(101->162) self.assertEqual(f2["counter"], f1["counter"] + 1) self.assertEqual(len(s1), 68) class AlgorithmGoldenTests(unittest.TestCase): """内联算法黄金值(与已逆向 core 实现逐位一致的固化输出)。""" def test_kwsg_10418_digest(self): from ksjsb import KWSG_10418_SIG3_HMAC_KEY, KWSG_10418_SIG3_PREFIX_CODE, _kwsg_tables, kwsg_10418_digest24, kwsg_10418_prefix8 t1, t2 = _kwsg_tables("B") got = kwsg_10418_digest24( "test-input-中文-10418".encode(), 7, 1787813423, t1, t2, prefix8=kwsg_10418_prefix8(KWSG_10418_SIG3_PREFIX_CODE, 0x0610FDA7), hmac_key=KWSG_10418_SIG3_HMAC_KEY, ).hex() self.assertEqual(got, "56473214b4ef0116181e1d1cecfaa8cc28d88a6e030f0117") def test_kwsg_10400_raw(self): from ksjsb import REWARD_SDK, _kwsg_tables, kwsg_10400_raw t1, t2 = _kwsg_tables("A") got = kwsg_10400_raw('{"x":"encData-黄金值"}'.encode(), REWARD_SDK, t1, t2, epoch_seconds=1787813423).hex() self.assertEqual( got, "5a54eecde4d4ea6193f79db96e3254547f6c770401437d63823030fad0f3d66671d4550214754963c06e257b78857c73a4fff39f4b37ea8996c1b0878f69cef97898e2bebf872583", ) def test_xfalcon_value(self): from ksjsb import xfalcon_value_from_input_bytes value = xfalcon_value_from_input_bytes(b"/rest/e/reward/mixed/adclient_key=2ac2a76dos=android") self.assertTrue(value.startswith("HUDR_sFnX+n5uAUNVsMPNK3DOP5wnti1Lc8Axjy5z88T61A==$TE_")) self.assertEqual( value.split("$TE_")[1], "1d029bfdfd210e0c6c7356210e0c6c7356575154566189cefaec418425e348f87e0fdf570c00688c2d0068ad56", ) def test_kww_generation(self): from ksjsb import kwf_generate_kww kww = kwf_generate_kww(3, 1787813423000) self.assertTrue(kww.startswith("PnGU+9+Y8008")) self.assertEqual( kww, "PnGU+9+Y8008S+nH0U+0mjPf8fP08f+98f+nLlwnrIP9Pl+9rh+/WA+/mD+9PAPAHh+/Z7G/SD8e+DPBLh8fzS8/rU8/SYGAPAP/GMPBr7+0pSGn8DweDUGAPAPepDG/qAG9+0w/LlG/40+Bz0+nzDweLM+0P7GfzDPe4jG0LA+/P=", ) def test_weapon_kas(self): from ksjsb import weapon_proof_headers headers = weapon_proof_headers("https://api.e.kuaishou.com/rest/e/reward/mixed/ad?__NS_sig3=abcd", "kaw-value") self.assertEqual(headers["kas"], "00418873e6dae68db455e61dca1fdfd426") def test_signed_api_url_sig3(self): from ksjsb import Kwsg10418State, signed_api_url state = Kwsg10418State(session_seed=0x0610FDA7, counter=0x60) url = signed_api_url( "/rest/e/reward/mixed/ad", {"client_key": "2ac2a76d", "os": "android", "ud": "5402308233"}, [("encData", "AAA"), ("kuaishou.api_st", "ST")], "salt", state, unix_time=1787813423, ) self.assertIn("sig=4df442297f332ac4bce562f7b720229c", url) self.assertIn("__NS_sig3=8d9ce9cf6f34dacda5c5c6c70c09b185f30351b5d8d4dacc", url) self.assertIn("__NStokensig=7e2cf6b51f8e8b86ea855ee1ed3e0be550eba81d5ef7effc76a54216fb65f2fa", url) self.assertIn("__NS_xfalcon=", url) class ApiParamSetTests(unittest.TestCase): """API query 参数集与真机抓包逐键一致(2026-08-27 在线 6001 排查结论)。""" CAPTURED_AD_QUERY_KEYS = [ "earphoneMode", "mod", "appver", "isp", "language", "ud", "did_tag", "egid", "thermal", "net", "kcv", "app", "kpf", "bottom_navigation", "ver", "android_os", "oDid", "boardPlatform", "kpn", "newOc", "androidApiLevel", "slh", "country_code", "nbh", "hotfix_ver", "did_gt", "keyconfig_state", "cdid_tag", "sys", "max_memory", "cold_launch_time_ms", "oc", "sh", "deviceBit", "browseType", "ddpi", "socName", "is_background", "c", "sw", "ftt", "apptype", "abi", "userRecoBit", "device_abi", "icaver", "totalMemory", "grant_browse_type", "iuid", "rdid", "sbh", "darkMode", "did", ] def test_ad_fetch_query_keys_match_capture(self): client = make_client() params = client._api_params() # 真机广告请求: 上述设备键 + sig/__NS_sig3/__NS_xfalcon/__NStokensig self.assertNotIn("sid", params) # 真机广告请求从不携带(2026-07 抓包实证) self.assertNotIn("deviceName", params) expected = set(self.CAPTURED_AD_QUERY_KEYS) self.assertTrue(expected.issubset(params), f"缺失: {expected - set(params)}") def test_uqatag_default_is_device_neutral_null_variant(self): """uQaTag 默认用 QoE 未采集回退形态(2026-07-08 抓包 result=1 接受)。""" from ksjsb import API_COMMON_BODY self.assertEqual(API_COMMON_BODY["uQaTag"], "1##DP:null#ecBl:00#ecPp:--#cmNt:-1#cmMnsl:-0") client = make_client() pairs = dict(client._api_body_common()) self.assertEqual(pairs["uQaTag"], API_COMMON_BODY["uQaTag"]) # 字段序对齐真机: cs, client_key, videoModelCrowdTag, os, api_st, uQaTag, token self.assertEqual( [k for k, _ in client._api_body_common()], ["cs", "client_key", "videoModelCrowdTag", "os", "kuaishou.api_st", "uQaTag", "token"], ) class WithSourceTests(unittest.TestCase): def test_appends_source_and_preserves_key_order(self): tracking = dict(TREASURE_TRACKING) out = with_source(tracking) self.assertEqual(list(out.keys()), list(tracking.keys())) self.assertEqual(list(out["extParams"].keys()), ["isServerRecordClickAction", "source"]) self.assertNotIn("source", tracking["extParams"]) class TaskSlotTests(unittest.TestCase): def test_update_ad_contexts_extracts_slots_and_scene(self): import base64 as b64 link_url = b64.b64encode( compact_json({"pageId": 11101, "subPageId": 100026367, "posId": 24067, "businessId": 672}).encode() ).decode() client = make_client() client._update_ad_contexts({ "data": { "dailyTasks": [ {"eventTrackingLogInfo": TASKLIST_AD_SLOT, "linkUrl": link_url, "extParam": {"descCountDown": 30000}}, {"extParam": {"eventTrackingLogInfo": TASKLIST_TREASURE_SLOT}}, ] } }) self.assertEqual(client.task_ad_slot, TASKLIST_AD_SLOT) self.assertEqual(client.task_treasure_slot, TASKLIST_TREASURE_SLOT) self.assertEqual(client.ad_fetch_scenes["normal"], (11101, 100026367, 24067)) self.assertEqual(client.ad_task_countdown_ms["normal"], 30000) def test_update_ad_contexts_extracts_hf_and_dsp_tasks(self): import base64 as b64 goods_link = b64.b64encode( compact_json({"pageId": 11101, "subPageId": 100074584, "posId": 96134, "businessId": 7038}).encode() ).decode() drama_link = b64.b64encode( compact_json({"pageId": 11101, "subPageId": 100134702, "posId": 175158, "businessId": 7052}).encode() ).decode() walk_slot = {"materialKey": "TASK_LIST_20048_WALKING_NOT_REWARDED", "eventTrackingTaskId": 20048} goods_slot = {"materialKey": "TASK_LIST_60151_PROGRESS", "eventTrackingTaskId": 60151} search_slot = {"materialKey": "TASK_LIST_176_ACTIVE_PASSIVE_TASK_UNFINISH", "eventTrackingTaskId": 176} dsp_slot = {"materialKey": "60494_TASK_LIST_EXCHANGE_DSP_1", "eventTrackingTaskId": 60494} dsp_task = { "name": "打开千问必得金币", "desc": "千问智能解答，高效省时间", "link": {"linkUrl": "tongyi://deeplink?x=1"}, "eventTrackingLogInfo": dsp_slot, "extParam": { "reportUrl": "/rest/n/nebula/exchange/dsp/report", "adId": "2162", "appIdentifier": "com.aliyun.tongyi", "downloadUrl": "https://aisite.wejianzhan.com/site/x", "reqId": "dsctREQ", "impId": "imp-1", "bidId": None, "adBidType": 1, "dspProvider": "happygame", "implTrackersUrl": ["http://t/1"], "downEndTrackersUrl": None, "dplkTrackersUrl": None, }, } client = make_client() client._update_ad_contexts({ "data": { "dailyTasks": [ {"eventTrackingLogInfo": walk_slot}, # linkType=101 + pageId=11101 + businessId 7038 -> goods 场景(搜好货) {"eventTrackingLogInfo": goods_slot, "linkUrl": goods_link, "linkType": 101, "name": "搜好货看广告"}, # 后出现的同形态任务(搜短剧 7052)覆盖 goods 场景(取最新) {"eventTrackingLogInfo": search_slot, "linkType": 101, "linkUrl": drama_link, "name": "搜短剧看广告"}, dsp_task, ] } }) self.assertEqual(client.hf_task_slots["TASK_LIST_20048_"], walk_slot) self.assertEqual(client.hf_task_slots["TASK_LIST_60151_"], goods_slot) self.assertEqual(client.hf_task_slots["TASK_LIST_176_"], search_slot) self.assertEqual(client.ad_fetch_scenes["goods"], (11101, 100134702, 175158)) self.assertEqual(client.ad_fetch_business["goods"], 7052) self.assertEqual(client.goods_task_name, "搜短剧看广告") self.assertEqual(len(client.dsp_tasks), 1) self.assertEqual(client.dsp_tasks[0]["task"], dsp_task) def test_dsp_body_assembly(self): """admInfo 拼装规则与 2026-08-27 抓包映射一致。""" entry = { "task": { "name": "打开千问必得金币", "desc": "千问智能解答，高效省时间", "link": {"linkUrl": "tongyi://www.tongyi.cn/x?src_pkg=zslytylhpl"}, }, "ext": { "adId": "2162", "appIdentifier": "com.aliyun.tongyi", "downloadUrl": "https://aisite.wejianzhan.com/site/y", "reqId": "dsctREQ", "impId": "imp-1", "bidId": None, "adBidType": 1, "dspProvider": "happygame", "implTrackersUrl": ["http://t/1"], "downEndTrackersUrl": None, "dplkTrackersUrl": None, }, } body = KsJsB.build_dsp_body(entry) self.assertEqual( list(body.keys()), ["reqId", "bidId", "impId", "admInfo", "adBidType", "dspProvider", "reportType", "downEndTrackersUrl", "dplkTrackersUrl", "implTrackersUrl"], ) self.assertEqual(body["reportType"], 1) self.assertEqual( body["admInfo"], { "adId": "2162", "title": "打开千问必得金币", "subTitle": "千问智能解答，高效省时间", "dpLink": "tongyi://www.tongyi.cn/x?src_pkg=zslytylhpl", "h5Link": "https://aisite.wejianzhan.com/site/y", "appIdentifier": "com.aliyun.tongyi", }, ) class SequentialRunTests(unittest.TestCase): def test_run_dry_run_completes_all_steps_without_network(self): from ksjsb import run env = {"ksck": f"test#{CAPTURED_COOKIE}#salt", "KS_DRY_RUN": "1", "KS_TIMEOUT": "5"} buf = io.StringIO() old = {k: os.environ.get(k) for k in env} os.environ.update(env) try: with contextlib.redirect_stdout(buf): code = run() finally: for k, v in old.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v self.assertEqual(code, 0) text = buf.getvalue() for marker in ( "1/7 查询金币/余额", "3/7 签到", "4/7 开宝箱", "5/7 高频任务", "6/7 看广告得金币", "7/7 搜好货看广告", # dry-run 下无任务列表, 标签为默认名 "最终余额", "任务执行完成", ): self.assertIn(marker, text) self.assertIn("开宝箱 dry-run method=POST", text) self.assertIn("/rest/e/reward/mixed/ad", text) self.assertIn("/rest/r/ad/task/report", text) self.assertIn("搜好货广告拉取", text) if __name__ == "__main__": unittest.main()