import base64 import json import tempfile import unittest from pathlib import Path from tools.extract_passport_wcfg import extract_wcfg_evidence, extract_wcfg_evidence_many, update_app_fields def _final_passport() -> str: raw = bytes.fromhex("203868e844b7900a75ef44ce0702671f") + b"ticket-body" return "VIMG_" + base64.b64encode(raw).decode("ascii") + "$AI_" + ("1" * 32) class ExtractPassportWcfgTests(unittest.TestCase): def test_extract_wcfg_evidence_selects_a_y_q_z_write(self): passport = _final_passport() events = [ {"tag": "SCRIPT_START", "seq": 1}, { "tag": "OKHTTP_BUILDER_URL", "seq": 2, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", }, { "tag": "WEAPON_B_UPLOAD_ENCRYPT", "seq": 3, "input": "UPLOAD_JSON", "value": "BASE64_UPLOAD_VALUE", }, { "tag": "EDITOR_PUT_STRING_IMPL", "seq": 4, "key": "a_y_q_z", "value": passport[:40] + "...", "value_full": passport, "value_len": len(passport), "stack": "java.lang.Exception\n\tat com.kuaishou.weapon.ks.z0.a(kSourceFile:278)", }, { "tag": "WEAPON_DD", "seq": 5, "type": 21, "value": passport[:40] + "...", "value_full": passport, "value_len": len(passport), }, ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_wcfg.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertTrue(result["chain_complete"]) self.assertEqual(result["passport_account_image"], passport) self.assertTrue(result["flags"]["has_fap_call"]) self.assertTrue(result["flags"]["has_upload_encrypt"]) self.assertTrue(result["flags"]["has_a_y_q_z_write"]) self.assertTrue(result["flags"]["has_dd21_read"]) self.assertEqual(result["selected"]["tag"], "EDITOR_PUT_STRING_IMPL") self.assertEqual(result["selected"]["diagnosis"]["format_kind"], "weapon_pr") self.assertEqual(result["app_fields_patch"]["passport_account_image"], passport) self.assertEqual(result["app_fields_patch"]["request_passport_account_image"], passport) self.assertEqual(result["app_fields_patch"]["checker_passport_account_image"], passport) self.assertGreaterEqual(len(result["timeline"]), 4) def test_extract_wcfg_evidence_marks_truncated_value_as_non_reusable(self): events = [ { "tag": "EDITOR_PUT_STRING_IMPL", "seq": 1, "key": "a_y_q_z", "value": "VIMG_abc...(len=1129)", } ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_wcfg.log" log.write_text("@@WCFG " + json.dumps(events[0]), encoding="utf-8") result = extract_wcfg_evidence(log) self.assertFalse(result["chain_complete"]) self.assertEqual(result["passport_account_image"], "") self.assertEqual(result["app_fields_patch"], {}) self.assertTrue(result["candidates"][0]["truncated"]) def test_extract_wcfg_evidence_can_select_fap_response_body_ticket(self): """当写入 wcfg 的 hook 没打到时，也要能从 /f/a/p 响应体提票据。""" passport = _final_passport() events = [ { "tag": "OKHTTP_REQUEST_BUILD", "seq": 1, "method": "POST", "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", "body": '{"data":"VIMG_UPLOAD_VALUE"}', }, { "tag": "OKHTTP_RESPONSE_BODY_STRING", "seq": 2, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", "body": json.dumps({"result": 1, "a_y_q_z": passport}, ensure_ascii=False), "body_full": json.dumps({"result": 1, "a_y_q_z": passport}, ensure_ascii=False), }, ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_wcfg.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertFalse(result["chain_complete"]) self.assertTrue(result["flags"]["has_fap_call"]) self.assertTrue(result["flags"]["has_fap_response_body"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "OKHTTP_RESPONSE_BODY_STRING") self.assertEqual(result["selected"]["source"], "body") self.assertEqual(result["app_fields_patch"]["passport_account_image"], passport) def test_extract_wcfg_evidence_can_select_weapon_i_return_ticket(self): """deep 脚本抓到 i.a(k1) 返回体时，也能提取 /f/a/p 下发票据。""" passport = _final_passport() ret_body = json.dumps({"result": 1, "a_y_q_z": passport}, ensure_ascii=False) events = [ { "tag": "I_A_K1_BEFORE", "seq": 1, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", "body": '{"data":"VIMG_UPLOAD_VALUE"}', "body_full": '{"data":"VIMG_UPLOAD_VALUE"}', }, { "tag": "I_A_K1_AFTER", "seq": 2, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", "ret": ret_body, "ret_full": ret_body, }, ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_deep_i.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertTrue(result["flags"]["has_fap_call"]) self.assertTrue(result["flags"]["has_fap_request_body"]) self.assertTrue(result["flags"]["has_fap_response_body"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "I_A_K1_AFTER") self.assertEqual(result["selected"]["source"], "ret") def test_extract_wcfg_evidence_can_select_weapon_y0_return_ticket(self): """deep 脚本抓到 y0.a(k1) 降级通道返回体时，也能提取票据。""" passport = _final_passport() ret_body = json.dumps({"result": 1, "data": {"a_y_q_z": passport}}, ensure_ascii=False) events = [ { "tag": "Y0_A_K1_AFTER", "seq": 1, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", "ret": ret_body, "ret_full": ret_body, } ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_deep_y0.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertTrue(result["flags"]["has_fap_call"]) self.assertTrue(result["flags"]["has_fap_response_body"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "Y0_A_K1_AFTER") self.assertEqual(result["selected"]["source"], "ret") def test_extract_wcfg_evidence_can_select_passport_form_add_ticket(self): """最小抓证脚本只抓到 FormBody.add 时，也要能提取最终票据。""" passport = _final_passport() events = [ { "tag": "OKHTTP_FORM_ADD", "seq": 1, "name": "passport_account_image", "value": passport[:40] + "...", "value_full": passport, "value_len": len(passport), } ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_min.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertFalse(result["chain_complete"]) self.assertTrue(result["flags"]["has_passport_form_add"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "OKHTTP_FORM_ADD") self.assertEqual(result["selected"]["name"], "passport_account_image") self.assertEqual(result["app_fields_patch"]["passport_account_image"], passport) def test_extract_wcfg_evidence_can_select_okhttp2_form_add_ticket(self): """兼容旧 OkHttp2 FormEncodingBuilder.add 的表单字段抓证。""" passport = _final_passport() events = [ { "tag": "OKHTTP2_FORM_ADD", "seq": 1, "name": "passport_account_image", "value": passport, "value_full": passport, "value_len": len(passport), } ] with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_passport_min_okhttp2.log" log.write_text( "\n".join("@@WCFG " + json.dumps(event, ensure_ascii=False) for event in events), encoding="utf-8", ) result = extract_wcfg_evidence(log) self.assertFalse(result["chain_complete"]) self.assertTrue(result["flags"]["has_passport_form_add"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "OKHTTP2_FORM_ADD") self.assertEqual(result["app_fields_patch"]["passport_account_image"], passport) def test_extract_wcfg_evidence_many_merges_multi_process_logs(self): """multi-attach 每个进程一个日志时，要能聚合后选出完整票据。""" passport = _final_passport() with tempfile.TemporaryDirectory() as td: main_log = Path(td) / "probe_multi_main.log" worker_log = Path(td) / "probe_multi_worker.log" worker_log.write_text( "\n".join( [ "@@WCFG " + json.dumps( { "tag": "PROCESS_NAME", "seq": 1, "process_name": "com.kuaishou.nebula:messagesdk", }, ensure_ascii=False, ), "@@WCFG " + json.dumps( { "tag": "PROCESS_SKIP", "seq": 2, "process_name": "com.kuaishou.nebula:messagesdk", "reason": "main_only", }, ensure_ascii=False, ), ] ), encoding="utf-8", ) main_log.write_text( "\n".join( "@@WCFG " + json.dumps(event, ensure_ascii=False) for event in [ { "tag": "PROCESS_NAME", "seq": 1, "process_name": "com.kuaishou.nebula", }, { "tag": "OKHTTP_REQUEST_BUILD", "seq": 2, "url": "https://gdfp.gifshow.com/f/a/p?appkey=20001", }, { "tag": "EDITOR_PUT_STRING_IMPL", "seq": 3, "key": "a_y_q_z", "value": passport, "value_full": passport, }, { "tag": "WEAPON_DD", "seq": 4, "type": 21, "value": passport, "value_full": passport, }, ] ), encoding="utf-8", ) result = extract_wcfg_evidence_many([worker_log, main_log]) self.assertTrue(result["chain_complete"]) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["source_log"], "") self.assertEqual(len(result["source_logs"]), 2) self.assertIn("com.kuaishou.nebula", result["process_names"]) self.assertIn("com.kuaishou.nebula:messagesdk", result["process_names"]) self.assertEqual(result["selected"]["source_log"], str(main_log)) self.assertEqual(result["app_fields_patch"]["checker_passport_account_image"], passport) def test_extract_wcfg_evidence_reads_mixed_utf8_utf16le_job_log(self): """PowerShell Start-Job/Tee 可能写出 UTF-8 头 + UTF-16LE Frida 行。""" passport = _final_passport() event = { "tag": "EDITOR_PUT_STRING_IMPL", "seq": 1, "key": "a_y_q_z", "value": passport, "value_full": passport, } line = "@@WCFG " + json.dumps(event, ensure_ascii=False) + "\r\n" with tempfile.TemporaryDirectory() as td: log = Path(td) / "probe_multi_mixed.log" log.write_bytes(b"\xef\xbb\xbfJOB_BEGIN pid=123\r\n" + line.encode("utf-16-le")) result = extract_wcfg_evidence(log) self.assertEqual(result["event_count"], 1) self.assertEqual(result["passport_account_image"], passport) self.assertEqual(result["selected"]["tag"], "EDITOR_PUT_STRING_IMPL") def test_update_app_fields_writes_selected_passport_fields(self): passport = _final_passport() with tempfile.TemporaryDirectory() as td: fields = Path(td) / "app_fields.json" fields.write_text(json.dumps({"host": "az2-api.ksapisrv.com"}), encoding="utf-8") updated = update_app_fields( fields, { "passport_account_image": passport, "request_passport_account_image": passport, "checker_passport_account_image": passport, }, ) saved = json.loads(fields.read_text(encoding="utf-8")) self.assertEqual(updated["passport_account_image"], passport) self.assertEqual(saved["checker_passport_account_image"], passport) self.assertEqual(saved["host"], "az2-api.ksapisrv.com") if __name__ == "__main__": unittest.main()