369 lines
15 KiB
Python
369 lines
15 KiB
Python
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"], "<multi:2 logs>")
|
|
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()
|